SDK Cookbook

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');

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);

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});

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);

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}

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());

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);

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}

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);

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});

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;

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

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);