#!/usr/bin/env python3
"""Verified cloud staging and fail-closed Borg backup. Standard library only."""
import datetime as dt
import fcntl
import hashlib
import json
import os
import re
from pathlib import Path, PurePosixPath
import shutil
import signal
import subprocess
import sys
import tempfile
import time


class BackupError(RuntimeError):
    pass


def log(message):
    print(f'{dt.datetime.now().astimezone().isoformat(timespec="seconds")} {message}', flush=True)


def atomic_json(path, data):
    descriptor, name = tempfile.mkstemp(prefix=path.name + '.', dir=path.parent)
    tmp = Path(name)
    try:
        with os.fdopen(descriptor, 'w') as stream:
            json.dump(data, stream, ensure_ascii=False, indent=2)
            stream.flush()
            os.fsync(stream.fileno())
        tmp.replace(path)
        directory = os.open(path.parent, os.O_RDONLY | os.O_DIRECTORY)
        try:
            os.fsync(directory)
        finally:
            os.close(directory)
    finally:
        tmp.unlink(missing_ok=True)


def read_json(path):
    if not path.exists():
        return {}
    return json.loads(path.read_text())


class Runner:
    def __init__(self):
        e = os.environ
        self.state = Path(e['STATE_DIR']).expanduser().absolute()
        self.repo = Path(e['REPO_DIR']).absolute()
        self.stage = Path(e['STAGING_DIR']).expanduser().absolute()
        self.remote = e['REMOTE']
        if not re.fullmatch(r'[A-Za-z0-9_.-]+', e['ARCHIVE_PREFIX']):
            raise BackupError('Unsafe ARCHIVE_PREFIX')
        self.borg = e.get('BORG_BIN', 'borg')
        self.timeout = int(e.get('COMMAND_TIMEOUT_SECONDS', '21600'))
        self.success_file = self.state / 'last-success.json'
        self.attempt_file = self.state / 'last-attempt.json'
        self.child = None
        self.phase = 'preflight'
        self.state.mkdir(parents=True, exist_ok=True, mode=0o700)
        self.success = read_json(self.success_file)
        # An old timestamp from a different repository/source must never suppress a backup.
        if self.success and (self.success.get('repo') != str(self.repo) or
                             self.success.get('remote') != self.remote):
            self.success = {}

    def run(self, args, *, capture=False, cwd=None, timeout=None):
        log('Running: ' + ' '.join(map(str, args)))
        self.child = subprocess.Popen(list(map(str, args)), cwd=cwd,
                                      stdout=subprocess.PIPE if capture else None,
                                      start_new_session=True, text=True)
        try:
            out, _ = self.child.communicate(timeout=timeout or self.timeout)
            if self.child.returncode:
                raise BackupError(f'{args[0]} failed (exit {self.child.returncode}); phase={self.phase}')
            return out
        except BaseException:
            self.stop_child()
            raise
        finally:
            self.child = None

    def stop_child(self):
        if self.child and self.child.poll() is None:
            os.killpg(self.child.pid, signal.SIGTERM)
            try:
                self.child.wait(timeout=10)
            except subprocess.TimeoutExpired:
                os.killpg(self.child.pid, signal.SIGKILL)
                self.child.wait()

    def cloud(self, *args, capture=False):
        return self.run(['rclone', *args, '--contimeout', '20s', '--timeout', '2m',
                         '--retries', '3', '--low-level-retries', '10',
                         '--log-level', 'INFO', '--stats', '1m'], capture=capture)

    def nas(self):
        # Trigger autofs, then verify the backing mount. Never mkdir or init on /net.
        self.run(['timeout', '60', 'test', '-f', self.repo / 'config'], timeout=70)
        mounts = json.loads(self.run(['findmnt', '--json', '--target', self.repo,
                                    '--output', 'SOURCE,FSTYPE,MAJ:MIN'], capture=True, timeout=60))['filesystems']
        device = self.repo.stat().st_dev
        backing_device = f'{os.major(device)}:{os.minor(device)}'
        mounts = [mount for mount in mounts if mount.get('maj:min') == backing_device]
        expected = os.environ.get('EXPECTED_NAS_SOURCE', 'thor:/volume1/NetBackup')
        fstype = os.environ.get('EXPECTED_NAS_FSTYPE', 'nfs4')
        if len(mounts) != 1 or mounts[0]['source'] != expected or mounts[0]['fstype'] != fstype:
            raise BackupError(f'Wrong NAS mount; expected {expected} ({fstype}), got {mounts}')
        free = shutil.disk_usage(self.repo).free
        if free < int(os.environ.get('MIN_FREE_REPO_GB', '5')) * 1024**3:
            raise BackupError('Insufficient free NAS space')

    def manifest(self):
        raw = json.loads(self.cloud('lsjson', self.remote, '--recursive', '--files-only',
                                    '--hash', '--no-mimetype', capture=True))
        result = {}
        for item in raw:
            name = item['Path']
            if (not name or name.startswith('/') or '..' in PurePosixPath(name).parts or
                    '\x00' in name or item.get('IsDir') or item['Size'] < 0 or name in result):
                raise BackupError('Unsafe, duplicate or unsupported cloud entry')
            result[name] = {'size': item['Size'], 'mtime': item.get('ModTime'),
                            'hashes': {k: v for k, v in item.get('Hashes', {}).items() if v}}
        return result

    def verify_local_hashes(self, manifest):
        # Independent validation avoids any size-only fallback chosen inside rclone.
        for name, item in manifest.items():
            path = self.stage / name
            if path.is_symlink() or not path.is_file() or path.stat().st_size != item['size']:
                raise BackupError(f'Staging file missing, unsafe or wrong size: {name}')
            hashes = {key.lower().replace('-', ''): value for key, value in item['hashes'].items()}
            algorithm = next((key for key in ('sha1', 'md5') if hashes.get(key)), None)
            if algorithm:
                digest = hashlib.new(algorithm, usedforsecurity=False)
                with path.open('rb') as stream:
                    while chunk := stream.read(8 * 1024 * 1024):
                        digest.update(chunk)
                if digest.hexdigest().lower() != hashes[algorithm].lower():
                    raise BackupError(f'Independent staging hash mismatch: {name}')
        log('Independent local size/hash verification passed')

    def validate_final_inventory(self, before, after):
        if before == after:
            return
        atomic_json(self.state / 'changed-inventory.json', after)
        # WebDAV checksum properties can appear/disappear without a file changing.
        # Changes to actual paths, sizes or modification times still require a new run.
        metadata = lambda values: {name: (item['size'], item['mtime']) for name, item in values.items()}
        if metadata(before) != metadata(after):
            raise BackupError('Cloud changed during sync/verification (paths, sizes or modification times); retry with a stable inventory')
        changed = {name: after[name] for name in before if before[name]['hashes'] != after[name]['hashes']}
        self.verify_local_hashes(changed)
        missing = [name for name, item in changed.items() if item['size'] and not any(
            key.lower().replace('-', '') in ('sha1', 'md5') for key in item['hashes'])]
        if missing:
            flags = []
            if not any('\n' in name or '\r' in name for name in missing):
                file_list = self.state / 'changed-hash-files.txt'
                file_list.write_text('\n'.join(sorted(missing)) + '\n')
                flags = ['--files-from-raw', str(file_list)]
            self.cloud('check', self.remote, str(self.stage), '--download', *flags)
        log(f'Checksum metadata changed on {len(changed)} files; current contents independently verified')

    @staticmethod
    def totals(manifest):
        return len(manifest), sum(x['size'] for x in manifest.values())

    def guard_source(self, manifest, baseline):
        count, size = self.totals(manifest)
        if count < int(os.environ.get('MIN_SOURCE_FILES', '1')) or size <= 0:
            raise BackupError('Empty or implausibly small cloud source; refusing backup')
        percent = float(os.environ.get('MAX_SOURCE_DROP_PERCENT', '20'))
        if not 0 <= percent <= 100:
            raise BackupError('MAX_SOURCE_DROP_PERCENT must be between 0 and 100')
        for key, current in [('files', count), ('bytes', size)]:
            previous = baseline.get(key, 0)
            if previous and current < previous * (1 - percent / 100):
                raise BackupError(f'Cloud {key} dropped by more than {percent}%: {previous} -> {current}')
        log(f'Cloud inventory: {count} files, {size} bytes')
        return count, size

    def staging_preflight(self, manifest):
        # Check a dedicated staging disk before mkdir or any sync. Without this
        # guard a missing mount could silently put the staging tree on root.
        stage_mount = os.environ.get('EXPECTED_STAGE_MOUNT', '')
        stage_uuid = os.environ.get('EXPECTED_STAGE_UUID', '')
        if bool(stage_mount) != bool(stage_uuid):
            raise BackupError('EXPECTED_STAGE_MOUNT and EXPECTED_STAGE_UUID must be configured together')
        if stage_mount:
            mount_path = Path(stage_mount).absolute()
            if mount_path not in self.stage.parents or mount_path.resolve() != mount_path:
                raise BackupError('Staging must be below the configured staging mount')
            mounts = json.loads(self.run(['findmnt', '--json', '--target', mount_path,
                                         '--output', 'TARGET,UUID,FSTYPE'], capture=True))['filesystems']
            if (len(mounts) != 1 or mounts[0].get('target') != str(mount_path)
                    or mounts[0].get('uuid') != stage_uuid
                    or mounts[0].get('fstype') not in ('ext4', 'btrfs', 'xfs', 'f2fs', 'zfs')):
                raise BackupError('Dedicated staging disk is missing or has the wrong identity')
        # Refuse symlinked, overlapping, mounted or foreign staging paths before rclone sync.
        for other in (self.repo, self.state, Path.home() / 'Nextcloud'):
            if self.stage == other or self.stage in other.parents or other in self.stage.parents:
                raise BackupError(f'Unsafe staging overlap: {other}')
        if self.stage.resolve() != self.stage:
            raise BackupError('Staging path contains a symlink')
        self.stage.mkdir(parents=True, exist_ok=True, mode=0o700)
        os.chmod(self.stage, 0o700)
        for base, dirs, files in os.walk(self.stage):
            for name in dirs + files:
                path = Path(base) / name
                if path.is_symlink() or path.is_mount():
                    raise BackupError(f'Unexpected symlink/mount in staging: {path}')
        if self.stage.is_mount():
            raise BackupError('Staging must be an ordinary local directory')
        owner_file = self.stage.parent / (self.stage.name + '.owner.json')
        expected = {'remote': self.remote, 'stage': str(self.stage)}
        if owner_file.exists():
            if read_json(owner_file) != expected:
                raise BackupError('Staging belongs to another cloud source')
        elif any(self.stage.iterdir()):
            raise BackupError('Nonempty unowned staging directory; refusing sync')
        else:
            atomic_json(owner_file, expected)
        mount = json.loads(self.run(['findmnt', '--json', '--target', self.stage,
                                    '--output', 'FSTYPE'], capture=True))['filesystems'][0]
        if mount['fstype'] not in ('ext4', 'btrfs', 'xfs', 'f2fs', 'zfs', 'tmpfs'):
            raise BackupError(f'Staging must be local; found {mount["fstype"]}')
        total = self.totals(manifest)[1]
        # Conservative full-copy headroom also covers temporary transfer files.
        need = total + int(os.environ.get('MIN_FREE_STAGE_GB', '5')) * 1024**3
        if shutil.disk_usage(self.stage).free < need:
            raise BackupError(f'Insufficient staging space: need {need} bytes free')

    def baseline(self):
        if self.success:
            return self.success
        data = json.loads(self.run([self.borg, 'info', '--json', '--last', '1',
                                   '--glob-archives', os.environ['ARCHIVE_PREFIX'] + '-*',
                                   self.repo], capture=True))
        archives = data.get('archives', [])
        if not archives:
            return {}
        stats = archives[0]['stats']
        return {'files': stats['nfiles'], 'bytes': stats['original_size']}

    def stale(self):
        age = time.time() - self.success.get('completed_at', 0)
        return age < 0 or age > float(os.environ.get('MAX_BACKUP_AGE_HOURS', '36')) * 3600

    def notify(self, message):
        log('WARNING: ' + message)
        stamp = self.state / 'last-notification.json'
        if time.time() - read_json(stamp).get('at', 0) < 6 * 3600:
            return
        if shutil.which('notify-send'):
            result = subprocess.run(['notify-send', '--urgency=critical',
                                     'Nextcloud NAS-Sicherung', message], timeout=15,
                                    stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
            if result.returncode == 0:
                atomic_json(stamp, {'at': time.time()})

    def backup(self, if_due=False):
        with Path(os.environ['LOCK_FILE']).open('a') as lock:
            try:
                fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
            except BlockingIOError:
                log('Another backup/maintenance command holds the lock; skipping')
                return
            age = time.time() - self.success.get('completed_at', 0)
            if if_due and 0 <= age < float(os.environ.get('BACKUP_INTERVAL_HOURS', '24')) * 3600:
                if self.success.get('maintenance_pending') or read_json(self.attempt_file).get('phase') == 'retention':
                    self.nas()
                    self.maintenance()
                    atomic_json(self.attempt_file, {'status': 'success', 'finished_at': time.time()})
                log('Verified backup is still current; no transfer needed')
                return
            attempt = {'started_at': time.time(), 'status': 'running'}
            atomic_json(self.attempt_file, attempt)
            try:
                self.perform()
            except BaseException as exc:
                atomic_json(self.attempt_file, {**attempt, 'finished_at': time.time(),
                                              'status': 'failed', 'phase': self.phase, 'error': str(exc)})
                try:
                    self.notify('Sicherung fehlgeschlagen: ' + str(exc) +
                                (' — Altersgrenze ohne bestätigtes Backup überschritten.' if self.stale() else ''))
                except Exception as notify_error:
                    log(f'Notification unavailable: {notify_error}')
                raise
            atomic_json(self.attempt_file, {**attempt, 'finished_at': time.time(), 'status': 'success'})

    def check_identity(self):
        identity = json.loads(self.run([self.borg, 'info', '--json', self.repo], capture=True))['repository']['id']
        bound = read_json(self.state / 'repository.json')
        if bound and bound != {'repo': str(self.repo), 'id': identity}:
            raise BackupError('Repository identity changed; refusing backup/retention')
        atomic_json(self.state / 'repository.json', {'repo': str(self.repo), 'id': identity})

    def perform(self):
        self.nas()
        self.check_identity()
        excludes = Path(os.environ['EXCLUDE_FILE'])
        if excludes.exists() and any(x.strip() and not x.lstrip().startswith('#')
                                     for x in excludes.read_text().splitlines()):
            raise BackupError('Active legacy Borg excludes found; resolve before full cloud backup')
        baseline = self.baseline()
        self.phase = 'cloud inventory'
        before = self.manifest()
        atomic_json(self.state / 'source-inventory.json', before)
        count, size = self.guard_source(before, baseline)
        self.staging_preflight(before)
        self.phase = 'cloud sync'
        common = {'md5', 'sha1'}
        unhashed = sorted(name for name, item in before.items() if item['size'] and not any(
            key.lower().replace('-', '') in common for key in item['hashes']))
        # rclone's raw file lists cannot encode newline names. Fall back to a full byte read.
        full_download = any('\n' in name or '\r' in name for name in unhashed)
        file_list = self.state / 'unhashed-files.txt'
        if unhashed and not full_download:
            file_list.write_text('\n'.join(unhashed) + '\n')
            log(f'Reading {len(unhashed)} files without server hashes in full')
            self.cloud('copy', self.remote, str(self.stage), '--ignore-times',
                       '--files-from-raw', str(file_list), '--transfers', '4')
        sync_flags = ['--ignore-times'] if full_download else ['--checksum']
        self.cloud('sync', self.remote, str(self.stage), '--check-first', '--delete-after',
                   '--max-delete', os.environ.get('MAX_STAGE_DELETES', '1000'),
                   '--create-empty-src-dirs', '--transfers', '4', '--checkers', '8', *sync_flags)
        self.phase = 'cloud verification'
        self.cloud('check', self.remote, str(self.stage), *(['--download'] if full_download else []))
        if unhashed and not full_download:
            self.cloud('check', self.remote, str(self.stage), '--download',
                       '--files-from-raw', str(file_list))
        self.verify_local_hashes(before)
        after = self.manifest()
        self.validate_final_inventory(before, after)
        atomic_json(self.state / 'verified-inventory.json', after)
        self.nas()
        self.phase = 'borg create'
        archive = os.environ['ARCHIVE_PREFIX'] + '-' + dt.datetime.now().strftime('%Y-%m-%d_%H-%M-%S')
        target = f'{self.repo}::{archive}'
        self.run([self.borg, 'create', '--stats', '--show-rc', '--compression',
                  os.environ['BORG_COMPRESSION'], target, '.'], cwd=self.stage)
        data = json.loads(self.run([self.borg, 'info', '--json', target], capture=True))
        stats = data['archives'][0]['stats']
        if stats['nfiles'] != count or stats['original_size'] != size:
            raise BackupError(f'Archive inventory mismatch: {stats}; expected {count} files/{size} bytes')
        self.phase = 'archive verification'
        self.run([self.borg, 'check', '--archives-only', '--verify-data',
                  '--glob-archives', archive, self.repo])
        self.run([self.borg, 'check', '--repository-only', self.repo])
        # Success means a complete verified archive. Retention failure must not lose this fact.
        self.success = {'completed_at': time.time(), 'archive': archive, 'files': count,
                        'bytes': size, 'repo': str(self.repo), 'remote': self.remote, 'stage': str(self.stage),
                        'archive_id': data['archives'][0]['id'], 'maintenance_pending': True}
        atomic_json(self.success_file, self.success)
        log(f'BACKUP VERIFIED: {archive} ({count} files, {size} bytes)')
        self.maintenance()
        log('Backup and maintenance finished successfully')

    def maintenance(self):
        self.phase = 'retention'
        self.check_identity()
        if self.success.get('archive'):
            archived = json.loads(self.run([self.borg, 'info', '--json',
                                  f'{self.repo}::{self.success["archive"]}'], capture=True))['archives'][0]
            if archived['id'] != self.success['archive_id']:
                raise BackupError('Verified archive identity changed; refusing retention')
        if os.environ.get('PRUNE_AFTER_BACKUP', 'true') == 'true':
            self.run([self.borg, 'prune', '--list', '--show-rc', '--glob-archives',
                      os.environ['ARCHIVE_PREFIX'] + '-*', '--keep-within',
                      os.environ['RETENTION_WITHIN'], '--keep-last', '7', self.repo])
            if os.environ.get('COMPACT_AFTER_PRUNE', 'true') == 'true':
                self.run([self.borg, 'compact', self.repo])
        if self.success:
            self.success['maintenance_pending'] = False
            atomic_json(self.success_file, self.success)
        log('Retention finished successfully')


def main():
    os.umask(0o077)
    runner = Runner()
    def interrupted(signum, _frame):
        runner.stop_child()
        raise BackupError(f'Interrupted by signal {signum}')
    signal.signal(signal.SIGTERM, interrupted)
    signal.signal(signal.SIGINT, interrupted)
    if sys.argv[1:] in (['health'], ['health', '--notify']):
        print(json.dumps({'healthy': not runner.stale(), 'last_success': runner.success,
                          'last_attempt': read_json(runner.attempt_file)}, ensure_ascii=False, indent=2))
        if runner.stale() and '--notify' in sys.argv:
            runner.notify('NAS-Sicherung überfällig: kein bestätigtes Backup innerhalb der Altersgrenze. Protokolle und health prüfen.')
        return 2 if runner.stale() else 0
    if sys.argv[1:] not in (['backup'], ['backup', '--if-due']):
        raise BackupError('Usage: runner backup [--if-due] | health')
    logs = Path(os.environ['LOG_DIR'])
    logs.mkdir(parents=True, exist_ok=True, mode=0o700)
    path = logs / ('backup-' + dt.datetime.now().strftime('%Y%m%d-%H%M%S') + '.log')
    # Journal + file, without an orphaned tee process on timeout/interruption.
    for old in logs.glob('backup-*.log'):
        if old.is_file() and not old.is_symlink() and old.stat().st_mtime < time.time() - 90 * 86400:
            old.unlink()
    with path.open('a', buffering=1) as stream:
        class Tee:
            def write(self, data):
                sys.__stdout__.write(data)
                stream.write(data)
            def flush(self):
                sys.__stdout__.flush()
                stream.flush()
        sys.stdout = Tee()
        # Child output goes to journal; lifecycle/phase/status additionally go to the run log.
        try:
            log('Verified cloud-to-NAS runner 2.0.0, revision 4')
            runner.backup('--if-due' in sys.argv)
        finally:
            sys.stdout = sys.__stdout__
    return 0


if __name__ == '__main__':
    try:
        sys.exit(main())
    except (Exception, KeyboardInterrupt) as error:
        log(f'ERROR: {error}')
        sys.exit(1)