monlet/server/monlet_server/api/events.py

import base64
import binascii
from datetime import datetime
from uuid import UUID

from fastapi import APIRouter, Depends, HTTPException, Query, status
from sqlalchemy import and_, or_, select
from sqlalchemy.ext.asyncio import AsyncSession

from .. import metrics
from ..agent_auth import AgentCredential, get_agent_credential
from ..db import get_session
from ..logging_config import get_logger
from ..models import Event
from ..schemas import (
    ID_PATTERN,
    EventBatchAcceptedResponse,
    EventBatchRequest,
    EventsPage,
    StoredCheckResultEvent,
)
from ..services.ingestion import accepted_agent_for_key, ingest_event
from ..timeutil import to_utc

router = APIRouter()

log = get_logger("monlet.events")


def _encode_cursor(received_at: datetime, event_id) -> str:
    raw = f"{to_utc(received_at).isoformat()}|{event_id}".encode()
    return base64.urlsafe_b64encode(raw).decode().rstrip("=")


def _decode_cursor(cursor: str) -> tuple[datetime, UUID]:
    pad = "=" * (-len(cursor) % 4)
    try:
        raw = base64.urlsafe_b64decode(cursor + pad).decode()
        ts_str, eid = raw.split("|", 1)
        # PH-010: decode event_id to UUID before comparing to the UUID column
        # to avoid silent type-coercion behavior across DB drivers.
        return to_utc(datetime.fromisoformat(ts_str)), UUID(eid)
    except (binascii.Error, ValueError) as exc:
        raise HTTPException(status_code=400, detail="invalid cursor") from exc


@router.post(
    "/events",
    response_model=EventBatchAcceptedResponse,
    status_code=status.HTTP_202_ACCEPTED,
)
async def events(
    body: EventBatchRequest,
    credential: AgentCredential = Depends(get_agent_credential),
    session: AsyncSession = Depends(get_session),
) -> EventBatchAcceptedResponse:
    agent = await accepted_agent_for_key(session, credential)
    if agent is None:
        raise HTTPException(status_code=403, detail="agent not accepted")
    accepted = 0
    deduplicated = 0
    for ev in body.events:
        try:
            async with session.begin_nested():
                ok = await ingest_event(session, agent.agent_id, ev)
        except HTTPException as exc:
            if exc.status_code != 400:
                raise
            log.warning(
                "event_rejected",
                check_id=ev.check_id,
                agent_id=agent.agent_id,
                code=exc.detail,
            )
            metrics.events_total.labels(result="rejected").inc()
            continue
        if ok:
            accepted += 1
        else:
            deduplicated += 1
    await session.commit()
    return EventBatchAcceptedResponse(accepted=accepted, deduplicated=deduplicated)


@router.get("/events/query", response_model=EventsPage)
async def query_events(
    agent_id: str | None = Query(default=None, max_length=128, pattern=ID_PATTERN),
    check_id: str | None = Query(default=None, max_length=128, pattern=ID_PATTERN),
    cursor: str | None = Query(default=None, max_length=512),
    back: bool = Query(default=False),
    limit: int = Query(default=100, ge=1, le=500),
    session: AsyncSession = Depends(get_session),
) -> EventsPage:
    # PH-review: bidirectional cursor. The natural sort is DESC on
    # (received_at, event_id); `back=True` reverses the query so we can return
    # the previous page without an unbounded cursor stack in the URL.
    if back:
        stmt = select(Event).order_by(Event.received_at.asc(), Event.event_id.asc())
    else:
        stmt = select(Event).order_by(Event.received_at.desc(), Event.event_id.desc())
    if agent_id is not None:
        stmt = stmt.where(Event.agent_id == agent_id)
    if check_id is not None:
        stmt = stmt.where(Event.check_id == check_id)
    if cursor is not None:
        ts, eid = _decode_cursor(cursor)
        if back:
            stmt = stmt.where(
                or_(
                    Event.received_at > ts,
                    and_(Event.received_at == ts, Event.event_id > eid),
                )
            )
        else:
            stmt = stmt.where(
                or_(
                    Event.received_at < ts,
                    and_(Event.received_at == ts, Event.event_id < eid),
                )
            )
    stmt = stmt.limit(limit + 1)
    rows = (await session.execute(stmt)).scalars().all()
    has_more = len(rows) > limit
    rows = rows[:limit]
    if back:
        rows = list(reversed(rows))
    next_cursor: str | None = None
    prev_cursor: str | None = None
    if rows:
        if back:
            next_cursor = _encode_cursor(rows[-1].received_at, rows[-1].event_id)
            if has_more:
                prev_cursor = _encode_cursor(rows[0].received_at, rows[0].event_id)
        else:
            if has_more:
                next_cursor = _encode_cursor(rows[-1].received_at, rows[-1].event_id)
            if cursor is not None:
                prev_cursor = _encode_cursor(rows[0].received_at, rows[0].event_id)
    items = [
        StoredCheckResultEvent(
            event_id=str(r.event_id),
            agent_id=r.agent_id,
            check_id=r.check_id,
            observed_at=to_utc(r.observed_at),
            received_at=to_utc(r.received_at),
            status=r.status,
            exit_code=r.exit_code,
            duration_ms=r.duration_ms,
            output=r.output,
            output_truncated=r.output_truncated,
            notifications_enabled=r.notifications_enabled,
            incident_key=r.incident_key,
        )
        for r in rows
    ]
    return EventsPage(items=items, next_cursor=next_cursor, prev_cursor=prev_cursor)