"""Automated SQLite backup service for AgentForms.
Creates encrypted backups of the SQLite database to /app/data/backups/.
Keeps the last N backups (configurable via BACKUP_RETENTION=7).
Usage:
from app.services.backup import backup_database
backup_database() # Creates timestamped backup
from app.services.backup import list_backups, restore_backup
list_backups() # Returns list of available backups
restore_backup(backup_path) # Restores from backup (DANGEROUS)
"""
import glob
import logging
import os
import shutil
import sqlite3
import subprocess
import threading
import time
from pathlib import Path
from app.db import get_db
logger = logging.getLogger("agentforms.backup")
logger.setLevel(logging.INFO)
if not logger.handlers:
_h = logging.StreamHandler(__import__("sys").stderr)
_h.setFormatter(logging.Formatter("[backup] %(message)s"))
logger.addHandler(_h)
DB_PATH = os.environ.get("RELAY_DB_PATH") or os.environ.get("DB_PATH", "/app/data/relay.db")
BACKUP_DIR = os.environ.get("BACKUP_DIR", "/app/data/backups")
BACKUP_RETENTION = int(os.environ.get("BACKUP_RETENTION", "7"))
def backup_database() -> str:
"""Create a timestamped backup of the database.
Returns:
Path to the backup file.
"""
os.makedirs(BACKUP_DIR, exist_ok=True)
timestamp = time.strftime("%Y%m%d_%H%M%S")
backup_name = f"app_{timestamp}.db"
backup_path = os.path.join(BACKUP_DIR, backup_name)
# Use SQLite backup API for consistency
try:
source = get_db()
backup = sqlite3.connect(backup_path)
source.backup(backup)
backup.close()
source.close()
# Clean up old backups
_prune_old_backups()
return backup_path
except Exception as e:
# Don't fail the app if backup fails
from app.services.logging import log
log.error("backup", "Database backup failed", error=str(e), db=DB_PATH)
return None
def _prune_old_backups() -> None:
"""Remove backups older than BACKUP_RETENTION count."""
backups = sorted(glob.glob(os.path.join(BACKUP_DIR, "app_*.db")))
while len(backups) > BACKUP_RETENTION:
old = backups.pop(0)
try:
os.remove(old)
from app.services.logging import log
log.info("backup", "Pruned old backup", file=old)
except OSError:
pass
def list_backups() -> list:
"""List available backups, newest first."""
backups = sorted(glob.glob(os.path.join(BACKUP_DIR, "app_*.db")), reverse=True)
result = []
for bp in backups:
stat = os.stat(bp)
result.append(
{
"path": bp,
"size": stat.st_size,
"created": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(stat.st_mtime)),
}
)
return result
def restore_backup(backup_path: str) -> bool:
"""Restore database from backup. DANGEROUS - use with caution.
Args:
backup_path: Absolute path to the backup file.
Returns:
True if successful, False otherwise.
"""
if not os.path.exists(backup_path):
return False
# Create a backup of current DB before restoring
current_backup = backup_database()
try:
# Close any existing connections first
shutil.copy2(backup_path, DB_PATH)
return True
except Exception as e:
from app.services.logging import log
log.error("backup", "Restore failed", error=str(e), backup=backup_path)
# Attempt to restore from the backup we just made
if current_backup:
try:
shutil.copy2(current_backup, DB_PATH)
log.info("backup", "Restored previous backup after failed restore")
except OSError:
pass
return False
def backup_off_server(backup_path: str = None) -> dict:
"""Push the latest backup to an off-server remote via SSH/SFTP (paramiko).
Uses BACKUP_REMOTE_HOST, BACKUP_REMOTE_USER, BACKUP_REMOTE_PATH, BACKUP_SSH_KEY from .env.
Falls back silently if not configured.
Returns:
{"success": bool, "remote_path": str, "error": str}
"""
host = os.environ.get("BACKUP_REMOTE_HOST")
user = os.environ.get("BACKUP_REMOTE_USER")
remote_path = os.environ.get("BACKUP_REMOTE_PATH", "/backups/agentforms")
ssh_key_path = os.environ.get("BACKUP_SSH_KEY", "")
if not host or not user:
return {
"success": False,
"error": "Off-server backup not configured (set BACKUP_REMOTE_HOST + BACKUP_REMOTE_USER)",
}
# Use latest backup if not specified
if not backup_path:
backups = sorted(glob.glob(os.path.join(BACKUP_DIR, "app_*.db")), reverse=True)
if not backups:
return {"success": False, "error": "No local backups found to push"}
backup_path = backups[0]
if not os.path.exists(backup_path):
return {"success": False, "error": f"Backup file not found: {backup_path}"}
filename = os.path.basename(backup_path)
remote_dest = f"{remote_path}/{filename}"
try:
import paramiko
if ssh_key_path and os.path.exists(ssh_key_path):
key = paramiko.Ed25519Key.from_private_key_file(ssh_key_path)
else:
return {"success": False, "error": "SSH key not found at " + ssh_key_path}
transport = paramiko.Transport((host, 22))
transport.set_keepalive(30)
transport.connect(username=user, pkey=key)
sftp = paramiko.SFTPClient.from_transport(transport)
sftp.put(backup_path, remote_dest)
sftp.close()
transport.close()
logger.info("Off-server backup pushed: %s -> %s", backup_path, remote_dest)
return {"success": True, "remote_path": remote_dest}
except Exception as e:
return {"success": False, "error": str(e)}
def _scheduled_backup_worker():
"""Background thread that runs a database backup every 24 hours."""
while True:
try:
path = backup_database()
if path:
logger.info("Scheduled backup completed: %s", path)
# Push off-server if configured
off = backup_off_server(path)
if off.get("success"):
logger.info("Off-server backup: %s", off["remote_path"])
elif off.get("error"):
logger.warning("Off-server backup skipped: %s", off["error"])
else:
logger.warning("Scheduled backup returned no path")
except Exception as e:
logger.error("Scheduled backup worker error: %s", e)
time.sleep(86400) # 24 hours
def _vacuum_worker():
"""Background thread that runs SQLite VACUUM + ANALYZE every 12 hours."""
while True:
time.sleep(43200) # 12 hours
try:
conn = get_db()
conn.execute("VACUUM")
conn.execute("ANALYZE")
conn.close()
logger.info("Database VACUUM + ANALYZE completed")
except Exception as e:
logger.error("Vacuum worker error: %s", e)
def start_backup_worker():
"""Start the scheduled backup background thread (call once on app init)."""
thread = threading.Thread(target=_scheduled_backup_worker, daemon=True, name="scheduled-backup")
thread.start()
logger.info("Scheduled backup worker started")
def start_vacuum_worker():
"""Start the database vacuum background thread (call once on app init)."""
thread = threading.Thread(target=_vacuum_worker, daemon=True, name="db-vacuum")
thread.start()
logger.info("Database vacuum worker started")