100 lines
4.0 KiB
Python
100 lines
4.0 KiB
Python
from typing import Literal
|
|
|
|
from fastapi import APIRouter, Depends, Query
|
|
from sqlalchemy import and_, or_, select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from ..cursors import decode, decode_datetime, encode
|
|
from ..db import get_session
|
|
from ..models import NotificationOutbox
|
|
from ..schemas import NotificationOutboxItem, OutboxPage
|
|
from ..timeutil import to_utc
|
|
|
|
router = APIRouter()
|
|
|
|
# PH-002/PH-011/PH-018: explicit sort + filter contracts.
|
|
OutboxState = Literal["pending", "sending", "sent", "retry", "failed", "discarded"]
|
|
OutboxEventType = Literal["firing", "resolved"]
|
|
OutboxNotifier = Literal["debug", "telegram", "webhook", "alertmanager"]
|
|
_SORT_COLUMNS = {
|
|
"created_at": NotificationOutbox.created_at,
|
|
"updated_at": NotificationOutbox.updated_at,
|
|
}
|
|
SortField = Literal["created_at", "updated_at"]
|
|
|
|
|
|
def _encode_outbox_cursor(row: NotificationOutbox, sort: SortField) -> str:
|
|
return encode(to_utc(getattr(row, sort)).isoformat(), row.id)
|
|
|
|
|
|
@router.get("/notifiers/outbox", response_model=OutboxPage)
|
|
async def list_outbox(
|
|
state: OutboxState | None = Query(default=None),
|
|
notifier: OutboxNotifier | None = Query(default=None),
|
|
event_type: OutboxEventType | None = Query(default=None),
|
|
sort: SortField = Query(default="updated_at"),
|
|
direction: Literal["asc", "desc"] = Query(default="desc"),
|
|
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),
|
|
) -> OutboxPage:
|
|
col = _SORT_COLUMNS[sort]
|
|
# PH-review: bidirectional cursor — for `back=True` we run the query in the
|
|
# opposite direction relative to the requested sort and reverse the result
|
|
# client-side. This avoids the unbounded cursor_stack in the URL.
|
|
effective_desc = (direction == "desc") ^ back
|
|
order_col = col.desc() if effective_desc else col.asc()
|
|
tiebreak = NotificationOutbox.id.desc() if effective_desc else NotificationOutbox.id.asc()
|
|
stmt = select(NotificationOutbox).order_by(order_col, tiebreak)
|
|
if state is not None:
|
|
stmt = stmt.where(NotificationOutbox.state == state)
|
|
if notifier is not None:
|
|
stmt = stmt.where(NotificationOutbox.notifier == notifier)
|
|
if event_type is not None:
|
|
stmt = stmt.where(NotificationOutbox.event_type == event_type)
|
|
if cursor is not None:
|
|
ts_str, last_id = decode(cursor, 2)
|
|
ts = decode_datetime(ts_str)
|
|
cmp = col < ts if effective_desc else col > ts
|
|
tie = NotificationOutbox.id < last_id if effective_desc else NotificationOutbox.id > last_id
|
|
stmt = stmt.where(or_(cmp, and_(col == ts, tie)))
|
|
rows = (await session.execute(stmt.limit(limit + 1))).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:
|
|
# Came from a forward page → there is always a next page (the one
|
|
# we just came from). `has_more` here means there are still earlier
|
|
# pages remaining.
|
|
next_cursor = _encode_outbox_cursor(rows[-1], sort)
|
|
if has_more:
|
|
prev_cursor = _encode_outbox_cursor(rows[0], sort)
|
|
else:
|
|
if has_more:
|
|
next_cursor = _encode_outbox_cursor(rows[-1], sort)
|
|
if cursor is not None:
|
|
prev_cursor = _encode_outbox_cursor(rows[0], sort)
|
|
return OutboxPage(
|
|
items=[
|
|
NotificationOutboxItem(
|
|
id=str(r.id),
|
|
notifier=r.notifier,
|
|
state=r.state,
|
|
incident_id=str(r.incident_id),
|
|
event_type=r.event_type,
|
|
attempts=r.attempts,
|
|
next_attempt_at=to_utc(r.next_attempt_at) if r.next_attempt_at else None,
|
|
last_error=r.last_error,
|
|
updated_at=to_utc(r.updated_at),
|
|
)
|
|
for r in rows
|
|
],
|
|
next_cursor=next_cursor,
|
|
prev_cursor=prev_cursor,
|
|
)
|