diff --git a/backend/src/media_library_viewer_api/services/settings_store.py b/backend/src/media_library_viewer_api/services/settings_store.py index 8b5186e..e1ba9d9 100644 --- a/backend/src/media_library_viewer_api/services/settings_store.py +++ b/backend/src/media_library_viewer_api/services/settings_store.py @@ -9,13 +9,14 @@ from __future__ import annotations import json import sqlite3 -from io import StringIO import time import uuid +from io import StringIO from pathlib import Path from typing import Any import paramiko + from media_library_viewer_api.config import get_settings DEFAULT_SETTINGS_PATH = Path(".cache/media_library_viewer/settings.sqlite") @@ -911,6 +912,299 @@ class SettingsStore: with self.connect() as conn: conn.execute("DELETE FROM dashboard_shortcuts WHERE id = ?", (shortcut_id,)) + # ------------------------------------------------------------------ + # Backup jobs + # ------------------------------------------------------------------ + + def _row_to_job(self, row: sqlite3.Row) -> dict[str, Any]: + return { + "id": row["id"], + "name": row["name"], + "source": row["source"], + "target": row["target"], + "schedule_interval_seconds": row["schedule_interval_seconds"], + "created_at": row["created_at"], + } + + def _normalize_backup_job_payload( + self, payload: dict[str, Any], job_id: str | None = None + ) -> dict[str, Any]: + current = self.get_backup_job(job_id) if job_id else None + job_id = str(payload.get("id") or job_id or uuid.uuid4().hex[:12]).strip() or uuid.uuid4().hex[:12] + name = str(payload.get("name") or (current or {}).get("name") or job_id).strip() or job_id + source = str(payload.get("source") if payload.get("source") is not None else (current or {}).get("source", "") or "").strip() + target = str(payload.get("target") if payload.get("target") is not None else (current or {}).get("target", "") or "").strip() + schedule_interval_seconds = payload.get("schedule_interval_seconds") + if schedule_interval_seconds is None: + schedule_interval_seconds = (current or {}).get("schedule_interval_seconds") + if schedule_interval_seconds is not None: + schedule_interval_seconds = int(schedule_interval_seconds) + return { + "id": job_id, + "name": name, + "source": source, + "target": target, + "schedule_interval_seconds": schedule_interval_seconds, + } + + def get_backup_job_by_name(self, name: str) -> dict[str, Any] | None: + self.init_schema() + with self.connect() as conn: + row = conn.execute("SELECT * FROM backup_jobs WHERE name = ?", (name,)).fetchone() + return self._row_to_job(row) if row else None + + def upsert_backup_job(self, payload: dict[str, Any]) -> dict[str, Any]: + self.init_schema() + job = self._normalize_backup_job_payload(payload) + now = int(time.time()) + with self.connect() as conn: + existing = conn.execute("SELECT created_at FROM backup_jobs WHERE id = ?", (job["id"],)).fetchone() + created_at = int(existing[0]) if existing else now + conn.execute( + """ + INSERT INTO backup_jobs (id, name, source, target, schedule_interval_seconds, created_at) + VALUES (?, ?, ?, ?, ?, ?) + ON CONFLICT(id) DO UPDATE SET + name = excluded.name, + source = excluded.source, + target = excluded.target, + schedule_interval_seconds = excluded.schedule_interval_seconds + ON CONFLICT(name) DO UPDATE SET + source = excluded.source, + target = excluded.target, + schedule_interval_seconds = excluded.schedule_interval_seconds + """, + ( + job["id"], + job["name"], + job["source"], + job["target"], + job["schedule_interval_seconds"], + created_at, + ), + ) + return self.get_backup_job(job["id"]) or job + + def get_backup_job(self, job_id: str | None) -> dict[str, Any] | None: + if not job_id: + return None + self.init_schema() + with self.connect() as conn: + row = conn.execute("SELECT * FROM backup_jobs WHERE id = ?", (job_id,)).fetchone() + return self._row_to_job(row) if row else None + + def list_backup_jobs(self) -> list[dict[str, Any]]: + self.init_schema() + with self.connect() as conn: + rows = conn.execute("SELECT * FROM backup_jobs ORDER BY created_at DESC").fetchall() + return [self._row_to_job(row) for row in rows] + + # ------------------------------------------------------------------ + # Backup runs + # ------------------------------------------------------------------ + + def _row_to_run(self, row: sqlite3.Row) -> dict[str, Any]: + details = json.loads(row["details_json"]) if row["details_json"] else {} + return { + "id": row["id"], + "job_id": row["job_id"], + "started_at": row["started_at"], + "ended_at": row["ended_at"], + "status": row["status"], + "bytes_transferred": row["bytes_transferred"], + "duration_ms": row["duration_ms"], + "error_message": row["error_message"], + "details": details, + "created_at": row["created_at"], + } + + def create_backup_run(self, payload: dict[str, Any]) -> dict[str, Any]: + self.init_schema() + run_id = str(payload.get("id") or uuid.uuid4().hex[:12]).strip() or uuid.uuid4().hex[:12] + now = int(time.time()) + job_id = str(payload["job_id"]) + started_at = int(payload["started_at"]) + ended_at = payload.get("ended_at") + if ended_at is not None: + ended_at = int(ended_at) + status = str(payload.get("status", "pending")) + bytes_transferred = payload.get("bytes_transferred") + if bytes_transferred is not None: + bytes_transferred = int(bytes_transferred) + duration_ms = payload.get("duration_ms") + if duration_ms is not None: + duration_ms = int(duration_ms) + error_message = str(payload.get("error_message") or "") + details = payload.get("details", {}) + details_json = json.dumps(details) if details else None + with self.connect() as conn: + conn.execute( + """ + INSERT INTO backup_runs (id, job_id, started_at, ended_at, status, bytes_transferred, duration_ms, error_message, details_json, created_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + (run_id, job_id, started_at, ended_at, status, bytes_transferred, duration_ms, error_message, details_json, now), + ) + return self.get_backup_run(run_id) or { + "id": run_id, + "job_id": job_id, + "started_at": started_at, + "ended_at": ended_at, + "status": status, + "bytes_transferred": bytes_transferred, + "duration_ms": duration_ms, + "error_message": error_message, + "details": details, + "created_at": now, + } + + def get_backup_run(self, run_id: str | None) -> dict[str, Any] | None: + if not run_id: + return None + self.init_schema() + with self.connect() as conn: + row = conn.execute("SELECT * FROM backup_runs WHERE id = ?", (run_id,)).fetchone() + return self._row_to_run(row) if row else None + + def list_backup_runs( + self, + job_id: str | None = None, + status: str | None = None, + limit: int = 50, + ) -> list[dict[str, Any]]: + self.init_schema() + clauses: list[str] = [] + params: list[Any] = [] + if job_id: + clauses.append("job_id = ?") + params.append(job_id) + if status: + clauses.append("status = ?") + params.append(status) + where = f"WHERE {' AND '.join(clauses)}" if clauses else "" + sql = f"SELECT * FROM backup_runs {where} ORDER BY created_at DESC LIMIT ?" + params.append(max(1, min(int(limit), 200))) + with self.connect() as conn: + rows = conn.execute(sql, params).fetchall() + return [self._row_to_run(row) for row in rows] + + def get_latest_backup_run(self, job_id: str) -> dict[str, Any] | None: + self.init_schema() + with self.connect() as conn: + row = conn.execute( + "SELECT * FROM backup_runs WHERE job_id = ? ORDER BY created_at DESC LIMIT 1", + (job_id,), + ).fetchone() + return self._row_to_run(row) if row else None + + # ------------------------------------------------------------------ + # Backup alerts + # ------------------------------------------------------------------ + + def _row_to_alert(self, row: sqlite3.Row) -> dict[str, Any]: + return { + "id": row["id"], + "job_id": row["job_id"], + "run_id": row["run_id"], + "alert_type": row["alert_type"], + "severity": row["severity"], + "message": row["message"], + "acknowledged": bool(row["acknowledged"]), + "resolved_at": row["resolved_at"], + "created_at": row["created_at"], + } + + def create_backup_alert(self, payload: dict[str, Any]) -> dict[str, Any]: + self.init_schema() + alert_id = str(payload.get("id") or uuid.uuid4().hex[:12]).strip() or uuid.uuid4().hex[:12] + now = int(time.time()) + job_id = str(payload["job_id"]) + run_id = payload.get("run_id") + if run_id is not None: + run_id = str(run_id) + alert_type = str(payload.get("alert_type", "generic")) + severity = str(payload.get("severity", "warning")) + message = str(payload.get("message", "")) + with self.connect() as conn: + conn.execute( + """ + INSERT INTO backup_alerts (id, job_id, run_id, alert_type, severity, message, acknowledged, resolved_at, created_at) + VALUES (?, ?, ?, ?, ?, ?, 0, NULL, ?) + """, + (alert_id, job_id, run_id, alert_type, severity, message, now), + ) + return self.get_backup_alert(alert_id) or { + "id": alert_id, + "job_id": job_id, + "run_id": run_id, + "alert_type": alert_type, + "severity": severity, + "message": message, + "acknowledged": False, + "resolved_at": None, + "created_at": now, + } + + def get_backup_alert(self, alert_id: str | None) -> dict[str, Any] | None: + if not alert_id: + return None + self.init_schema() + with self.connect() as conn: + row = conn.execute("SELECT * FROM backup_alerts WHERE id = ?", (alert_id,)).fetchone() + return self._row_to_alert(row) if row else None + + def list_backup_alerts( + self, + job_id: str | None = None, + acknowledged: bool | None = None, + severity: str | None = None, + ) -> list[dict[str, Any]]: + self.init_schema() + clauses: list[str] = [] + params: list[Any] = [] + if job_id: + clauses.append("job_id = ?") + params.append(job_id) + if acknowledged is not None: + clauses.append("acknowledged = ?") + params.append(1 if acknowledged else 0) + if severity: + clauses.append("severity = ?") + params.append(severity) + where = f"WHERE {' AND '.join(clauses)}" if clauses else "" + sql = f"SELECT * FROM backup_alerts {where} ORDER BY created_at DESC" + with self.connect() as conn: + rows = conn.execute(sql, params).fetchall() + return [self._row_to_alert(row) for row in rows] + + def acknowledge_backup_alert(self, alert_id: str) -> dict[str, Any] | None: + self.init_schema() + with self.connect() as conn: + conn.execute( + "UPDATE backup_alerts SET acknowledged = 1 WHERE id = ?", + (alert_id,), + ) + return self.get_backup_alert(alert_id) + + def resolve_backup_alerts_for_job(self, job_id: str, alert_type: str) -> int: + self.init_schema() + now = int(time.time()) + with self.connect() as conn: + cur = conn.execute( + "UPDATE backup_alerts SET resolved_at = ? WHERE job_id = ? AND alert_type = ? AND resolved_at IS NULL", + (now, job_id, alert_type), + ) + return int(cur.rowcount or 0) + + def prune_backup_alerts(self, cutoff_ts: int) -> int: + self.init_schema() + with self.connect() as conn: + cur = conn.execute( + "DELETE FROM backup_alerts WHERE created_at < ?", + (int(cutoff_ts),), + ) + return int(cur.rowcount or 0) + _store: SettingsStore | None = None diff --git a/backend/tests/test_backups.py b/backend/tests/test_backups.py index bfdae86..6c71afe 100644 --- a/backend/tests/test_backups.py +++ b/backend/tests/test_backups.py @@ -29,3 +29,29 @@ def test_backup_report_model(): ) assert report.name == "test-backup" assert report.status == "success" + + +def test_create_job_and_run(): + with tempfile.TemporaryDirectory() as tmpdir: + db_path = Path(tmpdir) / "test_settings.sqlite" + store = SettingsStore(db_path) + store.init_schema() + + job = store.upsert_backup_job({ + "name": "test-job", + "source": "server-a:/data", + "target": "server-b:/backups", + "schedule_interval_seconds": 86400, + }) + assert job["name"] == "test-job" + + run = store.create_backup_run({ + "job_id": job["id"], + "started_at": 1715392800, + "status": "success", + "ended_at": 1715393700, + "duration_ms": 900000, + "bytes_transferred": 1024, + }) + assert run["job_id"] == job["id"] + assert run["status"] == "success"