FaqSearch/app/main.py

"""FastAPI 应用入口。"""
from __future__ import annotations

import asyncio
import logging
from contextlib import asynccontextmanager
from typing import AsyncIterator

from fastapi import FastAPI

from app.exception_handlers import register_exception_handlers
from app.response import ApiResponse
from app.routers import search
from app.schemas import CacheRefreshResult, CacheStatus
from app.services.cache import (
    WORKER_ID,
    get_snapshot,
    leader_heartbeat,
    refresh_cache,
    scheduler_loop,
    try_become_leader,
    wait_cache_ready,
)
from app.services.redis_client import (
    KEY_LOCK_LEADER,
    close_redis,
    release_lock,
)

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
)
logger = logging.getLogger(__name__)


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    """启动：选举 leader → 仅 leader 刷新缓存与跑定时任务；关闭：释放资源。"""
    is_leader = await try_become_leader()
    tasks: list[asyncio.Task[None]] = []

    if is_leader:
        logger.info("[%s] 当选 leader，开始首次刷新", WORKER_ID)
        await refresh_cache()
        tasks.append(
            asyncio.create_task(scheduler_loop(), name="faq-scheduler")
        )
        tasks.append(
            asyncio.create_task(leader_heartbeat(), name="leader-heartbeat")
        )
        logger.info("[%s] 定时任务与 leader 心跳已启动", WORKER_ID)
    else:
        logger.info("[%s] 不是 leader，等待缓存就绪…", WORKER_ID)
        ready = await wait_cache_ready()
        if ready:
            logger.info("[%s] 共享缓存已就绪", WORKER_ID)
        else:
            logger.warning(
                "[%s] 等待缓存就绪超时，搜索将以空 sec_ids 降级运行",
                WORKER_ID,
            )

    try:
        yield
    finally:
        logger.info("[%s] 关闭：取消后台任务…", WORKER_ID)
        for task in tasks:
            task.cancel()
        for task in tasks:
            try:
                await task
            except asyncio.CancelledError:
                pass
        if is_leader:
            # 主动释放 leader 锁，让其他 worker 可以更快接管
            await release_lock(KEY_LOCK_LEADER, WORKER_ID)
        await close_redis()


app = FastAPI(
    title="Zendesk FAQ Keyword Search",
    description="封装 Zendesk Help Center 搜索；FAQ sec_ids 共享缓存（Redis），每日定时刷新。",
    version="1.0.0",
    lifespan=lifespan,
)

# 注册全局异常处理器（统一错误响应格式 {code, data, msg}）
register_exception_handlers(app)

# 路由
app.include_router(search.router)


@app.get(
    "/health",
    response_model=ApiResponse[dict[str, str]],
    tags=["meta"],
)
async def health() -> ApiResponse[dict[str, str]]:
    """健康检查。"""
    return ApiResponse.ok(data={"status": "ok"})


@app.get(
    "/health/cache",
    response_model=ApiResponse[CacheStatus],
    tags=["meta"],
)
async def cache_status() -> ApiResponse[CacheStatus]:
    """缓存状态查看。"""
    snap = await get_snapshot()
    return ApiResponse.ok(
        data=CacheStatus(
            sec_ids_count=len(snap["sec_ids"]),
            last_updated_at=snap["last_updated_at"],
            next_refresh_at=snap["next_refresh_at"],
        )
    )


@app.post(
    "/admin/cache/refresh",
    response_model=ApiResponse[CacheRefreshResult],
    tags=["meta"],
)
async def refresh_cache_now() -> ApiResponse[CacheRefreshResult]:
    """手动触发刷新（运维 / 测试用途）。

    使用 force=True 跳过去重锁，确保管理员请求立即生效。
    """
    await refresh_cache(force=True)
    snap = await get_snapshot()
    return ApiResponse.ok(
        data=CacheRefreshResult(
            sec_ids_count=len(snap["sec_ids"]),
            last_updated_at=snap["last_updated_at"],
        ),
        msg="缓存刷新成功",
    )