streaming.ts

/**
 * JavaScript and Node.js SDK for OpenFGA
 *
 * API version: 1.x
 * Website: https://openfga.dev
 * Documentation: https://openfga.dev/docs
 * Support: https://openfga.dev/community
 * License: [Apache-2.0](https://github.com/openfga/js-sdk/blob/main/LICENSE)
 *
 * NOTE: This file was auto generated by OpenAPI Generator (https://openapi-generator.tech). DO NOT EDIT.
 */

import type { Readable } from "node:stream";

// Helper: create async iterable from classic EventEmitter-style Readable streams
const createAsyncIterableFromReadable = (readable: any): AsyncIterable<any> => {
  return {
    [Symbol.asyncIterator](): AsyncIterator<any> {
      const chunkQueue: any[] = [];
      const pendings: Array<{ resolve: (v: IteratorResult<any>) => void; reject: (e?: any) => void }> = [];
      let ended = false;
      let error: any = null;

      const onData = (chunk: any) => {
        if (pendings.length > 0) {
          const { resolve } = pendings.shift()!;
          resolve({ value: chunk, done: false });
        } else {
          chunkQueue.push(chunk);
        }
      };

      const onEnd = () => {
        if (error) return; // Don't process end if error already occurred
        ended = true;
        while (pendings.length > 0) {
          const { resolve } = pendings.shift()!;
          resolve({ value: undefined, done: true });
        }
      };

      const onError = (err: any) => {
        error = err;
        while (pendings.length > 0) {
          const { reject } = pendings.shift()!;
          reject(err);
        }
        cleanup();
      };

      readable.on("data", onData);
      readable.once("end", onEnd);
      readable.once("error", onError);

      const cleanup = () => {
        readable.off("data", onData);
        readable.off("end", onEnd);
        readable.off("error", onError);
      };

      return {
        next(): Promise<IteratorResult<any>> {
          if (error) {
            return Promise.reject(error);
          }
          if (chunkQueue.length > 0) {
            const value = chunkQueue.shift();
            return Promise.resolve({ value, done: false });
          }
          if (ended) {
            cleanup();
            return Promise.resolve({ value: undefined, done: true });
          }
          return new Promise<IteratorResult<any>>((resolve, reject) => {
            pendings.push({ resolve, reject });
          });
        },
        return(): Promise<IteratorResult<any>> {
          try {
            cleanup();
          } finally {
            if (readable && typeof readable.destroy === "function") {
              readable.destroy();
            }
          }
          return Promise.resolve({ value: undefined, done: true });
        },
        throw(e?: any): Promise<IteratorResult<any>> {
          try {
            cleanup();
          } finally {
            if (readable && typeof readable.destroy === "function") {
              readable.destroy(e);
            }
          }
          return Promise.reject(e);
        }
      };
    }
  };
};

/**
 * Parse newline-delimited JSON (NDJSON) from a Node.js readable stream
 * @param stream - Node.js readable stream, AsyncIterable, string, or Buffer
 * @returns AsyncGenerator that yields parsed JSON objects
 */
export async function* parseNDJSONStream(
  stream: Readable | AsyncIterable<Uint8Array | string | Buffer> | string | Uint8Array | Buffer
): AsyncGenerator<any> {
  const decoder = new TextDecoder("utf-8");
  let buffer = "";

  // If stream is actually a string or Buffer-like, handle as whole payload
  const isString = typeof stream === "string";
  const isBuffer = typeof Buffer !== "undefined" && Buffer.isBuffer && Buffer.isBuffer(stream);
  const isUint8Array = typeof Uint8Array !== "undefined" && stream instanceof Uint8Array;

  if (isString || isBuffer || isUint8Array) {
    const text = isString
      ? (stream as string)
      : decoder.decode(isBuffer ? new Uint8Array(stream as Buffer) : (stream as Uint8Array));
    const lines = text.split("\n");

    for (const line of lines) {
      const trimmed = line.trim();
      if (!trimmed) {
        continue;
      }

      try {
        yield JSON.parse(trimmed);
      } catch (err) {
        console.warn("Failed to parse JSON line:", err);
      }
    }
    return;
  }

  const isAsyncIterable = stream && typeof (stream as any)[Symbol.asyncIterator] === "function";
  const source: AsyncIterable<any> = isAsyncIterable ? (stream as any) : createAsyncIterableFromReadable(stream as any);

  for await (const chunk of source) {
    // Node.js streams can return Buffer or string chunks
    // Convert to Uint8Array if needed for TextDecoder
    const uint8Chunk = typeof chunk === "string"
      ? new TextEncoder().encode(chunk)
      : chunk instanceof Buffer
        ? new Uint8Array(chunk)
        : chunk;

    // Append decoded chunk to buffer
    buffer += decoder.decode(uint8Chunk, { stream: true });

    // Split on newlines
    const lines = buffer.split("\n");

    // Keep the last (potentially incomplete) line in the buffer
    buffer = lines.pop() || "";

    // Parse and yield complete lines
    for (const line of lines) {
      const trimmed = line.trim();
      if (trimmed) {
        try {
          yield JSON.parse(trimmed);
        } catch (err) {
          console.warn("Failed to parse JSON line:", err);
        }
      }
    }
  }

  // Flush any remaining decoder state
  buffer += decoder.decode();

  // Handle any remaining data in buffer
  if (buffer.trim()) {
    try {
      yield JSON.parse(buffer);
    } catch (err) {
      console.warn("Failed to parse final JSON buffer:", err);
    }
  }
}