#!/usr/bin/env python3
"""Scoped release of the labelled 100-user fixture batch on the existing BaoTa host.

Run with the panel Python as root. DB credentials stay in memory; backups are
root-only. No admin login, fabricated sessions, or OAuth migration is performed.
"""
import argparse
import gzip
import hashlib
import json
import os
import pwd
import re
import shlex
import shutil
import subprocess
import sys
import time
import urllib.error
import urllib.request
import zipfile
from pathlib import Path

import pymysql

DOMAIN = 'https://im.bchongw.com'
BATCH = 'cn-adults-20260831-v1'
BT = Path('/www/server/xingyu-im/bt')
WEB = Path('/www/wwwroot/im.bchongw.com')
MEDIA = Path('/www/wwwroot/xingyu-data/uploads')
OLD_BINARY = '23ab263e16593b6f588e42bf480e08f364f8f46a12cf8c143077b767860b8b30'


def digest(path):
    return hashlib.sha256(Path(path).read_bytes()).hexdigest()


def env_values():
    values = {}
    for line in Path('/etc/xingyu-im-bt.env').read_text().splitlines():
        line = line[7:] if line.startswith('export ') else line
        if not line.strip() or line.lstrip().startswith('#'):
            continue
        key, value = line.split('=', 1)
        parts = shlex.split(value)
        assert len(parts) <= 1
        values[key] = parts[0] if parts else ''
    assert values['IM_ENV'] == 'production'
    assert values['IM_HOST'] == '127.0.0.1' and values['IM_PORT'] == '18888'
    return values


def database(values):
    match = re.fullmatch(r'([^:]+):(.*?)@tcp\(([^:]+):(\d+)\)/([^?]+)\?.*', values['IM_DB_DSN'])
    assert match and match[3] == '127.0.0.1' and match[4] == '3307' and match[5] == 'im'
    options = dict(host=match[3], port=int(match[4]), user=match[1], password=match[2], database=match[5], charset='utf8mb4', autocommit=True)
    return pymysql.connect(**options), options


def query(conn, sql, args=()):
    with conn.cursor() as cursor:
        cursor.execute(sql, args)
        return cursor.fetchall()


def fetch(path):
    with urllib.request.urlopen(DOMAIN + path, timeout=20) as response:
        return response.status, response.headers, response.read()


def health():
    for _ in range(100):
        try:
            status, _, data = fetch('/healthz')
            if status == 200 and json.loads(data)['data']['status'] == 'ok':
                return
        except Exception:
            pass
        time.sleep(0.5)
    raise RuntimeError('backend failed health check')


def restart():
    os.chdir('/www/server/panel')
    sys.path.insert(0, '/www/server/panel')
    sys.path.insert(0, '/www/server/panel/class')
    import public
    from projectModel.goModel import main as GoProject
    request = public.dict_obj()
    request.project_name = 'xingyu_im'
    manager = GoProject()
    project = manager.get_project_find('xingyu_im')
    assert project and project['project_config']['is_power_on'] == 1
    result = manager.restart_project(request)
    if not result.get('status'):
        recovered = manager.start_project(request)
        if not recovered.get('status'):
            raise RuntimeError('BaoTa could not start project')
    health()


def extract(archive, directory):
    directory.mkdir(mode=0o755, parents=True, exist_ok=False)
    with zipfile.ZipFile(str(archive)) as package:
        seen = set()
        for item in package.infolist():
            name = item.filename.replace('\\', '/')
            target = (directory / name).resolve()
            assert name not in seen and name and ':' not in name
            assert not name.startswith('/') and '..' not in Path(name).parts
            assert os.path.commonpath([str(directory.resolve()), str(target)]) == str(directory.resolve())
            assert (item.external_attr >> 16) & 0o170000 != 0o120000
            seen.add(name)
            if item.is_dir() or name.endswith('/'):
                target.mkdir(parents=True, exist_ok=True)
            else:
                target.parent.mkdir(parents=True, exist_ok=True)
                with package.open(item) as source, target.open('xb') as output:
                    shutil.copyfileobj(source, output)


def switch(entry, target, stamp):
    next_link = entry.with_name(entry.name + '.next-' + stamp)
    assert not next_link.exists() and not next_link.is_symlink()
    os.symlink(str(target), str(next_link))
    os.replace(str(next_link), str(entry))


def prepare_static(archive, kind, stamp):
    entry = WEB / ('admin' if kind == 'admin' else 'app')
    releases = Path('/www/wwwroot/xingyu-' + kind + '/releases')
    assert entry.is_symlink()
    previous = entry.resolve()
    assert previous.parent == releases
    release = releases / ('test-users-' + stamp)
    extract(archive, release)
    index = (release / 'index.html').read_text()
    assert ('/admin/jse/' if kind == 'admin' else '/app/assets/') in index
    if kind == 'admin':
        assert '"/admin/v1"' in (release / '_app.config.js').read_text()
    # Keep hashed resources used by tabs opened before this release.
    for folder in (['js', 'jse', 'css'] if kind == 'admin' else ['assets']):
        if not (previous / folder).is_dir():
            continue
        for source in (previous / folder).rglob('*'):
            if source.is_file():
                target = release / source.relative_to(previous)
                if not target.exists():
                    target.parent.mkdir(parents=True, exist_ok=True)
                    shutil.copy2(str(source), str(target))
    www = pwd.getpwnam('www')
    for path in [release] + list(release.rglob('*')):
        assert not path.is_symlink()
        os.chown(str(path), www.pw_uid, www.pw_gid)
        os.chmod(str(path), 0o755 if path.is_dir() else 0o644)
    return entry, previous, release


def verify(conn):
    rows = query(conn, "SELECT p.gender,COUNT(*) FROM users u JOIN user_profiles p ON p.user_id=u.id WHERE u.is_test=1 AND u.test_batch=%s AND u.deleted_at IS NULL GROUP BY p.gender ORDER BY p.gender", (BATCH,))
    assert rows == ((1, 50), (2, 50)), rows
    count = query(conn, "SELECT COUNT(*) FROM users WHERE is_test=1 AND test_batch=%s AND phone_hash IS NULL AND phone_cipher IS NULL AND password_hash='!TEST_PROFILE_NO_LOGIN'", (BATCH,))[0][0]
    assert count == 100
    avatars = query(conn, "SELECT DISTINCT p.avatar_url FROM users u JOIN user_profiles p ON p.user_id=u.id WHERE u.is_test=1 AND u.test_batch=%s", (BATCH,))
    assert len(avatars) == 10
    for (url,) in avatars:
        assert url.startswith(DOMAIN + '/uploads/')
        status, headers, body = fetch(url[len(DOMAIN):])
        assert status == 200 and headers.get_content_type() == 'image/png' and body.startswith(b'\x89PNG\r\n\x1a\n')
    for public_path in ['/admin/', '/app/']:
        status, headers, data = fetch(public_path)
        cache_control = ','.join(headers.get_all('Cache-Control', []))
        assert status == 200 and b'<html' in data.lower() and 'no-store' in cache_control
    for private_path in ['/admin/v1/users?userType=test', '/api/v1/discover/recommendations', '/ws']:
        try:
            fetch(private_path)
        except urllib.error.HTTPError as error:
            assert error.code == 401
        else:
            raise RuntimeError('private endpoint did not require authentication')
    health()
    return {'batch': BATCH, 'total': 100, 'male': 50, 'female': 50, 'avatarsHttp200': len(avatars), 'privateEndpointsRequireAuth': True}


def publish(archive, expected):
    assert os.geteuid() == 0
    archive = Path(archive).resolve()
    assert archive.parent == Path('/tmp') and re.fullmatch('[a-f0-9]{64}', expected)
    assert digest(archive) == expected
    values = env_values()
    conn, db_options = database(values)
    assert digest(BT / 'xingyu-api') == OLD_BINARY, 'online binary changed; re-audit before deployment'
    assert query(conn, "SELECT COUNT(*) FROM schema_migrations WHERE version='027_app_oauth_login.sql'")[0][0] == 0, 'OAuth release changed; do not replace it'
    assert query(conn, "SELECT COUNT(*) FROM users WHERE public_id LIKE 'TESTCN%%'")[0][0] == 0, 'test IDs already exist; inspect before retry'
    assert query(conn, "SELECT config_value FROM system_configs WHERE config_key='storage.local.directory'")[0][0] == './uploads'
    pid = int(Path('/var/tmp/gopids/xingyu_im.pid').read_text().strip())
    assert Path('/proc/{}/cwd'.format(pid)).resolve() == BT
    assert not (BT / 'uploads').exists() and not (BT / 'uploads').is_symlink()
    assert MEDIA.is_dir()
    health()
    stamp = time.strftime('%Y%m%d-%H%M%S')
    backup = Path('/www/backup/xingyu-test-users-' + stamp)
    backup.mkdir(mode=0o700)
    state = {
        'stamp': stamp, 'archiveSha256': expected, 'previousBackendSha256': OLD_BINARY,
        'usersBefore': query(conn, 'SELECT COUNT(*) FROM users')[0][0],
        'previousAdmin': str((WEB / 'admin').resolve()), 'previousH5': str((WEB / 'app').resolve()),
        'unchangedCounts': {table: query(conn, 'SELECT COUNT(*) FROM ' + table)[0][0] for table in ['orders', 'posts', 'im_messages', 'subscriptions', 'user_sessions', 'admin_users']},
    }
    # Dump only this application database using the matching MySQL 8 client.
    dump_env = dict(os.environ, MYSQL_PWD=db_options['password'])
    dump = subprocess.Popen(['/opt/mysql-8.4.11/bin/mysqldump', '-h', '127.0.0.1', '-P', '3307', '-u', db_options['user'], '--single-transaction', '--no-tablespaces', '--set-gtid-purged=OFF', '--routines', '--triggers', 'im'], env=dump_env, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    with gzip.open(str(backup / 'im.sql.gz'), 'wb') as output:
        shutil.copyfileobj(dump.stdout, output)
    _, errors = dump.communicate()
    if dump.returncode:
        raise RuntimeError('database backup failed: ' + errors.decode('utf-8', errors='replace'))
    os.chmod(str(backup / 'im.sql.gz'), 0o600)
    shutil.copy2(str(BT / 'xingyu-api'), str(backup / 'xingyu-api'))
    shutil.copy2('/etc/xingyu-im-bt.env', str(backup / 'environment.env'))
    os.chmod(str(backup / 'environment.env'), 0o600)
    (backup / 'state.json').write_text(json.dumps(state, indent=2))
    print('BACKUP=' + str(backup), flush=True)
    stage = Path('/www/server/xingyu-im/releases/test-users-' + stamp)
    extract(archive, stage)
    os.chmod(str(stage), 0o700)
    # Validate all deployed migrations, then add only the test-user migration.
    recorded = dict(query(conn, 'SELECT version,checksum FROM schema_migrations'))
    for file in (stage / 'migrations').glob('*.sql'):
        if file.name == '028_test_users.sql':
            continue
        assert recorded.get(file.name) == digest(file), 'migration history mismatch: ' + file.name
    migration = stage / 'migrations/028_test_users.sql'
    assert migration.is_file() and '028_test_users.sql' not in recorded
    sql = '\n'.join(line for line in migration.read_text().splitlines() if not line.lstrip().startswith('--'))
    for statement in sql.split(';'):
        if statement.strip():
            query(conn, statement)
    query(conn, 'INSERT INTO schema_migrations (version,checksum) VALUES (%s,%s)', (migration.name, digest(migration)))
    shutil.copy2(str(migration), str(BT / 'migrations' / migration.name))
    www = pwd.getpwnam('www')
    os.chown(str(BT / 'migrations' / migration.name), www.pw_uid, www.pw_gid)
    os.chmod(str(BT / 'migrations' / migration.name), 0o640)
    # The existing local-storage config is relative to the BaoTa working dir.
    # Preserve that config and make it point at the established persistent data dir.
    os.symlink(str(MEDIA), str(BT / 'uploads'))
    admin = prepare_static(stage / 'admin.zip', 'admin', stamp)
    h5 = prepare_static(stage / 'h5.zip', 'h5', stamp)
    switched = []
    try:
        new_binary = BT / ('xingyu-api.next-' + stamp)
        shutil.copy2(str(stage / 'xingyu-api'), str(new_binary))
        os.chown(str(new_binary), www.pw_uid, www.pw_gid)
        os.chmod(str(new_binary), 0o750)
        os.replace(str(new_binary), str(BT / 'xingyu-api'))
        restart()
        for entry, previous, release in [admin, h5]:
            switch(entry, release, stamp)
            switched.append((entry, previous))
        for _, _, release in [admin, h5]:
            prefix = '/admin/' if release == admin[2] else '/app/'
            assert fetch(prefix)[2] == (release / 'index.html').read_bytes()
        seeder = stage / 'seed-test-users'
        os.chmod(str(seeder), 0o700)
        seed_env = dict(os.environ, **values)
        command = [str(seeder), '--apply', '--allow-production-test-data', '--confirm-database', 'im', '--public-base', DOMAIN + '/uploads', '--avatars-dir', str(stage / 'avatars'), '--media-dir', str(MEDIA)]
        result = subprocess.run(command, env=seed_env, stdout=subprocess.PIPE, stderr=subprocess.PIPE, universal_newlines=True, timeout=90)
        if result.returncode:
            raise RuntimeError('seed failed: ' + result.stderr)
        imported = json.loads(result.stdout)
        assert imported['created'] == 100 and imported['male'] == 50 and imported['female'] == 50
        report = verify(conn)
        assert query(conn, 'SELECT COUNT(*) FROM users WHERE is_test=0')[0][0] == state['usersBefore']
        for table, count in state['unchangedCounts'].items():
            assert query(conn, 'SELECT COUNT(*) FROM ' + table)[0][0] == count, 'unexpected change to ' + table
        repeat = subprocess.run(command, env=seed_env, stdout=subprocess.PIPE, stderr=subprocess.PIPE, universal_newlines=True, timeout=90)
        assert repeat.returncode == 0
        repeated = json.loads(repeat.stdout)
        assert repeated['created'] == 0 and repeated['skipped'] == 100
        state.update(report)
        state.update({'backendSha256': digest(BT / 'xingyu-api'), 'admin': str(admin[2]), 'h5': str(h5[2]), 'idempotencySkipped': 100, 'seedResult': imported, 'backup': str(backup), 'stage': str(stage)})
        (backup / 'result.json').write_text(json.dumps(state, ensure_ascii=False, indent=2))
        print(json.dumps(state, ensure_ascii=False), flush=True)
    except Exception:
        # Keep additive schema and fixture data for diagnosis; never restore a
        # full DB dump over users who may have registered during deployment.
        for entry, previous in reversed(switched):
            switch(entry, previous, stamp + '-rollback')
        restore = BT / ('xingyu-api.rollback-' + stamp)
        shutil.copy2(str(backup / 'xingyu-api'), str(restore))
        os.chown(str(restore), www.pw_uid, www.pw_gid)
        os.chmod(str(restore), 0o750)
        os.replace(str(restore), str(BT / 'xingyu-api'))
        restart()
        print('APPLICATION_ROLLED_BACK; additive schema/data retained; backup=' + str(backup), flush=True)
        raise
    finally:
        conn.close()


def resume(backup_path):
    """Resume this exact backed-up release after an application-only rollback.

    No migration, initial import, environment change, or DB restoration occurs.
    The original preflight deliberately rejects retries after a partial release.
    """
    assert os.geteuid() == 0
    backup = Path(backup_path).resolve()
    assert backup.parent == Path('/www/backup')
    assert re.fullmatch(r'xingyu-test-users-\d{8}-\d{6}', backup.name)
    state = json.loads((backup / 'state.json').read_text())
    stamp = state['stamp']
    assert backup.name == 'xingyu-test-users-' + stamp
    stage = Path('/www/server/xingyu-im/releases/test-users-' + stamp)
    archive = Path('/tmp/xingyu-test-users-20260831.zip')
    assert digest(archive) == state['archiveSha256']
    with zipfile.ZipFile(str(archive)) as package:
        for name in ['xingyu-api', 'seed-test-users', 'admin.zip', 'h5.zip', 'migrations/028_test_users.sql']:
            matching = [item for item in package.infolist() if item.filename.replace('\\', '/') == name]
            assert len(matching) == 1
            assert digest(stage / name) == hashlib.sha256(package.read(matching[0])).hexdigest()
    assert digest(BT / 'xingyu-api') == OLD_BINARY == state['previousBackendSha256']
    assert (BT / 'uploads').is_symlink() and (BT / 'uploads').resolve() == MEDIA
    admin = (WEB / 'admin', Path(state['previousAdmin']), Path('/www/wwwroot/xingyu-admin/releases/test-users-' + stamp))
    h5 = (WEB / 'app', Path(state['previousH5']), Path('/www/wwwroot/xingyu-h5/releases/test-users-' + stamp))
    for entry, previous, release in [admin, h5]:
        assert entry.is_symlink() and entry.resolve() == previous
        assert previous.parent == release.parent
        assert (release / 'index.html').is_file()
    values = env_values()
    conn, _ = database(values)
    recorded = dict(query(conn, 'SELECT version,checksum FROM schema_migrations'))
    assert '027_app_oauth_login.sql' not in recorded
    assert recorded['028_test_users.sql'] == digest(stage / 'migrations/028_test_users.sql')
    health()
    # The complete fixture import must already exist. The seeder's transactional
    # repeat check is a no-op and refuses partial/modified/conflicting batches.
    assert query(conn, 'SELECT COUNT(*) FROM users WHERE is_test=1 AND test_batch=%s', (BATCH,))[0][0] == 100
    command = [str(stage / 'seed-test-users'), '--apply', '--allow-production-test-data', '--confirm-database', 'im', '--public-base', DOMAIN + '/uploads', '--avatars-dir', str(stage / 'avatars'), '--media-dir', str(MEDIA)]
    repeated = subprocess.run(command, env=dict(os.environ, **values), stdout=subprocess.PIPE, stderr=subprocess.PIPE, universal_newlines=True, timeout=90)
    assert repeated.returncode == 0, 'existing fixture batch failed repeat verification'
    imported = json.loads(repeated.stdout)
    assert imported['created'] == 0 and imported['skipped'] == 100
    switched = []
    www = pwd.getpwnam('www')
    try:
        new_binary = BT / ('xingyu-api.resume-' + stamp)
        assert not new_binary.exists()
        shutil.copy2(str(stage / 'xingyu-api'), str(new_binary))
        os.chown(str(new_binary), www.pw_uid, www.pw_gid)
        os.chmod(str(new_binary), 0o750)
        os.replace(str(new_binary), str(BT / 'xingyu-api'))
        restart()
        for entry, previous, release in [admin, h5]:
            switch(entry, release, stamp + '-resume')
            switched.append((entry, previous))
        assert fetch('/admin/')[2] == (admin[2] / 'index.html').read_bytes()
        assert fetch('/app/')[2] == (h5[2] / 'index.html').read_bytes()
        state.update(verify(conn))
        assert query(conn, 'SELECT COUNT(*) FROM users WHERE is_test=0')[0][0] == state['usersBefore']
        for table, count in state['unchangedCounts'].items():
            assert query(conn, 'SELECT COUNT(*) FROM ' + table)[0][0] == count, 'unexpected change to ' + table
        state.update({'backendSha256': digest(BT / 'xingyu-api'), 'admin': str(admin[2]), 'h5': str(h5[2]), 'idempotencySkipped': 100, 'repeatSeedResult': imported, 'backup': str(backup), 'stage': str(stage), 'resumedAfterCacheHeaderCheckFix': True})
        (backup / 'result.json').write_text(json.dumps(state, ensure_ascii=False, indent=2))
        print(json.dumps(state, ensure_ascii=False), flush=True)
    except Exception:
        for entry, previous in reversed(switched):
            switch(entry, previous, stamp + '-resume-rollback')
        restore = BT / ('xingyu-api.resume-rollback-' + stamp)
        shutil.copy2(str(backup / 'xingyu-api'), str(restore))
        os.chown(str(restore), www.pw_uid, www.pw_gid)
        os.chmod(str(restore), 0o750)
        os.replace(str(restore), str(BT / 'xingyu-api'))
        restart()
        print('APPLICATION_ROLLED_BACK; imported fixtures retained', flush=True)
        raise
    finally:
        conn.close()


if __name__ == '__main__':
    parser = argparse.ArgumentParser()
    parser.add_argument('archive', nargs='?')
    parser.add_argument('sha256', nargs='?')
    parser.add_argument('--resume-backup')
    args = parser.parse_args()
    if args.resume_backup:
        assert not args.archive and not args.sha256
        resume(args.resume_backup)
    else:
        assert args.archive and args.sha256
        publish(args.archive, args.sha256)