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.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 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 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 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:
            self.conn.execute("BEGIN IMMEDIATE")
            try:
                yield self.conn
                self.conn.commit()
            except Exception:
                self.conn.rollback()
                raise

    def execute(self, query, args=()):
        with self.lock, 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()