NixMsg/sdk/js/test/k02.test.ts

import { afterEach, describe, expect, it } from "vitest";
import { Client, disableJitterForTest, restoreJitterForTest } from "../src/index.js";
import { FakeTransport } from "../src/fake.js";

afterEach(() => {
  restoreJitterForTest();
});

async function connectFake(fake: FakeTransport): Promise<Client> {
  const c = new Client();
  await c.connect("ws://example.test/mqtt", "ep1", { password: "p" }, { transport: fake });
  return c;
}

describe("K02", () => {
  it("TestK01InflightResendAfterDisconnect", async () => {
    const fake = new FakeTransport();
    const c = await connectFake(fake);
    let firstRID = "";
    let firstID = "";
    let firstSendAt: unknown;
    const timer = setInterval(() => {
      const sends = fake.findUp("send");
      if (!sends.length) return;
      const last = sends[sends.length - 1]!;
      const rid = String(last.rid);
      if (!firstRID) {
        firstRID = rid;
        firstID = String(last.id);
        firstSendAt = last.send_at_ms;
        void fake.simulateReconnect();
        return;
      }
      if (rid !== firstRID) {
        expect(last.id).toBe(firstID);
        expect(last.send_at_ms).toBe(firstSendAt);
        fake.replyOK(rid, { id: firstID, send_at_ms: firstSendAt, state: "accepted" });
      }
    }, 5);
    const at = new Date(1_700_000_000_111);
    await c.send({ kind: "endpoint", id: "b" }, { enc: "utf8", data: "hi" }, { sendAt: at });
    clearInterval(timer);
    await c.close();
  }, 10000);

  it("unhandledRejection stays zero on ack publish throw", async () => {
    const rejections: unknown[] = [];
    const onRej = (e: unknown) => {
      rejections.push(e);
    };
    process.on("unhandledRejection", onRej);
    const fake = new FakeTransport();
    const c = await connectFake(fake);
    fake.publishUpImpl = async (s) => {
      if (s.includes('"type":"ack"')) throw new Error("ack publish fail");
    };
    c.onMessageHandler(() => {});
    fake.injectDown(
      JSON.stringify({
        v: 1,
        type: "msg",
        id: "m1",
        from: "a",
        to: { kind: "endpoint", id: "ep1" },
        body: { enc: "utf8", data: "hi" },
        send_at_ms: 1,
      }),
    );
    await new Promise((r) => setTimeout(r, 80));
    process.off("unhandledRejection", onRej);
    expect(rejections).toHaveLength(0);
    await c.close();
  });

  it("ack fail then same msg re-acks without second callback", async () => {
    const fake = new FakeTransport();
    const c = await connectFake(fake);
    let calls = 0;
    c.onMessageHandler(() => {
      calls++;
    });
    const msg = JSON.stringify({
      v: 1,
      type: "msg",
      id: "m1",
      from: "a",
      to: { kind: "endpoint", id: "ep1" },
      body: { enc: "utf8", data: "hi" },
      send_at_ms: 1,
    });
    fake.injectDown(msg);
    await new Promise((r) => setTimeout(r, 40));
    for (const fr of fake.findUp("ack")) {
      fake.replyErr(String(fr.rid), "not_found", "gone");
    }
    fake.injectDown(msg);
    await new Promise((r) => setTimeout(r, 80));
    expect(calls).toBe(1);
    expect(fake.findUp("ack").length).toBeGreaterThanOrEqual(2);
    await c.close();
  });

  it("revoked after msg skips callback", async () => {
    const fake = new FakeTransport();
    const c = await connectFake(fake);
    let msgs = 0;
    let revs = 0;
    c.onMessageHandler(() => {
      msgs++;
    });
    c.onRevokedHandler(() => {
      revs++;
    });
    fake.injectDown(
      JSON.stringify({
        v: 1,
        type: "msg",
        id: "m1",
        from: "a",
        to: { kind: "endpoint", id: "ep1" },
        body: { enc: "utf8", data: "hi" },
        send_at_ms: 1,
      }),
    );
    fake.injectDown(JSON.stringify({ v: 1, type: "revoked", id: "m1", from: "a", reason: "recalled" }));
    await new Promise((r) => setTimeout(r, 80));
    expect(msgs).toBe(0);
    expect(revs).toBe(1);
    await c.close();
  });

  it("TestK01WatchRestored", async () => {
    const fake = new FakeTransport();
    const c = await connectFake(fake);
    const timer = setInterval(() => {
      for (const fr of fake.findUp("presence.watch")) {
        fake.replyOK(String(fr.rid), {});
      }
    }, 5);
    await c.watchPresence(["a", "b"]);
    const n1 = fake.findUp("presence.watch").length;
    await fake.simulateReconnect();
    const deadline = Date.now() + 2000;
    while (Date.now() < deadline && fake.findUp("presence.watch").length <= n1) {
      await new Promise((r) => setTimeout(r, 10));
    }
    clearInterval(timer);
    expect(fake.findUp("presence.watch").length).toBeGreaterThan(n1);
    await c.close();
  });

  it("TestK01FatalOnce", async () => {
    const fake = new FakeTransport();
    const c = await connectFake(fake);
    let n = 0;
    c.onConnectionHandler((ev) => {
      if (ev.state === "auth_failed" && ev.reason === "disabled") n++;
    });
    const fatal = JSON.stringify({ v: 1, type: "fatal", reason: "disabled" });
    fake.injectDown(fatal);
    fake.injectDown(fatal);
    await new Promise((r) => setTimeout(r, 50));
    expect(n).toBe(1);
  });

  it("TestK01CallbackNoDeadlock", async () => {
    const fake = new FakeTransport();
    const c = await connectFake(fake);
    const timer = setInterval(() => {
      for (const typ of ["ack", "self.login_password"]) {
        for (const fr of fake.findUp(typ)) {
          fake.replyOK(String(fr.rid), typ === "ack" ? { result: "accepted" } : {});
        }
      }
    }, 5);
    let done = false;
    c.onMessageHandler(async (msg) => {
      await c.changeLoginPassword("old", "newpass12");
      await c.ack(msg);
      done = true;
    });
    fake.injectDown(
      JSON.stringify({
        v: 1,
        type: "msg",
        id: "m1",
        from: "a",
        to: { kind: "endpoint", id: "ep1" },
        body: { enc: "utf8", data: "hi" },
        send_at_ms: 1,
      }),
    );
    const deadline = Date.now() + 2000;
    while (Date.now() < deadline && !done) {
      await new Promise((r) => setTimeout(r, 10));
    }
    clearInterval(timer);
    expect(done).toBe(true);
    await c.close();
  });

  it("rate_limited rids unique with disabled jitter", async () => {
    disableJitterForTest();
    const fake = new FakeTransport();
    const c = await connectFake(fake);
    const rids: string[] = [];
    const timer = setInterval(() => {
      const sends = fake.findUp("send");
      if (!sends.length) return;
      const last = sends[sends.length - 1]!;
      const rid = String(last.rid);
      if (rids.includes(rid)) return;
      rids.push(rid);
      if (rids.length < 3) fake.replyErr(rid, "rate_limited", "slow");
      else fake.replyOK(rid, { id: last.id, send_at_ms: last.send_at_ms, state: "scheduled" });
    }, 5);
    await c.send(
      { kind: "endpoint", id: "b" },
      { enc: "utf8", data: "hi" },
      { sendAt: new Date(1_700_000_000_000) },
    );
    clearInterval(timer);
    expect(new Set(rids).size).toBe(3);
    await c.close();
  }, 10000);
});