"""V8.18 PostgreSQL disaster recovery, deliberately OFFLINE and fail-closed.

The management process and all Workers must be stopped while backing up the
application database and shared filesystem to guarantee an application-level
consistent pair. No switch-over to production is performed by this module.

pg_dump/pg_restore must be installed on the operator's administrative host.
Credentials are passed in child environment (not command arguments or logs).
"""
from __future__ import annotations

import hashlib
import io
import json
import os
import re
import shutil
import stat
import subprocess
import tempfile
import zipfile
from datetime import datetime, timezone
from pathlib import Path, PurePosixPath
from urllib.parse import parse_qsl, unquote, urlsplit

FORMAT = 'siteforge.pg.offline.v1'
MEDIA_ROOTS = ('media', 'legacy', 'builds', 'archives')
ALLOWED_EXTRA = {'manifest.json', 'database.dump'}
MAX_FILES = 250_000
MAX_UNCOMPRESSED = 30_000_000_000
MAX_RATIO = 1000

class RecoveryError(RuntimeError):
    pass


def connection_env(url: str, base: dict | None = None) -> dict:
    """PostgreSQL URL => libpq process env; never include password on argv."""
    p = urlsplit(url)
    if p.scheme not in ('postgresql', 'postgres') or not p.hostname or not p.path.strip('/'):
        raise RecoveryError('Require valid PostgreSQL URL with hostname and database')
    if '/' in unquote(p.path[1:]):
        raise RecoveryError('Invalid database name')
    opts = dict(parse_qsl(p.query, keep_blank_values=True))
    allowed = {'sslmode': 'PGSSLMODE', 'sslrootcert': 'PGSSLROOTCERT',
               'sslcert': 'PGSSLCERT', 'sslkey': 'PGSSLKEY',
               'connect_timeout': 'PGCONNECT_TIMEOUT'}
    if set(opts) - set(allowed):
        raise RecoveryError('Unsupported PostgreSQL URL options')
    try:
        port = str(p.port or 5432)
    except ValueError as exc:
        raise RecoveryError('Invalid PostgreSQL port') from exc
    env = dict(os.environ if base is None else base)
    env.update({'PGHOST': p.hostname, 'PGPORT': port,
                'PGDATABASE': unquote(p.path[1:]),
                'PGUSER': unquote(p.username or ''),
                'PGPASSWORD': unquote(p.password or ''),
                'PGAPPNAME': 'siteforge-v818-maintenance'})
    env.update({allowed[k]: v for k, v in opts.items()})
    # Maintenance must never silently fall back to a developer's ~/.pgpass.
    env['PGPASSFILE'] = os.devnull
    return env


def run_pg(argv: list[str], env: dict, timeout: int = 1800) -> subprocess.CompletedProcess:
    executable = argv[0]
    if shutil.which(executable) is None:
        raise RecoveryError(f'{executable} not installed on this host')
    try:
        result = subprocess.run(argv, env=env, stdout=subprocess.PIPE,
                                stderr=subprocess.PIPE, timeout=timeout, check=False)
    except subprocess.TimeoutExpired as exc:
        raise RecoveryError(f'{executable} timed out') from exc
    except OSError as exc:
        raise RecoveryError(f'{executable} execution failed') from exc
    if result.returncode:
        # pg tools can echo target location or credentials through external configs;
        # do not include untrusted tool stderr in user-facing errors.
        raise RecoveryError(f'{executable} returned non-zero exit code {result.returncode}')
    return result


def sha256_stream(stream) -> str:
    digest = hashlib.sha256()
    for chunk in iter(lambda: stream.read(1024 * 1024), b''):
        digest.update(chunk)
    return digest.hexdigest()


def sha256_path(path: Path) -> str:
    with path.open('rb') as f:
        return sha256_stream(f)


def media_files(source: Path):
    for folder in MEDIA_ROOTS:
        root = source / folder
        if root.is_symlink():
            raise RecoveryError(f'Symlinked storage root not allowed: {folder}')
        if not root.exists():
            continue
        for node in sorted(root.rglob('*')):
            if node.is_symlink():
                raise RecoveryError('Symlinked file in storage tree')
            if node.is_file():
                yield node, node.relative_to(source).as_posix()


def safe_member(name: str) -> bool:
    p = PurePosixPath(name)
    return bool(name and not name.endswith('/') and '\\' not in name and '\0' not in name
                and not p.is_absolute() and ':' not in name and
                all(part not in ('', '.', '..') for part in name.split('/')) and
                (name in ALLOWED_EXTRA or p.parts[0] in MEDIA_ROOTS))


def _inspect(path: Path):
    try:
        with zipfile.ZipFile(path) as z:
            infos = z.infolist()
            if len(infos) > MAX_FILES or sum(i.file_size for i in infos) > MAX_UNCOMPRESSED:
                raise RecoveryError('Backup resource limits exceeded')
            items = {}
            for info in infos:
                if not safe_member(info.filename) or info.filename in items:
                    raise RecoveryError('Unsafe or duplicate archive member')
                if stat.S_ISLNK(info.external_attr >> 16):
                    raise RecoveryError('Symlink in backup')
                if info.file_size > 5_000_000 and info.compress_size == 0:
                    raise RecoveryError('Invalid compressed member')
                if info.file_size > 5_000_000 and info.file_size / max(info.compress_size, 1) > MAX_RATIO:
                    raise RecoveryError('Unreasonable compression ratio')
                items[info.filename] = info
            if set(ALLOWED_EXTRA) - set(items):
                raise RecoveryError('Database dump or manifest missing')
            if items['manifest.json'].file_size > 2_000_000:
                raise RecoveryError('Manifest exceeds size limit')
            manifest = json.loads(z.read('manifest.json'))
            if manifest.get('format') != FORMAT or not isinstance(manifest.get('files'), dict):
                raise RecoveryError('Not a compatible SiteForge PG backup')
            documented = manifest['files']
            if set(documented) != set(items) - {'manifest.json'}:
                raise RecoveryError('Manifest and ZIP files differ')
            for name, rec in documented.items():
                if (not isinstance(rec, dict) or not isinstance(rec.get('size'), int)
                    or rec['size'] != items[name].file_size or
                    not re.fullmatch('[0-9a-f]{64}', str(rec.get('sha256', '')))):
                    raise RecoveryError('Invalid file metadata')
                with z.open(name) as f:
                    digest = sha256_stream(f)
                if digest != rec['sha256']:
                    raise RecoveryError('Checksum mismatch: ' + name[:100])
            return manifest
    except (zipfile.BadZipFile, json.JSONDecodeError, KeyError, UnicodeError) as exc:
        raise RecoveryError('Corrupted PostgreSQL backup') from exc


def verify(path: str | Path) -> dict:
    return _inspect(Path(path).resolve())


def backup(url: str, data: str | Path, archive: str | Path, *, confirmed_offline: bool = False,
           timeout: int = 1800) -> dict:
    if not confirmed_offline:
        raise RecoveryError('Stop all Web/Workers and pass --confirmed-offline')
    env = connection_env(url)
    source = Path(data).resolve()
    if not source.is_dir():
        raise RecoveryError('Shared storage directory does not exist')
    destination = Path(archive).resolve()
    if destination.exists():
        raise RecoveryError('Refusing to overwrite an existing backup')
    destination.parent.mkdir(parents=True, exist_ok=True)
    staging = destination.with_name(destination.name + '.partial')
    if staging.exists():
        raise RecoveryError('Partial backup file already exists')
    with tempfile.TemporaryDirectory(prefix='sf-pg-backup-') as temp:
        dump = Path(temp) / 'database.dump'
        run_pg(['pg_dump', '--format=custom', '--no-owner', '--no-acl',
                '--file', str(dump)], env, timeout)
        if not dump.is_file() or dump.stat().st_size == 0:
            raise RecoveryError('Empty database dump')
        run_pg(['pg_restore', '--list', str(dump)], env, min(120, timeout))
        members = [(dump, 'database.dump'), *media_files(source)]
        if len(members) > MAX_FILES:
            raise RecoveryError('Too many storage files')
        if sum(p.stat().st_size for p, _ in members) > MAX_UNCOMPRESSED:
            raise RecoveryError('Too much data for one backup')
        manifest = {'format': FORMAT, 'created_at': datetime.now(timezone.utc).isoformat(),
                    'database': 'postgresql', 'consistent': 'operator-confirmed-offline',
                    'files': {}}
        try:
            with zipfile.ZipFile(staging, 'w', compression=zipfile.ZIP_DEFLATED, compresslevel=5,
                                 allowZip64=True) as z:
                for path, name in members:
                    manifest['files'][name] = {'size': path.stat().st_size, 'sha256': sha256_path(path)}
                    z.write(path, name)
                z.writestr('manifest.json', json.dumps(manifest, ensure_ascii=False, sort_keys=True))
            _inspect(staging)
            staging.replace(destination)
        finally:
            staging.unlink(missing_ok=True)
    return {'archive': str(destination), 'files': len(members),
            'bytes': sum(x['size'] for x in manifest['files'].values())}


def restore_new(url: str, archive: str | Path, data: str | Path, *, confirmed_offline: bool = False,
                timeout: int = 1800) -> dict:
    """Restore ONLY into new, EMPTY PG database and non-existing storage path."""
    if not confirmed_offline:
        raise RecoveryError('Explicit offline confirmation is required')
    path = Path(archive).resolve()
    manifest = verify(path)
    dest = Path(data).resolve()
    if dest.exists():
        raise RecoveryError('Target storage must not exist (restore into new directory)')
    dest.parent.mkdir(parents=True, exist_ok=True)
    env = connection_env(url)
    # Empty target database check: exclude PostgreSQL system/extension internals.
    sql = "SELECT COUNT(*) FROM pg_catalog.pg_class WHERE relnamespace = 'public'::regnamespace AND relkind IN ('r','p','v','m','S','f')"
    result = run_pg(['psql', '--no-psqlrc', '--tuples-only', '--no-align', '--command', sql], env, 40)
    if result.stdout.strip() != b'0':
        raise RecoveryError('Target database public schema must be empty')
    with tempfile.TemporaryDirectory(prefix='.sf-pg-restore-', dir=dest.parent) as tmp:
        stage = Path(tmp) / 'siteforge'; stage.mkdir()
        with zipfile.ZipFile(path) as z:
            for name in manifest['files']:
                target = stage / name
                target.parent.mkdir(parents=True, exist_ok=True)
                with z.open(name) as src, target.open('wb') as out:
                    shutil.copyfileobj(src, out, 1024 * 1024)
        dump = stage / 'database.dump'
        run_pg(['pg_restore', '--exit-on-error', '--single-transaction', '--no-owner', '--no-acl',
                '--dbname', env['PGDATABASE'], str(dump)], env, timeout)
        dump.unlink()
        stage.replace(dest)
    return {'database': 'restored-into-empty-target', 'storage': str(dest),
            'media_files': len(manifest['files']) - 1}