ctms/backend/app/services/monitoring_retention.py

"""监测数据留存清理任务。"""

from __future__ import annotations

import asyncio
import logging
from dataclasses import dataclass, field
from datetime import datetime, timedelta, timezone
from typing import Any

from sqlalchemy import delete

from app.core.config import settings
from app.db.session import SessionLocal
from app.models.permission_access_log import PermissionAccessLog
from app.models.permission_metric_snapshot import PermissionMetricSnapshot
from app.models.security_access_log import SecurityAccessLog
from app.models.source_location_snapshot import SourceLocationSnapshot
from app.models.user_login_session import UserLoginSession

logger = logging.getLogger("ctms.monitoring_retention")


@dataclass
class MonitoringRetentionState:
    running: bool = False
    last_run_started_at: datetime | None = None
    last_success_at: datetime | None = None
    last_error_at: datetime | None = None
    last_error_type: str | None = None
    error_count: int = 0
    last_deleted: dict[str, int] = field(default_factory=dict)
    total_deleted: dict[str, int] = field(default_factory=dict)

    def snapshot(self) -> dict[str, Any]:
        return {
            "running": self.running,
            "access_log_retention_days": settings.MONITORING_ACCESS_LOG_RETENTION_DAYS,
            "metric_retention_days": settings.MONITORING_METRIC_RETENTION_DAYS,
            "login_activity_retention_days": settings.USER_LOGIN_ACTIVITY_RETENTION_DAYS,
            "interval_seconds": settings.MONITORING_RETENTION_INTERVAL_SECONDS,
            "last_run_started_at": (
                self.last_run_started_at.isoformat() if self.last_run_started_at else None
            ),
            "last_success_at": self.last_success_at.isoformat() if self.last_success_at else None,
            "last_error_at": self.last_error_at.isoformat() if self.last_error_at else None,
            "last_error_type": self.last_error_type,
            "error_count": self.error_count,
            "last_deleted": dict(self.last_deleted),
            "total_deleted": dict(self.total_deleted),
        }


_state = MonitoringRetentionState()


def get_monitoring_retention_status() -> dict[str, Any]:
    return _state.snapshot()


def _deleted_count(result: Any) -> int:
    rowcount = getattr(result, "rowcount", 0)
    return max(0, int(rowcount or 0))


async def purge_expired_monitoring_data(
    *, now: datetime | None = None
) -> dict[str, int]:
    reference_time = now or datetime.now(timezone.utc)
    access_cutoff = reference_time - timedelta(
        days=settings.MONITORING_ACCESS_LOG_RETENTION_DAYS
    )
    metric_cutoff = reference_time - timedelta(
        days=settings.MONITORING_METRIC_RETENTION_DAYS
    )
    login_activity_cutoff = reference_time - timedelta(
        days=settings.USER_LOGIN_ACTIVITY_RETENTION_DAYS
    )

    async with SessionLocal() as session:
        permission_result = await session.execute(
            delete(PermissionAccessLog).where(PermissionAccessLog.created_at < access_cutoff)
        )
        security_result = await session.execute(
            delete(SecurityAccessLog).where(SecurityAccessLog.created_at < access_cutoff)
        )
        metric_result = await session.execute(
            delete(PermissionMetricSnapshot).where(
                PermissionMetricSnapshot.bucket_time < metric_cutoff
            )
        )
        source_location_result = await session.execute(
            delete(SourceLocationSnapshot).where(
                SourceLocationSnapshot.bucket_time < access_cutoff
            )
        )
        login_session_result = await session.execute(
            delete(UserLoginSession).where(UserLoginSession.last_seen_at < login_activity_cutoff)
        )
        await session.commit()

    return {
        "permission_access_logs": _deleted_count(permission_result),
        "security_access_logs": _deleted_count(security_result),
        "permission_metric_snapshots": _deleted_count(metric_result),
        "source_location_snapshots": _deleted_count(source_location_result),
        "user_login_sessions": _deleted_count(login_session_result),
    }


async def run_monitoring_retention_once(
    *, now: datetime | None = None
) -> dict[str, int]:
    _state.last_run_started_at = now or datetime.now(timezone.utc)
    try:
        deleted = await purge_expired_monitoring_data(now=now)
    except Exception as exc:
        _state.last_error_at = datetime.now(timezone.utc)
        _state.last_error_type = type(exc).__name__
        _state.error_count += 1
        raise

    _state.last_success_at = datetime.now(timezone.utc)
    _state.last_deleted = deleted
    for key, count in deleted.items():
        _state.total_deleted[key] = _state.total_deleted.get(key, 0) + count
    return deleted


async def run_monitoring_retention(stop_event: asyncio.Event) -> None:
    logger.info("Monitoring retention task started")
    _state.running = True
    try:
        while not stop_event.is_set():
            try:
                deleted = await run_monitoring_retention_once()
                if any(deleted.values()):
                    logger.info("Purged expired monitoring data: %s", deleted)
            except Exception:
                logger.exception("Failed to purge expired monitoring data")

            try:
                await asyncio.wait_for(
                    stop_event.wait(),
                    timeout=settings.MONITORING_RETENTION_INTERVAL_SECONDS,
                )
            except asyncio.TimeoutError:
                continue
    finally:
        _state.running = False
        logger.info("Monitoring retention task stopped")