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

import hashlib
import json
import threading
import time
from datetime import UTC, datetime, timedelta
from uuid import uuid4

from pigway_frame_flow.drivers.base import DriverRegistry
from pigway_frame_flow.errors import ErrorCode as E
from pigway_frame_flow.errors import FlowError
from pigway_frame_flow.models import (
    DisplayRequest,
    FutureContent,
    ImageContent,
    PhotoSourceContent,
    TemplateContent,
    URLContent,
)
from pigway_frame_flow.providers.photo_sources import PhotoSources
from pigway_frame_flow.renderers.images import convert
from pigway_frame_flow.renderers.photo_caption import render_captioned_photo
from pigway_frame_flow.storage.database import encode, now_iso


class DisplayService:
    """One persistent queue; injected driver registry. No vendor imports."""

    def __init__(self, db, settings, registry: DriverRegistry, assets, renderer):
        self.photo_sources = PhotoSources(db, settings, assets)
        self.db, self.settings, self.registry, self.assets, self.renderer = (
            db,
            settings,
            registry,
            assets,
            renderer,
        )
        self.lock = threading.RLock()
        self.expiration_checked: dict[str, float] = {}
        # A killed write cannot be replayed safely; pre-write work can resume.
        db.execute(
            "UPDATE operations SET status='UNKNOWN', updated_at=? WHERE write_started=1 AND status IN ('UPLOADING','RENDERING')",
            (now_iso(),),
        )
        db.execute(
            "UPDATE operations SET status='QUEUED' WHERE write_started=0 AND status IN ('RENDERING','UPLOADING')"
        )

    def validate(self, request, writing=True):
        self.registry.get(request.device_id)
        if isinstance(request.content, FutureContent):
            raise FlowError(
                E.UNSUPPORTED_CAPABILITY, "Content type is unsupported", 422
            )
        if isinstance(request.content, TemplateContent):
            self.renderer.info(request.content.template)
        if isinstance(request.content, PhotoSourceContent):
            self.photo_sources.get(request.content.source_id)
        if writing and not self.settings.mock and not self.settings.write_enabled:
            raise FlowError(E.WRITE_DISABLED, "Real device writes are disabled", 403)

    def submit(self, action, payload, device_id, client_id, request_id, key=None):
        fingerprint = hashlib.sha256(
            encode({"action": action, "payload": payload, "device": device_id}).encode()
        ).hexdigest()
        with self.db.transaction() as tx:
            tx.execute("DELETE FROM idempotency WHERE expires_at<?", (now_iso(),))
            if key:
                existing = tx.execute(
                    "SELECT * FROM idempotency WHERE client_id=? AND key=?",
                    (client_id, key),
                ).fetchone()
                if existing:
                    if existing["fingerprint"] != fingerprint:
                        raise FlowError(
                            E.CONFLICT, "Idempotency key has a different request", 409
                        )
                    return existing["operation_id"]
            if (
                action not in ("render", "local-upload")
                and tx.execute(
                    "SELECT id FROM operations WHERE device_id=? AND status='UNKNOWN' AND write_started=1 LIMIT 1",
                    (device_id,),
                ).fetchone()
            ):
                raise FlowError(
                    E.WRITE_RESULT_UNKNOWN,
                    "Write result unknown; reconcile device state",
                    409,
                )
            id = uuid4().hex
            now = now_iso()
            tx.execute(
                "INSERT INTO operations(id,status,action,payload,device_id,client_id,request_id,created_at,updated_at) VALUES(?,?,?,?,?,?,?,?,?)",
                (
                    id,
                    "QUEUED",
                    action,
                    encode(self.settings.sanitize(payload)),
                    device_id,
                    client_id,
                    request_id,
                    now,
                    now,
                ),
            )
            if key:
                expires = (
                    datetime.now(UTC) + timedelta(hours=self.settings.idempotency_hours)
                ).isoformat()
                tx.execute(
                    "INSERT INTO idempotency VALUES(?,?,?,?,?)",
                    (client_id, key, fingerprint, id, expires),
                )
        return id

    def operation(self, id):
        op = self.db.one("SELECT * FROM operations WHERE id=?", (id,))
        if not op:
            raise FlowError(E.NOT_FOUND, "Operation not found", 404)
        return dict(
            operation_id=op["id"],
            status=op["status"],
            device_id=op["device_id"],
            created_at=op["created_at"],
            updated_at=op["updated_at"],
            result=json.loads(op["result"]) if op["result"] else None,
            error=json.loads(op["error"]) if op["error"] else None,
            request_id=op["request_id"],
            client_id=op["client_id"],
            action=op["action"],
        )

    def update(self, id, status, result=None, error=None):
        self.db.execute(
            "UPDATE operations SET status=?,updated_at=?,result=COALESCE(?,result),error=? WHERE id=?",
            (
                status,
                now_iso(),
                encode(result) if result else None,
                encode(error) if error else None,
                id,
            ),
        )

    def blocked(self, device_id):
        return (
            self.db.one(
                "SELECT id FROM operations WHERE device_id=? AND status='UNKNOWN' AND write_started=1 LIMIT 1",
                (device_id,),
            )
            is not None
        )

    def state(self, id):
        self.db.execute(
            "INSERT OR IGNORE INTO device_state(device_id) VALUES(?)", (id,)
        )
        return self.db.one("SELECT * FROM device_state WHERE device_id=?", (id,))

    def content_asset(self, content):
        if isinstance(content, TemplateContent):
            values = self.renderer.resolve(content.template, content.data)
            data = self.renderer.render(content.template, values, content.profile)
            asset = self.assets.put(data, content.template + ".jpg", "generated")
            source = {
                "type": "template",
                "template": content.template,
                "data": self.settings.sanitize(values),
                "profile": content.profile,
            }
            self.db.execute(
                "INSERT OR REPLACE INTO asset_sources VALUES(?,?)",
                (asset["asset_id"], encode(source)),
            )
            return asset
        if isinstance(content, PhotoSourceContent):
            data, name, caption = self.photo_sources.read(
                content.source_id, content.item_id
            )
            rendered = (
                render_captioned_photo(
                    data, caption, self.renderer.font, content.fit, content.profile
                )
                if caption
                else convert(data, content.fit, content.profile)
            )
            return self.assets.put(rendered, name)
        if isinstance(content, URLContent):
            data = self.assets.fetch_url(content.url)
        elif isinstance(content, ImageContent) and content.asset_id:
            return self.assets.info(content.asset_id)
        elif isinstance(content, ImageContent) and content.base64:
            data = self.assets.decode(content.base64)
        else:
            raise FlowError(E.INVALID_REQUEST, "Content requires renderable image data")
        return self.assets.put(
            convert(data, content.fit, content.profile), content.name
        )

    def current_content(self, device_id):
        device = self.registry.get(device_id).device()
        image_id = device.current_content.id
        row = self.db.one(
            "SELECT * FROM image_sources WHERE device_id=? AND image_id=?",
            (device_id, image_id),
        )
        source = json.loads(row["payload"]) if row else None
        # Acknowledged upload is not proof that the frame has refreshed yet.
        if row:
            op = self.operation(row["operation_id"])
            if op["status"] not in ("COMPLETED", "SYNCING"):
                source = None
        state = self.state(device_id)
        return {
            "device_id": device_id,
            "image_id": image_id,
            "synced": device.current_content.synced,
            "source_known": source is not None,
            "source": source,
            "asset_id": row["asset_id"] if row and source else None,
            "operation_id": row["operation_id"] if row and source else None,
            "preview_url": f"/api/v1/devices/{device_id}/images/{image_id}/file?thumbnail=true"
            if image_id
            else None,
            "previous_image_id": state["previous_image_id"],
            "overrides": self.db.all(
                "SELECT id,image_id,priority,expires_at FROM overrides WHERE device_id=? ORDER BY priority DESC,created_at DESC",
                (device_id,),
            ),
        }

    def top(self, device_id):
        return self.db.one(
            "SELECT * FROM overrides WHERE device_id=? AND expires_at>? ORDER BY priority DESC,created_at DESC LIMIT 1",
            (device_id, now_iso()),
        )

    def queued_operations(self):
        cursor_time, cursor_id = "", ""
        while True:
            rows = self.db.all(
                "SELECT * FROM operations WHERE status='QUEUED' AND "
                "(created_at>? OR (created_at=? AND id>?)) "
                "ORDER BY created_at,id LIMIT 50",
                (cursor_time, cursor_time, cursor_id),
            )
            if not rows:
                return
            yield from rows
            cursor_time, cursor_id = rows[-1]["created_at"], rows[-1]["id"]

    def process_one(self):
        with self.lock:
            for op in self.queued_operations():
                if op["action"] not in ("render", "local-upload") and self.blocked(
                    op["device_id"]
                ):
                    continue
                payload = json.loads(op["payload"])
                if op["action"] == "display":
                    priority = payload["options"]["priority"]
                    top = self.top(op["device_id"])
                    if top and priority < top["priority"]:
                        continue
                try:
                    self._execute(op, payload)
                except FlowError as exc:
                    uncertain = exc.code == E.WRITE_RESULT_UNKNOWN
                    error = {
                        "code": exc.code,
                        "message": exc.message,
                        "request_id": op["request_id"],
                        "details": {},
                    }
                    self.update(
                        op["id"], "UNKNOWN" if uncertain else "FAILED", error=error
                    )
                    self.db.history(op, exc.code)
                    self.db.event(
                        "operation.failed", {"operation_id": op["id"], "code": exc.code}
                    )
                except Exception:
                    wrote = self.db.one(
                        "SELECT write_started FROM operations WHERE id=?", (op["id"],)
                    )["write_started"]
                    self.update(
                        op["id"],
                        "UNKNOWN" if wrote else "FAILED",
                        error={
                            "code": E.WRITE_RESULT_UNKNOWN
                            if wrote
                            else E.INTERNAL_ERROR,
                            "message": "Write outcome unknown; reconcile device state"
                            if wrote
                            else "Operation failed",
                            "request_id": op["request_id"],
                            "details": {},
                        },
                    )
                    self.db.history(op, "UNKNOWN" if wrote else "FAILED")
                return True
        return False

    def _execute(self, op, payload):
        id, device_id, action = op["id"], op["device_id"], op["action"]
        driver = self.registry.get(device_id)
        if action in ("render", "local-upload"):
            self.update(id, "RENDERING")
            from pydantic import TypeAdapter

            from pigway_frame_flow.models import Content

            asset = self.content_asset(TypeAdapter(Content).validate_python(payload))
            if action == "local-upload":
                asset = self.assets.promote(asset["asset_id"])
            self.update(
                id, "COMPLETED", dict(asset_id=asset["asset_id"], url=asset["url"])
            )
            self.db.history(op, "COMPLETED")
            return
        device = driver.device(refresh=True)
        if device.status == "offline":
            raise FlowError(E.DEVICE_OFFLINE, "Target frame is offline", 409)
        # B101 layout is deliberately not transposed based on rotation.
        if (device.display.width, device.display.height) != (1200, 1600):
            raise FlowError(
                E.UNSUPPORTED_CAPABILITY, "Device geometry does not match renderer", 409
            )
        previous = device.current_content.id
        state = self.state(device_id)
        if action == "delete":
            image_id = payload["image_id"]
            protected = {state["previous_image_id"], state["base_image_id"]} | {
                x["image_id"]
                for x in self.db.all(
                    "SELECT image_id FROM overrides WHERE device_id=?", (device_id,)
                )
            }
            if image_id in protected:
                raise FlowError(E.CONFLICT, "Image is needed for restoration", 409)
            if image_id == previous and not payload["confirm_current"]:
                raise FlowError(E.CONFLICT, "Current image needs confirmation", 409)
            self._start_write(id)
            driver.delete(image_id)
            for suffix in [".original", "-thumb.jpg"]:
                (
                    self.settings.data_dir / "cache" / f"{device_id}-{image_id}{suffix}"
                ).unlink(missing_ok=True)
            self.db.execute(
                "DELETE FROM image_sources WHERE device_id=? AND image_id=?",
                (device_id, image_id),
            )
            driver.invalidate()
            self.update(id, "COMPLETED", {"image_id": image_id})
            self.db.history(op, "COMPLETED")
            return
        if action == "restore":
            target = (
                state["base_image_id"]
                if self.top(device_id)
                else state["previous_image_id"]
            )
            if target is None:
                raise FlowError(E.CONFLICT, "No previous image to restore", 409)
            self._start_write(id)
            driver.display(target)
            self.db.execute("DELETE FROM overrides WHERE device_id=?", (device_id,))
            self.db.execute(
                "UPDATE device_state SET base_image_id=NULL,managed_image_id=? WHERE device_id=?",
                (target, device_id),
            )
            driver.invalidate()
            self.update(id, "SYNCING", {"image_id": target})
            self.db.history(op, "ACCEPTED")
            return
        request = DisplayRequest.model_validate(payload)
        content = request.content
        self.update(id, "RENDERING")
        asset = None
        if isinstance(content, ImageContent) and (
            content.image_id or content.image_name
        ):
            image_id = content.image_id
            if content.image_name:
                matches = [x for x in driver.images() if x.name == content.image_name]
                if len(matches) != 1:
                    raise FlowError(
                        E.CONFLICT, "Name must identify exactly one gallery image", 409
                    )
                image_id = matches[0].image_id
        else:
            asset = self.content_asset(content)
            image_id = None
        self.db.execute(
            "UPDATE device_state SET previous_image_id=? WHERE device_id=?",
            (previous, device_id),
        )
        if request.options.restore_previous and not self.top(device_id):
            self.db.execute(
                "UPDATE device_state SET base_image_id=? WHERE device_id=?",
                (previous, device_id),
            )
        if request.options.restore_previous and previous is None:
            raise FlowError(
                E.CONFLICT, "Temporary override requires an existing image", 409
            )
        self._start_write(id)
        if asset:
            self.update(id, "UPLOADING")
            image_id = driver.upload(self.assets.path(asset["asset_id"]), asset["name"])
        elif isinstance(content, ImageContent) and content.image_name:
            driver.display_by_name(content.image_name)
        else:
            driver.display(image_id)
        result = {"image_id": image_id}
        if asset:
            self.assets.promote(asset["asset_id"])
            result["rendered_asset_id"] = asset["asset_id"]
            source = self.db.one(
                "SELECT payload FROM asset_sources WHERE asset_id=?",
                (asset["asset_id"],),
            )
            descriptor = source["payload"] if source else encode({"type": "image"})
            self.db.execute(
                "INSERT OR REPLACE INTO image_sources VALUES(?,?,?,?,?)",
                (device_id, image_id, asset["asset_id"], id, descriptor),
            )
        self.update(id, "SYNCING", result)
        driver.invalidate()
        if request.options.restore_previous:
            expires = (
                datetime.now(UTC) + timedelta(seconds=request.options.duration)
            ).isoformat()
            self.db.execute(
                "INSERT INTO overrides VALUES(?,?,?,?,?,?)",
                (id, device_id, image_id, request.options.priority, expires, now_iso()),
            )
        else:
            self.db.execute("DELETE FROM overrides WHERE device_id=?", (device_id,))
            self.db.execute(
                "UPDATE device_state SET base_image_id=NULL WHERE device_id=?",
                (device_id,),
            )
        self.db.execute(
            "UPDATE device_state SET managed_image_id=? WHERE device_id=?",
            (image_id, device_id),
        )
        self.db.history(op, "ACCEPTED")
        self.db.event(
            "display.changed",
            {"device_id": device_id, "image_id": image_id, "operation_id": id},
        )

    def _start_write(self, id):
        if not self.settings.mock and not self.settings.write_enabled:
            raise FlowError(E.WRITE_DISABLED, "Real device writes are disabled", 403)
        self.db.execute(
            "UPDATE operations SET write_started=1,status='UPLOADING',updated_at=? WHERE id=?",
            (now_iso(), id),
        )

    def sync(self):
        for op in self.db.all("SELECT * FROM operations WHERE status='SYNCING'"):
            try:
                device = self.registry.get(op["device_id"]).device()
                target = json.loads(op["result"])["image_id"]
                if (
                    device.current_content.id == target
                    and device.current_content.synced
                ):
                    self.update(op["id"], "COMPLETED")
                    self.db.history(op, "COMPLETED")
                    self.db.event("operation.completed", {"operation_id": op["id"]})
                elif (
                    datetime.now(UTC) - datetime.fromisoformat(op["updated_at"])
                ).total_seconds() > self.settings.sync_timeout:
                    self.update(
                        op["id"],
                        "UNKNOWN",
                        error={
                            "code": E.WRITE_RESULT_UNKNOWN,
                            "message": "Frame refresh was not confirmed",
                            "request_id": op["request_id"],
                            "details": {},
                        },
                    )
            except FlowError:
                pass

    def expire(self):
        if not self.settings.mock and not self.settings.write_enabled:
            return
        for device_id in self.registry.drivers:
            if self.blocked(device_id):
                continue
            expired = self.db.all(
                "SELECT * FROM overrides WHERE device_id=? AND expires_at<=?",
                (device_id, now_iso()),
            )
            if not expired:
                continue
            if not self.settings.mock:
                last = self.expiration_checked.get(device_id, 0)
                if time.monotonic() - last < 60:
                    continue
                self.expiration_checked[device_id] = time.monotonic()
            driver = self.registry.get(device_id)
            state = self.state(device_id)
            try:
                device = driver.device(refresh=True)
                if device.status == "offline":
                    continue
                top = self.top(device_id)
                target = top["image_id"] if top else state["base_image_id"]
                if target is None:
                    continue
                if device.current_content.id != state["managed_image_id"]:
                    # A user/app changed the artwork; do not overwrite it.
                    self.db.execute(
                        "DELETE FROM overrides WHERE device_id=?", (device_id,)
                    )
                    self.db.execute(
                        "UPDATE device_state SET base_image_id=NULL WHERE device_id=?",
                        (device_id,),
                    )
                    continue
                if target != device.current_content.id:
                    id = self.submit(
                        "auto-restore",
                        {"image_id": target},
                        device_id,
                        "scheduler",
                        "restore-" + uuid4().hex,
                    )
                    op = self.db.one("SELECT * FROM operations WHERE id=?", (id,))
                    self._start_write(id)
                    try:
                        driver.display(target)
                    except FlowError as exc:
                        self.update(
                            id,
                            "UNKNOWN"
                            if exc.code == E.WRITE_RESULT_UNKNOWN
                            else "FAILED",
                            error={
                                "code": exc.code,
                                "message": exc.message,
                                "request_id": op["request_id"],
                                "details": {},
                            },
                        )
                        continue
                    driver.invalidate()
                    self.update(id, "SYNCING", {"image_id": target})
                    self.db.history(op, "ACCEPTED")
                self.db.execute(
                    "DELETE FROM overrides WHERE device_id=? AND expires_at<=?",
                    (device_id, now_iso()),
                )
                self.db.execute(
                    "UPDATE device_state SET managed_image_id=?,base_image_id=? WHERE device_id=?",
                    (target, state["base_image_id"] if top else None, device_id),
                )
            except FlowError:
                continue

    def reconcile(self, id, acknowledge=False):
        with self.lock:
            op = self.operation(id)
            driver = self.registry.get(op["device_id"])
            device = driver.device(refresh=True)
            images = driver.images()
            expected = (op["result"] or {}).get("image_id")
            resolved = op["status"] != "UNKNOWN"
            if op["status"] == "UNKNOWN" and (
                acknowledge
                or (
                    expected
                    and expected == device.current_content.id
                    and device.current_content.synced
                )
            ):
                # Acknowledge releases the block; it never resends a write.
                self.update(
                    id,
                    "COMPLETED" if expected == device.current_content.id else "FAILED",
                    {"image_id": device.current_content.id},
                )
                resolved = True
                self.db.history(
                    dict(
                        id=id,
                        client_id="admin",
                        request_id=op["request_id"],
                        device_id=op["device_id"],
                        action="reconcile",
                    ),
                    "CONFIRMED"
                    if expected == device.current_content.id
                    else "ACKNOWLEDGED",
                )
            return dict(
                operation_id=id,
                current_image_id=device.current_content.id,
                expected_image_id=expected,
                cloud_images=images,
                resolved=resolved,
            )