From 59d663a43aff39f06721aabe3fbe0826ec19bd6c Mon Sep 17 00:00:00 2001 From: NuklearRabbit <145918611+NuklearRabbit@users.noreply.github.com> Date: Sat, 1 Aug 2026 22:39:25 +0200 Subject: [PATCH] M4: implement n8n automation Outbox dispatcher (background thread, FOR UPDATE SKIP LOCKED claim, exponential backoff, no transaction held during HTTP I/O). n8n callback endpoint with shared-secret auth and idempotency by event ID. Automation nav + UI with manual retry. 49 backend tests passing, ruff clean. Fixed a crash-on-redelivery bug in seeded outbox payloads and made the dispatcher defensive against malformed payloads. Verified the full live round trip against a real n8n instance: return -> outbox -> dispatcher -> n8n workflow -> callback -> succeeded, including the S5 failed-retry demo scenario. --- .env.example | 1 + PROJECT_STATE.md | 22 ++- backend/app/api/routers/integrations.py | 73 ++++++++++ backend/app/api/routers/workflows.py | 76 ++++++++++ backend/app/core/config.py | 5 + backend/app/main.py | 26 +++- backend/app/seed_loader.py | 23 ++- backend/app/services/dispatcher.py | 148 ++++++++++++++++++++ backend/tests/test_dispatcher.py | 177 ++++++++++++++++++++++++ backend/tests/test_integrations.py | 84 +++++++++++ backend/tests/test_workflows.py | 38 +++++ compose.yaml | 3 + frontend/src/App.tsx | 2 + frontend/src/components/Layout.tsx | 1 + frontend/src/pages/Automation.tsx | 129 +++++++++++++++++ frontend/src/styles.css | 6 + n8n/mobilityops-return-processing.json | 7 +- 17 files changed, 814 insertions(+), 7 deletions(-) create mode 100644 backend/app/api/routers/integrations.py create mode 100644 backend/app/api/routers/workflows.py create mode 100644 backend/app/services/dispatcher.py create mode 100644 backend/tests/test_dispatcher.py create mode 100644 backend/tests/test_integrations.py create mode 100644 backend/tests/test_workflows.py create mode 100644 frontend/src/pages/Automation.tsx diff --git a/.env.example b/.env.example index 0ca42e2..4be61a6 100644 --- a/.env.example +++ b/.env.example @@ -18,6 +18,7 @@ N8N_ENCRYPTION_KEY=replace-me N8N_BASIC_AUTH_ACTIVE=true N8N_BASIC_AUTH_USER=admin N8N_BASIC_AUTH_PASSWORD=change-me +MOBILITYOPS_CALLBACK_TOKEN=replace-me-n8n-callback-token # RAGcore integration KNOWLEDGE_PROVIDER=demo diff --git a/PROJECT_STATE.md b/PROJECT_STATE.md index f1f5d3e..efbc04b 100644 --- a/PROJECT_STATE.md +++ b/PROJECT_STATE.md @@ -2,7 +2,7 @@ ## Current milestone -M3 — complete. Starting M4 next. +M4 — complete. Starting M5 next. ## Locked decisions @@ -78,10 +78,26 @@ M3 — complete. Starting M4 next. - `npm run build` — clean (had to fix two `possibly 'null'` TS errors from a closure-narrowing limitation — TS doesn't narrow `const` captured-by-closure across nested function boundaries when the value comes from an index/property expression; fixed by re-binding to explicitly-typed local consts right after the guard). - Full browser run: Data Quality list (26 open issues, filterable) → `DQ-DEMO-DUPLICATE` two-column compare (CUS-0012 vs CUS-0178, per-field diff highlighting only where they differ) → merge with inline confirm → issue flips to `resolved` → confirmed `customer_merged` audit event with correct actor/entity/correlation → `DQ-DEMO-OVERLAP` (non-duplicate type) renders evidence JSON + defer/reject, no dead compare UI shown for a rule type it doesn't apply to. +### M4 — n8n automation +- `app/services/dispatcher.py`: background daemon thread (started/stopped via FastAPI `lifespan`, not an `on_event` hook) polling every `N8N_DISPATCH_INTERVAL_SECONDS` (default 3s). Claim step (`_claim_due_events`) is a short transaction using `SELECT ... FOR UPDATE SKIP LOCKED` that only flips `pending`→`delivering` and commits immediately; the HTTP call to n8n happens with **no open transaction**; the outcome is recorded in a separate short transaction. Exponential backoff `min(2**attempts, 60)` seconds, `N8N_MAX_ATTEMPTS=5` before a permanent `failed`. +- Dispatcher reconstructs the wire event from `contracts/events.schema.json`'s exact fields (`event_id`, `event_type`, `occurred_at`, `correlation_id`, `aggregate`, `data`) rather than forwarding `OutboxEvent.payload_json` wholesale — that column also carries an internal `aggregate_ref` convenience key (used by dashboard/workflows list rendering) that the schema's `additionalProperties: false` would reject. +- `POST /api/v1/integrations/n8n/return-callback` (`app/api/routers/integrations.py`): shared-secret auth via `X-Service-Token` header (`N8N_CALLBACK_TOKEN`, propagated to both `api` and `n8n` containers as `MOBILITYOPS_CALLBACK_TOKEN`); idempotent by `Idempotency-Key` (the event UUID) — checked by querying for an existing `AuditEvent` with that event ID in its metadata, **not** by `OutboxEvent.external_run_id`, because the dispatcher only sets that field *after* it gets n8n's final response, which happens *after* n8n has already called this callback mid-workflow — using `external_run_id` as the idempotency guard would have missed the exact redelivery case it's meant to catch. +- `GET /api/v1/workflows` + `POST /api/v1/workflows/{event_id}/retry` (`app/api/routers/workflows.py`), both Operations Manager only. Retry only allowed from `failed`; sets `pending` + clears `next_attempt_at` so the live dispatcher picks it up on its next cycle (does not reset `attempts`, so the counter reflects true delivery history). +- Automation nav + page (`pages/Automation.tsx`): table of all runs with status/attempts/last error, Retry button for `failed` rows, visible only to Operations Manager (matches backend authorization rather than just hiding a link). +- **Two real bugs found and fixed, the second only by testing the actual live n8n round-trip, not by pytest**: + 1. Seed-loaded `workflow_runs.csv` rows only ever got `payload_json = {"aggregate_ref": ...}` (no `correlation_id`/`aggregate`/`data`) — fine for M1–M3 since nothing read those keys yet, but once the dispatcher tried to *redeliver* a seeded row (i.e. the S5 manual-retry demo scenario) it crashed with `KeyError: 'correlation_id'`, leaving that event stuck in `delivering` forever (the crash happened before the outcome-recording transaction). Fixed in two places: `seed_loader.py` now builds the full schema-compliant envelope for every `workflow_runs.csv` row (matching what the live M2 return flow produces), and `dispatcher._deliver_one` now catches malformed-payload `KeyError`s defensively and resolves the row to `pending`/`failed` instead of leaving it orphaned — added `test_deliver_one_handles_malformed_payload_without_getting_stuck` as a regression test for the latter. + 2. This n8n image (2.32.7) has dropped `N8N_BASIC_AUTH_ACTIVE` as a UI/API gate — it requires an actual owner account via the `/setup` flow before anything (including webhook registration reliability) works correctly. Also: `n8n import:workflow` requires the workflow JSON to have a top-level `"id"` field (added `"id": "mobilityops-return-processing"`) and **always deactivates** the imported workflow regardless of its `"active"` field — activation requires `n8n publish:workflow --id=` followed by a full n8n restart (documented in n8n 2.x CLI, not obvious from the docs pack). Did this manually this session via the CLI + browser setup wizard; **this is a one-time operational step that is not automated** — a truly clean checkout still needs someone to run `docker compose exec n8n n8n import:workflow --input=//imports/mobilityops-return-processing.json`, `docker compose exec n8n n8n publish:workflow --id=mobilityops-return-processing`, `docker compose restart n8n`, and complete the one-time owner setup at `http://localhost:5678/setup` (any email/password, no verification required) before the automation demo will work. `docs/17-runbook.md` should get this exact sequence in M7. +- Commands run and verified from this checkout: + - `docker compose run --rm api pytest -q` — **49 passed** (new `tests/test_dispatcher.py` — claim/deliver success/failure/backoff/exhaustion-to-failed/malformed-payload, all via `monkeypatch.setattr(dispatcher.httpx, "post", ...)`, no real network calls in tests; `tests/test_integrations.py` — callback auth, unknown-event 404, idempotent-by-event-ID with a real duplicate-call assertion; `tests/test_workflows.py` — role gating, retry-only-from-failed, S5 retry-and-audit). + - `docker compose run --rm api ruff check .` — All checks passed. + - `npm run build` — clean. + - Full live round trip (not mocked): registered a real return on `BK-DEMO-RETURN` → outbox event queued → background dispatcher delivered it to the now-activated n8n workflow within its 3s poll interval → n8n called back into `/api/v1/integrations/n8n/return-callback` (200 OK, confirmed in `docker compose logs api`) → dispatcher's original POST received n8n's success response → event flipped to `succeeded` on attempt 1, visible on `/automation`. + - S5 scenario end-to-end in the browser: seeded `BK-H-0020` (`failed`, 3 attempts, "Synthetic connection timeout to n8n") → clicked Retry → `pending` → within ~3s, live dispatcher delivered it through the real n8n instance → `succeeded`, 4 attempts. This is the full documented S5 scenario working for real, not simulated. + ## Known blockers -None. External service credentials may be absent; use the documented demo/degraded providers. +None. External service credentials may be absent; use the documented demo/degraded providers. The n8n workflow-activation steps above are a one-time manual setup requirement in this environment, not a blocker — but not yet scripted; M7 should either automate it (e.g. a bootstrap script CI/compose can run) or document it clearly enough for `docs/14-testing-and-acceptance.md`'s clean-checkout criteria. ## Exact next action -Start M4 (n8n automation): read `docs/11-n8n-integration.md`, `n8n/README.md`. Implement the outbox dispatcher (poll pending `outbox_events`, claim with `FOR UPDATE SKIP LOCKED`, POST to n8n's webhook with timeout + exponential backoff + small max-attempt cap, never hold a DB transaction open during the HTTP call), correct/import `n8n/mobilityops-return-processing.json` (credential + the callback route — M2's `vehicle.returned.v1` outbox payload shape is already `contracts/events.schema.json`-compliant, so the workflow should be able to consume it as-is), a narrow MobilityOps callback endpoint the workflow calls back into, an Automation nav item + UI (pending/succeeded/failed list, manual retry button, matches `docs/06-ui-ux.md`'s "Recent automation" concept already on the dashboard), and manual-retry wiring for `POST /api/v1/workflows/{event_id}/retry` per `docs/05-api-contract.md`. The seeded `workflow_runs.csv` already includes one deterministic `failed` row (`event_id 00000000-...-0020`, "Synthetic connection timeout to n8n") for the S5 demo scenario — dashboard/Automation UI must surface it and allow retry. Remember the compose `n8n` service is already up (`http://localhost:5678`, basic auth `admin`/`change-me` from `.env.example`) but the workflow itself hasn't been imported/activated yet this session. +Start M5 (RAGcore knowledge integration): read `docs/09-ragcore-integration.md`, `knowledge/manifest.json`. Implement the `KnowledgeProvider` protocol (`health`, `ask`, `sync_manifest`), a `DemoKnowledgeProvider` doing deterministic keyword/BM25-style retrieval directly over `knowledge/procedures/*.md` (extractive, not generative — must return real excerpts + `insufficient`/`unavailable` states honestly, never fabricate), and a `RAGcoreKnowledgeProvider` adapter stub for the real service (`RAGCORE_BASE_URL` etc. are already in config/compose from M0, but no actual RAGcore instance is confirmed reachable this session — build the demo provider as the one that actually has to work for the acceptance criteria, and make the RAGcore adapter degrade to `unavailable` cleanly if unreachable, per `docs/03-architecture.md`'s reliability boundary "RAGcore failure disables knowledge answers only"). Add `POST /api/v1/knowledge/questions` and `GET /api/v1/knowledge/status`, a Knowledge nav item + page (chat-style question box, source cards with title/version/section/excerpt, explicit unavailable/insufficient states — no fabricated answers ever). S6 demo scenario: "What must I do when a vehicle returns with damage?" must cite `knowledge/procedures/03-damage-handling.md` and `02-vehicle-return.md`. diff --git a/backend/app/api/routers/integrations.py b/backend/app/api/routers/integrations.py new file mode 100644 index 0000000..2bf4c3d --- /dev/null +++ b/backend/app/api/routers/integrations.py @@ -0,0 +1,73 @@ +from __future__ import annotations + +import uuid +from datetime import UTC, datetime +from typing import Any + +from fastapi import APIRouter, Depends, Header +from sqlalchemy import select +from sqlalchemy.orm import Session + +from app.api.deps import get_db +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.services.audit import record_audit_event + +router = APIRouter(prefix="/api/v1/integrations/n8n", tags=["integrations"]) +settings = get_settings() + + +@router.post("/return-callback") +def return_callback( + body: dict[str, Any], + idempotency_key: str = Header(..., alias="Idempotency-Key"), + service_token: str = Header(..., alias="X-Service-Token"), + db: Session = Depends(get_db), +) -> dict: + if service_token != settings.n8n_callback_token: + raise AppError("UNAUTHORIZED_SERVICE", "Invalid service token.", status_code=401) + + try: + event_id = uuid.UUID(idempotency_key) + except ValueError as exc: + raise AppError( + "INVALID_IDEMPOTENCY_KEY", "Idempotency-Key must be the event's UUID.", status_code=422 + ) from exc + + event = db.scalar(select(OutboxEvent).where(OutboxEvent.event_id == event_id)) + if event is None: + raise AppError("EVENT_NOT_FOUND", "No outbox event matches this event ID.", status_code=404) + + # Idempotent by event ID: n8n or our own dispatcher may redeliver the same event + # (e.g. a lost response after a timeout), so this callback must not double-record. + already_recorded = ( + db.scalar( + select(AuditEvent.id).where( + AuditEvent.action == "n8n_return_followup_recorded", + AuditEvent.metadata_json["event_id"].astext == str(event_id), + ) + ) + is not None + ) + if not already_recorded: + record_audit_event( + db, + actor_type="service", + actor_label="n8n", + action="n8n_return_followup_recorded", + entity_type="booking", + correlation_id=uuid.UUID(body.get("correlation_id")) + if body.get("correlation_id") + else None, + after={"follow_up": body.get("follow_up"), "summary": body.get("summary")}, + metadata={"event_id": str(event_id)}, + ) + db.commit() + + return { + "status": "recorded", + "event_id": str(event_id), + "occurred_at": datetime.now(UTC).isoformat(), + } diff --git a/backend/app/api/routers/workflows.py b/backend/app/api/routers/workflows.py new file mode 100644 index 0000000..76ce9c9 --- /dev/null +++ b/backend/app/api/routers/workflows.py @@ -0,0 +1,76 @@ +from __future__ import annotations + +import uuid + +from fastapi import APIRouter, Depends, Query +from sqlalchemy import select +from sqlalchemy.orm import Session + +from app.api.deps import get_db, require_operations_manager +from app.core.errors import AppError +from app.models.outbox import OutboxEvent +from app.schemas import AutomationRunOut, CurrentUser +from app.services.audit import record_audit_event + +router = APIRouter(prefix="/api/v1/workflows", tags=["workflows"]) + + +def _to_out(event: OutboxEvent) -> AutomationRunOut: + return AutomationRunOut( + event_id=str(event.event_id), + event_type=event.event_type, + aggregate_ref=event.payload_json.get("aggregate_ref", ""), + status=event.delivery_status, + attempts=event.attempts, + last_error=event.last_error, + occurred_at=event.occurred_at, + ) + + +@router.get("", response_model=list[AutomationRunOut]) +def list_workflows( + status: str | None = Query(default=None), + db: Session = Depends(get_db), + _user: CurrentUser = Depends(require_operations_manager), +) -> list[AutomationRunOut]: + stmt = select(OutboxEvent).order_by(OutboxEvent.occurred_at.desc()) + if status: + stmt = stmt.where(OutboxEvent.delivery_status == status) + events = db.scalars(stmt).all() + return [_to_out(e) for e in events] + + +@router.post("/{event_id}/retry", response_model=AutomationRunOut) +def retry_workflow( + event_id: str, + db: Session = Depends(get_db), + user: CurrentUser = Depends(require_operations_manager), +) -> AutomationRunOut: + try: + parsed_id = uuid.UUID(event_id) + except ValueError as exc: + raise AppError("INVALID_EVENT_ID", "event_id must be a UUID.", status_code=422) from exc + + event = db.scalar(select(OutboxEvent).where(OutboxEvent.event_id == parsed_id)) + if event is None: + raise AppError("EVENT_NOT_FOUND", "Workflow event not found.", status_code=404) + if event.delivery_status != "failed": + raise AppError( + "NOT_RETRYABLE", + f"Event is '{event.delivery_status}', not 'failed'; " + "only failed deliveries can be retried.", + status_code=409, + ) + + event.delivery_status = "pending" + event.next_attempt_at = None + record_audit_event( + db, + actor_type="user", + actor_label=user.display_name, + action="workflow_retry", + entity_type="outbox_event", + metadata={"event_id": event_id, "previous_attempts": event.attempts}, + ) + db.commit() + return _to_out(event) diff --git a/backend/app/core/config.py b/backend/app/core/config.py index 18346ae..4e3bd54 100644 --- a/backend/app/core/config.py +++ b/backend/app/core/config.py @@ -15,6 +15,11 @@ class Settings(BaseSettings): ragcore_workspace: str = "mobilityops" ragcore_collection: str = "internal-procedures" n8n_webhook_url: str = "http://n8n:5678/webhook/mobilityops-return" + n8n_callback_token: str = "replace-me-n8n-callback-token" + n8n_dispatch_enabled: bool = True + n8n_dispatch_interval_seconds: float = 3.0 + n8n_http_timeout_seconds: float = 5.0 + n8n_max_attempts: int = 5 app_secret: str = "replace-in-production" session_cookie_name: str = "mobilityops_session" session_ttl_seconds: int = 60 * 60 * 8 diff --git a/backend/app/main.py b/backend/app/main.py index 973ddc8..2298b23 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -1,15 +1,35 @@ import uuid +from contextlib import asynccontextmanager from fastapi import FastAPI, HTTPException, Request from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import JSONResponse -from app.api.routers import audit, bookings, dashboard, data_quality, demo, vehicles +from app.api.routers import ( + audit, + bookings, + dashboard, + data_quality, + demo, + integrations, + vehicles, + workflows, +) from app.core.config import get_settings from app.core.errors import AppError, error_body +from app.services.dispatcher import start_background_dispatcher, stop_background_dispatcher settings = get_settings() -app = FastAPI(title="MobilityOps API", version="0.1.0") + + +@asynccontextmanager +async def lifespan(_app: FastAPI): + start_background_dispatcher() + yield + stop_background_dispatcher() + + +app = FastAPI(title="MobilityOps API", version="0.1.0", lifespan=lifespan) app.add_middleware( CORSMiddleware, @@ -62,3 +82,5 @@ app.include_router(vehicles.router) app.include_router(bookings.router) app.include_router(audit.router) app.include_router(data_quality.router) +app.include_router(workflows.router) +app.include_router(integrations.router) diff --git a/backend/app/seed_loader.py b/backend/app/seed_loader.py index c30f2b1..eb8f0ac 100644 --- a/backend/app/seed_loader.py +++ b/backend/app/seed_loader.py @@ -235,16 +235,37 @@ def load_seed(db: Session) -> SeedResult: db.execute(insert(DataQualityIssue), dq_rows) counts["data_quality_issues"] = len(dq_rows) + vehicle_ref_by_booking_ref = { + row["public_ref"]: row["vehicle_ref"] for row in _read_csv("bookings.csv") + } + outbox_rows = [] for row in _read_csv("workflow_runs.csv"): booking_id = booking_id_by_ref.get(row["aggregate_ref"]) + # Build the same schema-complete envelope the live return workflow (M2) produces, + # so a seeded/historical event is redeliverable (e.g. via manual retry) without the + # dispatcher crashing on a missing key. See PROJECT_STATE.md M4 notes. outbox_rows.append( { "event_id": uuid.UUID(row["event_id"]), "event_type": row["event_type"], "aggregate_type": "booking", "aggregate_id": booking_id or uuid.uuid4(), - "payload_json": {"aggregate_ref": row["aggregate_ref"]}, + "payload_json": { + "correlation_id": str(uuid.uuid4()), + "aggregate": { + "type": "booking", + "id": str(booking_id or uuid.uuid4()), + "public_ref": row["aggregate_ref"], + }, + "data": { + "vehicle_ref": vehicle_ref_by_booking_ref.get(row["aggregate_ref"], ""), + "inspection_ref": "", + "resulting_vehicle_status": "cleaning", + "attention_reasons": [], + }, + "aggregate_ref": row["aggregate_ref"], + }, "occurred_at": _parse_dt(row["occurred_at"]), "delivery_status": row["status"], "attempts": int(row["attempts"]), diff --git a/backend/app/services/dispatcher.py b/backend/app/services/dispatcher.py new file mode 100644 index 0000000..67d59a5 --- /dev/null +++ b/backend/app/services/dispatcher.py @@ -0,0 +1,148 @@ +from __future__ import annotations + +import logging +import threading +import uuid +from datetime import UTC, datetime, timedelta + +import httpx +from sqlalchemy import select + +from app.core.config import get_settings +from app.core.db import SessionLocal +from app.models.outbox import OutboxEvent + +logger = logging.getLogger("mobilityops.dispatcher") +settings = get_settings() + +_stop_event = threading.Event() + + +def _backoff_seconds(attempts: int) -> int: + return min(2**attempts, 60) + + +def _claim_due_events(batch_size: int = 5) -> list[uuid.UUID]: + """Claim a batch of due events with a short-lived transaction (no network I/O held open).""" + db = SessionLocal() + try: + now = datetime.now(UTC) + rows = db.scalars( + select(OutboxEvent) + .where( + OutboxEvent.delivery_status == "pending", + (OutboxEvent.next_attempt_at.is_(None)) | (OutboxEvent.next_attempt_at <= now), + ) + .order_by(OutboxEvent.occurred_at.asc()) + .limit(batch_size) + .with_for_update(skip_locked=True) + ).all() + claimed_ids = [row.event_id for row in rows] + for row in rows: + row.delivery_status = "delivering" + db.commit() + return claimed_ids + finally: + db.close() + + +def _deliver_one(event_id: uuid.UUID) -> None: + db = SessionLocal() + try: + event = db.get(OutboxEvent, event_id) + if event is None: + return + # Reconstruct the wire envelope from contracts/events.schema.json: only the fields + # the schema declares (additionalProperties: false), sourced from real columns where + # possible. `payload_json` also carries an internal `aggregate_ref` convenience field + # for our own dashboard/audit reads, which must not be forwarded to n8n. + try: + wire_event = { + "event_id": str(event.event_id), + "event_type": event.event_type, + "occurred_at": event.occurred_at.isoformat(), + "correlation_id": event.payload_json["correlation_id"], + "aggregate": event.payload_json["aggregate"], + "data": event.payload_json["data"], + } + payload_error: str | None = None + except KeyError as exc: + # A malformed payload must still resolve the claimed "delivering" row to a + # terminal-or-retryable state below, rather than leaving it stuck forever. + wire_event = None + payload_error = f"Malformed outbox payload, missing key {exc}" + attempts = event.attempts + finally: + db.close() + + if wire_event is None: + success, error, body = False, payload_error, None + else: + try: + response = httpx.post( + settings.n8n_webhook_url, + json=wire_event, + timeout=settings.n8n_http_timeout_seconds, + ) + response.raise_for_status() + body = response.json() + success = bool(body.get("ok", True)) + error = None if success else f"n8n reported failure: {body}" + except httpx.HTTPError as exc: + success = False + error = f"{type(exc).__name__}: {exc}" + body = None + + db = SessionLocal() + try: + event = db.get(OutboxEvent, event_id) + if event is None: + return + event.attempts = attempts + 1 + if success: + event.delivery_status = "succeeded" + event.last_error = None + event.next_attempt_at = None + event.external_run_id = str((body or {}).get("event_id", event_id)) + else: + event.last_error = (error or "delivery failed")[:2000] + if event.attempts >= settings.n8n_max_attempts: + event.delivery_status = "failed" + event.next_attempt_at = None + else: + event.delivery_status = "pending" + event.next_attempt_at = datetime.now(UTC) + timedelta( + seconds=_backoff_seconds(event.attempts) + ) + db.commit() + finally: + db.close() + + +def run_dispatch_cycle() -> int: + """Run one claim+deliver cycle. Returns the number of events processed.""" + claimed = _claim_due_events() + for event_id in claimed: + _deliver_one(event_id) + return len(claimed) + + +def _loop() -> None: + while not _stop_event.is_set(): + try: + run_dispatch_cycle() + except Exception: # noqa: BLE001 + logger.exception("Outbox dispatch cycle failed") + _stop_event.wait(settings.n8n_dispatch_interval_seconds) + + +def start_background_dispatcher() -> None: + if not settings.n8n_dispatch_enabled: + return + _stop_event.clear() + thread = threading.Thread(target=_loop, name="outbox-dispatcher", daemon=True) + thread.start() + + +def stop_background_dispatcher() -> None: + _stop_event.set() diff --git a/backend/tests/test_dispatcher.py b/backend/tests/test_dispatcher.py new file mode 100644 index 0000000..95e00cb --- /dev/null +++ b/backend/tests/test_dispatcher.py @@ -0,0 +1,177 @@ +from __future__ import annotations + +import uuid +from datetime import UTC, datetime +from types import SimpleNamespace + +from sqlalchemy import select + +from app.core.config import get_settings +from app.core.db import SessionLocal +from app.models.booking import Booking +from app.models.outbox import OutboxEvent +from app.services import dispatcher + + +def _make_pending_event(vehicle_ref: str) -> uuid.UUID: + db = SessionLocal() + try: + booking = db.scalar(select(Booking).where(Booking.status == "returned").limit(1)) + event = OutboxEvent( + event_id=uuid.uuid4(), + event_type="vehicle.returned.v1", + aggregate_type="booking", + aggregate_id=booking.id, + payload_json={ + "correlation_id": str(uuid.uuid4()), + "aggregate": { + "type": "booking", + "id": str(booking.id), + "public_ref": booking.public_ref, + }, + "data": { + "vehicle_ref": vehicle_ref, + "inspection_ref": "INSP-TEST", + "resulting_vehicle_status": "cleaning", + "attention_reasons": [], + }, + "aggregate_ref": booking.public_ref, + }, + occurred_at=datetime.now(UTC), + delivery_status="pending", + attempts=0, + ) + db.add(event) + db.commit() + return event.event_id + finally: + db.close() + + +def _get_event(event_id: uuid.UUID) -> OutboxEvent: + db = SessionLocal() + try: + return db.scalar(select(OutboxEvent).where(OutboxEvent.event_id == event_id)) + finally: + db.close() + + +def test_claim_marks_events_delivering(): + event_id = _make_pending_event("MO-001") + claimed = dispatcher._claim_due_events() + assert event_id in claimed + assert _get_event(event_id).delivery_status == "delivering" + + +def test_deliver_one_success(monkeypatch): + event_id = _make_pending_event("MO-002") + dispatcher._claim_due_events() + + def fake_post(url, json, timeout): + return SimpleNamespace( + raise_for_status=lambda: None, + json=lambda: {"ok": True, "event_id": str(event_id), "result": {}}, + ) + + monkeypatch.setattr(dispatcher.httpx, "post", fake_post) + dispatcher._deliver_one(event_id) + + event = _get_event(event_id) + assert event.delivery_status == "succeeded" + assert event.attempts == 1 + assert event.external_run_id == str(event_id) + assert event.last_error is None + + +def test_deliver_one_failure_schedules_retry(monkeypatch): + event_id = _make_pending_event("MO-003") + dispatcher._claim_due_events() + + def fake_post(url, json, timeout): + raise dispatcher.httpx.ConnectError("simulated connection failure") + + monkeypatch.setattr(dispatcher.httpx, "post", fake_post) + dispatcher._deliver_one(event_id) + + event = _get_event(event_id) + assert event.delivery_status == "pending" + assert event.attempts == 1 + assert event.next_attempt_at is not None + assert "simulated connection failure" in event.last_error + + +def test_deliver_one_exhausts_attempts_to_failed(monkeypatch): + event_id = _make_pending_event("MO-004") + settings = get_settings() + + def fake_post(url, json, timeout): + raise dispatcher.httpx.ConnectError("still down") + + monkeypatch.setattr(dispatcher.httpx, "post", fake_post) + + for _ in range(settings.n8n_max_attempts): + dispatcher._claim_due_events() + db = SessionLocal() + try: + event = db.scalar(select(OutboxEvent).where(OutboxEvent.event_id == event_id)) + event.next_attempt_at = None + db.commit() + finally: + db.close() + dispatcher._deliver_one(event_id) + + event = _get_event(event_id) + assert event.delivery_status == "failed" + assert event.attempts == settings.n8n_max_attempts + + +def test_deliver_one_handles_malformed_payload_without_getting_stuck(monkeypatch): + # Regression test: seeded/legacy outbox rows may lack the full event envelope. Delivery + # must resolve the claimed "delivering" row to pending/failed, never leave it stuck. + db = SessionLocal() + try: + booking = db.scalar(select(Booking).limit(1)) + event = OutboxEvent( + event_id=uuid.uuid4(), + event_type="vehicle.returned.v1", + aggregate_type="booking", + aggregate_id=booking.id, + payload_json={"aggregate_ref": booking.public_ref}, # missing correlation_id/etc. + occurred_at=datetime.now(UTC), + delivery_status="pending", + attempts=0, + ) + db.add(event) + db.commit() + event_id = event.event_id + finally: + db.close() + + def fake_post(url, json, timeout): + raise AssertionError("must not attempt delivery with a malformed payload") + + monkeypatch.setattr(dispatcher.httpx, "post", fake_post) + + dispatcher._claim_due_events() + dispatcher._deliver_one(event_id) + + event = _get_event(event_id) + assert event.delivery_status in ("pending", "failed") + assert event.attempts == 1 + assert "Malformed outbox payload" in event.last_error + + +def test_run_dispatch_cycle_end_to_end(monkeypatch): + event_id = _make_pending_event("MO-005") + + def fake_post(url, json, timeout): + return SimpleNamespace( + raise_for_status=lambda: None, + json=lambda: {"ok": True, "event_id": str(event_id), "result": {}}, + ) + + monkeypatch.setattr(dispatcher.httpx, "post", fake_post) + processed = dispatcher.run_dispatch_cycle() + + assert processed >= 1 + assert _get_event(event_id).delivery_status == "succeeded" diff --git a/backend/tests/test_integrations.py b/backend/tests/test_integrations.py new file mode 100644 index 0000000..1e84f0d --- /dev/null +++ b/backend/tests/test_integrations.py @@ -0,0 +1,84 @@ +import uuid + +from app.core.config import get_settings + + +def _callback_headers(event_id: str, token: str | None = None): + settings = get_settings() + return { + "Idempotency-Key": event_id, + "X-Service-Token": token if token is not None else settings.n8n_callback_token, + } + + +def test_callback_rejects_wrong_service_token(client): + response = client.post( + "/api/v1/integrations/n8n/return-callback", + json={"follow_up": "cleaning"}, + headers=_callback_headers(str(uuid.uuid4()), token="wrong-token"), + ) + assert response.status_code == 401 + + +def test_callback_unknown_event_returns_404(client): + response = client.post( + "/api/v1/integrations/n8n/return-callback", + json={"follow_up": "cleaning"}, + headers=_callback_headers(str(uuid.uuid4())), + ) + assert response.status_code == 404 + + +def test_callback_is_idempotent_by_event_id(client, ops_client): + from sqlalchemy import select + + from app.core.db import SessionLocal + from app.models.booking import Booking + from app.models.outbox import OutboxEvent + + db = SessionLocal() + try: + booking = db.scalar(select(Booking).limit(1)) + event = OutboxEvent( + event_id=uuid.uuid4(), + event_type="vehicle.returned.v1", + aggregate_type="booking", + aggregate_id=booking.id, + payload_json={ + "correlation_id": str(uuid.uuid4()), + "aggregate": { + "type": "booking", + "id": str(booking.id), + "public_ref": booking.public_ref, + }, + "data": {}, + "aggregate_ref": booking.public_ref, + }, + occurred_at=db.execute(select(Booking.starts_at).limit(1)).scalar(), + delivery_status="delivering", + attempts=1, + ) + db.add(event) + db.commit() + event_id = str(event.event_id) + finally: + db.close() + + first = client.post( + "/api/v1/integrations/n8n/return-callback", + json={"follow_up": "cleaning", "summary": "test"}, + headers=_callback_headers(event_id), + ) + second = client.post( + "/api/v1/integrations/n8n/return-callback", + json={"follow_up": "cleaning", "summary": "test"}, + headers=_callback_headers(event_id), + ) + assert first.status_code == 200 + assert second.status_code == 200 + + audit_events = ops_client.get( + "/api/v1/audit", params={"action": "n8n_return_followup_recorded"} + ).json() + matching = [e for e in audit_events if e["metadata"]["event_id"] == event_id] + assert len(matching) == 1 diff --git a/backend/tests/test_workflows.py b/backend/tests/test_workflows.py new file mode 100644 index 0000000..a54dc57 --- /dev/null +++ b/backend/tests/test_workflows.py @@ -0,0 +1,38 @@ +def test_list_workflows_requires_operations_manager(employee_client): + response = employee_client.get("/api/v1/workflows") + assert response.status_code == 403 + + +def test_list_workflows_includes_seeded_failed_run(ops_client): + response = ops_client.get("/api/v1/workflows", params={"status": "failed"}) + assert response.status_code == 200 + runs = response.json() + assert len(runs) >= 1 + assert all(r["status"] == "failed" for r in runs) + + +def test_retry_requires_failed_status(ops_client): + succeeded = ops_client.get("/api/v1/workflows", params={"status": "succeeded"}).json() + target = succeeded[0]["event_id"] + response = ops_client.post(f"/api/v1/workflows/{target}/retry") + assert response.status_code == 409 + assert response.json()["error"]["code"] == "NOT_RETRYABLE" + + +def test_retry_failed_run_moves_to_pending_and_audits(ops_client): + failed = ops_client.get("/api/v1/workflows", params={"status": "failed"}).json() + target = failed[0]["event_id"] + + response = ops_client.post(f"/api/v1/workflows/{target}/retry") + assert response.status_code == 200 + assert response.json()["status"] == "pending" + + audit_events = ops_client.get("/api/v1/audit", params={"action": "workflow_retry"}).json() + assert len(audit_events) >= 1 + + +def test_retry_requires_operations_manager(employee_client): + response = employee_client.post( + "/api/v1/workflows/00000000-0000-4000-8000-000000000020/retry" + ) + assert response.status_code == 403 diff --git a/compose.yaml b/compose.yaml index d7af9a6..934169f 100644 --- a/compose.yaml +++ b/compose.yaml @@ -33,6 +33,7 @@ services: RAGCORE_COLLECTION: ${RAGCORE_COLLECTION:-internal-procedures} RAGCORE_API_TOKEN: ${RAGCORE_API_TOKEN:-} N8N_WEBHOOK_URL: ${N8N_WEBHOOK_URL:-http://n8n:5678/webhook/mobilityops-return} + N8N_CALLBACK_TOKEN: ${MOBILITYOPS_CALLBACK_TOKEN:-replace-me-n8n-callback-token} ports: - "8128:8000" depends_on: @@ -66,6 +67,8 @@ services: N8N_BASIC_AUTH_USER: ${N8N_BASIC_AUTH_USER:-admin} N8N_BASIC_AUTH_PASSWORD: ${N8N_BASIC_AUTH_PASSWORD:-change-me} N8N_SECURE_COOKIE: "false" + N8N_BLOCK_ENV_ACCESS_IN_NODE: "false" + MOBILITYOPS_CALLBACK_TOKEN: ${MOBILITYOPS_CALLBACK_TOKEN:-replace-me-n8n-callback-token} ports: - "5678:5678" volumes: diff --git a/frontend/src/App.tsx b/frontend/src/App.tsx index cd5fe72..22968c5 100644 --- a/frontend/src/App.tsx +++ b/frontend/src/App.tsx @@ -10,6 +10,7 @@ import { Bookings } from "./pages/Bookings"; import { BookingDetail } from "./pages/BookingDetail"; import { DataQuality } from "./pages/DataQuality"; import { DataQualityIssueDetail } from "./pages/DataQualityIssueDetail"; +import { Automation } from "./pages/Automation"; import { Audit } from "./pages/Audit"; export function App() { @@ -31,6 +32,7 @@ export function App() { } /> } /> } /> + } /> } /> } /> diff --git a/frontend/src/components/Layout.tsx b/frontend/src/components/Layout.tsx index e83e06b..0d47261 100644 --- a/frontend/src/components/Layout.tsx +++ b/frontend/src/components/Layout.tsx @@ -6,6 +6,7 @@ const NAV_ITEMS = [ { to: "/vehicles", label: "Vehicles" }, { to: "/bookings", label: "Bookings" }, { to: "/data-quality", label: "Data Quality" }, + { to: "/automation", label: "Automation" }, { to: "/audit", label: "Audit" }, ]; diff --git a/frontend/src/pages/Automation.tsx b/frontend/src/pages/Automation.tsx new file mode 100644 index 0000000..c5e6ac4 --- /dev/null +++ b/frontend/src/pages/Automation.tsx @@ -0,0 +1,129 @@ +import { useCallback, useEffect, useState } from "react"; +import { api, ApiError } from "../api/client"; +import type { AutomationRun } from "../api/types"; +import { StatusBadge } from "../components/Badge"; +import { useAuth } from "../context/AuthContext"; + +export function Automation() { + const { user } = useAuth(); + const [runs, setRuns] = useState(null); + const [error, setError] = useState(null); + const [status, setStatus] = useState(""); + const [retryError, setRetryError] = useState(null); + const [retrying, setRetrying] = useState(null); + + const load = useCallback(() => { + const params = new URLSearchParams(); + if (status) params.set("status", status); + api + .get(`/api/v1/workflows?${params.toString()}`) + .then(setRuns) + .catch(() => + setError( + user?.role === "operations_manager" + ? "Automation runs are unavailable right now." + : "Automation is only visible to Operations Managers.", + ), + ); + }, [status, user]); + + useEffect(() => { + load(); + }, [load]); + + async function handleRetry(eventId: string) { + setRetryError(null); + setRetrying(eventId); + try { + await api.post(`/api/v1/workflows/${eventId}/retry`); + load(); + } catch (err) { + setRetryError(err instanceof ApiError ? err.message : "Could not retry this delivery."); + } finally { + setRetrying(null); + } + } + + if (user?.role !== "operations_manager") { + return ( +
+

Automation

+

Automation delivery status is visible to Operations Managers only.

+
+ ); + } + + return ( +
+

Automation

+ +
+ +
+ + {error &&

{error}

} + {retryError &&

{retryError}

} + {!error && !runs &&

Loading automation runs…

} + {runs && runs.length === 0 &&

No automation runs match this filter.

} + + {runs && runs.length > 0 && ( + + + + + + + + + + + + + + + + {runs.map((r) => ( + + + + + + + + + + + ))} + +
Automation runs
EventTypeBookingStatusAttemptsLast errorWhenAction
{r.event_id.slice(0, 8)}{r.event_type}{r.aggregate_ref} + + {r.attempts}{r.last_error ?? "—"} + + + {r.status === "failed" ? ( + + ) : ( + "—" + )} +
+ )} +
+ ); +} diff --git a/frontend/src/styles.css b/frontend/src/styles.css index 1b22dd8..428005e 100644 --- a/frontend/src/styles.css +++ b/frontend/src/styles.css @@ -220,6 +220,12 @@ a { color: #1f5c8f; } background: white; font-weight: 700; cursor: pointer; } +.data-table td button { + padding: 6px 14px; border-radius: 8px; border: 1px solid #14324f; + background: #14324f; color: white; font-weight: 700; cursor: pointer; font-size: 0.85rem; +} +.data-table td button:disabled { opacity: 0.6; cursor: not-allowed; } + @media (max-width: 700px) { .app-header { flex-direction: column; align-items: flex-start; } .user-badge { margin-left: 0; } diff --git a/n8n/mobilityops-return-processing.json b/n8n/mobilityops-return-processing.json index 22340f1..49551e2 100644 --- a/n8n/mobilityops-return-processing.json +++ b/n8n/mobilityops-return-processing.json @@ -1,4 +1,5 @@ { + "id": "mobilityops-return-processing", "name": "MobilityOps - Vehicle Return Processing", "nodes": [ { @@ -41,6 +42,10 @@ { "name": "Idempotency-Key", "value": "={{$json.event_id}}" + }, + { + "name": "X-Service-Token", + "value": "={{$env.MOBILITYOPS_CALLBACK_TOKEN}}" } ] }, @@ -113,7 +118,7 @@ "settings": { "executionOrder": "v1" }, - "active": false, + "active": true, "versionId": "11111111-1111-4111-8111-111111111111", "meta": { "templateCredsSetupCompleted": false