#!/usr/bin/env python3
"""WJ-021: real packaged application and H2, production deployment file operations."""
import argparse
import importlib.util
import json
import os
from pathlib import Path
import shutil
import signal
import socket
import subprocess
import sys
import tarfile
import tempfile
import threading
import time
import urllib.error
import urllib.request
import zipfile
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

ROOT = Path(__file__).resolve().parents[1]
spec = importlib.util.spec_from_file_location('deployment', ROOT / 'scripts/deploy-update-remote.py')
deployment = importlib.util.module_from_spec(spec)
spec.loader.exec_module(deployment)


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('jar', type=Path)
    parser.add_argument('--report', type=Path, required=True)
    parser.add_argument('--failures', type=int, choices=[1, 2], default=1)
    parser.add_argument('--fail-rollback', action='store_true')
    parser.add_argument('--failure-worker', action='store_true', help=argparse.SUPPRESS)
    parser.add_argument('--upload-only', action='store_true')
    args = parser.parse_args()
    jar = args.jar.resolve(strict=True)
    java = (Path(os.environ['JAVA_HOME']) / 'bin/java').resolve(strict=True)
    args.report.parent.mkdir(parents=True, exist_ok=True)
    args.report.unlink(missing_ok=True)
    if args.fail_rollback and args.failures != 1:
        parser.error('--fail-rollback requires one candidate')
    if args.upload_only and (args.fail_rollback or args.failures != 1):
        parser.error('--upload-only requires one candidate and no rollback failure')
    if args.fail_rollback and not args.failure_worker:
        output = args.report.with_name(args.report.stem + '-failure.log')
        with output.open('wb') as log:
            worker = subprocess.Popen([sys.executable, str(Path(__file__).resolve()), str(jar),
                '--report', str(args.report), '--fail-rollback', '--failure-worker'],
                stdout=log, stderr=subprocess.STDOUT, start_new_session=True)
            try:
                # Above the readiness waits this worker performs, or raising those would only
                # move the failure here, where it reads as a hung fixture instead of a slow host.
                worker.wait(timeout=600)
            finally:
                if worker.poll() is None:
                    os.killpg(worker.pid, signal.SIGTERM)
                    try:
                        worker.wait(timeout=5)
                    finally:
                        # The worker and every JVM share this private process group.
                        try:
                            os.killpg(worker.pid, signal.SIGKILL)
                        except ProcessLookupError:
                            pass
                        worker.wait()
        assert worker.returncode != 0, 'Failed rollback worker incorrectly returned success'
        failure_log = output.read_text()
        assert 'Expected candidate migration failure' in failure_log
        assert 'Expected rollback readiness failure' in failure_log
        report = json.loads(args.report.read_text())
        assert report['result'] == 'EXPECTED_FAILURE'
        assert report['maintenanceRetained'] and report['journalRetained'] and report['rollbackStopped']
        report.update(result='PASS', failureExitCode=worker.returncode, bothFailuresLogged=True)
        args.report.write_text(json.dumps(report, indent=2) + '\n')
        print(json.dumps(report))
        return
    logs = args.report.parent / (args.report.stem + '-logs')
    logs.mkdir(exist_ok=True)
    with tempfile.TemporaryDirectory(prefix='werkjournal-rollback-') as temporary:
        home = Path(temporary)
        app = home / 'opt/werkjournal'
        app.mkdir(parents=True)
        with zipfile.ZipFile(jar) as archive:
            h2_member, = [name for name in archive.namelist() if name.startswith('BOOT-INF/lib/h2-')]
            h2 = home / 'h2.jar'
            h2.write_bytes(archive.read(h2_member))
            properties = archive.read('META-INF/build-info.properties').decode()
            version = next(line.split('=', 1)[1] for line in properties.splitlines() if line.startswith('build.version='))
        old = 'a' * 40
        candidates = [letter * 40 for letter in 'bc'[:args.failures]]
        for commit in [old, *candidates]:
            release = app / 'releases' / commit
            release.mkdir(parents=True)
            target = release / 'werkjournal.jar'
            if commit == old:
                shutil.copyfile(jar, target)
            else:
                with zipfile.ZipFile(jar) as source, zipfile.ZipFile(target, 'w') as dest:
                    for entry in source.infolist():
                        data = source.read(entry.filename)
                        if entry.filename == 'BOOT-INF/classes/db/changelog.sql':
                            data += b'\n--changeset rollback-test:partial-failure dbms:h2 runInTransaction:false\nCREATE TABLE rollback_probe (id INTEGER PRIMARY KEY);\nINSERT INTO rollback_probe VALUES (1);\nSELECT * FROM intentionally_missing_rollback_table;\n'
                        dest.writestr(entry, data)
            digest = deployment.checksum(target)
            manifest = dict(schemaVersion=1, runtime='java-25', version=version, commit=commit,
                            sha256=digest, files={'werkjournal.jar': digest}, ancestors=[] if commit == old else [old])
            (release / 'manifest.json').write_text(json.dumps(manifest))
        (app / 'current').symlink_to(Path('releases') / old)
        shutil.copyfile(app / 'current/manifest.json', app / 'deployed.json')
        packages = {}
        for candidate in candidates:
            package = home / f'{candidate}.tar'
            with tarfile.open(package, 'w') as archive:
                for name in ['werkjournal.jar', 'manifest.json']:
                    archive.add(app / 'releases' / candidate / name, arcname=name)
            packages[candidate] = package
            # stage() must actually unpack and validate the candidate.
            shutil.rmtree(app / 'releases' / candidate)
        env = {'PATH': '/usr/bin:/bin', 'LANG': 'C.UTF-8', 'HOME': str(home),
               'WERKJOURNAL_GUEST_MODE': 'true', 'WERKJOURNAL_PUSH_ENABLED': 'false'}
        trap = home / 'sendmail'
        trap.write_text('#!/bin/sh\ntouch "' + str(home / 'unexpected-mail') + '"\nexit 1\n')
        trap.chmod(0o700)
        env['WERKJOURNAL_SENDMAIL'] = str(trap)
        with socket.socket() as sock:
            sock.bind(('127.0.0.1', 0))
            port = sock.getsockname()[1]
        base = f'http://127.0.0.1:{port}'

        def sql(statement):
            result = subprocess.run([str(java), '-cp', str(h2), 'org.h2.tools.Shell', '-url',
                'jdbc:h2:file:' + str(app / 'data/werkjournal'), '-user', 'sa', '-password', '',
                '-sql', statement], env=env, capture_output=True, text=True, timeout=60, check=True)
            if 'Error:' in result.stdout:
                raise AssertionError(result.stdout)
            return result.stdout

        class LocalDeployment(deployment.Deployment):
            process = None
            log = None
            starts = 0
            migration_failed = False
            matched_backup = False
            ready_with_maintenance = False
            blocked_port = None

            def preflight(self, check_data=True):
                # systemd configuration is covered by operations tests; this fixture owns its process.
                assert self.home == home and self.app == app
                assert (self.app / 'data/werkjournal.mv.db').is_file()

            def stop(self):
                if self.process is not None:
                    if self.process.poll() is None:
                        self.process.terminate()
                        try:
                            # Generous, because this runs beside the browser suites on a shared
                            # host: the same release closed its database in 0.2 s when it had the
                            # machine to itself. Only a starved host waits anywhere near this.
                            self.process.wait(timeout=60)
                        except subprocess.TimeoutExpired:
                            self.process.kill()
                            self.process.wait(timeout=20)
                            raise AssertionError('Application failed to close its database gracefully')
                    assert self.process.poll() is not None
                    self.log.close()
                    self.process = None

            def start(self):
                if self.migration_failed:
                    assert (self.app / 'current').resolve().name == old
                    state = json.loads(self.pending.read_text())
                    inventory = json.loads((self.app / 'rollbacks' / state['backup'] / 'inventory.json').read_text())
                    assert all(deployment.checksum(self.app / 'data' / name) == digest for name, digest in inventory.items())
                    self.matched_backup = True
                    if args.fail_rollback:
                        self.blocked_port = socket.socket()
                        self.blocked_port.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
                        self.blocked_port.bind(('127.0.0.1', port))
                        self.blocked_port.listen()
                self.starts += 1
                self.logpath = logs / f'start-{self.starts}.log'
                self.log = self.logpath.open('wb')
                self.process = subprocess.Popen([str(java), '-Xmx256m', '-XX:ActiveProcessorCount=2',
                    '-jar', str(self.app / 'current/werkjournal.jar'), '--server.address=127.0.0.1',
                    f'--server.port={port}'], cwd=self.app, env=env, stdout=self.log, stderr=subprocess.STDOUT)

            def ready(self, expected):
                # Three minutes for a start that takes one to three seconds on an idle machine.
                # On 2026-10-12 this host needed 17.1 s for one start and had not finished the
                # migration of the next after 25 s, with Firefox and its own applications on the
                # same three threads; the branch build failed with nothing wrong in the release.
                # The proof is unchanged: the application must become ready, and this only grants
                # a starved host the time to get there.
                deadline = time.monotonic() + 180
                while time.monotonic() < deadline:
                    if self.process.poll() is not None:
                        if args.fail_rollback and self.migration_failed and expected['commit'] == old:
                            assert self.process.returncode != 0
                            assert 'already in use' in self.logpath.read_text().lower()
                            raise RuntimeError('Expected rollback readiness failure')
                        assert expected['commit'] in candidates and self.process.returncode != 0
                        assert 'intentionally_missing_rollback_table' in self.logpath.read_text()
                        assert 'PARTIAL_MIGRATION_CONFIRMED' in sql("SELECT 'PARTIAL_MIGRATION_CONFIRMED' FROM rollback_probe WHERE id=1")
                        self.migration_failed = True
                        raise RuntimeError('Expected candidate migration failure')
                    try:
                        with urllib.request.urlopen(base + '/api/v1/info', timeout=2) as response:
                            assert json.load(response) == {'application': 'werkjournal', 'version': version}
                        with urllib.request.urlopen(base + '/?v-r=init&location=&query=', timeout=3) as response:
                            assert isinstance(json.load(response).get('appConfig'), dict)
                        if self.migration_failed:
                            assert self.flag.exists()
                            try:
                                urllib.request.urlopen(self.url, timeout=2)
                                raise AssertionError('Public access opened before readiness')
                            except urllib.error.HTTPError as error:
                                assert error.code == 503
                            self.ready_with_maintenance = True
                        return
                    except (urllib.error.URLError, TimeoutError):
                        time.sleep(.2)
                raise AssertionError('Application readiness timed out')

        runner = LocalDeployment(home)
        runner.site.mkdir(parents=True)

        class Gateway(BaseHTTPRequestHandler):
            def do_GET(self):
                body = runner.flag.read_bytes() if runner.flag.exists() else b'public access available'
                self.send_response(503 if runner.flag.exists() else 200)
                self.end_headers()
                self.wfile.write(body)

            def log_message(self, *unused):
                pass

        gateway = ThreadingHTTPServer(('127.0.0.1', 0), Gateway)
        threading.Thread(target=gateway.serve_forever, daemon=True).start()
        runner.url = f'http://127.0.0.1:{gateway.server_port}'
        try:
            runner.start()
            runner.ready({'commit': old})
            runner.stop()
            ids = [f'10000000-0000-0000-0000-{n:012d}' for n in range(1, 5)]
            user, customer, project, entry = ids
            sql(f"INSERT INTO app_user(id,display_name) VALUES('{user}','Rollback fixture');"
                f"INSERT INTO customer(id,owner_id,name) VALUES('{customer}','{user}','Rollback customer');"
                f"INSERT INTO work_project(id,owner_id,customer_id,name) VALUES('{project}','{user}','{customer}','Rollback project');"
                f"INSERT INTO work_entry(id,owner_id,customer_id,project_id,work_date,hours,description) VALUES('{entry}','{user}','{customer}','{project}',DATE '2026-09-01',3,'Preserve this booking');")
            runner.start()
            runner.ready({'commit': old})
            if args.upload_only:
                client = home / 'client'
                (client / 'tools').mkdir(parents=True)
                (client / 'scripts').mkdir()
                for name in ['tools/remote', 'scripts/lib-env.sh', 'scripts/deploy-prototype.py',
                             'scripts/deploy-update-remote.py']:
                    shutil.copy2(ROOT / name, client / name)
                artifact = client / 'artifact'
                artifact.mkdir()
                with tarfile.open(packages[candidates[0]]) as archive:
                    for name in ['manifest.json', 'werkjournal.jar']:
                        with archive.extractfile(name) as source, (artifact / name).open('wb') as target:
                            shutil.copyfileobj(source, target)
                transport = home / 'transport'
                transport.mkdir()
                captured = transport / 'upload.tar'
                ssh = transport / 'ssh'
                ssh.write_text("#!/usr/bin/env python3\nimport pathlib, shlex, sys\n"
                    "assert sys.argv[1:4] == ['-o', 'BatchMode=yes', 'fixture@invalid']\n"
                    "assert len(sys.argv) == 5 and shlex.split(sys.argv[4])[-1] == '--upload'\n"
                    "target = pathlib.Path(__file__).with_name('upload.tar')\n"
                    "with target.open('xb') as output:\n"
                    " import shutil\n shutil.copyfileobj(sys.stdin.buffer, output)\n")
                ssh.chmod(0o700)
                client_env = dict(env, PATH=str(transport) + ':/usr/bin:/bin', BACKEND_SSH='fixture@invalid')
                active_pid = runner.process.pid
                selected = (app / 'current').readlink()
                deployed_bytes = (app / 'deployed.json').read_bytes()
                subprocess.run([str(client / 'tools/remote'), 'backend', 'upload', '-y',
                    '--artifact', str(artifact)], env=client_env, cwd=client,
                    check=True, capture_output=True, timeout=30)
                # Execute the real server upload path on the exact bytes sent by remote.
                with captured.open('rb') as stream:
                    uploaded = runner.upload(stream)
                assert uploaded['commit'] == candidates[0]
                assert deployment.manifest(app / 'releases' / candidates[0]) == json.loads((artifact / 'manifest.json').read_text())
                assert runner.process.pid == active_pid and runner.process.poll() is None
                assert (app / 'current').readlink() == selected
                assert (app / 'deployed.json').read_bytes() == deployed_bytes
                assert not runner.flag.exists() and not runner.pending.exists()
                # Use the actual restart control with only the process adapter replaced.
                controls = dict(vars(deployment), __name__='service_control_fixture')
                exec(compile((ROOT / 'scripts/service-control-remote.py').read_text(),
                             'service-control-remote.py', 'exec'), controls)
                assert controls['control'](runner, 'restart') == 0
                assert runner.process.pid != active_pid and runner.process.poll() is None
                assert (app / 'current').readlink() == selected
                assert (app / 'deployed.json').read_bytes() == deployed_bytes
                assert not runner.flag.exists() and not runner.pending.exists()
                runner.stop()
                assert 'BOOKING_PRESERVED' in sql(f"SELECT 'BOOKING_PRESERVED' FROM work_entry WHERE id='{entry}' AND hours=3")
                assert 'CANDIDATE_NOT_RUN' in sql("SELECT 'CANDIDATE_NOT_RUN' WHERE NOT EXISTS(SELECT 1 FROM information_schema.tables WHERE table_name='ROLLBACK_PROBE')")
                assert not (home / 'unexpected-mail').exists()
                report = dict(result='PASS', uploadedCommit=candidates[0], activeCommit=old,
                    uploadKeptProcess=True, restartKeptRelease=True, bookingPreserved=True,
                    candidateMigrationNotRun=True, mailInvocations=0, pushDisabled=True)
                args.report.write_text(json.dumps(report, indent=2) + '\n')
                print(json.dumps(report))
                return
            rounds = []
            failed_rollback = None
            for index, candidate in enumerate(candidates):
                runner.migration_failed = runner.matched_backup = runner.ready_with_maintenance = False
                with packages[candidate].open('rb') as archive:
                    try:
                        runner.deploy(archive)
                        raise AssertionError('Failed migration was reported as successful')
                    except RuntimeError as error:
                        if args.fail_rollback:
                            assert str(error) == 'Expected rollback readiness failure'
                            assert str(error.__context__) == 'Expected candidate migration failure'
                            failed_rollback = error
                        else:
                            assert str(error) == 'Expected candidate migration failure'
                if args.fail_rollback:
                    assert runner.migration_failed and runner.matched_backup
                    assert runner.process is None, 'Failed rollback left a running process'
                    assert runner.flag.exists() and runner.pending.exists()
                    state = json.loads(runner.pending.read_text())
                    assert state['old']['commit'] == old and state['new']['commit'] == candidate
                    assert state['phase'] == 'switched'
                    assert (app / 'current').resolve().name == old
                    assert 'BOOKING_PRESERVED' in sql(f"SELECT 'BOOKING_PRESERVED' FROM work_entry WHERE id='{entry}' AND hours=3")
                    try:
                        urllib.request.urlopen(runner.url, timeout=2)
                        raise AssertionError('Failed rollback exposed the application')
                    except urllib.error.HTTPError as error:
                        assert error.code == 503 and 'Wartungsmodus' in error.read().decode()
                    break
                assert runner.migration_failed and runner.matched_backup and runner.ready_with_maintenance
                assert not runner.flag.exists() and not runner.pending.exists()
                assert json.loads((app / 'deployed.json').read_text())['commit'] == old
                assert (app / 'current').resolve().name == old
                with urllib.request.urlopen(runner.url, timeout=2) as response:
                    assert response.status == 200
                runner.stop()
                assert 'BOOKING_PRESERVED' in sql(f"SELECT 'BOOKING_PRESERVED' FROM work_entry WHERE id='{entry}' AND hours=3 AND description='Preserve this booking'")
                assert 'SCHEMA_RESTORED' in sql("SELECT 'SCHEMA_RESTORED' WHERE NOT EXISTS(SELECT 1 FROM information_schema.tables WHERE table_name='ROLLBACK_PROBE')")
                assert 'HISTORY_RESTORED' in sql("SELECT 'HISTORY_RESTORED' WHERE NOT EXISTS(SELECT 1 FROM databasechangelog WHERE author='rollback-test')")
                if index:
                    assert 'INTERMEDIATE_BOOKING_PRESERVED' in sql("SELECT 'INTERMEDIATE_BOOKING_PRESERVED' FROM work_entry WHERE description='Written after first rollback' AND hours=2")
                rounds.append(dict(candidate=candidate, restoredCommit=old, bookingPreserved=True,
                                   intermediateBookingPreserved=index > 0))
                if index + 1 < len(candidates):
                    # New data after B's rollback must survive C as well. Reusing
                    # the first deployment's snapshot would lose this entry.
                    sql(f"INSERT INTO work_entry(id,owner_id,customer_id,project_id,work_date,hours,description) VALUES('10000000-0000-0000-0000-000000000005','{user}','{customer}','{project}',DATE '2026-09-02',2,'Written after first rollback')")
                    runner.migration_failed = False
                    runner.start()
                    runner.ready({'commit': old})
            assert not (home / 'unexpected-mail').exists()
            if args.fail_rollback:
                args.report.write_text(json.dumps(dict(result='EXPECTED_FAILURE', maintenanceRetained=True,
                    journalRetained=True, rollbackStopped=True, bookingPreserved=True,
                    mailInvocations=0, pushDisabled=True)) + '\n')
                assert failed_rollback is not None
                raise failed_rollback
            report = dict(result='PASS', partialMigration=True, candidateStopped=True,
                          matchingBackupRestored=True, bookingPreserved=True, schemaRestored=True,
                          maintenanceUntilReady=True, mailInvocations=0, pushDisabled=True, rounds=rounds)
            args.report.write_text(json.dumps(report, indent=2) + '\n')
            print(json.dumps(report))
        finally:
            runner.stop()
            if runner.blocked_port is not None:
                runner.blocked_port.close()
            gateway.shutdown()
            gateway.server_close()


if __name__ == '__main__':
    main()