feat(FN-3713): migrate WhatsApp plugin to Baileys pairing flow
- Add Baileys-based connection/auth modules and pairing route support for QR, pair code, status, and logout - Refactor plugin entrypoint and split reply/auth/connection logic into focused modules - Update plugin metadata/docs/settings schema to pairing-era configuration and remove legacy webhook key usage - Add targeted tests for auth state, connection handling, reply flow, and updated schema expectations Fusion-Task-Id: FN-3713
This commit is contained in:
Fusion
@@ -1,32 +1,51 @@
|
||||
# WhatsApp Chat Plugin
|
||||
|
||||
Bridges Meta WhatsApp Cloud webhooks to a Fusion AI session so you can chat with your configured Fusion assistant from WhatsApp.
|
||||
WhatsApp Web (Baileys) bridge for Fusion. It pairs with your phone (QR or pairing code), keeps a background connection alive, forwards inbound direct text messages to a Fusion AI session, and sends the assistant reply back to WhatsApp.
|
||||
|
||||
No Meta Cloud app, webhook URL, verify token, or Graph API credentials are required.
|
||||
|
||||
## Setup
|
||||
|
||||
1. Enable/install the plugin in Plugin Manager.
|
||||
2. Configure `allowedSenders` (empty means **nobody is allowed**).
|
||||
3. Choose `pairingMode`:
|
||||
- `qr` (default): fetch QR from `/api/plugins/fusion-plugin-whatsapp-chat/qr` and scan in WhatsApp.
|
||||
- `code`: set `pairingPhoneNumber` (E.164 digits without `+`) and request code via `/pair-code`.
|
||||
4. Confirm `/status` reports `connected`.
|
||||
|
||||
## Settings
|
||||
|
||||
- `verifyToken`: Webhook verify token configured in Meta app settings.
|
||||
- `appSecret`: Meta app secret used for `x-hub-signature-256` validation.
|
||||
- `accessToken`: WhatsApp Cloud API token used to send replies.
|
||||
- `phoneNumberId`: WhatsApp phone number ID used for Graph API sends.
|
||||
- `graphApiVersion`: Graph API version (default `v21.0`).
|
||||
- `allowedSenders`: Optional allowlist of sender phone numbers.
|
||||
- `agentSystemPrompt`: Optional system prompt for generated replies.
|
||||
- `pairingMode`: `qr` or `code`.
|
||||
- `pairingPhoneNumber`: E.164 digits without `+` (used for `code` mode).
|
||||
- `allowedSenders`: allowed WhatsApp JIDs or E.164 digits.
|
||||
- `agentSystemPrompt`: system prompt for replies.
|
||||
- `historyTurnLimit`: persisted turn window (default `40`).
|
||||
|
||||
## Webhook routes
|
||||
## Routes
|
||||
|
||||
- `GET /api/plugins/fusion-plugin-whatsapp-chat/webhook` verification challenge.
|
||||
- `POST /api/plugins/fusion-plugin-whatsapp-chat/webhook` signed event ingress.
|
||||
- `GET /api/plugins/fusion-plugin-whatsapp-chat/status`
|
||||
- `GET /api/plugins/fusion-plugin-whatsapp-chat/qr`
|
||||
- `POST /api/plugins/fusion-plugin-whatsapp-chat/pair-code`
|
||||
- `POST /api/plugins/fusion-plugin-whatsapp-chat/logout`
|
||||
|
||||
## Behavior
|
||||
## Storage and lifecycle
|
||||
|
||||
- Validates Meta signature against raw request body.
|
||||
- Ignores unsupported/non-text messages.
|
||||
- Deduplicates inbound WhatsApp message IDs.
|
||||
- Persists sender transcript history to keep multi-turn continuity.
|
||||
- Sends reply chunks up to WhatsApp text limits.
|
||||
- Starts socket on `onLoad`, stops on `onUnload`.
|
||||
- Persists transcript and dedupe state in:
|
||||
- `whatsapp_chat_sessions`
|
||||
- `whatsapp_chat_dedupe`
|
||||
- Persists Baileys auth state in:
|
||||
- `whatsapp_auth_creds`
|
||||
- `whatsapp_auth_keys`
|
||||
- After restart, plugin reconnects automatically when auth is valid.
|
||||
|
||||
## Troubleshooting
|
||||
|
||||
- 401 on webhook POST: check `appSecret` and raw-body signature header.
|
||||
- 403 on webhook GET: verify token mismatch.
|
||||
- No replies: ensure sender is in `allowedSenders` (or clear allowlist) and `accessToken`/`phoneNumberId` are valid.
|
||||
- Stuck `awaiting-qr`: fetch a fresh QR and scan promptly.
|
||||
- `loggedOut`: call `/logout` (or wait for clear) and re-pair.
|
||||
- Pair code not generated: ensure `pairingPhoneNumber` is E.164 digits without `+`.
|
||||
- No replies: check `allowedSenders`; empty list blocks all inbound messages by design.
|
||||
|
||||
## Compliance warning
|
||||
|
||||
Baileys is an unofficial WhatsApp Web protocol client. Use may violate WhatsApp Terms of Service. This plugin is intended for self-hosted, single-user use at your own risk.
|
||||
|
||||
@@ -2,7 +2,7 @@
|
||||
"id": "fusion-plugin-whatsapp-chat",
|
||||
"name": "WhatsApp Chat",
|
||||
"version": "0.1.0",
|
||||
"description": "Bridge WhatsApp Cloud webhook messages to a Fusion agent conversation",
|
||||
"description": "WhatsApp Web (multi-device) bridge — pairs via QR/code, no Meta Cloud webhook required.",
|
||||
"author": "Fusion Team",
|
||||
"fusionVersion": ">=0.1.0"
|
||||
}
|
||||
|
||||
@@ -2,7 +2,7 @@
|
||||
"name": "@fusion-plugin-examples/whatsapp-chat",
|
||||
"version": "0.1.0",
|
||||
"type": "module",
|
||||
"description": "WhatsApp Cloud webhook bridge for Fusion agents",
|
||||
"description": "WhatsApp Web (Baileys) chat bridge for Fusion agents",
|
||||
"keywords": [
|
||||
"fusion-plugin",
|
||||
"whatsapp",
|
||||
@@ -21,7 +21,10 @@
|
||||
"test": "vitest run --silent=passed-only --reporter=dot"
|
||||
},
|
||||
"dependencies": {
|
||||
"@fusion/plugin-sdk": "workspace:*"
|
||||
"@fusion/plugin-sdk": "workspace:*",
|
||||
"@whiskeysockets/baileys": "^6.7.21",
|
||||
"pino": "^9.9.0",
|
||||
"qrcode": "^1.5.4"
|
||||
},
|
||||
"devDependencies": {
|
||||
"@types/node": "^25.5.2",
|
||||
|
||||
@@ -0,0 +1,99 @@
|
||||
import { describe, expect, it } from "vitest";
|
||||
import { clearAuthState, createPluginDbAuthState } from "../auth-state.js";
|
||||
|
||||
function createInMemoryDb() {
|
||||
const creds = new Map<string, string>();
|
||||
const keys = new Map<string, string>();
|
||||
const makeKey = (category: string, id: string) => `${category}:${id}`;
|
||||
|
||||
return {
|
||||
prepare(sql: string) {
|
||||
return {
|
||||
get: (...args: unknown[]) => {
|
||||
if (sql.includes("FROM whatsapp_auth_creds")) {
|
||||
const value = creds.get("creds");
|
||||
return value ? { value } : undefined;
|
||||
}
|
||||
if (sql.includes("FROM whatsapp_auth_keys")) {
|
||||
const key = makeKey(args[0] as string, args[1] as string);
|
||||
const value = keys.get(key);
|
||||
return value ? { value } : undefined;
|
||||
}
|
||||
return undefined;
|
||||
},
|
||||
run: (...args: unknown[]) => {
|
||||
if (sql.includes("INSERT INTO whatsapp_auth_creds")) {
|
||||
creds.set("creds", args[0] as string);
|
||||
}
|
||||
if (sql.includes("DELETE FROM whatsapp_auth_creds")) {
|
||||
creds.clear();
|
||||
}
|
||||
if (sql.includes("INSERT INTO whatsapp_auth_keys")) {
|
||||
keys.set(makeKey(args[0] as string, args[1] as string), args[2] as string);
|
||||
}
|
||||
if (sql.includes("DELETE FROM whatsapp_auth_keys WHERE category")) {
|
||||
keys.delete(makeKey(args[0] as string, args[1] as string));
|
||||
}
|
||||
if (sql.includes("DELETE FROM whatsapp_auth_keys")) {
|
||||
keys.clear();
|
||||
}
|
||||
},
|
||||
};
|
||||
},
|
||||
exec() {},
|
||||
_creds: creds,
|
||||
_keys: keys,
|
||||
};
|
||||
}
|
||||
|
||||
describe("auth-state", () => {
|
||||
it("round-trips creds", async () => {
|
||||
const db = createInMemoryDb();
|
||||
const auth = createPluginDbAuthState(db as any);
|
||||
auth.state.creds.me = { id: "123@s.whatsapp.net", name: "Fusion" } as any;
|
||||
await auth.saveCreds();
|
||||
|
||||
const next = createPluginDbAuthState(db as any);
|
||||
expect(next.state.creds.me?.id).toBe("123@s.whatsapp.net");
|
||||
});
|
||||
|
||||
it("sets, gets, and deletes key categories", async () => {
|
||||
const db = createInMemoryDb();
|
||||
const auth = createPluginDbAuthState(db as any);
|
||||
|
||||
await auth.state.keys.set({
|
||||
session: { alpha: { foo: "bar" } as any },
|
||||
"sender-key": { beta: { baz: "qux" } as any },
|
||||
});
|
||||
|
||||
const loaded = await auth.state.keys.get("session", ["alpha", "missing"]);
|
||||
expect((loaded as any).alpha.foo).toBe("bar");
|
||||
expect((loaded as any).missing).toBeUndefined();
|
||||
|
||||
await auth.state.keys.set({ session: { alpha: null } });
|
||||
const removed = await auth.state.keys.get("session", ["alpha"]);
|
||||
expect((removed as any).alpha).toBeUndefined();
|
||||
});
|
||||
|
||||
it("clears auth state", async () => {
|
||||
const db = createInMemoryDb();
|
||||
const auth = createPluginDbAuthState(db as any);
|
||||
auth.state.creds.me = { id: "123@s.whatsapp.net", name: "Fusion" } as any;
|
||||
await auth.saveCreds();
|
||||
await auth.state.keys.set({ session: { alpha: { ok: true } as any } });
|
||||
|
||||
clearAuthState(db as any);
|
||||
|
||||
expect(db._creds.size).toBe(0);
|
||||
expect(db._keys.size).toBe(0);
|
||||
});
|
||||
|
||||
it("handles corrupt json gracefully", async () => {
|
||||
const db = createInMemoryDb();
|
||||
db._keys.set("session:bad", "not-json");
|
||||
|
||||
const auth = createPluginDbAuthState(db as any);
|
||||
const loaded = await auth.state.keys.get("session", ["bad"]);
|
||||
expect((loaded as any).bad).toBeUndefined();
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,145 @@
|
||||
import { beforeEach, describe, expect, it, vi } from "vitest";
|
||||
|
||||
const mockState = vi.hoisted(() => {
|
||||
const handlers = new Map<string, (payload: any) => void>();
|
||||
const sendMessage = vi.fn();
|
||||
const end = vi.fn();
|
||||
const logout = vi.fn();
|
||||
const requestPairingCode = vi.fn().mockResolvedValue("123-456");
|
||||
const makeWASocket = vi.fn(() => ({
|
||||
ev: {
|
||||
on: (name: string, handler: (payload: any) => void) => handlers.set(name, handler),
|
||||
off: (name: string) => handlers.delete(name),
|
||||
},
|
||||
user: { id: "15550001111@s.whatsapp.net" },
|
||||
sendMessage,
|
||||
end,
|
||||
logout,
|
||||
requestPairingCode,
|
||||
}));
|
||||
return { handlers, sendMessage, end, logout, requestPairingCode, makeWASocket };
|
||||
});
|
||||
|
||||
vi.mock("@whiskeysockets/baileys", () => ({
|
||||
default: mockState.makeWASocket,
|
||||
makeWASocket: mockState.makeWASocket,
|
||||
DisconnectReason: { loggedOut: 401 },
|
||||
BufferJSON: { reviver: undefined, replacer: undefined },
|
||||
initAuthCreds: () => ({}),
|
||||
}));
|
||||
|
||||
vi.mock("qrcode", () => ({
|
||||
default: { toDataURL: vi.fn().mockResolvedValue("data:image/png;base64,abc") },
|
||||
}));
|
||||
|
||||
import { WhatsAppConnection } from "../connection.js";
|
||||
|
||||
function createInMemoryDb() {
|
||||
const sessions = new Map<string, string>();
|
||||
const dedupe = new Set<string>();
|
||||
const creds = new Map<string, string>();
|
||||
const keys = new Map<string, string>();
|
||||
return {
|
||||
exec() {},
|
||||
prepare(sql: string) {
|
||||
return {
|
||||
get: (...args: unknown[]) => {
|
||||
if (sql.includes("FROM whatsapp_chat_sessions")) {
|
||||
const history = sessions.get(args[0] as string);
|
||||
return history ? { history } : undefined;
|
||||
}
|
||||
if (sql.includes("FROM whatsapp_chat_dedupe")) return dedupe.has(args[0] as string) ? { found: 1 } : undefined;
|
||||
if (sql.includes("FROM whatsapp_auth_creds")) return creds.get("creds") ? { value: creds.get("creds") } : undefined;
|
||||
if (sql.includes("FROM whatsapp_auth_keys")) return keys.get(`${args[0]}:${args[1]}`) ? { value: keys.get(`${args[0]}:${args[1]}`) } : undefined;
|
||||
return undefined;
|
||||
},
|
||||
run: (...args: unknown[]) => {
|
||||
if (sql.includes("whatsapp_chat_sessions")) sessions.set(args[0] as string, args[1] as string);
|
||||
if (sql.includes("whatsapp_chat_dedupe")) dedupe.add(args[0] as string);
|
||||
if (sql.includes("INSERT INTO whatsapp_auth_creds")) creds.set("creds", args[0] as string);
|
||||
if (sql.includes("DELETE FROM whatsapp_auth_creds")) creds.clear();
|
||||
if (sql.includes("INSERT INTO whatsapp_auth_keys")) keys.set(`${args[0]}:${args[1]}`, args[2] as string);
|
||||
if (sql.includes("DELETE FROM whatsapp_auth_keys WHERE category")) keys.delete(`${args[0]}:${args[1]}`);
|
||||
if (sql.includes("DELETE FROM whatsapp_auth_keys")) keys.clear();
|
||||
},
|
||||
};
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
function makeCtx(settings: Record<string, unknown> = {}) {
|
||||
return {
|
||||
pluginId: "fusion-plugin-whatsapp-chat",
|
||||
settings: { allowedSenders: ["15550001111"], ...settings },
|
||||
taskStore: { getRootDir: () => "/tmp", getPluginStore: () => ({}) },
|
||||
logger: { info: vi.fn(), warn: vi.fn(), error: vi.fn(), debug: vi.fn() },
|
||||
emitEvent: vi.fn(),
|
||||
} as any;
|
||||
}
|
||||
|
||||
describe("WhatsAppConnection", () => {
|
||||
beforeEach(() => {
|
||||
mockState.handlers.clear();
|
||||
mockState.makeWASocket.mockClear();
|
||||
mockState.sendMessage.mockClear();
|
||||
mockState.end.mockClear();
|
||||
});
|
||||
|
||||
it("starts and stops idempotently", async () => {
|
||||
const connection = new WhatsAppConnection(makeCtx(), "0.1.0", vi.fn().mockResolvedValue("reply"), createInMemoryDb() as any);
|
||||
await connection.start();
|
||||
await connection.stop();
|
||||
await connection.stop();
|
||||
expect(mockState.makeWASocket).toHaveBeenCalledTimes(1);
|
||||
expect(mockState.end).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it("exposes qr updates", async () => {
|
||||
const connection = new WhatsAppConnection(makeCtx(), "0.1.0", vi.fn().mockResolvedValue("reply"), createInMemoryDb() as any);
|
||||
await connection.start();
|
||||
await mockState.handlers.get("connection.update")?.({ qr: "abc" });
|
||||
expect(connection.getStatus()).toMatchObject({ state: "awaiting-qr", qr: "abc" });
|
||||
});
|
||||
|
||||
it("reconnects on close unless logged out", async () => {
|
||||
vi.useFakeTimers();
|
||||
const connection = new WhatsAppConnection(makeCtx(), "0.1.0", vi.fn().mockResolvedValue("reply"), createInMemoryDb() as any);
|
||||
await connection.start();
|
||||
await mockState.handlers.get("connection.update")?.({ connection: "close", lastDisconnect: { error: new Error("boom") } });
|
||||
vi.advanceTimersByTime(1000);
|
||||
expect(mockState.makeWASocket).toHaveBeenCalledTimes(2);
|
||||
|
||||
await mockState.handlers.get("connection.update")?.({ connection: "close", lastDisconnect: { error: { output: { statusCode: 401 } } } });
|
||||
vi.advanceTimersByTime(1000);
|
||||
expect(mockState.makeWASocket).toHaveBeenCalledTimes(2);
|
||||
vi.useRealTimers();
|
||||
});
|
||||
|
||||
it("drops unsupported inbound traffic", async () => {
|
||||
const reply = vi.fn().mockResolvedValue("hello");
|
||||
const connection = new WhatsAppConnection(makeCtx(), "0.1.0", reply, createInMemoryDb() as any);
|
||||
await connection.start();
|
||||
const upsert = mockState.handlers.get("messages.upsert")!;
|
||||
await upsert({ type: "notify", messages: [{ key: { remoteJid: "abc@g.us", id: "1", fromMe: false }, message: { conversation: "hi" } }] });
|
||||
await upsert({ type: "notify", messages: [{ key: { remoteJid: "15550001111@s.whatsapp.net", id: "2", fromMe: true }, message: { conversation: "hi" } }] });
|
||||
await upsert({ type: "notify", messages: [{ key: { remoteJid: "15550001111@s.whatsapp.net", id: "3", fromMe: false }, message: {} }] });
|
||||
expect(reply).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("dedupes and handles reply failure with fallback", async () => {
|
||||
const reply = vi.fn().mockRejectedValue(new Error("nope"));
|
||||
const connection = new WhatsAppConnection(makeCtx(), "0.1.0", reply, createInMemoryDb() as any);
|
||||
await connection.start();
|
||||
const payload = { type: "notify", messages: [{ key: { remoteJid: "15550001111@s.whatsapp.net", id: "m-1", fromMe: false }, message: { conversation: "hi" } }] };
|
||||
await mockState.handlers.get("messages.upsert")?.(payload);
|
||||
await mockState.handlers.get("messages.upsert")?.(payload);
|
||||
expect(reply).toHaveBeenCalledTimes(1);
|
||||
expect(mockState.sendMessage).toHaveBeenCalledWith("15550001111@s.whatsapp.net", { text: "Sorry, I hit an internal error while processing that message." });
|
||||
});
|
||||
|
||||
it("splits oversized messages", () => {
|
||||
const chunks = WhatsAppConnection.splitMessageForWhatsapp("x".repeat(9000));
|
||||
expect(chunks.length).toBeGreaterThan(2);
|
||||
expect(chunks[0].length).toBeLessThanOrEqual(4096);
|
||||
});
|
||||
});
|
||||
@@ -1,186 +1,28 @@
|
||||
import { createHmac } from "node:crypto";
|
||||
import { describe, expect, it, vi, beforeEach } from "vitest";
|
||||
import plugin, { splitMessageForWhatsapp, verifyMetaSignature, webhookGetHandler, webhookPostHandler } from "../index.js";
|
||||
|
||||
function createInMemoryDb() {
|
||||
const sessions = new Map<string, string>();
|
||||
const dedupe = new Set<string>();
|
||||
return {
|
||||
exec: vi.fn(),
|
||||
prepare: vi.fn((sql: string) => ({
|
||||
get: (value: string) => {
|
||||
if (sql.includes("FROM whatsapp_chat_sessions")) {
|
||||
const history = sessions.get(value);
|
||||
return history ? { history } : undefined;
|
||||
}
|
||||
if (sql.includes("FROM whatsapp_chat_dedupe")) {
|
||||
return dedupe.has(value) ? { found: 1 } : undefined;
|
||||
}
|
||||
return undefined;
|
||||
},
|
||||
run: (...args: unknown[]) => {
|
||||
if (sql.includes("whatsapp_chat_sessions")) {
|
||||
sessions.set(args[0] as string, args[1] as string);
|
||||
}
|
||||
if (sql.includes("whatsapp_chat_dedupe")) {
|
||||
dedupe.add(args[0] as string);
|
||||
}
|
||||
},
|
||||
})),
|
||||
get history() {
|
||||
return sessions;
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
function makeCtx(overrides: Partial<any> = {}) {
|
||||
const db = createInMemoryDb();
|
||||
const ctx: any = {
|
||||
settings: {
|
||||
appSecret: "secret",
|
||||
verifyToken: "verify-me",
|
||||
accessToken: "token",
|
||||
phoneNumberId: "123",
|
||||
allowedSenders: ["15551234567"],
|
||||
},
|
||||
taskStore: {
|
||||
getRootDir: () => "/tmp/project",
|
||||
getPluginStore: () => ({ db }),
|
||||
},
|
||||
createAiSession: vi.fn().mockResolvedValue({
|
||||
session: {
|
||||
prompt: vi.fn().mockResolvedValue(undefined),
|
||||
state: { messages: [{ role: "assistant", content: "hello from fusion" }] },
|
||||
},
|
||||
}),
|
||||
logger: { info: vi.fn(), warn: vi.fn(), error: vi.fn(), debug: vi.fn() },
|
||||
emitEvent: vi.fn(),
|
||||
...overrides,
|
||||
};
|
||||
return { ctx, db };
|
||||
}
|
||||
|
||||
function signBody(secret: string, body: string): string {
|
||||
return `sha256=${createHmac("sha256", secret).update(Buffer.from(body)).digest("hex")}`;
|
||||
}
|
||||
import { describe, expect, it, vi } from "vitest";
|
||||
import plugin, { splitMessageForWhatsapp } from "../index.js";
|
||||
|
||||
describe("whatsapp plugin", () => {
|
||||
beforeEach(() => {
|
||||
vi.restoreAllMocks();
|
||||
});
|
||||
|
||||
it("registers schema init hook", () => {
|
||||
expect(plugin.hooks?.onSchemaInit).toBeDefined();
|
||||
});
|
||||
|
||||
it("verifies GET webhook challenge", async () => {
|
||||
const { ctx } = makeCtx();
|
||||
const result = await webhookGetHandler({ query: { "hub.mode": "subscribe", "hub.verify_token": "verify-me", "hub.challenge": "abc" } }, ctx);
|
||||
expect(result).toEqual({ status: 200, body: "abc" });
|
||||
it("registers pairing routes", () => {
|
||||
const paths = (plugin.routes ?? []).map((route) => `${route.method} ${route.path}`);
|
||||
expect(paths).toContain("GET /status");
|
||||
expect(paths).toContain("GET /qr");
|
||||
expect(paths).toContain("POST /pair-code");
|
||||
expect(paths).toContain("POST /logout");
|
||||
});
|
||||
|
||||
it("rejects invalid signatures", async () => {
|
||||
const { ctx } = makeCtx();
|
||||
const body = JSON.stringify({ entry: [] });
|
||||
const res = await webhookPostHandler({ rawBody: Buffer.from(body), headers: { "x-hub-signature-256": "sha256=badsig" }, body: { entry: [] } }, ctx);
|
||||
expect(res.status).toBe(401);
|
||||
});
|
||||
|
||||
it("dedupes repeated inbound messages", async () => {
|
||||
const { ctx } = makeCtx();
|
||||
vi.stubGlobal("fetch", vi.fn().mockResolvedValue({ ok: true, text: vi.fn().mockResolvedValue("") }));
|
||||
|
||||
const payload = {
|
||||
entry: [{ changes: [{ value: { messages: [{ id: "wamid.1", from: "15551234567", type: "text", text: { body: "Hi" } }] } }] }],
|
||||
};
|
||||
const raw = JSON.stringify(payload);
|
||||
const req = {
|
||||
rawBody: Buffer.from(raw),
|
||||
headers: { "x-hub-signature-256": signBody("secret", raw) },
|
||||
body: payload,
|
||||
};
|
||||
|
||||
const first = await webhookPostHandler(req, ctx);
|
||||
const second = await webhookPostHandler(req, ctx);
|
||||
|
||||
expect(first.status).toBe(200);
|
||||
expect(second.status).toBe(200);
|
||||
expect(ctx.createAiSession).toHaveBeenCalledTimes(1);
|
||||
expect((global.fetch as any)).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it("preserves transcript continuity across turns", async () => {
|
||||
const { ctx } = makeCtx();
|
||||
const promptSpy = vi.fn().mockResolvedValue(undefined);
|
||||
ctx.createAiSession = vi.fn().mockResolvedValue({
|
||||
session: {
|
||||
prompt: promptSpy,
|
||||
state: { messages: [{ role: "assistant", content: "reply" }] },
|
||||
},
|
||||
});
|
||||
vi.stubGlobal("fetch", vi.fn().mockResolvedValue({ ok: true, text: vi.fn().mockResolvedValue("") }));
|
||||
|
||||
const makeReq = (id: string, text: string) => {
|
||||
const payload = { entry: [{ changes: [{ value: { messages: [{ id, from: "15551234567", type: "text", text: { body: text } }] } }] }] };
|
||||
const raw = JSON.stringify(payload);
|
||||
return { rawBody: Buffer.from(raw), headers: { "x-hub-signature-256": signBody("secret", raw) }, body: payload };
|
||||
};
|
||||
|
||||
await webhookPostHandler(makeReq("wamid.1", "First"), ctx);
|
||||
await webhookPostHandler(makeReq("wamid.2", "Second"), ctx);
|
||||
|
||||
const secondPrompt = promptSpy.mock.calls[1][0] as string;
|
||||
expect(secondPrompt).toContain("User: First");
|
||||
expect(secondPrompt).toContain("Assistant: reply");
|
||||
expect(secondPrompt).toContain("User: Second");
|
||||
});
|
||||
|
||||
it("formats outbound payloads and chunks long replies", async () => {
|
||||
const longReply = "a".repeat(5000);
|
||||
const { ctx } = makeCtx({
|
||||
createAiSession: vi.fn().mockResolvedValue({
|
||||
session: {
|
||||
prompt: vi.fn().mockResolvedValue(undefined),
|
||||
state: { messages: [{ role: "assistant", content: longReply }] },
|
||||
},
|
||||
}),
|
||||
});
|
||||
const fetchMock = vi.fn().mockResolvedValue({ ok: true, text: vi.fn().mockResolvedValue("") });
|
||||
vi.stubGlobal("fetch", fetchMock);
|
||||
|
||||
const payload = { entry: [{ changes: [{ value: { messages: [{ id: "wamid.1", from: "15551234567", type: "text", text: { body: "hi" } }] } }] }] };
|
||||
const raw = JSON.stringify(payload);
|
||||
|
||||
await webhookPostHandler({ rawBody: Buffer.from(raw), headers: { "x-hub-signature-256": signBody("secret", raw) }, body: payload }, ctx);
|
||||
|
||||
expect(fetchMock).toHaveBeenCalledTimes(2);
|
||||
const body1 = JSON.parse(fetchMock.mock.calls[0][1].body as string);
|
||||
expect(body1.messaging_product).toBe("whatsapp");
|
||||
expect(body1.to).toBe("15551234567");
|
||||
});
|
||||
|
||||
it("sends fallback error response when AI/outbound flow fails", async () => {
|
||||
const { ctx } = makeCtx({
|
||||
createAiSession: vi.fn().mockRejectedValue(new Error("boom")),
|
||||
});
|
||||
const fetchMock = vi.fn().mockResolvedValue({ ok: true, text: vi.fn().mockResolvedValue("") });
|
||||
vi.stubGlobal("fetch", fetchMock);
|
||||
|
||||
const payload = { entry: [{ changes: [{ value: { messages: [{ id: "wamid.1", from: "15551234567", type: "text", text: { body: "help" } }] } }] }] };
|
||||
const raw = JSON.stringify(payload);
|
||||
|
||||
await webhookPostHandler({ rawBody: Buffer.from(raw), headers: { "x-hub-signature-256": signBody("secret", raw) }, body: payload }, ctx);
|
||||
|
||||
const fallbackBody = JSON.parse(fetchMock.mock.calls[0][1].body as string);
|
||||
expect(fallbackBody.text.body).toContain("Sorry");
|
||||
expect(ctx.logger.error).toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("has signature helper coverage", () => {
|
||||
const body = Buffer.from("{}");
|
||||
const signature = signBody("secret", "{}");
|
||||
expect(verifyMetaSignature(body, signature, "secret")).toBe(true);
|
||||
expect(verifyMetaSignature(body, signature, "wrong")).toBe(false);
|
||||
it("uses only pairing-era settings", () => {
|
||||
const schema = plugin.manifest.settingsSchema ?? {};
|
||||
expect(Object.keys(schema).sort()).toEqual([
|
||||
"agentSystemPrompt",
|
||||
"allowedSenders",
|
||||
"historyTurnLimit",
|
||||
"pairingMode",
|
||||
"pairingPhoneNumber",
|
||||
]);
|
||||
});
|
||||
|
||||
it("splits oversized messages", () => {
|
||||
|
||||
@@ -0,0 +1,52 @@
|
||||
import { describe, expect, it, vi } from "vitest";
|
||||
import { generateReply } from "../reply.js";
|
||||
|
||||
function makeCtx(overrides: Record<string, unknown> = {}) {
|
||||
const prompt = vi.fn();
|
||||
const createAiSession = vi.fn().mockResolvedValue({
|
||||
session: {
|
||||
prompt,
|
||||
state: { messages: [{ role: "assistant", content: "hello back" }] },
|
||||
},
|
||||
});
|
||||
|
||||
return {
|
||||
settings: {},
|
||||
taskStore: { getRootDir: () => "/repo" },
|
||||
createAiSession,
|
||||
...overrides,
|
||||
} as any;
|
||||
}
|
||||
|
||||
describe("generateReply", () => {
|
||||
it("creates ai session with readonly tools and cwd root", async () => {
|
||||
const ctx = makeCtx();
|
||||
await expect(generateReply(ctx, "1555", "hi", [])).resolves.toBe("hello back");
|
||||
expect(ctx.createAiSession).toHaveBeenCalledWith(expect.objectContaining({ cwd: "/repo", tools: "readonly" }));
|
||||
});
|
||||
|
||||
it("uses system prompt override and transcript continuity", async () => {
|
||||
const prompt = vi.fn();
|
||||
const ctx = makeCtx({
|
||||
settings: { agentSystemPrompt: "custom prompt" },
|
||||
createAiSession: vi.fn().mockResolvedValue({ session: { prompt, state: { messages: [{ role: "assistant", content: "ok" }] } } }),
|
||||
});
|
||||
|
||||
await generateReply(ctx, "1555", "new", [{ role: "user", text: "old", createdAt: "t" }]);
|
||||
expect(ctx.createAiSession).toHaveBeenCalledWith(expect.objectContaining({ systemPrompt: "custom prompt" }));
|
||||
expect(prompt.mock.calls[0][0]).toContain("User: old");
|
||||
expect(prompt.mock.calls[0][0]).toContain("User: new");
|
||||
});
|
||||
|
||||
it("throws on empty assistant content", async () => {
|
||||
const ctx = makeCtx({
|
||||
createAiSession: vi.fn().mockResolvedValue({ session: { prompt: vi.fn(), state: { messages: [{ role: "assistant", content: " " }] } } }),
|
||||
});
|
||||
await expect(generateReply(ctx, "1555", "hi", [])).rejects.toThrow("no assistant text");
|
||||
});
|
||||
|
||||
it("throws when createAiSession missing", async () => {
|
||||
const ctx = makeCtx({ createAiSession: undefined });
|
||||
await expect(generateReply(ctx, "1555", "hi", [])).rejects.toThrow("AI session factory unavailable");
|
||||
});
|
||||
});
|
||||
89
plugins/fusion-plugin-whatsapp-chat/src/auth-state.ts
Normal file
89
plugins/fusion-plugin-whatsapp-chat/src/auth-state.ts
Normal file
@@ -0,0 +1,89 @@
|
||||
import { BufferJSON, initAuthCreds, type AuthenticationState, type AuthenticationCreds, type SignalDataSet, type SignalDataTypeMap } from "@whiskeysockets/baileys";
|
||||
import type { PluginDb } from "./index.js";
|
||||
|
||||
type AuthStateResult = {
|
||||
state: AuthenticationState;
|
||||
saveCreds: () => Promise<void>;
|
||||
};
|
||||
|
||||
type AuthRow = { value: string };
|
||||
|
||||
function parseStoredValue<T>(value: string): T | null {
|
||||
try {
|
||||
return JSON.parse(value, BufferJSON.reviver) as T;
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
function serialize(value: unknown): string {
|
||||
return JSON.stringify(value, BufferJSON.replacer);
|
||||
}
|
||||
|
||||
function loadCreds(db: PluginDb): AuthenticationCreds {
|
||||
const row = db.prepare("SELECT value FROM whatsapp_auth_creds WHERE id = 'creds'").get() as AuthRow | undefined;
|
||||
if (!row) return initAuthCreds();
|
||||
return parseStoredValue<AuthenticationCreds>(row.value) ?? initAuthCreds();
|
||||
}
|
||||
|
||||
export function clearAuthState(db: PluginDb): void {
|
||||
db.prepare("DELETE FROM whatsapp_auth_creds").run();
|
||||
db.prepare("DELETE FROM whatsapp_auth_keys").run();
|
||||
}
|
||||
|
||||
export function createPluginDbAuthState(db: PluginDb): AuthStateResult {
|
||||
const state: AuthenticationState = {
|
||||
creds: loadCreds(db),
|
||||
keys: {
|
||||
get: async <T extends keyof SignalDataTypeMap>(type: T, ids: string[]) => {
|
||||
const result: Record<string, SignalDataTypeMap[T]> = {};
|
||||
const select = db.prepare("SELECT value FROM whatsapp_auth_keys WHERE category = ? AND keyId = ?");
|
||||
for (const id of ids) {
|
||||
const row = select.get(type, id) as AuthRow | undefined;
|
||||
if (!row) continue;
|
||||
const parsed = parseStoredValue<SignalDataTypeMap[T]>(row.value);
|
||||
if (parsed != null) {
|
||||
result[id] = parsed;
|
||||
}
|
||||
}
|
||||
return result;
|
||||
},
|
||||
set: async (data: SignalDataSet) => {
|
||||
const upsert = db.prepare(`
|
||||
INSERT INTO whatsapp_auth_keys(category, keyId, value, updatedAt)
|
||||
VALUES(?, ?, ?, ?)
|
||||
ON CONFLICT(category, keyId)
|
||||
DO UPDATE SET value = excluded.value, updatedAt = excluded.updatedAt
|
||||
`);
|
||||
const remove = db.prepare("DELETE FROM whatsapp_auth_keys WHERE category = ? AND keyId = ?");
|
||||
const now = new Date().toISOString();
|
||||
|
||||
for (const category of Object.keys(data) as Array<keyof SignalDataSet>) {
|
||||
const categoryEntries = data[category];
|
||||
if (!categoryEntries) continue;
|
||||
for (const id of Object.keys(categoryEntries)) {
|
||||
const value = categoryEntries[id];
|
||||
if (value == null) {
|
||||
remove.run(category, id);
|
||||
continue;
|
||||
}
|
||||
upsert.run(category, id, serialize(value), now);
|
||||
}
|
||||
}
|
||||
},
|
||||
},
|
||||
};
|
||||
|
||||
return {
|
||||
state,
|
||||
saveCreds: async () => {
|
||||
const now = new Date().toISOString();
|
||||
db.prepare(`
|
||||
INSERT INTO whatsapp_auth_creds(id, value, updatedAt)
|
||||
VALUES('creds', ?, ?)
|
||||
ON CONFLICT(id)
|
||||
DO UPDATE SET value = excluded.value, updatedAt = excluded.updatedAt
|
||||
`).run(serialize(state.creds), now);
|
||||
},
|
||||
};
|
||||
}
|
||||
243
plugins/fusion-plugin-whatsapp-chat/src/connection.ts
Normal file
243
plugins/fusion-plugin-whatsapp-chat/src/connection.ts
Normal file
@@ -0,0 +1,243 @@
|
||||
import { DisconnectReason, makeWASocket, type ConnectionState, type WAMessage, type WAMessageContent, type WASocket } from "@whiskeysockets/baileys";
|
||||
import type { PluginContext } from "@fusion/plugin-sdk";
|
||||
import pino from "pino";
|
||||
import qrcode from "qrcode";
|
||||
import { clearAuthState, createPluginDbAuthState } from "./auth-state.js";
|
||||
import { getAllowedSenders, getHistoryTurnLimit, loadHistory, markProcessed, saveHistory, wasProcessed, type ChatTurn, type PluginDb } from "./index.js";
|
||||
|
||||
export type ConnectionStatus = {
|
||||
state: "starting" | "awaiting-qr" | "awaiting-code" | "connected" | "disconnected" | "error";
|
||||
qr?: string;
|
||||
qrDataUrl?: string;
|
||||
pairingCode?: string;
|
||||
lastError?: string;
|
||||
jid?: string;
|
||||
};
|
||||
|
||||
export type ReplyGenerator = (ctx: PluginContext, sender: string, text: string, history: ChatTurn[]) => Promise<string>;
|
||||
|
||||
const BACKOFF_MS = [1000, 2000, 5000, 15000, 30000];
|
||||
const FALLBACK_TEXT = "Sorry, I hit an internal error while processing that message.";
|
||||
const MAX_WHATSAPP_MESSAGE_CHARS = 4096;
|
||||
|
||||
function extractText(message?: WAMessageContent | null): string | null {
|
||||
const text = message?.conversation ?? message?.extendedTextMessage?.text;
|
||||
if (!text || !text.trim()) return null;
|
||||
return text.trim();
|
||||
}
|
||||
|
||||
function normalizeSender(jid: string): string {
|
||||
return jid.split("@")[0]?.replace(/\D+/g, "") ?? "";
|
||||
}
|
||||
|
||||
function isLoggedOutDisconnect(error: unknown): boolean {
|
||||
const statusCode = (error as { output?: { statusCode?: unknown } })?.output?.statusCode;
|
||||
return statusCode === DisconnectReason.loggedOut;
|
||||
}
|
||||
|
||||
function splitMessageForWhatsapp(text: string): string[] {
|
||||
if (text.length <= MAX_WHATSAPP_MESSAGE_CHARS) return [text];
|
||||
|
||||
const chunks: string[] = [];
|
||||
let remaining = text;
|
||||
while (remaining.length > 0) {
|
||||
if (remaining.length <= MAX_WHATSAPP_MESSAGE_CHARS) {
|
||||
chunks.push(remaining);
|
||||
break;
|
||||
}
|
||||
const candidate = remaining.slice(0, MAX_WHATSAPP_MESSAGE_CHARS);
|
||||
const splitAt = Math.max(candidate.lastIndexOf("\n"), candidate.lastIndexOf(" "));
|
||||
const breakpoint = splitAt > 0 ? splitAt : MAX_WHATSAPP_MESSAGE_CHARS;
|
||||
chunks.push(remaining.slice(0, breakpoint).trim());
|
||||
remaining = remaining.slice(breakpoint).trimStart();
|
||||
}
|
||||
|
||||
return chunks.filter(Boolean);
|
||||
}
|
||||
|
||||
export class WhatsAppConnection {
|
||||
private sock: WASocket | null = null;
|
||||
private status: ConnectionStatus = { state: "disconnected" };
|
||||
private reconnectTimer: ReturnType<typeof setTimeout> | null = null;
|
||||
private reconnectAttempt = 0;
|
||||
private stopped = true;
|
||||
private authState: ReturnType<typeof createPluginDbAuthState>;
|
||||
|
||||
public constructor(
|
||||
private readonly ctx: PluginContext,
|
||||
private readonly fusionVersion: string,
|
||||
private readonly generateReply: ReplyGenerator,
|
||||
private readonly db: PluginDb,
|
||||
) {
|
||||
this.authState = createPluginDbAuthState(this.db);
|
||||
}
|
||||
|
||||
public async start(): Promise<void> {
|
||||
this.stopped = false;
|
||||
this.status = { state: "starting" };
|
||||
await this.connect();
|
||||
}
|
||||
|
||||
public async stop(): Promise<void> {
|
||||
if (this.stopped) return;
|
||||
this.stopped = true;
|
||||
this.clearReconnectTimer();
|
||||
|
||||
const socket = this.sock;
|
||||
this.sock = null;
|
||||
this.status = { state: "disconnected" };
|
||||
|
||||
if (socket) {
|
||||
socket.ev.off("creds.update", this.authState.saveCreds);
|
||||
socket.ev.off("connection.update", this.onConnectionUpdate);
|
||||
socket.ev.off("messages.upsert", this.onMessagesUpsert);
|
||||
await socket.end(undefined);
|
||||
}
|
||||
}
|
||||
|
||||
public getStatus(): ConnectionStatus {
|
||||
return { ...this.status };
|
||||
}
|
||||
|
||||
public async requestPairingCode(phoneNumberE164: string): Promise<string> {
|
||||
if (!this.sock) throw new Error("WhatsApp socket not initialized");
|
||||
const pairingCode = await this.sock.requestPairingCode(phoneNumberE164);
|
||||
this.status = { ...this.status, state: "awaiting-code", pairingCode };
|
||||
return pairingCode;
|
||||
}
|
||||
|
||||
public async logout(): Promise<void> {
|
||||
try {
|
||||
await this.sock?.logout();
|
||||
} finally {
|
||||
clearAuthState(this.db);
|
||||
this.authState = createPluginDbAuthState(this.db);
|
||||
this.status = { state: "disconnected" };
|
||||
}
|
||||
}
|
||||
|
||||
private async connect(): Promise<void> {
|
||||
if (this.stopped) return;
|
||||
|
||||
this.status = { state: "starting" };
|
||||
const socket = makeWASocket({
|
||||
auth: this.authState.state,
|
||||
printQRInTerminal: false,
|
||||
browser: ["Fusion", "Chrome", this.fusionVersion],
|
||||
logger: pino({ level: "silent" }),
|
||||
});
|
||||
|
||||
this.sock = socket;
|
||||
socket.ev.on("creds.update", this.authState.saveCreds);
|
||||
socket.ev.on("connection.update", this.onConnectionUpdate);
|
||||
socket.ev.on("messages.upsert", this.onMessagesUpsert);
|
||||
}
|
||||
|
||||
private readonly onConnectionUpdate = async (update: Partial<ConnectionState>): Promise<void> => {
|
||||
if (update.qr) {
|
||||
const qrDataUrl = await qrcode.toDataURL(update.qr);
|
||||
this.ctx.logger.info("WhatsApp pairing QR updated", update.qr);
|
||||
this.status = { state: "awaiting-qr", qr: update.qr, qrDataUrl };
|
||||
}
|
||||
|
||||
if (update.connection === "open") {
|
||||
this.reconnectAttempt = 0;
|
||||
this.status = { state: "connected", jid: this.sock?.user?.id };
|
||||
return;
|
||||
}
|
||||
|
||||
if (update.connection === "close") {
|
||||
if (isLoggedOutDisconnect(update.lastDisconnect?.error)) {
|
||||
clearAuthState(this.db);
|
||||
this.authState = createPluginDbAuthState(this.db);
|
||||
this.status = { state: "disconnected", lastError: "loggedOut" };
|
||||
return;
|
||||
}
|
||||
|
||||
const closeError = update.lastDisconnect?.error;
|
||||
this.status = {
|
||||
state: "disconnected",
|
||||
lastError: closeError instanceof Error ? closeError.message : "connection closed",
|
||||
};
|
||||
this.scheduleReconnect();
|
||||
}
|
||||
};
|
||||
|
||||
private readonly onMessagesUpsert = async (upsert: { type?: string; messages?: WAMessage[] }): Promise<void> => {
|
||||
if (upsert.type !== "notify") return;
|
||||
|
||||
for (const message of upsert.messages ?? []) {
|
||||
const jid = message.key.remoteJid;
|
||||
const messageId = message.key.id;
|
||||
if (!jid || !messageId) continue;
|
||||
if (jid.endsWith("@g.us") || jid.endsWith("@broadcast") || jid === "status@broadcast") continue;
|
||||
if (message.key.fromMe) continue;
|
||||
|
||||
const text = extractText(message.message);
|
||||
if (!text) continue;
|
||||
|
||||
const sender = normalizeSender(jid);
|
||||
const allowedSenders = getAllowedSenders(this.ctx.settings);
|
||||
if (allowedSenders.size === 0 || (!allowedSenders.has(sender) && !allowedSenders.has(jid))) continue;
|
||||
if (wasProcessed(this.db, messageId)) continue;
|
||||
|
||||
markProcessed(this.db, messageId, sender);
|
||||
|
||||
try {
|
||||
const history = loadHistory(this.db, sender);
|
||||
const reply = await this.generateReply(this.ctx, sender, text, history);
|
||||
const now = new Date().toISOString();
|
||||
const nextHistory: ChatTurn[] = [
|
||||
...history,
|
||||
{ role: "user" as const, text, createdAt: now },
|
||||
{ role: "assistant" as const, text: reply, createdAt: now },
|
||||
].slice(-getHistoryTurnLimit(this.ctx.settings));
|
||||
saveHistory(this.db, sender, nextHistory);
|
||||
|
||||
for (const chunk of splitMessageForWhatsapp(reply)) {
|
||||
await this.sock?.sendMessage(jid, { text: chunk });
|
||||
}
|
||||
} catch (error) {
|
||||
this.ctx.logger.error("WhatsApp chat processing failed", error);
|
||||
try {
|
||||
await this.sock?.sendMessage(jid, { text: FALLBACK_TEXT });
|
||||
} catch {
|
||||
// no-op
|
||||
}
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
private scheduleReconnect(): void {
|
||||
if (this.stopped || this.reconnectTimer) return;
|
||||
const delay = BACKOFF_MS[Math.min(this.reconnectAttempt, BACKOFF_MS.length - 1)] ?? 30000;
|
||||
this.reconnectAttempt += 1;
|
||||
|
||||
this.reconnectTimer = setTimeout(async () => {
|
||||
this.reconnectTimer = null;
|
||||
await this.connect();
|
||||
}, delay);
|
||||
}
|
||||
|
||||
private clearReconnectTimer(): void {
|
||||
if (!this.reconnectTimer) return;
|
||||
clearTimeout(this.reconnectTimer);
|
||||
this.reconnectTimer = null;
|
||||
}
|
||||
|
||||
public static splitMessageForWhatsapp(text: string, max = 4096): string[] {
|
||||
const chunks = splitMessageForWhatsapp(text);
|
||||
if (max === 4096) return chunks;
|
||||
return chunks.flatMap((chunk) => {
|
||||
if (chunk.length <= max) return [chunk];
|
||||
const split: string[] = [];
|
||||
let remaining = chunk;
|
||||
while (remaining.length > max) {
|
||||
split.push(remaining.slice(0, max));
|
||||
remaining = remaining.slice(max);
|
||||
}
|
||||
if (remaining.length) split.push(remaining);
|
||||
return split;
|
||||
});
|
||||
}
|
||||
}
|
||||
@@ -1,22 +1,13 @@
|
||||
import { createHmac, timingSafeEqual } from "node:crypto";
|
||||
import { definePlugin } from "@fusion/plugin-sdk";
|
||||
import type { FusionPlugin, PluginContext, PluginRouteDefinition, PluginRouteResponse, PluginSettingSchema } from "@fusion/plugin-sdk";
|
||||
import { WhatsAppConnection } from "./connection.js";
|
||||
import { generateReply } from "./reply.js";
|
||||
|
||||
const MAX_WHATSAPP_MESSAGE_CHARS = 4096;
|
||||
const DEFAULT_HISTORY_TURN_LIMIT = 40;
|
||||
|
||||
const settingsSchema: Record<string, PluginSettingSchema> = {
|
||||
verifyToken: { type: "password", label: "Verify Token", required: true },
|
||||
appSecret: { type: "password", label: "App Secret", required: true },
|
||||
accessToken: { type: "password", label: "Access Token", required: true },
|
||||
phoneNumberId: { type: "string", label: "Phone Number ID", required: true },
|
||||
graphApiVersion: { type: "string", label: "Graph API Version", defaultValue: "v21.0" },
|
||||
allowedSenders: { type: "array", label: "Allowed WhatsApp Senders", itemType: "string" },
|
||||
agentSystemPrompt: { type: "string", label: "Agent System Prompt", multiline: true, defaultValue: "You are a helpful assistant replying in WhatsApp chats." },
|
||||
};
|
||||
export type ChatTurn = { role: "user" | "assistant"; text: string; createdAt: string };
|
||||
|
||||
type ChatTurn = { role: "user" | "assistant"; text: string; createdAt: string };
|
||||
|
||||
type PluginDb = {
|
||||
export type PluginDb = {
|
||||
exec(sql: string): void;
|
||||
prepare(sql: string): {
|
||||
get(...args: unknown[]): unknown;
|
||||
@@ -24,81 +15,58 @@ type PluginDb = {
|
||||
};
|
||||
};
|
||||
|
||||
type IncomingMessage = { id: string; from: string; text: string };
|
||||
|
||||
type PluginRequest = {
|
||||
headers?: Record<string, string | string[] | undefined>;
|
||||
query?: Record<string, string | string[] | undefined>;
|
||||
body?: unknown;
|
||||
rawBody?: Buffer;
|
||||
const settingsSchema: Record<string, PluginSettingSchema> = {
|
||||
pairingMode: {
|
||||
type: "enum",
|
||||
label: "Pairing Mode",
|
||||
enumValues: ["qr", "code"],
|
||||
defaultValue: "qr",
|
||||
},
|
||||
pairingPhoneNumber: {
|
||||
type: "string",
|
||||
label: "Pairing Phone Number",
|
||||
description: "E.164 digits without + (required when pairingMode is code)",
|
||||
},
|
||||
allowedSenders: { type: "array", label: "Allowed WhatsApp Senders", itemType: "string" },
|
||||
agentSystemPrompt: {
|
||||
type: "string",
|
||||
label: "Agent System Prompt",
|
||||
multiline: true,
|
||||
defaultValue: "You are a helpful assistant replying in WhatsApp chats.",
|
||||
},
|
||||
historyTurnLimit: {
|
||||
type: "number",
|
||||
label: "History Turn Limit",
|
||||
defaultValue: DEFAULT_HISTORY_TURN_LIMIT,
|
||||
},
|
||||
};
|
||||
|
||||
function getSettingString(settings: Record<string, unknown>, key: string): string | undefined {
|
||||
const connections = new Map<string, WhatsAppConnection>();
|
||||
|
||||
export function getSettingString(settings: Record<string, unknown>, key: string): string | undefined {
|
||||
const value = settings[key];
|
||||
return typeof value === "string" && value.trim() ? value.trim() : undefined;
|
||||
}
|
||||
|
||||
function getAllowedSenders(settings: Record<string, unknown>): Set<string> {
|
||||
export function getAllowedSenders(settings: Record<string, unknown>): Set<string> {
|
||||
const senders = settings.allowedSenders;
|
||||
if (!Array.isArray(senders)) return new Set<string>();
|
||||
return new Set(senders.filter((entry): entry is string => typeof entry === "string" && entry.trim().length > 0).map((entry) => entry.trim()));
|
||||
}
|
||||
|
||||
function splitMessageForWhatsapp(text: string): string[] {
|
||||
if (text.length <= MAX_WHATSAPP_MESSAGE_CHARS) {
|
||||
return [text];
|
||||
export function getHistoryTurnLimit(settings: Record<string, unknown>): number {
|
||||
const value = settings.historyTurnLimit;
|
||||
if (typeof value !== "number" || !Number.isFinite(value) || value <= 0) {
|
||||
return DEFAULT_HISTORY_TURN_LIMIT;
|
||||
}
|
||||
|
||||
const chunks: string[] = [];
|
||||
let remaining = text;
|
||||
while (remaining.length > 0) {
|
||||
if (remaining.length <= MAX_WHATSAPP_MESSAGE_CHARS) {
|
||||
chunks.push(remaining);
|
||||
break;
|
||||
}
|
||||
const candidate = remaining.slice(0, MAX_WHATSAPP_MESSAGE_CHARS);
|
||||
const splitAt = Math.max(candidate.lastIndexOf("\n"), candidate.lastIndexOf(" "));
|
||||
const breakpoint = splitAt > 0 ? splitAt : MAX_WHATSAPP_MESSAGE_CHARS;
|
||||
chunks.push(remaining.slice(0, breakpoint).trim());
|
||||
remaining = remaining.slice(breakpoint).trimStart();
|
||||
}
|
||||
return chunks.filter(Boolean);
|
||||
return Math.floor(value);
|
||||
}
|
||||
|
||||
function verifyMetaSignature(rawBody: Buffer, signatureHeader: string | undefined, appSecret: string): boolean {
|
||||
if (!signatureHeader?.startsWith("sha256=")) {
|
||||
return false;
|
||||
}
|
||||
|
||||
const expected = createHmac("sha256", appSecret).update(rawBody).digest("hex");
|
||||
const provided = signatureHeader.slice("sha256=".length);
|
||||
const expectedBuf = Buffer.from(expected, "hex");
|
||||
const providedBuf = Buffer.from(provided, "hex");
|
||||
if (expectedBuf.length !== providedBuf.length) {
|
||||
return false;
|
||||
}
|
||||
return timingSafeEqual(expectedBuf, providedBuf);
|
||||
export function splitMessageForWhatsapp(text: string): string[] {
|
||||
return WhatsAppConnection.splitMessageForWhatsapp(text);
|
||||
}
|
||||
|
||||
function extractIncomingMessages(payload: unknown): IncomingMessage[] {
|
||||
const body = payload as {
|
||||
entry?: Array<{ changes?: Array<{ value?: { messages?: Array<{ id?: string; from?: string; text?: { body?: string }; type?: string }> } }> }>;
|
||||
};
|
||||
|
||||
const messages: IncomingMessage[] = [];
|
||||
for (const entry of body.entry ?? []) {
|
||||
for (const change of entry.changes ?? []) {
|
||||
for (const message of change.value?.messages ?? []) {
|
||||
if (message.type !== "text") continue;
|
||||
if (!message.id || !message.from || !message.text?.body?.trim()) continue;
|
||||
messages.push({ id: message.id, from: message.from, text: message.text.body.trim() });
|
||||
}
|
||||
}
|
||||
}
|
||||
return messages;
|
||||
}
|
||||
|
||||
function ensureSchema(db: PluginDb): void {
|
||||
export function ensureSchema(db: PluginDb): void {
|
||||
db.exec(`
|
||||
CREATE TABLE IF NOT EXISTS whatsapp_chat_sessions (
|
||||
sender TEXT PRIMARY KEY,
|
||||
@@ -111,10 +79,24 @@ function ensureSchema(db: PluginDb): void {
|
||||
sender TEXT NOT NULL,
|
||||
receivedAt TEXT NOT NULL
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS whatsapp_auth_creds (
|
||||
id TEXT PRIMARY KEY,
|
||||
value TEXT NOT NULL,
|
||||
updatedAt TEXT NOT NULL
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS whatsapp_auth_keys (
|
||||
category TEXT NOT NULL,
|
||||
keyId TEXT NOT NULL,
|
||||
value TEXT NOT NULL,
|
||||
updatedAt TEXT NOT NULL,
|
||||
PRIMARY KEY (category, keyId)
|
||||
);
|
||||
`);
|
||||
}
|
||||
|
||||
function loadHistory(db: PluginDb, sender: string): ChatTurn[] {
|
||||
export function loadHistory(db: PluginDb, sender: string): ChatTurn[] {
|
||||
const row = db.prepare("SELECT history FROM whatsapp_chat_sessions WHERE sender = ?").get(sender) as { history: string } | undefined;
|
||||
if (!row) return [];
|
||||
try {
|
||||
@@ -125,7 +107,7 @@ function loadHistory(db: PluginDb, sender: string): ChatTurn[] {
|
||||
}
|
||||
}
|
||||
|
||||
function saveHistory(db: PluginDb, sender: string, history: ChatTurn[]): void {
|
||||
export function saveHistory(db: PluginDb, sender: string, history: ChatTurn[]): void {
|
||||
const now = new Date().toISOString();
|
||||
db.prepare(`
|
||||
INSERT INTO whatsapp_chat_sessions(sender, history, updatedAt)
|
||||
@@ -134,80 +116,17 @@ function saveHistory(db: PluginDb, sender: string, history: ChatTurn[]): void {
|
||||
`).run(sender, JSON.stringify(history), now);
|
||||
}
|
||||
|
||||
function wasProcessed(db: PluginDb, messageId: string): boolean {
|
||||
export function wasProcessed(db: PluginDb, messageId: string): boolean {
|
||||
const row = db.prepare("SELECT 1 as found FROM whatsapp_chat_dedupe WHERE messageId = ?").get(messageId) as { found: number } | undefined;
|
||||
return Boolean(row?.found);
|
||||
}
|
||||
|
||||
function markProcessed(db: PluginDb, messageId: string, sender: string): void {
|
||||
export function markProcessed(db: PluginDb, messageId: string, sender: string): void {
|
||||
db.prepare("INSERT INTO whatsapp_chat_dedupe(messageId, sender, receivedAt) VALUES(?, ?, ?)").run(messageId, sender, new Date().toISOString());
|
||||
}
|
||||
|
||||
async function sendWhatsappText(ctx: PluginContext, to: string, text: string): Promise<void> {
|
||||
const accessToken = getSettingString(ctx.settings, "accessToken");
|
||||
const phoneNumberId = getSettingString(ctx.settings, "phoneNumberId");
|
||||
const graphApiVersion = getSettingString(ctx.settings, "graphApiVersion") ?? "v21.0";
|
||||
if (!accessToken || !phoneNumberId) {
|
||||
throw new Error("WhatsApp plugin missing accessToken or phoneNumberId settings");
|
||||
}
|
||||
|
||||
for (const chunk of splitMessageForWhatsapp(text)) {
|
||||
const response = await fetch(`https://graph.facebook.com/${graphApiVersion}/${phoneNumberId}/messages`, {
|
||||
method: "POST",
|
||||
headers: {
|
||||
Authorization: `Bearer ${accessToken}`,
|
||||
"Content-Type": "application/json",
|
||||
},
|
||||
body: JSON.stringify({
|
||||
messaging_product: "whatsapp",
|
||||
to,
|
||||
type: "text",
|
||||
text: { body: chunk },
|
||||
}),
|
||||
});
|
||||
|
||||
if (!response.ok) {
|
||||
const errorText = await response.text();
|
||||
throw new Error(`WhatsApp Graph API request failed (${response.status}): ${errorText}`);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async function generateReply(ctx: PluginContext, sender: string, text: string, history: ChatTurn[]): Promise<string> {
|
||||
if (!ctx.createAiSession) {
|
||||
throw new Error("AI session factory unavailable: engine not registered");
|
||||
}
|
||||
|
||||
const systemPrompt = getSettingString(ctx.settings, "agentSystemPrompt") ?? "You are a helpful assistant replying in WhatsApp chats.";
|
||||
const sessionResult = await ctx.createAiSession({
|
||||
cwd: ctx.taskStore.getRootDir(),
|
||||
systemPrompt,
|
||||
tools: "readonly",
|
||||
});
|
||||
|
||||
const promptLines = [
|
||||
"Continue this WhatsApp conversation.",
|
||||
...history.map((turn) => `${turn.role === "user" ? "User" : "Assistant"}: ${turn.text}`),
|
||||
`User: ${text}`,
|
||||
"Assistant:",
|
||||
];
|
||||
|
||||
await sessionResult.session.prompt(promptLines.join("\n"));
|
||||
const assistantMessages = sessionResult.session.state.messages.filter((message) => message.role === "assistant");
|
||||
const latest = assistantMessages[assistantMessages.length - 1];
|
||||
const content = latest?.content;
|
||||
if (typeof content === "string" && content.trim()) return content.trim();
|
||||
if (Array.isArray(content)) {
|
||||
const textParts = content
|
||||
.map((part) => (part && typeof part === "object" && "text" in part && typeof (part as { text?: unknown }).text === "string") ? (part as { text: string }).text : "")
|
||||
.filter(Boolean);
|
||||
if (textParts.length > 0) return textParts.join("\n").trim();
|
||||
}
|
||||
|
||||
throw new Error("AI session returned no assistant text");
|
||||
}
|
||||
|
||||
async function getDbFromTaskStore(ctx: PluginContext): Promise<PluginDb> {
|
||||
function getDbFromTaskStore(ctx: PluginContext): PluginDb {
|
||||
const pluginStore = ctx.taskStore.getPluginStore();
|
||||
const db = (pluginStore as unknown as { db?: PluginDb }).db;
|
||||
if (!db) {
|
||||
@@ -216,67 +135,70 @@ async function getDbFromTaskStore(ctx: PluginContext): Promise<PluginDb> {
|
||||
return db;
|
||||
}
|
||||
|
||||
async function webhookGetHandler(req: PluginRequest, ctx: PluginContext): Promise<PluginRouteResponse> {
|
||||
const verifyToken = getSettingString(ctx.settings, "verifyToken");
|
||||
const mode = req.query?.["hub.mode"];
|
||||
const challenge = req.query?.["hub.challenge"];
|
||||
const token = req.query?.["hub.verify_token"];
|
||||
|
||||
if (mode === "subscribe" && verifyToken && token === verifyToken && typeof challenge === "string") {
|
||||
return { status: 200, body: challenge };
|
||||
function getConnectionOrResponse(ctx: PluginContext): { connection?: WhatsAppConnection; error?: PluginRouteResponse } {
|
||||
const connection = connections.get(ctx.pluginId);
|
||||
if (!connection) {
|
||||
return { error: { status: 503, body: { error: "WhatsApp connection is not initialized" } } };
|
||||
}
|
||||
|
||||
return { status: 403, body: { error: "Verification failed" } };
|
||||
}
|
||||
|
||||
async function webhookPostHandler(req: PluginRequest, ctx: PluginContext): Promise<PluginRouteResponse> {
|
||||
const appSecret = getSettingString(ctx.settings, "appSecret");
|
||||
if (!appSecret) {
|
||||
return { status: 500, body: { error: "appSecret is not configured" } };
|
||||
}
|
||||
|
||||
const signatureHeader = (req.headers?.["x-hub-signature-256"] ?? req.headers?.["X-Hub-Signature-256"]) as string | undefined;
|
||||
if (!req.rawBody || !verifyMetaSignature(req.rawBody, signatureHeader, appSecret)) {
|
||||
return { status: 401, body: { error: "Invalid webhook signature" } };
|
||||
}
|
||||
|
||||
const db = await getDbFromTaskStore(ctx);
|
||||
const allowedSenders = getAllowedSenders(ctx.settings);
|
||||
const inboundMessages = extractIncomingMessages(req.body);
|
||||
|
||||
for (const inbound of inboundMessages) {
|
||||
if (wasProcessed(db, inbound.id)) {
|
||||
continue;
|
||||
}
|
||||
markProcessed(db, inbound.id, inbound.from);
|
||||
|
||||
if (allowedSenders.size > 0 && !allowedSenders.has(inbound.from)) {
|
||||
continue;
|
||||
}
|
||||
|
||||
try {
|
||||
const history = loadHistory(db, inbound.from);
|
||||
const reply = await generateReply(ctx, inbound.from, inbound.text, history);
|
||||
const now = new Date().toISOString();
|
||||
const nextHistory: ChatTurn[] = [
|
||||
...history,
|
||||
{ role: "user" as const, text: inbound.text, createdAt: now },
|
||||
{ role: "assistant" as const, text: reply, createdAt: now },
|
||||
].slice(-40);
|
||||
saveHistory(db, inbound.from, nextHistory);
|
||||
await sendWhatsappText(ctx, inbound.from, reply);
|
||||
} catch (error) {
|
||||
ctx.logger.error("WhatsApp chat processing failed", error);
|
||||
await sendWhatsappText(ctx, inbound.from, "Sorry, I hit an internal error while processing that message.");
|
||||
}
|
||||
}
|
||||
|
||||
return { status: 200, body: { processed: inboundMessages.length } };
|
||||
return { connection };
|
||||
}
|
||||
|
||||
const routes: PluginRouteDefinition[] = [
|
||||
{ method: "GET", path: "/webhook", handler: webhookGetHandler as unknown as PluginRouteDefinition["handler"] },
|
||||
{ method: "POST", path: "/webhook", handler: webhookPostHandler as unknown as PluginRouteDefinition["handler"] },
|
||||
{
|
||||
method: "GET",
|
||||
path: "/status",
|
||||
handler: async (_req, ctx) => {
|
||||
const { connection, error } = getConnectionOrResponse(ctx);
|
||||
if (!connection) return error as PluginRouteResponse;
|
||||
const status = connection.getStatus();
|
||||
return {
|
||||
status: 200,
|
||||
body: {
|
||||
status: status.state,
|
||||
jid: status.jid,
|
||||
allowedSenders: Array.from(getAllowedSenders(ctx.settings)),
|
||||
},
|
||||
};
|
||||
},
|
||||
},
|
||||
{
|
||||
method: "GET",
|
||||
path: "/qr",
|
||||
handler: async (_req, ctx) => {
|
||||
const { connection, error } = getConnectionOrResponse(ctx);
|
||||
if (!connection) return error as PluginRouteResponse;
|
||||
const status = connection.getStatus();
|
||||
if (status.state !== "awaiting-qr" || !status.qrDataUrl || !status.qr) {
|
||||
return { status: 409, body: { error: "QR is not currently available" } };
|
||||
}
|
||||
return { status: 200, body: { qrDataUrl: status.qrDataUrl, qr: status.qr } };
|
||||
},
|
||||
},
|
||||
{
|
||||
method: "POST",
|
||||
path: "/pair-code",
|
||||
handler: async (req, ctx) => {
|
||||
const { connection, error } = getConnectionOrResponse(ctx);
|
||||
if (!connection) return error as PluginRouteResponse;
|
||||
const body = (req as { body?: { phoneNumber?: unknown } })?.body;
|
||||
const phoneNumber = typeof body?.phoneNumber === "string" ? body.phoneNumber.trim() : "";
|
||||
if (!phoneNumber) {
|
||||
return { status: 400, body: { error: "phoneNumber is required" } };
|
||||
}
|
||||
const pairingCode = await connection.requestPairingCode(phoneNumber);
|
||||
return { status: 200, body: { pairingCode } };
|
||||
},
|
||||
},
|
||||
{
|
||||
method: "POST",
|
||||
path: "/logout",
|
||||
handler: async (_req, ctx) => {
|
||||
const { connection, error } = getConnectionOrResponse(ctx);
|
||||
if (!connection) return error as PluginRouteResponse;
|
||||
await connection.logout();
|
||||
return { status: 200, body: { ok: true } };
|
||||
},
|
||||
},
|
||||
];
|
||||
|
||||
const plugin: FusionPlugin = definePlugin({
|
||||
@@ -284,7 +206,7 @@ const plugin: FusionPlugin = definePlugin({
|
||||
id: "fusion-plugin-whatsapp-chat",
|
||||
name: "WhatsApp Chat",
|
||||
version: "0.1.0",
|
||||
description: "Bridge WhatsApp Cloud webhook messages to a Fusion agent conversation",
|
||||
description: "WhatsApp Web (multi-device) bridge that pairs via QR/code and forwards messages to Fusion AI",
|
||||
author: "Fusion Team",
|
||||
settingsSchema,
|
||||
},
|
||||
@@ -292,17 +214,21 @@ const plugin: FusionPlugin = definePlugin({
|
||||
routes,
|
||||
hooks: {
|
||||
onSchemaInit: (db) => {
|
||||
ensureSchema(db);
|
||||
ensureSchema(db as PluginDb);
|
||||
},
|
||||
onLoad: async (ctx) => {
|
||||
const db = getDbFromTaskStore(ctx);
|
||||
const connection = new WhatsAppConnection(ctx, plugin.manifest.version, generateReply, db);
|
||||
connections.set(ctx.pluginId, connection);
|
||||
await connection.start();
|
||||
},
|
||||
onUnload: async () => {
|
||||
for (const [pluginId, connection] of connections.entries()) {
|
||||
await connection.stop();
|
||||
connections.delete(pluginId);
|
||||
}
|
||||
},
|
||||
},
|
||||
});
|
||||
|
||||
export default plugin;
|
||||
export {
|
||||
extractIncomingMessages,
|
||||
generateReply,
|
||||
splitMessageForWhatsapp,
|
||||
verifyMetaSignature,
|
||||
webhookGetHandler,
|
||||
webhookPostHandler,
|
||||
};
|
||||
|
||||
1
plugins/fusion-plugin-whatsapp-chat/src/qrcode.d.ts
vendored
Normal file
1
plugins/fusion-plugin-whatsapp-chat/src/qrcode.d.ts
vendored
Normal file
@@ -0,0 +1 @@
|
||||
declare module "qrcode";
|
||||
37
plugins/fusion-plugin-whatsapp-chat/src/reply.ts
Normal file
37
plugins/fusion-plugin-whatsapp-chat/src/reply.ts
Normal file
@@ -0,0 +1,37 @@
|
||||
import type { PluginContext } from "@fusion/plugin-sdk";
|
||||
import { getSettingString, type ChatTurn } from "./index.js";
|
||||
|
||||
export async function generateReply(ctx: PluginContext, _sender: string, text: string, history: ChatTurn[]): Promise<string> {
|
||||
if (!ctx.createAiSession) {
|
||||
throw new Error("AI session factory unavailable: engine not registered");
|
||||
}
|
||||
|
||||
const systemPrompt = getSettingString(ctx.settings, "agentSystemPrompt") ?? "You are a helpful assistant replying in WhatsApp chats.";
|
||||
const sessionResult = await ctx.createAiSession({
|
||||
cwd: ctx.taskStore.getRootDir(),
|
||||
systemPrompt,
|
||||
tools: "readonly",
|
||||
});
|
||||
|
||||
const promptLines = [
|
||||
"Continue this WhatsApp conversation.",
|
||||
...history.map((turn) => `${turn.role === "user" ? "User" : "Assistant"}: ${turn.text}`),
|
||||
`User: ${text}`,
|
||||
"Assistant:",
|
||||
];
|
||||
|
||||
await sessionResult.session.prompt(promptLines.join("\n"));
|
||||
const assistantMessages = sessionResult.session.state.messages.filter((message) => message.role === "assistant");
|
||||
const latest = assistantMessages[assistantMessages.length - 1];
|
||||
const content = latest?.content;
|
||||
|
||||
if (typeof content === "string" && content.trim()) return content.trim();
|
||||
if (Array.isArray(content)) {
|
||||
const textParts = content
|
||||
.map((part) => (part && typeof part === "object" && "text" in part && typeof (part as { text?: unknown }).text === "string") ? (part as { text: string }).text : "")
|
||||
.filter(Boolean);
|
||||
if (textParts.length > 0) return textParts.join("\n").trim();
|
||||
}
|
||||
|
||||
throw new Error("AI session returned no assistant text");
|
||||
}
|
||||
@@ -1,6 +1,12 @@
|
||||
import { fileURLToPath } from "node:url";
|
||||
import { defineConfig } from "vitest/config";
|
||||
|
||||
export default defineConfig({
|
||||
resolve: {
|
||||
alias: {
|
||||
"@fusion/plugin-sdk": fileURLToPath(new URL("../../packages/plugin-sdk/src/index.ts", import.meta.url)),
|
||||
},
|
||||
},
|
||||
test: {
|
||||
environment: "node",
|
||||
include: ["src/__tests__/**/*.test.ts"],
|
||||
|
||||
Reference in New Issue
Block a user