"""风控规则扫描定时 Worker。

该进程不依赖 Web 进程生命周期，通过 MySQL 咨询锁保证同一时刻只有一个
调度者执行扫描。默认关闭，配置开启后才执行。
"""

from __future__ import annotations

import argparse
import asyncio
import logging
from collections.abc import AsyncIterator, Awaitable, Callable
from contextlib import AbstractAsyncContextManager, asynccontextmanager
from datetime import UTC, datetime, timedelta
from typing import Any
from uuid import uuid4

from sqlalchemy import text

from app.core.config import get_settings
from app.core.contracts import RequestContext
from app.infrastructure.db import SessionFactory, engine
from app.model.audit import InteractionAudit
from app.service.risk_scan_schedule_config import (
    RiskScanScheduleConfig,
    load_risk_scan_schedule_config,
)
from app.service.risk_scan_service import RiskScanService

SCAN_LOCK_NAME = "jr_risk_scan_schedule"
logger = logging.getLogger(__name__)

ConfigLoader = Callable[[], Awaitable[RiskScanScheduleConfig]]
ScanExecutor = Callable[[], Awaitable[dict[str, int | str]]]
AuditWriter = Callable[[str, dict[str, Any]], Awaitable[None]]
LockFactory = Callable[[], AbstractAsyncContextManager[bool]]


@asynccontextmanager
async def mysql_scan_lock() -> AsyncIterator[bool]:
    """使用 MySQL 连接级咨询锁约束跨进程并发。"""
    async with SessionFactory() as session:
        acquired = bool(
            await session.scalar(
                text("SELECT GET_LOCK(:name, 0)"),
                {"name": SCAN_LOCK_NAME},
            )
        )
        try:
            yield acquired
        finally:
            if acquired:
                await session.scalar(
                    text("SELECT RELEASE_LOCK(:name)"),
                    {"name": SCAN_LOCK_NAME},
                )


async def default_scan_executor() -> dict[str, int | str]:
    context = RequestContext(
        user_id="0",
        trace_id=f"risk-scan-schedule-{uuid4()}",
        roles=("system",),
        permissions=("risk:alert:scan",),
        data_scope="all",
        portal="worker",
    )
    async with SessionFactory() as session:
        return await RiskScanService(session).scan(context)


async def default_audit_writer(status: str, detail: dict[str, Any]) -> None:
    async with SessionFactory() as session, session.begin():
        session.add(
            InteractionAudit(
                actor_type="system",
                actor_id=None,
                target_customer_id=None,
                portal="worker",
                action_type=f"risk_scan_scheduled_{status}",
                detail=detail,
                created_at=datetime.now(UTC).replace(tzinfo=None),
            )
        )


class RiskScanSchedulerWorker:
    def __init__(
        self,
        *,
        config_loader: ConfigLoader = load_risk_scan_schedule_config,
        scan_executor: ScanExecutor = default_scan_executor,
        audit_writer: AuditWriter = default_audit_writer,
        lock_factory: LockFactory = mysql_scan_lock,
        now: Callable[[], datetime] | None = None,
    ) -> None:
        self.config_loader = config_loader
        self.scan_executor = scan_executor
        self.audit_writer = audit_writer
        self.lock_factory = lock_factory
        self.now = now or (lambda: datetime.now(UTC))
        self.last_run_at: datetime | None = None

    async def run_once(self, *, force: bool = False) -> bool:
        config = await self.config_loader()
        if not config.enabled:
            return False
        current = self.now()
        if not force and not self._is_due(config, current):
            return False

        async with self.lock_factory() as acquired:
            if not acquired:
                logger.info("风控定时扫描由其他 Worker 执行，本轮跳过")
                return False
            return await self._execute_with_retry(config)

    def _is_due(
        self,
        config: RiskScanScheduleConfig,
        current: datetime,
    ) -> bool:
        if self.last_run_at is None:
            return config.run_immediately
        return current - self.last_run_at >= timedelta(minutes=config.interval_minutes)

    async def _execute_with_retry(
        self,
        config: RiskScanScheduleConfig,
    ) -> bool:
        attempts = config.retry_limit + 1
        last_error: Exception | None = None
        for attempt in range(1, attempts + 1):
            try:
                result = await self.scan_executor()
                self.last_run_at = self.now()
                await self.audit_writer(
                    "succeeded",
                    {
                        **result,
                        "attempt": attempt,
                        "trace_id": f"risk-scan-schedule-{uuid4()}",
                    },
                )
                return True
            except Exception as error:
                last_error = error
                logger.warning(
                    "风控定时扫描失败 attempt=%s/%s",
                    attempt,
                    attempts,
                    exc_info=True,
                )
        await self.audit_writer(
            "failed",
            {
                "attempts": attempts,
                "error_type": type(last_error).__name__ if last_error else "unknown",
                "trace_id": f"risk-scan-schedule-{uuid4()}",
            },
        )
        return False


async def serve(*, once: bool = False, force: bool = False) -> None:
    worker = RiskScanSchedulerWorker()
    try:
        while True:
            try:
                await worker.run_once(force=force)
            except Exception:
                logger.warning("风控定时扫描 Worker 轮次失败", exc_info=True)
                if once:
                    raise
            if once:
                return
            await asyncio.sleep(get_settings().risk_scan_poll_seconds)
    finally:
        await engine.dispose()


def main() -> None:
    parser = argparse.ArgumentParser(description="奶龙风控规则定时扫描 Worker")
    parser.add_argument("--once", action="store_true", help="执行一轮后退出")
    parser.add_argument("--force", action="store_true", help="忽略间隔，立即执行一次")
    args = parser.parse_args()
    logging.basicConfig(level=logging.INFO)
    try:
        asyncio.run(serve(once=args.once, force=args.force))
    except KeyboardInterrupt:
        pass


if __name__ == "__main__":
    main()