Advanced Realtime Mode

Use advanced mode when you want protocol events instead of a pre-materialized row set.

If your UI only needs current rows, use live() or liveTable() first. They expose onCheckpoint too, so durable resume does not require raw event handling.

liveEvents(sql, onEvent, options?) is the low-level API.

Low-level SQL events

1import { ChangeType, MessageType } from '@kalamdb/client';2 3const stop = await client.liveEvents(4  'SELECT * FROM app.messages WHERE conversation_id = 42',5  (event) => {6    if (event.type !== MessageType.Change) {7      return;8    }9 10    if (event.change_type === ChangeType.Insert) {11      console.log(event.rows);12    }13  },14  {15    batchSize: 100,16    lastRows: 50,17  },18);19 20await stop();

Use this path for protocol tooling, debugging, custom reconciliation, or event processors that need direct access to subscription_ack, initial_data_batch, change, and error frames.

Options

1{2  batchSize?: number;3  lastRows?: number;4  from?: SeqId | number | string;5  autoFetchBatches?: boolean;6  onCheckpoint?: ({ lastSeqId, subscriptionId }) => void;7  onError?: (event) => void;8}

• batchSize chunks the initial snapshot from the server.
• lastRows includes the last N rows before live changes begin.
• from resumes from a known SeqId checkpoint.
• autoFetchBatches controls whether the client asks for remaining startup batches automatically.
• onCheckpoint fires whenever the SDK observes a newer applied sequence ID.
• onError receives protocol error frames without forcing your event callback to branch on every error.

Checkpoints

1let latestSeq;2 3const stop = await client.liveEvents(4  'SELECT id, body FROM app.messages WHERE conversation_id = 42',5  (event) => {6    if (event.type === MessageType.Change) {7      applyChange(event.change_type, event.rows, event.old_values);8    }9  },10  {11    lastRows: 20,12    from: latestSeq,13    onCheckpoint: ({ lastSeqId }) => {14      latestSeq = lastSeqId;15      localStorage.setItem('messages.seq', lastSeqId.toString());16    },17  },18);

The checkpoint is a single SeqId. Snapshot and commit boundaries stay server-owned.

Event model

Use runtime enums from @kalamdb/client:

1import { MessageType, ChangeType } from '@kalamdb/client';

Common sequence:

1. subscription_ack
2. one or more initial_data_batch
3. change events with insert, update, or delete
4. optional error

SQL safety for dynamic filters

liveEvents() accepts a SQL string. For dynamic values:

• Prefer validated numeric IDs before interpolation.
• Do not interpolate raw user text directly into SQL.

1function buildConversationSql(conversationId: number): string {2  if (!Number.isInteger(conversationId) || conversationId <= 0) {3    throw new Error('conversationId must be a positive integer');4  }5 6  return `SELECT * FROM app.messages WHERE conversation_id = ${conversationId}`;7}

Related docs

• Live Queries
• Simple Realtime Mode

SHARED table ownership and resume domains

Clustered servers default to shared_shards = 4. Each SHARED table owns exactly one Raft group. CREATE persists table_options.shared_shard_id and a fresh history_incarnation in replicated table metadata. Increasing the configured count places new tables across the expanded set; existing owners do not change. Do not remove a shard that owns tables. Legacy metadata without an owner stays on shard 0. Moving existing data requires explicit export/import into a fresh history.

_version = (raft_log_index << 16) | ordinal is ordered only within its table’s domain. Different tables on the same shard can interleave, so per-table versions can have gaps. Transactions spanning different owner groups are rejected.

A SHARED subscription acknowledgement includes batch_control.version_domain with history_incarnation, table_id, scope_id (the shared owner), and partition_id (currently null). Persist this complete envelope together with the last safely consumed version. Resume by supplying both options.from and options.version_domain. Missing envelopes, another table/group/history, and unsupported partitions are rejected as stale cursors. Take a fresh snapshot after CURSOR_EXPIRED; do not reuse the old domain.

The Rust transport preserves the acknowledgement domain for automatic reconnect. In TypeScript, use versionDomain with from when restoring a persisted checkpoint; it serializes as version_domain. Treat the domain as opaque and never infer it from the numeric version.