#!/usr/bin/env python3
"""U38 SAGA 独立实验。默认离线；只有显式确认后才启动自己的临时 TC。"""
import datetime
import hashlib
import json
import os
import pathlib
import re
import shutil
import signal
import socket
import subprocess
import sys
import tempfile
import time
import xml.etree.ElementTree as ET
import zipfile

ROOT = pathlib.Path(__file__).resolve().parent
EVIDENCE = ROOT / 'evidence/u38-saga.json'
MARKERS = ('U38_SAGA normal=PASS', 'U38_SAGA compensationFailure=PASS', 'U38_SAGA manualRecovery=PASS')
CASES = {'normalPathPersistsBusinessAndState', 'failedCompensationRemainsVisibleThenManualRetryRestores'}
PINNED_JARS = {
    'target/seata-server.jar': '1eee5700a5651ca6bedb4b36a3c10287244488d6cb0dbb00c60c17fd026c945a',
    'lib/seata-core-2.5.0.jar': '6000b0640a4c0a74614b5e2bd785e1a45a91793af98c93e9307ce05b8f1b1d40',
    'lib/asm-6.0.jar': 'dd8971c74a4e697899a8e95caae4ea8760ea6c486dc6b97b1795e75760420461',
}


def is_port_open():
    try:
        with socket.create_connection(('127.0.0.1', 18091), timeout=1): return True
    except OSError: return False


def now(): return datetime.datetime.now().astimezone().isoformat()


def save_evidence(value):
    def redact(item):
        if isinstance(item, dict): return {k: redact(v) for k, v in item.items()}
        if isinstance(item, list): return [redact(v) for v in item]
        if isinstance(item, str):
            item = re.sub(r'/Users/[^\s"\']+', '<local-path>', item)
            item = re.sub(r'/var/folders/[^\s"\']+', '<temporary-path>', item)
            return re.sub(r'\b(?:\d{1,3}\.){3}\d{1,3}\b',
                          lambda m: m.group() if m.group() == '127.0.0.1' else '<local-ip>', item)
        return item
    EVIDENCE.parent.mkdir(exist_ok=True)
    with tempfile.NamedTemporaryFile('w', encoding='utf8', dir=EVIDENCE.parent, delete=False) as f:
        json.dump(redact(value), f, ensure_ascii=False, indent=2); f.write('\n'); temporary = pathlib.Path(f.name)
    temporary.replace(EVIDENCE)


def files_sha():
    files = [ROOT / 'run.py', ROOT / 'pom.xml', ROOT / 'README.md']
    files += sorted(p for p in (ROOT / 'src').rglob('*') if p.is_file())
    files += sorted(p for p in (ROOT / '../seata/loopback').glob('*') if p.is_file() and p.suffix in {'.py', '.sh', '.java'})
    return {str(p.resolve().relative_to(ROOT.parent.parent)): hashlib.sha256(p.read_bytes()).hexdigest() for p in files}


def validate_jars(env):
    default = pathlib.Path.home() / '.cache/software-engineering-study/java-depth-impl/u38/dist/extracted-bin/apache-seata-2.5.0-incubating-bin/seata-server'
    home = pathlib.Path(env.get('U38_SEATA_SERVER_DIR') or default).expanduser().resolve()
    checked = {}
    for relative, expected in PINNED_JARS.items():
        jar = home / relative
        if not jar.is_file() or hashlib.sha256(jar.read_bytes()).hexdigest() != expected:
            raise RuntimeError('固定版本实验 JAR 缺失或摘要不符: ' + str(jar))
        checked[relative] = expected
    driver = pathlib.Path(env.get('U38_MYSQL_DRIVER') or pathlib.Path.home() / '.m2/repository/com/mysql/mysql-connector-j/9.7.0/mysql-connector-j-9.7.0.jar').expanduser().resolve()
    with zipfile.ZipFile(driver) as archive:
        manifest = archive.read('META-INF/MANIFEST.MF').decode('utf8')
        if ('com/mysql/cj/jdbc/Driver.class' not in archive.namelist()
                or not re.search(r'^Implementation-Version: 9[.]7[.]0\s*$', manifest, re.MULTILINE)):
            raise RuntimeError('MySQL 驱动版本不是 9.7.0')
    env['U38_SEATA_SERVER_DIR'] = str(home)
    env['U38_MYSQL_DRIVER'] = str(driver)
    return {'seataJarsSha256': checked, 'mysqlDriverSha256': hashlib.sha256(driver.read_bytes()).hexdigest(),
            'limits': '仅校验列出 JAR 和驱动，未认证完整发行包其他文件'}


def mysql_query(sql, env):
    default = '/Users/Shared/DBngin/mysql/8.4.7_arm64/bin/mysql'
    client = env.get('U38_SAGA_MYSQL_BIN') or (default if pathlib.Path(default).is_file() else shutil.which('mysql'))
    if not client: raise RuntimeError('无法定位 MySQL 命令行工具；保留所有权记录供人工核查')
    p = subprocess.run([client, '--protocol=tcp', '--host=127.0.0.1', '--port=3306', '--user=root', '--connect-timeout=2', '-N', '-B'],
                       input=sql, text=True, capture_output=True, timeout=15, env=env)
    if p.returncode: raise RuntimeError('MySQL 清理核验失败: ' + p.stderr.strip())
    return p.stdout.strip()


def cleanup_owned(root, env):
    for marker in sorted((root / 'ownership').iterdir()):
        name = marker.name
        if marker.is_symlink() or not marker.is_file() or not re.fullmatch(r'study_u38_saga(?:tc|biz)_[0-9a-f]{32}', name):
            raise RuntimeError('无效所有权文件，保留供人工核查: ' + str(marker))
        token = marker.read_text(encoding='utf8').strip()
        if not re.fullmatch(r'[0-9a-f]{32}', token) or not name.endswith('_' + token):
            raise RuntimeError('所有权令牌不匹配，保留供人工核查: ' + name)
        exists = mysql_query("SELECT COUNT(*) FROM information_schema.schemata WHERE schema_name='" + name + "';", env)
        if exists == '0': continue
        if exists != '1': raise RuntimeError('数据库状态异常，保留: ' + name)
        try: actual = mysql_query('SELECT token FROM `' + name + '`.u38_saga_owner;', env)
        except Exception as e: raise RuntimeError('所有权不明确，保留数据库 ' + name + ': ' + str(e)) from e
        if actual != token: raise RuntimeError('数据库所有权不匹配，保留: ' + name)
        mysql_query('DROP DATABASE `' + name + '`;', env)
    if is_port_open(): raise RuntimeError('18091 端口仍被占用，保留记录供人工核查')


def run_child(args, env, timeout):
    process = subprocess.Popen(args, cwd=ROOT, env=env, start_new_session=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True)
    try:
        output, _ = process.communicate(timeout=timeout)
        # Maven 退出不应遗留属于本次实验的 Surefire 或 TC 子进程。
        try: os.killpg(process.pid, 0)
        except ProcessLookupError: pass
        else:
            try: os.killpg(process.pid, signal.SIGTERM)
            except ProcessLookupError: pass
            time.sleep(1)
            try: os.killpg(process.pid, signal.SIGKILL)
            except ProcessLookupError: pass
            raise RuntimeError('Maven 退出后仍有本次进程组子进程，已尝试关闭并将实验标为失败')
        return process.returncode, output
    except BaseException:
        try: os.killpg(process.pid, signal.SIGTERM)
        except ProcessLookupError: pass
        try: process.communicate(timeout=12)
        except subprocess.TimeoutExpired: pass
        finally:
            try: os.killpg(process.pid, signal.SIGKILL)
            except ProcessLookupError: pass
            process.communicate()
        raise


def main():
    if sys.argv[1:] != ['--allow-start-tc']:
        print('仅在已获用户明确授权后执行: python3 labs/seata-saga/run.py --allow-start-tc', file=sys.stderr); return 2
    if is_port_open():
        print('18091 端口已被占用，未启动任何服务', file=sys.stderr); return 1
    mvn = shutil.which('mvn')
    if not mvn: print('缺少 Maven，未启动任何服务', file=sys.stderr); return 1
    env = os.environ.copy()
    env['MAVEN_OPTS'] = (env.get('MAVEN_OPTS', '') + ' -Xms32m -Xmx256m -XX:ActiveProcessorCount=2').strip()
    env['JAVA_TOOL_OPTIONS'] = '-Xms32m -Xmx192m -XX:ActiveProcessorCount=2 -Dio.netty.eventLoopThreads=2'
    evidence = {'ticket': 'U38 #75 SAGA', 'status': 'running', 'startedAt': now(), 'exitCode': None,
                'artifactSha256': files_sha(), 'commands': [], 'observed': [],
                'limits': ['固定 2.5.0 的 TC 单机进程；agent 仅将监听改为回环',
                           '同一 JVM 手动 compensate 并从数据库 reload，不证明跨进程自动恢复',
                           'sagaBranchRegisterEnable=false；不证明已向 TC 注册 SAGA 分支',
                           '正常/失败/恢复为有界合成业务与本机 MySQL，不代表生产或跨机器结果']}
    root = pathlib.Path(tempfile.mkdtemp(prefix='u38-saga-'))
    (root / 'ownership').mkdir(mode=0o700)
    env['U38_SAGA_WORK_DIR'] = str(root)
    env['U38_SAGA_ALLOW_TC'] = 'yes'
    def interrupt(signum, frame): raise InterruptedError('signal ' + str(signum))
    signal.signal(signal.SIGINT, interrupt); signal.signal(signal.SIGTERM, interrupt)
    exit_code = 1
    try:
        save_evidence(evidence)
        evidence['runtimeJars'] = validate_jars(env)
        evidence['mysqlVersion'] = mysql_query('SELECT VERSION();', env)
        java = shutil.which('java', path=env.get('PATH'))
        if not java: raise RuntimeError('未找到 JDK 21')
        version = subprocess.run([java, '-version'], env=env, capture_output=True, text=True, timeout=15)
        evidence['jdk'] = (version.stdout + version.stderr).strip()
        if version.returncode or not re.search(r'version "21[.\"]', evidence['jdk']):
            raise RuntimeError('需要 JDK 21')
        # 预先离线编译，缺失工件时尚未创建数据库或启动 TC。
        code, output = run_child([mvn, '-o', '-B', '-ntp', 'test-compile'], env, 240)
        evidence['commands'].append({'stage': 'offline-test-compile', 'exitCode': code, 'tail': output[-4000:]})
        if code: raise RuntimeError('离线编译失败；依赖缺失时需获准下载，未启动 TC')
        tree = root / 'dependencies.txt'
        code, output = run_child([mvn, '-o', '-B', '-ntp', 'dependency:tree', '-DoutputFile=' + str(tree)], env, 240)
        evidence['commands'].append({'stage': 'offline-dependency-tree', 'exitCode': code, 'tail': output[-4000:]})
        if code or not tree.is_file(): raise RuntimeError('离线依赖树不可用，未启动 TC')
        dependencies = dict(re.findall(r'(org\.apache\.seata:[\w-]+|org\.springframework:spring-context|'
                                       r'com\.mysql:mysql-connector-j|com\.fasterxml\.jackson\.core:jackson-databind)'
                                       r':jar:([^:\s]+)', tree.read_text(encoding='utf8')))
        saga_modules = {'seata-saga-spring', 'seata-saga-engine', 'seata-saga-engine-store',
                        'seata-saga-rm', 'seata-saga-statelang', 'seata-saga-processctrl'}
        if any(dependencies.get('org.apache.seata:' + name) != '2.5.0' for name in saga_modules) or any(
                version != '2.5.0' for name, version in dependencies.items() if name.startswith('org.apache.seata:')):
            raise RuntimeError('Seata 依赖缺失或版本不一致，未启动 TC')
        evidence['resolvedDependencies'] = dependencies
        start_ns = time.time_ns()
        code, output = run_child([mvn, '-o', '-B', '-ntp', 'test', '-DfailIfNoTests=true'], env, 420)
        evidence['commands'].append({'stage': 'test', 'exitCode': code, 'tail': output[-8000:]})
        if code: raise RuntimeError('测试失败')
        report = ROOT / 'target/surefire-reports/TEST-study.saga.SagaLabTest.xml'
        if not report.is_file() or report.stat().st_mtime_ns < start_ns: raise RuntimeError('Surefire 报告缺失或非本轮')
        suite = ET.parse(report).getroot()
        if (suite.get('name') != 'study.saga.SagaLabTest' or suite.get('tests') != '2' or
                any(suite.get(k) != '0' for k in ('failures', 'errors', 'skipped'))):
            raise RuntimeError('Surefire 失败或数量不符')
        cases = suite.findall('testcase')
        if len(cases) != 2 or {c.get('name') for c in cases} != CASES or any(
                c.find(k) is not None for c in cases for k in ('failure', 'error', 'skipped')):
            raise RuntimeError('Surefire 用例清单异常')
        observed = [line.strip() for line in output.splitlines() if line.startswith('U38_SAGA ')]
        if sorted(observed) != sorted(MARKERS): raise RuntimeError('观察行缺失或重复: ' + repr(observed))
        if files_sha() != evidence['artifactSha256']: raise RuntimeError('实验文件在执行期间发生变化')
        evidence['observed'] = observed
        evidence['testNames'] = sorted(CASES)
        evidence['surefireSha256'] = hashlib.sha256(report.read_bytes()).hexdigest()
        evidence['listener'] = '已用 lsof 核验实验 TC PID 仅在 127.0.0.1:18091 监听'
        evidence['status'] = 'passed'; exit_code = 0
    except BaseException as e:
        evidence['status'] = 'failed'; evidence['failure'] = str(e)
        exit_code = 130 if isinstance(e, (KeyboardInterrupt, InterruptedError)) else 1
        print('实验失败:', e, file=sys.stderr)
    finally:
        signal.signal(signal.SIGINT, signal.SIG_IGN); signal.signal(signal.SIGTERM, signal.SIG_IGN)
        if evidence['status'] != 'passed' and (root / 'tc.log').is_file():
            evidence['tcLogTail'] = (root / 'tc.log').read_text(encoding='utf8', errors='replace')[-12000:]
        try:
            cleanup_owned(root, env)
            shutil.rmtree(root)
            evidence['cleanup'] = '已核验本次数据库及临时目录清理，TC 端口空闲'
        except Exception as e:
            evidence['status'] = 'failed'; evidence['cleanupFailure'] = str(e); evidence['ownershipDir'] = str(root)
            exit_code = 1
            print('清理受阻，保留所有权记录:', root, e, file=sys.stderr)
        evidence['finishedAt'] = now(); evidence['exitCode'] = exit_code
        save_evidence(evidence)
    return exit_code

if __name__ == '__main__': sys.exit(main())