Topic Consumers & ACK

@kalamdb/consumer is the worker package. Use it for topic consumption, ACKs, and retrying workers. 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 workers with retries and failure hooks, see:

• /docs/ts-sdk/agents

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  onConnect: () => {8    console.log('worker client is ready');9  },10  onConnectionError: (event) => {11    console.error(event.message, { recoverable: event.recoverable, attempt: event.attempt });12  },13});

The consumer client now exposes the same connection lifecycle hooks as the base app client:

• onConnect(callback) for the first healthy connection and later recoveries
• onConnectionError(callback) for topic/auth/transport failures
• onError(callback) as a compatibility alias

If you do not register an error callback, createConsumerClient() logs connection failures to the console by default.

onConnectionError receives a richer worker-oriented event than the base client callback:

• message
• recoverable
• attempt
• backoffMs when a retry is scheduled
• context when the client knows which step failed
• raw error

Fatal configuration failures such as invalid URLs are marked recoverable: false. Unreachable servers and transient transport failures are reported as recoverable.

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
• handle.run(handler, hooks?) accepts optional lifecycle hooks such as onBatchSuccess({ nextOffset, hasMore, messageCount }) after each successful poll

The consumer client delegates base connection handling to its internal @kalamdb/client instance, so connection/auth failures during login, refresh, or topic auth normalization also flow through onConnectionError.

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.

For the table-type and impersonation rules behind this pattern, see /docs/server/architecture/table-types and /docs/server/sql-reference/impersonation.

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, onConnect, onConnectionError, and onFailed lifecycle hooks. It owns ack timing internally, so there is no auto_ack option on RunConsumerOptions.