// node:worker_threads ports. A pair is one native MessageChannel plus a shared queue per
// direction (a growable SharedArrayBuffer). A message without transferables is cloned
// into the queue (clone.ts) and the native port only rings the receiver's doorbell; a
// message with transferables, or holding what only a native clone can carry, rides the
// native port itself. Every message takes the next number in its direction's sequence,
// so the receiver delivers both paths in send order. The queue is what lets
// receiveMessageOnPort take a message synchronously, from a thread that never yields to
// its event loop (esbuild-wasm's sync API blocks in Atomics.wait, then reads its port).
//
// Records, not bare values, travel the pair: user messages and, on a Worker's channel,
// the thread's lifecycle (online, stdio, its uncaught error) and port close.
import type { PortDescriptor } from "../shared/protocol";
import { capture, runInFrame, type Context } from "./async_hooks";
import { NeedsNative, deserialize, kTransferOnly, serialize } from "./clone";
import { EventEmitter, type EventEmitterConstructor } from "./events";
import type { Loop } from "./loop";

export type PortRecord =
  | { k: "m"; v: unknown }
  | { k: "close" }
  | { k: "online" }
  | { k: "stdio"; fd: 1 | 2; bytes: Uint8Array }
  | { k: "error"; v: unknown };

/** What crosses the native port: a doorbell for the queue, or a whole record. */
type Wire = { bell: true } | { seq: number; rec: PortRecord };

// ---- the shared queue: one direction of a pair ----
// Int32 header [0] lock [1] head [2] tail, then records [u32 length][u32 seq][bytes]
// from byte 16. The writer compacts, then grows, when a record does not fit; the reader
// resets both ends when it empties the queue. Critical sections are copies, so the
// lock spins with a short wait.

const DATA = 16;
const INITIAL = 64 * 1024;
const MAX = 256 * 1024 * 1024;

export function newQueue(): SharedArrayBuffer {
  return new SharedArrayBuffer(DATA + INITIAL, { maxByteLength: DATA + MAX });
}

function lock(h: Int32Array): void {
  while (Atomics.compareExchange(h, 0, 0, 1) !== 0) Atomics.wait(h, 0, 1, 1);
}

function unlock(h: Int32Array): void {
  Atomics.store(h, 0, 0);
  Atomics.notify(h, 0, 1);
}

function push(sab: SharedArrayBuffer, seq: number, bytes: Uint8Array): void {
  const h = new Int32Array(sab, 0, 4);
  lock(h);
  try {
    let head = h[1]!;
    let tail = h[2]!;
    const need = 8 + bytes.length;
    if (tail + need > sab.byteLength - DATA && head > 0) {
      new Uint8Array(sab, DATA).copyWithin(0, head, tail);
      tail -= head;
      head = 0;
    }
    if (tail + need > sab.byteLength - DATA) {
      const want = DATA + Math.max(2 * (sab.byteLength - DATA), tail + need);
      if (want > sab.maxByteLength) throw new RangeError(`ERR_NIB_PORT_FULL: a port's queue cannot hold another ${bytes.length} bytes`);
      sab.grow(want);
    }
    const view = new DataView(sab, DATA);
    view.setUint32(tail, bytes.length, true);
    view.setUint32(tail + 4, seq, true);
    new Uint8Array(sab, DATA + tail + 8, bytes.length).set(bytes);
    h[1] = head;
    h[2] = tail + need;
  } finally {
    unlock(h);
  }
}

/** The record at the head when it is number `seq`, else null (empty, or `seq` is a
 *  native record still in flight). */
function shift(sab: SharedArrayBuffer, seq: number): Uint8Array | null {
  const h = new Int32Array(sab, 0, 4);
  lock(h);
  try {
    const head = h[1]!;
    const tail = h[2]!;
    if (head === tail) return null;
    const view = new DataView(sab, DATA);
    if (view.getUint32(head + 4, true) !== seq) return null;
    const len = view.getUint32(head, true);
    // slice() copies out of shared memory, which TextDecoder will not read.
    const bytes = new Uint8Array(sab, DATA + head + 8, len).slice();
    const next = head + 8 + len;
    h[1] = next === tail ? 0 : next;
    if (next === tail) h[2] = 0;
    return bytes;
  } finally {
    unlock(h);
  }
}

function queued(sab: SharedArrayBuffer): boolean {
  const h = new Int32Array(sab, 0, 4);
  return Atomics.load(h, 1) !== Atomics.load(h, 2);
}

// ---- the runtime's loop, set once at boot ----

let loop: Loop | null = null;
export function installPorts(l: Loop): void {
  loop = l;
}

type NativePort = globalThis.MessagePort;
const BELL: Wire = { bell: true };

/** Chromium's DataCloneError names the call it came from; Node's does not. */
function nodeCloneError(e: unknown): unknown {
  if (e instanceof DOMException && e.name === "DataCloneError") {
    return new DOMException(e.message.replace(/^Failed to execute 'postMessage' on '\w+': /, ""), "DataCloneError");
  }
  return e;
}

export class MessagePort extends (EventEmitter as unknown as EventEmitterConstructor) {
  declare [kTransferOnly]: true;
  #native: NativePort | null;
  #rx: SharedArrayBuffer;
  #tx: SharedArrayBuffer;
  #rxSeq: number;
  #txSeq: number;
  /** Records that came natively, by sequence number, until their turn. */
  #inbox = new Map<number, PortRecord>();
  /** Records taken from the queues and not yet delivered. */
  #pending: PortRecord[] = [];
  #started = false;
  #closed = false;
  #ref = true;
  #holding = false;
  #drainQueued = false;
  /** addEventListener's registry, keyed by type then the original listener, so
   *  removeEventListener finds the EventEmitter listener it wraps (DESIGN: ports had
   *  no EventTarget API; this is the thin layer over the existing on()/emit()). */
  #wrapped = new Map<string, Map<EventListenerOrEventListenerObject, (...args: unknown[]) => void>>();
  #onmessage: ((ev: MessageEvent) => void) | null = null;
  #onmessageerror: ((ev: MessageEvent) => void) | null = null;
  /** A Worker's end of its channel: every record goes here, lifecycle records ahead
   *  of the messages taken with them, as Node drains a worker's internal port before
   *  its public one. */
  onRecord: ((rec: PortRecord) => void) | null = null;
  /** The frame this port was created in: worker_threads.ts overrides delivery via
   *  onRecord (and captures its own, Worker-scoped frame there), so this only governs
   *  a plain MessageChannel's own 'message' event. */
  readonly #frame: Context;

  /** @internal Ports come from MessageChannel or arrive in a message. */
  constructor(desc: PortDescriptor) {
    super();
    Object.defineProperty(this, kTransferOnly, { value: true });
    this.#frame = capture();
    this.#native = desc.native;
    this.#rx = desc.rx;
    this.#tx = desc.tx;
    this.#rxSeq = desc.rxSeq;
    this.#txSeq = desc.txSeq;
    this.#native.onmessage = (ev: MessageEvent<Wire>) => this.#arrived(ev.data);
    // Node's ports start, and hold the loop, while they have a 'message' listener.
    this.on("newListener", (event: string | symbol) => {
      if (event === "message" && this.listenerCount("message") === 0) {
        this.#ref = true;
        this.start();
      }
    });
    this.on("removeListener", (event: string | symbol) => {
      if (event === "message" && this.listenerCount("message") === 0) {
        this.#started = false;
        this.#ref = false;
        this.#updateHold();
      }
    });
  }

  postMessage(value: unknown, transferList?: readonly unknown[] | { transfer?: readonly unknown[] }): void {
    const transfer = Array.isArray(transferList) ? transferList : (transferList as { transfer?: unknown[] } | undefined)?.transfer ?? [];
    if (this.#closed) return;
    this.send({ k: "m", v: value }, transfer);
  }

  /** @internal Send one record, by the queue when it can go there. */
  send(rec: PortRecord, transfer: readonly unknown[] = []): void {
    if (!this.#native) return;
    if (transfer.length === 0) {
      let bytes: Uint8Array | null = null;
      try {
        bytes = serialize(rec, { errorProps: rec.k === "error" });
      } catch (e) {
        if (!(e instanceof NeedsNative)) throw e;
      }
      if (bytes) {
        push(this.#tx, this.#txSeq, bytes);
        this.#txSeq++;
        this.#native.postMessage(BELL);
        return;
      }
    }
    const { value, natives } = exportPorts(rec, transfer);
    try {
      this.#native.postMessage({ seq: this.#txSeq, rec: value as PortRecord } satisfies Wire, natives as Transferable[]);
    } catch (e) {
      throw nodeCloneError(e);
    }
    this.#txSeq++;
  }

  start(): void {
    if (this.#closed) return;
    this.#started = true;
    this.#updateHold();
    this.#queueDrain();
  }

  close(): void {
    if (this.#closed) return;
    this.send({ k: "close" });
    this.#shut();
  }

  ref(): this {
    this.#ref = true;
    this.#updateHold();
    return this;
  }

  unref(): this {
    this.#ref = false;
    this.#updateHold();
    return this;
  }

  hasRef(): boolean {
    return this.#holding;
  }

  /** The EventTarget half of a port (Node's MessagePort is both): wraps a listener
   *  to receive a MessageEvent, backed by the same on()/emit() the Worker's internal
   *  record delivery and the 'newListener' auto-start already use. */
  addEventListener(type: string, listener: EventListenerOrEventListenerObject | null, options?: boolean | AddEventListenerOptions): void {
    if (listener === null || (typeof listener !== "function" && typeof listener.handleEvent !== "function")) return;
    let byType = this.#wrapped.get(type);
    if (!byType) {
      byType = new Map();
      this.#wrapped.set(type, byType);
    }
    if (byType.has(listener)) return; // DOM: adding the same listener twice is a no-op.
    const once = typeof options === "object" && options?.once === true;
    const call = typeof listener === "function" ? listener : listener.handleEvent.bind(listener);
    const wrapped = (data: unknown) => {
      if (once) this.removeEventListener(type, listener);
      call(new MessageEvent(type, { data }));
    };
    byType.set(listener, wrapped);
    this.on(type, wrapped);
  }

  removeEventListener(type: string, listener: EventListenerOrEventListenerObject | null): void {
    const wrapped = listener === null ? undefined : this.#wrapped.get(type)?.get(listener);
    if (!wrapped) return;
    this.#wrapped.get(type)!.delete(listener!);
    this.off(type, wrapped);
  }

  dispatchEvent(event: Event): boolean {
    this.emit(event.type, (event as MessageEvent).data);
    return true;
  }

  /** `port.onmessage = fn` is Node's sugar for addEventListener("message", fn):
   *  one slot, replaced on reassignment, which is also why it starts the port (the
   *  same 'newListener' hook that a plain on("message", ...) trips). */
  get onmessage(): ((ev: MessageEvent) => void) | null {
    return this.#onmessage;
  }

  set onmessage(fn: ((ev: MessageEvent) => void) | null) {
    if (this.#onmessage) this.removeEventListener("message", this.#onmessage as EventListener);
    this.#onmessage = typeof fn === "function" ? fn : null;
    if (this.#onmessage) this.addEventListener("message", this.#onmessage as EventListener);
  }

  get onmessageerror(): ((ev: MessageEvent) => void) | null {
    return this.#onmessageerror;
  }

  set onmessageerror(fn: ((ev: MessageEvent) => void) | null) {
    if (this.#onmessageerror) this.removeEventListener("messageerror", this.#onmessageerror as EventListener);
    this.#onmessageerror = typeof fn === "function" ? fn : null;
    if (this.#onmessageerror) this.addEventListener("messageerror", this.#onmessageerror as EventListener);
  }

  /** @internal receiveMessageOnPort: the next message, taken now, or undefined. */
  receiveSync(): { message: unknown } | undefined {
    for (;;) {
      const rec = this.#pending.shift() ?? this.#take();
      if (!rec) return undefined;
      if (rec.k === "m") return { message: rec.v };
      this.#deliver(rec);
    }
  }

  /** @internal Hand this end to another thread: it stops here. */
  detach(): PortDescriptor {
    const native = this.#native;
    if (!native) throw new DOMException("MessagePort was already transferred", "DataCloneError");
    native.onmessage = null;
    this.#native = null;
    this.#started = false;
    this.#closed = true;
    this.#updateHold();
    return { nibPort: true, native, rx: this.#rx, tx: this.#tx, rxSeq: this.#rxSeq, txSeq: this.#txSeq };
  }

  /** @internal Records are waiting on a native one still in flight. */
  stalled(): boolean {
    return queued(this.#rx) || this.#inbox.size > 0;
  }

  /** @internal Deliver everything available now; the Worker calls it when its thread
   *  has ended, before 'exit'. */
  drainNow(): void {
    for (let rec = this.#pending.shift() ?? this.#take(); rec; rec = this.#pending.shift() ?? this.#take()) this.#deliver(rec);
  }

  #arrived(wire: Wire): void {
    if (!("bell" in wire)) this.#inbox.set(wire.seq, wire.rec);
    this.#queueDrain();
    this.onWake?.();
  }

  /** @internal Called on every arrival: a Worker waiting to emit 'exit' retries. */
  onWake: (() => void) | null = null;

  #take(): PortRecord | undefined {
    const native = this.#inbox.get(this.#rxSeq);
    if (native) {
      this.#inbox.delete(this.#rxSeq);
      this.#rxSeq++;
      return importPorts(native) as PortRecord;
    }
    const bytes = shift(this.#rx, this.#rxSeq);
    if (!bytes) return undefined;
    this.#rxSeq++;
    return deserialize(bytes) as PortRecord;
  }

  #queueDrain(): void {
    if (this.#drainQueued || !this.#started || !loop) return;
    this.#drainQueued = true;
    loop.enqueueCompletion(() => {
      this.#drainQueued = false;
      this.#drainOne();
    });
  }

  /** One record per loop callback, so ticks and microtasks run between messages. */
  #drainOne(): void {
    if (!this.#started) return;
    if (this.#pending.length === 0) {
      const batch: PortRecord[] = [];
      for (let rec = this.#take(); rec; rec = this.#take()) batch.push(rec);
      // Lifecycle records first: Node drains a worker's internal port before its public one.
      this.#pending = this.onRecord ? [...batch.filter((r) => r.k !== "m"), ...batch.filter((r) => r.k === "m")] : batch;
    }
    const rec = this.#pending.shift();
    if (rec) this.#deliver(rec);
    // Later arrivals ring their own doorbell.
    if (this.#pending.length) this.#queueDrain();
  }

  #deliver(rec: PortRecord): void {
    if (this.onRecord) return this.onRecord(rec);
    switch (rec.k) {
      case "m":
        runInFrame(this.#frame, () => this.emit("message", rec.v));
        return;
      case "close":
        this.#shut();
        return;
      default:
        return;
    }
  }

  #shut(): void {
    if (this.#closed) return;
    this.#closed = true;
    this.#started = false;
    this.#updateHold();
    this.#native?.close();
    this.#native = null;
    loop?.enqueueCompletion(() => this.emit("close"));
  }

  #updateHold(): void {
    const hold = this.#started && this.#ref && !this.#closed;
    if (hold === this.#holding || !loop) return;
    this.#holding = hold;
    if (hold) loop.hold();
    else loop.release();
  }
}

/** A connected pair, as `new MessageChannel()` gives the guest. */
export function portPair(): [MessagePort, MessagePort] {
  const ch = new globalThis.MessageChannel();
  const a = newQueue();
  const b = newQueue();
  return [
    new MessagePort({ nibPort: true, native: ch.port1, rx: a, tx: b, rxSeq: 0, txSeq: 0 }),
    new MessagePort({ nibPort: true, native: ch.port2, rx: b, tx: a, rxSeq: 0, txSeq: 0 }),
  ];
}

export class MessageChannel {
  readonly port1: MessagePort;
  readonly port2: MessagePort;
  constructor() {
    [this.port1, this.port2] = portPair();
  }
}

/** Replace transferred node ports in `value` with descriptors, and list the native
 *  objects a native postMessage must transfer. */
export function exportPorts(value: unknown, transfer: readonly unknown[]): { value: unknown; natives: unknown[] } {
  const swapped = new Map<object, PortDescriptor>();
  const natives = transfer.map((t) => {
    if (!(t instanceof MessagePort)) return t;
    const desc = t.detach();
    swapped.set(t, desc);
    return desc.native;
  });
  return { value: swapped.size ? mapPlain(value, (o) => swapped.get(o)) : value, natives };
}

/** Turn descriptors in a natively received value back into node ports. */
export function importPorts(value: unknown): unknown {
  return mapPlain(value, (o) => ((o as Partial<PortDescriptor>).nibPort === true ? new MessagePort(o as PortDescriptor) : undefined));
}

/** Copy arrays and plain objects in `value`, replacing each object `swap` answers for. */
function mapPlain(value: unknown, swap: (o: object) => unknown): unknown {
  const memo = new Map<object, unknown>();
  const walk = (v: unknown): unknown => {
    if (v === null || typeof v !== "object") return v;
    const replaced = swap(v);
    if (replaced !== undefined) return replaced;
    if (memo.has(v)) return memo.get(v);
    if (Array.isArray(v)) {
      const out: unknown[] = [];
      memo.set(v, out);
      for (const x of v) out.push(walk(x));
      return out;
    }
    const proto = Object.getPrototypeOf(v);
    if (proto !== Object.prototype && proto !== null) return v;
    const out: Record<string, unknown> = {};
    memo.set(v, out);
    for (const [k, x] of Object.entries(v)) out[k] = walk(x);
    return out;
  };
  return walk(value);
}