import { existsSync, mkdirSync, readFileSync, writeFileSync } from "node:fs";
import { fireAndForget } from "./async-guard";
import { homedir } from "node:os";
import { dirname, join } from "node:path";
import { errMessage, isRecord, SHARED_CALLMUX_TOOL_CALL_TIMEOUT_MS } from "agent-relay-sdk";
import { createListener, type CallmuxConfig, type CreateListenerOptions, type ListenerHealthSnapshot, type ProgrammaticListener } from "callmux";
import type { OrchestratorConfig } from "./config";
import { agentRelayHome } from "./config";
import { PORTABLE_SHARED_MCP_SERVER_NAMES, provisionPortableSharedMcpServers, type PortableMcpProvisioningResult } from "./mcp-provisioning";

export const SHARED_MCP_URL_ENV = "AGENT_RELAY_SHARED_MCP_URL";
export const SHARED_CALLMUX_HOST_ENV = "AGENT_RELAY_SHARED_CALLMUX_HOST";
export const SHARED_CALLMUX_PORT_ENV = "AGENT_RELAY_SHARED_CALLMUX_PORT";
export const SHARED_CALLMUX_CONFIG_ENV = "AGENT_RELAY_SHARED_CALLMUX_CONFIG";
export const SHARED_CALLMUX_SOURCE_CONFIG_ENV = "AGENT_RELAY_SHARED_CALLMUX_SOURCE_CONFIG";
export const SHARED_CALLMUX_ENABLE_ENV = "AGENT_RELAY_SHARED_CALLMUX_ENABLE";

export const DEFAULT_SHARED_CALLMUX_HOST = "127.0.0.1";
export const DEFAULT_SHARED_CALLMUX_PORT = 4861;

// #1574 — how long a github/vent downstream is allowed to run on a captured
// `gh auth token` before the supervisor recycles it to re-mint a fresh one.
export const DEFAULT_GITHUB_TOKEN_REFRESH_MS = 30 * 60_000;

const CONFIG_SCHEMA = "callmux/schema.json";
const GITHUB_TOOLS = [
  "issue_read",
  "issue_write",
  "list_issues",
  "add_issue_comment",
  "search_issues",
  "search_code",
  "get_file_contents",
  "sub_issue_write",
];

export interface SharedCallmuxOptions {
  host: string;
  port: number;
  url: string;
  configPath: string;
  sourceConfigPath: string;
  enabled: boolean;
}

export interface SharedCallmuxSupervisorDeps {
  createListener(options: CreateListenerOptions): Promise<ProgrammaticListener>;
  setInterval(fn: () => void, ms: number): Timer;
  clearInterval(timer: Timer): void;
  setTimeout(fn: () => void, ms: number): Timer;
  clearTimeout(timer: Timer): void;
  fetch(input: string | URL | Request, init?: RequestInit): Promise<Response>;
  provisionServers(names?: Iterable<string>): PortableMcpProvisioningResult;
  log(message: string): void;
  report(snapshot: SharedCallmuxHealthSnapshot): void;
}

export interface SharedCallmuxHealthSnapshot {
  state: "disabled" | "starting" | "running" | "unhealthy" | "restarting" | "stopped";
  url: string;
  reason?: string;
  downstream?: ListenerHealthSnapshot["downstream"];
}

export function sharedCallmuxOptionsFromEnv(env: Record<string, string | undefined> = process.env): SharedCallmuxOptions {
  const explicitUrl = env[SHARED_MCP_URL_ENV];
  const parsed = explicitUrl ? parseListenerUrl(explicitUrl) : undefined;
  const host = env[SHARED_CALLMUX_HOST_ENV] ?? parsed?.hostname ?? DEFAULT_SHARED_CALLMUX_HOST;
  const port = numberEnv(env[SHARED_CALLMUX_PORT_ENV]) ?? parsed?.port ?? DEFAULT_SHARED_CALLMUX_PORT;
  const url = explicitUrl ?? `http://${host}:${port}/mcp`;
  return {
    host,
    port,
    url,
    configPath: env[SHARED_CALLMUX_CONFIG_ENV] || join(agentRelayHome(), "callmux", "shared-listener.json"),
    sourceConfigPath: env[SHARED_CALLMUX_SOURCE_CONFIG_ENV] || env.CALLMUX_CONFIG || join(homedir(), ".config", "callmux", "config.json"),
  enabled: !envOff(env[SHARED_CALLMUX_ENABLE_ENV]),
  };
}

export function sharedMcpListenerUrl(): string {
  return sharedCallmuxOptionsFromEnv().url;
}

export function writeSharedCallmuxConfig(opts: Pick<SharedCallmuxOptions, "configPath" | "sourceConfigPath"> & { registryServers?: CallmuxConfig["servers"] }): CallmuxConfig {
  const source = readJsonObject(opts.sourceConfigPath);
  const sourceServers = isRecord(source.servers) ? source.servers : {};
  // An omitted registry selection enables the legacy/source fallback. An
  // explicitly empty selection means provisioning was attempted and must stay
  // empty rather than resurrecting author-host descriptors.
  const rawServers = opts.registryServers === undefined
    ? fallbackSharedServers(sourceServers)
    : cloneServers(opts.registryServers);
  const servers = withSharedCallTimeoutFloor(rawServers);
  const generated: CallmuxConfig = {
    servers,
    cacheTtlSeconds: numberFromRecord(source, "cacheTtlSeconds") ?? 10,
    maxConcurrency: numberFromRecord(source, "maxConcurrency") ?? 20,
    callTimeoutMs: Math.max(
      numberFromRecord(source, "callTimeoutMs") ?? SHARED_CALLMUX_TOOL_CALL_TIMEOUT_MS,
      SHARED_CALLMUX_TOOL_CALL_TIMEOUT_MS,
    ),
    outputFormat: outputFormatFromRecord(source, "outputFormat") ?? "auto",
    // Relay workers consume only proxied tokenlean+github tools; suppress callmux meta-tools.
    exposeMetaTools: false,
  };
  const persisted = { $schema: CONFIG_SCHEMA, ...generated };
  mkdirSync(dirname(opts.configPath), { recursive: true });
  writeFileSync(opts.configPath, JSON.stringify(persisted, null, 2) + "\n", { mode: 0o600 });
  return generated;
}

export async function fetchSharedCallmuxRegistryServers(
  config: Pick<OrchestratorConfig, "relayUrl" | "token">,
  deps: Pick<SharedCallmuxSupervisorDeps, "fetch" | "log"> = { fetch: globalThis.fetch.bind(globalThis), log: () => {} },
): Promise<CallmuxConfig["servers"] | undefined> {
  try {
    const url = new URL("/api/provisioning/capabilities?kind=mcp", config.relayUrl);
    const headers: Record<string, string> = {};
    if (config.token) headers.Authorization = `Bearer ${config.token}`;
    const res = await deps.fetch(url, { headers });
    if (!res.ok) {
      deps.log(`[orchestrator] Shared callmux registry fetch failed: ${res.status}`);
      return undefined;
    }
    const payload = await res.json().catch(() => null) as unknown;
    const capabilities = isRecord(payload) && Array.isArray(payload.capabilities) ? payload.capabilities : [];
    const servers: CallmuxConfig["servers"] = {};
    for (const capability of capabilities) {
      if (!isRecord(capability) || !Array.isArray(capability.variants)) continue;
      for (const variant of capability.variants) {
        const entry = callmuxServerFromProvisioningVariant(variant);
        if (entry) servers[entry.name] = entry.server;
      }
    }
    return Object.keys(servers).length > 0 ? servers : undefined;
  } catch (err) {
    deps.log(`[orchestrator] Shared callmux registry fetch failed: ${errMessage(err)}`);
    return undefined;
  }
}

export class SharedCallmuxSupervisor {
  private listener: ProgrammaticListener | null = null;
  private readonly onStatus = (snapshot: ListenerHealthSnapshot) => this.reportCallmuxHealth(snapshot);
  private healthTimer: Timer | null = null;
  private restartTimer: Timer | null = null;
  private tokenRefreshTimer: Timer | null = null;
  private stopping = false;
  private backoffMs: number;

  constructor(
    private readonly config: OrchestratorConfig,
    private readonly opts = sharedCallmuxOptionsFromEnv(),
    private readonly deps: SharedCallmuxSupervisorDeps = defaultDeps(),
    private readonly timing: { healthIntervalMs?: number; restartBaseMs?: number; restartMaxMs?: number; tokenRefreshMs?: number } = {},
  ) {
    this.backoffMs = timing.restartBaseMs ?? 1_000;
  }

  start(): void {
    if (!this.opts.enabled) {
      this.deps.log("[orchestrator] shared callmux listener disabled by AGENT_RELAY_SHARED_CALLMUX_ENABLE=0");
      this.report("disabled", "global kill-switch");
      return;
    }
    this.deps.log(`[orchestrator] Shared callmux listener: ${this.opts.url}`);
    this.report("starting");
    void fireAndForget("Shared callmux open", () => this.open(), this.deps.log);
    this.healthTimer = this.deps.setInterval(() => {
      void fireAndForget("Shared callmux health check", () => this.checkHealth(), this.deps.log);
    }, this.timing.healthIntervalMs ?? 10_000);
    // #1574 — the github/vent downstream launchers capture `gh auth token`
    // once, at process spawn, and then run for the life of the shared
    // listener. A rotated or expired token never gets re-resolved on its own,
    // so agents see silent 401s until something restarts the listener. Recycle
    // it on a TTL through the existing restart/backoff path so the launchers
    // re-run and pick up a fresh token.
    this.tokenRefreshTimer = this.deps.setInterval(() => {
      this.refreshGithubToken();
    }, this.timing.tokenRefreshMs ?? DEFAULT_GITHUB_TOKEN_REFRESH_MS);
  }

  stop(): void {
    this.stopping = true;
    if (this.healthTimer) this.deps.clearInterval(this.healthTimer);
    if (this.restartTimer) this.deps.clearTimeout(this.restartTimer);
    if (this.tokenRefreshTimer) this.deps.clearInterval(this.tokenRefreshTimer);
    this.healthTimer = null;
    this.restartTimer = null;
    this.tokenRefreshTimer = null;
    if (this.listener) {
      const listener = this.listener;
      this.listener = null;
      listener.off("status", this.onStatus);
      void listener.stop().catch((err) => this.deps.log(`[orchestrator] Shared callmux listener stop failed: ${errMessage(err)}`));
    }
  }

  async checkHealth(): Promise<boolean> {
    if (!this.opts.enabled) return true;
    const snapshot = this.listener?.health();
    if (!snapshot) return false;
    this.reportCallmuxHealth(snapshot);
    if (snapshot.state === "running" || snapshot.state === "degraded" || snapshot.state === "reloading") {
      this.backoffMs = this.timing.restartBaseMs ?? 1_000;
      // callmux only recomputes its lifecycle state on start/reload, so a
      // "degraded" latch is not re-evaluated when a downstream reconnects at
      // runtime. Trust the live downstream counts instead: a degraded listener
      // with every downstream connected is healthy again (#1515).
      return mapCallmuxState(snapshot) === "running";
    }
    this.deps.log(`[orchestrator] Shared callmux listener is ${snapshot.state}; restarting`);
    this.restart(`listener ${snapshot.state}`);
    return false;
  }

  private refreshGithubToken(): void {
    if (this.stopping || !this.listener) return;
    this.deps.log("[orchestrator] Shared callmux gh-token TTL elapsed; recycling downstream servers to re-mint gh auth token");
    this.restart("gh token TTL refresh");
  }

  private async open(): Promise<void> {
    if (this.stopping || this.listener) return;
    try {
      // #1331 — the registry selects the surface, but never gets to replay its
      // author-host command paths. Known core entries are materialized as portable
      // host units (and qmd's SSH descriptor) under AGENT_RELAY_HOME instead.
      const registryServers = await fetchSharedCallmuxRegistryServers(this.config, this.deps);
      const provisioned = this.deps.provisionServers(registryServers ? Object.keys(registryServers) : undefined);
      for (const failure of provisioned.failures) this.deps.log(`[orchestrator] Shared MCP ${failure.name} unavailable: ${failure.reason}`);
      const effectiveServers = registryServers ? cloneServers(registryServers) : {};
      // Replace only the core descriptors we own. Registry-selected extensions
      // remain intact; a failed core provision never falls back to its
      // author-host path.
      for (const name of PORTABLE_SHARED_MCP_SERVER_NAMES) delete effectiveServers[name];
      Object.assign(effectiveServers, provisioned.servers);
      const persistedConfig = writeSharedCallmuxConfig({ ...this.opts, registryServers: effectiveServers });
      const config = withRuntimeEnv(persistedConfig, {
        ...this.config.env,
        [SHARED_MCP_URL_ENV]: this.opts.url,
      });
      const listener = await this.deps.createListener({
        host: this.opts.host,
        port: this.opts.port,
        config,
        configPath: this.opts.configPath,
      });
      if (this.stopping) {
        await listener.stop();
        return;
      }
      this.listener = listener;
      listener.on("status", this.onStatus);
      this.deps.log(`[orchestrator] Started shared callmux listener at ${listener.mcpUrl}`);
      this.reportCallmuxHealth(listener.health());
    } catch (err) {
      this.deps.log(`[orchestrator] Shared callmux listener failed to start: ${errMessage(err)}; scheduling restart`);
      this.report("restarting", errMessage(err));
      this.scheduleRestart();
    }
  }

  private restart(reason: string): void {
    if (this.listener) {
      const listener = this.listener;
      this.deps.log(`[orchestrator] Stopping shared callmux listener: ${reason}`);
      this.listener = null;
      listener.off("status", this.onStatus);
      void listener.stop().catch((err) => this.deps.log(`[orchestrator] Shared callmux listener stop failed: ${errMessage(err)}`));
    }
    this.report("restarting", reason);
    this.scheduleRestart();
  }

  private report(state: SharedCallmuxHealthSnapshot["state"], reason?: string, downstream?: ListenerHealthSnapshot["downstream"]): void {
    this.deps.report({
      state,
      url: this.opts.url,
      ...(reason ? { reason } : {}),
      ...(downstream ? { downstream } : {}),
    });
  }

  private reportCallmuxHealth(snapshot: ListenerHealthSnapshot): void {
    const state = mapCallmuxState(snapshot);
    // Only surface a failure reason while the mapped state is actually
    // unhealthy — a recovered "degraded" latch must not carry a stale reason
    // into a healthy report (#1515).
    const reason = state === "unhealthy"
      ? (snapshot.reason ?? (snapshot.downstream.failed > 0 ? `${snapshot.downstream.failed} downstream server(s) failed` : undefined))
      : undefined;
    this.deps.report({
      state,
      url: snapshot.mcpUrl,
      ...(reason ? { reason } : {}),
      downstream: snapshot.downstream,
    });
  }

  private scheduleRestart(): void {
    if (this.stopping || this.restartTimer) return;
    const delay = this.backoffMs;
    this.backoffMs = Math.min(this.backoffMs * 2, this.timing.restartMaxMs ?? 30_000);
    this.restartTimer = this.deps.setTimeout(() => {
      this.restartTimer = null;
      void fireAndForget("Shared callmux restart", () => this.open(), this.deps.log);
    }, delay);
  }
}

function defaultDeps(): SharedCallmuxSupervisorDeps {
  return {
    createListener,
    setInterval: (fn, ms) => setInterval(fn, ms),
    clearInterval: (timer) => clearInterval(timer),
    setTimeout: (fn, ms) => setTimeout(fn, ms),
    clearTimeout: (timer) => clearTimeout(timer),
    fetch: globalThis.fetch.bind(globalThis),
    provisionServers: (names) => provisionPortableSharedMcpServers({ names }),
    log: (message) => console.error(message),
    report: (snapshot) => console.error(`[orchestrator] shared callmux status ${snapshot.state}${snapshot.reason ? `: ${snapshot.reason}` : ""}`),
  };
}

function mapCallmuxState(snapshot: ListenerHealthSnapshot): SharedCallmuxHealthSnapshot["state"] {
  if (snapshot.state === "running") return "running";
  if (snapshot.state === "starting" || snapshot.state === "reloading") return "starting";
  if (snapshot.state === "stopped") return "stopped";
  // callmux computes its lifecycle state ("running"/"degraded") only on
  // start/reload and never re-derives it when a downstream reconnects at
  // runtime, so a "degraded" listener stays latched even after every downstream
  // is serving tools again. The live downstream counts in the snapshot are
  // always fresh — treat a degraded listener with no currently-failed
  // downstreams as healthy so the status latches back to running on
  // reconnect (#1515). A genuine live failure still reports unhealthy.
  if (snapshot.state === "degraded" && snapshot.downstream.failed === 0) return "running";
  return "unhealthy";
}

function envOff(value: string | undefined): boolean {
  if (value === undefined || value === null || value === "") return false;
  return ["0", "false", "off", "no"].includes(value.trim().toLowerCase());
}

function readJsonObject(path: string): Record<string, unknown> {
  if (!existsSync(path)) return {};
  const parsed = JSON.parse(readFileSync(path, "utf8"));
  return isRecord(parsed) ? parsed : {};
}

function cloneServer(value: unknown): CallmuxConfig["servers"][string] | undefined {
  if (!isRecord(value)) return undefined;
  return JSON.parse(JSON.stringify(value)) as CallmuxConfig["servers"][string];
}

function cloneServers(value: CallmuxConfig["servers"]): CallmuxConfig["servers"] {
  return Object.fromEntries(Object.entries(value).map(([name, server]) => [name, cloneServer(server) ?? server]));
}

function fallbackSharedServers(sourceServers: Record<string, unknown>): CallmuxConfig["servers"] {
  return {
    tokenlean: cloneServer(sourceServers.tokenlean) ?? defaultTokenleanServer(),
    github: cloneServer(sourceServers.github) ?? defaultGithubServer(),
  };
}

function withSharedCallTimeoutFloor(servers: CallmuxConfig["servers"]): CallmuxConfig["servers"] {
  return Object.fromEntries(Object.entries(servers).map(([name, server]) => {
    const callTimeoutMs = Math.max(
      server.callTimeoutMs ?? SHARED_CALLMUX_TOOL_CALL_TIMEOUT_MS,
      SHARED_CALLMUX_TOOL_CALL_TIMEOUT_MS,
    );
    return [name, { ...server, callTimeoutMs }];
  }));
}

function callmuxServerFromProvisioningVariant(value: unknown): { name: string; server: CallmuxConfig["servers"][string] } | null {
  if (!isRecord(value) || value.enabled === false || value.approvalStatus === "pending" || value.validationStatus === "invalid") return null;
  if (typeof value.name !== "string" || !value.name) return null;
  const definition = value.definition;
  if (!isRecord(definition) || definition.kind !== "mcp" || !isRecord(definition.server)) return null;
  const metadata = isRecord(definition.metadata) ? definition.metadata : {};
  const provenance = isRecord(value.provenance) ? value.provenance : {};
  if (metadata.sharedListenerEligible !== true && provenance.sharedListenerEligible !== true) return null;
  const rawCallmuxServer = isRecord(metadata.callmuxServer) ? metadata.callmuxServer : callmuxServerFromMcpServer(definition.server);
  const server = cloneServer(mergeConfigEnv(rawCallmuxServer));
  return server ? { name: value.name, server } : null;
}

function mergeConfigEnv(server: Record<string, unknown>): Record<string, unknown> {
  if (!isRecord(server.configEnv)) return server;
  const { configEnv, ...rest } = server;
  return {
    ...rest,
    env: { ...(configEnv as Record<string, string>), ...(isRecord(rest.env) ? (rest.env as Record<string, string>) : {}) },
  };
}

function callmuxServerFromMcpServer(server: Record<string, unknown>): Record<string, unknown> {
  const out: Record<string, unknown> = {};
  if (typeof server.command === "string") {
    out.command = server.command;
    if (Array.isArray(server.args)) out.args = server.args.filter((arg): arg is string => typeof arg === "string");
    if (isRecord(server.env)) out.env = server.env;
    if (isRecord(server.configEnv)) out.configEnv = server.configEnv;
  } else if (typeof server.url === "string") {
    out.url = server.url;
    if (server.type === "sse") out.transport = "sse";
    if (isRecord(server.headers)) out.headers = server.headers;
    if (isRecord(server.configEnv)) out.configEnv = server.configEnv;
  }
  return out;
}

function withRuntimeEnv(config: CallmuxConfig, env: Record<string, string | undefined>): CallmuxConfig {
  const inheritedEnv: Record<string, string> = {};
  for (const [key, value] of Object.entries(env)) {
    if (typeof value === "string") inheritedEnv[key] = value;
  }
  if (Object.keys(inheritedEnv).length === 0) return config;
  const servers = Object.fromEntries(Object.entries(config.servers).map(([name, server]) => {
    const rawHeaders = "headers" in server && isRecord(server.headers) ? server.headers as Record<string, string> : undefined;
    const headers = rawHeaders ? expandStringRecordPlaceholders(rawHeaders, inheritedEnv) : undefined;
    if (!("command" in server)) return [name, { ...server, ...(headers ? { headers } : {}) }];
    const serverEnv = isRecord(server.env) ? expandStringRecordPlaceholders(server.env as Record<string, string>, inheritedEnv) : {};
    return [name, {
      ...server,
      env: { ...inheritedEnv, ...serverEnv },
      ...(headers ? { headers } : {}),
    }];
  }));
  return { ...config, servers };
}

function expandStringRecordPlaceholders(value: Record<string, string>, env: Record<string, string>): Record<string, string> {
  return Object.fromEntries(Object.entries(value).map(([key, raw]) => [key, expandEnvPlaceholders(raw, env)]));
}

function expandEnvPlaceholders(value: string, env: Record<string, string>): string {
  return value.replace(/\$\{?([A-Z_][A-Z0-9_]*)\}?/g, (match, key: string) => env[key] ?? match);
}

function defaultTokenleanServer(): CallmuxConfig["servers"][string] {
  return {
    command: "tl-mcp",
    prefix: "",
    alwaysLoad: ["tl_symbols", "tl_snippet", "tl_pack", "tl_run", "tl_guard", "tl_lookup"],
    requireSessionCwd: true,
  };
}

function defaultGithubServer(): CallmuxConfig["servers"][string] {
  return {
    command: "github-mcp-server",
    args: ["stdio"],
    prefix: "gh",
    callTimeoutMs: SHARED_CALLMUX_TOOL_CALL_TIMEOUT_MS,
    tools: GITHUB_TOOLS,
    cachePolicy: { allowTools: ["issue_read", "list_issues", "search_issues", "search_code", "get_file_contents"] },
  };
}

function numberEnv(value: string | undefined): number | undefined {
  if (!value) return undefined;
  const parsed = Number(value);
  return Number.isFinite(parsed) && parsed > 0 ? parsed : undefined;
}

function numberFromRecord(value: Record<string, unknown>, key: string): number | undefined {
  return typeof value[key] === "number" ? value[key] : undefined;
}

function stringFromRecord(value: Record<string, unknown>, key: string): string | undefined {
  return typeof value[key] === "string" ? value[key] : undefined;
}

function outputFormatFromRecord(value: Record<string, unknown>, key: string): CallmuxConfig["outputFormat"] | undefined {
  const outputFormat = stringFromRecord(value, key);
  return outputFormat === "auto" || outputFormat === "json" || outputFormat === "toon" ? outputFormat : undefined;
}

function parseListenerUrl(value: string): { hostname: string; port: number } | undefined {
  try {
    const url = new URL(value);
    const port = Number(url.port) || (url.protocol === "https:" ? 443 : 80);
    return { hostname: url.hostname, port };
  } catch {
    return undefined;
  }
}