deno_node 0.192.0

// deno-lint-ignore-file
// Copyright 2018-2026 the Deno authors. MIT license.
// This file is derived from the original SonicBoom module
// MIT License
// Copyright (c) 2017 Matteo Collina

import { core, primordials } from "ext:core/mod.js";
const { Buffer } = core.loadExtScript("ext:deno_node/internal/buffer.mjs");
const { EventEmitter } = core.loadExtScript("ext:deno_node/_events.mjs");
// Lazy node:fs import: this module is loaded via the fs.Utf8Stream getter,
// which runs during node:fs's own body evaluation. A static `import * as fs
// from "node:fs"` here re-enters the still-loading node:fs namespace, whose
// `export const Utf8Stream = mod.Utf8Stream` line then re-triggers the
// getter while `lazyUtf8Stream().default` is in TDZ. Defer until first use.
const lazyFs = core.createLazyLoader("node:fs");
import * as path from "node:path";
import process from "node:process";
import { setInterval, setTimeout } from "node:timers";
const {
  ERR_INVALID_ARG_TYPE,
  ERR_INVALID_ARG_VALUE,
  ERR_INVALID_STATE,
} = core.loadExtScript("ext:deno_node/internal/errors.ts");
const {
  validateBoolean,
  validateFunction,
  validateObject,
  validateOneOf,
  validateString,
  validateUint32,
} = core.loadExtScript("ext:deno_node/internal/validators.mjs");

const {
  ArrayPrototypePush,
  MathMax,
  Number,
  ObjectAssign,
  SymbolDispose,
} = primordials;

const BUSY_WRITE_TIMEOUT = 100;
const kEmptyBuffer = Buffer.allocUnsafe(0);

// Sleep helper used in sync busy-write retries. In Node this uses
// Atomics.wait on a SharedArrayBuffer; here we fall back to a busy spin
// because the test suite does not exercise EAGAIN/EBUSY paths and a real
// SAB requires --unstable.
function sleep(ms) {
  const valid = ms > 0 && ms < Infinity;
  if (valid === false) {
    if (typeof ms !== "number" && typeof ms !== "bigint") {
      throw new ERR_INVALID_ARG_TYPE("ms", ["number", "bigint"], ms);
    }
    throw new ERR_INVALID_ARG_VALUE.RangeError(
      "ms",
      ms,
      "must be a number greater than 0 and less than Infinity",
    );
  }
  const target = Date.now() + Number(ms);
  while (Date.now() < target) { /* spin */ }
}

const kMaxWrite = 16 * 1024;
const kContentModeBuffer = "buffer";
const kContentModeUtf8 = "utf8";
const kNullPrototype = { __proto__: null };

class Utf8Stream extends EventEmitter {
  #len = 0;
  #fd = -1;
  #bufs = [];
  #lens = [];
  #writing = false;
  #ending = false;
  #reopening = false;
  #asyncDrainScheduled = false;
  #flushPending = false;
  #hwm = 16387;
  #file = null;
  #destroyed = false;
  #minLength = 0;
  #maxLength = 0;
  #maxWrite = kMaxWrite;
  #opening = false;
  #periodicFlush = 0;
  #periodicFlushTimer = undefined;
  #sync = false;
  #fsync = false;
  #append = true;
  #mode = 0o666;
  #retryEAGAIN = () => true;
  #mkdir = false;
  #writingBuf = "";
  #write;
  #flush;
  #flushSync;
  #actualWrite;
  #fsWriteSync;
  #fsWrite;
  #fs;

  constructor(options = kNullPrototype) {
    validateObject(options, "options");
    let { fd } = options;
    const {
      dest,
      minLength,
      maxLength,
      maxWrite,
      periodicFlush,
      sync,
      append = true,
      mkdir,
      retryEAGAIN,
      fsync,
      contentMode = kContentModeUtf8,
      mode,
      fs: overrideFs = {},
    } = options;

    super();

    fd ??= dest;

    validateObject(overrideFs, "options.fs");
    this.#fs = ObjectAssign({}, lazyFs(), overrideFs);
    validateFunction(this.#fs.write, "options.fs.write");
    validateFunction(this.#fs.writeSync, "options.fs.writeSync");
    validateFunction(this.#fs.fsync, "options.fs.fsync");
    validateFunction(this.#fs.fsyncSync, "options.fs.fsyncSync");
    validateFunction(this.#fs.close, "options.fs.close");
    validateFunction(this.#fs.open, "options.fs.open");
    validateFunction(this.#fs.mkdir, "options.fs.mkdir");
    validateFunction(this.#fs.mkdirSync, "options.fs.mkdirSync");

    this.#hwm = MathMax(minLength || 0, this.#hwm);
    this.#minLength = minLength || 0;
    this.#maxLength = maxLength || 0;
    this.#maxWrite = maxWrite || kMaxWrite;
    this.#periodicFlush = periodicFlush || 0;
    this.#sync = sync || false;
    this.#fsync = fsync || false;
    this.#append = append || false;
    this.#mode = mode;
    this.#retryEAGAIN = retryEAGAIN || (() => true);
    this.#mkdir = mkdir || false;

    validateUint32(this.#hwm, "options.hwm");
    validateUint32(this.#minLength, "options.minLength");
    validateUint32(this.#maxLength, "options.maxLength");
    validateUint32(this.#maxWrite, "options.maxWrite");
    validateUint32(this.#periodicFlush, "options.periodicFlush");
    validateBoolean(this.#sync, "options.sync");
    validateBoolean(this.#fsync, "options.fsync");
    validateBoolean(this.#append, "options.append");
    validateBoolean(this.#mkdir, "options.mkdir");
    validateFunction(this.#retryEAGAIN, "options.retryEAGAIN");
    validateOneOf(contentMode, "options.contentMode", [
      kContentModeBuffer,
      kContentModeUtf8,
    ]);

    if (contentMode === kContentModeBuffer) {
      this.#writingBuf = kEmptyBuffer;
      this.#write = (...args) => this.#writeBuffer(...args);
      this.#flush = (...args) => this.#flushBuffer(...args);
      this.#flushSync = (...args) => this.#flushBufferSync(...args);
      this.#actualWrite = (...args) => this.#actualWriteBuffer(...args);
      this.#fsWriteSync = () => this.#fs.writeSync(this.#fd, this.#writingBuf);
      this.#fsWrite = () =>
        this.#fs.write(this.#fd, this.#writingBuf, (...args) => {
          return this.#release(...args);
        });
    } else {
      this.#writingBuf = "";
      this.#write = (...args) => this.#writeUtf8(...args);
      this.#flush = (...args) => this.#flushUtf8(...args);
      this.#flushSync = (...args) => this.#flushSyncUtf8(...args);
      this.#actualWrite = (...args) => this.#actualWriteUtf8(...args);
      this.#fsWriteSync = () =>
        this.#fs.writeSync(this.#fd, this.#writingBuf, "utf8");
      this.#fsWrite = () =>
        this.#fs.write(this.#fd, this.#writingBuf, "utf8", (...args) => {
          return this.#release(...args);
        });
    }

    if (typeof fd === "number") {
      this.#fd = fd;
      process.nextTick(() => this.emit("ready"));
    } else if (typeof fd === "string") {
      this.#openFile(fd);
    } else {
      throw new ERR_INVALID_ARG_TYPE("fd", ["number", "string"], fd);
    }
    if (this.#minLength >= this.#maxWrite) {
      throw new ERR_INVALID_ARG_VALUE.RangeError(
        "minLength",
        this.#minLength,
        `should be smaller than maxWrite (${this.#maxWrite})`,
      );
    }

    this.on("newListener", (name) => {
      if (name === "drain") {
        this.#asyncDrainScheduled = false;
      }
    });

    if (this.#periodicFlush !== 0) {
      this.#periodicFlushTimer = setInterval(
        () => this.flush(null),
        this.#periodicFlush,
      );
      this.#periodicFlushTimer.unref();
    }
  }

  write(data) {
    return this.#write(data);
  }

  flush(cb = function (_err) {}) {
    this.#flush(cb);
  }

  flushSync() {
    return this.#flushSync();
  }

  reopen(file) {
    if (this.#destroyed) {
      throw new ERR_INVALID_STATE("Utf8Stream is destroyed");
    }

    if (this.#opening) {
      this.once("ready", () => this.reopen(file));
      return;
    }

    if (this.#ending) {
      return;
    }

    if (!this.#file) {
      throw new ERR_INVALID_STATE(
        "Unable to reopen a file descriptor, you must pass a file to Utf8Stream",
      );
    }

    if (file) {
      this.#file = file;
    }
    this.#reopening = true;

    if (this.#writing) {
      return;
    }

    const fd = this.#fd;
    this.once("ready", () => {
      if (fd !== this.#fd) {
        this.#fs.close(fd, (err) => {
          if (err) {
            return this.emit("error", err);
          }
        });
      }
    });

    this.#openFile(this.#file);
  }

  end() {
    if (this.#destroyed) {
      throw new ERR_INVALID_STATE("Utf8Stream is destroyed");
    }

    if (this.#opening) {
      this.once("ready", () => {
        this.end();
      });
      return;
    }

    if (this.#ending) {
      return;
    }

    this.#ending = true;

    if (this.#writing) {
      return;
    }

    if (this.#len > 0 && this.#fd >= 0) {
      this.#actualWrite();
    } else {
      this.#actualClose();
    }
  }

  destroy() {
    if (this.#destroyed) {
      return;
    }
    this.#actualClose();
  }

  get mode() {
    return this.#mode;
  }
  get file() {
    return this.#file;
  }
  get fd() {
    return this.#fd;
  }
  get minLength() {
    return this.#minLength;
  }
  get maxLength() {
    return this.#maxLength;
  }
  get writing() {
    return this.#writing;
  }
  get sync() {
    return this.#sync;
  }
  get fsync() {
    return this.#fsync;
  }
  get append() {
    return this.#append;
  }
  get periodicFlush() {
    return this.#periodicFlush;
  }
  get contentMode() {
    return this.#writingBuf instanceof Buffer
      ? kContentModeBuffer
      : kContentModeUtf8;
  }
  get mkdir() {
    return this.#mkdir;
  }

  [SymbolDispose]() {
    this.destroy();
  }

  #release(err, n) {
    if (err) {
      if (
        (err.code === "EAGAIN" || err.code === "EBUSY") &&
        this.#retryEAGAIN(
          err,
          this.#writingBuf.length,
          this.#len - this.#writingBuf.length,
        )
      ) {
        if (this.#sync) {
          try {
            sleep(BUSY_WRITE_TIMEOUT);
            this.#release(undefined, 0);
          } catch (err) {
            this.#release(err);
          }
        } else {
          setTimeout(() => this.#fsWrite(), BUSY_WRITE_TIMEOUT);
        }
      } else {
        this.#writing = false;
        this.emit("error", err);
      }
      return;
    }

    this.emit("write", n);
    const releasedBufObj = releaseWritingBuf(this.#writingBuf, this.#len, n);
    this.#len = releasedBufObj.len;
    this.#writingBuf = releasedBufObj.writingBuf;

    if (this.#writingBuf.length) {
      if (!this.#sync) {
        this.#fsWrite();
        return;
      }

      try {
        do {
          const n = this.#fsWriteSync();
          const releasedBufObj = releaseWritingBuf(
            this.#writingBuf,
            this.#len,
            n,
          );
          this.#len = releasedBufObj.len;
          this.#writingBuf = releasedBufObj.writingBuf;
        } while (this.#writingBuf.length);
      } catch (err) {
        this.#release(err);
        return;
      }
    }

    if (this.#fsync) {
      this.#fs.fsyncSync(this.#fd);
    }

    const len = this.#len;
    if (this.#reopening) {
      this.#writing = false;
      this.#reopening = false;
      this.reopen();
    } else if (len > this.#minLength) {
      this.#actualWrite();
    } else if (this.#ending) {
      if (len > 0) {
        this.#actualWrite();
      } else {
        this.#writing = false;
        this.#actualClose();
      }
    } else {
      this.#writing = false;
      if (this.#sync) {
        if (!this.#asyncDrainScheduled) {
          this.#asyncDrainScheduled = true;
          process.nextTick(() => this.#emitDrain());
        }
      } else {
        this.emit("drain");
      }
    }
  }

  #openFile(file) {
    this.#opening = true;
    this.#writing = true;
    this.#asyncDrainScheduled = false;

    const fileOpened = (err, fd) => {
      if (err) {
        this.#reopening = false;
        this.#writing = false;
        this.#opening = false;

        if (this.#sync) {
          process.nextTick(() => {
            if (this.listenerCount("error") > 0) {
              this.emit("error", err);
            }
          });
        } else {
          this.emit("error", err);
        }
        return;
      }

      const reopening = this.#reopening;

      this.#fd = fd;
      this.#file = file;
      this.#reopening = false;
      this.#opening = false;
      this.#writing = false;

      if (this.#sync) {
        process.nextTick(() => this.emit("ready"));
      } else {
        this.emit("ready");
      }

      if (this.#destroyed) {
        return;
      }

      if (
        (!this.#writing && this.#len > this.#minLength) || this.#flushPending
      ) {
        this.#actualWrite();
      } else if (reopening) {
        process.nextTick(() => this.emit("drain"));
      }
    };

    const flags = this.#append ? "a" : "w";
    const mode = this.#mode;

    if (this.#sync) {
      try {
        if (this.#mkdir) {
          this.#fs.mkdirSync(path.dirname(file), { recursive: true });
        }
        const fd = this.#fs.openSync(file, flags, mode);
        fileOpened(null, fd);
      } catch (err) {
        fileOpened(err);
        throw err;
      }
    } else if (this.#mkdir) {
      this.#fs.mkdir(path.dirname(file), { recursive: true }, (err) => {
        if (err) return fileOpened(err);
        this.#fs.open(file, flags, mode, fileOpened);
      });
    } else {
      this.#fs.open(file, flags, mode, fileOpened);
    }
  }

  #emitDrain() {
    const hasListeners = this.listenerCount("drain") > 0;
    if (!hasListeners) return;
    this.#asyncDrainScheduled = false;
    this.emit("drain");
  }

  #actualClose() {
    if (this.#fd === -1) {
      this.once("ready", () => this.#actualClose());
      return;
    }

    if (this.#periodicFlushTimer !== undefined) {
      clearInterval(this.#periodicFlushTimer);
    }

    this.#destroyed = true;
    this.#bufs = [];
    this.#lens = [];

    const done = (err) => {
      if (err) {
        this.emit("error", err);
        return;
      }

      if (this.#ending && !this.#writing) {
        this.emit("finish");
      }
      this.emit("close");
    };

    const closeWrapped = () => {
      if (this.#fd !== 1 && this.#fd !== 2) {
        this.#fs.close(this.#fd, done);
      } else {
        done();
      }
    };

    try {
      this.#fs.fsync(this.#fd, closeWrapped);
    } catch {
      // Intentionally empty.
    }
  }

  #actualWriteBuffer() {
    this.#writing = true;
    this.#writingBuf = this.#writingBuf.length
      ? this.#writingBuf
      : mergeBuf(this.#bufs.shift(), this.#lens.shift());

    if (this.#sync) {
      try {
        const written = this.#fs.writeSync(this.#fd, this.#writingBuf);
        this.#release(null, written);
      } catch (err) {
        this.#release(err);
      }
    } else {
      this.#writingBuf = Buffer.from(this.#writingBuf);
      this.#fs.write(
        this.#fd,
        this.#writingBuf,
        (...args) => this.#release(...args),
      );
    }
  }

  #actualWriteUtf8() {
    this.#writing = true;
    this.#writingBuf ||= this.#bufs.shift() || "";

    if (this.#sync) {
      try {
        const written = this.#fs.writeSync(this.#fd, this.#writingBuf, "utf8");
        this.#release(null, written);
      } catch (err) {
        this.#release(err);
      }
    } else {
      this.#fs.write(
        this.#fd,
        this.#writingBuf,
        "utf8",
        (...args) => this.#release(...args),
      );
    }
  }

  #flushBufferSync() {
    if (this.#destroyed) {
      throw new ERR_INVALID_STATE("Utf8Stream is destroyed");
    }

    if (this.#fd < 0) {
      throw new ERR_INVALID_STATE("Invalid file descriptor");
    }

    if (!this.#writing && this.#writingBuf.length > 0) {
      this.#bufs.unshift([this.#writingBuf]);
      this.#writingBuf = kEmptyBuffer;
    }

    let buf = kEmptyBuffer;
    while (this.#bufs.length || buf.length) {
      if (buf.length <= 0) {
        buf = mergeBuf(this.#bufs[0], this.#lens[0]);
      }
      try {
        const n = this.#fs.writeSync(this.#fd, buf);
        buf = buf.subarray(n);
        this.#len = MathMax(this.#len - n, 0);
        if (buf.length <= 0) {
          this.#bufs.shift();
          this.#lens.shift();
        }
      } catch (err) {
        const shouldRetry = err.code === "EAGAIN" || err.code === "EBUSY";
        if (
          shouldRetry &&
          !this.#retryEAGAIN(err, buf.length, this.#len - buf.length)
        ) {
          throw err;
        }

        sleep(BUSY_WRITE_TIMEOUT);
      }
    }
  }

  #flushSyncUtf8() {
    if (this.#destroyed) {
      throw new ERR_INVALID_STATE("Utf8Stream is destroyed");
    }

    if (this.#fd < 0) {
      throw new ERR_INVALID_STATE("Invalid file descriptor");
    }

    if (!this.#writing && this.#writingBuf.length > 0) {
      this.#bufs.unshift(this.#writingBuf);
      this.#writingBuf = "";
    }

    let buf = "";
    while (this.#bufs.length || buf) {
      if (buf.length <= 0) {
        buf = this.#bufs[0];
      }
      try {
        const n = this.#fs.writeSync(this.#fd, buf, "utf8");
        const releasedBufObj = releaseWritingBuf(buf, this.#len, n);
        buf = releasedBufObj.writingBuf;
        this.#len = releasedBufObj.len;
        if (buf.length <= 0) {
          this.#bufs.shift();
        }
      } catch (err) {
        const shouldRetry = err.code === "EAGAIN" || err.code === "EBUSY";
        if (
          shouldRetry &&
          !this.#retryEAGAIN(err, buf.length, this.#len - buf.length)
        ) {
          throw err;
        }

        sleep(BUSY_WRITE_TIMEOUT);
      }
    }

    try {
      this.#fs.fsyncSync(this.#fd);
    } catch {
      // Skip the error. The fd might not support fsync.
    }
  }

  #callFlushCallbackOnDrain(cb) {
    this.#flushPending = true;
    const onDrain = () => {
      if (!this.#fsync && !this.#destroyed) {
        try {
          this.#fs.fsync(this.#fd, (err) => {
            this.#flushPending = false;
            if (err?.code === "EBADF") {
              cb();
              return;
            }
            cb(err);
          });
        } catch (err) {
          this.#flushPending = false;
          cb(err);
        }
      } else {
        this.#flushPending = false;
        cb();
      }
      this.off("error", onError);
    };
    const onError = (err) => {
      this.#flushPending = false;
      cb(err);
      this.off("drain", onDrain);
    };

    this.once("drain", onDrain);
    this.once("error", onError);
  }

  #flushBuffer(cb) {
    validateFunction(cb, "cb");

    if (this.#destroyed) {
      const error = new ERR_INVALID_STATE("Utf8Stream is destroyed");
      if (cb) {
        cb(error);
        return;
      }

      throw error;
    }

    if (this.#minLength <= 0) {
      cb?.();
      return;
    }

    if (cb) {
      this.#callFlushCallbackOnDrain(cb);
    }

    if (this.#writing) {
      return;
    }

    if (this.#bufs.length === 0) {
      ArrayPrototypePush(this.#bufs, []);
      ArrayPrototypePush(this.#lens, 0);
    }

    this.#actualWrite();
  }

  #flushUtf8(cb) {
    validateFunction(cb, "cb");

    if (this.#destroyed) {
      const error = new ERR_INVALID_STATE("Utf8Stream is destroyed");
      if (cb) {
        cb(error);
        return;
      }

      throw error;
    }

    if (this.#minLength <= 0) {
      cb?.();
      return;
    }

    if (cb) {
      this.#callFlushCallbackOnDrain(cb);
    }

    if (this.#writing) {
      return;
    }

    if (this.#bufs.length === 0) {
      ArrayPrototypePush(this.#bufs, "");
    }

    this.#actualWrite();
  }

  #writeBuffer(data) {
    if (this.#destroyed) {
      throw new ERR_INVALID_STATE("Utf8Stream is destroyed");
    }

    if (!Buffer.isBuffer(data)) {
      throw new ERR_INVALID_ARG_TYPE("data", "Buffer", data);
    }

    const len = this.#len + data.length;
    const bufs = this.#bufs;
    const lens = this.#lens;

    if (this.#maxLength && len > this.#maxLength) {
      this.emit("drop", data);
      return this.#len < this.#hwm;
    }

    if (
      bufs.length === 0 ||
      lens[lens.length - 1] + data.length > this.#maxWrite
    ) {
      ArrayPrototypePush(bufs, []);
      ArrayPrototypePush(lens, data.length);
    } else {
      ArrayPrototypePush(bufs[bufs.length - 1], data);
      lens[lens.length - 1] += data.length;
    }

    this.#len = len;

    if (!this.#writing && this.#len >= this.#minLength) {
      this.#actualWrite();
    }

    return this.#len < this.#hwm;
  }

  #writeUtf8(data) {
    if (this.#destroyed) {
      throw new ERR_INVALID_STATE("Utf8Stream is destroyed");
    }
    validateString(data, "data");

    const len = this.#len + data.length;
    const bufs = this.#bufs;

    if (this.#maxLength && len > this.#maxLength) {
      this.emit("drop", data);
      return this.#len < this.#hwm;
    }

    if (
      bufs.length === 0 ||
      bufs[bufs.length - 1].length + data.length > this.#maxWrite
    ) {
      ArrayPrototypePush(bufs, "" + data);
    } else {
      bufs[bufs.length - 1] += data;
    }

    this.#len = len;

    if (!this.#writing && this.#len >= this.#minLength) {
      this.#actualWrite();
    }

    return this.#len < this.#hwm;
  }
}

function releaseWritingBuf(writingBuf, len, n) {
  if (typeof writingBuf === "string") {
    const byteLength = Buffer.byteLength(writingBuf);
    if (byteLength !== n) {
      const buf = Buffer.from(writingBuf);
      while (n > 0 && (buf[n] & 0xC0) === 0x80) {
        n--;
      }
      n = buf.subarray(0, n).toString().length;
    }
  }
  len = MathMax(len - n, 0);
  writingBuf = writingBuf.slice(n);
  return { writingBuf, len };
}

function mergeBuf(bufs, len) {
  if (bufs.length === 0) {
    return kEmptyBuffer;
  }

  if (bufs.length === 1) {
    return bufs[0];
  }

  return Buffer.concat(bufs, len);
}

export default Utf8Stream;
export { Utf8Stream };