/**
 * Unit tests for the GENERIC domain-event source (TASK-0094; generalizes
 * the TASK-0092 6b `page-updated-source`).
 *
 * The stream read is faked (a hand-rolled {@link DomainEventStreamClient});
 * the decisions are pure and asserted here without Redis: the generic
 * source delivers EVERY event (routing happens in the server's sink),
 * skips malformed entries, and advances the cursor.
 */
import { describe, expect, it } from "vitest";

import {
  createDomainEventSource,
  DOMAIN_EVENTS_STREAM,
  type DomainEventStreamClient,
} from "../src/event-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: DomainEventStreamClient = {
    async xread(...args: (string | number)[]) {
      calls.push(args);
      if (queue.length === 0) {
        await new Promise((r) => setTimeout(r, blockMs));
        return [];
      }
      return queue.shift() ?? [];
    },
  };
  return {
    client,
    calls,
    prime(result: Array<[string, Array<[string, string[]]>]> | null): void {
      queue.push(result);
    },
  };
}

const sleep = (ms: number): Promise<void> => new Promise((r) => setTimeout(r, ms));
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-27T00:00:00.000Z",
    "id",
    `evt_${id}`,
  ],
];

describe("createDomainEventSource (TASK-0094)", () => {
  it("reads the factory:events stream (the API bus's stream)", async () => {
    const fake = makeFakeClient();
    const source = createDomainEventSource(() => {}, { client: fake.client });
    source.start();
    await settle();
    await source.stop();
    expect(fake.calls.length).toBeGreaterThanOrEqual(1);
    const first = fake.calls[0]!;
    // XREAD COUNT n BLOCK ms STREAMS key lastId
    expect(first[0]).toBe("COUNT");
    expect(first[2]).toBe("BLOCK");
    expect(first[4]).toBe("STREAMS");
    expect(first[5]).toBe(DOMAIN_EVENTS_STREAM);
  });

  it("delivers EVERY event name (the server's sink routes; the source does not filter)", async () => {
    const fake = makeFakeClient();
    const seen: DomainEvent[] = [];
    const source = createDomainEventSource((e) => seen.push(e), { client: fake.client });
    source.start();
    fake.prime([
      [
        DOMAIN_EVENTS_STREAM,
        [
          entry("1", "page.updated", { pageId: "pg_1" }),
          entry("2", "comment.created", { commentId: "c1", pageId: "pg_1" }),
          entry("3", "notification.created", { notificationId: "n1", userId: "u1" }),
          entry("4", "record.created", { recordId: "r1" }),
        ],
      ],
    ]);
    await settle();
    await source.stop();
    expect(seen.map((e) => e.name)).toEqual([
      "page.updated",
      "comment.created",
      "notification.created",
      "record.created",
    ]);
    expect((seen[1]!.payload as { commentId: string }).commentId).toBe("c1");
  });

  it("keeps the parsed payload + ts + id of the envelope", async () => {
    const fake = makeFakeClient();
    const seen: DomainEvent[] = [];
    const source = createDomainEventSource((e) => seen.push(e), { client: fake.client });
    source.start();
    fake.prime([
      [
        DOMAIN_EVENTS_STREAM,
        [
          entry("7", "comment.created", {
            commentId: "c9",
            pageId: "pg_9",
            actorId: "u9",
            body: "hi",
          }),
        ],
      ],
    ]);
    await settle();
    await source.stop();
    expect(seen).toHaveLength(1);
    expect(seen[0]!.payload).toEqual({
      commentId: "c9",
      pageId: "pg_9",
      actorId: "u9",
      body: "hi",
    });
    expect(seen[0]!.ts).toBe("2026-09-27T00:00:00.000Z");
    expect(seen[0]!.id).toBe("evt_7");
  });

  it("skips a MALFORMED payload (logged via onError, never crashes the loop)", async () => {
    const fake = makeFakeClient();
    const seen: DomainEvent[] = [];
    const errors: unknown[] = [];
    const source = createDomainEventSource((e) => seen.push(e), {
      client: fake.client,
      onError: (err) => errors.push(err),
    });
    source.start();
    fake.prime([
      [
        DOMAIN_EVENTS_STREAM,
        [["6", ["event", "comment.created", "payload", "{not-json", "ts", "t"]]],
      ],
    ]);
    await settle();
    fake.prime([
      [DOMAIN_EVENTS_STREAM, [entry("8", "comment.created", { commentId: "ok", pageId: "p" })]],
    ]);
    await settle();
    await source.stop();
    expect(errors).toHaveLength(1);
    expect(seen).toHaveLength(1);
    expect((seen[0]!.payload as { commentId: string }).commentId).toBe("ok");
  });

  it("skips an entry WITHOUT an event name", async () => {
    const fake = makeFakeClient();
    const seen: DomainEvent[] = [];
    const source = createDomainEventSource((e) => seen.push(e), { client: fake.client });
    source.start();
    fake.prime([[DOMAIN_EVENTS_STREAM, [["11", ["payload", "{}", "ts", "t"]]]]]);
    await settle();
    await source.stop();
    expect(seen).toHaveLength(0);
  });

  it("advances the cursor past delivered entries (XREAD starts at the last id)", async () => {
    const fake = makeFakeClient();
    const source = createDomainEventSource(() => {}, { client: fake.client });
    source.start();
    fake.prime([
      [DOMAIN_EVENTS_STREAM, [entry("42", "comment.created", { commentId: "c", pageId: "p" })]],
    ]);
    await settle();
    await source.stop();
    const reads = fake.calls;
    expect(reads.length).toBeGreaterThanOrEqual(2);
    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 = createDomainEventSource((e) => seen.push(e), { client: fake.client });
    source.start();
    await source.stop();
    fake.prime([
      [DOMAIN_EVENTS_STREAM, [entry("9", "comment.created", { commentId: "c", pageId: "p" })]],
    ]);
    await settle();
    expect(seen).toHaveLength(0);
  });

  it("a dead client (throwing xread) back-offs and does not crash the loop", async () => {
    const errors: unknown[] = [];
    const client: DomainEventStreamClient = {
      async xread() {
        throw new Error("redis down");
      },
    };
    const seen: DomainEvent[] = [];
    const source = createDomainEventSource((e) => seen.push(e), {
      client,
      onError: (err) => errors.push(err),
    });
    source.start();
    await sleep(30);
    await source.stop();
    expect(errors.length).toBeGreaterThanOrEqual(1);
    expect(seen).toHaveLength(0);
  });
});