#!/usr/bin/env python3 """Publish nearby-distance support and backfill only the labelled fixture batch.""" import base64 import fcntl import hashlib import hmac import json import os import pwd import shutil import sys import time import urllib.error import urllib.request import zipfile from pathlib import Path sys.path.insert(0, '/www/server/xingyu-im/ops') from baota_publish_test_users import BT, WEB, database, digest, env_values, extract, query OLD = 'e8344e81cc828e6817bc7c232a261884fbb5d7e8123420da121ede908502329e' BASE = Path('/www/server/xingyu-im/releases/read-badge-20260901-183536') BATCH = 'cn-adults-20260831-v1' LOCAL = 'http://127.0.0.1:18888' ALLOWED = { 'backend/internal/app/social.go', 'backend/internal/testusers/fixtures.go', 'backend/internal/testusers/seed.go', 'fixtures/test-users/users.json', } def fetch(path): with urllib.request.urlopen(LOCAL + 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 local 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 source_hashes(archive): with zipfile.ZipFile(str(archive)) as package: result = {} for item in package.infolist(): if item.is_dir(): continue name = item.filename.replace('\\', '/') assert name not in result and not name.startswith('/') and '..' not in Path(name).parts result[name] = hashlib.sha256(package.read(item)).hexdigest() return result def running(expected): pids = [] for proc in Path('/proc').iterdir(): if not proc.name.isdigit(): continue try: if os.readlink(str(proc / 'exe')) == str(BT / 'xingyu-api'): assert digest(proc / 'exe') == expected pids.append(int(proc.name)) except (FileNotFoundError, PermissionError): pass assert len(pids) == 1, 'expected one running project executable' return pids def replace(source, stamp): temporary = BT / ('xingyu-api.nearby-distance-' + stamp) assert not temporary.exists() shutil.copy2(str(source), str(temporary)) www = pwd.getpwnam('www') os.chown(str(temporary), www.pw_uid, www.pw_gid) os.chmod(str(temporary), 0o750) os.replace(str(temporary), str(BT / 'xingyu-api')) def encode_json(value): raw = json.dumps(value, separators=(',', ':'), ensure_ascii=False).encode() return base64.urlsafe_b64encode(raw).rstrip(b'=').decode() def user_token(secret, user_id, nickname, version): now = int(time.time()) header = encode_json({'alg': 'HS256', 'typ': 'JWT'}) payload = encode_json({'role': 'user', 'name': nickname, 'ver': version, 'sub': str(user_id), 'iat': now, 'exp': now + 300}) signing_input = (header + '.' + payload).encode() signature = base64.urlsafe_b64encode(hmac.new(secret.encode(), signing_input, hashlib.sha256).digest()).rstrip(b'=').decode() return header + '.' + payload + '.' + signature def fixture_rows(conn): return query(conn, """SELECT user.id,user.public_id,profile.nickname FROM users user JOIN user_profiles profile ON profile.user_id=user.id WHERE user.is_test=1 AND user.test_batch=%s AND user.status=1 AND user.deleted_at IS NULL ORDER BY user.id""", (BATCH,)) def privacy_rows(conn): return query(conn, """SELECT privacy.user_id,privacy.distance_visible, DATE_FORMAT(privacy.updated_at,'%%Y-%%m-%%d %%H:%%i:%%s.%%f') FROM user_privacy_settings privacy JOIN users user ON user.id=privacy.user_id WHERE user.is_test=1 AND user.test_batch=%s AND user.deleted_at IS NULL ORDER BY privacy.user_id""", (BATCH,)) def location_rows(conn): return query(conn, """SELECT location.user_id,location.city_code,location.location_cell, CAST(location.latitude AS CHAR),CAST(location.longitude AS CHAR), DATE_FORMAT(location.last_location_at,'%%Y-%%m-%%d %%H:%%i:%%s.%%f'),location.source, DATE_FORMAT(location.created_at,'%%Y-%%m-%%d %%H:%%i:%%s.%%f'), DATE_FORMAT(location.updated_at,'%%Y-%%m-%%d %%H:%%i:%%s.%%f') FROM user_location_states location JOIN users user ON user.id=location.user_id WHERE user.is_test=1 AND user.test_batch=%s AND user.deleted_at IS NULL ORDER BY location.user_id""", (BATCH,)) def restore_data(conn, privacy_before, locations_before): fixture_ids = [row[0] for row in privacy_before] assert len(fixture_ids) == 100 placeholders = ','.join(['%s'] * len(fixture_ids)) with conn.cursor() as cursor: cursor.executemany( 'UPDATE user_privacy_settings SET distance_visible=%s,updated_at=%s WHERE user_id=%s', [(distance, updated, user_id) for user_id, distance, updated in privacy_before], ) cursor.execute('DELETE FROM user_location_states WHERE user_id IN (' + placeholders + ')', fixture_ids) if locations_before: cursor.executemany("""INSERT INTO user_location_states (user_id,city_code,location_cell,latitude,longitude,last_location_at,source,created_at,updated_at) VALUES(%s,%s,%s,%s,%s,%s,%s,%s,%s)""", locations_before) def verify_api(conn, values): selected = query(conn, """SELECT user.id,profile.nickname,security.token_version FROM users user JOIN user_profiles profile ON profile.user_id=user.id JOIN user_security_controls security ON security.user_id=user.id WHERE user.is_test=1 AND user.test_batch=%s AND user.status=1 AND user.deleted_at IS NULL ORDER BY user.id LIMIT 1""", (BATCH,)) assert selected user_id, nickname, version = selected[0] token = user_token(values['IM_JWT_SECRET'], user_id, nickname, version) request = urllib.request.Request( LOCAL + '/api/v1/nearby/users?page=1&pageSize=20', headers={'Authorization': 'Bearer ' + token, 'Accept': 'application/json'}, ) with urllib.request.urlopen(request, timeout=20) as response: assert response.status == 200 result = json.loads(response.read())['data'] assert result.get('locationReady') is True distance_items = [item for item in result.get('items', []) if item.get('distanceText') and float(item.get('distance', 0)) >= 0] assert distance_items, 'nearby API returned no visible distances' return { 'viewerUserId': user_id, 'items': len(result.get('items', [])), 'itemsWithDistance': len(distance_items), 'sampleDistanceText': distance_items[0]['distanceText'], } def publish(archive, expected): assert os.geteuid() == 0 archive = Path(archive).resolve() assert archive.parent == Path('/tmp') and digest(archive) == expected lock = open('/www/server/xingyu-im/ops/application-release.lock', 'a') fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB) assert digest(BT / 'xingyu-api') == OLD, 'production changed; re-audit required' running(OLD) health() stamp = time.strftime('%Y%m%d-%H%M%S') stage = Path('/www/server/xingyu-im/releases/nearby-distance-' + stamp) extract(archive, stage) os.chmod(str(stage), 0o700) old_source = source_hashes(BASE / 'backend-source.zip') new_source = source_hashes(stage / 'backend-source.zip') changed = {name for name in set(old_source) | set(new_source) if old_source.get(name) != new_source.get(name)} assert changed == ALLOWED, 'unreviewed source changes: ' + repr(changed) new_hash = digest(stage / 'xingyu-api') assert (stage / 'xingyu-api').read_bytes()[:4] == b'\x7fELF' and new_hash != OLD manifest = json.loads((stage / 'manifest.json').read_text()) for name in ['xingyu-api', 'backend-source.zip', 'fixture-locations.json']: assert digest(stage / name) == manifest['files'][name] profiles = json.loads((stage / 'fixture-locations.json').read_text()) assert len(profiles) == 100 profile_map = {item['publicId']: item for item in profiles} assert len(profile_map) == 100 assert all(item['testBatch'] == BATCH and item['isTest'] is True for item in profiles) assert len({(item['latitude'], item['longitude']) for item in profiles}) == 100 assert all(18 < item['latitude'] < 54 and 73 < item['longitude'] < 136 for item in profiles) values = env_values() conn, _ = database(values) try: migrations = dict(query(conn, 'SELECT version,checksum FROM schema_migrations')) for name, sha in migrations.items(): assert new_source['backend/migrations/' + name] == sha assert len(migrations) == len([name for name in new_source if name.startswith('backend/migrations/')]) frontend = {name: str((WEB / name).resolve()) for name in ['admin', 'app']} fixtures = fixture_rows(conn) assert len(fixtures) == 100 and {row[1] for row in fixtures} == set(profile_map) privacy_before = privacy_rows(conn) locations_before = location_rows(conn) assert len(privacy_before) == 100 backup = Path('/www/backup/xingyu-nearby-distance-' + stamp) backup.mkdir(mode=0o700) shutil.copy2(str(BT / 'xingyu-api'), str(backup / 'xingyu-api-original')) for name, rows in [('privacy-before.json', privacy_before), ('locations-before.json', locations_before)]: path = backup / name path.write_text(json.dumps(rows, ensure_ascii=False, default=str)) os.chmod(str(path), 0o600) state = { 'backup': str(backup), 'stage': str(stage), 'previousBackendSha256': OLD, 'backendSha256': new_hash, 'archiveSha256': expected, 'changedSourceFiles': sorted(changed), 'frontendBefore': frontend, 'databaseMigrationsApplied': [], 'businessDataWrites': True, 'dataScope': 'is_test=1 and test_batch=' + BATCH, 'locationRowsBefore': len(locations_before), } (backup / 'before.json').write_text(json.dumps(state, indent=2)) print('BACKUP_VERIFIED=' + str(backup), flush=True) changed_binary = False data_changed = False try: replace(stage / 'xingyu-api', stamp) changed_binary = True restart() state['runningPids'] = running(new_hash) assert digest(BT / 'xingyu-api') == new_hash assert dict(query(conn, 'SELECT version,checksum FROM schema_migrations')) == migrations assert {name: str((WEB / name).resolve()) for name in frontend} == frontend with conn.cursor() as cursor: cursor.execute("""UPDATE user_privacy_settings privacy JOIN users user ON user.id=privacy.user_id SET privacy.distance_visible=1 WHERE user.is_test=1 AND user.test_batch=%s AND user.deleted_at IS NULL""", (BATCH,)) state['privacyRowsChanged'] = cursor.rowcount data_changed = True existing_ids = {row[0] for row in locations_before} inserts = [] for user_id, public_id, _ in fixtures: if user_id in existing_ids: continue item = profile_map[public_id] inserts.append((user_id, item['cityCode'], item['latitude'], item['longitude'])) if inserts: with conn.cursor() as cursor: cursor.executemany("""INSERT INTO user_location_states (user_id,city_code,location_cell,latitude,longitude,source) VALUES(%s,%s,'fixture',%s,%s,'fixture')""", inserts) state['locationRowsInserted'] = len(inserts) after_locations = location_rows(conn) assert len(after_locations) == 100 assert all(int(row[1]) == 1 for row in privacy_rows(conn)) before_by_id = {row[0]: row for row in locations_before} after_by_id = {row[0]: row for row in after_locations} assert all(after_by_id[user_id] == row for user_id, row in before_by_id.items()) for path in ['/api/v1/nearby/users', '/admin/v1/users', '/ws']: try: fetch(path) except urllib.error.HTTPError as error: assert error.code == 401 else: raise RuntimeError('private endpoint became public') state['nearbyApi'] = verify_api(conn, values) state['healthAndAuthChecks'] = True state['locationRowsAfter'] = len(after_locations) (backup / 'result.json').write_text(json.dumps(state, indent=2)) print(json.dumps(state), flush=True) except Exception: if data_changed: restore_data(conn, privacy_before, locations_before) if changed_binary: replace(backup / 'xingyu-api-original', stamp + '-rollback') restart() running(OLD) (backup / 'rolled-back.txt').write_text('Original binary and scoped fixture distance data restored.\n') raise finally: conn.close() if __name__ == '__main__': publish(*sys.argv[1:])