import { spawn as nodeSpawn } from "node:child_process";
import { createHash } from "node:crypto";
import type { StatusEnvelope, StatusState } from "./status-sync.ts";

interface StateWriterStdin {
  readonly writable: boolean;
  write(chunk: string): unknown;
  end(): unknown;
  on(event: "error", listener: (error: Error) => void): this;
}

export interface StateWriterChild {
  readonly stdin: StateWriterStdin;
  on(event: "error", listener: (error: Error) => void): this;
  on(event: "exit", listener: (code: number | null, signal: NodeJS.Signals | null) => void): this;
  kill(signal?: NodeJS.Signals): boolean;
}

export type StateWriterSpawn = (
  command: string,
  args: readonly string[],
  options: Record<string, unknown>,
) => StateWriterChild;

export interface CreatePiStatefileWriterOptions {
  command: string;
  brokerRoot: string;
  sessionId: string;
  alias: string;
  extensionVersion: string;
  env?: NodeJS.ProcessEnv;
  pid?: number;
  now?: () => number;
  heartbeatIntervalMs?: number;
  retryDelayMs?: number;
  killGraceMs?: number;
  /**
   * Prefer one process-wide multi-key sink (`stream-write-statefiles`) shared by
   * all writers using the same command. Falls back to per-identity
   * `stream-write-statefile` when multi is disabled or marked unsupported.
   * Default true.
   */
  preferMultiKey?: boolean;
  /** Test/support override: force multi unsupported so single path is used. */
  multiKeySupported?: boolean;
  spawn?: StateWriterSpawn;
  setInterval?: (fn: () => void, ms: number) => ReturnType<typeof setInterval>;
  clearInterval?: (handle: ReturnType<typeof setInterval>) => void;
  setTimeout?: (fn: () => void, ms: number) => ReturnType<typeof setTimeout>;
  clearTimeout?: (handle: ReturnType<typeof setTimeout>) => void;
}

export interface PiStatefileWriter {
  readonly instanceKey: string;
  /** True when this writer shares a multi-key child process. */
  readonly multiKey: boolean;
  record(envelope: StatusEnvelope): void;
  updateAlias(alias: string): void;
  dispose(): void;
}

interface WriterState {
  sessionId: string;
  alias: string;
  extensionVersion: string;
  instanceKey: string;
  pid: number;
  startedAt: string;
  heartbeatIntervalMs: number;
  status: { state: StatusState; since: string };
}

/** Peer-confirmed safe key grammar for multi-key sink path components. */
const SAFE_INSTANCE_KEY = /^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$/;

/**
 * Validate an instance key as a single safe path component (c2c multi-key contract).
 * Rejects empty, `.`, `..`, and path separators.
 */
export function isSafeInstanceKey(key: string): boolean {
  if (!key || key === "." || key === "..") return false;
  if (key.includes("/") || key.includes("\\") || key.includes("\0")) return false;
  return SAFE_INSTANCE_KEY.test(key);
}

/**
 * Derive a host-safe c2c statefile instance key for one broker session.
 * Aliases are intentionally excluded because they are mutable and only
 * broker-local unique.
 */
export function deriveStatefileInstanceKey(brokerRoot: string, sessionId: string): string {
  const digest = createHash("sha256")
    .update(brokerRoot)
    .update("\0")
    .update(sessionId)
    .digest("hex");
  return `pi-${digest.slice(0, 20)}`;
}

function buildSnapshot(state: WriterState, emittedAt: string): Record<string, unknown> {
  const activity = {
    source_type: "pi",
    first_active_at: state.startedAt,
    last_active_at: emittedAt,
    heartbeat_interval_ms: state.heartbeatIntervalMs,
  };
  return {
    event: "state.snapshot",
    ts: emittedAt,
    state: {
      c2c_session_id: state.sessionId,
      c2c_alias: state.alias,
      state_last_updated_at: emittedAt,
      activity_sources: { pi: activity },
      agent: {
        is_idle: state.status.state === "idle",
        last_step: {
          event_type: `pi.${state.status.state}`,
          at: state.status.since,
          details: { state: state.status.state },
        },
      },
      pi: {
        schema_version: 1,
        pid: state.pid,
        extension_version: state.extensionVersion,
        instance_key: state.instanceKey,
        started_at: state.startedAt,
      },
    },
  };
}

const defaultSpawn: StateWriterSpawn = (command, args, options) =>
  nodeSpawn(command, [...args], options) as unknown as StateWriterChild;

/** Process-wide multi-key support cache per command path. */
const multiSupportByCommand = new Map<string, boolean>();

/** Process-wide hubs: one multi-key child per c2c command binary. */
const multiHubs = new Map<string, MultiKeyHub>();

/** Test helper: clear process-wide multi-key hub/support state. */
export function resetStatefileWriterGlobalsForTests(): void {
  for (const hub of multiHubs.values()) hub.forceDisposeAll();
  multiHubs.clear();
  multiSupportByCommand.clear();
}

interface MultiKeyHub {
  readonly command: string;
  attach(writer: SingleWriterHandle): void;
  detach(instanceKey: string): void;
  writeSnapshot(instanceKey: string, snapshot: Record<string, unknown>): void;
  forceDisposeAll(): void;
}

interface SingleWriterHandle {
  instanceKey: string;
  state: WriterState;
  disposed: boolean;
  writeCurrent: () => void;
  /** Switch this identity to a per-key single writer (multi unsupported). */
  fallBackToSingle: () => void;
}

function createMultiKeyHub(
  command: string,
  options: {
    env?: NodeJS.ProcessEnv;
    spawn: StateWriterSpawn;
    retryDelayMs: number;
    killGraceMs: number;
    setTimeout: (fn: () => void, ms: number) => ReturnType<typeof setTimeout>;
    clearTimeout: (handle: ReturnType<typeof setTimeout>) => void;
  },
): MultiKeyHub {
  const clients = new Map<string, SingleWriterHandle>();
  let disposed = false;
  let generation = 0;
  let child: StateWriterChild | null = null;
  let retryTimer: ReturnType<typeof setTimeout> | null = null;
  const retirementTimers = new Map<StateWriterChild, ReturnType<typeof setTimeout>>();
  let consecutiveConnectFailures = 0;

  function notifyAllFallback(): void {
    multiSupportByCommand.set(command, false);
    const snapshot = [...clients.values()];
    clients.clear();
    if (retryTimer) {
      options.clearTimeout(retryTimer);
      retryTimer = null;
    }
    const active = child;
    child = null;
    if (active) retire(active);
    disposed = true;
    multiHubs.delete(command);
    for (const client of snapshot) {
      if (!client.disposed) client.fallBackToSingle();
    }
  }

  function scheduleRetry(): void {
    if (disposed || retryTimer || clients.size === 0) return;
    retryTimer = options.setTimeout(() => {
      retryTimer = null;
      connect();
    }, options.retryDelayMs);
  }

  function finishRetirement(retiredChild: StateWriterChild): void {
    const timer = retirementTimers.get(retiredChild);
    if (!timer) return;
    options.clearTimeout(timer);
    retirementTimers.delete(retiredChild);
  }

  function retire(retiredChild: StateWriterChild): void {
    if (retirementTimers.has(retiredChild)) return;
    try { retiredChild.stdin.end(); } catch { /* best-effort */ }
    const timer = options.setTimeout(() => {
      retirementTimers.delete(retiredChild);
      try { retiredChild.kill("SIGTERM"); } catch { /* best-effort */ }
    }, options.killGraceMs);
    retirementTimers.set(retiredChild, timer);
    (timer as ReturnType<typeof setTimeout> & { unref?: () => void }).unref?.();
  }

  function fail(
    failedChild: StateWriterChild,
    failedGeneration: number,
    needsRetirement = true,
  ): void {
    if (disposed || failedGeneration !== generation || child !== failedChild) return;
    child = null;
    if (needsRetirement) retire(failedChild);
    consecutiveConnectFailures += 1;
    // Unknown command / hard fail: mark multi unsupported after first exit without clients writing.
    if (consecutiveConnectFailures >= 2) {
      notifyAllFallback();
      return;
    }
    scheduleRetry();
  }

  function writeLine(instanceKey: string, snapshot: Record<string, unknown>): void {
    const target = child;
    if (disposed || !target?.stdin.writable) return;
    if (!isSafeInstanceKey(instanceKey)) return;
    try {
      target.stdin.write(`${JSON.stringify({ ...snapshot, instance_key: instanceKey })}\n`);
      consecutiveConnectFailures = 0;
    } catch {
      fail(target, generation);
    }
  }

  function flushAll(): void {
    for (const client of clients.values()) {
      if (!client.disposed) client.writeCurrent();
    }
  }

  function connect(): void {
    if (disposed || clients.size === 0) return;
    const currentGeneration = ++generation;
    let next: StateWriterChild;
    try {
      const env = { ...(options.env ?? process.env) };
      delete env.C2C_INSTANCE_NAME;
      next = options.spawn(
        command,
        ["oc-plugin", "stream-write-statefiles"],
        {
          env,
          shell: false,
          stdio: ["pipe", "ignore", "ignore"],
        },
      );
    } catch {
      consecutiveConnectFailures += 1;
      if (consecutiveConnectFailures >= 2) {
        notifyAllFallback();
        return;
      }
      scheduleRetry();
      return;
    }

    child = next;
    multiSupportByCommand.set(command, true);
    let failed = false;
    const handleFailure = (): void => {
      if (failed) return;
      failed = true;
      fail(next, currentGeneration, true);
    };
    next.on("error", handleFailure);
    next.stdin.on("error", handleFailure);
    next.on("exit", (code) => {
      finishRetirement(next);
      if (failed || disposed) return;
      failed = true;
      // Immediate exit with usage/error codes → multi unsupported.
      if (code === 124 || code === 123 || code === 1 || code === 2) {
        child = null;
        notifyAllFallback();
        return;
      }
      fail(next, currentGeneration, false);
    });
    flushAll();
  }

  function maybeConnect(): void {
    if (!child && !retryTimer && clients.size > 0) connect();
  }

  function shutdownIfEmpty(): void {
    if (clients.size > 0) return;
    disposed = true;
    if (retryTimer) {
      options.clearTimeout(retryTimer);
      retryTimer = null;
    }
    const active = child;
    child = null;
    if (active) retire(active);
    multiHubs.delete(command);
  }

  return {
    command,
    attach(writer: SingleWriterHandle): void {
      if (disposed) {
        // Hub was empty-shutdown; recreate is handled by getOrCreate.
        return;
      }
      clients.set(writer.instanceKey, writer);
      maybeConnect();
      writer.writeCurrent();
    },
    detach(instanceKey: string): void {
      clients.delete(instanceKey);
      shutdownIfEmpty();
    },
    writeSnapshot(instanceKey: string, snapshot: Record<string, unknown>): void {
      writeLine(instanceKey, snapshot);
    },
    forceDisposeAll(): void {
      clients.clear();
      shutdownIfEmpty();
    },
  };
}

function getOrCreateMultiHub(
  command: string,
  options: Parameters<typeof createMultiKeyHub>[1],
): MultiKeyHub {
  let hub = multiHubs.get(command);
  if (!hub) {
    hub = createMultiKeyHub(command, options);
    multiHubs.set(command, hub);
  }
  return hub;
}

function createSingleKeyWriter(
  options: CreatePiStatefileWriterOptions,
  instanceKey: string,
  state: WriterState,
): PiStatefileWriter {
  const now = options.now ?? Date.now;
  const spawnWriter = options.spawn ?? defaultSpawn;
  const scheduleHeartbeat = options.setInterval ?? setInterval;
  const clearHeartbeat = options.clearInterval ?? clearInterval;
  const scheduleTimeout = options.setTimeout ?? setTimeout;
  const clearScheduledTimeout = options.clearTimeout ?? clearTimeout;
  const retryDelayMs = options.retryDelayMs ?? 10_000;
  const killGraceMs = options.killGraceMs ?? 1_000;

  let disposed = false;
  let generation = 0;
  let child: StateWriterChild | null = null;
  let retryTimer: ReturnType<typeof setTimeout> | null = null;
  const retirementTimers = new Map<StateWriterChild, ReturnType<typeof setTimeout>>();

  function scheduleRetry(): void {
    if (disposed || retryTimer) return;
    retryTimer = scheduleTimeout(() => {
      retryTimer = null;
      connect();
    }, retryDelayMs);
  }

  function finishRetirement(retiredChild: StateWriterChild): void {
    const timer = retirementTimers.get(retiredChild);
    if (!timer) return;
    clearScheduledTimeout(timer);
    retirementTimers.delete(retiredChild);
  }

  function retire(retiredChild: StateWriterChild): void {
    if (retirementTimers.has(retiredChild)) return;
    try { retiredChild.stdin.end(); } catch { /* best-effort */ }
    const timer = scheduleTimeout(() => {
      retirementTimers.delete(retiredChild);
      try { retiredChild.kill("SIGTERM"); } catch { /* best-effort */ }
    }, killGraceMs);
    retirementTimers.set(retiredChild, timer);
    (timer as ReturnType<typeof setTimeout> & { unref?: () => void }).unref?.();
  }

  function fail(
    failedChild: StateWriterChild,
    failedGeneration: number,
    needsRetirement = true,
  ): void {
    if (disposed || failedGeneration !== generation || child !== failedChild) return;
    child = null;
    if (needsRetirement) retire(failedChild);
    scheduleRetry();
  }

  function writeCurrent(): void {
    const target = child;
    if (disposed || !target?.stdin.writable) return;
    const emittedAt = new Date(now()).toISOString();
    try {
      target.stdin.write(`${JSON.stringify(buildSnapshot(state, emittedAt))}\n`);
    } catch {
      fail(target, generation);
    }
  }

  function connect(): void {
    if (disposed) return;
    const currentGeneration = ++generation;
    let next: StateWriterChild;
    try {
      next = spawnWriter(
        options.command,
        ["oc-plugin", "stream-write-statefile"],
        {
          env: { ...(options.env ?? process.env), C2C_INSTANCE_NAME: instanceKey },
          shell: false,
          stdio: ["pipe", "ignore", "ignore"],
        },
      );
    } catch {
      scheduleRetry();
      return;
    }

    child = next;
    let failed = false;
    const handleFailure = (): void => {
      if (failed) return;
      failed = true;
      fail(next, currentGeneration, true);
    };
    next.on("error", handleFailure);
    next.stdin.on("error", handleFailure);
    next.on("exit", () => {
      finishRetirement(next);
      if (failed || disposed) return;
      failed = true;
      fail(next, currentGeneration, false);
    });
    writeCurrent();
  }

  connect();
  const heartbeatTimer = scheduleHeartbeat(writeCurrent, state.heartbeatIntervalMs);

  return {
    instanceKey,
    multiKey: false,
    record(envelope: StatusEnvelope): void {
      if (disposed) return;
      state.status = { state: envelope.state, since: envelope.since };
      writeCurrent();
    },
    updateAlias(alias: string): void {
      if (disposed || alias === state.alias) return;
      state.alias = alias;
      writeCurrent();
    },
    dispose(): void {
      if (disposed) return;
      disposed = true;
      clearHeartbeat(heartbeatTimer);
      if (retryTimer) {
        clearScheduledTimeout(retryTimer);
        retryTimer = null;
      }
      const active = child;
      child = null;
      if (active) retire(active);
    },
  };
}

export function createPiStatefileWriter(
  options: CreatePiStatefileWriterOptions,
): PiStatefileWriter {
  const now = options.now ?? Date.now;
  const startedAt = new Date(now()).toISOString();
  const instanceKey = deriveStatefileInstanceKey(options.brokerRoot, options.sessionId);
  const state: WriterState = {
    sessionId: options.sessionId,
    alias: options.alias,
    extensionVersion: options.extensionVersion,
    instanceKey,
    pid: options.pid ?? process.pid,
    startedAt,
    heartbeatIntervalMs: options.heartbeatIntervalMs ?? 10_000,
    status: { state: "idle", since: startedAt },
  };

  const preferMulti = options.preferMultiKey !== false;
  const supportKnown = options.multiKeySupported ?? multiSupportByCommand.get(options.command);
  const useMulti = preferMulti && supportKnown !== false && isSafeInstanceKey(instanceKey);

  if (!useMulti) {
    return createSingleKeyWriter(options, instanceKey, state);
  }

  const scheduleHeartbeat = options.setInterval ?? setInterval;
  const clearHeartbeat = options.clearInterval ?? clearInterval;
  const scheduleTimeout = options.setTimeout ?? setTimeout;
  const clearScheduledTimeout = options.clearTimeout ?? clearTimeout;
  const spawnWriter = options.spawn ?? defaultSpawn;

  let disposed = false;
  let usingMulti = true;
  let singleFallback: PiStatefileWriter | null = null;

  function fallBackToSingle(): void {
    if (!usingMulti || disposed) return;
    usingMulti = false;
    // Hub has already detached all clients when multi became unsupported.
    singleFallback = createSingleKeyWriter(options, instanceKey, state);
  }

  const handle: SingleWriterHandle = {
    instanceKey,
    state,
    disposed: false,
    writeCurrent(): void {
      if (disposed || !usingMulti) return;
      const emittedAt = new Date(now()).toISOString();
      const hub = multiHubs.get(options.command);
      hub?.writeSnapshot(instanceKey, buildSnapshot(state, emittedAt));
    },
    fallBackToSingle,
  };

  const hub = getOrCreateMultiHub(options.command, {
    env: options.env,
    spawn: spawnWriter,
    retryDelayMs: options.retryDelayMs ?? 10_000,
    killGraceMs: options.killGraceMs ?? 1_000,
    setTimeout: scheduleTimeout,
    clearTimeout: clearScheduledTimeout,
  });
  hub.attach(handle);

  const heartbeatTimer = scheduleHeartbeat(() => {
    if (disposed) return;
    if (usingMulti) handle.writeCurrent();
  }, state.heartbeatIntervalMs);

  return {
    instanceKey,
    get multiKey() {
      return usingMulti && !disposed;
    },
    record(envelope: StatusEnvelope): void {
      if (disposed) return;
      state.status = { state: envelope.state, since: envelope.since };
      if (usingMulti) handle.writeCurrent();
      else singleFallback?.record(envelope);
    },
    updateAlias(alias: string): void {
      if (disposed) return;
      if (usingMulti) {
        if (alias === state.alias) return;
        state.alias = alias;
        handle.writeCurrent();
      } else {
        // Must not pre-mutate state.alias: single updateAlias early-returns when
        // alias already matches and would skip emitting a snapshot.
        singleFallback?.updateAlias(alias);
      }
    },
    dispose(): void {
      if (disposed) return;
      disposed = true;
      handle.disposed = true;
      clearHeartbeat(heartbeatTimer);
      if (usingMulti) multiHubs.get(options.command)?.detach(instanceKey);
      singleFallback?.dispose();
      singleFallback = null;
    },
  };
}