Jens/MobilityOps
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

n8n: store cleaned workflow definitions as repo source of truth

Browse Source 
Move the two live-validated workflows into n8n/workflows/ (credential-
based auth referenced by name only, no secret values), add a manifest
covering all 4 canonical workflows and a read-only drift-check script
against n8n's Public API. Retire the pre-integration root-level starter
files that still carried the literal-token pattern, and repoint the
Unraid deploy scripts, Makefile targets and runbook at the new files.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
NuklearRabbit
2026-08-04 13:34:51 +02:00
co-authored by Claude Sonnet 5
parent 59cb4c062e
commit e0c107a94a
12 changed files with 410 additions and 165 deletions
Download Patch File Download Diff File Expand all files Collapse all files
n8n/workflows/MANIFEST.md
+59
View File
@@ -0,0 +1,59 @@
# n8n workflow manifest
Source of truth for the four canonical Fleet Ops n8n workflows. Definitions in this
directory are cleaned exports of the live workflows on `https://n8n.itworx.tech` —
credential values are never embedded; nodes reference named n8n credentials instead. Run
`n8n/workflows/check_drift.py` to compare a live workflow against its repo definition.
## 1. Fleet Ops — Vehicle Return Orchestration
| Field | Value |
|---|---|
| File | `fleet-ops-vehicle-return.json` |
| Purpose | Orchestrate the post-return follow-up (cleaning vs. attention-required) once Fleet Ops emits a `vehicle.returned.v1` outbox event, and report the result back to Fleet Ops. |
| Trigger | Production webhook, `POST /webhook/mobilityops-return`, Header Auth (`Fleet Ops Webhook Trigger Token`) |
| Event contract | `contracts/events.schema.json`, `event_type: vehicle.returned.v1` (envelope: `event_id`, `event_type`, `occurred_at`, `correlation_id`, `aggregate`, `data`) |
| Required credentials | `Fleet Ops Webhook Trigger Token` (Header Auth, on the trigger); `Fleet Ops Service Token` (Header Auth, on the outbound HTTP call) |
| Live workflow ID | `mobilityops-return-processing` |
| Active status (as of 2026-08-04) | Active / Published |
| Checksum (sha256) | `e13a3087269fc97019a7adf6c6a6a4ee4bd354c2dd7167d4966d4753a48e970e` |
## 2. Fleet Ops — Scheduled Data Quality Scan
| Field | Value |
|---|---|
| File | `fleet-ops-data-quality-scan.json` |
| Purpose | Periodically (and on-demand) run the Fleet Ops data-quality scan and summarize created-issue counts per rule. |
| Trigger | Schedule Trigger (hourly, Europe/Brussels instance timezone) + Manual Trigger for on-demand test runs |
| Event contract | N/A — HTTP-triggered scan call, no inbound event envelope. Request: `POST /api/v1/integrations/n8n/scheduled-scan`, Header Auth. |
| Required credentials | `Fleet Ops Service Token` (Header Auth, on the scan HTTP call) |
| Live workflow ID | `mobilityops-scheduled-quality-scan` |
| Active status (as of 2026-08-04) | Active / Published |
| Checksum (sha256) | `cc30b28b07dad9f9908a6ea0c564ec4c2f362a3ed71b7e97a7b6894408bb7e2e` |
## 3. Fleet Ops — RAGcore Procedure Sync
Not yet built. Blocked on a RAGcore application credential (scope `sources:sync`) for the
`fleet-ops` application, to be provided by the project owner. Will sync
`n8n/workflows/fleet-ops-ragcore-procedure-sync.json` against the real RAGcore contract
(`POST /v1/uploads`, `GET /v1/knowledge-spaces`, etc. — see
`contracts/ragcore-contract-assumptions.md` and the live inspection notes in
`docs/live-ai-integration/n8n-current-state.md`).
| Field | Value |
|---|---|
| File | `fleet-ops-ragcore-procedure-sync.json` (not yet created) |
| Live workflow ID | — |
| Active status | Not built |
## 4. Fleet Ops — Workflow Error Handler
Not yet built. Central error workflow to be attached to workflows 1-3 via n8n's
per-workflow "Error Workflow" setting. Reports bounded, secret-free failure details to a
new Fleet Ops automation-failure endpoint.
| Field | Value |
|---|---|
| File | `fleet-ops-error-handler.json` (not yet created) |
| Live workflow ID | — |
| Active status | Not built |
n8n/workflows/check_drift.py
+146
View File
@@ -0,0 +1,146 @@
#!/usr/bin/env python3
"""Report drift between the repo's cleaned workflow definitions and the live n8n instance.
Read-only: this script only issues GET requests against n8n's Public API. It never writes,
imports, activates, or otherwise modifies anything in n8n -- fixing drift is a deliberate,
reviewed action a human takes in the n8n UI (or via a separate, explicit import step), not
something this script does automatically.
Usage:
N8N_BASE_URL=https://n8n.itworx.tech N8N_API_KEY=... python n8n/workflows/check_drift.py
N8N_API_KEY must be an n8n Public API key (n8n UI -> Settings -> API), not a session cookie
and not a workflow credential. It is read from the environment only and is never printed.
Exit code is 0 when every checked workflow matches the live instance, 1 when any drift (or
a fetch error) is found, so this is safe to wire into CI as a non-blocking check.
"""
from __future__ import annotations
import json
import os
import sys
import urllib.error
import urllib.request
from pathlib import Path
from typing import Any
WORKFLOWS_DIR = Path(__file__).parent
# (repo file name, live workflow ID) -- kept in sync with MANIFEST.md by hand, since the
# manifest is the human-readable source of truth and this is just its machine-checkable echo.
KNOWN_WORKFLOWS = [
("fleet-ops-vehicle-return.json", "mobilityops-return-processing"),
("fleet-ops-data-quality-scan.json", "mobilityops-scheduled-quality-scan"),
]
# Fields that legitimately differ between a committed definition and the live instance
# (instance-assigned identifiers, timestamps, UI-only cosmetics) and must not be reported
# as drift.
_VOLATILE_TOP_LEVEL_KEYS = {
"versionId",
"createdAt",
"updatedAt",
"pinData",
"staticData",
"shared",
"triggerCount",
"isArchived",
}
_VOLATILE_NODE_KEYS = {"position", "webhookId"}
def _strip_credential_ids(value: Any) -> Any:
"""Live workflows carry instance-specific credential IDs alongside the credential name
(e.g. {"id": "3", "name": "Fleet Ops Service Token"}). The repo definitions intentionally
omit the id, since it's meaningless outside the instance that issued it. Drop it from both
sides so credential *references* are compared by name only."""
if isinstance(value, dict):
if set(value.keys()) <= {"id", "name"} and "name" in value:
return {"name": value["name"]}
return {k: _strip_credential_ids(v) for k, v in value.items()}
if isinstance(value, list):
return [_strip_credential_ids(v) for v in value]
return value
def _normalize_node(node: dict[str, Any]) -> dict[str, Any]:
cleaned = {k: v for k, v in node.items() if k not in _VOLATILE_NODE_KEYS}
return _strip_credential_ids(cleaned)
def _normalize_workflow(doc: dict[str, Any]) -> dict[str, Any]:
nodes_by_name = {n["name"]: _normalize_node(n) for n in doc.get("nodes", [])}
return {
"name": doc.get("name"),
"active": doc.get("active"),
"nodes": nodes_by_name,
"connections": doc.get("connections", {}),
"settings": doc.get("settings", {}),
}
def _diff(path: str, local: Any, live: Any, out: list[str]) -> None:
if isinstance(local, dict) and isinstance(live, dict):
for key in sorted(set(local) | set(live)):
if key in _VOLATILE_TOP_LEVEL_KEYS:
continue
if key not in live:
out.append(f"{path}.{key}: present in repo, missing live")
elif key not in local:
out.append(f"{path}.{key}: present live, missing in repo")
else:
_diff(f"{path}.{key}", local[key], live[key], out)
elif local != live:
out.append(f"{path}: repo={local!r} live={live!r}")
def fetch_live_workflow(base_url: str, api_key: str, workflow_id: str) -> dict[str, Any]:
url = f"{base_url.rstrip('/')}/api/v1/workflows/{workflow_id}"
request = urllib.request.Request(url, headers={"X-N8N-API-KEY": api_key, "Accept": "application/json"})
with urllib.request.urlopen(request, timeout=15) as response: # noqa: S310 (fixed https base url from env)
return json.load(response)
def main() -> int:
base_url = os.environ.get("N8N_BASE_URL")
api_key = os.environ.get("N8N_API_KEY")
if not base_url or not api_key:
print(
"Set N8N_BASE_URL and N8N_API_KEY (an n8n Public API key) in the environment.",
file=sys.stderr,
)
return 1
any_drift = False
for filename, workflow_id in KNOWN_WORKFLOWS:
repo_path = WORKFLOWS_DIR / filename
local_doc = json.loads(repo_path.read_text(encoding="utf-8"))
try:
live_doc = fetch_live_workflow(base_url, api_key, workflow_id)
except urllib.error.HTTPError as exc:
print(f"[{filename}] FAILED to fetch live workflow {workflow_id}: HTTP {exc.code}")
any_drift = True
continue
except urllib.error.URLError as exc:
print(f"[{filename}] FAILED to fetch live workflow {workflow_id}: {exc.reason}")
any_drift = True
continue
differences: list[str] = []
_diff(filename, _normalize_workflow(local_doc), _normalize_workflow(live_doc), differences)
if differences:
any_drift = True
print(f"[{filename}] DRIFT from live workflow {workflow_id}:")
for line in differences:
print(f" - {line}")
else:
print(f"[{filename}] matches live workflow {workflow_id}")
return 1 if any_drift else 0
if __name__ == "__main__":
raise SystemExit(main())
n8n/workflows/fleet-ops-data-quality-scan.json
+85
View File
@@ -0,0 +1,85 @@
{
"id": "mobilityops-scheduled-quality-scan",
"name": "Fleet Ops — Scheduled Data Quality Scan",
"nodes": [
{
"parameters": {
"rule": {
"interval": [
{
"field": "hours",
"hoursInterval": 1
}
]
}
},
"id": "schedule-node",
"name": "Hourly schedule",
"type": "n8n-nodes-base.scheduleTrigger",
"typeVersion": 1.2,
"position": [240, 220]
},
{
"parameters": {},
"id": "manual-node",
"name": "Manual test trigger",
"type": "n8n-nodes-base.manualTrigger",
"typeVersion": 1,
"position": [240, 400]
},
{
"parameters": {
"method": "POST",
"url": "http://192.168.10.150:1236/api/v1/integrations/n8n/scheduled-scan",
"authentication": "genericCredentialType",
"genericAuthType": "httpHeaderAuth",
"sendHeaders": true,
"headerParameters": {
"parameters": []
},
"options": {
"timeout": 15000
}
},
"id": "scan-node",
"name": "Run quality scan",
"type": "n8n-nodes-base.httpRequest",
"typeVersion": 4.2,
"position": [520, 300],
"credentials": {
"httpHeaderAuth": {
"name": "Fleet Ops Service Token"
}
}
},
{
"parameters": {
"jsCode": "const created = $json.created ?? {};\nconst total = Object.values(created).reduce((a, b) => a + b, 0);\nreturn [{json: {total_created: total, created_by_rule: created}}];"
},
"id": "summarize-node",
"name": "Summarize result",
"type": "n8n-nodes-base.code",
"typeVersion": 2,
"position": [780, 300]
}
],
"connections": {
"Hourly schedule": {
"main": [[{ "node": "Run quality scan", "type": "main", "index": 0 }]]
},
"Manual test trigger": {
"main": [[{ "node": "Run quality scan", "type": "main", "index": 0 }]]
},
"Run quality scan": {
"main": [[{ "node": "Summarize result", "type": "main", "index": 0 }]]
}
},
"settings": {
"executionOrder": "v1"
},
"active": true,
"meta": {
"templateCredsSetupCompleted": false
},
"tags": []
}
n8n/workflows/fleet-ops-vehicle-return.json
+99
View File
@@ -0,0 +1,99 @@
{
"id": "mobilityops-return-processing",
"name": "Fleet Ops — Vehicle Return Orchestration",
"nodes": [
{
"parameters": {
"httpMethod": "POST",
"path": "mobilityops-return",
"authentication": "headerAuth",
"responseMode": "responseNode",
"options": {}
},
"id": "webhook-node",
"name": "Return webhook",
"type": "n8n-nodes-base.webhook",
"typeVersion": 2,
"position": [240, 300],
"webhookId": "mobilityops-return",
"credentials": {
"httpHeaderAuth": {
"name": "Fleet Ops Webhook Trigger Token"
}
}
},
{
"parameters": {
"jsCode": "const e = $json.body ?? $json;\nif (e.event_type !== 'vehicle.returned.v1') throw new Error('Unsupported event type');\nconst reasons = e.data?.attention_reasons ?? [];\nreturn [{json: {event_id: e.event_id, event_type: e.event_type, aggregate_ref: e.aggregate.public_ref, follow_up: reasons.length ? 'attention_required' : 'cleaning', summary: reasons.join('; ') || 'Create cleaning follow-up'}}];"
},
"id": "validate-node",
"name": "Validate and derive follow-up",
"type": "n8n-nodes-base.code",
"typeVersion": 2,
"position": [500, 300]
},
{
"parameters": {
"method": "POST",
"url": "http://192.168.10.150:1236/api/v1/integrations/n8n/return-callback",
"authentication": "genericCredentialType",
"genericAuthType": "httpHeaderAuth",
"sendHeaders": true,
"headerParameters": {
"parameters": [
{
"name": "Idempotency-Key",
"value": "={{$json.event_id}}"
}
]
},
"sendBody": true,
"contentType": "raw",
"rawContentType": "application/json",
"body": "={{JSON.stringify($json)}}",
"options": {}
},
"id": "callback-node",
"name": "Record follow-up",
"type": "n8n-nodes-base.httpRequest",
"typeVersion": 4.2,
"position": [760, 300],
"credentials": {
"httpHeaderAuth": {
"name": "Fleet Ops Service Token"
}
}
},
{
"parameters": {
"respondWith": "json",
"responseBody": "={{ { ok: true, event_id: $('Validate and derive follow-up').item.json.event_id, result: $json } }}",
"options": {}
},
"id": "response-node",
"name": "Return result",
"type": "n8n-nodes-base.respondToWebhook",
"typeVersion": 1.4,
"position": [1020, 300]
}
],
"connections": {
"Return webhook": {
"main": [[{ "node": "Validate and derive follow-up", "type": "main", "index": 0 }]]
},
"Validate and derive follow-up": {
"main": [[{ "node": "Record follow-up", "type": "main", "index": 0 }]]
},
"Record follow-up": {
"main": [[{ "node": "Return result", "type": "main", "index": 0 }]]
}
},
"settings": {
"executionOrder": "v1"
},
"active": true,
"meta": {
"templateCredsSetupCompleted": false
},
"tags": []
}