feat(chatwoot-adapter): history backfill (lazy + opt-in bulk) (#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.2.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.3.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,7 +6,19 @@ All notable changes to the Chatwoot Adapter plugin are documented here. The form
 
 ## [Unreleased]
 
-## [0.2.0] — 2026-07-03
+## [0.3.0] — 2026-07-03
+
+### Added
+
+- **History backfill** so agents see prior WhatsApp context in Chatwoot instead of a conversation that
+  starts mid-thread (#609). Two composable modes, both off by default:
+  - **Lazy (`backfillLimit`)** — when a chat first opens as a Chatwoot conversation, its recent messages
+    (both directions, with media) are replayed oldest→newest before the triggering message, so the thread
+    reads in order. Deduped against the live path, so nothing double-posts.
+  - **Bulk (`backfillAllOnce`)** — a one-time sweep that imports the history of every existing chat on
+    setup, for mirroring a whole inbox. Sequential, best-effort, runs once per session.
+  - Business-side (`fromMe`) messages post as Chatwoot `outgoing`, contact messages as `incoming`.
+  - Requires OpenWA 0.8.5+ (the new `engine.getChatHistory` capability).
 
 ### Added
 

chatwoot-adapter/README.md

@@ -11,13 +11,13 @@ mirror it. Runs sandboxed in the plugin worker; no server, no extra port.
 | Field | Value |
 | ----- | ----- |
 | **Identifier** | `chatwoot-adapter` |
-| **Version** | 0.2.0 |
+| **Version** | 0.3.0 |
 | **Released** | 2026-07-03 |
 | **Status** | beta |
 | **Author** | Yudhi Armyndharis |
 | **License** | MIT |
 | **Type** | `extension` |
-| **Requires OpenWA** | ≥ 0.8.3 (tested 0.8.3) |
+| **Requires OpenWA** | ≥ 0.8.5 (tested 0.8.5) |
 | **Keywords** | chatwoot, helpdesk, inbox, handover, two-way, agent, whatsapp, openwa |
 | **Repository** | [OpenWA-plugins/chatwoot-adapter](https://github.com/rmyndharis/OpenWA-plugins/tree/main/chatwoot-adapter) |
 <!-- END DETAILS -->
@@ -66,6 +66,8 @@ sees it.
 | `inboxId` | number | yes | The API-channel inbox id this adapter posts into and relays replies from. |
 | `relayGroups` | boolean | no (default `true`) | Relay group chats (one synthetic contact per group, sender-prefixed). |
 | `relayMedia` | boolean | no (default `true`) | Upload inbound media to Chatwoot as attachments. |
+| `backfillLimit` | number | no (default `0`) | When a chat first opens in Chatwoot, import this many recent messages (both directions, with media) so agents see prior context. `0` disables it; clamped to 100 host-side. Needs OpenWA 0.8.5+. |
+| `backfillAllOnce` | boolean | no (default `false`) | Also run a one-time sweep importing every existing chat's history on setup. Needs `backfillLimit` > 0. Runs once per session. |
 
 The Chatwoot webhook **secret** and the instance's **session scope** are set when you mint the instance (they
 are not part of the config above).
@@ -85,8 +87,9 @@ public host is added to the outbound allowlist). To use a self-hosted Chatwoot:
 
 ## Compatibility
 
-- **OpenWA** ≥ 0.8.3 — needs Integration SDK v1 (webhook ingress, `ctx.mappings`, the session+chat handover
-  gate, `net.allowConfigHosts`) and the `conversation.send` media/voice types used for outbound attachments.
+- **OpenWA** ≥ 0.8.5 — needs Integration SDK v1 (webhook ingress, `ctx.mappings`, the session+chat handover
+  gate, `net.allowConfigHosts`), the `conversation.send` media/voice types for outbound attachments, and
+  `engine.getChatHistory` for the history backfill.
 - **Chatwoot** — account-level webhooks with timestamped HMAC signing (see Setup).
 
 ## Security

chatwoot-adapter/backfill.test.ts

@@ -0,0 +1,135 @@
+import { test } from 'node:test';
+import assert from 'node:assert/strict';
+import { backfillHistory, backfillAllChats } from './backfill.ts';
+import { KeyedAsyncLock } from './chat-lock.ts';
+import type { InboundDeps } from './relay.ts';
+import type { IncomingMessage } from '../types/openwa';
+
+// Capture what relayMessage posts by observing the client the shared relay calls (text path only here;
+// media rendering is covered by the inbound tests).
+function makeDeps(
+  over: {
+    engine?: Record<string, unknown>;
+    store?: Record<string, unknown>;
+    relayGroups?: boolean;
+    backfillLimit?: number;
+  } = {},
+) {
+  const posts: Array<{ conversationId: number; type: string; body: string }> = [];
+  const seen = new Set<string>();
+  const client = {
+    searchContact: async () => null,
+    createContact: async () => ({ id: 9, sourceId: 'src' }),
+    findOpenConversation: async () => null,
+    createConversation: async () => 55,
+    postText: async (conversationId: number, body: string, o: { messageType?: string }) => {
+      posts.push({ conversationId, type: o?.messageType ?? 'incoming', body });
+      return { id: 1 };
+    },
+    postMedia: async () => ({ id: 2 }),
+    updateContact: async () => {},
+  };
+  const store = {
+    hasSeen: async (_k: string, id: string) => seen.has(id),
+    markSeen: async (_k: string, id: string) => void seen.add(id),
+    getByChat: async () => null,
+    link: async () => {},
+    patch: async () => {},
+    isBulkBackfilled: async () => false,
+    setBulkBackfilled: async () => {},
+    ...over.store,
+  };
+  const deps = {
+    lock: new KeyedAsyncLock(),
+    client,
+    store,
+    engine: { getChatHistory: async () => [], getChats: async () => [], ...over.engine },
+    instanceId: 'inst',
+    relayGroups: over.relayGroups ?? true,
+    relayMedia: true,
+    backfillLimit: over.backfillLimit ?? 20,
+    backfillAllOnce: false,
+    log: () => {},
+  } as unknown as InboundDeps;
+  return { deps, posts };
+}
+
+const hist = (id: string, ts: number, fromMe: boolean, body: string): IncomingMessage =>
+  ({ id, from: 'x', to: 'y', chatId: 'c@c.us', body, type: 'chat', timestamp: ts, fromMe, isGroup: false }) as IncomingMessage;
+
+test('backfillHistory posts oldest→newest with fromMe as outgoing (#609)', async () => {
+  const history = [hist('m3', 30, false, 'third'), hist('m1', 10, true, 'first'), hist('m2', 20, false, 'second')];
+  const { deps, posts } = makeDeps({ engine: { getChatHistory: async () => history } });
+  await backfillHistory(deps, 'sess', 'c@c.us', 55);
+  assert.deepEqual(posts, [
+    { conversationId: 55, type: 'outgoing', body: 'first' }, // fromMe, oldest first
+    { conversationId: 55, type: 'incoming', body: 'second' },
+    { conversationId: 55, type: 'incoming', body: 'third' },
+  ]);
+});
+
+test('backfillHistory skips messages already seen (dedup with the live path)', async () => {
+  const seen = new Set(['m1']);
+  const history = [hist('m1', 10, false, 'dup'), hist('m2', 20, false, 'new')];
+  const { deps, posts } = makeDeps({
+    engine: { getChatHistory: async () => history },
+    store: {
+      hasSeen: async (_k: string, id: string) => seen.has(id),
+      markSeen: async (_k: string, id: string) => void seen.add(id),
+    },
+  });
+  await backfillHistory(deps, 'sess', 'c@c.us', 55);
+  assert.deepEqual(posts.map(p => p.body), ['new']);
+});
+
+test('backfillHistory swallows a getChatHistory failure (best-effort)', async () => {
+  const { deps, posts } = makeDeps({
+    engine: {
+      getChatHistory: async () => {
+        throw new Error('engine down');
+      },
+    },
+  });
+  await backfillHistory(deps, 'sess', 'c@c.us', 55); // must not throw
+  assert.equal(posts.length, 0);
+});
+
+test('backfillAllChats sweeps each chat once, skips groups when relayGroups is off, survives a failure (#609)', async () => {
+  const chats = [
+    { id: 'a@c.us', name: 'A', isGroup: false, unreadCount: 0, timestamp: 1 },
+    { id: 'g@g.us', name: 'G', isGroup: true, unreadCount: 0, timestamp: 2 },
+    { id: 'b@c.us', name: 'B', isGroup: false, unreadCount: 0, timestamp: 3 },
+  ];
+  const historyByChat: Record<string, IncomingMessage[]> = {
+    'a@c.us': [{ ...hist('a1', 10, false, 'from A'), chatId: 'a@c.us' }],
+    'b@c.us': [{ ...hist('b1', 10, false, 'from B'), chatId: 'b@c.us' }],
+  };
+  let bulkDone = false;
+  const { deps, posts } = makeDeps({
+    relayGroups: false, // the group chat must be skipped
+    engine: {
+      getChats: async () => chats,
+      getChatHistory: async (_s: string, chatId: string) => historyByChat[chatId] ?? [],
+    },
+    store: {
+      hasSeen: async () => false,
+      markSeen: async () => {},
+      getByChat: async () => null,
+      link: async () => {},
+      isBulkBackfilled: async () => bulkDone,
+      setBulkBackfilled: async () => {
+        bulkDone = true;
+      },
+    },
+  });
+  await backfillAllChats(deps, 'sessBulk');
+  assert.deepEqual(
+    posts.map(p => p.body).sort(),
+    ['from A', 'from B'], // group skipped
+  );
+  assert.equal(bulkDone, true);
+  // second call is a no-op once the durable marker is set
+  const before = posts.length;
+  await backfillAllChats(deps, 'sessBulk');
+  assert.equal(posts.length, before);
+});

chatwoot-adapter/backfill.ts

@@ -0,0 +1,57 @@
+import type { ChatSummary } from '../types/openwa';
+import { relayMessage, ensureConversation, type InboundDeps } from './relay.ts';
+
+// Replay a chat's recent history into its Chatwoot conversation: oldest→newest, both directions, deduped
+// against the same markSeen store the live path uses (so a live message is never doubled). The caller
+// holds the per-chat lock. Best-effort: any failure is logged, never thrown into the live relay.
+export async function backfillHistory(
+  deps: InboundDeps,
+  sessionId: string,
+  chatId: string,
+  conversationId: number,
+): Promise<void> {
+  try {
+    const history = await deps.engine.getChatHistory(sessionId, chatId, deps.backfillLimit, true);
+    const ordered = [...history].sort((a, b) => a.timestamp - b.timestamp);
+    for (const msg of ordered) {
+      if (await deps.store.hasSeen('wa', msg.id, sessionId)) continue;
+      await deps.store.markSeen('wa', msg.id, sessionId);
+      await relayMessage(deps, conversationId, msg, msg.fromMe ? 'outgoing' : 'incoming');
+    }
+  } catch (err) {
+    deps.log('history backfill failed', err);
+  }
+}
+
+// In-memory guard so rapid successive inbounds can't launch the one-time sweep twice for a session.
+const bulkInFlight = new Set<string>();
+
+// One-time bulk sweep (opt-in): create a Chatwoot conversation for every existing chat and backfill its
+// history. Sequential — no parallel fan-out at Chatwoot — and best-effort: a per-chat failure never
+// aborts the sweep. Runs once per session behind a durable marker plus the in-memory in-flight guard.
+export async function backfillAllChats(deps: InboundDeps, sessionId: string): Promise<void> {
+  // Add to the in-flight set BEFORE the first await, so a concurrent call from a rapid second inbound
+  // sees it synchronously and bails — otherwise both could pass the durable-marker check and double-sweep.
+  if (bulkInFlight.has(sessionId)) return;
+  bulkInFlight.add(sessionId);
+  try {
+    if (await deps.store.isBulkBackfilled(sessionId)) return;
+    const chats = (await deps.engine.getChats(sessionId)) as ChatSummary[];
+    for (const chat of chats) {
+      if (chat.isGroup && !deps.relayGroups) continue;
+      await deps.lock.run(`${sessionId}:${chat.id}`, async () => {
+        try {
+          const conversationId = await ensureConversation(deps, sessionId, chat.id, { name: chat.name || chat.id });
+          await backfillHistory(deps, sessionId, chat.id, conversationId);
+        } catch (err) {
+          deps.log(`bulk backfill failed for ${chat.id}`, err);
+        }
+      });
+    }
+    await deps.store.setBulkBackfilled(sessionId);
+  } catch (err) {
+    deps.log('bulk backfill sweep failed', err);
+  } finally {
+    bulkInFlight.delete(sessionId);
+  }
+}

chatwoot-adapter/chatwoot-client.test.ts

@@ -49,6 +49,16 @@ test('postText posts an incoming message with the api token header', async () =>
   assert.equal(last.init!.headers!['api_access_token'], 'tok');
 });
 
+test('postText and postMedia honor an outgoing message_type (backfilled business messages)', async () => {
+  const { fn, calls } = fakeFetch({ 'POST /api/v1/accounts/3/conversations/55/messages': { body: { id: 1 } } });
+  const c = new ChatwootClient(fn, cfg);
+  await c.postText(55, 'sent by agent', { messageType: 'outgoing' });
+  assert.equal(JSON.parse(calls.at(-1)!.init!.body as string).message_type, 'outgoing');
+  await c.postMedia(55, '', { filename: 'x.jpg', contentType: 'image/jpeg', data: new Uint8Array([1]) }, { messageType: 'outgoing' });
+  const raw = Buffer.from(calls.at(-1)!.init!.body as Uint8Array).toString('latin1');
+  assert.match(raw, /name="message_type"\r\n\r\noutgoing/);
+});
+
 test('updateContact PUTs the new name to the contact', async () => {
   const { fn, calls } = fakeFetch({ 'PUT /api/v1/accounts/3/contacts/9': { body: { id: 9 } } });
   await new ChatwootClient(fn, cfg).updateContact(9, 'Budi');

chatwoot-adapter/chatwoot-client.ts

@@ -12,6 +12,9 @@ export interface ChatwootConfig {
 export interface MessagePostOptions {
   sourceId?: string;
   inReplyToExternalId?: string;
+  // Chatwoot direction. Live inbound is always 'incoming'; history backfill posts business-side
+  // (fromMe) messages as 'outgoing' so the reconstructed thread reads correctly. Default 'incoming'.
+  messageType?: 'incoming' | 'outgoing';
 }
 
 // The slice of ctx.net.fetch this client needs (host-proxied, SSRF-guarded). Injectable for tests.
@@ -111,7 +114,7 @@ export class ChatwootClient {
   }
 
   async postText(conversationId: number, content: string, opts: MessagePostOptions = {}): Promise<{ id: number }> {
-    const payload: Record<string, unknown> = { content, message_type: 'incoming', private: false };
+    const payload: Record<string, unknown> = { content, message_type: opts.messageType ?? 'incoming', private: false };
     if (opts.sourceId) payload.source_id = opts.sourceId;
     if (opts.inReplyToExternalId) payload.content_attributes = { in_reply_to_external_id: opts.inReplyToExternalId };
     const { data } = await this.json<{ id: number }>(`${this.base()}/conversations/${conversationId}/messages`, {
@@ -130,7 +133,7 @@ export class ChatwootClient {
     const boundary = `----cw${conversationId}${file.data.byteLength}`;
     const fields = [
       { name: 'content', value: content },
-      { name: 'message_type', value: 'incoming' },
+      { name: 'message_type', value: opts.messageType ?? 'incoming' },
     ];
     // Rails parses bracket notation into nested params; source_id + in_reply_to_external_id give Chatwoot
     // the threading it uses for the native WhatsApp integration, is_voice_message renders a voice bubble.

chatwoot-adapter/inbound.test.ts

@@ -68,7 +68,7 @@ test('forwards the quote context (source_id + in_reply_to_external_id) on a repl
   });
   const reply = { ...msg, id: 'r1', quotedMessage: { id: 'orig', body: 'earlier' } } as IncomingMessage;
   await handleInbound(d, 'sess', 'Engine', reply);
-  assert.deepEqual(opts, { sourceId: 'r1', inReplyToExternalId: 'orig' });
+  assert.deepEqual(opts, { sourceId: 'r1', inReplyToExternalId: 'orig', messageType: 'incoming' });
 });
 
 test('relays an inbound voice note as a Chatwoot voice message (#607)', async () => {