Advance M7 workflows and release operations

web/protocol/file-byte-reader.ts

@@ -0,0 +1,132 @@
+export const LARGE_FILE_IMPORT_BYTES = 512 * 1024;
+export const DEFAULT_FILE_READ_PROGRESS_CHUNK_BYTES = 1024 * 1024;
+export const DEFAULT_FILE_READ_YIELD_BYTES = 4 * 1024 * 1024;
+
+export type FileReadPhase = "STARTED" | "READING" | "COMPLETED" | "CANCELLED";
+
+export interface FileReadProgress {
+  phase: FileReadPhase;
+  bytesRead: number;
+  totalBytes: number;
+  fraction: number;
+}
+
+export interface FileByteSource {
+  readonly size: number;
+  stream(): ReadableStream<Uint8Array>;
+}
+
+export interface FileByteReadOptions {
+  signal: AbortSignal;
+  onProgress?: (progress: FileReadProgress) => void;
+  onResourceState?: (state: FileByteReadResourceState) => void;
+  yieldControl?: () => Promise<void>;
+  progressChunkBytes?: number;
+  yieldEveryBytes?: number;
+}
+
+export interface FileByteReadResourceState {
+  liveReaders: number;
+  liveInputBytes: number;
+  liveStagingFiles: number;
+}
+
+export type FileByteReadErrorCode =
+  | "FILE_READ_CANCELLED"
+  | "FILE_READ_ALLOCATION_FAILED"
+  | "FILE_READ_SIZE_MISMATCH";
+
+export class FileByteReadError extends Error {
+  readonly code: FileByteReadErrorCode;
+
+  constructor(code: FileByteReadErrorCode, message: string) {
+    super(`${code}: ${message}`);
+    this.name = "FileByteReadError";
+    this.code = code;
+  }
+}
+
+function progress(phase: FileReadPhase, bytesRead: number, totalBytes: number): FileReadProgress {
+  return {
+    phase,
+    bytesRead,
+    totalBytes,
+    fraction: totalBytes === 0 ? (phase === "COMPLETED" ? 1 : 0) : bytesRead / totalBytes,
+  };
+}
+
+export async function readFileBytes(source: FileByteSource, options: FileByteReadOptions): Promise<ArrayBuffer> {
+  const { signal, onProgress, onResourceState, yieldControl } = options;
+  const totalBytes = source.size;
+  const progressChunkBytes = options.progressChunkBytes ?? DEFAULT_FILE_READ_PROGRESS_CHUNK_BYTES;
+  const yieldEveryBytes = options.yieldEveryBytes ?? DEFAULT_FILE_READ_YIELD_BYTES;
+  if (!Number.isSafeInteger(totalBytes) || totalBytes < 0) {
+    throw new FileByteReadError("FILE_READ_SIZE_MISMATCH", "file size is not a non-negative safe integer");
+  }
+  if (!Number.isSafeInteger(progressChunkBytes) || progressChunkBytes <= 0 || !Number.isSafeInteger(yieldEveryBytes) || yieldEveryBytes <= 0) {
+    throw new FileByteReadError("FILE_READ_SIZE_MISMATCH", "progress or yield byte interval is invalid");
+  }
+
+  let outputBuffer: ArrayBuffer;
+  try {
+    outputBuffer = new ArrayBuffer(totalBytes);
+  }
+  catch {
+    throw new FileByteReadError("FILE_READ_ALLOCATION_FAILED", `could not allocate ${totalBytes} bytes`);
+  }
+  const output = new Uint8Array(outputBuffer);
+
+  const reader = source.stream().getReader();
+  onResourceState?.({ liveReaders: 1, liveInputBytes: totalBytes, liveStagingFiles: 0 });
+  let bytesRead = 0;
+  let nextYield = yieldEveryBytes;
+  let aborted = signal.aborted;
+  let completed = false;
+  const abort = (): void => {
+    aborted = true;
+    void reader.cancel("FILE_READ_CANCELLED").catch(() => undefined);
+  };
+  signal.addEventListener("abort", abort, { once: true });
+  onProgress?.(progress("STARTED", 0, totalBytes));
+
+  try {
+    while (true) {
+      if (aborted) throw new FileByteReadError("FILE_READ_CANCELLED", `cancelled after ${bytesRead} bytes`);
+      const { done, value } = await reader.read();
+      if (aborted) throw new FileByteReadError("FILE_READ_CANCELLED", `cancelled after ${bytesRead} bytes`);
+      if (done) break;
+      if (!value || bytesRead + value.byteLength > totalBytes) {
+        throw new FileByteReadError("FILE_READ_SIZE_MISMATCH", "stream exceeded the declared file size");
+      }
+      for (let offset = 0; offset < value.byteLength; offset += progressChunkBytes) {
+        if (aborted) throw new FileByteReadError("FILE_READ_CANCELLED", `cancelled after ${bytesRead} bytes`);
+        const chunk = value.subarray(offset, Math.min(value.byteLength, offset + progressChunkBytes));
+        output.set(chunk, bytesRead);
+        bytesRead += chunk.byteLength;
+        onProgress?.(progress("READING", bytesRead, totalBytes));
+        if (yieldControl && bytesRead >= nextYield && bytesRead < totalBytes) {
+          nextYield = bytesRead + yieldEveryBytes;
+          await yieldControl();
+        }
+      }
+    }
+    if (bytesRead !== totalBytes) {
+      throw new FileByteReadError("FILE_READ_SIZE_MISMATCH", `stream ended at ${bytesRead} of ${totalBytes} bytes`);
+    }
+    onProgress?.(progress("COMPLETED", bytesRead, totalBytes));
+    completed = true;
+    return outputBuffer;
+  }
+  catch (error) {
+    if (error instanceof FileByteReadError && error.code === "FILE_READ_CANCELLED") {
+      onProgress?.(progress("CANCELLED", bytesRead, totalBytes));
+    }
+    throw error;
+  }
+  finally {
+    signal.removeEventListener("abort", abort);
+    reader.releaseLock();
+    if (!completed) output.fill(0);
+    onResourceState?.({ liveReaders: 0, liveInputBytes: 0, liveStagingFiles: 0 });
+  }
+}