Mutual_Fund/service/advisor/event_consumer.py

"""Redis 事件消费者（投顾域事件：rebalance_draft_created / profile_update）。

双写原则（common_const §9）：Agent 双写 event_log + Redis Pub/Sub；Pub/Sub 仅实时通知，
消费以 event_id 幂等，补拉靠 event_log 扫描兜底。process_event 是幂等入口，实时订阅与
补拉两条路径共用。
"""
from __future__ import annotations

import asyncio
import json
import logging

from common_const import (
    ADVISOR_EVENTS,
    EVENT_PROFILE_UPDATE,
    EVENT_REBALANCE_DRAFT_CREATED,
    EVENT_STATUS_CONSUMED,
    TODO_SOURCE_AGENT_EVENT,
    TODO_TYPE_NEW_REBALANCE_DRAFT,
)
from model.advisor_todo import AdvisorTodo
from repositories.advisor_todo import AdvisorTodoRepo
from repositories.event_log import EventLogRepo

logger = logging.getLogger("service.advisor.event_consumer")


async def handle_rebalance_created(db, payload: dict) -> None:
    """消费 rebalance_draft_created → 生成待办（uk_todo 三元组去重，幂等）。"""
    advisor_id = payload.get("advisor_id")
    customer_id = payload.get("customer_id")
    draft_id = payload.get("draft_id")
    if not advisor_id:
        logger.warning("rebalance_draft_created 缺少 advisor_id，忽略: %s", payload)
        return
    repo = AdvisorTodoRepo(db)
    existing = await repo.get_by_unique(TODO_TYPE_NEW_REBALANCE_DRAFT, customer_id, draft_id)
    if existing is not None:
        return  # 已存在，幂等跳过
    todo = AdvisorTodo(
        todo_type=TODO_TYPE_NEW_REBALANCE_DRAFT,
        customer_id=customer_id,
        advisor_id=int(advisor_id),
        source=TODO_SOURCE_AGENT_EVENT,
        biz_id=str(draft_id) if draft_id else None,
    )
    await repo.add(todo)  # add 内 commit + refresh


async def handle_profile_update(db, payload: dict) -> None:
    """消费 profile_update → 失效该客户 Redis 画像缓存（不落库，画像只读）。

    注：V1.0 工作台尚未建设 profile:{customer_id} 画像热缓存读取，此处仅占位；后续
    360 读画像接入缓存后生效。事件仍需标记已消费。
    """
    return


async def process_event(db, *, event_id: str | None, event_name: str | None, payload: dict) -> None:
    """处理单条事件（幂等入口，实时订阅与补拉共用）。"""
    # 幂等兜底：event_log 已消费则跳过（实时订阅与补拉并发时防重）
    if event_id:
        existing = await EventLogRepo(db).get_by_event_id(event_id)
        if existing is not None and existing.status == EVENT_STATUS_CONSUMED:
            return

    if event_name == EVENT_REBALANCE_DRAFT_CREATED:
        await handle_rebalance_created(db, payload)
    elif event_name == EVENT_PROFILE_UPDATE:
        await handle_profile_update(db, payload)

    if event_id:
        await EventLogRepo(db).mark_consumed(event_id)


def parse_message(raw: str) -> tuple[str | None, str | None, dict] | None:
    """解析 Redis Pub/Sub 消息为 (event_id, event_name, payload)。

    容忍两种结构：带内层 payload 的标准结构；无内层 payload 时把外层整体当 payload。
    """
    try:
        msg = json.loads(raw)
    except (ValueError, TypeError):
        return None
    if not isinstance(msg, dict):
        return None
    event_name = msg.get("event_name")
    event_id = msg.get("event_id") or msg.get("id")
    payload = msg.get("payload")
    if not isinstance(payload, dict):
        payload = msg
    return event_id, event_name, payload


async def pull_pending_events(db) -> int:
    """补拉 event_log 中未消费的投顾域事件（Pub/Sub 丢消息兜底），返回处理条数。"""
    events = await EventLogRepo(db).list_pending(ADVISOR_EVENTS, limit=100)
    for ev in events:
        await process_event(
            db, event_id=ev.event_id, event_name=ev.event_name, payload=ev.payload or {}
        )
    return len(events)


class EventConsumer:
    """Redis 订阅循环（后台任务，dev 默认关闭，避免 --reload 重复订阅）。"""

    def __init__(self, redis):
        self.redis = redis
        self._task: asyncio.Task | None = None

    async def _run(self) -> None:
        from config.database.mysql import get_session_factory

        pubsub = self.redis.pubsub()
        await pubsub.subscribe(*ADVISOR_EVENTS)
        try:
            async for message in pubsub.listen():
                if message.get("type") != "message":
                    continue
                parsed = parse_message(message.get("data"))
                if parsed is None:
                    continue
                event_id, event_name, payload = parsed
                if event_name not in ADVISOR_EVENTS:
                    continue
                # 每事件开独立 session，避免长事务占用连接
                async with get_session_factory()() as session:
                    try:
                        await process_event(
                            session, event_id=event_id, event_name=event_name, payload=payload
                        )
                    except Exception:
                        logger.exception("process event failed: %s %s", event_name, event_id)
        finally:
            await pubsub.unsubscribe(*ADVISOR_EVENTS)
            await pubsub.aclose()

    def start(self) -> None:
        if self._task is None:
            self._task = asyncio.create_task(self._run())

    async def stop(self) -> None:
        if self._task is not None:
            self._task.cancel()
            try:
                await self._task
            except asyncio.CancelledError:
                pass
            self._task = None