# Raw subscriptions (/docs/client/realtime/raw-api) --- title: Raw subscriptions description: client.realtime is the untyped layer under live(). Build a topic yourself when the collection name is a variable, or when you want the keyed events instead of whole snapshots. kind: guide package: questpie --- `client.collections.posts.live()` knows its collection at compile time. Some code does not. An admin screen, a generic table component or a devtool picks a collection at runtime, so it needs to build the topic itself. `client.realtime` is that seam. It is on every client and it shares the connection `live()` uses. It hands you `unknown` where the typed wrapper hands you a row. ## What is on it | Member | What it does | | -------------------------------- | -------------------------------------------------------- | | `subscribe(topic, cb, …)` | Callback per materialized value. Returns unsubscribe | | `stream(topic, signal?, id?)` | The same values as an async generator | | `streamEvents(topic, …)` | The raw frames, including `up-to-date` | | `awaitTxId(txid, signal?)` | Resolves once realtime has caught up to that transaction | | `awaitMutation(result, signal?)` | The same, reading the txid off a mutation result | | `destroy()` | Stops every subscription opened through this API | | `topicCount`, `subscriberCount` | Read-only counters, for diagnostics | `subscribe` takes its callbacks positionally, which is the one place this API is harder to read than `live()`. ```ts import { buildCollectionTopic } from "questpie/client"; const stop = client.realtime.subscribe( buildCollectionTopic("posts", { where: { published: true } }), (data) => render(data), undefined, // signal undefined, // custom topic id (error) => console.error(error), ); ``` `TData` is yours to declare. Nothing checks it, so pass the type the typed wrapper would have inferred. ## Build a topic Two builders cover every topic `live()` can express. ```ts import { buildCollectionTopic, buildGlobalTopic } from "questpie/client"; buildCollectionTopic("posts", { where: { published: true }, limit: 10 }); buildCollectionTopic("posts", { where: { published: true } }, "count"); buildCollectionTopic("posts", { id: postId, with: { author: true } }, "get"); buildGlobalTopic("settings", { with: { logo: true } }); ``` The third argument picks the operation, and each one takes a different option set. | Operation | Options it reads | | --------------- | ------------------------------------------------------- | | `find`, default | `where`, `with`, `limit`, `offset`, `orderBy`, `locale` | | `count` | `where`, `locale` | | `get` | `id` (required), `with`, `locale` | | Global `get` | `where`, `with`, `locale` | Anything else you put in the object is dropped. The builders copy named fields, they do not spread. A `TopicConfig` also allows `columns` and `mode` on a collection `find`, and no builder emits either. Write the object by hand when you need one. [Row deltas](/docs/client/realtime/deltas) shows that form. Two subscriptions with the same topic fields share one topic on the wire. That is how `live()` deduplicates, and it applies here too. Pass a custom id only when you need to address one subscription by name. ## The event frames `streamEvents()` yields the wire union instead of finished values. Use it to feed a store that wants to apply changes rather than replace a page of rows. ```ts for await (const event of client.realtime.streamEvents( buildCollectionTopic("posts"), )) { if (event.type === "snapshot") store.replace(event.data); if (event.type === "insert") store.add(event.key, event.row); if (event.type === "delete") store.remove(event.key); } ``` | `type` | Fields it carries | | ------------------ | -------------------------------------------- | | `snapshot` | `data`, plus `reset` and `upToDate` when set | | `insert`, `update` | `key`, `row`, `index?`, `txid?` | | `delete` | `key`, `txid?` | | `up-to-date` | `upToDate?`, `meta.totalDocs?`, `txid?` | Every frame carries `topicId` and `seq`. A collection `find` topic gives you keyed events even when the server is sending whole snapshots. The client diffs consecutive snapshots and emits the difference for you, keyed by `String(row.id)`. So a store built on these frames works before anyone enables server-side deltas. See [Row deltas](/docs/client/realtime/deltas) for the server half. ### Turning frames back into a value Three reducers ship with the client, and they are what `subscribe` uses internally. | Function | For | | -------------------------- | ------------------------------------ | | `applyRealtimeFindEvent` | A `find` result, windowed or not | | `applyRealtimeScalarEvent` | A `count` topic, which is one number | | `applyRealtimeSingleEvent` | A collection `get` or a global | They all take `(current, event)` and return the next value. Reach for them when you consume `streamEvents()` but still want a whole result at the end. `applyRealtimeFindEvent` passes a snapshot straight through, and the first one carries the page window. Every later frame keeps that window, because `limit` and `offset` belong to the topic and cannot change under a subscription. A keyed event moves rows and shifts `totalDocs` by however many it added or removed. A totals frame then replaces that count with the server's. `totalPages`, `hasNextPage` and the rest are recomputed from the window each time, by the same formula the server's paginator uses, so a live `find` with a `limit` reports the page counts a `find()` would. ## Wait for your own write A mutation returns before its change has travelled back through realtime. So an optimistic UI can flicker. You write, the local state updates, and then a slightly older snapshot lands and undoes it. `awaitMutation()` closes that gap. QUESTPIE attaches the transaction id to every mutation result, and this reads it back off. ```ts const post = await client.collections.posts.create({ title: "Hello" }); await client.realtime.awaitMutation(post); // Every open topic has now seen this transaction. ``` It resolves when that exact transaction arrives, or when every open topic reports a watermark past it. Use `awaitTxId(txid)` when you carry the id yourself. Reconciliation is measured against your live subscriptions. With none open there is nothing to wait on, and the promise sits there. Pass a signal so you can time it out. ## Next **[Row deltas](/docs/client/realtime/deltas)** covers the server-side keyed events, which query shapes qualify, and the rollout gate in front of them.