Building open-source software for the glory of the Lord Jesus Christ ☦️
With love for developers of all backgrounds around the world ❤️

Subscription

Category: Points

A subscription streams values from the server: its loader is an async generator, and every yield travels to the client the moment it happens. The transport is plain HTTP — an NDJSON response on a real endpoint, not a WebSocket — so there are no tickets, no rooms, no socket to keep alive: the cheapest way to watch a long task, tail a log, or follow a feed.

import { root } from '@/lib/root'
import { z } from 'zod'

export const taskProgressSubscription = root.lets
  .subscription()
  .input(z.object({ taskId: z.string() }))
  .loader(async function* ({ input, signal }) {
    // cut from the client bundle — the generator runs on the server
    for await (const percent of watchProgress(input.taskId, { signal })) {
      yield { percent } // one streamed value per yield
    }
  })
  .subscription()

export const TaskProgress = ({ taskId }: { taskId: string }) => {
  const { data, status } = taskProgressSubscription.useSubscription(
    { taskId },
    { lastMessageFromServerAsData: true }, // keep the latest message in `data`
  )
  if (status === 'closed') return <span>Done</span> // the generator returned
  return <progress value={data?.percent ?? 0} max={100} />
}

Declaring a subscription

Open with .subscription(), declare input and a generator loader, close with .subscription(options?):

export const countSubscription = root.lets
  .subscription() // open
  .input(z.object({ upTo: z.number().max(10) }))
  .loader(async function* ({ input }) {
    for (let n = 1; n <= input.upTo; n++) {
      yield { n } // each yield is one streamed value
    }
    // the generator returns → the stream completes, consumers see 'closed'
  })
  .subscription({ reconnect: false }) // close, with point-level options

The loader is an async generator — that's the whole declaration of the data: each yield is one streamed value, a return completes the stream, and the data type is the union of the yields — nothing is declared. A plain (non-generator) loader can't close with .subscription(), and a generator loader can't close with .query() or .mutation() — both are type errors.

Beyond signal (covered below), the generator receives the standard loader surface — the parsed input, any ctx from .ctx/plugins, the request, a set helper, and points. .input(schema) takes any Standard Schema and merges down the chain, exactly as on a query.

A real endpoint

A subscription is served at GET /_point0/<scope>/subscription/<kebab-name> — a GET with the input JSON-encoded in the ?input= search param, same as a query; an input that carries a File/Blob or overflows the URL falls back to a POST body automatically. The response is NDJSON (application/x-ndjson): one transformer-serialized line per yield, closed by a terminal line that says whether the generator completed or threw. While the generator sits quiet between yields, blank-line heartbeats keep idle proxies from reaping the stream. You never write the path — useSubscription / fetchSubscription route to it for you.

The connect is a full request: middleware, plugins, .ctx, and input validation all run, like on any endpoint. An input that fails its schema answers as a plain typed error (status 422) before anything streams; a loader that throws mid-stream delivers the typed error after the values already yielded — see error handling.

Consuming with the hook

useSubscription(input?, options?) holds the stream while the component is mounted. By default messages never re-render the component — react to them through the onMessageFromServer callback:

const { error, status, isLoading } = taskProgressSubscription.useSubscription(
  { taskId },
  { onMessageFromServer: (tick) => setPercent(tick.percent) }, // fires on every message
)

status // 'connecting' | 'open' | 'closed' | 'error'
isLoading // status === 'connecting'

When you want to render the latest message directly, opt into the query-shaped result with lastMessageFromServerAsData: true — the result gains data (typed as the union of the yields) and the component re-renders on every message:

const { data } = taskProgressSubscription.useSubscription(
  { taskId },
  { lastMessageFromServerAsData: true },
)
data // the latest streamed value — undefined until the first one

Pick by what you do with a message: a callback for reactions (update a store, toast, feed a query cache — no render churn), lastMessageFromServerAsData for rendering the latest value (a ticker, a progress bar — the re-render is the point). status walks connecting → open → closed (the generator completed) or error (a typed error — final, never restarted); error is typed by the point's error class, as everywhere. All options together:

taskProgressSubscription.useSubscription(
  { taskId },
  {
    enabled: !!taskId, // false → nothing opens, status is 'closed'
    onMessageFromServer: (tick) => console.info(tick.percent), // fires on every message
    lastMessageFromServerAsData: true, // keep the latest message in `data`; default false
    reconnect: { delay: 1000 }, // per-call override — see Reconnect
  },
)

A listener can also live on the point itself — .subscription({ onMessageFromServer }) runs for every consumer's every message (per-call listeners add up, nothing doubles). It's client code: the compiler cuts it from the server bundle with its imports.

Scope-wide defaults live on .subscriptionOptions({...}) — declarable on root, base, or a plugin, like .queryOptions: the same reconnect and onMessageFromServer shape, applied to every subscription in scope. Options resolve lowest-to-highest — .subscriptionOptions(), then the closing .subscription({...}), then the useSubscription call — and onMessageFromServer listeners at every level all run, in that order (nothing doubles).

Changing the input closes the old stream and opens a new one (the input's serialized form is the identity). Each hook holds its own stream — two components subscribing with the same input are two independent streams, with no cross-component dedup (an HTTP stream is per-reader by nature — the deliberate contrast with channel connections and space memberships, which dedupe by input). The values live in the hook, not in the react-query cache: unmount and they're gone.

During SSR nothing streams — the hook renders connecting on the server and does the real work after mount. Nothing is prefetched or dehydrated.

The imperative consumer: `for await`

fetchSubscription(input?, options?) returns an AsyncIterable of the streamed values:

for await (const { percent } of taskProgressSubscription.fetchSubscription({
  taskId,
})) {
  render(percent)
}
// the loop ends when the generator completes

A typed error — or a broken stream — throws out of the loop; there is no auto-restart (an imperative consumer loops on its own terms). Breaking out of the loop cancels the stream — the server generator's signal fires. To cancel from outside the loop, pass a signal:

const controller = new AbortController()
for await (const value of feedSubscription.fetchSubscription(undefined, {
  signal: controller.signal,
})) {
  // controller.abort() anywhere stops the stream
}

fetchSubscription is client-only — a runtime error on the server.

Stopping the stream — the `signal`

The generator receives signal — an AbortSignal that fires when the consumer unsubscribes (the component unmounts, the for await loop breaks, the tab closes). Hand it to whatever feeds the generator, or check it in the loop — without it, a generator waiting on an external source outlives its consumer forever.

The server pulls its own source — poll, sleep, check the signal:

export const logTailSubscription = root.lets
  .subscription()
  .input(z.object({ file: z.string() }))
  .loader(async function* ({ input, signal }) {
    let offset = 0
    while (!signal.aborted) {
      const { lines, nextOffset } = await readNewLines(input.file, offset)
      offset = nextOffset
      if (lines.length > 0) yield { lines }
      await sleep(1000)
    }
  })
  .subscription()

Outside events drive the stream — on from node:events turns an emitter into an async iterable and takes the signal directly, so a mutation firing an event anywhere on the server streams to every subscribed client, still without a WebSocket:

import { on } from 'node:events'
import { postsEmitter } from '@/lib/posts' // .emit('add', post) in a mutation

export const newPostsSubscription = root.lets
  .subscription()
  .loader(async function* ({ signal }) {
    for await (const [post] of on(postsEmitter, 'add', { signal })) {
      yield { post }
    }
  })
  .subscription()

The teardown is symmetric: when the consumer goes away, the server aborts the signal and closes the generator, so finally blocks in the loader run and nothing leaks.

Reconnect

Only a broken stream restarts — the bytes stopped without a terminal line (a network drop, a server restart, a dev reload), or a line arrived that the transformer cannot parse, which says the same thing: what comes out of this connection is no longer the framing. The first retry fires immediately (a drop is usually accidental); from there the waits run delay, delay·backoff, delay·backoff², … capped at maxDelay — by default 300 ms doubling up to 5 s. The attempt counter resets whenever a stream opens successfully. While reconnecting, status is connecting and data keeps the last streamed value.

A completed stream (the generator returned) and a typed error (the loader threw, the input failed validation) are answers — they never restart.

.subscription({ reconnect: true }) // the default — same as omitting it
.subscription({ reconnect: { immediately: false, delay: 1000 } }) // never retry instantly
.subscription({ reconnect: { backoff: 1 } }) // constant 300ms waits, no exponent
.subscription({ reconnect: { retries: 3 } }) // give up after 3 attempts
.subscription({ reconnect: false }) // a break surfaces as a typed error

The policy lives on the closing .subscription({...}) and every useSubscription call can override it. With reconnect: false (or the retries cap exhausted) a break reports status: 'error' with a typed POINT0_SUBSCRIPTION_LOST error. fetchSubscription never reconnects — a break throws, and the for await loop retries if you wrap it in one.

A reconnect restarts the generator from scratch — whatever streamed during the gap is lost unless the point resumes instead: see Resuming after a break.

Resuming after a break

Declare the tracked-cursor pair on the closing .subscription({...}) and a broken stream resumes where it left off: the client plucks a cursor out of every delivered value, and when the auto-reconnect redials it writes the last one into the input it resubscribes with.

export const feedSubscription = root.lets
  .subscription()
  .input(
    z.object({ channelId: z.string(), lastEventId: z.string().optional() }),
  )
  .loader(async function* ({ input, signal }) {
    // a fresh start and a resume read the cursor the SAME way — out of the input
    for (const missed of await db.eventsAfter(
      input.channelId,
      input.lastEventId,
    )) {
      yield missed // { id, ... }
    }
    for await (const event of live(input.channelId, { signal })) {
      yield event
    }
  })
  .subscription({
    cursorParamFromInput: 'lastEventId', // the input field the reconnect rewrites
    cursorParamFromData: 'id', // where the cursor lives inside each streamed value
  })

cursorParamFromData names the field inside each streamed value the client remembers — the data of a tracked subscription is an object carrying its own cursor. There is no separately computed cursor: an event that needs one gets a field.
cursorParamFromInput names the input field the auto-reconnect rewrites with the last delivered cursor before resubscribing. The first subscribe sends your input untouched — leave the field unset to start from the top, or set it to start "from N". Make the field optional in the schema, or give it a meaning on the first subscribe.
Both are paths, dots included ('filter.since', 'meta.at') — the same rule as pageParamFromInput on an infinite query — and both are typed: each path must exist in its shape, and the data cursor's type must fit the input field it lands in, all checked at the close. The pair comes only together, and only here — not per call, not in .subscriptionOptions().

The feature is entirely client-side. The server, the wire and the loader don't change: the loader reads the cursor out of its input — identically on a fresh start and on a resume — and serves the missed events itself (nothing is buffered for replay). The transformer covers the cursor like any input: a Date plucked from the data arrives in the loader a Date. Serve strictly-after the cursor and nothing duplicates across the break.

The tRPC contrast: there resumption is tracked(id, data) plus the Last-Event-ID HTTP header — a string outside the transformer, under an imposed name. Here the cursor is a field of your input: typed by your schema, serialized by the transformer, named whatever you like.

fetchSubscription never reconnects, so it never rewrites anything — the pair changes nothing for the imperative consumer, which loops with its own cursor.

Lifecycle callbacks

The same three callbacks a channel declares, in the subscription vocabulary — on the closing .subscription({...}) (fires for every consumer) and per useSubscription call; both levels run, point level first:

export const taskProgressSubscription = root.lets
  .subscription()
  .input(z.object({ taskId: z.string() }))
  .loader(async function* ({ input, signal }) {
    yield* watchProgress(input.taskId, { signal })
  })
  .subscription({
    onConnect: ({ connectionIndex }) => console.log('open', connectionIndex), // 0 — the first connect, > 0 — a re-open after a break
    onDisconnect: ({ connectionIndex }) =>
      console.log('stopped', connectionIndex),
    onError: ({ error }) => toast.error(error.message),
  })

onConnect — the stream opened, the first time and on every successful re-open after a break alike; connectionIndex counts the opens before this one (0 = the first, > 0 = a re-open). Per consumer: two hooks are two streams, each fires its own.
onDisconnect — an open stream stopped: a break (the reconnect policy redials next), a completion, or a mid-stream typed error. A stream that never opened fires nothing, and neither does your own teardown — unmount and an early break are you leaving, there is nobody left to tell.
onError — the terminal error status: a typed error or a break the policy gave up on. A request that never opened (a 401, failed validation) fires onError alone.

The props carry { point, input, connectionIndex } (error too on onError) — the input identifies the stream, since a subscription has no shared facade. All three are client code, cut from the server bundle with their imports. fetchSubscription fires the point-level callbacks too (one attempt, connectionIndex: 0); its call site takes no callbacks — the imperative consumer IS the loop.

A clientHandler's pushes have none of these on purpose: they have no connection of their own — the moments belong to the channel connection or the space membership that carries them (onConnect on the channel, onEnter on the space).

A stream on your own URL

Open with .action(method, path) instead — a generator loader closed with .subscription() streams on the method and path you declared:

export const feedSubscription = root.lets
  .action('GET', '/api/feed')
  .loader(async function* ({ signal }) {
    for await (const item of watchFeed({ signal })) {
      yield { item }
    }
  })
  .subscription()

Closing a generator loader with .action() (or .query() / .mutation()) is a type error — a stream closes with .subscription() only. The endpoint answers the declared method only (no POST fallback), and the input rides the ACTION shape: route params fill the path, .search keys ride the query string, a .body rides the body — useSubscription/fetchSubscription take the same { params, search, body } input the action's own callers would. On this form signal is typed AbortSignal | undefined — an action opener also accepts plain loaders, so the two share one loader shape — but for a generator loader it is always there at runtime.

Foreign clients: SSE

The wire is content-negotiated. Point0's own client asks for the native NDJSON framing (Accept: application/x-ndjson); any other client — including a plain browser EventSource — gets the SAME envelopes in SSE framing (Content-Type: text/event-stream, one data: event per envelope, comment lines as heartbeats). No option to set — the endpoint answers whatever asks:

const source = new EventSource(
  '/_point0/root/subscription/task-progress?input=' + input,
)
source.onmessage = (event) => {
  const envelope = JSON.parse(event.data) // { v: value } | { e: error } | { d: true }
}

The envelope semantics survive the framing: { v } is one value, { e } is a typed failure, { d: true } is completion — an SSE stream ending without a terminal envelope is a break, same as NDJSON. SSE id:/event: fields are unused: resumption does not ride the SSE protocol — it is the cursor pair, which travels as a field of the input, so a foreign client resumes the same way Point0's own does (put the cursor in the input); a foreign reconnect without one restarts the generator. Responses carry Vary: Accept so caches never mix the framings.

Subscription vs channel

Both push live; they solve opposite directions.

A subscription is the server pulling or generating a stream for one consumer: the progress of a long task, a tail, a feed. One-way, one HTTP request per consumer, nothing shared — no socket, no tickets, no rooms. When the reader is what starts the flow, reach here first: it's the cheaper machine.
A channel is push from outside to many clients: chats, presence, notifications. Two-way messages, spaces of rooms deciding who's together, one WebSocket carrying everything — a bigger machine for a bigger job.

The line between them is the REQUEST: a subscription is one, a clientHandler's pushes are not — which is why there is no .subscription() on socket points. The socket side consumes its pushes with useOnMessageFromServer / onMessageFromServer and iterates them with iterateMessagesFromServer.

The usual point machinery

Stripping — the generator .loader is cut from the client bundle: its body and the imports it uses are removed, so it never ships to the browser (it runs server-side) — the standard server-only loader rule. .input is cut from the client bundle likewise. The closing .subscription(...) is not cut from either bundle — kept in both (isomorphic): it has to resolve the point on the server and the client alike.
Transformer — every streamed line goes through the point's transformer, like query payloads: set .transformer(superjson) up the chain and Dates or Maps survive the stream too.
Middleware, ctx, plugins — run per subscription request like on any endpoint; the connect is a full request.
OpenAPI — the endpoint appears in the generated OpenAPI spec under filter: 'all'; the stream's line-by-line body is not expressible in OpenAPI, so the operation describes the endpoint, not the values.
No react-query cache — a subscription's values live in the hook that holds the stream, not in the query cache; there is nothing to invalidate or dehydrate.
Events — the side-split pointSubscriptionServer* / pointSubscriptionClient* families fire per stream attempt on both sides: Start (the client one's attempt counts its reconnects — 0 is the first; the server cannot know the retry count and carries none), Data on every streamed value, Settled with the outcome (completed / failed / broken), and Error whenever the consumer sees an error. Deliberately stopping the stream (unmount, breaking out of the loop) emits nothing further.

Reference

Statuses

Status	Meaning
`connecting`	the request is in flight — also during a reconnect wait, on the server during SSR
`open`	the stream answered; values are arriving
`closed`	the generator completed (`return`) — final; also the state under `enabled: false`
`error`	a typed error (a loader throw, failed validation, retries exhausted) — final, never auto-restarted

Options

.subscription({...}) — point-level defaults, overridable per useSubscription call (except the cursor pair — it names schema fields, a point contract, so neither the call site nor .subscriptionOptions() carries it):

Option	Default	What it does
`reconnect`	on	restart a broken stream — a boolean or `{ enabled, immediately, delay, backoff, maxDelay, retries }` (first retry immediate, then 300 ms doubling up to 5 s); completed streams and typed errors never restart
`cursorParamFromInput`	—	the input field (a dot-path) the auto-reconnect rewrites with the last delivered cursor — resumes instead of restarting; always paired with `cursorParamFromData`, see Resuming after a break
`cursorParamFromData`	—	the field (a dot-path) inside each streamed value the client plucks the cursor from; always paired with `cursorParamFromInput`
`onMessageFromServer`	—	module-level listener — fires for every consumer's every message; client code, cut from the server bundle
`onConnect`	—	a consumer's stream opened — every successful open, first and re-opens alike (`connectionIndex`: `0` = the first, `> 0` = a re-open) — see Lifecycle callbacks
`onDisconnect`	—	an open stream stopped (break / completion / mid-stream error); never on the consumer's own teardown
`onError`	—	the terminal `error` status — the typed error in the props

useSubscription(input, {...}) — on top of the point-level options:

Option	Default	What it does
`enabled`	`true`	subscribe or not — `false` renders `closed` and opens nothing
`onMessageFromServer`	—	fires on every message — the reactive path, no re-renders
`lastMessageFromServerAsData`	`false`	keep the latest message in state as `data`, re-rendering per message (query-shaped result)
the three lifecycle callbacks	—	`onConnect` / `onDisconnect` / `onError`, per call — compose with (never replace) the point-level ones

fetchSubscription(input, {...}):

Option	Default	What it does
`signal`	—	abort to stop the stream — the iterator ends, the server `signal` fires

Method surface

Both take the input first (omit it when the subscription declares none). The hook is kept in both bundles (it renders the connecting state during SSR); the imperative consumer is client-only.

Method	Signature	Returns
`useSubscription`	`(input, options?)`	`{ error, status, isLoading }`; `+ data` (latest message) with `lastMessageFromServerAsData`
`fetchSubscription`	`(input, options?)`	`AsyncIterable<data>` — a typed error or a break throws

A subscription also exposes the standard id (scope:type:name — e.g. root:subscription:taskProgress), type ('subscription'), tags, point, and Infer.

Enjoying Point0?

Give it a star on GitHub, ask questions, and follow what's new

Star Point0

A star on GitHub helps more developers find Point0

Star on GitHub

Start0

Point0 integrated into the Start0 SaaS boilerplate — the best way to use it

Explore Start0

YouTube

Tutorials, walkthroughs, and updates

YouTube

Discord

Questions and chat in English

Join Discord

News channel and chat in Russian

Channel Chat

Twitter

Posts and updates

Twitter

Building open-source software for the glory of the Lord Jesus Christ ☦️
With love for developers of all backgrounds around the world ❤️

Subscription

Category: Points

import { root } from '@/lib/root'
import { z } from 'zod'

export const taskProgressSubscription = root.lets
  .subscription()
  .input(z.object({ taskId: z.string() }))
  .loader(async function* ({ input, signal }) {
    // cut from the client bundle — the generator runs on the server
    for await (const percent of watchProgress(input.taskId, { signal })) {
      yield { percent } // one streamed value per yield
    }
  })
  .subscription()

export const TaskProgress = ({ taskId }: { taskId: string }) => {
  const { data, status } = taskProgressSubscription.useSubscription(
    { taskId },
    { lastMessageFromServerAsData: true }, // keep the latest message in `data`
  )
  if (status === 'closed') return <span>Done</span> // the generator returned
  return <progress value={data?.percent ?? 0} max={100} />
}

Declaring a subscription

Open with .subscription(), declare input and a generator loader, close with .subscription(options?):

export const countSubscription = root.lets
  .subscription() // open
  .input(z.object({ upTo: z.number().max(10) }))
  .loader(async function* ({ input }) {
    for (let n = 1; n <= input.upTo; n++) {
      yield { n } // each yield is one streamed value
    }
    // the generator returns → the stream completes, consumers see 'closed'
  })
  .subscription({ reconnect: false }) // close, with point-level options

A real endpoint

Consuming with the hook

const { error, status, isLoading } = taskProgressSubscription.useSubscription(
  { taskId },
  { onMessageFromServer: (tick) => setPercent(tick.percent) }, // fires on every message
)

status // 'connecting' | 'open' | 'closed' | 'error'
isLoading // status === 'connecting'

const { data } = taskProgressSubscription.useSubscription(
  { taskId },
  { lastMessageFromServerAsData: true },
)
data // the latest streamed value — undefined until the first one

taskProgressSubscription.useSubscription(
  { taskId },
  {
    enabled: !!taskId, // false → nothing opens, status is 'closed'
    onMessageFromServer: (tick) => console.info(tick.percent), // fires on every message
    lastMessageFromServerAsData: true, // keep the latest message in `data`; default false
    reconnect: { delay: 1000 }, // per-call override — see Reconnect
  },
)

During SSR nothing streams — the hook renders connecting on the server and does the real work after mount. Nothing is prefetched or dehydrated.

The imperative consumer: `for await`

fetchSubscription(input?, options?) returns an AsyncIterable of the streamed values:

for await (const { percent } of taskProgressSubscription.fetchSubscription({
  taskId,
})) {
  render(percent)
}
// the loop ends when the generator completes

const controller = new AbortController()
for await (const value of feedSubscription.fetchSubscription(undefined, {
  signal: controller.signal,
})) {
  // controller.abort() anywhere stops the stream
}

fetchSubscription is client-only — a runtime error on the server.

Stopping the stream — the `signal`

The server pulls its own source — poll, sleep, check the signal:

export const logTailSubscription = root.lets
  .subscription()
  .input(z.object({ file: z.string() }))
  .loader(async function* ({ input, signal }) {
    let offset = 0
    while (!signal.aborted) {
      const { lines, nextOffset } = await readNewLines(input.file, offset)
      offset = nextOffset
      if (lines.length > 0) yield { lines }
      await sleep(1000)
    }
  })
  .subscription()

import { on } from 'node:events'
import { postsEmitter } from '@/lib/posts' // .emit('add', post) in a mutation

export const newPostsSubscription = root.lets
  .subscription()
  .loader(async function* ({ signal }) {
    for await (const [post] of on(postsEmitter, 'add', { signal })) {
      yield { post }
    }
  })
  .subscription()

The teardown is symmetric: when the consumer goes away, the server aborts the signal and closes the generator, so finally blocks in the loader run and nothing leaks.

Reconnect

A completed stream (the generator returned) and a typed error (the loader threw, the input failed validation) are answers — they never restart.

.subscription({ reconnect: true }) // the default — same as omitting it
.subscription({ reconnect: { immediately: false, delay: 1000 } }) // never retry instantly
.subscription({ reconnect: { backoff: 1 } }) // constant 300ms waits, no exponent
.subscription({ reconnect: { retries: 3 } }) // give up after 3 attempts
.subscription({ reconnect: false }) // a break surfaces as a typed error

A reconnect restarts the generator from scratch — whatever streamed during the gap is lost unless the point resumes instead: see Resuming after a break.

Resuming after a break

export const feedSubscription = root.lets
  .subscription()
  .input(
    z.object({ channelId: z.string(), lastEventId: z.string().optional() }),
  )
  .loader(async function* ({ input, signal }) {
    // a fresh start and a resume read the cursor the SAME way — out of the input
    for (const missed of await db.eventsAfter(
      input.channelId,
      input.lastEventId,
    )) {
      yield missed // { id, ... }
    }
    for await (const event of live(input.channelId, { signal })) {
      yield event
    }
  })
  .subscription({
    cursorParamFromInput: 'lastEventId', // the input field the reconnect rewrites
    cursorParamFromData: 'id', // where the cursor lives inside each streamed value
  })

cursorParamFromData names the field inside each streamed value the client remembers — the data of a tracked subscription is an object carrying its own cursor. There is no separately computed cursor: an event that needs one gets a field.
cursorParamFromInput names the input field the auto-reconnect rewrites with the last delivered cursor before resubscribing. The first subscribe sends your input untouched — leave the field unset to start from the top, or set it to start "from N". Make the field optional in the schema, or give it a meaning on the first subscribe.
Both are paths, dots included ('filter.since', 'meta.at') — the same rule as pageParamFromInput on an infinite query — and both are typed: each path must exist in its shape, and the data cursor's type must fit the input field it lands in, all checked at the close. The pair comes only together, and only here — not per call, not in .subscriptionOptions().

fetchSubscription never reconnects, so it never rewrites anything — the pair changes nothing for the imperative consumer, which loops with its own cursor.

Lifecycle callbacks

export const taskProgressSubscription = root.lets
  .subscription()
  .input(z.object({ taskId: z.string() }))
  .loader(async function* ({ input, signal }) {
    yield* watchProgress(input.taskId, { signal })
  })
  .subscription({
    onConnect: ({ connectionIndex }) => console.log('open', connectionIndex), // 0 — the first connect, > 0 — a re-open after a break
    onDisconnect: ({ connectionIndex }) =>
      console.log('stopped', connectionIndex),
    onError: ({ error }) => toast.error(error.message),
  })

onConnect — the stream opened, the first time and on every successful re-open after a break alike; connectionIndex counts the opens before this one (0 = the first, > 0 = a re-open). Per consumer: two hooks are two streams, each fires its own.
onDisconnect — an open stream stopped: a break (the reconnect policy redials next), a completion, or a mid-stream typed error. A stream that never opened fires nothing, and neither does your own teardown — unmount and an early break are you leaving, there is nobody left to tell.
onError — the terminal error status: a typed error or a break the policy gave up on. A request that never opened (a 401, failed validation) fires onError alone.

A stream on your own URL

Open with .action(method, path) instead — a generator loader closed with .subscription() streams on the method and path you declared:

export const feedSubscription = root.lets
  .action('GET', '/api/feed')
  .loader(async function* ({ signal }) {
    for await (const item of watchFeed({ signal })) {
      yield { item }
    }
  })
  .subscription()

Foreign clients: SSE

const source = new EventSource(
  '/_point0/root/subscription/task-progress?input=' + input,
)
source.onmessage = (event) => {
  const envelope = JSON.parse(event.data) // { v: value } | { e: error } | { d: true }
}

Subscription vs channel

Both push live; they solve opposite directions.

A subscription is the server pulling or generating a stream for one consumer: the progress of a long task, a tail, a feed. One-way, one HTTP request per consumer, nothing shared — no socket, no tickets, no rooms. When the reader is what starts the flow, reach here first: it's the cheaper machine.
A channel is push from outside to many clients: chats, presence, notifications. Two-way messages, spaces of rooms deciding who's together, one WebSocket carrying everything — a bigger machine for a bigger job.

The usual point machinery

Stripping — the generator .loader is cut from the client bundle: its body and the imports it uses are removed, so it never ships to the browser (it runs server-side) — the standard server-only loader rule. .input is cut from the client bundle likewise. The closing .subscription(...) is not cut from either bundle — kept in both (isomorphic): it has to resolve the point on the server and the client alike.
Transformer — every streamed line goes through the point's transformer, like query payloads: set .transformer(superjson) up the chain and Dates or Maps survive the stream too.
Middleware, ctx, plugins — run per subscription request like on any endpoint; the connect is a full request.
OpenAPI — the endpoint appears in the generated OpenAPI spec under filter: 'all'; the stream's line-by-line body is not expressible in OpenAPI, so the operation describes the endpoint, not the values.
No react-query cache — a subscription's values live in the hook that holds the stream, not in the query cache; there is nothing to invalidate or dehydrate.
Events — the side-split pointSubscriptionServer* / pointSubscriptionClient* families fire per stream attempt on both sides: Start (the client one's attempt counts its reconnects — 0 is the first; the server cannot know the retry count and carries none), Data on every streamed value, Settled with the outcome (completed / failed / broken), and Error whenever the consumer sees an error. Deliberately stopping the stream (unmount, breaking out of the loop) emits nothing further.

Reference

Statuses

Status	Meaning
`connecting`	the request is in flight — also during a reconnect wait, on the server during SSR
`open`	the stream answered; values are arriving
`closed`	the generator completed (`return`) — final; also the state under `enabled: false`
`error`	a typed error (a loader throw, failed validation, retries exhausted) — final, never auto-restarted

Options

Option	Default	What it does
`reconnect`	on	restart a broken stream — a boolean or `{ enabled, immediately, delay, backoff, maxDelay, retries }` (first retry immediate, then 300 ms doubling up to 5 s); completed streams and typed errors never restart
`cursorParamFromInput`	—	the input field (a dot-path) the auto-reconnect rewrites with the last delivered cursor — resumes instead of restarting; always paired with `cursorParamFromData`, see Resuming after a break
`cursorParamFromData`	—	the field (a dot-path) inside each streamed value the client plucks the cursor from; always paired with `cursorParamFromInput`
`onMessageFromServer`	—	module-level listener — fires for every consumer's every message; client code, cut from the server bundle
`onConnect`	—	a consumer's stream opened — every successful open, first and re-opens alike (`connectionIndex`: `0` = the first, `> 0` = a re-open) — see Lifecycle callbacks
`onDisconnect`	—	an open stream stopped (break / completion / mid-stream error); never on the consumer's own teardown
`onError`	—	the terminal `error` status — the typed error in the props

useSubscription(input, {...}) — on top of the point-level options:

Option	Default	What it does
`enabled`	`true`	subscribe or not — `false` renders `closed` and opens nothing
`onMessageFromServer`	—	fires on every message — the reactive path, no re-renders
`lastMessageFromServerAsData`	`false`	keep the latest message in state as `data`, re-rendering per message (query-shaped result)
the three lifecycle callbacks	—	`onConnect` / `onDisconnect` / `onError`, per call — compose with (never replace) the point-level ones

fetchSubscription(input, {...}):

Option	Default	What it does
`signal`	—	abort to stop the stream — the iterator ends, the server `signal` fires

Method surface

Both take the input first (omit it when the subscription declares none). The hook is kept in both bundles (it renders the connecting state during SSR); the imperative consumer is client-only.

Method	Signature	Returns
`useSubscription`	`(input, options?)`	`{ error, status, isLoading }`; `+ data` (latest message) with `lastMessageFromServerAsData`
`fetchSubscription`	`(input, options?)`	`AsyncIterable<data>` — a typed error or a break throws

A subscription also exposes the standard id (scope:type:name — e.g. root:subscription:taskProgress), type ('subscription'), tags, point, and Infer.

Enjoying Point0?

Give it a star on GitHub, ask questions, and follow what's new

Star Point0

A star on GitHub helps more developers find Point0

Star on GitHub

Start0

Point0 integrated into the Start0 SaaS boilerplate — the best way to use it

Explore Start0

YouTube

Tutorials, walkthroughs, and updates

YouTube

Discord

Questions and chat in English

Join Discord

News channel and chat in Russian

Channel Chat

Twitter

Posts and updates

Twitter

Subscription

Declaring a subscription

A real endpoint

Consuming with the hook

The imperative consumer: for await

Stopping the stream — the signal

Reconnect

Resuming after a break

Lifecycle callbacks

A stream on your own URL

Foreign clients: SSE

Subscription vs channel

The usual point machinery

Reference

Statuses

Options

Method surface

Enjoying Point0?

Star Point0

Start0

YouTube

Discord

Telegram

Twitter

Subscription

Declaring a subscription

A real endpoint

Consuming with the hook

The imperative consumer: for await

Stopping the stream — the signal

Reconnect

Resuming after a break

Lifecycle callbacks

A stream on your own URL

Foreign clients: SSE

Subscription vs channel

The usual point machinery

Reference

Statuses

Options

Method surface

Enjoying Point0?

Star Point0

Start0

YouTube

Discord

Telegram

Twitter

The imperative consumer: `for await`

Stopping the stream — the `signal`

The imperative consumer: `for await`

Stopping the stream — the `signal`