feat(po-email): pre-flight duplicate check and filed labels

Checks for a duplicate before filing anything: the same document at the same site is
archived, and a PO number already in the system is tagged PO Duplicate for a human
instead of being filed twice. Successfully filed mail is tagged PO Ingested and moved
out of the inbox; PO Needs Review and PO Duplicate stay in the inbox for review and are
skipped on later runs.

docs/n8n/po-email-intake.md

@@ -2,7 +2,7 @@
 
 **Status: built and tested. INACTIVE — waiting on one thing from you (step 1 below).**
 
-Workflow: [MEGA - PO Email Intake (Microsoft Outlook)](https://wfdash.n8n.apphub.co.za/workflow/ARfOsAOy2KDUm1rp)
+Workflow: [MEGA - PO Email Intake (Microsoft Outlook)](https://wfdash.n8n.apphub.co.za/workflow/gpKjNsXoqNUzsMb4)
 Source: [`po-email-intake.workflow.js`](./po-email-intake.workflow.js) (rebuildable from this file)
 
 ---
@@ -89,10 +89,11 @@ credential `Microsoft Outlook (MEGA PO)` so it lines up with the workflow.
 
 ## Step 2 — bind the credential (once)
 
-Select `Microsoft Outlook (MEGA PO)` on these six nodes:
+Select `Microsoft Outlook (MEGA PO)` on these nine nodes:
 
 `List Inbox POs` · `List Attachments` · `Download Attachment` · `Archive Message` ·
-`Archive Message (already filed)` · `Flag Needs Review` · `Flag Needs Review (no site)`
+`Archive Message (already filed)` · `Label PO Filed` · `Flag Needs Review` ·
+`Flag Needs Review (no site)` · `Flag Needs Review (number in use)`
 
 (The three PocketBase nodes — `Read Sites`, `Duplicate Check`, `Create PO Row` — are already
 bound to the existing **`MEGAINFRASTRUCTURE`** credential. No secret sits inline in the workflow.)
@@ -157,19 +158,43 @@ review rather than filed. The other three POs that week named their site in the
 > also close the gap above, because the PO's own `Subject` line (`MEGCREQ11913 (Nellmapius EXT25)`)
 > is far more reliable than the email text.
 
-## Duplicate protection
+## Duplicate protection — checked *before* anything is filed
+
+The workflow never creates a row without checking first:
+
+| Check | How | What happens |
+| --- | --- | --- |
+| **Same document, same site** | `file_hash` + `site` on an existing row | Already filed → the email is archived, nothing is filed twice |
+| **PO number already in the system** | the number is read from the attachment filename or the subject (e.g. `MEGCPO9524.pdf`) and looked up in `purchase_orders` | Tagged **`PO Duplicate`** and left in the inbox for a human — a differing file can be a legitimate *revision* of that PO, and the same number at another site is not proof of a duplicate |
+
+The app checks both again **after** OCR, in `process-batch`: a same-file match is deleted as a
+duplicate, and a number-only match is marked `po_number_conflict` with
+`number_conflict_candidate` and left for review. So a number can never quietly end up booked
+twice — the pre-check catches it before a row exists, and the post-check catches it if the
+number only becomes readable after OCR.
+
+The placeholder `purchase_order_number` (`PENDING-…`, derived from the file hash) is also
+protected by PocketBase's unique index, so two identical uploads cannot both create a row.
+
+## What happens to the email
+
+| Outcome | Label | Where the mail ends up |
+| --- | --- | --- |
+| Filed successfully | `PO Ingested` | moved out of the inbox (Archive) |
+| Same document already filed | — | moved out of the inbox (Archive) |
+| PO number already in the system | `PO Duplicate` | stays in the inbox, for review |
+| No site could be read from the email | `PO Needs Review` | stays in the inbox, for review |
+| Filing failed | `PO Needs Review` | stays in the inbox, for review |
+
+Messages carrying `PO Ingested`, `PO Duplicate` or `PO Needs Review` are skipped on later
+runs, so a flagged email is never re-processed every 5 minutes. In Outlook, filter the inbox by
+category to see everything that still needs a human.
 
-Before filing, the workflow looks for an existing `purchase_orders` row with the **same file hash
-at the same site** (same rule the browser upload uses). A re-send of the same document is
-archived rather than filed twice. The placeholder `purchase_order_number` is derived from the
-file hash, so PocketBase's unique index is a second line of defence.
 
 ## What it deliberately does not do
 
 - Only the inbox is read — not other folders. 25 most recent messages per run.
 - Only PDF / JPEG / PNG / WEBP attachments are considered. Everything else is ignored.
-- Messages already tagged `PO Needs Review` or `PO Ingested` are skipped on later runs, so a
-  flagged email is not re-processed every 5 minutes.
 - Non-PO attachments that merely look PO-ish are flagged for review, not filed — MEGA's own
   document-kind check then refuses anything that turns out to be an invoice.
 
@@ -194,4 +219,5 @@ behave exactly like browser uploads.
 | No emails picked up | n8n execution log for `List Inbox POs` — a 401/403 means the credential is not connected or the consent was refused |
 | POs tagged `PO Needs Review` | the email did not name a site; add the site/project code to the subject or check the site table in `Read Sites` |
 | Row created but never finalised | the existing `MEGA - Purchase Order Ingest` workflow — check its `Fetch Pending Batches` / `Escalate Failure` nodes |
-| Same PO filed twice | `Duplicate Check` compares `file_hash` + `site`; a different site for the same document is deliberately not treated as a duplicate |
+| Same PO filed twice | `Duplicate Check` compares `file_hash` + `site`, and the PO number against the system. A number that is already used is tagged `PO Duplicate` rather than filed |
+| Mail not moved out of the inbox | check the `Archive Message` / `Label PO Filed` nodes in the execution log — a failed Graph call leaves the mail in the inbox and it will be retried |

docs/n8n/po-email-intake.workflow.js

@@ -5,13 +5,27 @@
  * (validate first), or rebuild by hand in the n8n UI using
  * docs/n8n/po-email-intake.md as the wiring reference.
  *
- * Live workflow: https://wfdash.n8n.apphub.co.za/workflow/ARfOsAOy2KDUm1rp
+ * Live workflow: https://wfdash.n8n.apphub.co.za/workflow/gpKjNsXoqNUzsMb4
  *
  * What it does: polls ai@megainfra.co.za for purchase-order emails, downloads the PO
  * attachment, and files it into the same PocketBase pending queue the browser upload
  * button uses, so the existing "MEGA - Purchase Order Ingest" workflow OCRs and
- * finalises it. The source email is archived once the row is filed; anything that
+ * finalises it.
- * cannot be matched to a site is tagged "PO Needs Review" in Outlook instead.
+ *
+ * Duplicate protection runs BEFORE a row is created:
+ *   - same file hash at the same site -> the identical document is already filed, the
+ *     email is archived and nothing is filed twice;
+ *   - PO number already in the system (read from the attachment filename or subject,
+ *     e.g. MEGCPO9524.pdf) -> tagged "PO Duplicate" and left in the inbox for a human,
+ *     because a differing file can be a legitimate revision.
+ * The app also checks both again after OCR (process-batch) and flags po_number_conflict.
+ *
+ * Mail handling once filed:
+ *   - success        -> tagged "PO Ingested" and moved out of the inbox (Archive)
+ *   - number in use  -> tagged "PO Duplicate", stays in the inbox for review
+ *   - no site found  -> tagged "PO Needs Review", stays in the inbox for review
+ *   - filing failed  -> tagged "PO Needs Review", stays in the inbox for review
+ * Messages carrying any of those three labels are skipped on later runs.
  *
  * Credentials referenced (none of them contain an inline secret):
  *   - "Microsoft Outlook (MEGA PO)" (microsoftOutlookOAuth2Api) -> n8n's hosted Microsoft
@@ -19,9 +33,10 @@
  *     Swap to microsoftEntraServicePrincipalApi for unattended app-only access.
  *   - "MEGAINFRASTRUCTURE" (pocketbaseHttpApi) -> already exists on this instance.
  *
- * Set the Microsoft credential on these seven nodes after creating it:
+ * Set the Microsoft credential on these eight nodes after creating it:
  *   List Inbox POs, List Attachments, Download Attachment, Archive Message,
- *   Archive Message (already filed), Flag Needs Review, Flag Needs Review (no site)
+ *   Archive Message (already filed), Label PO Filed, Flag Needs Review,
+ *   Flag Needs Review (no site), Flag Needs Review (number in use)
  */
 
 import { workflow, node, trigger, ifElse, splitInBatches, nextBatch, newCredential, expr } from '@n8n/workflow-sdk';
@@ -104,6 +119,7 @@ const selectMsgs = node({
 let sites = [];
 try { sites = ($('Collect Sites').item.json.sites) || []; } catch (e) { sites = []; }
 const POISH = /(\\bp\\/?o\\b|\\bpo\\b|purchase\\s*order|\\border\\b)/i;
+const SKIP = ['PO Needs Review', 'PO Ingested', 'PO Duplicate'];
 function norm(s) { return String(s || '').toUpperCase().replace(/\\s+/g, ' ').trim(); }
 function matchSite(text) {
   const t = norm(text);
@@ -120,7 +136,9 @@ for (const it of items) {
   const m = it && it.json ? it.json : {};
   if (!m.id || m.hasAttachments !== true) continue;
   const cats = Array.isArray(m.categories) ? m.categories : [];
-  if (cats.indexOf('PO Needs Review') >= 0 || cats.indexOf('PO Ingested') >= 0) continue;
+  let skip = false;
+  for (const c of SKIP) { if (cats.indexOf(c) >= 0) skip = true; }
+  if (skip) continue;
   const subject = m.subject || '';
   const preview = m.bodyPreview || '';
   const from = (m.from && m.from.emailAddress && m.from.emailAddress.address) || '';
@@ -189,7 +207,12 @@ const selectAtts = node({
       jsCode: `const items = $input.all();
 const DOC_TYPES = ['application/pdf', 'image/jpeg', 'image/png', 'image/webp'];
 let messageId = '';
-try { messageId = $('Select PO Messages').item.json.messageId || ''; } catch (e) { messageId = ''; }
+let subject = '';
+try {
+  const mm = $('Select PO Messages').item.json || {};
+  messageId = mm.messageId || '';
+  subject = mm.subject || '';
+} catch (e) { messageId = ''; subject = ''; }
 if (!messageId) { try { messageId = $('Loop Messages').item.json.messageId || ''; } catch (e) { messageId = ''; } }
 const out = [];
 for (const it of items) {
@@ -199,12 +222,16 @@ for (const it of items) {
   const ct = String(a.contentType || '').toLowerCase();
   const name = String(a.name || '');
   if (DOC_TYPES.indexOf(ct) < 0) continue;
+  let numberHint = '';
+  const hit = (name + ' ' + subject).match(/MEG[A-Z]*PO[-_ ]?\\d{2,}/i);
+  if (hit) numberHint = hit[0].replace(/[^A-Za-z0-9]/g, '').toUpperCase();
   out.push({ json: {
     messageId: messageId,
     attachmentId: a.id,
     filename: name,
     contentType: ct,
-    size: a.size || 0
+    size: a.size || 0,
+    numberHint: numberHint
   } });
 }
 return out;`
@@ -264,6 +291,7 @@ for (const item of items) {
       fileHash: fileHash,
       siteId: msgMeta.siteId || '',
       siteCode: msgMeta.siteCode || '',
+      numberHint: attMeta.numberHint || '',
       purchaseOrderNumber: fileHash ? ('PENDING-' + fileHash.slice(0, 12).toUpperCase()) : ('PENDING-' + stamp),
       emailMeta: {
         source: 'email',
@@ -293,22 +321,72 @@ const dupCheck = node({
       resource: 'purchase_orders',
       operation: 'search',
       parameters: {
-        elementsPerPage: 1,
+        elementsPerPage: 5,
-        filter: expr('"=file_hash=\'" + $("Capture File").item.json.fileHash + "\' && site=\'" + $("Capture File").item.json.siteId + "\' && processing_status!=\'failed\'"')
+        filter: expr('"(file_hash=\'" + $("Capture File").item.json.fileHash + "\' && site=\'" + $("Capture File").item.json.siteId + "\') || (purchase_order_number=\'" + ($("Capture File").item.json.numberHint || "") + "\')"')
       }
     },
     credentials: PB_CRED
   }
 });
 
-const alreadyFiled = ifElse({
+const classifyDup = node({
+  type: 'n8n-nodes-base.code',
+  version: 2,
+  config: {
+    name: 'Classify Duplicate',
+    parameters: {
+      mode: 'runOnceForAllItems',
+      jsCode: `const rows = $input.all();
+let cap = {};
+try { cap = $('Capture File').item.json || {}; } catch (e) { cap = {}; }
+let verdict = 'none';
+let hitId = '';
+let hitNumber = '';
+for (const r of rows) {
+  const j = r && r.json ? r.json : {};
+  if (!j.id) continue;
+  if (j.file_hash && cap.fileHash && j.file_hash === cap.fileHash) { verdict = 'same_file'; hitId = j.id; hitNumber = j.purchase_order_number || ''; break; }
+  if (!hitId) { hitId = j.id; hitNumber = j.purchase_order_number || ''; }
+}
+if (verdict !== 'same_file' && hitId) verdict = 'same_number';
+return [{ json: {
+  verdict: verdict,
+  duplicateOf: hitId,
+  duplicateNumber: hitNumber,
+  fileHash: cap.fileHash || '',
+  siteId: cap.siteId || '',
+  siteCode: cap.siteCode || '',
+  numberHint: cap.numberHint || '',
+  purchaseOrderNumber: cap.purchaseOrderNumber || '',
+  messageId: cap.messageId || '',
+  emailMeta: cap.emailMeta || {}
+} }];`
+    }
+  }
+});
+
+const sameFileFiled = ifElse({
   version: 2.2,
   config: {
-    name: 'Already Filed?',
+    name: 'Same File Filed?',
     parameters: {
       conditions: {
         options: { caseSensitive: true, leftValue: '', typeValidation: 'loose' },
-        conditions: [{ leftValue: expr('{{ $json.id }}'), operator: { type: 'string', operation: 'notEmpty' }, rightValue: '' }],
+        conditions: [{ leftValue: expr('{{ $json.verdict }}'), operator: { type: 'string', operation: 'equals' }, rightValue: 'same_file' }],
+        combinator: 'and'
+      }
+    }
+  }
+});
+
+const numberUsed = ifElse({
+  version: 2.2,
+  config: {
+    name: 'Number Already Used?',
+    parameters: {
+      conditions: {
+        options: { caseSensitive: true, leftValue: '', typeValidation: 'loose' },
+        conditions: [{ leftValue: expr('{{ $json.verdict }}'), operator: { type: 'string', operation: 'equals' }, rightValue: 'same_number' }],
         combinator: 'and'
       }
     }
@@ -381,6 +459,22 @@ const filedOk = ifElse({
   }
 });
 
+const labelFiled = node({
+  type: 'n8n-nodes-base.microsoftOutlook',
+  version: 2,
+  config: {
+    name: 'Label PO Filed',
+    parameters: {
+      resource: 'message',
+      operation: 'update',
+      authentication: 'microsoftOutlookOAuth2Api',
+      messageId: { __rl: true, mode: 'id', value: '={{ $("Capture File").item.json.messageId }}' },
+      updateFields: { categories: ['PO Ingested'], isRead: true }
+    },
+    credentials: GRAPH_CRED
+  }
+});
+
 const archiveNew = node({
   type: 'n8n-nodes-base.microsoftOutlook',
   version: 2,
@@ -413,6 +507,22 @@ const archiveDup = node({
   }
 });
 
+const flagNumber = node({
+  type: 'n8n-nodes-base.microsoftOutlook',
+  version: 2,
+  config: {
+    name: 'Flag Needs Review (number in use)',
+    parameters: {
+      resource: 'message',
+      operation: 'update',
+      authentication: 'microsoftOutlookOAuth2Api',
+      messageId: { __rl: true, mode: 'id', value: '={{ $("Capture File").item.json.messageId }}' },
+      updateFields: { categories: ['PO Duplicate'], isRead: true }
+    },
+    credentials: GRAPH_CRED
+  }
+});
+
 const flagReview = node({
   type: 'n8n-nodes-base.microsoftOutlook',
   version: 2,
@@ -458,7 +568,7 @@ return [{ json: { finished: true, processed: items.length } }];`
   }
 });
 
-export default workflow('mega-po-email-intake-v2', 'MEGA - PO Email Intake (Microsoft Outlook)')
+export default workflow('mega-po-email-intake-v3', 'MEGA - PO Email Intake (Microsoft Outlook)')
   .add(schedule)
   .to(readSites)
   .to(collectSites)
@@ -472,12 +582,14 @@ export default workflow('mega-po-email-intake-v2', 'MEGA - PO Email Intake (Micr
         .to(download
           .to(capture)
           .to(dupCheck)
-          .to(alreadyFiled
+          .to(classifyDup)
+          .to(sameFileFiled
             .onTrue(archiveDup.to(nextBatch(loop)))
-            .onFalse(createPo
+            .onFalse(numberUsed
-              .to(checkResult)
+              .onTrue(flagNumber.to(nextBatch(loop)))
-              .to(filedOk
+              .onFalse(createPo
-                .onTrue(archiveNew.to(nextBatch(loop)))
+                .to(checkResult)
-                .onFalse(flagReview.to(nextBatch(loop))))))))
+                .to(filedOk
-      .onFalse(flagNoSite.to(nextBatch(loop))))
+                  .onTrue(labelFiled.to(archiveNew.to(nextBatch(loop))))
-  );
+                  .onFalse(flagReview.to(nextBatch(loop)))))))))
+      .onFalse(flagNoSite.to(nextBatch(loop)))));