"""账号池：候选筛选、两梯队排序、租约领取与释放。

分两层守：

- **纯函数**（充值量、排序、拆单）不碰数据库，直接喂数据断言 ——
  这几条是业务规则的集中地，改错了必须立刻变红；
- **租约**（唯一约束、并发抢号、按任务释放）必须打真实 MySQL ——
  「一个账号同时只能有一个有效租约」是**数据库索引**给的保证，
  用内存假库测不出并发下的双发。
"""

from __future__ import annotations

from datetime import UTC, datetime, timedelta

import pytest

from tests.support.db import get_test_engine, make_test_session
from web.backend.models import Account, AccountEvent, AccountLease, DouyuTask, User
from web.backend.services import account_pool as pool

NOW = datetime.now(UTC).replace(microsecond=0)
COMMODITY = "20260120QYOOB_bag10"


# ── 纯函数 ────────────────────────────────────────────────────────────────


class TestTopupMath:
    def test_rounds_gap_up_to_whole_gifts(self):
        assert pool.topup_gifts_for(300, 1280) == 98  # 缺口 980 → 98 个令
        assert pool.topup_gifts_for(1279, 1280) == 1
        assert pool.topup_gifts_for(1280, 1280) == 0
        assert pool.topup_gifts_for(2000, 1280) == 0

    def test_none_points_is_treated_as_zero(self):
        assert pool.topup_gifts_for(None, 20) == 2

    def test_exact_multiple_does_not_add_a_gift(self):
        """缺口正好是 10 的倍数时不能多给一个令（多买就是浪费钱）。"""
        assert pool.topup_gifts_for(0, 100) == 10


class TestRechargeSplit:
    def test_no_split_below_limit(self):
        assert pool.recharge_amounts_for(0) == []
        assert pool.recharge_amounts_for(98) == [98]
        assert pool.recharge_amounts_for(800) == [800]

    def test_splits_above_limit(self):
        assert pool.recharge_amounts_for(1700) == [800, 800, 100]
        assert pool.recharge_amounts_for(1600) == [800, 800]

    def test_validate_rejects_over_limit(self):
        assert pool.validate_recharge_amount(800) == 800
        with pytest.raises(ValueError, match="800"):
            pool.validate_recharge_amount(801)
        with pytest.raises(ValueError):
            pool.validate_recharge_amount(0)


class TestRanking:
    def _cand(self, account_id, points, score=1280, expire=None):
        return pool.make_candidate(
            account_id=account_id,
            username=f"acc{account_id}",
            points=points,
            score=score,
            points_expire_at=expire,
        )

    def test_tier_a_before_tier_b(self):
        rich = self._cand(1, 2000)
        poor = self._cand(2, 0)
        ranked = pool.rank_candidates([poor, rich])
        assert [c.account_id for c in ranked] == [1, 2]
        assert ranked[0].tier == "A" and ranked[1].tier == "B"

    def test_tier_a_spends_expiring_points_first(self):
        """A 梯队先花**快过期**的积分 —— 那是已经买过、不用就作废的钱。"""
        soon = self._cand(1, 2000, expire=NOW + timedelta(days=1))
        later = self._cand(2, 2000, expire=NOW + timedelta(days=30))
        never = self._cand(3, 2000, expire=None)
        ranked = pool.rank_candidates([never, later, soon])
        assert [c.account_id for c in ranked] == [1, 2, 3]

    def test_tier_a_then_prefers_smaller_balance(self):
        """都没有到期时间时先用小余额，把大额账号留给更贵的商品。"""
        small = self._cand(1, 1300)
        big = self._cand(2, 9999)
        ranked = pool.rank_candidates([big, small])
        assert [c.account_id for c in ranked] == [1, 2]

    def test_tier_b_prefers_least_topup(self):
        """B 梯队先选**充值最少**的，省真金白银。"""
        near = self._cand(1, 1200)  # 只差 8 个令
        far = self._cand(2, 0)  # 差 128 个令
        ranked = pool.rank_candidates([far, near])
        assert [c.account_id for c in ranked] == [1, 2]
        assert ranked[0].topup_gifts == 8

    def test_order_is_stable_for_equal_candidates(self):
        same = [self._cand(i, 0) for i in (5, 3, 9)]
        assert [c.account_id for c in pool.rank_candidates(same)] == [3, 5, 9]


# ── 数据库 ────────────────────────────────────────────────────────────────


class _PoolBase:
    def setup_method(self):
        self.engine = get_test_engine()
        self.session = make_test_session(self.engine)
        self.admin = User(username="pool-admin", password_hash="h", role="super_admin")
        self.support = User(username="pool-kf", password_hash="h", role="support")
        self.session.add_all([self.admin, self.support])
        self.session.flush()
        self.admin_id = self.admin.id
        self.support_id = self.support.id

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

    def make_account(
        self,
        name: str,
        *,
        points: int | None = 0,
        opened: bool = True,
        ck: bool = True,
        status: str = "active",
        assigned_to: int | None = None,
        expire_at: datetime | None = None,
        auto_managed: bool = True,
    ) -> Account:
        account = Account(
            username=name,
            password="p",
            email="e@e.com",
            email_password="ep",
            status=status,
            ck_check_status="valid" if ck else "",
            cookie="PHPSESSID=x" if ck else "",
            elite_opened=opened,
            points=points,
            assigned_to=assigned_to,
            points_expire_at=expire_at,
            auto_managed=auto_managed,
        )
        self.session.add(account)
        self.session.commit()
        return account

    def add_exchange(self, account_id: int, commodity_id: str, days_ago: int = 1):
        self.session.add(
            AccountEvent(
                account_id=account_id,
                event_type="exchange",
                category="business",
                occurred_at=NOW - timedelta(days=days_ago),
                title="兑换",
                subject_type="commodity",
                subject_key=commodity_id,
                source="official_sync",
            )
        )
        self.session.commit()

    def eligible(self, *, score=1280, limit=None):
        return pool.eligible_accounts(
            self.session,
            score=score,
            commodity_id=COMMODITY,
            now=NOW,
            limit=limit,
        )


class TestEligibilityFilters(_PoolBase):
    def test_unassigned_and_operator_owned_are_pool_members(self):
        operator = User(username="pool-op", password_hash="h", role="operation")
        self.session.add(operator)
        self.session.flush()
        free = self.make_account("free")
        owned = self.make_account("owned", assigned_to=operator.id)
        ids = {c.account_id for c in self.eligible()}
        assert {free.id, owned.id} <= ids

    def test_support_owned_accounts_are_not_in_pool(self):
        """第 1 条：不跨归属 —— 客服名下账号不进池。"""
        mine = self.make_account("kf-owned", assigned_to=self.support_id)
        assert mine.id not in {c.account_id for c in self.eligible()}

    def test_not_opened_is_excluded(self):
        """开通精英宝典要人工扫码，未开通的账号进池也白进。"""
        closed = self.make_account("closed", opened=False, points=5000)
        assert closed.id not in {c.account_id for c in self.eligible()}

    def test_unknown_opened_is_excluded(self):
        """`NULL` = 没查过，不能当「已开通」放行。"""
        unknown = self.make_account("unknown")
        unknown.elite_opened = None
        self.session.commit()
        assert unknown.id not in {c.account_id for c in self.eligible()}

    def test_exchanged_within_cooldown_is_excluded(self):
        done = self.make_account("done", points=5000)
        self.add_exchange(done.id, COMMODITY, days_ago=10)
        assert done.id not in {c.account_id for c in self.eligible()}

    def test_exchange_outside_cooldown_is_allowed_again(self):
        old = self.make_account("old", points=5000)
        self.add_exchange(old.id, COMMODITY, days_ago=pool.EXCHANGE_COOLDOWN_DAYS + 1)
        assert old.id in {c.account_id for c in self.eligible()}

    def test_other_commodity_history_does_not_block(self):
        other = self.make_account("other", points=5000)
        self.add_exchange(other.id, "20260120QYOOB_bag5", days_ago=1)
        assert other.id in {c.account_id for c in self.eligible()}

    def test_leased_account_is_excluded(self):
        account = self.make_account("leased", points=5000)
        pool.lease_accounts(
            self.session,
            holder_user_id=self.support_id,
            candidates=[
                pool.make_candidate(
                    account_id=account.id,
                    username=account.username,
                    points=5000,
                    score=1280,
                )
            ],
            commodity_id=COMMODITY,
            now=NOW,
        )
        assert account.id not in {c.account_id for c in self.eligible()}

    def test_no_ck_or_inactive_is_excluded(self):
        no_ck = self.make_account("no-ck", ck=False, points=5000)
        inactive = self.make_account("inactive", status="imported", points=5000)
        ids = {c.account_id for c in self.eligible()}
        assert no_ck.id not in ids and inactive.id not in ids

    def test_auto_managed_false_is_excluded(self):
        special = self.make_account("special", points=5000, auto_managed=False)
        assert special.id not in {c.account_id for c in self.eligible()}

    def test_unknown_points_is_excluded(self):
        """积分没查过 → 充值量算不出来，先别放出去。"""
        unknown = self.make_account("no-points", points=None)
        assert unknown.id not in {c.account_id for c in self.eligible()}

    def test_tiers_are_assigned_by_score(self):
        rich = self.make_account("rich", points=5000)
        poor = self.make_account("poor", points=100)
        by_id = {c.account_id: c for c in self.eligible(score=1280)}
        assert by_id[rich.id].tier == "A"
        assert by_id[poor.id].tier == "B"
        assert by_id[poor.id].topup_gifts == 118  # (1280-100)/10


class TestSupplyEstimate(_PoolBase):
    def test_counts_add_up(self):
        self.make_account("rich", points=5000)
        poor = self.make_account("poor", points=0)
        done = self.make_account("done", points=5000)
        self.add_exchange(done.id, COMMODITY, days_ago=1)
        self.make_account("closed", opened=False, points=5000)

        supply = pool.estimate_supply(
            self.session, score=1280, commodity_id=COMMODITY, now=NOW
        )
        assert supply["zero_topup"] == 1  # rich
        assert supply["after_topup"] == 2  # rich + poor
        assert supply["blocked"]["elite_not_opened"] == 1
        assert supply["blocked"]["redeemed_within_cooldown"] == 1
        # poor 差 128 个令；rich 已够，不产生缺口
        assert supply["cost"]["est_gifts"] == 128
        assert poor.id  # 保留引用，避免上面的构造被优化掉


class TestLeasing(_PoolBase):
    def test_lease_creates_active_row(self):
        account = self.make_account("a1", points=5000)
        leases = pool.lease_accounts(
            self.session,
            holder_user_id=self.support_id,
            candidates=[
                pool.make_candidate(
                    account_id=account.id, username="a1", points=5000, score=1280
                )
            ],
            commodity_id=COMMODITY,
            now=NOW,
        )
        assert len(leases) == 1
        assert leases[0].status == "active"
        assert leases[0].expires_at == NOW + timedelta(
            minutes=pool.DEFAULT_LEASE_TTL_MINUTES
        )

    def test_second_lease_on_same_account_is_rejected(self):
        """唯一索引裁决：同一账号第二个租约必须失败（这是并发安全的地基）。"""
        account = self.make_account("a2", points=5000)
        candidate = pool.make_candidate(
            account_id=account.id, username="a2", points=5000, score=1280
        )
        first = pool.lease_accounts(
            self.session,
            holder_user_id=self.support_id,
            candidates=[candidate],
            commodity_id=COMMODITY,
            now=NOW,
        )
        assert len(first) == 1
        second = pool.lease_accounts(
            self.session,
            holder_user_id=self.admin_id,
            candidates=[candidate],
            commodity_id=COMMODITY,
            now=NOW,
        )
        assert second == [], "同一个账号不该被第二个租约抢到"

    def test_concurrent_lease_from_two_sessions_only_one_wins(self):
        """两个独立会话抢同一账号 —— 真实并发形态。"""
        account = self.make_account("a3", points=5000)
        candidate = pool.make_candidate(
            account_id=account.id, username="a3", points=5000, score=1280
        )
        other = make_test_session(self.engine)
        try:
            first = pool.lease_accounts(
                self.session,
                holder_user_id=self.support_id,
                candidates=[candidate],
                commodity_id=COMMODITY,
                now=NOW,
            )
            second = pool.lease_accounts(
                other,
                holder_user_id=self.admin_id,
                candidates=[candidate],
                commodity_id=COMMODITY,
                now=NOW,
            )
        finally:
            other.close()
        assert len(first) == 1
        assert second == []
        assert (
            self.session.query(AccountLease)
            .filter(AccountLease.account_id == account.id)
            .count()
            == 1
        )

    def test_quantity_stops_early(self):
        accounts = [self.make_account(f"q{i}", points=5000) for i in range(3)]
        candidates = [
            pool.make_candidate(
                account_id=a.id, username=a.username, points=5000, score=1280
            )
            for a in accounts
        ]
        leases = pool.lease_accounts(
            self.session,
            holder_user_id=self.support_id,
            candidates=candidates,
            commodity_id=COMMODITY,
            quantity=2,
            now=NOW,
        )
        assert len(leases) == 2

    def test_release_for_task_only_frees_its_own_account(self):
        """一个批次里每个账号各有一条租约，不能因为一条任务结束就整批释放。"""
        accounts = [self.make_account(f"t{i}", points=5000) for i in range(2)]
        candidates = [
            pool.make_candidate(
                account_id=a.id, username=a.username, points=5000, score=1280
            )
            for a in accounts
        ]
        leases = pool.lease_accounts(
            self.session,
            holder_user_id=self.support_id,
            candidates=candidates,
            commodity_id=COMMODITY,
            batch_id="batch-1",
            now=NOW,
        )
        pool.bind_leases_to_batch(self.session, leases, "batch-1")
        task = DouyuTask(
            batch_id="batch-1",
            account_id=accounts[0].id,
            task_type="exchange_goods",
            handbook_scope="elite",
            status="success",
            created_by=self.support_id,
        )
        self.session.add(task)
        self.session.commit()

        assert pool.release_for_task(self.session, task, reason="task_success") is True
        self.session.expire_all()
        states = {
            lease.account_id: lease.status
            for lease in self.session.query(AccountLease)
            .filter(AccountLease.batch_id == "batch-1")
            .all()
        }
        assert states[accounts[0].id] == "released"
        assert states[accounts[1].id] == "active", "同批次另一个账号的租约不该被释放"

    def test_release_is_idempotent(self):
        account = self.make_account("i1", points=5000)
        lease = pool.lease_accounts(
            self.session,
            holder_user_id=self.support_id,
            candidates=[
                pool.make_candidate(
                    account_id=account.id, username="i1", points=5000, score=1280
                )
            ],
            commodity_id=COMMODITY,
            now=NOW,
        )[0]
        assert pool.release_lease(self.session, lease, reason="done") is True
        assert pool.release_lease(self.session, lease, reason="done") is False

    def test_released_account_becomes_available_again(self):
        account = self.make_account("r1", points=5000)
        lease = pool.lease_accounts(
            self.session,
            holder_user_id=self.support_id,
            candidates=[
                pool.make_candidate(
                    account_id=account.id, username="r1", points=5000, score=1280
                )
            ],
            commodity_id=COMMODITY,
            now=NOW,
        )[0]
        assert account.id not in {c.account_id for c in self.eligible()}
        pool.release_lease(self.session, lease, reason="done")
        assert account.id in {c.account_id for c in self.eligible()}

    def test_reap_expired_only_touches_past_due(self):
        due = self.make_account("due", points=5000)
        live = self.make_account("live", points=5000)
        candidates = [
            pool.make_candidate(
                account_id=a.id, username=a.username, points=5000, score=1280
            )
            for a in (due, live)
        ]
        pool.lease_accounts(
            self.session,
            holder_user_id=self.support_id,
            candidates=[candidates[0]],
            commodity_id=COMMODITY,
            ttl_minutes=1,
            now=NOW,
        )
        pool.lease_accounts(
            self.session,
            holder_user_id=self.support_id,
            candidates=[candidates[1]],
            commodity_id=COMMODITY,
            ttl_minutes=600,
            now=NOW,
        )
        reaped = pool.reap_expired(self.session, now=NOW + timedelta(minutes=5))
        assert reaped == 1
        self.session.expire_all()
        rows = {
            lease.account_id: lease.status
            for lease in self.session.query(AccountLease).all()
        }
        assert rows[due.id] == "expired"
        assert rows[live.id] == "active"

    def test_leased_account_ids_lists_holder_only(self):
        account = self.make_account("h1", points=5000)
        pool.lease_accounts(
            self.session,
            holder_user_id=self.support_id,
            candidates=[
                pool.make_candidate(
                    account_id=account.id, username="h1", points=5000, score=1280
                )
            ],
            commodity_id=COMMODITY,
            now=NOW,
        )
        assert pool.leased_account_ids(
            self.session, holder_user_id=self.support_id
        ) == [account.id] or account.id in pool.leased_account_ids(
            self.session, holder_user_id=self.support_id
        )
        assert pool.leased_account_ids(self.session, holder_user_id=self.admin_id) == []