feat: handle huge stream inputs

Author: DoneDeal0

src/lib/stream-list-diff-2/emitter.ts

@@ -0,0 +1,46 @@
+import { StreamListDiff } from "@models/stream";
+
+type Listener<T extends unknown[]> = (...args: T) => void;
+
+export enum StreamEvent {
+  Data = "data",
+  Finish = "finish",
+  Error = "error",
+}
+
+export type Emitter<T extends Record<string, unknown>> = EventEmitter<{
+  data: [StreamListDiff<T>[]];
+  error: [Error];
+  finish: [];
+}>;
+
+export class EventEmitter<Events extends Record<string, unknown[]>> {
+  private events: Record<string, Listener<unknown[]>[]> = {};
+
+  on<E extends keyof Events>(event: E, listener: Listener<Events[E]>): this {
+    if (!this.events[event as string]) {
+      this.events[event as string] = [];
+    }
+    this.events[event as string].push(listener as Listener<unknown[]>);
+    return this;
+  }
+
+  emit<E extends keyof Events>(event: E, ...args: Events[E]): void {
+    if (this.events[event as string]) {
+      this.events[event as string].forEach((listener) => listener(...args));
+    }
+  }
+}
+
+export type EmitterEvents<T extends Record<string, unknown>> = {
+  data: [StreamListDiff<T>[]];
+  error: [Error];
+  finish: [];
+};
+
+export interface StreamListener<T extends Record<string, unknown>> {
+  on<E extends keyof EmitterEvents<T>>(
+    event: E,
+    listener: Listener<EmitterEvents<T>[E]>,
+  ): this;
+}

src/lib/stream-list-diff-2/index.ts

@@ -0,0 +1,322 @@
+import {
+  DataBuffer,
+  DEFAULT_LIST_STREAM_OPTIONS,
+  ListStreamOptions,
+  ReferenceProperty,
+  StreamListDiff,
+} from "@models/stream";
+import { LIST_STATUS } from "@models/list";
+import { isObject } from "@lib/utils";
+import {
+  Emitter,
+  EmitterEvents,
+  EventEmitter,
+  StreamListener,
+  StreamEvent,
+} from "./emitter";
+import type { Readable } from "stream";
+
+const isNode =
+  typeof process !== "undefined" &&
+  process.release &&
+  process?.release?.name === "node";
+
+// let ReadableStreamImpl: unknown;
+// if (isNode) {
+//   // eslint-disable-next-line @typescript-eslint/no-require-imports
+//   ReadableStreamImpl = require("stream").Readable;
+// } else {
+//   ReadableStreamImpl = globalThis.ReadableStream;
+// }
+
+// async function streamToList<T>(stream: any): Promise<T[]> {
+//   const list: T[] = [];
+//   if (isNode) {
+//     // Node.js stream handling
+//     for await (const chunk of stream) {
+//       list.push(chunk);
+//     }
+//   } else {
+//     // Web Streams API handling (browser)
+//     const reader = stream.getReader();
+//     let done: boolean | undefined = false;
+//     while (!done) {
+//       const { value, done: streamDone } = await reader.read();
+//       done = streamDone;
+//       if (value) list.push(value);
+//     }
+//   }
+//   return list;
+// }
+
+function outputDiffChunk<T extends Record<string, unknown>>(
+  emitter: Emitter<T>,
+) {
+  let chunks: StreamListDiff<T>[] = [];
+
+  function handleDiffChunk(
+    chunk: StreamListDiff<T>,
+    options: ListStreamOptions,
+  ): void {
+    const showChunk = options?.showOnly
+      ? options?.showOnly.includes(chunk.status)
+      : true;
+    if (!showChunk) {
+      return;
+    }
+    if ((options.chunksSize as number) > 0) {
+      chunks.push(chunk);
+      if (chunks.length >= (options.chunksSize as number)) {
+        const output = chunks;
+        chunks = [];
+        return emitter.emit(StreamEvent.Data, output);
+      } else {
+        return;
+      }
+    }
+    return emitter.emit(StreamEvent.Data, [chunk]);
+  }
+
+  function releaseLastChunks() {
+    if (chunks.length > 0) {
+      const output = chunks;
+      chunks = [];
+      return emitter.emit(StreamEvent.Data, output);
+    }
+  }
+
+  return {
+    handleDiffChunk,
+    releaseLastChunks,
+  };
+}
+
+function isValidChunkSize(
+  chunksSize: ListStreamOptions["chunksSize"],
+): boolean {
+  if (!chunksSize) return true;
+  const sign = String(Math.sign(chunksSize));
+  return sign !== "-1" && sign !== "NaN";
+}
+
+function isDataValid<T extends Record<string, unknown>>(
+  data: T,
+  referenceProperty: ReferenceProperty<T>,
+  listType: "prevList" | "nextList",
+): { isValid: boolean; message?: string } {
+  if (!isObject(data)) {
+    return {
+      isValid: false,
+      message: `Your ${listType} must only contain valid objects. Found '${data}'`,
+    };
+  }
+  if (!Object.hasOwn(data, referenceProperty)) {
+    return {
+      isValid: false,
+      message: `The reference property '${String(referenceProperty)}' is not available in all the objects of your ${listType}.`,
+    };
+  }
+  return {
+    isValid: true,
+    message: "",
+  };
+}
+
+async function getDiffChunks<T extends Record<string, unknown>>(
+  prevStream: Readable,
+  nextStream: Readable,
+  referenceProperty: ReferenceProperty<T>,
+  emitter: Emitter<T>,
+  options: ListStreamOptions = DEFAULT_LIST_STREAM_OPTIONS,
+): Promise<void> {
+  if (!isValidChunkSize(options?.chunksSize)) {
+    return emitter.emit(
+      StreamEvent.Error,
+      new Error(
+        `The chunk size can't be negative. You entered the value '${options.chunksSize}'`,
+      ),
+    );
+  }
+  const { handleDiffChunk, releaseLastChunks } = outputDiffChunk<T>(emitter);
+  const prevDataBuffer: DataBuffer<T> = new Map();
+  const nextDataBuffer: DataBuffer<T> = new Map();
+  let currentPrevIndex = 0;
+  let currentNextIndex = 0;
+
+  async function processPrevStreamChunk(chunk: T) {
+    const { isValid, message } = isDataValid(
+      chunk,
+      referenceProperty,
+      "prevList",
+    );
+    if (!isValid) {
+      emitter.emit(StreamEvent.Error, new Error(message));
+      emitter.emit(StreamEvent.Finish);
+      return;
+    }
+    const ref = chunk[referenceProperty] as ReferenceProperty<T>;
+    const relatedChunk = nextDataBuffer.get(ref);
+
+    if (relatedChunk) {
+      nextDataBuffer.delete(ref);
+      const isDataEqual =
+        JSON.stringify(chunk) === JSON.stringify(relatedChunk.data);
+      const indexDiff = (relatedChunk.index as number) - currentPrevIndex;
+      if (isDataEqual) {
+        handleDiffChunk(
+          {
+            previousValue: chunk,
+            currentValue: relatedChunk.nextData,
+            prevIndex: currentPrevIndex,
+            newIndex: relatedChunk.nextIndex,
+            indexDiff,
+            status:
+              indexDiff === 0
+                ? LIST_STATUS.EQUAL
+                : options.considerMoveAsUpdate
+                  ? LIST_STATUS.UPDATED
+                  : LIST_STATUS.MOVED,
+          },
+          options,
+        );
+      } else {
+        handleDiffChunk(
+          {
+            previousValue: chunk,
+            currentValue: relatedChunk.nextData,
+            prevIndex: currentPrevIndex,
+            newIndex: relatedChunk.nextIndex,
+            indexDiff,
+            status: LIST_STATUS.UPDATED,
+          },
+          options,
+        );
+      }
+    } else {
+      prevDataBuffer.set(ref, { data: chunk, index: currentPrevIndex });
+    }
+    currentPrevIndex++;
+  }
+
+  async function processNextStreamChunk(chunk: T) {
+    const { isValid, message } = isDataValid(
+      chunk,
+      referenceProperty,
+      "nextList",
+    );
+    if (!isValid) {
+      emitter.emit(StreamEvent.Error, new Error(message));
+      emitter.emit(StreamEvent.Finish);
+      return;
+    }
+    const ref = chunk[referenceProperty] as ReferenceProperty<T>;
+    const relatedChunk = prevDataBuffer.get(ref);
+
+    if (relatedChunk) {
+      prevDataBuffer.delete(ref);
+      const isDataEqual =
+        JSON.stringify(chunk) === JSON.stringify(relatedChunk.data);
+      const indexDiff = currentNextIndex - (relatedChunk.index as number);
+      if (isDataEqual) {
+        handleDiffChunk(
+          {
+            previousValue: relatedChunk.data,
+            currentValue: chunk,
+            prevIndex: relatedChunk.index,
+            newIndex: currentNextIndex,
+            indexDiff,
+            status:
+              indexDiff === 0
+                ? LIST_STATUS.EQUAL
+                : options.considerMoveAsUpdate
+                  ? LIST_STATUS.UPDATED
+                  : LIST_STATUS.MOVED,
+          },
+          options,
+        );
+      } else {
+        handleDiffChunk(
+          {
+            previousValue: relatedChunk.data,
+            currentValue: chunk,
+            prevIndex: relatedChunk.index,
+            newIndex: currentNextIndex,
+            indexDiff,
+            status: LIST_STATUS.UPDATED,
+          },
+          options,
+        );
+      }
+    } else {
+      nextDataBuffer.set(ref, { data: chunk, index: currentNextIndex });
+    }
+    currentNextIndex++;
+  }
+
+  const prevStreamReader = async () => {
+    for await (const chunk of prevStream) {
+      await processPrevStreamChunk(chunk);
+    }
+  };
+
+  const nextStreamReader = async () => {
+    for await (const chunk of nextStream) {
+      await processNextStreamChunk(chunk);
+    }
+  };
+  await Promise.all([prevStreamReader(), nextStreamReader()]);
+
+  for (const [key, chunk] of prevDataBuffer.entries()) {
+    handleDiffChunk(
+      {
+        previousValue: chunk.data,
+        currentValue: null,
+        prevIndex: chunk.index,
+        newIndex: null,
+        indexDiff: null,
+        status: LIST_STATUS.DELETED,
+      },
+      options,
+    );
+    prevDataBuffer.delete(key);
+  }
+  for (const [key, chunk] of nextDataBuffer.entries()) {
+    handleDiffChunk(
+      {
+        previousValue: null,
+        currentValue: chunk.data,
+        prevIndex: null,
+        newIndex: chunk.index,
+        indexDiff: null,
+        status: LIST_STATUS.ADDED,
+      },
+      options,
+    );
+    nextDataBuffer.delete(key);
+  }
+  releaseLastChunks();
+  return emitter.emit(StreamEvent.Finish);
+}
+
+export function streamListDiff2<T extends Record<string, unknown>>(
+  prevStream: Readable,
+  nextStream: Readable,
+  referenceProperty: ReferenceProperty<T>,
+  options: ListStreamOptions = DEFAULT_LIST_STREAM_OPTIONS,
+): StreamListener<T> {
+  const emitter = new EventEmitter<EmitterEvents<T>>();
+  setTimeout(async () => {
+    try {
+      await getDiffChunks(
+        prevStream,
+        nextStream,
+        referenceProperty,
+        emitter,
+        options,
+      );
+    } catch (err) {
+      return emitter.emit(StreamEvent.Error, err as Error);
+    }
+  }, 0);
+  return emitter as StreamListener<T>;
+}