Add bounded-concurrency parsePdfBatch API

README.md

@@ -41,6 +41,47 @@ ESM:
 import { initParser, parsePdf } from '@captain-sdk/pdf-parser';
 ```
 
+## High-throughput batch parsing
+
+For very large queues, use `parsePdfBatch` to bound concurrency and avoid
+starting too many parses at once.
+
+```js
+const fs = require('fs');
+const { initParser, parsePdfBatch } = require('@captain-sdk/pdf-parser');
+
+(async () => {
+  await initParser();
+
+  const inputs = [
+    fs.readFileSync('invoice-1.pdf'),
+    fs.readFileSync('invoice-2.pdf'),
+    fs.readFileSync('invoice-3.pdf'),
+  ];
+
+  const summary = await parsePdfBatch(inputs, {
+    concurrency: 8,
+    stopOnError: false,
+    collectResults: true,
+    onProgress: ({ completed, total, succeeded, failed }) => {
+      if (completed % 100 === 0 || completed === total) {
+        console.log({ completed, total, succeeded, failed });
+      }
+    },
+  });
+
+  console.log(summary.total, summary.succeeded, summary.failed);
+})();
+```
+
+`parsePdfBatch` options:
+
+- `concurrency` (default `4`): maximum in-flight parses.
+- `stopOnError` (default `false`): when `true`, rejects on first failed item.
+- `collectResults` (default `true`): set `false` to reduce memory use for very
+  large batches when you only need counters/progress.
+- `onProgress(event)`: callback after each item completes.
+
 ## Output shape
 
 ```ts

index.d.ts

@@ -27,6 +27,37 @@ export interface ParseResult {
   blocks: Block[];
 }
 
+export interface BatchProgressEvent {
+  index: number;
+  completed: number;
+  total: number;
+  succeeded: number;
+  failed: number;
+  ok: boolean;
+}
+
+export interface ParsePdfBatchOptions extends ParseOptions {
+  /** Maximum number of in-flight parses. Defaults to 4. */
+  concurrency?: number;
+  /** Stop after the first failed parse and reject. Defaults to false. */
+  stopOnError?: boolean;
+  /** Include per-item outputs in the return object. Defaults to true. */
+  collectResults?: boolean;
+  /** Called after each item finishes. */
+  onProgress?: (event: BatchProgressEvent) => void;
+}
+
+export type ParsePdfBatchItemResult =
+  | { ok: true; result: ParseResult }
+  | { ok: false; error: Error };
+
+export interface ParsePdfBatchSummary {
+  total: number;
+  succeeded: number;
+  failed: number;
+  results?: ParsePdfBatchItemResult[];
+}
+
 /** Load and instantiate the engine (PDFium + parser). The first call loads ~6 MB
  *  of WebAssembly and is the slow one; call once at startup so your first parse is
  *  fast. Optional: parsePdf() calls it for you. */
@@ -38,3 +69,9 @@ export function parsePdf(
   bytes: Uint8Array | ArrayBuffer,
   options?: ParseOptions,
 ): Promise<ParseResult>;
+
+/** Parse many PDFs with bounded concurrency. */
+export function parsePdfBatch(
+  items: Array<Uint8Array | ArrayBuffer>,
+  options?: ParsePdfBatchOptions,
+): Promise<ParsePdfBatchSummary>;

index.js

@@ -103,4 +103,96 @@ async function parsePdf(bytes, options) {
   return JSON.parse(jsonStr);
 }
 
-module.exports = { parsePdf, initParser };
+/**
+ * Parse many PDFs with bounded concurrency.
+ * @param {Array<Uint8Array|Buffer|ArrayBuffer>} items
+ * @param {{
+ *   assetsDir?: string,
+ *   concurrency?: number,
+ *   stopOnError?: boolean,
+ *   collectResults?: boolean,
+ *   onProgress?: (event: {
+ *     index: number,
+ *     completed: number,
+ *     total: number,
+ *     succeeded: number,
+ *     failed: number,
+ *     ok: boolean
+ *   }) => void
+ * }} [options]
+ * @returns {Promise<{
+ *   total: number,
+ *   succeeded: number,
+ *   failed: number,
+ *   results?: Array<
+ *     | { ok: true, result: import('./index').ParseResult }
+ *     | { ok: false, error: Error }
+ *   >
+ * }>}
+ */
+async function parsePdfBatch(items, options) {
+  if (!Array.isArray(items)) {
+    throw new TypeError('parsePdfBatch expects an array of Uint8Array, Buffer, or ArrayBuffer.');
+  }
+  const total = items.length;
+  const concurrencyRaw = options && options.concurrency != null ? options.concurrency : 4;
+  const concurrency = Math.max(1, Math.floor(concurrencyRaw));
+  const stopOnError = Boolean(options && options.stopOnError);
+  const collectResults = options && options.collectResults != null ? Boolean(options.collectResults) : true;
+  const onProgress = options && typeof options.onProgress === 'function' ? options.onProgress : null;
+  const results = collectResults ? new Array(total) : undefined;
+
+  await ensureReady(options);
+
+  let nextIndex = 0;
+  let completed = 0;
+  let succeeded = 0;
+  let failed = 0;
+  let stoppedError = null;
+
+  async function worker() {
+    for (;;) {
+      if (stoppedError) return;
+      const index = nextIndex++;
+      if (index >= total) return;
+
+      try {
+        const result = await parsePdf(items[index], options);
+        succeeded += 1;
+        if (results) results[index] = { ok: true, result };
+        if (onProgress) {
+          onProgress({ index, completed: completed + 1, total, succeeded, failed, ok: true });
+        }
+      } catch (err) {
+        failed += 1;
+        const error = err instanceof Error ? err : new Error(String(err));
+        if (results) results[index] = { ok: false, error };
+        if (onProgress) {
+          onProgress({ index, completed: completed + 1, total, succeeded, failed, ok: false });
+        }
+        if (stopOnError && !stoppedError) {
+          stoppedError = { index, error };
+        }
+      } finally {
+        completed += 1;
+      }
+    }
+  }
+
+  const workers = [];
+  const workerCount = Math.min(concurrency, total || 1);
+  for (let i = 0; i < workerCount; i += 1) workers.push(worker());
+  await Promise.all(workers);
+
+  if (stoppedError) {
+    throw new Error(
+      `parsePdfBatch failed at index ${stoppedError.index}: ${stoppedError.error.message}`
+    );
+  }
+
+  const summary = { total, succeeded, failed };
+  if (results) summary.results = results;
+  return summary;
+}
+
+module.exports = { parsePdf, parsePdfBatch, initParser };