[RFC] add concurrency strategies for overlapping messages (queue, debounce, concurrent)

[cramforce]

Summary

• Add concurrency option to ChatConfig with four strategies: drop (default), queue, debounce, concurrent
• Add lockScope option (thread | channel) to control lock granularity — defaults to channel for WhatsApp and Telegram (breaking change)
• Add enqueue/dequeue/queueDepth to StateAdapter interface with implementations across all state adapters
• Add MessageContext to handler signatures so handlers can see skipped messages

Problem

Two independent problems with the current lock-and-drop model:

1. Overlapping messages are silently lost. When a message arrives while a handler is processing, the SDK throws LockError and the message is discarded. For AI chatbots, this means user follow-ups sent while the model is streaming are permanently lost.

2. Channel-based platforms lock too broadly. The lock key is always threadId. For WhatsApp and Telegram, threadId === channelId, so the entire conversation is locked. For Telegram forum topics, each topic gets its own lock even though the group chat is the natural conversation unit.

Concurrency Strategies

`drop` (default, unchanged)

Lock acquired or `LockError` thrown. Backward compatible.

`queue`

Messages enqueued while a handler is running. On completion, the queue is drained: only the latest message is dispatched, with all intermediate messages provided as `context.skipped`:

const chat = new Chat({ concurrency: 'queue', ... });

chat.onSubscribedMessage(async (thread, message, context) => {
  if (context?.skipped.length) {
    const allMessages = [...context.skipped, message];
  }
});

`debounce`

Every message (including the first) starts/resets a debounce timer. Only the final message in a burst is processed. Uses `waitUntil` to keep the function alive during the debounce window.

const chat = new Chat({
  concurrency: { strategy: 'debounce', debounceMs: 1500 },
  ...
});

`concurrent`

No locking. Every message processed immediately in parallel.

Fine-grained config

interface ConcurrencyConfig {
  strategy: 'drop' | 'queue' | 'debounce' | 'concurrent';
  maxQueueSize?: number;            // Default: 10
  onQueueFull?: 'drop-oldest' | 'drop-newest';  // Default: 'drop-oldest'
  queueEntryTtlMs?: number;         // Default: 90_000 (90s)
  debounceMs?: number;               // Default: 1500
  maxConcurrent?: number;            // Default: Infinity
}

Lock Scope

Problem

For channel-based platforms (WhatsApp, Telegram), `threadId === channelId` in most cases. For Telegram forum topics, each topic previously got an independent lock (`telegram:-100:456`), but the natural conversation unit is the group chat (`telegram:-100`).

Solution

New `lockScope` property on adapters and `ChatConfig`:

• `thread` (default): lock per `threadId` — per reply chain (Slack), per topic (Telegram forums)
• `channel`: lock per `channelId` — per channel (Slack), per group chat (Telegram)

Breaking change: WhatsApp and Telegram default to `channel`

// adapter-whatsapp
readonly lockScope = 'channel';

// adapter-telegram
readonly lockScope = 'channel';

For WhatsApp DMs, this is a no-op (`threadId === channelId` already).

For Telegram forum topics, this changes behavior: all topics in a group now share one lock instead of independent per-topic locks. This matches the mental model that the group chat is a single conversation.

Users can override per-instance:

const chat = new Chat({
  lockScope: 'thread',  // Override adapter default
  ...
});

Or dynamically with an async resolver:

const chat = new Chat({
  lockScope: async ({ adapter, isDM, threadId, channelId }) => {
    if (adapter.name === 'telegram' && isDM) return 'thread';
    return 'channel';
  },
  ...
});

StateAdapter additions

interface StateAdapter {
  enqueue(threadId: string, entry: QueueEntry, maxSize: number): Promise<number>;
  dequeue(threadId: string): Promise<QueueEntry | null>;
  queueDepth(threadId: string): Promise<number>;
}

Implemented in all four adapters:

• MemoryStateAdapter — in-process array
• RedisStateAdapter — Lua script (RPUSH + LTRIM + PEXPIRE)
• IoRedisStateAdapter — same Lua approach
• PostgresStateAdapter — `chat_state_queues` table with atomic dequeue, expired entry purging

Handler signature change

All message handler types now accept optional `MessageContext`:

type MentionHandler = (thread, message, context?) => void | Promise<void>;
type SubscribedMessageHandler = (thread, message, context?) => void | Promise<void>;
type DirectMessageHandler = (thread, message, channel, context?) => void | Promise<void>;
type MessageHandler = (thread, message, context?) => void | Promise<void>;

Existing handlers that don't use `context` are unaffected.

Observability

All strategies emit structured log events at `info` level:

Event	Strategy	Data
`message-queued`	queue	threadId, lockKey, messageId, queueDepth
`message-dequeued`	queue, debounce	threadId, lockKey, messageId, skippedCount
`message-dropped`	drop, queue (full)	threadId, lockKey, messageId, reason
`message-expired`	queue, debounce	threadId, lockKey, messageId
`message-superseded`	debounce	threadId, lockKey, droppedId
`message-debouncing`	debounce	threadId, lockKey, messageId, debounceMs

Backward Compatibility

• Default concurrency remains `drop` — no behavior change unless configured
• `onLockConflict` is deprecated but continues to work
• Handler signatures are backward-compatible (new `context` param is optional)
• Deduplication always runs regardless of strategy
• Breaking: WhatsApp and Telegram adapters now default to `lockScope: 'channel'`

Test plan

• Queue: processes latest message with skipped context after handler finishes
• Queue: enqueues when lock is held, drains on release
• Queue: drop-newest when queue is full
• Queue: drop-oldest evicts oldest entries
• Queue: expired entries skipped during drain
• Queue: works with `onSubscribedMessage` handlers
• Queue: works with `onNewMessage` pattern handlers
• Debounce: first message debounces (not processed immediately)
• Debounce: only last message in burst is processed
• Concurrent: processes without acquiring lock
• Lock scope: thread scope uses threadId as lock key
• Lock scope: channel scope uses channelId as lock key (adapter default)
• Lock scope: channel scope uses channelId as lock key (config override)
• Lock scope: async resolver function
• Lock scope: queue operations use channel-scoped lock key
• State adapters: enqueue/dequeue/queueDepth (memory — 8 behavioral tests)
• State adapters: enqueue/dequeue/queueDepth (pg — 12 mock-based tests)
• State adapters: method existence (redis, ioredis)
• Existing `onLockConflict` tests still pass

[vercel]

The latest updates on your projects. Learn more about Vercel for GitHub.

Project	Deployment	Actions	Updated (UTC)
chat	Ready	Preview, Comment, Open in v0	Mar 21, 2026 2:01am
chat-sdk-nextjs-chat	Ready	Preview, Comment, Open in v0	Mar 21, 2026 2:01am