MobilityOps/backend/app/services/integration_status.py

from __future__ import annotations

import threading
import time
from datetime import UTC, datetime, timedelta
from typing import Literal

import httpx
from sqlalchemy import func, select
from sqlalchemy.orm import Session

from app.core.config import get_settings
from app.models.audit import AuditEvent
from app.models.outbox import DEMO_SCENARIO_ERROR_CODE, OutboxEvent
from app.schemas import (
    McpHubIntegrationStatus,
    N8nErrorHandlerStatus,
    N8nIntegrationStatus,
    N8nWorkflowEvidence,
)

settings = get_settings()
_hub_health_lock = threading.Lock()
_hub_health_cached_at = 0.0
_hub_health_cached_value: bool | None = None

# The 4 canonical Fleet Ops n8n workflows (see n8n/workflows/MANIFEST.md). All 4 are
# built (all with their full node set saved).
_CANONICAL_WORKFLOWS = (
    "Fleet Ops — Vehicle Return Orchestration",
    "Fleet Ops — Scheduled Data Quality Scan",
    "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:
    counts: dict[str, int] = dict(
        db.execute(
            select(OutboxEvent.delivery_status, func.count()).group_by(OutboxEvent.delivery_status)
        ).all()  # type: ignore[arg-type]
    )
    pending = counts.get("pending", 0)
    delivering = counts.get("delivering", 0)
    failed = counts.get("failed", 0)
    succeeded = counts.get("succeeded", 0)

    # Prepared demo failures are props, not health signals. They stay visible and
    # counted -- hiding them would be its own kind of lie -- but they are counted
    # *separately*, and only genuinely unexpected failures are allowed to move n8n off
    # "operational". Without this split the demo seed's single staged failure pins the
    # integration to "degraded" forever, which tells a viewer something untrue about
    # the automation.
    demo_scenario_failed = (
        db.scalar(
            select(func.count())
            .select_from(OutboxEvent)
            .where(
                OutboxEvent.delivery_status == "failed",
                OutboxEvent.last_error_code == DEMO_SCENARIO_ERROR_CODE,
            )
        )
        or 0
    )
    unexpected_failed = max(failed - demo_scenario_failed, 0)

    latest_success_at = db.scalar(
        select(func.max(OutboxEvent.updated_at)).where(OutboxEvent.delivery_status == "succeeded")
    )
    # Health talks about real failures only, so the "latest failure" a health reader
    # sees must exclude the staged one too.
    latest_failure_at = db.scalar(
        select(func.max(OutboxEvent.updated_at)).where(
            OutboxEvent.delivery_status == "failed",
            OutboxEvent.last_error_code != DEMO_SCENARIO_ERROR_CODE,
        )
    )
    latest_demo_scenario_at = db.scalar(
        select(func.max(OutboxEvent.updated_at)).where(
            OutboxEvent.delivery_status == "failed",
            OutboxEvent.last_error_code == DEMO_SCENARIO_ERROR_CODE,
        )
    )

    state: Literal["disabled", "unavailable", "degraded", "operational", "no_evidence"]
    if not settings.n8n_dispatch_enabled:
        state = "disabled"
    elif unexpected_failed > 0 and succeeded == 0:
        state = "unavailable"
    elif unexpected_failed > 0:
        state = "degraded"
    elif succeeded > 0 or pending > 0 or delivering > 0:
        state = "operational"
    else:
        state = "no_evidence"

    # Scheduled scan evidence: only service-triggered runs count as n8n evidence, not
    # runs an operator triggered manually from the Data Quality page.
    latest_scan_at = db.scalar(
        select(func.max(AuditEvent.occurred_at)).where(
            AuditEvent.action == "data_quality_scan_run",
            AuditEvent.actor_type == "service",
        )
    )

    # RAGcore Procedure Sync evidence: result reports posted by the workflow itself once
    # it finishes uploading procedures to RAGcore (app/api/routers/integrations.py::
    # procedures_sync_result), the same "the workflow's own callback is the evidence"
    # pattern the scheduled scan and error handler already use below.
    latest_procedure_sync_at = db.scalar(
        select(func.max(AuditEvent.occurred_at)).where(AuditEvent.action == "n8n_procedures_synced")
    )

    # Error handler evidence: registrations posted by the "Fleet Ops — Workflow Error
    # Handler" n8n workflow itself, which also doubles as proof that workflow is wired
    # up and firing correctly.
    total_failures_registered = (
        db.scalar(
            select(func.count(AuditEvent.id)).where(
                AuditEvent.action == "n8n_workflow_failure_registered"
            )
        )
        or 0
    )
    latest_failure_row = db.execute(
        select(AuditEvent.occurred_at, AuditEvent.after_json)
        .where(AuditEvent.action == "n8n_workflow_failure_registered")
        .order_by(AuditEvent.occurred_at.desc())
        .limit(1)
    ).first()
    latest_handler_failure_at = latest_failure_row[0] if latest_failure_row else None
    latest_handler_failure_workflow = (
        (latest_failure_row[1] or {}).get("workflow_name") if latest_failure_row else None
    )

    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,
    }
    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
        )
        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),
        dispatch_enabled=settings.n8n_dispatch_enabled,
        state=state,
        pending=pending,
        delivering=delivering,
        failed=failed,
        unexpected_failed=unexpected_failed,
        demo_scenario_failed=demo_scenario_failed,
        succeeded=succeeded,
        latest_success_at=latest_success_at,
        latest_failure_at=latest_failure_at,
        latest_demo_scenario_at=latest_demo_scenario_at,
        expected_workflow_count=len(_CANONICAL_WORKFLOWS),
        known_workflow_count=sum(1 for w in workflows if w.last_seen_at is not None),
        workflows=workflows,
        error_handler=N8nErrorHandlerStatus(
            total_failures_registered=total_failures_registered,
            latest_failure_at=latest_handler_failure_at,
            latest_failure_workflow=latest_handler_failure_workflow,
        ),
    )


def derive_mcp_hub_status(db: Session) -> McpHubIntegrationStatus:
    """Evidence-based MCP Hub status: real tool-call audit history, not just the
    `MCP_HUB_REGISTRATION_ENABLED` flag flipped on. Every `mcp_tool_request` call
    already writes an `AuditEvent` (see `app/api/routers/mcp_integrations.py`)."""
    total_calls = (
        db.scalar(select(func.count(AuditEvent.id)).where(AuditEvent.action == "mcp_tool_request"))
        or 0
    )
    latest_call_row = db.execute(
        select(AuditEvent.occurred_at, AuditEvent.actor_label, AuditEvent.metadata_json)
        .where(AuditEvent.action == "mcp_tool_request")
        .order_by(AuditEvent.occurred_at.desc())
        .limit(1)
    ).first()
    last_called_at = latest_call_row[0] if latest_call_row else None
    last_client = latest_call_row[1] if latest_call_row else None
    last_tool = (latest_call_row[2] or {}).get("tool") if latest_call_row else None

    state: Literal["not_configured", "no_evidence", "operational"]
    if not settings.mcp_hub_registration_enabled:
        state = "not_configured"
    elif total_calls > 0:
        state = "operational"
    else:
        state = "no_evidence"

    hub_reachable = _check_hub_reachable()

    return McpHubIntegrationStatus(
        registration_enabled=settings.mcp_hub_registration_enabled,
        state=state,
        total_calls=total_calls,
        last_tool=last_tool,
        last_client=last_client,
        last_called_at=last_called_at,
        hub_reachable=hub_reachable,
    )


def _check_hub_reachable() -> bool | None:
    """Real Hub-side health signal (MCP Hub's own registration is catalog-driven on
    its side, so this is the only thing Fleet Ops itself can honestly check).
    `None` means not configured / not checked, never a guess."""
    if not settings.mcp_hub_base_url:
        return None
    global _hub_health_cached_at, _hub_health_cached_value
    now = time.monotonic()
    with _hub_health_lock:
        if now - _hub_health_cached_at < settings.mcp_hub_health_cache_seconds:
            return _hub_health_cached_value
        try:
            response = httpx.get(f"{settings.mcp_hub_base_url.rstrip('/')}/health", timeout=1.5)
            _hub_health_cached_value = response.status_code == 200
        except httpx.HTTPError:
            _hub_health_cached_value = False
        _hub_health_cached_at = time.monotonic()
        return _hub_health_cached_value