signal-platform/app/services/system_event_service.py

"""Persist and query operational system events (warnings / errors)."""

from __future__ import annotations

import logging
from datetime import datetime, timedelta, timezone

from sqlalchemy import select, update
from sqlalchemy.ext.asyncio import AsyncSession

from app.models.system_event import SystemEvent

logger = logging.getLogger(__name__)

DEFAULT_LOOKBACK_DAYS = 7
DEFAULT_DEDUP_HOURS = 24
SEVERITIES = frozenset({"warning", "error"})


async def log_event(
    db: AsyncSession,
    *,
    severity: str,
    source: str,
    code: str,
    message: str,
    symbol: str | None = None,
    dedup_key: str | None = None,
    dedup_hours: int = DEFAULT_DEDUP_HOURS,
) -> SystemEvent | None:
    """Insert a system event, optionally de-duplicating recent identical keys.

    Returns the new row, or None when a recent dedup_key already exists.
    Commits the session.
    """
    severity = (severity or "").strip().lower()
    if severity not in SEVERITIES:
        severity = "warning"
    source = (source or "system")[:64]
    code = (code or "unknown")[:64]
    message = (message or "").strip() or code
    symbol = symbol.strip().upper()[:20] if symbol else None
    dedup_key = dedup_key[:200] if dedup_key else None

    if dedup_key:
        cutoff = datetime.now(timezone.utc) - timedelta(hours=max(1, dedup_hours))
        existing = await db.execute(
            select(SystemEvent.id)
            .where(
                SystemEvent.dedup_key == dedup_key,
                SystemEvent.created_at >= cutoff,
            )
            .limit(1)
        )
        if existing.scalar_one_or_none() is not None:
            return None

    row = SystemEvent(
        severity=severity,
        source=source,
        code=code,
        message=message[:4000],
        symbol=symbol,
        dedup_key=dedup_key,
        created_at=datetime.now(timezone.utc),
    )
    db.add(row)
    await db.commit()
    await db.refresh(row)
    return row


async def log_event_standalone(
    *,
    severity: str,
    source: str,
    code: str,
    message: str,
    symbol: str | None = None,
    dedup_key: str | None = None,
) -> None:
    """Open a short-lived session and log an event (for scheduler / fire-and-forget)."""
    try:
        from app.database import async_session_factory

        async with async_session_factory() as db:
            await log_event(
                db,
                severity=severity,
                source=source,
                code=code,
                message=message,
                symbol=symbol,
                dedup_key=dedup_key,
            )
    except Exception:
        logger.exception("Failed to persist system event %s/%s", source, code)


async def list_events(
    db: AsyncSession,
    *,
    days: int = DEFAULT_LOOKBACK_DAYS,
    severity: str | None = None,
    unacknowledged_only: bool = False,
    limit: int = 200,
) -> list[SystemEvent]:
    days = max(1, min(int(days), 30))
    cutoff = datetime.now(timezone.utc) - timedelta(days=days)
    stmt = (
        select(SystemEvent)
        .where(SystemEvent.created_at >= cutoff)
        .order_by(SystemEvent.created_at.desc())
        .limit(max(1, min(limit, 500)))
    )
    if severity in SEVERITIES:
        stmt = stmt.where(SystemEvent.severity == severity)
    if unacknowledged_only:
        stmt = stmt.where(SystemEvent.acknowledged_at.is_(None))
    result = await db.execute(stmt)
    return list(result.scalars().all())


async def summary(
    db: AsyncSession,
    *,
    days: int = DEFAULT_LOOKBACK_DAYS,
) -> dict:
    """Counts for badge + admin header."""
    days = max(1, min(int(days), 30))
    cutoff = datetime.now(timezone.utc) - timedelta(days=days)
    rows = (
        await db.execute(
            select(SystemEvent.severity, SystemEvent.acknowledged_at).where(
                SystemEvent.created_at >= cutoff
            )
        )
    ).all()
    total = len(rows)
    unacked = 0
    errors = 0
    warnings = 0
    for severity, acknowledged_at in rows:
        if acknowledged_at is not None:
            continue
        unacked += 1
        if severity == "error":
            errors += 1
        elif severity == "warning":
            warnings += 1
    return {
        "days": days,
        "total": total,
        "unacknowledged": unacked,
        "unacknowledged_errors": errors,
        "unacknowledged_warnings": warnings,
    }


async def acknowledge_all(
    db: AsyncSession,
    *,
    days: int = DEFAULT_LOOKBACK_DAYS,
) -> int:
    """Mark unacknowledged events in the lookback window as dismissed. Returns count."""
    days = max(1, min(int(days), 30))
    cutoff = datetime.now(timezone.utc) - timedelta(days=days)
    now = datetime.now(timezone.utc)
    result = await db.execute(
        update(SystemEvent)
        .where(
            SystemEvent.created_at >= cutoff,
            SystemEvent.acknowledged_at.is_(None),
        )
        .values(acknowledged_at=now)
    )
    await db.commit()
    return int(result.rowcount or 0)


def event_to_dict(row: SystemEvent) -> dict:
    return {
        "id": row.id,
        "severity": row.severity,
        "source": row.source,
        "code": row.code,
        "message": row.message,
        "symbol": row.symbol,
        "created_at": row.created_at.isoformat() if row.created_at else None,
        "acknowledged_at": row.acknowledged_at.isoformat() if row.acknowledged_at else None,
    }