# Subscriptions

> 

## Introduction

Subscriptions are a type of real-time event stream between the client and server. Use subscriptions when you need to push real-time updates to the client.

With tRPC's subscriptions, the client establishes and maintains a persistent connection to the server plus automatically attempts to reconnect and recover gracefully if disconnected with the help of [`tracked()`](#tracked) events.

## WebSockets or Server-sent Events?

You can either use WebSockets or [Server-sent Events](https://en.wikipedia.org/wiki/Server-sent_events) (SSE) to setup real-time subscriptions in tRPC.

- For WebSockets, see [the WebSockets page](https://trpc.io/docs/server/websockets.md)
- For SSE, see the [httpSubscriptionLink](https://trpc.io/docs/client/links/httpSubscriptionLink.md)

If you are unsure which one to use, we recommend using SSE for subscriptions as it's easier to setup and doesn't require setting up a WebSocket server.

## Reference projects

| Type       | Example Type                            | Link                                                                                                                       |
| ---------- | --------------------------------------- | -------------------------------------------------------------------------------------------------------------------------- |
| WebSockets | Bare-minimum Node.js WebSockets example | [/examples/standalone-server](https://github.com/trpc/trpc/tree/main/examples/standalone-server)                           |
| SSE        | Full-stack SSE implementation           | [github.com/trpc/examples-tanstack-start-sse-chat](https://github.com/trpc/examples-tanstack-start-sse-chat)               |
| WebSockets | Full-stack WebSockets implementation    | [github.com/trpc/examples-next-prisma-websockets-starter](https://github.com/trpc/examples-next-prisma-starter-websockets) |

## Basic example

> [!TIP]
> For a full example, see [our full-stack SSE example](https://github.com/trpc/examples-tanstack-start-sse-chat).


```ts title="server.ts"
import EventEmitter, { on } from 'node:events';
import { initTRPC } from '@trpc/server';

const t = initTRPC.create();

type Post = { id: string; title: string };
const ee = new EventEmitter();

export const appRouter = t.router({
  onPostAdd: t.procedure.subscription(async function* (opts) {
    // listen for new events
    for await (const [data] of on(ee, 'add', {
      // Passing the AbortSignal from the request automatically cancels the event emitter when the request is aborted
      signal: opts.signal,
    })) {
      const post = data as Post;
      yield post;
    }
  }),
});
```

## Automatic tracking of id using `tracked()` (recommended) {#tracked}

If you `yield` an event using our `tracked()`-helper and include an `id`, the client will automatically reconnect when it gets disconnected and send the last known ID.

You can send an initial `lastEventId` when initializing the subscription and it will be automatically updated as the browser receives data.

- For SSE, this is part of the [`EventSource`-spec](https://html.spec.whatwg.org/multipage/server-sent-events.html#the-last-event-id-header) and will be propagated through `lastEventId` in your `.input()`.
- For WebSockets, our `wsLink` will automatically send the last known ID and update it as the browser receives data.

> [!TIP]
> If you're fetching data based on the `lastEventId`, and capturing all events is critical, make sure you set up the event listener before fetching events from your database as is done in [our full-stack SSE example](https://github.com/trpc/examples-tanstack-start-sse-chat), this can prevent newly emitted events being ignored while yield'ing the original batch based on `lastEventId`.


```ts
import EventEmitter, { on } from 'node:events';
import { initTRPC, tracked } from '@trpc/server';
import { z } from 'zod';

class IterableEventEmitter extends EventEmitter {
  toIterable(eventName: string, opts?: { signal?: AbortSignal }) {
    return on(this, eventName, opts);
  }
}

type Post = { id: string; title: string };

const t = initTRPC.create();
const publicProcedure = t.procedure;
const router = t.router;

const ee = new IterableEventEmitter();

export const subRouter = router({
  onPostAdd: publicProcedure
    .input(
      z
        .object({
          // lastEventId is the last event id that the client has received
          // On the first call, it will be whatever was passed in the initial setup
          // If the client reconnects, it will be the last event id that the client received
          lastEventId: z.string().nullish(),
        })
        .optional(),
    )
    .subscription(async function* (opts) {
      // We start by subscribing to the ee so that we don't miss any new events while fetching
      const iterable = ee.toIterable('add', {
        // Passing the AbortSignal from the request automatically cancels the event emitter when the request is aborted
        signal: opts.signal,
      });

      if (opts.input?.lastEventId) {
        // [...] get the posts since the last event id and yield them
        // const items = await db.post.findMany({ ... })
        // for (const item of items) {
        //   yield tracked(item.id, item);
        // }
      }
      // listen for new events from the iterable we set up above
      for await (const [data] of iterable) {
        const post = data as Post;
        // tracking the post id ensures the client can reconnect at any time and get the latest events since this id
        yield tracked(post.id, post);
      }
    }),
});
```

## Subscribing from the client

Route subscriptions to [`httpSubscriptionLink`](https://trpc.io/docs/client/links/httpSubscriptionLink.md) (SSE) or [`wsLink`](https://trpc.io/docs/client/links/wsLink.md) with a [`splitLink`](https://trpc.io/docs/client/links/splitLink.md), then call `.subscribe()` with the input and your callbacks. Tracked events arrive in `onData` as `{ id, data }`.

```ts title="client.ts"
import {
  createTRPCClient,
  httpBatchLink,
  httpSubscriptionLink,
  splitLink,
} from '@trpc/client';
import type { AppRouter } from './server';

const client = createTRPCClient<AppRouter>({
  links: [
    splitLink({
      condition: (op) => op.type === 'subscription',
      true: httpSubscriptionLink({ url: 'http://localhost:3000' }),
      false: httpBatchLink({ url: 'http://localhost:3000' }),
    }),
  ],
});

const subscription = client.onPostAdd.subscribe(
  { lastEventId: null },
  {
    onData(event) {
      // (parameter) event: {
      //     data: Post;
      //     id: string;
      // }
      console.log(event.id, event.data.title);
    },
    onError(err) {
      console.error(err);
    },
  },
);

// stop listening
subscription.unsubscribe();
```

For React, see [`subscriptionOptions`](https://trpc.io/docs/client/tanstack-react-query/usage.md#subscriptionOptions) with TanStack React Query, or the classic [`useSubscription()`](https://trpc.io/docs/client/react/useSubscription.md).

## Pull data in a loop

This recipe is useful when you want to periodically check for new data from a source like a database and push it to the client.

```ts title="server.ts"
import { tracked } from '@trpc/server';
import { z } from 'zod';
import { publicProcedure, router } from './trpc';

export const subRouter = router({
  onPostAdd: publicProcedure
    .input(
      z.object({
        // lastEventId is the last event id that the client has received
        // On the first call, it will be whatever was passed in the initial setup
        // If the client reconnects, it will be the last event id that the client received
        // The id is the createdAt of the post
        lastEventId: z.coerce.date().nullish(),
      }),
    )
    .subscription(async function* (opts) {
      // `opts.signal` is an AbortSignal that will be aborted when the client disconnects.
      let lastEventId = opts.input?.lastEventId ?? null;

      // We use a `while` loop that checks `!opts.signal.aborted`
      while (!opts.signal!.aborted) {
        const posts = await db.post.findMany({
          // If we have a `lastEventId`, we only fetch posts created after it.
          where: lastEventId
            ? {
                createdAt: {
                  gt: lastEventId,
                },
              }
            : undefined,
          orderBy: {
            createdAt: 'asc',
          },
        });

        for (const post of posts) {
          // `tracked` is a helper that sends an `id` with each event.
          // This allows the client to resume from the last received event upon reconnection.
          yield tracked(post.createdAt.toJSON(), post);
          lastEventId = post.createdAt;
        }

        // Wait for a bit before polling again to avoid hammering the database.
        await sleep(1_000);
      }
    }),
});
```

## Stopping a subscription from the server {#stopping-from-server}

If you need to stop a subscription from the server, simply `return` in the generator function.

```ts
export const subRouter = router({
  onPostAdd: publicProcedure
    .input(
      z.object({
        lastEventId: z.coerce.number().min(0).optional(),
      }),
    )
    .subscription(async function* (opts) {
      let index = opts.input.lastEventId ?? 0;
      while (!opts.signal!.aborted) {
        const idx = index++;
        if (idx > 100) {
          // With this, the subscription will stop and the client will disconnect
          return;
        }
        await new Promise((resolve) => setTimeout(resolve, 10));
      }
    }),
});
```

On the client, you just `.unsubscribe()` the subscription.

## Cleanup of side effects

If you need to clean up any side-effects of your subscription you can use the [`try...finally`](https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Global_Objects/Generator/return#using_return_with_try...finally) pattern, as `trpc` invokes the `.return()` of the Generator Instance when the subscription stops for any reason.

```ts
import EventEmitter, { on } from 'events';
import { initTRPC } from '@trpc/server';

type Post = { id: string; title: string };

const t = initTRPC.create();
const publicProcedure = t.procedure;
const router = t.router;

const ee = new EventEmitter();

export const subRouter = router({
  onPostAdd: publicProcedure.subscription(async function* (opts) {
    let timeout: ReturnType<typeof setTimeout> | undefined;
    try {
      for await (const [data] of on(ee, 'add', {
        signal: opts.signal,
      })) {
        timeout = setTimeout(() => console.log('Pretend like this is useful'));
        const post = data as Post;
        yield post;
      }
    } finally {
      if (timeout) clearTimeout(timeout);
    }
  }),
});
```

## Error handling

Throwing an error in a generator function propagates to `trpc`'s `onError()` on the backend.

If the error thrown is a 5xx error, the client will automatically attempt to reconnect based on the last event id that is [tracked using `tracked()`](#tracked). For other errors, the subscription will be cancelled and propagate to the `onError()` callback.

## Output validation {#output-validation}

Since subscriptions are async iterators, you have to go through the iterator to validate the output.

### Example with Zod v4

```ts title="zAsyncIterable.ts"
import type { TrackedEnvelope } from '@trpc/server';
import { isTrackedEnvelope, tracked } from '@trpc/server';
import { z } from 'zod';

function isAsyncIterable<TValue, TReturn = unknown>(
  value: unknown,
): value is AsyncIterable<TValue, TReturn> {
  return !!value && typeof value === 'object' && Symbol.asyncIterator in value;
}
const trackedEnvelopeSchema =
  z.custom<TrackedEnvelope<unknown>>(isTrackedEnvelope);

/**
 * A Zod schema helper designed specifically for validating async iterables. This schema ensures that:
 * 1. The value being validated is an async iterable.
 * 2. Each item yielded by the async iterable conforms to a specified type.
 * 3. The return value of the async iterable, if any, also conforms to a specified type.
 */
export function zAsyncIterable<
  TYieldIn,
  TYieldOut,
  TReturnIn = void,
  TReturnOut = void,
  Tracked extends boolean = false,
>(opts: {
  /**
   * Validate the value yielded by the async generator
   */
  yield: z.ZodType<TYieldOut, TYieldIn>;
  /**
   * Validate the return value of the async generator
   * @remarks not applicable for subscriptions
   */
  return?: z.ZodType<TReturnOut, TReturnIn>;
  /**
   * Whether the yielded values are tracked
   * @remarks only applicable for subscriptions
   */
  tracked?: Tracked;
}) {
  return z
    .custom<
      AsyncIterable<
        Tracked extends true ? TrackedEnvelope<TYieldIn> : TYieldIn,
        TReturnIn
      >
    >((val) => isAsyncIterable(val))
    .transform(async function* (iter) {
      const iterator = iter[Symbol.asyncIterator]();

      try {
        let next;
        while ((next = await iterator.next()) && !next.done) {
          if (opts.tracked) {
            const [id, data] = trackedEnvelopeSchema.parse(next.value);
            yield tracked(id, await opts.yield.parseAsync(data));
            continue;
          }
          yield opts.yield.parseAsync(next.value);
        }
        if (opts.return) {
          return await opts.return.parseAsync(next.value);
        }
        return;
      } finally {
        await iterator.return?.();
      }
    }) as z.ZodType<
    AsyncIterable<
      Tracked extends true ? TrackedEnvelope<TYieldIn> : TYieldIn,
      TReturnIn,
      unknown
    >,
    AsyncIterable<
      Tracked extends true ? TrackedEnvelope<TYieldOut> : TYieldOut,
      TReturnOut,
      unknown
    >
  >;
}
```

Now you can use this helper to validate the output of your subscription procedures:

```ts title="_app.ts"
import { tracked } from '@trpc/server';
import { z } from 'zod';
import { publicProcedure, router } from './trpc';
import { zAsyncIterable } from './zAsyncIterable';

export const appRouter = router({
  mySubscription: publicProcedure
    .input(
      z.object({
        lastEventId: z.coerce.number().min(0).optional(),
      }),
    )
    .output(
      zAsyncIterable({
        yield: z.object({
          count: z.number(),
        }),
        tracked: true,
      }),
    )
    .subscription(async function* (opts) {
      let index = opts.input.lastEventId ?? 0;
      while (true) {
        index++;
        yield tracked(String(index), {
          count: index,
        });

        await new Promise((resolve) => setTimeout(resolve, 1000));
      }
    }),
});
```

## Common mistakes

- **Not passing `opts.signal` on.** Pass it to what you're waiting on, like `on(ee, 'add', { signal: opts.signal })`, or check `opts.signal.aborted` in loops. Otherwise your generator can keep waiting, with its listeners attached, after the client is gone.
- **Fetching missed events before listening for new ones.** Events emitted in between are lost. Start listening first, as in the [`tracked()` example](#tracked).
- **Tracking an event with an empty id.** `tracked('', data)` throws, since an empty id means "no id" in SSE.
- **Returning an `observable()`.** Observable subscriptions are deprecated and will be removed in v12. Use an async generator.
- **Pinging less often than the client reconnects.** With SSE, keep `sse.ping.intervalMs` lower than `sse.client.reconnectAfterInactivityMs`, or the client reconnects between pings. tRPC throws when the ping interval is the longer one. See [Server Ping Configuration](https://trpc.io/docs/client/links/httpSubscriptionLink.md#server-ping).