fix(chatwoot-adapter): auto-recover from deleted Chatwoot conversations

chatwoot-adapter/chatwoot-client.test.ts

@@ -88,3 +88,15 @@ test('postMedia marks a voice note and threads it (is_voice_message + source_id
   assert.match(raw, /name="source_id"\r\n\r\nwa5/);
   assert.match(raw, /filename="voice.ogg"/);
 });
+
+test('postMedia on a non-ok response rejects with err.status set (so the 404 mapping-rebuild path can branch on it)', async () => {
+  // Inbound/sent recovery branches on `err.status === 404` to rebuild the dangling conversation mapping.
+  // json() already attaches status; postMedia did NOT — a media 404 would silently dead-letter. This test
+  // pins the status-attaching behaviour in place.
+  const { fn } = fakeFetch({ 'POST /api/v1/accounts/3/conversations/55/messages': { status: 404, body: 'gone' } });
+  const c = new ChatwootClient(fn, cfg);
+  await assert.rejects(
+    c.postMedia(55, 'hi', { filename: 'a.jpg', contentType: 'image/jpeg', data: new Uint8Array([1]) }),
+    (err: Error & { status?: number }) => err.status === 404,
+  );
+});

chatwoot-adapter/chatwoot-client.ts

@@ -151,7 +151,14 @@ export class ChatwootClient {
       headers: this.headers({ 'Content-Type': `multipart/form-data; boundary=${boundary}` }),
       body,
     });
-    if (!res.ok) throw new Error(`Chatwoot postMedia -> ${res.status}`);
+    if (!res.ok) {
+      // Mirror json(): attach err.status so the caller (inbound/sent recovery) can branch on the HTTP
+      // code (the 404 mapping-rebuild path needs this — the bare Error below would otherwise be invisible
+      // to the `err.status === 404` check and a media 404 would dead-letter instead of recover).
+      const e = new Error(`Chatwoot postMedia -> ${res.status}`) as Error & { status?: number };
+      e.status = res.status;
+      throw e;
+    }
     return JSON.parse(res.body || '{}') as { id: number };
   }
 

chatwoot-adapter/inbound.test.ts

@@ -9,7 +9,14 @@ const msg = {
   timestamp: 0, fromMe: false, isGroup: false, senderPhone: '+621', contact: { pushName: 'Budi' },
 } as IncomingMessage;
 
-function deps(over: { client?: Record<string, unknown>; store?: Record<string, unknown>; engine?: Record<string, unknown> } = {}) {
+function deps(
+  over: {
+    client?: Record<string, unknown>;
+    store?: Record<string, unknown>;
+    engine?: Record<string, unknown>;
+    log?: (m: string, e?: unknown) => void;
+  } = {},
+) {
   let contacts = 0;
   let convs = 0;
   const posted: Array<{ id: number; c: string }> = [];
@@ -23,19 +30,25 @@ function deps(over: { client?: Record<string, unknown>; store?: Record<string, u
     postMedia: async () => ({ id: 2 }),
     ...over.client,
   };
+  // unlinkByChatId + unlinkByConversationId default to actual map deletes against the SAME backing `store`
+  // the default getByChat / link use, so a recovery path that calls one of them actually clears the
+  // cached row (matches the real PluginStorage's semantics). Tests that need to introspect the call can
+  // still override them.
   const mapping = {
     getByChat: async (s: string, c: string) => store.get(`${s}:${c}`) ?? null,
     getByConversation: async () => null,
     link: async (s: string, c: string, _i: string, l: unknown) => void store.set(`${s}:${c}`, l),
     hasSeen: async () => false,
     markSeen: async () => {},
+    unlinkByChatId: async (s: string, c: string) => void store.delete(`${s}:${c}`),
+    unlinkByConversationId: async () => {},
     ...over.store,
   };
   // Default: identity canonicalization (@lid resolution exercised explicitly below).
   const engine = { canonicalChatId: async (_s: string, c: string) => c, ...over.engine };
   const d = {
     lock: new KeyedAsyncLock(), client, store: mapping, engine, instanceId: 'inst',
-    relayGroups: true, relayMedia: true, log: () => {},
+    relayGroups: true, relayMedia: true, log: over.log ?? (() => {}),
   } as unknown as InboundDeps;
   return { deps: d, counts: () => ({ contacts, convs }), posted };
 }
@@ -210,3 +223,97 @@ test('never renames a group contact from a member pushName (#609)', async () =>
   await handleInbound(d, 'sess', 'Engine', grp);
   assert.equal(updates.length, 0);
 });
+
+test('a 404 on inbound relay mid-flight rebuilds the mapping and retries once (no retry-queue entry, no duplicate conversation)', async () => {
+  // Operator deleted the Chatwoot conversation out-of-band: the cached conv id is dead. The plugin must
+  // drop the stale mapping (forward + scoped + legacy reverse), re-resolve through ensureConversation,
+  // and re-post the message into a fresh conversation instead of looping the dead id forever.
+  const unlinksChat: string[] = [];
+  const unlinksConv: number[] = [];
+  const postedTo: number[] = [];
+  const enqueued: string[] = [];
+  // Seed a stale mapping under the chatId that `handleInbound` will look up (msg.chatId = '621@c.us').
+  const seeded = new Map<string, unknown>([
+    ['sess:621@c.us', { conversationId: 191, contactId: 9, sourceId: 'src' }],
+  ]);
+  const { deps: d } = deps({
+    store: {
+      getByChat: async (s: string, c: string) => seeded.get(`${s}:${c}`) ?? null,
+      link: async (s: string, c: string, _i: string, l: unknown) => void seeded.set(`${s}:${c}`, l),
+      // The default unlinkByChatId in deps() clears the local map — match that semantics against the
+      // seeded map so the rebuild's getByChat actually misses.
+      unlinkByChatId: async (s: string, c: string) => { unlinksChat.push(c); seeded.delete(`${s}:${c}`); },
+      unlinkByConversationId: async (_s: string, conv: number) => void unlinksConv.push(conv),
+      enqueueRetry: async (e: { msg: { id: string } }) => void enqueued.push(e.msg.id),
+    },
+    client: {
+      searchContact: async () => ({ id: 9, sourceId: 'src' }), // contact already on Chatwoot — no createContact
+      findOpenConversation: async () => null, // the old conv is deleted; no orphan open conv
+      createConversation: async () => 192,
+      postText: async (id: number) => {
+        postedTo.push(id);
+        if (id === 191) {
+          const e = new Error('Chatwoot POST .../conversations/191/messages -> 404') as Error & { status?: number };
+          e.status = 404;
+          throw e;
+        }
+        return { id: 1 };
+      },
+    },
+  });
+  await handleInbound(d, 'sess', 'Engine', msg);
+  assert.deepEqual(unlinksChat, ['621@c.us'], 'forward mapping dropped');
+  assert.deepEqual(unlinksConv, [191], 'scoped + legacy reverse mappings dropped');
+  assert.deepEqual(postedTo, [191, 192], 'first post 404s against the dead id; second post succeeds against the rebuilt one');
+  assert.deepEqual(enqueued, [], 'no retry-queue entry — the 404 was recovered in-band');
+  // The rebuild reuses the existing Chatwoot contact (searchContact hit) so no second createContact fires,
+  // and exactly one new conversation is minted for the contact. counts() counts both via the default
+  // createContact/createConversation wrappers on the CLIENT overrides above — when an override replaces
+  // them, the default counters stop incrementing, so we assert by the recorded calls instead.
+  assert.ok(seeded.has('sess:621@c.us'), 'the fresh link is written back under the same chatId');
+  const fresh = seeded.get('sess:621@c.us') as { conversationId: number };
+  assert.equal(fresh.conversationId, 192, 'mapping now points at the new conversation id');
+});
+
+test('a 404 on the REBUILT inbound conversation propagates to the retry queue (no silent loop)', async () => {
+  // Catastrophic case: even the freshly-created conversation is already gone, or Chatwoot is fully down.
+  // The recovery must NOT swallow the second 404 on the rebuilt conv — the failure has to surface so the
+  // existing enqueueRetry + drainRetries (5-attempt cap) takes over.
+  const unlinksConv: number[] = [];
+  const postedTo: number[] = [];
+  const enqueued: string[] = [];
+  const logCalls: string[] = [];
+  // Seed the SAME stale mapping shape as the previous test.
+  const seeded = new Map<string, unknown>([
+    ['sess:621@c.us', { conversationId: 55, contactId: 9, sourceId: 'src' }],
+  ]);
+  const { deps: d } = deps({
+    store: {
+      getByChat: async (s: string, c: string) => seeded.get(`${s}:${c}`) ?? null,
+      link: async (s: string, c: string, _i: string, l: unknown) => void seeded.set(`${s}:${c}`, l),
+      unlinkByChatId: async (s: string, c: string) => { seeded.delete(`${s}:${c}`); },
+      unlinkByConversationId: async (_s: string, conv: number) => void unlinksConv.push(conv),
+      enqueueRetry: async (e: { msg: { id: string } }) => void enqueued.push(e.msg.id),
+    },
+    client: {
+      searchContact: async () => ({ id: 9, sourceId: 'src' }),
+      findOpenConversation: async () => null,
+      createConversation: async () => 200,
+      postText: async (id: number) => {
+        postedTo.push(id);
+        const e = new Error(`Chatwoot POST .../conversations/${id}/messages -> 404`) as Error & { status?: number };
+        e.status = 404;
+        throw e; // BOTH the dead id AND the rebuilt id 404 — the recovery cannot succeed
+      },
+    },
+    log: (m: string) => { logCalls.push(m); },
+  });
+  await handleInbound(d, 'sess', 'Engine', { ...msg, id: 'm-recover-twice' });
+  assert.deepEqual(unlinksConv, [55], 'still unlinks the dead conv mapping once');
+  assert.deepEqual(postedTo, [55, 200], 'posts were attempted against the dead id and then the rebuilt one');
+  assert.deepEqual(enqueued, ['m-recover-twice'], 'the original 404 surfaces to the retry queue (with the backoff cap)');
+  assert.ok(
+    logCalls.some(l => /inbound 404-recovery failed/.test(l)),
+    'logs the recovery failure so ops can see why the rebuild itself failed',
+  );
+});

chatwoot-adapter/inbound.ts

@@ -7,8 +7,17 @@ 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> {
+// handler and the retry drain (retry.ts) so a retried message follows the exact same path. `opts.recoverOn404`
+// defaults TRUE for the live path; the retry drain (index.ts) flips it FALSE so drain attempts against a
+// still-dangling mapping burn their retry-budget instead of minting a fresh orphan conversation per
+// attempt — the next LIVE inbound recovers the chat.
+export async function relayInbound(
+  deps: InboundDeps,
+  sessionId: string,
+  msg: IncomingMessage,
+  opts: { recoverOn404?: boolean } = {},
+): Promise<void> {
+  const recoverOn404 = opts.recoverOn404 ?? true;
   // Best-effort @lid -> <phone>@c.us for the LOOKUP only (never the lock — see handleInbound). Raw-fallback:
   // canonicalChatId needs a live engine, but the retry drain re-relays messages whose WA session may be
   // offline (the relay is a Chatwoot post that doesn't need it), so a failure must not block the relay.
@@ -18,14 +27,48 @@ export async function relayInbound(deps: InboundDeps, sessionId: string, msg: In
   } catch {
     /* session down / unresolvable — fall back to the raw id; dedup is best-effort */
   }
-  const { conversationId, created } = await resolveConversation(deps, sessionId, msg, canonical);
+  const { conversationId, created, foundKey } = await resolveConversation(deps, sessionId, msg, canonical);
   // 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, sessionId, conversationId, msg, 'incoming');
+  try {
+    await relayMessage(deps, sessionId, conversationId, msg, 'incoming');
+  } catch (err) {
+    // A 404 mid-relay means the CACHED conversation id is gone — almost always because an operator
+    // (or Chatwoot) deleted the conversation out-of-band, leaving the conv<->WA mapping dangling. Drop
+    // the stale forward + scoped + legacy reverse keys and rebuild via the same resolveConversation ->
+    // ensureConversation path; then retry once. A second failure (404 or anything) survives the rebuild
+    // and falls through to the regular enqueueRetry -> drainRetries pipeline with its 5-attempt cap, so
+    // a wedged Chatwoot or a misconfigured inbox cannot loop here.
+    if ((err as { status?: number } | null)?.status !== 404) throw err;
+    // Drain path: same 404, but the drain is already a re-attempt of a previously failed relay. Rebuilding
+    // here would mint a fresh orphan conversation per attempt (up to 5) and never converge — better to let
+    // the retry budget run out and let the NEXT live inbound self-heal the chat.
+    if (!recoverOn404) throw err;
+    const originalErr = err;
+    try {
+      // Unlink the key the mapping actually lived under (resolveConversation returns it), not msg.chatId.
+      // A migrated contact's mapping is often keyed under the canonical @c.us while msg.chatId is @lid —
+      // unlinking the raw key would miss entirely, resolveConversation would re-find the stale row via its
+      // canonical fallback, the second relay would 404 again, and the message would dead-letter.
+      await deps.store.unlinkByChatId(sessionId, foundKey);
+      await deps.store.unlinkByConversationId(sessionId, conversationId);
+      const re = await resolveConversation(deps, sessionId, msg, canonical);
+      // The deleted Chatwoot conversation is gone, so its history is gone too. When the rebuild mints a
+      // brand-new conversation, re-import prior history under the same gate the happy path uses — a fresh
+      // Chatwoot thread without context defeats the point of (lazy) backfill.
+      if (re.created && deps.backfillLimit > 0) {
+        await backfillHistory(deps, sessionId, msg.chatId, re.conversationId);
+      }
+      await relayMessage(deps, sessionId, re.conversationId, msg, 'incoming');
+    } catch (rebuildErr) {
+      deps.log('inbound 404-recovery failed; surfacing original 404 to retry queue', rebuildErr);
+      throw originalErr;
+    }
+  }
 }
 
 // WhatsApp → Chatwoot. Filter, then run resolve+post under the per-chat lock so two near-simultaneous
@@ -74,11 +117,13 @@ async function resolveConversation(
   sessionId: string,
   msg: IncomingMessage,
   canonicalChatId: string,
-): Promise<{ conversationId: number; created: boolean }> {
+): Promise<{ conversationId: number; created: boolean; foundKey: string }> {
   // Dual lookup (re-read inside the lock): the raw chatId finds a mapping keyed by @lid, the canonical
   // chatId finds one keyed by @c.us (a contact that has since migrated to @lid, when the lid resolves) —
   // so a migrated contact's inbound lands in its EXISTING conversation instead of splitting a duplicate.
-  // `foundKey` is the key the mapping actually lives under, so refreshContactName patches the right doc.
+  // `foundKey` is the key the mapping actually lives under, so refreshContactName patches the right doc
+  // AND the 404-recovery path knows which forward key to unlink (a migrated contact's stale row is under
+  // the canonical key, NOT msg.chatId — unlinking the raw id would miss it entirely).
   let existing = await deps.store.getByChat(sessionId, msg.chatId);
   let foundKey = msg.chatId;
   if (!existing && canonicalChatId !== msg.chatId) {
@@ -87,7 +132,7 @@ async function resolveConversation(
   }
   if (existing) {
     await refreshContactName(deps, sessionId, msg, existing, foundKey);
-    return { conversationId: existing.conversationId, created: false };
+    return { conversationId: existing.conversationId, created: false, foundKey };
   }
   const name = msg.isGroup
     ? `Group ${msg.chatId}`
@@ -96,5 +141,8 @@ async function resolveConversation(
     name,
     phone: msg.isGroup ? undefined : msg.senderPhone ?? undefined,
   });
-  return { conversationId, created: true };
+  // A freshly-created mapping lives under msg.chatId (ensureConversation links raw). The recovery path
+  // reads foundKey off this return only when `created === false` (a 404 implies a row existed); the value
+  // here is still defined so the return shape is uniform.
+  return { conversationId, created: true, foundKey: msg.chatId };
 }

chatwoot-adapter/mapping-store.ts

@@ -207,4 +207,13 @@ export class MappingStore {
   async countRetries(): Promise<number> {
     return (await this.retryKeys()).length;
   }
+
+  async unlinkByChatId(sessionId: string, chatId: string) {
+    await this.storage.delete(this.fwdKey(sessionId, chatId));
+  }
+
+  async unlinkByConversationId(sessionId: string, conversationId: number) {
+    await this.storage.delete(this.revKey(sessionId, conversationId));
+    await this.storage.delete(this.legacyRevKey(conversationId));
+  }
 }