leocrm/app/routes/outbox.py

"""Outbox monitoring and management endpoints — DLQ, stats, replay, consumer registry.

All endpoints require authentication and admin role.
Tenant context is set automatically via the ``get_current_user`` dependency
(used internally by ``require_admin``), which calls ``set_tenant_context``
on the shared database session.  RLS policies then filter all outbox queries
automatically.
"""

from __future__ import annotations

import logging
import uuid
from typing import Any

from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.db import get_db
from app.core.outbox import (
    cleanup_published_events,
    get_consumer_registry,
    get_failed_events,
    get_outbox_stats,
    recover_stuck_events,
    replay_all_failed_events,
    replay_failed_event,
)
from app.deps import require_admin

logger = logging.getLogger(__name__)

router = APIRouter(prefix="/api/v1/outbox", tags=["outbox"])


@router.get("/stats")
async def outbox_stats(
    db: AsyncSession = Depends(get_db),
    current_user: dict[str, Any] = Depends(require_admin),
):
    """Outbox statistics: counts per status, oldest pending age, total events.

    Returns a JSON object with:
    - ``counts``: dict mapping each status to its count (pending, processing,
      published, failed, no_handlers).
    - ``total``: total number of events for the current tenant.
    - ``oldest_pending_age_seconds``: age in seconds of the oldest pending
      event, or ``null`` if none pending.
    """
    return await get_outbox_stats(db)


@router.get("/failed")
async def outbox_failed(
    limit: int = Query(50, ge=1, le=500),
    offset: int = Query(0, ge=0),
    db: AsyncSession = Depends(get_db),
    current_user: dict[str, Any] = Depends(require_admin),
):
    """List failed outbox events with error details (paginated).

    Each event includes: id, tenant_id, event_name, error_message,
    failed_at, attempts, created_at, status.
    """
    events = await get_failed_events(db, limit=limit, offset=offset)
    return {
        "events": events,
        "limit": limit,
        "offset": offset,
        "count": len(events),
    }


@router.post("/replay/{event_id}")
async def outbox_replay_single(
    event_id: str,
    db: AsyncSession = Depends(get_db),
    current_user: dict[str, Any] = Depends(require_admin),
):
    """Replay a single failed outbox event.

    Resets the event to ``pending`` status with attempts=0,
    error_message=NULL, next_retry_at=NULL.
    """
    try:
        eid = uuid.UUID(event_id)
    except ValueError:
        raise HTTPException(
            status_code=400,
            detail={
                "detail": "Invalid event ID format",
                "code": "invalid_uuid",
            },
        )

    replayed = await replay_failed_event(db, eid)
    if not replayed:
        raise HTTPException(
            status_code=404,
            detail={
                "detail": "Failed event not found or not in 'failed' status",
                "code": "not_found",
            },
        )
    return {"replayed": True, "event_id": event_id}


@router.post("/replay-all")
async def outbox_replay_all(
    db: AsyncSession = Depends(get_db),
    current_user: dict[str, Any] = Depends(require_admin),
):
    """Replay all failed outbox events for the current tenant.

    Resets all failed events to ``pending`` status.
    Returns the number of events replayed.
    """
    tenant_id = uuid.UUID(current_user["tenant_id"])
    count = await replay_all_failed_events(db, tenant_id)
    return {"replayed_count": count, "tenant_id": str(tenant_id)}


@router.get("/consumer-registry")
async def outbox_consumer_registry(
    current_user: dict[str, Any] = Depends(require_admin),
):
    """List registered event handlers from the in-process event bus.

    Returns a mapping of ``event_name`` to a list of consumer (handler) names.
    This is read from ``event_bus._handlers`` at request time.
    """
    return {"registry": get_consumer_registry()}


@router.post("/recover-stuck")
async def outbox_recover_stuck(
    timeout_seconds: int = Query(120, ge=10, le=3600),
    db: AsyncSession = Depends(get_db),
    current_user: dict[str, Any] = Depends(require_admin),
):
    """Reset events stuck in 'processing' status back to 'pending'.

    If a worker crashes mid-processing, events remain in 'processing' forever.
    This endpoint resets events that have been in 'processing' longer than
    *timeout_seconds* back to 'pending' so they can be retried.
    """
    count = await recover_stuck_events(db, timeout_seconds=timeout_seconds)
    return {"recovered_count": count, "timeout_seconds": timeout_seconds}


@router.post("/cleanup-published")
async def outbox_cleanup_published(
    retention_days: int = Query(30, ge=1, le=365),
    db: AsyncSession = Depends(get_db),
    current_user: dict[str, Any] = Depends(require_admin),
):
    """Delete published events older than *retention_days*.

    Prevents the outbox table from growing indefinitely.
    """
    count = await cleanup_published_events(db, retention_days=retention_days)
    return {"deleted_count": count, "retention_days": retention_days}