"""L3 监测画像写入（B3 · PRD FR-7 / R-05 最小写入）。

合并规则（防降级，PRD FR-7）：monitor_tier 取最高档（normal < watch < high），
monitor_tags 追加合并不覆盖，last_alert_id 传入时联动最新预警（未传保留旧值，
防调用方漏传抹掉），computed_at 每次写当前时间（列 NOT NULL，毫秒截断对齐
DATETIME(3)，保证乐观锁读写比对一致）。

risk_score 口径（评审 P3-4 · 用户拍板 2026-09-06）：一期**不写**（保持 NULL）。
该列是 R-05 动态评分模型的客户风险分，与 risk_alert.risk_score（预警单严重度）
语义不同；由 R-05 首写，避免静态分造成 tier/score 错位与二期语义污染。
注意：update 为整行覆盖，历史行非 NULL 值会被置 NULL（一期无其他写入方，无影响）；
**R-05 接入时须改 merge_l3 的 risk_score 为保留 existing 值，或 update SQL 跳过该列**。

并发（B3 评审 P1-1 修复）：进程内锁按 customer_id 串行减少冲突（多进程部署换
Redis SET NX，接口不变）；锁超时降级与跨进程竞态由乐观锁兜底——update 比对
读时 computed_at，未命中或 insert 撞主键则重读合并重试，3 次仍冲突抛错
（风控数据宁失败不静默覆盖）。
"""

from __future__ import annotations

import logging
from datetime import datetime
from typing import Any

from sqlalchemy.exc import IntegrityError

from app.repository.risk_repository import RiskRepository
from app.service.risk import redis_gateway
from app.service.risk.locks import run_locked

logger = logging.getLogger(__name__)

TIER_ORDER = ("normal", "watch", "high")
L3_CACHE_KEY = "profile:l3:{customer_id}"  # redis-keys 手册：String(JSON) · 5m · l3 UPDATE 时 DEL
ALERT_TYPE_TIER = {
    "aml": "high",
    "pattern": "watch",
    "large_amount": "watch",
    "freq_trade": "watch",
    "suitability": "normal",
}
AML_PENDING_TAG = "aml_hit_pending_review"
_MAX_RETRIES = 3


def _ms(dt: datetime) -> datetime:
    """毫秒截断（对齐表列 DATETIME(3)；否则乐观锁 expected 比对永不命中）。"""
    return dt.replace(microsecond=(dt.microsecond // 1000) * 1000)


def tier_of(alert_type: str) -> str:
    """alert_type → L3 档位映射（PRD FR-7 固定映射；未知类型拒绝写入）。"""
    tier = ALERT_TYPE_TIER.get(alert_type)
    if tier is None:
        raise ValueError(f"unknown alert_type for L3 mapping: {alert_type}")
    return tier


def highest_tier(a: str, b: str) -> str:
    """取最高档（防降级核心原语）。"""
    return a if TIER_ORDER.index(a) >= TIER_ORDER.index(b) else b


def merge_l3(
    existing: dict[str, Any] | None,
    *,
    mapped_tier: str,
    monitor_tags: list[str],
    last_alert_id: str | None,
    score_dimensions: dict[str, Any] | None = None,
    computed_at: datetime | None = None,
) -> dict[str, Any]:
    """纯函数：existing 行（repo.get_l3 输出）与新事件合并后的 L3 行。

    existing 为 None 表示新客户首写；risk_score 一期恒 None（不读不写，归 R-05）。
    """
    if existing is None:
        tier, old_tags, old_dims, old_alert = "normal", [], {}, None
    else:
        tier = existing.get("monitor_tier") or "normal"
        old_tags = existing.get("monitor_tags") or []
        old_dims = existing.get("score_dimensions") or {}
        old_alert = existing.get("last_alert_id")

    return {
        "monitor_tier": highest_tier(tier, mapped_tier),
        "risk_score": None,
        "monitor_tags": sorted(set(old_tags) | set(monitor_tags)),
        "score_dimensions": score_dimensions if score_dimensions is not None else old_dims,
        "last_alert_id": last_alert_id if last_alert_id is not None else old_alert,
        "computed_at": _ms(computed_at or datetime.now()),
    }


def upsert_profile_l3(
    customer_id: str,
    alert_type: str,
    monitor_tags: list[str] | None = None,
    last_alert_id: str | None = None,
    score_dimensions: dict[str, Any] | None = None,
    risk_repo: RiskRepository | None = None,
    computed_at: datetime | None = None,
) -> dict[str, Any]:
    """预警事件 → L3 upsert（aml 自动追加待复核标签）。

    返回合并后的 L3 行（含 customer_id）。MySQL 写成功后 DEL Redis 读缓存
    `profile:l3:{customer_id}`（B7 挂账②落地，PRD §5.1；DEL 失败降级 TTL 过期，
    不阻塞业务）。

    已知窗口（B7 自查留痕）：DEL 在锁外执行，存在 cache-aside 经典竞态
    （读方 miss 读旧值 → 写方 DEL → 读方回填旧值），TTL 5 分钟兜底。一期
    无读路径写缓存（get_profile_l3 直读 MySQL），窗口无实际影响；**对话线
    接入 Redis 热读缓存时须改延迟双删或写后比对**。
    """
    repo = risk_repo or RiskRepository()
    mapped_tier = tier_of(alert_type)
    tags = list(monitor_tags or [])
    if alert_type == "aml" and AML_PENDING_TAG not in tags:
        tags.append(AML_PENDING_TAG)

    def _write(locked: bool) -> dict[str, Any]:  # 锁原语签名要求（是否获得锁）；冲突安全由乐观锁重试保证
        for attempt in range(1, _MAX_RETRIES + 1):
            existing = repo.get_l3(customer_id)
            merged = merge_l3(
                existing,
                mapped_tier=mapped_tier,
                monitor_tags=tags,
                last_alert_id=last_alert_id,
                score_dimensions=score_dimensions,
                computed_at=computed_at,
            )
            fields = (
                merged["monitor_tier"],
                merged["risk_score"],
                merged["score_dimensions"],
                merged["monitor_tags"],
                merged["last_alert_id"],
                merged["computed_at"],
            )
            if existing is None:
                try:
                    repo.insert_l3(customer_id, *fields)
                    merged["customer_id"] = customer_id
                    return merged
                except IntegrityError:
                    logger.warning("L3 insert race, retrying (attempt %d): %s", attempt, customer_id)
                    continue
            if repo.update_l3(customer_id, *fields, expected_computed_at=existing["computed_at"]):
                merged["customer_id"] = customer_id
                return merged
            logger.warning("L3 update lost race, retrying (attempt %d): %s", attempt, customer_id)
        raise RuntimeError(f"L3 upsert conflicted after {_MAX_RETRIES} retries: {customer_id}")

    merged = run_locked(f"l3:{customer_id}", _write)
    redis_gateway.cache_delete(L3_CACHE_KEY.format(customer_id=customer_id))
    return merged


def get_profile_l3(customer_id: str, risk_repo: RiskRepository | None = None) -> dict[str, Any] | None:
    """L3 只读薄封装（对话线/引擎复用；Redis 热读缓存归对话线接入时实现）。"""
    return (risk_repo or RiskRepository()).get_l3(customer_id)