Topic Consumers & ACK

@kalamdb/consumer is the worker package. Use it for topic consumption, ACKs, and retrying agents. Keep it out of browser code.

The consumer client wraps an internal @kalamdb/client instance, so workers still have query(), queryOne(), queryAll(), role-matrix executeAsUser(), login(), refreshToken(), and disconnect() available alongside the topic APIs.

For high-level row-processing agents with retries and failure hooks, see:

• Consumer Runtime

runConsumer() already deserializes standard KalamDB topic row payloads and unwraps legacy { row: ... } envelopes. Use a custom changeParser only for non-row payload modes or custom envelopes.

Create a worker client with the same auth options you would use for @kalamdb/client:

1import { Auth } from '@kalamdb/client';2import { createConsumerClient } from '@kalamdb/consumer';3 4const client = createConsumerClient({5  url: 'http://localhost:2900',6  authProvider: async () => Auth.basic('worker', 'Secret123!'),7});

Consumer builder API

Copy this when you want a long-running worker loop.

1const handle = client.consumer({2  topic: 'orders',3  group_id: 'billing',4  auto_ack: true,5  batch_size: 10,6});7 8await handle.run(async (ctx) => {9  console.log(10    ctx.message.partition_id,11    ctx.message.offset,12    ctx.message.op,13    ctx.message.user,14    ctx.message.payload,15  );16});17 18handle.stop();

Source behavior (client.ts):

• loops consume(options) until stop() requested
• if auto_ack and handler didn’t manually ctx.ack(), SDK auto-acks
• when no more messages, sleeps ~1s before polling again

Manual ACK mode

Use manual ACK when the message should be committed only after your handler finishes.

1const handle = client.consumer({2  topic: 'orders',3  group_id: 'billing-manual',4  auto_ack: false,5});6 7await handle.run(async (ctx) => {8  await processOrder(ctx.message.payload);9  await ctx.ack();10});

ctx contains:

• user (typed branded user identifier)
• message (topic, group, partition, offset, key, op, timestamp, decoded payload)
• ack(): Promise<void>

Topic payload contract in handlers

ctx.message.payload is the decoded JSON payload from the topic source. In the current backend, WITH (payload = 'full') publishes the changed row object directly and injects _table with the source table id.

Most handlers can work with the payload directly:

• direct row object for current full payloads
• optional legacy wrapper object with row under payload.row if you are still consuming older message shapes

For lower-level consumer().run() handlers, use a defensive parser when you need to normalize multiple payload shapes yourself:

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

If your topic source uses another payload mode (key / diff), treat both as still in development. If you are testing them anyway, align parser logic with the current payload shape and remember that _table lives inside the decoded payload, not in a separate source_table field.

One-shot batch consume

1const batch = await client.consumeBatch({2  topic: 'orders',3  group_id: 'billing',4  batch_size: 20,5  start: 'earliest',6});7 8console.log(batch.messages, batch.next_offset, batch.has_more);

If you know the payload shape ahead of time, type it once and keep the whole consume flow strongly typed:

1type OrderEventPayload = {2  order_id: string;3  status: string;4  amount: number;5  _table: string;6};7 8const batch = await client.consumeBatch<OrderEventPayload>({9  topic: 'orders',10  group_id: 'billing',11  batch_size: 20,12  start: 'earliest',13});14 15for (const message of batch.messages) {16  console.log(message.payload.status);17}18 19const handle = client.consumer<OrderEventPayload>({20  topic: 'orders',21  group_id: 'billing',22});23 24await handle.run(async (ctx) => {25  console.log(ctx.message.payload.status);26});

Each message includes:

• topic, group_id, partition_id, offset
• optional key
• optional user
• optional op
• optional timestamp_ms
• decoded payload

message.value still exists as a deprecated alias for message.payload.

Low-level ack API

1await client.ack('orders', 'billing', 0, 42);

This acknowledges up to and including offset 42.

ConsumeRequest options

1{2  topic: string;3  group_id: string;4  start?: string;5  partition_id?: number;6  batch_size?: number;7  timeout_seconds?: number;8  auto_ack?: boolean;9  concurrency_per_partition?: number;10}

start accepts backend-supported start modes (for example earliest or latest) and can include offset-style values depending on server routing.

Current backend notes:

• the HTTP topic API reads one partition per request
• partition_id: 0 is the safe default today
• timeout_seconds is accepted by the API shape but does not long-poll yet; the SDK keeps polling explicitly

USER and STREAM table write pattern in workers

When consuming events and writing into TYPE = 'USER' or TYPE = 'STREAM' tables, use executeAsUser() from an authorized service, DBA, or system account:

1await client.executeAsUser(2  'INSERT INTO chat.messages (conversation_id, role, content) VALUES ($1, $2, $3)',3  String(ctx.user ?? ''),4  [conversationId, 'assistant', reply],5);

This preserves per-user isolation in service-side writes through KalamDB’s explicit delegation boundary.

When to use which API

• SQL subscriptions: row-level table/query reactivity.
• consumer() from @kalamdb/consumer: explicit queue or partition processing with ack semantics.
• runConsumer() from @kalamdb/consumer: higher-level runtime with retry and onFailed lifecycle hooks.