import { setTimeout as delay } from "node:timers/promises";

import type {
  AmqCliAdapter,
  AmqCommandResult,
} from "./amq-cli-adapter.js";
import {
  parseAmqReadOutcome,
  parseMessageList,
  parseWatchCohort,
  sortAndDeduplicateCandidates,
  type AmqMessage,
  type WatchCandidate,
} from "./amq-message.js";
import {
  BridgeJournal,
  type BridgeBinding,
  type BridgeDelivery,
  type DiskProjectionEvidence,
} from "./bridge-journal.js";
import {
  AmqDlqRetryAuditVerifier,
  type DlqRetryAuditVerifier,
} from "./dlq-retry-audit.js";
import {
  PriorityDeliveryScheduler,
  type DeliveryBatchOutcome,
} from "./priority-delivery-scheduler.js";
import { ReceiverLeaseManager } from "./receiver-lease.js";
import { InboxStatusStore } from "./runtime-status.js";

export interface InboxBinding {
  readonly root: string;
  readonly rootId: string;
  readonly handle: string;
}

export type InboxAuthorityResult =
  | { readonly kind: "valid" }
  | { readonly kind: "invalid"; readonly diagnostic: string };

export interface InboxAuthority {
  revalidate(signal?: AbortSignal): Promise<InboxAuthorityResult>;
}

export interface SessionProjectionLookup {
  find(
    messageId: string,
    targetSessionId: string,
    signal?: AbortSignal,
  ): Promise<DiskProjectionEvidence | undefined>;
}

export interface IncomingFreshSessionDispatcher {
  dispatch(
    message: AmqMessage,
    signal?: AbortSignal,
  ): Promise<
    | { readonly kind: "ordinary" }
    | { readonly kind: "refused"; readonly reason: string }
    | { readonly kind: "pending"; readonly reason: string; readonly action: "/amq ingest" }
    | { readonly kind: "dispatched"; readonly projection: DiskProjectionEvidence }
  >;
}

type JsonCommandRunner = Pick<AmqCliAdapter, "runJson">;

export interface InboxPumpOptions {
  readonly adapter: JsonCommandRunner;
  readonly binding: InboxBinding;
  readonly receiverGeneration: string;
  readonly targetSessionId: string;
  readonly authority: InboxAuthority;
  readonly journal: BridgeJournal;
  readonly scheduler: PriorityDeliveryScheduler;
  readonly status?: InboxStatusStore;
  readonly projectionLookup?: SessionProjectionLookup;
  readonly freshSessionDispatcher?: IncomingFreshSessionDispatcher;
  readonly retryAuditVerifier?: DlqRetryAuditVerifier;
  readonly now?: () => Date;
  readonly random?: () => number;
  readonly sleep?: (milliseconds: number, signal?: AbortSignal) => Promise<void>;
  readonly backoff?: {
    readonly initialMs?: number;
    readonly maximumMs?: number;
    readonly multiplier?: number;
    readonly jitterRatio?: number;
  };
}

class TerminalPumpError extends Error {
  constructor(message: string) {
    super(message);
    this.name = "TerminalPumpError";
  }
}

class RecoverablePumpError extends Error {
  readonly needsRecovery: boolean;

  constructor(message: string, options: { readonly needsRecovery: boolean; readonly cause?: unknown }) {
    super(message, options.cause === undefined ? undefined : { cause: options.cause });
    this.name = "RecoverablePumpError";
    this.needsRecovery = options.needsRecovery;
  }
}

/** Supervises one non-consuming watch child and the exact-ID claim pipeline. */
export class InboxPump {
  readonly #adapter: JsonCommandRunner;
  readonly #binding: InboxBinding;
  readonly #receiverGeneration: string;
  readonly #targetSessionId: string;
  readonly #authority: InboxAuthority;
  readonly #journal: BridgeJournal;
  readonly #scheduler: PriorityDeliveryScheduler;
  readonly #status: InboxStatusStore;
  readonly #projectionLookup: SessionProjectionLookup;
  readonly #freshSessionDispatcher: IncomingFreshSessionDispatcher | undefined;
  readonly #retryAuditVerifier: DlqRetryAuditVerifier;
  readonly #now: () => Date;
  readonly #random: () => number;
  readonly #sleep: (milliseconds: number, signal?: AbortSignal) => Promise<void>;
  readonly #initialBackoffMs: number;
  readonly #maximumBackoffMs: number;
  readonly #backoffMultiplier: number;
  readonly #jitterRatio: number;
  #serial: Promise<void> = Promise.resolve();

  constructor(options: InboxPumpOptions) {
    requireNonEmpty(options.binding.root, "binding root");
    requireNonEmpty(options.binding.rootId, "binding root ID");
    requireNonEmpty(options.binding.handle, "binding handle");
    requireNonEmpty(options.receiverGeneration, "receiver generation");
    requireNonEmpty(options.targetSessionId, "target session ID");
    this.#adapter = options.adapter;
    this.#binding = options.binding;
    this.#receiverGeneration = options.receiverGeneration;
    this.#targetSessionId = options.targetSessionId;
    this.#authority = options.authority;
    this.#journal = options.journal;
    this.#scheduler = options.scheduler;
    this.#status = options.status ?? new InboxStatusStore();
    this.#projectionLookup = options.projectionLookup ?? NO_PROJECTIONS;
    this.#freshSessionDispatcher = options.freshSessionDispatcher;
    this.#retryAuditVerifier = options.retryAuditVerifier ?? new AmqDlqRetryAuditVerifier({
      runner: options.adapter,
      root: options.binding.root,
      handle: options.binding.handle,
    });
    this.#now = options.now ?? (() => new Date());
    this.#random = options.random ?? Math.random;
    this.#sleep = options.sleep ?? defaultSleep;
    this.#initialBackoffMs = positiveNumber(options.backoff?.initialMs ?? 100, "initial backoff");
    this.#maximumBackoffMs = positiveNumber(options.backoff?.maximumMs ?? 5_000, "maximum backoff");
    this.#backoffMultiplier = positiveNumber(options.backoff?.multiplier ?? 2, "backoff multiplier");
    this.#jitterRatio = ratio(options.backoff?.jitterRatio ?? 0.2, "backoff jitter ratio");
    if (this.#maximumBackoffMs < this.#initialBackoffMs) {
      throw new TypeError("maximum backoff must be at least the initial backoff");
    }
  }

  async run(signal?: AbortSignal): Promise<void> {
    this.#status.update({
      receiver: "receiving",
      watch: "recovering",
      receiverGeneration: this.#receiverGeneration,
      lastError: null,
    });
    let recover = true;
    let backoffAttempt = 0;

    while (!signal?.aborted) {
      try {
        if (recover) {
          this.#status.update({ watch: "recovering", nextBackoffMs: null });
          await this.#reconcileStartup(signal);
          recover = false;
        }
        await this.#assertAuthority(signal);
        this.#status.update({
          watch: "watching",
          backoffAttempt: 0,
          nextBackoffMs: null,
        });
        const result = await this.#adapter.runJson(
          [
            "watch",
            "--root",
            this.#binding.root,
            "--me",
            this.#binding.handle,
            "--timeout",
            "0",
          ],
          { ...(signal ? { signal } : {}), parse: parseWatchCohort },
        );
        if (result.category === "cancelled") break;
        if (result.category === "context-mismatch") {
          throw new TerminalPumpError(literalFailure(result, "AMQ watch context mismatch"));
        }
        if (result.json?.event === "timeout" && result.category === "timeout") {
          backoffAttempt = 0;
          continue;
        }
        if (!result.ok || !result.json) {
          throw new RecoverablePumpError(literalFailure(result, "AMQ watch failed"), {
            needsRecovery: false,
          });
        }
        await this.#serialized(() => this.#processCandidates(result.json!.messages, signal));
        await this.#refreshStagedCounts();
        backoffAttempt = 0;
      } catch (error) {
        if (signal?.aborted) break;
        if (error instanceof TerminalPumpError) {
          this.#status.update({
            watch: "disabled",
            nextBackoffMs: null,
            lastError: error.message,
          });
          return;
        }
        const recoverable = error instanceof RecoverablePumpError
          ? error
          : new RecoverablePumpError(toError(error).message, { needsRecovery: true, cause: error });
        recover ||= recoverable.needsRecovery;
        backoffAttempt += 1;
        const waitMs = this.#backoffDelay(backoffAttempt);
        this.#status.update({
          watch: "backoff",
          backoffAttempt,
          nextBackoffMs: waitMs,
          lastError: recoverable.message,
        });
        try {
          await this.#sleep(waitMs, signal);
        } catch (sleepError) {
          if (!signal?.aborted) throw sleepError;
          break;
        }
      }
    }
    this.#status.update({ watch: "stopped", nextBackoffMs: null });
  }

  async claimExact(id: string, signal?: AbortSignal): Promise<AmqMessage> {
    let claimed: AmqMessage | undefined;
    await this.#serialized(async () => {
      const messages = await this.#processCandidates([{
        id,
        priority: "normal",
        created: this.#now().toISOString(),
      }], signal);
      claimed = messages.find((message) => message.header.id === id);
      if (!claimed) claimed = await this.#readCurrentMessage(id, signal);
    });
    if (!claimed) throw new Error(`AMQ message ${id} was not available after receiver claim`);
    return claimed;
  }

  async ingestStaged(signal?: AbortSignal): Promise<number> {
    let count = 0;
    await this.#serialized(async () => {
      const staged = (await this.#journal.list()).filter(
        (record) => record.phase === "pending"
          && (record.delivery === "nextTurn" || record.delivery === "fresh-session"),
      );
      const messages: AmqMessage[] = [];
      for (const record of staged) {
        const message = await this.#readCurrentMessage(record.id, signal);
        if (!message) continue;
        await this.#journal.setTarget(record.id, this.#receiverGeneration, {
          targetSessionId: this.#targetSessionId,
          delivery: "followUp",
        });
        messages.push(record.delivery === "fresh-session"
          ? withFreshSessionOutcome(
              message,
              "pending-capability",
              "Fresh-session automation was unavailable; delivered by explicit /amq ingest",
            )
          : message);
      }
      if (messages.length === 0) return;
      const outcomes = await this.#scheduler.scheduleAs(
        messages,
        this.#targetSessionId,
        { deliverAs: "followUp", triggerTurn: true },
        signal,
      );
      await this.#applyDeliveryOutcomes(outcomes);
      count = outcomes.flatMap((outcome) => outcome.request.message.details.messageIds).length;
      await this.#refreshStagedCounts();
    });
    return count;
  }

  async #reconcileStartup(signal?: AbortSignal): Promise<void> {
    await this.#assertAuthority(signal);
    const binding: BridgeBinding = {
      rootId: this.#binding.rootId,
      handle: this.#binding.handle,
    };
    const records = await this.#journal.list();
    const unresolved: string[] = [];
    let diagnostic: string | null = null;

    for (const original of records) {
      throwIfAborted(signal);
      if (original.phase === "dlq") continue;
      if (original.phase === "injected") {
        if (original.targetSessionId) {
          const projection = await this.#projectionLookup.find(
            original.id,
            original.targetSessionId,
            signal,
          );
          if (!projection?.diskVisible) {
            diagnostic = `Injected AMQ projection is missing for ${original.id}; it was not resurrected`;
          }
        }
        continue;
      }

      const record = await this.#journal.adoptPending(
        original.id,
        this.#receiverGeneration,
        binding,
      );
      if (record.targetSessionId) {
        const projection = await this.#projectionLookup.find(
          record.id,
          record.targetSessionId,
          signal,
        );
        if (projection?.diskVisible) {
          await this.#journal.markInjected(record.id, this.#receiverGeneration, projection);
          continue;
        }
      }
      unresolved.push(record.id);
    }

    if (diagnostic) this.#status.update({ lastError: diagnostic });
    if (unresolved.length === 0) {
      await this.#refreshStagedCounts();
      return;
    }

    const listed = await this.#listRecoveryCandidates(signal);
    const byId = new Map(listed.map((candidate) => [candidate.id, candidate]));
    const candidates = unresolved.map((id): WatchCandidate => byId.get(id) ?? {
      id,
      priority: "normal",
      created: "",
      parseError: "pending journal ID is not listed in AMQ new/cur",
    });
    await this.#serialized(() => this.#processCandidates(candidates, signal));
    await this.#refreshStagedCounts();
  }

  async #listRecoveryCandidates(signal?: AbortSignal): Promise<readonly WatchCandidate[]> {
    const candidates: WatchCandidate[] = [];
    for (const box of ["--new", "--cur"] as const) {
      await this.#assertAuthority(signal);
      const result = await this.#adapter.runJson(
        [
          "list",
          "--root",
          this.#binding.root,
          "--me",
          this.#binding.handle,
          box,
        ],
        { ...(signal ? { signal } : {}), parse: parseMessageList },
      );
      if (result.category === "context-mismatch") {
        throw new TerminalPumpError(literalFailure(result, "AMQ recovery list context mismatch"));
      }
      if (!result.ok || !result.json) {
        throw new RecoverablePumpError(literalFailure(result, "AMQ recovery list failed"), {
          needsRecovery: true,
        });
      }
      candidates.push(...result.json);
    }
    return sortAndDeduplicateCandidates(candidates);
  }

  async #processCandidates(
    candidates: readonly WatchCandidate[],
    signal?: AbortSignal,
  ): Promise<readonly AmqMessage[]> {
    const messages: AmqMessage[] = [];
    const missingIds: string[] = [];
    for (const candidate of sortAndDeduplicateCandidates(candidates)) {
      throwIfAborted(signal);
      let record = await this.#journal.read(candidate.id);
      if (record?.phase === "injected") continue;
      if (record?.phase === "dlq") {
        const audits = await this.#retryAuditVerifier.verify(candidate.id, signal);
        let promoted = false;
        for (const audit of audits) {
          try {
            record = await this.#journal.promoteVerifiedRetry(
              candidate.id,
              this.#receiverGeneration,
              audit,
            );
            promoted = true;
            break;
          } catch (error) {
            this.#status.update({
              lastError: `DLQ retry audit ${audit.auditId} refused for ${candidate.id}: ${toError(error).message}`,
            });
          }
        }
        if (!promoted) {
          this.#status.update({
            lastError: `DLQ tombstone for ${candidate.id} remains terminal: visible same-ID mail has no unused delivered AMQ retry audit`,
          });
          continue;
        }
      }
      if (!record) {
        await this.#assertAuthority(signal);
        record = await this.#journal.createPending({
          id: candidate.id,
          attempt: 1,
          binding: { rootId: this.#binding.rootId, handle: this.#binding.handle },
          leaseGeneration: this.#receiverGeneration,
        });
      } else if (record.leaseGeneration !== this.#receiverGeneration) {
        record = await this.#journal.adoptPending(
          record.id,
          this.#receiverGeneration,
          { rootId: this.#binding.rootId, handle: this.#binding.handle },
        );
      }

      await this.#assertAuthority(signal);
      const result = await this.#adapter.runJson(
        [
          "read",
          "--root",
          this.#binding.root,
          "--me",
          this.#binding.handle,
          "--id",
          candidate.id,
        ],
        { ...(signal ? { signal } : {}), parse: parseAmqReadOutcome },
      );
      if (result.category === "cancelled") throwIfAborted(signal);
      if (result.category === "context-mismatch") {
        throw new TerminalPumpError(literalFailure(result, "AMQ read context mismatch"));
      }
      if (result.json?.kind === "dlq") {
        await this.#journal.markDlq(candidate.id, this.#receiverGeneration, {
          messageId: result.json.id,
          diagnostic: result.json.diagnostic,
        });
        continue;
      }
      if (result.json?.kind === "message"
        && (result.category === "success" || result.category === "error")) {
        if (result.json.message.header.id !== candidate.id) {
          throw new RecoverablePumpError(
            `AMQ read returned ${result.json.message.header.id} for requested ${candidate.id}`,
            { needsRecovery: true },
          );
        }
        await this.#journal.recordClaim(candidate.id, this.#receiverGeneration, {
          claimedAt: this.#now().toISOString(),
          receiptWarning: result.ok ? null : literalFailure(result, "AMQ claim committed with warning"),
        });
        if (!result.ok) {
          this.#status.update({
            lastError: `Claim committed; drained receipt audit failed for ${candidate.id}: ${literalFailure(result, "AMQ receipt audit failed")}`,
          });
        }
        const fresh = await this.#freshSessionDispatcher?.dispatch(result.json.message, signal);
        if (fresh?.kind === "pending") {
          await this.#journal.setTarget(candidate.id, this.#receiverGeneration, {
            targetSessionId: this.#targetSessionId,
            delivery: "fresh-session",
          });
          continue;
        }
        if (fresh?.kind === "dispatched") {
          await this.#journal.setTarget(candidate.id, this.#receiverGeneration, {
            targetSessionId: fresh.projection.sessionId,
            delivery: "fresh-session",
          });
          if (fresh.projection.diskVisible) {
            await this.#journal.markInjected(
              candidate.id,
              this.#receiverGeneration,
              fresh.projection,
            );
          }
          continue;
        }
        messages.push(fresh?.kind === "refused"
          ? withFreshSessionOutcome(result.json.message, "refused", fresh.reason)
          : result.json.message);
        continue;
      }
      if (result.category === "not-found") {
        missingIds.push(candidate.id);
        continue;
      }
      throw new RecoverablePumpError(literalFailure(result, `AMQ read failed for ${candidate.id}`), {
        needsRecovery: true,
      });
    }

    if (messages.length > 0) {
      for (const message of messages) {
        await this.#journal.setTarget(message.header.id, this.#receiverGeneration, {
          targetSessionId: this.#targetSessionId,
          delivery: bridgeDelivery(message.header.priority),
        });
      }
      const outcomes = await this.#scheduler.schedule(messages, this.#targetSessionId, signal);
      await this.#applyDeliveryOutcomes(outcomes);
    }
    if (missingIds.length > 0) {
      throw new RecoverablePumpError(
        `Pending AMQ messages were not found by exact ID: ${missingIds.join(", ")}`,
        { needsRecovery: true },
      );
    }
    return messages;
  }

  async #readCurrentMessage(id: string, signal?: AbortSignal): Promise<AmqMessage | undefined> {
    await this.#assertAuthority(signal);
    const result = await this.#adapter.runJson(
      ["read", "--root", this.#binding.root, "--me", this.#binding.handle, "--id", id],
      { ...(signal ? { signal } : {}), parse: parseAmqReadOutcome },
    );
    if (result.json?.kind === "message"
      && (result.category === "success" || result.category === "error")) {
      return result.json.message;
    }
    if (result.category === "not-found") return undefined;
    throw new RecoverablePumpError(literalFailure(result, `AMQ read failed for ${id}`), {
      needsRecovery: true,
    });
  }

  async #serialized<T>(operation: () => Promise<T>): Promise<T> {
    const task = this.#serial.then(operation);
    this.#serial = task.then(() => undefined, () => undefined);
    return task;
  }

  async #applyDeliveryOutcomes(outcomes: readonly DeliveryBatchOutcome[]): Promise<void> {
    for (const outcome of outcomes) {
      if (outcome.result.kind !== "accepted") {
        throw new RecoverablePumpError(
          `Pi delivery ${outcome.result.kind}: ${outcome.result.error.message}`,
          { needsRecovery: true, cause: outcome.result.error },
        );
      }
      const ids = new Set(outcome.request.message.details.messageIds);
      for (const projection of outcome.result.projections) {
        if (!ids.has(projection.messageId)) {
          throw new RecoverablePumpError(
            `Pi delivery returned projection for unrelated AMQ ID ${projection.messageId}`,
            { needsRecovery: true },
          );
        }
        if (!projection.diskVisible) continue;
        await this.#journal.markInjected(
          projection.messageId,
          this.#receiverGeneration,
          projection,
        );
      }
    }
  }

  async #refreshStagedCounts(): Promise<void> {
    const counts = { urgent: 0, normal: 0, low: 0 };
    for (const record of await this.#journal.list()) {
      if (record.phase !== "pending" || record.delivery === null) continue;
      switch (record.delivery) {
        case "steer":
          counts.urgent += 1;
          break;
        case "followUp":
        case "fresh-session":
          counts.normal += 1;
          break;
        case "nextTurn":
          counts.low += 1;
          break;
      }
    }
    this.#status.update({ staged: counts });
  }

  async #assertAuthority(signal?: AbortSignal): Promise<void> {
    throwIfAborted(signal);
    const authority = await this.#authority.revalidate(signal);
    if (authority.kind === "invalid") throw new TerminalPumpError(authority.diagnostic);
  }

  #backoffDelay(attempt: number): number {
    const base = Math.min(
      this.#maximumBackoffMs,
      this.#initialBackoffMs * this.#backoffMultiplier ** (attempt - 1),
    );
    const jitter = 1 + ((this.#random() * 2) - 1) * this.#jitterRatio;
    return Math.max(1, Math.min(this.#maximumBackoffMs, Math.round(base * jitter)));
  }
}

export class LeaseBackedInboxAuthority implements InboxAuthority {
  readonly #lease: ReceiverLeaseManager;
  readonly #generation: string;
  readonly #verifyBinding: (signal?: AbortSignal) => Promise<InboxAuthorityResult>;

  constructor(options: {
    readonly lease: ReceiverLeaseManager;
    readonly generation: string;
    readonly verifyBinding: (signal?: AbortSignal) => Promise<InboxAuthorityResult>;
  }) {
    this.#lease = options.lease;
    this.#generation = requireNonEmpty(options.generation, "receiver generation");
    this.#verifyBinding = options.verifyBinding;
  }

  async revalidate(signal?: AbortSignal): Promise<InboxAuthorityResult> {
    throwIfAborted(signal);
    const binding = await this.#verifyBinding(signal);
    if (binding.kind === "invalid") return binding;
    try {
      await this.#lease.withCurrentLease(this.#generation, async () => undefined);
      return { kind: "valid" };
    } catch (error) {
      return { kind: "invalid", diagnostic: `Receiver lease revalidation failed: ${toError(error).message}` };
    }
  }
}

const NO_PROJECTIONS: SessionProjectionLookup = {
  async find(): Promise<undefined> {
    return undefined;
  },
};

function bridgeDelivery(priority: "urgent" | "normal" | "low"): BridgeDelivery {
  switch (priority) {
    case "urgent":
      return "steer";
    case "normal":
      return "followUp";
    case "low":
      return "nextTurn";
  }
}

function withFreshSessionOutcome(
  message: AmqMessage,
  outcome: "refused" | "pending-capability",
  reason: string,
): AmqMessage {
  const previous = message.header.context.pi_amq;
  const piAmq = previous && typeof previous === "object" && !Array.isArray(previous)
    ? previous as Readonly<Record<string, unknown>>
    : {};
  return {
    ...message,
    header: {
      ...message.header,
      context: {
        ...message.header.context,
        pi_amq: {
          ...piAmq,
          fresh_session_outcome: outcome,
          fresh_session_reason: reason,
        },
      },
    },
  };
}

function literalFailure<T>(result: AmqCommandResult<T>, fallback: string): string {
  const literal = result.stderr.trim() || result.jsonError?.message || result.processError?.message;
  return literal ? `${fallback}: ${literal}` : `${fallback} (${result.category})`;
}

async function defaultSleep(milliseconds: number, signal?: AbortSignal): Promise<void> {
  if (signal) {
    await delay(milliseconds, undefined, { signal });
    return;
  }
  await delay(milliseconds);
}

function throwIfAborted(signal: AbortSignal | undefined): void {
  if (!signal?.aborted) return;
  throw signal.reason instanceof Error ? signal.reason : new Error("Inbox pump was cancelled");
}

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

function ratio(value: number, name: string): number {
  if (!Number.isFinite(value) || value < 0 || value > 1) {
    throw new TypeError(`${name} must be between 0 and 1`);
  }
  return value;
}

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

function toError(error: unknown): Error {
  return error instanceof Error ? error : new Error(String(error));
}