---
status: draft
last_updated: 2026-04-30
---

# API Surface

Core pubsub creation, types, and operators. No transport dependencies.

## `createPubSub`

```ts
function createPubSub<TPubSubPublishArgsByKey extends PubSubPublishArgsByKey>(
  config?: PubSubConfig<TPubSubPublishArgsByKey>,
): PubSub<TPubSubPublishArgsByKey>;
```

Factory function. Accepts an optional `eventTarget` config. If none is provided, uses `new EventTarget()` (in-process).

### Topic Scoping

Topics can be scoped with an id:

- `pubsub.publish("session.status", projectId, payload)` → dispatches to topic `session.status:{projectId}`
- `pubsub.subscribe("session.status", projectId)` → subscribes to topic `session.status:{projectId}` only
- `pubsub.publish("session.status", payload)` → dispatches to topic `session.status` (unscoped)
- `pubsub.subscribe("session.status")` → subscribes to topic `session.status` (unscoped)

The topic string is either the routing key directly (unscoped) or `{routingKey}:{id}` (scoped). This maps naturally to Redis channel naming and WebSocket message routing.

### `PubSubPublishArgsByKey`

The type parameter that defines the event map:

```ts
type PubSubPublishArgsByKey = {
  [key: string]: [] | [unknown] | [number | string, unknown];
};
```

- `[]` — event with no payload (trigger only)
- `[payload]` — unscoped event with payload
- `[id, payload]` — scoped event with id and payload

### `PubSub.subscribe()`

Returns a `Repeater<unknown>` (async iterable). Consumers iterate with `for await`:

```ts
for await (const payload of pubsub.subscribe("session.status")) {
  // handle payload
}
```

The `Repeater` automatically cleans up its `addEventListener` when the consumer breaks out of the loop (the `stop` promise resolves).

## Types

| Export | Source | Description |
|--------|--------|-------------|
| `TypedEvent<TType, TDetail>` | `types.ts` | Event with typed `type` and `detail`. Omits `CustomEvent`'s untyped fields. |
| `TypedEventTarget<TEvent>` | `types.ts` | Extends `EventTarget` with typed `addEventListener`, `dispatchEvent`, `removeEventListener`. |
| `TypedEventListener<TEvent>` | `types.ts` | `(evt: TEvent) => void` |
| `TypedEventListenerObject<TEvent>` | `types.ts` | `{ handleEvent(object: TEvent): void }` |
| `TypedEventListenerOrEventListenerObject<TEvent>` | `types.ts` | Union of the above |
| `PubSub<TPubSubPublishArgsByKey>` | `create_pubsub.ts` | `{ publish, subscribe }` |
| `PubSubConfig<TPubSubPublishArgsByKey>` | `create_pubsub.ts` | `{ eventTarget?: PubSubEventTarget }` |
| `PubSubEvent<TPubSubPublishArgsByKey, TKey>` | `create_pubsub.ts` | Derived `TypedEvent` for a specific event key |
| `PubSubEventTarget<TPubSubPublishArgsByKey>` | `create_pubsub.ts` | `TypedEventTarget<PubSubEvent<...>>` |

## Operators

All operators return `Repeater` instances and work with any async iterable.

### `filter`

```ts
function filter<T>(filterFn: (value: T) => Promise<boolean> | boolean): (source: AsyncIterable<T>) => Repeater<T>;
```

Type-narrowing overload available: `filter<T, U extends T>(fn: (input: T) => input is U)`.

### `map`

```ts
function map<T, O>(mapper: (input: T) => Promise<O> | O): (source: AsyncIterable<T>) => Repeater<O>;
```

### `pipe`

```ts
function pipe<A, B>(a: A, ab: (a: A) => B): B;
function pipe<A, B, C>(a: A, ab: (a: A) => B, bc: (b: B) => C): C;
// up to 5 arguments
```

Compose operators: `pipe(pubsub.subscribe("myEvent"), filter(isRelevant), map(transform))`

## Attribution

`createPubSub` and operators are adapted from `@graphql-yoga/subscription` (MIT). `TypedEventTarget` types are adapted from `@graphql-yoga/typed-event-target` (MIT). See file headers for full license text.