import type { KeyObject } from "node:crypto";
import { sandboxRelay, supervisorSandboxEnabled } from "./sandbox";
import { verifyCommand } from "../bus/envelope";
import { DurableFence } from "../bus/fence";
import type { ObservedStore } from "../bus/redis";
import type { ChannelState, DesiredChannel, ObservedChannel } from "../types";

export interface TransitionLogger {
  transition(event: { workspaceId: string; channelId: string; from: ChannelState; to: ChannelState; reason: string }): void;
}
interface ChannelRuntime {
  desired: DesiredChannel;
  state: ChannelState;
  reason: string;
  realizedVersion: number | null;
  relay: RelayJob | null;
  retries: number;
  retryTimer: { cancel(): void } | null;
  /** Consecutive failed swaps for the current desired version (the previous version stays on air). */
  swapFailures: number;
}
export interface RelayJob {
  start(channel: DesiredChannel): Promise<void>;
  stop(): Promise<void>;
  /**
   * Optional in-place change (production jobs: an A/B compositor swap behind the
   * frame gate). Resolves true once the new version is on air; false means the
   * change needs a restart (sources, output or profile changed).
   */
  reconfigure?(channel: DesiredChannel): Promise<boolean>;
}
export interface RetryScheduler { schedule(action: () => Promise<void>, delayMs: number): { cancel(): void } }

/**
 * A swap that fails (B never went on air; A is still on air) is retried this many times, 1 s then
 * 2 s later, before the supervisor falls back to a full restart. The full restart converges too,
 * but it takes the encoder and the push session down (WP9 "output that does not stop").
 */
export const SWAP_RETRIES = 2;
const defaultRetryScheduler: RetryScheduler = {
  schedule(action, delayMs) {
    const timer = setTimeout(() => { void action(); }, delayMs);
    return { cancel: () => clearTimeout(timer) };
  },
};

/** Accepted versions are durable; realized versions describe only this process lifetime. */
export class MediaSupervisor {
  private readonly channels = new Map<string, ChannelRuntime>();
  private readonly operations = new Map<string, Promise<unknown>>();
  private heartbeatTimer: ReturnType<typeof setInterval> | null = null;
  /** Set by `stop()`: late retries and relaunches become no-ops. */
  private stopped = false;
  constructor(
    private readonly workerId: string,
    desired: DesiredChannel[],
    private readonly store: ObservedStore,
    private readonly fence: DurableFence,
    private readonly controlKeys: ReadonlyMap<string, KeyObject>,
    private readonly createRelay: (channel: DesiredChannel, onExit: (reason: string) => void) => RelayJob,
    private readonly logger: TransitionLogger,
    private readonly retryScheduler: RetryScheduler = defaultRetryScheduler,
  ) {
    supervisorSandboxEnabled(); // Production cannot silently opt out of confinement.
    for (const channel of desired) {
      if (channel.workerId !== workerId) throw new Error("config worker mismatch");
      this.channels.set(this.key(channel.workspaceId, channel.channelId), {
        desired: channel, state: "STOPPED", reason: "worker boot", realizedVersion: null,
        relay: null, retries: 0, retryTimer: null, swapFailures: 0,
      });
    }
  }

  private key(workspaceId: string, channelId: string): string { return `${workspaceId}:${channelId}`; }
  status(): ObservedChannel[] {
    return [...this.channels.values()].map((runtime) => ({
      workspaceId: runtime.desired.workspaceId, channelId: runtime.desired.channelId,
      workerId: this.workerId, assignmentEpoch: runtime.desired.assignmentEpoch,
      state: runtime.state, realizedVersion: runtime.realizedVersion,
      updatedAt: new Date().toISOString(), reason: runtime.reason,
    }));
  }
  private async transition(runtime: ChannelRuntime, to: ChannelState, reason: string): Promise<void> {
    const from = runtime.state;
    runtime.state = to;
    runtime.reason = reason;
    this.logger.transition({ workspaceId: runtime.desired.workspaceId, channelId: runtime.desired.channelId, from, to, reason });
    await this.store.put(this.status().find((item) => item.workspaceId === runtime.desired.workspaceId && item.channelId === runtime.desired.channelId)!);
  }

  async start(): Promise<void> {
    for (const runtime of this.channels.values()) {
      const old = await this.store.get(runtime.desired.workspaceId, runtime.desired.channelId);
      await this.transition(runtime, "STOPPED", old
        ? `restart reconcile: prior ${old.state}; awaiting fresh signed snapshot`
        : "start reconcile: awaiting signed snapshot");
    }
    await this.beat();
    this.heartbeatTimer = setInterval(() => { void this.beat().catch((error: unknown) => this.logger.transition({ workspaceId: "", channelId: "", from: "DEGRADED", to: "DEGRADED", reason: `heartbeat failed: ${String(error)}` })); }, 5_000);
  }

  async beat(): Promise<void> { await this.store.heartbeat(this.workerId, this.status()); }

  async handle(raw: unknown, now = Date.now()): Promise<"accepted" | "replay" | "stale"> {
    const command = verifyCommand(raw, this.controlKeys, this.workerId, now);
    const body = command.body;
    return this.serialize(this.key(body.workspaceId, body.channelId), () => this.apply(body, now));
  }

  /** One operation at a time per channel (commands and swap retries). */
  private serialize<T>(key: string, operation: () => Promise<T>): Promise<T> {
    const previous = this.operations.get(key) ?? Promise.resolve();
    const task = previous.catch(() => undefined).then(operation);
    this.operations.set(key, task);
    void task.then(() => { if (this.operations.get(key) === task) this.operations.delete(key); },
      () => { if (this.operations.get(key) === task) this.operations.delete(key); });
    return task;
  }

  private async apply(body: ReturnType<typeof verifyCommand>["body"], now: number): Promise<"accepted" | "replay" | "stale"> {
    const runtime = this.channels.get(this.key(body.workspaceId, body.channelId));
    if (!runtime || runtime.desired.assignmentEpoch !== body.assignmentEpoch) throw new Error("unassigned channel or epoch");
    const result = this.fence.accept(body, now);
    if (result !== "accepted") return result;
    // RUNNING → RUNNING on a job that can swap: the encoder and push session stay up,
    // and the version is realized only once the new compositor is on air.
    const relay = runtime.relay;
    if (runtime.state === "RUNNING" && body.snapshot.desiredStatus === "RUNNING" && relay?.reconfigure) {
      runtime.desired = body.snapshot;
      runtime.retryTimer?.cancel();
      runtime.retryTimer = null;
      runtime.swapFailures = 0;
      if (await this.swap(runtime, relay) !== "restart") return result;
    }
    runtime.desired = body.snapshot;
    runtime.swapFailures = 0;
    runtime.realizedVersion = null;
    runtime.retryTimer?.cancel();
    runtime.retryTimer = null;
    await runtime.relay?.stop();
    runtime.relay = null;
    if (body.snapshot.desiredStatus === "STOPPED") {
      await this.transition(runtime, "STOPPED", "signed desired state stopped");
      runtime.realizedVersion = body.desiredVersion;
      await this.beat();
      return result;
    }
    runtime.retries = 0;
    await this.launch(runtime);
    return result;
  }

  /**
   * One swap attempt for `runtime.desired`. "done": on air (or the job is gone and its own exit
   * path owns the channel); "retrying": B failed, A stays on air (`realizedVersion` keeps naming
   * what viewers see) and a retry is scheduled; "restart": not swappable, or out of retries.
   */
  private async swap(runtime: ChannelRuntime, relay: RelayJob): Promise<"done" | "retrying" | "restart"> {
    const version = runtime.desired.desiredVersion;
    try {
      if (!(await relay.reconfigure?.(runtime.desired))) return "restart";
      if (runtime.relay !== relay) return "done";
      runtime.swapFailures = 0;
      runtime.realizedVersion = version;
      await this.transition(runtime, "RUNNING", "compositor swapped at a frame boundary");
      await this.beat();
      return "done";
    } catch (error: unknown) {
      if (runtime.relay !== relay) return "done";
      const name = error instanceof Error ? error.name : "unknown error";
      runtime.swapFailures += 1;
      if (runtime.swapFailures > SWAP_RETRIES) {
        runtime.reason = `swap failed: ${name}`;
        return "restart";
      }
      const delay = 1_000 * 2 ** (runtime.swapFailures - 1);
      await this.transition(runtime, "RUNNING", `swap to v${version} failed (${name}); v${runtime.realizedVersion ?? "?"} stays on air; retry ${runtime.swapFailures}/${SWAP_RETRIES} in ${delay}ms`);
      const key = this.key(runtime.desired.workspaceId, runtime.desired.channelId);
      runtime.retryTimer = this.retryScheduler.schedule(() => this.serialize(key, async () => {
        runtime.retryTimer = null;
        if (this.stopped) return;
        // A newer command or an exit has taken over since.
        if (runtime.desired.desiredVersion !== version || runtime.relay !== relay || runtime.state !== "RUNNING") return;
        if (await this.swap(runtime, relay) === "restart") await this.restart(runtime);
      }), delay);
      return "retrying";
    }
  }

  /** Full restart for the current desired snapshot (the swap path gave up). */
  private async restart(runtime: ChannelRuntime): Promise<void> {
    runtime.realizedVersion = null;
    runtime.swapFailures = 0;
    await runtime.relay?.stop();
    runtime.relay = null;
    runtime.retries = 0;
    await this.launch(runtime);
  }

  private async launch(runtime: ChannelRuntime): Promise<void> {
    await this.transition(runtime, runtime.retries ? "RESTARTING" : "STARTING", runtime.retries ? "relay retry" : "signed desired state running");
    try {
      const relay = sandboxRelay(this.createRelay(runtime.desired, (reason) => {
        if (runtime.relay === relay) void this.relayExited(runtime, reason);
      }), runtime.desired);
      runtime.relay = relay;
      await relay.start(runtime.desired);
      if (runtime.relay !== relay) return;
      runtime.realizedVersion = runtime.desired.desiredVersion;
      await this.transition(runtime, "RUNNING", "ffmpeg spawned with local relay pipes");
      await this.beat();
    } catch (error: unknown) {
      runtime.relay = null;
      await this.relayExited(runtime, `start failed: ${error instanceof Error ? error.name : "unknown error"}`);
    }
  }

  private async relayExited(runtime: ChannelRuntime, reason: string): Promise<void> {
    if (this.stopped || runtime.desired.desiredStatus !== "RUNNING") return;
    runtime.realizedVersion = null;
    runtime.relay = null;
    await this.transition(runtime, "DEGRADED", reason);
    if (runtime.retries >= 5) { await this.transition(runtime, "FAILED", "relay retry budget exhausted"); return; }
    const delay = Math.min(30_000, 1_000 * 2 ** runtime.retries);
    runtime.retries += 1;
    await this.transition(runtime, "RESTARTING", `retry in ${delay}ms`);
    runtime.retryTimer = this.retryScheduler.schedule(() => { runtime.retryTimer = null; return this.stopped ? Promise.resolve() : this.launch(runtime); }, delay);
  }

  /**
   * Worker shutdown. In-flight per-channel operations (a command, a swap retry, a relaunch) finish
   * first, so none of them can start a job after the channels were stopped.
   */
  async stop(): Promise<void> {
    this.stopped = true;
    if (this.heartbeatTimer) clearInterval(this.heartbeatTimer);
    this.heartbeatTimer = null;
    for (const runtime of this.channels.values()) runtime.retryTimer?.cancel();
    await Promise.allSettled([...this.operations.values()]);
    for (const runtime of this.channels.values()) {
      runtime.retryTimer?.cancel();
      runtime.retryTimer = null;
      runtime.desired = { ...runtime.desired, desiredStatus: "STOPPED" };
      await runtime.relay?.stop();
      runtime.realizedVersion = null;
      await this.transition(runtime, "STOPPED", "worker shutdown");
    }
  }
}