Row collections

A row collection is a table: many small rows, each validated against a schema you declare on the contract. You filter them, subscribe to subsets, mutate them in atomic batches, and secure them per row. This is the common case — messages, users, tasks, orders.

New to collections? Start with the overview for the rows-vs-documents split, or build one hands-on in Tutorial 4.

Declare rows on the contract

Each entry is a Standard Schema (Zod/Valibot/ArkType), the primary-key field, and optional advisory foreign keys:

import { defineContract } from '@super-line/core'
import * as z from 'zod'

export const api = defineContract({
  collections: {
    users: { schema: z.object({ id: z.string(), name: z.string() }), key: 'id' },
    messages: {
      schema: z.object({
        id: z.string(),
        channelId: z.string(),
        authorId: z.string(),
        text: z.string(),
        createdAt: z.number(),
      }),
      key: 'id',
      references: { authorId: 'users' }, // advisory FK: messages.authorId → users
    },
  },
  roles: { user: { clientToServer: { /* … */ } } },
})

Row types flow end-to-end with no codegen — RowOf<typeof api, 'messages'> is { id, channelId, authorId, text, createdAt } on the server handle, the client handle, and the TanStack collection. The server validates every row write against the schema.

Give the server a backend

Hand the server one collection backend — it serves every collection, in a single transaction domain, so cross-collection batches are atomic. Pair it with a row policy per collection (deny-by-default):

import { createSuperLineServer } from '@super-line/server'
import { memoryCollections } from '@super-line/collections-memory'
import { isIn } from '@super-line/core'

const srv = createSuperLineServer(api, {
  transports: [/* … */],
  authenticate: (h) => ({ role: 'user' as const, ctx: { userId: h.query.userId, channels: [/* … */] } }),
  identify: (conn) => conn.ctx.userId,
  collections: memoryCollections(),
  policies: {
    users: { read: () => undefined, write: () => true }, // world-readable directory
    messages: {
      read: (principal, ctx) => isIn('channelId', ctx.channels), // you only ever see your channels
      write: (principal, op, next, prev, ctx) =>
        op === 'delete' ? prev?.authorId === principal : next?.authorId === principal, // author-only
    },
  },
})

The read/write policies are the server-authoritative half a client query engine can't do on its own — row-level security enforced at the source. Which backend you pick (in-memory, SQLite, or the self-clustering Postgres tier) is a one-line swap.

Server code can co-write, bypassing policy but still schema-validated — the door for business-logic mutations a request handler owns:

await srv.collection('messages').insert({
  id: 'm1', channelId: 'general', authorId: 'system', text: 'welcome', createdAt: Date.now(),
})

Several co-writes, across collections, commit as one atomic batch with srv.batch — all of them or none. srv.conditionalBatch applies its batch only while every condition's stored row still matches its filter, and resolves false otherwise:

await srv.batch([
  { op: 'delete', collection: 'messages', id: 'm1' },
  { op: 'insert', collection: 'messages', row: { id: 'm2', channelId: 'general', authorId: 'system', text: 'hello', createdAt: Date.now() } },
])
const moved = await srv.conditionalBatch(
  [{ collection: 'messages', id: 'm2', filter: eq('channelId', 'general') }],
  [{ op: 'update', collection: 'messages', row: { id: 'm2', channelId: 'random', authorId: 'system', text: 'hello', createdAt: Date.now() } }],
)

An update that leaves a row exactly as it was reaches no subscriber. pgliteCollections is the exception: its change feed carries no prior row to compare against.

client.collection(name) is typed by the contract. subscribe(query) opens a live row-set — an initial snapshot, then per-row change events, auto-resubscribed and re-diffed across reconnects:

import { eq } from '@super-line/core'

const messages = client.collection('messages')
const sub = messages.subscribe({
  filter: eq('channelId', 'general'),
  orderBy: [{ field: 'createdAt', dir: 'asc' }],
  limit: 50,
})
await sub.ready                 // frames process concurrently — await before you depend on live delivery
sub.rows()                      // current rows, ordered + limited
sub.subscribe((ev) => { /* { type: 'insert'|'update'|'delete', id, row } */ })

await messages.insert({ id: 'm2', channelId: 'general', authorId: 'me', text: 'hi', createdAt: Date.now() })
await messages.batch([          // one atomic batch, all-or-nothing on the server
  { type: 'update', row: { /* … */ } },
  { type: 'delete', id: 'm1' },
])

Await sub.ready

The subscription's frames process concurrently, so sub.ready is the barrier before you can trust live delivery. This is a hard rule for the raw sync layer.

The query IR

The subscription carries a small query IR — filter (and/or/not, eq/neq/comparisons/in/like/ilike/has), orderBy, and limit/offset — built with the helpers exported from @super-line/core:

import { and, eq, gt, like } from '@super-line/core'

messages.subscribe({
  filter: and(eq('channelId', 'general'), gt('createdAt', since), like('text', '%deploy%')),
})

has(field, value) matches a row whose array field holds an element strictly equal to value (has('labels', 'vip')); it never matches a field that isn't an array. Strings compare by Unicode code point, which is the order of SQL's binary collation, and like's % spans newlines, as in SQL.

The server pushes down what it can and re-checks the exact predicate; the client re-filters too, so an over-approximate result is always safe. orderBy/limit shape the initial snapshot; the live phase then streams every matching change and the consumer owns its window. The full operator set is in the query IR reference and the same helpers power TanStack DB queries.

Page a list and show its total

offset/limit page the snapshot. The total the page belongs to comes from the server: srv.collection(n).count(filter?) counts matching rows without reading them. A SQL backend answers it with one COUNT(*) when the filter compiles exactly. Like the other srv.collection calls it skips policy, so AND in what the caller may read yourself. End orderBy on a unique field, so rows that tie on the sort value can't move between pages:

const filter = and(eq('channelId', channelId), has('labels', 'lobby'), ilike('name', `%${term}%`))
const page = client.collection('screens').subscribe({
  filter,
  orderBy: [{ field: 'name', dir: 'asc' }, { field: 'id', dir: 'asc' }],
  offset: 50,
  limit: 50,
})

// server: a request handler rebuilds the filter from its input and scopes it to the caller
const total = await srv.collection('screens').count(and(filter, eq('orgId', ctx.orgId)))

For a filter rail, srv.collection(n).countBy(field, filter?, { order?, limit? }) returns how many rows hold each value of a field: [{ value, count }], most rows first (order: 'value' sorts by value instead), with limit keeping the top N. Null is its own group; a missing field joins none. An array field counts each row once under every distinct element, the rows has(field, value) would match, so clicking a facet value filters to exactly the count it showed. A SQL backend answers with one GROUP BY when the filter compiles exactly:

const scope = eq('orgId', ctx.orgId)
const byLabel = await srv.collection('screens').countBy('labels', scope, { limit: 10 }) // [{ value: 'lobby', count: 412 }, …]
const byVersion = await srv.collection('screens').countBy('appVersion', and(scope, eq('channelId', channelId)))

The primitive is non-optimistic

client.collection(name) is the raw sync layer: a write appears in rows() when the server confirms it. Optimism — instant local application, rollback on error — is TanStack DB's job, layered on top.

React

useCollection is a thin, typed filtered-list hook for simple cases. For joins and complex live queries, point TanStack DB at the collection instead:

tsx

const { rows, insert, update, delete: del } = useCollection('messages', { filter: eq('channelId', id) })

Watch it in the Control Center

Beyond the Collections schema graph + row browser — which lists each row's created / updated timestamps alongside its data — mounting inspector() streams every subscribe and write to the live feed's Collections filter (collection.sub / collection.write / collection.change) — expand a row to see the written data, redacted per your inspector({ redact }) config. Those created/updated timestamps are inspector-only: they never travel over client.collection(n).subscribe(), so sub.rows() stays exactly your schema.

Run it

examples/collections — a runnable tsx tracer: RLS pushdown, a messages ⋈ users join, and optimistic writes with rollback, in ~120 lines against a real server.
examples/collections-chat — a full Slack-like React app: four collections (users/channels/memberships/messages), per-channel row-level security with a join gate, optimistic sends, and the Control Center Collections view.

Row-level security & policies — lock down who reads and writes each row.
Querying with TanStack DB — joins, live queries, and optimistic mutations.
Backends & clustering — pick a backend and go multi-node.

Row collections ​

Declare rows on the contract ​

Give the server a backend ​

Client: subscribe to subsets, mutate in batches ​

The query IR ​

Page a list and show its total ​

React ​

Run it ​

Next ​