From 84cf64770fc6470ad07d70eaeef2143e24e1057d Mon Sep 17 00:00:00 2001 From: Peter Lee Date: Sun, 21 Jun 2026 09:32:49 -0500 Subject: [PATCH] fix(whatsapp): wire missing Baileys retry/cache hooks for group message reliability (#94338) Merged via squash. Prepared head SHA: ee6de071f728d2a56e3b186fe2ac2fa86699d787 Co-authored-by: xialonglee <22994703+xialonglee@users.noreply.github.com> Co-authored-by: mcaxtr <7562095+mcaxtr@users.noreply.github.com> Reviewed-by: @mcaxtr --- extensions/whatsapp/src/auto-reply/monitor.ts | 21 +- .../whatsapp/src/connection-controller.ts | 6 +- extensions/whatsapp/src/inbound/monitor.ts | 208 ++++++++++++- ...x.streams-inbound-messages.test-support.ts | 275 +++++++++++++++++- extensions/whatsapp/src/session.ts | 5 + 5 files changed, 494 insertions(+), 21 deletions(-) diff --git a/extensions/whatsapp/src/auto-reply/monitor.ts b/extensions/whatsapp/src/auto-reply/monitor.ts index 668d78195473..8a50c6a561ea 100644 --- a/extensions/whatsapp/src/auto-reply/monitor.ts +++ b/extensions/whatsapp/src/auto-reply/monitor.ts @@ -1,4 +1,5 @@ // Whatsapp plugin module implements monitor behavior. +import type { WAMessageKey } from "baileys"; import { resolveAccountEntry } from "openclaw/plugin-sdk/account-core"; import { CHANNEL_APPROVAL_NATIVE_RUNTIME_CONTEXT_CAPABILITY } from "openclaw/plugin-sdk/approval-handler-runtime"; import { resolveInboundDebounceMs } from "openclaw/plugin-sdk/channel-inbound-debounce"; @@ -27,7 +28,13 @@ import { } from "../connection-controller.js"; import { resolveWhatsAppInboundPolicy } from "../inbound-policy.js"; import { normalizeWebInboundMessage } from "../inbound/message-aliases.js"; -import { attachWebInboxToSocket, type WhatsAppGroupMetadataCache } from "../inbound/monitor.js"; +import { + attachWebInboxToSocket, + readWhatsAppBaileysCacheEntry, + type WhatsAppBaileysGroupMetadataCache, + type WhatsAppBaileysMessageCache, + type WhatsAppGroupMetadataCache, +} from "../inbound/monitor.js"; import type { WebInboundMessageInput } from "../inbound/types.js"; import { newConnectionId, @@ -193,6 +200,8 @@ export async function monitorWebChannel( >(); const groupMemberNames = new Map>(); const groupMetadataCache: WhatsAppGroupMetadataCache = new Map(); + const recentMessageKeys: WhatsAppBaileysMessageCache = new Map(); + const baileysGroupMetaCache: WhatsAppBaileysGroupMetadataCache = new Map(); const echoTracker = createEchoTracker({ maxItems: 100, logVerbose }); const sleep = @@ -268,6 +277,14 @@ export async function monitorWebChannel( try { connection = await controller.openConnection({ connectionId, + getMessage: async (key: WAMessageKey) => + key.id && key.remoteJid + ? readWhatsAppBaileysCacheEntry(recentMessageKeys, `${key.remoteJid}:${key.id}`) + : undefined, + cachedGroupMetadata: async (jid: string) => { + const meta = readWhatsAppBaileysCacheEntry(baileysGroupMetaCache, jid); + return meta?.participants?.length ? meta : undefined; + }, createListener: async ({ sock, connection: connectionLocal }) => { const onMessage = createWebOnMessageHandler({ cfg, @@ -303,6 +320,8 @@ export async function monitorWebChannel( disconnectRetryPolicy: reconnectPolicy, disconnectRetryAbortSignal: controller.getDisconnectRetryAbortSignal(), groupMetadataCache, + recentMessageKeys, + baileysGroupMetaCache, onMessage: async (msg: WebInboundMessageInput) => { const normalized = normalizeWebInboundMessage(msg); const inboundAt = Date.now(); diff --git a/extensions/whatsapp/src/connection-controller.ts b/extensions/whatsapp/src/connection-controller.ts index 3f7e52be3c66..6a7483a809f8 100644 --- a/extensions/whatsapp/src/connection-controller.ts +++ b/extensions/whatsapp/src/connection-controller.ts @@ -1,5 +1,5 @@ // Whatsapp plugin module implements connection controller behavior. -import type { WASocket } from "baileys"; +import type { GroupMetadata, WASocket, WAMessageKey, proto } from "baileys"; import { info } from "openclaw/plugin-sdk/runtime-env"; import type { RuntimeEnv } from "openclaw/plugin-sdk/runtime-env"; import { @@ -528,6 +528,8 @@ export class WhatsAppConnectionController { }) => Promise; onHeartbeat?: (snapshot: WhatsAppConnectionSnapshot) => void; onWatchdogTimeout?: (snapshot: WhatsAppConnectionSnapshot) => void; + getMessage?: (key: WAMessageKey) => Promise; + cachedGroupMetadata?: (jid: string) => Promise; }): Promise { if (this.current) { await this.closeCurrentConnection(); @@ -539,6 +541,8 @@ export class WhatsAppConnectionController { sock = await createWaSocket(false, this.verbose, { authDir: this.authDir, ...this.socketTiming, + ...(params.getMessage ? { getMessage: params.getMessage } : {}), + ...(params.cachedGroupMetadata ? { cachedGroupMetadata: params.cachedGroupMetadata } : {}), }); await waitForWaConnection(sock, { timeoutMs: this.socketTiming.connectTimeoutMs }); diff --git a/extensions/whatsapp/src/inbound/monitor.ts b/extensions/whatsapp/src/inbound/monitor.ts index 90db847a0496..f718580e1115 100644 --- a/extensions/whatsapp/src/inbound/monitor.ts +++ b/extensions/whatsapp/src/inbound/monitor.ts @@ -5,6 +5,7 @@ import type { proto, GroupMetadata, WAMessage, + WAMessageKey, WASocket, } from "baileys"; import { recordChannelActivity } from "openclaw/plugin-sdk/channel-activity-runtime"; @@ -97,7 +98,8 @@ import type { const LOGGED_OUT_STATUS = DisconnectReason?.loggedOut ?? 401; const RECONNECT_IN_PROGRESS_ERROR = "no active socket - reconnection in progress"; -const GROUP_META_TTL_MS = 5 * 60 * 1000; // 5 minutes +const GROUP_META_TTL_MS = 5 * 60 * 1000; +const BAILEYS_MESSAGE_TTL_MS = 10 * 60 * 1000; const INBOUND_CLOSE_DRAIN_TIMEOUT_MS = 5_000; export const WHATSAPP_GROUP_METADATA_CACHE_MAX_ENTRIES = 500; @@ -106,6 +108,15 @@ type WhatsAppGroupMetadataCacheEntry = { expires: number; }; export type WhatsAppGroupMetadataCache = Map; +export type WhatsAppBaileysCacheEntry = { + expiresAt: number; + value: T; +}; +export type WhatsAppBaileysMessageCache = Map>; +export type WhatsAppBaileysGroupMetadataCache = Map< + string, + WhatsAppBaileysCacheEntry +>; type LocalGroupMetadataCacheEntry = WhatsAppGroupMetadataCacheEntry & { participants?: string[]; mentionParticipants?: WhatsAppOutboundMentionParticipant[]; @@ -171,6 +182,48 @@ function readGroupMetadataCacheEntry( return entry; } +function rememberWhatsAppBaileysCacheEntry( + cache: Map> | undefined, + key: string, + value: T, + ttlMs: number, +): void { + if (!cache) { + return; + } + if (cache.has(key)) { + cache.delete(key); + } + cache.set(key, { + expiresAt: Date.now() + ttlMs, + value, + }); + while (cache.size > WHATSAPP_GROUP_METADATA_CACHE_MAX_ENTRIES) { + const oldest = cache.keys().next(); + if (oldest.done) { + break; + } + cache.delete(oldest.value); + } +} + +export function readWhatsAppBaileysCacheEntry( + cache: Map>, + key: string, +): T | undefined { + const entry = cache.get(key); + if (!entry) { + return undefined; + } + if (entry.expiresAt <= Date.now()) { + cache.delete(key); + return undefined; + } + cache.delete(key); + cache.set(key, entry); + return entry.value; +} + function logWhatsAppVerbose(enabled: boolean | undefined, message: string) { if (!enabled) { return; @@ -242,6 +295,8 @@ type MonitorWebInboxOptions = { disconnectRetryAbortSignal?: AbortSignal; /** Shared group metadata cache used only for inbound metadata fallback after fetch failures. */ groupMetadataCache?: WhatsAppGroupMetadataCache; + recentMessageKeys?: WhatsAppBaileysMessageCache; + baileysGroupMetaCache?: WhatsAppBaileysGroupMetadataCache; }; type AttachWebInboxToSocketOptions = Omit< @@ -495,10 +550,29 @@ export async function attachWebInboxToSocket( const groupMetadataCache = options.groupMetadataCache ?? new Map(); const groupMetaCache = new Map(); const lidLookup = sock.signalRepository?.lidMapping; + const publishedGroupMetadataJids = new Set(); + const invalidatedGroupMetadataJids = new Set(); + let groupMetadataCacheClosed = false; const resolveInboundJid = async (jid: string | null | undefined): Promise => resolveJidToE164(jid, { authDir: options.authDir, lidLookup }); + const rememberBaileysMessage = ( + remoteJid: string | null | undefined, + messageId: string | null | undefined, + message: proto.IMessage | null | undefined, + ) => { + if (!options.recentMessageKeys || !remoteJid || !messageId || !message) { + return; + } + rememberWhatsAppBaileysCacheEntry( + options.recentMessageKeys, + `${remoteJid}:${messageId}`, + message, + BAILEYS_MESSAGE_TTL_MS, + ); + }; + const rememberOutboundMessage = (remoteJid: string, result: unknown) => { const messageId = typeof result === "object" && result && "key" in result @@ -512,6 +586,11 @@ export async function attachWebInboxToSocket( remoteJid, messageId, }); + const message = + typeof result === "object" && result && "message" in result + ? (result as { message?: proto.IMessage }).message + : undefined; + rememberBaileysMessage(remoteJid, messageId, message); }; const trackLateAcceptedSend = (jid: string, promise: Promise) => { // The local send has failed terminally, but Baileys may still deliver it. @@ -629,6 +708,13 @@ export async function attachWebInboxToSocket( } try { const meta = await (getCurrentSock() ?? sock).groupMetadata(jid); + rememberWhatsAppBaileysCacheEntry( + options.baileysGroupMetaCache, + jid, + meta, + GROUP_META_TTL_MS, + ); + publishedGroupMetadataJids.add(jid); const entry = await summarizeGroupMeta(meta); rememberGroupMetadataCacheEntry(groupMetadataCache, jid, { subject: entry.subject, @@ -1260,6 +1346,8 @@ export async function attachWebInboxToSocket( return; } for (const msg of upsert.messages ?? []) { + rememberBaileysMessage(msg.key?.remoteJid, msg.key?.id, msg.message); + const receiveOrder = nextReceiveOrder++; if ( await maybeResolveWhatsAppApprovalReaction({ @@ -1309,6 +1397,7 @@ export async function attachWebInboxToSocket( } }; const drainInboundBeforeSocketClose = async () => { + groupMetadataCacheClosed = true; await waitForPendingMessageHandlers(); await drainDebouncedInboundMessages(); }; @@ -1352,25 +1441,82 @@ export async function attachWebInboxToSocket( resolveClose({ status: undefined, isLoggedOut: false, error: err }); } }; - const detachMessagesUpsert = attachEmitterListener( - sock.ev as unknown as { - on: (event: string, listener: (...args: unknown[]) => void) => void; - off?: (event: string, listener: (...args: unknown[]) => void) => void; - removeListener?: (event: string, listener: (...args: unknown[]) => void) => void; - }, + const attachSockListener = (event: string, listener: (...args: unknown[]) => void) => + attachEmitterListener( + sock.ev as unknown as { + on: (event: string, listener: (...args: unknown[]) => void) => void; + off?: (event: string, listener: (...args: unknown[]) => void) => void; + removeListener?: (event: string, listener: (...args: unknown[]) => void) => void; + }, + event, + listener, + ); + const detachMessagesUpsert = attachSockListener( "messages.upsert", handleMessagesUpsertEvent as unknown as (...args: unknown[]) => void, ); - const detachConnectionUpdate = attachEmitterListener( - sock.ev as unknown as { - on: (event: string, listener: (...args: unknown[]) => void) => void; - off?: (event: string, listener: (...args: unknown[]) => void) => void; - removeListener?: (event: string, listener: (...args: unknown[]) => void) => void; - }, + const detachConnectionUpdate = attachSockListener( "connection.update", handleConnectionUpdate as unknown as (...args: unknown[]) => void, ); + const isFullGroupMetadataUpdate = (update: Partial): update is GroupMetadata => + typeof update.id === "string" && + typeof update.subject === "string" && + Array.isArray(update.participants); + + const rememberFullGroupMetadataUpdate = (jid: string, meta: GroupMetadata) => { + if (groupMetadataCacheClosed) { + return; + } + rememberWhatsAppBaileysCacheEntry(options.baileysGroupMetaCache, jid, meta, GROUP_META_TTL_MS); + publishedGroupMetadataJids.add(jid); + invalidatedGroupMetadataJids.delete(jid); + rememberGroupMetadataCacheEntry( + groupMetadataCache, + jid, + summarizeGroupMetaForReconnectCache(meta), + ); + groupMetaCache.delete(jid); + }; + + const forgetFullGroupMetadata = (jid: string) => { + options.baileysGroupMetaCache?.delete(jid); + groupMetadataCache.delete(jid); + groupMetaCache.delete(jid); + publishedGroupMetadataJids.delete(jid); + invalidatedGroupMetadataJids.add(jid); + }; + + const detachGroupsUpsert = attachSockListener("groups.upsert", ((groups: GroupMetadata[]) => { + for (const group of groups) { + if (group.id) { + rememberFullGroupMetadataUpdate(group.id, group); + } + } + }) as unknown as (...args: unknown[]) => void); + + const detachGroupsUpdate = attachSockListener("groups.update", (( + updates: Partial[], + ) => { + for (const update of updates) { + if (!update.id) { + continue; + } + if (isFullGroupMetadataUpdate(update)) { + rememberFullGroupMetadataUpdate(update.id, update); + continue; + } + forgetFullGroupMetadata(update.id); + } + }) as unknown as (...args: unknown[]) => void); + + const detachGroupParticipantsUpdate = attachSockListener("group-participants.update", ((update: { + id: string; + }) => { + forgetFullGroupMetadata(update.id); + }) as unknown as (...args: unknown[]) => void); + const replayTask = replayPendingDurableInboundMessages().catch((err: unknown) => { inboundLogger.error({ error: String(err) }, "failed replaying durable WhatsApp inbound"); inboundConsoleLog.error(`Failed replaying durable WhatsApp inbound: ${String(err)}`); @@ -1380,16 +1526,30 @@ export async function attachWebInboxToSocket( pendingMessageHandlers.delete(replayTask); }); - void (async () => { + const groupHydrationTask = (async () => { try { const groups = await sock.groupFetchAllParticipating(); + if (groupMetadataCacheClosed) { + return; + } for (const [jid, meta] of Object.entries(groups ?? {})) { - if (meta) { + if ( + meta && + !publishedGroupMetadataJids.has(jid) && + !invalidatedGroupMetadataJids.has(jid) + ) { rememberGroupMetadataCacheEntry( groupMetadataCache, jid, summarizeGroupMetaForReconnectCache(meta), ); + rememberWhatsAppBaileysCacheEntry( + options.baileysGroupMetaCache, + jid, + meta, + GROUP_META_TTL_MS, + ); + publishedGroupMetadataJids.add(jid); } } logWhatsAppVerbose( @@ -1406,6 +1566,7 @@ export async function attachWebInboxToSocket( ); } })(); + void groupHydrationTask; const sendApi = createWebSendApi({ sock: sendApiSocketOperations, @@ -1419,6 +1580,9 @@ export async function attachWebInboxToSocket( try { detachMessagesUpsert(); detachConnectionUpdate(); + detachGroupsUpsert(); + detachGroupsUpdate(); + detachGroupParticipantsUpdate(); await drainInboundBeforeSocketCloseWithTimeout(); } catch (err) { logWhatsAppVerbose(options.verbose, `Inbound close drain failed: ${String(err)}`); @@ -1442,9 +1606,21 @@ export async function attachWebInboxToSocket( export async function monitorWebInbox(options: MonitorWebInboxOptions) { const socketTiming = options.socketTiming ?? resolveWhatsAppSocketTiming(options.cfg); + const recentMessageKeys: WhatsAppBaileysMessageCache = options.recentMessageKeys ?? new Map(); + const baileysGroupMetaCache: WhatsAppBaileysGroupMetadataCache = + options.baileysGroupMetaCache ?? new Map(); + const sock = await createWaSocket(false, options.verbose, { authDir: options.authDir, ...socketTiming, + getMessage: async (key: WAMessageKey) => + key.id && key.remoteJid + ? readWhatsAppBaileysCacheEntry(recentMessageKeys, `${key.remoteJid}:${key.id}`) + : undefined, + cachedGroupMetadata: async (jid: string) => { + const meta = readWhatsAppBaileysCacheEntry(baileysGroupMetaCache, jid); + return meta?.participants?.length ? meta : undefined; + }, }); try { await waitForWaConnection(sock, { timeoutMs: socketTiming.connectTimeoutMs }); @@ -1469,5 +1645,7 @@ export async function monitorWebInbox(options: MonitorWebInboxOptions) { : undefined, socketTiming, sock, + recentMessageKeys, + baileysGroupMetaCache, }); } diff --git a/extensions/whatsapp/src/monitor-inbox.streams-inbound-messages.test-support.ts b/extensions/whatsapp/src/monitor-inbox.streams-inbound-messages.test-support.ts index ebdad42084be..6562b9a786a5 100644 --- a/extensions/whatsapp/src/monitor-inbox.streams-inbound-messages.test-support.ts +++ b/extensions/whatsapp/src/monitor-inbox.streams-inbound-messages.test-support.ts @@ -1,6 +1,7 @@ // Whatsapp plugin module implements monitor inbox.streams inbound messages support behavior. import fsSync from "node:fs"; import path from "node:path"; +import type { GroupMetadata, WAMessageKey } from "baileys"; import "./monitor-inbox.test-harness.js"; import { defaultRuntime } from "openclaw/plugin-sdk/runtime-env"; import { beforeEach, describe, expect, it, vi } from "vitest"; @@ -9,7 +10,12 @@ import { unregisterWhatsAppConnectionController, } from "./connection-controller-registry.js"; import { WhatsAppRetryableInboundError } from "./inbound/dedupe.js"; -import { WHATSAPP_GROUP_METADATA_CACHE_MAX_ENTRIES } from "./inbound/monitor.js"; +import { + readWhatsAppBaileysCacheEntry, + type WhatsAppBaileysGroupMetadataCache, + type WhatsAppBaileysMessageCache, + WHATSAPP_GROUP_METADATA_CACHE_MAX_ENTRIES, +} from "./inbound/monitor.js"; import type { WebInboundMessage } from "./inbound/types.js"; import { type InboxMonitorOptions, @@ -106,17 +112,70 @@ async function expectSocketOperationTimeout( await rejection; } +function groupMetadata(params: { + id?: string; + subject: string; + participants?: string[]; +}): GroupMetadata { + return { + id: params.id ?? "123@g.us", + subject: params.subject, + owner: undefined, + participants: (params.participants ?? ["555@s.whatsapp.net"]).map((id) => ({ id })), + }; +} + +function createBaileysCacheSupport() { + const recentMessageKeys: WhatsAppBaileysMessageCache = new Map(); + const baileysGroupMetaCache: WhatsAppBaileysGroupMetadataCache = new Map(); + const socketOptions = { + getMessage: async (key: WAMessageKey) => + key.id && key.remoteJid + ? readWhatsAppBaileysCacheEntry(recentMessageKeys, `${key.remoteJid}:${key.id}`) + : undefined, + cachedGroupMetadata: async (jid: string) => { + const meta = readWhatsAppBaileysCacheEntry(baileysGroupMetaCache, jid); + return meta?.participants?.length ? meta : undefined; + }, + }; + return { recentMessageKeys, baileysGroupMetaCache, socketOptions }; +} + +async function startInboxMonitorWithBaileysCache( + options: Partial> = {}, +) { + const baileysCache = createBaileysCacheSupport(); + const started = await startInboxMonitor(vi.fn(async () => {}) as InboxOnMessage, { + ...options, + recentMessageKeys: baileysCache.recentMessageKeys, + baileysGroupMetaCache: baileysCache.baileysGroupMetaCache, + }); + return { ...started, baileysCache }; +} + +async function expectCachedGroupMetadata( + baileysCache: ReturnType, + expected: Pick, +) { + await expect(baileysCache.socketOptions.cachedGroupMetadata(expected.id)).resolves.toMatchObject( + expected, + ); +} + async function primeInboundReplyHandle(params: { onMessage: ReturnType; socketRef: NonNullable; upsertId: string; retryPolicy: NonNullable; + baileysCache?: ReturnType; useCurrentSock?: boolean; }) { const { listener, sock } = await startInboxMonitor(params.onMessage as InboxOnMessage, { socketRef: params.socketRef, shouldRetryDisconnect: () => true, disconnectRetryPolicy: params.retryPolicy, + recentMessageKeys: params.baileysCache?.recentMessageKeys, + baileysGroupMetaCache: params.baileysCache?.baileysGroupMetaCache, }); const sourceSock = params.useCurrentSock ? getSock() : sock; sourceSock.ev.emit( @@ -480,6 +539,174 @@ describe("web monitor inbox", () => { await second.listener.close(); }); + it("keeps full participating group metadata available to Baileys", async () => { + const sock = getSock(); + sock.groupFetchAllParticipating.mockResolvedValueOnce({ + "123@g.us": groupMetadata({ + subject: "Recovered Group", + participants: ["444@s.whatsapp.net"], + }), + }); + + const { listener, baileysCache } = await startInboxMonitorWithBaileysCache(); + + await vi.waitFor(async () => { + await expectCachedGroupMetadata(baileysCache, { + id: "123@g.us", + subject: "Recovered Group", + participants: [{ id: "444@s.whatsapp.net" }], + }); + }); + + await listener.close(); + }); + + it("invalidates cached group metadata on partial group and participant updates", async () => { + const groupMetadataCache: NonNullable = new Map(); + const { listener, sock, baileysCache } = await startInboxMonitorWithBaileysCache({ + groupMetadataCache, + }); + sock.ev.emit("groups.update", [ + groupMetadata({ + subject: "Fresh Group", + }), + ]); + await expectCachedGroupMetadata(baileysCache, { + id: "123@g.us", + subject: "Fresh Group", + participants: [{ id: "555@s.whatsapp.net" }], + }); + expect(groupMetadataCache.has("123@g.us")).toBe(true); + + sock.ev.emit("groups.update", [{ id: "123@g.us" }]); + expect(groupMetadataCache.has("123@g.us")).toBe(false); + await expect( + baileysCache.socketOptions.cachedGroupMetadata("123@g.us"), + ).resolves.toBeUndefined(); + sock.ev.emit("groups.update", [ + groupMetadata({ + subject: "Fresh Again", + }), + ]); + expect(groupMetadataCache.has("123@g.us")).toBe(true); + sock.ev.emit("group-participants.update", { id: "123@g.us" }); + expect(groupMetadataCache.has("123@g.us")).toBe(false); + await expect( + baileysCache.socketOptions.cachedGroupMetadata("123@g.us"), + ).resolves.toBeUndefined(); + + await listener.close(); + }); + + it("expires Baileys retry and group metadata cache entries", async () => { + const now = vi.spyOn(Date, "now").mockReturnValue(1_700_000_000_000); + const baileysCache = createBaileysCacheSupport(); + const onMessage = vi.fn(async (_msg: Parameters[0]) => {}); + const { listener, sock } = await startInboxMonitor(onMessage as InboxOnMessage, { + recentMessageKeys: baileysCache.recentMessageKeys, + baileysGroupMetaCache: baileysCache.baileysGroupMetaCache, + }); + const messageId = nextMessageId("baileys-expiry"); + try { + sock.ev.emit( + "messages.upsert", + buildNotifyMessageUpsert({ + id: messageId, + remoteJid: "999@s.whatsapp.net", + text: "retry me", + timestamp: 1_700_000_000, + pushName: "Tester", + }), + ); + sock.ev.emit("groups.update", [ + groupMetadata({ + subject: "Expiring Group", + }), + ]); + await waitForMessageCalls(onMessage, 1); + + await expect( + baileysCache.socketOptions.getMessage({ + id: messageId, + remoteJid: "999@s.whatsapp.net", + }), + ).resolves.toEqual({ conversation: "retry me" }); + await expectCachedGroupMetadata(baileysCache, { + id: "123@g.us", + subject: "Expiring Group", + participants: [{ id: "555@s.whatsapp.net" }], + }); + + now.mockReturnValue(1_700_000_000_000 + 5 * 60 * 1000 + 1); + await expect( + baileysCache.socketOptions.cachedGroupMetadata("123@g.us"), + ).resolves.toBeUndefined(); + + now.mockReturnValue(1_700_000_000_000 + 10 * 60 * 1000 + 1); + await expect( + baileysCache.socketOptions.getMessage({ + id: messageId, + remoteJid: "999@s.whatsapp.net", + }), + ).resolves.toBeUndefined(); + } finally { + now.mockRestore(); + await listener.close(); + } + }); + + it("does not republish invalidated group metadata from pending hydration", async () => { + const groupMetadataCache: NonNullable = new Map(); + const baileysCache = createBaileysCacheSupport(); + const sock = getSock(); + let resolveHydration!: (groups: Record) => void; + sock.groupFetchAllParticipating.mockImplementationOnce( + async () => + await new Promise>((resolve) => { + resolveHydration = resolve; + }), + ); + + const { listener } = await startInboxMonitor(vi.fn(async () => {}) as InboxOnMessage, { + groupMetadataCache, + recentMessageKeys: baileysCache.recentMessageKeys, + baileysGroupMetaCache: baileysCache.baileysGroupMetaCache, + }); + sock.ev.emit("groups.update", [{ id: "123@g.us" }]); + + resolveHydration({ + "123@g.us": groupMetadata({ + subject: "Stale Hydration Group", + }), + }); + await settleInboundWork(); + + expect(groupMetadataCache.has("123@g.us")).toBe(false); + await expect( + baileysCache.socketOptions.cachedGroupMetadata("123@g.us"), + ).resolves.toBeUndefined(); + + await listener.close(); + }); + + it("cleans up Baileys group metadata listeners on close", async () => { + const baileysCache = createBaileysCacheSupport(); + const { listener, sock } = await startInboxMonitor(vi.fn(async () => {}) as InboxOnMessage, { + recentMessageKeys: baileysCache.recentMessageKeys, + baileysGroupMetaCache: baileysCache.baileysGroupMetaCache, + }); + + expect(sock.ev.listenerCount("groups.upsert")).toBe(1); + expect(sock.ev.listenerCount("groups.update")).toBe(1); + expect(sock.ev.listenerCount("group-participants.update")).toBe(1); + + await listener.close(); + + expect(sock.ev.listenerCount("groups.upsert")).toBe(0); + expect(sock.ev.listenerCount("groups.update")).toBe(0); + expect(sock.ev.listenerCount("group-participants.update")).toBe(0); + }); + it("bounds cached group metadata kept across reconnects", async () => { const groupMetadataCache: NonNullable = new Map(); const groups = Object.fromEntries( @@ -934,16 +1161,50 @@ describe("web monitor inbox", () => { } }); - it("suppresses self-echo when a timed-out socket send is later accepted", async () => { + it("records outbound replies for Baileys retry lookup", async () => { const onMessage = vi.fn(async () => undefined); const socketRef = createSocketRef(); + const baileysCache = createBaileysCacheSupport(); + const message = { conversation: "pong" }; const { listener, sock, inbound } = await primeInboundReplyHandle({ onMessage, socketRef, + baileysCache, + upsertId: "outbound-retry-cache", + retryPolicy: fastReconnectPolicy(2), + }); + sock.sendMessage.mockResolvedValueOnce({ + key: { id: "outbound-cached" }, + message, + }); + + await inbound.platform.reply("pong"); + + await expect( + baileysCache.socketOptions.getMessage({ + id: "outbound-cached", + remoteJid: "999@s.whatsapp.net", + }), + ).resolves.toBe(message); + + await listener.close(); + }); + + it("suppresses self-echo when a timed-out socket send is later accepted", async () => { + const onMessage = vi.fn(async () => undefined); + const socketRef = createSocketRef(); + const baileysCache = createBaileysCacheSupport(); + const { listener, sock, inbound } = await primeInboundReplyHandle({ + onMessage, + socketRef, + baileysCache, upsertId: "late-accept", retryPolicy: fastReconnectPolicy(2), }); - let acceptLateSend: ((value: { key: { id: string } }) => void) | undefined; + const message = { conversation: "pong" }; + let acceptLateSend: + | ((value: { key: { id: string }; message: { conversation: string } }) => void) + | undefined; vi.useFakeTimers(); try { sock.sendMessage.mockImplementationOnce( @@ -959,8 +1220,14 @@ describe("web monitor inbox", () => { vi.useRealTimers(); } - acceptLateSend?.({ key: { id: "late-accepted" } }); + acceptLateSend?.({ key: { id: "late-accepted" }, message }); await settleInboundWork(); + await expect( + baileysCache.socketOptions.getMessage({ + id: "late-accepted", + remoteJid: "999@s.whatsapp.net", + }), + ).resolves.toBe(message); sock.ev.emit("messages.upsert", { type: "notify", messages: [ diff --git a/extensions/whatsapp/src/session.ts b/extensions/whatsapp/src/session.ts index 7f8101165cd8..5c4a06cb80b4 100644 --- a/extensions/whatsapp/src/session.ts +++ b/extensions/whatsapp/src/session.ts @@ -1,6 +1,7 @@ // Whatsapp plugin module implements session behavior. import { randomUUID } from "node:crypto"; import type { Agent } from "node:https"; +import type { GroupMetadata, WAMessageKey, proto } from "baileys"; import { formatCliCommand } from "openclaw/plugin-sdk/cli-runtime"; import { VERSION } from "openclaw/plugin-sdk/cli-runtime"; import { @@ -134,6 +135,8 @@ export async function createWaSocket( opts: { authDir?: string; onQr?: (qr: string) => void; + getMessage?: (key: WAMessageKey) => Promise; + cachedGroupMetadata?: (jid: string) => Promise; } & WhatsAppSocketTimingOptions = {}, ): Promise> { const baseLogger = getChildLogger( @@ -185,6 +188,8 @@ export async function createWaSocket( // Baileys types still model `fetchAgent` as a Node agent even though the // runtime path accepts an undici dispatcher for upload fetches. fetchAgent: fetchAgent as Agent | undefined, + ...(opts.getMessage ? { getMessage: opts.getMessage } : {}), + ...(opts.cachedGroupMetadata ? { cachedGroupMetadata: opts.cachedGroupMetadata } : {}), }); sock.ev.on("creds.update", () => enqueueSaveCreds(authDir, saveCreds, sessionLogger));