Real-time updates with durable streams

In chapter 4 the table was live on the client — it refetched and mutated without a navigation. But it still only knows what it fetched, and it only learns anything by asking. This chapter builds the other direction: a durable change-stream that the server pushes into an open page, read by a useLiveQuery hook that re-renders when data arrives — no polling loop, no refresh button, no refetch.

Be clear-eyed about how far the shipped plumbing carries that today. The subscription half is complete: a StreamDB handle, a live query, and rows that appear in an already-open page without a reload. The producer half is where the current runtime stops short — the sagas plugin mirrors its instances into the stream once, when its service starts, rather than on every transition. So you will build a genuinely live view and drive a genuinely pushed update through it, and you will see exactly where the seam ends.

What you will build

A live monitor island that subscribes to a durable StreamDB and re-renders when new events arrive. You will open a StreamDB handle pointed at the streams runtime, drive a table with useLiveQuery, and mount it from a definePage page that resolves the stream address on the server. The worked example is the sagas stream — the durable change-stream the sagas plugin ships with a ready-made typed collection.

Why the example uses the sagas stream

NetScript's durable-streams runtime mirrors execution state — saga instances, worker executions — into change-streams that the frontend can subscribe to. The sagas plugin ships the ready-made typed collection for this: createSagasStreamDB gives useLiveQuery a StreamDB it can query with full types, which is why this chapter's worked live table grounds there. The pattern is identical for any StreamDB collection; once you have it, pointing a live table at your own stream is the same three moves — and the producer half of that story is already in reach, because the streams plugin scaffolds a user-owned durable stream into your workspace (see the closing section). To follow this chapter against running data, the workspace needs the sagas plugin and its streams runtime — add the published package with netscript plugin install @netscript/plugin-sagas if it is not already installed.

Before you begin

You should have completed chapter 4: the orders page rendering through definePage with a hydrated QueryIsland. The live layer needs the streams runtime reachable. With aspire start up, find the streams resource in the dashboard resource list, copy its endpoint, and confirm it answers:

curl <streams-endpoint>/health

Its host port was picked by the installer, not fixed by you — the resource list is the only place to read it. A healthy response means the durable-streams producer runtime is live. If it is dead, the sagas plugin (which brings the stream) is not installed or Aspire has not finished booting it — check the dashboard resource list at :18888.

Step 1 — Open a StreamDB handle

createSagasStreamDB from @plugins/sagas/streams opens a typed StreamDB client against the streams runtime. You give it the runtime's baseUrl; it gives you typed collections you can query. Build it inside the island, memoized on the URL, and manage its lifecycle:

// apps/dashboard/routes/(dashboard)/dashboard/sagas/(_islands)/SagasLiveIsland.tsx (the StreamDB handle)
import { useEffect, useMemo } from 'preact/hooks';
import { createSagasStreamDB, type SagaInstance } from '@plugins/sagas/streams';

function SagasLiveInner(props: { streamsBaseUrl: string }) {
  const sagasDb = useMemo(
    () => createSagasStreamDB({ baseUrl: props.streamsBaseUrl }),
    [props.streamsBaseUrl],
  );

  // Preload the stream on mount; close it on unmount.
  useEffect(() => {
    void sagasDb.preload();
    return () => sagasDb.close();
  }, [sagasDb]);

  // … useLiveQuery below
}

preload() warms the stream so the first frame has data; close() tears the subscription down when the island unmounts. Always pair them — a leaked subscription keeps a connection open.

Step 2 — Drive a table with useLiveQuery

useLiveQuery from @netscript/fresh/query runs a query against a StreamDB collection and re-renders whenever the underlying data changes — no polling, no manual refetch. Query the sagaInstance collection:

// apps/dashboard/routes/(dashboard)/dashboard/sagas/(_islands)/SagasLiveIsland.tsx (the live query)
import { useLiveQuery } from '@netscript/fresh/query';

const { data: instanceRows = [] } = useLiveQuery(
  (query) => query.from({ instance: sagasDb.collections.sagaInstance }),
  [sagasDb],
);

const instances = instanceRows as SagaInstance[];
// Render `instances` as a table — the array is replaced whenever the stream
// pushes a change for one of these rows, and the table re-renders.

The callback shape is a tiny query builder: query.from({ instance: <collection> }) selects rows from the sagaInstance collection. When the server pushes a change for any of those rows, useLiveQuery returns the new array and the table re-renders. That is the entire push path on the client — it reacts to whatever the producer sends, as soon as it arrives.

Step 3 — Seed the island from the server

The island needs one thing from the server before it can subscribe: the streams runtime's address. Resolve it in a definePage page — the same shape as chapter 4, a request-scoped .withResource feeding a .withLayer loader — so the live monitor gets a real page rather than a loose helper function. getStreamsUrl from @netscript/plugin-streams-core resolves the runtime address:

// apps/dashboard/routes/(dashboard)/dashboard/sagas/index.tsx
import { definePage } from '@app/utils.ts';
import { getStreamsUrl } from '@netscript/plugin-streams-core';
import SagasLiveIsland from './(_islands)/SagasLiveIsland.tsx';

export const sagasMonitorPage = definePage()
  .withTelemetry({ enabled: true, spanName: 'dashboard.sagas.live' })
  // Resolved once per request; the layer loader below awaits it.
  .withResource('streamsBaseUrl', () => getStreamsUrl())
  .withLayer('monitor', SagasLiveIsland, {
    loader: async (ctx) => ({
      streamsBaseUrl: await ctx.resource('streamsBaseUrl'),
    }),
  })
  .withLayout((slots) => <main class='ns-page'>{slots.monitor()}</main>)
  .withMeta(() => ({ title: 'Saga monitor', description: 'Live saga instances.' }))
  .build();

export const { handler, default: page } = sagasMonitorPage;
export { page as default };

The layer loader hands the island the one prop Step 1's StreamDB handle needs. Note what is not here: no withPolicy, no partial, no staleTime, and no dehydrated query cache. Chapter 4 needed all of that because its rows arrive by request and must survive a cold cache; these rows arrive by push, so the same machinery would be dead weight. A live page is the lighter of the two — the builder does not force you to carry what you are not using. Partials covers the other side of that choice: what a partial layer buys a region whose data arrives by request, and why a pushed region does not want it.

Step 4 — Wrap the island in QueryIsland

useLiveQuery needs the TanStack Query context, so the live monitor lives inside a QueryIsland exactly like chapter 4's orders island:

// apps/dashboard/routes/(dashboard)/dashboard/sagas/(_islands)/SagasLiveIsland.tsx (the island boundary)
import { QueryIsland } from '@netscript/fresh/query';

export default function SagasLiveIsland(props: { streamsBaseUrl: string }) {
  return (
    <QueryIsland>
      <SagasLiveInner streamsBaseUrl={props.streamsBaseUrl} />
    </QueryIsland>
  );
}

// SagasLiveInner holds Step 1's StreamDB handle and Step 2's useLiveQuery.

The boundary is deliberately thin: QueryIsland supplies the context, and SagasLiveInner does the work — open the handle, preload(), subscribe with useLiveQuery, close() on unmount. The island is now complete — prove it compiles as a unit:

deno check 'apps/dashboard/routes/(dashboard)/dashboard/sagas/(_islands)/SagasLiveIsland.tsx' --unstable-kv

A clean check confirms the StreamDB handle and the live query line up.

Point it at your own stream

Everything above consumes a stream the framework produces for you. The producer half of the seam is just as close: if your workspace has the streams plugin installed (netscript plugin install streams), its scaffolder wrote streams/notifications-stream.ts — a user-owned durable stream you edit like any other source file. It is the same two primitives this whole chapter rides on:

// streams/notifications-stream.ts (scaffolded — yours to edit)
// defineStreamSchema declares the typed collections; createDurableStream
// returns the producer. Swap the sample event shape for your own domain
// events — an order.cancelled event is one zod object away.
import { createDurableStream, defineStreamSchema } from '@netscript/plugin-streams-core';

The producer publishes with upsert(collection, row) and await flush(); a browser consumes a user-defined stream over HTTP at its streamPath (long-poll subscription by default, SSE available). The full producer surface, URL resolution, and current limitations are on the Durable streams page — when you outgrow the sagas monitor, that is where your own order-events stream starts.

Verify your progress

With aspire start up, open the live monitor at /dashboard/sagas/ — the route Step 3's page declares. Get the app's host from the Aspire dashboard resource list rather than a memorized port; a scaffolded Fresh app pins no host port, so Aspire allocates one at each start.

The table starts empty, because nothing in this track has created a saga instance yet — chapter 2 built a plain oRPC read-model, not a saga producer, so posting an order does not publish a saga message. You create one yourself, and the order of the steps is what makes the push observable.

The live producer runs in the saga processor after each durable transition, on either the Prisma or KV store backend. Prisma additionally enables startup reconciliation of instances created before the stream producer was available; install it with netscript plugin install @netscript/plugin-sagas --saga-store-backend prisma, then netscript db init, when you need that historical backfill.

# Install the shorthand once (see the storefront track for the full form):
#   deno install -gArf -n ns-sagas jsr:@netscript/plugin-sagas@0.0.6/cli

# 1. Register a throwaway saga, then restart the graph — the processor loads
#    its registry at boot, so a saga added while it runs is not yet known.
ns-sagas add saga demo --message-type=DemoStarted --durability=t1 --topic=demo

# 2. With the graph back up, open the monitor page and leave it connected.

# 3. Publish a message. The engine durably records the transition and then
#    upserts the latest instance state into the live stream.
ns-sagas publish DemoStarted --payload='{ "id": "demo_1" }' --correlation-key=demo_1

# 4. Confirm the instance exists in the durable store.
ns-sagas list --instances --saga=DemoSaga --json

With the subscription live, the first transition upsert arrives over the open connection and the row appears without a page reload. Every later step—including failure and compensation steps—updates the same sagaInstance row after its transition is durable. Restarting the saga processor does not erase canonical progress; on Prisma, the service's startup reconciliation also backfills historical rows into the stream.

Type-check the new files:

deno task check

[ ] curl <streams-endpoint>/health (endpoint from the Aspire resource list) returns healthy.
[ ] The live island opens a createSagasStreamDB handle and queries it with useLiveQuery.
[ ] A definePage page at dashboard/sagas/index.tsx resolves getStreamsUrl() in a .withResource and hands it to the island through a .withLayer loader.
[ ] With the page open, publishing and advancing a saga updates its row through the subscription, with no reload.
[ ] deno task check is clean.

No row after the restart?

Work down the path in order. ns-sagas list --instances --json tells you whether the instance exists at all — if it does not, the saga never registered (ns-sagas list --registered --json) or the publish failed. If the instance exists but no row arrives, check that the streams runtime is up and that both the processor's durable-stream producer and the page's getStreamsUrl() resolved the streams service. A dead streams endpoint (such as the :4437 port assigned in this tutorial's scaffold; note that each scaffold allocates its own randomized ports) means no transition event can reach the stream. The Prisma-only Saga Prisma delegates unavailable startup warning means historical reconciliation is skipped; it does not disable live per-transition upserts on KV.

What you built

A server-pushed table: a durable StreamDB handle (createSagasStreamDB) driving useLiveQuery, wrapped in a QueryIsland and mounted by a definePage page whose .withResource resolves the streams address. You proved the push end to end — state that entered the database through a saga publish arrived in an already-open browser page with no polling, no refetch and no reload — and you saw precisely how far the shipped producer carries it: one reconciliation per sagas-service start. The client half is the durable part; when a plugin's producer grows a per-transition feed, this page gets finer-grained updates without a line of change. Next you run the whole graph locally under Aspire.