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 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 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 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()