diff --git a/docs/cli/directory.md b/docs/cli/directory.md index 38c1224247ef..1d54c1423d23 100644 --- a/docs/cli/directory.md +++ b/docs/cli/directory.md @@ -23,6 +23,7 @@ Default (non-JSON) output is `id` (and sometimes `name`) separated by a tab. ## Notes - For many channels, results are config-backed (allowlists / configured groups) rather than a live provider directory. +- WhatsApp group listing is live. Gateway lookups reuse its owned connection; a standalone command opens the linked session only when no other process owns that account and otherwise reports that live groups are unavailable. - An already-installed channel plugin can lack directory support. In that case the command reports the unsupported operation; it does not try to reinstall or upgrade the plugin to add support. ## Using results with `message send` diff --git a/extensions/whatsapp/src/channel.ts b/extensions/whatsapp/src/channel.ts index 470e1679354c..e26b958489e5 100644 --- a/extensions/whatsapp/src/channel.ts +++ b/extensions/whatsapp/src/channel.ts @@ -156,6 +156,8 @@ export const whatsappPlugin: ChannelPlugin = (await loadWhatsAppDirectoryConfig()).listWhatsAppDirectoryPeersFromConfig(params), listGroups: async (params) => (await loadWhatsAppDirectoryConfig()).listWhatsAppDirectoryGroupsFromConfig(params), + listGroupsLive: async (params) => + (await loadWhatsAppDirectoryConfig()).listWhatsAppDirectoryGroupsLive(params), }, actions: { describeMessageTool: ({ cfg, accountId }) => diff --git a/extensions/whatsapp/src/connection-controller-runtime-context.ts b/extensions/whatsapp/src/connection-controller-runtime-context.ts index a2e357d2d013..bc135146b93d 100644 --- a/extensions/whatsapp/src/connection-controller-runtime-context.ts +++ b/extensions/whatsapp/src/connection-controller-runtime-context.ts @@ -6,6 +6,7 @@ import type { ActiveWebListener } from "./inbound/types.js"; import { getOptionalWhatsAppRuntime } from "./runtime.js"; export const WHATSAPP_CONNECTION_CONTROLLER_CAPABILITY = "connection-controller"; +export const WHATSAPP_CONNECTION_OWNER_PENDING_CAPABILITY = "connection-owner-pending"; type WhatsAppConnectionControllerHandle = { getActiveListener(): ActiveWebListener | null; @@ -24,3 +25,14 @@ export function getWhatsAppConnectionController( }); return (context as WhatsAppConnectionControllerHandle | undefined) ?? null; } + +export function hasPendingWhatsAppConnectionOwner(accountId: string): boolean { + return Boolean( + getChannelRuntimeContext({ + channelRuntime: getOptionalWhatsAppRuntime()?.channel, + channelId: "whatsapp", + accountId, + capability: WHATSAPP_CONNECTION_OWNER_PENDING_CAPABILITY, + }), + ); +} diff --git a/extensions/whatsapp/src/connection-controller.test.ts b/extensions/whatsapp/src/connection-controller.test.ts index df4985961f7e..e60068c0a79f 100644 --- a/extensions/whatsapp/src/connection-controller.test.ts +++ b/extensions/whatsapp/src/connection-controller.test.ts @@ -16,6 +16,7 @@ import { createWaSocket, logoutWeb, readWebAuthExistsForDecision, + waitForCredsSaveQueueWithTimeout, waitForWaConnection, } from "./session.js"; import { DEFAULT_WHATSAPP_SOCKET_TIMING } from "./socket-timing.js"; @@ -28,6 +29,7 @@ vi.mock("./session.js", async () => { waitForWaConnection: vi.fn(), logoutWeb: vi.fn(async () => true), readWebAuthExistsForDecision: vi.fn(async () => ({ outcome: "stable" as const, exists: true })), + waitForCredsSaveQueueWithTimeout: vi.fn(async () => "drained" as const), }; }); @@ -36,6 +38,11 @@ const runtimeContextMocks = vi.hoisted(() => ({ register: vi.fn(), })); +const connectionOwnerMocks = vi.hoisted(() => ({ + acquire: vi.fn(), + release: vi.fn(), +})); + vi.mock("openclaw/plugin-sdk/channel-runtime-context", () => { return { getChannelRuntimeContext: vi.fn(), @@ -47,10 +54,15 @@ vi.mock("./runtime.js", () => ({ getWhatsAppRuntime: () => ({ channel: runtimeContextMocks.channelRuntime }), })); +vi.mock("./connection-owner.js", () => ({ + acquireWhatsAppGatewayConnectionOwner: connectionOwnerMocks.acquire, +})); + const createWaSocketMock = vi.mocked(createWaSocket); const waitForWaConnectionMock = vi.mocked(waitForWaConnection); const logoutWebMock = vi.mocked(logoutWeb); const readWebAuthExistsForDecisionMock = vi.mocked(readWebAuthExistsForDecision); +const waitForCredsSaveQueueWithTimeoutMock = vi.mocked(waitForCredsSaveQueueWithTimeout); const registerChannelRuntimeContextMock = runtimeContextMocks.register; function createListenerStub(messageId = "ok") { @@ -63,10 +75,19 @@ function createListenerStub(messageId = "ok") { } function createSocketWithTransportEmitter() { - const ws = new EventEmitter() as EventEmitter & { close: ReturnType }; - ws.close = vi.fn(); + let closed = false; + const ws = new EventEmitter() as EventEmitter & { + close: ReturnType; + readonly isClosed: boolean; + }; + Object.defineProperty(ws, "isClosed", { get: () => closed }); + ws.close = vi.fn(async () => { + closed = true; + }); return { - end: vi.fn(), + end: vi.fn(async (_error?: Error) => { + closed = true; + }), ws, }; } @@ -143,10 +164,13 @@ describe("WhatsAppConnectionController", () => { beforeEach(() => { vi.clearAllMocks(); registerChannelRuntimeContextMock.mockReturnValue({ dispose: vi.fn() }); + connectionOwnerMocks.acquire.mockResolvedValue({ release: connectionOwnerMocks.release }); + connectionOwnerMocks.release.mockResolvedValue(undefined); logoutWebMock.mockResolvedValue(true); readWebAuthExistsForDecisionMock .mockReset() .mockResolvedValue({ outcome: "stable", exists: true }); + waitForCredsSaveQueueWithTimeoutMock.mockReset().mockResolvedValue("drained"); controller = new WhatsAppConnectionController({ accountId: "work", authDir: "/tmp/wa-auth", @@ -171,12 +195,7 @@ describe("WhatsAppConnectionController", () => { }); it("closes the socket when open fails before listener creation", async () => { - const sock = { - end: vi.fn(), - ws: { - close: vi.fn(), - }, - }; + const sock = createSocketWithTransportEmitter(); const createListener = vi.fn(); createWaSocketMock.mockResolvedValueOnce(sock as never); @@ -207,11 +226,26 @@ describe("WhatsAppConnectionController", () => { expect(sock.ws.close).toHaveBeenCalledOnce(); }); + it("keeps asynchronous fallback close failures best-effort", async () => { + const sock = { + end: vi.fn().mockRejectedValue(new Error("end failed")), + ws: { close: vi.fn(() => Promise.reject(new Error("websocket close failed"))) }, + }; + + closeWaSocket(sock); + await new Promise((resolve) => { + setTimeout(resolve, 0); + }); + + expect(sock.end).toHaveBeenCalledOnce(); + expect(sock.ws.close).toHaveBeenCalledOnce(); + }); + it("lets createWaSocket own the auth barrier before opening a socket", async () => { const callOrder: string[] = []; createWaSocketMock.mockImplementationOnce(async () => { callOrder.push("create"); - return { ws: { close: vi.fn() } } as never; + return createSocketWithTransportEmitter() as never; }); waitForWaConnectionMock.mockImplementationOnce(async () => { callOrder.push("wait-for-connection"); @@ -598,7 +632,7 @@ describe("WhatsAppConnectionController", () => { } }); - it("keeps the previous registered controller until a replacement listener is ready", async () => { + it("keeps the ready controller published while a different-auth replacement connects", async () => { const disposeRuntimeContext = vi.fn(); registerChannelRuntimeContextMock.mockReturnValueOnce({ dispose: disposeRuntimeContext }); const liveController = new WhatsAppConnectionController({ @@ -619,22 +653,13 @@ describe("WhatsAppConnectionController", () => { }, }); const liveListener = createListenerStub("live"); - createWaSocketMock.mockResolvedValueOnce({ ws: { close: vi.fn() } } as never); + createWaSocketMock.mockResolvedValueOnce(createSocketWithTransportEmitter() as never); waitForWaConnectionMock.mockResolvedValueOnce(undefined); await liveController.openConnection({ connectionId: "live-conn", createListener: async () => liveListener, }); - expect(registerChannelRuntimeContextMock).toHaveBeenCalledWith({ - channelRuntime: runtimeContextMocks.channelRuntime, - channelId: "whatsapp", - accountId: "work", - capability: "connection-controller", - context: liveController, - abortSignal: undefined, - }); - const replacement = new WhatsAppConnectionController({ accountId: "work", authDir: "/tmp/wa-auth-2", @@ -654,7 +679,7 @@ describe("WhatsAppConnectionController", () => { }); try { - createWaSocketMock.mockResolvedValueOnce({ ws: { close: vi.fn() } } as never); + createWaSocketMock.mockResolvedValueOnce(createSocketWithTransportEmitter() as never); waitForWaConnectionMock.mockRejectedValueOnce(new Error("replacement failed")); await expect( @@ -664,7 +689,20 @@ describe("WhatsAppConnectionController", () => { }), ).rejects.toThrow("replacement failed"); - expect(registerChannelRuntimeContextMock).toHaveBeenCalledTimes(1); + expect(registerChannelRuntimeContextMock).toHaveBeenCalledTimes(3); + expect(registerChannelRuntimeContextMock).toHaveBeenLastCalledWith({ + channelRuntime: runtimeContextMocks.channelRuntime, + channelId: "whatsapp", + accountId: "work", + capability: "connection-owner-pending", + context: true, + abortSignal: undefined, + }); + const activeControllerRegistrations = registerChannelRuntimeContextMock.mock.calls.filter( + ([registration]) => registration.capability === "connection-controller", + ); + expect(activeControllerRegistrations).toHaveLength(1); + expect(activeControllerRegistrations[0]?.[0].context).toBe(liveController); } finally { await replacement.shutdown(); await liveController.shutdown(); @@ -672,6 +710,212 @@ describe("WhatsAppConnectionController", () => { expect(disposeRuntimeContext).toHaveBeenCalledOnce(); }); + it("releases connection ownership only after the Baileys socket closes", async () => { + const order: string[] = []; + let closed = false; + const sock = { + end: vi.fn(async () => { + closed = true; + order.push("socket-close"); + }), + ws: { + close: vi.fn(async () => { + closed = true; + }), + get isClosed() { + return closed; + }, + }, + }; + connectionOwnerMocks.release.mockImplementationOnce(async () => { + order.push("owner-release"); + }); + createWaSocketMock.mockResolvedValueOnce(sock as never); + waitForWaConnectionMock.mockResolvedValueOnce(undefined); + + await controller.openConnection({ + connectionId: "owned-conn", + createListener: async () => createListenerStub() as never, + }); + await controller.shutdown(); + + expect(order).toEqual(["socket-close", "owner-release"]); + }); + + it("joins pending ownership acquisition before shutdown returns", async () => { + let resolveOwner = (_lease: { release: () => Promise }) => {}; + connectionOwnerMocks.acquire.mockReturnValueOnce( + new Promise((resolve) => { + resolveOwner = resolve; + }), + ); + const openPromise = controller.openConnection({ + connectionId: "pending-owner", + createListener: async () => createListenerStub() as never, + }); + const shutdownPromise = controller.shutdown(); + + resolveOwner({ release: connectionOwnerMocks.release }); + + await expect(openPromise).rejects.toThrow("controller is shutting down"); + await expect(shutdownPromise).resolves.toBeUndefined(); + expect(registerChannelRuntimeContextMock).not.toHaveBeenCalled(); + expect(createWaSocketMock).not.toHaveBeenCalled(); + expect(connectionOwnerMocks.release).toHaveBeenCalledOnce(); + }); + + it("cancels handshake setup and closes its socket before shutdown returns", async () => { + const sock = createSocketWithTransportEmitter(); + createWaSocketMock.mockResolvedValueOnce(sock as never); + waitForWaConnectionMock.mockReturnValueOnce(new Promise(() => {})); + const openPromise = controller.openConnection({ + connectionId: "pending-handshake", + createListener: async () => createListenerStub() as never, + }); + await vi.waitFor(() => expect(waitForWaConnectionMock).toHaveBeenCalledOnce()); + + await expect(controller.shutdown()).resolves.toBeUndefined(); + await expect(openPromise).rejects.toThrow("controller is shutting down"); + expect(sock.end).toHaveBeenCalledOnce(); + expect(registerChannelRuntimeContextMock).toHaveBeenCalledTimes(1); + expect(registerChannelRuntimeContextMock).toHaveBeenLastCalledWith( + expect.objectContaining({ capability: "connection-owner-pending" }), + ); + expect(connectionOwnerMocks.release).toHaveBeenCalledOnce(); + }); + + it("does not publish a listener that resolves after setup is cancelled", async () => { + const sock = createSocketWithTransportEmitter(); + const lateListener = { ...createListenerStub(), close: vi.fn(async () => {}) }; + let resolveListener = (_listener: ReturnType) => {}; + const listenerPromise = new Promise>((resolve) => { + resolveListener = resolve; + }); + createWaSocketMock.mockResolvedValueOnce(sock as never); + waitForWaConnectionMock.mockResolvedValueOnce(undefined); + const openPromise = controller.openConnection({ + connectionId: "pending-listener", + createListener: async () => await listenerPromise, + }); + await vi.waitFor(() => expect(waitForWaConnectionMock).toHaveBeenCalledOnce()); + + await expect(controller.shutdown()).resolves.toBeUndefined(); + await expect(openPromise).rejects.toThrow("controller is shutting down"); + resolveListener(lateListener); + await listenerPromise; + await vi.waitFor(() => expect(lateListener.close).toHaveBeenCalledOnce()); + expect(controller.getActiveListener()).toBeNull(); + expect(controller.getCurrentSock()).toBeNull(); + expect(registerChannelRuntimeContextMock).toHaveBeenCalledTimes(1); + expect(connectionOwnerMocks.release).toHaveBeenCalledOnce(); + }); + + it("closes a listener resolved at the setup cancellation boundary", async () => { + const sock = createSocketWithTransportEmitter(); + const listener = { ...createListenerStub(), close: vi.fn(async () => {}) }; + let shutdownPromise: Promise | undefined; + createWaSocketMock.mockResolvedValueOnce(sock as never); + waitForWaConnectionMock.mockResolvedValueOnce(undefined); + const openPromise = controller.openConnection({ + connectionId: "resolved-listener", + createListener: async () => { + queueMicrotask(() => { + shutdownPromise = controller.shutdown(); + }); + return listener as never; + }, + }); + + await expect(openPromise).rejects.toThrow("controller is shutting down"); + await vi.waitFor(() => expect(shutdownPromise).toBeDefined()); + await shutdownPromise; + expect(listener.close).toHaveBeenCalledOnce(); + expect(controller.getActiveListener()).toBeNull(); + expect(connectionOwnerMocks.release).toHaveBeenCalledOnce(); + }); + + it("retains connection ownership when socket close cannot be confirmed", async () => { + let closed = false; + const sock = { + end: vi + .fn() + .mockRejectedValueOnce(new Error("end failed")) + .mockImplementationOnce(async () => { + closed = true; + }), + ws: { + close: vi.fn().mockRejectedValueOnce(new Error("websocket close failed")), + get isClosed() { + return closed; + }, + }, + }; + createWaSocketMock.mockResolvedValueOnce(sock as never); + waitForWaConnectionMock.mockResolvedValueOnce(undefined); + await controller.openConnection({ + connectionId: "uncertain-close", + createListener: async () => createListenerStub() as never, + }); + + await expect(controller.shutdown()).rejects.toThrow("socket close could not be confirmed"); + expect(connectionOwnerMocks.release).not.toHaveBeenCalled(); + }); + + it("retains connection ownership until queued credentials drain", async () => { + let closed = false; + const sock = { + end: vi.fn(async () => { + closed = true; + }), + ws: { + close: vi.fn(async () => { + closed = true; + }), + get isClosed() { + return closed; + }, + }, + }; + createWaSocketMock.mockResolvedValueOnce(sock as never); + waitForWaConnectionMock.mockResolvedValueOnce(undefined); + waitForCredsSaveQueueWithTimeoutMock + .mockResolvedValueOnce("timed_out") + .mockResolvedValueOnce("drained"); + await controller.openConnection({ + connectionId: "pending-creds", + createListener: async () => createListenerStub() as never, + }); + + await expect(controller.shutdown()).rejects.toThrow("credential persistence did not drain"); + expect(connectionOwnerMocks.release).not.toHaveBeenCalled(); + expect(sock.end).toHaveBeenCalledOnce(); + expect(controller.getActiveListener()).toBeNull(); + expect(controller.getCurrentSock()).toBeNull(); + + await expect(controller.shutdown()).resolves.toBeUndefined(); + expect(connectionOwnerMocks.release).toHaveBeenCalledOnce(); + }); + + it("retains connection ownership until a failed release can be retried", async () => { + const sock = createSocketWithTransportEmitter(); + connectionOwnerMocks.release + .mockRejectedValueOnce(new Error("owner release failed")) + .mockResolvedValueOnce(undefined); + createWaSocketMock.mockResolvedValueOnce(sock as never); + waitForWaConnectionMock.mockResolvedValueOnce(undefined); + await controller.openConnection({ + connectionId: "release-retry", + createListener: async () => createListenerStub() as never, + }); + + await expect(controller.shutdown()).rejects.toThrow("owner release failed"); + expect(controller.getActiveListener()).toBeNull(); + expect(controller.getCurrentSock()).toBeNull(); + + await expect(controller.shutdown()).resolves.toBeUndefined(); + expect(connectionOwnerMocks.release).toHaveBeenCalledTimes(2); + }); + it("tracks real websocket frame activity in the connection snapshot", async () => { vi.useFakeTimers(); const controllerValue = new WhatsAppConnectionController({ diff --git a/extensions/whatsapp/src/connection-controller.ts b/extensions/whatsapp/src/connection-controller.ts index 3ec1796666c0..a26975672732 100644 --- a/extensions/whatsapp/src/connection-controller.ts +++ b/extensions/whatsapp/src/connection-controller.ts @@ -3,7 +3,14 @@ import type { GroupMetadata, WASocket, WAMessageKey, proto } from "baileys"; import { registerChannelRuntimeContext } from "openclaw/plugin-sdk/channel-runtime-context"; import { info } from "openclaw/plugin-sdk/runtime-env"; import type { RuntimeEnv } from "openclaw/plugin-sdk/runtime-env"; -import { WHATSAPP_CONNECTION_CONTROLLER_CAPABILITY } from "./connection-controller-runtime-context.js"; +import { + WHATSAPP_CONNECTION_CONTROLLER_CAPABILITY, + WHATSAPP_CONNECTION_OWNER_PENDING_CAPABILITY, +} from "./connection-controller-runtime-context.js"; +import { + acquireWhatsAppGatewayConnectionOwner, + type WhatsAppConnectionOwnerLease, +} from "./connection-owner.js"; import { resolveComparableIdentity, type WhatsAppSelfIdentity } from "./identity.js"; import type { ActiveWebListener, WebListenerCloseReason } from "./inbound/types.js"; import { computeBackoff, sleepWithAbort, type ReconnectPolicy } from "./reconnect.js"; @@ -14,9 +21,11 @@ import { getStatusCode, logoutWeb, readWebAuthExistsForDecision, + waitForCredsSaveQueueWithTimeout, waitForWaConnection, WhatsAppAuthUnstableError, } from "./session.js"; +import { closeWhatsAppSocketAndWait } from "./socket-close.js"; import { DEFAULT_WHATSAPP_SOCKET_TIMING, type WhatsAppSocketTimingOptions, @@ -66,6 +75,21 @@ type WhatsAppLiveConnection = { backgroundTasks: Set>; closePromise: Promise; resolveClose: (reason: WebListenerCloseReason) => void; + socketClosed: boolean; +}; + +type WhatsAppSocketCleanup = Pick; + +type WhatsAppOpenConnectionParams = { + connectionId: string; + createListener: (context: { + sock: WASocket; + connection: WhatsAppLiveConnection; + }) => Promise; + onHeartbeat?: (snapshot: WhatsAppConnectionSnapshot) => void; + onWatchdogTimeout?: (snapshot: WhatsAppConnectionSnapshot) => void; + getMessage?: (key: WAMessageKey) => Promise; + cachedGroupMetadata?: (jid: string) => Promise; }; type WhatsAppConnectionSnapshot = { @@ -163,13 +187,22 @@ function createLiveConnection(params: { backgroundTasks: new Set>(), closePromise, resolveClose: resolveClosePromise, + socketClosed: false, }; } +async function closeWebSocketBestEffort(sock: { ws?: { close?: () => void | Promise } }) { + try { + await sock.ws?.close?.(); + } catch { + // ignore best-effort shutdown failures + } +} + export function closeWaSocket( sock: | { - end?: (error: Error | undefined) => void; + end?: (error: Error | undefined) => void | Promise; ws?: { close?: () => void }; } | null @@ -177,15 +210,25 @@ export function closeWaSocket( ): void { try { if (typeof sock?.end === "function") { - sock.end(new Error("OpenClaw WhatsApp socket close")); + void Promise.resolve(sock.end(new Error("OpenClaw WhatsApp socket close"))).catch( + async () => await closeWebSocketBestEffort(sock), + ); return; } - sock?.ws?.close?.(); + if (sock) { + void closeWebSocketBestEffort(sock); + } } catch { - // ignore best-effort shutdown failures + if (sock) { + void closeWebSocketBestEffort(sock); + } } } +function stoppedControllerError(): Error { + return new Error("WhatsApp connection controller is shutting down"); +} + export function closeWaSocketSoon( sock: | { @@ -426,6 +469,15 @@ export class WhatsAppConnectionController { private current: WhatsAppLiveConnection | null = null; private runtimeContextLease: { dispose: () => void } | null = null; + private pendingOwnerContextLease: { dispose: () => void } | null = null; + private connectionOwnerLease: WhatsAppConnectionOwnerLease | null = null; + private retainedOwnerReleaseLease: WhatsAppConnectionOwnerLease | null = null; + private connectionOwnerLeasePromise: Promise | null = null; + private connectionSetupPromise: Promise | null = null; + private pendingSocketCleanup: WhatsAppSocketCleanup | null = null; + private shuttingDown = false; + private readonly ownerAcquireAbortController = new AbortController(); + private readonly setupAbortController = new AbortController(); private reconnectAttempts = 0; private lastHandledInboundAt: number | null = null; @@ -470,10 +522,18 @@ export class WhatsAppConnectionController { if (params.abortSignal?.aborted) { this.stopDisconnectRetries(); + this.ownerAcquireAbortController.abort(params.abortSignal.reason); + this.setupAbortController.abort(params.abortSignal.reason); } else { - params.abortSignal?.addEventListener("abort", () => this.stopDisconnectRetries(), { - once: true, - }); + params.abortSignal?.addEventListener( + "abort", + () => { + this.stopDisconnectRetries(); + this.ownerAcquireAbortController.abort(params.abortSignal?.reason); + this.setupAbortController.abort(params.abortSignal?.reason); + }, + { once: true }, + ); } } @@ -567,20 +627,35 @@ export class WhatsAppConnectionController { this.current.unregisterUnhandled = unregister; } - async openConnection(params: { - connectionId: string; - createListener: (context: { - sock: WASocket; - connection: WhatsAppLiveConnection; - }) => Promise; - onHeartbeat?: (snapshot: WhatsAppConnectionSnapshot) => void; - onWatchdogTimeout?: (snapshot: WhatsAppConnectionSnapshot) => void; - getMessage?: (key: WAMessageKey) => Promise; - cachedGroupMetadata?: (jid: string) => Promise; - }): Promise { + async openConnection(params: WhatsAppOpenConnectionParams): Promise { + if (this.shuttingDown) { + throw stoppedControllerError(); + } + if (this.connectionSetupPromise) { + throw new Error("WhatsApp connection setup is already in progress"); + } + const setupPromise = this.openConnectionOwned(params); + this.connectionSetupPromise = setupPromise; + try { + return await setupPromise; + } finally { + if (this.connectionSetupPromise === setupPromise) { + this.connectionSetupPromise = null; + } + } + } + + private async openConnectionOwned( + params: WhatsAppOpenConnectionParams, + ): Promise { + await this.ensureConnectionOwnership(); + this.throwIfSetupStopped(); + await this.finishPendingSocketCleanup(); + this.throwIfSetupStopped(); if (this.current) { await this.closeCurrentConnection(); } + this.throwIfSetupStopped(); let sock: WaSocket | null = null; let connection: WhatsAppLiveConnection | null = null; @@ -591,9 +666,11 @@ export class WhatsAppConnectionController { ...(params.getMessage ? { getMessage: params.getMessage } : {}), ...(params.cachedGroupMetadata ? { cachedGroupMetadata: params.cachedGroupMetadata } : {}), }); - await waitForWaConnection(sock, { timeoutMs: this.socketTiming.connectTimeoutMs }); - const channelRuntime = getWhatsAppRuntime().channel; - + this.throwIfSetupStopped(); + await this.waitForSetupStep( + waitForWaConnection(sock, { timeoutMs: this.socketTiming.connectTimeoutMs }), + ); + this.throwIfSetupStopped(); this.socketRef.current = sock; const placeholderListener = {} as ManagedWhatsAppListener; connection = createLiveConnection({ @@ -602,38 +679,72 @@ export class WhatsAppConnectionController { listener: placeholderListener, openedAfterRecentInbound: this.isOpeningAfterRecentInbound(), }); - const listener = await params.createListener({ sock, connection }); + const listenerTask = params.createListener({ sock, connection }); + let listener: ManagedWhatsAppListener; + try { + listener = await this.waitForSetupStep(listenerTask); + } catch (error) { + if (this.setupAbortController.signal.aborted) { + void listenerTask.then((lateListener) => lateListener.close?.()).catch(() => {}); + } + throw error; + } connection.listener = listener; + this.throwIfSetupStopped(); this.current = connection; connection.unregisterTransportActivity = this.attachTransportActivityListener(sock); const previousRuntimeContextLease = this.runtimeContextLease; - // Outbound adapters and agent tools read this context outside the gateway account task, - // so the plugin-injected runtime is the shared owner for this internal capability. + // Publish only after the listener is ready. Lease tokens make disposal of this + // controller's previous registration unable to remove a newer replacement. this.runtimeContextLease = registerChannelRuntimeContext({ - channelRuntime, + channelRuntime: getWhatsAppRuntime().channel, channelId: "whatsapp", accountId: this.accountId, capability: WHATSAPP_CONNECTION_CONTROLLER_CAPABILITY, context: this, abortSignal: this.abortSignal, }); - // Publish the ready replacement before releasing the old lease. Runtime-context - // lease tokens keep stale disposal from unregistering the replacement. previousRuntimeContextLease?.dispose(); + this.pendingOwnerContextLease?.dispose(); + this.pendingOwnerContextLease = null; this.startTimers(connection, { onHeartbeat: params.onHeartbeat, onWatchdogTimeout: params.onWatchdogTimeout, }); return connection; } catch (err) { - if (this.socketRef.current === sock) { - this.socketRef.current = null; + try { + if (connection && this.current === connection) { + await this.closeCurrentConnection(); + } else { + try { + await connection?.listener.close?.(); + } catch { + // Socket closure remains authoritative for ownership release. + } + if (this.socketRef.current === sock) { + this.socketRef.current = null; + } + if (sock) { + const cleanup = connection ?? { sock, socketClosed: false }; + await this.finishSocketCleanup(cleanup); + } + if (connection?.unregisterUnhandled) { + connection.unregisterUnhandled(); + } + connection?.unregisterTransportActivity?.(); + } + } catch (closeError) { + if (sock && (!connection || this.current !== connection)) { + this.pendingSocketCleanup = connection ?? { sock, socketClosed: false }; + } + const aggregateError = new AggregateError( + [err, closeError], + "WhatsApp connection setup and close failed", + { cause: err }, + ); + throw aggregateError; } - closeWaSocket(sock); - if (connection?.unregisterUnhandled) { - connection.unregisterUnhandled(); - } - connection?.unregisterTransportActivity?.(); throw err; } } @@ -773,8 +884,10 @@ export class WhatsAppConnectionController { if (!connection) { return; } + // Stop exposing the listener before any fallible teardown. The cleanup record + // remains reachable so later reconnect/shutdown attempts can finish safely. this.current = null; - + this.pendingSocketCleanup = connection; if (this.socketRef.current === connection.sock) { this.socketRef.current = null; } @@ -795,7 +908,10 @@ export class WhatsAppConnectionController { } catch { // best-effort close } - closeWaSocket(connection.sock); + await this.finishSocketCleanup(connection); + if (this.pendingSocketCleanup === connection) { + this.pendingSocketCleanup = null; + } } async waitBeforeRetry(delayMs: number): Promise { @@ -803,10 +919,130 @@ export class WhatsAppConnectionController { } async shutdown(): Promise { + this.shuttingDown = true; + const stoppedError = stoppedControllerError(); + this.ownerAcquireAbortController.abort(stoppedError); + this.setupAbortController.abort(stoppedError); this.stopDisconnectRetries(); + await this.connectionSetupPromise?.catch(() => {}); + await this.connectionOwnerLeasePromise?.catch(() => {}); await this.closeCurrentConnection(); - this.runtimeContextLease?.dispose(); - this.runtimeContextLease = null; + await this.finishPendingSocketCleanup(); + try { + this.runtimeContextLease?.dispose(); + } finally { + this.runtimeContextLease = null; + try { + this.pendingOwnerContextLease?.dispose(); + } finally { + this.pendingOwnerContextLease = null; + // Contexts stay unpublished even when a retryable owner release fails. + // Keeping that lease reference preserves fail-closed process ownership. + if (this.connectionOwnerLease) { + await this.connectionOwnerLease.release(); + this.connectionOwnerLease = null; + } + if (this.retainedOwnerReleaseLease) { + await this.retainedOwnerReleaseLease.release(); + this.retainedOwnerReleaseLease = null; + } + } + } + } + + private async finishSocketCleanup(cleanup: WhatsAppSocketCleanup): Promise { + if (!cleanup.socketClosed) { + await closeWhatsAppSocketAndWait(cleanup.sock, "OpenClaw WhatsApp socket close"); + cleanup.socketClosed = true; + } + const queueResult = await waitForCredsSaveQueueWithTimeout(this.authDir); + if (queueResult === "timed_out") { + throw new Error("WhatsApp credential persistence did not drain before owner release"); + } + } + + private async finishPendingSocketCleanup(): Promise { + const cleanup = this.pendingSocketCleanup; + if (!cleanup) { + return; + } + await this.finishSocketCleanup(cleanup); + if (this.pendingSocketCleanup === cleanup) { + this.pendingSocketCleanup = null; + } + } + + private throwIfSetupStopped(): void { + if (this.shuttingDown || this.setupAbortController.signal.aborted) { + throw stoppedControllerError(); + } + } + + private async waitForSetupStep(task: Promise): Promise { + this.throwIfSetupStopped(); + const signal = this.setupAbortController.signal; + let onAbort: (() => void) | undefined; + const aborted = new Promise((_resolve, reject) => { + onAbort = () => reject(stoppedControllerError()); + signal.addEventListener("abort", onAbort, { once: true }); + }); + try { + return await Promise.race([task, aborted]); + } finally { + if (onAbort) { + signal.removeEventListener("abort", onAbort); + } + } + } + + private async ensureConnectionOwnership(): Promise { + if (this.retainedOwnerReleaseLease) { + await this.retainedOwnerReleaseLease.release(); + this.retainedOwnerReleaseLease = null; + } + if (this.connectionOwnerLease) { + return; + } + if (!this.connectionOwnerLeasePromise) { + this.connectionOwnerLeasePromise = (async () => { + const ownerLease = await acquireWhatsAppGatewayConnectionOwner( + this.authDir, + this.ownerAcquireAbortController.signal, + ); + try { + if (this.shuttingDown) { + throw stoppedControllerError(); + } + // Keep a pending marker separate from the ready controller capability. This + // blocks a second socket without replacing a still-working controller on reload. + this.pendingOwnerContextLease = registerChannelRuntimeContext({ + channelRuntime: getWhatsAppRuntime().channel, + channelId: "whatsapp", + accountId: this.accountId, + capability: WHATSAPP_CONNECTION_OWNER_PENDING_CAPABILITY, + context: true, + abortSignal: this.abortSignal, + }); + this.connectionOwnerLease = ownerLease; + } catch (error) { + try { + await ownerLease.release(); + } catch (releaseError) { + this.retainedOwnerReleaseLease = ownerLease; + const aggregateError = new AggregateError( + [error, releaseError], + "WhatsApp connection ownership setup and release failed", + { cause: error }, + ); + throw aggregateError; + } + throw error; + } + })().finally(() => { + this.connectionOwnerLeasePromise = null; + }); + } + await this.connectionOwnerLeasePromise; } private startTimers( diff --git a/extensions/whatsapp/src/connection-owner.test.ts b/extensions/whatsapp/src/connection-owner.test.ts new file mode 100644 index 000000000000..3ef83fa4bbee --- /dev/null +++ b/extensions/whatsapp/src/connection-owner.test.ts @@ -0,0 +1,100 @@ +// Whatsapp tests cover exclusive auth-backed connection ownership. +import fs from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { afterEach, describe, expect, it } from "vitest"; +import { + acquireWhatsAppGatewayConnectionOwner, + acquireWhatsAppStandaloneConnectionOwner, +} from "./connection-owner.js"; + +describe("WhatsApp connection owner", () => { + const tempDirs: string[] = []; + + afterEach(async () => { + await Promise.all( + tempDirs.splice(0).map((dir) => fs.rm(dir, { recursive: true, force: true })), + ); + }); + + it("rejects a second process-local owner until the first lease is released", async () => { + const parent = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-wa-owner-")); + tempDirs.push(parent); + const authDir = path.join(parent, "auth"); + await fs.mkdir(authDir); + + const gatewayOwner = await acquireWhatsAppGatewayConnectionOwner(authDir); + await expect(acquireWhatsAppStandaloneConnectionOwner(authDir)).rejects.toMatchObject({ + code: "whatsapp_connection_owner_busy", + authDir, + }); + + await gatewayOwner.release(); + const standaloneOwner = await acquireWhatsAppStandaloneConnectionOwner(authDir); + await standaloneOwner.release(); + }); + + it("lets the gateway wait for a process-local standalone owner", async () => { + const parent = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-wa-owner-")); + tempDirs.push(parent); + const authDir = path.join(parent, "auth"); + await fs.mkdir(authDir); + + const standaloneOwner = await acquireWhatsAppStandaloneConnectionOwner(authDir); + const gatewayOwnerPromise = acquireWhatsAppGatewayConnectionOwner(authDir); + await standaloneOwner.release(); + + const gatewayOwner = await gatewayOwnerPromise; + await gatewayOwner.release(); + }); + + it.runIf(process.platform !== "win32")( + "treats symlink aliases as the same process-local owner", + async () => { + const parent = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-wa-owner-")); + tempDirs.push(parent); + const authDir = path.join(parent, "auth"); + const authAlias = path.join(parent, "auth-alias"); + await fs.mkdir(authDir); + await fs.symlink(authDir, authAlias, "dir"); + + const gatewayOwner = await acquireWhatsAppGatewayConnectionOwner(authDir); + await expect(acquireWhatsAppStandaloneConnectionOwner(authAlias)).rejects.toMatchObject({ + code: "whatsapp_connection_owner_busy", + authDir: authAlias, + }); + await gatewayOwner.release(); + }, + ); + + it("recovers an unchanged lock owned by a definitely dead process", async () => { + const parent = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-wa-owner-")); + tempDirs.push(parent); + const authDir = path.join(parent, "auth"); + await fs.mkdir(authDir); + await fs.writeFile( + `${authDir}.lock`, + `${JSON.stringify({ pid: 2_147_483_647, createdAt: new Date().toISOString() })}\n`, + ); + + const owner = await acquireWhatsAppStandaloneConnectionOwner(authDir); + await owner.release(); + }); + + it("cancels cross-process owner retries during shutdown", async () => { + const parent = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-wa-owner-")); + tempDirs.push(parent); + const authDir = path.join(parent, "auth"); + await fs.mkdir(authDir); + await fs.writeFile( + `${authDir}.lock`, + `${JSON.stringify({ pid: process.pid, createdAt: new Date().toISOString() })}\n`, + ); + const abortController = new AbortController(); + + const ownerPromise = acquireWhatsAppGatewayConnectionOwner(authDir, abortController.signal); + abortController.abort(new Error("shutdown")); + + await expect(ownerPromise).rejects.toThrow("shutdown"); + }); +}); diff --git a/extensions/whatsapp/src/connection-owner.ts b/extensions/whatsapp/src/connection-owner.ts new file mode 100644 index 000000000000..8237f265bbb6 --- /dev/null +++ b/extensions/whatsapp/src/connection-owner.ts @@ -0,0 +1,216 @@ +import fs from "node:fs/promises"; +// Whatsapp connection-owner lease serializes auth-backed Baileys sockets across processes. +import { + acquireFileLock, + FILE_LOCK_STALE_ERROR_CODE, + FILE_LOCK_TIMEOUT_ERROR_CODE, + type FileLockHandle, +} from "openclaw/plugin-sdk/file-lock"; +import { resolveUserPath } from "openclaw/plugin-sdk/text-utility-runtime"; + +const WHATSAPP_CONNECTION_OWNER_BUSY_CODE = "whatsapp_connection_owner_busy"; + +export class WhatsAppConnectionOwnerBusyError extends Error { + readonly code = WHATSAPP_CONNECTION_OWNER_BUSY_CODE; + + constructor( + public readonly authDir: string, + options?: ErrorOptions, + ) { + super("Another process owns this WhatsApp connection.", options); + this.name = "WhatsAppConnectionOwnerBusyError"; + } +} + +export type WhatsAppConnectionOwnerLease = Pick; + +const OWNER_LOCK_STALE_MS = 5 * 60_000; +const GATEWAY_LOCAL_OWNER_WAIT_MS = 150_000; + +type ProcessOwner = { + released: Promise; + resolveReleased: () => void; + token: symbol; +}; + +const processOwners = new Map(); + +function ownershipCancelledError(signal?: AbortSignal): Error { + const reason = signal?.reason; + return reason instanceof Error + ? reason + : new Error( + "WhatsApp connection ownership cancelled", + reason === undefined ? {} : { cause: reason }, + ); +} + +async function waitForAbortableDelay(delayMs: number, signal?: AbortSignal): Promise { + if (signal?.aborted) { + throw ownershipCancelledError(signal); + } + await new Promise((resolve, reject) => { + const onAbort = () => { + clearTimeout(timer); + reject(ownershipCancelledError(signal)); + }; + const timer = setTimeout(() => { + signal?.removeEventListener("abort", onAbort); + resolve(); + }, delayMs); + signal?.addEventListener("abort", onAbort, { once: true }); + }); +} + +async function reserveProcessOwner(params: { + authDir: string; + ownerPath: string; + signal?: AbortSignal; + waitForLocalOwner: boolean; +}): Promise { + while (true) { + if (params.signal?.aborted) { + throw ownershipCancelledError(params.signal); + } + const current = processOwners.get(params.ownerPath); + if (!current) { + let resolveReleased = () => {}; + const owner: ProcessOwner = { + released: new Promise((resolve) => { + resolveReleased = resolve; + }), + resolveReleased, + token: Symbol(params.ownerPath), + }; + processOwners.set(params.ownerPath, owner); + return owner; + } + if (!params.waitForLocalOwner) { + throw new WhatsAppConnectionOwnerBusyError(params.authDir); + } + let timer: ReturnType | undefined; + let onAbort: (() => void) | undefined; + const outcome = await Promise.race([ + current.released.then(() => "released" as const), + new Promise<"timed_out">((resolve) => { + timer = setTimeout(() => resolve("timed_out"), GATEWAY_LOCAL_OWNER_WAIT_MS); + timer.unref?.(); + }), + new Promise<"aborted">((resolve) => { + onAbort = () => resolve("aborted"); + params.signal?.addEventListener("abort", onAbort, { once: true }); + }), + ]).finally(() => { + if (timer) { + clearTimeout(timer); + } + if (onAbort) { + params.signal?.removeEventListener("abort", onAbort); + } + }); + if (outcome === "aborted") { + throw ownershipCancelledError(params.signal); + } + if (outcome === "timed_out") { + throw new WhatsAppConnectionOwnerBusyError(params.authDir); + } + } +} + +function abandonProcessOwner(ownerPath: string, owner: ProcessOwner): void { + if (processOwners.get(ownerPath)?.token !== owner.token) { + return; + } + processOwners.delete(ownerPath); + owner.resolveReleased(); +} + +async function acquireOwnerLease(params: { + authDir: string; + retries: number; + signal?: AbortSignal; + waitForLocalOwner: boolean; +}): Promise { + const resolvedOwnerPath = resolveUserPath(params.authDir); + await fs.mkdir(resolvedOwnerPath, { recursive: true }); + const ownerPath = await fs.realpath(resolvedOwnerPath); + // Reserve before awaiting the filesystem so concurrent callers cannot use the + // underlying file lock's intentionally re-entrant mode for two sockets. + const processOwner = await reserveProcessOwner({ + authDir: params.authDir, + ownerPath, + signal: params.signal, + waitForLocalOwner: params.waitForLocalOwner, + }); + let fileLock: FileLockHandle; + let attempt = 0; + while (true) { + if (params.signal?.aborted) { + abandonProcessOwner(ownerPath, processOwner); + throw ownershipCancelledError(params.signal); + } + try { + fileLock = await acquireFileLock(ownerPath, { + retries: { retries: 0, factor: 1, minTimeout: 1, maxTimeout: 1 }, + stale: OWNER_LOCK_STALE_MS, + // The shared lock wrapper reclaims only a definitely dead PID and removes + // the exact unchanged sidecar. Live or ambiguous owners remain fail-closed. + staleRecovery: "remove-if-unchanged", + }); + break; + } catch (error) { + const code = (error as { code?: unknown }).code; + if (code === FILE_LOCK_STALE_ERROR_CODE) { + abandonProcessOwner(ownerPath, processOwner); + throw new WhatsAppConnectionOwnerBusyError(params.authDir, { cause: error }); + } + if (code !== FILE_LOCK_TIMEOUT_ERROR_CODE || attempt >= params.retries) { + abandonProcessOwner(ownerPath, processOwner); + if (code === FILE_LOCK_TIMEOUT_ERROR_CODE) { + throw new WhatsAppConnectionOwnerBusyError(params.authDir, { cause: error }); + } + throw error; + } + const delayMs = Math.min(100 * 1.5 ** attempt, 1_000); + attempt += 1; + await waitForAbortableDelay(delayMs, params.signal).catch((delayError: unknown) => { + abandonProcessOwner(ownerPath, processOwner); + throw delayError; + }); + } + } + let releasePromise: Promise | null = null; + return { + release: async () => { + if (!releasePromise) { + releasePromise = fileLock + .release() + .then(() => { + abandonProcessOwner(ownerPath, processOwner); + }) + .catch((releaseError: unknown) => { + releasePromise = null; + throw releaseError; + }); + } + await releasePromise; + }, + }; +} + +/** Gateway owner waits for a bounded standalone lookup to finish before startup. */ +export async function acquireWhatsAppGatewayConnectionOwner( + authDir: string, + signal?: AbortSignal, +): Promise { + // Gateway lifecycle stops an account before restarting it. A timed-out incumbent + // must keep same-auth restarts blocked; handoff would permit concurrent sockets. + return await acquireOwnerLease({ authDir, retries: 150, signal, waitForLocalOwner: true }); +} + +/** Standalone lookup fails quickly when a gateway already owns the account. */ +export async function acquireWhatsAppStandaloneConnectionOwner( + authDir: string, +): Promise { + return await acquireOwnerLease({ authDir, retries: 3, waitForLocalOwner: false }); +} diff --git a/extensions/whatsapp/src/directory-config.test.ts b/extensions/whatsapp/src/directory-config.test.ts index 534c761aab93..aa54872581f0 100644 --- a/extensions/whatsapp/src/directory-config.test.ts +++ b/extensions/whatsapp/src/directory-config.test.ts @@ -1,33 +1,126 @@ // Whatsapp tests cover directory config plugin behavior. import { createDirectoryTestRuntime } from "openclaw/plugin-sdk/channel-test-helpers"; -import { describe, expect, it } from "vitest"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { readWebAuthExistsForDecision } from "./auth-store.js"; +import { getWhatsAppConnectionController } from "./connection-controller-runtime-context.js"; import { + acquireWhatsAppStandaloneConnectionOwner, + WhatsAppConnectionOwnerBusyError, +} from "./connection-owner.js"; +import { + listWhatsAppDirectoryGroupsLive, listWhatsAppDirectoryGroupsFromConfig, listWhatsAppDirectoryPeersFromConfig, } from "./directory-config.js"; import type { OpenClawConfig } from "./runtime-api.js"; +import { + createWaDirectorySocket, + waitForCredsSaveQueueWithTimeout, + waitForWaConnection, +} from "./session.js"; + +const mocks = vi.hoisted(() => ({ + acquireOwner: vi.fn(), + createSocket: vi.fn(), + getController: vi.fn(), + hasPendingOwner: vi.fn(), + readAuth: vi.fn(), + releaseOwner: vi.fn(), + resolveAuthDir: vi.fn(), + waitForConnection: vi.fn(), + waitForCreds: vi.fn(), +})); + +vi.mock("./active-listener.js", () => ({ + resolveWebAccountId: () => "default", +})); + +vi.mock("./accounts.js", () => ({ + resolveWhatsAppAuthDir: mocks.resolveAuthDir, +})); + +vi.mock("./connection-controller-runtime-context.js", () => ({ + getWhatsAppConnectionController: mocks.getController, + hasPendingWhatsAppConnectionOwner: mocks.hasPendingOwner, +})); + +vi.mock("./connection-owner.js", async (importOriginal) => { + const actual = await importOriginal(); + return { + ...actual, + acquireWhatsAppStandaloneConnectionOwner: mocks.acquireOwner, + }; +}); + +vi.mock("./auth-store.js", async (importOriginal) => { + const actual = await importOriginal(); + return { + ...actual, + readWebAuthExistsForDecision: mocks.readAuth, + }; +}); + +vi.mock("./session.js", () => ({ + createWaDirectorySocket: mocks.createSocket, + waitForCredsSaveQueueWithTimeout: mocks.waitForCreds, + waitForWaConnection: mocks.waitForConnection, +})); + +const getControllerMock = vi.mocked(getWhatsAppConnectionController); +const acquireOwnerMock = vi.mocked(acquireWhatsAppStandaloneConnectionOwner); +const readAuthMock = vi.mocked(readWebAuthExistsForDecision); +const createSocketMock = vi.mocked(createWaDirectorySocket); +const waitForConnectionMock = vi.mocked(waitForWaConnection); +const waitForCredsMock = vi.mocked(waitForCredsSaveQueueWithTimeout); describe("whatsapp directory", () => { const runtimeEnv = createDirectoryTestRuntime() as never; - - it("lists peers and groups from config", async () => { - const cfg = { - channels: { - whatsapp: { - authDir: "/tmp/wa-auth", - allowFrom: [ - "whatsapp:+15551230001", - "15551230002@s.whatsapp.net", - "120363999999999999@g.us", - ], - groups: { - "120363111111111111@g.us": {}, - "120363222222222222@g.us": {}, - }, + const cfg = { + channels: { + whatsapp: { + authDir: "/tmp/wa-auth", + allowFrom: [ + "whatsapp:+15551230001", + "15551230002@s.whatsapp.net", + "120363999999999999@g.us", + ], + groups: { + "120363111111111111@g.us": {}, + "120363222222222222@g.us": {}, }, }, - } as unknown as OpenClawConfig; + }, + } as unknown as OpenClawConfig; + const makeParams = (overrides: { query?: string; limit?: number } = {}) => + ({ + cfg, + accountId: undefined, + query: overrides.query, + limit: overrides.limit, + runtime: runtimeEnv, + }) as never; + + beforeEach(() => { + vi.clearAllMocks(); + getControllerMock.mockReturnValue(null); + mocks.hasPendingOwner.mockReturnValue(false); + mocks.releaseOwner.mockResolvedValue(undefined); + mocks.resolveAuthDir.mockImplementation(({ accountId }: { accountId: string }) => ({ + authDir: accountId === "secondary" ? "/tmp/secondary-wa-auth" : "/tmp/wa-auth", + isLegacy: false, + })); + acquireOwnerMock.mockResolvedValue({ release: mocks.releaseOwner }); + readAuthMock.mockResolvedValue({ outcome: "stable", exists: true }); + waitForConnectionMock.mockResolvedValue(undefined); + waitForCredsMock.mockResolvedValue("drained"); + }); + + afterEach(() => { + vi.useRealTimers(); + }); + + it("lists peers and groups from config", async () => { await expect( listWhatsAppDirectoryPeersFromConfig({ cfg, @@ -54,4 +147,190 @@ describe("whatsapp directory", () => { { kind: "group", id: "120363222222222222@g.us" }, ]); }); + + it("uses the active gateway owner and applies deterministic filtering", async () => { + const sock = { + groupFetchAllParticipating: vi.fn().mockResolvedValue({ + "120363300000000000@g.us": { id: "120363300000000000@g.us", subject: "Beta" }, + "120363100000000000@g.us": { id: "120363100000000000@g.us", subject: "Beta Team" }, + }), + }; + getControllerMock.mockReturnValue({ getCurrentSock: () => sock } as never); + + await expect( + listWhatsAppDirectoryGroupsLive(makeParams({ query: "beta", limit: 1 })), + ).resolves.toEqual([{ kind: "group", id: "120363100000000000@g.us", name: "Beta Team" }]); + expect(acquireOwnerMock).not.toHaveBeenCalled(); + }); + + it("reports a typed error when the gateway owner has no active socket", async () => { + getControllerMock.mockReturnValue({ getCurrentSock: () => null } as never); + + await expect(listWhatsAppDirectoryGroupsLive(makeParams())).rejects.toMatchObject({ + code: "whatsapp_directory_unavailable", + reason: "active_owner_unavailable", + }); + expect(acquireOwnerMock).not.toHaveBeenCalled(); + }); + + it("does not start a standalone socket while the gateway owner is connecting", async () => { + mocks.hasPendingOwner.mockReturnValueOnce(true); + + await expect(listWhatsAppDirectoryGroupsLive(makeParams())).rejects.toMatchObject({ + code: "whatsapp_directory_unavailable", + reason: "active_owner_unavailable", + }); + expect(acquireOwnerMock).not.toHaveBeenCalled(); + }); + + it("does not substitute configured groups when the gateway lookup fails", async () => { + getControllerMock.mockReturnValue({ + getCurrentSock: () => ({ + groupFetchAllParticipating: vi.fn().mockRejectedValue(new Error("query failed")), + }), + } as never); + + await expect(listWhatsAppDirectoryGroupsLive(makeParams())).rejects.toMatchObject({ + code: "whatsapp_directory_unavailable", + reason: "lookup_failed", + }); + }); + + it("runs a standalone lookup under exclusive ownership and awaits cleanup", async () => { + const order: string[] = []; + let closed = false; + const sock = { + groupFetchAllParticipating: vi.fn().mockResolvedValue({ + "120363300000000000@g.us": { id: "120363300000000000@g.us", subject: "Three" }, + }), + end: vi.fn(async () => { + closed = true; + order.push("socket-ended"); + }), + ws: { + close: vi.fn(async () => { + closed = true; + }), + get isClosed() { + return closed; + }, + }, + }; + createSocketMock.mockResolvedValue(sock as never); + waitForCredsMock.mockImplementationOnce(async () => { + order.push("creds-drained"); + return "drained"; + }); + mocks.releaseOwner.mockImplementationOnce(async () => { + order.push("owner-released"); + }); + + await expect(listWhatsAppDirectoryGroupsLive(makeParams())).resolves.toEqual([ + { kind: "group", id: "120363300000000000@g.us", name: "Three" }, + ]); + expect(acquireOwnerMock).toHaveBeenCalledWith("/tmp/wa-auth"); + expect(waitForConnectionMock).toHaveBeenCalledWith(sock, { timeoutMs: 30_000 }); + expect(order).toEqual(["socket-ended", "creds-drained", "owner-released"]); + }); + + it("reports connection-owner contention without opening a socket", async () => { + acquireOwnerMock.mockRejectedValueOnce(new WhatsAppConnectionOwnerBusyError("/tmp/wa-auth")); + + await expect(listWhatsAppDirectoryGroupsLive(makeParams())).rejects.toMatchObject({ + code: "whatsapp_directory_unavailable", + reason: "connection_owner_busy", + }); + expect(createSocketMock).not.toHaveBeenCalled(); + }); + + it("reports unlinked auth and releases standalone ownership", async () => { + readAuthMock.mockResolvedValueOnce({ outcome: "stable", exists: false }); + + await expect(listWhatsAppDirectoryGroupsLive(makeParams())).rejects.toMatchObject({ + code: "whatsapp_directory_unavailable", + reason: "not_linked", + }); + expect(mocks.releaseOwner).toHaveBeenCalledOnce(); + }); + + it("resolves standalone credentials for the selected account", async () => { + readAuthMock.mockResolvedValueOnce({ outcome: "stable", exists: false }); + const namedCfg = { + channels: { whatsapp: { accounts: { secondary: { enabled: true } } } }, + } as unknown as OpenClawConfig; + + await expect( + listWhatsAppDirectoryGroupsLive({ + cfg: namedCfg, + accountId: "secondary", + query: undefined, + limit: undefined, + runtime: runtimeEnv, + } as never), + ).rejects.toMatchObject({ reason: "not_linked" }); + + expect(mocks.resolveAuthDir).toHaveBeenCalledWith({ + cfg: namedCfg, + accountId: "secondary", + }); + expect(acquireOwnerMock).toHaveBeenCalledWith("/tmp/secondary-wa-auth"); + }); + + it("closes a created socket when standalone connection setup fails", async () => { + let closed = false; + const sock = { + groupFetchAllParticipating: vi.fn(), + end: vi.fn(async () => { + closed = true; + }), + ws: { + close: vi.fn(async () => { + closed = true; + }), + get isClosed() { + return closed; + }, + }, + }; + createSocketMock.mockResolvedValueOnce(sock as never); + waitForConnectionMock.mockRejectedValueOnce(new Error("offline")); + + await expect(listWhatsAppDirectoryGroupsLive(makeParams())).rejects.toMatchObject({ + code: "whatsapp_directory_unavailable", + reason: "connection_failed", + }); + expect(sock.end).toHaveBeenCalledOnce(); + expect(waitForCredsMock).toHaveBeenCalledWith("/tmp/wa-auth"); + expect(mocks.releaseOwner).toHaveBeenCalledOnce(); + }); + + it("retries retained ownership after credential cleanup times out", async () => { + vi.useFakeTimers(); + let closed = false; + const sock = { + groupFetchAllParticipating: vi.fn().mockResolvedValue({}), + end: vi.fn(async () => { + closed = true; + }), + ws: { + close: vi.fn(async () => { + closed = true; + }), + get isClosed() { + return closed; + }, + }, + }; + createSocketMock.mockResolvedValueOnce(sock as never); + waitForCredsMock.mockResolvedValueOnce("timed_out"); + + await expect(listWhatsAppDirectoryGroupsLive(makeParams())).rejects.toMatchObject({ + code: "whatsapp_directory_unavailable", + reason: "cleanup_failed", + }); + expect(mocks.releaseOwner).not.toHaveBeenCalled(); + + await vi.advanceTimersByTimeAsync(1_000); + expect(mocks.releaseOwner).toHaveBeenCalledOnce(); + }); }); diff --git a/extensions/whatsapp/src/directory-config.ts b/extensions/whatsapp/src/directory-config.ts index 2afab47aac37..28263b6a7f13 100644 --- a/extensions/whatsapp/src/directory-config.ts +++ b/extensions/whatsapp/src/directory-config.ts @@ -2,11 +2,30 @@ import { listResolvedDirectoryGroupEntriesFromMapKeys, listResolvedDirectoryUserEntriesFromAllowFrom, + type ChannelDirectoryEntry, type DirectoryConfigParams, } from "openclaw/plugin-sdk/directory-config-runtime"; import { resolveMergedWhatsAppAccountConfig } from "./account-config.js"; import type { WhatsAppAccountConfig } from "./account-types.js"; +import { resolveWhatsAppAuthDir } from "./accounts.js"; +import { resolveWebAccountId } from "./active-listener.js"; +import { readWebAuthExistsForDecision } from "./auth-store.js"; +import { + getWhatsAppConnectionController, + hasPendingWhatsAppConnectionOwner, +} from "./connection-controller-runtime-context.js"; +import { + acquireWhatsAppStandaloneConnectionOwner, + WhatsAppConnectionOwnerBusyError, + type WhatsAppConnectionOwnerLease, +} from "./connection-owner.js"; import { isWhatsAppGroupJid, normalizeWhatsAppTarget } from "./normalize.js"; +import { + createWaDirectorySocket, + waitForCredsSaveQueueWithTimeout, + waitForWaConnection, +} from "./session.js"; +import { closeWhatsAppSocketAndWait } from "./socket-close.js"; type WhatsAppDirectoryAccount = WhatsAppAccountConfig & { accountId: string }; @@ -39,3 +58,284 @@ export async function listWhatsAppDirectoryGroupsFromConfig(params: DirectoryCon resolveGroups: (account) => account.groups, }); } + +export const WHATSAPP_DIRECTORY_UNAVAILABLE_CODE = "whatsapp_directory_unavailable"; + +export type WhatsAppDirectoryUnavailableReason = + | "active_owner_unavailable" + | "connection_owner_busy" + | "not_linked" + | "auth_unstable" + | "connection_failed" + | "lookup_failed" + | "cleanup_failed"; + +export class WhatsAppDirectoryUnavailableError extends Error { + readonly code = WHATSAPP_DIRECTORY_UNAVAILABLE_CODE; + + constructor( + public readonly reason: WhatsAppDirectoryUnavailableReason, + message: string, + options?: ErrorOptions, + ) { + super(message, options); + this.name = "WhatsAppDirectoryUnavailableError"; + } +} + +type GroupFetchSocket = { + groupFetchAllParticipating(): Promise< + Record + >; +}; + +async function fetchLiveGroups( + sock: GroupFetchSocket, + params: DirectoryConfigParams, +): Promise { + const groups = await sock.groupFetchAllParticipating(); + const query = params.query?.trim().toLowerCase() ?? ""; + const limit = typeof params.limit === "number" && params.limit > 0 ? params.limit : undefined; + const entries = Object.entries(groups) + .map(([jid, metadata]) => ({ + kind: "group" as const, + id: jid, + name: metadata?.subject?.trim() || undefined, + })) + .filter((entry) => { + if (!query) { + return true; + } + return entry.id.toLowerCase().includes(query) || entry.name?.toLowerCase().includes(query); + }) + .toSorted((left, right) => left.id.localeCompare(right.id)); + return limit ? entries.slice(0, limit) : entries; +} + +function unavailable( + reason: WhatsAppDirectoryUnavailableReason, + message: string, + cause?: unknown, +): WhatsAppDirectoryUnavailableError { + return new WhatsAppDirectoryUnavailableError(reason, message, cause ? { cause } : undefined); +} + +type StandaloneSocket = Awaited>; + +type ManagedStandaloneCleanup = { + authDir: string; + inFlight: Promise | null; + ownerLease: WhatsAppConnectionOwnerLease; + retryTimer: ReturnType | null; + sock: StandaloneSocket | null; + socketClosed: boolean; +}; + +const pendingStandaloneCleanups = new Map(); +const STANDALONE_CLEANUP_RETRY_MS = 1_000; + +async function completeStandaloneCleanup(cleanup: ManagedStandaloneCleanup): Promise { + if (cleanup.sock && !cleanup.socketClosed) { + await closeWhatsAppSocketAndWait( + cleanup.sock, + "OpenClaw WhatsApp standalone directory socket close", + ); + cleanup.socketClosed = true; + } + if (cleanup.sock) { + const queueResult = await waitForCredsSaveQueueWithTimeout(cleanup.authDir); + if (queueResult === "timed_out") { + throw new Error("WhatsApp credential persistence did not drain before socket release"); + } + } + await cleanup.ownerLease.release(); + if (pendingStandaloneCleanups.get(cleanup.authDir) === cleanup) { + pendingStandaloneCleanups.delete(cleanup.authDir); + } + if (cleanup.retryTimer) { + clearTimeout(cleanup.retryTimer); + cleanup.retryTimer = null; + } +} + +function runStandaloneCleanup(cleanup: ManagedStandaloneCleanup): Promise { + if (cleanup.inFlight) { + return cleanup.inFlight; + } + const task = completeStandaloneCleanup(cleanup).finally(() => { + if (cleanup.inFlight === task) { + cleanup.inFlight = null; + } + }); + cleanup.inFlight = task; + return task; +} + +function scheduleStandaloneCleanupRetry(cleanup: ManagedStandaloneCleanup): void { + if (cleanup.retryTimer) { + return; + } + cleanup.retryTimer = setTimeout(() => { + cleanup.retryTimer = null; + void runStandaloneCleanup(cleanup).catch(() => { + scheduleStandaloneCleanupRetry(cleanup); + }); + }, STANDALONE_CLEANUP_RETRY_MS); + // Gateway processes stay alive and retry; standalone CLI processes may exit. + // A later process safely reclaims the unchanged lock from the definitely dead PID. + cleanup.retryTimer.unref?.(); +} + +function retainStandaloneCleanup(cleanup: ManagedStandaloneCleanup): void { + pendingStandaloneCleanups.set(cleanup.authDir, cleanup); + scheduleStandaloneCleanupRetry(cleanup); +} + +async function finishStandaloneCleanupOrThrow( + cleanup: ManagedStandaloneCleanup, + operationError?: unknown, +): Promise { + try { + await runStandaloneCleanup(cleanup); + } catch (cleanupError) { + retainStandaloneCleanup(cleanup); + const cause = + operationError === undefined + ? cleanupError + : new AggregateError( + [operationError, cleanupError], + "WhatsApp live group lookup and cleanup failed", + { cause: operationError }, + ); + throw cleanupUnavailable(cause); + } +} + +function cleanupUnavailable(error: unknown): WhatsAppDirectoryUnavailableError { + return unavailable( + "cleanup_failed", + "WhatsApp live group lookup could not safely close its standalone connection.", + error, + ); +} + +async function finishPriorStandaloneCleanup(authDir: string): Promise { + const cleanup = pendingStandaloneCleanups.get(authDir); + if (!cleanup) { + return; + } + try { + await runStandaloneCleanup(cleanup); + } catch (error) { + scheduleStandaloneCleanupRetry(cleanup); + throw cleanupUnavailable(error); + } +} + +async function listGroupsThroughStandaloneOwner( + params: DirectoryConfigParams, +): Promise { + const account = resolveWhatsAppDirectoryAccount(params.cfg, params.accountId); + const authDir = resolveWhatsAppAuthDir({ + cfg: params.cfg, + accountId: account.accountId, + }).authDir; + await finishPriorStandaloneCleanup(authDir); + let ownerLease: WhatsAppConnectionOwnerLease; + try { + ownerLease = await acquireWhatsAppStandaloneConnectionOwner(authDir); + } catch (error) { + if (error instanceof WhatsAppConnectionOwnerBusyError) { + throw unavailable( + "connection_owner_busy", + "WhatsApp live groups are unavailable because the account is owned by another process.", + error, + ); + } + throw unavailable( + "connection_failed", + "WhatsApp live groups are unavailable because connection ownership failed.", + error, + ); + } + + const cleanup: ManagedStandaloneCleanup = { + authDir, + inFlight: null, + ownerLease, + retryTimer: null, + sock: null, + socketClosed: true, + }; + let groups: ChannelDirectoryEntry[]; + try { + let authState: Awaited>; + try { + authState = await readWebAuthExistsForDecision(authDir); + } catch (error) { + throw unavailable( + "auth_unstable", + "WhatsApp live groups are unavailable because linked credentials could not be read.", + error, + ); + } + if (authState.outcome === "unstable") { + throw unavailable( + "auth_unstable", + "WhatsApp live groups are unavailable while linked credentials are changing.", + ); + } + if (!authState.exists) { + throw unavailable( + "not_linked", + "WhatsApp live groups are unavailable because this account is not linked.", + ); + } + + try { + cleanup.sock = await createWaDirectorySocket(authDir); + cleanup.socketClosed = false; + await waitForWaConnection(cleanup.sock, { timeoutMs: 30_000 }); + } catch (error) { + throw unavailable( + "connection_failed", + "WhatsApp live groups are unavailable because the standalone connection failed.", + error, + ); + } + + try { + groups = await fetchLiveGroups(cleanup.sock, params); + } catch (error) { + throw unavailable("lookup_failed", "WhatsApp live group lookup failed.", error); + } + } catch (error) { + await finishStandaloneCleanupOrThrow(cleanup, error); + throw error; + } + await finishStandaloneCleanupOrThrow(cleanup); + return groups; +} + +export async function listWhatsAppDirectoryGroupsLive( + params: DirectoryConfigParams, +): Promise { + const accountId = resolveWebAccountId({ cfg: params.cfg, accountId: params.accountId }); + const controller = getWhatsAppConnectionController(accountId); + if (!controller && !hasPendingWhatsAppConnectionOwner(accountId)) { + return await listGroupsThroughStandaloneOwner(params); + } + + const sock = controller?.getCurrentSock(); + if (!sock) { + throw unavailable( + "active_owner_unavailable", + "WhatsApp live groups are unavailable while the gateway connection is offline.", + ); + } + try { + return await fetchLiveGroups(sock, params); + } catch (error) { + throw unavailable("lookup_failed", "WhatsApp live group lookup failed.", error); + } +} diff --git a/extensions/whatsapp/src/session.test.ts b/extensions/whatsapp/src/session.test.ts index 7c94e36f3756..9eb9335ecdfd 100644 --- a/extensions/whatsapp/src/session.test.ts +++ b/extensions/whatsapp/src/session.test.ts @@ -40,6 +40,7 @@ vi.mock("undici", async () => { const useMultiFileAuthStateMock = vi.mocked(baileys.useMultiFileAuthState); let createWaSocket: typeof import("./session.js").createWaSocket; +let createWaDirectorySocket: typeof import("./session.js").createWaDirectorySocket; let formatError: typeof import("./session.js").formatError; const OPENCLAW_WHATSAPP_WEB_SOCKET_URL_ENV = "OPENCLAW_WHATSAPP_WEB_SOCKET_URL"; let renderQrTerminalMock: ReturnType; @@ -152,6 +153,7 @@ function readLastSocketOptions(): { connectTimeoutMs?: number; defaultQueryTimeoutMs?: number; fetchAgent?: unknown; + fireInitQueries?: boolean; keepAliveIntervalMs?: number; printQRInTerminal?: boolean; waWebSocketUrl?: string | URL; @@ -169,6 +171,7 @@ function readLastSocketOptions(): { connectTimeoutMs?: number; defaultQueryTimeoutMs?: number; fetchAgent?: unknown; + fireInitQueries?: boolean; keepAliveIntervalMs?: number; printQRInTerminal?: boolean; waWebSocketUrl?: string | URL; @@ -224,6 +227,7 @@ function installUndiciRuntimeDeps(): void { describe("web session", () => { beforeAll(async () => { ({ + createWaDirectorySocket, createWaSocket, formatError, waitForWaConnection, @@ -259,6 +263,7 @@ describe("web session", () => { await createWaSocket(true, false, { authDir }); const passed = readLastSocketOptions(); expect(passed.printQRInTerminal).toBe(false); + expect(passed.fireInitQueries).toBe(true); expect(passed.keepAliveIntervalMs).toBe(DEFAULT_WHATSAPP_SOCKET_TIMING.keepAliveIntervalMs); expect(passed.connectTimeoutMs).toBe(DEFAULT_WHATSAPP_SOCKET_TIMING.connectTimeoutMs); expect(passed.defaultQueryTimeoutMs).toBe(DEFAULT_WHATSAPP_SOCKET_TIMING.defaultQueryTimeoutMs); @@ -278,6 +283,48 @@ describe("web session", () => { openMock.restore(); }); + it("creates standalone directory sockets without inbound message consumers", async () => { + const authDir = createTempAuthDir("openclaw-wa-directory-socket"); + const ws = new EventEmitter() as EventEmitter & { close: ReturnType }; + ws.close = vi.fn(); + ws.on("CB:message", vi.fn()); + ws.on("CB:call", vi.fn()); + ws.on("CB:receipt", vi.fn()); + ws.on("CB:notification", vi.fn()); + ws.on("CB:ack,class:message", vi.fn()); + ws.on("CB:presence", vi.fn()); + ws.on("CB:chatstate", vi.fn()); + ws.on("CB:ib,,dirty", vi.fn()); + ws.on("CB:ib,,offline_preview", vi.fn()); + ws.on("CB:ib,,offline", vi.fn()); + ws.on("CB:ib,,edge_routing", vi.fn()); + const sock = { + ev: new EventEmitter(), + ws, + groupFetchAllParticipating: vi.fn().mockResolvedValue({}), + }; + vi.mocked(baileys.makeWASocket).mockReturnValueOnce(sock as never); + + await createWaDirectorySocket(authDir); + + expect(readLastSocketOptions().fireInitQueries).toBe(false); + for (const event of [ + "CB:message", + "CB:call", + "CB:receipt", + "CB:notification", + "CB:ack,class:message", + "CB:presence", + "CB:chatstate", + "CB:ib,,dirty", + "CB:ib,,offline_preview", + "CB:ib,,offline", + "CB:ib,,edge_routing", + ]) { + expect(ws.listenerCount(event), event).toBe(0); + } + }); + it("prints compact terminal QR output when requested", async () => { const authDir = createTempAuthDir("openclaw-wa-terminal-qr"); const logSpy = vi.spyOn(console, "log").mockImplementation(() => {}); diff --git a/extensions/whatsapp/src/session.ts b/extensions/whatsapp/src/session.ts index 61182d06ccd9..18427c8d9531 100644 --- a/extensions/whatsapp/src/session.ts +++ b/extensions/whatsapp/src/session.ts @@ -212,6 +212,24 @@ export async function createWaSocket( cachedGroupMetadata?: (jid: string) => Promise; waWebSocketUrl?: string | URL; } & WhatsAppSocketTimingOptions = {}, +): Promise> { + return await createWaSocketInternal(printQr, verbose, opts, "normal"); +} + +async function createWaSocketInternal( + printQr: boolean, + verbose: boolean, + opts: { + authDir?: string; + onQr?: (qr: string) => void; + beforeCredentialPersistence?: () => Promise; + onCredentialPersistenceError?: (error: unknown) => void; + onCredentialPersistenceTask?: (task: Promise) => void; + getMessage?: (key: WAMessageKey) => Promise; + cachedGroupMetadata?: (jid: string) => Promise; + waWebSocketUrl?: string | URL; + } & WhatsAppSocketTimingOptions, + receiveMode: "normal" | "directory", ): Promise> { const baseLogger = getChildLogger( { module: "baileys" }, @@ -324,6 +342,7 @@ export async function createWaSocket( printQRInTerminal: false, browser: ["openclaw", "cli", VERSION], syncFullHistory: false, + fireInitQueries: receiveMode !== "directory", markOnlineOnConnect: false, ...socketTiming, agent, @@ -335,6 +354,25 @@ export async function createWaSocket( ...(opts.getMessage ? { getMessage: opts.getMessage } : {}), ...(opts.cachedGroupMetadata ? { cachedGroupMetadata: opts.cachedGroupMetadata } : {}), }); + if (receiveMode === "directory") { + // A standalone directory lookup must not consume, acknowledge, or react to user + // traffic. Keep only Baileys connection/query machinery for the group IQ request. + for (const event of [ + "CB:message", + "CB:call", + "CB:receipt", + "CB:notification", + "CB:ack,class:message", + "CB:presence", + "CB:chatstate", + "CB:ib,,dirty", + "CB:ib,,offline_preview", + "CB:ib,,offline", + "CB:ib,,edge_routing", + ]) { + sock.ws.removeAllListeners(event); + } + } socketRef.current = sock; if (pendingSocketAbort) { abortSocketAfterCredentialPersistenceFailure(sock, pendingSocketAbort.error); @@ -388,6 +426,12 @@ export async function createWaSocket( return sock; } +export async function createWaDirectorySocket( + authDir: string, +): Promise> { + return await createWaSocketInternal(false, false, { authDir }, "directory"); +} + async function resolveEnvProxyAgent( logger: ReturnType, ): Promise { diff --git a/extensions/whatsapp/src/socket-close.test.ts b/extensions/whatsapp/src/socket-close.test.ts new file mode 100644 index 000000000000..681058e61b18 --- /dev/null +++ b/extensions/whatsapp/src/socket-close.test.ts @@ -0,0 +1,96 @@ +// Whatsapp tests cover transport-confirmed Baileys socket shutdown. +import { EventEmitter } from "node:events"; +import { describe, expect, it, vi } from "vitest"; +import { closeWhatsAppSocketAndWait } from "./socket-close.js"; + +describe("closeWhatsAppSocketAndWait", () => { + it("closes the underlying transport when Baileys end resolves while it remains open", async () => { + let closed = false; + const sock = { + end: vi.fn().mockResolvedValue(undefined), + ws: { + close: vi.fn(async () => { + closed = true; + }), + get isClosed() { + return closed; + }, + }, + }; + + await closeWhatsAppSocketAndWait(sock as never, "test close"); + + expect(sock.end).toHaveBeenCalledOnce(); + expect(sock.ws.close).toHaveBeenCalledOnce(); + expect(sock.ws.isClosed).toBe(true); + }); + + it("waits for the asynchronous WebSocket close handshake", async () => { + let closed = false; + let closing = false; + const ws = new EventEmitter() as EventEmitter & { + close: ReturnType; + readonly isClosed: boolean; + readonly isClosing: boolean; + }; + Object.defineProperties(ws, { + isClosed: { get: () => closed }, + isClosing: { get: () => closing }, + }); + ws.close = vi.fn(() => { + closing = true; + setImmediate(() => { + closed = true; + closing = false; + ws.emit("close"); + }); + }); + const sock = { end: vi.fn(), ws }; + + await closeWhatsAppSocketAndWait(sock as never, "test close"); + + expect(sock.ws.close).toHaveBeenCalledOnce(); + expect(sock.ws.isClosed).toBe(true); + }); + + it("bounds Baileys teardown after the transport is already closed", async () => { + vi.useFakeTimers(); + const ws = new EventEmitter() as EventEmitter & { + close: ReturnType; + isClosed: boolean; + isClosing: boolean; + }; + ws.close = vi.fn(); + ws.isClosed = true; + ws.isClosing = false; + const sock = { end: vi.fn(() => new Promise(() => {})), ws }; + + try { + const closeTask = closeWhatsAppSocketAndWait(sock as never, "test close"); + await vi.advanceTimersByTimeAsync(15_000); + await expect(closeTask).resolves.toBeUndefined(); + expect(ws.close).not.toHaveBeenCalled(); + } finally { + vi.useRealTimers(); + } + }); + + it("rejects when neither Baileys nor the transport confirms closure", async () => { + const ws = new EventEmitter() as EventEmitter & { + close: ReturnType; + isClosed: boolean; + isClosing: boolean; + }; + ws.close = vi.fn().mockRejectedValue(new Error("close failed")); + ws.isClosed = false; + ws.isClosing = false; + const sock = { + end: vi.fn().mockResolvedValue(undefined), + ws, + }; + + await expect(closeWhatsAppSocketAndWait(sock as never, "test close")).rejects.toThrow( + "socket close could not be confirmed", + ); + }); +}); diff --git a/extensions/whatsapp/src/socket-close.ts b/extensions/whatsapp/src/socket-close.ts new file mode 100644 index 000000000000..394a809f905d --- /dev/null +++ b/extensions/whatsapp/src/socket-close.ts @@ -0,0 +1,80 @@ +// Whatsapp socket shutdown confirms the underlying Baileys transport is closed. +import type { WASocket } from "baileys"; + +const SOCKET_CLOSE_TIMEOUT_MS = 15_000; + +async function withCloseTimeout(task: Promise, operationName: string): Promise { + let timer: ReturnType | undefined; + await Promise.race([ + task, + new Promise((_resolve, reject) => { + timer = setTimeout( + () => reject(new Error(`WhatsApp ${operationName} timed out`)), + SOCKET_CLOSE_TIMEOUT_MS, + ); + }), + ]).finally(() => { + if (timer) { + clearTimeout(timer); + } + }); +} + +async function waitForTransportClose( + ws: Pick, + operation: unknown, + operationName: string, +): Promise { + let onClose: (() => void) | undefined; + const closed = ws.isClosed + ? Promise.resolve() + : new Promise((resolve) => { + onClose = resolve; + ws.once("close", onClose); + if (ws.isClosed) { + onClose(); + } + }); + try { + await withCloseTimeout(Promise.all([Promise.resolve(operation), closed]), operationName); + } finally { + if (onClose) { + ws.removeListener("close", onClose); + } + } +} + +/** Close Baileys and verify the transport state before connection ownership moves. */ +export async function closeWhatsAppSocketAndWait( + sock: Pick, + reason: string, +): Promise { + const errors: unknown[] = []; + try { + const endResult: unknown = sock.end(new Error(reason)); + if (sock.ws.isClosed) { + await waitForTransportClose(sock.ws, endResult, "socket end"); + return; + } + if (sock.ws.isClosing) { + await waitForTransportClose(sock.ws, endResult, "socket end"); + } else { + await withCloseTimeout(Promise.resolve(endResult), "socket end"); + } + } catch (error) { + errors.push(error); + } + if (sock.ws.isClosed) { + return; + } + try { + const closeResult: unknown = sock.ws.close(); + await waitForTransportClose(sock.ws, closeResult, "WebSocket close"); + } catch (error) { + errors.push(error); + } + if (sock.ws.isClosed) { + return; + } + throw new AggregateError(errors, "WhatsApp socket close could not be confirmed"); +}