import { createHash } from 'node:crypto'
import { readFile } from 'node:fs/promises'
type Outcome =
| { state: 'succeeded'; key: string; result: unknown; credits: number }
| { state: 'failed'; key: string; reason: string; requestId?: string }
const API = 'https://dokyumi.com/api/v1/extract'
const CONCURRENCY = 6 // keep C x 60 / latency under the key's rpm
const MAX_ATTEMPTS = 4
const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms))
async function extractOne(path: string, schema: string, apiKey: string): Promise<Outcome> {
const bytes = await readFile(path)
const key = createHash('sha256').update(bytes).digest('hex') // idempotency key
for (let attempt = 0; attempt < MAX_ATTEMPTS; attempt++) {
const form = new FormData()
form.append('file', new Blob([bytes]), path.split('/').pop() ?? 'document.pdf')
form.append('schema', schema)
const res = await fetch(API, {
method: 'POST',
headers: { Authorization: 'Bearer ' + apiKey },
body: form,
})
if (res.ok) {
const body = await res.json()
return { state: 'succeeded', key, result: body, credits: body?.meta?.credits_used ?? 0 }
}
const err = await res.json().catch(() => ({ error: 'HTTP ' + res.status }))
// Transient: a per-minute rate limit. Wait exactly as long as we are told.
if (res.status === 429 && err.code === 'rate_limited') {
const retryAfter = Number(res.headers.get('Retry-After'))
await sleep(Number.isFinite(retryAfter) ? Math.max(1000, retryAfter * 1000) : 1000 * 2 ** attempt)
continue
}
// Terminal: quota gone (429 quota_exceeded / 402) or the document was rejected (413).
if (res.status >= 400 && res.status < 500) {
return { state: 'failed', key, reason: err.code || err.error, requestId: err.request_id }
}
// Server-side: back off and try again.
await sleep(1000 * 2 ** attempt)
}
return { state: 'failed', key, reason: 'max attempts exceeded' }
}
export async function runBatch(paths: string[], schema: string, apiKey: string, done: Set<string>) {
const queue = [...paths]
const results: Outcome[] = []
const worker = async () => {
for (let path = queue.shift(); path; path = queue.shift()) {
const outcome = await extractOne(path, schema, apiKey)
// Stop the whole run on a terminal quota failure: retrying cannot help.
if (outcome.state === 'failed' && outcome.reason === 'quota_exceeded') {
queue.length = 0
}
if (outcome.state === 'succeeded') done.add(outcome.key)
results.push(outcome)
}
}
await Promise.all(Array.from({ length: CONCURRENCY }, worker))
const credits = results.reduce((n, r) => n + (r.state === 'succeeded' ? r.credits : 0), 0)
return { results, credits }
}