pigway-frame-flow/pigway_frame_flow/storage/database.py

import json
import sqlite3
import threading
from contextlib import contextmanager
from datetime import UTC, datetime
from uuid import uuid4


def now_iso():
    return datetime.now(UTC).isoformat()


def encode(value):
    return json.dumps(value, ensure_ascii=False, separators=(",", ":"))


class Database:
    def __init__(self, path):
        self.lock = threading.RLock()
        self._transaction_depth = 0
        self.conn = sqlite3.connect(path, check_same_thread=False)
        self.conn.row_factory = sqlite3.Row
        self.conn.executescript("""
        PRAGMA journal_mode=WAL;
        PRAGMA busy_timeout=5000;
        PRAGMA synchronous=FULL;
        CREATE TABLE IF NOT EXISTS managed_cloud_uploads (device_id TEXT, image_id INTEGER, asset_id TEXT, job_id TEXT, operation_id TEXT, uploaded_at TEXT, confirmed_at TEXT, cleanup_operation TEXT, PRIMARY KEY(device_id,image_id));
        CREATE TABLE IF NOT EXISTS photo_preferences (source_id TEXT NOT NULL, item_id TEXT NOT NULL, starred INTEGER NOT NULL DEFAULT 0, excluded INTEGER NOT NULL DEFAULT 0, PRIMARY KEY(source_id,item_id));
        CREATE TABLE IF NOT EXISTS asset_metadata (asset_id TEXT PRIMARY KEY, starred INTEGER NOT NULL DEFAULT 0, origin TEXT NOT NULL DEFAULT 'unknown', job_id TEXT, job_name TEXT);
        CREATE TABLE IF NOT EXISTS image_retention (device_id TEXT NOT NULL, image_id INTEGER NOT NULL, starred INTEGER NOT NULL DEFAULT 0, expires_at TEXT, upload_operation TEXT, cleanup_operation TEXT, PRIMARY KEY(device_id,image_id));
        CREATE TABLE IF NOT EXISTS photo_dimensions (path TEXT PRIMARY KEY, mtime INTEGER NOT NULL, bytes INTEGER NOT NULL, width INTEGER NOT NULL, height INTEGER NOT NULL);
        CREATE TABLE IF NOT EXISTS asset_sources (asset_id TEXT PRIMARY KEY, payload TEXT NOT NULL);
        CREATE TABLE IF NOT EXISTS image_sources (device_id TEXT, image_id INTEGER, asset_id TEXT, operation_id TEXT, payload TEXT NOT NULL, PRIMARY KEY(device_id,image_id));
        CREATE TABLE IF NOT EXISTS app_config (key TEXT PRIMARY KEY, payload TEXT NOT NULL);
        CREATE TABLE IF NOT EXISTS playlists (id TEXT PRIMARY KEY, payload TEXT, position INTEGER, next_run TEXT);
        CREATE TABLE IF NOT EXISTS operation_recovery (operation_id TEXT PRIMARY KEY, state TEXT NOT NULL, reason TEXT NOT NULL, first_checked_at TEXT NOT NULL, last_checked_at TEXT NOT NULL, checks INTEGER NOT NULL);
        CREATE TABLE IF NOT EXISTS operations (
            id TEXT PRIMARY KEY, status TEXT NOT NULL, action TEXT NOT NULL,
            payload TEXT NOT NULL, device_id TEXT NOT NULL, client_id TEXT NOT NULL,
            request_id TEXT NOT NULL, created_at TEXT NOT NULL, updated_at TEXT NOT NULL,
            result TEXT, error TEXT, write_started INTEGER DEFAULT 0);
        CREATE TABLE IF NOT EXISTS idempotency (
            client_id TEXT, key TEXT, fingerprint TEXT NOT NULL, operation_id TEXT NOT NULL,
            expires_at TEXT NOT NULL, PRIMARY KEY(client_id,key));
        CREATE TABLE IF NOT EXISTS jobs (id TEXT PRIMARY KEY, payload TEXT NOT NULL,
            enabled INTEGER NOT NULL, next_run TEXT, created_at TEXT, updated_at TEXT,
            last_run TEXT, last_result TEXT);
        CREATE TABLE IF NOT EXISTS job_runs (job_id TEXT, scheduled_at TEXT,
            operation_id TEXT, PRIMARY KEY(job_id,scheduled_at));
        CREATE TABLE IF NOT EXISTS hidden_history (operation_id TEXT PRIMARY KEY);
        CREATE TABLE IF NOT EXISTS history (id INTEGER PRIMARY KEY, client_id TEXT,
            request_id TEXT, operation_id TEXT, device_id TEXT, action TEXT,
            result TEXT, timestamp TEXT);
        CREATE TABLE IF NOT EXISTS device_state (device_id TEXT PRIMARY KEY,
            previous_image_id INTEGER, base_image_id INTEGER, managed_image_id INTEGER);
        CREATE TABLE IF NOT EXISTS overrides (id TEXT PRIMARY KEY, device_id TEXT,
            image_id INTEGER, priority INTEGER, expires_at TEXT, created_at TEXT);
        CREATE TABLE IF NOT EXISTS template_config (template_id TEXT PRIMARY KEY, payload TEXT);
        CREATE TABLE IF NOT EXISTS assets (id TEXT PRIMARY KEY, name TEXT, kind TEXT, created_at TEXT);
        CREATE TABLE IF NOT EXISTS clients (client_id TEXT PRIMARY KEY, token_hash TEXT,
            name TEXT, permissions TEXT, created_at TEXT, last_used_at TEXT, enabled INTEGER);
        CREATE TABLE IF NOT EXISTS events (id TEXT PRIMARY KEY, event_type TEXT,
            payload TEXT, created_at TEXT);
        CREATE TABLE IF NOT EXISTS webhook_subscriptions (id TEXT PRIMARY KEY,
            client_id TEXT, callback_url TEXT, event_types TEXT, enabled INTEGER,
            created_at TEXT);
        """)

    @contextmanager
    def transaction(self):
        with self.lock:
            nested = self._transaction_depth > 0
            savepoint = "tx_" + str(self._transaction_depth)
            self.conn.execute("SAVEPOINT " + savepoint if nested else "BEGIN IMMEDIATE")
            self._transaction_depth += 1
            try:
                yield self.conn
                if nested:
                    self.conn.execute("RELEASE SAVEPOINT " + savepoint)
                else:
                    self.conn.commit()
            except BaseException:
                if nested:
                    self.conn.execute("ROLLBACK TO SAVEPOINT " + savepoint)
                    self.conn.execute("RELEASE SAVEPOINT " + savepoint)
                else:
                    self.conn.rollback()
                raise
            finally:
                self._transaction_depth -= 1

    def execute(self, query, args=()):
        with self.lock:
            if self._transaction_depth:
                return self.conn.execute(query, args)
            with self.conn:
                return self.conn.execute(query, args)

    def one(self, query, args=()):
        with self.lock:
            row = self.conn.execute(query, args).fetchone()
            return dict(row) if row else None

    def all(self, query, args=()):
        with self.lock:
            return [dict(r) for r in self.conn.execute(query, args).fetchall()]

    def history(self, op, result):
        self.execute(
            "INSERT INTO history(client_id,request_id,operation_id,device_id,action,result,timestamp) VALUES(?,?,?,?,?,?,?)",
            (
                op["client_id"],
                op["request_id"],
                op.get("id"),
                op["device_id"],
                op["action"],
                result,
                now_iso(),
            ),
        )

    def event(self, event_type, payload):
        self.execute(
            "INSERT INTO events VALUES(?,?,?,?)",
            (uuid4().hex, event_type, encode(payload), now_iso()),
        )

    def close(self):
        self.conn.close()