#!/usr/bin/env python3
"""Closed-file H2 upgrades with a durable recovery journal. Hostsharing layout."""
import fcntl
import hashlib
import http.cookiejar
import html
import urllib.parse
import json
import os
from pathlib import Path
import re
import shutil
import shlex
import subprocess
import sys
import tarfile
import tempfile
import time
import urllib.error
import urllib.request
import uuid
import zipfile


def atomic(path, value):
    tmp = path.with_name(path.name + '.next')
    with tmp.open('w') as stream:
        stream.write(value)
        stream.flush()
        os.fsync(stream.fileno())
    tmp.chmod(0o600)
    tmp.replace(path)
    sync_directory(path.parent)


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


def durable_tree(folder):
    for path in folder.rglob('*'):
        if path.is_file():
            with path.open('rb') as stream:
                os.fsync(stream.fileno())
    for path in sorted((p for p in folder.rglob('*') if p.is_dir()), reverse=True):
        sync_directory(path)
    sync_directory(folder)


def replace_database(source, target, displaced):
    """Shared restore core: prepare a durable copy before displacing the closed DB."""
    staged = target.parent / ('restore-data-' + uuid.uuid4().hex)
    try:
        shutil.copytree(source, staged)
        durable_tree(staged)
        if target.exists():
            target.rename(displaced / ('displaced-data-' + uuid.uuid4().hex))
            sync_directory(displaced)
            sync_directory(target.parent)
        staged.rename(target)
        sync_directory(target.parent)
    finally:
        if staged.exists():
            shutil.rmtree(staged)


def link(path, target):
    tmp = path.with_name(path.name + '.next')
    if tmp.is_symlink():
        tmp.unlink()
    tmp.symlink_to(target)
    tmp.replace(path)
    sync_directory(path.parent)


def checksum(path):
    with path.open('rb') as stream:
        return hashlib.file_digest(stream, 'sha256').hexdigest()


def manifest(folder):
    value = json.loads((folder / 'manifest.json').read_text())
    digest = checksum(folder / 'werkjournal.jar')
    if (value.get('schemaVersion') != 1 or value.get('runtime') != 'java-25'
            or not re.fullmatch('[0-9a-f]{40}', value.get('commit', ''))
            or value.get('sha256') != digest or value.get('files') != {'werkjournal.jar': digest}):
        raise ValueError('Invalid release manifest or checksum')
    return value


class Deployment:
    service = 'werkjournal-backend.service'
    url = 'https://werkjournal.javagil.de'

    def __init__(self, home):
        self.home = home
        self.app = home / 'opt/werkjournal'
        self.site = home / 'doms/werkjournal.javagil.de/htdocs-ssl'
        self.flag = self.site / 'maintenance.flag'
        self.pending = self.app / 'pending-deployment.json'

    def command(self, *args):
        return subprocess.check_output(args, text=True, stderr=subprocess.PIPE).strip()

    def stop(self):
        self.command('systemctl', '--user', 'stop', self.service)
        if self.command('systemctl', '--user', 'show', self.service, '--property=ActiveState', '--value') not in {'inactive', 'failed'}:
            raise RuntimeError('Service did not stop; no database copy is safe')
        if self.command('systemctl', '--user', 'show', self.service, '--property=MainPID', '--value') != '0':
            raise RuntimeError('Service still has a process')

    def start(self):
        self.command('systemctl', '--user', 'start', self.service)

    def ready_once(self, expected):
        # Only this loopback probe may transmit Secure cookies over HTTP. Production
        # browser cookies remain Secure; TLS is terminated by the external proxy.
        class LoopbackCookies(http.cookiejar.DefaultCookiePolicy):
            def return_ok_secure(self, cookie, request):
                return request.host == '127.0.0.1:18090'
        jar = http.cookiejar.CookieJar(policy=LoopbackCookies())
        opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(jar))
        base = 'http://127.0.0.1:18090'
        logged_in = False

        def csrf():
            with opener.open(base + '/login', timeout=5) as response:
                page = response.read().decode()
            token = re.search(r'name="_csrf" value="([^"]+)"', page)
            if not token:
                raise ValueError('Login page has no CSRF token')
            return urllib.parse.urlencode({'_csrf': html.unescape(token.group(1))}).encode()

        try:
            with opener.open(base + '/api/v1/info', timeout=3) as response:
                needs_login = urllib.parse.urlsplit(response.url).path == '/login'
                info = None if needs_login else json.load(response)
            if not needs_login:
                if info != {'application': 'werkjournal', 'version': expected['version']}:
                    return False
                with opener.open(base + '/?v-r=init&location=&query=', timeout=5) as response:
                    needs_login = urllib.parse.urlsplit(response.url).path == '/login'
                    if not needs_login:
                        return ('application/json' in response.headers.get('Content-Type', '')
                                and isinstance(json.load(response).get('appConfig'), dict))
            if needs_login:
                with opener.open(base + '/login/guest', data=csrf(), timeout=5) as response:
                    response.read()
                logged_in = True
                try:
                    with opener.open(base + '/api/v1/info', timeout=3) as response:
                        info = json.load(response)
                except urllib.error.HTTPError as error:
                    if error.code != 403:
                        raise
                    # Historical authenticated releases denied this route even to
                    # logged-in users. Verify their packaged version for rollback.
                    with zipfile.ZipFile(self.app / 'current/werkjournal.jar') as jarfile:
                        properties = jarfile.read('META-INF/build-info.properties').decode()
                    version = next(line.removeprefix('build.version=') for line in properties.splitlines()
                                   if line.startswith('build.version='))
                    info = {'application': 'werkjournal', 'version': version}
            if info != {'application': 'werkjournal', 'version': expected['version']}:
                return False
            with opener.open(base + '/?v-r=init&location=&query=', timeout=5) as response:
                return ('application/json' in response.headers.get('Content-Type', '')
                        and isinstance(json.load(response).get('appConfig'), dict))
        finally:
            if logged_in:
                with opener.open(base + '/logout', data=csrf(), timeout=5) as response:
                    response.read()

    def ready(self, expected):
        deadline = time.monotonic() + 90
        while time.monotonic() < deadline:
            try:
                if self.ready_once(expected):
                    return
            except (OSError, ValueError):
                pass
            time.sleep(2)
        raise RuntimeError('Application or Flow initialization failed')

    def preflight(self, check_data=True, allow_restore=False):
        if not allow_restore and (self.app / 'pending-restore.json').exists():
            raise RuntimeError('Pending operator restore: run remote backend restore --recover first')
        if not self.app.is_dir() or not self.site.is_dir():
            raise RuntimeError('Expected Hostsharing directories are missing')
        # The initial installation uses this closed-file database. Refuse runtime
        # overrides until their backup paths have been explicitly configured.
        unit = self.home / '.config/systemd/user/werkjournal-backend.service'
        content = unit.read_text()
        lines = [line.strip() for line in content.splitlines() if line.strip() and not line.startswith(('#', '['))]
        expected_start = f'ExecStart={self.home}/opt/jdk25/bin/java -Xms32m -Xmx384m -XX:MaxMetaspaceSize=192m -XX:ActiveProcessorCount=2 -Djava.awt.headless=true -jar {self.app}/current/werkjournal.jar'
        if ([line for line in lines if line.startswith('WorkingDirectory=')] != [f'WorkingDirectory={self.app}']
                or [line for line in lines if line.startswith('ExecStart=')] != [expected_start]):
            raise RuntimeError('Unexpected service paths or JVM options')
        allowed = {'Description', 'After', 'Type', 'WorkingDirectory', 'ExecStart', 'Environment', 'EnvironmentFile', 'Restart', 'RestartSec', 'TimeoutStopSec', 'UMask', 'WantedBy'}
        if any(line.partition('=')[0] not in allowed for line in lines):
            raise RuntimeError('Unexpected service directives require review')
        expected_environment = {'Environment=BACKEND_PORT=18090', 'Environment=SERVER_ADDRESS=127.0.0.1',
            'Environment=SERVER_FORWARD_HEADERS_STRATEGY=native', 'Environment=SERVER_TOMCAT_REMOTEIP_HOST_HEADER=x-forwarded-host'}
        if set(line for line in lines if line.startswith('Environment=')) != expected_environment:
            raise RuntimeError('Unexpected inline environment')
        if [line for line in lines if line.startswith('EnvironmentFile=')] != ['EnvironmentFile=-%h/.config/werkjournal/environment']:
            raise RuntimeError('Unexpected environment file configuration')
        environment_file = self.home / '.config/werkjournal/environment'
        if environment_file.exists():
            if environment_file.stat().st_mode & 0o077:
                raise RuntimeError('Runtime environment file must be private to its owner')
            for line in environment_file.read_text().splitlines():
                parts = shlex.split(line, comments=True)
                if not parts:
                    continue
                if len(parts) != 1 or '=' not in parts[0]:
                    raise RuntimeError('Unsupported runtime environment file syntax')
                key, value = parts[0].split('=', 1)
                if key in {'WERKJOURNAL_DATABASE_URL', 'SPRING_DATASOURCE_URL'}:
                    if value != 'jdbc:h2:file:./data/werkjournal':
                        raise RuntimeError('Database override is outside the managed backup path')
                elif key == 'BACKEND_PORT':
                    if value != '18090':
                        raise RuntimeError('Unexpected backend port override')
                elif not (key.startswith('WERKJOURNAL_') or key.startswith('SPRING_SECURITY_OAUTH2_CLIENT_')):
                    raise RuntimeError('Unsupported runtime environment key: ' + key)
        for prop, expected in [('DropInPaths', ''), ('NeedDaemonReload', 'no')]:
            if self.command('systemctl', '--user', 'show', self.service, f'--property={prop}', '--value') != expected:
                raise RuntimeError('Unexpected service overrides or pending unit reload')
        environment = self.command('systemctl', '--user', 'show-environment')
        if any(line.startswith(('SPRING_', 'WERKJOURNAL_DATABASE_', 'JAVA_TOOL_OPTIONS=', 'JDK_JAVA_OPTIONS=', '_JAVA_OPTIONS=')) for line in environment.splitlines()):
            raise RuntimeError('Manager environment may override the database configuration')
        if not check_data:
            return
        if not (self.app / 'data/werkjournal.mv.db').is_file():
            raise RuntimeError('Expected H2 file is missing')
        if (self.app / 'data').is_symlink():
            raise RuntimeError('Database directory must not be a symlink')
        for path in (self.app / 'data').rglob('*'):
            if path.is_symlink() or not path.is_file():
                raise RuntimeError('Unexpected database directory layout')

    def stage(self, stream):
        stage = Path(tempfile.mkdtemp(prefix='candidate-', dir=self.app))
        try:
            seen = set()
            with tarfile.open(fileobj=stream, mode='r|') as archive:
                for entry in archive:
                    if entry.name not in {'manifest.json', 'werkjournal.jar'} or entry.name in seen or not entry.isfile() or entry.size > 300 * 1024 * 1024:
                        raise ValueError('Invalid artifact archive')
                    seen.add(entry.name)
                    with archive.extractfile(entry) as source, (stage / entry.name).open('wb') as output:
                        shutil.copyfileobj(source, output)
                        output.flush()
                        os.fsync(output.fileno())
            if len(seen) != 2:
                raise ValueError('Incomplete artifact archive')
            value = manifest(stage)
            release = self.app / 'releases' / value['commit']
            if release.exists():
                if manifest(release) != value:
                    raise ValueError('Release ID already has different contents')
            else:
                stage.rename(release)
                sync_directory(release.parent)
            return value
        finally:
            if stage.exists():
                shutil.rmtree(stage)

    def upload(self, stream):
        self.preflight()
        if self.pending.exists():
            raise RuntimeError('Resolve the pending deployment before uploading another candidate')
        candidate = self.stage(stream)
        return {'result': 'STAGED', 'commit': candidate['commit'],
                'message': 'Candidate stored; deploy is required for activation'}

    def journal(self, state, phase):
        state['phase'] = phase
        atomic(self.pending, json.dumps(state))

    def maintenance(self):
        atomic(self.flag, 'Wartungsmodus — bitte in wenigen Minuten erneut versuchen.\n')
        try:
            urllib.request.urlopen(self.url + '/', timeout=15)
            raise RuntimeError('Public maintenance did not return HTTP 503')
        except urllib.error.HTTPError as error:
            if error.code != 503 or 'Wartungsmodus' not in error.read().decode():
                raise

    def restore(self, state):
        try:
            self._restore(state)
        except BaseException:
            try:
                self.stop()
            except Exception:
                pass
            raise

    def _restore(self, state):
        self.stop()
        bundle = self.app / 'rollbacks' / state['backup']
        if state['phase'] in {'backed_up', 'switched', 'verified'}:
            inventory = json.loads((bundle / 'inventory.json').read_text())
            if 'werkjournal.mv.db' not in inventory or any(Path(name).name != name for name in inventory):
                raise RuntimeError('Invalid backup inventory')
            if any(path.is_symlink() or not path.is_file() for path in (bundle / 'data').iterdir()):
                raise RuntimeError('Invalid backup layout')
            for name, digest in inventory.items():
                if checksum(bundle / 'data' / name) != digest:
                    raise RuntimeError('Backup checksum mismatch; maintenance stays active')
            replace_database(bundle / 'data', self.app / 'data', bundle)
        old = self.app / 'releases' / state['old']['commit']
        if manifest(old) != state['old']:
            raise RuntimeError('Previous application failed verification')
        link(self.app / 'current', old.relative_to(self.app))
        atomic(self.app / 'deployed.json', json.dumps(state['old'], indent=2) + '\n')
        self.start()
        self.ready(state['old'])
        self.journal(state, 'restored')
        self.flag.unlink()
        sync_directory(self.site)
        self.pending.unlink()
        sync_directory(self.app)

    def recover(self):
        self.preflight(check_data=False)
        state = json.loads(self.pending.read_text())
        if (not re.fullmatch('[0-9a-f]{32}', state.get('backup', ''))
                or any(not re.fullmatch('[0-9a-f]{40}', state.get(key, {}).get('commit', '')) for key in ['old', 'new'])):
            raise RuntimeError('Invalid recovery journal')
        if state['phase'] == 'published':
            self.ready(state['new'])
            if self.flag.exists():
                self.flag.unlink()
                sync_directory(self.site)
            self.finalize(state)
            return
        if not self.flag.exists():
            if state['phase'] in {'prepared', 'restored'}:
                self.pending.unlink()
                return
            if state['phase'] != 'verified':
                raise RuntimeError('Unexpected missing maintenance flag; manual inspection required')
            # Publication may have happened before the process died. Never undo
            # writes accepted since then: finish bookkeeping, not rollback.
            self.finalize(state)
            return
        self.restore(state)

    def finalize(self, state):
        link(self.app / 'previous', Path('releases') / state['old']['commit'])
        link(self.app / 'rollback', Path('rollbacks') / state['backup'])
        self.pending.unlink(missing_ok=True)
        sync_directory(self.app)
        # Only deployment-owned UUID bundles and full-commit release directories
        # are eligible; unrelated operator files are never swept.
        for folder, retained, pattern in [
            (self.app / 'rollbacks', {state['backup']}, '[0-9a-f]{32}'),
            (self.app / 'releases', {state['old']['commit'], state['new']['commit']}, '[0-9a-f]{40}')]:
            for path in folder.iterdir():
                if path.name not in retained and re.fullmatch(pattern, path.name) and path.is_dir() and not path.is_symlink():
                    try:
                        shutil.rmtree(path)
                    except OSError:
                        print('Old deployment artifact cleanup deferred', file=sys.stderr)

    def deploy(self, stream):
        self.preflight()
        if self.pending.exists() or self.flag.exists():
            raise RuntimeError('Unresolved maintenance/deployment: use recovery first')
        old = json.loads((self.app / 'deployed.json').read_text())
        if not re.fullmatch('[0-9a-f]{40}', old.get('commit', '')):
            raise RuntimeError('Invalid deployed commit')
        old_release = self.app / 'releases' / old['commit']
        if (self.app / 'current').resolve() != old_release or manifest(old_release) != old:
            raise RuntimeError('Current release differs from its verified manifest')
        new = self.stage(stream)
        if new['commit'] == old['commit']:
            return {'result': 'UNCHANGED', 'commit': new['commit']}
        if old['commit'] not in new.get('ancestors', []):
            raise RuntimeError('Candidate does not descend from the deployed commit')
        state = {'old': old, 'new': new, 'backup': uuid.uuid4().hex}
        self.journal(state, 'prepared')
        try:
            self.maintenance()
            self.stop()
            bundle = self.app / 'rollbacks' / state['backup']
            bundle.mkdir(parents=True, mode=0o700)
            shutil.copytree(self.app / 'data', bundle / 'data')
            inventory = {}
            for path in (bundle / 'data').iterdir():
                with path.open('rb') as stream:
                    os.fsync(stream.fileno())
                inventory[path.name] = checksum(path)
            atomic(bundle / 'inventory.json', json.dumps(inventory))
            atomic(bundle / 'manifest.json', json.dumps(old))
            sync_directory(bundle / 'data')
            sync_directory(bundle.parent)
            self.journal(state, 'backed_up')
            link(self.app / 'current', Path('releases') / new['commit'])
            self.journal(state, 'switched')
            self.start()
            self.ready(new)
            atomic(self.app / 'deployed.json', json.dumps(new, indent=2) + '\n')
            self.journal(state, 'verified')
        except Exception:
            if self.flag.exists():
                self.restore(state)
            raise
        # Durably record publication intent BEFORE exposing the application.
        # Recovery must preserve new writes even if the flag reappears on reboot.
        self.journal(state, 'published')
        self.flag.unlink()
        sync_directory(self.site)
        self.finalize(state)
        return {'result': 'PASS', 'commit': new['commit'], 'url': self.url}


if __name__ == '__main__':
    os.umask(0o077)
    deployment = Deployment(Path.home())
    with (deployment.app / '.deploy.lock').open('a') as lock:
        fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
        if '--upload' in sys.argv:
            print(json.dumps(deployment.upload(sys.stdin.buffer)))
        elif '--recover' in sys.argv:
            deployment.recover()
            print(json.dumps({'result': 'RECOVERED'}))
        else:
            print(json.dumps(deployment.deploy(sys.stdin.buffer)))