pigway-frame-flow/pigway_frame_flow/display/recovery.py

"""Durable, read-only recovery progress for writes with uncertain outcomes."""

import json
from datetime import UTC, datetime
from uuid import uuid4

from pigway_frame_flow.errors import ErrorCode
from pigway_frame_flow.storage.database import encode, now_iso


class RecoveryMonitor:
    def __init__(self, service):
        self.service, self.db = service, service.db

    def observe(self, op, reason):
        timestamp = now_iso()
        age = (
            datetime.now(UTC) - datetime.fromisoformat(op["created_at"])
        ).total_seconds()
        review = age >= self.service.settings.sync_timeout + 300
        with self.db.transaction() as tx:
            previous = tx.execute(
                "SELECT * FROM operation_recovery WHERE operation_id=?", (op["id"],)
            ).fetchone()
            # Once escalated, wall-clock corrections must not clear the alert.
            review = review or bool(previous and previous["state"] == "manual_review")
            state = "manual_review" if review else "checking"
            tx.execute(
                "INSERT INTO operation_recovery VALUES(?,?,?,?,?,1) ON CONFLICT(operation_id) DO UPDATE SET state=excluded.state,reason=excluded.reason,last_checked_at=excluded.last_checked_at,checks=operation_recovery.checks+1",
                (op["id"], state, reason, timestamp, timestamp),
            )
            if review and (not previous or previous["state"] != "manual_review"):
                tx.execute(
                    "INSERT INTO events VALUES(?,?,?,?)",
                    (
                        uuid4().hex,
                        "operation.blocked",
                        encode(
                            {
                                "operation_id": op["id"],
                                "device_id": op["device_id"],
                                "reason": reason,
                                "requires_manual_review": True,
                            }
                        ),
                        timestamp,
                    ),
                )
                tx.execute(
                    "INSERT INTO history(client_id,request_id,operation_id,device_id,action,result,timestamp) VALUES(?,?,?,?,?,?,?)",
                    (
                        op["client_id"],
                        op["request_id"],
                        op["id"],
                        op["device_id"],
                        op["action"],
                        "MANUAL_REVIEW_REQUIRED",
                        timestamp,
                    ),
                )

    def resolve(self, operation_id):
        self.db.execute(
            "UPDATE operation_recovery SET state='resolved' WHERE operation_id=? AND state!='resolved'",
            (operation_id,),
        )

    def info(self, operation_id):
        row = self.db.one(
            "SELECT state,reason,first_checked_at,last_checked_at,checks FROM operation_recovery WHERE operation_id=?",
            (operation_id,),
        )
        if row:
            row["requires_manual_review"] = row["state"] == "manual_review"
        return row

    def observe_device(self, op, *, device=None, error=None):
        row = self.db.one("SELECT result FROM operations WHERE id=?", (op["id"],))
        result = json.loads(row["result"] or "{}")
        observation = {"checked_at": now_iso()}
        if device is not None:
            observation.update(
                image_id=device.current_content.id,
                synced=device.current_content.synced,
                online=device.status == "online",
                heartbeat=device.last_seen_at,
            )
        else:
            observation["read_error"] = {
                "code": error.code,
                **{
                    k: v
                    for k, v in error.details.items()
                    if k
                    in ("transport_error", "http_status", "attempts", "retry_exhausted")
                },
            }
        samples = result.get("sync_observations", [])
        if samples:
            old = samples[-1]
            changed = {k: v for k, v in old.items() if k != "checked_at"} != {
                k: v for k, v in observation.items() if k != "checked_at"
            }
            if (
                not changed
                and (
                    datetime.now(UTC) - datetime.fromisoformat(old["checked_at"])
                ).total_seconds()
                < 60
            ):
                return observation
        samples.append(observation)
        result["sync_observations"] = (
            [samples[0], *samples[-11:]] if len(samples) > 12 else samples
        )
        # Evidence must not reset updated_at: that is the sync timeout's clock.
        self.db.execute(
            "UPDATE operations SET result=? WHERE id=?", (encode(result), op["id"])
        )
        return observation

    def close_superseded_displays(self):
        for op in self.db.all(
            "SELECT * FROM operations WHERE action='display' AND write_started=1 "
            "AND status IN ('SYNCING','UNKNOWN')"
        ):
            result = json.loads(op["result"] or "{}")
            if not result.get("image_id") or json.loads(op["payload"]).get(
                "options", {}
            ).get("restore_previous"):
                continue
            newer = self.db.one(
                "SELECT id,result FROM operations WHERE device_id=? AND action='display' "
                "AND status='COMPLETED' AND write_started=1 AND created_at>? "
                "AND COALESCE(json_extract(payload,'$.options.restore_previous'),0)=0 "
                "AND json_extract(result,'$.image_id') IS NOT NULL "
                "ORDER BY created_at DESC LIMIT 1",
                (op["device_id"], op["created_at"]),
            )
            if (
                not newer
                or json.loads(newer["result"])["image_id"] == result["image_id"]
            ):
                continue
            # A later confirmed permanent display ends this attempt, but cannot
            # prove whether the earlier image was ever shown. Never replay it.
            self.service.update(
                op["id"],
                "FAILED",
                error={
                    "code": ErrorCode.DISPLAY_NOT_CONFIRMED,
                    "message": "Superseded by a later confirmed display; earlier display unconfirmed",
                    "request_id": op["request_id"],
                    "details": {
                        "superseded_by_operation_id": newer["id"],
                        "previous_error": json.loads(op["error"] or "null"),
                    },
                },
            )
            self.db.history(op, "DISPLAY_NOT_CONFIRMED")
            self.db.event(
                "operation.failed",
                {"operation_id": op["id"], "code": ErrorCode.DISPLAY_NOT_CONFIRMED},
            )