siftlode/backend/app/routes/audit.py

"""Admin audit log API: read the append-only who/what/when trail, and clear it.

Admin-only. The activity page reads it a page at a time, newest first, filtered on the server like
the server log (`_keyset` paging), so the whole trail is reachable however long it grows."""

from dataclasses import dataclass
from datetime import datetime
from typing import Literal

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

from app import audit
from app.audit import AuditAction
from app.auth import admin_user
from app.db import get_db
from app.models import AuditLog, User
from app.routes._keyset import PAGE_MAX, accounts, older_than, page, utc
from app.utils import iso, like_escape

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

# `none` selects the rows no account made (the background) — a value to filter on, not "any".
NONE = "none"


def _serialize(row: AuditLog) -> dict:
    return {
        "id": row.id,
        "created_at": iso(row.created_at),
        # NULL = the background did it (the client shows "System"); the address is in `accounts`.
        "actor_id": row.actor_id,
        "action": row.action,
        "target_type": row.target_type,
        "target_id": row.target_id,
        "summary": row.summary,
        "before": row.before,
        "after": row.after,
        "request_id": row.request_id,
    }


@dataclass(frozen=True)
class AuditFilters:
    action: str | None
    actor: str | None
    q: str | None
    since: datetime | None
    until: datetime | None

    def conditions(self, skip: Literal["action"] | None = None) -> list:
        """The WHERE conditions, ANDed. `skip="action"` leaves the action filter out, so the action
        menu counts every action under the other filters (the way back from a pick)."""
        out = []
        if self.action is not None and skip != "action":
            out.append(AuditLog.action == self.action)
        if self.actor is not None:
            out.append(
                AuditLog.actor_id.is_(None)
                if self.actor == NONE
                else AuditLog.actor_id == int(self.actor)
            )
        if self.since is not None:
            out.append(AuditLog.created_at >= utc(self.since))
        if self.until is not None:
            out.append(AuditLog.created_at < utc(self.until))
        if self.q and self.q.strip():
            like = f"%{like_escape(self.q.strip())}%"
            out.append(
                or_(
                    AuditLog.summary.ilike(like),
                    AuditLog.target_id.ilike(like),
                    AuditLog.action.ilike(like),
                    AuditLog.request_id.ilike(like),
                    cast(AuditLog.before, Text).ilike(like),
                    cast(AuditLog.after, Text).ilike(like),
                )
            )
        return out


def audit_filters(
    action: str | None = Query(default=None, max_length=48, pattern=r"^[a-z][a-z0-9_]*$"),
    user: str | None = Query(default=None, pattern=r"^(none|[0-9]{1,10})$"),
    q: str | None = Query(default=None, max_length=200),
    since: datetime | None = None,
    until: datetime | None = None,
) -> AuditFilters:
    return AuditFilters(action, user, q, since, until)


@router.get("")
def list_audit(
    filters: AuditFilters = Depends(audit_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; `before` continues from a
    previous page's `next_cursor`."""
    query = select(AuditLog).where(*filters.conditions())
    if before is not None:
        query = query.where(older_than(AuditLog.created_at, AuditLog.id, before))
    order = (AuditLog.created_at.desc(), AuditLog.id.desc())
    rows, cursor = page(
        list(db.execute(query.order_by(*order).limit(limit + 1)).scalars().all()),
        limit,
        lambda r: r.created_at,
    )
    return {
        "items": [_serialize(r) for r in rows],
        "accounts": accounts(db, {r.actor_id for r in rows if r.actor_id is not None}),
        "next_cursor": cursor,
    }


@router.get("/facets")
def audit_facets(
    filters: AuditFilters = Depends(audit_filters),
    _: User = Depends(admin_user),
    db: Session = Depends(get_db),
) -> dict:
    """The actions present under the other filters, with counts — what the action menu offers."""
    n = func.count().label("n")
    stmt = (
        select(AuditLog.action, n)
        .where(*filters.conditions("action"))
        .group_by(AuditLog.action)
        .order_by(n.desc(), AuditLog.action)
    )
    return {"actions": [{"value": a, "count": c} for a, c in db.execute(stmt).all()]}


@router.delete("")
def clear_audit(admin: User = Depends(admin_user), db: Session = Depends(get_db)) -> dict:
    """Clear the audit log. Leaves ONE row recording who cleared it + how many (tamper-evidence)."""
    n = db.execute(delete(AuditLog)).rowcount or 0
    audit.record(
        db,
        admin.id,
        AuditAction.AUDIT_CLEAR,
        summary=f"Cleared {n} audit entries",
        after={"cleared": n},  # the count as data: the log names it in the reader's language
    )
    db.commit()
    return {"ok": True, "cleared": n}