"""profile_l3 单测（B3 · 最高档合并防降级 / AML 后大额不回落 / 并发首单）。

sqlite StaticPool 单连接共享内存库（同 test_alert_service 模式）；
IntegrityError 兜底用 monkeypatch 模拟跨进程首写竞态。
"""

from datetime import datetime

import pytest
from sqlalchemy import create_engine, text
from sqlalchemy.pool import StaticPool

from app.repository.risk_repository import RiskRepository
from app.service.risk.profile_l3 import (
    AML_PENDING_TAG,
    ALERT_TYPE_TIER,
    get_profile_l3,
    highest_tier,
    merge_l3,
    tier_of,
    upsert_profile_l3,
)


@pytest.fixture()
def env():
    engine = create_engine(
        "sqlite://",
        poolclass=StaticPool,
        connect_args={"check_same_thread": False},
    )
    with engine.begin() as conn:
        conn.execute(
            text(
                """
                CREATE TABLE customer_profile_l3 (
                    customer_id VARCHAR(64) PRIMARY KEY,
                    monitor_tier VARCHAR(16) NOT NULL,
                    risk_score INTEGER,
                    score_dimensions TEXT,
                    monitor_tags TEXT,
                    last_alert_id VARCHAR(64),
                    computed_at TIMESTAMP NOT NULL,
                    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
                )
                """
            )
        )
    repo = RiskRepository(engine=engine)
    yield repo, engine
    engine.dispose()


def _upsert(repo, cid, alert_type, score=None, alert_id=None, tags=None, dims=None):
    return upsert_profile_l3(
        cid,
        alert_type,
        risk_score=score,
        monitor_tags=tags,
        last_alert_id=alert_id,
        score_dimensions=dims,
        risk_repo=repo,
    )


def _row(engine, cid):
    with engine.connect() as conn:
        return conn.execute(
            text(
                "SELECT monitor_tier, risk_score, monitor_tags, last_alert_id, computed_at"
                " FROM customer_profile_l3 WHERE customer_id = :cid"
            ),
            {"cid": cid},
        ).mappings().one()


# ---------- 映射与纯函数 ----------


@pytest.mark.parametrize(("alert_type", "tier"), sorted(ALERT_TYPE_TIER.items()))
def test_alert_type_tier_mapping(alert_type, tier):
    assert tier_of(alert_type) == tier


def test_unknown_alert_type_rejected(env):
    repo, _ = env
    with pytest.raises(ValueError):
        _upsert(repo, "C1", "unknown_type", 70)


def test_highest_tier_order():
    assert highest_tier("normal", "watch") == "watch"
    assert highest_tier("watch", "high") == "high"
    assert highest_tier("high", "normal") == "high"
    assert highest_tier("normal", "normal") == "normal"


def test_merge_new_customer():
    merged = merge_l3(None, mapped_tier="watch", risk_score=70, monitor_tags=["t1"],
                      last_alert_id="ALT-1")
    assert merged["monitor_tier"] == "watch"
    assert merged["risk_score"] == 70
    assert merged["monitor_tags"] == ["t1"]
    assert merged["last_alert_id"] == "ALT-1"
    assert merged["computed_at"] is not None


def test_merge_both_scores_none_stays_none():
    existing = {"monitor_tier": "watch", "risk_score": None, "monitor_tags": [],
                "score_dimensions": {}}
    merged = merge_l3(existing, mapped_tier="watch", risk_score=None, monitor_tags=[],
                      last_alert_id="ALT-2")
    assert merged["risk_score"] is None


# ---------- upsert：首写与合并 ----------


def test_first_event_inserts(env):
    repo, engine = env
    merged = _upsert(repo, "C1", "large_amount", 70, alert_id="ALT-1")
    assert merged["monitor_tier"] == "watch" and merged["customer_id"] == "C1"
    row = _row(engine, "C1")
    assert row["monitor_tier"] == "watch" and row["risk_score"] == 70
    assert row["last_alert_id"] == "ALT-1" and row["computed_at"] is not None
    assert get_profile_l3("C1", risk_repo=repo)["monitor_tier"] == "watch"


def test_normal_upgrades_to_watch(env):
    repo, _ = env
    _upsert(repo, "C1", "suitability", 90, alert_id="ALT-0")
    merged = _upsert(repo, "C1", "pattern", 80, alert_id="ALT-1")
    assert merged["monitor_tier"] == "watch"  # normal → watch 升档


def test_watch_does_not_degrade_to_normal(env):
    """ Suitability 映射 normal：已 watch 的客户不被 suitability 事件拉低。"""
    repo, engine = env
    _upsert(repo, "C1", "pattern", 80, alert_id="ALT-1")
    merged = _upsert(repo, "C1", "suitability", 90, alert_id="ALT-2")
    assert merged["monitor_tier"] == "watch"
    assert _row(engine, "C1")["monitor_tier"] == "watch"


def test_aml_marks_high_with_pending_review_tag(env):
    repo, engine = env
    merged = _upsert(repo, "C1", "aml", 95, alert_id="ALT-1")
    assert merged["monitor_tier"] == "high"
    assert AML_PENDING_TAG in merged["monitor_tags"]
    assert AML_PENDING_TAG in _row(engine, "C1")["monitor_tags"]


def test_large_amount_after_aml_does_not_fall_back(env):
    """B3 验收：AML 后大额 → tier 仍 high、tags 并集、score 取 max。"""
    repo, engine = env
    _upsert(repo, "C1", "aml", 95, alert_id="ALT-1")
    merged = _upsert(repo, "C1", "large_amount", 70, alert_id="ALT-2", tags=["manual_review"])
    assert merged["monitor_tier"] == "high"
    assert merged["risk_score"] == 95  # 取 max，不降
    assert set(merged["monitor_tags"]) == {AML_PENDING_TAG, "manual_review"}
    assert merged["last_alert_id"] == "ALT-2"  # 联动最新
    row = _row(engine, "C1")
    assert row["monitor_tier"] == "high" and row["risk_score"] == 95


def test_tags_accumulate_not_overwrite(env):
    repo, _ = env
    _upsert(repo, "C1", "pattern", 80, alert_id="ALT-1", tags=["freq"])
    merged = _upsert(repo, "C1", "pattern", 80, alert_id="ALT-2", tags=["freq"])
    assert merged["monitor_tags"] == ["freq"]  # set 合并去重


def test_score_dimensions_replaced_only_when_passed(env):
    repo, _ = env
    _upsert(repo, "C1", "aml", 95, alert_id="ALT-1", dims={"amount": 1})
    merged = _upsert(repo, "C1", "large_amount", 70, alert_id="ALT-2")
    assert merged["score_dimensions"] == {"amount": 1}  # 未传保留旧值
    merged = _upsert(repo, "C1", "large_amount", 70, alert_id="ALT-3", dims={"amount": 2})
    assert merged["score_dimensions"] == {"amount": 2}  # 传入则替换


# ---------- 并发与竞态 ----------


def test_concurrent_first_upsert_single_row(env):
    """并发冒烟：两线程同客户首单（aml + 大额）→ 1 行、high、tags 并集。"""
    import json
    from threading import Thread

    repo, engine = env
    errors = []

    def worker(alert_type, alert_id):
        try:
            _upsert(repo, "C1", alert_type, 70, alert_id=alert_id)
        except Exception as exc:  # pragma: no cover
            errors.append(exc)

    threads = [
        Thread(target=worker, args=("aml", "ALT-A")),
        Thread(target=worker, args=("large_amount", "ALT-B")),
    ]
    for t in threads:
        t.start()
    for t in threads:
        t.join()
    assert not errors, errors
    with engine.connect() as conn:
        count, payload = conn.execute(
            text(
                "SELECT COUNT(*), GROUP_CONCAT(monitor_tags) FROM customer_profile_l3"
                " WHERE customer_id = 'C1'"
            )
        ).fetchone()
    assert count == 1
    tags = json.loads(payload)
    assert AML_PENDING_TAG in tags


def test_integrity_error_falls_back_to_remerge(env, monkeypatch):
    """跨进程竞态（评审 P2-7③）：首读 None、insert 撞主键 → 重读合并转更新，不丢对方写入。"""
    repo, engine = env
    from sqlalchemy.exc import IntegrityError

    # 对方进程已写入 high（模拟 AML 先落库）
    repo.insert_l3("C1", "high", 95, {}, [AML_PENDING_TAG], "ALT-AML", datetime.now())

    calls = {"n": 0}
    orig_get = type(repo).get_l3

    def racing_get(self, cid):
        calls["n"] += 1
        if calls["n"] == 1:
            return None  # 本进程读写在对方 insert 之前发生
        return orig_get(self, cid)

    monkeypatch.setattr(type(repo), "get_l3", racing_get)
    try:
        merged = _upsert(repo, "C1", "large_amount", 70, alert_id="ALT-2")
    finally:
        monkeypatch.undo()

    assert merged["monitor_tier"] == "high"  # 重读后合并，不降级
    assert merged["risk_score"] == 95
    assert merged["last_alert_id"] == "ALT-2"
    row = _row(engine, "C1")
    assert row["monitor_tier"] == "high" and row["last_alert_id"] == "ALT-2"