/**
 * apps/worker — domain-event stream consumer (Redis Streams, XREADGROUP).
 *
 * The API publishes every domain event to the stream `factory:events` as a
 * flat field map `{event, payload (JSON string), ts}`. Each queue gets its
 * OWN consumer group (`worker-<queue>`), so queues can lag independently
 * without head-of-line blocking each other.
 *
 * Delivery semantics:
 *  - XGROUP CREATE ... MKSTREAM (tolerates BUSYGROUP on restart),
 *  - loop: XREADGROUP GROUP g c COUNT 50 BLOCK 5000 → onEvent(id, envelope)
 *    → XACK on success; on error the entry stays PENDING (not acked) and is
 *    redelivered by the XAUTOCLAIM loop below — at-least-once, exactly-once
 *    is the processor's job (all handlers must be idempotent).
 *  - XAUTOCLAIM loop every 60s claims entries idle > 60s (crashed consumer /
 *    died between processing and ACK), max 20 per pass.
 *  - stop() resolves cleanly; in-flight entries finish or stay pending.
 */
import type { Redis } from "ioredis";
import { getLogger } from "../logger.js";

export interface StreamEntry {
  id: string;
  fields: Record<string, string | undefined>;
}

/**
 * XREADGROUP reply: an array of [streamName, entries] pairs where entries is
 * a list of [id, [field, value, …]] pairs — or `null` when the BLOCK window
 * expires with no new entries. (ioredis types the command loosely; we model
 * the shape we actually consume.)
 */
type ReadStreamResult = [string, [string, string[]][]][] | null;

export type StreamEventHandler = (entry: StreamEntry) => Promise<void>;

interface ClaimedEntry {
  id: string;
  fields: Record<string, string | undefined>;
}

const COUNT = 50;
const BLOCK_MS = 5000;
const CLAIM_INTERVAL_MS = 60_000;
const CLAIM_MIN_IDLE_MS = 60_000;
const CLAIM_BATCH = 20;

const log = getLogger("stream");

export class StreamConsumer {
  private stopped = false;
  private processing = 0;
  private claimTimer: NodeJS.Timeout | null = null;
  // Reserved for unblocking an in-flight XREADGROUP on stop(). Currently
  // always null: the read loop exits via the `stopped` flag + the command's
  // own BLOCK expiry, so no resolver is wired. Kept so `stop()`'s guard
  // compiles and a future fast-stop can assign a resolver here.
  private readResolve: (() => void) | null = null;

  constructor(
    private redis: Redis,
    private stream: string,
    private group: string,
    private consumerId: string,
    private onEvent: StreamEventHandler
  ) {}

  private get log() {
    return log.child({ group: this.group });
  }

  /** Idempotent: safe to call stop() multiple times. */
  get isStopped(): boolean {
    return this.stopped;
  }

  async start(): Promise<void> {
    // 1) Ensure the consumer group exists (BUSYGROUP = already there).
    try {
      await this.redis.xgroup("CREATE", this.stream, this.group, "0", "MKSTREAM");
      this.log.info(`consumer group created: ${this.group} @ ${this.stream}`);
    } catch (err) {
      const msg = String((err as Error).message ?? err);
      if (!msg.includes("BUSYGROUP")) throw err;
    }

    // 2) XAUTOCLAIM maintenance loop.
    this.claimTimer = setInterval(() => {
      void this.claimStale();
    }, CLAIM_INTERVAL_MS);
    if (typeof this.claimTimer.unref === "function") this.claimTimer.unref();

    // 3) Read loop.
    void this.readLoop();
  }

  async stop(): Promise<void> {
    if (this.stopped) return;
    this.stopped = true;
    if (this.claimTimer) clearInterval(this.claimTimer);
    // Unblock an in-flight XREADGROUP so the loop can exit.
    if (this.readResolve) {
      this.readResolve();
      this.readResolve = null;
    }
    // Give in-flight handlers a moment to finish (they XACK or stay pending).
    for (let i = 0; i < 50 && this.processing > 0; i++) {
      await new Promise((r) => setTimeout(r, 100));
    }
    this.log.info("consumer stopped");
  }

  private async readLoop(): Promise<void> {
    const lastId = ">";
    while (!this.stopped) {
      let entries: [string, string[]][];
      try {
        const res = await this.readWithStopSignal(lastId);
        // ioredis returns [[streamName, [[id, [k,v,...]], ...]]] (one pair)
        // or null when BLOCK expires with no new entries.
        entries = res?.[0]?.[1] ?? [];
      } catch (err) {
        if (this.stopped) break;
        this.log.warn(
          { err: (err as Error).message },
          "XREADGROUP failed — backoff 2s"
        );
        await sleep(2000);
        continue;
      }
      if (entries.length === 0) continue;

      for (const [id, fields] of entries) {
        if (this.stopped) break;
        const map: Record<string, string | undefined> = {};
        for (let i = 0; i < fields.length; i += 2) {
          const key = fields[i];
          if (key === undefined) continue; // defensive: malformed field list
          map[key] = fields[i + 1];
        }
        const entry: StreamEntry = { id, fields: map };
        this.processing++;
        try {
          await this.onEvent(entry);
          await this.redis.xack(this.stream, this.group, id);
        } catch (err) {
          // NOT acked: entry stays pending → redelivered by XAUTOCLAIM.
          this.log.error(
            { err: (err as Error).message, id, group: this.group },
            "handler failed — leaving entry pending for redelivery"
          );
        } finally {
          this.processing--;
        }
      }
    }
  }

  /**
   * XREADGROUP with the stop-signal trick: a racing `stop()` can't wedge the
   * loop on a 5s BLOCK, because we race the command against a stop flag
   * (the command itself keeps running on the socket, harmlessly).
   */
  private readWithStopSignal(lastId: string): Promise<ReadStreamResult> {
    return new Promise((resolve, reject) => {
      if (this.stopped) {
        resolve(null);
        return;
      }
      const cmd = this.redis.xreadgroup(
        "GROUP",
        this.group,
        this.consumerId,
        "COUNT",
        COUNT,
        "BLOCK",
        BLOCK_MS,
        "STREAMS",
        this.stream,
        lastId
      ) as Promise<ReadStreamResult>;
      cmd.then(resolve).catch(reject);
    });
  }

  /**
   * Claim entries that went idle (crashed / pending > 60s) and re-run them
   * through the same handler. XAUTOCLAIM returns [nextStartId, entries,
   * deletedIds].
   */
  private async claimStale(): Promise<void> {
    if (this.stopped) return;
    try {
      const res = await this.redis.call(
        "XAUTOCLAIM",
        this.stream,
        this.group,
        this.consumerId,
        CLAIM_MIN_IDLE_MS,
        "0",
        "COUNT",
        CLAIM_BATCH
      ) as [string, ClaimedEntry[], (string | null)[]];
      const entries = Array.isArray(res?.[1]) ? res[1] : [];
      if (entries.length === 0) return;
      this.log.warn({ count: entries.length }, "claiming stale pending entries");
      for (const e of entries) {
        if (this.stopped) break;
        this.processing++;
        try {
          await this.onEvent(e);
          await this.redis.xack(this.stream, this.group, e.id);
        } catch (err) {
          this.log.error(
            { err: (err as Error).message, id: e.id },
            "claimed entry failed again — stays pending"
          );
        } finally {
          this.processing--;
        }
      }
    } catch (err) {
      this.log.warn(
        { err: (err as Error).message },
        "XAUTOCLAIM failed (stream may not exist yet)"
      );
    }
  }
}

function sleep(ms: number): Promise<void> {
  return new Promise((r) => setTimeout(r, ms));
}