#!/usr/bin/env python3
"""Host-only backup policy and systemd installation, using only the stdlib."""
import argparse
from contextlib import contextmanager
from datetime import datetime, timezone
import fcntl
import hashlib
import json
import os
from pathlib import Path
import pwd
import re
import subprocess
import sys
import tempfile
import time
import uuid

DEPLOY = Path(__file__).resolve().parent.parent
STATE = DEPLOY / ".auto-backup"
RETRIES = 3
RETRY_SECONDS = 900
MAX_AGE = 86400


def utc(epoch=None):
    return datetime.fromtimestamp(time.time() if epoch is None else epoch, timezone.utc).isoformat()


def read_json(path):
    with path.open() as stream:
        return json.load(stream)


def atomic_json(path, data):
    fd, temp = tempfile.mkstemp(prefix=path.name + ".part.", dir=path.parent)
    try:
        with os.fdopen(fd, "w") as stream:
            json.dump(data, stream, ensure_ascii=False, indent=2)
            stream.write("\n")
            stream.flush()
            os.fsync(stream.fileno())
        os.replace(temp, path)
        sync_dir(path.parent)
    finally:
        Path(temp).unlink(missing_ok=True)


def sync_dir(path):
    fd = os.open(path, os.O_RDONLY | os.O_DIRECTORY)
    try:
        os.fsync(fd)
    finally:
        os.close(fd)


def digest(path):
    h = hashlib.sha256()
    with path.open("rb") as stream:
        for chunk in iter(lambda: stream.read(1024 * 1024), b""):
            h.update(chunk)
    return h.hexdigest()


def component(value):
    # Hash prevents collisions between DB names with punctuation or Unicode.
    return re.sub(r"[^A-Za-z0-9_.-]", "_", value)[:60] + "-" + hashlib.sha256(value.encode()).hexdigest()[:12]


@contextmanager
def lock(path):
    with path.open("a") as stream:
        try:
            fcntl.flock(stream, fcntl.LOCK_EX | fcntl.LOCK_NB)
        except BlockingIOError as exc:
            raise RuntimeError("another automatic backup or configuration operation is running") from exc
        yield


def mount_check(config):
    mount = config.get("mount")
    if mount and subprocess.run(["mountpoint", "-q", "--", mount]).returncode:
        raise RuntimeError("required backup mount is disconnected: " + mount)


@contextmanager
def storage(config):
    """Anchor writes to an open directory: unmount must never fall back to local storage."""
    root = Path(config["backup_root"])
    mount_check(config)
    anchor = Path(config["mount"]) if config.get("mount") else root
    if not config.get("mount"):
        anchor.mkdir(parents=True, exist_ok=True, mode=0o700)
    old = os.open(".", os.O_RDONLY | os.O_DIRECTORY)
    fd = os.open(anchor, os.O_RDONLY | os.O_DIRECTORY)
    try:
        mount_check(config)
        if os.fstat(fd).st_dev != anchor.stat().st_dev:
            raise RuntimeError("backup mount changed during access")
        os.fchdir(fd)
        relative = root.relative_to(anchor) / component(config["project"]) / component(config["database"])
        relative.mkdir(parents=True, exist_ok=True, mode=0o700)
        os.chdir(relative)
        yield root / component(config["project"]) / component(config["database"])
        mount_check(config)
    finally:
        os.fchdir(old)
        os.close(fd)
        os.close(old)


def prune(config, latest):
    cutoff = time.time() - config["retention_days"] * 86400
    # Only a complete, matching auto metadata record authorizes deletion.
    for sidecar in Path(".").glob("auto-*.dump.json"):
        if sidecar.is_symlink():
            continue
        try:
            record = read_json(sidecar)
            name = sidecar.name[:-5]
            archive = Path(name)
            if (name == latest or not re.fullmatch(r"auto-\d{8}T\d{6}Z-[0-9a-f]{32}\.dump", name)
                    or record.get("kind") != "automatic-postgresql-v1"
                    or record.get("project") != config["project"]
                    or record.get("database") != config["database"]
                    or Path(record.get("file", "")).name != name
                    or record["completed_epoch"] >= cutoff or archive.is_symlink()):
                continue
            mount_check(config)
            archive.unlink(missing_ok=True)
            sidecar.unlink()
        except (ValueError, KeyError, TypeError):
            continue  # Unknown/manual files are never adopted.


def attempt(config):
    started = time.time()
    name = "auto-" + datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") + "-" + uuid.uuid4().hex + ".dump"
    with storage(config) as directory:
        env = {**os.environ, "ENGINE": config["engine"], "ENV_FILE": config["env_file"],
               "AUTO_BACKUP_EXPECT_PROJECT": config["project"], "AUTO_BACKUP_EXPECT_DB": config["database"]}
        subprocess.run(["bash", str(DEPLOY / "bin/backup-db.sh"), config["engine"], name], env=env, check=True)
        archive = Path(name)
        with archive.open("rb") as stream:
            os.fsync(stream.fileno())
        checksum = digest(archive)
        completed = time.time()
        record = {"kind": "automatic-postgresql-v1", "project": config["project"],
                  "database": config["database"], "file": str(directory / name),
                  "started_at": utc(started), "completed_at": utc(completed),
                  "started_epoch": started, "completed_epoch": completed,
                  "duration_seconds": round(completed - started, 3),
                  "size_bytes": archive.stat().st_size, "sha256": checksum}
        mount_check(config)
        atomic_json(Path(name + ".json"), record)
        atomic_json(STATE / "last-success.json", record)
        # Once success is durable, retention is allowed. Pruning errors remain visible.
        prune(config, name)
    return record


def run(config):
    if os.geteuid() != config["uid"]:
        raise RuntimeError("run as the configured backup account: " + config["user"])
    with lock(STATE / "run.lock"):
        for number in range(RETRIES + 1):
            try:
                record = attempt(config)
                print("backup succeeded: " + record["file"], flush=True)
                return 0
            except (OSError, ValueError, RuntimeError, subprocess.SubprocessError) as exc:
                failure = {"failed_at": utc(), "failed_epoch": time.time(), "attempt": number + 1,
                           "error": str(exc)}
                atomic_json(STATE / "last-failure.json", failure)
                print(json.dumps(failure, ensure_ascii=False), file=sys.stderr, flush=True)
                if number < RETRIES:
                    time.sleep(RETRY_SECONDS)
        return 1


def status(config, check=False, verify_checksum=False):
    success_path, failure_path = STATE / "last-success.json", STATE / "last-failure.json"
    success = read_json(success_path) if success_path.exists() else None
    failure = read_json(failure_path) if failure_path.exists() else None
    errors = []
    age = None
    try:
        mount_check(config)
    except RuntimeError as exc:
        errors.append(str(exc))
    if not success:
        errors.append("no successful automatic backup")
    else:
        age = round(time.time() - success["started_epoch"], 1)
        archive = Path(success["file"])
        expected_dir = Path(config["backup_root"]) / component(config["project"]) / component(config["database"])
        if archive.parent != expected_dir:
            errors.append("latest archive belongs to a different backup destination")
        if success["project"] != config["project"] or success["database"] != config["database"]:
            errors.append("backup identity differs from configuration")
        if age > MAX_AGE or age < 0:
            errors.append("backup snapshot is older than 24 hours or clock moved backwards")
        if not archive.is_file() or archive.is_symlink() or archive.stat().st_size != success["size_bytes"]:
            errors.append("latest archive missing or size changed")
        elif verify_checksum and digest(archive) != success["sha256"]:
            errors.append("latest archive checksum mismatch")
        sidecar = Path(str(archive) + ".json")
        if not sidecar.is_file() or read_json(sidecar) != success:
            errors.append("latest archive metadata missing or changed")
    if failure and (not success or failure["failed_epoch"] > success["completed_epoch"]):
        errors.append("recent backup attempt failed")
    result = {"enabled": config["enabled"], "unit": config["unit"], "scope": config["scope"],
              "last_success": success, "last_failure": failure, "age_seconds": age, "errors": errors}
    print(json.dumps(result, ensure_ascii=False, indent=2))
    return 1 if errors and check else 0


def systemctl(config, *args):
    command = ["systemctl"] + (["--user"] if config["scope"] == "user" else [])
    subprocess.run([*command, *args], check=True)


def unit_quote(value, command=False):
    if any(ord(c) < 32 for c in value):
        raise ValueError("control characters are not allowed in systemd settings")
    value = value.replace("\\", "\\\\").replace('"', '\\"').replace("%", "%%")
    return '"' + (value.replace("$", "$$") if command else value) + '"'


def unit_directory(scope, account):
    return Path("/etc/systemd/system") if scope == "system" else Path(account.pw_dir) / ".config/systemd/user"


def detect_engine(account, project):
    """Probe running deployment containers as the eventual backup account."""
    prefix = [] if os.geteuid() == account.pw_uid else ["runuser", "-u", account.pw_name, "--"]
    name_pattern = re.compile(re.escape(project) + r"(?:-postgres(?:-[0-9]+)?|_postgres_[0-9]+)")
    candidates = []
    for engine in ("docker", "podman"):
        try:
            result = subprocess.run([*prefix, engine, "ps", "--format", "{{.Names}}"],
                                    capture_output=True, text=True, timeout=10)
        except (OSError, subprocess.TimeoutExpired):
            continue
        if result.returncode == 0:
            candidates.extend((engine, name) for name in result.stdout.splitlines() if name_pattern.fullmatch(name))
    if len(candidates) != 1:
        raise RuntimeError("cannot uniquely detect the deployment PostgreSQL engine as " + account.pw_name
                           + "; check running containers/permissions or specify --engine docker|podman")
    print("Detected backup engine: " + candidates[0][0] + " (" + candidates[0][1] + ")")
    return candidates[0][0]


def install(args):
    account = pwd.getpwnam(args.user)
    if args.scope == "system" and os.geteuid() != 0:
        raise RuntimeError("system timer installation requires root; use sudo")
    if args.scope == "user" and (os.geteuid() != account.pw_uid or account.pw_uid == 0):
        raise RuntimeError("install a user timer while logged in as its non-root execution account")
    engine = detect_engine(account, os.environ["AUTO_BACKUP_PROJECT"]) if args.engine == "auto" else args.engine
    if engine == "podman" and account.pw_uid != 0 and args.scope != "user":
        raise RuntimeError("rootless Podman requires --scope user and that user's login session")
    if args.retention_days < 1:
        raise ValueError("retention days must be positive")
    unit_quote(args.calendar)
    subprocess.run(["systemd-analyze", "calendar", args.calendar], check=True, stdout=subprocess.DEVNULL)
    project, database = os.environ["AUTO_BACKUP_PROJECT"], os.environ["AUTO_BACKUP_DB"]
    root = Path(args.backup_dir or DEPLOY / "backups/auto").resolve()
    for protected in ("bin", "conf", ".auto-backup", ".bundle-state"):
        if root == DEPLOY / protected or DEPLOY / protected in root.parents:
            raise ValueError("backup destination overlaps deployment code/configuration/state")
    mount = Path(args.require_mount).resolve() if args.require_mount else None
    if mount:
        root.relative_to(mount)  # Reject NAS guard unrelated to the destination.
    unit = "pi-db-backup-" + hashlib.sha256(str(DEPLOY).encode()).hexdigest()[:16]
    unit_dir = unit_directory(args.scope, account)
    config = {"version": 1, "project": project, "database": database, "engine": engine,
              "env_file": str(Path(os.environ["AUTO_BACKUP_ENV_FILE"]).resolve()),
              "backup_root": str(root), "mount": str(mount) if mount else None,
              "retention_days": args.retention_days, "calendar": args.calendar,
              "user": args.user, "uid": account.pw_uid, "scope": args.scope,
              "unit": unit, "unit_dir": str(unit_dir), "enabled": False}
    mount_check(config)
    STATE.mkdir(mode=0o700, exist_ok=True)
    with lock(STATE / "run.lock"):
        if (STATE / "config.json").exists():
            raise RuntimeError("configuration already exists; uninstall before changing installation settings")
        if args.scope == "user":
            # Root authorization, when required by the host, is handled by loginctl.
            subprocess.run(["loginctl", "enable-linger", args.user], check=True)
        unit_dir.mkdir(parents=True, exist_ok=True)
        replacements = {"NAME": unit, "USER": "User=" + args.user if args.scope == "system" else "",
                        "COMMAND": unit_quote(str(DEPLOY / "bin/auto-backup-db.sh"), command=True),
                        "CALENDAR": args.calendar.replace("%", "%%")}
        created = []
        try:
            for kind in ("service", "timer"):
                content = (DEPLOY / "bin/systemd" / ("auto-backup." + kind + ".in")).read_text()
                for key, value in replacements.items():
                    content = content.replace("@" + key + "@", value)
                path = unit_dir / (unit + "." + kind)
                with path.open("x") as stream:
                    created.append(path)
                    stream.write(content)
                path.chmod(0o644)
            atomic_json(STATE / "config.json", config)
            # This account also runs manual backup/restore to share the same lock inode.
            shared_lock = DEPLOY / ".db-operations.lock"
            shared_lock.touch(exist_ok=True, mode=0o600)
            if os.geteuid() == 0:
                for path in (STATE, STATE / "run.lock", STATE / "config.json", shared_lock):
                    os.chown(path, account.pw_uid, account.pw_gid)
            systemctl(config, "daemon-reload")
        except BaseException:
            for path in created:
                path.unlink(missing_ok=True)
            (STATE / "config.json").unlink(missing_ok=True)
            raise
    print("Installed disabled: " + unit + ".timer; run first backup and isolated restore, then enable --restore-verified")


def lifecycle(config, command, restore_verified):
    if config["scope"] == "user" and os.geteuid() != config["uid"]:
        raise RuntimeError("manage this user timer as " + config["user"])
    if config["scope"] == "system" and os.geteuid() != 0:
        raise RuntimeError("manage this system timer with sudo")
    if command == "enable":
        if not restore_verified:
            raise RuntimeError("enable requires --restore-verified after an isolated restore drill")
        with lock(STATE / "run.lock"):
            if status(config, check=True, verify_checksum=True):
                raise RuntimeError("a healthy first backup is required before activation")
            config["enabled"] = True
            # Persist monitoring intent first; failed activation must not hide missing backups.
            atomic_json(STATE / "config.json", config)
            if os.geteuid() == 0:
                account = pwd.getpwnam(config["user"])
                os.chown(STATE / "config.json", account.pw_uid, account.pw_gid)
        # Persistent timers can immediately launch the runner. Release its lock first.
        systemctl(config, "enable", "--now", config["unit"] + ".timer")
    else:
        systemctl(config, "disable", "--now", config["unit"] + ".timer")
        # Do not interrupt a dump. A running service must finish before uninstall.
        with lock(STATE / "run.lock"):
            config["enabled"] = False
            atomic_json(STATE / "config.json", config)
            if command == "uninstall":
                for kind in ("service", "timer"):
                    (Path(config["unit_dir"]) / (config["unit"] + "." + kind)).unlink(missing_ok=True)
                systemctl(config, "daemon-reload")
                (STATE / "config.json").rename(STATE / "uninstalled-config.json")
                # Backups, success/failure evidence and linger (possibly shared) are retained.
    if os.geteuid() == 0 and (STATE / "config.json").exists():
        account = pwd.getpwnam(config["user"])
        os.chown(STATE / "config.json", account.pw_uid, account.pw_gid)


def main():
    os.umask(0o077)
    parser = argparse.ArgumentParser(description=__doc__)
    commands = parser.add_subparsers(dest="command", required=True)
    p = commands.add_parser("install", help="install disabled units and site configuration")
    p.add_argument("--engine", choices=("auto", "docker", "podman"), default="auto",
                   help="default: detect the running PostgreSQL container as --user; persist the chosen engine")
    p.add_argument("--user", required=True, help="backup execution account with engine and backup-directory access")
    p.add_argument("--scope", choices=("system", "user"), required=True,
                   help="system: host timer installed as root; user: account timer (required for rootless Podman)")
    p.add_argument("--backup-dir")
    p.add_argument("--require-mount")
    p.add_argument("--calendar", default="*-*-* 03:00:00 Asia/Seoul")
    p.add_argument("--retention-days", type=int, default=30)
    commands.add_parser("run", help="one backup cycle, including up to three retries")
    p = commands.add_parser("status")
    p.add_argument("--check", action="store_true")
    p.add_argument("--verify-checksum", action="store_true")
    p.add_argument("--if-enabled", action="store_true")
    p = commands.add_parser("enable")
    p.add_argument("--restore-verified", action="store_true")
    commands.add_parser("disable")
    commands.add_parser("uninstall")
    args = parser.parse_args()
    try:
        if args.command == "install":
            install(args)
            return 0
        if args.command == "status" and args.if_enabled and not (STATE / "config.json").exists():
            print("automatic backup is not installed")
            return 0
        config = read_json(STATE / "config.json")
        if args.command == "run":
            return run(config)
        if args.command == "status":
            if args.if_enabled and not config["enabled"]:
                print("automatic backup is disabled")
                return 0
            return status(config, args.check, args.verify_checksum)
        lifecycle(config, args.command, getattr(args, "restore_verified", False))
        return 0
    except (OSError, ValueError, KeyError, RuntimeError, subprocess.SubprocessError) as exc:
        print("automatic backup: " + str(exc), file=sys.stderr)
        return 1


if __name__ == "__main__":
    sys.exit(main())