Add database backup and restore

scripts/backup-database.py

@@ -0,0 +1,63 @@
+#!/usr/bin/env python3
+"""Create a consistent SQLite backup from the running Docker service."""
+
+import argparse
+import subprocess
+import sys
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+from uuid import uuid4
+
+CONTAINER_BACKUP = """import sqlite3
+source = sqlite3.connect('/data/eternity.db')
+destination = sqlite3.connect('/tmp/eternity-backup.db')
+with destination:
+    source.backup(destination)
+destination.close()
+source.close()
+"""
+
+
+def run(command, **kwargs):
+    return subprocess.run(command, check=True, text=True, **kwargs)
+
+
+def main():
+    parser = argparse.ArgumentParser(description="Back up the live Eternity Project SQLite database.")
+    parser.add_argument("--output-dir", default="backups", help="Directory for timestamped backup files")
+    parser.add_argument("--keep-days", type=int, default=30, help="Delete backups older than this many days; use 0 to keep all")
+    args = parser.parse_args()
+
+    output_dir = Path(args.output_dir).resolve()
+    output_dir.mkdir(parents=True, exist_ok=True)
+    timestamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
+    destination = output_dir / f"eternity-{timestamp}.db"
+    temporary_path = f"/tmp/eternity-backup-{uuid4().hex}.db"
+
+    try:
+        container_id = run(["docker", "compose", "ps", "-q", "eternityproject"], capture_output=True).stdout.strip()
+        if not container_id:
+            raise RuntimeError("The eternityproject Docker service must be running.")
+        run(["docker", "exec", container_id, "python", "-c", CONTAINER_BACKUP.replace("/tmp/eternity-backup.db", temporary_path)])
+        run(["docker", "cp", f"{container_id}:{temporary_path}", str(destination)])
+    except (OSError, subprocess.CalledProcessError, RuntimeError) as error:
+        print(f"DATABASE_BACKUP_FAILED: {error}", file=sys.stderr)
+        return 1
+    finally:
+        if "container_id" in locals() and container_id:
+            subprocess.run(["docker", "exec", container_id, "rm", "-f", temporary_path], check=False)
+
+    print(f"DATABASE_BACKUP_CREATED={destination}")
+    if args.keep_days > 0:
+        cutoff = datetime.now(timezone.utc) - timedelta(days=args.keep_days)
+        removed = 0
+        for backup in output_dir.glob("eternity-*.db"):
+            if backup != destination and datetime.fromtimestamp(backup.stat().st_mtime, timezone.utc) < cutoff:
+                backup.unlink()
+                removed += 1
+        print(f"DATABASE_BACKUP_RETENTION: removed={removed} keep_days={args.keep_days}")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/verify-backup.py

@@ -0,0 +1,42 @@
+#!/usr/bin/env python3
+"""Restore a backup to an isolated temporary database and verify its integrity."""
+
+import argparse
+import shutil
+import sqlite3
+import sys
+import tempfile
+from pathlib import Path
+
+
+def main():
+    parser = argparse.ArgumentParser(description="Test an Eternity Project database backup by restoring it safely.")
+    parser.add_argument("backup", type=Path, help="Path to a .db backup created by backup-database.py")
+    args = parser.parse_args()
+    source = args.backup.resolve()
+    if not source.is_file():
+        print(f"BACKUP_VERIFY_FAILED: Backup does not exist: {source}", file=sys.stderr)
+        return 2
+
+    with tempfile.TemporaryDirectory(prefix="eternity-restore-") as directory:
+        restored = Path(directory) / "eternity-restored.db"
+        shutil.copy2(source, restored)
+        try:
+            database = sqlite3.connect(f"file:{restored}?mode=ro", uri=True)
+            integrity = database.execute("PRAGMA integrity_check").fetchone()[0]
+            users = database.execute("SELECT COUNT(*) FROM users").fetchone()[0]
+            posts = database.execute("SELECT COUNT(*) FROM posts").fetchone()[0]
+            database.close()
+        except sqlite3.DatabaseError as error:
+            print(f"BACKUP_VERIFY_FAILED: {error}", file=sys.stderr)
+            return 1
+
+    if integrity != "ok":
+        print(f"BACKUP_VERIFY_FAILED: integrity_check returned {integrity}", file=sys.stderr)
+        return 1
+    print(f"BACKUP_RESTORE_VERIFIED: users={users} posts={posts} source={source}")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())