/**
 * The `page.updated` event source (TASK-0092 slice 6b — the (b) producer
 * seam).
 *
 * The API publishes every block mutation (REST write AND the CRDT flush)
 * as a `page.updated` domain event onto the Redis stream `factory:events`
 * (apps/api `EventBus`, the same stream the worker consumes). The realtime
 * server consumes that stream and fans a `block:changed` REFETCH-HINT out
 * to the page's control sockets — the web overlay then re-reads the DB
 * (the source of truth at rest, correct timestamps).
 *
 * The source is an INJECTABLE SEAM (the server tests pass a fake; the
 * production wiring in index.ts lazily loads ioredis, mirroring the
 * `RedisUpdateLog` pattern in update-log.ts — the ioredis client
 * satisfies {@link PageUpdatedStreamClient} structurally). The stream
 * read itself is the Redis-gated part; every decision made from an event
 * (which sockets get the hint) is pure and tested without Redis.
 */
import {
  DomainEventName,
  type DomainEvent,
} from "@factory/shared";
import { BlockChangedFrameSchema, type BlockChangedFrame } from "@factory/shared/realtime/frames";

/** The Redis stream the API's event bus appends to (single source of the
 *  name — keep in lockstep with `EVENTS_STREAM` in apps/api constants). */
export const PAGE_UPDATED_STREAM = "factory:events";

/** The minimal ioredis surface the source needs (a real `Redis` satisfies
 *  it structurally; the integration tests inject a fake). The signature is
 *  the RAW keyword form ioredis forwards to Redis (`XREAD COUNT n BLOCK ms
 *  STREAMS key lastId`) — ioredis's typed overloads accept the literal
 *  `"STREAMS"`/`"BLOCK"` tokens as the leading args, and that is exactly
 *  what `readOnce` below sends. */
export interface PageUpdatedStreamClient {
  /** `XREAD COUNT {count} BLOCK {block} STREAMS {key} {lastId}` →
   *  `[[key, [[id, [f1,v1,...]]]], ...]` (an empty array on BLOCK timeout). */
  xread(
    ...args: (string | number)[]
  ): Promise<Array<[string, Array<[string, string[]]>]> | null>;
}

export interface PageUpdatedSourceOptions {
  readonly client: PageUpdatedStreamClient;
  /** `XREAD BLOCK` wait (ms). Default 5000. */
  readonly blockMs?: number;
  /** Max entries per read. Default 100. */
  readonly count?: number;
  /** Log line for errors (a throw would kill the loop). */
  readonly onError?: (err: unknown) => void;
  /** The last stream entry id consumed (`$` = start from now). */
  readonly lastId?: () => string;
}

export interface PageUpdatedSource {
  /** Begin the poll loop (no-op if already started). Returns when the
   *  source is disposed. */
  start(): void;
  /** Stop the loop (resolves when the in-flight read settles). */
  stop(): Promise<void>;
}

/**
 * A stream-polling `page.updated` source: `XREAD` in a loop, parse each
 * envelope (`{ event, payload(JSON), ts, id }`), keep the `page.updated`
 * ones (a `pageId` in the payload), and hand the parsed
 * {@link DomainEvent} to `onPageUpdated`.
 *
 * Envelope parsing mirrors the API's `EventBus.publish` field names and
 * the worker's `eventEnvelopeSchema` (event / payload / ts). `payload` is
 * re-parsed here and each consumer validates its own shape (defense in
 * depth) — a malformed entry is logged + SKIPPED, never a crash.
 */
export function createPageUpdatedSource(
  onPageUpdated: (event: DomainEvent) => void,
  opts: PageUpdatedSourceOptions,
): PageUpdatedSource {
  const blockMs = opts.blockMs ?? 5000;
  const count = opts.count ?? 100;
  const onError = opts.onError ?? ((err: unknown) => console.error("page-updated-source:", err));
  const lastId = opts.lastId ?? (() => "$");

  let running = false;
  let stopped = false;
  let cursor = lastId();
  let inFlight: Promise<void> | null = null;

  const readOnce = async (): Promise<void> => {
    // Raw keyword form — ioredis forwards the args to Redis verbatim
    // (XREAD COUNT n BLOCK ms STREAMS key lastId); the fake and a real
    // client both see the same shape. Empty array / null = BLOCK timeout.
    const res = await opts.client.xread(
      "COUNT",
      count,
      "BLOCK",
      blockMs,
      "STREAMS",
      PAGE_UPDATED_STREAM,
      cursor,
    );
    if (res === null || res.length === 0) return;
    for (const [, entries] of res) {
      for (const [id, fields] of entries) {
        cursor = id;
        const body: Record<string, string> = {};
        for (let i = 0; i + 1 < fields.length; i += 2) {
          body[fields[i]!] = fields[i + 1]!;
        }
        if (body.event !== DomainEventName.PAGE_UPDATED) continue;
        let payload: Record<string, unknown>;
        try {
          payload = JSON.parse(body.payload ?? "{}") as Record<string, unknown>;
        } catch (err) {
          onError(err);
          continue; // malformed payload — skip, never crash the loop
        }
        const pageId = payload.pageId;
        if (typeof pageId !== "string" || pageId.length === 0) continue;
        onPageUpdated({
          id: body.id ?? id,
          name: DomainEventName.PAGE_UPDATED,
          payload,
          ts: body.ts ?? new Date().toISOString(),
        });
      }
    }
  };

  const tick = async (): Promise<void> => {
    while (running && !stopped) {
      try {
        await readOnce();
      } catch (err) {
        onError(err);
        // Back off so a dead Redis doesn't hot-loop the error path.
        await new Promise((r) => setTimeout(r, 1000));
      }
    }
  };

  return {
    start(): void {
      if (running || stopped) return;
      running = true;
      inFlight = tick();
      inFlight.finally(() => {
        inFlight = null;
      });
    },
    async stop(): Promise<void> {
      if (stopped) return;
      stopped = true;
      running = false;
      if (inFlight !== null) await inFlight;
    },
  };
}

/**
 * The (b) fan-out: the `block:changed` REFETCH-HINT for a `page.updated`
 * event. Pure — no sockets, no Redis (unit-testable):
 *
 *  - a `page.updated` WITHOUT a `pageId` is a no-op (defensive; every
 *    producer carries one);
 *  - `version` is ADVISORY (int ≥ 1 only) — the client MUST NOT gate on
 *    it (the refetch is authoritative); a missing/non-integer version is
 *    dropped from the frame, never a reason to skip the hint;
 *  - the returned frame is the SHARED `BlockChangedFrame` (the single
 *    source of truth the server emits and the web consumes — slice 6a).
 */
export function blockChangedForPageUpdated(event: DomainEvent): BlockChangedFrame | null {
  if (event.name !== DomainEventName.PAGE_UPDATED) return null;
  const payload = event.payload as { pageId?: unknown; version?: unknown } | null;
  const pageId = payload?.pageId;
  if (typeof pageId !== "string" || pageId.length === 0) return null;
  const version = payload?.version;
  const frame = BlockChangedFrameSchema.safeParse(
    version !== undefined ? { type: "block:changed", pageId, version } : { type: "block:changed", pageId },
  );
  if (frame.success) return frame.data;
  // A non-integer version is advisory noise — DROP it, still emit the
  // hint (a bad version is never a reason to skip the refetch).
  const fallback = BlockChangedFrameSchema.safeParse({ type: "block:changed", pageId });
  return fallback.success ? fallback.data : null;
}