SMB finance teams manually key invoice data from PDF attachments into NetSuite, wasting hours and introducing errors. A reliable document pipeline that handles messy scans and formats saves time and prevents AP mistakes.
A complete, working implementation of this recipe — downloadable as a zip or browsable file by file. Generated by our build pipeline; tested with full coverage before publishing.
This tutorial walks you through building an automated invoice processing pipeline for NetSuite. You’ll create a Next.js API endpoint that receives invoice PDFs from a NetSuite webhook, extracts text using unpdf and tesseract.js, parses structured data with Mistral AI, repairs any malformed JSON with @reaatech/structured-repair-core, posts the result to NetSuite as a draft vendor bill, and deduplicates incoming invoices using @reaatech/agent-memory — all traced with Langfuse observability.
Prerequisites
Node.js >= 22 and pnpm 10.x installed
A Mistral AI API key — set it as MISTRAL_API_KEY
An OpenAI API key for agent-memory embeddings — set it as OPENAI_API_KEY
NetSuite REST API credentials: account ID, consumer key/secret, token ID/secret, and base URL
(Optional) Langfuse project keys (LANGFUSE_PUBLIC_KEY, LANGFUSE_SECRET_KEY) for observability
Basic familiarity with TypeScript, Next.js App Router, and NetSuite’s REST record API
Step 1: Scaffold the project
Start by creating a new Next.js project and installing the recipe’s dependencies. The recipe ships with pre-configured package.json, tsconfig.json, vitest.config.ts, and next.config.ts.
terminal
mkdir invoice-pipeline && cd invoice-pipeline
The starter files are included in the recipe download. Once you have them in place, install the dependencies:
terminal
pnpm install
Expected output: All packages install without errors. The project uses Node.js >= 22 with exact version pins — no ^ or ~ ranges.
Step 2: Configure environment variables
The pipeline reads all its secrets and endpoints from environment variables at startup. Copy the .env.example file and fill in your real values.
terminal
cp .env.example .env
Here is what your .env file should look like with placeholders filled in:
Expected output: A .env file with all 14 variables set. The file is listed in .gitignore so you never commit real secrets.
Step 3: Define the invoice schemas with Zod
Start with the data contracts. You’ll define four Zod schemas that describe the shape of an invoice line item, a vendor, a full invoice document, and a slimmed-down extraction variant.
Expected output: The file compiles without errors. Note total: z.coerce.number() — this lets Mistral return "1499.99" (a string) and Zod coerces it to 1499.99. The InvoiceExtractionSchema omits the notes field because Mistral shouldn’t be asked to produce it.
Step 4: Create the SHA-256 hashing utility
You need a deterministic way to fingerprint each incoming PDF so the pipeline can detect duplicates before processing. This utility computes a SHA-256 hex digest of a buffer.
Create src/lib/hash-util.ts:
ts
import { createHash } from "node:crypto";export function sha256Hex(data: Buffer | Uint8Array): string { return createHash("sha256").update(data).digest("hex");}export function hashInvoicePdf(pdfBuffer: Buffer): string { return sha256Hex(pdfBuffer);}
Expected output:sha256Hex(Buffer.from("hello")) returns "2cf24dba5fb0a30e26e83b2ac5b9e29e1b161e5c1fa7425e73043362938b9824". Empty buffers produce the known SHA-256 of an empty input.
Step 5: Build the PDF text extraction service
This service handles both text-based and scanned PDFs. For text PDFs it uses unpdf to extract the text directly. For scanned documents it preprocesses each page image with sharp (grayscale + normalize) and runs OCR with tesseract.js.
Create src/services/pdf-extractor.ts:
ts
import { extractText, getDocumentProxy } from "unpdf";import sharp from "sharp";import { createWorker } from "tesseract.js";export class EmptyPdfError extends Error { constructor(message: string = "PDF is empty or contains no pages") { super(message); this.name = "EmptyPdfError"; }}export class ExtractionError extends Error { constructor(message: string, cause?: unknown) { super(message); this.name = "ExtractionError"; this.cause = cause; }}export async function extractTextFromPdf(pdfBuffer: Buffer): Promise<string> { if (pdfBuffer.length === 0) { throw new EmptyPdfError(); } try { const pdf = await getDocumentProxy(new Uint8Array(pdfBuffer)); const { totalPages, text } = await extractText(pdf, { mergePages: true }); if (totalPages === 0) { throw new EmptyPdfError(); } return text; } catch (error) { if (error instanceof EmptyPdfError) throw error; throw new ExtractionError("Failed to extract text from PDF", error); }}export async function preprocessImage(imageBuffer: Buffer): Promise<Buffer> { try { return await sharp(imageBuffer).grayscale().normalize().png().toBuffer(); } catch (error) { throw new ExtractionError("Failed to preprocess image", error); }}export async function ocrImage(imageBuffer: Buffer, lang?: string): Promise<string> { let worker; try { worker = await createWorker(lang ?? "eng"); const ret = await worker.recognize(imageBuffer); return ret.data.text; } catch (error) { throw new ExtractionError("OCR processing failed", error); } finally { if (worker) { await worker.terminate(); } }}export async function extractTextFromScannedPdf(imageBuffers: Buffer[]): Promise<string> { if (imageBuffers.length === 0) { throw new EmptyPdfError("No image buffers provided for scanned PDF"); } const results: string[] = []; for (const imageBuffer of imageBuffers) { const processed = await preprocessImage(imageBuffer); const text = await ocrImage(processed); results.push(text); } return results.join("\n\n");}export async function extractInvoiceText(pdfBuffer: Buffer, isScanned: boolean, imageBuffers?: Buffer[]): Promise<string> { if (isScanned) { if (!imageBuffers || imageBuffers.length === 0) { throw new ExtractionError("Scanned PDF extraction requires image buffers"); } return extractTextFromScannedPdf(imageBuffers); } return extractTextFromPdf(pdfBuffer);}
Expected output: A service that dispatches correctly — extractInvoiceText(buf, false) calls extractTextFromPdf, while extractInvoiceText(buf, true, [imgBuf]) preprocesses and OCRs each image page.
Step 6: Orchestrate Mistral AI document extraction
This step wires Mistral’s chat API to extract structured invoice data from raw text. It also initializes the @reaatech/media-pipeline-mcp-doc-extraction package for document pipeline operations.
Create src/services/document-extractor.ts:
ts
import { Mistral } from "@mistralai/mistralai";import { createDocumentExtractionOperations } from "@reaatech/media-pipeline-mcp-doc-extraction";import { ArtifactRegistry } from "@reaatech/media-pipeline-mcp-core";import { createStorage } from "@reaatech/media-pipeline-mcp-storage";const INVOICE_EXTRACTION_PROMPT = `You are an invoice data extraction assistant. Extract structured JSON from the following invoice text.Return a JSON object with these fields:- invoiceNumber (string)- invoiceDate (string)- dueDate (string, optional)- vendor: { name (string), taxId (string, optional), address (string, optional), email (string, optional) }- subtotal (number)- tax (number, optional)- total (number)- currency (string, optional)- lineItems: array of { description (string), quantity (number), unitPrice (number), amount (number), taxCode (string, optional) }Return ONLY valid JSON with no additional text or markdown formatting.`;export class MistralParser { private mistral: Mistral; constructor(apiKey: string | undefined) { this.mistral = new Mistral({ apiKey: apiKey ?? process.env["MISTRAL_API_KEY"] ?? "" }); const registry = new ArtifactRegistry(); const storage = createStorage({ type: "local", config: { basePath: "/tmp" } }); createDocumentExtractionOperations(registry, storage); } async parseInvoiceText(rawText: string): Promise<string> { const result = await this.mistral.chat.complete({ model: "mistral-large-latest", messages: [ { role: "system", content: INVOICE_EXTRACTION_PROMPT }, { role: "user", content: `Extract invoice data from this text:\n\n${rawText}` }, ], responseFormat: { type: "json_object" }, }); const choice = result.choices[0]; if (!choice.message) { return ""; } const messageContent = choice.message.content; if (messageContent === null || messageContent === undefined) { return ""; } if (typeof messageContent === "string") { return messageContent; } return ""; } async extractFieldsFromText(rawText: string, schema: Record<string, unknown>): Promise<string> { const result = await this.mistral.chat.complete({ model: "mistral-large-latest", messages: [ { role: "system", content: "Extract structured data from the following text according to the provided JSON schema. Return ONLY valid JSON matching the schema." }, { role: "user", content: rawText }, ], responseFormat: { type: "json_schema", jsonSchema: { name: "extracted_fields", schemaDefinition: schema, strict: false, }, }, }); const choice = result.choices[0]; if (!choice.message) { return ""; } const messageContent = choice.message.content; if (messageContent === null || messageContent === undefined) { return ""; } if (typeof messageContent === "string") { return messageContent; } return ""; }}
Expected output: A parser that sends the invoice text to Mistral’s mistral-large-latest model with responseFormat: { type: "json_object" }. The response is valid JSON (or an empty string on failure).
Step 7: Repair malformed JSON with structured-repair-core
LLM output is rarely perfect JSON on the first try. The @reaatech/structured-repair-core package applies a battery of strategies — stripping markdown fences, fixing broken syntax, coercing types, fuzzy-matching keys, and removing hallucinated fields — to turn Mistral’s raw output into schema-conforming data.
Expected output: A repairer that can fix JSON like ```json {"invoiceNumber":"INV-001"...} ``` (fence-stripped), {'invoiceNumber':'INV-001'} (single-quotes fixed), and JSON with trailing commas.
Step 8: Add idempotency with agent-memory
Use @reaatech/agent-memory to track which invoices have already been processed. Each PDF’s SHA-256 hash is stored in memory so the pipeline can short-circuit duplicates before spending tokens on LLM calls.
Expected output: A memory service that returns true for isDuplicate after you’ve called markProcessed with the same hash.
Step 9: Integrate Langfuse observability
Trace every stage of the pipeline with Langfuse. The observability module initializes a Langfuse client and provides helpers to create traces, spans, and finalize them with status.
Create src/lib/observability.ts:
ts
import Langfuse from "langfuse";let langfuseClient: Langfuse | null = null;const traceMap: Map<string, LangfuseTraceClient> = new Map();const spanMap: Map<string, LangfuseSpanClient> = new Map();type LangfuseTraceClient = ReturnType<Langfuse["trace"]>;type LangfuseSpanClient = ReturnType<LangfuseTraceClient["span"]>;export function initObservability(): void { if (langfuseClient) return; const publicKey = process.env["LANGFUSE_PUBLIC_KEY"]; const secretKey = process.env["LANGFUSE_SECRET_KEY"]; if (!publicKey || !secretKey) { console.warn("Langfuse keys not configured — observability disabled"); return; } langfuseClient = new Langfuse({ publicKey, secretKey });}export function createTrace(name: string, metadata?: Record<string, unknown>) { if (!langfuseClient) return null; const trace = langfuseClient.trace({ name, metadata }); traceMap.set(trace.id, trace); return trace.id;}export function createSpan(traceId: string, name: string, metadata?: Record<string, unknown>) { if (!langfuseClient) return null; const trace = traceMap.get(traceId); if (!trace) { console.warn(`Trace ${traceId} not found`); return null; } const span = trace.span({ name, metadata }); spanMap.set(span.id, span); return span.id;}export function finalizeSpan(spanId: string, status?: "ok" | "error", metadata?: unknown): void { const span = spanMap.get(spanId); if (!span) { console.warn(`Span ${spanId} not found`); return; } if (status) { console.debug(`Span ${spanId} finalized with status: ${status}`, metadata); } span.end(); spanMap.delete(spanId);}
Expected output: If LANGFUSE_PUBLIC_KEY and LANGFUSE_SECRET_KEY are set, the pipeline emits traces and spans. If they’re missing, observability is silently disabled and createTrace/createSpan return null.
Step 10: Create the NetSuite API client with OAuth 1.0
NetSuite’s REST API uses OAuth 1.0 with HMAC-SHA256 signing. This client constructs the authorization header, signs each request, and provides typed methods for vendor bill operations.
Expected output: A client that authenticates every request with an OAuth 1.0 Authorization header containing a realm, consumer key, token, signature, timestamp, and nonce. Non-2xx responses throw a NetSuiteApiError with the status code and response body.
Step 11: Wire the pipeline orchestrator
The InvoicePipeline class ties everything together. It runs each stage in order — hash, dedup check, PDF extraction, Mistral parsing, JSON repair, NetSuite posting, memory recording — with per-step timing and Langfuse tracing.
Create src/services/invoice-pipeline.ts:
ts
import { z } from "zod";import { sha256Hex } from "../lib/hash-util.js";import { createTrace, createSpan, finalizeSpan } from "../lib/observability.js";import { extractInvoiceText } from "./pdf-extractor.js";import { MistralParser } from "./document-extractor.js";import { InvoiceRepairer, UnrepairableError } from "./invoice-repairer.js";import { NetSuiteApiClient } from "../lib/netsuite-api.js";import { InvoiceMemoryService } from "./memory-service.js";import { InvoiceDataSchema } from "../schemas/invoice-schema.js";export interface ProcessInvoiceResult { status: "success" |
Expected output: A fully wired pipeline. A successful run returns { status: "success", invoiceId: "123", hash: "abc..." }. A duplicate returns { status: "duplicate", hash: "abc..." }. Any failure returns { status: "error", stage: "<name>", error: "..." }.
Step 12: Create the webhook route handler and instrumentation
The API route receives multipart form data containing a PDF file, validates the webhook secret, instantiates the pipeline, and returns the result. The instrumentation file wires Langfuse initialization at server startup.
export async function register(): Promise<void> { if (process.env.NEXT_RUNTIME !== "nodejs") return; const { initObservability } = await import("./lib/observability.js"); initObservability();}
Now enable the instrumentation hook in next.config.ts. The starter config ships with an empty object — add the experimental.instrumentationHook flag so the register() function fires at startup:
ts
import type { NextConfig } from "next";const nextConfig: NextConfig = { experimental: { instrumentationHook: true, },};export default nextConfig;
Expected output: A POST /api/invoice-webhook endpoint that accepts multipart form data with a pdf field and returns { status: "success", invoiceId: "123" }. The GET endpoint returns { status: "ok" } as a health check. Langfuse initializes at server start if the environment keys are present.
Step 13: Run the tests
The project includes a comprehensive test suite with 116 tests and coverage thresholds at 90% across all categories. Run it to verify everything works:
terminal
pnpm test
Expected output: All 116 tests pass, with coverage satisfying:
Expected output: Both commands exit without errors.
Next steps
Switch agent-memory to a persistent store. Replace { provider: "memory" } with a database-backed provider (PostgreSQL, SQLite) so deduplication survives restarts.
Add a dashboard page. Create an App Router page under app/dashboard/ that lists recent invoices using the observability data.
Extend to purchase orders. The same pipeline pattern works for PO PDFs — just add a new Zod schema and a new webhook endpoint.
