FabledSteward/steward/backup/job.py

"""Steward's own backups as a scheduled job: make the file, send it to the
bucket, record how it went (#5673).

Once a day at the hour chosen in Settings → Backups, in the zone of whoever
saved it. A day missed while Steward was down is made up on the next tick;
a failed run is tried again an hour later, never sooner, so a target that is
down isn't pressed. Each run is kept in `backup.history` (newest first, the
last HISTORY_KEEP), audited, and leaves two metrics under source "steward",
resource "backup": backup_age_secs (since the last success, every tick while
backups are on, so an alert rule can watch it) and backup_bytes.
"""
from __future__ import annotations

import asyncio
import logging
import tempfile
import time
from datetime import datetime, timedelta, timezone
from pathlib import Path

from sqlalchemy import text
from sqlalchemy.ext.asyncio import create_async_engine

from .dump import BackupError, make_backup
from .s3 import S3Error, S3Target, get_file, list_objects, put_file

logger = logging.getLogger(__name__)

TICK_SECONDS = 15 * 60
RETRY_AFTER = timedelta(hours=1)
HISTORY_KEEP = 30
PREFIX = "steward-"

# One backup at a time, whether the schedule or a person started it.
_running = asyncio.Lock()
_background: set[asyncio.Task] = set()


async def migration_heads(database_url: str) -> list[str]:
    """The migration revisions the database is at, for the manifest."""
    engine = create_async_engine(database_url)
    try:
        async with engine.connect() as conn:
            return list((await conn.execute(text("SELECT version_num FROM alembic_version"))).scalars())
    finally:
        await engine.dispose()


def target_from(settings: dict) -> S3Target | None:
    """The bucket the settings name, or None while they're incomplete."""
    endpoint = (settings.get("backup.s3_endpoint") or "").strip()
    bucket = (settings.get("backup.s3_bucket") or "").strip()
    access = (settings.get("backup.s3_access_key") or "").strip()
    secret = settings.get("backup.s3_secret_key") or ""
    if not (endpoint and bucket and access and secret):
        return None
    return S3Target(endpoint, bucket, access, secret,
                    region=(settings.get("backup.s3_region") or "us-east-1").strip(),
                    verify_tls=bool(settings.get("backup.s3_verify_tls", True)))


def _moment(raw) -> datetime | None:
    try:
        return datetime.fromisoformat(raw) if raw else None
    except (TypeError, ValueError):
        return None


def last_slot(now: datetime, hour: int, tz) -> datetime:
    """The latest scheduled moment at or before `now`: today's hour in `tz`,
    or yesterday's when today's hasn't come yet."""
    local = now.astimezone(tz)
    slot = local.replace(hour=hour, minute=0, second=0, microsecond=0)
    if slot > local:
        slot -= timedelta(days=1)
    return slot.astimezone(timezone.utc)


def is_due(now: datetime, hour: int, tz, last_success: datetime | None,
           last_attempt: datetime | None) -> bool:
    """Due when no backup has succeeded since the latest slot, and the last
    try (if it failed) was at least RETRY_AFTER ago."""
    slot = last_slot(now, hour, tz)
    if last_success is not None and last_success >= slot:
        return False
    return last_attempt is None or now - last_attempt >= RETRY_AFTER


def history_with(history: list, entry: dict) -> list:
    return ([entry] + list(history or []))[:HISTORY_KEEP]


async def _record(app, entry: dict, user_id: str | None, username: str) -> None:
    from steward.core.alerts import MetricBatch
    from steward.core.audit import log_audit
    from steward.core.settings import get_setting, set_setting

    async with app.db_sessionmaker() as db:
        async with db.begin():
            history = await get_setting(db, "backup.history") or []
            await set_setting(db, "backup.history", history_with(history, entry))
            await set_setting(db, "backup.last_attempt_at", entry["at"])
            if entry["ok"]:
                await set_setting(db, "backup.last_success_at", entry["at"])
                batch = MetricBatch()
                batch.add(source_module="steward", resource_name="backup",
                          metric_name="backup_bytes", value=float(entry["size"]))
                batch.add(source_module="steward", resource_name="backup",
                          metric_name="backup_age_secs", value=0.0)
                await batch.record(db)
    await log_audit(app, user_id, username, "backup.made" if entry["ok"] else "backup.failed",
                    entity_type="backup", entity_id=entry.get("name") or None,
                    detail={k: entry[k] for k in ("name", "size", "secs", "error", "by") if entry.get(k)})


async def run_backup(app, by: str = "schedule", user_id: str | None = None) -> dict:
    """Make one backup and send it. Returns the history entry it recorded."""
    from steward.core.settings import get_all_settings

    async with _running:
        started = time.monotonic()
        at = datetime.now(timezone.utc)
        entry = {"at": at.isoformat(), "ok": False, "by": by, "name": "", "size": 0,
                 "secs": 0.0, "error": ""}
        async with app.db_sessionmaker() as db:
            settings = await get_all_settings(db)
        target = target_from(settings)
        try:
            if target is None:
                raise BackupError("The bucket isn't set up: fill in the endpoint, bucket and keys.")
            db_url = app.config["DATABASE_URL"]
            heads = await migration_heads(db_url)
            with tempfile.TemporaryDirectory(prefix="steward-backup-") as tmp:
                made = await make_backup(db_url, app.config["SECRET_KEY"], heads, Path(tmp), now=at)
                await put_file(target, made.name, made.path)
                entry.update(ok=True, name=made.name, size=made.size)
                entry["check"] = await _self_check(app, made.path, made.name)
        except (BackupError, S3Error) as exc:
            entry["error"] = str(exc)
        except Exception as exc:     # recorded and logged: a run never takes the scheduler down
            logger.exception("Backup failed")
            entry["error"] = f"Unexpected error: {exc}"
        entry["secs"] = round(time.monotonic() - started, 1)
        if entry["ok"]:
            logger.info("Backup %s sent (%d bytes, %.1fs)", entry["name"], entry["size"], entry["secs"])
        else:
            logger.warning("Backup failed: %s", entry["error"])
        await _record(app, entry, user_id, by if by != "schedule" else "")
        return entry


async def _self_check(app, path: Path, name: str) -> dict:
    """A backup just sent, read back and compared with what it was made from
    (#5823): proof the file restores, kept on its history row. A check that
    can't run says why; the backup was sent either way."""
    try:
        report = await check_file(app, path, name)
    except Exception as exc:
        logger.warning("Backup %s couldn't be checked: %s", name, exc)
        return {"ok": False, "changes": 0, "rows": 0, "summary": f"Couldn't check it: {exc}"}
    return report.as_entry()


async def check_file(app, path: Path, name: str = ""):
    """Compare a backup file with this Steward's database (check.py)."""
    from .check import check_backup
    return await check_backup(path, app.config["DATABASE_URL"], app.config["SECRET_KEY"],
                              plugin_dirs=[Path(d) for d in app.config["PLUGIN_DIRS"]],
                              name=name or path.name)


def is_backup_name(name: str) -> bool:
    """A file name Steward's backups have, and nothing else: no path, no
    other object in the bucket."""
    import re
    return bool(re.fullmatch(r"steward-\d{8}T\d{6}Z\.tar", name or ""))


async def check_stored(app, name: str):
    """Fetch one of this Steward's backups from the bucket and check it.
    Raises BackupError or S3Error with what went wrong."""
    from steward.core.settings import get_all_settings
    if not is_backup_name(name):
        raise BackupError(f"{name!r} isn't the name of a Steward backup.")
    async with app.db_sessionmaker() as db:
        target = target_from(await get_all_settings(db))
    if target is None:
        raise BackupError("The bucket isn't set up: fill in the endpoint, bucket and keys.")
    with tempfile.TemporaryDirectory(prefix="steward-check-") as tmp:
        path = Path(tmp) / name
        await get_file(target, name, path)
        return await check_file(app, path, name)


def start_backup(app, by: str, user_id: str | None) -> bool:
    """Start a backup in the background for a person; False if one is running."""
    if _running.locked():
        return False
    task = asyncio.get_running_loop().create_task(run_backup(app, by=by, user_id=user_id))
    _background.add(task)
    task.add_done_callback(_background.discard)
    return True


def is_running() -> bool:
    return _running.locked()


async def tick(app, now: datetime | None = None) -> None:
    """The scheduled task: record how old the newest backup is, and make one
    when it is due."""
    from steward.core.alerts import MetricBatch
    from steward.core.formatting import zone
    from steward.core.settings import get_all_settings

    now = now or datetime.now(timezone.utc)
    async with app.db_sessionmaker() as db:
        settings = await get_all_settings(db)
    if not settings.get("backup.enabled"):
        return
    last_success = _moment(settings.get("backup.last_success_at"))
    if last_success is not None:
        async with app.db_sessionmaker() as db:
            async with db.begin():
                batch = MetricBatch()
                batch.add(source_module="steward", resource_name="backup",
                          metric_name="backup_age_secs", value=(now - last_success).total_seconds())
                await batch.record(db)
    if _running.locked():
        return
    if is_due(now, int(settings.get("backup.schedule_hour", 3)), zone(settings.get("backup.schedule_tz")),
              last_success, _moment(settings.get("backup.last_attempt_at"))):
        await run_backup(app)


async def test_target(target: S3Target) -> tuple[bool, str]:
    """Settings → Backups' Test button: can Steward reach the bucket with
    these keys? A listing refused as AccessDenied still proves the keys:
    an append-only key may not list, and uploads anyway."""
    try:
        found = await list_objects(target, PREFIX, limit=1000)
    except S3Error as exc:
        said = str(exc)
        if "AccessDenied" in said:
            return True, ("Reached the bucket and the keys were accepted. They may not list "
                          "what is there, which an append-only key can't; uploads still work.")
        return False, said
    if not found:
        return True, f"Reached {target.bucket}; no Steward backups there yet."
    newest = max(found, key=lambda o: o.key)
    return True, f"Reached {target.bucket}: {len(found)} Steward backups, the newest {newest.key}."


# ── What Settings → Backups shows ────────────────────────────────────────────

def shown_hour(hour: int, saved_tz, view_tz, now: datetime | None = None) -> int:
    """The schedule's hour as a reader in `view_tz` sees it today."""
    local = (now or datetime.now(timezone.utc)).astimezone(saved_tz)
    return local.replace(hour=hour, minute=0, second=0, microsecond=0).astimezone(view_tz).hour


def history_rows(history: list) -> list[dict]:
    """The stored history with its times read back as datetimes."""
    return [entry | {"at": _moment(entry.get("at"))} for entry in history or []]


def verdict(settings: dict, running: bool) -> dict:
    """One line for the top of the page: {"tone": ok|warn|bad|off, "text", "moment"}.
    `moment` is the time the line is about, shown as "ago" by the template,
    which ends the sentence; a failure also carries `failed`, its `error` and
    `good` (the newest success, or None)."""
    rows = history_rows(settings.get("backup.history") or [])
    last_ok = _moment(settings.get("backup.last_success_at"))
    if running:
        return {"tone": "ok", "text": "A backup is being made now", "moment": None}
    if not settings.get("backup.enabled"):
        if target_from(settings) is None:
            return {"tone": "off", "text": "Backups are off, and no bucket is set up", "moment": None}
        return {"tone": "off", "text": "Scheduled backups are off; Back up now still works", "moment": None}
    if target_from(settings) is None:
        return {"tone": "bad", "text": "Backups are on, but the bucket isn't set up", "moment": None}
    if rows and not rows[0]["ok"]:
        # `good`: when the newest backup that did go was sent (None: never).
        return {"tone": "bad", "text": "The last backup failed", "moment": rows[0]["at"],
                "error": rows[0]["error"], "good": last_ok, "failed": True}
    if last_ok is None:
        return {"tone": "warn", "text": "No backup has been sent yet; the first goes within 15 minutes",
                "moment": None}
    return {"tone": "ok", "text": f"The newest backup is {rows[0]['name']} ({_size(rows[0]['size'])}), sent",
            "moment": last_ok}


def _size(n) -> str:
    from steward.core.formatting import human_bytes
    return human_bytes(n)