Consumer Runtime

@kalamdb/consumer exposes a high-level worker runtime on top of topic consumers:

• runConsumer() for row/change-driven handlers with retry/failure hooks
• runAgent() as a deprecated compatibility alias for the same lifecycle model
• createLangChainAdapter() for plugging in LangChain chat models

These APIs are exported directly from @kalamdb/consumer.

The client you pass to these helpers should be created with createConsumerClient() or otherwise satisfy the worker-side ConsumerClientLike contract.

runConsumer() quick start

1import { Auth } from '@kalamdb/client';2import { createConsumerClient, runConsumer } from '@kalamdb/consumer';3 4const client = createConsumerClient({5  url: 'http://127.0.0.1:2900',6  authProvider: async () => Auth.basic('root', 'kalamdb123'),7});8 9await runConsumer<Record<string, unknown>>({10  client,11  name: 'blog-summarizer',12  topic: 'blog.summarizer',13  groupId: 'blog-summarizer-agent',14  start: 'earliest',15  batchSize: 20,16  retry: {17    maxAttempts: 3,18    initialBackoffMs: 250,19    maxBackoffMs: 1500,20    multiplier: 2,21  },22  onChange: async (ctx, change) => {23    const row = change.data;24    const blogId = row.blog_id;25    if (typeof blogId !== 'string' && typeof blogId !== 'number') {26      return;27    }28 29    await ctx.sql(30      'UPDATE blog.blogs SET updated = NOW() WHERE blog_id = $1',31      [String(blogId)],32    );33  },34  onFailed: async (ctx, change) => {35    await ctx.sql(36      'INSERT INTO blog.summary_failures (run_key, blog_id, error, created, updated) VALUES ($1, $2, $3, NOW(), NOW())',37      [ctx.runKey, String(change.data.blog_id ?? 'unknown'), String(ctx.error ?? 'unknown')],38    );39  },40  ackOnFailed: true,41});

Handler context (ConsumerRunContext)

onChange(ctx, change) receives:

• runtime metadata: ctx.name, ctx.runKey, ctx.attempt, and ctx.maxAttempts
• source data: change.data is the decoded row/event payload
• change metadata: change.user, typed change.op, change.key, change.timestampMs, change.partitionId, change.offset, change.topic, change.groupId, and metadata-only change.message
• helpers: sql(), queryOne(), queryAll(), ack()
• LLM helper: llm (when llm adapter + optional systemPrompt are configured)

Use change.data for the decoded changed row/event and use the other change.* fields for metadata about that same event. High-level ctx is intentionally reserved for runtime execution state and helpers, so it does not expose message, change, user, op, or offset duplicates. The high-level change.message object intentionally omits payload, deprecated value, and raw transport change fields, so the decoded row lives in exactly one place: change.data. change.user is populated only when the topic event carries a subject and is expected to be undefined for shared-table routes such as the summarizer example. The older onRow(ctx, row) alias remains deprecated; use onChange(ctx, change) or onFailed(ctx, change) for the stable shape.

Generated ORM row types use the same interface. Pass the generated row type as the first runConsumer<T>() generic, read the row from change.data, and read event metadata from change.op, change.user, change.offset, or change.message. No ORM-specific wrapper is needed.

runKey

Default format is:

<name>:<topic>:<partition_id>:<offset>

Override with runKeyFactory when integrating custom idempotency keys.

Retry and ACK behavior

runConsumer() does not expose an auto_ack option. Internally it creates the underlying low-level consumer with auto_ack: false so the runtime can control retries and ack timing explicitly:

• onChange succeeds: message is acked.
• onChange throws: retried based on retry policy.
• retries exhausted:
  • if onFailed is missing, message is not acked.
  • if onFailed succeeds and ackOnFailed !== false, message is acked.
  • if onFailed throws, message is not acked.

If you need to choose auto_ack yourself, use client.consumer() directly instead of runConsumer().

Hooks:

• onRetry({ error, attempt, maxAttempts, backoffMs, runKey, message })
• onError({ error, runKey, message })
• onConnectionRetry({ error, attempt, maxAttempts, backoffMs })
• onConnectionRestored({ attempt })
• onConnectionError({ error, attempt })

Retry policy options

The retry object supports additional tuning:

• jitterRatio — add randomness to backoff (0 disables jitter)
• shouldRetry(error, attempt) — classify retryable failures (defaults to retrying)

1retry: {2  maxAttempts: 5,3  initialBackoffMs: 250,4  maxBackoffMs: 5_000,5  multiplier: 2,6  jitterRatio: 0.1,7  shouldRetry: (error) => {8    // Example: don't retry on validation errors9    return !String(error).includes('invalid-input');10  },11}

Other useful runConsumer options include partitionId, timeoutSeconds, and stopSignal for graceful shutdown.

runConsumer() also supervises the underlying consumer loop. If the server is temporarily down or the connection drops, it retries with exponential backoff and jitter until stopSignal aborts. Configure this with connectionRetry:

1connectionRetry: {2  initialBackoffMs: 500,3  maxBackoffMs: 30_000,4  multiplier: 1.8,5  jitterRatio: 0.2,6}

Use the connection hooks when you want operators to see the worker lifecycle clearly in logs:

1await runConsumer({2  client,3  name: 'blog-summarizer',4  topic: 'blog.summarizer',5  groupId: 'blog-summarizer-agent',6  onConnectionRetry: ({ error, attempt, maxAttempts, backoffMs }) => {7    console.warn('worker cannot reach KalamDB:', String(error));8    console.warn(`retrying in ${backoffMs}ms (attempt ${maxAttempts ? `${attempt}/${maxAttempts}` : attempt})`);9  },10  onConnectionRestored: ({ attempt }) => {11    console.log(`worker reconnected after ${attempt} retry attempt${attempt === 1 ? '' : 's'}`);12  },13  onConnectionError: ({ error, attempt }) => {14    console.error(`worker stopped reconnecting after ${attempt} attempts:`, error);15  },16  onChange: async () => {},17});

onConnectionRestored() fires once after a retriable connection failure when the runtime successfully reaches KalamDB again. With the default createConsumerClient() implementation, this happens after the first successful poll, even if that poll returns no messages.

Row parsing

By default, the runtime parses KalamDB topic rows like this:

• it reads message.payload
• if message.payload has a nested row object, it uses payload.row for backward compatibility
• otherwise it uses message.payload directly

That means apps using the ORM generator can pass the generated row type directly and skip parser boilerplate:

1import { runConsumer } from '@kalamdb/consumer';2import type { ChatDemoMessages } from './schema.generated';3 4await runConsumer<ChatDemoMessages>({5  client,6  name: 'chat-demo-agent',7  topic: 'chat_demo.ai_inbox',8  groupId: 'chat-demo-agent',9  onChange: async (_ctx, change) => {10    const row = change.data;11    if (row.role !== 'user') return;12    console.log(row.content);13  },14});

You can override with changeParser(message):

1changeParser: (message) => {2  const payload = message.payload as Record<string, unknown> | null;3  if (!payload) return null;4  if (typeof payload.row === 'object' && payload.row) return payload.row as Record<string, unknown>;5  return payload;6}

Return null only when you intentionally want to skip agent handling for a message.

Message-Level Handling

Use onMessage only for older single-argument handlers that still want the retry/failure lifecycle:

1import { runConsumer } from '@kalamdb/consumer';2 3type OrderEventPayload = {4  order_id: string;5  status: string;6  amount: number;7  _table: string;8};9 10await runConsumer<OrderEventPayload>({11  client,12  name: 'orders-worker',13  topic: 'orders.events',14  groupId: 'orders-group',15  onMessage: async (_ctx, change) => {16    console.log(change.offset, change.op, change.data.status);17  },18});

onMessage remains as a deprecated compatibility hook. New workers should use onChange(ctx, change) so change data and per-change metadata stay together.

runConsumer() acks after successful handler completion. Call ctx.ack() only if you explicitly want to ack before returning.

LangChain integration

llm and systemPrompt are still supported optional RunConsumerOptions fields. If you do not need model-assisted workers, leave them undefined and ctx.llm will be null.

createLangChainAdapter() accepts a duck-typed chat model with invoke() and optional stream():

1import { createLangChainAdapter, runConsumer } from '@kalamdb/consumer';2import { ChatOpenAI } from '@langchain/openai';3 4const llm = createLangChainAdapter(5  new ChatOpenAI({ apiKey: process.env.OPENAI_API_KEY, model: 'gpt-4o-mini' }),6);7 8await runConsumer({9  client,10  name: 'summary',11  topic: 'blog.summarizer',12  groupId: 'summary-group',13  llm,14  systemPrompt: 'Write one concise sentence.',15  onChange: async (ctx, _change) => {16    const summary = await ctx.llm?.complete('Summarize this row');17    console.log(summary);18  },19});

ctx.llm.complete() accepts either a plain string prompt or structured message input.

Production notes

• Prefer idempotent writes keyed by runKey.
• Keep retry.maxAttempts bounded and tune backoff for your latency profile.
• Use stopSignal for graceful shutdown in containerized workers.
• Leave the default connection retry enabled for long-running workers so transient restarts do not exit the process.
• Persist terminal failures (onFailed) to a table for replay/inspection.

Related docs

• Topic Consumers & ACK
• TypeScript Summarizer Example