"""S5：业务事实直发事件。

取代了原来的「收集器事后扫 `douyu_tasks.result` 反推事件」。本文件守住三件事：

1. **命名构造函数的字段口径**（金额符号、detail、subject）；
2. **幂等**：`source_ref=task_ref(task_id)` —— 让直发与仍在跑的收集器能互相去重，
   也让「runner 轮询」与「供应商回调」不会对同一笔充值重复记账；
3. **发射顺序**：事件必须在 `_mark_task` **之前**发。`_mark_task` 内部 commit，
   而 `emit` 只 RELEASE SAVEPOINT、不提交外层事务，runner 的 `finally` 只 close
   （未提交即回滚）—— 发在后面会被静默丢弃，且 `emit` 仍返回 True。
"""

from datetime import UTC, datetime

import pytest

from tests.support.db import get_test_engine, make_test_session
from web.backend.models import Account, AccountEvent, DouyuTask, User
from web.backend.services import account_events_service as svc
from web.backend.services import business_events as biz

NOW = datetime(2026, 10, 4, 12, 0, 0, tzinfo=UTC)


class _Base:
    def setup_method(self):
        self.engine = get_test_engine()
        self.session = make_test_session(self.engine)
        self.session.query(AccountEvent).delete()
        self.user = User(username="biz-admin", password_hash="h", role="super_admin")
        self.session.add(self.user)
        self.session.flush()
        self.account = Account(
            username="biz-acc", password="p", email="b@e.com", email_password="ep"
        )
        self.session.add(self.account)
        self.session.commit()

    def teardown_method(self):
        self.session.close()

    def events(self, event_type: str) -> list[AccountEvent]:
        return (
            self.session.query(AccountEvent)
            .filter(
                AccountEvent.account_id == self.account.id,
                AccountEvent.event_type == event_type,
            )
            .order_by(AccountEvent.id.asc())
            .all()
        )


class TestConstructors(_Base):
    def test_recharge_records_amount_and_channel(self):
        assert (
            biz.record_recharge(
                self.session,
                self.account,
                task_id=101,
                amount=100,
                channel="wechat_qr",
                douyu_uid="u1",
            )
            is True
        )
        self.session.commit()
        events = self.events(svc.EVENT_RECHARGE)
        assert len(events) == 1
        event = events[0]
        assert event.amount == 100
        assert event.category == svc.CATEGORY_BUSINESS
        assert event.source_ref == "douyu_task:101"
        assert event.detail["channel"] == "wechat_qr"
        assert event.detail["confirmed"] is True

    def test_recharge_without_amount_still_records(self):
        """金额拿不到时宁可记一条金额为空的事件，也不要凭空编一个数字。"""
        biz.record_recharge(self.session, self.account, task_id=102, amount=None)
        self.session.commit()
        event = self.events(svc.EVENT_RECHARGE)[0]
        assert event.amount is None

    def test_recharge_marks_supplier_confirmation_honestly(self):
        """供应商终态成功 ≠ 斗鱼余额已变，必须如实标注。"""
        biz.record_recharge(
            self.session,
            self.account,
            task_id=103,
            amount=50,
            channel="supplier_api",
            confirmed=False,
        )
        self.session.commit()
        event = self.events(svc.EVENT_RECHARGE)[0]
        assert event.detail["confirmed"] is False
        assert "供应商回执" in event.title

    def test_exchange_amount_is_negative_and_subject_is_commodity(self):
        biz.record_exchange(
            self.session,
            self.account,
            task_id=201,
            name="兔兔白日梦",
            score=1200,
            commodity_id="sku-9",
            points_after=800,
            scope="elite",
        )
        self.session.commit()
        event = self.events(svc.EVENT_EXCHANGE)[0]
        assert event.amount == -1200
        assert event.balance_after == 800
        assert event.subject_type == "commodity"
        assert event.subject_key == "sku-9"
        assert event.detail["scope"] == "elite"

    def test_exchange_without_score_keeps_amount_null(self):
        biz.record_exchange(
            self.session, self.account, task_id=202, name="x", score=None
        )
        self.session.commit()
        assert self.events(svc.EVENT_EXCHANGE)[0].amount is None

    def test_points_credit_marks_unconfirmed(self):
        """未确认到账也记，但必须标出来 —— 不能把「预期」写成「已到账」。"""
        biz.record_points_credit(
            self.session,
            self.account,
            task_id=301,
            gain=980,
            points_after=100,
            confirmed=False,
        )
        self.session.commit()
        event = self.events(svc.EVENT_POINTS_CREDIT)[0]
        assert event.amount == 980
        assert event.detail["confirmed"] is False
        assert "未确认" in event.title

    def test_points_credit_confirmed(self):
        biz.record_points_credit(
            self.session, self.account, task_id=302, gain=10, confirmed=True
        )
        self.session.commit()
        event = self.events(svc.EVENT_POINTS_CREDIT)[0]
        assert event.detail["confirmed"] is True
        assert "未确认" not in event.title

    def test_bind_records_scope_and_rebind(self):
        biz.record_bind(
            self.session,
            self.account,
            task_id=401,
            scope="esports",
            role_name="小明",
            rebind=True,
        )
        self.session.commit()
        event = self.events(svc.EVENT_BIND_OK)[0]
        assert event.subject_type == "esports_role"
        assert event.subject_key == "小明"
        assert event.detail["rebind"] is True
        assert "换绑" in event.title

    def test_bind_weak_is_flagged(self):
        biz.record_bind(
            self.session,
            self.account,
            task_id=402,
            scope="elite",
            role_name="小华",
            weak=True,
        )
        self.session.commit()
        event = self.events(svc.EVENT_BIND_OK)[0]
        assert event.detail["confirmed"] is False
        assert "未回查确认" in event.title


class TestIdempotency(_Base):
    """同一个 task 只能对应一条事实 —— 这是直发与收集器/回调共存的基石。"""

    def test_same_task_id_does_not_duplicate(self):
        biz.record_recharge(self.session, self.account, task_id=500, amount=10)
        self.session.commit()
        # 模拟「另一个路径也观测到同一笔」（收集器重跑 / 供应商回调）
        biz.record_recharge(self.session, self.account, task_id=500, amount=10)
        self.session.commit()
        assert len(self.events(svc.EVENT_RECHARGE)) == 1

    def test_different_task_id_is_a_new_fact(self):
        biz.record_recharge(self.session, self.account, task_id=501, amount=10)
        self.session.commit()
        biz.record_recharge(self.session, self.account, task_id=502, amount=10)
        self.session.commit()
        assert len(self.events(svc.EVENT_RECHARGE)) == 2

    def test_task_ref_matches_old_collector_key(self):
        """必须与旧收集器的 key 完全一致，否则迁移窗口内会重复记账。"""
        assert biz.task_ref(123) == "douyu_task:123"


class TestEmitSurvivesWorkerSessionClose(_Base):
    """发射顺序的**行为验证**：事件要能挺过 runner 的 `finally: worker_db.close()`。

    runner 的每个任务在 `_execute_one` 的 finally 里只 close 不 commit。
    `_mark_task` 内部有 commit，所以「紧贴 _mark_task 之前 emit」会被那次 commit
    一起提交；若把 emit 放在 _mark_task 之后，它就落在一个没人提交的事务里，
    close 时被回滚 —— 而且 `emit` 照样返回 True，没有任何日志。
    """

    def test_event_committed_together_with_mark_task(self):
        # 模拟 runner：emit 后由 _mark_task 的 commit 收尾
        biz.record_recharge(self.session, self.account, task_id=600, amount=7)
        self.session.commit()  # 等价于 _mark_task 内部的 commit
        self.session.close()  # 等价于 runner 的 finally

        fresh = make_test_session(self.engine)
        try:
            rows = (
                fresh.query(AccountEvent)
                .filter(AccountEvent.source_ref == "douyu_task:600")
                .all()
            )
            assert len(rows) == 1, "事件没有挺过 session close —— 发射顺序错了"
        finally:
            fresh.close()

    def test_emit_without_commit_is_lost_on_close(self):
        """反证：不 commit 就 close，事件确实会丢（这就是顺序规则存在的理由）。"""
        biz.record_recharge(self.session, self.account, task_id=601, amount=7)
        self.session.close()  # 故意不 commit

        fresh = make_test_session(self.engine)
        try:
            rows = (
                fresh.query(AccountEvent)
                .filter(AccountEvent.source_ref == "douyu_task:601")
                .all()
            )
            assert rows == [], "若不 commit 也能落库，说明事务边界与预期不符"
        finally:
            fresh.close()


class TestCollectorIsGone:
    """收集器必须被彻底删掉，不留半条命。"""

    def test_module_removed(self):
        import importlib

        try:
            importlib.import_module(
                "web.backend.services.automation.jobs.account_events_collect"
            )
        except ModuleNotFoundError:
            return
        raise AssertionError("收集器模块还在")

    def test_no_collector_symbols_left(self):
        assert not hasattr(svc, "COLLECTED_TASK_TYPES")
        assert not hasattr(svc, "_task_to_event")

    def test_job_key_not_registered(self):
        """⚠️ `registered_jobs()` 返回的是 **dict**（key → JobSpec）。

        写成 `{spec.key for spec in registered_jobs()}` 会遍历出字符串键、
        `spec.key` 直接报错；而如果它恰好是空字典，推导式又会得到空集让断言**假绿**。
        所以这里显式 register 后再按键判断。
        """
        from web.backend.services.automation import (
            register_default_jobs,
            registered_jobs,
        )

        register_default_jobs()
        keys = set(registered_jobs())
        assert "account_events_collect" not in keys
        # 反向确认这个断言不是假绿：新作业确实在里面
        assert "douyu_tasks_retention" in keys


class TestEmitOrderingInRunners:
    """源码扫描：runner 里的 `record_*` 必须在 `_mark_task(..., "success")` **之前**。

    这条规则很隐蔽（违反了不会报错、只是事件静默消失），所以用用例钉住。
    做法：对每个 `record_*` 调用，找它之后出现的第一个 `_mark_task(`，该调用里
    必须带 `"success"`；如果事件被写在 mark 之后，那么它后面第一个 `_mark_task`
    要么是另一个分支的、要么根本不存在 —— 两种情况都会让用例变红。
    """

    RUNNER_FILES = (
        "web/backend/services/douyu_runner_goods.py",
        "web/backend/services/douyu_runner_gold.py",
        "web/backend/services/douyu_runner_bind.py",
        "web/backend/services/douyu_runner_donate.py",
        "web/backend/services/douyu_runner_xpd.py",
    )

    def test_record_calls_precede_success_mark(self):
        import re
        from pathlib import Path

        root = Path(__file__).resolve().parents[1]
        record_re = re.compile(r"\brecord_(recharge|exchange|points_credit|bind)\(")
        offenders: list[str] = []
        checked = 0

        for rel in self.RUNNER_FILES:
            source = (root / rel).read_text(encoding="utf-8")
            for match in record_re.finditer(source):
                checked += 1
                tail = source[match.end() :]
                mark = tail.find("_mark_task(")
                line_no = source[: match.start()].count("\n") + 1
                if mark == -1:
                    offenders.append(f"{rel}:{line_no} 之后没有 _mark_task")
                    continue
                # 取这次 _mark_task 调用的前若干字符，判断是不是 success 分支
                snippet = tail[mark : mark + 400]
                if '"success"' not in snippet:
                    offenders.append(
                        f"{rel}:{line_no} 之后的第一个 _mark_task 不是 success"
                    )

        assert checked >= 10, f"只扫到 {checked} 个 record_ 调用，扫描规则可能失效了"
        assert not offenders, (
            "record_* 必须在 _mark_task(success) **之前**调用"
            "（否则事件会被 runner 的 close 静默回滚）：\n" + "\n".join(offenders)
        )


class TestTaskReadyForApi(_Base):
    def test_douyu_task_still_works(self):
        """兜底：删掉收集器不应影响 douyu_tasks 本身的读写。"""
        task = DouyuTask(
            batch_id="b1",
            account_id=self.account.id,
            task_type="exchange_goods",
            status="success",
            result={"commodity_name": "x"},
            created_by=self.user.id,
        )
        self.session.add(task)
        self.session.commit()
        assert task.id is not None


class TestAssetOutflow(_Base):
    """资产**出项**也要有金额 —— 否则进项有账、出项没账，对不了账。

    这是审「哪些操作需要进账号生命周期」时发现的：充值/积分到账都有金额，
    而开通（花积分）、送礼（花鱼翅）几乎没有 —— 能算出「一共充了多少」，
    算不出「花哪儿了」。
    """

    def test_gold_spent_is_negative(self):
        assert (
            biz.record_gold_spent(
                self.session,
                self.account,
                task_id=701,
                amount=100,
                reason="赠送精英令 x1",
                source_label="赠送精英令",
                balance_after=50,
            )
            is True
        )
        self.session.commit()
        event = self.events(svc.EVENT_GOLD_SPENT)[0]
        assert event.amount == -100  # 出项用负数，和 exchange 一致
        assert event.balance_after == 50
        assert event.category == svc.CATEGORY_BUSINESS
        assert event.source_ref == "douyu_task:701:gold"

    def test_gold_spent_skips_non_positive(self):
        """余额没减少（或拿不到）就不发 —— 不要记一笔没发生的消耗。"""
        for bad in (None, 0, -5):
            assert (
                biz.record_gold_spent(
                    self.session, self.account, task_id=702, amount=bad
                )
                is False
            )
        self.session.commit()
        assert self.events(svc.EVENT_GOLD_SPENT) == []

    def test_points_expire_defaults_to_unconfirmed(self):
        """到期金额默认是**预计**的 —— 我们并没有亲眼看到扣减。"""
        biz.record_points_expire(
            self.session,
            self.account,
            amount=500,
            expires_at=NOW,
        )
        self.session.commit()
        event = self.events(svc.EVENT_POINTS_EXPIRE)[0]
        assert event.amount == -500
        assert event.detail["confirmed"] is False
        assert "预计" in event.title

    def test_points_expire_can_be_confirmed(self):
        biz.record_points_expire(self.session, self.account, amount=500, confirmed=True)
        self.session.commit()
        event = self.events(svc.EVENT_POINTS_EXPIRE)[0]
        assert event.detail["confirmed"] is True
        assert "预计" not in event.title

    def test_two_events_from_one_task_do_not_collide(self):
        """同一个任务既要「积分到账」又要「消耗鱼翅」——

        共用 `task_ref` 会撞唯一键、后写的那条被**静默丢弃**。副事件带后缀避免它。
        """
        biz.record_points_credit(
            self.session, self.account, task_id=703, gain=10, source_label="精英令"
        )
        biz.record_gold_spent(
            self.session,
            self.account,
            task_id=703,
            amount=100,
            source_label="赠送精英令",
        )
        self.session.commit()
        assert len(self.events(svc.EVENT_POINTS_CREDIT)) == 1
        assert len(self.events(svc.EVENT_GOLD_SPENT)) == 1, (
            "副事件被主事件的幂等键挤掉了"
        )


class TestPointsExpireJob(_Base):
    """到期补记作业：扫「已到期」的账号，写一条事件并清标记。"""

    def _job(self):
        from web.backend.models import AutomationJob

        return AutomationJob(
            key="points_expire_check",
            title="积分到期补记",
            interval_seconds=86400,
            batch_size=200,
            concurrency=1,
        )

    def _run(self):
        from web.backend.services.automation.jobs import points_expire

        out = points_expire.run_points_expire_check(self.session, self._job())
        self.session.commit()
        return out

    def test_expired_points_are_recorded_once(self):
        from datetime import timedelta

        now = datetime.now(UTC)
        self.account.points = 500
        self.account.points_expire_at = now - timedelta(days=1)
        self.session.commit()

        out = self._run()
        assert out["written"] == 1
        events = self.events(svc.EVENT_POINTS_EXPIRE)
        assert len(events) == 1
        assert events[0].amount == -500
        assert events[0].detail["confirmed"] is False
        # 标记被清掉 → 再跑一次不会重复记
        self.session.refresh(self.account)
        assert self.account.points_expire_at is None
        assert self._run()["written"] == 0

    def test_not_yet_expired_is_left_alone(self):
        from datetime import timedelta

        now = datetime.now(UTC)
        self.account.points = 500
        self.account.points_expire_at = now + timedelta(days=3)
        self.session.commit()

        assert self._run()["written"] == 0
        self.session.refresh(self.account)
        assert self.account.points_expire_at is not None, "没到期的标记不该被清"

    def test_zero_points_clears_marker_without_event(self):
        """积分已经为 0 就没有东西可到期 —— 写一条 -0 只是噪声。"""
        from datetime import timedelta

        now = datetime.now(UTC)
        self.account.points = 0
        self.account.points_expire_at = now - timedelta(days=1)
        self.session.commit()

        out = self._run()
        assert out["written"] == 0
        assert out["skipped_zero"] == 1
        assert self.events(svc.EVENT_POINTS_EXPIRE) == []
        self.session.refresh(self.account)
        assert self.account.points_expire_at is None


class TestExpireAtParsing:
    """斗鱼给的到期时间形态不定（时间戳或字符串），两种都要兜住。"""

    @pytest.mark.parametrize(
        "value",
        [
            1791115200,  # 秒级时间戳
            "1791115200",  # 字符串时间戳
            1791115200000,  # 毫秒时间戳
            "2026-10-04 12:00:00",  # 常见字符串
            "2026-10-04T12:00:00",
        ],
    )
    def test_parses_known_shapes(self, value):
        from web.backend.services.douyu_runner_donate import _parse_expire_at

        parsed = _parse_expire_at(value)
        assert parsed is not None
        assert parsed.tzinfo is not None, "必须是 aware UTC，否则入库会偏"

    @pytest.mark.parametrize("value", [None, "", "不是什么时间", True, {}])
    def test_unknown_shapes_return_none(self, value):
        from web.backend.services.douyu_runner_donate import _parse_expire_at

        assert _parse_expire_at(value) is None, "认不出来就返回 None，不要猜一个时刻"