From 0da52515241f35514bdc91a0d0684e0050b9a625 Mon Sep 17 00:00:00 2001 From: NuklearRabbit <145918611+NuklearRabbit@users.noreply.github.com> Date: Tue, 4 Aug 2026 19:47:39 +0200 Subject: [PATCH] n8n: add backend endpoints for the RAGcore Procedure Sync workflow GET /api/v1/integrations/n8n/procedures lists every procedure Markdown file Fleet Ops ships (all languages) with a stable per-document id and content hash, ready for workflow 3 to push into RAGcore. POST .../procedures-sync-result records the sync outcome as an idempotent audit event, matching the existing return-callback/workflow-error pattern. Extracted frontmatter parsing out of the demo knowledge provider into a shared module so both read the same source of truth. --- backend/app/api/routers/integrations.py | 76 +++++++++++++++++++- backend/app/schemas.py | 26 +++++++ backend/app/services/knowledge/demo.py | 20 +----- backend/app/services/knowledge/procedures.py | 71 ++++++++++++++++++ backend/tests/test_integrations.py | 66 +++++++++++++++++ 5 files changed, 240 insertions(+), 19 deletions(-) create mode 100644 backend/app/services/knowledge/procedures.py diff --git a/backend/app/api/routers/integrations.py b/backend/app/api/routers/integrations.py index 86b8271..16824d1 100644 --- a/backend/app/api/routers/integrations.py +++ b/backend/app/api/routers/integrations.py @@ -2,6 +2,7 @@ from __future__ import annotations import uuid from datetime import UTC, datetime +from pathlib import Path from typing import Any from fastapi import APIRouter, Depends, Header @@ -13,9 +14,18 @@ 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, WorkflowErrorReportIn, WorkflowErrorReportResult +from app.schemas import ( + ProcedureDocumentOut, + ProcedureListOut, + ProcedureSyncResultIn, + ProcedureSyncResultResult, + ScanResultOut, + WorkflowErrorReportIn, + WorkflowErrorReportResult, +) from app.services.audit import record_audit_event from app.services.data_quality import run_scan +from app.services.knowledge.procedures import iter_procedure_documents router = APIRouter(prefix="/api/v1/integrations/n8n", tags=["integrations"]) settings = get_settings() @@ -146,3 +156,67 @@ def workflow_error( execution_id=body.execution_id, occurred_at=datetime.now(UTC), ) + + +@router.get("/procedures", response_model=ProcedureListOut) +def list_procedures(service_token: str = Header(..., alias="X-Service-Token")) -> ProcedureListOut: + """Read-only source list for the RAGcore Procedure Sync workflow: every procedure + Markdown file Fleet Ops ships, across every supported language, with a stable + per-document id (source_id) and a content hash so the caller can detect changes + without re-fetching content it already has.""" + if service_token != settings.n8n_callback_token: + raise AppError("UNAUTHORIZED_SERVICE", "Invalid service token.", status_code=401) + + documents = [ + ProcedureDocumentOut( + id=doc.source_id, + language=doc.language, + document_id=doc.document_id, + title=doc.title, + version=doc.version, + content=doc.content, + content_hash=doc.content_hash, + ) + for doc in iter_procedure_documents(Path(settings.knowledge_dir)) + ] + return ProcedureListOut(documents=documents) + + +@router.post("/procedures-sync-result", response_model=ProcedureSyncResultResult) +def procedures_sync_result( + body: ProcedureSyncResultIn, + service_token: str = Header(..., alias="X-Service-Token"), + db: Session = Depends(get_db), +) -> ProcedureSyncResultResult: + """Receives a summary (counts only, no document content) from the n8n "Fleet Ops -- + RAGcore Procedure Sync" workflow once it finishes uploading procedures to RAGcore. + Idempotent on execution_id, matching the workflow-error and return-callback pattern.""" + if service_token != settings.n8n_callback_token: + raise AppError("UNAUTHORIZED_SERVICE", "Invalid service token.", status_code=401) + + already_recorded = ( + db.scalar( + select(AuditEvent.id).where( + AuditEvent.action == "n8n_procedures_synced", + AuditEvent.metadata_json["execution_id"].astext == body.execution_id, + ) + ) + is not None + ) + if not already_recorded: + record_audit_event( + db, + actor_type="service", + actor_label="n8n procedure sync", + action="n8n_procedures_synced", + entity_type="automation", + after={"synced": body.synced, "failed": body.failed}, + metadata={"execution_id": body.execution_id}, + ) + db.commit() + + return ProcedureSyncResultResult( + status="already_registered" if already_recorded else "registered", + execution_id=body.execution_id, + occurred_at=datetime.now(UTC), + ) diff --git a/backend/app/schemas.py b/backend/app/schemas.py index 8ce6126..af3775a 100644 --- a/backend/app/schemas.py +++ b/backend/app/schemas.py @@ -164,6 +164,32 @@ class WorkflowErrorReportResult(BaseModel): occurred_at: datetime +class ProcedureDocumentOut(BaseModel): + id: str + language: str + document_id: str + title: str + version: str + content: str + content_hash: str + + +class ProcedureListOut(BaseModel): + documents: list[ProcedureDocumentOut] + + +class ProcedureSyncResultIn(BaseModel): + execution_id: str = Field(max_length=120) + synced: int = Field(ge=0) + failed: int = Field(default=0, ge=0) + + +class ProcedureSyncResultResult(BaseModel): + status: Literal["registered", "already_registered"] + execution_id: str + occurred_at: datetime + + class ProvideFieldsRequest(BaseModel): fields: dict[str, str] diff --git a/backend/app/services/knowledge/demo.py b/backend/app/services/knowledge/demo.py index 0e4aa16..04312e0 100644 --- a/backend/app/services/knowledge/demo.py +++ b/backend/app/services/knowledge/demo.py @@ -7,6 +7,7 @@ from pathlib import Path from app.core.config import get_settings from app.services.knowledge import GroundedAnswer, KnowledgeHealth, SourceCard +from app.services.knowledge.procedures import parse_frontmatter SUPPORTED_LANGUAGES = ("nl-BE", "en-GB", "fr-BE") DEFAULT_LANGUAGE = "en-GB" @@ -74,23 +75,6 @@ class ScoredSection: body_tokens: set[str] -def _parse_frontmatter(raw: str) -> tuple[dict[str, str], str]: - if not raw.startswith("---"): - return {}, raw - end = raw.find("\n---", 3) - if end == -1: - return {}, raw - block = raw[3:end].strip() - body = raw[end + 4 :].lstrip("\n") - meta: dict[str, str] = {} - for line in block.splitlines(): - if ":" not in line: - continue - key, _, value = line.partition(":") - meta[key.strip()] = value.strip().strip('"') - return meta, body - - def _split_sections(body: str) -> list[tuple[str, str]]: sections: list[tuple[str, str]] = [] current_heading = "Overview" @@ -114,7 +98,7 @@ def _load_sections(procedures_dir: Path, language: str) -> list[ScoredSection]: sections: list[ScoredSection] = [] for path in sorted(procedures_dir.glob("*.md")): raw = path.read_text(encoding="utf-8") - meta, body = _parse_frontmatter(raw) + meta, body = parse_frontmatter(raw) title = meta.get("title", path.stem) doc = Document( document_id=meta.get("document_id", path.stem), diff --git a/backend/app/services/knowledge/procedures.py b/backend/app/services/knowledge/procedures.py new file mode 100644 index 0000000..a4a995e --- /dev/null +++ b/backend/app/services/knowledge/procedures.py @@ -0,0 +1,71 @@ +from __future__ import annotations + +import hashlib +import uuid +from dataclasses import dataclass +from pathlib import Path + +SUPPORTED_LANGUAGES = ("nl-BE", "en-GB", "fr-BE") + +# Stable across runs (and across which language ships first) so a document's RAGcore +# source_id never changes just because the sync ran on a different day or in a +# different order -- required for RAGcore's upload idempotency to work per document. +_SOURCE_ID_NAMESPACE = uuid.uuid5(uuid.NAMESPACE_URL, "https://mobilityops.internal/knowledge/procedures") + + +def parse_frontmatter(raw: str) -> tuple[dict[str, str], str]: + if not raw.startswith("---"): + return {}, raw + end = raw.find("\n---", 3) + if end == -1: + return {}, raw + block = raw[3:end].strip() + body = raw[end + 4 :].lstrip("\n") + meta: dict[str, str] = {} + for line in block.splitlines(): + if ":" not in line: + continue + key, _, value = line.partition(":") + meta[key.strip()] = value.strip().strip('"') + return meta, body + + +@dataclass(frozen=True) +class ProcedureDocument: + source_id: str + language: str + document_id: str + title: str + version: str + content: str + content_hash: str + + +def iter_procedure_documents(knowledge_dir: Path) -> list[ProcedureDocument]: + """Read every procedure Markdown file Fleet Ops ships, across every supported + language, as a flat list ready for external sync (e.g. into RAGcore). Frontmatter + fields (title, version) come from the same files the demo knowledge provider + already reads -- see parse_frontmatter -- so the two never drift apart.""" + + documents: list[ProcedureDocument] = [] + for language in SUPPORTED_LANGUAGES: + language_dir = knowledge_dir / language + if not language_dir.is_dir(): + continue + for path in sorted(language_dir.glob("*.md")): + raw = path.read_text(encoding="utf-8") + meta, body = parse_frontmatter(raw) + document_id = meta.get("document_id", path.stem) + content = body.strip() + documents.append( + ProcedureDocument( + source_id=str(uuid.uuid5(_SOURCE_ID_NAMESPACE, f"{language}:{document_id}")), + language=language, + document_id=document_id, + title=meta.get("title", path.stem), + version=meta.get("version", "1.0"), + content=content, + content_hash=hashlib.sha256(content.encode("utf-8")).hexdigest(), + ) + ) + return documents diff --git a/backend/tests/test_integrations.py b/backend/tests/test_integrations.py index 58c8bdc..0ced5ae 100644 --- a/backend/tests/test_integrations.py +++ b/backend/tests/test_integrations.py @@ -191,3 +191,69 @@ def test_workflow_error_bounds_summary_length(client): headers={"X-Service-Token": settings.n8n_callback_token}, ) assert response.status_code == 422 + + +def test_procedures_rejects_wrong_service_token(client): + response = client.get( + "/api/v1/integrations/n8n/procedures", headers={"X-Service-Token": "wrong-token"} + ) + assert response.status_code == 401 + + +def test_procedures_lists_every_language_with_stable_ids(client): + settings = get_settings() + response = client.get( + "/api/v1/integrations/n8n/procedures", + headers={"X-Service-Token": settings.n8n_callback_token}, + ) + assert response.status_code == 200 + documents = response.json()["documents"] + assert len(documents) > 0 + assert {d["language"] for d in documents} == {"en-GB", "nl-BE", "fr-BE"} + checkout_docs = [d for d in documents if d["document_id"] == "vehicle-checkout-procedure"] + assert len(checkout_docs) == 3 # one per language + assert all(d["content"] and d["content_hash"] for d in checkout_docs) + # Same document_id, different language, must not collide on id. + assert len({d["id"] for d in checkout_docs}) == 3 + + second_response = client.get( + "/api/v1/integrations/n8n/procedures", + headers={"X-Service-Token": settings.n8n_callback_token}, + ) + second_ids = {d["id"] for d in second_response.json()["documents"]} + assert second_ids == {d["id"] for d in documents} # ids are stable across requests + + +def test_procedures_sync_result_rejects_wrong_service_token(client): + response = client.post( + "/api/v1/integrations/n8n/procedures-sync-result", + json={"execution_id": str(uuid.uuid4()), "synced": 5, "failed": 0}, + headers={"X-Service-Token": "wrong-token"}, + ) + assert response.status_code == 401 + + +def test_procedures_sync_result_registers_and_is_idempotent(client, ops_client): + settings = get_settings() + headers = {"X-Service-Token": settings.n8n_callback_token} + execution_id = str(uuid.uuid4()) + body = {"execution_id": execution_id, "synced": 33, "failed": 1} + + first = client.post( + "/api/v1/integrations/n8n/procedures-sync-result", json=body, headers=headers + ) + second = client.post( + "/api/v1/integrations/n8n/procedures-sync-result", json=body, headers=headers + ) + + assert first.status_code == 200 + assert first.json()["status"] == "registered" + assert second.status_code == 200 + assert second.json()["status"] == "already_registered" + + audit_events = ops_client.get( + "/api/v1/audit", params={"action": "n8n_procedures_synced"} + ).json() + matching = [e for e in audit_events if e["metadata"]["execution_id"] == execution_id] + assert len(matching) == 1 + assert matching[0]["after"] == {"synced": 33, "failed": 1}