// Whatsapp plugin module implements monitor inbox.streams inbound messages support behavior.
import fsSync from "node:fs";
import path from "node:path";
import "./monitor-inbox.test-harness.js";
import { beforeEach, describe, expect, it, vi } from "vitest";
import {
  registerWhatsAppConnectionController,
  unregisterWhatsAppConnectionController,
} from "./connection-controller-registry.js";
import { WhatsAppRetryableInboundError } from "./inbound/dedupe.js";
import { WHATSAPP_GROUP_METADATA_CACHE_MAX_ENTRIES } from "./inbound/monitor.js";
import type { WebInboundMessage } from "./inbound/types.js";
import {
  type InboxMonitorOptions,
  buildNotifyMessageUpsert,
  DEFAULT_ACCOUNT_ID,
  failNextWhatsAppPluginStateRegisterIfAbsent,
  getAuthDir,
  getSock,
  installWebMonitorInboxUnitTestHooks,
  resetWebInboundDedupeForTests,
  settleInboundWork,
  startInboxMonitor,
  waitForMessageCalls,
} from "./monitor-inbox.test-harness.js";
import type { InboxOnMessage } from "./monitor-inbox.test-harness.js";

const { imageOps, sleepWithAbortMock } = vi.hoisted(() => ({
  imageOps: {
    getImageMetadata: vi.fn(),
    resizeToJpeg: vi.fn(),
  },
  sleepWithAbortMock: vi.fn(async (_ms: number, _signal?: AbortSignal) => undefined),
}));

vi.mock("openclaw/plugin-sdk/media-runtime", async () => {
  const actual = await vi.importActual<typeof import("openclaw/plugin-sdk/media-runtime")>(
    "openclaw/plugin-sdk/media-runtime",
  );
  return {
    ...actual,
    getImageMetadata: imageOps.getImageMetadata,
    resizeToJpeg: imageOps.resizeToJpeg,
  };
});

vi.mock("./reconnect.js", async () => {
  const actual = await vi.importActual<typeof import("./reconnect.js")>("./reconnect.js");
  return {
    ...actual,
    sleepWithAbort: (ms: number, signal?: AbortSignal) => sleepWithAbortMock(ms, signal),
  };
});

let nextMessageSequence = 0;

function nextMessageId(label: string): string {
  nextMessageSequence += 1;
  return `${label}-${nextMessageSequence}`;
}

function createSocketRef(): NonNullable<InboxMonitorOptions["socketRef"]> {
  return { current: null };
}

function inboundMessage(onMessage: ReturnType<typeof vi.fn>, index = 0): WebInboundMessage {
  const msg = onMessage.mock.calls[index]?.[0];
  expect(msg).toBeDefined();
  return msg as WebInboundMessage;
}

async function primeInboundReplyHandle(params: {
  onMessage: ReturnType<typeof vi.fn>;
  socketRef: NonNullable<InboxMonitorOptions["socketRef"]>;
  upsertId: string;
  retryPolicy: NonNullable<InboxMonitorOptions["disconnectRetryPolicy"]>;
  useCurrentSock?: boolean;
}) {
  const { listener, sock } = await startInboxMonitor(params.onMessage as InboxOnMessage, {
    socketRef: params.socketRef,
    shouldRetryDisconnect: () => true,
    disconnectRetryPolicy: params.retryPolicy,
  });
  const sourceSock = params.useCurrentSock ? getSock() : sock;
  sourceSock.ev.emit(
    "messages.upsert",
    buildNotifyMessageUpsert({
      id: nextMessageId(params.upsertId),
      remoteJid: "999@s.whatsapp.net",
      text: "ping",
      timestamp: 1_700_000_000,
      pushName: "Tester",
    }),
  );
  await waitForMessageCalls(params.onMessage, 1);

  const inbound = inboundMessage(params.onMessage);

  return { listener, sock, inbound };
}

describe("web monitor inbox", () => {
  installWebMonitorInboxUnitTestHooks();

  beforeEach(() => {
    imageOps.getImageMetadata.mockReset();
    imageOps.getImageMetadata.mockResolvedValue(null);
    imageOps.resizeToJpeg.mockReset();
    imageOps.resizeToJpeg.mockRejectedValue(new Error("unexpected thumbnail generation"));
    sleepWithAbortMock.mockReset();
    sleepWithAbortMock.mockImplementation(async (_ms: number, _signal?: AbortSignal) => undefined);
  });

  async function expectQuotedReplyContext(quotedMessage: unknown) {
    const onMessage = vi.fn(async (msg) => {
      await msg.platform.reply("pong");
    });

    const { listener, sock } = await startInboxMonitor(onMessage as InboxOnMessage);
    const upsert = {
      type: "notify",
      messages: [
        {
          key: {
            id: nextMessageId("quoted"),
            fromMe: false,
            remoteJid: "999@s.whatsapp.net",
          },
          message: {
            extendedTextMessage: {
              text: "reply",
              contextInfo: {
                stanzaId: "q1",
                participant: "111@s.whatsapp.net",
                quotedMessage,
              },
            },
          },
          messageTimestamp: 1_700_000_000,
          pushName: "Tester",
        },
      ],
    };

    sock.ev.emit("messages.upsert", upsert);
    await waitForMessageCalls(onMessage, 1);

    const inbound = inboundMessage(onMessage);
    expect(inbound.quote?.id).toBe("q1");
    expect(inbound.quote?.body).toBe("original");
    expect(inbound.quote?.sender?.displayName).toBe("+111");
    const sender = inbound.platform.sender as { e164?: string; name?: string };
    expect(sender.e164).toBe("+999");
    expect(sender.name).toBe("Tester");
    const replyTo = inbound.quote?.context as {
      body?: string;
      id?: string;
      sender?: { e164?: string; jid?: string; label?: string };
    };
    expect(replyTo.id).toBe("q1");
    expect(replyTo.body).toBe("original");
    expect(replyTo.sender?.jid).toBe("111@s.whatsapp.net");
    expect(replyTo.sender?.e164).toBe("+111");
    expect(replyTo.sender?.label).toBe("+111");
    const self = inbound.platform.self as { e164?: string; jid?: string };
    expect(self.jid).toBe("123@s.whatsapp.net");
    expect(self.e164).toBe("+123");
    expect(sock.sendMessage).toHaveBeenCalledWith("999@s.whatsapp.net", {
      text: "pong",
    });

    await listener.close();
  }

  it("streams inbound messages", async () => {
    const onMessage = vi.fn(async (msg) => {
      await msg.sendComposing();
      await msg.reply("flat reply works");
      await msg.sendMedia({ text: "flat media works" });
    });

    const { listener, sock } = await startInboxMonitor(onMessage as InboxOnMessage);
    expect(sock.sendPresenceUpdate).toHaveBeenNthCalledWith(1, "available");
    const messageId = nextMessageId("stream");
    const upsert = buildNotifyMessageUpsert({
      id: messageId,
      remoteJid: "999@s.whatsapp.net",
      text: "ping",
      timestamp: 1_700_000_000,
      pushName: "Tester",
    });

    sock.ev.emit("messages.upsert", upsert);
    await waitForMessageCalls(onMessage, 1);

    const inbound = inboundMessage(onMessage);
    expect(inbound.payload.body).toBe("ping");
    expect(inbound.from).toBe("+999");
    expect(inbound.platform.recipientJid).toBe("+123");
    expect(sock.readMessages).toHaveBeenCalledWith([
      {
        remoteJid: "999@s.whatsapp.net",
        id: messageId,
        participant: undefined,
        fromMe: false,
      },
    ]);
    expect(sock.sendPresenceUpdate).toHaveBeenCalledWith("available");
    expect(sock.sendPresenceUpdate).toHaveBeenCalledWith("composing", "999@s.whatsapp.net");
    expect(sock.sendMessage).toHaveBeenNthCalledWith(1, "999@s.whatsapp.net", {
      text: "flat reply works",
    });
    expect(sock.sendMessage).toHaveBeenNthCalledWith(2, "999@s.whatsapp.net", {
      text: "flat media works",
    });

    await listener.close();
  });

  it("delays read receipts until inbound handlers complete", async () => {
    let finishMessage: (() => void) | undefined;
    const handlerGate = new Promise<void>((resolve) => {
      finishMessage = resolve;
    });
    const onMessage = vi.fn(async () => {
      await handlerGate;
    });

    const { listener, sock } = await startInboxMonitor(onMessage as InboxOnMessage);
    const messageId = nextMessageId("delayed-read");

    sock.ev.emit(
      "messages.upsert",
      buildNotifyMessageUpsert({
        id: messageId,
        remoteJid: "999@s.whatsapp.net",
        text: "ping",
        timestamp: 1_700_000_000,
        pushName: "Tester",
      }),
    );
    await waitForMessageCalls(onMessage, 1);

    expect(sock.readMessages).not.toHaveBeenCalled();
    finishMessage?.();
    await vi.waitFor(() => {
      expect(sock.readMessages).toHaveBeenCalledWith([
        {
          remoteJid: "999@s.whatsapp.net",
          id: messageId,
          participant: undefined,
          fromMe: false,
        },
      ]);
    });

    await listener.close();
  });

  it("continues live delivery when durable persistence rejects a message", async () => {
    failNextWhatsAppPluginStateRegisterIfAbsent(new Error("PLUGIN_STATE_LIMIT_EXCEEDED"));
    const onMessage = vi.fn(async () => undefined);

    const { listener, sock } = await startInboxMonitor(onMessage as InboxOnMessage);
    const messageId = nextMessageId("durable-fallback");

    sock.ev.emit(
      "messages.upsert",
      buildNotifyMessageUpsert({
        id: messageId,
        remoteJid: "999@s.whatsapp.net",
        text: "ping",
        timestamp: 1_700_000_000,
        pushName: "Tester",
      }),
    );
    await waitForMessageCalls(onMessage, 1);

    expect(inboundMessage(onMessage).payload.body).toBe("ping");
    await vi.waitFor(() => {
      expect(sock.readMessages).toHaveBeenCalledWith([
        {
          remoteJid: "999@s.whatsapp.net",
          id: messageId,
          participant: undefined,
          fromMe: false,
        },
      ]);
    });

    await listener.close();
  });

  it("does not dispatch live duplicates that already have pending durable delivery", async () => {
    let finishMessage: (() => void) | undefined;
    const handlerGate = new Promise<void>((resolve) => {
      finishMessage = resolve;
    });
    const onMessage = vi.fn(async () => {
      await handlerGate;
    });

    const { listener, sock } = await startInboxMonitor(onMessage as InboxOnMessage);
    const messageId = nextMessageId("durable-pending");
    const upsert = buildNotifyMessageUpsert({
      id: messageId,
      remoteJid: "999@s.whatsapp.net",
      text: "ping",
      timestamp: 1_700_000_000,
      pushName: "Tester",
    });

    sock.ev.emit("messages.upsert", upsert);
    await waitForMessageCalls(onMessage, 1);

    resetWebInboundDedupeForTests();
    sock.ev.emit("messages.upsert", upsert);
    await settleInboundWork();
    expect(onMessage).toHaveBeenCalledTimes(1);

    finishMessage?.();
    await listener.close();
  });

  it("stays unavailable on connect in self-chat mode", async () => {
    const { listener, sock } = await startInboxMonitor(vi.fn(async () => {}) as InboxOnMessage, {
      selfChatMode: true,
    });

    expect(sock.sendPresenceUpdate).toHaveBeenNthCalledWith(1, "unavailable");

    await listener.close();
  });

  it("hydrates participating groups once after connect", async () => {
    const { listener, sock } = await startInboxMonitor(vi.fn(async () => {}) as InboxOnMessage);

    expect(sock.groupFetchAllParticipating).toHaveBeenCalledTimes(1);

    await listener.close();
  });

  it("continues when group hydration fails on connect", async () => {
    const sock = getSock();
    sock.groupFetchAllParticipating.mockRejectedValueOnce(new Error("no groups"));

    const { listener } = await startInboxMonitor(vi.fn(async () => {}) as InboxOnMessage);

    expect(sock.groupFetchAllParticipating).toHaveBeenCalledTimes(1);
    expect(sock.sendPresenceUpdate).toHaveBeenNthCalledWith(1, "available");

    await listener.close();
  });

  it("omits group context when a group message has no group facts", async () => {
    const sock = getSock();
    sock.groupFetchAllParticipating.mockRejectedValueOnce(new Error("no groups"));
    const onMessage = vi.fn(async () => {});
    const { listener } = await startInboxMonitor(onMessage as InboxOnMessage);
    sock.groupMetadata.mockRejectedValueOnce(new Error("group metadata unavailable"));

    sock.ev.emit(
      "messages.upsert",
      buildNotifyMessageUpsert({
        id: nextMessageId("group-no-facts"),
        remoteJid: "123@g.us",
        participant: "444@s.whatsapp.net",
        text: "ping",
        timestamp: 1_700_000_000,
      }),
    );

    await waitForMessageCalls(onMessage, 1);
    const inbound = inboundMessage(onMessage);
    expect(inbound.chatType).toBe("group");
    expect(inbound.group).toBeUndefined();

    await listener.close();
  });

  it("keeps group inbound alive with cached metadata after reconnect-time metadata fetch failures", async () => {
    const groupMetadataCache: NonNullable<InboxMonitorOptions["groupMetadataCache"]> = new Map();
    const onMessage = vi.fn(async (_msg: Parameters<InboxOnMessage>[0]) => {});

    const firstSock = getSock();
    firstSock.groupFetchAllParticipating.mockResolvedValueOnce({
      "123@g.us": {
        id: "123@g.us",
        subject: "Recovered Group",
        owner: undefined,
        participants: [{ id: "444@s.whatsapp.net" }],
      },
    });
    const first = await startInboxMonitor(onMessage as InboxOnMessage, {
      groupMetadataCache,
    });
    await vi.waitFor(() => {
      expect(groupMetadataCache.get("123@g.us")?.subject).toBe("Recovered Group");
    });
    expect(
      (groupMetadataCache.get("123@g.us") as Record<string, unknown>)?.participants,
    ).toBeUndefined();
    await first.listener.close();

    const second = await startInboxMonitor(onMessage as InboxOnMessage, {
      groupMetadataCache,
    });
    second.sock.groupMetadata.mockRejectedValueOnce(new Error("408 timed out"));
    second.sock.ev.emit(
      "messages.upsert",
      buildNotifyMessageUpsert({
        id: nextMessageId("group-reconnect-cache"),
        remoteJid: "123@g.us",
        participant: "444@s.whatsapp.net",
        text: "ping",
        timestamp: 1_700_000_000,
      }),
    );

    await waitForMessageCalls(onMessage, 1);
    const inbound = inboundMessage(onMessage);
    expect(inbound.payload.body).toBe("ping");
    expect(inbound.from).toBe("123@g.us");
    expect(inbound.group?.subject).toBe("Recovered Group");
    expect(inbound.platform.senderE164).toBe("+444");
    expect(inbound.chatType).toBe("group");
    expect(inbound.group?.participants).toBeUndefined();

    await second.listener.close();
  });

  it("bounds cached group metadata kept across reconnects", async () => {
    const groupMetadataCache: NonNullable<InboxMonitorOptions["groupMetadataCache"]> = new Map();
    const groups = Object.fromEntries(
      Array.from({ length: WHATSAPP_GROUP_METADATA_CACHE_MAX_ENTRIES + 2 }, (_, index) => [
        `${index}@g.us`,
        {
          id: `${index}@g.us`,
          subject: `Group ${index}`,
          owner: undefined,
          participants: [],
        },
      ]),
    );
    const sock = getSock();
    sock.groupFetchAllParticipating.mockResolvedValueOnce(groups);

    const { listener } = await startInboxMonitor(vi.fn(async () => {}) as InboxOnMessage, {
      groupMetadataCache,
    });

    await vi.waitFor(() => {
      expect(groupMetadataCache.size).toBe(WHATSAPP_GROUP_METADATA_CACHE_MAX_ENTRIES);
    });
    expect(groupMetadataCache.has("0@g.us")).toBe(false);
    expect(groupMetadataCache.has(`${WHATSAPP_GROUP_METADATA_CACHE_MAX_ENTRIES + 1}@g.us`)).toBe(
      true,
    );

    await listener.close();
  });

  it("does not keep reconnect group metadata when the expiry would exceed a valid Date", async () => {
    const groupMetadataCache: NonNullable<InboxMonitorOptions["groupMetadataCache"]> = new Map();
    const dateNow = vi.spyOn(Date, "now").mockReturnValue(8_640_000_000_000_000);
    try {
      const sock = getSock();
      sock.groupFetchAllParticipating.mockResolvedValueOnce({
        "123@g.us": {
          id: "123@g.us",
          subject: "Boundary Group",
          owner: undefined,
          participants: [],
        },
      });

      const { listener } = await startInboxMonitor(vi.fn(async () => {}) as InboxOnMessage, {
        groupMetadataCache,
      });

      await vi.waitFor(() => {
        expect(sock.groupFetchAllParticipating).toHaveBeenCalledTimes(1);
      });
      expect(groupMetadataCache.has("123@g.us")).toBe(false);

      await listener.close();
    } finally {
      dateNow.mockRestore();
    }
  });

  it("does not block inbound listeners while group hydration is pending", async () => {
    let resolveHydration!: () => void;
    const sock = getSock();
    const pendingHydration = new Promise<Record<string, never>>((resolve) => {
      resolveHydration = () => resolve({});
    });
    sock.groupFetchAllParticipating.mockImplementationOnce(() => pendingHydration);
    const onMessage = vi.fn(async () => {});

    const { listener } = await startInboxMonitor(onMessage as InboxOnMessage);
    sock.ev.emit(
      "messages.upsert",
      buildNotifyMessageUpsert({
        id: nextMessageId("pending-hydration"),
        remoteJid: "999@s.whatsapp.net",
        text: "ping",
        timestamp: 1_700_000_000,
        pushName: "Tester",
      }),
    );
    await waitForMessageCalls(onMessage, 1);

    resolveHydration();
    await listener.close();
  });

  it("uses a replacement socket for replies created before reconnect", async () => {
    const onMessage = vi.fn(async () => undefined);
    const socketRef: NonNullable<InboxMonitorOptions["socketRef"]> = { current: null };

    const { listener, sock } = await startInboxMonitor(onMessage as InboxOnMessage, { socketRef });
    sock.ev.emit(
      "messages.upsert",
      buildNotifyMessageUpsert({
        id: nextMessageId("replacement-socket"),
        remoteJid: "999@s.whatsapp.net",
        text: "ping",
        timestamp: 1_700_000_000,
        pushName: "Tester",
      }),
    );
    await waitForMessageCalls(onMessage, 1);

    const inbound = inboundMessage(onMessage);

    const replacementSock = {
      sendMessage: vi.fn(async () => undefined),
      sendPresenceUpdate: vi.fn(async () => undefined),
    };
    socketRef.current = replacementSock as unknown as NonNullable<
      InboxMonitorOptions["socketRef"]
    >["current"];

    await inbound.platform.reply("pong");
    await inbound.platform.sendMedia({ text: "after-reconnect" });
    await inbound.platform.sendComposing();

    expect(replacementSock.sendMessage).toHaveBeenNthCalledWith(1, "999@s.whatsapp.net", {
      text: "pong",
    });
    expect(replacementSock.sendMessage).toHaveBeenNthCalledWith(2, "999@s.whatsapp.net", {
      text: "after-reconnect",
    });
    expect(replacementSock.sendPresenceUpdate).toHaveBeenCalledWith(
      "composing",
      "999@s.whatsapp.net",
    );
    expect(sock.sendMessage).not.toHaveBeenCalled();

    await listener.close();
  });

  it("prepopulates image previews for inbound sendMedia replies", async () => {
    const onMessage = vi.fn(async () => undefined);
    const { listener, sock } = await startInboxMonitor(onMessage as InboxOnMessage);
    sock.ev.emit(
      "messages.upsert",
      buildNotifyMessageUpsert({
        id: nextMessageId("image-preview"),
        remoteJid: "999@s.whatsapp.net",
        text: "ping",
        timestamp: 1_700_000_000,
        pushName: "Tester",
      }),
    );
    await waitForMessageCalls(onMessage, 1);

    const inbound = inboundMessage(onMessage);
    const image = Buffer.from("img");
    const thumbnail = Buffer.from("thumb");
    imageOps.getImageMetadata.mockResolvedValueOnce({ width: 640, height: 480 });
    imageOps.resizeToJpeg.mockResolvedValueOnce(thumbnail);

    await inbound.platform.sendMedia({ image, caption: "cap", mimetype: "image/png" });

    expect(imageOps.resizeToJpeg).toHaveBeenCalledWith({
      buffer: image,
      maxSide: 32,
      quality: 50,
      withoutEnlargement: true,
    });
    expect(sock.sendMessage).toHaveBeenCalledWith("999@s.whatsapp.net", {
      image,
      caption: "cap",
      mimetype: "image/png",
      width: 640,
      height: 480,
      jpegThumbnail: thumbnail.toString("base64"),
    });

    await listener.close();
  });

  it("waits for a replacement socket before sending replies", async () => {
    const onMessage = vi.fn(async () => undefined);
    const socketRef = createSocketRef();
    const { listener, sock, inbound } = await primeInboundReplyHandle({
      onMessage,
      socketRef,
      upsertId: "reconnect-gap",
      retryPolicy: {
        initialMs: 10,
        maxMs: 10,
        factor: 1,
        jitter: 0,
        maxAttempts: 2,
      },
    });

    const replacementSock = {
      sendMessage: vi.fn(async () => undefined),
      sendPresenceUpdate: vi.fn(async () => undefined),
    };
    socketRef.current = null;
    sleepWithAbortMock.mockImplementationOnce(async () => {
      socketRef.current = replacementSock as unknown as NonNullable<
        InboxMonitorOptions["socketRef"]
      >["current"];
    });

    await inbound?.platform.reply("pong");

    expect(sleepWithAbortMock).toHaveBeenCalledWith(10, undefined);
    expect(replacementSock.sendMessage).toHaveBeenCalledWith("999@s.whatsapp.net", {
      text: "pong",
    });
    expect(sock.sendMessage).not.toHaveBeenCalled();

    await listener.close();
  });

  it("flushes pending debounced inbound batches after close", async () => {
    vi.useFakeTimers();
    try {
      const onMessage = vi.fn(async () => undefined);
      const { listener, sock } = await startInboxMonitor(onMessage as InboxOnMessage, {
        debounceMs: 50,
      });
      sock.ev.emit(
        "messages.upsert",
        buildNotifyMessageUpsert({
          id: nextMessageId("debounce-close-1"),
          remoteJid: "999@s.whatsapp.net",
          text: "first",
          timestamp: 1_700_000_000,
          pushName: "Tester",
        }),
      );
      sock.ev.emit(
        "messages.upsert",
        buildNotifyMessageUpsert({
          id: nextMessageId("debounce-close-2"),
          remoteJid: "999@s.whatsapp.net",
          text: "second",
          timestamp: 1_700_000_001,
          pushName: "Tester",
        }),
      );

      await listener.close();
      await vi.advanceTimersByTimeAsync(50);
      await waitForMessageCalls(onMessage, 1);
      const inbound = inboundMessage(onMessage);
      expect(inbound.payload.body).toBe("first\nsecond");
      expect(inbound.chatType).toBe("direct");
    } finally {
      vi.useRealTimers();
    }
  });

  it("lets a drained debounced inbound reply before closing the socket", async () => {
    vi.useFakeTimers();
    try {
      const onMessage = vi.fn(async (msg) => {
        await msg.platform.reply("pong");
        await msg.platform.sendMedia({ text: "media" });
      });
      const { listener, sock } = await startInboxMonitor(onMessage as InboxOnMessage, {
        debounceMs: 50,
      });
      sock.ev.emit(
        "messages.upsert",
        buildNotifyMessageUpsert({
          id: nextMessageId("debounce-close-reply-1"),
          remoteJid: "999@s.whatsapp.net",
          text: "first",
          timestamp: 1_700_000_000,
          pushName: "Tester",
        }),
      );
      sock.ev.emit(
        "messages.upsert",
        buildNotifyMessageUpsert({
          id: nextMessageId("debounce-close-reply-2"),
          remoteJid: "999@s.whatsapp.net",
          text: "second",
          timestamp: 1_700_000_001,
          pushName: "Tester",
        }),
      );

      await listener.close();

      expect(onMessage).toHaveBeenCalledTimes(1);
      expect(inboundMessage(onMessage).payload.body).toBe("first\nsecond");
      expect(sock.sendMessage).toHaveBeenNthCalledWith(1, "999@s.whatsapp.net", {
        text: "pong",
      });
      expect(sock.sendMessage).toHaveBeenNthCalledWith(2, "999@s.whatsapp.net", {
        text: "media",
      });
      expect(sock.end).toHaveBeenCalledTimes(1);
      expect(sock.sendMessage.mock.invocationCallOrder.at(-1)).toBeLessThan(
        sock.end.mock.invocationCallOrder.at(0),
      );
    } finally {
      vi.useRealTimers();
    }
  });

  it("waits for in-flight inbound handlers before draining on close", async () => {
    let releaseHandler: (() => void) | undefined;
    const handlerGate = new Promise<void>((resolve) => {
      releaseHandler = resolve;
    });
    let markHandlerStarted: (() => void) | undefined;
    const handlerStarted = new Promise<void>((resolve) => {
      markHandlerStarted = resolve;
    });
    const onMessage = vi.fn(async (msg) => {
      await msg.platform.reply("pong");
      markHandlerStarted?.();
      await handlerGate;
    });
    const { listener, sock } = await startInboxMonitor(onMessage as InboxOnMessage);

    sock.ev.emit(
      "messages.upsert",
      buildNotifyMessageUpsert({
        id: nextMessageId("close-inflight"),
        remoteJid: "999@s.whatsapp.net",
        text: "first",
        timestamp: 1_700_000_000,
        pushName: "Tester",
      }),
    );

    await handlerStarted;
    const closePromise = listener.close();
    await Promise.resolve();

    expect(sock.end).not.toHaveBeenCalled();

    if (!releaseHandler) {
      throw new Error("Expected handler release callback to be initialized");
    }
    releaseHandler();
    await closePromise;

    expect(onMessage).toHaveBeenCalledTimes(1);
    expect(sock.sendMessage).toHaveBeenCalledWith("999@s.whatsapp.net", {
      text: "pong",
    });
    expect(sock.end).toHaveBeenCalledTimes(1);
    expect(sock.sendMessage.mock.invocationCallOrder.at(0)).toBeLessThan(
      sock.end.mock.invocationCallOrder.at(0),
    );
  });

  it("retries timed-out sends on the same socket without clearing the socket ref", async () => {
    const onMessage = vi.fn(async () => undefined);
    const socketRef = createSocketRef();
    const { listener, sock, inbound } = await primeInboundReplyHandle({
      onMessage,
      socketRef,
      upsertId: "timeout-retry",
      retryPolicy: {
        initialMs: 1,
        maxMs: 1,
        factor: 1,
        jitter: 0,
        maxAttempts: 2,
      },
    });

    sock.sendMessage
      .mockRejectedValueOnce(new Error("operation timed out"))
      .mockResolvedValueOnce({ key: { id: "after-timeout" } });

    await inbound?.platform.reply("pong");

    expect(sock.sendMessage).toHaveBeenNthCalledWith(1, "999@s.whatsapp.net", {
      text: "pong",
    });
    expect(sock.sendMessage).toHaveBeenNthCalledWith(2, "999@s.whatsapp.net", {
      text: "pong",
    });
    expect(socketRef.current).toBe(sock);
    expect(sleepWithAbortMock).toHaveBeenCalledTimes(1);

    await listener.close();
  });

  it("bounds reconnect-gap retries even when reconnect attempts are unlimited", async () => {
    const onMessage = vi.fn(async () => undefined);
    const socketRef = createSocketRef();
    const { listener, inbound } = await primeInboundReplyHandle({
      onMessage,
      socketRef,
      upsertId: "unlimited-reconnect-send-bound",
      retryPolicy: {
        initialMs: 1,
        maxMs: 1,
        factor: 1,
        jitter: 0,
        maxAttempts: 0,
      },
      useCurrentSock: true,
    });

    socketRef.current = null;

    await expect(inbound?.platform.reply("pong")).rejects.toThrow(
      "no active socket - reconnection in progress",
    );
    expect(sleepWithAbortMock).toHaveBeenCalledTimes(11);

    await listener.close();
  });

  // VE-513: when the gateway's channel-health-monitor tears down controller A
  // mid-run (shutdown nulls A.socketRef and aborts A.disconnectRetries), then a
  // fresh controller B registers for the same accountId, an in-flight inbound's
  // captured reply closure must route through B's socket via the registry instead
  // of throwing RECONNECT_IN_PROGRESS. Before the registry fallback was added,
  // sendTrackedMessage only knew its own controller's socketRef and the captured
  // reply was permanently broken.
  it("routes the captured reply through a successor controller when the original controller's socket is gone", async () => {
    const onMessage = vi.fn(async () => undefined);
    const socketRefA = createSocketRef();

    let aShouldRetryDisconnect = true;
    const { listener: listenerA, sock: sockA } = await startInboxMonitor(
      onMessage as InboxOnMessage,
      {
        socketRef: socketRefA,
        shouldRetryDisconnect: () => aShouldRetryDisconnect,
        disconnectRetryPolicy: {
          initialMs: 1,
          maxMs: 1,
          factor: 1,
          jitter: 0,
          maxAttempts: 1,
        },
      },
    );

    sockA.ev.emit(
      "messages.upsert",
      buildNotifyMessageUpsert({
        id: nextMessageId("outbound-successor"),
        remoteJid: "999@s.whatsapp.net",
        text: "ping",
        timestamp: 1_700_000_000,
        pushName: "Tester",
      }),
    );
    await waitForMessageCalls(onMessage, 1);
    const inbound = inboundMessage(onMessage);

    // The mock harness socket exposes user.id = "123@s.whatsapp.net"; the
    // successor handle must report a self identity that overlaps that JID
    // so the session-safety guard accepts the fallback.
    const handleA = {
      getActiveListener: () => null,
      getCurrentSock: () => null,
      getSelfIdentity: () => null,
    } as never;
    registerWhatsAppConnectionController(DEFAULT_ACCOUNT_ID, handleA);

    // === Simulate health-monitor-driven shutdown of controller A ===
    socketRefA.current = null;
    aShouldRetryDisconnect = false;
    unregisterWhatsAppConnectionController(DEFAULT_ACCOUNT_ID, handleA);

    // === Successor controller B comes up with its OWN socket and registers ===
    const sockB = {
      sendMessage: vi.fn(async () => ({ key: { id: "post-restart-msg-id" } })),
    };
    const handleB = {
      getActiveListener: () => null,
      getCurrentSock: () => sockB as never,
      getSelfIdentity: () => ({ jid: "123@s.whatsapp.net", lid: null }),
    } as never;
    registerWhatsAppConnectionController(DEFAULT_ACCOUNT_ID, handleB);

    try {
      await inbound.reply("pong");
      await inbound.sendMedia({ text: "media after restart" });

      // Captured A reply routed through B via the registry handle.
      expect(sockB.sendMessage).toHaveBeenCalledTimes(2);
      expect(sockB.sendMessage).toHaveBeenNthCalledWith(1, "999@s.whatsapp.net", {
        text: "pong",
      });
      expect(sockB.sendMessage).toHaveBeenNthCalledWith(2, "999@s.whatsapp.net", {
        text: "media after restart",
      });
    } finally {
      unregisterWhatsAppConnectionController(DEFAULT_ACCOUNT_ID, handleB);
      await listenerA.close();
    }
  });

  // VE-513 PN/LID normalization: Baileys sockets may expose self identity in
  // different forms across reconnects (PN JID `<n>@s.whatsapp.net` vs LID
  // `<x>@lid`). The fallback must recognize the same account when one side
  // reports only PN and the other only LID, because the captured reply
  // shouldn't be dropped just because the identity form rotated.
  it("accepts successor-controller fallback when only the LID-vs-PN form differs for the same account e164", async () => {
    const onMessage = vi.fn(async () => undefined);
    const socketRefA = createSocketRef();

    let aShouldRetryDisconnect = true;
    const { listener: listenerA, sock: sockA } = await startInboxMonitor(
      onMessage as InboxOnMessage,
      {
        socketRef: socketRefA,
        shouldRetryDisconnect: () => aShouldRetryDisconnect,
        disconnectRetryPolicy: {
          initialMs: 1,
          maxMs: 1,
          factor: 1,
          jitter: 0,
          maxAttempts: 1,
        },
      },
    );

    sockA.ev.emit(
      "messages.upsert",
      buildNotifyMessageUpsert({
        id: nextMessageId("outbound-successor-lid"),
        remoteJid: "999@s.whatsapp.net",
        text: "ping",
        timestamp: 1_700_000_000,
        pushName: "Tester",
      }),
    );
    await waitForMessageCalls(onMessage, 1);
    const inbound = inboundMessage(onMessage);

    socketRefA.current = null;
    aShouldRetryDisconnect = false;

    // The mock harness socket has user.id="123@s.whatsapp.net" (PN JID, no
    // lid). The successor reports only the LID form. Both resolve to the same
    // synthetic e164 via resolveComparableIdentity, so identitiesOverlap
    // accepts the fallback. (resolveComparableIdentity preserves an explicit
    // e164 on the input over deriving from the JID via authDir lookup.)
    const sharedE164 = "+123";
    const sockB = {
      sendMessage: vi.fn(async () => ({ key: { id: "post-restart-lid-msg-id" } })),
    };
    const handleB = {
      getActiveListener: () => null,
      getCurrentSock: () => sockB as never,
      getSelfIdentity: () => ({ jid: null, lid: "12300:1@lid", e164: sharedE164 }),
    } as never;
    registerWhatsAppConnectionController(DEFAULT_ACCOUNT_ID, handleB);

    try {
      await inbound.reply("pong");
      expect(sockB.sendMessage).toHaveBeenCalledTimes(1);
      expect(sockB.sendMessage).toHaveBeenCalledWith("999@s.whatsapp.net", { text: "pong" });
    } finally {
      unregisterWhatsAppConnectionController(DEFAULT_ACCOUNT_ID, handleB);
      await listenerA.close();
    }
  });

  // VE-513 session-safety guard: if the registered successor controller has been
  // re-linked to a different WhatsApp identity (different self JID), the
  // captured reply must fail closed rather than route through the wrong number.
  it("refuses successor-controller fallback when the registered controller's self JID does not match", async () => {
    const onMessage = vi.fn(async () => undefined);
    const socketRefA = createSocketRef();

    let aShouldRetryDisconnect = true;
    const { listener: listenerA, sock: sockA } = await startInboxMonitor(
      onMessage as InboxOnMessage,
      {
        socketRef: socketRefA,
        shouldRetryDisconnect: () => aShouldRetryDisconnect,
        disconnectRetryPolicy: {
          initialMs: 1,
          maxMs: 1,
          factor: 1,
          jitter: 0,
          maxAttempts: 1,
        },
      },
    );

    sockA.ev.emit(
      "messages.upsert",
      buildNotifyMessageUpsert({
        id: nextMessageId("outbound-successor-mismatch"),
        remoteJid: "999@s.whatsapp.net",
        text: "ping",
        timestamp: 1_700_000_000,
        pushName: "Tester",
      }),
    );
    await waitForMessageCalls(onMessage, 1);
    const inbound = inboundMessage(onMessage);

    // A's shutdown sequence.
    socketRefA.current = null;
    aShouldRetryDisconnect = false;

    // === Successor is registered but with a DIFFERENT self identity (relink/repair) ===
    const sockB = {
      sendMessage: vi.fn(async () => ({ key: { id: "should-not-be-called" } })),
    };
    const handleBMismatch = {
      getActiveListener: () => null,
      getCurrentSock: () => sockB as never,
      getSelfIdentity: () => ({ jid: "456@s.whatsapp.net", lid: null }),
    } as never;
    registerWhatsAppConnectionController(DEFAULT_ACCOUNT_ID, handleBMismatch);

    try {
      await expect(inbound.reply("pong")).rejects.toThrow(
        "no active socket - reconnection in progress",
      );

      // The mismatched successor's socket was never used.
      expect(sockB.sendMessage).not.toHaveBeenCalled();
    } finally {
      unregisterWhatsAppConnectionController(DEFAULT_ACCOUNT_ID, handleBMismatch);
      await listenerA.close();
    }
  });

  it("deduplicates redelivered messages by id", async () => {
    const onMessage = vi.fn(async () => {});

    const { listener, sock } = await startInboxMonitor(onMessage as InboxOnMessage);
    const upsert = buildNotifyMessageUpsert({
      id: nextMessageId("dedupe"),
      remoteJid: "999@s.whatsapp.net",
      text: "ping",
      timestamp: 1_700_000_000,
      pushName: "Tester",
    });

    sock.ev.emit("messages.upsert", upsert);
    sock.ev.emit("messages.upsert", upsert);
    await waitForMessageCalls(onMessage, 1);

    expect(onMessage).toHaveBeenCalledTimes(1);

    await listener.close();
  });

  it("retries redelivered messages after an explicit retryable inbound failure", async () => {
    let attempts = 0;
    const onMessage = vi.fn(async () => {
      attempts += 1;
      if (attempts === 1) {
        throw new WhatsAppRetryableInboundError("retry me");
      }
    });

    const { listener, sock } = await startInboxMonitor(onMessage as InboxOnMessage);
    const upsert = buildNotifyMessageUpsert({
      id: nextMessageId("retryable-dedupe"),
      remoteJid: "999@s.whatsapp.net",
      text: "ping",
      timestamp: 1_700_000_000,
      pushName: "Tester",
    });

    sock.ev.emit("messages.upsert", upsert);
    await waitForMessageCalls(onMessage, 1);
    expect(sock.readMessages).not.toHaveBeenCalled();

    sock.ev.emit("messages.upsert", upsert);
    await waitForMessageCalls(onMessage, 2);
    await vi.waitFor(() => {
      expect(sock.readMessages).toHaveBeenCalledTimes(1);
    });

    await listener.close();
  });

  it("resolves LID JIDs using Baileys LID mapping store", async () => {
    const onMessage = vi.fn(async () => {});

    const { listener, sock } = await startInboxMonitor(onMessage as InboxOnMessage);
    const getPNForLID = vi.spyOn(sock.signalRepository.lidMapping, "getPNForLID");
    sock.signalRepository.lidMapping.getPNForLID.mockResolvedValueOnce("999:0@s.whatsapp.net");
    const upsert = buildNotifyMessageUpsert({
      id: nextMessageId("lid-store"),
      remoteJid: "999@lid",
      text: "ping",
      timestamp: 1_700_000_000,
      pushName: "Tester",
    });

    sock.ev.emit("messages.upsert", upsert);
    await waitForMessageCalls(onMessage, 1);

    expect(getPNForLID).toHaveBeenCalledWith("999@lid");
    const inbound = inboundMessage(onMessage);
    expect(inbound.payload.body).toBe("ping");
    expect(inbound.from).toBe("+999");
    expect(inbound.platform.recipientJid).toBe("+123");

    await listener.close();
  });

  it("resolves LID JIDs via authDir mapping files", async () => {
    const onMessage = vi.fn(async () => {});
    fsSync.writeFileSync(
      path.join(getAuthDir(), "lid-mapping-555_reverse.json"),
      JSON.stringify("1555"),
    );

    const { listener, sock } = await startInboxMonitor(onMessage as InboxOnMessage);
    const getPNForLID = vi.spyOn(sock.signalRepository.lidMapping, "getPNForLID");
    const upsert = buildNotifyMessageUpsert({
      id: nextMessageId("lid-authdir"),
      remoteJid: "555@lid",
      text: "ping",
      timestamp: 1_700_000_000,
      pushName: "Tester",
    });

    sock.ev.emit("messages.upsert", upsert);
    await waitForMessageCalls(onMessage, 1);

    const inbound = inboundMessage(onMessage);
    expect(inbound.payload.body).toBe("ping");
    expect(inbound.from).toBe("+1555");
    expect(inbound.platform.recipientJid).toBe("+123");
    expect(getPNForLID).not.toHaveBeenCalled();

    await listener.close();
  });

  it("resolves group participant LID JIDs via Baileys mapping", async () => {
    const onMessage = vi.fn(async () => {});

    const { listener, sock } = await startInboxMonitor(onMessage as InboxOnMessage);
    const getPNForLID = vi.spyOn(sock.signalRepository.lidMapping, "getPNForLID");
    sock.signalRepository.lidMapping.getPNForLID.mockResolvedValueOnce("444:0@s.whatsapp.net");
    const upsert = buildNotifyMessageUpsert({
      id: nextMessageId("group-lid"),
      remoteJid: "123@g.us",
      participant: "444@lid",
      text: "ping",
      timestamp: 1_700_000_000,
    });

    sock.ev.emit("messages.upsert", upsert);
    await waitForMessageCalls(onMessage, 1);

    expect(getPNForLID).toHaveBeenCalledWith("444@lid");
    const inbound = inboundMessage(onMessage);
    expect(inbound.payload.body).toBe("ping");
    expect(inbound.from).toBe("123@g.us");
    expect(inbound.platform.senderE164).toBe("+444");
    expect(inbound.chatType).toBe("group");

    await listener.close();
  });

  it("does not block follow-up messages when handler is pending", async () => {
    let resolveFirst: (() => void) | null = null;
    const onMessage = vi.fn(async () => {
      if (!resolveFirst) {
        await new Promise<void>((resolve) => {
          resolveFirst = resolve;
        });
      }
    });

    const { listener, sock } = await startInboxMonitor(onMessage as InboxOnMessage);
    const upsert = {
      type: "notify",
      messages: [
        {
          key: { id: "abc1", fromMe: false, remoteJid: "999@s.whatsapp.net" },
          message: { conversation: "ping" },
          messageTimestamp: 1_700_000_000,
        },
        {
          key: { id: "abc2", fromMe: false, remoteJid: "999@s.whatsapp.net" },
          message: { conversation: "pong" },
          messageTimestamp: 1_700_000_001,
        },
      ],
    };

    sock.ev.emit("messages.upsert", upsert);
    await waitForMessageCalls(onMessage, 2);

    expect(onMessage).toHaveBeenCalledTimes(2);

    (resolveFirst as (() => void) | null)?.();
    await listener.close();
  });

  it("captures reply context from quoted messages", async () => {
    await expectQuotedReplyContext({ conversation: "original" });
  });

  it("captures reply context from wrapped quoted messages", async () => {
    await expectQuotedReplyContext({
      viewOnceMessageV2Extension: {
        message: { conversation: "original" },
      },
    });
  });

  it("captures reply context from botInvokeMessage wrapped quoted messages", async () => {
    await expectQuotedReplyContext({
      botInvokeMessage: {
        message: { conversation: "original" },
      },
    });
  });

  it("captures reply context from groupMentionedMessage wrapped quoted messages", async () => {
    await expectQuotedReplyContext({
      groupMentionedMessage: {
        message: { conversation: "original" },
      },
    });
  });
});
