import { describe, expect, test } from "vitest";

import {
  Communicator,
  NotConnectedError,
  type Transport,
  parserFromSchema,
  type MessageParser,
} from "../src/communicator.js";
import {
  type DecodedEnvelope,
  decodePacketBaseEnvelope,
  FrameReorderBuffer,
  WorkerGateway,
} from "../src/inbound_gateway.js";
import { createNodeWorkerGateway } from "../src/node_inbound_gateway.js";
import {
  create,
  fromBinary,
  PacketBaseSchema,
  TestDataSchema,
  toBinary,
  type PacketBase,
} from "../src/packets/message_common_pb.js";

const MAX_NONCE = 2_147_483_647;

class MockTransport implements Transport {
  packets: PacketBase[] = [];

  async writePacket(base: PacketBase): Promise<void> {
    this.packets.push(base);
  }

  async close(): Promise<void> {}
}

class BlockingTransport implements Transport {
  packets: PacketBase[] = [];
  private writeResolvers: Array<() => void> = [];

  async writePacket(base: PacketBase): Promise<void> {
    this.packets.push(base);
    await new Promise<void>((resolve) => {
      this.writeResolvers.push(resolve);
    });
  }

  releaseNext(): void {
    this.writeResolvers.shift()?.();
  }

  releaseAll(): void {
    while (this.writeResolvers.length > 0) {
      this.releaseNext();
    }
  }

  async close(): Promise<void> {}
}

function parserMap() {
  return new Map([[TestDataSchema.typeName, parserFromSchema(TestDataSchema)]]);
}

function packet(nonce: number): PacketBase {
  return create(PacketBaseSchema, {
    typeName: TestDataSchema.typeName,
    nonce,
    data: new Uint8Array(),
  });
}

async function waitFor(predicate: () => boolean): Promise<void> {
  for (let i = 0; i < 20; i += 1) {
    if (predicate()) {
      return;
    }
    await Promise.resolve();
  }
  throw new Error("condition not met");
}

async function waitUntil(predicate: () => boolean, timeoutMs = 2000): Promise<void> {
  const deadline = Date.now() + timeoutMs;
  while (Date.now() < deadline) {
    if (predicate()) {
      return;
    }
    await new Promise<void>((r) => setTimeout(r, 5));
  }
  throw new Error("condition not met");
}

function rawFrame(index: number, nonce: number, responseNonce = 0): Uint8Array {
  const data = toBinary(TestDataSchema, create(TestDataSchema, { index }));
  return toBinary(
    PacketBaseSchema,
    create(PacketBaseSchema, {
      typeName: TestDataSchema.typeName,
      nonce,
      responseNonce,
      data,
    }),
  );
}

describe("Communicator", () => {
  test("addListener/addRequestListener mutual exclusion", () => {
    const comm = new Communicator();
    comm.initialize(new MockTransport(), parserMap());

    comm.addListener(TestDataSchema.typeName, () => {});
    expect(() =>
      comm.addRequestListener(TestDataSchema.typeName, () =>
        create(TestDataSchema, { index: 1, message: "x" }),
      ),
    ).toThrow(/already registered with addListener/);

    const other = new Communicator();
    other.initialize(new MockTransport(), parserMap());
    other.addRequestListener(TestDataSchema.typeName, () =>
      create(TestDataSchema, { index: 1, message: "x" }),
    );
    expect(() => other.addListener(TestDataSchema.typeName, () => {})).toThrow(
      /already registered with addRequestListener/,
    );
  });

  test("send increments nonce", async () => {
    const transport = new MockTransport();
    const comm = new Communicator();
    comm.initialize(transport, parserMap());

    await comm.send(create(TestDataSchema, { index: 1, message: "first" }));
    await comm.send(create(TestDataSchema, { index: 2, message: "second" }));

    expect(transport.packets).toHaveLength(2);
    expect(transport.packets[0]?.nonce).toBe(1);
    expect(transport.packets[1]?.nonce).toBe(2);
  });

  test("nonce wraps after int32 max without emitting zero", async () => {
    const transport = new MockTransport();
    const comm = new Communicator();
    comm.initialize(transport, parserMap());
    (comm as unknown as { nonce: number }).nonce = MAX_NONCE - 1;

    await comm.send(create(TestDataSchema, { index: 1, message: "first" }));
    await comm.send(create(TestDataSchema, { index: 2, message: "second" }));

    expect(transport.packets.map((sentPacket) => sentPacket.nonce)).toEqual([MAX_NONCE, 1]);
    expect(transport.packets.every((sentPacket) => sentPacket.nonce !== 0)).toBe(true);
  });

  test("sendRequest resolves on response", async () => {
    const transport = new MockTransport();
    const comm = new Communicator();
    comm.initialize(transport, parserMap());

    const pending = comm.sendRequest(
      create(TestDataSchema, { index: 21, message: "hello" }),
      TestDataSchema,
      1000,
    );

    const request = transport.packets[0];
    expect(request?.nonce).toBe(1);

    comm.onReceivedData(
      TestDataSchema.typeName,
      toBinary(TestDataSchema, create(TestDataSchema, { index: 42, message: "echo: hello" })),
      0,
      request?.nonce ?? 0,
    );

    await expect(pending).resolves.toMatchObject({
      index: 42,
      message: "echo: hello",
    });
  });

  test("sendRequest nonce wraps at int32 max", async () => {
    const transport = new MockTransport();
    const comm = new Communicator();
    comm.initialize(transport, parserMap());
    (comm as unknown as { nonce: number }).nonce = MAX_NONCE - 1;

    const pending = comm.sendRequest(
      create(TestDataSchema, { index: 1, message: "max" }),
      TestDataSchema,
      1000,
    );

    const request = transport.packets[0];
    expect(request?.nonce).toBe(MAX_NONCE);
    expect(request?.nonce).not.toBe(0);

    comm.onReceivedData(
      TestDataSchema.typeName,
      toBinary(TestDataSchema, create(TestDataSchema, { index: 2, message: "max response" })),
      0,
      request?.nonce ?? 0,
    );

    await expect(pending).resolves.toMatchObject({
      index: 2,
      message: "max response",
    });
  });

  test("nextNonce skips active pending request nonce after wrap", async () => {
    const transport = new MockTransport();
    const comm = new Communicator();
    comm.initialize(transport, parserMap());
    (comm as unknown as { nonce: number }).nonce = MAX_NONCE;

    const pending = comm.sendRequest(
      create(TestDataSchema, { index: 1, message: "pending-one" }),
      TestDataSchema,
      10000,
    );

    const firstRequest = transport.packets[0];
    expect(firstRequest?.nonce).toBe(1);

    await comm.send(create(TestDataSchema, { index: 2, message: "skip-one" }));
    expect(transport.packets[1]?.nonce).toBe(2);

    comm.onReceivedData(
      TestDataSchema.typeName,
      toBinary(TestDataSchema, create(TestDataSchema, { index: 3, message: "pending-one response" })),
      0,
      1,
    );

    await expect(pending).resolves.toMatchObject({
      index: 3,
      message: "pending-one response",
    });
  });

  test("sendRequest rejects on timeout", async () => {
    const comm = new Communicator();
    comm.initialize(new MockTransport(), parserMap());

    await expect(
      comm.sendRequest(create(TestDataSchema, { index: 1, message: "wait" }), TestDataSchema, 10),
    ).rejects.toThrow(/request timeout for nonce 1/);
  });

  test("sendRequest는 write가 blocked인 동안 response timeout을 시작하지 않는다", async () => {
    const transport = new BlockingTransport();
    const comm = new Communicator();
    comm.initialize(transport, parserMap());

    let settled = false;
    const pending = comm
      .sendRequest(create(TestDataSchema, { index: 1, message: "blocked" }), TestDataSchema, 20)
      .then(
        () => {
          settled = true;
        },
        () => {
          settled = true;
        },
      );

    // write가 transport에 도달했지만 아직 완료되지 않았다(blocked). request timeout(20ms)이
    // 두 번 지나도 timer가 시작되지 않았으므로 reject되지 않아야 한다.
    await waitFor(() => transport.packets.length === 1);
    await new Promise<void>((r) => setTimeout(r, 60));
    expect(settled).toBe(false);

    // write를 완료시키면 이제 response timer가 시작되고 응답이 없으므로 timeout으로 reject된다.
    transport.releaseAll();
    await expect(pending).resolves.toBeUndefined();
    expect(settled).toBe(true);
  });

  test("sendRequest는 write 완료 후 response timeout으로 reject된다", async () => {
    const transport = new BlockingTransport();
    const comm = new Communicator();
    comm.initialize(transport, parserMap());

    const pending = comm.sendRequest(
      create(TestDataSchema, { index: 1, message: "wait" }),
      TestDataSchema,
      10,
    );
    await waitFor(() => transport.packets.length === 1);
    transport.releaseAll();

    await expect(pending).rejects.toThrow(/request timeout for nonce 1/);
  });

  test("shutdown cancels pending requests", async () => {
    const comm = new Communicator();
    comm.initialize(new MockTransport(), parserMap());

    const pending = comm.sendRequest(
      create(TestDataSchema, { index: 1, message: "wait" }),
      TestDataSchema,
      1000,
    );
    comm.shutdown();

    await expect(pending).rejects.toBeInstanceOf(NotConnectedError);
  });

  test("queuePacket preserves order for concurrent sends", async () => {
    const transport = new MockTransport();
    const comm = new Communicator();
    comm.initialize(transport, parserMap());

    await Promise.all([
      comm.send(create(TestDataSchema, { index: 1, message: "one" })),
      comm.send(create(TestDataSchema, { index: 2, message: "two" })),
      comm.send(create(TestDataSchema, { index: 3, message: "three" })),
    ]);

    expect(transport.packets.map((packet) => packet.nonce)).toEqual([1, 2, 3]);
    expect(transport.packets.map((packet) => packet.typeName)).toEqual([
      TestDataSchema.typeName,
      TestDataSchema.typeName,
      TestDataSchema.typeName,
    ]);
  });

  test("write queue가 MAX_WRITE_QUEUE_SIZE 초과 시 대기 후 순서대로 처리된다", async () => {
    const transport = new BlockingTransport();
    const comm = new Communicator();
    comm.initialize(transport, parserMap());
    const state = comm as unknown as {
      writeQueue: unknown[];
      writeQueueWaiters: unknown[];
    };
    const promises: Array<Promise<unknown>> = [];

    for (let nonce = 1; nonce <= 65; nonce += 1) {
      promises.push(
        comm.queuePacket(packet(nonce)).catch((err: unknown) => err),
      );
    }

    expect(transport.packets.map((sentPacket) => sentPacket.nonce)).toEqual([
      1,
    ]);
    expect(state.writeQueue).toHaveLength(64);
    expect(state.writeQueueWaiters).toHaveLength(0);

    promises.push(comm.queuePacket(packet(66)).catch((err: unknown) => err));
    await waitFor(() => state.writeQueueWaiters.length === 1);

    expect(transport.packets.map((sentPacket) => sentPacket.nonce)).toEqual([
      1,
    ]);
    expect(state.writeQueue).toHaveLength(64);

    transport.releaseNext();
    await waitFor(
      () =>
        transport.packets.length === 2 &&
        state.writeQueue.length === 64 &&
        state.writeQueueWaiters.length === 0,
    );

    expect(transport.packets.map((sentPacket) => sentPacket.nonce)).toEqual([
      1, 2,
    ]);

    comm.shutdown();
    transport.releaseAll();
    await Promise.allSettled(promises);
  });

  test("sendRequest stores protobuf packet payload", async () => {
    const transport = new MockTransport();
    const comm = new Communicator();
    comm.initialize(transport, parserMap());

    const promise = comm.sendRequest(
      create(TestDataSchema, { index: 7, message: "payload" }),
      TestDataSchema,
      1000,
    );

    const packet = transport.packets[0];
    expect(packet).toMatchObject({
      typeName: TestDataSchema.typeName,
      nonce: 1,
      responseNonce: 0,
    });
    const request = packet
      ? toBinary(PacketBaseSchema, packet)
      : new Uint8Array();
    expect(request.byteLength).toBeGreaterThan(0);

    comm.shutdown();
    await expect(promise).rejects.toBeInstanceOf(NotConnectedError);
  });

  test("inbound queue FIFO: onReceivedData 순서대로 listener가 호출된다", async () => {
    const transport = new MockTransport();
    const comm = new Communicator();
    comm.initialize(transport, parserMap());

    const received: number[] = [];
    comm.addListener(TestDataSchema.typeName, (msg) => {
      received.push((msg as unknown as { index: number }).index);
    });

    for (let i = 1; i <= 5; i++) {
      const data = toBinary(TestDataSchema, create(TestDataSchema, { index: i }));
      comm.onReceivedData(TestDataSchema.typeName, data, i, 0);
    }

    // inbound queue worker가 처리하도록 마이크로태스크 양보
    for (let i = 0; i < 20; i++) {
      if (received.length === 5) break;
      await Promise.resolve();
    }

    expect(received).toEqual([1, 2, 3, 4, 5]);
    await comm.close();
  });

  test("slow request handler 자동 응답이 FIFO를 보존한다", async () => {
    const transport = new MockTransport();
    const comm = new Communicator();
    comm.initialize(transport, parserMap());

    comm.addRequestListener(TestDataSchema.typeName, async (msg, nonce) => {
      const m = msg as unknown as { index: number };
      if (m.index === 1) {
        // 첫 번째 요청 50ms 지연
        await new Promise<void>((r) => setTimeout(r, 50));
      }
      return create(TestDataSchema, { index: m.index + 100, message: "echo" });
    });

    const data1 = toBinary(TestDataSchema, create(TestDataSchema, { index: 1 }));
    const data2 = toBinary(TestDataSchema, create(TestDataSchema, { index: 2 }));
    comm.onReceivedData(TestDataSchema.typeName, data1, 10, 0);
    comm.onReceivedData(TestDataSchema.typeName, data2, 20, 0);

    // 두 응답이 전송될 때까지 대기 (최대 2초)
    for (let i = 0; i < 400; i++) {
      if (transport.packets.length >= 2) break;
      await new Promise<void>((r) => setTimeout(r, 5));
    }

    expect(transport.packets.length).toBeGreaterThanOrEqual(2);
    // FIFO: 첫 번째 응답이 responseNonce=10, 두 번째가 responseNonce=20
    expect(transport.packets[0]?.responseNonce).toBe(10);
    expect(transport.packets[1]?.responseNonce).toBe(20);
    await comm.close();
  });

  test("close 시 pending request가 NotConnectedError로 거부된다", async () => {
    const comm = new Communicator();
    comm.initialize(new MockTransport(), parserMap());

    const pending = comm.sendRequest(
      create(TestDataSchema, { index: 1, message: "wait" }),
      TestDataSchema,
      5000,
    );

    comm.shutdown();

    await expect(pending).rejects.toBeInstanceOf(NotConnectedError);
    expect(comm.isAlive()).toBe(false);
  });

  test("inbound queue full 시 backpressure가 가동한다", async () => {
    const transport = new MockTransport();
    const comm = new Communicator();
    comm.initialize(transport, parserMap());

    const data = toBinary(TestDataSchema, create(TestDataSchema, { index: 1, message: "heavy" }));

    let resolveHandler: (() => void) | null = null;
    const handlerPromise = new Promise<void>((r) => {
      resolveHandler = r;
    });

    // 지연 request 핸들러 등록
    comm.addRequestListener(TestDataSchema.typeName, async () => {
      await handlerPromise;
      return create(TestDataSchema, { index: 100, message: "done" });
    });

    // 1번째 호출: 핸들러가 handlerPromise에 의해 블로킹됨
    await comm.enqueueInbound(TestDataSchema.typeName, data, 1, 0);

    // 2번째부터 65번째(총 64개)를 enqueue: 첫번째 아이템은 디스패치 중이므로 큐에는 현재 0개 있음
    // 2 ~ 65번째(총 64개) 추가 시 큐가 꽉 차게 됨
    for (let i = 2; i <= 65; i++) {
      await comm.enqueueInbound(TestDataSchema.typeName, data, i, 0);
    }

    // 66번째 enqueue 시도: 큐가 가득 찼으므로( capacity 64 ) 대기(블로킹)해야 함
    let enqueueDone = false;
    const enqueuePromise = comm.enqueueInbound(TestDataSchema.typeName, data, 66, 0).then(() => {
      enqueueDone = true;
    });

    await new Promise<void>((r) => setTimeout(r, 20));
    expect(enqueueDone).toBe(false);

    // 핸들러를 완료시켜 큐를 하나 소모하게 함
    resolveHandler!();

    await enqueuePromise;
    expect(enqueueDone).toBe(true);

    comm.shutdown();
  });

  test("graceful close 시 in-flight request handler가 무사히 완주하여 자동 응답을 송신한다", async () => {
    const transport = new MockTransport();
    const comm = new Communicator();
    comm.initialize(transport, parserMap());

    let handlerStarted = false;
    let handlerFinished = false;

    comm.addRequestListener(TestDataSchema.typeName, async (msg, nonce) => {
      handlerStarted = true;
      await new Promise<void>((r) => setTimeout(r, 50));
      handlerFinished = true;
      return create(TestDataSchema, { index: 100, message: "graceful" });
    });

    const data = toBinary(TestDataSchema, create(TestDataSchema, { index: 1 }));
    comm.onReceivedData(TestDataSchema.typeName, data, 1, 0);

    // 핸들러가 구동될 때까지 대기
    for (let i = 0; i < 20; i++) {
      if (handlerStarted) break;
      await new Promise<void>((r) => setTimeout(r, 2));
    }
    expect(handlerStarted).toBe(true);

    // 이 상태에서 close() 호출
    const closePromise = comm.close();

    // close 완료 전에 핸들러가 완주했는지 확인
    await closePromise;
    expect(handlerFinished).toBe(true);

    // 결과 검증: close 완료 후에도 핸들러 자동 응답이 정상적으로 전송 큐에 잘 적재되었는지 확인
    expect(transport.packets.length).toBe(1);
    expect(transport.packets[0]?.responseNonce).toBe(1);
    expect(comm.isAlive()).toBe(false);
  });

  test("worker gateway: out-of-order seq를 reorder한 뒤 입력 순서로 dispatch한다", async () => {
    const transport = new MockTransport();
    const comm = new Communicator();
    comm.initialize(transport, parserMap());

    const received: number[] = [];
    comm.addListener(TestDataSchema.typeName, (msg) => {
      received.push((msg as unknown as { index: number }).index);
    });

    // Fake gateway: out-of-order(seq=2, seq=1)로 전달된 아이템
    const items = [
      { seq: 2, index: 2, nonce: 2 },
      { seq: 1, index: 1, nonce: 1 },
    ];

    // coordinator로 밀어넣기 전에 seq 오름차순으로 reorder
    items.sort((a, b) => a.seq - b.seq);

    for (const item of items) {
      const data = toBinary(TestDataSchema, create(TestDataSchema, { index: item.index }));
      comm.onReceivedData(TestDataSchema.typeName, data, item.nonce, 0);
    }

    // inbound queue worker가 처리하도록 마이크로태스크 양보
    for (let i = 0; i < 20; i++) {
      if (received.length === 2) break;
      await Promise.resolve();
    }

    expect(received).toEqual([1, 2]);
    await comm.close();
  });
});

describe("FrameReorderBuffer", () => {
  test("out-of-order 결과를 contiguous prefix가 채워질 때만 seq 순서로 release한다", () => {
    const buffer = new FrameReorderBuffer();

    expect(buffer.release({ seq: 3, frame: null, error: null })).toHaveLength(0);
    expect(buffer.release({ seq: 2, frame: null, error: null })).toHaveLength(0);

    const flushed = buffer.release({ seq: 1, frame: null, error: null });
    expect(flushed.map((r) => r.seq)).toEqual([1, 2, 3]);

    // nextSeq 아래의 seq는 중복/지연 도착이므로 무시한다.
    expect(buffer.release({ seq: 1, frame: null, error: null })).toHaveLength(0);
  });
});

describe("Communicator inbound gateway", () => {
  test("gateway on: worker가 out-of-order로 완료해도 입력 seq 순서로 dispatch한다", async () => {
    const comm = new Communicator();
    comm.initialize(new MockTransport(), parserMap());

    const received: number[] = [];
    comm.addListener(TestDataSchema.typeName, (msg) => {
      received.push((msg as unknown as { index: number }).index);
    });

    const total = 6;
    // 앞선 seq일수록 더 오래 decode시켜 완료 순서를 역전시킨다.
    comm.enableInboundGateway({
      workers: 4,
      decode: async (bytes) => {
        const env = decodePacketBaseEnvelope(bytes);
        await new Promise<void>((r) => setTimeout(r, (total - env.incomingNonce) * 5));
        return env;
      },
    });

    for (let i = 1; i <= total; i += 1) {
      comm.onReceivedFrame(rawFrame(i, i));
    }

    await waitUntil(() => received.length === total, 3000);
    expect(received).toEqual([1, 2, 3, 4, 5, 6]);
    await comm.close();
  });

  test("gateway 미설정 fallback: onReceivedFrame이 inline decode로 coordinator path를 유지한다", async () => {
    const comm = new Communicator();
    comm.initialize(new MockTransport(), parserMap());

    const received: number[] = [];
    comm.addListener(TestDataSchema.typeName, (msg) => {
      received.push((msg as unknown as { index: number }).index);
    });

    for (let i = 1; i <= 5; i += 1) {
      comm.onReceivedFrame(rawFrame(i, i));
    }

    for (let i = 0; i < 20; i += 1) {
      if (received.length === 5) break;
      await Promise.resolve();
    }

    expect(received).toEqual([1, 2, 3, 4, 5]);
    await comm.close();
  });

  test("gateway decode 오류는 frame error handler로 보고되고 이후 프레임을 막지 않는다", async () => {
    const comm = new Communicator();
    comm.initialize(new MockTransport(), parserMap());

    const received: number[] = [];
    comm.addListener(TestDataSchema.typeName, (msg) => {
      received.push((msg as unknown as { index: number }).index);
    });
    const errors: Error[] = [];
    comm.setFrameErrorHandler((err) => {
      errors.push(err);
    });

    comm.enableInboundGateway({ workers: 2 });

    comm.onReceivedFrame(rawFrame(1, 1));
    comm.onReceivedFrame(new Uint8Array([0xff, 0xff, 0xff])); // 디코드 불가 프레임
    comm.onReceivedFrame(rawFrame(3, 3));

    await waitUntil(() => received.length === 2, 2000);
    expect(received).toEqual([1, 3]);
    expect(errors).toHaveLength(1);
    await comm.close();
  });

  // 이 테스트는 기본 단일 경로(default path)가 아닌, 명시적으로 worker_threads 백엔드를
  // 사용하는 경우의 입력 순서 보존 계약을 검증합니다.
  test("Node worker_threads gateway (Explicit Path): off-thread decode가 입력 순서를 보존한다", async () => {
    const comm = new Communicator();
    comm.initialize(new MockTransport(), parserMap());

    const received: number[] = [];
    comm.addListener(TestDataSchema.typeName, (msg) => {
      received.push((msg as unknown as { index: number }).index);
    });

    comm.attachInboundGateway(
      createNodeWorkerGateway({
        sink: (frame) =>
          comm.enqueueInbound(
            frame.typeName,
            frame.data,
            frame.incomingNonce,
            frame.responseNonce,
          ),
        workers: 4,
      }),
    );

    const total = 8;
    for (let i = 1; i <= total; i += 1) {
      comm.onReceivedFrame(rawFrame(i, i));
    }

    await waitUntil(() => received.length === total, 8000);
    expect(received).toEqual([1, 2, 3, 4, 5, 6, 7, 8]);
    await comm.close();
  }, 10000);

  test("attachInboundGateway 경계: 닫힌 이전 gateway의 늦은 결과가 현재 communicator로 들어오지 않는다", async () => {
    const comm = new Communicator();
    comm.initialize(new MockTransport(), parserMap());

    const received: number[] = [];
    comm.addListener(TestDataSchema.typeName, (msg) => {
      received.push((msg as unknown as { index: number }).index);
    });

    let resolveDecode!: (env: DecodedEnvelope) => void;
    comm.enableInboundGateway({
      workers: 1,
      decode: () =>
        new Promise<DecodedEnvelope>((resolve) => {
          resolveDecode = resolve;
        }),
    });
    comm.onReceivedFrame(rawFrame(1, 1));
    await waitFor(() => resolveDecode !== undefined);

    // 새 gateway attach가 이전 gateway를 close한다.
    comm.attachInboundGateway(
      new WorkerGateway({
        sink: (frame) =>
          comm.enqueueInbound(
            frame.typeName,
            frame.data,
            frame.incomingNonce,
            frame.responseNonce,
          ),
      }),
    );

    // 이전 gateway의 pending decode를 close 이후 완료시킨다.
    const data = toBinary(TestDataSchema, create(TestDataSchema, { index: 1 }));
    resolveDecode({
      typeName: TestDataSchema.typeName,
      data,
      incomingNonce: 1,
      responseNonce: 0,
    });
    await new Promise<void>((r) => setTimeout(r, 10));

    expect(received).toEqual([]);
    await comm.close();
  });

  test("Alias collision during initialization triggers error", () => {
    const comm = new Communicator();
    const badParserMap = new Map<string, MessageParser>();
    badParserMap.set("a.b.TestData", parserFromSchema(TestDataSchema));
    badParserMap.set("x.y.TestData", parserFromSchema(TestDataSchema));
    
    expect(() => {
      comm.initialize(new MockTransport(), badParserMap);
    }).toThrow();
  });

  test("Legacy simple-name alias routing and pending matching", async () => {
    const comm = new Communicator();
    comm.initialize(new MockTransport(), parserMap());

    // 1. Test Listener routing with legacy simple name
    const received: number[] = [];
    comm.addListener(TestDataSchema.typeName, (msg) => {
      received.push((msg as unknown as { index: number }).index);
    });

    const testMsgBytes = toBinary(TestDataSchema, create(TestDataSchema, { index: 42 }));
    comm.onReceivedData("TestData", testMsgBytes, 1, 0);

    for (let i = 0; i < 20; i += 1) {
      if (received.length === 1) break;
      await Promise.resolve();
    }
    expect(received).toEqual([42]);

    // 2. Test Pending request matching with legacy simple name
    const transport = new MockTransport();
    const comm2 = new Communicator();
    comm2.initialize(transport, parserMap());

    const task = comm2.sendRequest(create(TestDataSchema, { index: 1 }), TestDataSchema, 1000);

    await waitUntil(() => transport.packets.length === 1, 1000);
    const request = transport.packets[0];

    comm2.onReceivedData("TestData", testMsgBytes, 0, request.nonce);

    const result = await task;
    expect((result as unknown as { index: number }).index).toBe(42);
    await comm.close();
    await comm2.close();
  });

  test("Request handler registered with full name handles simple-name inbound request", async () => {
    const transport = new MockTransport();
    const comm = new Communicator();
    comm.initialize(transport, parserMap());

    comm.addRequestListener(TestDataSchema.typeName, async (msg) => {
      const m = msg as unknown as { index: number };
      return create(TestDataSchema, { index: m.index + 1, message: "echo" });
    });

    const reqBytes = toBinary(TestDataSchema, create(TestDataSchema, { index: 7, message: "req" }));
    comm.onReceivedData("TestData", reqBytes, 42, 0);

    await waitUntil(() => transport.packets.length === 1, 1000);

    expect(transport.packets.length).toBe(1);
    expect(transport.packets[0]?.responseNonce).toBe(42);
    const decoded = fromBinary(TestDataSchema, transport.packets[0]!.data);
    expect((decoded as unknown as { index: number }).index).toBe(8);
    await comm.close();
  });
});

describe("WorkerGateway lifecycle", () => {
  test("close() 이후 pending decode result는 sink로 전달되지 않는다", async () => {
    let resolveDecode!: (env: DecodedEnvelope) => void;
    const delivered: number[] = [];
    const gateway = new WorkerGateway({
      workers: 1,
      decode: () =>
        new Promise<DecodedEnvelope>((resolve) => {
          resolveDecode = resolve;
        }),
      sink: (frame) => {
        delivered.push(frame.seq);
      },
    });

    gateway.submit({ seq: 1, bytes: new Uint8Array([1]) });
    await waitFor(() => resolveDecode !== undefined);
    gateway.close();
    resolveDecode({ typeName: "t", data: new Uint8Array(), incomingNonce: 0, responseNonce: 0 });
    await new Promise<void>((r) => setTimeout(r, 0));

    expect(delivered).toEqual([]);
  });

  test("close() 이후 pending decode reject는 onError로 전달되지 않는다", async () => {
    let rejectDecode!: (err: Error) => void;
    const errors: Error[] = [];
    const gateway = new WorkerGateway({
      workers: 1,
      decode: () =>
        new Promise<DecodedEnvelope>((_resolve, reject) => {
          rejectDecode = reject;
        }),
      sink: () => {},
      onError: (err) => {
        errors.push(err);
      },
    });

    gateway.submit({ seq: 1, bytes: new Uint8Array([1]) });
    await waitFor(() => rejectDecode !== undefined);
    gateway.close();
    rejectDecode(new Error("decode failed"));
    await new Promise<void>((r) => setTimeout(r, 0));

    expect(errors).toEqual([]);
  });
});

describe("WorkerGateway cleanup stats", () => {
  test("close()는 queued/reorder/sink backlog를 비우고 closed=true로 만든다", async () => {
    let resolveDecode!: (env: DecodedEnvelope) => void;
    const delivered: number[] = [];
    const gateway = new WorkerGateway({
      workers: 1,
      decode: () =>
        new Promise<DecodedEnvelope>((resolve) => {
          resolveDecode = resolve;
        }),
      sink: (frame) => {
        delivered.push(frame.seq);
      },
    });

    // workers=1이므로 첫 frame은 in-flight decode, 나머지 둘은 queued backlog가 된다.
    gateway.submit({ seq: 1, bytes: new Uint8Array([1]) });
    gateway.submit({ seq: 2, bytes: new Uint8Array([2]) });
    gateway.submit({ seq: 3, bytes: new Uint8Array([3]) });
    await waitFor(() => resolveDecode !== undefined);

    const before = gateway.stats();
    expect(before.active).toBe(1);
    expect(before.queued).toBe(2);
    expect(before.reorderPending).toBe(0);
    expect(before.sinkPending).toBe(0);
    expect(before.closed).toBe(false);

    gateway.close();
    const after = gateway.stats();
    expect(after.queued).toBe(0);
    expect(after.reorderPending).toBe(0);
    expect(after.sinkPending).toBe(0);
    expect(after.workerPending).toBe(0);
    expect(after.closed).toBe(true);

    // close 후 완료되는 in-flight decode는 sink로 전달되지 않고 active를 0으로 drain한다.
    resolveDecode({ typeName: "t", data: new Uint8Array(), incomingNonce: 0, responseNonce: 0 });
    await waitFor(() => gateway.stats().active === 0);
    expect(delivered).toEqual([]);
  });

  test("stats()는 out-of-order 결과의 reorder pending을 관측하고 close가 비운다", async () => {
    const resolvers = new Map<number, (env: DecodedEnvelope) => void>();
    const delivered: number[] = [];
    const gateway = new WorkerGateway({
      workers: 4,
      decode: (bytes) =>
        new Promise<DecodedEnvelope>((resolve) => {
          resolvers.set(bytes[0]!, resolve);
        }),
      sink: (frame) => {
        delivered.push(frame.seq);
      },
    });

    gateway.submit({ seq: 1, bytes: new Uint8Array([1]) });
    gateway.submit({ seq: 2, bytes: new Uint8Array([2]) });
    await waitFor(() => resolvers.size === 2);

    // seq 2를 먼저 완료시키면 reorder가 선행 seq 1을 기다리며 결과를 보류한다.
    resolvers.get(2)!({ typeName: "t", data: new Uint8Array(), incomingNonce: 0, responseNonce: 0 });
    await waitFor(() => gateway.stats().reorderPending === 1);
    expect(delivered).toEqual([]);

    gateway.close();
    const after = gateway.stats();
    expect(after.reorderPending).toBe(0);
    expect(after.queued).toBe(0);
    expect(after.closed).toBe(true);
  });

  // 이 테스트는 명시적으로 worker_threads 백엔드를 사용하는 경우의 자원 정리 및
  // workerPending 계측이 정상적으로 동작하는지 검증합니다.
  test("Node worker gateway (Explicit Path): pending decode가 있는 상태에서 close 후 workerPending=0", async () => {
    const gateway = createNodeWorkerGateway({
      sink: () => {},
      workers: 2,
    });

    const total = 6;
    for (let i = 1; i <= total; i += 1) {
      gateway.submit({ seq: i, bytes: rawFrame(i, i) });
    }

    // 동기 제출 직후에는 worker pool 응답이 아직 없어 in-flight decode와 queued backlog가 관측된다.
    const before = gateway.stats?.();
    expect(before?.workerPending).toBeGreaterThan(0);
    expect(before?.queued).toBeGreaterThan(0);
    expect(before?.closed).toBe(false);

    gateway.close();
    const after = gateway.stats?.();
    expect(after?.workerPending).toBe(0);
    expect(after?.queued).toBe(0);
    expect(after?.closed).toBe(true);
  });

  test("close()는 양수 sinkPending을 비우고 release된 sink가 stale delivery를 만들지 않는다", async () => {
    let releaseSink!: () => void;
    const sinkBlock = new Promise<void>((resolve) => {
      releaseSink = resolve;
    });
    const delivered: number[] = [];
    let sinkCalls = 0;
    const gateway = new WorkerGateway({
      workers: 2,
      // decode는 즉시 완료되어 두 결과가 reorder를 통해 sink chain에 스케줄된다.
      decode: (bytes): DecodedEnvelope => ({
        typeName: "t",
        data: new Uint8Array(),
        incomingNonce: bytes[0]!,
        responseNonce: 0,
      }),
      sink: (frame) => {
        sinkCalls += 1;
        delivered.push(frame.seq);
        // 첫 sink 호출을 unresolved promise로 막아 sinkChain을 보류시킨다 → sinkPending이 양수가 된다.
        if (sinkCalls === 1) {
          return sinkBlock;
        }
        return undefined;
      },
    });

    gateway.submit({ seq: 1, bytes: new Uint8Array([1]) });
    gateway.submit({ seq: 2, bytes: new Uint8Array([2]) });

    // 첫 sink가 block되어 후속 결과가 sink chain에 적체된다.
    await waitFor(() => gateway.stats().sinkPending > 0);
    expect(delivered).toEqual([1]);

    gateway.close();
    const after = gateway.stats();
    expect(after.sinkPending).toBe(0);
    expect(after.queued).toBe(0);
    expect(after.reorderPending).toBe(0);
    expect(after.workerPending).toBe(0);
    expect(after.closed).toBe(true);

    // blocked sink를 release해도 close 이후에는 두 번째 frame이 sink로 전달되지 않는다.
    releaseSink();
    await new Promise<void>((r) => setTimeout(r, 10));
    expect(delivered).toEqual([1]);
  });
});

