From e115031a57e3c6de69afa550fa7e815ecfd1872f Mon Sep 17 00:00:00 2001 From: NuklearRabbit <145918611+NuklearRabbit@users.noreply.github.com> Date: Sun, 2 Aug 2026 06:59:17 +0200 Subject: [PATCH] feat(n8n): add scheduled quality-scan workflow The original docs described two n8n workflows but the repository only ever shipped one (return-processing); the sketched second workflow (knowledge sync) depends on RAGcore, which isn't connected here, so it stays deferred. Add POST /api/v1/integrations/n8n/scheduled-scan (X-Service-Token protected, same pattern as the return callback), calling the same run_scan() the manual "Run quality scan" UI action uses and recording a service-actor data_quality_scan_run audit event. run_scan() already only creates an issue for a condition without one open, so overlapping triggers do no duplicate domain work. n8n/mobilityops-scheduled-quality-scan.json (hourly schedule + manual test trigger, both feeding the same HTTP call) ships "active": false so it can't fire anywhere until deliberately published. Verified live against the local n8n instance via the Manual test trigger: full green execution, and the resulting data_quality_scan_run audit event (actor_type=service, actor_label="n8n scheduled scan") confirms the real round trip, not just a contract test. deploy/unraid/setup-scheduled-scan.sh mirrors the existing return-workflow publish script for the shared Unraid n8n. --- Makefile | 9 +- backend/app/api/routers/data_quality.py | 2 +- backend/app/api/routers/integrations.py | 18 +++ backend/app/services/data_quality.py | 10 +- backend/tests/test_integrations.py | 40 +++++++ deploy/unraid/setup-scheduled-scan.sh | 46 ++++++++ docs/11-n8n-integration.md | 32 +++++- docs/17-runbook.md | 37 ++++++ n8n/mobilityops-scheduled-quality-scan.json | 120 ++++++++++++++++++++ 9 files changed, 306 insertions(+), 8 deletions(-) create mode 100644 deploy/unraid/setup-scheduled-scan.sh create mode 100644 n8n/mobilityops-scheduled-quality-scan.json diff --git a/Makefile b/Makefile index fb83afa..ea80ee5 100644 --- a/Makefile +++ b/Makefile @@ -1,4 +1,4 @@ -.PHONY: up down logs test lint seed reset n8n-setup demo e2e +.PHONY: up down logs test lint seed reset n8n-setup n8n-setup-scan demo e2e up: docker compose up --build -d @@ -32,6 +32,13 @@ n8n-setup: docker compose exec n8n n8n publish:workflow --id=mobilityops-return-processing docker compose restart n8n +# One-time per environment: imports and activates the scheduled quality-scan workflow. +# Same owner-account precondition as n8n-setup above. +n8n-setup-scan: + docker compose exec n8n n8n import:workflow --input=//imports/mobilityops-scheduled-quality-scan.json + docker compose exec n8n n8n publish:workflow --id=mobilityops-scheduled-quality-scan + docker compose restart n8n + # Full deterministic demo bootstrap: build, migrate (automatic on api startup), seed. demo: up docker compose exec api python -m app.cli seed --reset diff --git a/backend/app/api/routers/data_quality.py b/backend/app/api/routers/data_quality.py index e8bd1e7..52f3d8c 100644 --- a/backend/app/api/routers/data_quality.py +++ b/backend/app/api/routers/data_quality.py @@ -260,5 +260,5 @@ def scan( db: Session = Depends(get_db), user: CurrentUser = Depends(require_operations_manager), ) -> ScanResultOut: - result = run_scan(db, actor=user) + result = run_scan(db, actor_label=user.display_name, actor_type="user") return ScanResultOut(created=result.created) diff --git a/backend/app/api/routers/integrations.py b/backend/app/api/routers/integrations.py index 2bf4c3d..b3a421a 100644 --- a/backend/app/api/routers/integrations.py +++ b/backend/app/api/routers/integrations.py @@ -13,7 +13,9 @@ from app.core.config import get_settings from app.core.errors import AppError from app.models.audit import AuditEvent from app.models.outbox import OutboxEvent +from app.schemas import ScanResultOut from app.services.audit import record_audit_event +from app.services.data_quality import run_scan router = APIRouter(prefix="/api/v1/integrations/n8n", tags=["integrations"]) settings = get_settings() @@ -71,3 +73,19 @@ def return_callback( "event_id": str(event_id), "occurred_at": datetime.now(UTC).isoformat(), } + + +@router.post("/scheduled-scan", response_model=ScanResultOut) +def scheduled_scan( + service_token: str = Header(..., alias="X-Service-Token"), + db: Session = Depends(get_db), +) -> ScanResultOut: + """Triggered by the scheduled n8n quality-scan workflow. Narrow, read-mostly, and + safe to call repeatedly: run_scan() only ever creates an issue for a condition that + doesn't already have one open, so a duplicate or overlapping trigger does no + duplicate domain work -- it just reports zero new issues for anything already known.""" + if service_token != settings.n8n_callback_token: + raise AppError("UNAUTHORIZED_SERVICE", "Invalid service token.", status_code=401) + + result = run_scan(db, actor_label="n8n scheduled scan", actor_type="service") + return ScanResultOut(created=result.created) diff --git a/backend/app/services/data_quality.py b/backend/app/services/data_quality.py index e1f1e25..e8610d8 100644 --- a/backend/app/services/data_quality.py +++ b/backend/app/services/data_quality.py @@ -299,18 +299,20 @@ def _scan_odometer_regressions(db: Session, scan: ScanResult) -> None: break -def run_scan(db: Session, *, actor: CurrentUser | None = None) -> ScanResult: +def run_scan( + db: Session, *, actor_label: str | None = None, actor_type: str = "user" +) -> ScanResult: scan = ScanResult() _scan_duplicate_customers(db, scan) _scan_missing_required_fields(db, scan) _scan_odometer_regressions(db, scan) _scan_booking_overlaps(db, scan) _scan_vehicle_status_conflicts(db, scan) - if actor is not None: + if actor_label is not None: record_audit_event( db, - actor_type="user", - actor_label=actor.display_name, + actor_type=actor_type, + actor_label=actor_label, action="data_quality_scan_run", entity_type="system", metadata={"created": scan.created}, diff --git a/backend/tests/test_integrations.py b/backend/tests/test_integrations.py index 1e84f0d..1699bca 100644 --- a/backend/tests/test_integrations.py +++ b/backend/tests/test_integrations.py @@ -82,3 +82,43 @@ def test_callback_is_idempotent_by_event_id(client, ops_client): ).json() matching = [e for e in audit_events if e["metadata"]["event_id"] == event_id] assert len(matching) == 1 + + +def test_scheduled_scan_rejects_wrong_service_token(client): + response = client.post( + "/api/v1/integrations/n8n/scheduled-scan", + headers={"X-Service-Token": "wrong-token"}, + ) + assert response.status_code == 401 + + +def test_scheduled_scan_requires_service_token_header(client): + response = client.post("/api/v1/integrations/n8n/scheduled-scan") + assert response.status_code == 422 + + +def test_scheduled_scan_runs_and_returns_counts_by_rule(client, ops_client): + settings = get_settings() + response = client.post( + "/api/v1/integrations/n8n/scheduled-scan", + headers={"X-Service-Token": settings.n8n_callback_token}, + ) + assert response.status_code == 200 + assert response.json() == {"created": {}} # already-seeded conditions, nothing new + + audit_events = ops_client.get( + "/api/v1/audit", params={"action": "data_quality_scan_run"} + ).json() + service_triggered = [e for e in audit_events if e["actor_type"] == "service"] + assert len(service_triggered) >= 1 + assert service_triggered[0]["actor_label"] == "n8n scheduled scan" + + +def test_scheduled_scan_is_idempotent_across_repeated_triggers(client): + settings = get_settings() + headers = {"X-Service-Token": settings.n8n_callback_token} + first = client.post("/api/v1/integrations/n8n/scheduled-scan", headers=headers) + second = client.post("/api/v1/integrations/n8n/scheduled-scan", headers=headers) + assert first.status_code == 200 + assert second.status_code == 200 + assert second.json()["created"] == {} diff --git a/deploy/unraid/setup-scheduled-scan.sh b/deploy/unraid/setup-scheduled-scan.sh new file mode 100644 index 0000000..4c39c5d --- /dev/null +++ b/deploy/unraid/setup-scheduled-scan.sh @@ -0,0 +1,46 @@ +#!/bin/sh +set -eu + +container_name="${1:-n8n}" +scan_url="${2:-http://192.168.10.150:1236/api/v1/integrations/n8n/scheduled-scan}" +source_workflow="${3:-n8n/mobilityops-scheduled-quality-scan.json}" + +if [ ! -f .env ]; then + echo "Missing deployment .env" >&2 + exit 1 +fi +if [ ! -f "$source_workflow" ]; then + echo "Missing workflow export: $source_workflow" >&2 + exit 1 +fi +if ! docker inspect "$container_name" >/dev/null 2>&1; then + echo "Existing n8n container not found: $container_name" >&2 + exit 1 +fi + +callback_token="$(sed -n 's/^MOBILITYOPS_CALLBACK_TOKEN=//p' .env | tail -n 1)" +if [ -z "$callback_token" ]; then + echo "MOBILITYOPS_CALLBACK_TOKEN is empty" >&2 + exit 1 +fi + +temporary_workflow="$(mktemp /tmp/mobilityops-n8n-workflow.XXXXXX.json)" +container_workflow="/tmp/mobilityops-scheduled-quality-scan.json" +cleanup() { + rm -f "$temporary_workflow" + docker exec "$container_name" rm -f "$container_workflow" >/dev/null 2>&1 || true +} +trap cleanup EXIT INT TERM + +jq --arg scan_url "$scan_url" --arg callback_token "$callback_token" ' + (.nodes[] | select(.id == "scan-node") | .parameters.url) = $scan_url | + (.nodes[] | select(.id == "scan-node") | .parameters.headerParameters.parameters[] | + select(.name == "X-Service-Token") | .value) = $callback_token +' "$source_workflow" > "$temporary_workflow" + +docker cp "$temporary_workflow" "$container_name:$container_workflow" >/dev/null +docker exec "$container_name" n8n import:workflow --input="$container_workflow" +docker exec "$container_name" n8n publish:workflow --id=mobilityops-scheduled-quality-scan +docker restart "$container_name" >/dev/null + +echo "Published MobilityOps scheduled quality-scan workflow to existing container ${container_name}" diff --git a/docs/11-n8n-integration.md b/docs/11-n8n-integration.md index 5aa135c..156210c 100644 --- a/docs/11-n8n-integration.md +++ b/docs/11-n8n-integration.md @@ -18,7 +18,34 @@ Steps: The starter export is `n8n/mobilityops-return-processing.json`. Claude may correct its credentials and callback route but must preserve idempotency. -## Optional second workflow: knowledge sync +## Second live workflow: scheduled quality scan + +RAGcore is not connected in this environment, so the originally sketched "knowledge sync" +workflow below remains deferred (see "Deferred: knowledge sync"). The second implemented +workflow does not depend on RAGcore or MCP Hub, so it is not blocked by them. + +Input: hourly schedule trigger, or a manual trigger for on-demand testing. + +Steps: + +1. call the narrow, service-token-protected `POST + /api/v1/integrations/n8n/scheduled-scan` endpoint; +2. the endpoint runs the same deterministic `run_scan()` domain function the manual + "Run quality scan" UI action uses, and records a `data_quality_scan_run` audit event + with `actor_type=service`; +3. return counts of newly created issues per rule type. + +`run_scan()` only ever creates an issue for a condition that does not already have one +open, so a duplicate or overlapping trigger (a manual test run firing close to the +scheduled one, or a retried HTTP call) does no duplicate domain work. + +The starter export is `n8n/mobilityops-scheduled-quality-scan.json`, imported and +published the same way as the return-processing workflow (see +`deploy/unraid/setup-scheduled-scan.sh` and `docs/17-runbook.md`). It ships with +`"active": false` so it cannot fire against any environment until deliberately +published with a real service token. + +## Deferred: knowledge sync Input: manual trigger or manifest-changed event. @@ -28,7 +55,8 @@ Steps: 2. call RAGcore ingestion/sync API; 3. record per-document results through MobilityOps integration status API. -This workflow is useful but must not delay the core demo if RAGcore's final API is not ready. +Deferred until RAGcore's live ingestion API is available in this environment; must not +delay or block the core demo. ## Outbox dispatcher diff --git a/docs/17-runbook.md b/docs/17-runbook.md index 27b8f92..8a56b8c 100644 --- a/docs/17-runbook.md +++ b/docs/17-runbook.md @@ -62,6 +62,35 @@ curl -b cookies.txt http://localhost:8128/api/v1/workflows | grep succeeded A failed/offline n8n does not roll back the return — the outbox event simply stays `pending`/`failed` and is safely retryable from the Automation page. +### Second workflow: scheduled quality scan + +Import and publish the same way: + +```bash +make n8n-setup-scan +``` + +which runs: + +```bash +docker compose exec n8n n8n import:workflow --input=//imports/mobilityops-scheduled-quality-scan.json +docker compose exec n8n n8n publish:workflow --id=mobilityops-scheduled-quality-scan +docker compose restart n8n +``` + +Verify: + +```bash +curl -X POST http://localhost:8128/api/v1/integrations/n8n/scheduled-scan \ + -H "X-Service-Token: " +# {"created": {...}} +``` + +Trigger a live run from n8n's own UI ("Manual test trigger" node → Execute Workflow) to +confirm the round trip without waiting for the hourly schedule. It does not depend on +RAGcore or MCP Hub and ships `"active": false`, so it never fires anywhere until +deliberately published with a real service token. + ### Existing shared n8n on the Unraid review server The Unraid deployment uses the existing n8n at `http://192.168.10.150:5678`; it does not @@ -85,6 +114,14 @@ inside n8n's protected application data. The callback travels through the Mobili proxy, so the shared n8n container does not need direct database access or membership of the MobilityOps Docker network. +Publish the scheduled quality-scan workflow the same way: + +```bash +./deploy/unraid/setup-scheduled-scan.sh \ + n8n \ + http://192.168.10.150:1236/api/v1/integrations/n8n/scheduled-scan +``` + ## Required operational checks - API and web health (`GET /health`, web root `200`); diff --git a/n8n/mobilityops-scheduled-quality-scan.json b/n8n/mobilityops-scheduled-quality-scan.json new file mode 100644 index 0000000..5bdbaa7 --- /dev/null +++ b/n8n/mobilityops-scheduled-quality-scan.json @@ -0,0 +1,120 @@ +{ + "id": "mobilityops-scheduled-quality-scan", + "name": "MobilityOps - Scheduled 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": "={{$env.MOBILITYOPS_API_URL || 'http://api:8000'}}/api/v1/integrations/n8n/scheduled-scan", + "sendHeaders": true, + "headerParameters": { + "parameters": [ + { + "name": "X-Service-Token", + "value": "={{$env.MOBILITYOPS_CALLBACK_TOKEN}}" + } + ] + }, + "options": { + "timeout": 15000 + } + }, + "id": "scan-node", + "name": "Run quality scan", + "type": "n8n-nodes-base.httpRequest", + "typeVersion": 4.2, + "position": [ + 520, + 300 + ] + }, + { + "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": false, + "versionId": "22222222-2222-4222-8222-222222222222", + "meta": { + "templateCredsSetupCompleted": false + }, + "tags": [] +}