backend/worker_main.py

"""
Worker Main App — lightweight FastAPI for background game sync and pre-cache.

Endpoints:
  GET  /health  — MongoDB ping, last cycle summary, cycle_running flag
  POST /trigger — token-protected, starts a worker cycle as background task
  GET  /logs    — token-protected, read structured log tail
"""

import asyncio
import logging
import os
import uuid
from contextlib import asynccontextmanager
from datetime import datetime, timezone
from typing import Any

from fastapi import FastAPI, Query, Request
from fastapi.responses import JSONResponse

from app.core.config import settings
from app.core.worker_logging import (
    AsyncTimingContext,
    WORKER_LOG_WHITELIST,
    log_structured,
    read_log_tail,
    resolve_log_path,
    set_cycle_id,
    setup_app_logging,
    setup_structured_logger,
)
from app.db.mongodb import mongodb
from app.services.game_sync_service import GameSyncService
from app.services.nlp_service import NLPService
from app.services.precache_service import PreCacheService
from app.services.steam_service import SteamService
from app.services.priority_refresh_service import PriorityRefreshService
from app.services.update_detection_service import UpdateDetectionService

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

# Cycle state
_cycle_running = False
_last_cycle_summary: dict[str, Any] = {}


@asynccontextmanager
async def lifespan(app: FastAPI):
    """Connect MongoDB on startup, disconnect on shutdown."""
    await mongodb.connect()
    setup_structured_logger("worker")
    setup_app_logging()
    logger.info("Worker started — MongoDB connected, structured logging initialized")
    yield
    await mongodb.disconnect()
    logger.info("Worker shutting down")


app = FastAPI(title="SentimentStream Worker", lifespan=lifespan)


@app.get("/health")
async def health():
    """Health check with cycle status."""
    mongo_ok = False
    try:
        if mongodb.client:
            await mongodb.client.admin.command("ping")
            mongo_ok = True
    except Exception:
        pass

    return {
        "status": "ok" if mongo_ok else "degraded",
        "mongodb": "connected" if mongo_ok else "disconnected",
        "cycle_running": _cycle_running,
        "last_cycle": _last_cycle_summary,
    }


def _check_bearer_token(request: Request) -> bool:
    """Validate Bearer token from Authorization header."""
    auth = request.headers.get("Authorization", "")
    expected = settings.worker_trigger_token
    return bool(expected and auth.startswith("Bearer ") and auth[7:] == expected)


@app.post("/trigger")
async def trigger(request: Request):
    """Token-protected trigger to start a worker cycle."""
    global _cycle_running

    if not _check_bearer_token(request):
        return JSONResponse(status_code=401, content={"detail": "Unauthorized"})

    if _cycle_running:
        return JSONResponse(status_code=503, content={"detail": "Cycle already running"})

    asyncio.create_task(_run_cycle())
    return {"status": "started"}


@app.get("/logs")
async def get_logs(
    request: Request,
    lines: int = Query(default=100, ge=1, le=1000),
    level: str | None = Query(default=None),
    event: str | None = Query(default=None),
    file: str = Query(default="worker"),
):
    """Token-protected endpoint to read structured log tail."""
    if not _check_bearer_token(request):
        return JSONResponse(status_code=401, content={"detail": "Unauthorized"})

    log_path = resolve_log_path(file, WORKER_LOG_WHITELIST)
    if log_path is None:
        return JSONResponse(
            status_code=400,
            content={"detail": f"Unknown log file: '{file}'. Valid: {list(WORKER_LOG_WHITELIST.keys())}"},
        )

    entries = read_log_tail(log_path, lines=lines, level=level, event=event)
    return {"entries": entries, "count": len(entries)}


async def _run_cycle() -> None:
    """Execute a full worker cycle."""
    global _cycle_running, _last_cycle_summary
    _cycle_running = True
    started = datetime.now(timezone.utc)
    summary: dict[str, Any] = {"started_at": started.isoformat()}

    cycle_id = uuid.uuid4().hex[:8]
    set_cycle_id(cycle_id)
    log_structured("cycle_start", cycle_id=cycle_id)

    steam_svc = SteamService()
    nlp_svc = NLPService()
    game_sync_svc = GameSyncService()
    priority_svc = PriorityRefreshService()
    update_svc = UpdateDetectionService()

    try:
        # 1. Game sync (if enabled and not synced recently)
        if settings.game_sync_enabled:
            top_games = await mongodb.get_top_games_by_reviews(1)
            last_synced = top_games[0].get("synced_at") if top_games else None
            hours_since_sync = None
            if last_synced:
                delta = datetime.now(timezone.utc) - last_synced
                hours_since_sync = delta.total_seconds() / 3600

            if hours_since_sync is None or hours_since_sync > 20:
                async with AsyncTimingContext() as t_sync:
                    logger.info("Starting game sync...")
                    upserted, modified = await game_sync_svc.sync_all_games()
                    summary["game_sync"] = {"upserted": upserted, "modified": modified}
                log_structured("game_sync", elapsed_s=t_sync.elapsed_s,
                               detail=summary["game_sync"])

                async with AsyncTimingContext() as t_details:
                    enriched = await game_sync_svc.sync_top_game_details()
                    summary["game_details"] = {"enriched": enriched}
                log_structured("game_details", elapsed_s=t_details.elapsed_s,
                               detail=summary["game_details"])
            else:
                summary["game_sync"] = "skipped (recent)"
                log_structured("game_sync", detail="skipped (recent)")

            # ALWAYS enrich CN names if sync is enabled, even if main sync skipped
            async with AsyncTimingContext() as t_cn:
                cn_processed = await game_sync_svc.enrich_cn_names()
                summary["cn_enrichment"] = {"processed": cn_processed}
            log_structured("cn_enrichment", elapsed_s=t_cn.elapsed_s,
                           detail=summary["cn_enrichment"])

            async with AsyncTimingContext() as t_app_types:
                app_types_processed = await game_sync_svc.enrich_app_types()
                summary["app_type_enrichment"] = {"processed": app_types_processed}
            log_structured("app_type_enrichment", elapsed_s=t_app_types.elapsed_s,
                           detail=summary["app_type_enrichment"])

        # 1b. Priority refresh
        async with AsyncTimingContext() as t_priority:
            priority_result = await priority_svc.refresh_priorities()
            summary["priority_refresh"] = priority_result
        log_structured("priority_refresh", elapsed_s=t_priority.elapsed_s, detail=priority_result)

        # 2. Update detection
        async with AsyncTimingContext() as t_update:
            top_games = await mongodb.get_priority_games_for_analysis()
            updated_games = await update_svc.check_for_updates(top_games)
            summary["updates_detected"] = len(updated_games)
        log_structured("update_detection", elapsed_s=t_update.elapsed_s,
                       detail={"updates_detected": len(updated_games)})

        # 3. Create schedules for updated games
        precache_svc = PreCacheService(steam_svc, nlp_svc)

        async with AsyncTimingContext() as t_sched:
            if updated_games:
                await precache_svc.create_schedules_for_updates(updated_games)
        log_structured("create_schedules", elapsed_s=t_sched.elapsed_s,
                       detail={"updated_games": len(updated_games) if updated_games else 0})

        # 4. Bootstrap missing analyses
        async with AsyncTimingContext() as t_boot:
            bootstrapped = await precache_svc.bootstrap_missing_analyses(top_games)
            summary["bootstrapped"] = bootstrapped
        log_structured("bootstrap_missing", elapsed_s=t_boot.elapsed_s,
                       detail={"bootstrapped": bootstrapped})

        # 5. Process due analyses
        if settings.precache_enabled:
            async with AsyncTimingContext() as t_analyses:
                executed = await precache_svc.process_due_analyses()
                summary["analyses_executed"] = executed
            log_structured("process_due_analyses", elapsed_s=t_analyses.elapsed_s,
                           detail={"executed": executed})
        else:
            summary["precache"] = "disabled"

    except Exception as e:
        logger.error(f"Cycle error: {e}", exc_info=True)
        summary["error"] = str(e)
        log_structured("cycle_error", level=logging.ERROR, error=str(e))
    finally:
        await game_sync_svc.close()
        await priority_svc.close()
        await update_svc.close()
        await steam_svc.close()

        elapsed = (datetime.now(timezone.utc) - started).total_seconds()
        summary["elapsed_seconds"] = round(elapsed, 1)
        _last_cycle_summary = summary
        _cycle_running = False
        log_structured("cycle_end", elapsed_s=round(elapsed, 1),
                       detail=summary)
        set_cycle_id(None)
        logger.info(f"Cycle complete in {elapsed:.1f}s: {summary}")