import { availableParallelism } from "node:os";
import { fileURLToPath } from "node:url";
import { Worker } from "node:worker_threads";

import {
  type DecodedEnvelope,
  type FrameDecoder,
  type InboundFrame,
  type InboundGateway,
  type InboundGatewayStats,
  WorkerGateway,
} from "./inbound_gateway.js";

interface PendingDecode {
  resolve: (env: DecodedEnvelope) => void;
  reject: (err: Error) => void;
}

interface WorkerReply {
  id: number;
  env?: DecodedEnvelope;
  error?: string;
}

/**
 * `worker_threads` 기반 PacketBase decode pool이다. N개 worker에 frame을 round-robin으로 분배해 off-thread로
 * decode하고, 결과를 Promise로 돌려준다. pool 자체는 stateful dispatch나 순서 복원을 소유하지 않는다. 순서
 * 복원과 sink 직렬화는 이 pool을 `decode` 백엔드로 사용하는 {@link WorkerGateway}가 담당한다.
 *
 * decode된 payload는 owned buffer로 transfer되어 결과 path의 clone 복사를 줄인다. 입력 raw frame은 source가
 * buffer 전체를 소유할 때만 transfer하고, pool(공유) buffer view이면 owned buffer로 복사해 detach 부작용을
 * 피한다.
 */
class WorkerThreadDecodePool {
  private readonly workers: Worker[];
  private readonly pending = new Map<number, PendingDecode>();
  private nextId = 0;
  private rr = 0;
  private closed = false;

  constructor(workerCount: number) {
    this.workers = Array.from({ length: workerCount }, () => {
      const worker = spawnDecodeWorker();
      worker.on("message", (reply: WorkerReply) => this.onReply(reply));
      worker.on("error", (err: Error) => this.failAll(err));
      worker.unref();
      return worker;
    });
  }

  decode(bytes: Uint8Array): Promise<DecodedEnvelope> {
    if (this.closed) {
      return Promise.reject(new Error("inbound gateway worker pool is closed"));
    }
    const id = this.nextId++;
    const worker = this.workers[this.rr % this.workers.length];
    this.rr += 1;
    return new Promise<DecodedEnvelope>((resolve, reject) => {
      this.pending.set(id, { resolve, reject });
      const owned = detachableBuffer(bytes);
      worker?.postMessage({ id, bytes: new Uint8Array(owned) }, [owned]);
    });
  }

  close(): void {
    if (this.closed) {
      return;
    }
    this.closed = true;
    this.failAll(new Error("inbound gateway worker pool is closed"));
    for (const worker of this.workers) {
      void worker.terminate();
    }
  }

  /** worker에 제출됐지만 아직 응답이 오지 않은 decode 수다. cleanup 계측의 workerPending으로 관측한다. */
  pendingCount(): number {
    return this.pending.size;
  }

  private onReply(reply: WorkerReply): void {
    const pending = this.pending.get(reply.id);
    if (pending === undefined) {
      return;
    }
    this.pending.delete(reply.id);
    if (reply.error !== undefined) {
      pending.reject(new Error(reply.error));
      return;
    }
    if (reply.env !== undefined) {
      pending.resolve(reply.env);
    }
  }

  private failAll(err: Error): void {
    const pending = Array.from(this.pending.values());
    this.pending.clear();
    for (const p of pending) {
      p.reject(err);
    }
  }
}

/**
 * source view가 buffer 전체를 소유하면 그 ArrayBuffer를 그대로(transfer 대상) 반환하고, pool/공유 view이면
 * owned buffer로 복사해 반환한다. 후자에서만 복사가 발생한다.
 */
function detachableBuffer(bytes: Uint8Array): ArrayBuffer {
  if (
    bytes.byteOffset === 0 &&
    bytes.byteLength === bytes.buffer.byteLength &&
    bytes.buffer instanceof ArrayBuffer
  ) {
    return bytes.buffer;
  }
  // bytes가 더 큰(공유/pool) buffer의 view이면 해당 구간만 owned buffer로 복사한다. Node Buffer의 slice()는
  // 복사가 아니라 subarray view라 .buffer가 원본 전체 buffer를 가리키므로, Uint8Array 복사로 명시적으로
  // 정확한 구간만 떼어낸다(transport read loop가 넘기는 Buffer subarray frame을 안전하게 transfer하기 위함).
  const copy = new Uint8Array(bytes.byteLength);
  copy.set(bytes);
  return copy.buffer;
}

/**
 * decode worker를 띄운다. 빌드 산출물(.js)에서는 worker 모듈을 직접 로드한다. 빌드 전(.ts) 실행 환경(vitest/
 * tsx 등)에서는 worker `execArgv`가 `--import`를 허용하지 않으므로, worker 안에서 tsx loader를 register한 뒤
 * `.ts` worker를 동적 import하는 부트스트랩을 eval로 실행한다. tsx는 devDependency이며 이 경로는 빌드 전
 * 환경에서만 사용된다.
 */
function spawnDecodeWorker(): Worker {
  const here = fileURLToPath(import.meta.url);
  if (here.endsWith(".ts")) {
    const workerUrl = new URL("./node_inbound_gateway_worker.ts", import.meta.url);
    const bootstrap =
      `import { register } from 'tsx/esm/api';\n` +
      `register();\n` +
      `await import(${JSON.stringify(workerUrl.href)});\n`;
    return new Worker(bootstrap, { eval: true });
  }
  return new Worker(new URL("./node_inbound_gateway_worker.js", import.meta.url));
}

/**
 * Node `worker_threads`로 PacketBase decode를 off-thread 처리하는 inbound gateway를 만든다.
 * 이 함수는 기본 gateway 경로가 아니며, explicit Node worker backend 성능 튜닝이나 실험용으로만 명시적으로 attach한다.
 * 결과는 내부 `seq`로 reorder된 뒤 입력 순서로 sink에 전달되므로, 수신 coordinator가 FIFO dispatch와
 * stateful 처리를 단독 소유한다는 계약을 유지한다.
 *
 * `workers <= 0`이면 사용 가능한 병렬도로 기본값을 적용한다. transferable `ArrayBuffer`로 decode 결과 payload
 * 복사를 줄인다. browser 등 `worker_threads` 미지원 환경에서는 이 모듈을 import하지 말고, inline decode를
 * 쓰는 {@link WorkerGateway}(또는 `Communicator.enableInboundGateway`)를 safe fallback으로 사용한다.
 */
export function createNodeWorkerGateway(options: {
  sink: (frame: { typeName: string; data: Uint8Array; incomingNonce: number; responseNonce: number }) => void | Promise<void>;
  onError?: (err: Error) => void;
  workers?: number;
}): InboundGateway {
  const requested = options.workers ?? 0;
  const workerCount = requested > 0 ? requested : Math.max(1, availableParallelism());
  const pool = new WorkerThreadDecodePool(workerCount);
  const decode: FrameDecoder = (bytes) => pool.decode(bytes);

  const gateway = new WorkerGateway({
    workers: workerCount,
    decode,
    sink: options.sink,
    onError: options.onError,
  });

  return {
    submit(frame: InboundFrame): void {
      gateway.submit(frame);
    },
    close(): void {
      gateway.close();
      pool.close();
    },
    stats(): InboundGatewayStats {
      // reorder/sink/queue/active backlog는 WorkerGateway가 소유하고, off-thread decode pending은 pool이
      // 소유하므로 둘을 합성한다. close 후에는 pool.failAll로 pending이 비고 workerPending=0이 된다.
      return { ...gateway.stats(), workerPending: pool.pendingCount() };
    },
  };
}
