Subscriptions

From https://trpc.io/docs/server/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() events.

WebSockets or Server-sent Events?

You can either use WebSockets or Server-sent Events (SSE) to setup real-time subscriptions in tRPC.

For WebSockets, see the WebSockets page
For SSE, see the httpSubscriptionLink

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
SSE	Full-stack SSE implementation	github.com/trpc/examples-tanstack-start-sse-chat
WebSockets	Full-stack WebSockets implementation	github.com/trpc/examples-next-prisma-websockets-starter

Basic example

Tip

For a full example, see our full-stack SSE example.

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)

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 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, this can prevent newly emitted events being ignored while yield'ing the original batch based on lastEventId.

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 (SSE) or wsLink with a splitLink, then call .subscribe() with the input and your callbacks. Tracked events arrive in onData as { id, data }.

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) {
      console.log(event.id, event.data.title);
    },
    onError(err) {
      console.error(err);
    },
  },
);

// stop listening
subscription.unsubscribe();

For React, see subscriptionOptions with TanStack React Query, or the classic useSubscription().

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.

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

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

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 pattern, as trpc invokes the .return() of the Generator Instance when the subscription stops for any reason.

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(). For other errors, the subscription will be cancelled and propagate to the onError() callback.

Output validation

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

Example with Zod v4

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:

_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.
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.

References

WebSockets
HTTP Subscription Link: A terminating link for subscriptions over Server-sent Events
WebSocket Link: A terminating link that sends tRPC operations over a WebSocket

server-setup (https://trpc.io/docs/server/overview): npx @tanstack/intent@latest load @trpc/server#server-setup
links (https://trpc.io/docs/client/links): npx @tanstack/intent@latest load @trpc/client#links

@trpc/server

subscriptions

Subscriptions#

Introduction#

WebSockets or Server-sent Events?#

Reference projects#

Basic example#

Automatic tracking of id using tracked() (recommended)#

Subscribing from the client#

Pull data in a loop#

Stopping a subscription from the server#

Cleanup of side effects#

Error handling#

Output validation#

Example with Zod v4#

Common mistakes#

References#

Related skills#