# Watch (/docs/guides/events/watch) > Stream a tenant's events live over Server-Sent Events with handleWatch, resume from any cursor with Last-Event-ID, filter by type, and authorize every stream. `handleWatch` streams one tenant's events as [Server-Sent Events](https://html.spec.whatwg.org/multipage/server-sent-events.html), read straight from the outbox. A client opens one long `GET`, receives each committed event as it becomes visible, and, when the connection drops, reconnects with the id of the last event it saw and continues exactly after it: no gap, no duplicate from the server's side. It is the right tool for anything that follows events while it runs: the console's live stores, a dashboard, a script that waits for a reservation to become active. For a service that must see every event even while it is down, use [webhooks](/docs/guides/events/webhooks), which keep the position on the server. ## Mount the endpoint [#mount-the-endpoint] `handleWatch(request, options)` is a Fetch-style handler from `better-compute/events`. The compute server does not mount it: the host that serves the API does, next to it, and supplies the authorization. That keeps the long-lived stream out of the JSON API's envelope and lets each host choose how streams are authenticated. **serve (host export):** ```ts title="better-compute.config.ts" import { defineHost, watchAuthorizer } from 'better-compute/cli'; import { createWatchRegistry, handleWatch } from 'better-compute/events'; export const host = defineHost(({ compute }) => { const authorize = watchAuthorizer(compute); // decided exactly like events.list const streams = createWatchRegistry(); // so IAM revocations close open streams at once return { iam: { bridge, streams: [streams] }, // bridge: your createComputeIamBridge() http: { routes: [ { path: '/watch', method: 'GET', handler: (request: Request) => handleWatch(request, { sql: compute.context.sql, authorize, registry: streams }), }, ], }, }; }); ``` `better-compute serve` answers the host export's routes beside the API, so this serves `GET /watch` on the API's listener. `watchAuthorizer(compute, { cookie? })` decides the stream through the server's own envelope: it calls `events.list` for one event as the caller (authentication, the tenant pre-check, one `compute:events:read` decision on `compute-tenant/{t}`, and the caches), with `?tenantId=` defaulting to the caller's tenant. It returns the caller's subject and, when the IAM port has `authorizeSession`, a re-authorizer, so the stream can be closed or re-checked later. Without `cookie: true` it accepts only the `Authorization` header. **Next.js route:** ```ts title="app/api/compute/watch/route.ts" import { ComputeError, resourceType } from 'better-compute/core'; import { handleWatch } from 'better-compute/events'; import { compute, origin } from '@/lib/compute'; export const runtime = 'nodejs'; export const dynamic = 'force-dynamic'; export async function GET(request: Request): Promise { return handleWatch(request, { sql: compute.context.sql, authorize: async (watch) => { // Cookie-authenticated streams need a CSRF guard: require the custom header, check the origin. if (watch.headers.get('x-better-compute') !== '1') throw new ComputeError('CSRF_REJECTED', 'Send X-Better-Compute: 1'); const site = watch.headers.get('sec-fetch-site'); const from = watch.headers.get('origin'); if ((site && site !== 'same-origin') || (from && from !== origin)) throw new ComputeError('UNTRUSTED_ORIGIN', 'Origin is not trusted'); const credential = { headers: watch.headers }; const caller = await compute.context.iam.authenticate(credential); const tenantId = new URL(watch.url).searchParams.get('tenantId') ?? caller.tenantId; await compute.context.iam.require(credential, { tenantId, action: 'compute:events:read', resource: { type: resourceType('tenant'), id: tenantId }, }); return { tenantId }; }, }); } ``` This is how the console serves `/api/compute/watch`. **Any Fetch server:** ```ts import { handleWatch } from 'better-compute/events'; const watch = (request: Request) => handleWatch(request, { sql: compute.context.sql, authorize: authorizeWatch }); // Bun, Deno, Hono, or anything that maps a Request to a Response: if (new URL(request.url).pathname === '/watch') return watch(request); ``` ## Authorization [#authorization] `authorize` is the whole security boundary of the stream, so decide it the way the API decides `events.list`: `compute:events:read` on `compute-tenant/{tenantId}`. The stream then carries only rows whose `tenantId` is the one `authorize` returned; the filter is part of the outbox query, not applied after it. * Authenticate with the same credentials the API accepts (a bearer token, a session JWT, or on the same origin the IAM cookie) by passing the request's headers to `compute.context.iam.authenticate`. * Default to the caller's own tenant. Accepting `?tenantId=` lets a platform administrator follow another tenant, and the `require` call decides whether they may. * When the stream accepts a cookie, add a CSRF guard like the Next.js tab above. `GET` requests carry no custom header by default, and browsers omit `Origin` on same-origin `GET`s, so the console requires `X-Better-Compute: 1`, which a cross-origin page cannot send without a preflight nothing approves. > **Authenticating is not enough.** An `authorize` that only authenticates lets any member of a tenant follow all of that tenant's events. Decide `compute:events:read` as above, or use `watchAuthorizer`, as the example deployments (`examples/shared/compose.ts`) do. Without a grant, a member gets 403; a caller without a token, 401. ### Revocation [#revocation] A stream is authorized when it opens, so the grant says how to end it early: * **`reauthorize`** runs every `reauthorizeMs` (60 s). `false` closes the stream; a throw counts as IAM unreachable, and after `iamOutageMs` (2 minutes) of that the stream closes. * **`registry`** lets a revocation close matching streams at once: `closeWhere({ agentId | delegationId | sessionId | identityId | tenantId + anyAgent }, reason)`. A tenant alone is refused, since it would close too much. With `host.iam: { bridge }` and the registry in `host.streams` (or `host.iam.streams`), `serve` installs handlers that close an agent's or delegation's streams on `delegation:revoke` and `agent:suspend`, every agent's in the tenant on `agent:suspend-all`, and an identity's on its offboarding or session revocation (the client reconnects with `Last-Event-ID` and is decided again). `dev --sim` passes `host.streams` to its embedded IAM's handlers too, so the examples' `/watch` streams close on revocation in development as well. A closed stream ends with `event: error` carrying `{ "code": "ACCESS_DENIED", "message": }`, and events read during the revocation are not sent. `tests/e2e/access.test.ts` checks that a revoked delegation and a member's sign-out close `/watch` within 5 seconds. A refusal, or any error before the stream starts, is a JSON error with the code's status, like the API's: `{ "error": { "code": "ACCESS_DENIED", "message": "…" } }` with 403, `INVALID_INPUT` with 400 for a bad cursor or type list, and `METHOD_NOT_ALLOWED` with 405 for anything but `GET`. ## The stream [#the-stream] ```text GET /watch?types=reservation.*,node.connected Accept: text/event-stream Authorization: Bearer biam_… Last-Event-ID: 88213.51119 retry: 2000 id: 88213.51120 event: reservation.active data: {"id":"evt_01k5y3v0m6y8x7w5t4s3r2p1n0","type":"reservation.active","tenantId":"acme","subject":"reservation/rsv_01k5y3a0b1c2d3e4f5g6h7j8k9","time":"2026-09-24T10:15:02.114Z","sequence":"88213.51120","data":{"state":"active","poolId":"pool_01k5y1z2x3w4v5t6s7r8q9p0n1"}} : keep-alive 88213.51187 ``` The response is `text/event-stream` with `Cache-Control: no-cache, no-transform` and `X-Accel-Buffering: no`, so nginx-style proxies do not buffer it. It opens with `retry:`, then sends one block per event: | Field | Value | | ------- | --------------------------------------------------------------------------------------- | | `id` | the event's outbox position, `txid.seq`, equal to its `sequence` | | `event` | the event type, so `EventSource.addEventListener('reservation.active', …)` works | | `data` | the whole event as one line of JSON, the [event shape](/docs/guides/events#event-shape) | A comment line `: keep-alive ` goes out every `heartbeatMs`. If reading the outbox fails mid-stream, the server sends `event: error` with `data: {"message": …}` and closes; the client reconnects and resumes. After `maxDurationMs` the server closes the stream on purpose, which also makes clients re-authenticate at least once an hour. ## Cursors and resume [#cursors-and-resume] The stream's cursor is the outbox position `.`, with a dot. It is every event's `sequence` and every SSE `id:`. | Request | Starts after | | ----------------------------------- | --------------------------------------------------------------------------------------------- | | no cursor | the current tail: only events committed from now on; history is not replayed | | `Last-Event-ID: 88213.51119` header | that event | | `?lastEventId=88213.51119` | that event (for clients that cannot set the header, such as a first `EventSource` connection) | | `?lastEventId=0.0` | the start of the outbox: everything retention has kept | The header wins when both are sent. A cursor that is not two groups of 1 to 20 digits joined by a dot is `INVALID_INPUT`. Note that `events.list` writes the same position with a colon (`88213:51119`); replace the colon with a dot to move from history to the live stream. The outbox is read with the xmin rule: only transactions older than the snapshot's xmin, in `(txid, seq)` order. No event can therefore commit behind a cursor the stream has already handed out, which is what makes resuming exact. The cost is that an event becomes visible only when every older transaction has finished; see [commit order](/docs/guides/events#commit-order-and-the-xmin-cursor). A cursor older than what [retention](/docs/operations/retention) has pruned continues with the oldest row still there, without an error. To backfill and then follow without a gap, page history with `events.list`, then open the stream with the last item's `sequence`, colon replaced by a dot, as `Last-Event-ID`. ## Filters [#filters] `?types=` takes 1 to 20 comma-separated patterns of up to 128 characters, with the same matching as webhooks: `*` (the default), a family such as `reservation.*` or `node.gpu.*`, or an exact type. An empty list is `INVALID_INPUT`. Unlike webhooks, patterns are not checked against the event catalog, so a typo matches nothing: take family names from the [event reference](/docs/reference/events). Filtering happens after the tenant filter, and the cursor still advances past events that do not match, so a filtered stream resumes correctly too. ## Consume it [#consume-it] **The typed client: `$watch`.** `client.$watch({ types?, tenantId?, lastEventId?, signal?, path?, maxBackoffMs? })` from `@better-compute/client` is an async iterable over the stream. It sends the client's token and headers, yields each event with its cursor as `sequence`, reconnects with `Last-Event-ID` after the server's `retry:` delay (or with back-off, up to 30 seconds by default, after failures), stops on `break` or its `signal`, and throws `UNAUTHENTICATED` on a 401 and `ACCESS_DENIED` on a 403. It reads `{basePath}/watch` by default (`/api/compute/watch`); pass `path: '/watch'` for a stream mounted at the root, as in the `serve` tab above. `tenantId` (e2fe6f8) sends `?tenantId=`, which the host's `watchAuthorizer` decides; without it the stream is the caller's own tenant. ```ts for await (const event of client.$watch({ types: ['reservation.*'], lastEventId: savedCursor, path: '/watch' })) { handle(event); savedCursor = event.sequence; } ``` See [the TypeScript client](/docs/guides/clients/typescript-client#watching-events) for its options. **The CLI: `events watch`.** `better-compute events watch --url … [--tenant T] [--types …] [--since CURSOR] [--count N]` follows the same stream with `$watch` and prints one terminal-safe JSON line per event; its `--path` defaults to `/watch` (the console app serves `/api/compute/watch`). `--since` takes the stream's own `.` cursor, the `sequence` of a printed line (`--since 184467.12`, 86706b4), and also the colon form of `events.list` pages (`184467:12`), the same outbox position, which it sends as the dotted one (867c837); `events list --cursor` takes only the colon form. See [follow events live](/docs/guides/clients/cli#follow-events-live). **Browsers and Node.js: console-kit.** `watchEvents` from `@better-compute/console-kit` reads the stream over `fetch` with callbacks instead of a loop, and can also send an `Authorization` header, which `EventSource` cannot. It keeps the last `id`, reconnects after the server's `retry:` delay when the server closes the stream, backs off (doubling, up to 60 seconds) after failures, and sends `Last-Event-ID` on every reconnect. ```ts import { watchEvents } from '@better-compute/console-kit'; const watcher = watchEvents({ url: 'https://compute.example.com/watch', types: ['reservation.*'], lastEventId: savedCursor, // optional: resume across restarts headers: async () => ({ authorization: `Bearer ${await getToken()}` }), // fresh per connection onEvent: (event) => { handle(event); savedCursor = event.sequence as string; }, onStatus: (status, detail) => console.log(status, detail?.retryInMs), }); ``` `startLiveConsole(client, { tenantId, watchUrl, headers, permissions })` goes further: it loads the tenant's stores and routes every event to them. See [console](/docs/guides/clients/console#live-updates). **The console** uses `watchEvents` against its own `/api/compute/watch` with the IAM cookie and `X-Better-Compute: 1`, shows a "live" badge while the stream is open, and falls back to refreshing loaded stores every 20 seconds while it is not. **`EventSource`** works when the stream accepts a cookie and needs no custom header, or when you put the credential somewhere your `authorize` reads other than a header. It reconnects with `Last-Event-ID` by itself. **curl** is enough to look at a stream: ```bash curl -N https://compute.example.com/watch?types=node.* \ -H "Authorization: Bearer $BETTER_COMPUTE_TOKEN" -H "Accept: text/event-stream" ``` ## Delivery guarantees [#delivery-guarantees] * **At least once across reconnects.** The server never skips or repeats an event after the cursor it was given. A client that crashes between handling an event and saving its cursor sees it again, so deduplicate on `id`. * **In commit order**, within one stream. * **One tenant per stream.** A platform view of several tenants opens several streams, or uses an [event sink](/docs/extending/sinks). * **No back-pressure on the database.** Each open stream polls the outbox every `pollMs` while idle and reads in batches of `batch` while behind. Many idle streams mean many small queries; put the watch route on processes sized for it.