/**
 * Realtime integration (TASK-0091, acceptance criteria 2 + 3): heartbeat
 * / prune and the message/payload rate limits.
 *
 * Time model: the server runs on REAL time (the fixture's ping interval,
 * prune threshold, and GC delay are shortened real timers — the same
 * timers production uses, only faster). The heartbeat test uses a raw TCP
 * socket that NEVER answers the server's protocol pings (the `ws` client
 * auto-pongs, which would defeat the scenario).
 */
import { describe, expect, it } from "vitest";
import * as Y from "yjs";
import {
  buildRealtimeFixture,
  connectClient,
  connectRawSocket,
  makeTextUpdate,
} from "./rt-helpers";
import { WS_CLOSE, REALTIME_LIMITS } from "@factory/shared/realtime/frames";

describe("heartbeat + prune (R4)", () => {
  it("a socket that never pongs is terminated with 4000 (heartbeat_timeout)", async () => {
    // Short real cadences: ping every 300 ms, prune after 1.5 s silence.
    const fx = await buildRealtimeFixture({
      pages: new Set(["p-it"]),
      workspaces: { "p-it": "w1" },
      levels: { "page:p-it": "edit" },
      pingIntervalMs: 300,
      pruneSilenceMs: 1_500,
    });
    try {
      const raw = await connectRawSocket(fx, { room: "page:p-it" });
      // Wait for the prune: the raw socket never pongs → 90 s (here 1.5 s)
      // of silence → close 4000. The raw scanner records the unmasked
      // close frame code from the wire (it ignores the 101 response and
      // the protocol ping frames — see rt-helpers' close-frame check).
      const deadline = Date.now() + 8_000;
      while (Date.now() < deadline && !raw.closeCodes().includes(WS_CLOSE.ProtocolOrHeartbeat)) {
        await new Promise((r) => setTimeout(r, 100));
      }
      expect(raw.closeCodes(), "server should have sent the close frame").toContain(
        WS_CLOSE.ProtocolOrHeartbeat,
      );
      raw.destroy();
    } finally {
      await fx.stop();
    }
  }, 30_000);

  it("a socket that pongs stays alive past the prune window (10× the window)", async () => {
    const fx = await buildRealtimeFixture({
      pages: new Set(["p-it"]),
      workspaces: { "p-it": "w1" },
      levels: { "page:p-it": "edit" },
      pingIntervalMs: 300,
      pruneSilenceMs: 1_500,
    });
    try {
      const c = await connectClient(fx, { room: "page:p-it" });
      await c.nextFrame("hello");
      // The `ws` client auto-pongs every server ping — no app-level
      // activity needed. Outlive 4× the prune window.
      await new Promise((r) => setTimeout(r, 6_000));
      expect(c.isOpen()).toBe(true);
      c.close();
    } finally {
      await fx.stop();
    }
  });

  it("an application-level ping frame gets a ping-pong echo", async () => {
    const fx = await buildRealtimeFixture({
      pages: new Set(["p-it"]),
      workspaces: { "p-it": "w1" },
      levels: { "page:p-it": "edit" },
    });
    try {
      const c = await connectClient(fx, { room: "page:p-it" });
      await c.nextFrame("hello");
      c.sendText({ type: "ping", ts: 12345 });
      const pong = await c.nextFrame("ping-pong");
      expect(pong.ts).toBe(12345);
      c.close();
    } finally {
      await fx.stop();
    }
  });
});

describe("rate limits (R1/R3)", () => {
  it("a 300 KB frame kills the socket with 4009 payload_too_large + audit", async () => {
    const fx = await buildRealtimeFixture({
      pages: new Set(["p-it"]),
      workspaces: { "p-it": "w1" },
      levels: { "page:p-it": "edit" },
    });
    try {
      const c = await connectClient(fx, { room: "page:p-it" });
      await c.nextFrame("hello");
      // 300 KB of raw bytes with the 0x01 update prefix (invalid Yjs —
      // irrelevant: the payload check fires before validation).
      const big = new Uint8Array(300 * 1024);
      for (let i = 0; i < big.length; i += 1) big[i] = i & 0xff;
      c.sendBinary(1, big);
      const closed = await c.onClose(5_000);
      expect(closed.code).toBe(WS_CLOSE.PayloadTooLarge);
      expect(fx.audit.abuse("frame_too_large")).toBe(1);
    } finally {
      await fx.stop();
    }
  });

  it("120 msgs/s sustained kills the socket with 4008 rate_limited + audit", async () => {
    const fx = await buildRealtimeFixture({
      pages: new Set(["p-it"]),
      workspaces: { "p-it": "w1" },
      levels: { "page:p-it": "edit" },
    });
    try {
      const c = await connectClient(fx, { room: "page:p-it" });
      await c.nextFrame("hello");
      const doc = new Y.Doc();
      const u = makeTextUpdate(doc, "a"); // a valid small update (1–3 KB)
      doc.destroy();

      // Fire 120 messages in a tight burst (well above the 100/s
      // sustained budget; the bucket refills continuously, so a burst of
      // >100 in ~100 ms is the deterministic overflow).
      const closePromise = c.onClose(10_000);
      for (let i = 0; i < 120; i += 1) {
        if (!c.isOpen()) break;
        c.sendBinary(1, u);
        if (i % 20 === 0) await new Promise((r) => setTimeout(r, 10));
      }
      const closed = await closePromise;
      expect(closed.code).toBe(WS_CLOSE.RateOrPolicy);
      expect(fx.audit.abuse("msg_rate_limited")).toBe(1);
    } finally {
      await fx.stop();
    }
  });

  it("10 corrupt frames within the strike window → 4008 strike_limit + audit", async () => {
    const fx = await buildRealtimeFixture({
      pages: new Set(["p-it"]),
      workspaces: { "p-it": "w1" },
      levels: { "page:p-it": "edit" },
    });
    try {
      const c = await connectClient(fx, { room: "page:p-it" });
      await c.nextFrame("hello");
      const garbage = new Uint8Array([0xde, 0xad, 0xbe, 0xef]);
      const closePromise = c.onClose(10_000);
      for (let i = 0; i < REALTIME_LIMITS.strikeLimit + 1; i += 1) {
        if (!c.isOpen()) break;
        c.sendBinary(1, garbage);
        await new Promise((r) => setTimeout(r, 20));
      }
      const closed = await closePromise;
      expect(closed.code).toBe(WS_CLOSE.RateOrPolicy);
      expect(fx.audit.abuse()).toBeGreaterThanOrEqual(1);
      expect(fx.audit.records.some((r) => String(r.payload.cause).startsWith("strikes:"))).toBe(true);
    } finally {
      await fx.stop();
    }
  });

  it("an unknown JSON frame type gets error unknown_frame (no close below the strike limit)", async () => {
    const fx = await buildRealtimeFixture({
      pages: new Set(["p-it"]),
      workspaces: { "p-it": "w1" },
      levels: { "page:p-it": "edit" },
    });
    try {
      const c = await connectClient(fx, { room: "page:p-it" });
      await c.nextFrame("hello");
      c.sendText({ type: "teleport", payload: "nope" });
      const err = await c.nextFrame("error");
      expect(err.code).toBe("unknown_frame");
      // One bad frame — the socket survives (the 10-strike policy).
      expect(c.isOpen()).toBe(true);
      c.close();
    } finally {
      await fx.stop();
    }
  });

  it("a corrupt Yjs binary (valid frame, invalid update) gets error bad_update", async () => {
    const fx = await buildRealtimeFixture({
      pages: new Set(["p-it"]),
      workspaces: { "p-it": "w1" },
      levels: { "page:p-it": "edit" },
    });
    try {
      const c = await connectClient(fx, { room: "page:p-it" });
      await c.nextFrame("hello");
      // 0x01 prefix is applied by sendBinary; the payload is not a
      // Yjs update → the throwaway applyUpdate probe throws.
      c.sendBinary(1, new Uint8Array([1, 2, 3, 4, 5]));
      const err = await c.nextFrame("error");
      expect(err.code).toBe("bad_update");
      c.close();
    } finally {
      await fx.stop();
    }
  });
});

describe("awareness relay (R4 fan-out, the TASK-0093 seam)", () => {
  it("an awareness frame is relayed to the other peers (not the sender)", async () => {
    const fx = await buildRealtimeFixture({
      pages: new Set(["p-it"]),
      workspaces: { "p-it": "w1" },
      levels: { "page:p-it": "edit" },
    });
    try {
      const a = await connectClient(fx, { room: "page:p-it" });
      const b = await connectClient(fx, { room: "page:p-it" });
      await a.nextFrame("hello");
      await b.nextFrame("hello");
      const payload = new Uint8Array([0x00, 0x01, 0x02]);
      a.sendBinary(2, payload);
      const got = await b.nextBinary();
      expect(got.kind).toBe("awareness");
      expect(Buffer.from(got.bytes).equals(Buffer.from(payload))).toBe(true);
      a.close();
      b.close();
    } finally {
      await fx.stop();
    }
  });
});