live-hub-py/tests/test_account_events.py

"""账号事件流回归：幂等发射、旁路隔离、任务映射、收集器水位线。

三条必须守住：

1. **发射是旁路**：`emit` 任何失败都不能影响主流程（登录/兑换照常）；
2. **幂等**：同一个 `source_ref` 重复发射只落一条 —— 收集器重跑的可靠性全押在这上面；
3. **水位线只进不退**：解析失败要跳过而不是卡住，否则一行坏数据会让事件流永久停摆。
"""

import os
import time
from collections.abc import Iterator
from contextlib import contextmanager
from datetime import UTC, datetime

from sqlalchemy.orm import sessionmaker

import web.backend.database as database_module
from tests.support.db import get_test_engine, make_test_session
from web.backend.models import Account, AccountEvent, AutomationJob, DouyuTask, User
from web.backend.services import account_events_service as svc
from web.backend.services.automation.jobs import account_events_collect as collect_job


@contextmanager
def _force_non_utc_timezone(name: str = "Asia/Shanghai") -> Iterator[None]:
    """临时把进程时区切到非 UTC，让「naive 被当成本地时间」这类 bug 必然暴露。

    ⚠️ 不加这一段，时区相关的回归在 **UTC 机器上（CI / 生产容器）会假绿**：
    旧实现 `naive.astimezone(UTC)` 在 TZ=UTC 下恰好等于正确结果，
    只有开发机（UTC+8）才现形 —— 而「只在某些机器上失败」的测试等于没有测试。
    """
    previous = os.environ.get("TZ")
    os.environ["TZ"] = name
    time.tzset()
    try:
        yield
    finally:
        if previous is None:
            os.environ.pop("TZ", None)
        else:
            os.environ["TZ"] = previous
        time.tzset()


class _Base:
    def setup_method(self):
        self.engine = get_test_engine()
        database_module.SessionLocal.configure(bind=self.engine)
        self.session = make_test_session(self.engine)
        self.session.query(AccountEvent).delete()
        self.session.query(DouyuTask).delete()
        self.session.query(AutomationJob).filter(
            AutomationJob.key == collect_job.JOB_KEY
        ).delete()
        self.session.commit()

        user = User(username="evt-admin", password_hash="h", role="super_admin")
        self.session.add(user)
        self.session.flush()
        self.user_id = user.id
        self.account = Account(
            username="事件测试账号",
            password="p",
            email="evt@example.com",
            email_password="ep",
        )
        self.session.add(self.account)
        self.session.commit()
        self.account_id = self.account.id

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

    def _fresh(self):
        return sessionmaker(bind=self.engine)()


class TestEmit(_Base):
    def test_naive_occurred_at_is_treated_as_utc(self):
        """⚠️ 回归：naive datetime 必须按「已经是 UTC」处理，不能当本地时间。

        仓库的既定约定写在 `database.UTCDateTime` 里：
        **写：aware → 转 UTC 去 tzinfo；naive → 视为已是 UTC，原样存**。
        `scheduler._as_utc` / `yyb_service._as_utc` / `audit.py` 都遵守了这条。

        但 `emit()` 曾经直接 `.astimezone(UTC)` —— 对 naive 值，Python 会把它
        当作**服务器本地时间**再换算。本机是 UTC+8 时，事件时间会被整整扣掉
        8 小时（实测库里 `account_events.occurred_at` 比源 `douyu_tasks.finished_at`
        早 8 小时）。而 `occurred_at` 是「业务发生时间」，错了整条时间线都是错的。

        naive 值的来源很具体：收集器用**裸 SQL** 读 `douyu_tasks`，绕过了
        `UTCDateTime` 装饰器，拿到的就是 naive。

        用例显式切到 UTC+8 执行：在 UTC 机器上旧实现与正确实现结果相同，
        不切时区就会假绿。
        """
        with _force_non_utc_timezone():
            # 故意造 naive 值：本用例就是要验证 naive 的处理
            naive_probe = datetime(2026, 10, 3, 11, 46, 7)  # noqa: DTZ001
            assert naive_probe.astimezone(UTC).replace(tzinfo=None) != naive_probe, (
                "时区未生效，本用例失去意义"
            )
            db = self._fresh()
            try:
                # 模拟从 DATETIME 列读出来的裸值（语义是 UTC，但没有 tzinfo）
                naive_utc = datetime(2026, 10, 3, 11, 46, 7)  # noqa: DTZ001
                assert naive_utc.tzinfo is None
                ok = svc.emit(
                    db,
                    account_id=self.account_id,
                    event_type=svc.EVENT_RECHARGE,
                    title="充值 98",
                    amount=98,
                    occurred_at=naive_utc,
                    source="task",
                    source_ref="naive-utc-regression",
                )
                assert ok
                db.commit()
            finally:
                db.close()

        db = self._fresh()
        try:
            event = (
                db.query(AccountEvent)
                .filter(AccountEvent.source_ref == "naive-utc-regression")
                .one()
            )
            # 存进去的应当是同一个「墙上时间」，不能被时区换算挪动
            assert event.occurred_at == datetime(2026, 10, 3, 11, 46, 7, tzinfo=UTC)
        finally:
            db.close()

    def test_aware_occurred_at_is_converted_to_utc(self):
        """aware 值仍要正常换算（别为了修 naive 把这条路弄坏）。"""
        from datetime import timedelta, timezone

        db = self._fresh()
        try:
            ok = svc.emit(
                db,
                account_id=self.account_id,
                event_type=svc.EVENT_RECHARGE,
                title="充值 1",
                occurred_at=datetime(
                    2026, 10, 3, 19, 46, 7, tzinfo=timezone(timedelta(hours=8))
                ),
                source="task",
                source_ref="aware-utc-regression",
            )
            assert ok
            db.commit()
        finally:
            db.close()

        db = self._fresh()
        try:
            event = (
                db.query(AccountEvent)
                .filter(AccountEvent.source_ref == "aware-utc-regression")
                .one()
            )
            assert event.occurred_at == datetime(2026, 10, 3, 11, 46, 7, tzinfo=UTC)
        finally:
            db.close()

    def test_stores_category_subject_and_actor(self):
        """台账检索列（category / subject_* / actor_*）要真的能写进去。

        S1 的迁移 0042 建了这 5 列与复合索引 `ix_account_events_ledger`，但 `emit()`
        一直没有暴露它们 —— 列建好了却没有写入方（和 `status` 一样的「空壳」）。
        这条用例锁住「能写」，S3/S5 才能靠索引做「该账号该商品最近一次兑换」这类查询。
        """
        db = self._fresh()
        try:
            ok = svc.emit(
                db,
                account_id=self.account_id,
                event_type=svc.EVENT_EXCHANGE,
                title="兑换 兔兔白日梦",
                category=svc.CATEGORY_BUSINESS,
                subject_type="commodity",
                subject_key="20260120QYOOB_bag10",
                actor_type="system",
                actor_id=7,
                source_ref="ledger-columns-regression",
            )
            assert ok
            db.commit()
        finally:
            db.close()

        db = self._fresh()
        try:
            event = (
                db.query(AccountEvent)
                .filter(AccountEvent.source_ref == "ledger-columns-regression")
                .one()
            )
            assert event.category == "business"
            assert event.subject_type == "commodity"
            assert event.subject_key == "20260120QYOOB_bag10"
            assert event.actor_type == "system"
            assert event.actor_id == 7
        finally:
            db.close()

    def test_writes_event(self):
        db = self._fresh()
        try:
            ok = svc.emit(
                db,
                account_id=self.account_id,
                event_type=svc.EVENT_EXCHANGE,
                title="兑换 套装-兔兔白日梦",
                amount=-1280,
                balance_after=0,
                detail={"commodity_name": "套装-兔兔白日梦"},
                source="manual",
            )
            db.commit()
        finally:
            db.close()
        assert ok is True

        db = self._fresh()
        try:
            row = db.query(AccountEvent).one()
            assert row.account_id == self.account_id
            assert row.event_type == svc.EVENT_EXCHANGE
            assert row.amount == -1280
            assert row.balance_after == 0
            assert row.detail["commodity_name"] == "套装-兔兔白日梦"
        finally:
            db.close()

    def test_same_source_ref_is_idempotent(self):
        """同一 source_ref 只能落一条 —— 收集器重跑靠它兜住。"""
        db = self._fresh()
        try:
            for _ in range(3):
                svc.emit(
                    db,
                    account_id=self.account_id,
                    event_type=svc.EVENT_RECHARGE,
                    title="充值 98",
                    amount=98,
                    source="task",
                    source_ref="douyu_task:1001",
                )
            db.commit()
        finally:
            db.close()

        db = self._fresh()
        try:
            assert db.query(AccountEvent).count() == 1
        finally:
            db.close()

    def test_multiple_null_source_refs_coexist(self):
        """没有幂等键的事件（手工记录）不能互相冲突 —— 这正是 source_ref 要可空的原因。"""
        db = self._fresh()
        try:
            for _ in range(3):
                svc.emit(
                    db,
                    account_id=self.account_id,
                    event_type=svc.EVENT_ASSIGNED,
                    title="分配客服",
                    source="manual",
                )
            db.commit()
        finally:
            db.close()

        db = self._fresh()
        try:
            assert db.query(AccountEvent).count() == 3
        finally:
            db.close()

    def test_failure_is_swallowed(self):
        """旁路原则：account_id 不存在（外键失败）也不能抛出去。"""
        db = self._fresh()
        try:
            ok = svc.emit(
                db,
                account_id=99999999,
                event_type=svc.EVENT_LOGIN_OK,
                title="登录成功",
                source="login",
            )
            db.commit()
        finally:
            db.close()
        assert ok is False

    def test_event_does_not_break_caller_transaction(self):
        """emit 失败后，调用方在同一事务里已做的写入仍然能提交。

        注意必须用**同一个 session**：换 session 就测不到「SAVEPOINT 隔离」这件事
        （改动会落在另一个事务里，断言必然失败）。
        """
        db = self._fresh()
        try:
            account = db.query(Account).filter(Account.id == self.account_id).first()
            account.remark = "改动要保住"
            db.flush()
            # 这条会因外键失败
            svc.emit(
                db,
                account_id=99999999,
                event_type=svc.EVENT_LOGIN_OK,
                title="x",
            )
            db.commit()
        finally:
            db.close()

        db = self._fresh()
        try:
            row = db.query(Account).filter(Account.id == self.account_id).first()
            assert row.remark == "改动要保住"
        finally:
            db.close()

    def test_empty_account_or_type_is_noop(self):
        db = self._fresh()
        try:
            assert svc.emit(db, account_id=0, event_type="x", title="y") is False
            assert (
                svc.emit(db, account_id=self.account_id, event_type="", title="y")
                is False
            )
        finally:
            db.close()


class TestTaskMapping:
    def test_exchange_maps_amount_negative(self):
        payload = svc._task_to_event(
            "exchange_goods",
            {
                "commodity_name": "套装-狂战萌兔",
                "commodity_id": "20260120QYOOB_bag5",
                "score": 480,
                "points_query": {"points": 0},
                "order_id": "147874",
                "game_name": "窝补誓渣男",
                "game_channel": "微信 / 安卓(android)",
            },
        )
        assert payload["event_type"] == svc.EVENT_EXCHANGE
        assert payload["amount"] == -480, "消耗要记成负数，否则累计统计会反向"
        assert payload["balance_after"] == 0
        assert payload["detail"]["game_name"] == "窝补誓渣男"

    def test_recharge_maps_amount(self):
        payload = svc._task_to_event(
            "create_gold_qr",
            {
                "buy_num": 98,
                "order_id": "DYGF5C24C5E4EA84T238035",
                "supplier_order": {
                    "order_id": "200000229425",
                    "finish_time": "2026-10-03 11:46:07",
                },
                "recharge_channel": "supplier_api",
                "douyu_uid": "678503799",
            },
        )
        assert payload["event_type"] == svc.EVENT_RECHARGE
        assert payload["amount"] == 98
        assert payload["detail"]["supplier_order_id"] == "200000229425"

    def test_points_credit_keeps_expire_time(self):
        payload = svc._task_to_event(
            "donate_elite_gift",
            {
                "gift_points_expected_gain": 980,
                "points": 1280,
                "gift_points_baseline": 300,
                "gift_points_target": 1280,
                "points_query": {"raw": {"data": {"willExpireTime": 1798819199}}},
            },
        )
        assert payload["event_type"] == svc.EVENT_POINTS_CREDIT
        assert payload["amount"] == 980
        assert payload["balance_after"] == 1280
        assert payload["detail"]["points_expire_at"] == 1798819199

    def test_unparsable_result_returns_none(self):
        assert svc._task_to_event("exchange_goods", "{not json") is None
        assert svc._task_to_event("exchange_goods", None) is None
        assert svc._task_to_event("unknown_type", {"a": 1}) is None


class TestCollector(_Base):
    def _make_job(self, watermark=None):
        job = AutomationJob(
            key=collect_job.JOB_KEY,
            title="账号事件收集",
            enabled=True,
            interval_seconds=600,
            batch_size=100,
            concurrency=1,
            config={"watermark": watermark} if watermark is not None else None,
            updated_at=datetime.now(UTC),
        )
        self.session.add(job)
        self.session.commit()
        return job

    def _add_task(self, task_type, result, status="success", account_id=None):
        task = DouyuTask(
            batch_id="evt-batch",
            account_id=account_id or self.account_id,
            task_type=task_type,
            status=status,
            result=result,
            created_by=self.user_id,
        )
        self.session.add(task)
        self.session.commit()
        return task.id

    def test_first_run_sets_watermark_without_backfill(self):
        """首次运行只定水位线，不回溯历史（2026-10-03 已定不回填）。"""
        self._add_task(
            "exchange_goods",
            {"commodity_name": "旧兑换", "score": 100, "points_query": {"points": 0}},
        )
        job = self._make_job()
        db = self._fresh()
        try:
            result = collect_job.run_account_events_collect(db, job)
        finally:
            db.close()
        assert result["detail"]["first_run"] is True
        assert result["scanned"] == 0

        db = self._fresh()
        try:
            assert db.query(AccountEvent).count() == 0, "首次运行不应回填历史事件"
        finally:
            db.close()

    def test_collects_only_new_rows_and_advances_watermark(self):
        first = self._add_task(
            "exchange_goods",
            {"commodity_name": "老兑换", "score": 100, "points_query": {"points": 5}},
        )
        job = self._make_job(watermark=first)
        new_id = self._add_task(
            "create_gold_qr",
            {"buy_num": 68, "order_id": "DYGF-x", "supplier_order": {}},
        )

        db = self._fresh()
        try:
            result = collect_job.run_account_events_collect(db, job)
        finally:
            db.close()

        assert result["detail"]["written"] == 1
        assert result["detail"]["watermark"] == new_id

        db = self._fresh()
        try:
            rows = db.query(AccountEvent).all()
            assert len(rows) == 1
            assert rows[0].event_type == svc.EVENT_RECHARGE
            assert rows[0].source_ref == f"douyu_task:{new_id}"
        finally:
            db.close()

    def test_rerun_does_not_duplicate(self):
        """重复跑不能造重复事件（source_ref 兜底）。"""
        job = self._make_job(watermark=0)
        self._add_task(
            "create_gold_qr",
            {"buy_num": 98, "order_id": "o1", "supplier_order": {}},
        )
        db = self._fresh()
        try:
            collect_job.run_account_events_collect(db, job)
        finally:
            db.close()

        # 把水位线倒回去，模拟「重复处理同一批」
        job2 = self._fresh()
        try:
            row = (
                job2.query(AutomationJob)
                .filter(AutomationJob.key == collect_job.JOB_KEY)
                .first()
            )
            row.config = {"watermark": 0}
            job2.commit()
        finally:
            job2.close()

        db = self._fresh()
        try:
            job3 = (
                db.query(AutomationJob)
                .filter(AutomationJob.key == collect_job.JOB_KEY)
                .first()
            )
            collect_job.run_account_events_collect(db, job3)
        finally:
            db.close()

        db = self._fresh()
        try:
            assert db.query(AccountEvent).count() == 1
        finally:
            db.close()

    def test_failed_task_produces_no_event(self):
        """失败的兑换不该记成「消耗了积分」。"""
        job = self._make_job(watermark=0)
        self._add_task(
            "exchange_goods",
            {"commodity_name": "失败兑换", "score": 100},
            status="failed",
        )
        db = self._fresh()
        try:
            result = collect_job.run_account_events_collect(db, job)
        finally:
            db.close()
        assert result["detail"]["written"] == 0
        assert result["skipped"] == 1

    def test_unparsable_row_does_not_stall_watermark(self):
        """坏数据要跳过并推进水位线，不能让事件流永久停摆。"""
        job = self._make_job(watermark=0)
        bad_id = self._add_task("exchange_goods", "{not json")
        good_id = self._add_task(
            "create_gold_qr", {"buy_num": 1, "order_id": "o2", "supplier_order": {}}
        )
        db = self._fresh()
        try:
            result = collect_job.run_account_events_collect(db, job)
        finally:
            db.close()
        assert result["detail"]["written"] == 1
        assert result["detail"]["watermark"] == good_id
        assert good_id > bad_id

    def test_collected_occurred_at_matches_source_task_time(self):
        """⚠️ 端到端回归：收集器写的事件时间必须等于源任务的业务时间。

        这条用例存在的理由：收集器用**裸 SQL** 读 `douyu_tasks`，拿到的是 naive
        datetime（绕过 `UTCDateTime`）。旧实现把它当本地时间换算，导致事件时间
        整整早 8 小时（本机 UTC+8，库里实测过）——而「什么时候兑换/充值了」
        正是这张表存在的意义。

        旧用例只断言 written / watermark 的数量，从不看时间，所以这个 bug
        一路绿灯溜了过去。
        """
        from web.backend.models import DouyuTask

        task_id = self._add_task(
            "create_gold_qr",
            {"buy_num": 98, "order_id": "o-time", "supplier_order": {}},
        )
        # 给源任务一个明确的业务完成时间（UTC 语义）
        source = self.session.query(DouyuTask).filter(DouyuTask.id == task_id).one()
        source.finished_at = datetime(2026, 10, 3, 11, 46, 7)  # noqa: DTZ001
        self.session.commit()

        job = self._make_job(watermark=task_id - 1)
        # 显式切到 UTC+8：这个 bug 在 UTC 机器上不可见，不切时区会假绿。
        with _force_non_utc_timezone():
            db = self._fresh()
            try:
                result = collect_job.run_account_events_collect(db, job)
            finally:
                db.close()
        assert result["detail"]["written"] == 1

        db = self._fresh()
        try:
            event = (
                db.query(AccountEvent)
                .filter(AccountEvent.source_ref == f"douyu_task:{task_id}")
                .one()
            )
            assert event.occurred_at == datetime(2026, 10, 3, 11, 46, 7, tzinfo=UTC)
        finally:
            db.close()


class TestAccountApi(_Base):
    """账号详情接口：概览统计 + 时间线分页 + 删除账号时清理事件。"""

    def _admin(self):
        return self.session.query(User).filter(User.username == "evt-admin").first()

    def _seed(self):
        """造三条不同类型的事件，供统计与筛选断言。"""
        db = self._fresh()
        try:
            svc.emit(
                db,
                account_id=self.account_id,
                event_type=svc.EVENT_RECHARGE,
                title="充值 98",
                amount=98,
                occurred_at=datetime(2026, 10, 1, 3, 0, tzinfo=UTC),
                source="task",
                source_ref="douyu_task:9001",
            )
            svc.emit(
                db,
                account_id=self.account_id,
                event_type=svc.EVENT_EXCHANGE,
                title="兑换 兔兔白日梦",
                amount=-1280,
                balance_after=0,
                occurred_at=datetime(2026, 10, 2, 3, 0, tzinfo=UTC),
                source="task",
                source_ref="douyu_task:9002",
            )
            svc.emit(
                db,
                account_id=self.account_id,
                event_type=svc.EVENT_ELITE_OPENED,
                title="精英宝典开通",
                occurred_at=datetime(2026, 10, 3, 3, 0, tzinfo=UTC),
                source="manual",
                source_ref="elite_opened:1:1",
            )
            db.commit()
        finally:
            db.close()
        # ⚠️ MySQL 默认 REPEATABLE READ：self.session 在 setup 阶段就开了事务，
        # 快照早于上面那批写入，不结束事务就查不到刚写的事件（P1b 踩过同一个坑）。
        self.session.commit()

    def test_overview_aggregates_stats(self):
        from web.backend.routers import accounts as api

        self._seed()
        data = api.account_overview(
            account_id=self.account_id, db=self.session, current=self._admin()
        )
        assert data["username"] == "事件测试账号"
        assert data["created_at"] is not None, "导入时间必须返回"
        stats = data["stats"]
        assert stats["recharge_total"] == 98
        assert stats["exchange_count"] == 1
        assert stats["points_spent"] == 1280, "消耗要按正数展示给用户"
        assert stats["event_count"] == 3
        # 开通时间由事件表提供（不回填历史，所以只有上线后开通的才有）
        assert data["opened_at"]["elite"] is not None

    def test_events_paginated_desc(self):
        from web.backend.routers import accounts as api

        self._seed()
        page = api.account_events(
            account_id=self.account_id,
            page=1,
            page_size=2,
            db=self.session,
            current=self._admin(),
        )
        assert page["total"] == 3
        assert len(page["items"]) == 2
        # 倒序：最新的精英宝典开通在最前
        assert page["items"][0]["event_type"] == svc.EVENT_ELITE_OPENED

    def test_events_filter_by_type(self):
        from web.backend.routers import accounts as api

        self._seed()
        page = api.account_events(
            account_id=self.account_id,
            event_type=svc.EVENT_EXCHANGE,
            db=self.session,
            current=self._admin(),
        )
        assert page["total"] == 1
        assert page["items"][0]["title"] == "兑换 兔兔白日梦"

    def test_events_isolated_between_accounts(self):
        """事件必须按账号隔离 —— 串号会让客服看到别人的记录。"""
        from web.backend.routers import accounts as api

        self._seed()
        other = Account(
            username="另一个账号",
            password="p",
            email="o@example.com",
            email_password="ep",
        )
        self.session.add(other)
        self.session.commit()
        db = self._fresh()
        try:
            svc.emit(
                db,
                account_id=other.id,
                event_type=svc.EVENT_RECHARGE,
                title="别人的充值",
                amount=1,
                source="task",
                source_ref="douyu_task:9003",
            )
            db.commit()
        finally:
            db.close()

        page = api.account_events(
            account_id=self.account_id, db=self.session, current=self._admin()
        )
        assert all("别人" not in item["title"] for item in page["items"])

    def test_delete_account_cleans_events(self):
        """事件表有指向 accounts 的外键：删账号必须连带清事件，否则删不掉。"""
        from web.backend.routers import accounts as api

        self._seed()
        api.delete_account(
            account_id=self.account_id, db=self.session, current=self._admin()
        )

        db = self._fresh()
        try:
            left = (
                db.query(AccountEvent)
                .filter(AccountEvent.account_id == self.account_id)
                .count()
            )
            assert left == 0
            assert (
                db.query(Account).filter(Account.id == self.account_id).first() is None
            )
        finally:
            db.close()