pigway-frame-flow/pigway_frame_flow/extensions/official.py

"""Explicit widget mutations, queued once with conservative uncertain-write handling."""

from fastapi import Depends, Header, Request
from pydantic import Field

from pigway_frame_flow.core.auth import require
from pigway_frame_flow.errors import ErrorCode as E
from pigway_frame_flow.errors import FlowError
from pigway_frame_flow.models import Accepted, StrictModel
from pigway_frame_flow.storage.database import encode


class WidgetSelection(StrictModel):
    cover_id: int | None = Field(default=None, gt=0)
    args: dict[str, str] = Field(default_factory=dict, max_length=30)


def install_official(app, service):
    import json

    for device_id, connection in service.registry.official.items():
        row = service.db.one(
            "SELECT payload FROM app_config WHERE key=?", ("mock-widget:" + device_id,)
        )
        if connection.reader.mock and row:
            connection.reader.mock_selection = json.loads(row["payload"])["cover_id"]

    def check(device_id, payload):
        service.registry.get(device_id)
        connection = service.registry.official.get(device_id)
        if not connection:
            raise FlowError(
                E.UNSUPPORTED_CAPABILITY, "This driver has no official widgets", 422
            )
        if not service.settings.mock and not service.settings.write_enabled:
            raise FlowError(E.WRITE_DISABLED, "Device writes are disabled", 403)
        view = connection.widgets()
        if not view.configured:
            raise FlowError(
                E.OFFICIAL_AUTH_REQUIRED,
                "Connect and associate an official account first",
                401,
            )
        if payload.cover_id is not None:
            widget = next((x for x in view.catalog if x.id == payload.cover_id), None)
            if not widget:
                raise FlowError(
                    E.INVALID_REQUEST, "Widget is not in the authenticated catalog"
                )
            if set(payload.args) - set(widget.arg_keys):
                raise FlowError(E.INVALID_REQUEST, "Unknown widget parameter")
            if any(len(k) > 120 or len(v) > 2000 for k, v in payload.args.items()):
                raise FlowError(E.INVALID_REQUEST, "Widget parameter is too long")
        elif payload.args:
            raise FlowError(
                E.INVALID_REQUEST, "Cancellation does not accept parameters"
            )
        return connection

    @app.post(
        "/api/v1/devices/{device_id}/widgets/selection",
        response_model=Accepted,
        status_code=202,
        tags=["Official"],
        dependencies=[Depends(require("admin"))],
    )
    def select(
        device_id: str,
        payload: WidgetSelection,
        request: Request,
        idempotency_key: str | None = Header(default=None),
    ):
        check(device_id, payload)
        id = service.submit(
            "official-widget",
            payload.model_dump(),
            device_id,
            request.state.client_id,
            request.state.request_id,
            idempotency_key,
        )
        return Accepted(operation_id=id, job_id=id, status="accepted")

    def execute(op, payload):
        selection = WidgetSelection.model_validate(payload)
        connection = check(op["device_id"], selection)
        service.update(op["id"], "UPLOADING")
        service.db.execute(
            "UPDATE operations SET write_started=1 WHERE id=?", (op["id"],)
        )
        with connection.lock:
            if connection.reader.mock:
                connection.reader.mock_selection = selection.cover_id
                service.db.execute(
                    "INSERT OR REPLACE INTO app_config VALUES(?,?)",
                    ("mock-widget:" + op["device_id"], encode(selection.model_dump())),
                )
            else:
                connection.reader.rpc.set_widget(
                    connection.reader._robot_id, selection.cover_id, selection.args
                )
            connection.reader._cached = None
            try:
                view = connection.widgets()
            except Exception:
                raise FlowError(
                    E.WRITE_RESULT_UNKNOWN,
                    "Widget write sent but read-only confirmation failed",
                    409,
                ) from None
        if (
            not view.current_verified
            or (view.current.id if view.current else None) != selection.cover_id
        ):
            raise FlowError(
                E.WRITE_RESULT_UNKNOWN,
                "Widget write acknowledged but selection is not confirmed",
                409,
            )
        service.update(op["id"], "COMPLETED", {"cover_id": selection.cover_id})
        service.db.history(op, "COMPLETED")
        service.db.event(
            "display.changed",
            {
                "device_id": op["device_id"],
                "operation_id": op["id"],
                "cover_id": selection.cover_id,
            },
        )

    service.operation_handlers["official-widget"] = execute

    def reconcile(op, acknowledge):
        id = op["operation_id"]
        row = service.db.one("SELECT payload FROM operations WHERE id=?", (id,))
        expected_cover = json.loads(row["payload"])["cover_id"]
        connection = service.registry.official[op["device_id"]]
        with connection.lock:
            connection.reader._cached = None
            view = connection.widgets()
        confirmed = (
            view.current_verified
            and (view.current.id if view.current else None) == expected_cover
        )
        resolved = op["status"] != "UNKNOWN"
        if op["status"] == "UNKNOWN" and (confirmed or acknowledge):
            service.update(
                id,
                "COMPLETED" if confirmed else "FAILED",
                {"cover_id": expected_cover} if confirmed else None,
            )
            service.db.history(
                dict(
                    id=id,
                    client_id="admin",
                    request_id=op["request_id"],
                    device_id=op["device_id"],
                    action="reconcile",
                ),
                "CONFIRMED" if confirmed else "ACKNOWLEDGED",
            )
            resolved = True
        return dict(
            operation_id=id,
            resolved=resolved,
            current_image_id=None,
            expected_image_id=None,
            cloud_images=[],
        )

    service.reconciliation_handlers["official-widget"] = reconcile