import { fileURLToPath } from "node:url";
import { describe, expect, it } from "vitest";
import { WorkerConnection, workerLines } from "../src/worker.ts";

const fixture = fileURLToPath(new URL("./fixtures/worker.ts", import.meta.url));
const limits = { maxRequestBytes: 4096, maxResponseBytes: 4096 };
function connection() {
  return new WorkerConnection({ cmd: process.execPath, args: [fixture] });
}
const request = (task: string) => ({ id: "1", task, input: "中文" });

describe("worker 管道与回收", () => {
  it("拆开的 UTF-8 和多行消息仍能正确分帧", async () => {
    const bytes = new TextEncoder().encode("中文\n第二行\n");
    const stream = new ReadableStream<Uint8Array>({
      start(c) {
        for (const b of bytes) c.enqueue(Uint8Array.of(b));
        c.close();
      },
    });
    const lines = [];
    for await (const line of workerLines(stream.getReader(), 64)) lines.push(line);
    expect(lines).toEqual(["中文", "第二行"]);
  });

  it("转发进度、返回结果，并持续排空大量 stderr", async () => {
    const seen: unknown[] = [];
    expect(await connection().request(request("normal"), { ...limits, onProgress: (p) => seen.push(p) })).toEqual({
      id: "1",
      type: "result",
      result: "中文",
    });
    expect(seen).toEqual([{ bytes: 2 }]);
    expect(await connection().request(request("stderr"), limits)).toMatchObject({ result: true });
  });

  it("大结果跨多个管道分块也完整返回", async () => {
    const reply = await connection().request(request("large"), { ...limits, maxResponseBytes: 2 * 1024 * 1024 });
    expect(reply).toEqual({ id: "1", type: "result", result: "中文".repeat(200_000) });
  });

  it.each([
    ["wrong-id", "protocol"],
    ["truncated", "protocol"],
    ["oversized", "size"],
    ["after-result", "protocol"],
    ["failed-exit", "exit"],
  ])("拒绝 %s，而不接受部分或错误响应", async (task, kind) => {
    await expect(connection().request(request(task), limits)).rejects.toMatchObject({ kind });
  });

  it("请求大小超限也回收已启动的进程", async () => {
    const c = connection();
    await expect(c.request({ ...request("hang"), input: "x".repeat(8192) }, limits)).rejects.toMatchObject({
      kind: "size",
    });
  });

  it("启动失败保留 spawn 分类", async () => {
    const c = new WorkerConnection({ cmd: "/nonexistent/umt-worker", args: [] });
    await expect(c.request(request("normal"), limits)).rejects.toMatchObject({ kind: "spawn" });
  });

  it("执行中的请求可终止，返回调用方的原因", async () => {
    const c = connection();
    const pending = c.request(request("hang"), limits);
    const reason = new Error("调用方取消");
    c.terminate(reason, 50);
    await expect(pending).rejects.toBe(reason);
  });

  it("进度消费者抛错时回收连接，不让任务继续等待", async () => {
    await expect(
      connection().request(request("normal"), {
        ...limits,
        onProgress: () => {
          throw new Error("进度无效");
        },
      }),
    ).rejects.toMatchObject({ kind: "protocol", message: expect.stringContaining("进度无效") });
  });
});