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:

subscription_ack
one or more initial_data_batch
change events with insert, update, or delete
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}