"""Encrypted-at-rest-by-FileVault-free policy: backups are rsync'ed over SSH (encrypted in transit) to a second
LAN node, plus a pg_dump of the metadata database. Restore drill: `findata backup restore-drill`."""
from __future__ import annotations

import os
import subprocess
from datetime import UTC, datetime
from pathlib import Path
from typing import Any

from findata_core.config import get_settings
from findata_core.db import transaction

from findata_ingest.lake import Lake


def _env() -> dict[str, str]:
    env = dict(os.environ)
    env["PATH"] = "/opt/homebrew/opt/postgresql@17/bin:/opt/homebrew/bin:/usr/bin:/bin:/usr/sbin:/sbin"
    return env


def run(job: dict[str, Any], lake: Lake, log) -> dict[str, Any]:
    s = get_settings()
    target = (job.get("params") or {}).get("target") or s.backup_target
    if not target:
        raise RuntimeError("FINDATA_BACKUP_TARGET not configured")
    host, _, remote_dir = target.partition(":")
    stamp = datetime.now(UTC).strftime("%Y%m%dT%H%M%SZ")
    dumps = lake.root / "_pgdump"
    dumps.mkdir(exist_ok=True)
    dump = dumps / f"findata-{stamp}.sql.zst"
    url = s.database_url.get_secret_value()
    cmd = f"pg_dump --no-owner --no-privileges '{url}' | zstd -q -T0 -o '{dump}'"
    r = subprocess.run(["bash", "-c", cmd], capture_output=True, text=True, env=_env(), timeout=3600)
    if r.returncode != 0:
        raise RuntimeError(f"pg_dump failed: {r.stderr[-500:]}")
    for old in sorted(dumps.glob("findata-*.sql.zst"))[:-14]:
        old.unlink()
    subprocess.run(["ssh", "-o", "BatchMode=yes", host, f"mkdir -p {remote_dir}/lake {remote_dir}/pgdump"], check=True, timeout=60, env=_env())
    excludes = ["--exclude", "_tmp/", "--exclude", "_retired/", "--exclude", "exports/", "--exclude", ".writer.lock"]
    rs = subprocess.run(["rsync", "-a", "--delete", "--stats", *excludes, f"{lake.root}/", f"{host}:{remote_dir}/lake/"],
                        capture_output=True, text=True, env=_env(), timeout=6 * 3600)
    if rs.returncode != 0:
        raise RuntimeError(f"rsync failed: {rs.stderr[-500:]}")
    rs2 = subprocess.run(["rsync", "-a", f"{dumps}/", f"{host}:{remote_dir}/pgdump/"], capture_output=True, text=True, env=_env(), timeout=3600)
    if rs2.returncode != 0:
        raise RuntimeError(f"rsync pgdump failed: {rs2.stderr[-500:]}")
    # verification: remote file count + bytes of curated/
    v = subprocess.run(["ssh", "-o", "BatchMode=yes", host, f"find {remote_dir}/lake/curated -type f | wc -l; du -sk {remote_dir}/lake/curated | cut -f1"],
                       capture_output=True, text=True, env=_env(), timeout=600)
    remote_files, remote_kb = [x.strip() for x in v.stdout.strip().splitlines()[:2]] if v.returncode == 0 else ("?", "?")
    local_files = sum(1 for p in (lake.root / "curated").rglob("*") if p.is_file())
    stats = {"at": stamp, "target": target, "pg_dump": dump.name, "pg_dump_bytes": dump.stat().st_size,
             "rsync_summary": [ln for ln in rs.stdout.splitlines() if "Number of files" in ln or "Total transferred" in ln or "Total file size" in ln],
             "remote_curated_files": remote_files, "local_curated_files": local_files, "remote_curated_kb": remote_kb,
             "verified": str(local_files) == str(remote_files)}
    with transaction() as conn:
        conn.execute("INSERT INTO settings_kv(key, value) VALUES ('last_backup', %s) ON CONFLICT (key) DO UPDATE SET value=EXCLUDED.value, updated_at=now()",
                     (__import__("json").dumps(stats, default=str),))
    log.info("backup.done", **{k: v for k, v in stats.items() if k != "rsync_summary"})
    return stats


def restore_drill(target: str, dest: Path, dataset_id: str = "market_bars.stock.1m.raw") -> dict[str, Any]:
    """Restore the latest pg_dump into a scratch database and the lake into `dest`, then verify manifest hashes."""
    import hashlib
    s = get_settings()
    host, _, remote_dir = target.partition(":")
    dest.mkdir(parents=True, exist_ok=True)
    subprocess.run(["rsync", "-a", f"{host}:{remote_dir}/lake/curated/{dataset_id}/", f"{dest}/curated/{dataset_id}/"], check=True, env=_env(), timeout=3600)
    latest = subprocess.run(["ssh", "-o", "BatchMode=yes", host, f"ls -1 {remote_dir}/pgdump | sort | tail -1"], capture_output=True, text=True, env=_env(), timeout=60).stdout.strip()
    subprocess.run(["rsync", "-a", f"{host}:{remote_dir}/pgdump/{latest}", f"{dest}/"], check=True, env=_env(), timeout=600)
    base_url = s.database_url.get_secret_value().rsplit("/", 1)[0]
    scratch = f"findata_restore_{datetime.now(UTC).strftime('%H%M%S')}"
    env = _env()
    env["FD_PG_ADMIN"] = f"{base_url}/postgres"
    env["FD_PG_SCRATCH"] = f"{base_url}/{scratch}"
    env["FD_DUMP"] = f"{dest}/{latest}"

    def _sh(cmd: str, timeout: int) -> subprocess.CompletedProcess:
        r = subprocess.run(["bash", "-c", cmd], capture_output=True, text=True, env=env, timeout=timeout)
        if r.returncode != 0:
            from findata_core.redact import redact
            raise RuntimeError(redact(f"restore step failed: {r.stderr.strip()[-400:]}", s.secret_values()))
        return r

    _sh('psql "$FD_PG_ADMIN" -qc "CREATE DATABASE \"$(basename "$FD_PG_SCRATCH")\"" && zstd -dc "$FD_DUMP" | psql -q "$FD_PG_SCRATCH" >/dev/null', 3600)
    out = _sh(f'psql -tA "$FD_PG_SCRATCH" -c "SELECT count(*) FROM lake_files WHERE dataset_id=\'{dataset_id}\' AND version_removed IS NULL"', 120).stdout.strip()
    listing = _sh(f'psql -tA -F "|" "$FD_PG_SCRATCH" -c "SELECT path, sha256 FROM lake_files WHERE dataset_id=\'{dataset_id}\' AND version_removed IS NULL"', 600).stdout
    verified = 0
    missing = 0
    files = [ln.split("|") for ln in listing.strip().splitlines() if "|" in ln]
    for path, sha in files:
        p = dest / path
        if not p.exists():
            missing += 1
            continue
        h = hashlib.sha256(p.read_bytes()).hexdigest()
        verified += int(h == sha)
    subprocess.run(["bash", "-c", 'psql "$FD_PG_ADMIN" -qc "DROP DATABASE \"$(basename "$FD_PG_SCRATCH")\""'], env=env, timeout=120, capture_output=True)
    return {"pg_dump": latest, "restored_lake_files_in_db": out, "files_listed": len(files),
            "files_hash_verified": verified, "files_missing": missing, "dest": str(dest)}