// node:worker_threads. A Worker is a kernel thread in this process (DESIGN §Processes,
// threads, signals): the kernel starts a runtime worker for it and says when it ends;
// the Worker object holds this thread's loop until then. Everything else travels the
// port pair the Worker creates (ports.ts): messages both ways, and from the thread its
// 'online', its stdout and stderr (written to this thread's process.stdout/stderr, as
// Node pipes them) and its uncaught error.
import type { HostToken } from "../shared/vurls";
import type { KernelToRuntimeInit, RuntimeToKernel, SyscallSpec, ThreadExit } from "../shared/protocol";
import { capture, runInFrame, type Context } from "./async_hooks";
import { Buffer } from "./buffer";
import { uncloneable } from "./clone";
import { invalidArgType } from "./errors";
import { EventEmitter, type EventEmitterConstructor } from "./events";
import type { Loop } from "./loop";
import { MessageChannel, MessagePort, exportPorts, importPorts, portPair, type PortRecord } from "./ports";
import type { GuestProcess } from "./process";
import { unsupported } from "./refuse";
import { Readable } from "./stream";

export interface ThreadsDeps {
  loop: Loop;
  init: KernelToRuntimeInit;
  host: HostToken;
  callSync<K extends keyof SyscallSpec>(op: K, args: SyscallSpec[K]["args"]): SyscallSpec[K]["ret"];
  post(msg: RuntimeToKernel, transfer: Transferable[]): void;
  process(): GuestProcess;
  resolvePath(p: string): string;
}

export interface WorkerThreads {
  module: Record<string, unknown>;
  /** This thread's end of its parent's channel; null on the main thread. */
  parentPort: MessagePort | null;
  threadExited(exit: ThreadExit): void;
}

interface WorkerOptions {
  eval?: boolean;
  workerData?: unknown;
  argv?: unknown[];
  env?: Record<string, unknown> | symbol;
  transferList?: unknown[];
  name?: string;
  stdout?: boolean;
  stderr?: boolean;
  resourceLimits?: Record<string, number>;
}

const SHARE_ENV = Symbol.for("nodejs.worker_threads.SHARE_ENV");
const EVAL_NAME = "[worker eval]";

function workerPathError(filename: string): TypeError {
  return Object.assign(new TypeError(
    "The worker script or module filename must be an absolute path or a relative path starting with './' or '../'."
    + ` Received "${filename}"`,
  ), { code: "ERR_WORKER_PATH" });
}

export function makeWorkerThreads(deps: ThreadsDeps): WorkerThreads {
  const { loop, init } = deps;
  const self = init.thread;
  const workers = new Map<number, Worker>();
  const environmentData = new Map<unknown, unknown>(
    self ? (importPorts(self.environmentData) as Array<[unknown, unknown]>) : [],
  );
  const untransferable = new WeakSet<object>();
  const parentPort = self ? new MessagePort(self.port) : null;

  const PROFILER_WHY = "no per-worker V8 heap or profiler from a page";

  class Worker extends (EventEmitter as unknown as EventEmitterConstructor) {
    readonly threadId: number;
    readonly stdin = null;
    readonly stdout = new Readable({ read() {} });
    readonly stderr = new Readable({ read() {} });
    #resourceLimits: Record<string, number>;
    #port: MessagePort;
    #pipe: { stdout: boolean; stderr: boolean };
    #exit: ThreadExit | null = null;
    #done = false;
    #ref = true;
    #holding = false;
    #exited: Array<(code: number) => void> = [];
    #threadName: string;
    /** new Worker() creates the handle (probed): 'online', 'message' and 'exit' all
     *  run in the frame current here, not in whatever frame happens to be current
     *  when the thread's port wakes this one natively. */
    readonly #frame: Context = capture();

    constructor(filename: string | URL, options: WorkerOptions = {}) {
      super();
      const isEval = options.eval === true;
      let entry: string;
      if (isEval) {
        if (typeof filename !== "string") throw invalidArgType("filename", "of type string", filename);
        entry = EVAL_NAME;
      } else if (filename instanceof URL) {
        if (filename.protocol !== "file:") throw Object.assign(new TypeError(`The URL must be of scheme file`), { code: "ERR_INVALID_URL_SCHEME" });
        entry = decodeURIComponent(filename.pathname);
      } else if (typeof filename === "string") {
        if (filename.startsWith("file:")) entry = decodeURIComponent(new URL(filename).pathname);
        else if (filename.startsWith("/")) entry = filename;
        else if (filename.startsWith("./") || filename.startsWith("../")) entry = deps.resolvePath(filename);
        else throw workerPathError(filename);
      } else {
        throw invalidArgType("filename", "of type string or an instance of URL", filename);
      }
      if (options.env === SHARE_ENV) {
        throw unsupported("worker_threads", "SHARE_ENV", "a thread's env is a snapshot taken when it starts; the kernel keeps no live environment to share");
      }
      const proc = deps.process();
      const source = options.env === undefined || options.env === null ? proc.env : options.env as Record<string, unknown>;
      const env = Object.fromEntries(Object.entries(source).map(([k, v]) => [k, String(v)]));
      const argv = (options.argv ?? []).map(String);

      this.threadId = deps.callSync("newThread", {}).tid;
      this.#threadName = typeof options.name === "string" && options.name !== "" ? options.name : "WorkerThread";
      this.#resourceLimits = { ...(options.resourceLimits ?? {}) };
      this.#pipe = { stdout: options.stdout === true, stderr: options.stderr === true };

      const [mine, theirs] = portPair();
      this.#port = mine;
      mine.onRecord = (rec) => runInFrame(this.#frame, () => this.#record(rec));
      mine.onWake = () => {
        if (this.#exit) loop.enqueueCompletion(() => runInFrame(this.#frame, () => this.#tryFinish()));
      };
      // The Worker, not its port, decides whether the loop stays alive.
      mine.unref();
      mine.start();

      const { value, natives } = exportPorts(
        { workerData: options.workerData, environmentData: [...environmentData] },
        options.transferList ?? [],
      );
      const shared = value as { workerData: unknown; environmentData: Array<[unknown, unknown]> };
      const port = theirs.detach();
      const transfer = [port.native, ...natives] as Transferable[];
      deps.post({
        t: "startThread", entry, argv, env, cwd: proc.cwd(),
        thread: {
          tid: this.threadId, name: this.#threadName, eval: isEval ? filename as string : null,
          workerData: shared.workerData, environmentData: shared.environmentData, port,
        },
        transfer,
      }, transfer);
      workers.set(this.threadId, this);
      this.#updateHold();
    }

    /** The limits this Worker was given, as Node's getter reads back the constructor
     *  option; Node's own defaults and post-exit `{}` are V8-internal and not modeled. */
    get resourceLimits(): Record<string, number> {
      return { ...this.#resourceLimits };
    }

    /** The name option this Worker was started with, as the thread sees itself
     *  through worker_threads.threadName; "WorkerThread" is Node's default. */
    get threadName(): string {
      return this.#threadName;
    }

    cpuUsage(): never {
      throw unsupported("worker_threads", "Worker.cpuUsage()", PROFILER_WHY);
    }

    startCpuProfile(): never {
      throw unsupported("worker_threads", "Worker.startCpuProfile()", PROFILER_WHY);
    }

    startHeapProfile(): never {
      throw unsupported("worker_threads", "Worker.startHeapProfile()", PROFILER_WHY);
    }

    postMessage(value: unknown, transferList?: readonly unknown[]): void {
      if (!this.#done) this.#port.postMessage(value, transferList);
    }

    terminate(): Promise<number | undefined> {
      if (this.#done) return Promise.resolve(undefined);
      deps.post({ t: "terminateThread", tid: this.threadId }, []);
      return new Promise((resolve) => this.#exited.push(resolve));
    }

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

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

    getHeapSnapshot(): never {
      throw unsupported("worker_threads", "Worker.getHeapSnapshot()", "the browser exposes no heap snapshots");
    }

    getHeapStatistics(): never {
      throw unsupported("worker_threads", "Worker.getHeapStatistics()", PROFILER_WHY);
    }

    /** @internal The kernel says the thread ended. */
    ended(exit: ThreadExit): void {
      this.#exit = exit;
      loop.enqueueCompletion(() => runInFrame(this.#frame, () => this.#tryFinish()));
    }

    #record(rec: PortRecord): void {
      switch (rec.k) {
        case "online":
          this.emit("online");
          return;
        case "m":
          this.emit("message", rec.v);
          return;
        case "error":
          this.emit("error", rec.v);
          return;
        case "stdio": {
          const name = rec.fd === 1 ? "stdout" : "stderr";
          if (this.#pipe[name]) this[name].push(Buffer.from(rec.bytes));
          else deps.process()[name].write(rec.bytes);
          return;
        }
        case "close":
          return;
      }
    }

    /** 'exit' comes after everything the thread sent; a record sent natively may
     *  still be in flight, and its arrival retries. */
    #tryFinish(): void {
      if (!this.#exit || this.#done) return;
      this.#port.drainNow();
      if (this.#port.stalled() && !this.#exit.terminated) return;
      this.#done = true;
      workers.delete(this.threadId);
      this.#updateHold();
      this.#port.close();
      if (this.#pipe.stdout) this.stdout.push(null);
      if (this.#pipe.stderr) this.stderr.push(null);
      const { code } = this.#exit;
      this.emit("exit", code);
      for (const resolve of this.#exited.splice(0)) resolve(code);
    }

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

  /** Node's BroadcastChannel: an EventTarget over the browser's, scoped to this process
   *  so two guests on one origin never hear each other. Open channels hold the loop. */
  class BroadcastChannel extends EventTarget {
    // Node's name/onmessage/onmessageerror are prototype accessors (WebIDL legacy
    // event-handler attributes), not own instance fields — getters below, backed by
    // private state, so they show up the same way on the class.
    #name: string;
    #native: globalThis.BroadcastChannel | null;
    #ref = true;
    #holding = false;
    #onmessage: ((ev: MessageEvent) => void) | null = null;
    #onmessageerror: ((ev: MessageEvent) => void) | null = null;
    readonly #frame: Context;

    constructor(name: unknown) {
      super();
      if (name === undefined) throw Object.assign(new TypeError('The "name" argument must be specified'), { code: "ERR_MISSING_ARGS" });
      this.#name = String(name);
      this.#frame = capture();
      this.#native = new globalThis.BroadcastChannel(`nib:${deps.host}:${init.pid}:${this.#name}`);
      this.#native.onmessage = (ev) => loop.enqueueCompletion(() => runInFrame(this.#frame, () => {
        const out = new MessageEvent("message", { data: ev.data });
        this.#onmessage?.(out);
        this.dispatchEvent(out);
      }));
      this.#updateHold();
    }

    get name(): string {
      return this.#name;
    }

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

    set onmessage(fn: ((ev: MessageEvent) => void) | null) {
      this.#onmessage = typeof fn === "function" ? fn : null;
    }

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

    set onmessageerror(fn: ((ev: MessageEvent) => void) | null) {
      this.#onmessageerror = typeof fn === "function" ? fn : null;
    }

    postMessage(message: unknown): void {
      if (!this.#native) throw new DOMException("BroadcastChannel is closed.", "InvalidStateError");
      this.#native.postMessage(message);
    }

    close(): void {
      this.#native?.close();
      this.#native = null;
      this.#updateHold();
    }

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

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

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

  const module = {
    isMainThread: self === null,
    isInternalThread: false,
    threadId: self?.tid ?? 0,
    // Node's default display name for an unnamed thread; the main thread's is "".
    threadName: self ? (self.name || "WorkerThread") : "",
    workerData: self ? importPorts(self.workerData) : null,
    parentPort,
    resourceLimits: {},
    SHARE_ENV,
    Worker,
    MessageChannel,
    MessagePort,
    BroadcastChannel,
    // The Web Locks API re-export DESIGN calls for; cross-thread locks are the
    // browser's own, not a kernel reimplementation.
    locks: (globalThis as unknown as { navigator: { locks: unknown } }).navigator.locks,
    markAsUntransferable(value: unknown): void {
      if (value !== null && (typeof value === "object" || typeof value === "function")) untransferable.add(value);
    },
    isMarkedAsUntransferable(value: unknown): boolean {
      return value !== null && typeof value === "object" && untransferable.has(value);
    },
    /** Node checks this only for a plain object (clone.ts): Array, Map, Date and the
     *  rest already have their own clone path before the check and are unaffected. */
    markAsUncloneable(value: unknown): void {
      if (value !== null && (typeof value === "object" || typeof value === "function")) uncloneable.add(value as object);
    },
    /** Node routes this through the main thread's registry of every live thread in
     *  the process; this kernel only wires a thread to its direct parent/child ports
     *  (ports.ts), so only the same-thread validation can be honest here — reaching
     *  an arbitrary thread id needs the kernel to mediate (a missing primitive, not
     *  implementable from this module alone). */
    async postMessageToThread(threadId: unknown, _value: unknown, _transferListOrTimeout?: unknown, _timeout?: unknown): Promise<void> {
      if (Number(threadId) === module.threadId) {
        throw Object.assign(new Error("Cannot sent a message to the same thread"), { code: "ERR_WORKER_MESSAGING_SAME_THREAD" });
      }
      throw unsupported("worker_threads", "postMessageToThread()", "no process-wide thread registry: only direct parent/child ports are wired, so an arbitrary thread id can't be routed to");
    },
    /** The port's next message, taken now from its shared queue, or undefined. A
     *  message sent with transferables rides the native port, so it is here only once
     *  this thread's loop has seen it arrive. */
    receiveMessageOnPort(port: unknown): { message: unknown } | undefined {
      if (!(port instanceof MessagePort)) {
        throw Object.assign(new TypeError('The "port" argument must be a MessagePort instance'), { code: "ERR_INVALID_ARG_TYPE" });
      }
      return port.receiveSync();
    },
    getEnvironmentData(key: unknown): unknown {
      return environmentData.get(key);
    },
    setEnvironmentData(key: unknown, value: unknown): void {
      if (value === undefined) environmentData.delete(key);
      else environmentData.set(key, value);
    },
    moveMessagePortToContext(): never {
      throw unsupported("worker_threads", "moveMessagePortToContext()", "there is no node:vm context to move a port into");
    },
  };

  return {
    module,
    parentPort,
    threadExited(exit) {
      workers.get(exit.tid)?.ended(exit);
    },
  };
}