Designing fallback routing for disconnected field devices

from __future__ import annotations

import hashlib
import json
import logging
import sqlite3
from contextlib import contextmanager
from datetime import datetime, timezone
from typing import Any, Iterator, Literal

from pydantic import BaseModel, Field, field_validator

logger = logging.getLogger("construction.field_fallback")

DocType = Literal["change_order", "rfi", "daily_log", "submittal", "safety_incident"]
Discipline = Literal["ARCH", "STR", "MEP", "CIV", "ELEC", "PLMB"]
SchemaVersion = Literal["1.0", "1.1", "2.0"]
RoutingAction = Literal["auto_route", "human_review", "quarantine"]

# WBS elements follow PROJ-NNN-DIV-NN; MasterFormat divisions are XX XX XX.
_WBS = r"^[A-Z]{2,5}-\d{3}-\d{2}-\d{2}$"
_MASTERFORMAT = r"^\d{2} \d{2} \d{2}$"

# A payload below this confidence is never auto-posted on reconnect.
AUTO_ROUTE = 0.92
REVIEW_FLOOR = 0.75
# Beyond this skew the device clock is untrustworthy; force review.
MAX_TRUSTED_SKEW_SECONDS = 120


class FieldRecord(BaseModel):
    """A field-captured payload staged for routing once connectivity returns."""

    model_config = {"frozen": True, "extra": "forbid"}

    device_id: str = Field(min_length=3, max_length=48)
    doc_type: DocType
    discipline: Discipline
    wbs_element: str = Field(pattern=_WBS)
    masterformat_division: str = Field(pattern=_MASTERFORMAT)
    budget_impact_usd: float = Field(ge=0.0)
    schema_version: SchemaVersion
    captured_at: datetime
    clock_skew_seconds: float = Field(ge=0.0, default=0.0)
    body: dict[str, Any] = Field(default_factory=dict)

    @field_validator("captured_at")
    @classmethod
    def _require_utc(cls, v: datetime) -> datetime:
        # A naive timestamp from an offline device routes to the wrong day.
        if v.tzinfo is None:
            raise ValueError("captured_at must be timezone-aware UTC")
        return v.astimezone(timezone.utc)

    @property
    def idempotency_key(self) -> str:
        # Deterministic: flapping replays collapse to one server-side record.
        seed = f"{self.device_id}:{self.captured_at.isoformat()}:{self.doc_type}"
        return hashlib.sha256(seed.encode()).hexdigest()

    def routing(self, current_schema: SchemaVersion) -> tuple[RoutingAction, float, str]:
        """Score the record against the canonical confidence bands."""
        confidence = 1.0
        notes: list[str] = []
        if self.clock_skew_seconds > MAX_TRUSTED_SKEW_SECONDS:
            confidence -= 0.30
            notes.append(f"clock skew {self.clock_skew_seconds:.0f}s")
        if self.schema_version != current_schema:
            confidence -= 0.20
            notes.append(f"schema {self.schema_version} != {current_schema}")
        detail = "; ".join(notes) or "clean"

        # Safety incidents are escalated to a human regardless of score.
        if self.doc_type == "safety_incident":
            return "quarantine", confidence, "safety_incident: human review required"
        if confidence >= AUTO_ROUTE:
            return "auto_route", confidence, detail
        if confidence >= REVIEW_FLOOR:
            return "human_review", confidence, detail
        return "quarantine", confidence, detail


class FieldQueue:
    """Durable, replay-safe on-device queue backed by WAL-mode SQLite."""

    def __init__(self, db_path: str) -> None:
        self.db_path = db_path
        with self._conn() as c:
            c.execute("PRAGMA journal_mode=WAL")  # survive crashes mid-write
            c.execute(
                """CREATE TABLE IF NOT EXISTS queue (
                    key TEXT PRIMARY KEY,
                    priority INTEGER NOT NULL,
                    captured_at TEXT NOT NULL,
                    record TEXT NOT NULL,
                    synced INTEGER NOT NULL DEFAULT 0
                )"""
            )

    @contextmanager
    def _conn(self) -> Iterator[sqlite3.Connection]:
        conn = sqlite3.connect(self.db_path)
        try:
            yield conn
            conn.commit()
        except Exception:
            conn.rollback()
            raise
        finally:
            conn.close()

    def enqueue(self, record: FieldRecord) -> str:
        # Safety incidents (0) preempt change orders (1) preempt everything else (2).
        priority = 0 if record.doc_type == "safety_incident" else (
            1 if record.doc_type == "change_order" and record.budget_impact_usd > 50_000 else 2
        )
        with self._conn() as c:
            c.execute(
                "INSERT OR IGNORE INTO queue (key, priority, captured_at, record) "
                "VALUES (?, ?, ?, ?)",
                (record.idempotency_key, priority,
                 record.captured_at.isoformat(), record.model_dump_json()),
            )
        logger.info("queued %s key=%s", record.doc_type, record.idempotency_key[:12])
        return record.idempotency_key

    def drain(self, current_schema: SchemaVersion) -> list[dict[str, Any]]:
        """On reconnect: route pending records by priority, then capture order."""
        results: list[dict[str, Any]] = []
        with self._conn() as c:
            rows = c.execute(
                "SELECT key, record FROM queue WHERE synced = 0 "
                "ORDER BY priority ASC, captured_at ASC"
            ).fetchall()
            for key, raw in rows:
                record = FieldRecord.model_validate_json(raw)
                action, confidence, detail = record.routing(current_schema)
                if action == "auto_route":
                    # Real dispatch goes here; the key is the server dedupe token.
                    c.execute("UPDATE queue SET synced = 1 WHERE key = ?", (key,))
                    logger.info("auto_route %s conf=%.2f", key[:12], confidence)
                else:
                    logger.warning("%s %s conf=%.2f (%s)", action, key[:12], confidence, detail)
                results.append({"key": key, "action": action,
                                "confidence": round(confidence, 2), "detail": detail})
        return results