async-batch.mjs

/**
 * A folder of long PDFs, submitted to the queue and collected as they finish.
 *
 * Two things make this different from a loop of `analyze()` calls:
 *
 * 1. Nothing is held open for 60 seconds. A 50-page contract can take minutes, and a
 *    synchronous request would be killed at 60 s with a 504 `sync_timeout`.
 * 2. A stable `idempotencyKey` per file means re-running the script after a crash
 *    replays the stored responses instead of paying for the batch twice.
 *
 *   node examples/async-batch.mjs ./contracts 4
 */

import { readdir } from 'node:fs/promises';
import { basename, extname, join } from 'node:path';
import { VisionAPI, TaskFailedError, unwrap } from '@devrobotlabs/visionapi';

const [dir = './contracts', concurrency = '4'] = process.argv.slice(2);
const vision = new VisionAPI();

const files = (await readdir(dir))
  .filter((f) => extname(f).toLowerCase() === '.pdf')
  .map((f) => join(dir, f));

const BATCH = new Date().toISOString().slice(0, 10); // one namespace per day

async function process(path) {
  const { task_id } = await vision.analyzeAsync(
    { file: path, preset: 'contract', pages: '1-50' },
    // Derived from the file, not random: a re-run of the same batch must not re-charge.
    { idempotencyKey: `${BATCH}:${basename(path)}` },
  );

  try {
    const task = await vision.waitForTask(task_id, { pollInterval: 2_000, maxWait: 20 * 60_000 });
    return { path, credits: task.credits_used, data: unwrap(task.result, { dropNull: true }) };
  } catch (err) {
    // A failed task costs 0 credits — the reservation is released in full.
    if (err instanceof TaskFailedError) return { path, error: err.code };
    throw err;
  }
}

/** Bounded concurrency: N workers pulling from one queue, rather than N at a time. */
async function pool(items, limit, worker) {
  const results = [];
  const queue = [...items];
  await Promise.all(
    Array.from({ length: Math.min(limit, queue.length) }, async () => {
      for (let item = queue.shift(); item !== undefined; item = queue.shift()) {
        results.push(await worker(item));
      }
    }),
  );
  return results;
}

// The rate limit is 60 requests/minute per key, and polling counts. Four concurrent
// files at one poll every 2 s is comfortably inside it.
const results = await pool(files, Number(concurrency), process);

for (const r of results) {
  console.log(r.error ? `✗ ${r.path}: ${r.error}` : `✓ ${r.path}: ${r.credits} credits`);
}

const spent = results.reduce((sum, r) => sum + (r.credits ?? 0), 0);
console.log(`\n${results.length} files, ${spent} credits`);