"""Generate WFDASH v5 (Catch and Create) workflow from v4. Strategy: - Load v4 JSON untouched - Insert 3 new nodes right after "PocketBase Admin Auth" so binary is uploaded to PocketBase BEFORE any downstream Code node can strip it: Plan: Upload Attachments Early (Code: split binary -> N items) Upload Attachment Early (HTTP Request: multipart POST) Collect Attachment IDs (Code: aggregate N -> 1 item) - Modify "Parse PB Auth" to preserve binary and pass through attachment IDs - Replace the broken "Plan: Upload Attachments" -> "Upload Attachment" chain (which runs LATE, after binary is stripped) with: Link Attachments to Quote (Code: emit one item per PB id) PATCH Attachment (set quote) (HTTP Request: link each to quote) - Rename the broken nodes to "DISABLED - ..." prefix and detach them from connections (so they exist in the workflow but don't execute) - Save as v5_catch_and_create.json Idempotent: running twice produces identical output. """ import copy import json import sys from pathlib import Path V4_PATH = Path(__file__).parent.parent / "n8n_workfows" / "WFDASH — Quote Email Monitor (v4 direct-to-PB).json" OUT_PATH = Path(__file__).parent.parent / "n8n_workfows" / "WFDASH — Quote Email Monitor (v5 catch-and-create).json" # ----- New nodes to insert after PocketBase Admin Auth ----- NEW_NODES_AFTER_AUTH = [ { "parameters": { "mode": "runOnceForAllItems", "jsCode": ( "// Plan: Upload Attachments Early (Catch and Create)\n" "// CATCH the email binary immediately after IMAP + auth,\n" "// BEFORE any downstream Code node can strip it. We:\n" "// 1. Walk every binary entry on the email\n" "// 2. Emit one item per attachment, each carrying the\n" "// data needed by the HTTP POST that uploads it\n" "//\n" "// IMPORTANT: this node must run while binary is still\n" "// fresh. n8n strips binary from items after any HTTP\n" "// node that uses it, AND after Code nodes that return\n" "// { json: ... } without binary. Place this BEFORE any\n" "// such nodes.\n" "//\n" "// Context source: we read the full email context (pb_url,\n" "// message_id, from_email, etc.) from the LAST node that\n" "// had it on its output — \"PocketBase Connection\" — because\n" "// the immediate upstream ($json) is the PB admin auth\n" "// response which only has { token, admin }.\n" "//\n" "// On no-attachments emails, emits one placeholder item\n" "// with _attachment_record_ids=[] so downstream wiring\n" "// is unaffected.\n" "const inputs = $input.all();\n" "if (inputs.length === 0) {\n" " return [{ json: { _attachment_record_ids: [], _attachment_count: 0 } }];\n" "}\n" "\n" "// Pull email context from the Set node that holds pb_url,\n" "// pb_admin_email, pb_admin_password, message_id, etc.\n" "const ctx = $('PocketBase Connection').item.json || {};\n" "const pbUrl = String(ctx.pb_url || '');\n" "const pbToken = String($json.token || '');\n" "const pbAdminId = String(($json.admin && $json.admin.id) || '');\n" "const messageId = String(ctx.message_id || '');\n" "\n" "const out = [];\n" "let attachmentIndex = 0;\n" "for (const input of inputs) {\n" " const bin = input.binary || {};\n" " for (const key of Object.keys(bin)) {\n" " const b = bin[key];\n" " if (!b) continue;\n" " const filename = String(b.fileName || 'attachment');\n" " const contentType = String(b.mimeType || 'application/octet-stream');\n" " out.push({\n" " json: {\n" " // PB auth context — every emitted item needs this\n" " // so the next HTTP node can build URL + Bearer.\n" " pb_url: pbUrl,\n" " pb_token: pbToken,\n" " pb_admin_id: pbAdminId,\n" " // Attachment metadata — survives across HTTP call\n" " email_message_id: messageId,\n" " filename: filename,\n" " content_type: contentType,\n" " size_bytes: Number(b.fileSize || 0),\n" " // Reference back to where the bytes live on this item\n" " binary_key: key,\n" " _upload_index: attachmentIndex\n" " },\n" " // Re-anchor the binary under a stable property name\n" " // so the HTTP node's \"Send Binary Data\" lookup is\n" " // deterministic.\n" " binary: { attachment: b }\n" " });\n" " attachmentIndex++;\n" " }\n" "}\n" "\n" "// No attachments? Emit a single placeholder so the chain\n" "// continues. The HTTP POST will see no binary and PB will\n" "// reject — we use 'continueRegularOutput' onError so the\n" "// failure is non-fatal.\n" "if (out.length === 0) {\n" " out.push({\n" " json: {\n" " pb_url: pbUrl,\n" " pb_token: pbToken,\n" " pb_admin_id: pbAdminId,\n" " email_message_id: messageId,\n" " filename: '',\n" " content_type: '',\n" " size_bytes: 0,\n" " binary_key: '__none__',\n" " _upload_index: 0,\n" " _no_attachments: true\n" " },\n" " binary: {}\n" " });\n" "}\n" "\n" "return out;\n" ), }, "id": "f0a1c000-0001-4000-8000-000000000001", "name": "Plan: Upload Attachments Early", "type": "n8n-nodes-base.code", "typeVersion": 2, "position": [1120, 240], }, { "parameters": { "method": "POST", "url": "={{ $json.pb_url }}/api/collections/quote_attachments/records", "sendHeaders": True, "headerParameters": { "parameters": [ {"name": "Authorization", "value": "=Bearer {{ $json.pb_token }}"} ] }, "sendBody": True, "contentType": "multipart-form-data", "bodyParameters": { "parameters": [ {"name": "filename", "value": "={{ $json.filename }}"}, {"name": "content_type", "value": "={{ $json.content_type }}"}, {"name": "size_bytes", "value": "={{ $json.size_bytes }}"}, {"name": "email_message_id", "value": "={{ $json.email_message_id }}"}, {"name": "uploaded_at", "value": "={{ $now.toISO() }}"}, { "name": "file", "value": "={{ $binary.attachment }}", "inputDataFieldName": "attachment", }, ] }, "options": { "response": {"response": {"neverError": True, "responseFormat": "json"}} }, }, "id": "f0a1c000-0001-4000-8000-000000000002", "name": "Upload Attachment Early", "type": "n8n-nodes-base.httpRequest", "typeVersion": 4.4, "position": [1344, 240], "onError": "continueRegularOutput", }, { "parameters": { "mode": "runOnceForAllItems", "jsCode": ( "// Collect Attachment IDs\n" "// After N HTTP POSTs (one per attachment), each item has a\n" "// PB response with `id`. Aggregate them into one item with\n" "// an array of attachment record IDs, and merge back the\n" "// upstream email context so the chain can continue.\n" "//\n" "// Context source: the immediate upstream $json is the PB\n" "// upload response (just { id, collectionId, ... }). The\n" "// full email context (pb_url, message_id, etc.) is on\n" "// \"Plan: Upload Attachments Early\" — we read from there.\n" "//\n" "// Also preserve binary so the next Code nodes (Parse PB\n" "// Auth etc.) don't have to worry about it. Though by this\n" "// point we no longer NEED binary — the attachments are\n" "// already in PB — preserving it costs nothing and is\n" "// robust against future chain edits.\n" "const inputs = $input.all();\n" "const ids = [];\n" "const details = [];\n" "let upstream = {};\n" "const mergedBinary = {};\n" "\n" "// Pull email context from Plan: Upload Attachments Early.\n" "// That node emitted N items each carrying pb_url, pb_token,\n" "// message_id, etc. Read from its first item.\n" "try {\n" " upstream = $('Plan: Upload Attachments Early').first().json || {};\n" "} catch (e) {\n" " upstream = {};\n" "}\n" "\n" "for (const input of inputs) {\n" " const j = input.json || {};\n" " // PB returns { id, ... } on success. Skip failures (no id).\n" " if (j.id) {\n" " ids.push(String(j.id));\n" " details.push({\n" " id: String(j.id),\n" " filename: upstream.filename || j.filename || '',\n" " content_type: upstream.content_type || j.content_type || '',\n" " size_bytes: upstream.size_bytes || j.size_bytes || 0,\n" " email_message_id: upstream.email_message_id || j.email_message_id || ''\n" " });\n" " }\n" " // Merge binary in case anything downstream cares (it\n" " // shouldn't, but cheap insurance).\n" " if (input.binary) {\n" " for (const k of Object.keys(input.binary)) {\n" " if (!mergedBinary[k]) mergedBinary[k] = input.binary[k];\n" " }\n" " }\n" "}\n" "\n" "return [{\n" " json: Object.assign({}, upstream, {\n" " _attachment_record_ids: ids,\n" " _attachment_records: details,\n" " _attachment_count: ids.length\n" " }),\n" " binary: mergedBinary\n" "}];\n" ), }, "id": "f0a1c000-0001-4000-8000-000000000003", "name": "Collect Attachment IDs", "type": "n8n-nodes-base.code", "typeVersion": 2, "position": [1568, 240], }, ] # ----- New nodes to insert after Create Quote Item (replace broken chain) ----- NEW_NODES_AFTER_QUOTE_ITEM = [ { "parameters": { "mode": "runOnceForAllItems", "jsCode": ( "// Link Attachments to Quote\n" "// The PB attachment record IDs were captured EARLY in the\n" "// chain (right after IMAP). Now that the quote exists,\n" "// emit one item per attachment record so the next HTTP\n" "// PATCH can set `quote=` on each.\n" "//\n" "// If there are no attachment records (no-attachment\n" "// emails, or early upload failed), emit nothing —\n" "// downstream PATCH is skipped.\n" "const inputs = $input.all();\n" "if (inputs.length === 0) {\n" " return [{ json: { _skip: true, reason: 'no upstream' } }];\n" "}\n" "const ctx = inputs[0].json || {};\n" "const quoteId = String(ctx.quote_id || '');\n" "const ids = Array.isArray(ctx._attachment_record_ids)\n" " ? ctx._attachment_record_ids : [];\n" "\n" "if (!quoteId) {\n" " return [{ json: Object.assign({}, ctx, {\n" " _skip: true, reason: 'no quote_id'\n" " }) }];\n" "}\n" "\n" "if (ids.length === 0) {\n" " // No attachments — skip PATCH but pass context through\n" " // so Mark has_attachments still runs.\n" " return [{ json: Object.assign({}, ctx, {\n" " _skip: true, reason: 'no attachments to link',\n" " _attachment_record_ids: []\n" " }) }];\n" "}\n" "\n" "return ids.map(id => ({\n" " json: {\n" " pb_url: ctx.pb_url || '',\n" " pb_token: ctx.pb_token || '',\n" " attachment_id: id,\n" " quote_id: quoteId,\n" " quote_number: ctx.quote_number || ''\n" " }\n" "}));\n" ), }, "id": "f0a1c000-0001-4000-8000-000000000004", "name": "Link Attachments to Quote", "type": "n8n-nodes-base.code", "typeVersion": 2, "position": [6624, 560], }, { "parameters": { "method": "PATCH", "url": "={{ $json.pb_url }}/api/collections/quote_attachments/records/{{ $json.attachment_id }}", "sendHeaders": True, "headerParameters": { "parameters": [ {"name": "Authorization", "value": "=Bearer {{ $json.pb_token }}"}, {"name": "Content-Type", "value": "application/json"}, ] }, "sendBody": True, "specifyBody": "json", "jsonBody": "={{ { \"quote\": $json.quote_id } }}", "options": { "response": {"response": {"neverError": True, "responseFormat": "json"}} }, }, "id": "f0a1c000-0001-4000-8000-000000000005", "name": "PATCH Attachment (set quote)", "type": "n8n-nodes-base.httpRequest", "typeVersion": 4.4, "position": [6848, 560], "onError": "continueRegularOutput", }, ] # ----- Parse PB Auth: fixed to preserve binary + pass through attachment IDs ----- FIXED_PARSE_PB_AUTH = { "parameters": { "mode": "runOnceForEachItem", "jsCode": ( "// Parse PB Auth response (FIXED for v5 Catch and Create)\n" "// Now preserves binary AND reads the upstream email context\n" "// from \"Collect Attachment IDs\" (which in v5 is the last\n" "// node with full email context). The original v4 read from\n" "// \"PocketBase Connection\" — that still works as a fallback,\n" "// but Collect Attachment IDs carries the attachment IDs we\n" "// captured earlier.\n" "//\n" "// After POST /api/admins/auth-with-password, the response is\n" "// { token, admin: {...} }. We pass the token through on\n" "// $json.pb_token so every downstream HTTP node can use it\n" "// as Bearer auth.\n" "const body = $json || {};\n" "const token = String(body.token || '');\n" "\n" "// Read email context. Prefer Collect Attachment IDs (it has\n" "// the attachment IDs from the early upload); fall back to\n" "// PocketBase Connection if that lookup fails for any reason.\n" "let ctx = {};\n" "try {\n" " ctx = $('Collect Attachment IDs').first().json || {};\n" "} catch (e) {\n" " ctx = $('PocketBase Connection').item.json || {};\n" "}\n" "\n" "if (!token) {\n" " return {\n" " json: Object.assign({}, ctx, {\n" " pb_token: '',\n" " pb_auth_error: 'auth returned no token: ' + JSON.stringify(body)\n" " }),\n" " binary: $binary || {}\n" " };\n" "}\n" "return {\n" " json: Object.assign({}, ctx, {\n" " pb_token: token,\n" " pb_admin_id: (body.admin && body.admin.id) ? body.admin.id : ''\n" " }),\n" " binary: $binary || {}\n" "};\n" ), }, "id": "6e1f9349-0489-46a6-9e97-653e2e46202c", # same id as v4 — overwrites "name": "Parse PB Auth", "type": "n8n-nodes-base.code", "typeVersion": 2, "position": [1120, 464], } def find_node(nodes, name): for i, n in enumerate(nodes): if n.get("name") == name: return i, n raise KeyError(f"Node not found: {name}") def disable_node(node): """Mark a node as disabled by renaming and adding the disabled flag. n8n reads a per-node `disabled: true` field on the node object itself (sibling of `parameters`, `type`, etc.) — setting it makes the node render greyed-out in the UI and skip execution while preserving the workflow JSON. """ node["disabled"] = True node["name"] = "DISABLED - " + node["name"] def main(): if not V4_PATH.exists(): print(f"ERROR: v4 workflow not found at {V4_PATH}", file=sys.stderr) sys.exit(1) wf = json.loads(V4_PATH.read_text()) nodes = wf["nodes"] connections = wf["connections"] # 1. Disable the broken late-upload chain so it doesn't run. for name in ("Plan: Upload Attachments", "Upload Attachment"): _, n = find_node(nodes, name) disable_node(n) # Remove their connections entirely so nothing wires to them. for name in ("Plan: Upload Attachments", "Upload Attachment"): connections.pop(name, None) # Detach the upstream edge that USED to point at the first disabled node. # In v4: "Create Quote Item" -> "Plan: Upload Attachments" cqi = connections.get("Create Quote Item") if cqi and "main" in cqi and cqi["main"] and cqi["main"][0]: cqi["main"][0] = [ edge for edge in cqi["main"][0] if edge.get("node") != "Plan: Upload Attachments" ] # 2. Insert new "Link Attachments to Quote" + "PATCH Attachment" nodes. # Position them after the disabled ones (the disabled nodes still hold # their old coordinates — that's fine, the editor only renders live nodes). nodes.extend(NEW_NODES_AFTER_QUOTE_ITEM) # Wire Create Quote Item -> Link Attachments to Quote -> PATCH Attachment # -> Mark has_attachments (existing). cqi_main = connections.setdefault("Create Quote Item", {"main": [[]]})["main"] # Append the new edge (preserve any existing edges, e.g. disabled one already stripped) cqi_main[0].append({"node": "Link Attachments to Quote", "type": "main", "index": 0}) connections["Link Attachments to Quote"] = { "main": [[{"node": "PATCH Attachment (set quote)", "type": "main", "index": 0}]] } # The PATCH node's downstream is "Mark has_attachments" (existing). # Mark has_attachments already exists as a connection target from # "Upload Attachment" — but Upload Attachment is disabled, so its # connection was just removed. Add a fresh edge from PATCH Attachment. connections["PATCH Attachment (set quote)"] = { "main": [[{"node": "Mark has_attachments", "type": "main", "index": 0}]] } # 3. Modify Parse PB Auth to preserve binary + forward attachment IDs. _, parse_pb = find_node(nodes, "Parse PB Auth") parse_pb.update(FIXED_PARSE_PB_AUTH) # 4. Insert the 3-node Catch and Create branch between # "PocketBase Admin Auth" and "Parse PB Auth". nodes.extend(NEW_NODES_AFTER_AUTH) # Rewire: PocketBase Admin Auth -> Plan: Upload Attachments Early # -> Upload Attachment Early -> Collect Attachment IDs -> Parse PB Auth auth_main = connections.setdefault( "PocketBase Admin Auth", {"main": [[]]} )["main"] # Strip the existing direct link to Parse PB Auth (if present) auth_main[0] = [ edge for edge in auth_main[0] if edge.get("node") != "Parse PB Auth" ] auth_main[0].append( {"node": "Plan: Upload Attachments Early", "type": "main", "index": 0} ) connections["Plan: Upload Attachments Early"] = { "main": [[{"node": "Upload Attachment Early", "type": "main", "index": 0}]] } connections["Upload Attachment Early"] = { "main": [[{"node": "Collect Attachment IDs", "type": "main", "index": 0}]] } connections["Collect Attachment IDs"] = { "main": [[{"node": "Parse PB Auth", "type": "main", "index": 0}]] } # 5. Update workflow metadata. wf["name"] = "WFDASH — Quote Email Monitor (v5 catch-and-create)" wf["settings"]["binaryMode"] = "separate" # explicit; unchanged # Bump versionId so n8n treats this as a new workflow on import. wf["versionId"] = "v5-catch-and-create-001" OUT_PATH.write_text(json.dumps(wf, indent=2)) print(f"Wrote {OUT_PATH}") print(f" Total nodes: {len(nodes)}") print(f" Disabled (rollback safety):") for n in nodes: if n.get("disabled"): print(f" - {n['name']}") if __name__ == "__main__": main()