#!/usr/bin/env python3
"""U46 公开入口：真实 MySQL 上运行 Spring Batch 6.0.6 与 Quartz 2.5.2 的持久任务实验。

前置（不满足则明确受阻退出，不伪造通过）：
- 本机可连接的 MySQL 8.x（默认 127.0.0.1:3306 root 无密码；可用 LAB_DB_* 覆盖）
- JDK 21（JAVA_HOME 或 /usr/libexec/java_home -v 21）与 Maven 3.9.x
- Maven 依赖已在本机仓库缓存（入口以 -o 离线运行；首次准备可先在线 test-compile）

每次运行在 MySQL 上创建 study_u46_<runid>_<uuid> 独有库（每个测试一个），
结束时测试 fixture 自删；runner 再核对无残留。不触碰用户既有库与其他会话的库。
无论成功失败都写 evidence/u46-2026-10-31.json（含 exitCode、测试名、观察行与清理结果）。
"""
import glob
import hashlib
import json
import os
import pathlib
import platform
import re
import shutil
import signal
import subprocess
import sys
import tempfile
import time
import xml.etree.ElementTree as ET
from datetime import datetime

ROOT = pathlib.Path(__file__).resolve().parent
EVIDENCE = ROOT / 'evidence' / 'u46-2026-10-31.json'
MYSQL_BIN_CANDIDATES = [
    os.environ.get('MYSQL_BIN'),
    shutil.which('mysql'),
    '/Users/Shared/DBngin/mysql/8.4.7_arm64/bin/mysql',
    '/opt/homebrew/bin/mysql',
    '/usr/local/bin/mysql',
    '/usr/local/mysql/bin/mysql',
]
EXPECTED_TESTS = {
    'study.batchquartz.BatchPersistenceTest': {
        'jobIdentityIsNamePlusIdentifyingParameters',
        'failedChunkRestartResumesFromCheckpointAndReplaysUncommittedWork',
        'skipLimitWithoutFaultTolerantDoesNotSkip',
        'faultTolerantSkipsBadItemAndKeepsOthers',
        'abandonMakesExecutionUnrestartableAndRecoverUnsticksStarted',
        'completedStepIsSkippedOnRestartUnlessAllowStartIfComplete',
        'concurrentLaunchOfSameInstanceOnlyOneWins',
    },
    'study.batchquartz.QuartzPersistenceTest': {
        'jdbcStorePersistsAcrossSchedulerRestart',
        'clusteredSchedulersFireOnceThroughRowLock',
        'nonClusteredModeLeavesOrphanedTriggersStuck',
        'misfireCronFireOnceNowVersusDoNothing',
        'misfireSimpleSmartPolicyReschedulesNotCatchUp',
        'clusterTakesOverAfterInstanceDies',
        'failingJobIsNotRetriedByQuartz',
        'disallowConcurrentBlocksRatherThanRunsParallel',
        'shutdownWaitAllowsRunningJobToFinish',
    },
    'study.batchquartz.BootAutoConfigurationTest': {
        'batchDefaultsToResourcelessRepositoryWithoutDataSource',
        'batchWithDataSourceUsesJdbcRepositoryAndCreatesSchema',
        'quartzDefaultsToMemoryStoreAndJdbcRequiresExplicitProperty',
    },
}
EXPECTED_VERSIONS = {
    'org.springframework.boot:spring-boot': '4.0.9',
    'org.springframework.batch:spring-batch-core': '6.0.6',
    'org.springframework.batch:spring-batch-infrastructure': '6.0.6',
    'org.quartz-scheduler:quartz': '2.5.2',
    'org.springframework:spring-jdbc': '7.0.10',
    'com.mysql:mysql-connector-j': '9.7.0',
    'org.junit.jupiter:junit-jupiter': '6.0.3',
}
OBSERVATION_PREFIX = 'U46 '
MIN_OBSERVATIONS = 16


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


def write_json(path, value):
    path.parent.mkdir(parents=True, exist_ok=True)
    temporary = None
    try:
        with tempfile.NamedTemporaryFile(mode='w', encoding='utf-8', dir=path.parent,
                                         prefix=path.name + '.', delete=False) as stream:
            temporary = pathlib.Path(stream.name)
            json.dump(value, stream, ensure_ascii=False, indent=2)
            stream.write('\n')
        temporary.replace(path)
    finally:
        if temporary is not None:
            temporary.unlink(missing_ok=True)


def file_sha(path):
    return hashlib.sha256(path.read_bytes()).hexdigest()


def artifact_digests():
    sha = {}
    for name in ('pom.xml', 'run.py', 'README.md'):
        target = ROOT / name
        if target.exists():
            sha[name] = file_sha(target)
    for path in sorted((ROOT / 'src').rglob('*.java')):
        sha[str(path.relative_to(ROOT))] = file_sha(path)
    return sha


def java_home():
    if os.environ.get('JAVA_HOME'):
        return pathlib.Path(os.environ['JAVA_HOME'])
    if sys.platform == 'darwin':
        return pathlib.Path(subprocess.check_output(
            ['/usr/libexec/java_home', '-v', '21'], text=True).strip())
    raise SystemExit('需要 JDK 21：设置 JAVA_HOME')


def maven_binary():
    if os.environ.get('MAVEN_BIN'):
        return os.environ['MAVEN_BIN']
    if shutil.which('mvn'):
        return 'mvn'
    raise SystemExit('需要 Maven 3.9.x：设置 MAVEN_BIN 或在 PATH 中提供 mvn')


def mysql_binary():
    for candidate in MYSQL_BIN_CANDIDATES:
        if candidate and pathlib.Path(candidate).is_file():
            return candidate
    return None


def leftover_databases(mysql):
    if mysql is None:
        return None
    output = subprocess.run([mysql, '-h', os.environ.get('LAB_DB_HOST', '127.0.0.1'),
                             '-P', os.environ.get('LAB_DB_PORT', '3306'),
                             '-u', os.environ.get('LAB_DB_USER', 'root'), '-N', '-B',
                             '-e', "SELECT schema_name FROM information_schema.schemata"
                                   " WHERE schema_name LIKE 'study_u46%'"],
                            capture_output=True, text=True, timeout=30)
    if output.returncode != 0:
        return None
    return [line.strip() for line in output.stdout.splitlines() if line.strip()]


def live_group_pids(pgid):
    snapshot = subprocess.run(['/bin/ps', '-axo', 'pid=,pgid=,stat='],
                              capture_output=True, text=True, check=True, timeout=5)
    members = []
    for line in snapshot.stdout.splitlines():
        pid, group, state = line.split()
        if int(group) == pgid and not state.startswith('Z'):
            members.append(int(pid))
    return members


def cleanup_group(process):
    for signum, grace in ((signal.SIGTERM, 2), (signal.SIGKILL, 5)):
        process.poll()
        if not live_group_pids(process.pid):
            return
        try:
            os.killpg(process.pid, signum)
        except ProcessLookupError:
            return
        deadline = time.monotonic() + grace
        while time.monotonic() < deadline:
            process.poll()
            if not live_group_pids(process.pid):
                return
            time.sleep(0.05)
    remaining = live_group_pids(process.pid)
    if remaining:
        raise RuntimeError('进程组仍有存活进程: ' + str(remaining))


def run(name, args, logs, env, results, timeout=1800):
    log_path = logs / (name + '.log')
    started = now()
    process = None
    failure = None
    cleanup_error = None
    exit_code = 0
    with log_path.open('w', encoding='utf-8') as stream:
        try:
            process = subprocess.Popen(args, cwd=ROOT, env=env, text=True,
                                       stdout=stream, stderr=subprocess.STDOUT,
                                       start_new_session=True)
            code = process.wait(timeout=timeout)
            exit_code = code if code >= 0 else 128 - code
        except BaseException as exc:
            failure = exc
            if isinstance(exc, subprocess.TimeoutExpired):
                exit_code = 124
            elif isinstance(exc, KeyboardInterrupt):
                exit_code = 130
            else:
                exit_code = 1
        finally:
            handlers = {sig: signal.signal(sig, signal.SIG_IGN)
                        for sig in (signal.SIGINT, signal.SIGTERM)}
            try:
                if process is not None:
                    cleanup_group(process)
            except Exception as exc:
                cleanup_error = str(exc)
                exit_code = exit_code or 1
            finally:
                for sig, handler in handlers.items():
                    signal.signal(sig, handler)
    output = log_path.read_text(encoding='utf-8', errors='replace') if log_path.exists() else ''
    result = {
        'name': name, 'command': args, 'cwd': str(ROOT),
        'startedAt': started, 'finishedAt': now(),
        'exitCode': process.returncode if process is not None else None,
        'runnerExitCode': exit_code,
        'timedOut': isinstance(failure, subprocess.TimeoutExpired),
        'failure': str(failure) if failure is not None else None,
        'cleanupError': cleanup_error,
        'log': str(log_path),
        'logSha256': file_sha(log_path) if log_path.exists() else None,
        'observed': [line for line in output.splitlines() if line.startswith(OBSERVATION_PREFIX)],
    }
    results.append(result)
    write_json(logs / (name + '.status.json'), result)
    if exit_code:
        print(name + ' exit=' + str(exit_code) + '; raw log: ' + str(log_path)
              + ('; cleanup: ' + cleanup_error if cleanup_error else '')
              + '\n' + output[-3500:], file=sys.stderr, flush=True)
        raise SystemExit(exit_code)
    print('DONE ' + name + ' exit=0; log: ' + str(log_path), flush=True)
    return output


def validate_reports(logs, not_before_ns):
    report_dir = ROOT / 'target/surefire-reports'
    paths = sorted(report_dir.glob('TEST-*.xml'))
    expected_files = sorted('TEST-' + name + '.xml' for name in EXPECTED_TESTS)
    if [path.name for path in paths] != expected_files:
        raise ValueError('Surefire 报告缺失或多出报告: ' + str([path.name for path in paths]))
    reports = []
    for path in paths:
        if path.stat().st_mtime_ns < not_before_ns:
            raise ValueError('Surefire 报告不是本次测试生成: ' + str(path))
        suite = ET.parse(path).getroot()
        if suite.tag != 'testsuite':
            raise ValueError('Surefire testsuite 根节点错误: ' + path.name)
        class_name = suite.get('name')
        expected = EXPECTED_TESTS.get(class_name)
        if expected is None:
            raise ValueError('未登记的测试类: ' + str(class_name))
        report = {key: suite.attrib.get(key) for key in ('name', 'tests', 'failures', 'errors', 'skipped')}
        for field in ('failures', 'errors', 'skipped'):
            if int(report[field]) != 0:
                raise ValueError(class_name + ' ' + field + '=' + report[field])
        cases = suite.findall('testcase')
        names = [case.attrib['name'] for case in cases]
        if len(set(names)) != len(names) or set(names) != expected:
            raise ValueError(class_name + ' 测试名/数量不匹配: ' + str(sorted(names)))
        for case in cases:
            if case.get('classname') != class_name or any(
                    case.find(tag) is not None for tag in ('failure', 'error', 'skipped')):
                raise ValueError('Surefire 用例失败、跳过或类名不匹配: ' + case.attrib['name'])
        report['testNames'] = sorted(names)
        report['sha256'] = file_sha(path)
        reports.append(report)
    total = sum(int(report['tests']) for report in reports)
    expected_total = sum(len(names) for names in EXPECTED_TESTS.values())
    if total != expected_total:
        raise ValueError('测试总数不匹配: ' + str(total) + ' != ' + str(expected_total))
    write_json(logs / 'reports.json', reports)
    return reports


def validate_observations(observed):
    markers = [line for line in observed if line.startswith(OBSERVATION_PREFIX)]
    if len(markers) != len(set(markers)):
        duplicates = sorted({line for line in markers if markers.count(line) > 1})
        raise ValueError('观察行重复: ' + str(duplicates))
    if len(markers) < MIN_OBSERVATIONS:
        raise ValueError('观察行过少: ' + str(len(markers)) + ' < ' + str(MIN_OBSERVATIONS))
    return markers


def read_dependencies(path):
    versions = {}
    for line in path.read_text(encoding='utf-8').splitlines():
        fields = line.strip().split(' ', 1)[0].split(':')
        if len(fields) not in (5, 6) or fields[2] not in ('jar', 'pom', 'test-jar'):
            continue
        if fields[-1] not in ('compile', 'provided', 'runtime', 'test', 'system'):
            continue
        coordinate = ':'.join(fields[:2])
        version = fields[-2]
        if coordinate in versions and versions[coordinate] != version:
            raise ValueError('依赖版本冲突: ' + coordinate)
        versions[coordinate] = version
    for coordinate, expected in EXPECTED_VERSIONS.items():
        if versions.get(coordinate) != expected:
            raise ValueError('依赖版本不匹配: ' + coordinate + '=' + str(versions.get(coordinate))
                             + '; expected=' + expected)
    return versions


def main():
    logs = None
    evidence = {
        'ticket': 'U46 #83', 'status': 'running',
        'startedAt': now(), 'finishedAt': None,
        'input': 'python3 labs/batch-quartz/run.py', 'exitCode': None,
        'environment': {'os': platform.platform(), 'arch': platform.machine(),
                        'python': platform.python_version()},
        'expectedTestCount': sum(len(names) for names in EXPECTED_TESTS.values()),
        'expectedTestNames': {name: sorted(cases) for name, cases in sorted(EXPECTED_TESTS.items())},
        'reports': [], 'testCount': 0, 'observed': [], 'results': [],
        'artifacts': {}, 'logs': None,
        'cleanup': {'processGroups': '尚未完成',
                    'resources': '测试 fixture 关闭 Hikari 数据源并 DROP 每个独有库；runner 再核对 study_u46% 无残留'},
        'limits': [
            'Spring Batch 6.0.6 / Quartz 2.5.2 / Framework 7.0.10 / Boot 4.0.9 / MySQL 8.4.7 全部固定并实跑；'
            '中文注释与主题基线等价于仓库其余 lab，不是 mock 或内存替身',
            'Quartz 多实例是同一 JVM 内多个 Scheduler 对象共用同一 MySQL 库；证明行锁与集群接管语义，'
            '不等于多机进程隔离、网络分区或断电',
            'misfire 用回拨 NEXT_FIRE_TIME 与 2 秒阈值构造受控加速，不是生产时间窗；'
            '集群接管用回拨 LAST_CHECKIN_TIME 构造，不是真实进程死亡',
            '未运行：远程分区/远程 chunk（spring-batch-integration）、JSR-352、Quartz Terracotta/JTA、'
            'XA 跨库、数据库主从切换、Batch Mongo JobRepository、生产 TLS',
        ],
    }
    exit_code = 1
    mysql = mysql_binary()
    try:
        before = leftover_databases(mysql)
        evidence['cleanup']['databasesBefore'] = before
        if before:
            raise SystemExit('本机已有 study_u46% 库：' + str(before) + '；先确认并删除后重试')
        write_json(EVIDENCE, evidence)
        logs_root = pathlib.Path.home() / '.cache/software-engineering-study/java-depth-impl/u46'
        logs_root.mkdir(parents=True, exist_ok=True)
        logs = pathlib.Path(tempfile.mkdtemp(prefix='run-', dir=logs_root))
        evidence['logs'] = str(logs)
        print('LOGS ' + str(logs), flush=True)
        if len(sys.argv) != 1:
            raise ValueError('本入口不接受参数；使用 python3 labs/batch-quartz/run.py')
        source_files = sorted((ROOT / 'src').rglob('*.java'))
        if len(source_files) != 6:
            raise ValueError('测试源文件数量变化: ' + str([str(p.relative_to(ROOT)) for p in source_files]))
        evidence['artifacts'] = artifact_digests()
        digest_before = dict(evidence['artifacts'])
        env = dict(os.environ)
        home = java_home()
        env['JAVA_HOME'] = str(home)
        env['PATH'] = str(home / 'bin') + os.pathsep + env.get('PATH', '')
        mvn = maven_binary()
        run('java-version', [str(home / 'bin/java'), '-version'], logs, env, evidence['results'])
        run('maven-version', [mvn, '-version'], logs, env, evidence['results'])
        started_ns = time.time_ns()
        run('mvn-test', [mvn, '-B', '-ntp', '-o', '-f', str(ROOT / 'pom.xml'), 'test'], logs, env,
            evidence['results'])
        evidence['reports'] = validate_reports(logs, started_ns)
        evidence['testCount'] = sum(int(report['tests']) for report in evidence['reports'])
        dependency_file = logs / 'dependencies.txt'
        run('mvn-dependency', [mvn, '-B', '-ntp', '-o', '-f', str(ROOT / 'pom.xml'),
                               'dependency:list', '-DoutputFile=' + str(dependency_file)],
            logs, env, evidence['results'])
        evidence['environment']['dependencies'] = read_dependencies(dependency_file)
        observed = [line for result in evidence['results'] for line in result['observed']]
        evidence['observed'] = validate_observations(observed)
        after = artifact_digests()
        if after != digest_before:
            raise ValueError('运行期间源文件发生变化，证据不可信: ' + json.dumps(
                {key: (digest_before.get(key), after.get(key)) for key in set(after) | set(digest_before)
                 if digest_before.get(key) != after.get(key)}, ensure_ascii=False))
        remaining = leftover_databases(mysql)
        evidence['cleanup']['databasesAfter'] = remaining
        if remaining is None:
            raise ValueError('无法核对 MySQL 残留（mysql 客户端不可用）')
        if remaining:
            raise ValueError('存在未清理的 U46 库: ' + str(remaining))
        evidence['cleanup']['processGroups'] = ('本入口创建的进程组已逐条回收（每条命令后 cleanup_group；'
                                                '清理失败会置非零退出并阻止 passed）')
        evidence['cleanup']['databaseDropped'] = True
        evidence['status'] = 'passed'
        evidence['exitCode'] = 0
        evidence['finishedAt'] = now()
        write_json(EVIDENCE, evidence)
        print('WROTE labs/batch-quartz/evidence/u46-2026-10-31.json tests='
              + str(evidence['testCount']), flush=True)
        exit_code = 0
    except BaseException as exc:
        exit_code = 1 if not isinstance(exc, SystemExit) else (exc.code or 1)
        if isinstance(exc, subprocess.TimeoutExpired):
            exit_code = 124
        if isinstance(exc, KeyboardInterrupt):
            exit_code = 130
        evidence['status'] = 'failed'
        evidence['exitCode'] = exit_code
        evidence['failure'] = str(exc)[-3000:]
        evidence['finishedAt'] = now()
        write_json(EVIDENCE, evidence)
        print('FAILED: ' + str(exc), file=sys.stderr, flush=True)
    return exit_code


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