smartstream/ts/smartstream.nodewebhelpers.ts

import * as plugins from './smartstream.plugins.js';

/**
 * Creates a Web ReadableStream from a file using pull-based backpressure.
 *
 * @param filePath - The path to the file to be read
 * @returns A Web ReadableStream that reads the file in chunks
 */
export function createWebReadableStreamFromFile(filePath: string): ReadableStream<Uint8Array> {
  const fileStream = plugins.fs.createReadStream(filePath);

  return new ReadableStream({
    start(controller) {
      fileStream.on('error', (err) => {
        controller.error(err);
      });

      fileStream.on('end', () => {
        controller.close();
      });

      // Pause immediately — pull() will drive reads
      fileStream.pause();
    },
    pull(controller) {
      return new Promise<void>((resolve, reject) => {
        const chunk = fileStream.read();
        if (chunk !== null) {
          controller.enqueue(chunk as Uint8Array);
          resolve();
          return;
        }
        // No data available yet — wait for 'readable' or 'end'
        const onReadable = () => {
          cleanup();
          const data = fileStream.read();
          if (data !== null) {
            controller.enqueue(data as Uint8Array);
          }
          resolve();
        };
        const onEnd = () => {
          cleanup();
          resolve();
        };
        const onError = (err: Error) => {
          cleanup();
          reject(err);
        };
        const cleanup = () => {
          fileStream.removeListener('readable', onReadable);
          fileStream.removeListener('end', onEnd);
          fileStream.removeListener('error', onError);
        };
        fileStream.once('readable', onReadable);
        fileStream.once('end', onEnd);
        fileStream.once('error', onError);
      });
    },
    cancel() {
      fileStream.destroy();
    }
  });
}

/**
 * Converts a Web ReadableStream to a Node.js Readable stream.
 *
 * @param webStream - The Web ReadableStream to convert
 * @returns A Node.js Readable stream that reads data from the Web ReadableStream
 */
export function convertWebReadableToNodeReadable(webStream: ReadableStream<Uint8Array>): plugins.stream.Readable {
  const reader = webStream.getReader();

  return new plugins.stream.Readable({
    read() {
      reader.read().then(
        ({ value, done }) => {
          if (done) {
            this.push(null);
          } else {
            this.push(Buffer.from(value));
          }
        },
        (err) => {
          this.destroy(err);
        }
      );
    }
  });
}

/**
 * Converts a Node.js Readable stream to a Web ReadableStream using pull-based backpressure.
 *
 * @param nodeStream - The Node.js Readable stream to convert
 * @returns A Web ReadableStream that reads data from the Node.js Readable stream
 */
export function convertNodeReadableToWebReadable(nodeStream: plugins.stream.Readable): ReadableStream<Uint8Array> {
  return new ReadableStream({
    start(controller) {
      nodeStream.on('error', (err) => {
        controller.error(err);
      });

      nodeStream.on('end', () => {
        controller.close();
      });

      // Pause immediately — pull() will drive reads
      nodeStream.pause();
    },
    pull(controller) {
      return new Promise<void>((resolve, reject) => {
        const chunk = nodeStream.read();
        if (chunk !== null) {
          controller.enqueue(new Uint8Array(chunk));
          resolve();
          return;
        }
        // No data available yet — wait for 'readable' or 'end'
        const onReadable = () => {
          cleanup();
          const data = nodeStream.read();
          if (data !== null) {
            controller.enqueue(new Uint8Array(data));
          }
          resolve();
        };
        const onEnd = () => {
          cleanup();
          resolve();
        };
        const onError = (err: Error) => {
          cleanup();
          reject(err);
        };
        const cleanup = () => {
          nodeStream.removeListener('readable', onReadable);
          nodeStream.removeListener('end', onEnd);
          nodeStream.removeListener('error', onError);
        };
        nodeStream.once('readable', onReadable);
        nodeStream.once('end', onEnd);
        nodeStream.once('error', onError);
      });
    },
    cancel() {
      nodeStream.destroy();
    }
  });
}

/**
 * Converts a Web WritableStream to a Node.js Writable stream.
 *
 * @param webWritable - The Web WritableStream to convert
 * @returns A Node.js Writable stream that writes data to the Web WritableStream
 */
export function convertWebWritableToNodeWritable(webWritable: WritableStream<Uint8Array>): plugins.stream.Writable {
  const writer = webWritable.getWriter();

  return new plugins.stream.Writable({
    write(chunk, encoding, callback) {
      writer.write(new Uint8Array(chunk)).then(
        () => callback(),
        (err) => callback(err)
      );
    },
    final(callback) {
      writer.close().then(() => callback()).catch(callback);
    },
    destroy(err, callback) {
      if (err) {
        writer.abort(err).then(() => callback(err)).catch(() => callback(err));
      } else {
        // Clean destroy — just release the lock
        writer.releaseLock();
        callback(null);
      }
    }
  });
}

/**
 * Converts a Node.js Writable stream to a Web WritableStream.
 *
 * @param nodeWritable - The Node.js Writable stream to convert
 * @returns A Web WritableStream that writes data to the Node.js Writable stream
 */
export function convertNodeWritableToWebWritable(nodeWritable: plugins.stream.Writable): WritableStream<Uint8Array> {
  return new WritableStream({
    write(chunk) {
      return new Promise((resolve, reject) => {
        nodeWritable.write(Buffer.from(chunk), (err) => {
          if (err) {
            reject(err);
          } else {
            resolve();
          }
        });
      });
    },
    close() {
      return new Promise((resolve, reject) => {
        nodeWritable.end((err: Error | null) => {
          if (err) {
            reject(err);
          } else {
            resolve();
          }
        });
      });
    },
    abort(reason) {
      nodeWritable.destroy(reason instanceof Error ? reason : new Error(String(reason)));
    }
  });
}