Live collections and CDC event streams

For each query that you pass to createRuntime(), the runtime gives you two ways to get its data:

1. runtime.client.<queryName>() creates a live collection: the current rows of the query, kept up to date in memory.
2. runtime.events.<queryName>() subscribes to a CDC event stream: each row change of the query, one event at a time.

The examples on this page use the allOrders and ordersByStatus queries from Server setup.

Live collections

A live collection is like the result of a Drizzle select that stays up to date by itself. The runtime reads the rows that match the query once, keeps them in memory, and updates them after each commit that changes them. Use it when your code needs the current rows, for example to render a list or to answer requests from memory.

Create a live collection

Create a live collection with runtime.client.<queryName>(). If the query has arguments, pass them as the only parameter. The call returns a promise that resolves when the live collection has loaded its first rows. The promise rejects if the runtime is not started.

const orders = await runtime.client.ordersByStatus({ status: 'accepted' });
const everyOrder = await runtime.client.allOrders();

Each call creates a new, independent live collection. Create a live collection once, keep it for as long as you need it, and then dispose it.

Read rows

list() returns the current rows synchronously. The rows are sorted by primary key, in the direction that the query’s .order() sets. Call it whenever you need the rows, for example in a request handler.

const rows = orders.list();

Change notifications

onChange() calls your listener after each change to the rows. The listener gets three fields:

1. events: the changes that the runtime applied. Each change has op set to 'insert', 'update', or 'delete', as described in Events.
2. state: the rows after the change. It is the same as list().
3. lsn: the log sequence number of the commit.

orders.onChange(({ events, state, lsn }) => {
  console.log(`${events.length} changes, ${state.length} rows at ${lsn}`);
});

onChange() returns a function that removes the listener.

Resync

After a connection loss, the live collection loads its rows again. The runtime does not know which rows changed during the connection loss, so onChange() does not fire for those changes. Instead, the live collection calls the listeners that you register with onResync(). The listener gets the new state and the lsn from which the live collection continues.

orders.onResync(({ state }) => {
  // Send the full list again, for example to connected clients.
  broadcast(state);
});

onResync() also returns a function that removes the listener.

Errors

onError() calls your listener when the live collection fails after you create it. This happens in two cases:

1. The live collection cannot load its rows again after a reconnect.
2. Replication fails permanently.

onError() also returns a function that removes the listener.

orders.onError((error) => {
  console.error('Live collection failed', error);
});

Dispose

dispose() stops the live collection. After that, the runtime calls none of its listeners. You can call it more than once.

orders.dispose();

CDC event streams

A CDC event stream delivers the row changes for one query with its arguments, one by one. Use it when you want to react to changes, for example to send a notification or to update another system. The stream keeps no rows and has no first load. It reports only changes that happen after you subscribe.

Subscribe

Subscribe to a CDC event stream with runtime.events.<queryName>(). Pass these parameters in this order:

1. The arguments, if the query has them.
2. A callback.
3. Optional options.

The call returns a function that closes the stream. It throws an error if the runtime is not started.

const stop = runtime.events.ordersByStatus(
  { status: 'accepted' },
  (event, lsn) => {
    switch (event.op) {
      case 'insert':
        console.log('entered the result', event.row);
        break;
      case 'update':
        console.log('changed', event.oldRow, '->', event.row);
        break;
      case 'delete':
        console.log('left the result', event.oldRow);
        break;
    }
  },
  {
    onResync: () => {
      // Load your state again.
    },
    onError: (error) => {
      console.error('CDC event stream closed', error);
    },
  },
);

stop();

A query without arguments takes the callback first: runtime.events.allOrders((event, lsn) => { ... }).

The callback must handle its own errors, the same as the listeners of a live collection.

Events

Events describe changes to the query result. For this reason, one table update can produce different events for different queries. For example, if an order changes from requested to accepted, a query for accepted orders gets an insert and a query for requested orders gets a delete.

There are three kinds of events:

1. insert: a row starts to match the query, or a matching row is inserted. Fields: op, row, pk.
2. update: a row changes and still matches the query. Fields: op, row, oldRow, pk.
3. delete: a row stops matching the query, or a matching row is deleted. Fields: op, oldRow, pk.

Every row in an event is a full row with the columns of the query. Columns outside the query never reach your callback. Your callback cannot tell a table insert from a table update that made a row match.

Order and repeated delivery

Events arrive in commit order. All events of one transaction share one lsn. If a disconnect happens between a delivery and its acknowledgement, the runtime can deliver the same commit again after the reconnect. If your callback has side effects, use the pair (event.pk, lsn) as an idempotency key.

Resync after a reconnect

After a connection loss, the runtime creates a new replication slot. It does not deliver the changes from commits that happened during the outage. Instead, the runtime calls onResync. When this happens, load again any state that you build from the events.

Stream errors

The runtime calls onError when replication fails permanently. After this call, the runtime closes the CDC event stream.