import test from "node:test";
import assert from "node:assert/strict";
import { PassThrough } from "node:stream";
import { mkdtemp, writeFile, rm } from "node:fs/promises";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { fileURLToPath } from "node:url";
import { serveRestrictedRegistry } from "./service.mjs";

const workerd = process.env.QUIXOS_TEST_WORKERD;
if (!workerd) throw Error("Set QUIXOS_TEST_WORKERD to the pinned engine");
const worker = fileURLToPath(new URL("./worker.mjs", import.meta.url));
const abi = "quixos-transaction-ports-v1";
let next = 0;
async function fixture(source, options = {}) {
  const directory = await mkdtemp(join(tmpdir(), "qx-restricted-test-"));
  const module = join(directory, "module.mjs");
  await writeFile(module, source);
  const io = { stdin: new PassThrough(), stdout: new PassThrough(), exitCode: 0 };
  const waiting = [],
    queued = [];
  let buffer = "";
  io.stdout.on("data", (chunk) => {
    buffer += chunk;
    let index;
    while ((index = buffer.indexOf("\n")) >= 0) {
      const value = JSON.parse(buffer.slice(0, index));
      buffer = buffer.slice(index + 1);
      if (waiting.length) waiting.shift()(value);
      else queued.push(value);
    }
  });
  const receive = async () => {
    if (queued.length) return queued.shift();
    let timer;
    try {
      return await Promise.race([
        new Promise((resolve) => waiting.push(resolve)),
        new Promise((_, reject) => {
          timer = setTimeout(() => reject(Error("TEST_MESSAGE_TIMEOUT")), 5000);
        }),
      ]);
    } finally {
      clearTimeout(timer);
    }
  };
  const service = await serveRestrictedRegistry(
    { workerd, worker, module, registryExport: "registry", exports: ["test"], sdk: [], ...options },
    io,
  );
  assert.deepEqual(await receive(), { abi, kind: "ready", exports: ["test"] });
  const send = (value) => io.stdin.write(JSON.stringify({ abi, ...value }) + "\n");
  const start = (input = null) => {
    const invocation = "invoke-" + ++next,
      frame = "frame-" + next;
    send({
      kind: "invoke",
      invocation,
      frame,
      exportId: "test",
      workspace: "workspace:restricted",
      receiver: null,
      input,
      effect: "snapshot",
    });
    return { invocation, frame };
  };
  return {
    io,
    receive,
    send,
    start,
    async close() {
      await service.close();
      await rm(directory, { recursive: true, force: true });
    },
  };
}
const registry = (code) => `export const registry={exports:['test'],async invoke(request,channel){${code}}};`;

test("real engine returns results and resets all globals between frames", async () => {
  const f = await fixture(
    registry(
      "globalThis.retained=(globalThis.retained??0)+1; return {count:globalThis.retained,text:new Intl.Segmenter('zh',{granularity:'word'}).segment('雨后的书店').containing(3).segment};",
    ),
  );
  try {
    for (let n = 0; n < 3; n++) {
      const call = f.start();
      const result = await f.receive();
      assert.equal(result.invocation, call.invocation);
      assert.equal(result.status, "returned");
      assert.equal(result.output.count, 1);
      assert.equal(typeof result.output.text, "string");
    }
  } finally {
    await f.close();
  }
});
test("captured frame is the sole source of nested port identities", async () => {
  const f = await fixture(
    registry("return await channel({portId:'read',action:{kind:'state',primitive:'read'},input:null});"),
  );
  try {
    const call = f.start();
    const port = await f.receive();
    assert.equal(port.kind, "port");
    assert.equal(port.frame, call.frame);
    assert.equal(port.invocation, call.invocation);
    f.send({ kind: "port-result", ...call, sequence: port.sequence, status: "returned", output: 42 });
    const result = await f.receive();
    assert.equal(result.status, "returned");
    assert.equal(result.output, 42);
  } finally {
    await f.close();
  }
});
test("authored serialization cannot forge result routing or ABI fields", async () => {
  const f = await fixture(
    registry(
      "Object.prototype.toJSON=function(){return this.status==='returned'?{status:'returned',output:42,invocation:'forged',abi:'forged'}:this;};return 42;",
    ),
  );
  try {
    const call = f.start();
    const result = await f.receive();
    assert.equal(result.invocation, call.invocation);
    assert.equal(result.abi, abi);
    assert.equal(result.status, "failed");
    assert.equal(result.error.code, "RESTRICTED_PACKAGE_FAILED");
  } finally {
    await f.close();
  }
});
test("the checked portable SDK keeps codecs and nested graph ports inside the frame", async () => {
  const sdk = process.env.QUIXOS_TEST_SDK;
  assert.ok(sdk, "Set QUIXOS_TEST_SDK to the checked portable SDK");
  const C = await import(sdk),
    workspace = C.workspaceId("workspace:restricted");
  const integer = { kind: "scalar", name: "int64" },
    unit = { kind: "builtin", name: "unit" };
  const declarations = [
    {
      id: "test",
      name: "test",
      effect: "snapshot",
      receiver: null,
      input: integer,
      output: integer,
      methods: [
        {
          name: "get",
          portName: "count",
          portId: "count-read",
          action: { kind: "state", primitive: "read" },
          input: unit,
          output: integer,
          keyed: false,
        },
      ],
    },
  ];
  const source = `import {definePortableRegistry} from 'qx:core'; export const registry=definePortableRegistry(${JSON.stringify(declarations)},{test:async ctx=>ctx.input+await ctx.ports.count.get()});`;
  const f = await fixture(source, { sdk: [{ name: "qx:core", path: sdk }] });
  try {
    const call = f.start(C.encodeArgument(integer, 2n, { workspace }));
    const port = await f.receive();
    assert.equal(port.kind, "port");
    assert.equal(C.decodeArgument(unit, port.input, { workspace }), null);
    f.send({
      kind: "port-result",
      ...call,
      sequence: port.sequence,
      status: "returned",
      output: C.encodeArgument(integer, 4n, { workspace }),
    });
    const result = await f.receive();
    assert.equal(result.status, "returned");
    assert.equal(C.decodeArgument(integer, result.output, { workspace }), 6n);
  } finally {
    await f.close();
  }
});
test("ambient clock, entropy, network and native imports are denied", async () => {
  for (const code of [
    "Date.now()",
    "new Date()",
    "new Intl.DateTimeFormat().format()",
    "new Intl.DateTimeFormat().formatToParts()",
    "Math.random()",
    "crypto.randomUUID()",
    "fetch('https://example.com')",
    "(await import('node:fs')).readFileSync('/etc/passwd')",
    "(await import('./unlisted.mjs')).secret",
  ]) {
    const f = await fixture(registry(`return ${code};`));
    try {
      f.start();
      assert.equal((await f.receive()).status, "failed", code);
    } finally {
      await f.close();
    }
  }
});
test("explicit dates remain useful without an implicit clock", async () => {
  const f = await fixture(registry("return new Date('2026-01-01T00:00:00Z').toISOString();"));
  try {
    f.start();
    assert.equal((await f.receive()).output, "2026-01-01T00:00:00.000Z");
  } finally {
    await f.close();
  }
});
test("initialization and regex loops are killed, then a fresh frame can start", async () => {
  for (const source of [
    "while(true){};" + registry("return 1;"),
    registry("return /(a+)+$/.test('a'.repeat(100000)+'!');"),
  ]) {
    const f = await fixture(source, { deadlineMs: 350 });
    try {
      for (let n = 0; n < 2; n++) {
        f.start();
        const result = await f.receive();
        assert.equal(result.error.code, "RESTRICTED_DEADLINE");
      }
    } finally {
      await f.close();
    }
  }
});
test("cancellation kills the entire frame before acknowledgment and rejects late port replies", async () => {
  const f = await fixture(
    registry("return await channel({portId:'read',action:{kind:'state',primitive:'read'},input:null});"),
  );
  try {
    const call = f.start();
    const port = await f.receive();
    f.send({ kind: "cancel", ...call });
    const cancelled = await f.receive();
    assert.equal(cancelled.kind, "cancelled");
    f.send({ kind: "port-result", ...call, sequence: port.sequence, status: "returned", output: "late" });
    const next = f.start();
    const nextPort = await f.receive();
    assert.equal(nextPort.invocation, next.invocation);
    f.send({ kind: "port-result", ...next, sequence: nextPort.sequence, status: "returned", output: "fresh" });
    const result = await f.receive();
    assert.equal(result.output, "fresh");
    assert.equal(result.invocation, next.invocation);
  } finally {
    await f.close();
  }
});
test("cancellation during asynchronous preparation leaves no running frame", async () => {
  const f = await fixture(registry("return 1;"));
  try {
    for (let n = 0; n < 5; n++) {
      const call = f.start();
      f.send({ kind: "cancel", ...call });
      assert.equal((await f.receive()).kind, "cancelled");
    }
    f.start();
    assert.equal((await f.receive()).output, 1);
  } finally {
    await f.close();
  }
});

test("bounded FIFO rejects overload without losing active or queued frames", async () => {
  const f = await fixture(
    registry("return await channel({portId:'read',action:{kind:'state',primitive:'read'},input:request.input});"),
    { maxQueued: 2 },
  );
  try {
    const first = f.start(1),
      port = await f.receive();
    const second = f.start(2),
      third = f.start(3),
      overflow = f.start(4);
    const refused = await f.receive();
    assert.equal(refused.invocation, overflow.invocation);
    assert.equal(refused.error.code, "RESTRICTED_QUEUE_FULL");
    f.send({ kind: "cancel", ...second });
    assert.equal((await f.receive()).invocation, second.invocation);
    f.send({ kind: "port-result", ...first, sequence: port.sequence, status: "returned", output: 1 });
    assert.equal((await f.receive()).output, 1);
    const nextPort = await f.receive();
    assert.equal(nextPort.invocation, third.invocation);
    assert.equal(nextPort.input, 3);
    f.send({ kind: "port-result", ...third, sequence: nextPort.sequence, status: "returned", output: 3 });
    assert.equal((await f.receive()).output, 3);
    f.start(5);
    const fresh = await f.receive();
    f.send({
      kind: "port-result",
      invocation: fresh.invocation,
      frame: fresh.frame,
      sequence: fresh.sequence,
      status: "returned",
      output: 5,
    });
    assert.equal((await f.receive()).output, 5);
  } finally {
    await f.close();
  }
});

test("queue wait consumes the deadline and cancelled work never reaches its ports", async () => {
  const f = await fixture(
    registry("return await channel({portId:'read',action:{kind:'state',primitive:'read'},input:null});"),
    { deadlineMs: 350 },
  );
  try {
    const first = f.start();
    await f.receive();
    const queued = f.start();
    const replies = [await f.receive(), await f.receive()];
    assert.deepEqual(new Set(replies.map((x) => x.invocation)), new Set([first.invocation, queued.invocation]));
    assert.ok(replies.every((x) => x.error.code === "RESTRICTED_DEADLINE"));
    const fresh = f.start();
    const port = await f.receive();
    assert.equal(port.invocation, fresh.invocation);
    f.send({ kind: "port-result", ...fresh, sequence: port.sequence, status: "returned", output: "fresh" });
    assert.equal((await f.receive()).output, "fresh");
  } finally {
    await f.close();
  }
});