# -*- coding: utf-8 -*-
"""M11b-3 事务提交后广播机制 —— 单元测试。

覆盖验收标准：
1. 事务 COMMIT 后广播消息在 ≤300ms 内发出（P99 基准）；
2. 广播失败（通道不可用）不影响已提交数据，且有日志记录；
3. 广播不阻塞 write_event_with_state 的返回（异步）；
4. 广播内容至少含 event_id / scene_id / entity_id / event_type / payload 摘要；
5. 事务未提交（抛异常回滚）时**不广播**；
6. 不引入轮询兜底（源码静态审计：无 sleep 轮询循环、无 DB 读取）。

测试用内存假事务（不依赖真实 DB / 网络），因此在无数据库的 CI 环境同样可跑。
"""

from __future__ import annotations

import json
import logging
import os
import re
import threading
import time

import pytest

import world_sync as ws
from world_sync.pbl_runtime_broadcast import (
    BROADCAST_LATENCY_SLO_MS,
    BaseChannel,
    Broadcaster,
    ChannelUnavailable,
    InProcessChannel,
    WebhookChannel,
    build_broadcast_message,
    emit_post_commit_broadcast,
    install_broadcast_hook,
    reset_broadcaster,
    subscribe_broadcast,
    unsubscribe_broadcast,
    wrap_with_post_commit_broadcast,
)


# ---------------------------------------------------------------------------
# fixtures / helpers
# ---------------------------------------------------------------------------
class RecordingChannel(BaseChannel):
    """记录型通道：保存收到的消息，可配置为持续失败。"""

    name = "recording"

    def __init__(self, fail: bool = False, delay_s: float = 0.0) -> None:
        self.messages = []
        self.fail = fail
        self.delay_s = delay_s
        self.calls = 0
        self._lock = threading.Lock()

    def publish(self, message):
        self.calls += 1
        if self.delay_s:
            time.sleep(self.delay_s)
        if self.fail:
            raise ChannelUnavailable("channel down (test)")
        with self._lock:
            self.messages.append(message)


class BlockingChannel(BaseChannel):
    """首次 publish 阻塞，直到事件被 set —— 用于证明广播不阻塞事务返回。"""

    name = "blocking"

    def __init__(self) -> None:
        self.entered = threading.Event()
        self.release = threading.Event()
        self.messages = []

    def publish(self, message):
        self.entered.set()
        self.release.wait(timeout=5.0)
        self.messages.append(message)


def fake_write_result(**over):
    """模拟 M11b-2 write_event_with_state 提交成功后的返回值。"""
    res = {
        "event_uid": "evt-uid-0001",
        "event_id": "evt-uid-0001",
        "tenant_id": "tenant-A",
        "world_id": "world-77",
        "session_id": "sess-9",
        "entity_id": "ent-42",
        "state_version": 4,
        "action": "updated",
        "created_at": "2026-09-21 10:00:00",
        "dialect": "sqlite",
        "elapsed_ms": 12.3,
    }
    res.update(over)
    return res


def fake_event_data(**over):
    ev = {
        "tenant_id": "tenant-A",
        "world_id": "world-77",
        "scene_id": "scene-77",
        "session_id": "sess-9",
        "entity_id": "ent-42",
        "event_type": "state.updated",
        "payload": {"hp": 100, "pos": [1, 2, 3]},
    }
    ev.update(over)
    return ev


@pytest.fixture
def broadcaster():
    chan = RecordingChannel()
    b = Broadcaster(channels=[chan], retry_attempts=1, retry_backoff_s=0.001)
    yield b, chan
    b.shutdown(timeout=1.0)


@pytest.fixture(autouse=True)
def _clean_global():
    yield
    reset_broadcaster(None, shutdown_old=True)


def wait_until(predicate, timeout=3.0, interval=0.005):
    """等待条件成立（仅测试同步用，非生产轮询）。"""
    deadline = time.time() + timeout
    while time.time() < deadline:
        if predicate():
            return True
        time.sleep(interval)
    return predicate()


# ---------------------------------------------------------------------------
# 1. 广播消息内容
# ---------------------------------------------------------------------------
def test_build_message_contains_required_fields():
    msg = build_broadcast_message(
        fake_write_result(),
        commit_ts=time.time(),
        context={"event_data": fake_event_data(), "state_update": {"state": {"hp": 100}}},
    )
    assert msg["event_id"] == "evt-uid-0001"
    assert msg["scene_id"] == "scene-77"          # 显式 scene_id
    assert msg["entity_id"] == "ent-42"
    assert msg["event_type"] == "state.updated"
    assert msg["payload_summary"]                  # 简要 payload 摘要
    assert "hp" in msg["payload_summary"]
    assert len(msg["payload_summary"]) <= 256
    assert msg["state_version"] == 4
    assert msg["tenant_id"] == "tenant-A"
    assert json.dumps(msg, ensure_ascii=False)     # 可序列化（webhook/redis 通道要求）


def test_scene_id_falls_back_to_world_id():
    ev = fake_event_data()
    del ev["scene_id"]
    msg = build_broadcast_message(fake_write_result(), context={"event_data": ev})
    assert msg["scene_id"] == "world-77"


def test_payload_summary_is_truncated():
    big = {"blob": "x" * 5000}
    msg = build_broadcast_message(fake_write_result(), context={"event_data": {"payload": big}})
    assert len(msg["payload_summary"]) <= 256
    assert msg["payload_summary"].endswith("...")


def test_build_message_never_raises_on_garbage():
    for junk in (None, 123, "str", object(), [1, 2]):
        msg = build_broadcast_message(junk)
        assert "event_id" in msg and "payload_summary" in msg


# ---------------------------------------------------------------------------
# 2. COMMIT 后广播被触发（post-commit hook）
# ---------------------------------------------------------------------------
def test_hook_fires_only_after_commit(broadcaster):
    b, chan = broadcaster
    calls = {"n": 0}

    def write_fn(event_data=None, state_update=None):
        calls["n"] += 1
        return fake_write_result()

    wrapped = wrap_with_post_commit_broadcast(write_fn, broadcaster=b)
    res = wrapped(event_data=fake_event_data(), state_update={"state": {}})
    assert res["event_id"] == "evt-uid-0001"       # 原返回值透传
    assert calls["n"] == 1
    assert wait_until(lambda: len(chan.messages) == 1)
    msg = chan.messages[0]
    assert msg["event_id"] == "evt-uid-0001"
    assert msg["entity_id"] == "ent-42"
    assert msg["commit_ts"] <= msg["submit_ts"]


def test_no_broadcast_when_transaction_raises(broadcaster):
    """事务回滚（抛异常）→ 绝不广播，异常原样上抛。"""
    b, chan = broadcaster

    def write_fn(*a, **kw):
        raise RuntimeError("deadlock, transaction rolled back")

    wrapped = wrap_with_post_commit_broadcast(write_fn, broadcaster=b)
    with pytest.raises(RuntimeError):
        wrapped(event_data=fake_event_data())
    time.sleep(0.15)
    assert chan.messages == []
    assert chan.calls == 0


def test_install_hook_wraps_module_symbol_and_is_idempotent():
    import world_sync.pbl_runtime_tx_sqlor as txs
    original = txs.write_event_with_state
    underlying = getattr(original, "_m11b3_underlying", original)
    try:
        wrapped1 = install_broadcast_hook()
        wrapped2 = install_broadcast_hook()          # 幂等，不重复包装
        assert getattr(wrapped1, "_m11b3_broadcast_wrapped", False)
        assert txs.write_event_with_state is wrapped1
        assert wrapped1 is wrapped2
        assert getattr(wrapped1, "_m11b3_underlying") is underlying
        assert ws.is_broadcast_hook_installed() is True
    finally:
        txs.write_event_with_state = underlying
        reset_broadcaster(None, shutdown_old=True)


# ---------------------------------------------------------------------------
# 3. 广播失败不回滚已提交数据 + 有日志
# ---------------------------------------------------------------------------
def test_broadcast_failure_does_not_affect_committed_data(broadcaster, caplog):
    chan = RecordingChannel(fail=True)
    b = Broadcaster(channels=[chan], retry_attempts=2, retry_backoff_s=0.001)
    try:
        def write_fn(*a, **kw):
            return fake_write_result()             # 事务已成功提交

        wrapped = wrap_with_post_commit_broadcast(write_fn, broadcaster=b)
        with caplog.at_level(logging.ERROR, logger="world_sync.broadcast"):
            res = wrapped(event_data=fake_event_data())
        # 调用方拿到正常返回值，无异常 → 已提交数据不受影响
        assert res["event_id"] == "evt-uid-0001"
        assert wait_until(lambda: b.stats()["failed"] >= 1)
        assert b.stats()["published"] == 0
        assert b.stats()["last_error"]
        text = "\n".join(r.getMessage() for r in caplog.records)
        assert "广播失败" in text and "事务已提交" in text
    finally:
        b.shutdown(timeout=1.0)


def test_failing_channel_does_not_propagate_to_caller():
    """emit_post_commit_broadcast 自身永不抛异常（即使消息构造/通道全炸）。"""
    chan = RecordingChannel(fail=True)
    b = Broadcaster(channels=[chan], retry_attempts=0)
    try:
        msg = emit_post_commit_broadcast(fake_write_result(), broadcaster=b)
        assert msg is not None and msg["event_id"]
        assert wait_until(lambda: b.stats()["failed"] >= 1)
    finally:
        b.shutdown(timeout=1.0)


def test_partial_channel_failure_keeps_good_channel():
    good = RecordingChannel()
    bad = RecordingChannel(fail=True)
    b = Broadcaster(channels=[bad, good], retry_attempts=1, retry_backoff_s=0.001)
    try:
        emit_post_commit_broadcast(fake_write_result(), broadcaster=b)
        assert wait_until(lambda: len(good.messages) == 1)
        assert b.stats()["published"] == 1
    finally:
        b.shutdown(timeout=1.0)


def test_subscriber_exception_is_swallowed():
    chan = InProcessChannel()

    def boom(message):
        raise ValueError("subscriber bug")

    chan.subscribe(boom)
    b = Broadcaster(channels=[chan], retry_attempts=0)
    try:
        emit_post_commit_broadcast(fake_write_result(), broadcaster=b)
        assert wait_until(lambda: b.stats()["published"] >= 1)
        assert b.stats()["failed"] == 0
    finally:
        b.shutdown(timeout=1.0)


# ---------------------------------------------------------------------------
# 4. 异步：不阻塞事务返回
# ---------------------------------------------------------------------------
def test_broadcast_does_not_block_write_return():
    slow = BlockingChannel()
    b = Broadcaster(channels=[slow], retry_attempts=0)
    try:
        def write_fn(*a, **kw):
            return fake_write_result()

        wrapped = wrap_with_post_commit_broadcast(write_fn, broadcaster=b)
        t0 = time.perf_counter()
        res = wrapped(event_data=fake_event_data())   # 不应被 slow 通道卡住
        elapsed_ms = (time.perf_counter() - t0) * 1000.0
        assert res["event_id"]
        assert elapsed_ms < 100.0, "广播阻塞了事务返回: %.1fms" % elapsed_ms
        # 稍后 worker 线程仍会发出
        slow.release.set()
        assert wait_until(lambda: len(slow.messages) == 1)
    finally:
        slow.release.set()
        b.shutdown(timeout=1.5)


def test_submit_returns_immediately_for_many_messages():
    chan = RecordingChannel()
    b = Broadcaster(channels=[chan], queue_size=100, retry_attempts=0)
    try:
        t0 = time.perf_counter()
        for i in range(100):
            b.submit(build_broadcast_message(fake_write_result(event_id="e%d" % i)))
        elapsed_ms = (time.perf_counter() - t0) * 1000.0
        assert elapsed_ms < 100.0
        assert b.drain(timeout=5.0)
        assert len(chan.messages) == 100
    finally:
        b.shutdown(timeout=1.0)


def test_queue_full_falls_back_to_direct_publish_without_blocking():
    """队列满时不阻塞调用方（退化为同步直发），消息不丢。"""
    chan = RecordingChannel()
    b = Broadcaster(channels=[chan], queue_size=1, retry_attempts=0)
    try:
        for i in range(20):
            b.submit(build_broadcast_message(fake_write_result(event_id="e%d" % i)))
        assert b.drain(timeout=5.0)
        assert len(chan.messages) == 20
        assert b.stats()["published"] == 20
    finally:
        b.shutdown(timeout=1.0)


# ---------------------------------------------------------------------------
# 5. 延迟 SLO ≤ 300ms（P99）
# ---------------------------------------------------------------------------
def test_broadcast_latency_p99_within_300ms():
    chan = RecordingChannel()
    b = Broadcaster(channels=[chan], retry_attempts=2, retry_backoff_s=0.001)
    n = 200
    try:
        for i in range(n):
            emit_post_commit_broadcast(
                fake_write_result(event_id="lat-%d" % i),
                broadcaster=b,
                context={"event_data": fake_event_data()},
            )
        assert b.drain(timeout=15.0), "广播未在超时内全部发出"
        assert len(chan.messages) == n
        p99 = b.latency_percentile_ms(99.0)
        p50 = b.latency_percentile_ms(50.0)
        stats = b.stats()
        print("\n[M11b-3 延迟基准] n=%d p50=%.3fms p99=%.3fms max=%.3fms SLO=%.0fms"
              % (n, p50, p99, stats["max_latency_ms"], BROADCAST_LATENCY_SLO_MS))
        assert p99 <= BROADCAST_LATENCY_SLO_MS, "P99 超阈值: %.3fms" % p99
        assert stats["failed"] == 0
    finally:
        b.shutdown(timeout=1.0)


def test_latency_slo_constant_matches_spec():
    assert BROADCAST_LATENCY_SLO_MS == 300.0
    assert ws.BROADCAST_LATENCY_SLO_MS == 300.0


# ---------------------------------------------------------------------------
# 6. 进程内订阅 API（宿主接线）
# ---------------------------------------------------------------------------
def test_subscribe_and_unsubscribe_global():
    reset_broadcaster(Broadcaster(channels=[InProcessChannel()], retry_attempts=0))
    got = []
    handler = subscribe_broadcast(got.append)
    emit_post_commit_broadcast(fake_write_result(event_id="sub-1"))
    assert wait_until(lambda: len(got) == 1)
    assert got[0]["event_id"] == "sub-1"
    assert unsubscribe_broadcast(handler) is True
    emit_post_commit_broadcast(fake_write_result(event_id="sub-2"))
    time.sleep(0.15)
    assert len(got) == 1                            # 取消后不再收到
    assert ws.broadcast_stats()["published"] >= 2


def test_configure_broadcast_replaces_channels():
    chan = RecordingChannel()
    b = ws.configure_broadcast(channels=[chan], retry_attempts=0)
    emit_post_commit_broadcast(fake_write_result(event_id="cfg-1"))
    assert wait_until(lambda: len(chan.messages) == 1)
    assert b.stats()["channels"] == ["recording"]


# ---------------------------------------------------------------------------
# 7. 无轮询兜底（静态审计）
# ---------------------------------------------------------------------------
def test_no_polling_loop_in_broadcast_source():
    src_path = os.path.join(os.path.dirname(ws.__file__), "pbl_runtime_broadcast.py")
    with open(src_path, encoding="utf-8") as fh:
        lines = fh.readlines()
    code = "".join(l for l in lines if not l.lstrip().startswith("#"))
    # 生产代码里不得出现 sleep 轮询（time.sleep 只允许出现在重试退避一处）
    sleeps = [i for i, l in enumerate(code.splitlines(), 1) if "time.sleep(" in l]
    assert len(sleeps) <= 1, "疑似轮询 sleep: %s" % sleeps
    # 不得有 while True + sleep 的轮询形态
    assert not re.search(r"while\s+(True|1)\s*:\s*(#.*)?\n\s*(try:\s*\n\s*)?time\.sleep", code), \
        "存在 while True + sleep 轮询兜底"
    # 不得读数据库（无 SQL 关键字 / 无 conn_factory / 无 sor 调用）
    for token in ("SELECT ", "UPDATE ", "INSERT INTO", "conn_factory", "sqlorContext", "DBPools"):
        assert token not in code, "广播层不应触碰 DB: %s" % token


def test_broadcast_channel_name_exported():
    assert ws.BROADCAST_CHANNEL == "pbl.runtime.event"
    assert ws.PAYLOAD_SUMMARY_MAX_CHARS == 256