import assert from "node:assert/strict";
import { spawn } from "node:child_process";
import { once } from "node:events";
import { mkdtemp, readdir, readFile, rm } from "node:fs/promises";
import { createServer, get as httpGet } from "node:http";
import { createServer as createNetServer } from "node:net";
import { tmpdir } from "node:os";
import { join } from "node:path";

const root = await mkdtemp(join(tmpdir(), "nodedc-map-gateway-streaming-smoke-"));
const cacheDir = join(root, "cache");
const upstreamPort = await freePort();
const gatewayPort = await freePort();
const distinctCount = 12;
const state = {
  sharedRequests: 0,
  sharedEnded: false,
  abortRequests: 0,
  abortEnded: false,
  brokenRequests: 0,
  distinctRequests: 0,
  distinctResponses: [],
};
let upstream;
let gateway;

try {
  upstream = createServer((request, response) => {
    const requestUrl = new URL(request.url || "/", `http://127.0.0.1:${upstreamPort}`);
    if (requestUrl.pathname === "/slow/shared") {
      state.sharedRequests += 1;
      response.writeHead(200, { "content-type": "application/octet-stream", "cache-control": "max-age=3600" });
      response.write("first-");
      setTimeout(() => {
        state.sharedEnded = true;
        response.end("second");
      }, 250);
      return;
    }
    if (requestUrl.pathname === "/slow/abort") {
      state.abortRequests += 1;
      response.writeHead(200, { "content-type": "application/octet-stream", "cache-control": "max-age=3600" });
      response.write("abort-first-");
      setTimeout(() => {
        state.abortEnded = true;
        response.end("abort-second");
      }, 250);
      return;
    }
    if (requestUrl.pathname === "/broken") {
      state.brokenRequests += 1;
      response.writeHead(200, { "content-type": "application/octet-stream", "cache-control": "max-age=3600" });
      response.write("partial-must-not-commit");
      setTimeout(() => response.destroy(new Error("intentional_upstream_truncation")), 50);
      return;
    }
    if (requestUrl.pathname.startsWith("/distinct/")) {
      state.distinctRequests += 1;
      state.distinctResponses.push({ id: requestUrl.pathname.split("/").pop(), response });
      if (state.distinctResponses.length === distinctCount) {
        for (const item of state.distinctResponses) {
          item.response.writeHead(200, { "content-type": "text/plain", "cache-control": "max-age=3600" });
          item.response.end(`distinct-${item.id}`);
        }
      }
      return;
    }
    response.writeHead(404).end();
  });
  upstream.listen(upstreamPort, "127.0.0.1");
  await once(upstream, "listening");

  gateway = spawn(process.execPath, ["src/server.mjs"], {
    cwd: new URL("..", import.meta.url),
    env: {
      ...process.env,
      NODE_ENV: "test",
      PORT: String(gatewayPort),
      MAP_CACHE_DIR: cacheDir,
      MAP_CACHE_MODE: "readwrite",
      MAP_GATEWAY_ALLOW_ANONYMOUS: "true",
      MAP_GATEWAY_UPSTREAM_ALLOWLIST: "127.0.0.1",
      MAP_GATEWAY_TEST_ALLOW_HTTP_LOOPBACK: "true",
    },
    stdio: ["ignore", "pipe", "pipe"],
  });
  await waitForGateway(gatewayPort, gateway);

  const sharedSource = `http://127.0.0.1:${upstreamPort}/slow/shared`;
  const sharedUrl = gatewayUrl(gatewayPort, sharedSource);
  const leader = await fetch(sharedUrl);
  assert.equal(leader.status, 200);
  const reader = leader.body.getReader();
  const first = await reader.read();
  assert.equal(Buffer.from(first.value).toString("utf8"), "first-");
  assert.equal(state.sharedEnded, false, "first byte must reach the leader before upstream end");
  assert.equal(await indexEntryCount(cacheDir), 0, "partial body must not be published in index");

  const followerPromise = fetch(sharedUrl).then(async (response) => ({ status: response.status, body: await response.text() }));
  const leaderBody = `first-${await readRemaining(reader)}`;
  const follower = await followerPromise;
  assert.equal(leaderBody, "first-second");
  assert.deepEqual(follower, { status: 200, body: "first-second" });
  assert.equal(state.sharedRequests, 1, "same-key followers must share one upstream");
  assert.equal(await indexEntryCount(cacheDir), 1, "complete body must be committed");

  const abortSource = `http://127.0.0.1:${upstreamPort}/slow/abort`;
  const abortUrl = gatewayUrl(gatewayPort, abortSource);
  assert.equal(await abortAfterFirstChunk(abortUrl), "abort-first-");
  const afterAbort = await fetch(abortUrl);
  assert.equal(await afterAbort.text(), "abort-first-abort-second");
  assert.equal(state.abortEnded, true);
  assert.equal(state.abortRequests, 1, "client abort must not cancel the server-owned cache fill");
  assert.equal(await indexEntryCount(cacheDir), 2);

  const beforeBroken = await indexEntryCount(cacheDir);
  const brokenSource = `http://127.0.0.1:${upstreamPort}/broken`;
  let brokenRejected = false;
  try {
    const broken = await fetch(gatewayUrl(gatewayPort, brokenSource));
    await broken.arrayBuffer();
  } catch {
    brokenRejected = true;
  }
  assert.equal(brokenRejected, true);
  await waitFor(async () => (await tempFiles(cacheDir)).length === 0);
  assert.equal(await indexEntryCount(cacheDir), beforeBroken, "truncated upstream must not publish an index entry");
  assert.equal((await tempFiles(cacheDir)).length, 0, "truncated upstream must remove temp files");
  assert.equal((await fetch(`http://127.0.0.1:${gatewayPort}/healthz`)).status, 200, "stream failure must not stop Gateway");

  const diagnosticsBefore = (await (await fetch(`http://127.0.0.1:${gatewayPort}/healthz`)).json()).diagnostics;
  const distinctBodies = await Promise.all(Array.from({ length: distinctCount }, async (_, index) => {
    const response = await fetch(gatewayUrl(gatewayPort, `http://127.0.0.1:${upstreamPort}/distinct/${index}`));
    return response.text();
  }));
  assert.deepEqual(distinctBodies.sort(), Array.from({ length: distinctCount }, (_, index) => `distinct-${index}`).sort());
  await waitFor(async () => await indexEntryCount(cacheDir) === beforeBroken + distinctCount);
  const diagnosticsAfter = (await (await fetch(`http://127.0.0.1:${gatewayPort}/healthz`)).json()).diagnostics;
  const snapshotDelta = diagnosticsAfter.indexSnapshots - diagnosticsBefore.indexSnapshots;
  assert.equal(state.distinctRequests, distinctCount);
  assert.equal(snapshotDelta >= 1 && snapshotDelta <= 2, true, `expected <=2 coalesced snapshots, got ${snapshotDelta}`);
  console.log("ok: cold miss streams immediately, singleflight survives abort, partial bodies stay private, and index snapshots coalesce");
} finally {
  await stop(gateway);
  if (upstream?.listening) {
    upstream.close();
    await once(upstream, "close");
  }
  await rm(root, { recursive: true, force: true });
}

function gatewayUrl(port, source) {
  return `http://127.0.0.1:${port}/api/map/cache?url=${encodeURIComponent(source)}`;
}

async function readRemaining(reader) {
  const chunks = [];
  while (true) {
    const next = await reader.read();
    if (next.done) break;
    chunks.push(Buffer.from(next.value));
  }
  return Buffer.concat(chunks).toString("utf8");
}

function abortAfterFirstChunk(url) {
  return new Promise((resolve, reject) => {
    let resolved = false;
    const request = httpGet(url, (response) => {
      response.once("data", (chunk) => {
        resolved = true;
        resolve(String(chunk));
        response.destroy();
      });
      response.once("error", (error) => { if (!resolved) reject(error); });
    });
    request.once("error", (error) => { if (!resolved) reject(error); });
  });
}

async function indexEntryCount(dir) {
  try {
    const index = JSON.parse(await readFile(join(dir, "index.json"), "utf8"));
    return Object.keys(index.entries || {}).length;
  } catch (error) {
    if (error?.code === "ENOENT") return 0;
    throw error;
  }
}

async function tempFiles(dir) {
  const found = [];
  async function walk(current) {
    let entries;
    try { entries = await readdir(current, { withFileTypes: true }); }
    catch (error) { if (error?.code === "ENOENT") return; throw error; }
    for (const entry of entries) {
      const path = join(current, entry.name);
      if (entry.isDirectory()) await walk(path);
      else if (entry.name.includes(".tmp")) found.push(path);
    }
  }
  await walk(dir);
  return found;
}

async function waitFor(predicate) {
  for (let attempt = 0; attempt < 200; attempt += 1) {
    if (await predicate()) return;
    await new Promise((resolve) => setTimeout(resolve, 20));
  }
  throw new Error("condition_timeout");
}

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 waitForGateway(port, child) {
  let output = "";
  child.stderr.on("data", (chunk) => { output += String(chunk); });
  for (let attempt = 0; attempt < 80; attempt += 1) {
    if (child.exitCode !== null) throw new Error(`gateway_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((resolve) => setTimeout(resolve, 50));
  }
  throw new Error(`gateway_start_timeout:${output}`);
}

async function stop(child) {
  if (child && !child.killed) {
    child.kill("SIGTERM");
    await once(child, "exit").catch(() => undefined);
  }
}