import { mkdir, readdir } from "node:fs/promises";
import { isAbsolute, join, normalize } from "node:path";

import {
  readJson,
  writeAtomicJson,
  type AtomicFaultInjector,
} from "./atomic-json.js";
import { ReceiverLeaseManager } from "./receiver-lease.js";

const RECORD_KEYS = [
  "schema",
  "id",
  "attempt",
  "phase",
  "binding",
  "leaseGeneration",
  "targetSessionId",
  "delivery",
  "claimedAt",
  "receiptWarning",
  "updatedAt",
] as const;
const BINDING_KEYS = ["rootId", "handle"] as const;

export type BridgePhase = "pending" | "injected" | "dlq";
export type BridgeDelivery = "steer" | "followUp" | "nextTurn" | "fresh-session";

export interface BridgeBinding {
  readonly rootId: string;
  readonly handle: string;
}

export interface BridgeJournalRecordV1 {
  readonly schema: 1;
  readonly id: string;
  readonly attempt: number;
  readonly phase: BridgePhase;
  readonly binding: BridgeBinding;
  readonly leaseGeneration: string;
  readonly targetSessionId: string | null;
  readonly delivery: BridgeDelivery | null;
  readonly claimedAt: string | null;
  readonly receiptWarning: string | null;
  readonly updatedAt: string;
}

export interface DiskProjectionEvidence {
  readonly messageId: string;
  readonly sessionId: string;
  readonly diskVisible: boolean;
}

export interface BridgeRetryAuditEvidence {
  readonly auditId: string;
  readonly originalMessageId: string;
  readonly retryCount: number;
  readonly retryState: "delivered";
  readonly visibleMessageId: string;
}

export interface BridgeAttemptHistoryV1 {
  readonly schema: "pi-amq-attempt-history/v1";
  readonly record: BridgeJournalRecordV1;
  readonly retryAudit: BridgeRetryAuditEvidence;
}

export interface BridgeJournalOptions {
  readonly namespaceDirectory: string;
  readonly leaseManager: ReceiverLeaseManager;
  readonly now?: () => Date;
  readonly atomicFault?: AtomicFaultInjector;
}

export interface CreatePendingInput {
  readonly id: string;
  readonly attempt: number;
  readonly binding: BridgeBinding;
  readonly leaseGeneration: string;
}

export type BridgeRecoveryPlan =
  | { readonly kind: "exact-read"; readonly emitDrainedReceipt: boolean }
  | { readonly kind: "mark-injected" }
  | { readonly kind: "mark-dlq" }
  | { readonly kind: "terminal-dlq" }
  | { readonly kind: "already-injected" }
  | { readonly kind: "needs-attention"; readonly reason: "message-not-found" }
  | { readonly kind: "diagnostic"; readonly reason: "injected-projection-missing" };

export interface BridgeRecoveryFacts {
  readonly amqLocation: "new" | "cur" | "missing" | "dlq";
  readonly projection?: DiskProjectionEvidence;
}

/** Generation-fenced durable intent across the AMQ-to-pi delivery boundary. */
export class BridgeJournal {
  readonly #namespaceDirectory: string;
  readonly #leaseManager: ReceiverLeaseManager;
  readonly #now: () => Date;
  readonly #atomicFault: AtomicFaultInjector | undefined;

  constructor(options: BridgeJournalOptions) {
    if (!isAbsolute(options.namespaceDirectory)) {
      throw new TypeError("Bridge journal namespace must be absolute");
    }
    this.#namespaceDirectory = normalize(options.namespaceDirectory);
    this.#leaseManager = options.leaseManager;
    this.#now = options.now ?? (() => new Date());
    this.#atomicFault = options.atomicFault;
  }

  async createPending(input: CreatePendingInput): Promise<BridgeJournalRecordV1> {
    validateMessageId(input.id);
    requirePositiveInteger(input.attempt, "bridge attempt");
    const binding = parseBinding(input.binding);
    requireNonEmpty(input.leaseGeneration, "lease generation");
    return this.#leaseManager.withCurrentLease(input.leaseGeneration, async () => {
      await this.#initialize();
      const record: BridgeJournalRecordV1 = {
        schema: 1,
        id: input.id,
        attempt: input.attempt,
        phase: "pending",
        binding,
        leaseGeneration: input.leaseGeneration,
        targetSessionId: null,
        delivery: null,
        claimedAt: null,
        receiptWarning: null,
        updatedAt: this.#now().toISOString(),
      };
      await writeAtomicJson(this.#path(input.id), record, {
        createOnly: true,
        ...(this.#atomicFault ? { fault: this.#atomicFault } : {}),
      });
      return record;
    });
  }

  async read(id: string): Promise<BridgeJournalRecordV1 | undefined> {
    validateMessageId(id);
    const raw = await readJson(this.#path(id));
    return raw === undefined ? undefined : parseBridgeJournalRecordV1(raw);
  }

  async list(): Promise<readonly BridgeJournalRecordV1[]> {
    let entries: string[];
    try {
      entries = await readdir(join(this.#namespaceDirectory, "journal"));
    } catch (error) {
      if (isNodeError(error) && error.code === "ENOENT") return [];
      throw error;
    }
    const records: BridgeJournalRecordV1[] = [];
    for (const entry of entries.sort()) {
      if (!entry.endsWith(".json") || entry.startsWith(".")) continue;
      const raw = await readJson(join(this.#namespaceDirectory, "journal", entry));
      if (raw === undefined) continue;
      records.push(parseBridgeJournalRecordV1(raw));
    }
    return records.sort((left, right) => left.id.localeCompare(right.id));
  }

  async readHistory(id: string, attempt: number): Promise<BridgeAttemptHistoryV1 | undefined> {
    validateMessageId(id);
    requirePositiveInteger(attempt, "bridge history attempt");
    const raw = await readJson(this.#historyPath(id, attempt));
    return raw === undefined ? undefined : parseBridgeAttemptHistoryV1(raw);
  }

  async adoptPending(
    id: string,
    generation: string,
    binding: BridgeBinding,
  ): Promise<BridgeJournalRecordV1> {
    validateMessageId(id);
    requireNonEmpty(generation, "lease generation");
    const expectedBinding = parseBinding(binding);
    return this.#leaseManager.withCurrentLease(generation, async () => {
      const record = await this.read(id);
      if (!record) throw new Error(`Bridge journal record ${id} does not exist`);
      if (record.phase !== "pending") {
        throw new Error(`Only a pending bridge journal record can be adopted, got ${record.phase}`);
      }
      if (record.binding.rootId !== expectedBinding.rootId
        || record.binding.handle !== expectedBinding.handle) {
        throw new Error("Pending bridge journal binding does not match the current receiver");
      }
      if (record.leaseGeneration === generation) return record;
      const adopted = parseBridgeJournalRecordV1({
        ...record,
        leaseGeneration: generation,
        updatedAt: this.#now().toISOString(),
      });
      await writeAtomicJson(this.#path(id), adopted, {
        ...(this.#atomicFault ? { fault: this.#atomicFault } : {}),
      });
      return adopted;
    });
  }

  async recordClaim(
    id: string,
    generation: string,
    claim: { readonly claimedAt: string; readonly receiptWarning?: string | null },
  ): Promise<BridgeJournalRecordV1> {
    const claimedAt = requireIso(claim.claimedAt, "claimedAt");
    const receiptWarning = claim.receiptWarning === undefined
      ? undefined
      : optionalString(claim.receiptWarning, "receiptWarning");
    return this.#mutatePending(id, generation, (record) => ({
      ...record,
      claimedAt,
      ...(receiptWarning === undefined ? {} : { receiptWarning }),
      updatedAt: this.#now().toISOString(),
    }));
  }

  async setTarget(
    id: string,
    generation: string,
    target: { readonly targetSessionId: string; readonly delivery: BridgeDelivery },
  ): Promise<BridgeJournalRecordV1> {
    const targetSessionId = requireNonEmpty(target.targetSessionId, "target session ID");
    const delivery = parseDelivery(target.delivery);
    return this.#mutatePending(id, generation, (record) => ({
      ...record,
      targetSessionId,
      delivery,
      updatedAt: this.#now().toISOString(),
    }));
  }

  async markInjected(
    id: string,
    generation: string,
    evidence: DiskProjectionEvidence,
  ): Promise<BridgeJournalRecordV1> {
    return this.#mutatePending(id, generation, (record) => {
      if (!matchingDiskProjection(record, evidence)) {
        throw new Error("Injected transition requires matching disk-visible pi projection evidence");
      }
      return {
        ...record,
        phase: "injected",
        updatedAt: this.#now().toISOString(),
      };
    });
  }

  async markDlq(
    id: string,
    generation: string,
    evidence: { readonly messageId: string; readonly diagnostic: string },
  ): Promise<BridgeJournalRecordV1> {
    requireNonEmpty(evidence.diagnostic, "DLQ diagnostic");
    return this.#mutatePending(id, generation, (record) => {
      if (evidence.messageId !== record.id) {
        throw new Error("DLQ evidence message ID does not match the journal record");
      }
      return {
        ...record,
        phase: "dlq",
        updatedAt: this.#now().toISOString(),
      };
    });
  }

  async promoteVerifiedRetry(
    id: string,
    generation: string,
    evidence: BridgeRetryAuditEvidence,
  ): Promise<BridgeJournalRecordV1> {
    validateMessageId(id);
    requireNonEmpty(generation, "lease generation");
    const audit = parseBridgeRetryAuditEvidence(evidence);
    if (audit.originalMessageId !== id || audit.visibleMessageId !== id) {
      throw new Error("DLQ retry requires matching original and visibly redelivered message IDs");
    }
    return this.#leaseManager.withCurrentLease(generation, async () => {
      const record = await this.read(id);
      if (!record) throw new Error(`Bridge journal record ${id} does not exist`);
      if (record.leaseGeneration !== generation) {
        throw new Error("Bridge journal lease generation does not match the current retry");
      }

      if (record.phase === "pending") {
        const history = record.attempt > 1
          ? await this.readHistory(id, record.attempt - 1)
          : undefined;
        if (history && sameRetryAudit(history.retryAudit, audit)) return record;
        throw new Error(`Bridge journal pending attempt ${record.attempt} cannot be promoted again`);
      }
      if (record.phase !== "dlq") {
        throw new Error(`Bridge journal ${record.phase} phase cannot accept a DLQ retry`);
      }
      if (audit.retryCount !== record.attempt) {
        if (await this.#retryAuditWasConsumed(id, audit.auditId)) {
          throw new Error(`AMQ retry audit ${audit.auditId} was already consumed by an earlier bridge attempt`);
        }
        throw new Error(
          `AMQ retry count ${audit.retryCount} does not match terminal bridge attempt ${record.attempt}`,
        );
      }

      await this.#initialize();
      await mkdir(join(this.#namespaceDirectory, "history"), { recursive: true, mode: 0o700 });
      const history: BridgeAttemptHistoryV1 = {
        schema: "pi-amq-attempt-history/v1",
        record,
        retryAudit: audit,
      };
      const historyPath = this.#historyPath(id, record.attempt);
      const existing = await readJson(historyPath);
      if (existing === undefined) {
        if (await this.#retryAuditWasConsumed(id, audit.auditId)) {
          throw new Error(`AMQ retry audit ${audit.auditId} was already consumed by an earlier bridge attempt`);
        }
        await writeAtomicJson(historyPath, history, {
          createOnly: true,
          ...(this.#atomicFault ? { fault: this.#atomicFault } : {}),
        });
      } else {
        const parsed = parseBridgeAttemptHistoryV1(existing);
        if (!sameArchivedAttempt(parsed.record, record)
          || !sameRetryAudit(parsed.retryAudit, audit)) {
          throw new Error(`Bridge attempt history ${record.attempt} conflicts with the verified retry audit`);
        }
      }

      const pending: BridgeJournalRecordV1 = {
        schema: 1,
        id,
        attempt: record.attempt + 1,
        phase: "pending",
        binding: record.binding,
        leaseGeneration: generation,
        targetSessionId: null,
        delivery: null,
        claimedAt: null,
        receiptWarning: null,
        updatedAt: this.#now().toISOString(),
      };
      await writeAtomicJson(this.#path(id), pending, {
        ...(this.#atomicFault ? { fault: this.#atomicFault } : {}),
      });
      return pending;
    });
  }

  async #mutatePending(
    id: string,
    generation: string,
    mutation: (record: BridgeJournalRecordV1) => BridgeJournalRecordV1,
  ): Promise<BridgeJournalRecordV1> {
    validateMessageId(id);
    requireNonEmpty(generation, "lease generation");
    return this.#leaseManager.withCurrentLease(generation, async () => {
      const record = await this.read(id);
      if (!record) throw new Error(`Bridge journal record ${id} does not exist`);
      if (record.leaseGeneration !== generation) {
        throw new Error("Bridge journal lease generation does not match the current mutation");
      }
      if (record.phase !== "pending") {
        throw new Error(`Bridge journal ${record.phase} phase is terminal for attempt ${record.attempt}`);
      }
      const updated = parseBridgeJournalRecordV1(mutation(record));
      await writeAtomicJson(this.#path(id), updated, {
        ...(this.#atomicFault ? { fault: this.#atomicFault } : {}),
      });
      return updated;
    });
  }

  async #initialize(): Promise<void> {
    await mkdir(join(this.#namespaceDirectory, "journal"), { recursive: true, mode: 0o700 });
  }

  #path(id: string): string {
    return join(this.#namespaceDirectory, "journal", `${id}.json`);
  }

  #historyPath(id: string, attempt: number): string {
    return join(this.#namespaceDirectory, "history", `${id}.attempt-${attempt}.json`);
  }

  async #retryAuditWasConsumed(id: string, auditId: string): Promise<boolean> {
    let entries: string[];
    try {
      entries = await readdir(join(this.#namespaceDirectory, "history"));
    } catch (error) {
      if (isNodeError(error) && error.code === "ENOENT") return false;
      throw error;
    }
    for (const entry of entries) {
      if (!entry.startsWith(`${id}.attempt-`) || !entry.endsWith(".json")) continue;
      const raw = await readJson(join(this.#namespaceDirectory, "history", entry));
      if (raw && parseBridgeAttemptHistoryV1(raw).retryAudit.auditId === auditId) return true;
    }
    return false;
  }
}

export function planBridgeRecovery(
  record: BridgeJournalRecordV1,
  facts: BridgeRecoveryFacts,
): BridgeRecoveryPlan {
  const parsed = parseBridgeJournalRecordV1(record);
  if (parsed.phase === "dlq") return { kind: "terminal-dlq" };
  if (parsed.phase === "injected") {
    return facts.projection && matchingDiskProjection(parsed, facts.projection)
      ? { kind: "already-injected" }
      : { kind: "diagnostic", reason: "injected-projection-missing" };
  }
  if (facts.amqLocation === "dlq") return { kind: "mark-dlq" };
  if (facts.projection && matchingDiskProjection(parsed, facts.projection)) {
    return { kind: "mark-injected" };
  }
  if (facts.amqLocation === "new") return { kind: "exact-read", emitDrainedReceipt: true };
  if (facts.amqLocation === "cur") return { kind: "exact-read", emitDrainedReceipt: false };
  return { kind: "needs-attention", reason: "message-not-found" };
}

export function parseBridgeJournalRecordV1(value: unknown): BridgeJournalRecordV1 {
  const object = strictRecord(value, RECORD_KEYS, "bridge journal record");
  if (numeric(object, "schema") !== 1) throw new Error("bridge journal schema must be 1");
  const phase = object.phase;
  if (phase !== "pending" && phase !== "injected" && phase !== "dlq") {
    throw new Error("bridge journal phase is invalid");
  }
  const targetSessionId = optionalString(object.targetSessionId, "targetSessionId");
  const delivery = object.delivery === null ? null : parseDelivery(object.delivery);
  if ((targetSessionId === null) !== (delivery === null)) {
    throw new Error("targetSessionId and delivery must be present together");
  }
  if (phase === "injected" && targetSessionId === null) {
    throw new Error("injected bridge journal record requires a target session");
  }
  return {
    schema: 1,
    id: parseMessageId(object.id),
    attempt: requirePositiveInteger(numeric(object, "attempt"), "bridge attempt"),
    phase,
    binding: parseBinding(object.binding),
    leaseGeneration: string(object, "leaseGeneration"),
    targetSessionId,
    delivery,
    claimedAt: optionalIso(object.claimedAt, "claimedAt"),
    receiptWarning: optionalString(object.receiptWarning, "receiptWarning"),
    updatedAt: requireIso(string(object, "updatedAt"), "updatedAt"),
  };
}

export function parseBridgeAttemptHistoryV1(value: unknown): BridgeAttemptHistoryV1 {
  const object = strictRecord(value, ["schema", "record", "retryAudit"], "bridge attempt history");
  if (object.schema !== "pi-amq-attempt-history/v1") {
    throw new Error("bridge attempt history schema is invalid");
  }
  const record = parseBridgeJournalRecordV1(object.record);
  if (record.phase !== "dlq") throw new Error("bridge attempt history must archive a terminal DLQ record");
  return {
    schema: "pi-amq-attempt-history/v1",
    record,
    retryAudit: parseBridgeRetryAuditEvidence(object.retryAudit),
  };
}

function parseBridgeRetryAuditEvidence(value: unknown): BridgeRetryAuditEvidence {
  const object = strictRecord(
    value,
    ["auditId", "originalMessageId", "retryCount", "retryState", "visibleMessageId"],
    "bridge retry audit",
  );
  if (object.retryState !== "delivered") throw new Error("AMQ retry audit must be delivered");
  return {
    auditId: string(object, "auditId"),
    originalMessageId: parseMessageId(object.originalMessageId),
    retryCount: requirePositiveInteger(numeric(object, "retryCount"), "AMQ retry count"),
    retryState: "delivered",
    visibleMessageId: parseMessageId(object.visibleMessageId),
  };
}

function sameRetryAudit(left: BridgeRetryAuditEvidence, right: BridgeRetryAuditEvidence): boolean {
  return left.auditId === right.auditId
    && left.originalMessageId === right.originalMessageId
    && left.retryCount === right.retryCount
    && left.retryState === right.retryState
    && left.visibleMessageId === right.visibleMessageId;
}

function sameArchivedAttempt(left: BridgeJournalRecordV1, right: BridgeJournalRecordV1): boolean {
  return JSON.stringify(left) === JSON.stringify(right);
}

function matchingDiskProjection(
  record: BridgeJournalRecordV1,
  evidence: DiskProjectionEvidence,
): boolean {
  return evidence.diskVisible
    && evidence.messageId === record.id
    && record.targetSessionId !== null
    && evidence.sessionId === record.targetSessionId;
}

function parseBinding(value: unknown): BridgeBinding {
  const object = strictRecord(value, BINDING_KEYS, "bridge binding");
  return {
    rootId: string(object, "rootId"),
    handle: string(object, "handle"),
  };
}

function parseDelivery(value: unknown): BridgeDelivery {
  if (value === "steer" || value === "followUp" || value === "nextTurn" || value === "fresh-session") {
    return value;
  }
  throw new Error("bridge delivery is invalid");
}

function parseMessageId(value: unknown): string {
  if (typeof value !== "string") throw new Error("bridge message ID must be a string");
  validateMessageId(value);
  return value;
}

function validateMessageId(id: string): void {
  if (!/^[A-Za-z0-9][A-Za-z0-9._-]{0,254}$/.test(id)) {
    throw new TypeError("Bridge message ID is not safe for the extension namespace");
  }
}

function strictRecord(
  value: unknown,
  allowed: readonly string[],
  name: string,
): Record<string, unknown> {
  if (typeof value !== "object" || value === null || Array.isArray(value)) {
    throw new Error(`${name} must be an object`);
  }
  const object = value as Record<string, unknown>;
  const unknown = Object.keys(object).filter((key) => !allowed.includes(key));
  if (unknown.length > 0) throw new Error(`${name} contains unknown fields: ${unknown.join(", ")}`);
  return object;
}

function string(object: Record<string, unknown>, key: string): string {
  const value = object[key];
  if (typeof value !== "string" || value.length === 0) throw new Error(`${key} must be a non-empty string`);
  return value;
}

function numeric(object: Record<string, unknown>, key: string): number {
  const value = object[key];
  if (typeof value !== "number" || !Number.isFinite(value)) throw new Error(`${key} must be a number`);
  return value;
}

function optionalString(value: unknown, name: string): string | null {
  if (value === null) return null;
  if (typeof value !== "string" || value.length === 0) throw new Error(`${name} must be null or a non-empty string`);
  return value;
}

function optionalIso(value: unknown, name: string): string | null {
  const stringValue = optionalString(value, name);
  return stringValue === null ? null : requireIso(stringValue, name);
}

function requireIso(value: string, name: string): string {
  if (!Number.isFinite(Date.parse(value))) throw new Error(`${name} must be an ISO timestamp`);
  return value;
}

function requirePositiveInteger(value: number, name: string): number {
  if (!Number.isSafeInteger(value) || value <= 0) throw new TypeError(`${name} must be a positive safe integer`);
  return value;
}

function requireNonEmpty(value: string, name: string): string {
  if (value.length === 0) throw new TypeError(`${name} is required`);
  return value;
}

function isNodeError(error: unknown): error is NodeJS.ErrnoException {
  return error instanceof Error && "code" in error;
}