import type { Server } from 'node:net';
import { channel } from '../channel.ts';
import { shared } from '../shared/shared.ts';
import { int32atomic } from '../shared/descriptors.ts';
import type { WorkerHandle } from '../runner.ts';
import { PushChannel } from './push-channel.ts';
import { leastConns } from './strategies.ts';
import type { ListenArgs, ListenOptions, ServerThread, ServerThreads, ServerThreadsOptions } from './types.ts';

const DEFAULT_HIGH_WATER_MARK = 64;
const DEFAULT_DRAIN_TIMEOUT = 30_000;

export function serverThreads(
  workers: readonly WorkerHandle[],
  server: Server,
  opts: ServerThreadsOptions = {},
): ServerThreads {
  const balance = opts.balance ?? leastConns();
  const hwm = opts.highWaterMark ?? DEFAULT_HIGH_WATER_MARK;
  const listenOpts: Required<ListenOptions> = {
    drainTimeout: opts.listen?.drainTimeout ?? DEFAULT_DRAIN_TIMEOUT,
  };

  const n = workers.length;
  if (n === 0) throw new Error('serverThreads: workers array must be non-empty');

  // Per-worker fd push channels (each is an AsyncIterable<number>).
  const pushChannels = Array.from({ length: n }, () => new PushChannel<number>());
  // Wrap each PushChannel in moroutine's channel() so it is recognized by
  // moroutine's arg-preparation layer and piped cross-thread via MessagePort.
  const channels = pushChannels.map((pc) => channel<number>(pc, { highWaterMark: hwm }));
  // Shared atomic counter slots, one per worker.
  const counters = shared(Array.from({ length: n }, () => int32atomic));

  const onConnection = (socket: any) => {
    const origFd: number = socket._handle?.fd;
    if (typeof origFd !== 'number') {
      // Windows or unexpected handle type — drop hard.
      socket.destroy();
      return;
    }
    // Snapshot per-worker state, pick target.
    const threads: ServerThread[] = new Array(n);
    for (let i = 0; i < n; i++) threads[i] = { worker: workers[i], conns: counters.elements[i].load() };
    const selected = balance.select(threads);
    const idx = workers.indexOf(selected.worker);

    // Detach the socket from libuv without closing the fd. readStop()
    // deregisters the read watcher; no-oping close() prevents destroy()
    // from closing the underlying fd. The worker will open it fresh.
    const handle = socket._handle;
    handle.readStop();
    handle.close = (cb: any) => {
      if (cb) cb();
    };
    socket._handle = null;
    socket.destroy();

    counters.elements[idx].add(1);
    try {
      pushChannels[idx].send(origFd);
    } catch {
      counters.elements[idx].sub(1);
    }
  };

  server.on('connection', onConnection);

  let disposed = false;
  const dispose = () => {
    if (disposed) return;
    disposed = true;
    server.off('connection', onConnection);
    server.off('close', onClose);
    for (const pc of pushChannels) pc.close();
  };
  const onClose = () => dispose();
  server.once('close', onClose);

  // Build the indexable/iterable/disposable result.
  const entries = workers.map((w, i) => [w, [channels[i], counters.elements[i], listenOpts] as ListenArgs] as const);
  const result: readonly (readonly [WorkerHandle, ListenArgs])[] = entries;
  Object.defineProperty(result, Symbol.dispose, { value: dispose, enumerable: false });
  return result as unknown as ServerThreads;
}