FabledScribe/src/scribe/services/logging.py

"""Application logging service for audit, usage, and error events."""

import asyncio
import json
import logging
import traceback as tb_module
from datetime import datetime, timedelta, timezone

from sqlalchemy import delete, func, select, text

from scribe.config import Config
from scribe.models import async_session
from scribe.models.app_log import AppLog

logger = logging.getLogger(__name__)

_retention_task: asyncio.Task | None = None


async def log_audit(
    action: str,
    user_id: int | None = None,
    username: str | None = None,
    ip_address: str | None = None,
    details: dict | None = None,
) -> None:
    async with async_session() as session:
        log = AppLog(
            category="audit",
            user_id=user_id,
            username=username,
            action=action,
            ip_address=ip_address,
            details=json.dumps(details) if details else None,
        )
        session.add(log)
        await session.commit()


async def log_usage(
    user_id: int | None = None,
    username: str | None = None,
    endpoint: str | None = None,
    method: str | None = None,
    status_code: int | None = None,
    duration_ms: float | None = None,
) -> None:
    async with async_session() as session:
        log = AppLog(
            category="usage",
            user_id=user_id,
            username=username,
            endpoint=endpoint,
            method=method,
            status_code=status_code,
            duration_ms=duration_ms,
        )
        session.add(log)
        await session.commit()


async def log_error(
    user_id: int | None = None,
    username: str | None = None,
    endpoint: str | None = None,
    method: str | None = None,
    error_type: str | None = None,
    error_message: str | None = None,
    traceback: str | None = None,
) -> None:
    details = {}
    if error_type:
        details["error_type"] = error_type
    if error_message:
        details["error_message"] = error_message
    if traceback:
        details["traceback"] = traceback

    async with async_session() as session:
        log = AppLog(
            category="error",
            user_id=user_id,
            username=username,
            endpoint=endpoint,
            method=method,
            details=json.dumps(details) if details else None,
        )
        session.add(log)
        await session.commit()


def parse_filter_datetime(value: str | None, *, end_of_day: bool = False) -> datetime | None:
    """An ISO date/datetime from a query string as an AWARE UTC datetime.

    The admin log filters arrive as raw `request.args` strings and used to be
    compared straight against `AppLog.created_at`. asyncpg binds a str as
    VARCHAR and Postgres has no `timestamptz >= text` operator, so supplying
    either date filter raised — the same defect as #1727 in notifications, in a
    second place (#2254).

    Returns None for anything unparseable: a malformed filter should narrow
    nothing rather than 500 the log viewer.

    `end_of_day` matters for the upper bound. A bare "2026-07-30" parses to
    midnight, so `created_at <= that` would exclude the whole of the day the
    user asked for — the one day they most likely wanted. With this flag a
    date-only value is pushed to the last microsecond of that day; a value that
    already carries a time is left exactly as given.
    """
    raw = (value or "").strip()
    if not raw:
        return None
    try:
        parsed = datetime.fromisoformat(raw)
    except ValueError:
        return None
    if end_of_day and len(raw) == 10:   # date-only, no time component
        parsed = parsed.replace(hour=23, minute=59, second=59, microsecond=999999)
    if parsed.tzinfo is None:
        parsed = parsed.replace(tzinfo=timezone.utc)
    return parsed


async def get_logs(
    category: str | None = None,
    user_id: int | None = None,
    search: str | None = None,
    date_from: str | None = None,
    date_to: str | None = None,
    limit: int = 50,
    offset: int = 0,
) -> tuple[list[dict], int]:
    async with async_session() as session:
        query = select(AppLog)
        count_query = select(func.count(AppLog.id))

        if category:
            query = query.where(AppLog.category == category)
            count_query = count_query.where(AppLog.category == category)
        if user_id is not None:
            query = query.where(AppLog.user_id == user_id)
            count_query = count_query.where(AppLog.user_id == user_id)
        if search:
            pattern = f"%{search}%"
            search_filter = (
                AppLog.action.ilike(pattern)
                | AppLog.endpoint.ilike(pattern)
                | AppLog.username.ilike(pattern)
                | AppLog.details.ilike(pattern)
            )
            query = query.where(search_filter)
            count_query = count_query.where(search_filter)
        start = parse_filter_datetime(date_from)
        end = parse_filter_datetime(date_to, end_of_day=True)
        if start:
            query = query.where(AppLog.created_at >= start)
            count_query = count_query.where(AppLog.created_at >= start)
        if end:
            query = query.where(AppLog.created_at <= end)
            count_query = count_query.where(AppLog.created_at <= end)

        total = (await session.execute(count_query)).scalar() or 0

        query = query.order_by(AppLog.created_at.desc()).limit(limit).offset(offset)
        result = await session.execute(query)
        logs = [row.to_dict() for row in result.scalars().all()]

        return logs, total


async def get_log_stats() -> dict:
    async with async_session() as session:
        result = await session.execute(
            text(
                "SELECT category, COUNT(*) as count FROM app_logs GROUP BY category"
            )
        )
        stats = {row.category: row.count for row in result}
        return {
            "audit": stats.get("audit", 0),
            "usage": stats.get("usage", 0),
            "error": stats.get("error", 0),
            "total": sum(stats.values()),
        }


async def delete_old_logs(retention_days: int) -> int:
    cutoff = datetime.now(timezone.utc) - timedelta(days=retention_days)
    async with async_session() as session:
        result = await session.execute(
            delete(AppLog).where(AppLog.created_at < cutoff)
        )
        await session.commit()
        return result.rowcount


async def _retention_tick() -> None:
    deleted = await delete_old_logs(Config.LOG_RETENTION_DAYS)
    if deleted:
        logger.info("Log retention: deleted %d old log entries", deleted)


def start_log_retention_loop() -> None:
    global _retention_task
    if _retention_task is None or _retention_task.done():
        from scribe.services.background import start_periodic
        _retention_task = start_periodic(3600, _retention_tick, label="log_retention")  # hourly