/**
 * @factory/realtime — the generic `factory:events` stream source
 * (TASK-0094; generalizes the TASK-0092 6b `page-updated-source`).
 *
 * The API publishes EVERY domain event onto the Redis stream
 * `factory:events` (apps/api `EventBus`). The realtime app polls that ONE
 * stream and routes events to fan-outs:
 *   - `page.updated`         → `block:changed` refetch-hint (TASK-0092 6b)
 *   - `comment.*`            → `comment.event` room frames (TASK-0094 R1)
 *   - `notification.created` → `notification.push` control-room push (R7)
 *
 * This module is the shared STREAM POLLING core (the `XREAD` loop, the
 * envelope parse, the malformed-entry skip). It is an INJECTABLE SEAM (the
 * server tests pass a fake client; the production wiring in index.ts lazily
 * loads ioredis). The stream read is the Redis-gated part; every decision
 * made from an event is pure and tested without Redis (the per-event
 * builders live in `comments.ts` / `notifications.ts` /
 * `page-updated-source.ts`).
 *
 * The client interface is the SAME as the old `PageUpdatedStreamClient`
 * (kept exported from `page-updated-source.ts` for lockstep — a real
 * `ioredis` `Redis` satisfies it structurally).
 */

import type { DomainEvent } from "@factory/shared";

/** 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 DOMAIN_EVENTS_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`). */
export interface DomainEventStreamClient {
  /** `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 DomainEventSourceOptions {
  readonly client: DomainEventStreamClient;
  /** `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 DomainEventSource {
  /** Begin the poll loop (no-op if already started). Returns nothing. */
  start(): void;
  /** Stop the loop (resolves when the in-flight read settles). */
  stop(): Promise<void>;
}

/**
 * A stream-polling domain-event source: `XREAD` in a loop, parse each
 * envelope (`{ event, payload(JSON), ts, id }`), and hand the parsed
 * {@link DomainEvent} to `onEvent` for EVERY event name (the caller
 * routes by `event.name`).
 *
 * 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. The
 * envelope carries no top-level `actorId`/`workspaceId` (the API's bus
 * does not emit them); the per-event payload carries whatever the emitter
 * included (TASK-0094 emitters include `actorId` — recorded).
 */
export function createDomainEventSource(
  onEvent: (event: DomainEvent) => void,
  opts: DomainEventSourceOptions,
): DomainEventSource {
  const blockMs = opts.blockMs ?? 5000;
  const count = opts.count ?? 100;
  const onError = opts.onError ?? ((err: unknown) => console.error("domain-event-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",
      DOMAIN_EVENTS_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]!;
        }
        const name = body.event;
        if (typeof name !== "string" || name.length === 0) 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
        }
        onEvent({
          id: body.id ?? id,
          name: name as DomainEvent["name"],
          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;
    },
  };
}