feat(chatwoot-adapter): retry failed inbound relays instead of dropping them (#609)

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 |

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

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 |

chatwoot-adapter/inbound.test.ts

@@ -38,6 +38,16 @@ function deps(over: { client?: Record<string, unknown>; store?: Record<string, u
   return { deps: d, counts: () => ({ 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);

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<void> {
+  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<void> {
   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})`);
     }
   });
 }

chatwoot-adapter/index.test.ts

@@ -1,16 +1,23 @@
 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<string, unknown>) {
   const hooks: string[] = [];
   const routes: string[] = [];
   const cbs: Record<string, (h: unknown) => Promise<{ continue: boolean }>> = {};
   let fetches = 0;
+  const storageMap = new Map<string, unknown>();
   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<string, unknown>) {
     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/);

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<string, unknown>): ChatwootFullConfig {
 }
 
 export default class ChatwootAdapter implements IPlugin {
+  private retryTimer: ReturnType<typeof setInterval> | null = null;
+  private store: MappingStore | null = null;
+  private deadLetterCount = 0;
+  private draining = false;
+
   async onEnable(ctx: PluginContext): Promise<void> {
+    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<void> => {
+      // 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<void> {
+    this.clearRetryTimer();
+  }
+
+  async onUnload(): Promise<void> {
+    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 };
+  }
 }

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.",

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<string, unknown>();
   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']);
+});