import assert from "node:assert/strict";
import { spawn } from "node:child_process";
import { once } from "node:events";
import { mkdtemp, rm } from "node:fs/promises";
import { createServer as createHttpServer, request as httpRequest } from "node:http";
import { createServer as createNetServer } from "node:net";
import { tmpdir } from "node:os";
import { dirname, join, resolve } from "node:path";
import test from "node:test";
import { fileURLToPath } from "node:url";

const foundryRoot = resolve(dirname(fileURLToPath(import.meta.url)), "..");

test("Map Gateway BFF caches only confirmed hits, honors validators and aborts abandoned streams", async () => {
  const root = await mkdtemp(join(tmpdir(), "nodedc-foundry-map-proxy-"));
  const gatewayPort = await freePort();
  const foundryPort = await freePort();
  let conditionalHeader = "";
  let resolveSlowClosed;
  const slowClosed = new Promise((resolveClosed) => { resolveSlowClosed = resolveClosed; });
  let resolveIdleClosed;
  const idleClosed = new Promise((resolveClosed) => { resolveIdleClosed = resolveClosed; });
  const gateway = createHttpServer((request, response) => {
    const url = new URL(request.url || "/", "http://gateway.local");
    if (url.pathname === "/healthz") {
      response.writeHead(200, { "content-type": "application/json" });
      response.end(JSON.stringify({ cache: { persistent: true, entries: 1, bytes: 4 } }));
      return;
    }
    if (url.pathname.startsWith("/api/map/ion/assets/")) {
      response.writeHead(200, { "content-type": "application/json" });
      response.end(JSON.stringify({ ok: true, credentialMode: "gateway" }));
      return;
    }
    if (url.pathname !== "/api/map/cache") {
      response.writeHead(404);
      response.end();
      return;
    }
    const target = String(url.searchParams.get("url") || "");
    conditionalHeader = String(request.headers["if-none-match"] || conditionalHeader);
    if (target.includes("/provider-error")) {
      response.writeHead(401, { "content-type": "application/json" });
      response.end(JSON.stringify({ error: "provider_unauthorized" }));
      return;
    }
    const headers = {
      "content-type": "application/octet-stream",
      etag: '"tile-v1"',
      "x-nodedc-map-cache": target.includes("nodedc_cache_refresh=1") ? "live-stale-upstream-error" : "live-cache-hit",
    };
    if (target.includes("/headers-slow")) {
      const timer = setTimeout(() => {
        if (response.destroyed) return;
        response.writeHead(200, { ...headers, "content-length": "4" });
        response.end("tile");
      }, 180);
      response.once("close", () => clearTimeout(timer));
      return;
    }
    if (target.includes("/body-idle")) {
      response.writeHead(200, headers);
      response.write("start");
      const timer = setTimeout(() => response.end("late"), 500);
      response.once("close", () => {
        clearTimeout(timer);
        resolveIdleClosed();
      });
      return;
    }
    if (target.includes("/progress")) {
      response.writeHead(200, { ...headers, "content-length": String(8 * 256) });
      let chunks = 0;
      const interval = setInterval(() => {
        chunks += 1;
        response.write(Buffer.alloc(256, chunks));
        if (chunks === 8) {
          clearInterval(interval);
          response.end();
        }
      }, 30);
      response.once("close", () => clearInterval(interval));
      return;
    }
    if (target.includes("/slow")) {
      response.writeHead(200, headers);
      response.write(Buffer.alloc(1024, 1));
      const interval = setInterval(() => response.write(Buffer.alloc(1024, 2)), 20);
      response.once("close", () => {
        clearInterval(interval);
        resolveSlowClosed();
      });
      return;
    }
    response.writeHead(200, { ...headers, "content-length": "4" });
    response.end("tile");
  });
  let foundry;

  try {
    gateway.listen(gatewayPort, "127.0.0.1");
    await once(gateway, "listening");
    foundry = spawn(process.execPath, ["server/catalog-server.mjs"], {
      cwd: foundryRoot,
      env: {
        ...process.env,
        NODE_ENV: "development",
        HOST: "127.0.0.1",
        PORT: String(foundryPort),
        FOUNDRY_RUNTIME_DIR: root,
        NODEDC_FOUNDRY_AUTH_REQUIRED: "false",
        NODEDC_MAP_GATEWAY_INTERNAL_URL: `http://127.0.0.1:${gatewayPort}`,
        NODEDC_MAP_GATEWAY_HEADERS_TIMEOUT_MS: "100",
        NODEDC_MAP_GATEWAY_BODY_IDLE_TIMEOUT_MS: "80",
      },
      stdio: ["ignore", "pipe", "pipe"],
    });
    await waitForService(foundryPort, foundry);
    const base = `http://127.0.0.1:${foundryPort}`;
    const proxied = (target) => `${base}/api/map-gateway/api/map/cache?url=${encodeURIComponent(target)}`;

    const hit = await fetch(proxied("https://assets.ion.cesium.com/1/tile.bin"));
    assert.equal(hit.status, 200);
    assert.equal(await hit.text(), "tile");
    assert.equal(hit.headers.get("cache-control"), "private, max-age=300, stale-while-revalidate=60");
    assert.equal(hit.headers.get("vary"), "Cookie");
    assert.equal(hit.headers.get("etag"), '"tile-v1"');

    const conditional = await fetch(proxied("https://assets.ion.cesium.com/1/tile.bin"), {
      headers: { "if-none-match": 'W/"tile-v1"' },
    });
    assert.equal(conditional.status, 304);
    assert.equal(conditionalHeader, 'W/"tile-v1"');
    assert.equal(conditional.headers.get("cache-control"), "private, max-age=300, stale-while-revalidate=60");

    const refresh = await fetch(proxied("https://assets.ion.cesium.com/1/tile.bin?nodedc_cache_refresh=1"));
    assert.equal(refresh.status, 200);
    assert.equal(refresh.headers.get("cache-control"), "no-store");

    const providerError = await fetch(proxied("https://assets.ion.cesium.com/1/provider-error"));
    assert.equal(providerError.status, 401);
    assert.equal(providerError.headers.get("cache-control"), "no-store");

    const endpoint = await fetch(`${base}/api/map-gateway/api/map/ion/assets/1/endpoint`);
    assert.equal(endpoint.status, 200);
    assert.equal(endpoint.headers.get("cache-control"), "no-store");
    const health = await fetch(`${base}/api/map-gateway/healthz`);
    assert.equal(health.status, 200);
    assert.equal(health.headers.get("cache-control"), "no-store");

    const progressing = await fetch(proxied("https://assets.ion.cesium.com/1/progress"));
    assert.equal(progressing.status, 200);
    assert.equal((await progressing.arrayBuffer()).byteLength, 8 * 256);

    const headersTimeout = await fetch(proxied("https://assets.ion.cesium.com/1/headers-slow"));
    assert.equal(headersTimeout.status, 504);
    assert.equal((await headersTimeout.json()).error, "map_gateway_headers_timeout");

    const idle = await fetch(proxied("https://assets.ion.cesium.com/1/body-idle"));
    assert.equal(idle.status, 200);
    await assert.rejects(idle.arrayBuffer());
    await Promise.race([
      idleClosed,
      new Promise((_, reject) => setTimeout(() => reject(new Error("upstream_body_idle_abort_timeout")), 2_000)),
    ]);

    await abortAfterFirstChunk(proxied("https://assets.ion.cesium.com/1/slow"));
    await Promise.race([
      slowClosed,
      new Promise((_, reject) => setTimeout(() => reject(new Error("upstream_abort_timeout")), 2_000)),
    ]);
    const stillHealthy = await fetch(`${base}/healthz`);
    assert.equal(stillHealthy.status, 200);
  } finally {
    await stop(foundry);
    if (gateway.listening) {
      gateway.close();
      await once(gateway, "close");
    }
    await rm(root, { recursive: true, force: true });
  }
});

function abortAfterFirstChunk(url) {
  return new Promise((resolveAbort, rejectAbort) => {
    const request = httpRequest(url, (response) => {
      response.once("data", () => {
        response.destroy();
        request.destroy();
        resolveAbort();
      });
    });
    request.once("error", (error) => {
      if (error?.code === "ECONNRESET") resolveAbort();
      else rejectAbort(error);
    });
    request.end();
  });
}

async function freePort() {
  const server = createNetServer();
  server.listen(0, "127.0.0.1");
  await once(server, "listening");
  const address = server.address();
  assert(address && typeof address === "object");
  const port = address.port;
  server.close();
  await once(server, "close");
  return port;
}

async function waitForService(port, child) {
  let output = "";
  child.stderr.on("data", (chunk) => { output += String(chunk); });
  for (let attempt = 0; attempt < 100; attempt += 1) {
    if (child.exitCode !== null) throw new Error(`foundry_exited:${child.exitCode}:${output}`);
    try {
      const response = await fetch(`http://127.0.0.1:${port}/healthz`);
      if (response.ok) return;
    } catch { /* service is still starting */ }
    await new Promise((resolveWait) => setTimeout(resolveWait, 50));
  }
  throw new Error(`foundry_start_timeout:${output}`);
}

async function stop(child) {
  if (!child || child.exitCode !== null) return;
  child.kill("SIGTERM");
  await once(child, "exit").catch(() => undefined);
}