#!/usr/bin/env node
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import { StreamableHTTPServerTransport } from "@modelcontextprotocol/sdk/server/streamableHttp.js";
import express from "express";
import compression from "compression";
import cors from "cors";
import { execSync } from "node:child_process";
import fs from "node:fs";
import path from "node:path";
import { pipeline } from "node:stream/promises";
import { fileURLToPath } from "node:url";
import { createGunzip } from "node:zlib";

import { RijksmuseumApiClient } from "./api/RijksmuseumApiClient.js";
import { OaiPmhClient } from "./api/OaiPmhClient.js";
import { VocabularyDb } from "./api/VocabularyDb.js";
import { EmbeddingsDb } from "./api/EmbeddingsDb.js";
import { EmbeddingModel } from "./api/EmbeddingModel.js";
import { ResponseCache } from "./utils/ResponseCache.js";
import { UsageStats } from "./utils/UsageStats.js";
import {
  captureMemorySnapshot,
  formatMemorySnapshotDetailed,
  type DbHandle,
} from "./utils/MemoryStats.js";
import { registerAll, similarPages, enrichmentReviewPages } from "./registration.js";
import { isAllowedOrigin, parseMcpAllowedOrigins } from "./utils/origin.js";

const SERVER_NAME = "rijksmuseum-mcp+";

const __dirname = path.dirname(fileURLToPath(import.meta.url));
const pkg = JSON.parse(fs.readFileSync(path.join(__dirname, "..", "package.json"), "utf-8"));
const SERVER_VERSION: string = pkg.version;

function getGitCommit(): string {
  // Railway webhook deploys
  if (process.env.RAILWAY_GIT_COMMIT_SHA) {
    return process.env.RAILWAY_GIT_COMMIT_SHA.slice(0, 7);
  }
  // `railway up` / CLI deploys — read hash baked at build time
  try {
    const commitFile = new URL("commit.txt", import.meta.url);
    return fs.readFileSync(commitFile, "utf-8").trim();
  } catch { /* no build artifact */ }
  // Local dev with git repo available
  try {
    return execSync("git rev-parse --short HEAD", { encoding: "utf-8" }).trim();
  } catch {
    return "unknown";
  }
}

const GIT_COMMIT = getGitCommit();

// ─── Types ───────────────────────────────────────────────────────────

// ─── Determine transport mode ────────────────────────────────────────

function shouldUseHttp(): boolean {
  return process.argv.includes("--http") || !!process.env.PORT;
}

function getHttpPort(): number {
  return parseInt(process.env.PORT ?? "3000", 10);
}

// ─── Database download helpers (run once per volume lifetime) ────────

interface DbSpec {
  name: string;
  pathEnvVar: string;
  urlEnvVar: string;
  defaultFile: string;
  /** SQL query that must succeed for the DB to be considered valid. */
  validationQuery: string;
}

const VOCAB_DB_SPEC: DbSpec = {
  name: "Vocabulary",
  pathEnvVar: "VOCAB_DB_PATH",
  urlEnvVar: "VOCAB_DB_URL",
  defaultFile: "vocabulary.db",
  validationQuery: "SELECT 1 FROM vocab_term_counts LIMIT 1",
};

const EMBEDDINGS_DB_SPEC: DbSpec = {
  name: "Embeddings",
  pathEnvVar: "EMBEDDINGS_DB_PATH",
  urlEnvVar: "EMBEDDINGS_DB_URL",
  defaultFile: "embeddings.db",
  validationQuery: "SELECT 1 FROM artwork_embeddings LIMIT 1",
};

function resolveDbPathForSpec(spec: DbSpec): string {
  return process.env[spec.pathEnvVar] || path.join(__dirname, "..", "data", spec.defaultFile);
}

/**
 * Ensure a SQLite database exists and passes validation.
 * Downloads from the URL env var if missing or invalid.
 */
async function ensureDb(spec: DbSpec): Promise<void> {
  const dbPath = resolveDbPathForSpec(spec);

  if (fs.existsSync(dbPath)) {
    try {
      const { default: Database } = await import("better-sqlite3");
      const db = new Database(dbPath, { readonly: true });
      db.prepare(spec.validationQuery).get();
      db.close();
      return;
    } catch {
      console.error(`${spec.name} DB invalid or outdated — will re-download`);
    }
  }

  const url = process.env[spec.urlEnvVar];
  if (!url) return;

  console.error(`Downloading ${spec.name} DB...`);
  const dir = path.dirname(dbPath);
  if (!fs.existsSync(dir)) fs.mkdirSync(dir, { recursive: true });

  const tmpPath = dbPath + ".tmp";
  const controller = new AbortController();
  const downloadTimer = setTimeout(() => controller.abort(), 330_000);
  try {
    const res = await fetch(url, { redirect: "follow", signal: controller.signal });
    if (!res.ok || !res.body) throw new Error(`HTTP ${res.status} ${res.statusText}`);

    const dest = fs.createWriteStream(tmpPath);
    const isGzip = url.endsWith(".gz") || res.headers.get("content-type")?.includes("gzip");

    if (isGzip) {
      await pipeline(res.body, createGunzip(), dest);
    } else {
      await pipeline(res.body, dest);
    }

    fs.renameSync(tmpPath, dbPath);
    console.error(`${spec.name} DB ready: ${dbPath}`);
  } catch (err) {
    console.error(`Failed to download ${spec.name} DB: ${err instanceof Error ? err.message : err}`);
    if (fs.existsSync(tmpPath)) fs.unlinkSync(tmpPath);
  } finally {
    clearTimeout(downloadTimer);
  }
}

// ─── Shared database instances (one read-only instance each) ─────────

let vocabDb: VocabularyDb | null = null;
let embeddingsDb: EmbeddingsDb | null = null;
let embeddingModel: EmbeddingModel | null = null;

async function initDatabases(): Promise<void> {
  await ensureDb(VOCAB_DB_SPEC);
  vocabDb = new VocabularyDb();
  await ensureDb(EMBEDDINGS_DB_SPEC);
  embeddingsDb = new EmbeddingsDb();
  const needsModel = embeddingsDb.available;
  if (needsModel) {
    embeddingModel = new EmbeddingModel();
    const modelId = process.env.EMBEDDING_MODEL_ID ?? "Xenova/multilingual-e5-small";
    const targetDim = embeddingsDb.available ? embeddingsDb.vectorDimensions : 0;
    await embeddingModel.init(modelId, targetDim);
  }
}

function buildMemoryDbHandles(): DbHandle[] {
  return [
    { name: "vocab", filePath: vocabDb?.dbPath ?? null, db: vocabDb?.rawDb ?? null },
    { name: "embeddings", filePath: embeddingsDb?.dbPath ?? null, db: embeddingsDb?.rawDb ?? null },
  ];
}

// ─── Shared client instances (one per process) ──────────────────────

let sharedApiClient: RijksmuseumApiClient;
let sharedOaiClient: OaiPmhClient;

function initSharedClients(): void {
  const cache = new ResponseCache(1000, 5 * 60_000);
  sharedApiClient = new RijksmuseumApiClient(cache);
  sharedOaiClient = new OaiPmhClient();
}

// ─── Usage stats accumulator ─────────────────────────────────────────

let usageStats: UsageStats | undefined;

function initUsageStats(): void {
  usageStats = new UsageStats();
}

// ─── Create a configured McpServer ───────────────────────────────────

function createServer(httpPort?: number): McpServer {
  const server = new McpServer(
    { name: SERVER_NAME, version: SERVER_VERSION, title: "Rijksmuseum MCP+" },
    {
      capabilities: { tools: {}, resources: {}, prompts: {} },
      instructions:
        "Rijksmuseum collection explorer — circa 834,000 artworks from antiquity to the present day " +
        "spanning paintings, prints, drawings, photographs, furniture, ceramics, textiles, and more.\n\n" +

        "Search uses a vocabulary database. All filters combine freely; array values are AND-combined. " +
        "Vocabulary labels are bilingual (English/Dutch) — fall back to the Dutch term when an English query " +
        "returns nothing (e.g. 'fotograaf' for 'photographer').\n\n" +

        "Images are served via IIIF deep-zoom. Three viewer-family tools: get_artwork_image opens an interactive " +
        "viewer for the user; inspect_artwork_image returns base64 bytes for the LLM's own visual analysis; " +
        "navigate_viewer highlights regions in an already-open viewer.\n\n" +

        "For demographic person queries (gender, birth/death year/place, profession), use search_persons → feed " +
        "the returned vocabId into search_artwork({creator}). For artworks depicting a known person, use " +
        "search_artwork({aboutActor}) — broader recall than depictedPerson because it searches both subject and " +
        "creator vocabularies and tolerates cross-language name forms.\n\n" +

        "Place searches support depictedPlace and productionPlace as vocabulary filters. " +
        "For proximity, most place vocabulary entries lack coordinates — nearPlace only works for the subset that has been " +
        "authority-geocoded, but nearLat/nearLon with explicit coordinates always works for searching artworks near any point. " +
        "Multi-word place names like 'Oude Kerk Amsterdam' still resolve to a single vocabulary entry regardless of " +
        "coordinate coverage.\n\n" +

        "Iconclass covers ~39,800 base subject notations (1.3M+ including key-expanded variants). Use the Iconclass server's search tool to find notation codes by concept, " +
        "then pass them to search_artwork for precise iconographic filtering.\n\n" +

        "Two free-text corpora are searchable through search_artwork: descriptions (Dutch, cataloguer-written, " +
        "~61% coverage) and curatorialNarrative (English wall text, ~14K works). When neither corpus carries the " +
        "relevant language, prefer semantic_search.\n\n" +

        "Three search modes: search_artwork for structured filters, semantic_search for free-text concepts that " +
        "resist structured metadata, find_similar for artwork-to-artwork similarity from a known objectNumber. " +
        "For aggregate counts and distributions, use collection_stats instead of looping search_artwork calls.\n\n" +

        "Specialised tools: search_provenance for ownership-history questions across the ~48K artworks with " +
        "parsed provenance; list_curated_sets + browse_set for exhibition/thematic groupings curated by Rijksmuseum " +
        "staff; get_recent_changes for OAI-PMH delta tracking against a harvest checkpoint.",
    }
  );

  registerAll(server, sharedApiClient, sharedOaiClient, vocabDb, embeddingsDb, embeddingModel, httpPort, usageStats);
  return server;
}

// ─── Stdio mode ──────────────────────────────────────────────────────

async function runStdio(): Promise<void> {
  await initDatabases();
  initSharedClients();
  initUsageStats();
  const server = createServer();
  // Eager warm-up (~13s) is load-bearing for Claude Desktop's first-query latency but
  // pure waste for a short-lived CLI one-shot (caches build lazily on first use anyway).
  // The CLI's stdio transport sets MCP_SKIP_STARTUP_WARM=1; default stays eager.
  if (process.env.MCP_SKIP_STARTUP_WARM !== "1") {
    if (vocabDb?.available) { vocabDb.warmCorePages(); vocabDb.warmSimilarCaches(); vocabDb.ensureCuratedSetsCache(); }
    if (embeddingsDb?.available) embeddingsDb.warmCorePages();
  }
  const transport = new StdioServerTransport();
  await server.connect(transport);
  console.error("Rijksmuseum MCP server running on stdio");
}

// ─── HTTP mode ───────────────────────────────────────────────────────

async function runHttp(): Promise<void> {
  await initDatabases();
  initSharedClients();
  initUsageStats();
  const port = getHttpPort();
  const app = express();
  app.set("trust proxy", 1); // Railway reverse proxy sets X-Forwarded-For

  const allowedOrigins = process.env.ALLOWED_ORIGINS;
  app.use(
    cors({
      origin: allowedOrigins ? allowedOrigins.split(",") : "*",
    })
  );
  app.use(compression({ threshold: 1024 })); // gzip responses ≥ 1 KB (skips small payloads + base64 images)
  app.use(express.json());

  // ── MCP endpoint (stateless — no sessions, no SSE streams) ─────
  //
  // Each POST creates a fresh transport+server, processes the request, responds,
  // and closes both. A single shared McpServer cannot be used here: Protocol.connect()
  // requires _transport to be unset, so two overlapping requests would collide
  // ("Already connected to a transport") — and the ChatGPT app bridge fires
  // concurrent call_mcp requests (in-viewer related-variant nav + the poll loop).
  // No long-lived connections to time out (#41). createServer() is cheap — the DBs,
  // API clients, and embedding model are module-scope singletons; only tool/Zod
  // registration runs per request.

  // 30s safety net — respond 504 before Railway's proxy kills the connection silently
  app.use("/mcp", (_req: express.Request, res: express.Response, next: express.NextFunction) => {
    res.setTimeout(30_000, () => {
      if (!res.headersSent) {
        res.status(504).json({ error: "Request timeout" });
      }
    });
    next();
  });

  // ── Origin validation (DNS-rebinding mitigation, spec §Streamable HTTP) ─
  //
  // Spec: https://modelcontextprotocol.io/specification/2025-11-25/basic/transports
  // Allow categories: missing Origin (CLI/stdio), non-web schemes (app://,
  // chrome-extension://, file://, "null"), localhost/loopback, and the
  // configured web-host allowlist. Defaults cover claude.ai / chatgpt.com /
  // *.openai.com; override via MCP_ALLOWED_ORIGINS (comma-separated, or "*"
  // to disable web-host checking).
  const mcpOriginAllowlist = parseMcpAllowedOrigins(process.env.MCP_ALLOWED_ORIGINS);
  if (mcpOriginAllowlist === "*") {
    console.warn("[mcp-origin] WARNING: web-host validation disabled (MCP_ALLOWED_ORIGINS=*)");
  }
  const recentOriginRejections = new Map<string, number>();
  const ORIGIN_LOG_INTERVAL_MS = 60_000;
  app.use("/mcp", (req: express.Request, res: express.Response, next: express.NextFunction) => {
    const origin = req.get("origin");
    if (isAllowedOrigin(origin, mcpOriginAllowlist)) {
      next();
      return;
    }
    const now = Date.now();
    const lastLogged = recentOriginRejections.get(origin!) ?? 0;
    if (now - lastLogged >= ORIGIN_LOG_INTERVAL_MS) {
      recentOriginRejections.set(origin!, now);
      console.warn(`[mcp-origin] 403 rejected origin=${origin} ip=${req.ip}`);
    }
    res.status(403).json({
      jsonrpc: "2.0",
      error: { code: -32000, message: "Origin not allowed" },
    });
  });

  app.post("/mcp", async (req: express.Request, res: express.Response) => {
    const server = createServer(port);
    const transport = new StreamableHTTPServerTransport({
      sessionIdGenerator: undefined,
    });
    try {
      await server.connect(transport);
      await transport.handleRequest(req, res, req.body);
    } catch (err) {
      console.error("MCP endpoint error:", err);
      if (!res.headersSent) {
        res.status(500).json({ error: "Internal server error" });
      }
    } finally {
      await transport.close();
      await server.close();
    }
  });

  // Reject GET/DELETE/etc. — stateless mode has no SSE streams or sessions.
  //
  // Spec-conformant: §Streamable HTTP → Listening for Messages from the Server
  // says the server MUST either return Content-Type: text/event-stream OR
  // HTTP 405 Method Not Allowed, "indicating that the server does not offer an
  // SSE stream at this endpoint":
  //   https://modelcontextprotocol.io/specification/2025-11-25/basic/transports
  app.all("/mcp", (_req: express.Request, res: express.Response) => {
    res.setHeader("Allow", "POST");
    res.status(405).json({ error: "Method not allowed — this server is stateless (POST only)" });
  });

  // ── Similar artworks comparison page ────────────────────────────

  app.get("/similar/:uuid", (req: express.Request, res: express.Response) => {
    const page = similarPages.get(req.params.uuid as string);
    if (!page) {
      res.status(404).json({ error: "Page not found or expired (30 min TTL)" });
      return;
    }
    page.lastAccess = Date.now();
    res.type("html").send(page.html);
  });

  app.get("/enrichment-review/:uuid", (req: express.Request, res: express.Response) => {
    const page = enrichmentReviewPages.get(req.params.uuid as string);
    if (!page) {
      res.status(404).json({ error: "Page not found or expired (30 min TTL)" });
      return;
    }
    page.lastAccess = Date.now();
    res.type("html").send(page.html);
  });

  // ── Warm DB pages before accepting traffic ─────────────────────
  //
  // Page in critical mmap regions so the first real user query is fast.
  // This runs before app.listen(), so Railway's healthcheck only passes
  // once the DBs are warm.

  if (vocabDb?.available) { vocabDb.warmCorePages(); vocabDb.warmSimilarCaches(); vocabDb.ensureCuratedSetsCache(); }
  if (embeddingsDb?.available) embeddingsDb.warmCorePages();

  // ── Health + readiness ──────────────────────────────────────────
  //
  // /health is the Railway liveness probe — passes as soon as express is
  // listening (DB core pages already warmed above).
  // /ready is an informational readiness flag: flips true once the ONNX
  // embedding model has done one inference and the filtered-KNN path is
  // warm. Railway is NOT configured to gate on /ready — the flag is purely
  // for observability so cold-path spikes are attributable.

  let ready = false;

  app.get("/health", (_req: express.Request, res: express.Response) => {
    res.json({ status: "ok", server: SERVER_NAME, version: SERVER_VERSION, commit: GIT_COMMIT });
  });

  app.get("/ready", (_req: express.Request, res: express.Response) => {
    res.json({ ready, status: ready ? "warm" : "warming" });
  });

  // ── Memory observability (issue #272) ───────────────────────────
  //
  // /debug/memory returns the same snapshot used for startup + periodic logs.
  // Unauthenticated to match /health, /similar, /enrichment-review — the data
  // is operational signal, not sensitive.

  app.get("/debug/memory", (_req: express.Request, res: express.Response) => {
    res.json(captureMemorySnapshot(buildMemoryDbHandles()));
  });

  // /debug/slow-queries surfaces per-input p50/p90/max + repeat counts and the
  // search/semantic phase splits (#378 instrumentation). In-memory only — resets
  // on redeploy. Same unauthenticated operational-signal rationale as /debug/memory.
  app.get("/debug/slow-queries", (_req: express.Request, res: express.Response) => {
    res.json(usageStats?.slowQueries() ?? { perInput: {}, phases: {} });
  });

  // ── Start ───────────────────────────────────────────────────────

  httpServer = app.listen(port, () => {
    console.error(`Rijksmuseum MCP server listening on http://localhost:${port}`);
    console.error(`  MCP endpoint: POST /mcp`);
    console.error(`  Health:       GET  /health`);
    console.error(`  Ready:        GET  /ready`);

    // Post-listen warm-up — healthcheck already passes, so this doesn't delay
    // deployment. Covers the ONNX first-inference cost + filtered-KNN page
    // faults that warmCorePages() doesn't touch.
    void (async () => {
      const t0 = Date.now();
      try {
        let warmVec: Float32Array | undefined;
        if (embeddingModel?.available) {
          warmVec = await embeddingModel.embed("warmup");
          console.error(`  Embedding model first-inference warmed in ${Date.now() - t0}ms`);
        }
        if (embeddingsDb?.available) {
          const vec = warmVec ?? new Float32Array(embeddingsDb.vectorDimensions);
          embeddingsDb.warmFilteredPath(vec);
        }
        ready = true;
        console.error(`  Post-listen warmup complete in ${Date.now() - t0}ms — /ready now true`);
        console.error(formatMemorySnapshotDetailed(captureMemorySnapshot(buildMemoryDbHandles())));
      } catch (err) {
        console.error(`  Post-listen warmup failed: ${err instanceof Error ? err.message : err}`);
        ready = true; // don't leave /ready stuck — failure is logged
      }
    })();
  });
}


// ─── Graceful shutdown ───────────────────────────────────────────────

let httpServer: import("node:http").Server | undefined;

function shutdown() {
  console.error("Shutting down...");
  usageStats?.flush();
  httpServer?.close();
  process.exit(0);
}

process.on("SIGTERM", shutdown);
process.on("SIGINT", shutdown);

// ─── Entry point ─────────────────────────────────────────────────────

if (shouldUseHttp()) {
  runHttp().catch((err) => {
    console.error("Failed to start HTTP server:", err);
    process.exit(1);
  });
} else {
  runStdio().catch((err) => {
    console.error("Failed to start stdio server:", err);
    process.exit(1);
  });
}
