# Build an Invoice Pipeline in n8n With a Review Branch > Import a tested n8n workflow that validates extracted invoice fields and routes exceptions to review, then swap in Gmail, Datalab and Sheets. - Source: https://www.zarifautomates.com/blog/invoice-pipeline-in-n8n - Published: 2026-09-26 - Updated: 2026-09-26 - Pillar: Automation Workflows - Tags: document ai, document processing, ai invoice processing workflow, n8n workflow, invoice ocr - Author: Zarif --- An invoice pipeline is easy to demo and easy to get wrong. The extraction call is the simple part. The work is in deciding which results are allowed to reach the spreadsheet on their own and which ones a person has to look at first. This lesson gives you that decision as an importable n8n workflow. Eight simulated extraction results pass through a validation node and a routing node. Two land as accepted invoice rows. Six land in a review queue, each with the reason it was stopped. Then the second half of the lesson shows how to replace the simulated parts with a Gmail inbox, the Datalab extraction API and Google Sheets. The fixture workflow was imported and executed with **n8n 2.40.7 on Node 24.21.0 on September 26, 2026**. It uses the real n8n engine and JavaScript Code nodes. The extraction results are synthetic, shaped like the responses in Datalab's docs, so the run shows routing behavior, not extraction quality. The live nodes in the second half are configured from n8n's and Datalab's docs and were not executed. This is part 6 of the Document AI series. Parts 1 to 5 cover what these APIs do and how the vendors compare. This one puts one of them to work. If you have not picked an extractor yet, start with [Extend vs Datalab vs other parsers](/blog/extend-vs-datalab-vs-other-document-parsers). ## 1. The shape of the pipeline ```text Live version (you build this in section 5) Gmail Trigger → Submit to Datalab → Wait → Check extraction status → Attach source → Validate invoice fields → Needs review? ├─ true → Google Sheets: Review tab └─ false → Google Sheets: Invoices tab Lab version (you import this now) Run the lab → Simulated Datalab results → Validate invoice fields → Needs review? ├─ true → Review queue rows └─ false → Accepted invoice rows ``` The two versions share the middle. **Validate invoice fields** and **Needs review?** are identical in both, so the rules you test with fixtures are the rules that run on real invoices. Only the ends change: a Manual Trigger and a Code node full of fixtures become an email trigger and two HTTP calls, and the two terminal Code nodes become Google Sheets writes. ## 2. Import and run the lab workflow Copy the JSON at the end of this section into a file named `invoice-pipeline.json`. In the n8n editor, create a new workflow, open its menu, choose **Import from File** and pick the file. The [import guide](https://docs.n8n.io/build/manage-workflows/export-and-import) covers the options. The workflow has `active: false`, a Manual Trigger, no credentials and no HTTP Request node, so nothing leaves your machine. Select **Execute Workflow**. The **Needs review?** node should show 6 items on its true output and 2 on its false output. To run it from the command line instead, use an empty directory. n8n 2.40.7 requires Node 24 or later, and it refused to start on Node 22 in this run. ```bash node --version # must print v24 or later npm init -y npm install n8n@2.40.7 export N8N_USER_FOLDER="$PWD/n8n-test-state" export N8N_DIAGNOSTICS_ENABLED=false export N8N_VERSION_NOTIFICATIONS_ENABLED=false export N8N_TEMPLATES_ENABLED=false export N8N_PERSONALIZATION_ENABLED=false export N8N_RUNNERS_BROKER_LISTEN_ADDRESS=127.0.0.1 export N8N_RUNNERS_BROKER_PORT=15691 npx n8n import:workflow --input=invoice-pipeline.json npx n8n execute --id=ZarifInvoiceLab01 --rawOutput ``` The separate `N8N_USER_FOLDER` keeps this test out of any n8n database you already use, since importing a workflow ID that already exists can replace it. The [CLI docs](https://docs.n8n.io/deploy/host-n8n/configure-n8n/use-the-command-line) describe `import:workflow` and `execute`. Two practical notes from the run: if you installed n8n under an older Node and then switched, run `npm rebuild isolated-vm` or the expression engine fails to start, and a warning that the Python task runner is missing is harmless here because every Code node is JavaScript. The full export: ```json { "id": "ZarifInvoiceLab01", "name": "Invoice pipeline, local fixtures", "active": false, "nodes": [ { "id": "1", "name": "Run the lab", "type": "n8n-nodes-base.manualTrigger", "typeVersion": 1, "position": [ 0, 0 ], "parameters": {} }, { "id": "2", "name": "Simulated Datalab results", "type": "n8n-nodes-base.code", "typeVersion": 2, "position": [ 220, 0 ], "parameters": { "mode": "runOnceForAllItems", "jsCode": "// SIMULATED: each item mimics a completed Datalab /api/v1/extract poll\n// response in fast mode (fields, _citations, _score). No API is called.\nconst ok = (fields, scores) => {\n const out = {};\n for (const [key, value] of Object.entries(fields)) {\n out[key] = value;\n out[`${key}_citations`] = [`block_${key}`];\n if (scores) out[`${key}_score`] = { score: scores[key] ?? 5, reasoning: 'fixture' };\n }\n return out;\n};\nconst lines = [\n { description: 'Paper, A4, 10 boxes', quantity: 10, unit_price: 40, amount: 400 },\n { description: 'Toner cartridge', quantity: 4, unit_price: 150, amount: 600 },\n];\nconst base = { invoice_number: 'INV-1001', vendor_name: 'Acme Supply Co.', invoice_date: '2026-09-14', currency: 'USD', subtotal: 1000, tax: 80, total_amount: 1080, line_items: lines };\nconst cases = [\n ['clean', base, {}],\n ['total_mismatch', { ...base, invoice_number: 'INV-1002', total_amount: 1008 }, {}],\n ['missing_number', { ...base, invoice_number: null }, {}],\n ['low_confidence', { ...base, invoice_number: 'INV-1004' }, { total_amount: 2 }],\n ['duplicate', base, {}],\n ['failed', null, null],\n ['no_scores', { ...base, invoice_number: 'INV-1007' }, null],\n ['clean_eur', { ...base, invoice_number: 'G-7731', vendor_name: 'Globex GmbH', currency: 'EUR' }, {}],\n];\nreturn cases.map(([name, fields, scores], i) => {\n const source = { message_id: `msg-${i + 1}`, filename: `${name}.pdf` };\n if (!fields) {\n return { json: { source, poll: { status: 'failed', success: false, error: 'fixture: document could not be processed' } } };\n }\n const extraction = ok(fields, scores);\n const values = Object.entries(extraction).filter(([k]) => k.endsWith('_score')).map(([, v]) => v.score);\n return {\n json: {\n source,\n poll: {\n status: 'complete',\n success: true,\n page_count: 1,\n extraction_schema_json: JSON.stringify(extraction),\n extraction_score_average: values.length ? values.reduce((a, b) => a + b, 0) / values.length : null,\n },\n },\n };\n});\n" } }, { "id": "3", "name": "Validate invoice fields", "type": "n8n-nodes-base.code", "typeVersion": 2, "position": [ 440, 0 ], "parameters": { "mode": "runOnceForAllItems", "jsCode": "const REQUIRED = ['invoice_number', 'vendor_name', 'invoice_date', 'currency', 'total_amount'];\nconst CURRENCIES = ['USD', 'EUR'];\nconst MIN_SCORE = 4;\nconst cents = (n) => Math.round(Number(n) * 100);\nconst seen = new Set();\n\nreturn $input.all().map((item, index) => {\n const { source, poll } = item.json;\n const reasons = [];\n let fields = {};\n\n if (poll.status !== 'complete' || poll.success === false) {\n reasons.push(`extraction_failed: ${poll.error ?? poll.status}`);\n } else {\n try {\n fields = typeof poll.extraction_schema_json === 'string'\n ? JSON.parse(poll.extraction_schema_json)\n : poll.extraction_schema_json;\n } catch (error) {\n reasons.push('unparseable_extraction');\n }\n }\n\n if (reasons.length === 0) {\n for (const key of REQUIRED) {\n if (fields[key] === null || fields[key] === undefined || fields[key] === '') reasons.push(`missing:${key}`);\n }\n const date = fields.invoice_date;\n if (date && (!/^\\d{4}-\\d{2}-\\d{2}$/.test(date) || Number.isNaN(Date.parse(date)))) reasons.push('invalid_date');\n if (fields.currency && !CURRENCIES.includes(fields.currency)) reasons.push('unexpected_currency');\n\n const lineSum = (fields.line_items ?? []).reduce((sum, line) => sum + cents(line.amount), 0);\n if (fields.line_items?.length && lineSum !== cents(fields.subtotal)) reasons.push('line_items_do_not_sum_to_subtotal');\n if (cents(fields.subtotal) + cents(fields.tax ?? 0) !== cents(fields.total_amount)) reasons.push('subtotal_plus_tax_not_total');\n\n if (poll.extraction_score_average === null || poll.extraction_score_average === undefined) {\n reasons.push('scores_unavailable');\n } else {\n for (const key of REQUIRED) {\n const score = fields[`${key}_score`]?.score;\n if (typeof score === 'number' && score < MIN_SCORE) reasons.push(`low_confidence:${key}`);\n }\n }\n\n const key = `${String(fields.vendor_name).toLowerCase()}|${fields.invoice_number}`;\n if (fields.invoice_number && seen.has(key)) reasons.push('duplicate_in_batch');\n seen.add(key);\n }\n\n const invoice = Object.fromEntries(\n [...REQUIRED, 'subtotal', 'tax'].map((key) => [key, fields[key] ?? null]),\n );\n return {\n json: { ...source, ...invoice, reasons, needs_review: reasons.length > 0 },\n pairedItem: { item: index },\n };\n});\n" } }, { "id": "4", "name": "Needs review?", "type": "n8n-nodes-base.if", "typeVersion": 2.2, "position": [ 660, 0 ], "parameters": { "conditions": { "options": { "caseSensitive": true, "leftValue": "", "typeValidation": "strict", "version": 2 }, "conditions": [ { "id": "needs-review", "leftValue": "={{ $json.needs_review }}", "rightValue": true, "operator": { "type": "boolean", "operation": "true", "singleValue": true } } ], "combinator": "and" }, "options": {} } }, { "id": "5", "name": "Review queue rows", "type": "n8n-nodes-base.code", "typeVersion": 2, "position": [ 880, -100 ], "parameters": { "mode": "runOnceForAllItems", "jsCode": "return $input.all().map((item, index) => ({\n json: {\n message_id: item.json.message_id,\n source_file: item.json.filename,\n reasons: item.json.reasons.join('; '),\n extracted: JSON.stringify({\n invoice_number: item.json.invoice_number,\n vendor_name: item.json.vendor_name,\n total_amount: item.json.total_amount,\n currency: item.json.currency,\n }),\n status: 'awaiting_review',\n decision: '',\n reviewer: '',\n },\n pairedItem: { item: index },\n}));\n" } }, { "id": "6", "name": "Accepted invoice rows", "type": "n8n-nodes-base.code", "typeVersion": 2, "position": [ 880, 100 ], "parameters": { "mode": "runOnceForAllItems", "jsCode": "return $input.all().map((item, index) => ({\n json: {\n invoice_number: item.json.invoice_number,\n vendor_name: item.json.vendor_name,\n invoice_date: item.json.invoice_date,\n currency: item.json.currency,\n total_amount: item.json.total_amount,\n source_file: item.json.filename,\n message_id: item.json.message_id,\n status: 'accepted',\n },\n pairedItem: { item: index },\n}));\n" } } ], "connections": { "Run the lab": { "main": [ [ { "node": "Simulated Datalab results", "type": "main", "index": 0 } ] ] }, "Simulated Datalab results": { "main": [ [ { "node": "Validate invoice fields", "type": "main", "index": 0 } ] ] }, "Validate invoice fields": { "main": [ [ { "node": "Needs review?", "type": "main", "index": 0 } ] ] }, "Needs review?": { "main": [ [ { "node": "Review queue rows", "type": "main", "index": 0 } ], [ { "node": "Accepted invoice rows", "type": "main", "index": 0 } ] ] } }, "settings": { "executionOrder": "v1" }, "pinData": {}, "tags": [] } ``` ## 3. What each node does | Node | Type | What to inspect | | --- | --- | --- | | Run the lab | Manual Trigger | Starts one execution when you click | | Simulated Datalab results | Code | Eight fixtures shaped like a completed Datalab extract poll in fast mode. **Simulated.** | | Validate invoice fields | Code | Required fields, date format, currency list, arithmetic, confidence scores, duplicates in the batch | | Needs review? | If | Strict boolean check on `needs_review` | | Review queue rows | Code | One row per exception, with reasons joined and the extracted values kept | | Accepted invoice rows | Code | One clean row per invoice for the ledger sheet | The fixtures follow the response format in [Datalab's structured extraction docs](https://documentation.datalab.to/docs/recipes/structured-extraction/api-overview). A completed poll carries `status`, `success`, `page_count` and `extraction_schema_json`, a JSON string holding your fields plus a `_citations` list for each. In fast mode, [Datalab's confidence scoring](https://documentation.datalab.to/docs/recipes/structured-extraction/confidence-scoring) adds a `_score` object per field, from 1 for no clear evidence to 5 for high confidence, and an `extraction_score_average`. The docs, read on 2026-09-26, note that scoring is in beta, runs only in fast mode, and can fail on its own, so a missing average means "unavailable", not "zero". That last detail is why the validator treats a missing score as its own review reason. Treating it as a pass would let a result through with no confidence signal at all. Treating it as a failure would hide the difference between "the model was unsure" and "the score never arrived". Here is the validation logic from the Code node, formatted for reading. It runs once for all items, which is what lets it spot duplicates across the batch. ```javascript const REQUIRED = ['invoice_number', 'vendor_name', 'invoice_date', 'currency', 'total_amount']; const CURRENCIES = ['USD', 'EUR']; const MIN_SCORE = 4; const cents = (n) => Math.round(Number(n) * 100); const seen = new Set(); return $input.all().map((item, index) => { const { source, poll } = item.json; const reasons = []; let fields = {}; if (poll.status !== 'complete' || poll.success === false) { reasons.push(`extraction_failed: ${poll.error ?? poll.status}`); } else { try { fields = typeof poll.extraction_schema_json === 'string' ? JSON.parse(poll.extraction_schema_json) : poll.extraction_schema_json; } catch (error) { reasons.push('unparseable_extraction'); } } if (reasons.length === 0) { for (const key of REQUIRED) { if (fields[key] === null || fields[key] === undefined || fields[key] === '') reasons.push(`missing:${key}`); } const date = fields.invoice_date; if (date && (!/^\d{4}-\d{2}-\d{2}$/.test(date) || Number.isNaN(Date.parse(date)))) reasons.push('invalid_date'); if (fields.currency && !CURRENCIES.includes(fields.currency)) reasons.push('unexpected_currency'); const lineSum = (fields.line_items ?? []).reduce((sum, line) => sum + cents(line.amount), 0); if (fields.line_items?.length && lineSum !== cents(fields.subtotal)) reasons.push('line_items_do_not_sum_to_subtotal'); if (cents(fields.subtotal) + cents(fields.tax ?? 0) !== cents(fields.total_amount)) reasons.push('subtotal_plus_tax_not_total'); if (poll.extraction_score_average === null || poll.extraction_score_average === undefined) { reasons.push('scores_unavailable'); } else { for (const key of REQUIRED) { const score = fields[`${key}_score`]?.score; if (typeof score === 'number' && score < MIN_SCORE) reasons.push(`low_confidence:${key}`); } } const key = `${String(fields.vendor_name).toLowerCase()}|${fields.invoice_number}`; if (fields.invoice_number && seen.has(key)) reasons.push('duplicate_in_batch'); seen.add(key); } const invoice = Object.fromEntries( [...REQUIRED, 'subtotal', 'tax'].map((key) => [key, fields[key] ?? null]), ); return { json: { ...source, ...invoice, reasons, needs_review: reasons.length > 0 }, pairedItem: { item: index }, }; }); ``` Every check is deterministic. The model reads the invoice. Plain code decides whether the reading is fit to store. Amounts are compared in whole cents so that floating-point rounding cannot pass a total that is off by a fraction. ## 4. Read the eight results | Fixture | What is wrong | Route | Reason recorded | | --- | --- | --- | --- | | `clean.pdf` | Nothing | Accepted | | | `total_mismatch.pdf` | Total 1,008 where subtotal plus tax is 1,080 | Review | `subtotal_plus_tax_not_total` | | `missing_number.pdf` | No invoice number | Review | `missing:invoice_number` | | `low_confidence.pdf` | Total scored 2 of 5 | Review | `low_confidence:total_amount` | | `duplicate.pdf` | Same vendor and number as `clean.pdf` | Review | `duplicate_in_batch` | | `failed.pdf` | Extraction job failed | Review | `extraction_failed: ...` | | `no_scores.pdf` | Scoring never arrived | Review | `scores_unavailable` | | `clean_eur.pdf` | Nothing, in euros | Accepted | | An accepted row from the recorded run: ```json { "invoice_number": "INV-1001", "vendor_name": "Acme Supply Co.", "invoice_date": "2026-09-14", "currency": "USD", "total_amount": 1080, "source_file": "clean.pdf", "message_id": "msg-1", "status": "accepted" } ``` These counts describe the rules, not an extractor. Eight fixtures in, six to review, two accepted. A real inbox will have a very different mix, and [measuring extraction accuracy field by field](/blog/measuring-document-extraction-accuracy) is how you find out what yours is. **Change one rule and rerun.** Open **Validate invoice fields** and change `const MIN_SCORE = 4;` to `const MIN_SCORE = 2;`. Execute again. The low-confidence fixture now passes, and **Needs review?** shows 5 and 3. That one constant is your review threshold, and moving it trades reviewer time for risk. Restore it before you continue. ## 5. Swap in the live nodes This half is configuration drawn from the docs, not a recorded run. Build it in a copy of the workflow, keep the copy inactive, and test it with a handful of invoices you are allowed to send to a third-party API. **Intake: Gmail Trigger.** Replace **Run the lab** with a [Gmail Trigger](https://docs.n8n.io/integrations/builtin/trigger-nodes/n8n-nodes-base.gmailtrigger/). Use the **Search** filter to narrow it, for example to messages with a PDF attachment sent to an invoices address. Turn **Simplify** off. In the node's 2.40.7 source, the **Options** collection with **Download Attachments** only appears then. Switch **Download Attachments** on. The first attachment arrives as binary data named `attachment_0`, from the default **Attachment Prefix**. If invoices land in a folder instead, n8n's [Local File Trigger](https://docs.n8n.io/integrations/builtin/core-nodes/n8n-nodes-base.localfiletrigger/) watches a path, but its docs say it is self-hosted only and disabled by default from n8n 2.0. **Extract: HTTP Request to Datalab.** Replace **Simulated Datalab results** with three nodes. 1. **Submit to Datalab**, an [HTTP Request](https://docs.n8n.io/integrations/builtin/core-nodes/n8n-nodes-base.httprequest/) node. Method POST, URL `https://www.datalab.to/api/v1/extract`. Authentication: a Header Auth credential with the header name `X-API-Key`, which is the header Datalab's docs use. Body Content Type **Form-Data**. Add a parameter of type **n8n Binary File** named `file` with **Input Data Field Name** set to `attachment_0`. Add text parameters `page_schema`, holding your JSON schema as a string, and `extraction_mode` set to `fast` so scores come back. 2. **Wait**, a [Wait](https://docs.n8n.io/integrations/builtin/core-nodes/n8n-nodes-base.wait/) node set to **After Time Interval**, a few seconds. 3. **Check extraction status**, a second HTTP Request with method GET, the same credential, and the URL set to the `request_check_url` from the submit response. Follow it with an [If](https://docs.n8n.io/integrations/builtin/core-nodes/n8n-nodes-base.if/) node that sends anything whose `status` is not `complete` or `failed` back to **Wait**. Cap the loop with a counter so a stuck job becomes a review item instead of an endless execution. The Datalab docs also list a `webhook_url` parameter on the extract endpoint, which replaces polling with a callback into an n8n Webhook node. That splits the pipeline into two workflows, so start with polling. **Attach source.** The validator expects each item as `source` plus `poll`. Add an Edit Fields (Set) or Code node that builds `source` from the Gmail message ID and attachment file name, and sets `poll` to the status response. Keep the message ID: it is what a reviewer uses to find the original email. **Validate and route.** Leave **Validate invoice fields** and **Needs review?** as they are. Note that the duplicate check only sees one execution. Each email starts a fresh execution, so a resent invoice would pass. Before going live, look up the vendor and invoice number in the Invoices sheet or a database, and route matches to review. **Write: Google Sheets.** Replace both terminal Code nodes with [Google Sheets](https://docs.n8n.io/integrations/builtin/app-nodes/n8n-nodes-base.googlesheets/) nodes using the **Append Row** operation, one pointed at an Invoices tab and one at a Review tab. Keep the Code nodes' output shape as your column list. The review tab becomes the queue: a reviewer fills `decision` and `reviewer`, and a separate workflow picks up approved rows. The [review queue post](/blog/designing-the-document-review-queue) covers what that screen needs. **Errors.** Set **Retry On Fail** in the settings of both HTTP Request nodes for rate limits, and add an error workflow with an [Error Trigger](https://docs.n8n.io/integrations/builtin/core-nodes/n8n-nodes-base.errortrigger/) so a failed execution alerts someone. On 2026-09-26 Datalab's [limits page](https://documentation.datalab.to/docs/common/limits) listed 10 requests per minute for the free tier and its pricing page listed 25, so plan for the lower number until you confirm yours. The [ticket-triage lab](/blog/how-to-use-n8n-for-advanced-ai-workflow-automation) explains why an expected rejection and an unexpected crash deserve different handling. ## 6. Before real invoices go through - **Test the live half with the same eight cases.** Make PDFs that reproduce each fixture, send them through, and check you get the same routes. If a fixture that should go to review lands in the Invoices tab, stop there. - **Keep the original file.** Store the attachment or its Gmail message ID with every row, accepted or not. An auditor or a reviewer will need it. - **Decide what the sheet is.** A sheet is fine for a few dozen invoices a week. Past that, or once two people review at the same time, move the queue to a database with a unique key on vendor and invoice number. - **Check the data terms.** Sending invoices to any extraction API sends vendor names, bank details and amounts to a third party. Confirm retention and processing location with the vendor first. Your deliverable is the imported lab workflow, the 6 and 2 baseline, and the threshold change with its 5 and 3 result. Next in the series: [measure extraction accuracy field by field](/blog/measuring-document-extraction-accuracy), so the threshold you just moved rests on numbers. The terms used here are defined in the [Document AI glossary](/blog/document-ai-glossary). For background on Datalab's modes, see [Datalab in practice](/blog/datalab-in-practice-marker-surya-chandra).