MobilityOps/backend/app/services/integration_status.py

from __future__ import annotations

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()

# 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
    try:
        response = httpx.get(f"{settings.mcp_hub_base_url.rstrip('/')}/health", timeout=1.5)
        return response.status_code == 200
    except httpx.HTTPError:
        return False