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.

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)

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)

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},

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"] == {}