DEV Community: Pablo Grant

Building a Robust Document Parsing System with Cloudflare Queues and Workflows

Pablo Grant — Tue, 30 Dec 2025 04:28:41 +0000

Introduction

At pdfparse.net, we run a document parsing system that turns PDFs into structured, queryable data at scale. Our stack is Cloudflare-native, which means global distribution, predictable costs, and no servers to babysit.

This article focuses on invoice parsing because it combines structured fields, messy layouts, and long OCR runtimes. We will build a pipeline that accepts large batches, survives failures, and keeps users informed about progress end to end.

Architecture Overview

At a high level, the pipeline looks like this:

HTTP API → Queue → Workflow → Mistral Batch OCR → Database

API accepts document IDs and creates a job record.
Queues smooth bursts and split work into predictable batches.
Workflows orchestrate long-running OCR with retries and durable state.
Mistral performs OCR in batch mode.
Database stores job status and extracted invoice data.

We keep the pieces small and loosely coupled so each one can be retried safely without restarting the entire job.

Project Setup

Prerequisites

Cloudflare account with Workers paid plan (for Workflows)
Mistral AI API key (sign up)
Node.js 18+ and pnpm

Dependencies

We keep the dependency list focused:

{
  "dependencies": {
    "@mistralai/mistralai": "^1.0.0",
    "drizzle-orm": "^0.44.0",
    "effect": "^3.19.0",
    "zod": "^3.23.0"
  },
  "devDependencies": {
    "wrangler": "^3.90.0",
    "@cloudflare/workers-types": "^4.0.0"
  }
}

Project Structure

document-parser/
├── packages/
│   ├── db/                 # Shared database schema + migrations
│   ├── queue-manager/      # HTTP + queue consumer
│   └── workflows/          # Durable workflow logic

Database Setup (D1 + Drizzle)

We store jobs, batch processes, and extracted invoice fields in D1.

import { sqliteTable, text, integer, numeric } from "drizzle-orm/sqlite-core";

export const jobs = sqliteTable("jobs", {
  id: integer("id").primaryKey({ autoIncrement: true }),
  jobStatus: text("job_status").default("queued").notNull(),
  createdAt: integer("created_at", { mode: "timestamp" })
    .$defaultFn(() => new Date()),
});

export const batch_processes = sqliteTable("batch_processes", {
  id: integer("id").primaryKey({ autoIncrement: true }),
  batchId: text("batch_id").notNull().unique(),
  workflowInstanceId: text("workflow_instance_id").notNull().unique(),
  job_id: integer("job_id").references(() => jobs.id, { onDelete: "cascade" }),
  status: text("status").default("init"),
});

export const invoices = sqliteTable("invoices", {
  invoiceId: text("invoice_id").primaryKey(),
  vendorName: text("vendor_name").notNull(),
  invoiceDate: text("invoice_date").notNull(),
  dueDate: text("due_date").notNull(),
  totalAmountUsd: numeric("total_amount_usd").notNull()
});

Create the database and run migrations:

wrangler d1 create document-parser-db
pnpm drizzle-kit generate
wrangler d1 migrations apply document-parser-db --remote

Worker Bindings

Add bindings to both the queue worker and workflow:

{
  "r2_buckets": [
    { "binding": "DOCUMENT_BUCKET", "bucket_name": "<YOUR_BUCKET_NAME>" }
  ],
  "d1_databases": [
    {
      "binding": "DB",
      "database_name": "<YOUR_DB_NAME>",
      "database_id": "<YOUR_DB_ID>"
    }
  ],
  "queues": {
    "producers": [{ "binding": "JOB_QUEUE", "queue": "<YOUR_QUEUE_NAME>" }],
    "consumers": [{ "queue": "<YOUR_QUEUE_NAME>" }]
  },
  "workflows": [
    { "binding": "WORKFLOW", "name": "<YOUR_WORKFLOW_NAME>" }
  ]
}

Also add a MISTRAL_API_KEY secret and an R2 access URL for presigned links.

Queue Manager

The queue manager handles two things: accepting document IDs via HTTP and fanning them out into batches.

import { createDB, jobs } from "@repo/db";

type Env = {
  DB: D1Database;
  JOB_QUEUE: Queue;
  WORKFLOW: Workflow;
};

export default {
  async fetch(request: Request, env: Env): Promise<Response> {
    if (request.method !== "POST") {
      return new Response("Not Found", { status: 404 });
    }

    const { documentIds } = await request.json();
    const db = await createDB(env.DB);

    const [job] = await db
      .insert(jobs)
      .values({ jobStatus: "queued" })
      .returning();

    const batches = chunk(documentIds, 50);

    await env.JOB_QUEUE.sendBatch(
      batches.map((batch, index) => ({
        body: {
          jobId: job.id,
          documentKeys: batch,
          batchIndex: index,
          totalBatches: batches.length,
        },
      }))
    );

    return Response.json({ jobId: job.id });
  },

  async queue(batch: MessageBatch<any>, env: Env): Promise<void> {
    for (const message of batch.messages) {
      await env.WORKFLOW.create({ params: message.body });
      message.ack();
    }
  },
};

Why 50 documents per batch? It is a good balance between throughput and retry cost. If files are large or highly variable, use smaller batches to keep failure recovery cheap.

Workflows: Durable, Retryable OCR

Workflows are a great fit for OCR: they can pause, retry, and keep state across minutes or hours without manual orchestration.

Step 1: Submit a Batch (with Effect.ts)

We start by assembling a batch payload and handing it off to Mistral.
We use Effect.ts to make failure handling explicit and consistent. It keeps errors typed and composable, which matters when you are orchestrating network calls, file uploads, and database updates.

const batchJob = await step.do("submit-batch", async () => {
  const program = pipe(
    Effect.gen(function* (_) {
      const urls = yield* _(generatePresignedUrls(documentKeys, env.R2_URL));

      const requests = urls.map((url, i) => ({
        custom_id: documentKeys[i],
        body: {
          model: "mistral-ocr-latest",
          document: { type: "document_url", document_url: url },
        },
      }));

      const jsonl = requests.map((r) => JSON.stringify(r)).join("\n");
      const client = new Mistral({ apiKey: env.MISTRAL_API_KEY });
      const file = new File([jsonl], `batch_${jobId}.jsonl`, {
        type: "application/x-ndjson",
      });

      const uploadedFile = yield* _(
        Effect.tryPromise({
          try: () => client.files.upload({ file, purpose: "batch" }),
          catch: (error) => new BatchUploadError({ cause: error }),
        })
      );

      return yield* _(
        Effect.tryPromise({
          try: () =>
            client.batch.jobs.create({
              inputFiles: [uploadedFile.id],
              model: "mistral-ocr-latest",
              endpoint: "/v1/ocr",
              metadata: { jobId: String(jobId) },
            }),
          catch: (error) => new BatchCreationError({ cause: error }),
        })
      );
    })
  );

  return Effect.runPromise(program);
});

Step 2: Poll for Completion

We wait for Mistral to finish by polling inside the workflow. Sleeping steps are not billed, so long polling stays cost-effective. You can configure retries via the retries: {} option on step.do().

const results = await step.do("poll-status", {
  retries: { limit: 60, delay: "30 seconds" },
  async () => {
    const program = pipe(
      Effect.gen(function* (_) {
        const status = yield* _(
          Effect.tryPromise({
            try: () => client.batch.jobs.get({ jobId: batchJob.id }),
            catch: (error) => new NetworkError({ cause: error }),
          })
        );

        if (status.status !== "SUCCESS") {
          throw new Error("Still processing");
        }

        return parseJsonlResults(await downloadFile(status.outputFile));
      })
    );

    return Effect.runPromise(program);
  },
});

Step 3: Save Results

Once results are ready, we persist the extracted fields and keep writes simple.

await step.do("save-results", async () => {
  for (const result of results) {
    await db.insert(invoices).values({ /* ... */ });
  }
});

Step 4: Update Batch Status

After saving results, we update the batch record and mark the parent job complete once all batches finish.

await step.do("update-status", async () => {
  await db
    .update(batch_processes)
    .set({
      status: "completed",
      completedRequests: results.length,
    })
    .where(eq(batch_processes.batchId, batchJob.id));

  const allBatches = await db
    .select()
    .from(batch_processes)
    .where(eq(batch_processes.job_id, jobId));

  const allComplete = allBatches.every(
    (b) => b.status === "completed" || b.status === "failed"
  );

  if (allComplete) {
    await db
      .update(jobs)
      .set({ jobStatus: "completed" })
      .where(eq(jobs.id, jobId));
  }
});

Failure Modes We Plan For

OCR provider outages: workflow retries with backoff, status remains "processing."
Bad documents: mark failures per document, not per job.
Duplicate delivery: use idempotent keys (batch ID + document ID).
Database contention: chunk writes and keep transactions short.

This keeps the pipeline stable without manual cleanup.

Observability and Status

We expose job status directly from D1. Each job has a jobStatus and per-batch metadata, which lets us show progress like "12/24 batches complete." Internally, we log workflow step timings to identify slow OCR runs and throttling.

If you need an external-facing dashboard, this structure makes it trivial to compute percent complete without storing extra state.

Resources

Introducing PdfParse: Transform Documents into Structured Databases

Pablo Grant — Sun, 28 Dec 2025 21:29:37 +0000

We’re excited to introduce PdfParse - a novel approach to document data extraction that transforms unstructured PDFs into robust, normalized databases. Unlike traditional OCR tools that dump raw text or basic JSON, PdfParse leverages AI-powered extraction with automatic data normalization to create structured, queryable databases from your documents. And the best part? It’s incredibly cheap to use, with a free tier for evaluation and pay-per-page pricing that won’t break the bank. Start extracting structured PDFs for free or explore the product walk-through below.

Whether you're processing invoices, receipts, contracts, or custom forms, PdfParse handles the complexity of nested data relationships, table extraction, and schema validation - delivering production-ready SQLite databases you can download and integrate immediately.

Features

Structured User Generated Schemas

With PdfParse, you define custom structured schemas tailored to your specific document types. Need to extract invoice data from PDFs? Create a schema that maps your invoice fields - from customer information to line items - and PdfParse handles the rest. Our structured schema builder lets you:

Define custom tables with typed fields (text, numbers, dates, booleans)
Set validation rules and constraints
Create reusable templates for common document types
Support complex data types and nested relationships

Simply design your schema once, and extract consistent, validated data from every PDF you process. No more wrestling with inconsistent JSON formats or manual data cleaning.

Nested and Hierarchical Table Extraction

PdfParse excels at understanding complex document structures with nested relationships. Here's how our extraction pipeline works:

┌─────────────────────────────────────────────────────┐
│              Schema Definition                      │
│                                                     │
│  ┌──────────────┐         ┌─────────────────┐       │
│  │   Invoice    │────┬───→│    Customer     │       │
│  │              │    │    │                 │       │
│  │ - number     │    │    │ - name          │       │
│  │ - date       │    │    │ - address       │       │
│  │ - total      │    │    │ - email         │       │
│  └──────────────┘    │    └─────────────────┘       │
│         │            │                              │
│         │            │    ┌─────────────────┐       │
│         └────────────┴───→│  Invoice_Items  │       │
│                           │                 │       │
│                           │ - description   │       │
│                           │ - quantity      │       │
│                           │ - price         │       │
│                           └─────────────────┘       │
└─────────────────────────────────────────────────────┘

Once you’ve created your schema with parent-child table relationships (like invoice → customer and invoice → invoice_items), the extraction process is simple:

Upload your PDF files to your project (get started at /create-account)
Click the "Process Files" button
PdfParse extracts data according to your schema, populating parent and child tables
Any extraction issues are saved as errors for later review and retry (see how we handle this in the features section)

Here's what happens under the hood:

┌──────────────────────────┐
│     Invoice PDF          │
│                          │
│  INVOICE #12345          │
│  Date: 2025-01-15        │
│  Customer: Acme Corp     │
│                          │
│  Items:                  │
│  - Widget A   $100       │
│  - Widget B   $200       │
│  Total: $300             │
└──────────────────────────┘
            │
            ▼
    ┌───────────────┐
    │  AI Pipeline  │
    │  (Extraction) │
    └───────────────┘
            │
            ▼
┌─────────────────────────────────────┐
│       SQLite Database               │
│                                     │
│  invoices:                          │
│  ├─ id: 1                           │
│  ├─ number: "12345"                 │
│  ├─ date: "2025-01-15"              │
│  └─ total: 300.00                   │
│                                     │
│  customers:                         │
│  ├─ id: 1                           │
│  ├─ invoice_id: 1  (FK)             │
│  └─ name: "Acme Corp"               │
│                                     │
│  invoice_items:                     │
│  ├─ id: 1, invoice_id: 1            │
│  │  description: "Widget A"         │
│  │  price: 100.00                   │
│  └─ id: 2, invoice_id: 1            │
│     description: "Widget B"         │
│     price: 200.00                   │
└─────────────────────────────────────┘

SQLite as a Single Source of Truth

We chose SQLite over basic JSON for a compelling reason: it’s robust, portable, and fast - making it the backbone of most LLM agents and modern data pipelines. Here’s why SQLite makes sense:

Robust: ACID-compliant transactions, data integrity, and reliability
Portable: A single file containing your entire structured database
Fast: Efficient querying, indexing, and aggregations without external dependencies
Universal: Supported by virtually every programming language and framework
Downloadable: Export your complete database as a .sqlite file for offline use, backups, or integration with existing systems

Instead of parsing nested JSON structures, you get a real relational database with foreign keys, indexes, and SQL querying capabilities. Use it directly in your applications, connect it to BI tools, or feed it to AI agents for natural language queries.

Coming soon: API access and webhook integrations for real-time data synchronization with your systems.

Human Readable Errors

Extraction isn’t always perfect, especially with complex or low-quality documents. That’s why PdfParse provides human-readable error messages with a built-in review interface (see the error-handling walkthrough):

View detailed error messages for each failed extraction field
See exactly which document and field encountered issues
Review errors in context with the original PDF
Trigger individual or batch retries for failed extractions
Track error resolution progress across your projects

No more digging through logs or debugging cryptic error codes. PdfParse surfaces extraction issues in a clear, actionable format so you can quickly identify patterns, fix schema definitions, or manually correct edge cases.

SQL Table Based Navigation

Once your documents are processed, navigate your extracted data like a SQL explorer. PdfParse provides an intuitive table-based interface where you can:

Browse all tables in your database with familiar spreadsheet-like views
Search and filter data across any field
View parent-child relationships and navigate between linked records
Execute custom SQL queries for advanced analysis
Export individual tables or entire databases as CSV, JSON, or SQLite files

It's the power of a full database client, built directly into your extraction workflow. No need to download databases and open external tools - explore your data immediately after extraction.

Pricing

We believe document extraction should be affordable and accessible to everyone. That’s why we’ve designed competitive pricing that scales with your needs, without hidden fees or surprise charges. Compare plans on the main site’s pricing section or try the free tier right away.

Free Tier

Try PdfParse risk-free with our free evaluation tier. Test the product with real documents before committing to a paid plan:

┌──────────────┬─────────────────┬──────────────┐
│     Plan     │  Pages/Month    │  Price/Month │
├──────────────┼─────────────────┼──────────────┤
│     Free     │       20        │     $0.00    │
│     Basic    │      800        │    $29.99    │
│      Pro     │    3,000        │    $79.99    │
│  Enterprise  │   10,000        │   $129.99    │
└──────────────┴─────────────────┴──────────────┘

Transparent, Token-Based Pricing

Tokens deducted per page, not per document - a 10-page PDF uses 10 tokens
No charges for failed extractions - you only consume tokens for successful page processing
Fixed monthly allowance - each plan includes a set number of pages per month
Cancel anytime with no penalties or long-term commitments

Our pricing is designed to be cheap compared to traditional document processing services that charge per document or require expensive enterprise contracts. With PdfParse, you get production-ready extraction at a fraction of the cost.

Get Started Today

Ready to transform your document workflow? Sign up for a free account at pdfparse.net and start extracting structured data in minutes. Prefer to talk to a human? Contact us and we’ll help you set up your first project.

Questions? Reach out to us at support@pdfparse.net - we'd love to hear about your use case.

Deploying the Phoenix Api to DigitalOcean using mix release

Pablo Grant — Tue, 14 Jul 2020 10:15:25 +0000

This is a personalized report that explains how to deploy an api to a Digital Ocean ubuntu server with minimal technical explanation.

Log into your server with your ssh credentials.

e.g: ssh root@<yoururl>.com

pull your code from your preferred branch.
run mix deps.get to update your dependencies. You can then either run mix ecto.reset or mix ecto.migrate to reset/update your database depending on what you want to do. (You may need to disable any active services using the database during this time.)
run MIX_ENV=<env> mix release to update your release folder.
restart your systemctl service.
Check the endpoint to see if it works as intended by checking /

Quickly install elixir using asdf

Pablo Grant — Thu, 09 Apr 2020 04:54:56 +0000

ASDF is a cli used to manage the installation and runtime versions of several popular languages. I've recently found out about asdf, and used it to run a quick update on my local machine. This article will provide a quick tutorial on the installation and update process.

1. Install ASDF.

To install asdf, first clone the most recent stable release from the github repository:

git clone https://github.com/asdf-vm/asdf.git ~/.asdf --branch v0.7.8

Then, add it as an environment variable to bash

echo -e '\n. $HOME/.asdf/asdf.sh' >> ~/.bashrc

2. Install Erlang and Elixir

You can then install any language from these list of plugins. We will be installing the most recent version of erlang and elixir using these plugins.

Install erlang:

asdf plugin-add erlang https://github.com/asdf-vm/asdf-erlang.git

Install elixir:

To install elixir, the installation instructions dictate you must have unzip installed first. Run the following commands
sudo apt install unzip && asdf plugin-add elixir https://github.com/asdf-vm/asdf-elixir.git

Optional: Install hex and rebar.

You can install hex and rebar by running the following command
mix local.hex --if-missing --force && mix local.rebar --if-missing --force

Final: Set up your .tool-versions.

The .tool-versions file is the file which asdf uses to determine which installed version of the language your project uses. U can set up a global .tool-versions for asdf to fall back on if your current folder doesn't have a .tool-versions present. Run the following command.

echo -e '\n elixir 1.10.2 \n mix 1.10.2' >> $HOME/.tool-versions