import {
  ArrayPrototypePush as push,
  Error,
  FinalizationRegistry,
  ObjectCreate,
  ObjectDefineProperty as define,
  ObjectFreeze as freeze,
  ObjectGetOwnPropertyDescriptor,
  RangeError,
  ReflectApply as apply,
  ReflectOwnKeys,
  Set,
  SetPrototypeAdd as retain,
  SetPrototypeDelete as release,
  String,
  StringPrototypeSlice as slice,
  SymbolIterator,
  SymbolToStringTag,
  TypedArrayPrototypeGetBuffer as buffer,
  TypedArrayPrototypeGetByteLength as byteLength,
  TypedArrayPrototypeGetByteOffset as byteOffset,
  TypedArrayPrototypeGetTag as tag,
  TypeError,
  Uint8Array,
  Uint8ArrayPrototypeSet as setBytes,
  uncurryThis,
  WeakMap,
  WeakMapPrototypeSet as set,
} from "ext:core/primordials";
import {
  markPromiseHandled,
  promiseAll,
  promiseResolve,
  promiseThen as then,
  safePromise,
} from "ext:core/promises";
import {
  op_process_cancel_read,
  op_process_end,
  op_process_keepalive,
  op_process_kill,
  op_process_read,
  op_process_ref,
  op_process_release,
  op_process_resize,
  op_process_run_sync,
  op_process_spawn,
  op_process_stop,
  op_process_wait,
  op_process_write,
} from "ext:core/ops";
import {
  array,
  dictionary,
  isObject,
  required,
  state,
  usv,
} from "ext:web/idl.js";
import { type AbortSignal, addAbort, signals } from "ext:web/events.js";
import { registerUncloneable } from "ext:web/serialization.js";
import type { ReadableStreamDefaultReader } from "ext:webstreams/streams.js";
import {
  enqueue,
  finish,
  read,
  ReadableStream,
  readerCancel,
  release as releaseReader,
  streamLocked,
  streamReader,
  WritableStream,
} from "ext:webstreams/streams.js";
import { TextDecoder, TextEncoder } from "ext:encoding/main.js";
import { args, path, processHandle } from "ext:file/main.js";
import { addSignalListener, removeSignalListener } from "./signals.js";
import { errorCode, failure, result } from "ext:file/errors.js";
import type { URL } from "ext:url/main.js";
import type { ProcessOptions, RunSyncOptions, SpawnOptions } from "avril:std";
import type {
  NativeOptions,
  NativeStdio,
  ProcessBytes,
  ProcessOutputValue,
  ProcessStatusValue,
} from "./types.js";

const Number = globalThis.Number,
  safeInteger = Number.isSafeInteger,
  min = Math.min;
const decode = uncurryThis(TextDecoder.prototype.decode),
  encode = uncurryThis(TextEncoder.prototype.encode),
  encodingOf = uncurryThis(
    ObjectGetOwnPropertyDescriptor(TextDecoder.prototype, "encoding")!.get!,
  );
const encoder = new TextEncoder(), utf8 = new TextDecoder();
const parseJSON = JSON.parse,
  indexOf = uncurryThis(String.prototype.indexOf);
const writerOf = uncurryThis(WritableStream.prototype.getWriter),
  write = uncurryThis(WritableStreamDefaultWriter.prototype.write),
  closeWriter = uncurryThis(WritableStreamDefaultWriter.prototype.close),
  releaseWriter = uncurryThis(
    WritableStreamDefaultWriter.prototype.releaseLock,
  );
const asyncDispose: typeof Symbol.asyncDispose =
  globalThis.Symbol.asyncDispose;
const children = new WeakMap<ChildProcess, ProcessRecord>(),
  live = new Set<ChildProcess>(),
  collecting = new Set<ChildProcess>(),
  secret = ObjectCreate(null) as Secret;
registerUncloneable(children);
const register = uncurryThis(FinalizationRegistry.prototype.register),
  unregister = uncurryThis(FinalizationRegistry.prototype.unregister);
const finalizer = new FinalizationRegistry((id: number) =>
  op_process_release(id)
);

type ProcessRecord<T = ProcessBytes | string> = {
  __proto__?: null;
  id: number;
  pid: number;
  reads: Record<number, ReadSource | undefined>;
  writes: Record<number, WriteSource | undefined>;
  stdio: Record<number, ProcessStream>;
  terminal: ProcessTerminal | null;
  collectLimit: number;
  decoder: TextDecoder | null;
  decode: (bytes: ProcessBytes) => T;
  encoding: string;
  check: boolean;
  automaticInput: boolean;
  collected: boolean;
  disposed: boolean;
  finished: boolean;
  referenced: boolean;
  inputTask: Promise<void> | undefined;
  inputFailed: boolean;
  inputError: unknown;
  abortReason: unknown;
  status: Promise<ProcessStatusValue>;
  completed: Promise<ProcessStatusValue>;
};
type ProcessStream =
  | ReadableStream<ProcessBytes>
  | WritableStream<ProcessBytes>;
type ReadSource = {
  __proto__?: null;
  owner: ChildProcess | null;
  used: boolean;
  stream: ReadableStream<ProcessBytes>;
};
type WriteSource = {
  __proto__?: null;
  owner: ChildProcess | null;
  stream: WritableStream<ProcessBytes>;
};
type ProcessTerminal = {
  readable: ReadableStream<ProcessBytes>;
  writable: WritableStream<ProcessBytes>;
  resize(columns: number, rows: number): void;
};
type CollectionBudget = {
  __proto__?: null;
  limit: number;
  used: number;
  exceeded: boolean;
};
type Collection = {
  __proto__?: null;
  status: ProcessStatusValue;
  stdout: ProcessBytes | null;
  stderr: ProcessBytes | null;
  cause: unknown;
};
type LaunchedStream = {
  __proto__?: null;
  record: ProcessRecord;
  reader: ReadableStreamDefaultReader<ProcessBytes>;
  settled: Promise<Collection>;
};
interface ProcessPromise<T = ProcessBytes | string>
  extends Promise<ProcessOutputValue<T>> {
  bytes(): Promise<ProcessBytes>;
  text(): Promise<string>;
  json(): Promise<unknown>;
  lines(): AsyncIterableIterator<string>;
}
type Prepared<T = ProcessBytes | string> = {
  __proto__?: null;
  native: NativeOptions;
  input: ProcessBytes | ReadableStream<ProcessBytes> | undefined;
  stream: boolean;
  check: boolean;
  decoder: TextDecoder | null;
  decode: (bytes: ProcessBytes) => T;
  signal: unknown;
  maxCollectedBytes: number;
};

function own(object: object, key: PropertyKey, value: unknown): void {
  define(object, key, {
    __proto__: null,
    value,
    writable: true,
    configurable: true,
    enumerable: true,
  } as PropertyDescriptor);
}
class ProcessError extends Error {
  declare static readonly [Symbol.hasInstance]: (value: unknown) => value is ProcessError;
  declare readonly result: ProcessOutputValue | null;

  constructor(
    message: string,
    output: ProcessOutputValue | null = null,
    cause: unknown = undefined,
  ) {
    super(message);
    own(this, "name", "ProcessError");
    own(this, "result", output);
    if (cause !== undefined) own(this, "cause", cause);
  }
}
class OutputLimitError extends ProcessError {
  declare static readonly [Symbol.hasInstance]: (value: unknown) => value is OutputLimitError;
  constructor(output: ProcessOutputValue) {
    super("Subprocess output exceeded its configured limit", output);
    own(this, "name", "OutputLimitError");
  }
}
class ProcessTimeoutError extends ProcessError {
  declare static readonly [Symbol.hasInstance]: (value: unknown) => value is ProcessTimeoutError;
  constructor(output: ProcessOutputValue) {
    super("Subprocess timed out", output);
    own(this, "name", "ProcessTimeoutError");
  }
}
class ProcessAbortError extends ProcessError {
  declare static readonly [Symbol.hasInstance]: (value: unknown) => value is ProcessAbortError;
  constructor(output: ProcessOutputValue | null, cause?: unknown) {
    super("Subprocess was aborted", output, cause);
    own(this, "name", "ProcessAbortError");
  }
}
function checked<T extends Uint8Array | string>(
  output: ProcessOutputValue<T>,
  check: boolean,
  cause?: unknown,
): ProcessOutputValue<T> {
  if (output.reason === "output-limit") throw new OutputLimitError(output);
  if (output.reason === "timeout") throw new ProcessTimeoutError(output);
  if (output.reason === "abort") throw new ProcessAbortError(output, cause);
  if (output.reason !== null || (check && !output.success)) {
    throw new ProcessError(
      output.error ||
        (output.reason === null
          ? `Subprocess exited with ${output.signal || output.code}`
          : `Subprocess ${output.reason}`),
      output,
      cause,
    );
  }
  return output;
}
function keys(
  value: Record<PropertyKey, unknown>,
  allowed: readonly PropertyKey[],
): void {
  const names = ReflectOwnKeys(value);
  for (let i = 0; i < names.length; i++) {
    let found = false;
    for (let j = 0; j < allowed.length; j++) {
      if (names[i] === allowed[j]) found = true;
    }
    if (!found) throw new TypeError("Unknown subprocess option");
  }
}
function integer<F extends number | null | undefined>(
  value: unknown,
  fallback: F,
  minimum = 0,
  maximum = 2 ** 53 - 1,
): number | Extract<F, null | undefined> {
  if (value === undefined) {
    return fallback as number | Extract<F, null | undefined>;
  }
  if (
    typeof value !== "number" || !safeInteger(value) || value < minimum ||
    value > maximum
  ) throw new RangeError("Subprocess option is outside its integer range");
  return value;
}
function boolean(value: unknown, fallback: boolean): boolean {
  if (value === undefined) return fallback;
  if (typeof value !== "boolean") {
    throw new TypeError("Expected a boolean subprocess option");
  }
  return value;
}
function bytes(value: unknown, text = false): ProcessBytes {
  if (text && typeof value === "string") return encode(encoder, value);
  if (tag(value) !== "Uint8Array") {
    throw new TypeError("Subprocess input requires string or Uint8Array bytes");
  }
  const copy = new Uint8Array(byteLength(value as Uint8Array<ArrayBufferLike>));
  setBytes(copy, value as Uint8Array<ArrayBufferLike>);
  return copy;
}
function sequence(value: unknown): string[] {
  const output = array<string>();
  if (value === undefined) return output;
  if (!isObject(value)) {
    throw new TypeError("Arguments must be an iterable of strings");
  }
  const method = value[SymbolIterator];
  if (typeof method !== "function") {
    throw new TypeError("Arguments must be iterable");
  }
  const iterator = apply(method, value, []) as { next: unknown },
    next = iterator.next;
  if (!isObject(iterator) || typeof next !== "function") {
    throw new TypeError("Invalid argument iterator");
  }
  while (true) {
    const step = apply(next, iterator, []);
    if (!isObject(step)) {
      throw new TypeError("Invalid argument iterator result");
    }
    if (step.done) return output;
    push(output, usv(step.value));
  }
}
function stdio(
  value: unknown,
  fallback: string,
  input: boolean,
  stderr = false,
  extra = false,
): NativeStdio {
  const spec: NativeStdio = {
    __proto__: null,
    mode: fallback,
    file: null,
    fileMode: null,
    handle: null,
    direction: null,
  };
  if (typeof value === "string" || value === undefined) {
    if (value !== undefined) spec.mode = value;
  } else {
    const options = dictionary(value);
    keys(options, ["file", "mode", "handle", "pipe", "direction"]);
    const filename = options.file,
      mode = options.mode,
      handle = options.handle,
      pipe = options.pipe,
      direction = options.direction;
    if (extra) {
      spec.direction = (pipe === undefined ? direction : pipe) as string;
      input = spec.direction === "parent-to-child";
    }
    if (filename !== undefined) {
      if (handle !== undefined || pipe !== undefined) {
        throw new TypeError("Conflicting stdio options");
      }
      spec.mode = "file";
      spec.file = path(filename);
      if (input) {
        if (mode !== undefined) {
          throw new TypeError("Input redirection has no file mode");
        }
      } else spec.fileMode = mode === undefined ? "truncate" : usv(mode);
    } else if (handle !== undefined) {
      if (pipe !== undefined || mode !== undefined) {
        throw new TypeError("Conflicting stdio handle options");
      }
      spec.mode = "handle";
      spec.handle = processHandle(handle);
    } else if (extra && pipe !== undefined) {
      if (mode !== undefined || direction !== undefined) {
        throw new TypeError("Conflicting pipe options");
      }
      spec.mode = "piped";
    } else throw new TypeError("Expected a stdio mode, file, or handle");
  }
  if (
    spec.mode !== "piped" && spec.mode !== "inherit" && spec.mode !== "null" &&
    spec.mode !== "file" && spec.mode !== "handle" &&
    !(stderr && spec.mode === "stdout")
  ) throw new TypeError("Invalid stdio mode");
  if (extra && spec.mode === "inherit") {
    throw new TypeError("Extra stdio cannot inherit internal descriptors");
  }
  if (
    extra && spec.direction !== "parent-to-child" &&
    spec.direction !== "child-to-parent"
  ) throw new TypeError("Extra stdio needs a pipe direction");
  if (
    spec.fileMode !== null && spec.fileMode !== "append" &&
    spec.fileMode !== "truncate" && spec.fileMode !== "create-new"
  ) throw new TypeError("Invalid stdio file mode");
  return spec;
}
/** Splits `(args?, options?)`: an iterable second argument is the argument list. */
function invocation(
  second: unknown,
  third: unknown,
): { __proto__?: null; args: unknown; options: unknown } {
  if (isObject(second) && typeof second[SymbolIterator] === "function") {
    return { __proto__: null, args: second, options: third };
  }
  if (
    typeof second === "string" || (second !== undefined && third !== undefined)
  ) {
    throw new TypeError("Arguments must be an iterable of strings");
  }
  return { __proto__: null, args: undefined, options: second ?? third };
}
function normalize(
  command: unknown,
  positional: unknown,
  raw: unknown,
  capture: boolean,
  sync: boolean,
): Prepared {
  command = path(command);
  const options = dictionary(raw);
  keys(options, [
    "args",
    "cwd",
    "searchPath",
    "clearEnv",
    "env",
    "stdin",
    "stdout",
    "stderr",
    "input",
    "encoding",
    "check",
    "signal",
    "timeoutMs",
    "maxCollectedBytes",
    "maxBufferedBytes",
    "termination",
    "drainTimeoutMs",
    "extraStdio",
    "terminal",
  ]);
  if (positional !== undefined && options.args !== undefined) {
    throw new TypeError("Arguments were given both positionally and as args");
  }
  const argv = sequence(positional ?? options.args), cwdValue = options.cwd;
  const cwd = cwdValue === undefined ? null : path(cwdValue);
  const searchPath = boolean(options.searchPath, true),
    clearEnv = boolean(options.clearEnv, false);
  const environment = dictionary(options.env),
    env = array<[string, string]>(),
    names = ReflectOwnKeys(environment);
  for (let i = 0; i < names.length; i++) {
    const key = names[i];
    if (typeof key !== "string") {
      throw new TypeError("Environment names must be strings");
    }
    if (ObjectGetOwnPropertyDescriptor(environment, key)!.enumerable) {
      push(env, args(key, usv(environment[key])));
    }
  }
  let input: unknown = options.input, stream = false;
  if (input !== undefined) {
    if (typeof input !== "string" && tag(input) !== "Uint8Array") {
      try {
        if (streamLocked(input)) throw new TypeError("Input stream is locked");
        stream = true;
      } catch {
        throw new TypeError(
          "Input requires string, bytes, or an unlocked ReadableStream",
        );
      }
      if (sync) throw new TypeError("runSync does not accept streaming input");
    } else input = bytes(input, true);
  }
  const terminalValue = options.terminal;
  let terminal: NativeOptions["terminal"] = null;
  if (terminalValue !== undefined) {
    if (capture || sync) throw new TypeError("Terminals require spawn");
    const size = dictionary(terminalValue);
    keys(size, ["columns", "rows"]);
    terminal = {
      __proto__: null,
      columns: integer(size.columns, 80, 1, 65535),
      rows: integer(size.rows, 24, 1, 65535),
    };
  }
  const stdinValue = options.stdin,
    stdoutValue = options.stdout,
    stderrValue = options.stderr;
  if (
    input !== undefined && stdinValue !== undefined && stdinValue !== "piped"
  ) throw new TypeError("input requires piped stdin");
  const stdin = stdio(
      stdinValue,
      input !== undefined || terminal ? "piped" : "null",
      true,
    ),
    stdout = stdio(stdoutValue, "piped", false),
    stderr = stdio(stderrValue, terminal ? "stdout" : "piped", false, true);
  if (capture && stdin.mode === "piped" && input === undefined) {
    input = new Uint8Array(0);
  }
  const extra = dictionary(options.extraStdio),
    extraStdio = array<[number, NativeStdio]>(),
    descriptors = ReflectOwnKeys(extra);
  for (let i = 0; i < descriptors.length; i++) {
    const key = descriptors[i];
    if (typeof key !== "string") {
      throw new TypeError("Extra stdio descriptor must be an integer");
    }
    const fd = integer(Number(key), undefined, 3, 2147483615)!;
    if (String(fd) !== key) {
      throw new TypeError("Noncanonical stdio descriptor");
    }
    const value = dictionary(extra[key]);
    push(extraStdio, args(fd, stdio(value, "piped", false, false, true)));
  }
  if ((capture || sync) && extraStdio.length) {
    throw new TypeError("Additional streams require spawn");
  }
  const termination = dictionary(options.termination);
  keys(termination, ["scope", "signal", "graceMs"]);
  const scopeValue = termination.scope, signalValue = termination.signal;
  const scope = scopeValue === undefined ? "group" : usv(scopeValue),
    signalName = signalValue === undefined ? "SIGTERM" : usv(signalValue),
    graceMs = integer(termination.graceMs, 1000);
  if (scope !== "group" && scope !== "child") {
    throw new TypeError("Invalid termination scope");
  }
  const timeoutMs = integer(options.timeoutMs, null),
    maxCollectedBytes = integer(
      options.maxCollectedBytes,
      128 * 1024 * 1024,
      1,
    ),
    maxBufferedBytes = integer(options.maxBufferedBytes, 1024 * 1024, 1),
    drainTimeoutMs = integer(options.drainTimeoutMs, 1000);
  const check = boolean(options.check, true), encodingValue = options.encoding;
  const encoding = encodingValue === undefined ? null : usv(encodingValue);
  const decoder = encoding === null ? null : new TextDecoder(encoding);
  const signal = options.signal;
  if (signal !== undefined) {
    if (sync) throw new TypeError("runSync does not accept AbortSignal");
    const aborted = state(signals, signal);
    if (aborted.aborted) throw new ProcessAbortError(null, aborted.reason);
  }
  if (stream && streamLocked(input)) {
    throw new TypeError("Input stream is locked");
  }
  // run() always has a reader draining its output, collecting it or
  // streaming lines, so the child waits for that reader rather than racing
  // the unread limit.
  const backpressure = array<number>();
  if (capture && !sync) {
    if (stdout.mode === "piped") push(backpressure, 1);
    if (stderr.mode === "piped") push(backpressure, 2);
  }
  return {
    __proto__: null,
    native: {
      __proto__: null,
      command: command as string,
      args: argv,
      cwd,
      searchPath,
      clearEnv,
      env,
      stdin,
      stdout,
      stderr,
      extraStdio,
      terminal,
      timeoutMs,
      maxCollectedBytes: sync ? maxCollectedBytes : null,
      maxBufferedBytes,
      termination: { __proto__: null, scope, signal: signalName, graceMs },
      drainTimeoutMs,
      backpressure,
    },
    input: input as Prepared["input"],
    stream,
    check,
    decoder,
    decode: (bytes) => decoder === null ? bytes : decode(decoder, bytes),
    signal,
    maxCollectedBytes: maxCollectedBytes as number,
  };
}
function keepAlive(record: ProcessRecord, epoch: number): void {
  const pending = then(op_process_keepalive(args(record.id, epoch)), result);
  markPromiseHandled(pending);
}
function readable(record: ProcessRecord, fd: number): ReadSource {
  const source: ReadSource = {
    __proto__: null,
    owner: null,
    used: false,
    stream: null!,
  };
  source.stream = new ReadableStream<ProcessBytes>({
    __proto__: null,
    type: "bytes",
    async pull(controller) {
      source.used = true;
      if (record.disposed) {
        source.owner = null;
        finish(controller);
        return;
      }
      const chunk = result(
        await safePromise(op_process_read(args(record.id, fd))),
      );
      if (chunk === null) {
        source.owner = null;
        finish(controller);
        return;
      }
      enqueue(controller, chunk);
    },
    cancel() {
      source.used = true;
      source.owner = null;
      if (!record.disposed) result(op_process_cancel_read(args(record.id, fd)));
    },
  }, { __proto__: null, highWaterMark: 0 });
  return source;
}
function writable(record: ProcessRecord, fd: number): WriteSource {
  const source: WriteSource = { __proto__: null, owner: null, stream: null! };
  source.stream = new WritableStream<ProcessBytes>({
    __proto__: null,
    write(value) {
      if (record.disposed) {
        throw failure("BadResource", "Subprocess is disposed");
      }
      return then(op_process_write(args(record.id, fd, bytes(value))), result);
    },
    close() {
      source.owner = null;
      if (!record.disposed) {
        return then(op_process_end(args(record.id, fd)), result);
      }
    },
    abort() {
      source.owner = null;
      if (!record.disposed) {
        return then(op_process_end(args(record.id, fd)), result);
      }
    },
  }, {
    __proto__: null,
    highWaterMark: 65536,
    size(value) {
      return tag(value) === "Uint8Array" ? byteLength(value) : 1;
    },
  });
  return source;
}
async function feed(
  record: ProcessRecord,
  value: ProcessBytes | ReadableStream<ProcessBytes>,
  streaming: boolean,
): Promise<void> {
  const writer = writerOf(record.writes[0]!.stream);
  let reader: ReadableStreamDefaultReader<ProcessBytes> | null = null,
    writing = false;
  const cancellation = then(record.completed, () => {
    if (reader) markPromiseHandled(readerCancel(reader));
  });
  markPromiseHandled(cancellation);
  try {
    if (streaming) reader = streamReader(value as ReadableStream<ProcessBytes>);
    if (reader) {
      while (true) {
        writing = false;
        const step = await safePromise(read(reader));
        if (step.done) break;
        writing = true;
        await safePromise(write(writer, bytes(step.value)));
      }
    } else if (byteLength(value)) {
      writing = true;
      await safePromise(write(writer, value));
    }
    writing = true;
    await safePromise(closeWriter(writer));
  } catch (error) {
    if (!writing || errorCode(error) !== "BrokenPipe") {
      record.inputFailed = true;
      record.inputError = error;
      if (!record.finished && !record.disposed) {
        result(op_process_stop(args(record.id, "io-error")));
      }
    }
  } finally {
    if (reader) {
      markPromiseHandled(readerCancel(reader));
      releaseReader(reader);
    }
    releaseWriter(writer);
  }
}
async function collect(
  source: ReadSource | undefined,
  budget: CollectionBudget,
  record: ProcessRecord,
): Promise<ProcessBytes | null> {
  if (!source) return null;
  const reader = streamReader(source.stream), chunks = array<ProcessBytes>();
  let size = 0;
  try {
    while (true) {
      const step = await safePromise(read(reader));
      if (step.done) break;
      const count = byteLength(step.value as ProcessBytes),
        allowed = min(count, budget.limit - budget.used);
      if (allowed) {
        push(
          chunks,
          allowed === count ? step.value as ProcessBytes : new Uint8Array(
            buffer(step.value as ProcessBytes),
            byteOffset(step.value as ProcessBytes),
            allowed,
          ),
        );
        size += allowed;
        budget.used += allowed;
      }
      if (allowed < count && !budget.exceeded) {
        budget.exceeded = true;
        if (!record.disposed) {
          result(op_process_stop(args(record.id, "output-limit")));
        }
      }
    }
  } finally {
    releaseReader(reader);
  }
  const bytes = new Uint8Array(size);
  let offset = 0;
  for (let i = 0; i < chunks.length; i++) {
    setBytes(bytes, chunks[i]!, offset);
    offset += byteLength(chunks[i]!);
  }
  return bytes;
}
function output<T>(
  status: ProcessStatusValue,
  stdout: ProcessBytes | null,
  stderr: ProcessBytes | null,
  decodeOutput: (bytes: ProcessBytes) => T,
): ProcessOutputValue<T> {
  return freeze({
    __proto__: null,
    code: status.code,
    signal: status.signal,
    success: status.success,
    reason: status.reason,
    error: status.error,
    stdout: stdout === null ? null : decodeOutput(stdout),
    stderr: stderr === null ? null : decodeOutput(stderr),
  });
}
function claim(record: ProcessRecord): CollectionBudget {
  if (record.collected) {
    throw new TypeError("Subprocess output can only be collected once");
  }
  for (let fd = 1; fd <= 2; fd++) {
    const source = record.reads[fd];
    if (source && (source.used || streamLocked(source.stream))) {
      throw new TypeError("Subprocess output was already consumed or locked");
    }
  }
  record.collected = true;
  return {
    __proto__: null,
    limit: record.collectLimit,
    used: 0,
    exceeded: false,
  };
}
/** Settles a collection: the status reflects output and input failures. */
function settle(
  record: ProcessRecord,
  budget: CollectionBudget,
  values: [ProcessBytes | null, ProcessBytes | null, ProcessStatusValue, void],
): Collection {
  const status: ProcessStatusValue & { __proto__?: null } = budget.exceeded
    ? { __proto__: null, ...values[2], success: false, reason: "output-limit" }
    : record.inputFailed && values[2].reason === null
    ? {
      __proto__: null,
      ...values[2],
      success: false,
      reason: "io-error",
      error: "Subprocess input failed",
    }
    : values[2];
  return {
    __proto__: null,
    status,
    stdout: values[0],
    stderr: values[1],
    cause: status.reason === "abort" ? record.abortReason : record.inputError,
  };
}
/** Collects raw output and status, unchecked. */
function gather(
  child: ChildProcess,
  record: ProcessRecord,
): Promise<Collection> {
  const budget = claim(record);
  retain(collecting, child);
  const pending = then(
    promiseAll(
      [
        collect(record.reads[1], budget, record),
        collect(record.reads[2], budget, record),
        record.completed,
        record.inputTask,
      ] as const,
    ),
    (values) => settle(record, budget, values),
  );
  return then(pending, (value) => {
    release(collecting, child);
    return value;
  }, (error) => {
    release(collecting, child);
    throw error;
  });
}
/**
 * Collects like gather, but leaves stdout to lines(), which reads it as the
 * child writes. Settles independently of iteration.
 */
function stream(child: ChildProcess, record: ProcessRecord): LaunchedStream {
  const budget = claim(record);
  retain(collecting, child);
  const reader = streamReader(record.reads[1]!.stream);
  const pending = then(
    promiseAll(
      [
        null,
        collect(record.reads[2], budget, record),
        record.completed,
        record.inputTask,
      ] as const,
    ),
    (values) => settle(record, budget, values),
  );
  const settled = then(pending, (value) => {
    release(collecting, child);
    return value;
  }, (error) => {
    release(collecting, child);
    throw error;
  });
  return { __proto__: null, record, reader, settled };
}
async function* streamLines(launched: LaunchedStream): AsyncGenerator<string> {
  const record = launched.record, reader = launched.reader;
  const decoder = new TextDecoder(record.encoding),
    options = { __proto__: null, stream: true };
  let pending = "", finished = false, released = false, halted = false;
  // A stopped run (timeout, abort, termination) ends the iteration without
  // the lines still queued behind the reader.
  const halt = () => {
    halted = true;
    if (!released) markPromiseHandled(readerCancel(reader));
  };
  markPromiseHandled(then(launched.settled, (raw) => {
    if (raw.status.reason !== null) halt();
  }, halt));
  const line = (end: number) => {
    const text = slice(pending, 0, end);
    pending = slice(pending, end + 1);
    return unterminated(text);
  };
  try {
    lines: while (true) {
      const step = await safePromise(read(reader));
      if (halted) break;
      if (step.done) {
        pending += decode(decoder);
        if (pending !== "") yield line(pending.length);
        break;
      }
      pending += decode(decoder, step.value as ProcessBytes, options);
      for (let end; (end = indexOf(pending, "\n")) !== -1;) {
        if (halted) break lines;
        yield line(end);
      }
    }
    finished = true;
  } finally {
    if (!halted) markPromiseHandled(readerCancel(reader));
    if (!finished && !record.finished && !record.disposed) {
      result(op_process_stop(args(record.id, "terminated")));
    }
    released = true;
    releaseReader(reader);
  }
  const raw = await safePromise(launched.settled);
  checked(
    output(raw.status, null, raw.stderr, record.decode),
    record.check,
    raw.cause,
  );
}
function method<
  T extends object,
  K extends PropertyKey,
  V extends () => unknown,
>(
  object: T,
  name: K,
  value: V,
): asserts object is T & Record<K, V> {
  define(object, name, {
    __proto__: null,
    value,
    writable: true,
    configurable: true,
  } as PropertyDescriptor);
}
function split(text: string): string[] {
  const lines = array<string>();
  let start = 0;
  for (let end; (end = indexOf(text, "\n", start)) !== -1; start = end + 1) {
    push(lines, unterminated(slice(text, start, end)));
  }
  if (start < text.length) push(lines, unterminated(slice(text, start)));
  return lines;
}
function unterminated(line: string): string {
  return line[line.length - 1] === "\r" ? slice(line, 0, -1) : line;
}
/**
 * The run() promise, with methods reading stdout from the same result.
 * Collection starts on the next microtask unless lines() claims stdout
 * first, in which case it streams.
 */
function invoke<T extends ProcessBytes | string>(
  child: ChildProcess<T> | null,
  failed: boolean,
  failure: unknown,
): ProcessPromise<T> {
  const record = failed ? null : childRecord(child!);
  let streamed: LaunchedStream | null = null;
  const raw = then(promiseResolve(undefined), () => {
    if (failed) throw failure;
    return streamed ? streamed.settled : gather(child!, record!);
  });
  const settled = then(raw, (raw) => ({
    __proto__: null,
    output: checked(
      output(raw.status, raw.stdout, raw.stderr, record!.decode),
      record!.check,
      raw.cause,
    ),
    stdout: raw.stdout,
  }));
  markPromiseHandled(settled);
  const promise = then(settled, (value) => value.output);
  const stdout = () => {
    markPromiseHandled(promise);
    return then(settled, (value) => {
      if (value.stdout === null) {
        throw new TypeError(
          streamed
            ? "Subprocess stdout was streamed by lines()"
            : "Subprocess stdout is not piped",
        );
      }
      return value.stdout;
    });
  };
  const decodeText = (bytes: ProcessBytes) =>
    decode(record!.decoder ?? utf8, bytes);
  method(promise, "bytes", function bytes() {
    return stdout();
  });
  method(promise, "text", function text() {
    return then(stdout(), decodeText);
  });
  method(promise, "json", function json() {
    return then(stdout(), (bytes) => parseJSON(decodeText(bytes)));
  });
  method(promise, "lines", function lines() {
    if (failed) {
      markPromiseHandled(promise);
      return (async function* () {
        throw failure;
      })();
    }
    if (streamed) throw new TypeError("Subprocess lines can only be read once");
    if (record!.collected) {
      // Collection already began: split the buffered result instead.
      const pending = stdout();
      markPromiseHandled(pending);
      return (async function* () {
        const lines = split(decodeText(await safePromise(pending)));
        for (let i = 0; i < lines.length; i++) yield lines[i];
      })();
    }
    if (!record!.reads[1]) {
      throw new TypeError("Subprocess stdout is not piped");
    }
    streamed = stream(child!, record!);
    markPromiseHandled(promise);
    return streamLines(streamed);
  });
  return promise;
}
declare const secretBrand: unique symbol;
type Secret = { readonly [secretBrand]: true };
// Keep the output type tied to its private record across the shared resource table.
let childRecord: <T extends ProcessBytes | string>(
  child: ChildProcess<T>,
) => ProcessRecord<T>;
class ChildProcess<T extends ProcessBytes | string = ProcessBytes | string> {
  declare static readonly [Symbol.hasInstance]: (value: unknown) => value is ChildProcess;
  #record: ProcessRecord<T>;

  static {
    childRecord = (child) => {
      state(children, child);
      return child.#record;
    };
  }

  constructor(key: Secret, record: ProcessRecord<T>) {
    if (key !== secret) throw new TypeError("Illegal constructor");
    this.#record = record;
    set(children, this, record);
    register(finalizer, this, record.id, this);
  }
  get pid() {
    return state(children, this).pid;
  }
  get stdin() {
    const record = state(children, this);
    return record.automaticInput ? null : record.writes[0]?.stream ?? null;
  }
  get stdout() {
    return state(children, this).reads[1]?.stream ?? null;
  }
  get stderr() {
    return state(children, this).reads[2]?.stream ?? null;
  }
  get stdio() {
    return state(children, this).stdio;
  }
  get terminal() {
    return state(children, this).terminal;
  }
  get status() {
    return state(children, this).status;
  }
  get completed() {
    return state(children, this).completed;
  }
  output(): Promise<ProcessOutputValue<T>> {
    const record = childRecord(this);
    return then(
      gather(this, record),
      (raw) =>
        checked(
          output(raw.status, raw.stdout, raw.stderr, record.decode),
          record.check,
          raw.cause,
        ),
    );
  }
  kill(signal: string = "SIGTERM"): boolean {
    const record = state(children, this);
    return record.disposed
      ? false
      : result(op_process_kill(args(record.id, usv(signal))));
  }
  terminate(): Promise<ProcessStatusValue> {
    const record = state(children, this);
    if (!record.disposed) {
      result(op_process_stop(args(record.id, "terminated")));
    }
    return record.completed;
  }
  ref(): void {
    const record = state(children, this);
    if (!record.disposed && !record.referenced) {
      record.referenced = true;
      keepAlive(record, result(op_process_ref(args(record.id, true))));
    }
  }
  unref(): void {
    const record = state(children, this);
    if (!record.disposed && record.referenced) {
      record.referenced = false;
      result(op_process_ref(args(record.id, false)));
    }
  }
  async [asyncDispose](): Promise<void> {
    const record = state(children, this);
    if (!record.disposed) {
      result(op_process_stop(args(record.id, "dispose")));
      await safePromise(record.completed);
      record.disposed = true;
      unregister(finalizer, this);
      op_process_release(record.id);
    }
  }
}
define(ChildProcess.prototype, SymbolToStringTag, {
  value: "ChildProcess",
  configurable: true,
});
function start<T extends ProcessBytes | string>(
  options: Prepared<T>,
): ChildProcess<T> {
  const info = result(op_process_spawn(options.native));
  const record: ProcessRecord<T> = {
    __proto__: null,
    id: info.id,
    pid: info.pid,
    reads: ObjectCreate(null),
    writes: ObjectCreate(null),
    stdio: ObjectCreate(null),
    terminal: null,
    collectLimit: options.maxCollectedBytes,
    decoder: options.decoder,
    decode: options.decode,
    encoding: options.decoder === null ? "utf-8" : encodingOf(options.decoder),
    check: options.check,
    automaticInput: options.input !== undefined,
    collected: false,
    disposed: false,
    finished: false,
    referenced: true,
    inputTask: undefined,
    inputFailed: false,
    inputError: undefined,
    abortReason: undefined,
    status: undefined!,
    completed: undefined!,
  };
  const child = new ChildProcess(secret, record);
  for (let i = 0; i < info.readable.length; i++) {
    const fd = info.readable[i];
    const source = readable(record, fd);
    source.owner = child;
    record.reads[fd] = source;
    if (fd >= 3) own(record.stdio, fd, source.stream);
  }
  for (let i = 0; i < info.writable.length; i++) {
    const fd = info.writable[i];
    const source = writable(record, fd);
    source.owner = child;
    record.writes[fd] = source;
    if (fd >= 3) own(record.stdio, fd, source.stream);
  }
  freeze(record.stdio);
  record.status = then(
    op_process_wait(args(info.id, false)),
    (reply) => freeze(result(reply)),
  );
  record.completed = then(
    op_process_wait(args(info.id, true)),
    (reply) => freeze(result(reply)),
  );
  markPromiseHandled(record.status);
  markPromiseHandled(record.completed);
  keepAlive(record, 0);
  retain(live, child);
  let removeAbort: (() => void) | undefined;
  if (options.signal !== undefined) {
    removeAbort = addAbort(options.signal as AbortSignal, () => {
      record.abortReason = state(signals, options.signal).reason;
      if (!record.finished && !record.disposed) {
        result(op_process_stop(args(record.id, "abort")));
      }
    });
  }
  const cleanup = then(record.completed, () => {
    record.finished = true;
    removeAbort?.();
    release(live, child);
  });
  markPromiseHandled(cleanup);
  if (info.terminal) {
    record.terminal = freeze({
      readable: record.reads[1]!.stream,
      writable: record.writes[0]!.stream,
      resize(columns: number, rows: number) {
        required(arguments.length, 2);
        if (record.disposed) {
          throw failure("BadResource", "Subprocess is disposed");
        }
        result(
          op_process_resize(
            args(
              record.id,
              integer(columns, undefined, 1, 65535)!,
              integer(rows, undefined, 1, 65535)!,
            ),
          ),
        );
      },
    });
  }
  if (options.input !== undefined) {
    record.inputTask = feed(record, options.input!, options.stream);
    markPromiseHandled(record.inputTask);
  }
  return child;
}
// Static methods retain the non-constructible function values while allowing overloads.
class Launch {
  static spawn(
    command: string | URL,
    options?: SpawnOptions & { encoding?: undefined },
  ): ChildProcess<ProcessBytes>;
  static spawn(
    command: string | URL,
    options: SpawnOptions & { encoding: string },
  ): ChildProcess<string>;
  static spawn(
    command: string | URL,
    args: Iterable<string> | undefined,
    options?: SpawnOptions & { args?: undefined; encoding?: undefined },
  ): ChildProcess<ProcessBytes>;
  static spawn(
    command: string | URL,
    args: Iterable<string> | undefined,
    options: SpawnOptions & { args?: undefined; encoding: string },
  ): ChildProcess<string>;
  static spawn(
    command: string | URL,
    second: unknown = undefined,
    third: unknown = undefined,
  ): ChildProcess {
    required(arguments.length, 1);
    const call = invocation(second, third);
    return start(normalize(command, call.args, call.options, false, false));
  }
  static run(
    command: string | URL,
    options?: ProcessOptions & { encoding?: undefined },
  ): ProcessPromise<ProcessBytes>;
  static run(
    command: string | URL,
    options: ProcessOptions & { encoding: string },
  ): ProcessPromise<string>;
  static run(
    command: string | URL,
    args: Iterable<string> | undefined,
    options?: ProcessOptions & { args?: undefined; encoding?: undefined },
  ): ProcessPromise<ProcessBytes>;
  static run(
    command: string | URL,
    args: Iterable<string> | undefined,
    options: ProcessOptions & { args?: undefined; encoding: string },
  ): ProcessPromise<string>;
  static run(
    command: string | URL,
    second: unknown = undefined,
    third: unknown = undefined,
  ): ProcessPromise {
    let child: ChildProcess | null = null, failed = false, failure: unknown;
    try {
      required(arguments.length, 1);
      const call = invocation(second, third);
      child = start(normalize(command, call.args, call.options, true, false));
    } catch (error) {
      failed = true;
      failure = error;
    }
    return invoke(child, failed, failure);
  }
  static runSync(
    command: string | URL,
    options?: RunSyncOptions & { encoding?: undefined },
  ): ProcessOutputValue<ProcessBytes>;
  static runSync(
    command: string | URL,
    options: RunSyncOptions & { encoding: string },
  ): ProcessOutputValue<string>;
  static runSync(
    command: string | URL,
    args: Iterable<string> | undefined,
    options?: RunSyncOptions & { args?: undefined; encoding?: undefined },
  ): ProcessOutputValue<ProcessBytes>;
  static runSync(
    command: string | URL,
    args: Iterable<string> | undefined,
    options: RunSyncOptions & { args?: undefined; encoding: string },
  ): ProcessOutputValue<string>;
  static runSync(
    command: string | URL,
    second: unknown = undefined,
    third: unknown = undefined,
  ): ProcessOutputValue {
    required(arguments.length, 1);
    const call = invocation(second, third);
    const prepared = normalize(command, call.args, call.options, true, true);
    const reply = result(
      op_process_run_sync(
        args(
          prepared.native,
          prepared.input === undefined ? null : prepared.input as ProcessBytes,
        ),
      ),
    );
    return checked(
      output(reply.status, reply.stdout, reply.stderr, prepared.decode),
      prepared.check,
    );
  }
}
const { spawn, run, runSync } = Launch;
export {
  addSignalListener,
  ChildProcess,
  OutputLimitError,
  ProcessAbortError,
  ProcessError,
  ProcessTimeoutError,
  removeSignalListener,
  run,
  runSync,
  spawn,
};