siftlode/backend/app/routes/admin_logs.py

"""The admin diagnostics console's read side (E22.6): query `log_events`, the server's stored log.

Admin-only and read-only — rows leave the table through the retention job (`log_retention_days`),
never through here. Reading is not audited: it changes nothing, and the console's live tail polls,
so a row per read would bury the audit log's real entries.

What a row holds was redacted when it was written (`logs.record_fields`, and the ingest route for a
browser's rows), so this serves the columns as they are; a value the redactor took out reads as
`[redacted]` / `[email]`, which is the point — the console shows that something was there.

Paging is a keyset over `(ts desc, id desc)`: a browser row's `ts` is clamped into its consent
window, so `ts` alone is not unique, and an offset would skip or repeat rows while the writer adds
new ones at the top. The live tail re-reads the first page rather than asking for "ids after N":
several processes write this table, and a batch's ids are taken when it inserts, not when it
commits, so a slower batch can land below an id the tail already passed — and would never show.
"""

import base64
import binascii
import json
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Literal

from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy import Text, and_, cast, func, or_, select
from sqlalchemy.orm import Session

from app.auth import admin_user
from app.db import get_db
from app.models import LogEvent, User
from app.utils import iso, like_escape

router = APIRouter(prefix="/api/admin/logs", tags=["admin"])

PAGE_MAX = 200
# A facet list is a chip row, not a report: past this many distinct events the rest are reachable
# through the free-text search, and the response stays small whatever a self-host logs.
FACET_EVENTS_MAX = 100

Level = Literal["TRACE", "DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]
# `none` selects the rows that have no account / no event code (a scheduler line, an anonymous
# request) — a real value to filter on, not the absence of a filter.
NONE = "none"


def _utc(value: datetime | None) -> datetime | None:
    """A bound without a zone is UTC — the column is `timestamptz`, and a naive value would be read
    in the database session's zone instead."""
    if value is None or value.tzinfo is not None:
        return value
    return value.replace(tzinfo=timezone.utc)


def _encode_cursor(row: LogEvent) -> str:
    return base64.urlsafe_b64encode(json.dumps([row.ts.isoformat(), row.id]).encode()).decode()


def _decode_cursor(cursor: str) -> tuple[datetime, int]:
    try:
        ts, row_id = json.loads(base64.urlsafe_b64decode(cursor.encode()))
        return _utc(datetime.fromisoformat(ts)), int(row_id)
    except (binascii.Error, json.JSONDecodeError, ValueError, TypeError) as exc:
        raise HTTPException(status_code=400, detail={"code": "bad_cursor"}) from exc


def _serialize(row: LogEvent) -> dict:
    return {
        "id": row.id,
        "ts": iso(row.ts),
        "level": row.level,
        "source": row.source,
        "logger": row.logger,
        "event": row.event,
        "msg": row.msg,
        "attrs": row.attrs,
        "exc": row.exc,
        "request_id": row.request_id,
        "user_id": row.user_id,
        "job_id": row.job_id,
        "debug_session_id": row.debug_session_id,
        "version": row.version,
    }


def _accounts(db: Session, user_ids: set[int]) -> dict[str, str]:
    """The accounts a page names, by id → address. An id with no account left (erased after the
    line was written) is simply absent, and the client shows the bare id."""
    if not user_ids:
        return {}
    rows = db.execute(select(User.id, User.email).where(User.id.in_(user_ids))).all()
    return {str(uid): email for uid, email in rows}


@dataclass(frozen=True)
class LogFilters:
    """The console's filters, shared by the page and its facet counts."""

    levels: list[str]
    sources: list[str]
    user: str | None
    event: str | None
    request_id: str | None
    job_id: int | None
    q: str | None
    since: datetime | None
    until: datetime | None

    def conditions(self, skip: str | None = None) -> list:
        """The WHERE conditions, ANDed by the caller. `skip` leaves one filter out: a facet counts
        its own values under every OTHER filter, so picking a level does not zero the other
        levels' counts (the way back), while a request pivot does narrow them (honest counts)."""
        out = []
        if self.levels and skip != "levels":
            out.append(LogEvent.level.in_(self.levels))
        if self.sources and skip != "sources":
            out.append(LogEvent.source.in_(self.sources))
        if self.user is not None and skip != "user":
            out.append(
                LogEvent.user_id.is_(None)
                if self.user == NONE
                else LogEvent.user_id == int(self.user)
            )
        if self.event is not None and skip != "event":
            out.append(
                LogEvent.event.is_(None) if self.event == NONE else LogEvent.event == self.event
            )
        if self.request_id is not None:
            out.append(LogEvent.request_id == self.request_id)
        if self.job_id is not None:
            out.append(LogEvent.job_id == self.job_id)
        if self.since is not None:
            out.append(LogEvent.ts >= _utc(self.since))
        if self.until is not None:
            out.append(LogEvent.ts < _utc(self.until))
        if self.q and self.q.strip():
            like = f"%{like_escape(self.q.strip())}%"
            out.append(
                or_(
                    LogEvent.msg.ilike(like),
                    LogEvent.event.ilike(like),
                    LogEvent.request_id.ilike(like),
                    LogEvent.exc.ilike(like),
                    cast(LogEvent.attrs, Text).ilike(like),
                )
            )
        return out


def log_filters(
    levels: list[Level] = Query(default=[]),
    sources: list[str] = Query(default=[], max_length=8),
    user: str | None = Query(default=None, pattern=r"^(none|[0-9]{1,10})$"),
    event: str | None = Query(
        default=None, max_length=64, pattern=r"^(none|[a-z][a-z0-9_]*(\.[a-z0-9_]+)*)$"
    ),
    request_id: str | None = Query(default=None, pattern=r"^[A-Za-z0-9._-]{8,64}$"),
    job_id: int | None = Query(default=None, ge=1),
    q: str | None = Query(default=None, max_length=200),
    since: datetime | None = None,
    until: datetime | None = None,
) -> LogFilters:
    return LogFilters(list(levels), list(sources), user, event, request_id, job_id, q, since, until)


@router.get("")
def list_logs(
    filters: LogFilters = Depends(log_filters),
    before: str | None = Query(default=None, max_length=200),
    limit: int = Query(default=100, ge=1, le=PAGE_MAX),
    _: User = Depends(admin_user),
    db: Session = Depends(get_db),
) -> dict:
    """One page of rows, newest first, matching every filter given (they AND together); `before`
    continues from a previous page's `next_cursor`."""
    query = select(LogEvent).where(*filters.conditions())
    if before is not None:
        ts, row_id = _decode_cursor(before)
        # The `ts <=` conjunct is implied by the OR; it is there so the planner can bound the
        # `ts` index scan instead of filtering every newer row first.
        query = query.where(
            and_(
                LogEvent.ts <= ts,
                or_(LogEvent.ts < ts, and_(LogEvent.ts == ts, LogEvent.id < row_id)),
            )
        )

    # One more than asked, to know whether there is a next page without a second count query.
    rows = (
        db.execute(query.order_by(LogEvent.ts.desc(), LogEvent.id.desc()).limit(limit + 1))
        .scalars()
        .all()
    )
    more = len(rows) > limit
    rows = rows[:limit]
    return {
        "items": [_serialize(r) for r in rows],
        "accounts": _accounts(db, {r.user_id for r in rows if r.user_id is not None}),
        "next_cursor": _encode_cursor(rows[-1]) if more else None,
    }


@router.get("/facets")
def log_facets(
    filters: LogFilters = Depends(log_filters),
    _: User = Depends(admin_user),
    db: Session = Depends(get_db),
) -> dict:
    """What the filter chips offer, with counts: the levels, sources and accounts present, and the
    most frequent events — each counted under every filter but its own (`LogFilters.conditions`).
    `total` is the "all levels" count: every filter but the level one."""

    def counts(col, skip: str, cap: int | None = None) -> list[dict]:
        n = func.count().label("n")
        stmt = select(col, n).where(*filters.conditions(skip)).group_by(col).order_by(n.desc(), col)
        if cap is not None:
            stmt = stmt.limit(cap)
        return [
            {"value": NONE if value is None else value, "count": count}
            for value, count in db.execute(stmt).all()
        ]

    users = counts(LogEvent.user_id, "user")
    total = select(func.count()).select_from(LogEvent).where(*filters.conditions("levels"))
    return {
        "levels": counts(LogEvent.level, "levels"),
        "sources": counts(LogEvent.source, "sources"),
        "events": counts(LogEvent.event, "event", FACET_EVENTS_MAX),
        "users": [{**u, "value": str(u["value"])} for u in users],
        "accounts": _accounts(db, {u["value"] for u in users if u["value"] != NONE}),
        "total": db.scalar(total) or 0,
    }