From 9f3a31c87ef93b23f569b073407b17585d8d155c Mon Sep 17 00:00:00 2001 From: rmyndharis Date: Fri, 3 Jul 2026 20:42:23 +0700 Subject: [PATCH] feat(chatwoot-adapter): retry failed inbound relays instead of dropping them (#609) Inbound relay was at-most-once: a failed Chatwoot POST was logged and the message dropped. When Chatwoot is transiently unreachable this silently loses customer messages. Make inbound at-least-once via a durable, storage-backed retry queue drained on a timer. - inbound.ts: on a relay failure, enqueue the message (media over a size cap is stripped so an oversized value can't be rejected and lose the message) rather than only logging. markSeen stays before the relay (dedups WA re-deliveries + makes backfill skip the live message); the pending-queue entry is the not-yet-relayed signal. - retry.ts: drainRetries re-posts each queued message under its per-chat lock, streaming one entry at a time so a large media backlog is never fully resident in memory; a message that keeps failing is dead-lettered after 5 attempts. - mapping-store.ts: retry-queue persistence (individual retry:* keys, one file each; count/list-keys without loading blobs; overflow drops the oldest, set-before-delete). - index.ts: 30s .unref() drain timer with a single-flight guard, cleared on disable/unload and skipped while config is invalid; healthCheck surfaces the pending backlog + dead-letters and reads unhealthy when the queue saturates. At-least-once means a message that reached Chatwoot but whose response was lost may rarely re-post as a duplicate; posts carry source_id as a dedup handle. chatwoot-adapter 0.5.0; no core change (minOpenWAVersion stays 0.8.7). --- README.md | 2 +- chatwoot-adapter/CHANGELOG.md | 12 ++++ chatwoot-adapter/README.md | 2 +- chatwoot-adapter/inbound.test.ts | 10 +++ chatwoot-adapter/inbound.ts | 49 +++++++++----- chatwoot-adapter/index.test.ts | 44 ++++++++++++- chatwoot-adapter/index.ts | 66 ++++++++++++++++++- chatwoot-adapter/manifest.json | 2 +- chatwoot-adapter/mapping-store.test.ts | 46 ++++++++++++- chatwoot-adapter/mapping-store.ts | 89 ++++++++++++++++++++++++- chatwoot-adapter/retry.test.ts | 90 ++++++++++++++++++++++++++ chatwoot-adapter/retry.ts | 71 ++++++++++++++++++++ plugins.json | 4 +- 13 files changed, 461 insertions(+), 26 deletions(-) create mode 100644 chatwoot-adapter/retry.test.ts create mode 100644 chatwoot-adapter/retry.ts diff --git a/README.md b/README.md index aee382c..c2d6446 100644 --- a/README.md +++ b/README.md @@ -36,7 +36,7 @@ This repository provides: | ------ | ----------- | ------- | ------ | | [`after-hours`](./after-hours) | Auto-replies with a configurable away/closing message to messages received outside business hours. | 0.1.2 | stable | | [`chat-flow`](./chat-flow) | Interactive, stateful auto-reply: a trigger word starts a greeting + numbered menu, replies traverse a configurable menu tree, and per-chat state expires after 15 minutes. | 1.0.5 | stable | -| [`chatwoot-adapter`](./chatwoot-adapter) | Two-way sync between a WhatsApp session and a Chatwoot inbox: relays WhatsApp messages (1:1 and groups, with media) into Chatwoot as an API-channel inbox, sends agent replies back to WhatsApp, and hands a chat over to a human agent — silencing other OpenWA bots — when an agent takes it in Chatwoot. First consumer of the OpenWA Integration SDK v1; runs sandboxed in the plugin worker. | 0.4.0 | beta | +| [`chatwoot-adapter`](./chatwoot-adapter) | Two-way sync between a WhatsApp session and a Chatwoot inbox: relays WhatsApp messages (1:1 and groups, with media) into Chatwoot as an API-channel inbox, sends agent replies back to WhatsApp, and hands a chat over to a human agent — silencing other OpenWA bots — when an agent takes it in Chatwoot. First consumer of the OpenWA Integration SDK v1; runs sandboxed in the plugin worker. | 0.5.0 | beta | | [`faq-bot`](./faq-bot) | Auto-replies to inbound WhatsApp messages from configurable FAQ keyword/regex rules. | 0.1.6 | stable | | [`group-translate`](./group-translate) | Auto-translates group messages between participants' languages via a LibreTranslate backend. Configure in-chat with /tr commands. Admin-gated; disabled until enabled. | 1.0.5 | stable | | [`gsheets-logger`](./gsheets-logger) | Logs WhatsApp message events to a Google Sheet via a service account. | 0.2.3 | stable | diff --git a/chatwoot-adapter/CHANGELOG.md b/chatwoot-adapter/CHANGELOG.md index 9893e0d..c6b628c 100644 --- a/chatwoot-adapter/CHANGELOG.md +++ b/chatwoot-adapter/CHANGELOG.md @@ -6,6 +6,18 @@ All notable changes to the Chatwoot Adapter plugin are documented here. The form ## [Unreleased] +## [0.5.0] — 2026-07-03 + +### Added + +- **Inbound relay is now retried instead of dropped** when Chatwoot is transiently unreachable (#609). + A failed inbound message is held in a durable, storage-backed queue and re-posted on a timer until it + succeeds; a message that keeps failing is dead-lettered after several attempts. The plugin's health + check surfaces the pending backlog and any dead-lettered messages. + - This makes inbound delivery **at-least-once** (previously at-most-once — a failed post was logged and + dropped). As a result, a message that actually reached Chatwoot but whose response was lost may, on + rare occasions, be re-posted as a duplicate. + ## [0.4.0] — 2026-07-03 ### Added diff --git a/chatwoot-adapter/README.md b/chatwoot-adapter/README.md index c2546c3..4171871 100644 --- a/chatwoot-adapter/README.md +++ b/chatwoot-adapter/README.md @@ -11,7 +11,7 @@ mirror it. Runs sandboxed in the plugin worker; no server, no extra port. | Field | Value | | ----- | ----- | | **Identifier** | `chatwoot-adapter` | -| **Version** | 0.4.0 | +| **Version** | 0.5.0 | | **Released** | 2026-07-03 | | **Status** | beta | | **Author** | Yudhi Armyndharis | diff --git a/chatwoot-adapter/inbound.test.ts b/chatwoot-adapter/inbound.test.ts index 1b8e1ca..87658ee 100644 --- a/chatwoot-adapter/inbound.test.ts +++ b/chatwoot-adapter/inbound.test.ts @@ -38,6 +38,16 @@ function deps(over: { client?: Record; store?: Record ({ contacts, convs }), posted }; } +test('a failed relay queues the message for retry (at-least-once), not dropped', async () => { + const enqueued: string[] = []; + const { deps: d } = deps({ + client: { postText: async () => { throw new Error('chatwoot 503'); } }, + store: { enqueueRetry: async (e: { msg: { id: string } }) => void enqueued.push(e.msg.id) }, + }); + await handleInbound(d, 'sess', 'Engine', msg); + assert.deepEqual(enqueued, ['m1']); // conversation resolved, post failed → queued (not silently dropped) +}); + test('creates contact + conversation and posts an incoming message', async () => { const { deps: d, posted, counts } = deps(); await handleInbound(d, 'sess', 'Engine', msg); diff --git a/chatwoot-adapter/inbound.ts b/chatwoot-adapter/inbound.ts index 8e165d5..f8e33fa 100644 --- a/chatwoot-adapter/inbound.ts +++ b/chatwoot-adapter/inbound.ts @@ -2,12 +2,26 @@ import type { IncomingMessage } from '../types/openwa'; import { shouldRelayInbound } from './filters.ts'; import { relayMessage, ensureConversation, refreshContactName, type InboundDeps } from './relay.ts'; import { backfillHistory } from './backfill.ts'; +import { MAX_PENDING_RETRIES, slimForRetry } from './retry.ts'; export type { InboundDeps }; +// The resolve + backfill + relay core, lock-free, that THROWS on failure. Shared by the live inbound +// handler and the retry drain (retry.ts) so a retried message follows the exact same path. +export async function relayInbound(deps: InboundDeps, sessionId: string, msg: IncomingMessage): Promise { + const { conversationId, created } = await resolveConversation(deps, sessionId, msg); + // Lazy backfill: the first time this chat maps, replay its recent history (older messages, both + // directions, deduped) BEFORE posting this one — so the thread reads chronologically and this message's + // quote resolves against a just-posted source_id. This message is already markSeen, so backfill skips it. + if (created && deps.backfillLimit > 0) { + await backfillHistory(deps, sessionId, msg.chatId, conversationId); + } + await relayMessage(deps, conversationId, msg, 'incoming'); +} + // WhatsApp → Chatwoot. Filter, then run resolve+post under the per-chat lock so two near-simultaneous -// first messages can't create duplicate Chatwoot contacts/conversations. Dedup is a mark-before-act -// check-and-set inside the lock (at-most-once; a failed post won't re-relay). +// first messages can't create duplicate Chatwoot contacts/conversations. Inbound is AT-LEAST-ONCE: a +// failed relay is queued for retry (retry.ts drains it) rather than dropped. export async function handleInbound( deps: InboundDeps, sessionId: string, @@ -16,22 +30,25 @@ export async function handleInbound( ): Promise { if (!shouldRelayInbound(msg, source, deps.relayGroups)) return; await deps.lock.run(`${sessionId}:${msg.chatId}`, async () => { + // markSeen stays BEFORE the relay: it dedups WA re-deliveries and makes backfill skip this live + // message. It is NOT the "relayed" signal — a failed relay is enqueued below, and the pending-queue + // entry is what drives retry, independent of this marker. Scoped by session so two tenants' WA message + // ids can't collide in the shared plugin store. + if (await deps.store.hasSeen('wa', msg.id, sessionId)) return; + await deps.store.markSeen('wa', msg.id, sessionId); try { - // Mark-before-act: inbound is deliberately at-most-once (a failed post won't re-relay). Scoped by - // session so two tenants' WA message ids can't collide in the shared plugin store. - if (await deps.store.hasSeen('wa', msg.id, sessionId)) return; - await deps.store.markSeen('wa', msg.id, sessionId); - const { conversationId, created } = await resolveConversation(deps, sessionId, msg); - // Lazy backfill: the first time this chat maps, replay its recent history (older messages, both - // directions, deduped) BEFORE posting this one — so the thread reads chronologically and this - // message's quote resolves against a just-posted source_id. This message is already markSeen, so - // backfill skips it and it posts last, below. - if (created && deps.backfillLimit > 0) { - await backfillHistory(deps, sessionId, msg.chatId, conversationId); - } - await relayMessage(deps, conversationId, msg, 'incoming'); + await relayInbound(deps, sessionId, msg); } catch (err) { - deps.log('inbound relay failed', err); + deps.log('inbound relay failed; queued for retry', err); + // Strip an oversized media blob before persisting so a huge value can't be rejected by the storage + // layer (which would lose the message — it's already markSeen); the retry then posts a placeholder. + const dropped = await deps.store + .enqueueRetry({ sessionId, chatId: msg.chatId, msg: slimForRetry(msg), enqueuedAt: Date.now() }, MAX_PENDING_RETRIES) + .catch(e => { + deps.log('enqueue retry failed', e); + return null; + }); + if (dropped) deps.log(`retry queue full; dropped oldest pending inbound (msg ${dropped})`); } }); } diff --git a/chatwoot-adapter/index.test.ts b/chatwoot-adapter/index.test.ts index 02fe837..91038f7 100644 --- a/chatwoot-adapter/index.test.ts +++ b/chatwoot-adapter/index.test.ts @@ -1,6 +1,7 @@ import { test } from 'node:test'; import assert from 'node:assert/strict'; import ChatwootAdapter from './index.ts'; +import { MAX_PENDING_RETRIES } from './retry.ts'; import type { PluginContext } from '../types/openwa'; function fakeCtx(config: Record) { @@ -8,9 +9,15 @@ function fakeCtx(config: Record) { const routes: string[] = []; const cbs: Record Promise<{ continue: boolean }>> = {}; let fetches = 0; + const storageMap = new Map(); const ctx = { config, - storage: { get: async () => null, set: async () => {}, delete: async () => {}, list: async () => [] }, + storage: { + get: async (k: string) => (storageMap.has(k) ? storageMap.get(k) : null), + set: async (k: string, v: unknown) => void storageMap.set(k, v), + delete: async (k: string) => void storageMap.delete(k), + list: async () => [...storageMap.keys()], + }, mappings: { upsert: async () => {}, get: async () => null, getByProvider: async () => null }, net: { fetch: async () => { fetches++; return { ok: true, status: 200, headers: {}, body: '{}' }; } }, conversations: { send: async () => ({}) }, @@ -20,7 +27,7 @@ function fakeCtx(config: Record) { registerHook: (event: string, cb: (h: unknown) => Promise<{ continue: boolean }>) => { hooks.push(event); cbs[event] = cb; }, registerWebhook: (route: string) => void routes.push(route), } as unknown as PluginContext; - return { ctx, hooks, routes, cbs, fetches: () => fetches }; + return { ctx, hooks, routes, cbs, fetches: () => fetches, storageMap }; } const goodConfig = { baseUrl: 'https://chat.acme.com', apiToken: 'tok', accountId: 3, inboxId: 7 }; @@ -51,6 +58,39 @@ test('relayOwnMessages=false gates the message:sent handler off (no Chatwoot API assert.equal(fetches(), 0); }); +test('healthCheck reports the pending retry backlog (healthy — pending is transient)', async () => { + const { ctx, storageMap } = fakeCtx(goodConfig); + storageMap.set('retry:sess:m1', { sessionId: 'sess', chatId: 'c@wa', msg: { id: 'm1' }, attempts: 1, enqueuedAt: 1 }); + const adapter = new ChatwootAdapter(); + await adapter.onEnable(ctx); + const h = await adapter.healthCheck(); + assert.equal(h.healthy, true); + assert.match(h.message ?? '', /1 inbound message\(s\) pending retry/); + await adapter.onDisable(); +}); + +test('healthCheck is UNHEALTHY when the retry queue is saturated (at capacity → dropping oldest)', async () => { + const { ctx, storageMap } = fakeCtx(goodConfig); + for (let i = 0; i < MAX_PENDING_RETRIES; i++) { + storageMap.set(`retry:sess:m${i}`, { sessionId: 'sess', chatId: 'c', msg: { id: `m${i}` }, attempts: 1, enqueuedAt: i }); + } + const adapter = new ChatwootAdapter(); + await adapter.onEnable(ctx); + const h = await adapter.healthCheck(); + assert.equal(h.healthy, false); // active data loss must not read as healthy + assert.match(h.message ?? '', /queue full, dropping oldest/); + await adapter.onDisable(); +}); + +test('healthCheck is healthy with an empty queue; onDisable/onUnload run cleanly (timer cleared)', async () => { + const { ctx } = fakeCtx(goodConfig); + const adapter = new ChatwootAdapter(); + await adapter.onEnable(ctx); + assert.deepEqual(await adapter.healthCheck(), { healthy: true, message: undefined }); + await adapter.onDisable(); + await adapter.onUnload(); +}); + test('onEnable throws on missing / invalid config', async () => { const { ctx } = fakeCtx({ baseUrl: 'https://x' }); // missing apiToken, accountId, inboxId await assert.rejects(new ChatwootAdapter().onEnable(ctx), /missing\/invalid config/); diff --git a/chatwoot-adapter/index.ts b/chatwoot-adapter/index.ts index 3bb77aa..81e3c14 100644 --- a/chatwoot-adapter/index.ts +++ b/chatwoot-adapter/index.ts @@ -2,10 +2,11 @@ import type { IPlugin, PluginContext, IncomingMessage, HookContext, HookResult, import { ChatwootClient } from './chatwoot-client.ts'; import { MappingStore } from './mapping-store.ts'; import { KeyedAsyncLock } from './chat-lock.ts'; -import { handleInbound } from './inbound.ts'; +import { handleInbound, relayInbound } from './inbound.ts'; import { handleSent } from './sent.ts'; import { backfillAllChats } from './backfill.ts'; import { handleOutbound } from './outbound.ts'; +import { drainRetries, RETRY_INTERVAL_MS, MAX_RETRY_ATTEMPTS, MAX_PENDING_RETRIES } from './retry.ts'; interface ChatwootFullConfig { baseUrl: string; @@ -58,10 +59,17 @@ function readConfig(raw: Record): ChatwootFullConfig { } export default class ChatwootAdapter implements IPlugin { + private retryTimer: ReturnType | null = null; + private store: MappingStore | null = null; + private deadLetterCount = 0; + private draining = false; + async onEnable(ctx: PluginContext): Promise { + this.clearRetryTimer(); // idempotent re-enable: never leak a timer from a prior enable readConfig(ctx.config); // fail fast on the base config const lock = new KeyedAsyncLock(); const store = new MappingStore(ctx.storage, ctx.mappings); + this.store = store; // Re-read config per event so a per-session/instance override (PR E) is picked up live. const clientFor = () => new ChatwootClient(ctx.net.fetch.bind(ctx.net), readConfig(ctx.config)); @@ -131,6 +139,62 @@ export default class ChatwootAdapter implements IPlugin { ), ); + // Retry failed inbound relays (at-least-once). The durable, storage-backed queue is drained on a timer: + // each queued message is re-posted via the same inbound path, and a message that keeps failing is + // dead-lettered after MAX_RETRY_ATTEMPTS. Retries use the base config (per-session overrides aren't + // re-resolved outside a hook). .unref() so the timer never keeps the worker alive; cleared on disable. + const drain = (): Promise => { + // Single-flight: a slow drain (large backlog) must not overlap the next tick, or two runs would + // snapshot the same entries and double-post. Skip the tick if a drain is still in progress. + if (this.draining) return Promise.resolve(); + // Skip entirely if the config is currently invalid (e.g. apiToken cleared): every relay would throw + // on readConfig and walk the whole queue to dead-letter within a few ticks. Wait for valid config. + try { + readConfig(ctx.config); + } catch { + return Promise.resolve(); + } + this.draining = true; + return drainRetries( + { store, lock, log: (m, e) => ctx.logger.error(m, e) }, + (sessionId, _chatId, msg) => relayInbound(buildDeps(readConfig(ctx.config), sessionId), sessionId, msg), + MAX_RETRY_ATTEMPTS, + ) + .then( + ({ deadLettered }) => void (this.deadLetterCount += deadLettered), + e => ctx.logger.error('retry drain failed', e), + ) + .finally(() => void (this.draining = false)); + }; + this.retryTimer = setInterval(() => void drain(), RETRY_INTERVAL_MS); + this.retryTimer.unref?.(); + ctx.logger.log('chatwoot-adapter enabled'); } + + async onDisable(): Promise { + this.clearRetryTimer(); + } + + async onUnload(): Promise { + this.clearRetryTimer(); + } + + private clearRetryTimer(): void { + if (this.retryTimer) { + clearInterval(this.retryTimer); + this.retryTimer = null; + } + } + + // Surface the retry backlog + permanent failures in the dashboard's plugin health. A saturated queue is + // unhealthy: at capacity, every new failure drops the oldest pending message (active data loss). + async healthCheck(): Promise<{ healthy: boolean; message?: string }> { + const pending = this.store ? await this.store.countRetries() : 0; + const saturated = pending >= MAX_PENDING_RETRIES; + const parts: string[] = []; + if (pending > 0) parts.push(`${pending} inbound message(s) pending retry${saturated ? ' — queue full, dropping oldest' : ''}`); + if (this.deadLetterCount > 0) parts.push(`${this.deadLetterCount} dead-lettered after ${MAX_RETRY_ATTEMPTS} attempts`); + return { healthy: this.deadLetterCount === 0 && !saturated, message: parts.join('; ') || undefined }; + } } diff --git a/chatwoot-adapter/manifest.json b/chatwoot-adapter/manifest.json index 95b3218..4a790dc 100644 --- a/chatwoot-adapter/manifest.json +++ b/chatwoot-adapter/manifest.json @@ -1,7 +1,7 @@ { "id": "chatwoot-adapter", "name": "Chatwoot Adapter", - "version": "0.4.0", + "version": "0.5.0", "type": "extension", "main": "dist/index.js", "description": "Two-way sync between a WhatsApp session and a Chatwoot inbox: relays WhatsApp messages (1:1 and groups, with media) into Chatwoot as an API-channel inbox, sends agent replies back to WhatsApp, and hands a chat over to a human agent — silencing other OpenWA bots — when an agent takes it in Chatwoot. First consumer of the OpenWA Integration SDK v1; runs sandboxed in the plugin worker.", diff --git a/chatwoot-adapter/mapping-store.test.ts b/chatwoot-adapter/mapping-store.test.ts index a177ea5..64428c4 100644 --- a/chatwoot-adapter/mapping-store.test.ts +++ b/chatwoot-adapter/mapping-store.test.ts @@ -1,8 +1,11 @@ import { test } from 'node:test'; import assert from 'node:assert/strict'; -import type { PluginStorage, PluginMappingsCapability } from '../types/openwa'; +import type { PluginStorage, PluginMappingsCapability, IncomingMessage } from '../types/openwa'; import { MappingStore } from './mapping-store.ts'; +const msg = (id: string, chatId = 'c@wa'): IncomingMessage => + ({ id, from: 'x', to: 'y', chatId, body: 'hi', type: 'chat', timestamp: 0, fromMe: false, isGroup: false }) as IncomingMessage; + function fakeStorage(): PluginStorage { const m = new Map(); return { @@ -55,3 +58,44 @@ test('patch merges over the existing forward doc', async () => { await store.patch('s', 'c', { handoverState: 'human' }); assert.deepEqual(await store.getByChat('s', 'c'), { conversationId: 1, contactId: 2, sourceId: 'x', handoverState: 'human' }); }); + +test('enqueueRetry persists an entry; listRetries returns it; deleteRetry removes it; countRetries counts', async () => { + const store = new MappingStore(fakeStorage(), fakeMappings()); + const dropped = await store.enqueueRetry({ sessionId: 'sess', chatId: 'c@wa', msg: msg('m1'), enqueuedAt: 100 }, 500); + assert.equal(dropped, null); + assert.equal(await store.countRetries(), 1); + const [e] = await store.listRetries(); + assert.equal(e.msg.id, 'm1'); + assert.equal(e.attempts, 0); + assert.equal(e.sessionId, 'sess'); + await store.deleteRetry(e.key); + assert.equal(await store.countRetries(), 0); +}); + +test('enqueueRetry is a no-op for an already-queued message id (does not reset attempts)', async () => { + const store = new MappingStore(fakeStorage(), fakeMappings()); + await store.enqueueRetry({ sessionId: 'sess', chatId: 'c@wa', msg: msg('m1'), enqueuedAt: 100 }, 500); + const [e1] = await store.listRetries(); + await store.bumpRetryAttempts(e1.key, 3); + await store.enqueueRetry({ sessionId: 'sess', chatId: 'c@wa', msg: msg('m1'), enqueuedAt: 200 }, 500); // duplicate id + assert.equal(await store.countRetries(), 1); + assert.equal((await store.listRetries())[0].attempts, 3); // not reset +}); + +test('bumpRetryAttempts updates the attempt count in place', async () => { + const store = new MappingStore(fakeStorage(), fakeMappings()); + await store.enqueueRetry({ sessionId: 'sess', chatId: 'c@wa', msg: msg('m1'), enqueuedAt: 100 }, 500); + const [e] = await store.listRetries(); + await store.bumpRetryAttempts(e.key, 2); + assert.equal((await store.listRetries())[0].attempts, 2); +}); + +test('enqueueRetry drops the OLDEST entry (by enqueuedAt) when the queue is at capacity', async () => { + const store = new MappingStore(fakeStorage(), fakeMappings()); + await store.enqueueRetry({ sessionId: 's', chatId: 'c', msg: msg('old'), enqueuedAt: 100 }, 2); + await store.enqueueRetry({ sessionId: 's', chatId: 'c', msg: msg('mid'), enqueuedAt: 200 }, 2); + const dropped = await store.enqueueRetry({ sessionId: 's', chatId: 'c', msg: msg('new'), enqueuedAt: 300 }, 2); + assert.equal(dropped, 'old'); // returns the dropped id for logging + const ids = (await store.listRetries()).map(e => e.msg.id).sort(); + assert.deepEqual(ids, ['mid', 'new']); +}); diff --git a/chatwoot-adapter/mapping-store.ts b/chatwoot-adapter/mapping-store.ts index 905b532..29e3c17 100644 --- a/chatwoot-adapter/mapping-store.ts +++ b/chatwoot-adapter/mapping-store.ts @@ -1,4 +1,14 @@ -import type { PluginStorage, PluginMappingsCapability } from '../types/openwa'; +import type { PluginStorage, PluginMappingsCapability, IncomingMessage } from '../types/openwa'; + +// A failed inbound relay held for retry. The full message (incl. media) is stored so the retry re-posts +// it faithfully. `enqueuedAt` is a wall-clock ms used only to pick the oldest entry to drop on overflow. +export interface RetryEntry { + sessionId: string; + chatId: string; + msg: IncomingMessage; + attempts: number; + enqueuedAt: number; +} export interface ChatLink { conversationId: number; @@ -85,4 +95,81 @@ export class MappingStore { async setBulkBackfilled(sessionId: string): Promise { await this.storage.set(`backfill:all:${sessionId}`, 1); } + + // ---- Inbound retry queue (durable, over ctx.storage) -------------------------------------------- + // Individual keys `retry::` — one per failed relay — so concurrent enqueues never + // read-modify-write a shared array. The list is filtered by the `retry:` prefix defensively (the host + // list(prefix) already narrows, but a fake that ignores the arg would otherwise leak other keys). + + private retryKey(sessionId: string, msgId: string): string { + return `retry:${sessionId}:${msgId}`; + } + + private async retryKeys(): Promise { + return (await this.storage.list('retry:')).filter(k => k.startsWith('retry:')); + } + + private async readRetries(keys: string[]): Promise> { + const out: Array = []; + for (const key of keys) { + const e = await this.storage.get(key); + if (e) out.push({ ...e, key }); + } + return out; + } + + // Enqueue a failed inbound relay. No-op if this message id is already queued (never resets its attempt + // count). When the queue is at `maxPending`, drop the OLDEST entry (returns its msg id so the caller can + // log the loss) to bound storage. `attempts` starts at 0. + async enqueueRetry(entry: Omit, maxPending: number): Promise { + const key = this.retryKey(entry.sessionId, entry.msg.id); + if (await this.storage.get(key)) return null; + const keys = await this.retryKeys(); + let oldestKey: string | null = null; + let dropped: string | null = null; + if (keys.length >= maxPending) { + // Only NOW load the values (to pick the oldest). The common under-cap path never deserializes any + // queued blob — counting by key length keeps enqueue O(1) in payload size. + const pending = await this.readRetries(keys); + if (pending.length) { + const oldest = pending.reduce((a, b) => (a.enqueuedAt <= b.enqueuedAt ? a : b)); + oldestKey = oldest.key; + dropped = oldest.msg.id; + } + } + // Write the NEW entry BEFORE deleting the oldest: if the set rejects (e.g. an oversized value), the + // oldest is preserved instead of both being lost, and no drop is reported. + await this.storage.set(key, { ...entry, attempts: 0 } satisfies RetryEntry); + if (oldestKey) await this.storage.delete(oldestKey); + return dropped; + } + + async listRetries(): Promise> { + return this.readRetries(await this.retryKeys()); + } + + // Streaming primitives for the drain: list keys (a directory scan, no value loads), then fetch one + // entry at a time — so a saturated queue of media messages is never all resident in memory at once. + listRetryKeys(): Promise { + return this.retryKeys(); + } + + async getRetry(key: string): Promise<(RetryEntry & { key: string }) | null> { + const e = await this.storage.get(key); + return e ? { ...e, key } : null; + } + + async bumpRetryAttempts(key: string, attempts: number): Promise { + const e = await this.storage.get(key); + if (e) await this.storage.set(key, { ...e, attempts }); + } + + async deleteRetry(key: string): Promise { + await this.storage.delete(key); + } + + // Count by key only — never load the (media-bearing) values, so a large backlog can't spike memory. + async countRetries(): Promise { + return (await this.retryKeys()).length; + } } diff --git a/chatwoot-adapter/retry.test.ts b/chatwoot-adapter/retry.test.ts new file mode 100644 index 0000000..803abae --- /dev/null +++ b/chatwoot-adapter/retry.test.ts @@ -0,0 +1,90 @@ +import { test } from 'node:test'; +import assert from 'node:assert/strict'; +import { drainRetries, slimForRetry, RETRY_MAX_MEDIA_B64 } from './retry.ts'; +import { MappingStore } from './mapping-store.ts'; +import { KeyedAsyncLock } from './chat-lock.ts'; +import type { PluginStorage, PluginMappingsCapability, IncomingMessage } from '../types/openwa'; + +function fakeStorage(): PluginStorage { + const m = new Map(); + return { + get: async (k: string) => (m.has(k) ? (m.get(k) as T) : null), + set: async (k: string, v: unknown) => void m.set(k, v), + delete: async (k: string) => void m.delete(k), + list: async () => [...m.keys()], + }; +} +const fakeMappings: PluginMappingsCapability = { upsert: async () => {}, get: async () => null, getByProvider: async () => null }; +const msg = (id: string, chatId = 'c@wa'): IncomingMessage => + ({ id, from: 'x', to: 'y', chatId, body: 'hi', type: 'chat', timestamp: 0, fromMe: false, isGroup: false }) as IncomingMessage; + +function deps() { + const logs: string[] = []; + return { store: new MappingStore(fakeStorage(), fakeMappings), lock: new KeyedAsyncLock(), log: (m: string) => logs.push(m), logs }; +} + +test('drain: a successful relay removes the entry and calls relay with (sessionId, chatId, msg)', async () => { + const d = deps(); + await d.store.enqueueRetry({ sessionId: 'sess', chatId: 'c@wa', msg: msg('m1'), enqueuedAt: 1 }, 500); + const calls: string[] = []; + const r = await drainRetries(d, async (s, c, m) => void calls.push(`${s}:${c}:${m.id}`), 5); + assert.deepEqual(calls, ['sess:c@wa:m1']); + assert.equal(await d.store.countRetries(), 0); + assert.equal(r.deadLettered, 0); +}); + +test('drain: a failing relay increments attempts and keeps the entry (below max)', async () => { + const d = deps(); + await d.store.enqueueRetry({ sessionId: 'sess', chatId: 'c@wa', msg: msg('m1'), enqueuedAt: 1 }, 500); + const r = await drainRetries(d, async () => { throw new Error('chatwoot down'); }, 5); + assert.equal(await d.store.countRetries(), 1); + assert.equal((await d.store.listRetries())[0].attempts, 1); + assert.equal(r.deadLettered, 0); +}); + +test('drain: dead-letters (drops + logs + counts) once attempts reach maxAttempts', async () => { + const d = deps(); + await d.store.enqueueRetry({ sessionId: 'sess', chatId: 'c@wa', msg: msg('m1'), enqueuedAt: 1 }, 500); + await d.store.bumpRetryAttempts((await d.store.listRetries())[0].key, 4); // one below max=5 + const r = await drainRetries(d, async () => { throw new Error('chatwoot down'); }, 5); + assert.equal(await d.store.countRetries(), 0); // dropped from the queue + assert.equal(r.deadLettered, 1); + assert.match(d.logs.join('\n'), /dead-lettered after 5 attempts/); +}); + +test('slimForRetry strips an oversized media blob (so it retries as a placeholder) but keeps small media', () => { + const big = { ...msg('m1'), media: { mimetype: 'image/jpeg', data: 'A'.repeat(RETRY_MAX_MEDIA_B64 + 1) } } as IncomingMessage; + const slimmed = slimForRetry(big); + assert.equal(slimmed.media?.data, undefined); + assert.equal(slimmed.media?.omitted, true); + const small = { ...msg('m2'), media: { mimetype: 'image/jpeg', data: 'AAA' } } as IncomingMessage; + assert.equal(slimForRetry(small).media?.data, 'AAA'); // small media kept for a faithful retry + assert.equal(slimForRetry(msg('m3')).media, undefined); // no media → unchanged +}); + +test('drain: a successful relay whose deleteRetry throws is NOT re-posted or bumped (treated as delivered)', async () => { + let posts = 0; + let bumps = 0; + const entry = { key: 'retry:sess:m1', sessionId: 'sess', chatId: 'c@wa', msg: msg('m1'), attempts: 0, enqueuedAt: 1 }; + const store = { + listRetryKeys: async () => [entry.key], + getRetry: async () => entry, + deleteRetry: async () => { throw new Error('storage delete failed'); }, + bumpRetryAttempts: async () => void bumps++, + } as unknown as MappingStore; + const logs: string[] = []; + const r = await drainRetries({ store, lock: new KeyedAsyncLock(), log: (m: string) => logs.push(m) }, async () => void posts++, 5); + assert.equal(posts, 1); // relayed once + assert.equal(bumps, 0); // a delete failure after a successful post must NOT bump attempts / re-post + assert.equal(r.deadLettered, 0); + assert.match(logs.join('\n'), /deleteRetry after a successful relay failed/); +}); + +test('drain: processes multiple entries independently (one succeeds, one fails)', async () => { + const d = deps(); + await d.store.enqueueRetry({ sessionId: 'sess', chatId: 'a@wa', msg: msg('ok', 'a@wa'), enqueuedAt: 1 }, 500); + await d.store.enqueueRetry({ sessionId: 'sess', chatId: 'b@wa', msg: msg('bad', 'b@wa'), enqueuedAt: 2 }, 500); + await drainRetries(d, async (_s, _c, m) => { if (m.id === 'bad') throw new Error('down'); }, 5); + const ids = (await d.store.listRetries()).map(e => e.msg.id); + assert.deepEqual(ids, ['bad']); // 'ok' relayed + removed; 'bad' remains for retry +}); diff --git a/chatwoot-adapter/retry.ts b/chatwoot-adapter/retry.ts new file mode 100644 index 0000000..3efaf44 --- /dev/null +++ b/chatwoot-adapter/retry.ts @@ -0,0 +1,71 @@ +import type { IncomingMessage } from '../types/openwa'; +import type { MappingStore } from './mapping-store.ts'; +import type { KeyedAsyncLock } from './chat-lock.ts'; + +// How often the retry queue is drained; how many times a message is retried before it's dead-lettered; +// and the max pending entries kept (older ones are dropped on overflow to bound ctx.storage). +export const RETRY_INTERVAL_MS = 30_000; +export const MAX_RETRY_ATTEMPTS = 5; +export const MAX_PENDING_RETRIES = 500; +// Cap the media blob persisted per queued entry. A larger blob is stripped (retried as a placeholder) +// rather than stored — so a queued entry can't exceed the host's per-value storage cap and get rejected, +// which would lose the whole message (it's already markSeen). ~700 KB base64 ≈ ~512 KB binary. +export const RETRY_MAX_MEDIA_B64 = 700_000; + +// Return a copy safe to persist in the retry queue: an oversized media blob is dropped (marked omitted) +// so the retry posts a type placeholder instead of failing to persist. Small media is kept for a faithful +// retry. Pure. +export function slimForRetry(msg: IncomingMessage): IncomingMessage { + const media = msg.media; + if (media?.data && media.data.length > RETRY_MAX_MEDIA_B64) { + return { ...msg, media: { ...media, data: undefined, omitted: true } }; + } + return msg; +} + +export interface DrainDeps { + store: MappingStore; + lock: KeyedAsyncLock; + log: (m: string, e?: unknown) => void; +} + +// Drain the inbound retry queue: re-relay each queued failed message under its per-chat lock (so it +// serializes with a concurrent live inbound for the same chat). Success drops the entry; a failure bumps +// its attempt count, and once attempts reach `maxAttempts` the message is dead-lettered (logged + dropped) +// rather than retried forever. `relay` re-posts the message and throws on failure. Returns how many were +// dead-lettered this run, for the plugin health check. +export async function drainRetries( + deps: DrainDeps, + relay: (sessionId: string, chatId: string, msg: IncomingMessage) => Promise, + maxAttempts: number, +): Promise<{ deadLettered: number }> { + // Stream by key so a large media backlog is never fully resident: fetch one entry at a time. + const keys = await deps.store.listRetryKeys(); + let deadLettered = 0; + for (const key of keys) { + const e = await deps.store.getRetry(key); + if (!e) continue; // key vanished since the scan (already drained/dropped) — nothing to do + await deps.lock.run(`${e.sessionId}:${e.chatId}`, async () => { + let relayed = false; + try { + await relay(e.sessionId, e.chatId, e.msg); + relayed = true; + } catch (err) { + const attempts = e.attempts + 1; + if (attempts >= maxAttempts) { + deps.log(`inbound relay dead-lettered after ${attempts} attempts (chat ${e.chatId}, msg ${e.msg.id})`, err); + await deps.store.deleteRetry(e.key); + deadLettered++; + } else { + await deps.store.bumpRetryAttempts(e.key, attempts); + } + } + // Delete OUTSIDE the try: a storage.delete failure after a SUCCESSFUL post must not be caught as a + // relay failure (which would bump attempts and re-post the already-delivered message next drain). + if (relayed) { + await deps.store.deleteRetry(e.key).catch(err => deps.log('deleteRetry after a successful relay failed', err)); + } + }); + } + return { deadLettered }; +} diff --git a/plugins.json b/plugins.json index 0a1e1e5..ef7a3d0 100644 --- a/plugins.json +++ b/plugins.json @@ -369,7 +369,7 @@ { "id": "chatwoot-adapter", "name": "Chatwoot Adapter", - "version": "0.4.0", + "version": "0.5.0", "type": "extension", "status": "beta", "description": "Two-way sync between a WhatsApp session and a Chatwoot inbox: relays WhatsApp messages (1:1 and groups, with media) into Chatwoot as an API-channel inbox, sends agent replies back to WhatsApp, and hands a chat over to a human agent — silencing other OpenWA bots — when an agent takes it in Chatwoot. First consumer of the OpenWA Integration SDK v1; runs sandboxed in the plugin worker.", @@ -391,7 +391,7 @@ "repoPath": "chatwoot-adapter", "repoUrl": "https://github.com/rmyndharis/OpenWA-plugins", "homepage": "https://github.com/rmyndharis/OpenWA-plugins/tree/main/chatwoot-adapter", - "download": "https://github.com/rmyndharis/OpenWA-plugins/releases/download/chatwoot-adapter-v0.4.0/chatwoot-adapter.zip" + "download": "https://github.com/rmyndharis/OpenWA-plugins/releases/download/chatwoot-adapter-v0.5.0/chatwoot-adapter.zip" }, { "id": "faq-bot",