Streaming

A handler answers once. A stream keeps answering: it yields events for as long as the client is listening, over Server-Sent Events. Live lists, notifications, progress on a long job, and anything driven by a publisher are all the same shape.

Swap .handler() for .stream() and write an async generator:

import { ServerError } from 'typebase-io/server';
import { z } from 'zod';

import { action } from '../../_generated/server.ts';

export const progress = action
  .input(z.object({ id: z.number() }))
  .output(z.object({ done: z.number(), total: z.number() }))
  .stream(async function* ({ db, input, signal }) {
    while (signal?.aborted !== true) {
      const job = await db.query.jobs.findFirst({ where: { id: input.id } });

      if (!job) {
        throw new ServerError('NOT_FOUND');
      }

      yield { done: job.done, total: job.total };

      if (job.done === job.total) {
        return;
      }

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

Everything you know from .handler() carries over: .input() validates what comes in, errors are thrown the same way, and you get the whole handler context: db, env, auth, publisher, input, reqHeaders. A stream adds two values a handler has no use for:

lastEventId is undefined on a first connection. signal may be undefined when the client or runtime cannot provide a cancellation signal. A stream that ignores both still works, but it cannot stop work promptly on disconnect or resume missed events.

What .output() means on a stream

On a handler, .output() describes the response. On a stream, it describes one event, and every event is validated against it as you yield:

.output(z.object({ done: z.number(), total: z.number() }))

The client receives an async iterable of that shape. You don't wrap it in anything yourself; .stream() handles that. .output() stays optional, leave it off and the event type is inferred from what you yield, exactly as a handler's return type is inferred:

export const ticks = action.stream(async function* () {
  yield { at: Date.now() };
});

Rules the compiler enforces

Three, all reported at compile time:

A stream must yield. A generator that never yields sends the client nothing, so it's rejected rather than silently connecting to a stream that produces no events:

// Error: a stream must yield at least one event
export const nothing = action.output(shape).stream(async function* () {});

Events are yielded, never returned. A returned value ends the stream without sending anything, which is almost always a mistake:

export const wrong = action.output(shape).stream(async function* () {
  return { done: 1, total: 1 }; // Error
});

export const right = action.output(shape).stream(async function* () {
  yield { done: 1, total: 1 };

  return; // fine: a bare return just ends the stream
});

Every event matches .output(). Yield something else and it's a type error, the same as returning the wrong shape from a handler.

Ending a stream

A stream ends when the generator returns, or when the client goes away. Every stream receives signal: AbortSignal | undefined; when the runtime provides it, the signal fires on disconnect:

export const live = action.stream(async function* ({ publisher, signal }) {
  const created = await publisher.subscribe('todo.created', { signal });

  for await (const todo of created) {
    yield todo;
  }
});

Pass signal to anything that would otherwise keep running: a publisher.subscribe, a fetch, a database call that takes a while. Without it, a client closing the tab leaves your loop running until the platform kills the function.

Cleanup belongs in finally, which runs whether the client disconnected or the generator returned on its own:

export const watched = action.stream(async function* ({ signal }) {
  const watcher = openWatcher();

  try {
    while (signal?.aborted !== true) {
      yield await watcher.next();
    }
  } finally {
    watcher.close();
  }
});

Event metadata

Every event can carry an id and a retry, the two fields SSE defines. Attach them with withEventMeta:

import { withEventMeta } from 'typebase-io/server';

export const live = action.stream(async function* ({ db }) {
  for await (const row of rows(db)) {
    yield withEventMeta({ id: row.id, value: row.value }, { id: String(row.id), retry: 5_000 });
  }
});

withEventMeta returns the same value with the metadata attached, so what the client receives is unchanged. Reading it back is getEventMeta, which returns { id?, retry? } or undefined for an event that was never tagged:

import { getEventMeta, withEventMeta } from 'typebase-io/server';

for await (const todo of created) {
  const meta = getEventMeta(todo);

  yield withEventMeta(todo, { id: meta?.id });
}

A publisher already tags what it delivers with the event's row id, so streams built on publisher.subscribe get ids without doing anything.

Resuming after a disconnect

When a client reconnects, it sends the id of the last event it received. That arrives as lastEventId on the stream's context:

export const live = action.stream(async function* ({ publisher, signal, lastEventId }) {
  const created = await publisher.subscribe('todo.created', { signal, lastEventId });

  for await (const todo of created) {
    yield todo;
  }
});

It's undefined on a first connection and a string on a resume. Handing it to publisher.subscribe replays what was published while the client was away; ignoring it means the client silently misses those events. What you can replay depends on what you kept, which for the db publisher is however much of the events table you haven't pruned.

Consuming a stream

First create or import a client with createRouterClient or createTanstackQueryClient. On the client, a stream resolves to an async iterable. With the router client, call the action directly; with the TanStack client, use .call():

You can for await the resulting async iterable directly. For UI code, consumeStream inverts it into callbacks and hands you back an unsubscribe function. Consumption starts immediately, and every call opens a separate stream.

With createRouterClient

import { consumeStream, createRouterClient } from 'typebase-io/client';
import type { Router } from '../../typebase/_generated/server';

const client = createRouterClient<Router>({
  url: process.env.NEXT_PUBLIC_TYPEBASE_APP_URL || '',
});

const unsubscribe = consumeStream(client.queries.todos.getMany(), {
  onEvent: (todos) => render(todos),
  onError: (error) => console.error(error),
  onSuccess: () => console.log('stream ended'),
});

// When the code that owns this subscription stops
void unsubscribe();

With createTanstackQueryClient

import { consumeStream, createTanstackQueryClient } from 'typebase-io/client';
import type { Router } from '../../typebase/_generated/server';

const client = createTanstackQueryClient<Router>({
  url: process.env.NEXT_PUBLIC_TYPEBASE_APP_URL || '',
});

const unsubscribe = consumeStream(client.queries.todos.getMany.call(), {
  onEvent: (todos) => render(todos),
  onError: (error) => console.error(error),
  onSuccess: () => console.log('stream ended'),
});

// When the code that owns this subscription stops
void unsubscribe();

useEffect(() => {
  const unsubscribe = consumeStream(client.queries.todos.getMany.call(), {
    onEvent: (todos) => setTodos(todos),
    onError: (error) => console.error(error),
  });

  return () => {
    void unsubscribe();
  };
}, []);

Typing an event

A stream's entry in RouterOutputs is the async iterable, not the event. InferStreamEvent unwraps it:

import type { InferStreamEvent } from 'typebase-io/server';
import type { RouterOutputs } from '../typebase/_generated/server';

type Todos = InferStreamEvent<RouterOutputs['queries']['todos']['getMany']>;

This works whether or not the action declares .output(); both report the same type. See Typing action inputs and outputs.

Reconnecting

Add the retry plugin to your client to make dropped streams reconnect, and to make lastEventId reach the server:

import { createRouterClient } from 'typebase-io/client';
import { ClientRetryPlugin } from 'typebase-io/client/plugins';

import type { Router } from '../typebase/_generated/server.ts';

export const client = createRouterClient<Router>({
  url: process.env.NEXT_PUBLIC_TYPEBASE_URL,
  plugins: [new ClientRetryPlugin({ default: { retry: Number.POSITIVE_INFINITY } })],
});

retry is 0 by default, which is why a stream that drops simply stops. The delay between attempts comes from the last event's retry metadata, falling back to two seconds.

Regenerating types

Converting an action in an existing file from .handler() to .stream() does not require codegen; its type flows through the imports already in _generated/server.ts. Run codegen only if you created, removed, or renamed the action file itself:

npx typebase-io-cli codegen

npx typebase-io-cli deploy and generate-server rerun codegen before every build.