diff --git a/README.md b/README.md index 9f696c5..aee382c 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.3.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.4.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 cf26dac..9893e0d 100644 --- a/chatwoot-adapter/CHANGELOG.md +++ b/chatwoot-adapter/CHANGELOG.md @@ -6,6 +6,19 @@ All notable changes to the Chatwoot Adapter plugin are documented here. The form ## [Unreleased] +## [0.4.0] — 2026-07-03 + +### Added + +- **Relay your own outbound sends** into Chatwoot, so a conversation isn't one-sided when you reply from a + linked phone, the WhatsApp app, or the OpenWA API (#615). These mirror into the contact's **existing** + mapped Chatwoot conversation as `outgoing` messages (a send to a chat not yet in Chatwoot is skipped — + it appears once the contact replies, never as a duplicate conversation). Replies you send from within + Chatwoot are recognized and never duplicated. New `relayOwnMessages` setting, **on by default**; turn it + off to keep phone-composed messages out of the helpdesk. When the `@lid` mapping is resolvable, own + sends to a contact WhatsApp has migrated to `@lid` land in their existing conversation instead of a + duplicate, via the new host `canonicalChatId` resolver. Requires OpenWA 0.8.7+. + ## [0.3.0] — 2026-07-03 ### Added diff --git a/chatwoot-adapter/README.md b/chatwoot-adapter/README.md index d206097..c2546c3 100644 --- a/chatwoot-adapter/README.md +++ b/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.3.0 | +| **Version** | 0.4.0 | | **Released** | 2026-07-03 | | **Status** | beta | | **Author** | Yudhi Armyndharis | | **License** | MIT | | **Type** | `extension` | -| **Requires OpenWA** | ≥ 0.8.6 (tested 0.8.6) | +| **Requires OpenWA** | ≥ 0.8.7 (tested 0.8.7) | | **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) | diff --git a/chatwoot-adapter/filters.test.ts b/chatwoot-adapter/filters.test.ts index 6b7f5e1..400a145 100644 --- a/chatwoot-adapter/filters.test.ts +++ b/chatwoot-adapter/filters.test.ts @@ -1,6 +1,6 @@ import { test } from 'node:test'; import assert from 'node:assert/strict'; -import { shouldRelayInbound, shouldRelayOutbound } from './filters.ts'; +import { shouldRelayInbound, shouldRelayOutbound, shouldRelayOwn } from './filters.ts'; const base = { id: 'm1', @@ -22,6 +22,16 @@ test('inbound: Engine + not fromMe + has chatId; drops API, fromMe, and groups w assert.equal(shouldRelayInbound({ ...base, isGroup: true }, 'Engine', true), true); }); +test('own-outbound: Engine + fromMe + has chatId; drops inbound, API source, empty chatId, and groups when relayGroups=false', () => { + const own = { ...base, fromMe: true } as const; + assert.equal(shouldRelayOwn(own, 'Engine', true), true); + assert.equal(shouldRelayOwn({ ...own, fromMe: false }, 'Engine', true), false); // inbound is not an own send + assert.equal(shouldRelayOwn(own, 'API', true), false); // only engine-delivered creates + assert.equal(shouldRelayOwn({ ...own, chatId: '' }, 'Engine', true), false); // no target chat + assert.equal(shouldRelayOwn({ ...own, isGroup: true }, 'Engine', false), false); + assert.equal(shouldRelayOwn({ ...own, isGroup: true }, 'Engine', true), true); +}); + test('outbound: strict private — relay only outgoing + private===false in the configured inbox', () => { const ok = { message_type: 'outgoing', private: false, inbox: { id: 7 }, conversation: { id: 1 }, content: 'r' }; assert.equal(shouldRelayOutbound(ok, 7), true); diff --git a/chatwoot-adapter/filters.ts b/chatwoot-adapter/filters.ts index 89c18af..b091e8c 100644 --- a/chatwoot-adapter/filters.ts +++ b/chatwoot-adapter/filters.ts @@ -6,6 +6,14 @@ export function shouldRelayInbound(msg: IncomingMessage, source: string, relayGr return source === 'Engine' && !msg.fromMe && !!msg.chatId && (!msg.isGroup || relayGroups); } +// Relay the account's OWN outbound sends (composed on a linked phone or via the OpenWA API) so the +// Chatwoot thread mirrors WhatsApp (#615). The mirror of shouldRelayInbound with fromMe===true. The +// adapter's own Chatwoot-agent replies are ALSO fromMe and reach message:sent, but they're excluded +// out-of-band by the 'wa' send-id echo marker, not here. Pure — no ctx. +export function shouldRelayOwn(msg: IncomingMessage, source: string, relayGroups: boolean): boolean { + return source === 'Engine' && msg.fromMe && !!msg.chatId && (!msg.isGroup || relayGroups); +} + // The subset of a Chatwoot account-level webhook payload the adapter reads (message_created + // conversation_updated). Everything is optional — Chatwoot omits fields per event/version. export interface ChatwootWebhookMessage { diff --git a/chatwoot-adapter/index.test.ts b/chatwoot-adapter/index.test.ts index 3ce5c0e..02fe837 100644 --- a/chatwoot-adapter/index.test.ts +++ b/chatwoot-adapter/index.test.ts @@ -6,29 +6,51 @@ import type { PluginContext } from '../types/openwa'; function fakeCtx(config: Record) { const hooks: string[] = []; const routes: string[] = []; + const cbs: Record Promise<{ continue: boolean }>> = {}; + let fetches = 0; const ctx = { config, storage: { get: async () => null, set: async () => {}, delete: async () => {}, list: async () => [] }, mappings: { upsert: async () => {}, get: async () => null, getByProvider: async () => null }, - net: { fetch: async () => ({ ok: true, status: 200, headers: {}, body: '{}' }) }, + net: { fetch: async () => { fetches++; return { ok: true, status: 200, headers: {}, body: '{}' }; } }, conversations: { send: async () => ({}) }, handover: { set: async () => ({}) }, + engine: { canonicalChatId: async (_s: string, c: string) => c }, logger: { log: () => {}, debug: () => {}, warn: () => {}, error: () => {} }, - registerHook: (event: string) => void hooks.push(event), + 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 }; + return { ctx, hooks, routes, cbs, fetches: () => fetches }; } const goodConfig = { baseUrl: 'https://chat.acme.com', apiToken: 'tok', accountId: 3, inboxId: 7 }; -test('onEnable registers the message:received hook and the chatwoot ingress route', async () => { +test('onEnable registers the message:received + message:sent hooks and the chatwoot ingress route', async () => { const { ctx, hooks, routes } = fakeCtx(goodConfig); await new ChatwootAdapter().onEnable(ctx); - assert.deepEqual(hooks, ['message:received']); + assert.deepEqual(hooks, ['message:received', 'message:sent']); assert.deepEqual(routes, ['chatwoot']); }); +test('message:sent is registered even when relayOwnMessages=false (gate is per-event, not at enable)', async () => { + const { ctx, hooks } = fakeCtx({ ...goodConfig, relayOwnMessages: false }); + await new ChatwootAdapter().onEnable(ctx); + assert.ok(hooks.includes('message:sent')); +}); + +test('relayOwnMessages=false gates the message:sent handler off (no Chatwoot API call)', async () => { + const { ctx, cbs, fetches } = fakeCtx({ ...goodConfig, relayOwnMessages: false }); + await new ChatwootAdapter().onEnable(ctx); + const r = await cbs['message:sent']({ + sessionId: 'sess', + source: 'Engine', + data: { id: 'x', fromMe: true, chatId: 'c@wa', body: 'hi', type: 'chat', isGroup: false }, + }); + await new Promise(res => setImmediate(res)); // let any detached work settle (there should be none) + assert.deepEqual(r, { continue: true }); + assert.equal(fetches(), 0); +}); + 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 fe03176..3bb77aa 100644 --- a/chatwoot-adapter/index.ts +++ b/chatwoot-adapter/index.ts @@ -3,6 +3,7 @@ import { ChatwootClient } from './chatwoot-client.ts'; import { MappingStore } from './mapping-store.ts'; import { KeyedAsyncLock } from './chat-lock.ts'; import { handleInbound } from './inbound.ts'; +import { handleSent } from './sent.ts'; import { backfillAllChats } from './backfill.ts'; import { handleOutbound } from './outbound.ts'; @@ -13,6 +14,7 @@ interface ChatwootFullConfig { inboxId: number; relayGroups: boolean; relayMedia: boolean; + relayOwnMessages: boolean; backfillLimit: number; backfillAllOnce: boolean; } @@ -49,6 +51,7 @@ function readConfig(raw: Record): ChatwootFullConfig { inboxId, relayGroups: raw.relayGroups !== false, relayMedia: raw.relayMedia !== false, + relayOwnMessages: raw.relayOwnMessages !== false, backfillLimit: Number.isFinite(rawLimit) ? Math.max(0, Math.trunc(rawLimit)) : 0, backfillAllOnce: raw.backfillAllOnce === true || raw.backfillAllOnce === 'true', }; @@ -62,23 +65,27 @@ export default class ChatwootAdapter implements IPlugin { // 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)); + // Shared per-event dependency bag for the inbound and own-send relays (both render into Chatwoot the + // same way). instanceId = sessionId (a session-scoped instance is 1:1 with its session). + const buildDeps = (cfg: ChatwootFullConfig, sessionId: string) => ({ + lock, + client: clientFor(), + store, + engine: ctx.engine, + instanceId: sessionId, + relayGroups: cfg.relayGroups, + relayMedia: cfg.relayMedia, + backfillLimit: cfg.backfillLimit, + backfillAllOnce: cfg.backfillAllOnce, + log: (m: string, e?: unknown) => ctx.logger.error(m, e), + }); + ctx.registerHook('message:received', async (h: HookContext): Promise => { const sessionId = h.sessionId; const msg = h.data as IncomingMessage; if (sessionId && msg) { const cfg = readConfig(ctx.config); - const deps = { - lock, - client: clientFor(), - store, - engine: ctx.engine, - instanceId: sessionId, - relayGroups: cfg.relayGroups, - relayMedia: cfg.relayMedia, - backfillLimit: cfg.backfillLimit, - backfillAllOnce: cfg.backfillAllOnce, - log: (m: string, e?: unknown) => ctx.logger.error(m, e), - }; + const deps = buildDeps(cfg, sessionId); // Fire-and-forget off the hook so a slow/failing Chatwoot API never blocks the WA pipeline. The // mapping mirror is keyed on sessionId (a session-scoped instance is 1:1 with its session). void handleInbound(deps, sessionId, h.source, msg).catch(e => ctx.logger.error('inbound hook failed', e)); @@ -91,12 +98,31 @@ export default class ChatwootAdapter implements IPlugin { return { continue: true }; }); + // The account's OWN outbound sends (linked phone / WhatsApp app / OpenWA REST API) arrive on + // message:sent, not message:received. Relay them as 'outgoing' so the Chatwoot thread mirrors the full + // WhatsApp conversation (#615). The adapter's own Chatwoot-agent replies also surface here but are + // echo-suppressed inside handleSent. Gated per-event so a live relayOwnMessages flip applies at once. + ctx.registerHook('message:sent', async (h: HookContext): Promise => { + const sessionId = h.sessionId; + const msg = h.data as IncomingMessage; + if (sessionId && msg) { + const cfg = readConfig(ctx.config); + if (cfg.relayOwnMessages) { + void handleSent(buildDeps(cfg, sessionId), sessionId, h.source, msg).catch(e => + ctx.logger.error('sent hook failed', e), + ); + } + } + return { continue: true }; + }); + ctx.registerWebhook('chatwoot', async (req: WebhookRequest) => handleOutbound( { lock, conversations: ctx.conversations, handover: ctx.handover, + engine: ctx.engine, store, inboxId: readConfig(ctx.config).inboxId, log: (m, e) => ctx.logger.error(m, e), diff --git a/chatwoot-adapter/manifest.json b/chatwoot-adapter/manifest.json index aa8571e..95b3218 100644 --- a/chatwoot-adapter/manifest.json +++ b/chatwoot-adapter/manifest.json @@ -1,7 +1,7 @@ { "id": "chatwoot-adapter", "name": "Chatwoot Adapter", - "version": "0.3.0", + "version": "0.4.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.", @@ -11,14 +11,14 @@ "repository": "https://github.com/rmyndharis/OpenWA-plugins", "keywords": ["chatwoot", "helpdesk", "inbox", "handover", "two-way", "agent", "whatsapp", "openwa"], "status": "beta", - "minOpenWAVersion": "0.8.6", - "testedOpenWAVersion": "0.8.6", + "minOpenWAVersion": "0.8.7", + "testedOpenWAVersion": "0.8.7", "sdkVersion": "1", "permissions": ["net:fetch", "conversation:send", "webhook:ingress", "engine:read"], "net": { "allow": [], "allowConfigHosts": ["baseUrl"] }, "sessionScoped": true, "sessions": ["*"], - "hooks": ["message:received"], + "hooks": ["message:received", "message:sent"], "ingress": [ { "route": "chatwoot", @@ -78,6 +78,12 @@ "default": true, "description": "Upload inbound WhatsApp media (images, files, audio) to Chatwoot as attachments." }, + "relayOwnMessages": { + "type": "boolean", + "title": "Relay your own outbound sends", + "default": true, + "description": "Mirror messages you send from a linked phone / the WhatsApp app / the OpenWA API into Chatwoot as outgoing messages, so the thread isn't one-sided. Your Chatwoot agent replies are never duplicated. Turn off to keep phone-composed messages out of the helpdesk." + }, "backfillLimit": { "type": "number", "title": "History backfill (messages per chat)", diff --git a/chatwoot-adapter/outbound.test.ts b/chatwoot-adapter/outbound.test.ts index f624893..b47834f 100644 --- a/chatwoot-adapter/outbound.test.ts +++ b/chatwoot-adapter/outbound.test.ts @@ -35,6 +35,7 @@ function deps(over: { store?: Record } = {}) { lock: new KeyedAsyncLock(), conversations: { send: async (e: { sessionId?: string; chatId?: string; type: string; text?: string }) => void sent.push(e) }, handover: { set: async (k: unknown, s: string) => void handovers.push([k, s]) }, + engine: { canonicalChatId: async (_s: string, c: string) => c }, store: { hasSeen: async () => false, markSeen: async () => {}, @@ -124,6 +125,7 @@ test('cross-tenant: two accounts sharing conversation id 55 route to the correct lock: new KeyedAsyncLock(), conversations: { send: async (e: { sessionId?: string; chatId?: string; text?: string }) => void sent.push(e) }, handover: { set: async () => {} }, + engine: { canonicalChatId: async (_s: string, c: string) => c }, store, inboxId: 7, log: () => {}, @@ -151,6 +153,7 @@ test('a transient send failure does not poison dedup: the retry re-sends the rep }, }, handover: { set: async () => {} }, + engine: { canonicalChatId: async (_s: string, c: string) => c }, store, inboxId: 7, log: () => {}, @@ -162,6 +165,61 @@ test('a transient send failure does not poison dedup: the retry re-sends the rep assert.equal(sent[0].text, 'hi'); }); +test('marks the WA send id as seen (scoped by the WA session) so message:sent does not echo the agent reply', async () => { + const marks: Array<[string, string, string | undefined]> = []; + const d = { + lock: new KeyedAsyncLock(), + // conversations.send returns { messageId, timestamp } — the engine's serialized id of the sent + // message, which is the same id the subsequent message:sent event carries. + conversations: { send: async () => ({ messageId: 'WA_9', timestamp: 0 }) }, + handover: { set: async () => {} }, + engine: { canonicalChatId: async (_s: string, c: string) => c }, + store: { + hasSeen: async () => false, + markSeen: async (kind: string, id: string, scope?: string) => void marks.push([kind, id, scope]), + // The WA session that owns this chat (target.sessionId) differs from the delivery scope on purpose. + getByConversation: async () => ({ sessionId: 'sessX', chatId: 'c@wa' }), + }, + inboxId: 7, + log: () => {}, + } as unknown as OutboundDeps; + await handleOutbound( + d, + req({ event: 'message_created', message_type: 'outgoing', private: false, id: 5, content: 'reply', inbox: { id: 7 }, conversation: { id: 55 } }), + ); + assert.ok( + marks.some(([k, id, scope]) => k === 'wa' && id === 'WA_9' && scope === 'sessX'), + `expected a wa:WA_9 marker scoped by the WA session sessX, got ${JSON.stringify(marks)}`, + ); +}); + +test('relay canonicalizes the WA chat id (via engine.canonicalChatId) so its lock serializes with the own-send handler', async () => { + const calls: Array<[string, string]> = []; + const d = { + lock: new KeyedAsyncLock(), + conversations: { send: async () => ({ messageId: 'X', timestamp: 0 }) }, + handover: { set: async () => {} }, + engine: { + canonicalChatId: async (s: string, c: string) => { + calls.push([s, c]); + return c === '628@lid' ? '628@c.us' : c; + }, + }, + store: { + hasSeen: async () => false, + markSeen: async () => {}, + getByConversation: async () => ({ sessionId: 'sessX', chatId: '628@lid' }), + }, + inboxId: 7, + log: () => {}, + } as unknown as OutboundDeps; + await handleOutbound( + d, + req({ event: 'message_created', message_type: 'outgoing', private: false, id: 5, content: 'r', inbox: { id: 7 }, conversation: { id: 55 } }), + ); + assert.deepEqual(calls, [['sessX', '628@lid']]); // canonicalized with the target's session + raw chatId, before locking +}); + test('handover guards a non-object changed_attributes element (no throw / retry loop)', async () => { const { deps: d, handovers } = deps(); const r = await handleOutbound(d, req({ event: 'conversation_updated', conversation: { id: 55 }, changed_attributes: [null, 'x', { assignee_id: { current_value: 3 } }] })); diff --git a/chatwoot-adapter/outbound.ts b/chatwoot-adapter/outbound.ts index f04b1e6..2058be3 100644 --- a/chatwoot-adapter/outbound.ts +++ b/chatwoot-adapter/outbound.ts @@ -1,4 +1,10 @@ -import type { WebhookRequest, PluginConversationsCapability, PluginHandoverCapability, HandoverState } from '../types/openwa'; +import type { + WebhookRequest, + PluginConversationsCapability, + PluginHandoverCapability, + PluginEngineReadCapability, + HandoverState, +} from '../types/openwa'; import type { MappingStore } from './mapping-store.ts'; import type { KeyedAsyncLock } from './chat-lock.ts'; import { shouldRelayOutbound, type ChatwootWebhookMessage } from './filters.ts'; @@ -7,6 +13,9 @@ export interface OutboundDeps { lock: KeyedAsyncLock; conversations: PluginConversationsCapability; handover: PluginHandoverCapability; + // Resolves @lid -> @c.us so the per-chat lock key matches handleSent's for a migrated contact (the echo + // marker + own-send handler must serialize on one canonical key). + engine: PluginEngineReadCapability; store: MappingStore; inboxId: number; log: (m: string, e?: unknown) => void; @@ -50,13 +59,15 @@ async function relay(deps: OutboundDeps, sessionId: string | undefined, evt: Cha deps.log(`no WA mapping for conversation ${conversationId}`); return; } - await deps.lock.run(`${target.sessionId}:${target.chatId}`, async () => { + const lockKey = await deps.engine.canonicalChatId(target.sessionId, target.chatId); + await deps.lock.run(`${target.sessionId}:${lockKey}`, async () => { const id = evt.id !== undefined ? String(evt.id) : undefined; // Dedup, but mark only AFTER a successful send: a transient send failure must retry the reply, not be // silently suppressed as "already seen". Scope the marker by the delivery's session (F-02/F-03). if (id && (await deps.store.hasSeen('cw', id, sessionId))) return; + let res: unknown; if (media) { - await deps.conversations.send({ + res = await deps.conversations.send({ sessionId: target.sessionId, chatId: target.chatId, type: media.type, @@ -64,9 +75,22 @@ async function relay(deps: OutboundDeps, sessionId: string | undefined, evt: Cha text: text || undefined, }); } else { - await deps.conversations.send({ sessionId: target.sessionId, chatId: target.chatId, type: 'text', text }); + res = await deps.conversations.send({ sessionId: target.sessionId, chatId: target.chatId, type: 'text', text }); } if (id) await deps.store.markSeen('cw', id, sessionId); + // Echo guard for the own-send relay (#615): the message we just sent to WhatsApp will come back as a + // fromMe message:sent event. Mark its WA id seen — scoped by the WA session that will emit it + // (target.sessionId, NOT the delivery scope) — so handleSent recognizes it as ours and skips it. Held + // inside this lock, so the mark lands before message:sent can acquire the same per-chat lock. + const sentId = (res as { messageId?: string } | null)?.messageId; + if (sentId) { + await deps.store.markSeen('wa', sentId, target.sessionId); + } else { + // No id to key the echo guard on (an engine that returns an empty messageId — e.g. Baileys' `?? ''`). + // The reply's own message:sent could then be re-relayed as a duplicate; surface it rather than fail + // silently open. + deps.log('conversation.send returned no message id; own-send echo guard skipped for this reply'); + } }); } @@ -109,7 +133,11 @@ async function applyHandover(deps: OutboundDeps, sessionId: string | undefined, if (!state) return; // never infer human from status alone const resolved = state; // const so the closure keeps the non-null narrowing // instanceId in the mapping mirror = sessionId (a session-scoped instance is 1:1 with its session), so - // inbound's ctx.mappings.upsert and this handover.set address the SAME conversation-mapping row. + // inbound's ctx.mappings.upsert and this handover.set address the SAME conversation-mapping row. Lock on + // the RAW chatId (not canonicalized): a handover webhook arrives independently of WA session state, and + // canonicalChatId requires a live engine — resolving it here would wedge a handover while the session is + // offline. Handover (a state-flag write) and relay (a message send) are independent, so they don't need + // to serialize with each other; the raw key still serializes concurrent handover events for a chat. await deps.lock.run(`${target.sessionId}:${target.chatId}`, () => deps.handover.set({ sessionId: target.sessionId, chatId: target.chatId, instanceId: target.sessionId }, resolved), ); diff --git a/chatwoot-adapter/relay.ts b/chatwoot-adapter/relay.ts index e8c8e5d..1a8fe6a 100644 --- a/chatwoot-adapter/relay.ts +++ b/chatwoot-adapter/relay.ts @@ -41,8 +41,19 @@ function locationText(msg: IncomingMessage): string { // A short stand-in for a bodyless message we couldn't relay as media (e.g. a voice note or sticker whose // blob was omitted for size), so Chatwoot shows a meaningful line instead of an empty bubble. function placeholderFor(msg: IncomingMessage): string { + // Type-based markers first: a media message relayed via `message:sent` (own outbound send) carries NO + // media object at all — wwjs' message_create path is not enriched — so `msg.media` is absent and only + // `msg.type` distinguishes it. Without a per-type marker a caption-less photo/video/etc would post an + // empty Chatwoot bubble (or 422 → drop, since it's already markSeen). Location coords are likewise + // absent on that path. if (msg.type === 'voice') return '🎤 Voice message'; if (msg.type === 'sticker') return '🎨 Sticker'; + if (msg.type === 'location') return '📍 Location'; + if (msg.type === 'image') return '📷 Photo'; + if (msg.type === 'video') return '🎥 Video'; + if (msg.type === 'audio') return '🎵 Audio'; + if (msg.type === 'contact') return '👤 Contact'; + if (msg.type === 'document') return `📎 ${msg.media?.filename ?? 'Document'}`; if (msg.media) return `📎 ${msg.media.filename ?? 'Attachment'}`; return msg.body; } diff --git a/chatwoot-adapter/sent.test.ts b/chatwoot-adapter/sent.test.ts new file mode 100644 index 0000000..18e11a4 --- /dev/null +++ b/chatwoot-adapter/sent.test.ts @@ -0,0 +1,170 @@ +import { test } from 'node:test'; +import assert from 'node:assert/strict'; +import { handleSent } from './sent.ts'; +import type { InboundDeps } from './relay.ts'; +import { KeyedAsyncLock } from './chat-lock.ts'; +import type { IncomingMessage } from '../types/openwa'; + +// A fromMe send composed on a linked phone / the OpenWA API (chatId = the recipient). +const own = { + id: 'o1', from: 'me@c.us', to: '621@c.us', chatId: '621@c.us', body: 'from my phone', type: 'chat', + timestamp: 0, fromMe: true, isGroup: false, +} as IncomingMessage; + +function deps( + over: { + client?: Record; + store?: Record; + engine?: Record; + relayGroups?: boolean; + relayMedia?: boolean; + } = {}, +) { + let contacts = 0; + let convs = 0; + const posted: Array<{ id: number; c: string; type?: string }> = []; + const seen = new Set(); + const links = new Map(); + const client = { + searchContact: async () => null, + createContact: async () => { contacts++; return { id: 9, sourceId: 'src' }; }, + findOpenConversation: async () => null, + createConversation: async () => { convs++; return 55; }, + postText: async (id: number, c: string, o?: { messageType?: string }) => { posted.push({ id, c, type: o?.messageType }); return { id: 1 }; }, + postMedia: async (id: number, c: string, _f: unknown, o?: { messageType?: string }) => { posted.push({ id, c, type: o?.messageType }); return { id: 2 }; }, + ...over.client, + }; + const store = { + getByChat: async (s: string, c: string) => links.get(`${s}:${c}`) ?? null, + link: async (s: string, c: string, _i: string, l: unknown) => void links.set(`${s}:${c}`, l), + hasSeen: async (_k: string, id: string, scope?: string) => seen.has(`${scope ?? ''}:${id}`), + markSeen: async (_k: string, id: string, scope?: string) => void seen.add(`${scope ?? ''}:${id}`), + ...over.store, + }; + // Default: identity canonicalization (@lid resolution tested explicitly below). + const engine = { canonicalChatId: async (_s: string, c: string) => c, ...over.engine }; + const d = { + lock: new KeyedAsyncLock(), client, store, engine, instanceId: 'inst', + relayGroups: over.relayGroups ?? true, relayMedia: over.relayMedia ?? true, backfillLimit: 0, backfillAllOnce: false, + log: () => {}, + } as unknown as InboundDeps; + return { deps: d, counts: () => ({ contacts, convs }), posted, seen }; +} + +test('an own send to an UNMAPPED chat is dropped — relay-into-existing only, never creates (no split)', async () => { + const { deps: d, posted, counts } = deps(); // default store has no mapping for this chat + await handleSent(d, 'sess', 'Engine', own); + assert.equal(posted.length, 0); + assert.deepEqual(counts(), { contacts: 0, convs: 0 }); +}); + +test('reuses an existing mapped conversation and does NOT create a contact', async () => { + const { deps: d, posted, counts } = deps({ + store: { getByChat: async () => ({ conversationId: 77, contactId: 9, sourceId: 'src', name: 'Budi' }) }, + }); + await handleSent(d, 'sess', 'Engine', own); + assert.deepEqual(posted, [{ id: 77, c: 'from my phone', type: 'outgoing' }]); + assert.deepEqual(counts(), { contacts: 0, convs: 0 }); +}); + +test('skips when the WA id is already seen — an agent reply the adapter itself sent does not echo', async () => { + const { deps: d, posted } = deps(); + await d.store.markSeen('wa', 'o1', 'sess'); // outbound.relay marked it when it sent the agent reply + await handleSent(d, 'sess', 'Engine', own); + assert.equal(posted.length, 0); +}); + +test('is at-most-once: the same send processed twice posts once', async () => { + const { deps: d, posted } = deps({ + store: { getByChat: async () => ({ conversationId: 55, contactId: 9, sourceId: 'src', name: 'x' }) }, + }); + await handleSent(d, 'sess', 'Engine', own); + await handleSent(d, 'sess', 'Engine', own); + assert.equal(posted.length, 1); +}); + +test('relays own media as an OUTGOING attachment when relayMedia (postMedia, not postText)', async () => { + let mediaOpts: { messageType?: string } | undefined; + const { deps: d, posted } = deps({ + store: { getByChat: async () => ({ conversationId: 55, contactId: 9, sourceId: 'src', name: 'x' }) }, + client: { postMedia: async (_id: number, _c: string, _f: unknown, o: { messageType?: string }) => { mediaOpts = o; return { id: 2 }; } }, + }); + const media = { ...own, id: 'o2', body: '', type: 'image', media: { mimetype: 'image/jpeg', data: 'AAA' } } as IncomingMessage; + await handleSent(d, 'sess', 'Engine', media); + assert.equal(mediaOpts?.messageType, 'outgoing'); + assert.equal(posted.length, 0); // postText path was not taken +}); + +test('a phone-composed location (no coords in the message:sent payload) posts a location placeholder, not an empty bubble', async () => { + const { deps: d, posted } = deps({ + store: { getByChat: async () => ({ conversationId: 55, contactId: 9, sourceId: 'src', name: 'x' }) }, + }); + const loc = { ...own, id: 'o3', body: '', type: 'location' } as IncomingMessage; // no msg.location + await handleSent(d, 'sess', 'Engine', loc); + assert.deepEqual(posted, [{ id: 55, c: '📍 Location', type: 'outgoing' }]); +}); + +test('a caption-less own media send (no media object in the message:sent payload) posts a type placeholder, not an empty bubble', async () => { + const { deps: d, posted } = deps({ + store: { getByChat: async () => ({ conversationId: 55, contactId: 9, sourceId: 'src', name: 'x' }) }, + }); + // The real wwjs message:sent shape for a phone photo without a caption: type only, no media, no body. + const photo = { ...own, id: 'o9', body: '', type: 'image' } as IncomingMessage; + await handleSent(d, 'sess', 'Engine', photo); + assert.deepEqual(posted, [{ id: 55, c: '📷 Photo', type: 'outgoing' }]); +}); + +test('a migrated contact (@lid send, @c.us-keyed mapping) relays into the EXISTING conversation via dual-lookup, no split', async () => { + // The recipient migrated to @lid: this message:sent carries chatId=@lid, but the mapping was created + // under @c.us. Without the canonical fallback lookup this would create a duplicate @lid conversation. + const lidMsg = { ...own, id: 'o7', chatId: '628@lid' } as IncomingMessage; + const byChat = new Map([ + ['sess:628@c.us', { conversationId: 77, contactId: 9, sourceId: 'src', name: 'Budi' }], + ]); + const { deps: d, posted, counts } = deps({ + engine: { canonicalChatId: async (_s: string, c: string) => (c === '628@lid' ? '628@c.us' : c) }, + store: { getByChat: async (_s: string, c: string) => byChat.get(`sess:${c}`) ?? null }, + }); + await handleSent(d, 'sess', 'Engine', lidMsg); + assert.deepEqual(posted, [{ id: 77, c: 'from my phone', type: 'outgoing' }]); + assert.deepEqual(counts(), { contacts: 0, convs: 0 }); // found via the canonical lookup — nothing created +}); + +test('an agent reply to a migrated (@lid) contact does not echo: the id-keyed marker is still seen', async () => { + const lidMsg = { ...own, id: 'o8', chatId: '628@lid' } as IncomingMessage; + const { deps: d, posted } = deps({ + engine: { canonicalChatId: async (_s: string, c: string) => (c === '628@lid' ? '628@c.us' : c) }, + }); + await d.store.markSeen('wa', 'o8', 'sess'); // outbound.relay marked the WA id it sent (chatId-independent) + await handleSent(d, 'sess', 'Engine', lidMsg); + assert.equal(posted.length, 0); +}); + +test('cold lid->phone table (canonicalChatId cannot resolve @lid) DROPS the own send instead of splitting the @c.us conversation', async () => { + // Finding 1: on wwjs the lid->phone table is often cold, so canonicalChatId(@lid) returns @lid. The + // @c.us-keyed mapping is then invisible — but creating a @lid conversation here would split the thread. + const lidMsg = { ...own, id: 'o10', chatId: '628@lid' } as IncomingMessage; + const byChat = new Map([ + ['sess:628@c.us', { conversationId: 77, contactId: 9, sourceId: 'src', name: 'Budi' }], + ]); + const { deps: d, posted, counts } = deps({ + engine: { canonicalChatId: async (_s: string, c: string) => c }, // cold: @lid stays @lid + store: { getByChat: async (_s: string, c: string) => byChat.get(`sess:${c}`) ?? null }, + }); + await handleSent(d, 'sess', 'Engine', lidMsg); + assert.equal(posted.length, 0); // raw @lid misses; canonical==raw so no 2nd lookup → drop + assert.deepEqual(counts(), { contacts: 0, convs: 0 }); // crucially: NO duplicate conversation created +}); + +test('respects relayGroups=false for a fromMe group send', async () => { + const { deps: d, posted } = deps({ relayGroups: false }); + const grp = { ...own, id: 'o4', chatId: '12@g.us', isGroup: true } as IncomingMessage; + await handleSent(d, 'sess', 'Engine', grp); + assert.equal(posted.length, 0); +}); + +test('ignores a non-Engine source (defensive)', async () => { + const { deps: d, posted } = deps(); + await handleSent(d, 'sess', 'API', own); + assert.equal(posted.length, 0); +}); diff --git a/chatwoot-adapter/sent.ts b/chatwoot-adapter/sent.ts new file mode 100644 index 0000000..a781018 --- /dev/null +++ b/chatwoot-adapter/sent.ts @@ -0,0 +1,58 @@ +import type { IncomingMessage } from '../types/openwa'; +import { shouldRelayOwn } from './filters.ts'; +import { relayMessage, type InboundDeps } from './relay.ts'; + +// WhatsApp → Chatwoot for the account's OWN outbound sends — messages composed on a linked phone / the +// WhatsApp mobile app / the OpenWA REST API — so the Chatwoot thread mirrors the full WhatsApp +// conversation (#615). Posts 'outgoing' into a conversation the INBOUND path already mapped. +// +// Relay-into-existing-only, never create: an own send whose chat isn't mapped yet is dropped. Creating a +// conversation on a miss would split a contact who has migrated to @lid — its @c.us mapping is invisible +// to a raw @lid lookup, and @lid->@c.us resolution is only best-effort (the lid->phone table is cold on +// a phone-only-contacted wwjs contact), so a "new" chat can't be told apart from a migrated one. Dropping +// is strictly safe (no duplicate conversation); the chat still appears once the customer replies inbound. +// +// Echo guard: the adapter's own Chatwoot-agent replies are ALSO fromMe and arrive here as message:sent. +// outbound.relay marks the WA id of every reply it sends (markSeen('wa', …)) on the same canonical +// per-chat lock, so the hasSeen check below recognizes and skips them. At-most-once (a failed post is +// not retried), like inbound. +export async function handleSent( + deps: InboundDeps, + sessionId: string, + source: string, + msg: IncomingMessage, +): Promise { + if (!shouldRelayOwn(msg, source, deps.relayGroups)) return; + // Canonicalize @lid -> @c.us for the lock key. A send to a contact WhatsApp has migrated to @lid + // resolves to the @lid address, so this message:sent carries chatId=@lid while outbound.relay locked + // (and marked the echo on) the mapping's @c.us key. Locking on the canonical form makes both paths + // serialize on the SAME key (when resolvable), so the echo marker is observed before this handler checks. + const key = await deps.engine.canonicalChatId(sessionId, msg.chatId); + await deps.lock.run(`${sessionId}:${key}`, async () => { + try { + if (await deps.store.hasSeen('wa', msg.id, sessionId)) return; + const conversationId = await findMappedConversation(deps, sessionId, msg, key); + if (conversationId === null) return; // unmapped chat — drop, never create (no split) + await deps.store.markSeen('wa', msg.id, sessionId); + await relayMessage(deps, conversationId, msg, 'outgoing'); + } catch (err) { + deps.log('own-send relay failed', err); + } + }); +} + +// Find the Chatwoot conversation this chat is already mapped to, or null. Dual lookup: the raw chatId +// finds a mapping keyed by @lid (created post-migration), 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 own send +// lands in its EXISTING conversation instead of a duplicate. Never creates. +async function findMappedConversation( + deps: InboundDeps, + sessionId: string, + msg: IncomingMessage, + canonicalChatId: string, +): Promise { + const existing = + (await deps.store.getByChat(sessionId, msg.chatId)) ?? + (canonicalChatId !== msg.chatId ? await deps.store.getByChat(sessionId, canonicalChatId) : null); + return existing ? existing.conversationId : null; +} diff --git a/plugins.json b/plugins.json index 170899a..0a1e1e5 100644 --- a/plugins.json +++ b/plugins.json @@ -369,7 +369,7 @@ { "id": "chatwoot-adapter", "name": "Chatwoot Adapter", - "version": "0.3.0", + "version": "0.4.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.", @@ -385,13 +385,13 @@ "whatsapp", "openwa" ], - "minOpenWAVersion": "0.8.6", - "testedOpenWAVersion": "0.8.6", + "minOpenWAVersion": "0.8.7", + "testedOpenWAVersion": "0.8.7", "releasedAt": "2026-07-03", "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.3.0/chatwoot-adapter.zip" + "download": "https://github.com/rmyndharis/OpenWA-plugins/releases/download/chatwoot-adapter-v0.4.0/chatwoot-adapter.zip" }, { "id": "faq-bot", diff --git a/types/openwa.d.ts b/types/openwa.d.ts index 7488c96..705cd1a 100644 --- a/types/openwa.d.ts +++ b/types/openwa.d.ts @@ -71,6 +71,12 @@ export interface PluginEngineReadCapability { getChats(sessionId: string): Promise; /** Recent messages for a chat, both directions (v0.8.5+). The host clamps `limit` (max 100). */ getChatHistory(sessionId: string, chatId: string, limit?: number, includeMedia?: boolean): Promise; + /** + * Canonical (neutral) form of a chat id: resolves a `@lid` privacy id to its stable `@c.us` when + * the mapping is known, else returns the id unchanged. Lets a plugin key a chat by one identity across + * WhatsApp's `@lid` migration (best-effort). Available on OpenWA 0.8.7+. + */ + canonicalChatId(sessionId: string, chatId: string): Promise; } // ── v0.7: host-proxied, SSRF-guarded outbound HTTP ──────────────────────────────────────────────