SDK Cookbook

This page is a practical recipe collection built from real SDK source usage and the examples/chat-with-ai implementation.

Recipe 1: Chat with AI pipeline (topic + consumer + subscriptions)

This mirrors the chat-with-ai architecture:

User message inserted into chat.messages
CDC routes inserts into topic chat.ai_processing
Background service consumes topic and writes AI reply back to table
Frontend receives both user and AI rows instantly via live subscription

1) SQL setup

SQL

1CREATE NAMESPACE IF NOT EXISTS chat;2 3CREATE TABLE chat.messages (4  id BIGINT PRIMARY KEY DEFAULT SNOWFLAKE_ID(),5  conversation_id BIGINT NOT NULL,6  sender TEXT NOT NULL,7  role TEXT NOT NULL DEFAULT 'user',8  content TEXT NOT NULL,9  created_at TIMESTAMP NOT NULL DEFAULT NOW()10) WITH (TYPE = 'USER');11 12CREATE TOPIC chat.ai_processing;13ALTER TOPIC chat.ai_processing14  ADD SOURCE chat.messages ON INSERT WITH (payload = 'full');

2) Frontend subscription (React)

1const chatSql = `2  SELECT * FROM chat.messages3  WHERE conversation_id = ${conversationId}4`;5 6const unsub = await client.live(7  chatSql,8  (rows) => {9    // Let the SDK hand the UI the current materialized conversation state.10    setMessages(rows);11  },12  {13    subscriptionOptions: { last_rows: 200 },14  },15);

3) Node.js background processor (consumer)

1import { Auth } from '@kalamdb/client';2import { createConsumerClient } from '@kalamdb/consumer';3 4const worker = createConsumerClient({5  url: process.env.KALAMDB_URL ?? 'http://localhost:2900',6  authProvider: async () => Auth.basic(process.env.KALAMDB_USER!, process.env.KALAMDB_PASSWORD!),7});8 9const handle = worker.consumer({10  topic: 'chat.ai_processing',11  group_id: 'ai-processor-service',12  auto_ack: true,13  batch_size: 1,14});15 16await handle.run(async (ctx) => {17  const payload = ctx.message.value as Record<string, unknown>;18  const row = (payload.row ?? payload) as {19    conversation_id: string | number;20    role: string;21    content: string;22  };23 24  if (row.role !== 'user') return;25 26  const aiReply = await generateReply(row.content);27  const user = String(ctx.user ?? '');28  if (!user) return;29  const conversationId = Number(row.conversation_id);30  if (!Number.isInteger(conversationId)) return;31 32  await worker.executeAsUser(33    'INSERT INTO chat.messages (conversation_id, sender, role, content) VALUES ($1, $2, $3, $4)',34    user,35    [conversationId, 'AI Assistant', 'assistant', aiReply],36  );37});

4) Operational notes from source

executeAsUser(...) follows KalamDB’s role matrix; service workers can target service/user accounts, while direct USER-table queries remain scoped to the authenticated account.
executeAsUser(...) is valid for USER and STREAM tables; shared tables still rely on their table policy.
Typing indicator updates are done via a stream table (chat.typing_indicators).
Service loads WASM explicitly in Node contexts where needed.

The server-side table policy details are documented at /docs/server/architecture/table-types.

Recipe 2: Resumable subscriptions with sequence checkpoints

Use from plus the typed lastSeqId checkpoints returned by getSubscriptions().

1import { SeqId } from '@kalamdb/client';2 3let savedSeqId: SeqId | undefined;4const eventsSql = "SELECT * FROM app.events WHERE source = 'system'";5 6const unsub = await client.live(7  eventsSql,8  (rows) => {9    // process the fully reconciled event feed10    renderEvents(rows);11 12    const active = client.getSubscriptions().find((sub) => sub.tableName === eventsSql);13    if (active?.lastSeqId) {14      savedSeqId = active.lastSeqId;15    }16  },17  {18    subscriptionOptions: {19      ...(savedSeqId ? { from: savedSeqId } : {}),20    },21  },22);

Recipe 2.1: Payload parser guard for consumer handlers

When your topic source uses WITH (payload = 'full'), parse both wrapper and direct row forms:

1function extractMessageRow(value: unknown) {2  if (!value || typeof value !== 'object') return null;3  const record = value as Record<string, unknown>;4  const source =5    record.row && typeof record.row === 'object'6      ? (record.row as Record<string, unknown>)7      : record;8 9  const conversationId = Number(source.conversation_id);10  if (!Number.isInteger(conversationId)) return null;11  if (source.role !== 'user' && source.role !== 'assistant') return null;12  if (typeof source.content !== 'string') return null;13 14  return {15    conversation_id: conversationId,16    role: source.role,17    content: source.content,18  };19}

Recipe 3: Typed query helpers for safer app code

1interface Conversation {2  id: string;3  title: string;4  created_at: string;5}6 7const conversations = await client.queryAll(8  'SELECT id, title, created_at FROM chat.conversations ORDER BY updated_at DESC'9);10 11const firstConversation = await client.queryOne(12  'SELECT id, title, created_at FROM chat.conversations LIMIT 1'13);14 15conversations.forEach((row) => {16  console.log(row.id.asString(), row.title.asString(), row.created_at.asString());17});18 19console.log(firstConversation?.id.asString());

Why this pattern helps:

keeps SQL in one place
maps array-row response into typed objects
reduces unsafe indexing in UI code

Recipe 4: File attachments in messages

1const result = await client.queryWithFiles(2  'INSERT INTO chat.messages (conversation_id, sender, role, content, file_data) VALUES ($1, $2, $3, $4, FILE("attachment"))',3  { attachment: selectedFile },4  [conversationId, 'user', 'user', messageText],5  (progress) => {6    setUploadPercent(progress.percent);7  }8);

Then parse FILE metadata:

1import { parseFileRef } from '@kalamdb/client';2 3const ref = parseFileRef(row.file_data);4if (ref) {5  const downloadUrl = ref.getDownloadUrl('http://localhost:2900', 'chat', 'messages');6  console.log(downloadUrl, ref.formatSize(), ref.getTypeDescription());7}

Recipe 5: Connection bootstrap for apps

1const client = createClient({2  url: 'http://localhost:2900',3  authProvider: async () => Auth.basic(username, password),4  wasmUrl: '/wasm/kalam_client_bg.wasm',5});6 7await client.login();8client.setAutoReconnect(true);9client.setReconnectDelay(1000, 30000);10client.setMaxReconnectAttempts(0);

This mirrors the provider pattern in chat-with-ai: Basic credentials are exchanged for JWT automatically before the realtime connection. Most apps can let the first live/subscription call connect lazily; call connect() only when you want to establish the shared WebSocket eagerly.

Recipe 6: Graceful shutdown for worker services

1const handle = worker.consumer({ topic: 'orders', group_id: 'order-worker', auto_ack: true });2 3const runPromise = handle.run(async (ctx) => {4  await processMessage(ctx.message.value);5});6 7process.on('SIGINT', async () => {8  handle.stop();9  await runPromise.catch(() => undefined);10  await worker.disconnect();11  process.exit(0);12});

Recipe 7: Vector search for semantic retrieval

This matches the new vector-search flow tested in KalamDB core: embeddings live in regular SQL tables, vector indexes are created with DDL, and ranking happens in standard SQL.

1) Schema

SQL

1CREATE TABLE rag.documents (2  id BIGINT PRIMARY KEY,3  title TEXT NOT NULL,4  body TEXT NOT NULL5) WITH (TYPE = 'USER');6 7CREATE TABLE rag.documents_vectors (8  id BIGINT PRIMARY KEY,9  doc_embedding EMBEDDING(384)10) WITH (TYPE = 'USER');11 12ALTER TABLE rag.documents_vectors13  CREATE INDEX doc_embedding USING COSINE;

2) Insert or update embeddings

1await client.query(2  `INSERT INTO rag.documents_vectors (id, doc_embedding)3   VALUES ($1, $2)`,4  [docId, JSON.stringify(embedding)],5);

3) Query nearest documents

1const matches = await client.queryAll(2  `SELECT d.id, d.title, d.body3   FROM rag.documents AS d4   JOIN rag.documents_vectors AS v ON v.id = d.id5   ORDER BY COSINE_DISTANCE(v.doc_embedding, $1)6   LIMIT 5`,7  [JSON.stringify(queryEmbedding)],8);