// node:stream, the subset IncomingMessage, @hono/node-server, Express and zlib need:
// the legacy Stream base, a Readable with Node's flowing/paused modes, read(),
// end/close ordering, destroy(), pipe()/unpipe() into any writable-shaped emitter,
// async iteration and the toWeb/fromWeb adapters; Node's writable state machine as a
// Writable, and over that Readable as a Duplex (an upgraded socket) and a Transform.
import { Buffer } from "./buffer";
import { EventEmitter, type EventEmitterConstructor } from "./events";
import { nextTick } from "./ticks";

export interface StreamConstructor extends EventEmitterConstructor {
  new (options?: unknown): EventEmitter;
  Stream: StreamConstructor;
  Readable: typeof Readable;
  Writable: typeof Writable;
  Duplex: typeof Duplex;
  Transform: typeof Transform;
  PassThrough: typeof PassThrough;
  finished: typeof finished;
  pipeline: typeof pipeline;
  promises: typeof promises;
  addAbortSignal: typeof addAbortSignal;
  compose: typeof compose;
  destroy: typeof destroy;
  duplexPair: typeof duplexPair;
  getDefaultHighWaterMark: typeof getDefaultHighWaterMark;
  setDefaultHighWaterMark: typeof setDefaultHighWaterMark;
  isDestroyed: typeof isDestroyed;
  isDisturbed: typeof isDisturbed;
  isErrored: typeof isErrored;
  isReadable: typeof isReadable;
  isWritable: typeof isWritable;
}

/** Node's default highWaterMark since v22: 64 KiB, or 16 objects. Mutable through
 *  get/setDefaultHighWaterMark, so a later `new Readable()` picks up a changed value. */
let defaultHighWaterMarkBytes = 64 * 1024;
let defaultHighWaterMarkObjectMode = 16;

export function getDefaultHighWaterMark(objectMode = false): number {
  return objectMode ? defaultHighWaterMarkObjectMode : defaultHighWaterMarkBytes;
}

export function setDefaultHighWaterMark(objectMode: boolean, value: number): void {
  if (!Number.isInteger(value) || value < 0) {
    throw nodeError("ERR_OUT_OF_RANGE", `The value of "value" is out of range. It must be >= 0 and an integer. Received ${value}`);
  }
  if (objectMode) defaultHighWaterMarkObjectMode = value;
  else defaultHighWaterMarkBytes = value;
}

/** Node's literal "Received type ..." / "Received an instance of ..." tail, reused by
 *  every ERR_INVALID_ARG_TYPE here. */
function describeReceived(value: unknown): string {
  if (value === undefined) return " Received undefined";
  if (typeof value === "object" && value !== null) return ` Received an instance of ${(value as object).constructor?.name ?? "Object"}`;
  return ` Received type ${typeof value} (${typeof value === "string" ? `'${value}'` : String(value)})`;
}

/** Readable state a Transform in this module sets, kept off the public surface. */
let readableInternals!: {
  /** Node's Transform clears `sync` at construction: its pushes come from writes. */
  clearSync(r: Readable): void;
  /** A duplex closes only when both sides are done. */
  setCanAutoDestroy(r: Readable, ready: () => boolean): void;
};

/** The legacy base every Node stream inherits from (`instanceof Stream`). */
export const Stream = function Stream(this: EventEmitter, options?: unknown) {
  EventEmitter.call(this, options);
} as unknown as StreamConstructor;
Object.setPrototypeOf(Stream.prototype, EventEmitter.prototype);
Object.setPrototypeOf(Stream, EventEmitter);

function nodeError(code: string, message: string): Error {
  return Object.assign(new Error(message), { code });
}

export interface ReadableOptions {
  highWaterMark?: number;
  encoding?: string;
  objectMode?: boolean;
  autoDestroy?: boolean;
  read?(this: Readable, size: number): void;
  destroy?(this: Readable, error: Error | null, callback: (error?: Error | null) => void): void;
}

type Chunk = Buffer | string | unknown;

interface ReadableStateView {
  readonly objectMode: boolean;
  readonly highWaterMark: number;
  readonly length: number;
  readonly flowing: boolean | null;
  readonly ended: boolean;
  readonly endEmitted: boolean;
  readonly reading: boolean;
  readonly destroyed: boolean;
  readonly closed: boolean;
  readonly errored: Error | null;
}

/** What pipe() needs from a destination: Node accepts any writable-shaped emitter. */
export interface PipeDestination extends EventEmitter {
  write(chunk: unknown): boolean;
  end(): unknown;
}

export class Readable extends (Stream as unknown as EventEmitterConstructor) {
  static fromWeb: typeof fromWeb;
  static toWeb: typeof toWeb;
  static from: typeof readableFrom;
  static wrap: typeof readableWrapStatic;

  #readable = true;
  #readableObjectMode: boolean;
  #readableHighWaterMark: number;
  #buffer: Chunk[] = [];
  #length = 0; // bytes, or chunks in object mode
  #flowing: boolean | null = null;
  #ended = false; // push(null) seen
  #endEmitted = false;
  #endScheduled = false;
  #flowScheduled = false;
  #reading = false;
  /** Node's kNeedReadable/kEmittedReadable/kReadableListening: whether a push should
   *  bother emitting 'readable' at all (only true while something is actually waiting
   *  on it — a first 'readable' listener attaching, or read() having just found nothing),
   *  whether that emission is already scheduled (so a second push before it fires
   *  doesn't queue a redundant one), and whether the first 'readable' listener's
   *  one-time setup already ran. Without this gating every push while paused queues
   *  its own nextTick, even when an earlier one is still pending or nothing is
   *  listening — extra hops Node never takes (see lib/internal/streams/readable.js's
   *  `emitReadable`/`emitReadable_`), which was exactly the extra tick a composed,
   *  already-flowing Readable showed up behind other, unrelated stream setup. */
  #needReadable = false;
  #emittedReadable = false;
  #readableListening = false;
  /** Node's `state.sync`: true until the first _read returns and during each _read.
   *  A push while sync buffers instead of emitting 'data' on the caller's stack. */
  #sync = true;
  #readingMore = false;
  #canAutoDestroy: () => boolean = () => true;
  #didRead = false;
  #destroyed = false;
  #closed = false;
  #errored: Error | null = null;
  #encoding: string | null = null;
  #decoder: TextDecoder | null = null;
  #autoDestroy: boolean;
  #waiters: Array<() => void> = [];
  #pipes: Array<{ dest: PipeDestination; cleanup: () => void }> = [];
  #iterationGuarded = false;
  #stateView: ReadableStateView | null = null;

  static {
    readableInternals = {
      clearSync: (r) => {
        r.#sync = false;
      },
      setCanAutoDestroy: (r, ready) => {
        r.#canAutoDestroy = ready;
      },
    };
  }

  constructor(options: ReadableOptions = {}) {
    super();
    this.#readableObjectMode = options.objectMode ?? false;
    this.#readableHighWaterMark = options.highWaterMark ?? getDefaultHighWaterMark(this.#readableObjectMode);
    this.#autoDestroy = options.autoDestroy ?? true;
    if (options.read) this._read = options.read;
    if (options.destroy) this._destroy = options.destroy;
    if (options.encoding) this.setEncoding(options.encoding);
  }

  /** Node makes this settable too (legacy code sometimes flips it by hand). */
  get readable(): boolean {
    return this.#readable;
  }
  set readable(value: boolean) {
    this.#readable = value;
  }
  get readableObjectMode(): boolean {
    return this.#readableObjectMode;
  }
  get readableHighWaterMark(): number {
    return this.#readableHighWaterMark;
  }
  /** Node's deprecated direct view of the queue (undocumented but present). */
  get readableBuffer(): Chunk[] {
    return this.#buffer.slice();
  }
  get readableLength(): number {
    return this.#length;
  }
  get readableEnded(): boolean {
    return this.#endEmitted;
  }
  get readableDidRead(): boolean {
    return this.#didRead;
  }
  get readableEncoding(): string | null {
    return this.#encoding;
  }
  get readableFlowing(): boolean | null {
    return this.#flowing;
  }
  get readableAborted(): boolean {
    return (this.#destroyed || this.#errored !== null) && !this.#endEmitted;
  }
  get destroyed(): boolean {
    return this.#destroyed;
  }
  get closed(): boolean {
    return this.#closed;
  }
  get errored(): Error | null {
    return this.#errored;
  }
  /** A live view of the state fields npm code reads (ws: `endEmitted`, `length`). */
  get _readableState(): ReadableStateView {
    const self = this;
    return (this.#stateView ??= {
      get objectMode() { return self.readableObjectMode; },
      get highWaterMark() { return self.readableHighWaterMark; },
      get length() { return self.#length; },
      get flowing() { return self.#flowing; },
      get ended() { return self.#ended; },
      get endEmitted() { return self.#endEmitted; },
      get reading() { return self.#reading; },
      get destroyed() { return self.#destroyed; },
      get closed() { return self.#closed; },
      get errored() { return self.#errored; },
    });
  }

  _read(_size: number): void {
    this.destroy(nodeError("ERR_METHOD_NOT_IMPLEMENTED", "The _read() method is not implemented"));
  }

  _destroy(error: Error | null, callback: (error?: Error | null) => void): void {
    callback(error);
  }

  push(chunk: unknown, encoding?: string): boolean {
    this.#reading = false;
    if (chunk === null) {
      if (this.#ended) return false;
      this.#ended = true;
      this.#wake();
      if (this.#flowing && !this.#sync) this.#flow();
      else if (this.#flowing) this.#scheduleFlow();
      else if (this.#length === 0) this.#emitReadable();
      return false;
    }
    if (this.#ended) {
      nextTick(() => this.destroy(nodeError("ERR_STREAM_PUSH_AFTER_EOF", "stream.push() after EOF")));
      return false;
    }
    let item: Chunk = chunk;
    if (!this.readableObjectMode) {
      if (typeof chunk === "string") item = Buffer.from(chunk, encoding);
      else if (chunk instanceof Uint8Array && !(chunk instanceof Buffer)) {
        item = new Buffer(chunk.buffer as ArrayBuffer, chunk.byteOffset, chunk.byteLength);
      } else if (!(chunk instanceof Buffer)) {
        throw nodeError("ERR_INVALID_ARG_TYPE", 'The "chunk" argument must be of type string or an instance of Buffer or Uint8Array.');
      }
      if (this.#encoding !== null) item = this.#decode(item as Buffer);
      // A decoder holding a partial character yields "": nothing to queue yet.
      if ((item as Buffer | string).length === 0) return this.#length < this.readableHighWaterMark;
    }
    if (this.#flowing && this.#length === 0 && !this.#sync && this.listenerCount("data") > 0) {
      this.#didRead = true;
      this.emit("data", item);
    } else {
      this.#buffer.push(item);
      this.#length += this.readableObjectMode ? 1 : (item as Buffer | string).length;
      if (this.#flowing) this.#scheduleFlow();
      else if (this.#needReadable) this.#emitReadable();
    }
    this.#wake();
    this.#maybeReadMore();
    return this.#length < this.readableHighWaterMark || this.#length === 0;
  }

  unshift(chunk: unknown): void {
    if (chunk === null) return;
    const item = !this.readableObjectMode && typeof chunk === "string" && this.#encoding === null ? Buffer.from(chunk) : chunk;
    this.#buffer.unshift(item);
    this.#length += this.readableObjectMode ? 1 : (item as Buffer | string).length;
  }

  read(size?: number): Chunk | null {
    if (size === 0) {
      this.#maybeRead();
      return null;
    }
    // Node asks for more before taking, so a _read that pushes synchronously (a
    // Transform releasing a held write) contributes to this read.
    if (this.#length === 0 || this.#length - this.#howMuch(size) < this.readableHighWaterMark) this.#maybeRead(true);
    const chunk = this.#take(size);
    if (chunk === null) {
      // Node's howMuchToRead: nothing to give back means nothing is buffered (or not
      // enough while paused), so the next push should wake whoever's waiting again.
      if (!this.#flowing && !this.#ended) this.#needReadable = true;
      if (this.#ended) this.#scheduleEnd();
      return null;
    }
    this.#didRead = true;
    if (this.#length === 0) {
      if (this.#ended) this.#scheduleEnd();
      else if (!this.#flowing) this.#needReadable = true;
    }
    this.emit("data", chunk);
    return chunk;
  }

  setEncoding(encoding: string): this {
    this.#encoding = encoding.toLowerCase() === "utf-8" ? "utf8" : encoding.toLowerCase();
    this.#decoder = this.#encoding === "utf8" ? new TextDecoder() : null;
    this.#buffer = this.#buffer.map((c) => (c instanceof Buffer ? this.#decode(c) : c));
    return this;
  }

  override on(event: string | symbol, fn: (...args: any[]) => void): this {
    super.on(event, fn);
    if (event === "data" && this.#flowing !== false) this.resume();
    // Node's kReadableListening: this one-time setup (arming needReadable, forcing
    // paused mode) runs only for the *first* 'readable' listener — a second one
    // attaching later rides on the state the first already set up.
    if (event === "readable" && !this.#endEmitted && !this.#readableListening) {
      this.#readableListening = true;
      this.#needReadable = true;
      this.#flowing = false;
      this.#emittedReadable = false;
      if (this.#length > 0 || this.#ended) this.#emitReadable();
      // Node's nReadingNextTick: a 'readable' listener starts the source reading.
      else if (!this.#reading) nextTick(() => this.read(0));
    }
    return this;
  }
  override addListener(event: string | symbol, fn: (...args: any[]) => void): this {
    return this.on(event, fn);
  }

  resume(): this {
    if (this.#destroyed) return this;
    if (!this.#flowing) {
      this.#flowing = true;
      this.#scheduleFlow();
      nextTick(() => this.emit("resume"));
    }
    return this;
  }

  pause(): this {
    // Node's pause(): a destroyed stream (as `pt` here already is by the time its
    // 'finish'-triggered unpipe() calls this) never emits 'pause' — confirmed against
    // real Node 24.16.0 (lib/internal/streams/readable.js's `Readable.prototype.pause`),
    // where this guard is what keeps a pipe's auto-unpipe-on-finish from firing a
    // spurious 'pause' once the source has already ended and auto-destroyed.
    if (this.#destroyed) return this;
    if (this.#flowing !== false) {
      this.#flowing = false;
      this.emit("pause");
    }
    return this;
  }

  /** Node's pipe: write each chunk, pause while the destination asks for a drain,
   *  end it on 'end' unless `{ end: false }`, and unpipe when it closes or errors. */
  pipe<T extends PipeDestination>(dest: T, options?: { end?: boolean }): T {
    const onData = (chunk: unknown): void => {
      if (dest.write(chunk) === false) this.pause();
    };
    const onDrain = (): void => {
      this.resume();
    };
    const onEnd = (): void => {
      dest.end();
    };
    const onUnpipeTrigger = (): void => {
      this.unpipe(dest);
    };
    const onError = (err: unknown): void => {
      this.unpipe(dest);
      if (dest.listenerCount("error") === 0) throw err;
    };
    const onSourceEnd = options?.end === false ? onUnpipeTrigger : onEnd;
    this.on("data", onData);
    dest.on("drain", onDrain);
    dest.once("close", onUnpipeTrigger);
    dest.once("finish", onUnpipeTrigger);
    dest.prependListener("error", onError);
    this.once("end", onSourceEnd);
    this.#pipes.push({
      dest,
      cleanup: () => {
        this.off("data", onData);
        this.off("end", onSourceEnd);
        dest.off("drain", onDrain);
        dest.off("close", onUnpipeTrigger);
        dest.off("finish", onUnpipeTrigger);
        dest.off("error", onError);
      },
    });
    dest.emit("pipe", this);
    if (!this.#flowing) this.resume();
    return dest;
  }

  unpipe(dest?: PipeDestination): this {
    if (this.#pipes.length === 0) return this;
    const removed = dest === undefined ? this.#pipes : this.#pipes.filter((p) => p.dest === dest);
    if (removed.length === 0) return this;
    this.#pipes = dest === undefined ? [] : this.#pipes.filter((p) => p.dest !== dest);
    if (this.#pipes.length === 0) this.pause();
    for (const p of removed) {
      p.cleanup();
      p.dest.emit("unpipe", this, { hasUnpiped: false });
    }
    return this;
  }

  isPaused(): boolean {
    return this.#flowing === false;
  }

  destroy(error?: Error | null, callback?: (error?: Error | null) => void): this {
    if (this.#destroyed) {
      if (callback) nextTick(() => callback(this.#errored));
      return this;
    }
    this.#destroyed = true;
    this.readable = false;
    if (error) this.#errored = error;
    this._destroy(error ?? null, (err) => {
      if (err && !this.#errored) this.#errored = err;
      nextTick(() => {
        this.#closed = true;
        this.#wake();
        if (err) {
          markErrorEmitted(this);
          this.emit("error", err);
        }
        this.emit("close");
        callback?.(err);
      });
    });
    return this;
  }

  [Symbol.asyncIterator](): AsyncGenerator<Chunk> {
    return this.#iterate(true);
  }

  /** readable.iterator({ destroyOnReturn }): the same iteration, but a consumer that
   *  stops early (destroyOnReturn: false) leaves the stream alive instead of destroying it. */
  iterator(options?: { destroyOnReturn?: boolean }): AsyncGenerator<Chunk> {
    return this.#iterate(options?.destroyOnReturn ?? true);
  }

  #iterate(destroyOnReturn: boolean): AsyncGenerator<Chunk> {
    const self = this;
    // The generator reads #errored directly instead of listening for 'error'; this
    // listener exists only so a destroy() this iteration triggers (closing early, or
    // cascading into a piped-from source) never hits Node's "unhandled 'error' event"
    // default throw. Node's own async iterator does the same, kept for the stream's life.
    if (!self.#iterationGuarded) {
      self.#iterationGuarded = true;
      self.on("error", () => {});
    }
    return (async function* () {
      let finished = false;
      try {
        while (true) {
          const chunk = self.#destroyed ? null : self.read();
          if (chunk !== null) {
            yield chunk;
            continue;
          }
          if (self.#errored) throw self.#errored;
          if (self.#ended && self.#length === 0) {
            finished = true;
            return;
          }
          if (self.#destroyed) throw nodeError("ERR_STREAM_PREMATURE_CLOSE", "Premature close");
          await new Promise<void>((resolve) => self.#waiters.push(resolve));
        }
      } finally {
        // A consumer that stops early (break, return(), an uncaught throw further down
        // the chain) closes the source the way Node's own iterator helpers do: an abort,
        // not a silent destroy, unless it already ran to completion.
        if (!finished && destroyOnReturn) destroy(self);
      }
    })();
  }

  /** readable.wrap(stream): drive this Readable from an old-style (pre-streams3)
   *  readable: one that emits 'data'/'end' and exposes pause()/resume(). */
  wrap(stream: EventEmitter & { pause(): void; resume(): void }): this {
    let paused = false;
    stream.on("data", (chunk: unknown) => {
      if (this.readableObjectMode && chunk == null) return;
      if (!this.readableObjectMode && (!chunk || (chunk as { length: number }).length === 0)) return;
      if (!this.push(chunk) && !paused) {
        paused = true;
        stream.pause();
      }
    });
    stream.on("end", () => this.push(null));
    for (const event of ["error", "close", "destroy", "pause", "resume"]) {
      stream.on(event, (...args: unknown[]) => this.emit(event, ...args));
    }
    for (const key in stream as unknown as Record<string, unknown>) {
      if ((this as unknown as Record<string, unknown>)[key] === undefined && typeof (stream as unknown as Record<string, unknown>)[key] === "function") {
        (this as unknown as Record<string, unknown>)[key] = (...args: unknown[]) => (stream as unknown as Record<string, (...a: unknown[]) => unknown>)[key]!(...args);
      }
    }
    this._read = () => {
      if (paused) {
        paused = false;
        stream.resume();
      }
    };
    return this;
  }

  /** readable.compose(stream): pipe this readable through a stream, a generator
   *  function, or an async function, and return the combined Duplex. */
  compose(stream: unknown): Duplex {
    return compose(this, stream);
  }

  // map/filter/flatMap return a Readable synchronously, so (unlike forEach/some/every/
  // find/reduce/toArray, which validate lazily and reject) Node validates up front:
  // an async generator's body doesn't run at all until first consumed, so leaving
  // this to mapAsync/filterAsync/flatMapAsync would silently drop the error if the
  // result is never read.
  map<R>(fn: (value: Chunk, index: number) => R | Promise<R>, options?: IteratorHelperOptions): Readable {
    validateFn(fn);
    validateConcurrency(options);
    return readableFrom(mapAsync(this, fn, options));
  }

  filter(fn: (value: Chunk, index: number) => unknown, options?: IteratorHelperOptions): Readable {
    validateFn(fn);
    validateConcurrency(options);
    return readableFrom(filterAsync(this, fn, options));
  }

  flatMap<R>(fn: (value: Chunk, index: number) => R | Iterable<R> | AsyncIterable<R>, options?: IteratorHelperOptions): Readable {
    validateFn(fn);
    validateConcurrency(options);
    return readableFrom(flatMapAsync(this, fn, options));
  }

  take(limit: number): Readable {
    return readableFrom(takeAsync(this, validateTakeDrop(limit)));
  }

  drop(limit: number): Readable {
    return readableFrom(dropAsync(this, validateTakeDrop(limit)));
  }

  forEach(fn: (value: Chunk, index: number) => unknown, options?: IteratorHelperOptions): Promise<void> {
    return forEachAsync(this, fn, options);
  }

  toArray(options?: { signal?: AbortSignal }): Promise<Chunk[]> {
    return toArrayAsync(this, options);
  }

  some(fn: (value: Chunk, index: number) => unknown, options?: IteratorHelperOptions): Promise<boolean> {
    return someAsync(this, fn, options);
  }

  every(fn: (value: Chunk, index: number) => unknown, options?: IteratorHelperOptions): Promise<boolean> {
    return everyAsync(this, fn, options);
  }

  find<R = Chunk>(fn: (value: Chunk, index: number) => unknown, options?: IteratorHelperOptions): Promise<R | undefined> {
    return findAsync(this, fn, options) as Promise<R | undefined>;
  }

  reduce<R>(fn: (acc: R | Chunk, value: Chunk, index: number) => R | Promise<R>, initial?: R, options?: { signal?: AbortSignal }): Promise<R> {
    return reduceAsync(this, fn, initial, arguments.length > 1, options);
  }

  #decode(bytes: Buffer): string {
    return this.#decoder ? this.#decoder.decode(bytes, { stream: true }) : bytes.toString(this.#encoding ?? undefined);
  }

  /** Dequeue: one chunk while flowing, everything when paused, or exactly `size` bytes. */
  #take(size: number | undefined): Chunk | null {
    if (this.#length === 0) return null;
    if (this.readableObjectMode || (size === undefined && this.#flowing)) {
      const chunk = this.#buffer.shift()!;
      this.#length -= this.readableObjectMode ? 1 : (chunk as Buffer | string).length;
      return chunk;
    }
    const want = size === undefined || size >= this.#length ? this.#length : size;
    if (size !== undefined && size > this.#length && !this.#ended) return null;
    const parts: Array<Buffer | string> = [];
    let need = want;
    while (need > 0) {
      const head = this.#buffer[0] as Buffer | string;
      if (head.length <= need) {
        parts.push(head);
        this.#buffer.shift();
        need -= head.length;
      } else {
        parts.push(head.slice(0, need));
        this.#buffer[0] = head.slice(need);
        need = 0;
      }
    }
    this.#length -= want;
    if (this.#encoding !== null) return (parts as string[]).join("");
    return parts.length === 1 ? parts[0]! : Buffer.concat(parts as Buffer[]);
  }

  /** Node's howMuchToRead: what read(size) would take now. */
  #howMuch(size: number | undefined): number {
    if (this.#length === 0) return 0;
    if (this.readableObjectMode) return 1;
    if (size === undefined) return this.#flowing ? (this.#buffer[0] as Buffer | string).length : this.#length;
    if (size <= this.#length) return size;
    return this.#ended ? this.#length : 0;
  }

  #maybeRead(belowMark = this.#length < this.readableHighWaterMark): void {
    if (this.#ended || this.#reading || this.#destroyed || !belowMark) return;
    this.#reading = true;
    this.#sync = true;
    // Node's read(): an empty buffer about to call _read definitely needs a 'readable'
    // (or more 'data') once something arrives.
    if (this.#length === 0) this.#needReadable = true;
    this._read(this.readableHighWaterMark);
    this.#sync = false;
  }

  /** Node's maybeReadMore: after a push, keep reading up to the mark on a later tick. */
  #maybeReadMore(): void {
    if (this.#readingMore) return;
    this.#readingMore = true;
    nextTick(() => {
      while (!this.#reading && !this.#ended && !this.#destroyed &&
        (this.#length < this.readableHighWaterMark || (this.#flowing && this.#length === 0))) {
        const before = this.#length;
        this.#maybeRead(true);
        if (before === this.#length) break;
      }
      this.#readingMore = false;
    });
  }

  #flow(): void {
    while (this.#flowing && !this.#destroyed && this.read() !== null) { /* each read emits 'data' */ }
  }

  #scheduleFlow(): void {
    if (this.#flowScheduled) return;
    this.#flowScheduled = true;
    nextTick(() => {
      this.#flowScheduled = false;
      this.#flow();
    });
  }

  /** Node's emitReadable: marks a 'readable' emission as owed and, unless one is
   *  already pending (#emittedReadable), schedules the one that will pay it off. */
  #emitReadable(): void {
    this.#needReadable = false;
    if (!this.#emittedReadable) {
      this.#emittedReadable = true;
      nextTick(() => this.#emitReadableNow());
    }
  }

  /** Node's emitReadable_: fires 'readable' only if there's still something to show
   *  for it (a push drained in the meantime leaves this a no-op), then re-arms
   *  needReadable for whatever pushes next while paused and below the mark. */
  #emitReadableNow(): void {
    this.#emittedReadable = false;
    if (!this.#destroyed && (this.#length > 0 || this.#ended)) this.emit("readable");
    if (!this.#flowing && !this.#ended && this.#length <= this.readableHighWaterMark) this.#needReadable = true;
    if (this.#flowing) this.#flow();
  }

  #scheduleEnd(): void {
    if (this.#endEmitted || this.#endScheduled) return;
    this.#endScheduled = true;
    nextTick(() => {
      if (this.#endEmitted || this.#length !== 0 || this.#destroyed) return;
      this.#endEmitted = true;
      this.readable = false;
      this.emit("end");
      if (this.#autoDestroy && this.#canAutoDestroy()) this.destroy();
    });
  }

  #wake(): void {
    const waiters = this.#waiters;
    this.#waiters = [];
    for (const w of waiters) w();
  }
}

function validateFn(fn: unknown, name = "fn"): void {
  if (typeof fn !== "function") throw nodeError("ERR_INVALID_ARG_TYPE", `The "${name}" argument must be of type function.${describeReceived(fn)}`);
}

export interface IteratorHelperOptions {
  concurrency?: number;
  signal?: AbortSignal;
}

function validateConcurrency(options?: IteratorHelperOptions): number {
  const concurrency = options?.concurrency ?? 1;
  if (!Number.isInteger(concurrency) || concurrency < 1) {
    throw nodeError("ERR_OUT_OF_RANGE", `The value of "options.concurrency" is out of range. It must be >= 1 && <= 9007199254740991. Received ${concurrency}`);
  }
  return concurrency;
}

function abortError(): Error {
  return nodeError("ABORT_ERR", "The operation was aborted");
}

function checkAborted(signal: AbortSignal | undefined): void {
  if (signal?.aborted) throw abortError();
}

function validateTakeDrop(n: unknown): number {
  const value = Number(n);
  if (value < 0) throw nodeError("ERR_OUT_OF_RANGE", `The value of "number" is out of range. It must be >= 0. Received ${value}`);
  return value;
}

/** Node's iterator helpers run `fn` with up to `concurrency` calls in flight, in
 *  parallel, but always yield results in the source's order. Closing early (break,
 *  an error) closes `source`'s own iterator, which cascades into an abort on it. */
async function* mapAsync<R>(source: AsyncIterable<Chunk>, fn: (value: Chunk, index: number) => R | Promise<R>, options?: IteratorHelperOptions): AsyncGenerator<R> {
  validateFn(fn);
  const concurrency = validateConcurrency(options);
  const signal = options?.signal;
  const iterator = source[Symbol.asyncIterator]();
  const inFlight: Promise<{ value: R }>[] = [];
  let index = 0;
  let sourceDone = false;
  const pull = async (): Promise<void> => {
    const { value, done } = await iterator.next();
    if (done) {
      sourceDone = true;
      return;
    }
    const i = index++;
    inFlight.push(Promise.resolve(fn(value, i)).then((v) => ({ value: v })));
  };
  try {
    while (!sourceDone && inFlight.length < concurrency) {
      checkAborted(signal);
      await pull();
    }
    while (inFlight.length) {
      checkAborted(signal);
      const { value } = await inFlight.shift()!;
      if (!sourceDone && inFlight.length < concurrency) await pull();
      yield value;
    }
  } finally {
    await iterator.return?.();
  }
}

async function* filterAsync(source: AsyncIterable<Chunk>, fn: (value: Chunk, index: number) => unknown, options?: IteratorHelperOptions): AsyncGenerator<Chunk> {
  validateFn(fn);
  for await (const { value, keep } of mapAsync(source, async (value, index) => ({ value, keep: await fn(value, index) }), options)) {
    if (keep) yield value;
  }
}

function isIterable(value: unknown): value is Iterable<unknown> | AsyncIterable<unknown> {
  return value != null && (typeof (value as Record<PropertyKey, unknown>)[Symbol.asyncIterator] === "function" || typeof (value as Record<PropertyKey, unknown>)[Symbol.iterator] === "function");
}

async function* flatMapAsync(source: AsyncIterable<Chunk>, fn: (value: Chunk, index: number) => unknown, options?: IteratorHelperOptions): AsyncGenerator<Chunk> {
  validateFn(fn);
  for await (const mapped of mapAsync(source, fn, options)) {
    if (isIterable(mapped)) yield* mapped as AsyncIterable<Chunk> | Iterable<Chunk>;
    else yield mapped as Chunk;
  }
}

async function* takeAsync(source: AsyncIterable<Chunk>, limit: number): AsyncGenerator<Chunk> {
  if (!(limit > 0)) return;
  let n = 0;
  for await (const value of source) {
    yield value;
    if (++n >= limit) return;
  }
}

async function* dropAsync(source: AsyncIterable<Chunk>, limit: number): AsyncGenerator<Chunk> {
  let n = 0;
  for await (const value of source) {
    if (n++ < limit) continue;
    yield value;
  }
}

async function forEachAsync(source: AsyncIterable<Chunk>, fn: (value: Chunk, index: number) => unknown, options?: IteratorHelperOptions): Promise<void> {
  validateFn(fn);
  for await (const _ of mapAsync(source, fn, options)) {
    // Node discards each return value; the call is for its side effect.
  }
}

async function toArrayAsync(source: AsyncIterable<Chunk>, options?: { signal?: AbortSignal }): Promise<Chunk[]> {
  const out: Chunk[] = [];
  for await (const value of source) {
    checkAborted(options?.signal);
    out.push(value);
  }
  return out;
}

async function someAsync(source: AsyncIterable<Chunk>, fn: (value: Chunk, index: number) => unknown, options?: IteratorHelperOptions): Promise<boolean> {
  validateFn(fn);
  for await (const keep of mapAsync(source, fn, options)) {
    if (keep) return true;
  }
  return false;
}

async function everyAsync(source: AsyncIterable<Chunk>, fn: (value: Chunk, index: number) => unknown, options?: IteratorHelperOptions): Promise<boolean> {
  validateFn(fn);
  for await (const keep of mapAsync(source, fn, options)) {
    if (!keep) return false;
  }
  return true;
}

async function findAsync(source: AsyncIterable<Chunk>, fn: (value: Chunk, index: number) => unknown, options?: IteratorHelperOptions): Promise<Chunk | undefined> {
  validateFn(fn);
  let index = 0;
  for await (const value of mapAsync(source, async (value, i) => ({ value, i, keep: await fn(value, i) }), options)) {
    index = value.i;
    if (value.keep) return value.value;
  }
  void index;
  return undefined;
}

async function reduceAsync<R>(source: AsyncIterable<Chunk>, fn: (acc: R | Chunk, value: Chunk, index: number) => R | Promise<R>, initial: R | undefined, hasInitial: boolean, options?: { signal?: AbortSignal }): Promise<R> {
  validateFn(fn);
  let acc: R | Chunk | undefined = initial;
  let seeded = hasInitial;
  let index = 0;
  for await (const value of source) {
    checkAborted(options?.signal);
    acc = seeded ? await fn(acc as R, value, index) : value;
    seeded = true;
    index++;
  }
  if (!seeded) throw nodeError("ERR_MISSING_ARGS", "Reduce of an empty stream requires an initial value");
  return acc as R;
}

/** Readable.from(iterable): a Readable fed by pulling an iterable (sync or async) one
 *  value at a time. A string or Buffer is pushed whole, as Node does. */
function readableFrom(iterable: unknown, options: ReadableOptions = {}): Readable {
  if (typeof iterable === "string" || iterable instanceof Buffer) {
    const whole = iterable;
    return new Readable({
      objectMode: true,
      ...options,
      read() {
        this.push(whole);
        this.push(null);
      },
    });
  }
  if (!isIterable(iterable)) {
    throw nodeError("ERR_INVALID_ARG_TYPE", `The "iterable" argument must be an iterable.${describeReceived(iterable)}`);
  }
  const iterator = typeof (iterable as AsyncIterable<unknown>)[Symbol.asyncIterator] === "function"
    ? (iterable as AsyncIterable<unknown>)[Symbol.asyncIterator]()
    : (iterable as Iterable<unknown>)[Symbol.iterator]();
  let reading = false;
  const readable: Readable = new Readable({
    objectMode: true,
    highWaterMark: 1,
    ...options,
    read() {
      if (reading) return;
      reading = true;
      (async () => {
        try {
          while (true) {
            const { value, done } = await (iterator as AsyncIterator<unknown>).next();
            if (done) {
              readable.push(null);
              return;
            }
            if (value === null) throw nodeError("ERR_STREAM_NULL_VALUES", "May not write null values to stream");
            if (!readable.push(value)) {
              reading = false;
              return;
            }
          }
        } catch (err) {
          readable.destroy(err as Error);
        }
      })();
    },
    destroy(error, callback) {
      (async () => {
        try {
          if (typeof (iterator as AsyncIterator<unknown>).return === "function") await (iterator as AsyncIterator<unknown>).return!();
        } finally {
          callback(error);
        }
      })();
    },
  });
  return readable;
}

/** Static Readable.wrap(stream): Node's own `new Readable({...}).wrap(stream)`. */
function readableWrapStatic(stream: EventEmitter & { pause(): void; resume(): void; readableObjectMode?: boolean; objectMode?: boolean; destroy?(error?: Error): unknown }, options?: ReadableOptions): Readable {
  return new Readable({
    objectMode: stream.readableObjectMode ?? stream.objectMode ?? true,
    ...options,
    destroy(error, callback) {
      if (typeof stream.destroy === "function") stream.destroy(error ?? undefined);
      callback(error);
    },
  }).wrap(stream);
}

/** A push/end queue that is itself an async iterable: the bridge a generator-function
 *  stage (Duplex.from, stream.compose) reads its writable side's input through. */
class AsyncQueue implements AsyncIterable<unknown> {
  #items: unknown[] = [];
  #waiters: Array<(result: IteratorResult<unknown>) => void> = [];
  #ended = false;
  #error: Error | null = null;

  push(value: unknown): void {
    const waiter = this.#waiters.shift();
    if (waiter) waiter({ value, done: false });
    else this.#items.push(value);
  }

  end(): void {
    this.#ended = true;
    while (this.#waiters.length) this.#waiters.shift()!({ value: undefined, done: true });
  }

  fail(error: Error): void {
    this.#error = error;
    this.end();
  }

  [Symbol.asyncIterator](): AsyncIterator<unknown> {
    return {
      next: (): Promise<IteratorResult<unknown>> => {
        if (this.#items.length) return Promise.resolve({ value: this.#items.shift(), done: false });
        if (this.#error) return Promise.reject(this.#error);
        if (this.#ended) return Promise.resolve({ value: undefined, done: true });
        return new Promise((resolve) => this.#waiters.push(resolve));
      },
    };
  }
}

type GeneratorStage = (source: AsyncIterable<unknown>) => AsyncIterable<unknown> | Iterable<unknown>;

/** Wraps an async-generator-style stage (what Duplex.from and stream.compose accept as
 *  a function) as a Duplex: writes feed the function's input, its output is pushed. */
function fromGeneratorFunction(fn: GeneratorStage): Duplex {
  const input = new AsyncQueue();
  const duplex: Duplex = new Duplex({
    objectMode: true,
    read() {},
    write(chunk, _encoding, callback) {
      input.push(chunk);
      callback();
    },
    final(callback) {
      input.end();
      callback();
    },
    destroy(error, callback) {
      if (error) input.fail(error);
      else input.end();
      callback(error);
    },
  });
  (async () => {
    try {
      const output = fn(input);
      for await (const chunk of output) duplex.push(chunk);
      duplex.push(null);
    } catch (err) {
      duplex.destroy(err as Error);
    }
  })();
  return duplex;
}

/** Wraps a plain (non-Duplex) Readable as a Duplex whose readable side pulls from it
 *  directly — the same 'readable'-event/read() pull Node's _duplexify and this module's
 *  own compose() use for a tail stage — instead of draining it through async iteration.
 *  Async iteration would add a microtask hop per chunk (the generator's internal await)
 *  that Node's real Duplex.from(readable) never takes, which is exactly the extra tick
 *  that showed up composing an already-flowing Readable: confirmed by comparing
 *  lib/internal/streams/duplexify.js's `isReadableNodeStream(body)` fast path (checked
 *  before its generic `isIterable(body)` one) against real Node 24.16.0. */
function duplexFromReadable(r: Readable): Duplex {
  let onreadable: (() => void) | null = null;
  let duplex!: Duplex;
  const pull = (): void => {
    for (;;) {
      const buf = r.read();
      if (buf === null) {
        onreadable = pull;
        return;
      }
      if (!duplex.push(buf)) return;
    }
  };
  duplex = new Duplex({ objectMode: r.readableObjectMode, read: pull });
  r.on("readable", () => {
    const cb = onreadable;
    onreadable = null;
    cb?.();
  });
  r.once("end", () => duplex.push(null));
  r.once("error", (err: Error) => duplex.destroy(err));
  // Node's own Duplexify(options.writable === false): the writable side is already
  // over, so a write to it fails the normal "write after end" way.
  duplex.end();
  return duplex;
}

/** Wraps a plain (non-Duplex) Writable as a Duplex whose writable side forwards
 *  directly to it; the readable side is already over (nothing to read), matching
 *  Node's Duplexify(options.readable === false). */
function duplexFromWritable(w: Writable): Duplex {
  const duplex: Duplex = new Duplex({
    objectMode: w.writableObjectMode,
    read() {},
    write(chunk, encoding, callback) {
      if (w.write(chunk, encoding)) callback();
      else w.once("drain", () => callback());
    },
    final(callback) {
      w.end();
      w.once("finish", () => callback());
    },
  });
  w.once("error", (err: Error) => duplex.destroy(err));
  duplex.push(null);
  return duplex;
}

/** Duplex.from(body): a stream passed through, a plain Readable/Writable wrapped
 *  pull/push-style (never through async iteration — see duplexFromReadable), a
 *  generator function wrapped via fromGeneratorFunction, a WHATWG { readable, writable }
 *  pair bridged natively, or an iterable exposed read-only through Readable.from (Node
 *  leaves its writable side dead). */
function duplexFrom(body: unknown): Duplex {
  if (body instanceof Duplex) return body;
  if (body instanceof Readable) return duplexFromReadable(body);
  if (body instanceof Writable) return duplexFromWritable(body);
  if (typeof body === "function") return fromGeneratorFunction(body as GeneratorStage);
  if (body && typeof body === "object" && "readable" in body && "writable" in body
    && (body as { readable: unknown }).readable instanceof ReadableStream && (body as { writable: unknown }).writable instanceof WritableStream) {
    return duplexFromWeb(body as { readable: ReadableStream; writable: WritableStream });
  }
  if (isIterable(body) || typeof body === "string" || body instanceof Buffer) {
    // Readable-only: Node exposes it as a Duplex but its writable side is already
    // ended, so a write fails the normal "write after end" way.
    const duplex = fromGeneratorFunction(() => (isIterable(body) ? (body as AsyncIterable<unknown>) : [body]));
    duplex.end();
    return duplex;
  }
  throw nodeError("ERR_INVALID_ARG_TYPE", `The "body" argument must be a Stream, a Function, an Iterable, an AsyncIterable or an Object.${describeReceived(body)}`);
}

/** stream.compose(...stages): chains streams and generator functions into one Duplex
 *  whose writable side feeds the first stage and readable side yields the last stage's
 *  output; any stage erroring destroys every other stage.
 *
 *  The readable side is wired as a *pull*, the way Node's own internal/streams/compose.js
 *  does it: `last`'s data is read via `last.read()` from inside `proxy`'s own `_read`,
 *  re-armed by `last`'s 'readable' event, never by listening for `last`'s 'data' event.
 *  This isn't a cosmetic choice: Node's Readable only emits 'data' synchronously off a
 *  push() when something is already flowing *with a 'data' listener attached*; a
 *  'readable' listener instead puts the stream in paused mode, so every chunk `last`
 *  pushes first goes through its push()'s buffered path, whose `emitReadable` is always
 *  `process.nextTick`-deferred (lib/internal/streams/readable.js). That hop is what lets
 *  other already-queued synchronous work (e.g. node:test's harness starting the next
 *  test, which runs its body synchronously up to its first await) run before a composed
 *  reporter's formatted line actually reaches its destination — confirmed against real
 *  Node 24.16.0 by diffing node:test's own `compose(rootReporter, reporter).pipe(dest)`
 *  wiring (lib/internal/test_runner/utils.js's setupTestReporters) against a guest that
 *  logs between two tests (tests/fixtures/guests/builtin-test-order.cjs). */
export function compose(...stages: unknown[]): Duplex {
  if (stages.length === 0) throw nodeError("ERR_MISSING_ARGS", 'The "streams" argument must be specified');
  // Node's compose() only runs a stage through Duplex.from when it's a function; a
  // stage that's already a Readable/Writable/Duplex is pipe()d as-is (lib/internal/
  // streams/compose.js builds its internal `pipeline(streams, onfinished)` straight
  // from the given stages, only substituting `Duplex.from(streams[n])` when
  // `typeof streams[n] === 'function'`). Wrapping an already-a-stream stage anyway —
  // which our own Duplex.from does, correctly, for *its* callers — routes an
  // already-flowing Readable's data through an extra push()/'readable' round trip
  // real Node never takes for a stage that's already a stream: confirmed by comparing
  // NODE_DEBUG=stream output for this guest's `compose(src, upper())` (src's "composed 2"
  // push reaches upper()'s own push synchronously in Node, never through a deferred
  // 'readable') against this file's prior behavior of always wrapping every stage.
  const streams = stages.map((s) => (s instanceof Duplex || s instanceof Readable || s instanceof Writable ? s : duplexFrom(s))) as unknown as Duplex[];
  const destroyAll = (err: Error): void => {
    for (const s of streams) if (!s.destroyed) s.destroy(err);
  };
  for (const s of streams) s.on("error", destroyAll);
  for (let i = 0; i < streams.length - 1; i++) streams[i]!.pipe(streams[i + 1] as unknown as PipeDestination);

  const first = streams[0]!;
  const last = streams[streams.length - 1]!;
  // Node's compose(): `writable`/`readable` come from what `head`/`tail` actually are,
  // not from wrapping them — a bare Readable passed in (readable.compose(genFn), or
  // compose(someReadable, ...)) has no writable side at all.
  const writableSide = first instanceof Writable;
  const readableSide = last instanceof Readable;
  let onreadable: (() => void) | null = null;
  let proxy!: Duplex;
  const pull = (): void => {
    for (;;) {
      const buf = last.read();
      if (buf === null) {
        onreadable = pull;
        return;
      }
      if (!proxy.push(buf)) return;
    }
  };
  proxy = new Duplex({
    writableObjectMode: (first as unknown as { writableObjectMode?: boolean }).writableObjectMode ?? false,
    readableObjectMode: (last as unknown as { readableObjectMode?: boolean }).readableObjectMode ?? false,
    write(chunk, encoding, callback) {
      (first as unknown as Writable).write(chunk, encoding, callback);
    },
    final(callback) {
      (first as unknown as Writable).end(callback);
    },
    read: readableSide ? pull : () => {},
    destroy(error, callback) {
      if (error) destroyAll(error);
      callback(error);
    },
  });
  // Node's Duplexify(options.writable === false / options.readable === false): force
  // that side into its already-finished state directly, the way a stage that was
  // never writable (or never readable) reports it — not through write()/push(), which
  // would call methods `first`/`last` don't have.
  if (!writableSide) {
    const ws = (proxy as unknown as { _writableState: WritableState })._writableState;
    ws.ending = true;
    ws.ended = true;
    ws.finished = true;
  }
  if (!readableSide) proxy.push(null);
  if (readableSide) {
    last.on("readable", () => {
      const cb = onreadable;
      onreadable = null;
      cb?.();
    });
    last.once("end", () => proxy.push(null));
  }
  last.once("error", (err: Error) => proxy.destroy(err));
  return proxy;
}

/** duplexPair(): two Duplexes wired so each side's writes arrive as the other's reads,
 *  and destroying one destroys the other (without propagating its error). */
export function duplexPair(options?: DuplexOptions): [Duplex, Duplex] {
  let a!: Duplex;
  let b!: Duplex;
  const side = (other: () => Duplex): Duplex => new Duplex({
    ...options,
    read() {},
    write(chunk, _encoding, callback) {
      other().push(chunk);
      callback();
    },
    final(callback) {
      other().push(null);
      callback();
    },
    destroy(error, callback) {
      // Finish this side's own close first: Node closes the side destroy() was called
      // on before the other side it cascades into.
      callback(error);
      const o = other();
      if (o && !o.destroyed) o.destroy();
    },
  });
  a = side(() => b);
  b = side(() => a);
  return [a, b];
}

/** stream.addAbortSignal(signal, stream): destroy `stream` (with an AbortError) when
 *  `signal` aborts, now if it already has. */
export function addAbortSignal<T extends EventEmitter>(signal: AbortSignal, stream: T): T {
  if (!(signal instanceof AbortSignal)) {
    throw nodeError("ERR_INVALID_ARG_TYPE", `The "signal" argument must be an instance of AbortSignal.${describeReceived(signal)}`);
  }
  const onAbort = (): void => {
    destroy(stream, abortError());
  };
  if (signal.aborted) nextTick(onAbort);
  else signal.addEventListener("abort", onAbort, { once: true });
  return stream;
}

function isFinishedSync(stream: unknown): boolean {
  const s = stream as { readableEnded?: boolean; writableFinished?: boolean };
  const isR = stream instanceof Readable;
  const isW = stream instanceof Writable;
  return (!isR || s.readableEnded === true) && (!isW || s.writableFinished === true);
}

/** stream.isDestroyed/isDisturbed/isErrored/isReadable/isWritable: read-only
 *  predicates npm's `is-stream`-style helpers and `eos` build on. */
export function isDestroyed(stream: unknown): boolean | null {
  if (!(stream instanceof Stream)) return null;
  return !!(stream as { destroyed?: boolean }).destroyed;
}

export function isDisturbed(stream: unknown): boolean {
  const s = stream as { readableDidRead?: boolean; readableAborted?: boolean } | null | undefined;
  return !!(s && (s.readableDidRead || s.readableAborted));
}

export function isErrored(stream: unknown): boolean {
  return !!(stream as { errored?: Error | null } | null | undefined)?.errored;
}

export function isReadable(stream: unknown): boolean | null {
  const s = stream as { readable?: boolean; readableEnded?: boolean } | null | undefined;
  if (typeof s?.readable !== "boolean") return null;
  if (isDestroyed(stream)) return false;
  return s.readable && !s.readableEnded;
}

export function isWritable(stream: unknown): boolean | null {
  const s = stream as { writable?: boolean; writableEnded?: boolean } | null | undefined;
  if (typeof s?.writable !== "boolean") return null;
  if (isDestroyed(stream)) return false;
  return s.writable && !s.writableEnded;
}

/** stream.destroy(stream[, error]): the destroyer every helper above (and Node's own
 *  eos/pipeline) closes a stream through. No error and not yet finished synthesizes an
 *  AbortError, so an early-abandoned stream reports *why* it stopped. */
export function destroy<T>(stream: T, error?: Error): T {
  const s = stream as unknown as { destroyed?: boolean; destroy?: (error?: Error) => unknown; close?: () => void } | null | undefined;
  if (!s || isDestroyed(stream)) return stream;
  const err = error ?? (isFinishedSync(stream) ? undefined : abortError());
  if (typeof s.destroy === "function") s.destroy(err);
  else if (typeof s.close === "function") s.close();
  return stream;
}

type WriteCallback = (error?: Error | null) => void;
type TransformCallback = (error?: Error | null, data?: unknown) => void;

export interface WritableOptions {
  highWaterMark?: number;
  objectMode?: boolean;
  decodeStrings?: boolean;
  defaultEncoding?: string;
  autoDestroy?: boolean;
  write?(chunk: unknown, encoding: string, callback: WriteCallback): void;
  final?(callback: WriteCallback): void;
  destroy?(error: Error | null, callback: (error?: Error | null) => void): void;
}

interface PendingWrite {
  chunk: unknown;
  encoding: string;
  callback: WriteCallback;
}

const noop = (): void => {};

/** Node's WritableState, kept on `_writableState` because npm code reads it (ws checks
 *  `finished`, `errorEmitted`, `needDrain` and `length`). */
export class WritableState {
  readonly objectMode: boolean;
  readonly highWaterMark: number;
  readonly decodeStrings: boolean;
  readonly autoDestroy: boolean;
  defaultEncoding: string;
  length = 0;
  writing = false;
  /** True while _write runs: a callback it calls synchronously defers its effects a tick. */
  sync = false;
  corked = 0;
  needDrain = false;
  ending = false;
  ended = false;
  finalCalled = false;
  finished = false;
  errorEmitted = false;
  buffered: PendingWrite[] = [];
  onFinished: WriteCallback[] = [];
  /** Whether a finished stream may auto-destroy: a duplex also waits for its 'end'. */
  readableDone: () => boolean = () => true;

  constructor(options: WritableOptions, objectMode: boolean, highWaterMark: number | undefined) {
    this.objectMode = objectMode;
    this.highWaterMark = highWaterMark ?? getDefaultHighWaterMark(objectMode);
    this.decodeStrings = options.decodeStrings ?? true;
    this.autoDestroy = options.autoDestroy ?? true;
    this.defaultEncoding = options.defaultEncoding ?? "utf8";
  }
}

/** What the writable state machine drives: a Writable, or a Duplex's writable side. */
interface WritableHost extends EventEmitter {
  readonly _writableState: WritableState;
  _write(chunk: unknown, encoding: string, callback: WriteCallback): void;
  _final?(callback: WriteCallback): void;
  readonly destroyed: boolean;
  readonly errored: Error | null;
  destroy(error?: Error | null): unknown;
}

function writeChunk(stream: WritableHost, chunk: unknown, encoding?: string | WriteCallback | null, callback?: WriteCallback): boolean {
  const state = stream._writableState;
  if (typeof encoding === "function") {
    callback = encoding;
    encoding = null;
  }
  const cb = callback ?? noop;
  if (chunk === null) throw nodeError("ERR_STREAM_NULL_VALUES", "May not write null values to stream");
  let enc = encoding ?? state.defaultEncoding;
  if (!state.objectMode) {
    if (typeof chunk === "string") {
      if (state.decodeStrings) {
        chunk = Buffer.from(chunk, enc);
        enc = "buffer";
      }
    } else if (chunk instanceof Uint8Array) {
      if (!(chunk instanceof Buffer)) chunk = new Buffer(chunk.buffer as ArrayBuffer, chunk.byteOffset, chunk.byteLength);
      enc = "buffer";
    } else {
      throw nodeError("ERR_INVALID_ARG_TYPE", 'The "chunk" argument must be of type string or an instance of Buffer or Uint8Array.');
    }
  }
  const err = state.ending
    ? nodeError("ERR_STREAM_WRITE_AFTER_END", "write after end")
    : stream.destroyed ? nodeError("ERR_STREAM_DESTROYED", "Cannot call write after a stream was destroyed") : null;
  if (err) {
    nextTick(() => cb(err));
    stream.destroy(err);
    return false;
  }
  const len = state.objectMode ? 1 : (chunk as Buffer | string).length;
  state.length += len;
  const ok = state.length < state.highWaterMark;
  if (!ok) state.needDrain = true;
  if (state.writing || state.corked > 0 || state.buffered.length > 0) state.buffered.push({ chunk, encoding: enc, callback: cb });
  else doWrite(stream, state, chunk, enc, cb);
  return ok && !stream.errored && !stream.destroyed;
}

function endWritable(stream: WritableHost, chunk?: unknown, encoding?: string | WriteCallback | null, callback?: WriteCallback): void {
  const state = stream._writableState;
  if (typeof chunk === "function") {
    callback = chunk as WriteCallback;
    chunk = null;
  } else if (typeof encoding === "function") {
    callback = encoding;
    encoding = null;
  }
  if (chunk !== null && chunk !== undefined) writeChunk(stream, chunk, encoding);
  if (state.corked > 0) {
    state.corked = 0;
    clearBuffer(stream, state);
  }
  if (!state.ending && !stream.errored) {
    state.ending = true;
    state.ended = true;
    finishMaybe(stream, state);
  }
  if (callback) {
    const cb = callback;
    if (state.finished) nextTick(() => cb(null));
    else if (stream.errored || stream.destroyed) nextTick(() => cb(stream.errored ?? nodeError("ERR_STREAM_DESTROYED", "Cannot call end after a stream was destroyed")));
    else state.onFinished.push(cb);
  }
}

const sizeOf = (state: WritableState, chunk: unknown): number => (state.objectMode ? 1 : (chunk as Buffer | string).length);

function doWrite(stream: WritableHost, state: WritableState, chunk: unknown, encoding: string, cb: WriteCallback): void {
  const len = sizeOf(state, chunk);
  state.writing = true;
  state.sync = true;
  stream._write(chunk, encoding, (err) => onWrite(stream, state, err ?? null, len, cb));
  state.sync = false;
}

function onWrite(stream: WritableHost, state: WritableState, err: Error | null, len: number, cb: WriteCallback): void {
  const sync = state.sync;
  state.writing = false;
  state.length -= len;
  if (err) {
    const fail = (): void => {
      cb(err);
      stream.destroy(err);
    };
    if (sync) nextTick(fail);
    else fail();
    return;
  }
  clearBuffer(stream, state);
  if (sync) nextTick(() => afterWrite(stream, state, cb));
  else afterWrite(stream, state, cb);
}

function clearBuffer(stream: WritableHost, state: WritableState): void {
  if (state.writing || state.corked > 0) return;
  const next = state.buffered.shift();
  if (next) doWrite(stream, state, next.chunk, next.encoding, next.callback);
}

function afterWrite(stream: WritableHost, state: WritableState, cb: WriteCallback): void {
  if (!state.ending && !stream.destroyed && state.length === 0 && state.needDrain) {
    state.needDrain = false;
    stream.emit("drain");
  }
  cb(null);
  finishMaybe(stream, state);
}

/** Once ended with nothing in flight: _final (when there is one), 'prefinish', then
 *  'finish' a tick later. */
function finishMaybe(stream: WritableHost, state: WritableState): void {
  if (!state.ending || state.finalCalled || state.writing || state.buffered.length > 0 || stream.destroyed || stream.errored) return;
  state.finalCalled = true;
  const prefinish = (): void => {
    stream.emit("prefinish");
    nextTick(() => finish(stream, state));
  };
  if (typeof stream._final !== "function") {
    prefinish();
    return;
  }
  stream._final((err) => {
    if (err) stream.destroy(err);
    else prefinish();
  });
}

function finish(stream: WritableHost, state: WritableState): void {
  if (stream.destroyed && stream.errored) return;
  state.finished = true;
  const callbacks = state.onFinished;
  state.onFinished = [];
  for (const cb of callbacks) cb(null);
  stream.emit("finish");
  if (state.autoDestroy && state.readableDone() && !stream.destroyed) stream.destroy();
}

/** A stream that error-destroys records it, as Node's errorOrDestroy does. */
function markErrorEmitted(stream: object): void {
  const state = (stream as { _writableState?: WritableState })._writableState;
  if (state) state.errorEmitted = true;
}

/** Node's Writable. Its methods read only `_writableState`, so Duplex shares them. */
export class Writable extends (Stream as unknown as EventEmitterConstructor) implements WritableHost {
  static fromWeb: typeof writableFromWeb;
  static toWeb: typeof writableToWeb;

  readonly _writableState: WritableState;
  #destroyed = false;
  #closed = false;
  #errored: Error | null = null;

  /** A Duplex is a Writable too, as in Node: its prototype chain runs through Readable. */
  static [Symbol.hasInstance](value: unknown): boolean {
    if (Function.prototype[Symbol.hasInstance].call(this, value)) return true;
    if (this !== Writable) return false;
    return (value as { _writableState?: unknown } | null)?._writableState instanceof WritableState;
  }

  constructor(options: WritableOptions = {}) {
    super();
    this._writableState = new WritableState(options, options.objectMode ?? false, options.highWaterMark);
    if (options.write) this._write = options.write;
    if (options.final) this._final = options.final;
    if (options.destroy) this._destroy = options.destroy;
  }

  get writable(): boolean {
    const state = this._writableState;
    return !this.destroyed && !this.errored && !state.ending;
  }
  get writableLength(): number {
    return this._writableState.length;
  }
  get writableEnded(): boolean {
    return this._writableState.ending;
  }
  get writableFinished(): boolean {
    return this._writableState.finished;
  }
  get writableNeedDrain(): boolean {
    return this._writableState.needDrain;
  }
  get writableHighWaterMark(): number {
    return this._writableState.highWaterMark;
  }
  get writableObjectMode(): boolean {
    return this._writableState.objectMode;
  }
  get writableCorked(): number {
    return this._writableState.corked;
  }
  /** True once destroy() has run and the stream never reached 'finish'. */
  get writableAborted(): boolean {
    return this.#destroyed && !this._writableState.finished;
  }
  /** Node's deprecated direct view of the pending writes (undocumented but present). */
  get writableBuffer(): PendingWrite[] {
    return this._writableState.buffered.slice();
  }
  get destroyed(): boolean {
    return this.#destroyed;
  }
  get closed(): boolean {
    return this.#closed;
  }
  get errored(): Error | null {
    return this.#errored;
  }

  _write(_chunk: unknown, _encoding: string, _callback: WriteCallback): void {
    throw nodeError("ERR_METHOD_NOT_IMPLEMENTED", "The _write() method is not implemented");
  }

  _final?(callback: WriteCallback): void;

  _destroy(error: Error | null, callback: (error?: Error | null) => void): void {
    callback(error);
  }

  write(chunk: unknown, encoding?: string | WriteCallback | null, callback?: WriteCallback): boolean {
    return writeChunk(this, chunk, encoding, callback);
  }

  end(chunk?: unknown, encoding?: string | WriteCallback | null, callback?: WriteCallback): this {
    endWritable(this, chunk, encoding, callback);
    return this;
  }

  /** Writes made while corked wait for uncork() (or end()), then go out in order. */
  cork(): void {
    this._writableState.corked++;
  }

  uncork(): void {
    const state = this._writableState;
    if (state.corked === 0) return;
    state.corked--;
    clearBuffer(this, state);
  }

  setDefaultEncoding(encoding: string): this {
    if (!Buffer.isEncoding(encoding)) throw Object.assign(new TypeError(`Unknown encoding: ${encoding}`), { code: "ERR_UNKNOWN_ENCODING" });
    this._writableState.defaultEncoding = encoding.toLowerCase();
    return this;
  }

  destroy(error?: Error | null, callback?: (error?: Error | null) => void): this {
    if (this.#destroyed) {
      if (callback) nextTick(() => callback(this.#errored));
      return this;
    }
    this.#destroyed = true;
    if (error) this.#errored = error;
    this._destroy(error ?? null, (err) => {
      if (err && !this.#errored) this.#errored = err;
      nextTick(() => {
        this.#closed = true;
        if (err) {
          markErrorEmitted(this);
          this.emit("error", err);
        }
        this.emit("close");
        callback?.(err);
      });
    });
    return this;
  }
}

/** Writable's members that only read `_writableState`; Duplex.prototype gets them. */
// Node's Duplex inherits writableBuffer but not writableAborted (checked against real
// Node; an asymmetry in Node's own class wiring, not ours to fix).
const SHARED_WRITABLE = [
  "writable", "writableLength", "writableEnded", "writableFinished", "writableNeedDrain", "writableHighWaterMark",
  "writableObjectMode", "writableCorked", "writableBuffer", "write", "end", "cork", "uncork", "setDefaultEncoding",
] as const;

export interface DuplexOptions extends ReadableOptions, Omit<WritableOptions, "destroy"> {
  /** False ends the writable side once the readable side ends (net.Socket's default). */
  allowHalfOpen?: boolean;
  readableHighWaterMark?: number;
  writableHighWaterMark?: number;
  readableObjectMode?: boolean;
  writableObjectMode?: boolean;
}

export interface Duplex extends Pick<Writable, (typeof SHARED_WRITABLE)[number]> {}

/** A Readable with Writable's state machine in front, as Node builds it: the writable
 *  members are Writable's own, and it closes once both 'end' and 'finish' happened. */
/** Duplex.from/fromWeb/toWeb: distinct signatures from Readable's own static
 *  fromWeb/toWeb, so they're assigned (not declared) below, past the class body. */
export interface DuplexConstructor {
  from: typeof duplexFrom;
  fromWeb: typeof duplexFromWeb;
  toWeb: typeof duplexToWeb;
}

export class Duplex extends Readable implements WritableHost {
  readonly _writableState: WritableState;
  allowHalfOpen: boolean;

  constructor(options: DuplexOptions = {}) {
    const {
      write, final, allowHalfOpen, readableHighWaterMark, writableHighWaterMark, readableObjectMode, writableObjectMode,
      decodeStrings, defaultEncoding, ...readable
    } = options;
    const objectMode = options.objectMode ?? false;
    super({ ...readable, objectMode: readableObjectMode ?? objectMode, highWaterMark: readableHighWaterMark ?? options.highWaterMark });
    this._writableState = new WritableState(
      { decodeStrings, defaultEncoding, autoDestroy: options.autoDestroy },
      writableObjectMode ?? objectMode,
      writableHighWaterMark ?? options.highWaterMark,
    );
    this._writableState.readableDone = () => this.readableEnded;
    readableInternals.setCanAutoDestroy(this, () => this._writableState.finished);
    this.allowHalfOpen = allowHalfOpen ?? true;
    if (write) this._write = write;
    if (final) this._final = final;
    // Node's onReadableEnd: without half-open, the readable side's end ends the writable.
    this.once("end", () => {
      if (!this.allowHalfOpen && !this._writableState.ending) nextTick(() => this.end());
    });
  }

  _write(_chunk: unknown, _encoding: string, _callback: WriteCallback): void {
    throw nodeError("ERR_METHOD_NOT_IMPLEMENTED", "The _write() method is not implemented");
  }

  _final?(callback: WriteCallback): void;
}

for (const key of SHARED_WRITABLE) {
  Object.defineProperty(Duplex.prototype, key, Object.getOwnPropertyDescriptor(Writable.prototype, key)!);
}

export interface TransformOptions extends Omit<DuplexOptions, "read" | "write" | "final"> {
  transform?(this: Transform, chunk: unknown, encoding: string, callback: TransformCallback): void;
  flush?(this: Transform, callback: TransformCallback): void;
}

/** A Duplex whose writes become reads: a write whose output fills the readable side
 *  holds its callback until the next _read, and end() flushes through _flush. */
export class Transform extends Duplex {
  /** Node's kCallback: the write callback held while the readable side is full. */
  #held: (() => void) | null = null;

  constructor(options: TransformOptions = {}) {
    const { transform, flush, ...duplex } = options;
    super(duplex);
    if (transform) this._transform = transform;
    if (flush) this._flush = flush;
    readableInternals.clearSync(this);
    // Node's Transform prefinish: a subclass that replaces _final (zlib does) flushes
    // here instead, so its 'finish' can come before its output ends.
    this.on("prefinish", () => {
      if (this._final !== Transform.prototype._final) this.#flushThenEnd();
    });
  }

  _transform(_chunk: unknown, _encoding: string, _callback: TransformCallback): void {
    throw nodeError("ERR_METHOD_NOT_IMPLEMENTED", "The _transform() method is not implemented");
  }

  _flush?(callback: TransformCallback): void;

  override _read(_size: number): void {
    const held = this.#held;
    this.#held = null;
    held?.();
  }

  override _write(chunk: unknown, encoding: string, callback: WriteCallback): void {
    const before = this.readableLength;
    this._transform(chunk, encoding, (err, data) => {
      if (err) return callback(err);
      if (data != null) this.push(data);
      if (this._writableState.ending || before === this.readableLength || this.readableLength < this.readableHighWaterMark) callback();
      else this.#held = () => callback();
    });
  }

  /** Node's Transform final: flush, push what it returns, then EOF. */
  override _final(callback: WriteCallback): void {
    this.#flushThenEnd(callback);
  }

  #flushThenEnd(callback?: WriteCallback): void {
    const done: TransformCallback = (err, data) => {
      if (err) {
        if (callback) callback(err);
        else this.destroy(err);
        return;
      }
      if (data != null) this.push(data);
      this.push(null);
      callback?.();
    };
    if (this._flush && !this.destroyed) this._flush(done);
    else done();
  }
}

/** A Transform that passes every chunk through untouched. */
export class PassThrough extends Transform {
  constructor(options?: TransformOptions) {
    super({ ...options, transform(chunk, _encoding, callback) { callback(null, chunk); } });
  }
}

/** Wraps a callback so only its first call takes effect, as Node's `once` util does
 *  for every completion callback here (a stream can fire 'error' then 'close'). */
function once<T extends (...args: any[]) => void>(fn: T): T {
  let called = false;
  return ((...args: unknown[]) => {
    if (called) return;
    called = true;
    fn(...args);
  }) as T;
}

export interface FinishedOptions {
  error?: boolean;
  readable?: boolean;
  writable?: boolean;
  signal?: AbortSignal;
}
type FinishedCallback = (error?: Error | null) => void;

/** stream.finished: resolves once `stream` is fully done (both sides, when both apply),
 *  or reports why it stopped early. Node's eos checks the stream's *current* state
 *  first — errored, closed/destroyed, or already finished — and still calls back
 *  (on a later tick) for a stream that got there before this call. */
export function finished(stream: EventEmitter, options: FinishedOptions | FinishedCallback, callback?: FinishedCallback): () => void {
  if (typeof options === "function") {
    callback = options;
    options = {};
  }
  if (typeof callback !== "function") {
    throw nodeError("ERR_INVALID_ARG_TYPE", 'The "callback" argument must be of type function. Received undefined');
  }
  const opts = options as FinishedOptions;
  const checkError = opts.error ?? true;
  const isReadable = stream instanceof Readable;
  const isWritable = stream instanceof Writable;
  const wantReadable = opts.readable ?? isReadable;
  const wantWritable = opts.writable ?? isWritable;
  const s = stream as unknown as { readableEnded?: boolean; writableFinished?: boolean; errored?: Error | null; closed?: boolean; destroyed?: boolean };

  let readableEnded = !wantReadable || (isReadable && (stream as Readable).readableEnded);
  let writableFinished = !wantWritable || (isWritable && (stream as unknown as Writable).writableFinished);

  let cancelled = false;
  // Node calls the success path with no arguments at all, so `err` reads as
  // `undefined` (not `null`) to anything that logs or inspects it.
  const cb = once((err?: Error) => {
    if (cancelled) return;
    cleanup();
    if (err) callback!(err);
    else callback!();
  });

  // Already errored, closed or destroyed before this call: Node reports it
  // on a later tick instead of waiting for events that already happened.
  const existingError = checkError ? (s.errored ?? null) : null;
  if (existingError) {
    nextTick(() => cb(existingError));
  } else if (s.closed || s.destroyed) {
    const premature = (wantReadable && !readableEnded) || (wantWritable && !writableFinished);
    nextTick(() => cb(premature ? nodeError("ERR_STREAM_PREMATURE_CLOSE", "Premature close") : undefined));
  } else if (readableEnded && writableFinished) {
    nextTick(() => cb());
  }

  function maybeDone(): void {
    if (readableEnded && writableFinished) cb();
  }
  function onend(): void {
    readableEnded = true;
    maybeDone();
  }
  function onfinish(): void {
    writableFinished = true;
    maybeDone();
  }
  function onerror(err: Error): void {
    if (checkError) cb(err);
    else cb();
  }
  function onclose(): void {
    if (!readableEnded || !writableFinished) cb(nodeError("ERR_STREAM_PREMATURE_CLOSE", "Premature close"));
    else cb();
  }
  function onabort(): void {
    cb(nodeError("ABORT_ERR", "The operation was aborted"));
  }

  stream.on("end", onend);
  stream.on("finish", onfinish);
  stream.on("error", onerror);
  stream.on("close", onclose);
  const signal = opts.signal;
  if (signal) {
    if (signal.aborted) nextTick(onabort);
    else signal.addEventListener("abort", onabort);
  }

  function cleanup(): void {
    stream.off("end", onend);
    stream.off("finish", onfinish);
    stream.off("error", onerror);
    stream.off("close", onclose);
    if (signal) signal.removeEventListener("abort", onabort);
  }

  return () => {
    cancelled = true;
    cleanup();
  };
}

type PipelineCallback = (error?: Error | null) => void;

/** stream.pipeline: pipe()s each stream into the next, destroys every stream the
 *  moment any of them errors or closes early, and calls `callback` exactly once.
 *  Stages are stream instances only (an array, or spread arguments) — not yet the
 *  async-iterable/generator-function stages Node also accepts. */
export function pipeline(...args: unknown[]): EventEmitter {
  const last = args[args.length - 1];
  if (typeof last !== "function") {
    // Node's literal (unsubstituted) property name, down to the typo.
    throw nodeError("ERR_INVALID_ARG_TYPE", `The "streams[stream.length - 1]" property must be of type function.${describeReceived(last)}`);
  }
  const callback = args.pop() as PipelineCallback;
  const streams = (args.length === 1 && Array.isArray(args[0]) ? args[0] : args) as EventEmitter[];
  if (streams.length < 2) {
    throw nodeError("ERR_MISSING_ARGS", 'The "streams" argument must be specified');
  }
  for (const s of streams) {
    if (!(s instanceof EventEmitter) || (typeof (s as any).pipe !== "function" && typeof (s as any).write !== "function")) {
      throw nodeError("ERR_INVALID_ARG_TYPE", 'The "streams[...]" argument must be of type Stream.');
    }
  }

  const cb = once(callback);
  let settled = false;
  let pending = streams.length;
  const cleanups = streams.map((s) =>
    finished(s, { error: true }, (err) => {
      if (settled) return;
      if (err) {
        settled = true;
        // destroy() emits 'error'/'close' a tick later, so keep each finished()
        // listener alive until then — removing it now would leave that emit with
        // no listener and throw uncaught (Node's default EventEmitter behavior).
        for (const other of streams) if (!(other as any).destroyed) (other as any).destroy(err);
        nextTick(() => { for (const c of cleanups) c(); });
        cb(err);
        return;
      }
      pending--;
      if (pending === 0) {
        settled = true;
        for (const c of cleanups) c();
        cb();
      }
    }),
  );

  for (let i = 0; i < streams.length - 1; i++) {
    (streams[i] as unknown as Readable).pipe(streams[i + 1] as unknown as PipeDestination);
  }

  return streams[streams.length - 1]!;
}

export const promises = {
  finished(stream: EventEmitter, options?: FinishedOptions): Promise<void> {
    return new Promise((resolve, reject) => {
      finished(stream, options ?? {}, (err) => (err ? reject(err) : resolve()));
    });
  },
  pipeline(...streams: unknown[]): Promise<void> {
    return new Promise((resolve, reject) => {
      pipeline(...streams, (err?: Error | null) => (err ? reject(err) : resolve()));
    });
  },
};

/** Readable.toWeb: a byte ReadableStream that pauses the source under backpressure. */
function toWeb(readable: Readable): ReadableStream<Uint8Array> {
  let controller!: ReadableStreamDefaultController<Uint8Array>;
  const onData = (chunk: unknown): void => {
    const bytes = typeof chunk === "string" ? Buffer.from(chunk) : (chunk as Uint8Array);
    controller.enqueue(new Uint8Array(bytes));
    if ((controller.desiredSize ?? 1) <= 0) readable.pause();
  };
  return new ReadableStream<Uint8Array>({
    start(c) {
      controller = c;
      readable.on("data", onData);
      readable.once("end", () => controller.close());
      readable.once("error", (e: unknown) => controller.error(e));
    },
    pull() {
      readable.resume();
    },
    cancel(reason) {
      readable.off("data", onData);
      readable.destroy(reason instanceof Error ? reason : null);
    },
  });
}

/** Readable.fromWeb: each _read pulls one chunk from the web stream's reader. */
function fromWeb(stream: ReadableStream, options: ReadableOptions = {}): Readable {
  const reader = stream.getReader();
  return new Readable({
    ...options,
    read() {
      reader.read().then(
        ({ done, value }) => {
          this.push(done ? null : value);
        },
        (e: Error) => this.destroy(e),
      );
    },
    destroy(error, callback) {
      reader.cancel(error ?? undefined).then(() => callback(error), (e: Error) => callback(e));
    },
  });
}

/** Writable.toWeb: a byte sink that writes/closes/aborts through the Writable's own
 *  state machine, so backpressure and 'finish'/'close' still come from one place. */
function writableToWeb(writable: Writable): WritableStream<Uint8Array> {
  return new WritableStream<Uint8Array>({
    write(chunk) {
      return new Promise<void>((resolve, reject) => {
        writable.write(chunk, (err) => (err ? reject(err) : resolve()));
      });
    },
    close() {
      return new Promise<void>((resolve, reject) => {
        writable.end((err?: Error | null) => (err ? reject(err) : resolve()));
      });
    },
    abort(reason) {
      writable.destroy(reason instanceof Error ? reason : undefined);
    },
  });
}

/** Writable.fromWeb: each write/end/destroy goes through the web stream's writer. */
function writableFromWeb(writableStream: WritableStream, options: WritableOptions = {}): Writable {
  const writer = writableStream.getWriter();
  return new Writable({
    ...options,
    write(chunk, encoding, callback) {
      const bytes = typeof chunk === "string" ? Buffer.from(chunk, encoding) : (chunk as Uint8Array);
      writer.write(bytes).then(() => callback(), (e: Error) => callback(e));
    },
    final(callback) {
      writer.close().then(() => callback(), (e: Error) => callback(e));
    },
    destroy(error, callback) {
      writer.abort(error ?? undefined).then(() => callback(error), (e: Error) => callback(e));
    },
  });
}

/** Duplex.fromWeb: bridges both sides of a { readable, writable } web pair at once. */
function duplexFromWeb(pair: { readable: ReadableStream; writable: WritableStream }, options: DuplexOptions = {}): Duplex {
  const reader = pair.readable.getReader();
  const writer = pair.writable.getWriter();
  return new Duplex({
    ...options,
    read() {
      reader.read().then(
        ({ done, value }) => this.push(done ? null : value),
        (e: Error) => this.destroy(e),
      );
    },
    write(chunk, encoding, callback) {
      const bytes = typeof chunk === "string" ? Buffer.from(chunk, encoding) : (chunk as Uint8Array);
      writer.write(bytes).then(() => callback(), (e: Error) => callback(e));
    },
    final(callback) {
      writer.close().then(() => callback(), (e: Error) => callback(e));
    },
    destroy(error, callback) {
      Promise.allSettled([reader.cancel(error ?? undefined), writer.abort(error ?? undefined)]).then(() => callback(error));
    },
  });
}

/** Duplex.toWeb: the readable side via Readable.toWeb, the writable side via
 *  Writable.toWeb — both driven off the same Duplex. */
function duplexToWeb(duplex: Duplex): { readable: ReadableStream<Uint8Array>; writable: WritableStream<Uint8Array> } {
  return { readable: toWeb(duplex), writable: writableToWeb(duplex as unknown as Writable) };
}

/** The legacy (pre-streams3) Stream.prototype.pipe: any EventEmitter that emits
 *  'data'/'end' and exposes pause()/resume(), piped into any writable-shaped emitter.
 *  Readable overrides this with its own (identical in spirit, streams3-aware) pipe. */
type LegacyPipe = (this: EventEmitter & { pause?(): void; resume?(): void }, dest: PipeDestination, options?: { end?: boolean }) => PipeDestination;
(Stream.prototype as unknown as { pipe: LegacyPipe }).pipe = function (this: EventEmitter & { pause?(): void; resume?(): void }, dest: PipeDestination, options?: { end?: boolean }): PipeDestination {
  const source = this;
  let didOnEnd = false;
  const onData = (chunk: unknown): void => {
    if ((dest as unknown as { writable?: boolean }).writable !== false && dest.write(chunk) === false && source.pause) source.pause();
  };
  const onDrain = (): void => {
    if (source.resume) source.resume();
  };
  const onEnd = (): void => {
    if (didOnEnd) return;
    didOnEnd = true;
    dest.end();
  };
  const onClose = (): void => {
    if (didOnEnd) return;
    didOnEnd = true;
    if (typeof (dest as unknown as { destroy?(): void }).destroy === "function") (dest as unknown as { destroy(): void }).destroy();
  };
  const onError = (err: unknown): void => {
    cleanup();
    if (source.listenerCount("error") === 0) throw err;
  };
  source.on("data", onData);
  dest.on("drain", onDrain);
  if (!options || options.end !== false) {
    source.on("end", onEnd);
    source.on("close", onClose);
  }
  source.once("error", onError);
  dest.once("error", onError);
  source.once("end", cleanup);
  source.once("close", cleanup);
  dest.once("close", cleanup);
  function cleanup(): void {
    source.removeListener("data", onData);
    dest.removeListener("drain", onDrain);
    source.removeListener("end", onEnd);
    source.removeListener("close", onClose);
    source.removeListener("error", onError);
    dest.removeListener("error", onError);
  }
  dest.emit("pipe", source);
  return dest;
};

// ReadableStream.from(iterable) is missing from the Worker global in the Chromium
// build this kernel currently targets (present on `window`, absent on
// DedicatedWorkerGlobalScope) even though it's a plain WHATWG static. It's small
// enough to patch onto the native constructor rather than carry a todo for it.
if (typeof (ReadableStream as unknown as { from?: unknown }).from !== "function") {
  (ReadableStream as unknown as { from(source: unknown): ReadableStream }).from = function (source: unknown): ReadableStream {
    const iterator = isIterable(source)
      ? (typeof (source as AsyncIterable<unknown>)[Symbol.asyncIterator] === "function" ? (source as AsyncIterable<unknown>)[Symbol.asyncIterator]() : (source as Iterable<unknown>)[Symbol.iterator]())
      : undefined;
    if (!iterator) throw new TypeError("The provided value is not iterable");
    return new ReadableStream({
      async pull(controller) {
        const { value, done } = await (iterator as AsyncIterator<unknown>).next();
        if (done) controller.close();
        else controller.enqueue(value);
      },
      async cancel(reason) {
        await (iterator as AsyncIterator<unknown>).return?.(reason);
      },
    });
  };
}

/** stream/web: Node re-exports the platform's own WHATWG stream constructors here;
 *  so do we, straight off globalThis. */
export const streamWeb = {
  ReadableStream,
  ReadableStreamDefaultReader,
  ReadableStreamBYOBReader,
  ReadableStreamBYOBRequest,
  ReadableByteStreamController,
  ReadableStreamDefaultController,
  WritableStream,
  WritableStreamDefaultWriter,
  WritableStreamDefaultController,
  TransformStream,
  TransformStreamDefaultController,
  ByteLengthQueuingStrategy,
  CountQueuingStrategy,
  TextEncoderStream,
  TextDecoderStream,
  CompressionStream,
  DecompressionStream,
};

/** stream/consumers: drain a Readable/ReadableStream/async-iterable into bytes (via
 *  this Readable's own async iteration, already reliable), then hand that buffer to a
 *  Response so decoding (text/json) and the typed results (Blob/ArrayBuffer) come
 *  from the platform. Not Response fed the *live* stream: in this kernel's Worker,
 *  Response's internal consumption of a hand-built pull-based ReadableStream never
 *  settles its last read, which leaves the guest unable to exit even once the value
 *  it's waiting on has resolved. A plain buffer has no such read to hang on. */
async function collectBytes(stream: unknown): Promise<Uint8Array> {
  if (stream instanceof Readable) {
    const chunks: Buffer[] = [];
    for await (const chunk of stream) chunks.push(typeof chunk === "string" ? Buffer.from(chunk) : Buffer.from(chunk as Uint8Array));
    return Buffer.concat(chunks);
  }
  if (stream instanceof ReadableStream) {
    const reader = stream.getReader();
    const chunks: Uint8Array[] = [];
    for (let r = await reader.read(); !r.done; r = await reader.read()) chunks.push(r.value as Uint8Array);
    return Buffer.concat(chunks.map((c) => Buffer.from(c)));
  }
  if (isIterable(stream)) return collectBytes(readableFrom(stream));
  throw nodeError("ERR_INVALID_ARG_TYPE", `The "stream" argument must be an instance of Readable, ReadableStream, or an AsyncIterator.${describeReceived(stream)}`);
}

export const streamConsumers = {
  async buffer(stream: unknown): Promise<Buffer> {
    return Buffer.from(await collectBytes(stream));
  },
  async text(stream: unknown): Promise<string> {
    return new Response(await collectBytes(stream) as BodyInit).text();
  },
  async json(stream: unknown): Promise<unknown> {
    return new Response(await collectBytes(stream) as BodyInit).json();
  },
  async arrayBuffer(stream: unknown): Promise<ArrayBuffer> {
    return new Response(await collectBytes(stream) as BodyInit).arrayBuffer();
  },
  async blob(stream: unknown): Promise<Blob> {
    return new Response(await collectBytes(stream) as BodyInit).blob();
  },
  async bytes(stream: unknown): Promise<Uint8Array> {
    return new Uint8Array(await collectBytes(stream));
  },
};

Readable.toWeb = toWeb;
Readable.fromWeb = fromWeb;
Readable.from = readableFrom;
Readable.wrap = readableWrapStatic;
Writable.fromWeb = writableFromWeb;
Writable.toWeb = writableToWeb;
(Duplex as unknown as DuplexConstructor).from = duplexFrom;
(Duplex as unknown as DuplexConstructor).fromWeb = duplexFromWeb;
(Duplex as unknown as DuplexConstructor).toWeb = duplexToWeb;
Stream.Stream = Stream;
Stream.Readable = Readable;
Stream.Writable = Writable;
Stream.Duplex = Duplex;
Stream.Transform = Transform;
Stream.PassThrough = PassThrough;
Stream.finished = finished;
Stream.pipeline = pipeline;
Stream.promises = promises;
Stream.addAbortSignal = addAbortSignal;
Stream.compose = compose;
Stream.destroy = destroy;
Stream.duplexPair = duplexPair;
Stream.getDefaultHighWaterMark = getDefaultHighWaterMark;
Stream.setDefaultHighWaterMark = setDefaultHighWaterMark;
Stream.isDestroyed = isDestroyed;
Stream.isDisturbed = isDisturbed;
Stream.isErrored = isErrored;
Stream.isReadable = isReadable;
Stream.isWritable = isWritable;