import { spawnSync } from "node:child_process";
import { createHash } from "node:crypto";
import {
  chmodSync,
  closeSync,
  existsSync,
  fsyncSync,
  mkdirSync,
  openSync,
  readFileSync,
  renameSync,
  statSync,
  unlinkSync,
  writeFileSync,
} from "node:fs";
import { hostname } from "node:os";
import { dirname, join, resolve } from "node:path";
import { fileURLToPath } from "node:url";
import { fsyncParentDirectory } from "../../workflow/durable-record.ts";
import { acquireNoClobberLease } from "../../workflow/workspaces/leases.ts";
import { canonicalJsonV3, sha256V3 } from "./canonical.ts";
import { ADAPTER_CAPABILITY_PROFILES_V3 } from "./capabilities.ts";
import {
  type ActivationManifestV3,
  buildActivationManifestV3,
  buildCandidateGenesisManifestV3,
  type CandidateGenesisManifestV3,
  EVENT_V3_ACTIVATION_MANIFEST,
  EVENT_V3_GENESIS_MANIFEST,
  type EventV3ControlState,
  readEventV3ControlState,
} from "./control.ts";
import { repairEventV3ControlPair } from "./control-writer.ts";
import { loadOrCreateFingerprintKeyStoreV3 } from "./fingerprint-keys.ts";
import { EVENT_V3_SCHEMA_DIGEST } from "./generated.ts";
import { reanchorArchivedHookProducersV3 } from "./producers/recorder.ts";
import { resolveEventLedgerRotateActiveBytesV3 } from "./rotation-config.ts";
import { currentHarneryRuntimeBuild, liveEventV3BuildId, livePlatformV3 } from "./runtime-build.ts";
import { drainReadyEventsV3 } from "./writer.ts";

export interface InitializeEventLedgerV3Input {
  coordRoot: string;
  harneryBuild: string;
  hostBuild: string;
  configDigest: `sha256:${string}`;
  approvalRecordId: string;
  forceNewEpoch?: boolean;
  /**
   * Activate a candidate epoch that was already created for this root.
   *
   * Unset, this activates a candidate that carries this initializer's producer
   * identity and refuses every other one. Creation and activation are one
   * locked step here, so such a candidate is the residue of a crash or a
   * failed activation, never work in progress; a candidate from another
   * producer is a deliberate cutover state. Set it explicitly to force either
   * answer.
   */
  resumeCandidate?: boolean;
  now?: () => Date;
}

export interface InitializeEventLedgerV3Result {
  state: "active";
  initialized: boolean;
  archived_epoch?: string;
  control: Extract<EventV3ControlState, { state: "active" }>;
}

export interface RefreshIncompatibleEventLedgerV3Result {
  state: "current" | "refreshed";
  archived_epoch?: string;
  control: Extract<EventV3ControlState, { state: "active" }>;
}

/**
 * The control producer every epoch created by this initializer carries.
 *
 * It separates a stranded epoch from a deliberate one. A candidate minted here
 * was always meant to be activated in the same locked step, so finishing it is
 * a repair. A candidate published by any other producer (an operator cutover or
 * rollback rehearsal) is a state someone chose, and automatic activation would
 * overrule them.
 */
const BOOTSTRAP_CONTROL_PRODUCER_ID = "prd_harnery-init" as const;

const BOOTSTRAP_LEASE_RETRIES = 12;
const BOOTSTRAP_LEASE_RETRY_MS = 25;
const BOOTSTRAP_LEASE_STALE_MS = 10_000;
const bootstrapSleepCell = new Int32Array(new SharedArrayBuffer(4));

/**
 * Ensure programmatic Harnery entry points have the same universal V3
 * boundary as `harn init`. Only a genuinely absent control pair is created;
 * candidate, damaged, or incompatible state still fails closed.
 */
export function ensureEventLedgerV3(
  coordRoot: string,
  approvalRecordId = "harnery-runtime-v3-universal",
): Extract<EventV3ControlState, { state: "active" }> {
  const root = resolve(coordRoot);
  const current = readEventV3ControlState(root);
  if (current.state === "active") return current;
  if (current.state !== "closed") {
    throw new Error(`event_v3_control_unavailable:${current.state}`);
  }
  const configPath = join(root, ".harnery", "config.jsonc");
  return initializeEventLedgerV3({
    coordRoot: root,
    harneryBuild: repositoryBuild(rootOfHarnery()),
    hostBuild: repositoryBuild(root),
    configDigest: sha256V3(existsSync(configPath) ? readFileSync(configPath) : Buffer.from("{}\n")),
    approvalRecordId,
  }).control;
}

/**
 * Ensure a root has one current V3 epoch. An incompatible or explicitly
 * replaced epoch is moved intact to the V3 archive before the new control
 * pair is published. No historical ledger bytes are rewritten or deleted.
 */
export function initializeEventLedgerV3(
  input: InitializeEventLedgerV3Input,
): InitializeEventLedgerV3Result {
  return withBootstrapLease(input.coordRoot, () => initializeEventLedgerV3Locked(input));
}

/**
 * Replace only a runtime-incompatible epoch that the current code can name
 * exactly. Corruption and ambiguous control failures remain closed for the
 * explicit recovery command.
 */
export function refreshIncompatibleEventLedgerV3(
  coordRoot: string,
): RefreshIncompatibleEventLedgerV3Result {
  const root = resolve(coordRoot);
  return withBootstrapLease(root, () => {
    const current = readEventV3ControlState(root);
    if (current.state === "active" && runtimeCapabilityProfileCurrent(current)) {
      return { state: "current", control: current };
    }
    const refreshable =
      (current.state === "invalid" && current.reason === "genesis_schema_digest_incompatible") ||
      ((current.state === "candidate" || current.state === "active") &&
        !runtimeCapabilityProfileCurrent(current));
    if (!refreshable) {
      const reason = "reason" in current ? `${current.state}:${current.reason}` : current.state;
      throw new Error(`event_v3_runtime_refresh_refused:${reason}`);
    }
    const configPath = join(root, ".harnery", "config.jsonc");
    const initialized = initializeEventLedgerV3Locked({
      coordRoot: root,
      harneryBuild: currentHarneryRuntimeBuild(),
      hostBuild: repositoryBuild(root),
      configDigest: sha256V3(
        existsSync(configPath) ? readFileSync(configPath) : Buffer.from("{}\n"),
      ),
      approvalRecordId: "harnery-runtime-v3-auto-refresh",
      forceNewEpoch: true,
    });
    return {
      state: "refreshed",
      ...(initialized.archived_epoch ? { archived_epoch: initialized.archived_epoch } : {}),
      control: initialized.control,
    };
  });
}

export interface RepairStrandedEventLedgerV3CandidateResult {
  state: "repaired" | "not_stranded" | "unavailable";
  reason?: string;
  control: EventV3ControlState;
}

/** The approval record every automatic stranded-candidate repair is bound to. */
export const EVENT_V3_STRANDED_CANDIDATE_APPROVAL_RECORD_ID =
  "harnery-runtime-v3-stranded-candidate" as const;

/**
 * Complete an epoch that was created but never activated.
 *
 * This initializer creates a candidate and activates it inside one lease, so a
 * candidate carrying its producer identity is residue: some process published
 * the genesis and died or failed before publishing the activation, and every
 * hook afterwards served that half-built epoch with a candidate-only write
 * gate. Finishing it is a repair, not a decision, and it is exactly as safe as
 * the creation it completes: the same lease serializes it, the activation is
 * derived from the immutable candidate packet, and no ledger row is edited or
 * synthesized.
 *
 * A candidate from any other producer is a deliberate cutover state and is
 * left untouched. A live bootstrap lease makes this a no-op rather than a
 * failure, so the next boundary retries.
 */
export function repairStrandedEventLedgerV3Candidate(
  coordRoot: string,
): RepairStrandedEventLedgerV3CandidateResult {
  const root = resolve(coordRoot);
  const observed = readEventV3ControlState(root);
  const deliberateCandidate = observed.state === "candidate";
  if (!strandedBootstrapCandidate(observed)) {
    return {
      state: "not_stranded",
      control: observed,
      ...(deliberateCandidate ? { reason: "candidate_not_bootstrap_created" } : {}),
    };
  }
  try {
    return withBootstrapLease(root, () => {
      const current = readEventV3ControlState(root);
      if (!strandedBootstrapCandidate(current)) return { state: "not_stranded", control: current };
      const activated = activateCandidateEpoch(
        root,
        current.genesis,
        EVENT_V3_STRANDED_CANDIDATE_APPROVAL_RECORD_ID,
      );
      return { state: "repaired", control: activated.control };
    });
  } catch (error) {
    return {
      state: "unavailable",
      reason: error instanceof Error ? error.message : String(error),
      control: observed,
    };
  }
}

function strandedBootstrapCandidate(
  control: EventV3ControlState,
): control is Extract<EventV3ControlState, { state: "candidate" }> {
  return (
    control.state === "candidate" &&
    control.genesis.event.producer.producer_id === BOOTSTRAP_CONTROL_PRODUCER_ID
  );
}

export interface RotateOversizedEventLedgerV3Result {
  state: "rotated" | "not_oversized" | "not_active" | "disabled";
  active_bytes: number;
  threshold_bytes: number;
  archived_epoch?: string;
  control?: Extract<EventV3ControlState, { state: "active" }>;
}

/**
 * Archive a valid oversized epoch intact and start a fresh one.
 *
 * Every canonical read validates the complete epoch, and hook producers are
 * one-shot processes, so an unbounded active segment makes every hook's cold
 * read scale with all history. Rotation bounds that cost: the replaced epoch
 * (events, spool, producer states, manifests) moves whole into the V3
 * archive, live sessions re-onboard into the new epoch on their next signal,
 * and the writer's epoch fence keeps in-flight producers of the old epoch
 * from ever committing into the new one. A candidate epoch rotates on the same
 * terms as an active one: its control pair is valid, every reader still
 * validates all of its history, and refusing to bound it is how a stranded
 * epoch grew to 100 MB. Integrity failures stay closed for the explicit
 * recovery command. The threshold comes from the isolated Event V3 config
 * resolver unless the caller pins one; a non-positive threshold disables
 * rotation.
 */
export function rotateOversizedEventLedgerV3(
  coordRoot: string,
  options: { thresholdBytes?: number } = {},
): RotateOversizedEventLedgerV3Result {
  const root = resolve(coordRoot);
  const threshold = options.thresholdBytes ?? resolveEventLedgerRotateActiveBytesV3(root);
  const activePath = join(root, ".harnery", "ledgers", "v3", "active.ndjson");
  const measure = () => {
    try {
      return statSync(activePath).size;
    } catch {
      return 0;
    }
  };
  if (threshold <= 0) {
    return { state: "disabled", active_bytes: measure(), threshold_bytes: threshold };
  }
  if (measure() < threshold) {
    return { state: "not_oversized", active_bytes: measure(), threshold_bytes: threshold };
  }
  return withBootstrapLease(root, () => {
    const activeBytes = measure();
    if (activeBytes < threshold) {
      return { state: "not_oversized", active_bytes: activeBytes, threshold_bytes: threshold };
    }
    const current = readEventV3ControlState(root);
    if (current.state !== "active" && current.state !== "candidate") {
      return { state: "not_active", active_bytes: activeBytes, threshold_bytes: threshold };
    }
    // Flush durable ready rows into the epoch they were produced for, so the
    // archive carries them committed instead of stranded in its spool.
    try {
      drainReadyEventsV3(root);
    } catch {
      // A busy append lease only leaves rows in the archived spool; rotation
      // itself stays safe.
    }
    const configPath = join(root, ".harnery", "config.jsonc");
    const initialized = initializeEventLedgerV3Locked({
      coordRoot: root,
      harneryBuild: currentHarneryRuntimeBuild(),
      hostBuild: repositoryBuild(root),
      configDigest: sha256V3(
        existsSync(configPath) ? readFileSync(configPath) : Buffer.from("{}\n"),
      ),
      approvalRecordId: "harnery-runtime-v3-size-rotation",
      forceNewEpoch: true,
    });
    if (initialized.archived_epoch) {
      reanchorArchivedHookProducersV3({
        coordRoot: root,
        archivedEpoch: initialized.archived_epoch,
        mode: initialized.control.state,
        build_id: liveEventV3BuildId(currentHarneryRuntimeBuild()),
        platform: livePlatformV3(),
      });
    }
    return {
      state: "rotated",
      active_bytes: activeBytes,
      threshold_bytes: threshold,
      ...(initialized.archived_epoch ? { archived_epoch: initialized.archived_epoch } : {}),
      control: initialized.control,
    };
  });
}

function initializeEventLedgerV3Locked(
  input: InitializeEventLedgerV3Input,
): InitializeEventLedgerV3Result {
  const root = resolve(input.coordRoot);
  let current = readEventV3ControlState(root);
  const resumeCandidate = input.resumeCandidate ?? strandedBootstrapCandidate(current);
  if (resumeCandidate && current.state === "repairable") {
    current = repairEventV3ControlPair(root);
  }
  if (!input.forceNewEpoch && current.state === "active") {
    return { state: "active", initialized: false, control: current };
  }
  if (!input.forceNewEpoch && resumeCandidate && current.state === "candidate") {
    return activateCandidateEpoch(root, current.genesis, input.approvalRecordId);
  }
  if (!input.forceNewEpoch && current.state === "candidate") {
    throw new Error("event_v3_candidate_requires_explicit_activation_or_epoch_replacement");
  }

  const now = input.now ?? (() => new Date());
  const createdAt = now().toISOString();
  const keys = loadOrCreateFingerprintKeyStoreV3(root, now);
  const harneryBuild = safeBuild(input.harneryBuild);
  const hostBuild = safeBuild(input.hostBuild);
  const producerId = BOOTSTRAP_CONTROL_PRODUCER_ID;
  const bootId =
    `boot_${createHash("sha256").update(`${root}\0${createdAt}`).digest("hex")}` as const;
  const buildId = liveEventV3BuildId(harneryBuild);
  const rootId = `root_${createHash("sha256").update(root).digest("hex")}` as const;
  const instanceId =
    `inst_init_${createHash("sha256").update(`${root}\0${createdAt}`).digest("hex")}` as const;
  const capabilityDigests = Object.values(ADAPTER_CAPABILITY_PROFILES_V3)
    .map((profile) => sha256V3(canonicalJsonV3(profile)))
    .sort();
  const candidate = buildCandidateGenesisManifestV3({
    profile: {
      initial_schema_digest: EVENT_V3_SCHEMA_DIGEST,
      contract_source_digest: EVENT_V3_SCHEMA_DIGEST,
      harnery_commit: harneryBuild,
      host_repository_commit: hostBuild,
      producer_build_ids: [buildId],
      adapter_capability_profile_digests: capabilityDigests,
      config_digest: input.configDigest,
      canonicalizer_version: "harnery-jcs-nfc-v1",
      fingerprint_version: "hmac-sha256-v1",
      privacy_key_epoch: keys.active_epoch_id,
      candidate_created_at: createdAt,
    },
    root_id: rootId,
    instance_id: instanceId,
    producer: {
      producer_id: producerId,
      boot_id: bootId,
      sequence: 1,
      build_id: buildId,
      platform: livePlatformV3(),
    },
  });
  // Mint the complete control pair before any epoch bytes move. Every approval
  // and validation failure then happens while the current epoch is still
  // whole, so it can never leave a replaced epoch stranded in candidate state.
  const activation = buildActivationManifestV3(activationInput(candidate, input.approvalRecordId));

  const archivedEpoch = archiveCurrentEpoch(root, createdAt);
  publishControlFile(join(root, EVENT_V3_GENESIS_MANIFEST), candidate);
  let activated: InitializeEventLedgerV3Result;
  try {
    activated = completeCandidateActivation(root, activation);
  } catch (error) {
    // The epoch has already been replaced, so returning here would strand it.
    // Retry the completion inside the same lease and, failing that, name the
    // stranded epoch rather than reporting a successful initialization.
    activated = completeStrandedCandidateActivation(root, activation, error);
  }
  return {
    ...activated,
    ...(archivedEpoch ? { archived_epoch: archivedEpoch } : {}),
  };
}

function completeStrandedCandidateActivation(
  root: string,
  activation: ActivationManifestV3,
  cause: unknown,
): InitializeEventLedgerV3Result {
  try {
    return completeCandidateActivation(root, activation);
  } catch {
    throw new Error(
      `event_v3_epoch_stranded_before_activation:${cause instanceof Error ? cause.message : String(cause)}`,
    );
  }
}

function runtimeCapabilityProfileCurrent(
  control: Extract<EventV3ControlState, { state: "candidate" | "active" }>,
): boolean {
  const expected = Object.values(ADAPTER_CAPABILITY_PROFILES_V3).map((profile) =>
    sha256V3(canonicalJsonV3(profile)),
  );
  const approved = control.genesis.profile.adapter_capability_profile_digests;
  const expectedDigests = new Set<string>(expected);
  return control.state === "candidate"
    ? approved.some((digest) => expectedDigests.has(digest))
    : expected.every((digest) => approved.includes(digest));
}

function withBootstrapLease<T>(coordRoot: string, operation: () => T): T {
  const root = resolve(coordRoot);
  const authority = createHash("sha256").update(root).digest("hex");
  let lease: ReturnType<typeof acquireNoClobberLease> | undefined;
  for (let attempt = 0; attempt < BOOTSTRAP_LEASE_RETRIES; attempt += 1) {
    try {
      lease = acquireNoClobberLease({
        path: join(root, ".harnery", "private", "event-v3-bootstrap-lease"),
        scope: "event-v3-bootstrap",
        authoritySha256: authority,
        staleAfterMs: BOOTSTRAP_LEASE_STALE_MS,
        validateStaleOwner: (owner) => owner.host === hostname() && !pidIsAlive(owner.pid),
      });
      break;
    } catch (error) {
      if (attempt === BOOTSTRAP_LEASE_RETRIES - 1) throw error;
      Atomics.wait(bootstrapSleepCell, 0, 0, BOOTSTRAP_LEASE_RETRY_MS);
    }
  }
  if (!lease) throw new Error("event_v3_bootstrap_lease_busy");
  try {
    return operation();
  } finally {
    lease.release();
  }
}

function pidIsAlive(pid: number): boolean {
  if (!Number.isSafeInteger(pid) || pid < 1) return false;
  try {
    process.kill(pid, 0);
    return true;
  } catch (error) {
    return (error as NodeJS.ErrnoException).code !== "ESRCH";
  }
}

function activateCandidateEpoch(
  root: string,
  candidate: CandidateGenesisManifestV3,
  approvalRecordId: string,
): InitializeEventLedgerV3Result {
  return completeCandidateActivation(
    root,
    buildActivationManifestV3(activationInput(candidate, approvalRecordId)),
  );
}

/** The activation packet one exact candidate deserves, derived and pure. */
function activationInput(candidate: CandidateGenesisManifestV3, approvalRecordId: string) {
  const producer = candidate.event.producer;
  return {
    candidate,
    approval_record_id: approvalRecordId,
    activation_approved_at: candidate.profile.candidate_created_at,
    producer: {
      producer_id: producer.producer_id as `prd_${string}`,
      boot_id: producer.boot_id as `boot_${string}`,
      sequence: producer.sequence + 1,
      build_id: producer.build_id as `build_${string}`,
      platform: producer.platform,
      ...(producer.bridge ? { bridge: producer.bridge } : {}),
    },
  };
}

/**
 * Bring a published candidate to active: append the pre-minted genesis event
 * if a crash left it missing, publish the activation manifest, then append its
 * pre-minted event. Every step is idempotent, so the caller may retry it, and
 * anything short of `active` throws with the state that blocked it.
 */
function completeCandidateActivation(
  root: string,
  activation: ActivationManifestV3,
): InitializeEventLedgerV3Result {
  const candidateState = repairEventV3ControlPair(root);
  if (candidateState.state !== "candidate") {
    throw new Error(`event_v3_candidate_initialization_failed:${candidateState.state}`);
  }
  publishControlFile(join(root, EVENT_V3_ACTIVATION_MANIFEST), activation);
  const active = repairEventV3ControlPair(root);
  if (active.state !== "active") {
    throw new Error(`event_v3_activation_failed:${active.state}`);
  }
  return { state: "active", initialized: true, control: active };
}

function archiveCurrentEpoch(root: string, createdAt: string): string | undefined {
  const current = join(root, ".harnery", "ledgers", "v3");
  if (!existsSync(current)) return undefined;
  const archives = join(root, ".harnery", "ledgers", "v3-archives");
  mkdirSync(archives, { recursive: true, mode: 0o700 });
  const stamp = createdAt.replace(/[^0-9]/g, "");
  let target = join(archives, `epoch-${stamp}`);
  let suffix = 0;
  while (existsSync(target)) target = join(archives, `epoch-${stamp}-${++suffix}`);
  renameSync(current, target);
  fsyncParentDirectory(target);
  return target;
}

function publishControlFile(path: string, value: unknown): void {
  mkdirSync(dirname(path), { recursive: true, mode: 0o700 });
  const temp = `${path}.tmp.${process.pid}`;
  // A prior failed publish by this same pid would otherwise make every retry
  // fail on the exclusive create, which is exactly how one transient error
  // turns into a permanently stranded epoch.
  if (existsSync(temp)) unlinkSync(temp);
  const fd = openSync(temp, "wx", 0o600);
  try {
    writeFileSync(fd, `${canonicalJsonV3(value)}\n`, "utf8");
    fsyncSync(fd);
  } finally {
    closeSync(fd);
  }
  chmodSync(temp, 0o600);
  renameSync(temp, path);
  fsyncParentDirectory(path);
}

function safeBuild(value: string): string {
  const normalized = value.normalize("NFC").trim();
  if (/^[a-zA-Z0-9._-]{1,120}$/.test(normalized)) return normalized;
  return createHash("sha256")
    .update(normalized || "unknown")
    .digest("hex");
}

function rootOfHarnery(): string {
  return resolve(dirname(fileURLToPath(import.meta.url)), "../../../..");
}

function repositoryBuild(root: string): string {
  const result = spawnSync("git", ["-C", root, "rev-parse", "HEAD"], {
    encoding: "utf8",
    timeout: 5000,
  });
  const commit = result.status === 0 ? result.stdout.trim() : "";
  if (/^[0-9a-f]{40,64}$/.test(commit)) return commit;
  return createHash("sha256").update(resolve(root)).digest("hex");
}