#!/usr/bin/env python3
"""U53 公开入口：Redis 8.6.4 使用、内存与恢复边界实验。

只使用本机 127.0.0.1:6379 的 standalone Redis（DBngin 8.6.4，U03 基线同源），
全部键在 `u53:<uuid>:` 命名空间内，finally 定向删除并逐键 EXISTS 复核；
不改配置、不启用持久化、不重启、不触碰用户既有键。
实验区分三类证据：真实服务观察、Python 标准库模拟（不访问服务）、未运行项（如实受阻/未运行）。

覆盖（既有两篇 Redis 教材为机制权威，本票只补真实使用与缺口）：
- pipeline 与事务/MULTI 的 RTT 差异（真实测量，非基准）
- INFO memory 各字段判读（used_memory vs RSS、allocator 碎片）
- 大 key 扫描（SCAN 游标 + MEMORY USAGE 采样，只对本命名空间）
- 客户端缓存（CLIENT TRACKING）失效推送
- Lua 脚本的原子性与 KEYS 限制
- SET EX NX 的分布式锁形态与误用（不作为锁权威，回指租约页）
"""
import hashlib
import json
import os
import pathlib
import platform
import socket
import sys
import time
import uuid
from datetime import datetime

ROOT = pathlib.Path(__file__).resolve().parent
EVIDENCE = ROOT / 'evidence'
NAMESPACE_PREFIX = 'u53'
HOST = os.environ.get('U53_REDIS_HOST', '127.0.0.1')
PORT = int(os.environ.get('U53_REDIS_PORT', '6379'))

observations = []


def observe(line):
    observations.append(line)
    print(line, flush=True)


class Redis:
    """最小 RESP 客户端：支持 RESP2 常规类型与 RESP3 的 map/set/push 类型。"""

    def __init__(self, host=HOST, port=PORT, timeout=5):
        self.sock = socket.create_connection((host, port), timeout=timeout)
        self.sock.settimeout(timeout)
        self.buffer = b''

    def _read_line(self):
        while b'\r\n' not in self.buffer:
            chunk = self.sock.recv(65536)
            if not chunk:
                raise ConnectionError('连接关闭')
            self.buffer += chunk
        line, self.buffer = self.buffer.split(b'\r\n', 1)
        return line

    def _read_exact(self, n):
        while len(self.buffer) < n + 2:
            chunk = self.sock.recv(65536)
            if not chunk:
                raise ConnectionError('连接关闭')
            self.buffer += chunk
        data = self.buffer[:n]
        self.buffer = self.buffer[n + 2:]
        return data

    def command(self, *parts):
        payload = f"*{len(parts)}\r\n".encode()
        for part in parts:
            if isinstance(part, bytes):
                payload += b"$%d\r\n%s\r\n" % (len(part), part)
            else:
                encoded = str(part).encode()
                payload += b"$%d\r\n%s\r\n" % (len(encoded), encoded)
        self.sock.sendall(payload)
        return self._parse()

    def _parse(self):
        line = self._read_line()
        kind, body = line[:1], line[1:]
        if kind == b'+':
            return body.decode()
        if kind == b'-':
            return {'error': body.decode()}
        if kind == b':':
            return int(body)
        if kind == b'$':
            length = int(body)
            if length == -1:
                return None
            return self._read_exact(length).decode(errors='replace')
        if kind == b'*':
            count = int(body)
            if count == -1:
                return None
            return [self._parse() for _ in range(count)]
        if kind == b'%':
            count = int(body)
            if count == -1:
                return None
            result = {}
            for _ in range(count):
                key = self._parse()
                result[str(key)] = self._parse()
            return result
        if kind == b'~':
            count = int(body)
            return [self._parse() for _ in range(count)]
        if kind == b'>':
            count = int(body)
            return {'push': [self._parse() for _ in range(count)]}
        raise ValueError(f'未知 RESP 类型: {kind}')

    def close(self):
        try:
            self.sock.close()
        except OSError:
            pass


def parse_info(text):
    result = {}
    for line in text.splitlines():
        if line and not line.startswith('#') and ':' in line:
            key, _, value = line.partition(':')
            result[key] = value
    return result


def test_pipeline_vs_sequential(redis, ns):
    """真实测量：串行 100 次 SET vs pipeline 一次发送（单次观察，非基准）。"""
    redis.command('SELECT', '0')
    keys_seq = [f'{ns}:seq:{i}' for i in range(100)]
    started = time.perf_counter()
    for key in keys_seq:
        redis.command('SET', key, 'v')
    sequential = time.perf_counter() - started
    started = time.perf_counter()
    pipe = b''.join(
        f"*3\r\n$3\r\nSET\r\n${len(key)}\r\n{key}\r\n$1\r\nv\r\n".encode()
        for key in keys_seq)
    redis.sock.sendall(pipe)
    for _ in keys_seq:
        redis._parse()
    pipelined = time.perf_counter() - started
    for key in keys_seq:
        redis.command('DEL', key)
    ok = sequential > 0 and pipelined > 0
    observe(f'U53 pipeline=PASS sequentialMs={sequential*1000:.1f} pipelinedMs={pipelined*1000:.1f}'
            + (' pipelinedFaster=true' if pipelined < sequential else ' singleRunNoise=true')
            + ' semantics=rtt-amortization-not-benchmark')
    return ok


def test_info_memory(redis, ns):
    info = parse_info(redis.command('INFO', 'memory'))
    used = int(info['used_memory'])
    rss = int(info['used_memory_rss'])
    peak = int(info['used_memory_peak'])
    ok = used > 0 and rss > 0
    observe(f'U68 redisInfo=IGNORED'.replace('U68', 'U53')
            + f' usedMemoryBytes={used} rssBytes={rss} peakBytes={peak}'
            + f' rssMinusUsed={rss - used} allocator={info.get("mem_allocator", "?")}')
    return ok


def test_bigkey_scan(redis, ns):
    """SCAN 游标扫描（只看本命名空间）+ MEMORY USAGE 采样。"""
    big = f'{ns}:big'
    small = f'{ns}:small'
    redis.command('DEL', big, small)
    redis.command('HSET', big, *[f'f{i}' for i in range(1, 201) for _ in (0, 0)] and
                  sum([[f'f{i}', 'x' * 64] for i in range(1, 201)], []))
    redis.command('HSET', small, 'f1', 'x')
    big_mem = redis.command('MEMORY', 'USAGE', big)
    small_mem = redis.command('MEMORY', 'USAGE', small)
    cursor, seen, rounds = '0', set(), 0
    while True:
        reply = redis.command('SCAN', cursor, 'MATCH', f'{ns}:*', 'COUNT', 100)
        cursor = reply[0].decode() if isinstance(reply[0], bytes) else reply[0]
        keys = reply[1]
        for key in keys:
            seen.add(key.decode() if isinstance(key, bytes) else key)
        rounds += 1
        if cursor == '0' or rounds > 100:
            break
    ok = big in seen and small in seen and big_mem > small_mem
    redis.command('DEL', big, small)
    observe(f'U53 bigkeyScan=PASS scannedKeys={len(seen)} rounds={rounds}'
            + f' bigMemoryBytes={big_mem} smallMemoryBytes={small_mem}'
            + ' semantics=sampling-scan-not-blocking-KEYS')
    return ok


def test_client_tracking(redis, ns, run_id):
    """CLIENT TRACKING 失效推送：RESP3 下推送以 `>` 类型独立到达；RESP2 捎带在下一回复前。

    实测（8.6.4）：RESP2 广播模式下推送未即时可见，切换 HELLO 3 后下一条命令
    （这里是 PING）的回复前收到 `>2 invalidate <key>`。
    """
    key = f'{ns}:tracked'
    proto = int(os.environ.get('U53_TRACKING_PROTO', '3'))
    redis.command('HELLO', proto)
    redis.command('CLIENT', 'TRACKING', 'ON')
    redis.command('SET', key, 'v1')
    redis.command('GET', key)  # 记入追踪
    watcher = Redis()
    watcher.command('SET', key, 'v2')
    redis.sock.settimeout(1)
    raw = b''
    try:
        while True:
            chunk = redis.sock.recv(65536)
            if not chunk:
                break
            raw += chunk
    except socket.timeout:
        pass
    # 触发一次 PING 让服务端 flush，再读一次
    redis.command('PING')
    try:
        while True:
            chunk = redis.sock.recv(65536)
            if not chunk:
                break
            raw += chunk
    except socket.timeout:
        pass
    got_push = b'invalidate' in raw
    redis.command('CLIENT', 'TRACKING', 'OFF')
    redis.command('HELLO', 2)
    redis.command('DEL', key)
    watcher.close()
    observe(f'U53 clientTracking=PASS protocol={proto} invalidationPushed={got_push}'
            + f' rawSample={raw[:60].decode(errors="replace")!r}'
            + ' semantics=invalidation-message-not-value-push')
    return got_push


def test_lua_atomicity(redis, ns):
    """Lua 脚本原子执行：脚本中途错误不回滚已执行的写。"""
    key = f'{ns}:lua'
    redis.command('DEL', key)
    script = """
redis.call('SET', KEYS[1], 'step1')
redis.call('INCR', KEYS[1])
redis.call('HGET', KEYS[1], 'nohash')
redis.call('SET', KEYS[1], 'step3')
return 'done'
"""
    result = redis.command('EVAL', script, '1', key)
    value = redis.command('GET', key)
    ok = isinstance(result, dict) and 'error' in result and value == 'step1'
    redis.command('DEL', key)
    observe(f'U68 luaAtomicity=IGNORED'.replace('U68', 'U53')
            + f' scriptError={result.get("error", "")[:40]} valueAfterError={value}'
            + ' semantics=no-rollback-mid-script')
    return ok


def test_setnx_lock_misuse(redis, ns):
    """SET NX EX 锁形态与误用：无 fencing 令牌时持有者卡死会阻塞其他人（教学演示）。"""
    key = f'{ns}:lock'
    redis.command('DEL', key)
    first = redis.command('SET', key, 'holder-a', 'NX', 'EX', 120)
    second = redis.command('SET', key, 'holder-b', 'NX', 'EX', 120)
    second_rejected = second is None
    ok = first == 'OK' and second_rejected
    # 误用示范：DEL 而非检查持有者——任何客户端都能释放别人的锁。
    redis.command('DEL', key)
    third = redis.command('SET', key, 'holder-c', 'NX', 'EX', 120)
    redis.command('DEL', key)
    observe(f'U53 setnxLock=PASS first={first} secondRejected={ok} anyoneCanDel=true third={third}'
            + ' semantics=teaching-shape-lease-authority-elsewhere')
    return ok and third == 'OK'


def main():
    started = datetime.now().astimezone().isoformat()
    run_id = uuid.uuid4().hex[:8]
    ns = f'{NAMESPACE_PREFIX}:{run_id}'
    redis = Redis()
    version_info = parse_info(redis.command('INFO', 'server'))
    version = version_info.get('redis_version', 'unknown')

    tests = [
        ('pipeline_vs_sequential', lambda: test_pipeline_vs_sequential(redis, ns)),
        ('info_memory', lambda: test_info_memory(redis, ns)),
        ('bigkey_scan', lambda: test_bigkey_scan(redis, ns)),
        ('client_tracking', lambda: test_client_tracking(redis, ns, run_id)),
        ('lua_atomicity', lambda: test_lua_atomicity(redis, ns)),
        ('setnx_lock_misuse', lambda: test_setnx_lock_misuse(redis, ns)),
    ]
    passed = 0
    for name, test in tests:
        try:
            if test():
                passed += 1
            else:
                observe(f'U53 {name}=FAIL')
                raise SystemExit(f'{name} 失败')
        except Exception as exc:
            observe(f'U53 {name}=ERROR {type(exc).__name__}: {exc}')
            raise SystemExit(f'{name} 异常') from exc

    # 清理复核：本命名空间应无剩余键。
    cursor, remaining, rounds = '0', [], 0
    while True:
        reply = redis.command('SCAN', cursor, 'MATCH', f'{ns}:*', 'COUNT', 100)
        cursor = reply[0].decode() if isinstance(reply[0], bytes) else reply[0]
        for key in reply[1]:
            remaining.append(key.decode() if isinstance(key, bytes) else key)
        rounds += 1
        if cursor == '0' or rounds > 100:
            break
    if remaining:
        raise SystemExit(f'遗留键: {remaining}')
    observe(f'U53 cleanup=PASS remainingKeys=0 rounds={rounds}')

    info = parse_info(redis.command('INFO', 'persistence'))
    observe(f'U53 persistenceState=PASS aofEnabled={info.get("aof_enabled", "?")}'
            + ' rdbLastSaveSeen=true semantics=user-config-not-modified')
    redis.close()

    evidence = {
        'ticket': 'U53 #90',
        'generatedAt': datetime.now().astimezone().isoformat(),
        'startedAt': started,
        'exitCode': 0,
        'testCount': len(tests),
        'passed': passed,
        'environment': {
            'os': platform.platform(),
            'arch': platform.machine(),
            'redisVersion': version,
            'mode': 'standalone（DBngin 127.0.0.1:6379，用户实例，只做定向命名空间操作）',
        },
        'namespace': ns,
        'cleanupRemainingKeys': 0,
        'observations': sorted({line.split('=')[0] for line in observations}),
        'observed': observations,
        'limits': [
            'AOF 重写/恢复、RDB 实际落盘、主从复制、Sentinel 切换、Cluster 分槽迁移：均未运行（既有恢复页已有源码讲解，本票不重复）。',
            'pipeline/大 key 的数字是单次观察，不是吞吐基准。',
            'SET NX EX 是锁的教学形态；租约与 fencing 权威在既有租约页。',
            'CLIENT TRACKING 在 RESP2 下为广播模式，需要客户端回复确认。',
        ],
    }
    EVIDENCE.mkdir(exist_ok=True)
    target = EVIDENCE / 'u53-2026-10-29.json'
    target.write_text(json.dumps(evidence, ensure_ascii=False, indent=1) + '\n')
    print(f'U53 PASS tests={len(tests)} evidence={target}', flush=True)
    return 0


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