/**
 * Unit tests for the `page.updated` control source (TASK-0092 6b).
 *
 * The stream read is faked (a hand-rolled {@link PageUpdatedStreamClient});
 * every decision the server makes from an event — envelope parsing, the
 * `page.updated` filter, the `block:changed` frame derivation — is pure and
 * asserted here without Redis.
 */
import { describe, expect, it } from "vitest";

import { DomainEventName } from "@factory/shared";
import { createPageUpdatedSource, blockChangedForPageUpdated, type PageUpdatedStreamClient } from "../src/page-updated-source.js";
import type { DomainEvent } from "@factory/shared";

/** A controllable fake Redis stream client (records every XREAD). Honors
 *  `BLOCK`: when nothing is queued it waits `blockMs` (a real
 *  `XREAD BLOCK` blocks up to blockMs; a queued entry is returned
 *  immediately). This paces the poll loop exactly like production. */
function makeFakeClient(blockMs = 3) {
  const calls: (string | number)[][] = [];
  const queue: (Array<[string, Array<[string, string[]]>]> | null)[] = [];
  const client: PageUpdatedStreamClient = {
    // Raw keyword form (XREAD COUNT n BLOCK ms STREAMS key lastId) — the
    // last arg is the cursor, so the cursor-advance assertions below hold.
    async xread(...args: (string | number)[]) {
      calls.push(args);
      if (queue.length === 0) {
        await new Promise((r) => setTimeout(r, blockMs));
        return []; // BLOCK timeout — no entries (ioredis returns [])
      }
      return queue.shift() ?? [];
    },
  };
  return {
    client,
    calls,
    /** Queue one read result (consumed by the next read). */
    prime(result: Array<[string, Array<[string, string[]]>]> | null): void {
      queue.push(result);
    },
  };
}

const sleep = (ms: number): Promise<void> => new Promise((r) => setTimeout(r, ms));
/** Let the (paced) poll loop consume a primed read + settle. */
const settle = (): Promise<void> => sleep(40);

const entry = (id: string, event: string, payload: Record<string, unknown>): [string, string[]] => [
  id,
  ["event", event, "payload", JSON.stringify(payload), "ts", "2026-09-26T00:00:00.000Z", "id", `evt_${id}`],
];

describe("createPageUpdatedSource", () => {
  it("delivers a page.updated entry as a parsed DomainEvent (pageId kept in the payload)", async () => {
    const fake = makeFakeClient();
    const seen: DomainEvent[] = [];
    const source = createPageUpdatedSource((e) => seen.push(e), {
      client: fake.client,
    });
    source.start();
    fake.prime([["factory:events", [entry("1", "page.updated", { pageId: "pg_1", version: 7 })]]]);
    await settle();
    await source.stop();
    expect(seen).toHaveLength(1);
    expect(seen[0]?.name).toBe(DomainEventName.PAGE_UPDATED);
    expect(seen[0]?.payload).toEqual({ pageId: "pg_1", version: 7 });
    expect(seen[0]?.ts).toBe("2026-09-26T00:00:00.000Z");
  });

  it("ignores non-page.updated events (the stream carries the whole domain event bus)", async () => {
    const fake = makeFakeClient();
    const seen: DomainEvent[] = [];
    const source = createPageUpdatedSource((e) => seen.push(e), { client: fake.client });
    source.start();
    fake.prime([
      [
        "factory:events",
        [
          entry("2", "page.created", { pageId: "pg_1" }),
          entry("3", "comment.created", { commentId: "c1" }),
          entry("4", "page.updated", { pageId: "pg_2", version: 1 }),
        ],
      ],
    ]);
    await settle();
    await source.stop();
    expect(seen).toHaveLength(1);
    expect((seen[0]?.payload as { pageId: string }).pageId).toBe("pg_2");
  });

  it("skips a page.updated WITHOUT a pageId (defensive — no crash, no event)", async () => {
    const fake = makeFakeClient();
    const seen: DomainEvent[] = [];
    const source = createPageUpdatedSource((e) => seen.push(e), { client: fake.client });
    source.start();
    fake.prime([["factory:events", [entry("5", "page.updated", { version: 9 })]]]);
    await settle();
    await source.stop();
    expect(seen).toHaveLength(0);
  });

  it("skips a MALFORMED payload (logged via onError, never crashes the loop)", async () => {
    const fake = makeFakeClient();
    const seen: DomainEvent[] = [];
    const errors: unknown[] = [];
    const source = createPageUpdatedSource((e) => seen.push(e), {
      client: fake.client,
      onError: (err) => errors.push(err),
    });
    source.start();
    fake.prime([["factory:events", [["6", ["event", "page.updated", "payload", "{not-json", "ts", "t"]]]]]);
    await settle();
    fake.prime([["factory:events", [entry("7", "page.updated", { pageId: "pg_ok", version: 1 })]]]);
    await settle();
    await source.stop();
    expect(errors).toHaveLength(1);
    // The loop survived the bad entry and still delivered the next one.
    expect(seen).toHaveLength(1);
    expect((seen[0]?.payload as { pageId: string }).pageId).toBe("pg_ok");
  });

  it("advances the cursor past delivered entries (XREAD starts at the last id)", async () => {
    const fake = makeFakeClient();
    const source = createPageUpdatedSource(() => {}, { client: fake.client });
    source.start();
    fake.prime([["factory:events", [entry("42", "page.updated", { pageId: "pg_1" })]]]);
    await settle();
    await source.stop();
    // ioredis form: [count, block, key, lastId] — the last entry is the cursor.
    const reads = fake.calls;
    expect(reads.length).toBeGreaterThanOrEqual(2);
    // The read AFTER the entry must start at "42" (the cursor advanced).
    const after = reads.find((c) => c[c.length - 1] === "42");
    expect(after).toBeDefined();
    expect(reads[0]![reads[0]!.length - 1]).toBe("$");
  });

  it("does nothing after stop() (the loop exits)", async () => {
    const fake = makeFakeClient();
    const seen: DomainEvent[] = [];
    const source = createPageUpdatedSource((e) => seen.push(e), { client: fake.client });
    source.start();
    await source.stop();
    fake.prime([["factory:events", [entry("9", "page.updated", { pageId: "pg_1" })]]]);
    await settle();
    expect(seen).toHaveLength(0);
  });
});

describe("blockChangedForPageUpdated (the (b) fan-out derivation)", () => {
  const event = (payload: Record<string, unknown>, name: string = DomainEventName.PAGE_UPDATED): DomainEvent => ({
    id: "evt_1",
    name: name as DomainEvent["name"],
    payload,
    ts: "2026-09-26T00:00:00.000Z",
  });

  it("derives a block:changed refetch-hint with the advisory version", () => {
    expect(blockChangedForPageUpdated(event({ pageId: "pg_1", version: 7 }))).toEqual({
      type: "block:changed",
      pageId: "pg_1",
      version: 7,
    });
  });

  it("derives the hint WITHOUT a version when the event has none (a REST write)", () => {
    expect(blockChangedForPageUpdated(event({ pageId: "pg_1" }))).toEqual({ type: "block:changed", pageId: "pg_1" });
  });

  it("drops a non-integer version (advisory only — the hint still goes out)", () => {
    expect(blockChangedForPageUpdated(event({ pageId: "pg_1", version: 1.5 }))).toEqual({
      type: "block:changed",
      pageId: "pg_1",
    });
  });

  it("returns null for a non-page.updated event", () => {
    expect(blockChangedForPageUpdated(event({ pageId: "pg_1" }, DomainEventName.PAGE_CREATED))).toBeNull();
  });

  it("returns null when the payload has no pageId", () => {
    expect(blockChangedForPageUpdated(event({ version: 3 }))).toBeNull();
    expect(blockChangedForPageUpdated(event({ pageId: "" }))).toBeNull();
  });
});