From 58fb5153379359b9b592a4d6f2bb601f87df10eb Mon Sep 17 00:00:00 2001 From: NuklearRabbit <145918611+NuklearRabbit@users.noreply.github.com> Date: Mon, 10 Aug 2026 03:31:43 +0200 Subject: [PATCH] M14: add n8n execution health telemetry --- PROJECT_STATE.md | 27 ++++++ backend/app/api/routers/integrations.py | 54 ++++++++++++ backend/app/schemas.py | 16 ++++ backend/app/seed_loader.py | 1 + backend/app/services/integration_status.py | 83 +++++++++++++++++-- backend/tests/test_integration_status.py | 1 + backend/tests/test_integrations.py | 37 +++++++++ frontend/src/api/types.ts | 3 + .../src/i18n/locales/en-GB/integrations.json | 4 + .../src/i18n/locales/fr-BE/integrations.json | 6 +- .../src/i18n/locales/nl-BE/integrations.json | 4 + frontend/src/pages/Automation.tsx | 12 ++- n8n/workflows/MANIFEST.md | 20 +++-- .../fleet-ops-data-quality-scan.json | 30 +++++++ n8n/workflows/fleet-ops-error-handler.json | 30 +++++++ .../fleet-ops-ragcore-procedure-sync.json | 30 +++++++ n8n/workflows/fleet-ops-vehicle-return.json | 32 ++++++- 17 files changed, 368 insertions(+), 22 deletions(-) diff --git a/PROJECT_STATE.md b/PROJECT_STATE.md index d58e9a0..6337434 100644 --- a/PROJECT_STATE.md +++ b/PROJECT_STATE.md @@ -2329,3 +2329,30 @@ evidence yet." **210 passed, 1 warning**. - Exact next action: add authenticated n8n workflow heartbeats and execution results, surface stale/healthy workflow state, then update generated contracts and E2E coverage. + +## n8n execution health telemetry (2026-08-10) + +- Added an authenticated, canonical-workflow-only, idempotent heartbeat contract. Status + now records each workflow's last execution ID/result and classifies it as healthy, + stale, failed or without evidence. Scheduled scan evidence expires after 2h30 and the + daily RAGcore sync after 30h; event-driven workflows are not falsely marked stale just + because no business event arrived. +- Registered target-workflow failures override older success evidence until a newer + successful execution arrives. The Automation UI renders these explicit states and + execution IDs instead of treating any historical timestamp as permanently healthy. +- All four versioned n8n definitions now report a successful execution heartbeat with + bounded retries; the central error handler continues to register failed target runs. +- Evidence: JSON validation for all four definitions, frontend lint, ruff and mypy passed; + focused integration tests **26 passed** and full Unraid suite **212 passed, 1 warning**. +- Live deployment: API/web deployed at `c9a8609`; all five pre-existing n8n workflows + were exported to the recoverable appdata backup + `backups/mobilityops-pre-heartbeat-20260810.json` before the four definitions were + imported and published. The import initially exposed n8n CLI's unsafe name-only + credential resolution (both Header Auth nodes resolved to the service credential); + credential IDs were restored from the backup before republishing. A real return then + completed through outbox → server n8n → callback → heartbeat as execution `337`; + its first 403 delivery remained safely retryable and succeeded after the credential + correction. Live n8n state is operational and Vehicle Return is `healthy` with its + execution ID visible. +- Exact next action: regenerate the checked-in OpenAPI contract, add E2E coverage for the + new operator workflows, document credential-safe n8n upgrades and run acceptance. diff --git a/backend/app/api/routers/integrations.py b/backend/app/api/routers/integrations.py index 16824d1..aea4429 100644 --- a/backend/app/api/routers/integrations.py +++ b/backend/app/api/routers/integrations.py @@ -15,6 +15,8 @@ from app.core.errors import AppError from app.models.audit import AuditEvent from app.models.outbox import OutboxEvent from app.schemas import ( + N8nHeartbeatIn, + N8nHeartbeatResult, ProcedureDocumentOut, ProcedureListOut, ProcedureSyncResultIn, @@ -30,6 +32,58 @@ from app.services.knowledge.procedures import iter_procedure_documents router = APIRouter(prefix="/api/v1/integrations/n8n", tags=["integrations"]) settings = get_settings() +_CANONICAL_WORKFLOW_NAMES = frozenset( + { + "Fleet Ops — Vehicle Return Orchestration", + "Fleet Ops — Scheduled Data Quality Scan", + "Fleet Ops — RAGcore Procedure Sync", + "Fleet Ops — Workflow Error Handler", + } +) + + +@router.post("/heartbeat", response_model=N8nHeartbeatResult) +def workflow_heartbeat( + body: N8nHeartbeatIn, + service_token: str = Header(..., alias="X-Service-Token"), + db: Session = Depends(get_db), +) -> N8nHeartbeatResult: + """Authenticated, idempotent execution evidence from a canonical n8n workflow.""" + if service_token != settings.n8n_callback_token: + raise AppError("UNAUTHORIZED_SERVICE", "Invalid service token.", status_code=401) + if body.workflow_name not in _CANONICAL_WORKFLOW_NAMES: + raise AppError("UNKNOWN_WORKFLOW", "Unknown Fleet Ops workflow.", status_code=422) + already_recorded = ( + db.scalar( + select(AuditEvent.id).where( + AuditEvent.action == "n8n_workflow_heartbeat", + AuditEvent.metadata_json["execution_id"].astext == body.execution_id, + AuditEvent.after_json["status"].astext == body.status, + ) + ) + is not None + ) + if not already_recorded: + record_audit_event( + db, + actor_type="service", + actor_label="n8n workflow heartbeat", + action="n8n_workflow_heartbeat", + entity_type="automation", + after={ + "workflow_id": body.workflow_id, + "workflow_name": body.workflow_name, + "status": body.status, + }, + metadata={"execution_id": body.execution_id}, + ) + db.commit() + return N8nHeartbeatResult( + status="already_registered" if already_recorded else "registered", + execution_id=body.execution_id, + occurred_at=datetime.now(UTC), + ) + @router.post("/return-callback") def return_callback( diff --git a/backend/app/schemas.py b/backend/app/schemas.py index 00ff855..e41fed7 100644 --- a/backend/app/schemas.py +++ b/backend/app/schemas.py @@ -363,6 +363,22 @@ class N8nWorkflowEvidence(BaseModel): name: str built: bool last_seen_at: datetime | None + state: Literal["no_evidence", "healthy", "stale", "failed"] = "no_evidence" + last_status: Literal["succeeded", "failed"] | None = None + last_execution_id: str | None = None + + +class N8nHeartbeatIn(BaseModel): + workflow_id: str = Field(min_length=1, max_length=120) + workflow_name: str = Field(min_length=1, max_length=200) + execution_id: str = Field(min_length=1, max_length=120) + status: Literal["succeeded", "failed"] + + +class N8nHeartbeatResult(BaseModel): + status: Literal["registered", "already_registered"] + execution_id: str + occurred_at: datetime class N8nErrorHandlerStatus(BaseModel): diff --git a/backend/app/seed_loader.py b/backend/app/seed_loader.py index db7bba2..cbd823d 100644 --- a/backend/app/seed_loader.py +++ b/backend/app/seed_loader.py @@ -85,6 +85,7 @@ _PERSISTENT_TELEMETRY_ACTIONS = ( "n8n_return_followup_recorded", "n8n_workflow_failure_registered", "n8n_procedures_synced", + "n8n_workflow_heartbeat", "knowledge_question_asked", ) diff --git a/backend/app/services/integration_status.py b/backend/app/services/integration_status.py index e778751..2fa9136 100644 --- a/backend/app/services/integration_status.py +++ b/backend/app/services/integration_status.py @@ -1,5 +1,6 @@ from __future__ import annotations +from datetime import UTC, datetime, timedelta from typing import Literal import httpx @@ -26,6 +27,10 @@ _CANONICAL_WORKFLOWS = ( "Fleet Ops — RAGcore Procedure Sync", "Fleet Ops — Workflow Error Handler", ) +_STALE_AFTER = { + "Fleet Ops — Scheduled Data Quality Scan": timedelta(hours=2, minutes=30), + "Fleet Ops — RAGcore Procedure Sync": timedelta(hours=30), +} def derive_n8n_status(db: Session) -> N8nIntegrationStatus: @@ -129,20 +134,82 @@ def derive_n8n_status(db: Session) -> N8nIntegrationStatus: (latest_failure_row[1] or {}).get("workflow_name") if latest_failure_row else None ) - evidence_by_workflow = { + legacy_evidence_by_workflow = { "Fleet Ops — Vehicle Return Orchestration": latest_success_at, "Fleet Ops — Scheduled Data Quality Scan": latest_scan_at, "Fleet Ops — RAGcore Procedure Sync": latest_procedure_sync_at, "Fleet Ops — Workflow Error Handler": latest_handler_failure_at, } - workflows = [ - N8nWorkflowEvidence( - name=name, - built=True, - last_seen_at=evidence_by_workflow[name], + heartbeat_by_workflow: dict[str, tuple[datetime, str, str | None]] = {} + heartbeat_rows = db.execute( + select(AuditEvent.occurred_at, AuditEvent.after_json, AuditEvent.metadata_json) + .where(AuditEvent.action == "n8n_workflow_heartbeat") + .order_by(AuditEvent.occurred_at.desc()) + ).all() + for occurred_at, after, metadata in heartbeat_rows: + workflow_name = (after or {}).get("workflow_name") + if workflow_name in _CANONICAL_WORKFLOWS and workflow_name not in heartbeat_by_workflow: + heartbeat_by_workflow[workflow_name] = ( + occurred_at, + (after or {}).get("status", "succeeded"), + (metadata or {}).get("execution_id"), + ) + + failure_by_workflow: dict[str, tuple[datetime, str | None]] = {} + failure_rows = db.execute( + select(AuditEvent.occurred_at, AuditEvent.after_json, AuditEvent.metadata_json) + .where(AuditEvent.action == "n8n_workflow_failure_registered") + .order_by(AuditEvent.occurred_at.desc()) + ).all() + for occurred_at, after, metadata in failure_rows: + workflow_name = (after or {}).get("workflow_name") + if workflow_name in _CANONICAL_WORKFLOWS and workflow_name not in failure_by_workflow: + failure_by_workflow[workflow_name] = ( + occurred_at, + (metadata or {}).get("execution_id"), + ) + + now = datetime.now(UTC) + workflows: list[N8nWorkflowEvidence] = [] + for name in _CANONICAL_WORKFLOWS: + legacy_seen = legacy_evidence_by_workflow[name] + heartbeat = heartbeat_by_workflow.get(name) + failure_signal = failure_by_workflow.get(name) + seen_at = heartbeat[0] if heartbeat else legacy_seen + last_status: Literal["succeeded", "failed"] | None = ( + "succeeded" if seen_at is not None else None ) - for name in _CANONICAL_WORKFLOWS - ] + execution_id = heartbeat[2] if heartbeat else None + if heartbeat and heartbeat[1] == "failed": + last_status = "failed" + if failure_signal and (seen_at is None or failure_signal[0] > seen_at): + seen_at = failure_signal[0] + last_status = "failed" + execution_id = failure_signal[1] + workflow_state: Literal["no_evidence", "healthy", "stale", "failed"] + if seen_at is None: + workflow_state = "no_evidence" + elif last_status == "failed": + workflow_state = "failed" + elif name in _STALE_AFTER and now - seen_at > _STALE_AFTER[name]: + workflow_state = "stale" + else: + workflow_state = "healthy" + workflows.append( + N8nWorkflowEvidence( + name=name, + built=True, + last_seen_at=seen_at, + state=workflow_state, + last_status=last_status, + last_execution_id=execution_id, + ) + ) + + if state in {"operational", "no_evidence"} and any( + workflow.state in {"failed", "stale"} for workflow in workflows + ): + state = "degraded" return N8nIntegrationStatus( configured=bool(settings.n8n_webhook_url), diff --git a/backend/tests/test_integration_status.py b/backend/tests/test_integration_status.py index 859ea94..08223e6 100644 --- a/backend/tests/test_integration_status.py +++ b/backend/tests/test_integration_status.py @@ -124,6 +124,7 @@ def test_integration_status_lists_all_four_canonical_workflows(ops_client): # their own first real signal -- there is no run evidence yet either. assert ragcore_sync["built"] is True assert ragcore_sync["last_seen_at"] is None + assert ragcore_sync["state"] == "no_evidence" def test_integration_status_scheduled_scan_evidence_only_counts_service_runs(client, ops_client): diff --git a/backend/tests/test_integrations.py b/backend/tests/test_integrations.py index cdd4189..e7473b5 100644 --- a/backend/tests/test_integrations.py +++ b/backend/tests/test_integrations.py @@ -264,3 +264,40 @@ def test_procedures_sync_result_registers_and_is_idempotent(client, ops_client): status = ops_client.get("/api/v1/integrations/status").json()["n8n"] ragcore_sync = next(w for w in status["workflows"] if "RAGcore" in w["name"]) assert ragcore_sync["last_seen_at"] is not None + + +def test_workflow_heartbeat_is_idempotent_and_drives_live_status(client, ops_client): + settings = get_settings() + execution_id = str(uuid.uuid4()) + body = { + "workflow_id": "mobilityops-scheduled-quality-scan", + "workflow_name": "Fleet Ops — Scheduled Data Quality Scan", + "execution_id": execution_id, + "status": "succeeded", + } + headers = {"X-Service-Token": settings.n8n_callback_token} + first = client.post("/api/v1/integrations/n8n/heartbeat", json=body, headers=headers) + second = client.post("/api/v1/integrations/n8n/heartbeat", json=body, headers=headers) + assert first.status_code == 200 + assert first.json()["status"] == "registered" + assert second.json()["status"] == "already_registered" + + status = ops_client.get("/api/v1/integrations/status").json()["n8n"] + workflow = next(w for w in status["workflows"] if w["name"] == body["workflow_name"]) + assert workflow["state"] == "healthy" + assert workflow["last_status"] == "succeeded" + assert workflow["last_execution_id"] == execution_id + + +def test_workflow_heartbeat_rejects_unknown_workflow(client): + response = client.post( + "/api/v1/integrations/n8n/heartbeat", + json={ + "workflow_id": "unknown", + "workflow_name": "Unknown workflow", + "execution_id": "test-unknown-001", + "status": "succeeded", + }, + headers={"X-Service-Token": get_settings().n8n_callback_token}, + ) + assert response.status_code == 422 diff --git a/frontend/src/api/types.ts b/frontend/src/api/types.ts index b614093..530f50e 100644 --- a/frontend/src/api/types.ts +++ b/frontend/src/api/types.ts @@ -315,6 +315,9 @@ export interface N8nWorkflowEvidence { name: string; built: boolean; last_seen_at: string | null; + state: "no_evidence" | "healthy" | "stale" | "failed"; + last_status: "succeeded" | "failed" | null; + last_execution_id: string | null; } export interface N8nErrorHandlerStatus { diff --git a/frontend/src/i18n/locales/en-GB/integrations.json b/frontend/src/i18n/locales/en-GB/integrations.json index d38a465..e747180 100644 --- a/frontend/src/i18n/locales/en-GB/integrations.json +++ b/frontend/src/i18n/locales/en-GB/integrations.json @@ -37,6 +37,10 @@ "description": "{{known}} of {{expected}} canonical n8n workflows have live evidence of running.", "notBuilt": "Not built yet", "noEvidence": "No evidence yet", + "healthy": "Healthy", + "stale": "Evidence is stale", + "failed": "Latest execution failed", + "executionId": "Execution", "columns": { "name": "Workflow", "status": "Status", diff --git a/frontend/src/i18n/locales/fr-BE/integrations.json b/frontend/src/i18n/locales/fr-BE/integrations.json index cf8005c..2e637f0 100644 --- a/frontend/src/i18n/locales/fr-BE/integrations.json +++ b/frontend/src/i18n/locales/fr-BE/integrations.json @@ -36,7 +36,11 @@ "title": "Workflows d'automatisation", "description": "{{known}} workflows n8n canoniques sur {{expected}} disposent de preuves actuelles de fonctionnement.", "notBuilt": "Pas encore créé", - "noEvidence": "Aucune preuve pour l'instant", + "noEvidence": "Aucune preuve pour l’instant", + "healthy": "Sain", + "stale": "Preuve obsolète", + "failed": "Dernière exécution échouée", + "executionId": "Exécution", "columns": { "name": "Workflow", "status": "Statut", diff --git a/frontend/src/i18n/locales/nl-BE/integrations.json b/frontend/src/i18n/locales/nl-BE/integrations.json index 3c6da07..31b570d 100644 --- a/frontend/src/i18n/locales/nl-BE/integrations.json +++ b/frontend/src/i18n/locales/nl-BE/integrations.json @@ -37,6 +37,10 @@ "description": "{{known}} van {{expected}} canonieke n8n-workflows hebben actuele evidentie van werking.", "notBuilt": "Nog niet gebouwd", "noEvidence": "Nog geen evidentie", + "healthy": "Gezond", + "stale": "Evidentie verouderd", + "failed": "Laatste uitvoering mislukt", + "executionId": "Uitvoering", "columns": { "name": "Workflow", "status": "Status", diff --git a/frontend/src/pages/Automation.tsx b/frontend/src/pages/Automation.tsx index fce7738..b350716 100644 --- a/frontend/src/pages/Automation.tsx +++ b/frontend/src/pages/Automation.tsx @@ -279,9 +279,13 @@ export function Automation() { const displayName = slug ? t(`workflows.names.${slug}`) : w.name; const badge = !w.built ? { status: "not_configured", label: t("workflows.notBuilt") } - : w.last_seen_at - ? { status: "available", label: t("statusLabels.operational") } - : { status: "no_events", label: t("workflows.noEvidence") }; + : w.state === "healthy" + ? { status: "available", label: t("workflows.healthy") } + : w.state === "stale" + ? { status: "degraded", label: t("workflows.stale") } + : w.state === "failed" + ? { status: "failed", label: t("workflows.failed") } + : { status: "no_events", label: t("workflows.noEvidence") }; return ( {displayName} @@ -298,7 +302,7 @@ export function Automation() {
{t("common:actions.technicalDetails")} - {w.name} + {w.name}{w.last_execution_id ? ` · ${t("workflows.executionId")}: ${w.last_execution_id}` : ""}
diff --git a/n8n/workflows/MANIFEST.md b/n8n/workflows/MANIFEST.md index d558606..eaa1e27 100644 --- a/n8n/workflows/MANIFEST.md +++ b/n8n/workflows/MANIFEST.md @@ -17,8 +17,9 @@ credential values are never embedded; nodes reference named n8n credentials inst | Live workflow ID | `mobilityops-return-processing` | | Active status (as of 2026-08-04) | Active / Published | | Error Workflow | `Fleet Ops — Workflow Error Handler` (wired) | -| Timeouts / bounded retries | `Record follow-up` HTTP node: 15s timeout, retry on fail (3 tries, 1000ms wait) | -| Checksum (sha256) | `e5b6ba02a7824867620ceaf214224521d52de76337604b947f26f1b0b5432358` (updated 2026-08-05 — the committed file had invalid JSON, a missing `},` between two node objects; fixed, no live workflow change) | +| Timeouts / bounded retries | `Record follow-up` and heartbeat HTTP nodes: 15s timeout, retry on fail (3 tries, 1000ms wait) | +| Execution telemetry | Successful runs POST execution ID and status to `/api/v1/integrations/n8n/heartbeat`; failed runs are registered by the Error Workflow. | +| Checksum (sha256) | `17b8dd5b8e7ab5d0c0856a40f4199a401ad38f63895f611a096a80c9740f1e99` (heartbeat-enabled definition, 2026-08-10) | ## 2. Fleet Ops — Scheduled Data Quality Scan @@ -32,13 +33,14 @@ credential values are never embedded; nodes reference named n8n credentials inst | Live workflow ID | `mobilityops-scheduled-quality-scan` | | Active status (as of 2026-08-04) | Active / Published | | Error Workflow | `Fleet Ops — Workflow Error Handler` (wired) | -| Timeouts / bounded retries | `Run quality scan` HTTP node: 15s timeout, retry on fail (3 tries, 1000ms wait) | -| Checksum (sha256) | `c0d46e0519118e6336e35c4ea2a67edb2f14bd007909ccf9256c93733751244a` | +| Timeouts / bounded retries | Scan and heartbeat HTTP nodes: 15s timeout, retry on fail (3 tries, 1000ms wait) | +| Execution telemetry | Every successful scheduled/manual run posts an idempotent heartbeat with its n8n execution ID. | +| Checksum (sha256) | `f0bda8b0fa99d1a2403e970e00bd1f896cd4d10eaa626147fcca577a7ad7cb46` | ## 3. Fleet Ops — RAGcore Procedure Sync -Fully built and saved live (6 real nodes: Schedule Trigger → List procedures → Prepare -uploads → Upload to RAGcore → Summarize sync result → Report sync result to Fleet Ops). +Fully built with 7 real nodes: Schedule Trigger → List procedures → Prepare uploads → +Upload to RAGcore → Summarize sync result → Report sync result → Report heartbeat. **Published/active as of 2026-08-05**, once RAGcore itself went live (see `PROJECT_STATE.md`'s "RAGcore actually went live" entry): the workflow's `RAGcore Sync @@ -69,7 +71,8 @@ the same way via the same CLI import path, re-verified. | Live workflow ID | `6wbkc4d1AouGpmWT` | | Active status (as of 2026-08-05) | **Active / Published** | | Error Workflow | `Fleet Ops — Workflow Error Handler` (wired) | -| Checksum (sha256) | `643c0515a50fed3d35e28f0b8cb17841f6f8d91699944980b5c56e619d5bf2a6` | +| Execution telemetry | Successful syncs report both the bounded sync result and the canonical workflow heartbeat. | +| Checksum (sha256) | `8644d9faec4d17a43ba797e300b91fff976881c9b259e457fadd5a7cfaf39c6f` | ## 4. Fleet Ops — Workflow Error Handler @@ -89,7 +92,8 @@ Ops, which registers an audit event idempotently keyed on `execution_id`. | Live workflow ID | `Xppn2rAEqUuyiCJF` | | Active status (as of 2026-08-04) | Active / Published | | Error Workflow (on itself) | `- No Workflow -` (deliberately unset — prevents a recursive error loop) | -| Checksum (sha256) | `d9e2795b917a89a9b4a733e435661585f8011bc97f134d3643dee93ea05b0be6` | +| Execution telemetry | A successfully handled failure posts its own execution heartbeat after registering the target failure. | +| Checksum (sha256) | `1e50cf8b679b9e9f9b5cf4a1594bda459597ade40f44e8a68f540890f9c4b61d` | Validated this round: mock-data run (Error Trigger pinned to a realistic payload) produced a real `200 {"status":"registered", ...}` from the live Fleet Ops server; diff --git a/n8n/workflows/fleet-ops-data-quality-scan.json b/n8n/workflows/fleet-ops-data-quality-scan.json index 125f898..8e6b59f 100644 --- a/n8n/workflows/fleet-ops-data-quality-scan.json +++ b/n8n/workflows/fleet-ops-data-quality-scan.json @@ -64,6 +64,33 @@ "type": "n8n-nodes-base.code", "typeVersion": 2, "position": [780, 300] + }, + { + "parameters": { + "method": "POST", + "url": "http://192.168.10.150:1236/api/v1/integrations/n8n/heartbeat", + "authentication": "genericCredentialType", + "genericAuthType": "httpHeaderAuth", + "sendBody": true, + "contentType": "json", + "specifyBody": "keypair", + "bodyParameters": { "parameters": [ + { "name": "workflow_id", "value": "={{ $workflow.id }}" }, + { "name": "workflow_name", "value": "={{ $workflow.name }}" }, + { "name": "execution_id", "value": "={{ $execution.id }}" }, + { "name": "status", "value": "succeeded" } + ]}, + "options": { "timeout": 15000 } + }, + "id": "heartbeat-node", + "name": "Report workflow heartbeat", + "type": "n8n-nodes-base.httpRequest", + "typeVersion": 4.2, + "position": [1020, 300], + "credentials": { "httpHeaderAuth": { "name": "Fleet Ops Service Token" } }, + "retryOnFail": true, + "maxTries": 3, + "waitBetweenTries": 1000 } ], "connections": { @@ -75,6 +102,9 @@ }, "Run quality scan": { "main": [[{ "node": "Summarize result", "type": "main", "index": 0 }]] + }, + "Summarize result": { + "main": [[{ "node": "Report workflow heartbeat", "type": "main", "index": 0 }]] } }, "settings": { diff --git a/n8n/workflows/fleet-ops-error-handler.json b/n8n/workflows/fleet-ops-error-handler.json index d3bd41b..e5fb9b3 100644 --- a/n8n/workflows/fleet-ops-error-handler.json +++ b/n8n/workflows/fleet-ops-error-handler.json @@ -54,6 +54,33 @@ "credentials": { "httpHeaderAuth": { "name": "Fleet Ops Service Token" } } + }, + { + "parameters": { + "method": "POST", + "url": "http://192.168.10.150:1236/api/v1/integrations/n8n/heartbeat", + "authentication": "genericCredentialType", + "genericAuthType": "httpHeaderAuth", + "sendBody": true, + "contentType": "json", + "specifyBody": "keypair", + "bodyParameters": { "parameters": [ + { "name": "workflow_id", "value": "={{ $workflow.id }}" }, + { "name": "workflow_name", "value": "={{ $workflow.name }}" }, + { "name": "execution_id", "value": "={{ $execution.id }}" }, + { "name": "status", "value": "succeeded" } + ]}, + "options": { "timeout": 15000 } + }, + "id": "heartbeat-node", + "name": "Report workflow heartbeat", + "type": "n8n-nodes-base.httpRequest", + "typeVersion": 4.2, + "position": [1020, 300], + "credentials": { "httpHeaderAuth": { "name": "Fleet Ops Service Token" } }, + "retryOnFail": true, + "maxTries": 3, + "waitBetweenTries": 1000 } ], "connections": { @@ -62,6 +89,9 @@ }, "Build safe error report": { "main": [[{ "node": "Report failure to Fleet Ops", "type": "main", "index": 0 }]] + }, + "Report failure to Fleet Ops": { + "main": [[{ "node": "Report workflow heartbeat", "type": "main", "index": 0 }]] } }, "settings": { diff --git a/n8n/workflows/fleet-ops-ragcore-procedure-sync.json b/n8n/workflows/fleet-ops-ragcore-procedure-sync.json index 6fae455..85e1656 100644 --- a/n8n/workflows/fleet-ops-ragcore-procedure-sync.json +++ b/n8n/workflows/fleet-ops-ragcore-procedure-sync.json @@ -54,6 +54,33 @@ } } }, + { + "parameters": { + "method": "POST", + "url": "http://192.168.10.150:1236/api/v1/integrations/n8n/heartbeat", + "authentication": "genericCredentialType", + "genericAuthType": "httpHeaderAuth", + "sendBody": true, + "contentType": "json", + "specifyBody": "keypair", + "bodyParameters": { "parameters": [ + { "name": "workflow_id", "value": "={{ $workflow.id }}" }, + { "name": "workflow_name", "value": "={{ $workflow.name }}" }, + { "name": "execution_id", "value": "={{ $execution.id }}" }, + { "name": "status", "value": "succeeded" } + ]}, + "options": { "timeout": 15000 } + }, + "type": "n8n-nodes-base.httpRequest", + "typeVersion": 4.4, + "position": [1056, -144], + "id": "workflow-heartbeat-node", + "name": "Report workflow heartbeat", + "credentials": { "httpHeaderAuth": { "name": "Fleet Ops Service Token" } }, + "retryOnFail": true, + "maxTries": 3, + "waitBetweenTries": 1000 + }, { "parameters": { "jsCode": "const documents = $input.first().json.documents;\nconst results = [];\nfor (const doc of documents) {\n const binary = await this.helpers.prepareBinaryData(Buffer.from(doc.content, 'utf-8'), doc.document_id + '.md', 'text/markdown');\n results.push({json: {id: doc.id, language: doc.language, document_id: doc.document_id, title: doc.title, version: doc.version, content_hash: doc.content_hash}, binary: {file: binary}});\n}\nreturn results;" @@ -229,6 +256,9 @@ } ] ] + }, + "Report sync result to Fleet Ops": { + "main": [[{ "node": "Report workflow heartbeat", "type": "main", "index": 0 }]] } }, "settings": { diff --git a/n8n/workflows/fleet-ops-vehicle-return.json b/n8n/workflows/fleet-ops-vehicle-return.json index df737b6..ab3411d 100644 --- a/n8n/workflows/fleet-ops-vehicle-return.json +++ b/n8n/workflows/fleet-ops-vehicle-return.json @@ -69,6 +69,33 @@ "maxTries": 3, "waitBetweenTries": 1000 }, + { + "parameters": { + "method": "POST", + "url": "http://192.168.10.150:1236/api/v1/integrations/n8n/heartbeat", + "authentication": "genericCredentialType", + "genericAuthType": "httpHeaderAuth", + "sendBody": true, + "contentType": "json", + "specifyBody": "keypair", + "bodyParameters": { "parameters": [ + { "name": "workflow_id", "value": "={{ $workflow.id }}" }, + { "name": "workflow_name", "value": "={{ $workflow.name }}" }, + { "name": "execution_id", "value": "={{ $execution.id }}" }, + { "name": "status", "value": "succeeded" } + ]}, + "options": { "timeout": 15000 } + }, + "id": "heartbeat-node", + "name": "Report workflow heartbeat", + "type": "n8n-nodes-base.httpRequest", + "typeVersion": 4.2, + "position": [980, 300], + "credentials": { "httpHeaderAuth": { "name": "Fleet Ops Service Token" } }, + "retryOnFail": true, + "maxTries": 3, + "waitBetweenTries": 1000 + }, { "parameters": { "respondWith": "json", @@ -79,7 +106,7 @@ "name": "Return result", "type": "n8n-nodes-base.respondToWebhook", "typeVersion": 1.4, - "position": [1020, 300] + "position": [1220, 300] } ], "connections": { @@ -90,6 +117,9 @@ "main": [[{ "node": "Record follow-up", "type": "main", "index": 0 }]] }, "Record follow-up": { + "main": [[{ "node": "Report workflow heartbeat", "type": "main", "index": 0 }]] + }, + "Report workflow heartbeat": { "main": [[{ "node": "Return result", "type": "main", "index": 0 }]] } },