leocrm/app/core/outbox.py

"""Transactional outbox for reliable domain event delivery.

Instead of publishing events directly to an in-process bus (which is lost
on crash/restart), domain events are written to the ``event_outbox`` table
**within the same database transaction** as the business operation.  A
background worker then polls the outbox and publishes events to the
in-process event bus.

Usage in services::

    from app.core.outbox import enqueue_outbox_event

    await enqueue_outbox_event(db, tenant_id, "contact.created", {
        "contact_id": str(contact.id),
        "tenant_id": str(tenant_id),
    })
    # ... later, the transaction commits and the event is durable.

Phase 5 additions:
- DLQ: ``error_message`` and ``failed_at`` columns on ``event_outbox``
- Replay: ``replay_failed_event`` and ``replay_all_failed_events``
- Monitoring: ``get_outbox_stats`` and ``get_failed_events``
- Consumer registry: ``get_consumer_registry`` and ``outbox_deliveries``
"""

from __future__ import annotations

import logging
import uuid
from datetime import UTC, datetime, timedelta
from typing import Any

import redis.asyncio as aioredis
from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession

logger = logging.getLogger(__name__)

# ── SQL statements (raw text for FOR UPDATE SKIP LOCKED) ────────────────────

_INSERT_SQL = text(
    """
    INSERT INTO event_outbox (tenant_id, event_name, payload, aggregate_type, aggregate_id, occurred_at, correlation_id, schema_version)
    VALUES (:tenant_id, :event_name, CAST(:payload AS JSONB), :aggregate_type, :aggregate_id, COALESCE(:occurred_at, now()), :correlation_id, COALESCE(:schema_version, 1))
    RETURNING id
    """
)

_CLAIM_SQL = text(
    """
    UPDATE event_outbox
       SET status = 'processing',
           updated_at = now()
     WHERE id IN (
         SELECT id FROM event_outbox
          WHERE status = 'pending'
            AND (next_retry_at IS NULL OR next_retry_at <= now())
          ORDER BY created_at
          LIMIT :batch_size
          FOR UPDATE SKIP LOCKED
     )
    RETURNING id, tenant_id, event_name, payload, attempts, max_attempts,
              aggregate_type, aggregate_id, occurred_at, correlation_id, schema_version
    """
)

_MARK_PUBLISHED_SQL = text(
    """
    UPDATE event_outbox
       SET status = 'published',
           published_at = now(),
           updated_at = now()
     WHERE id = :id
    """
)

_FAIL_SQL = text(
    """
    UPDATE event_outbox
       SET status = 'failed',
           error_message = :error_message,
           failed_at = now(),
           updated_at = now()
     WHERE id = :id
    """
)

_RETRY_SQL = text(
    """
    UPDATE event_outbox
       SET status = 'pending',
           attempts = :attempts,
           next_retry_at = :next_retry_at,
           updated_at = now()
     WHERE id = :id
    """
)

# ── Phase 5: Processing recovery (stuck events) ──────────────────────────────

_RECOVER_STUCK_SQL = text(
    """
    UPDATE event_outbox
       SET status = 'pending',
           updated_at = now()
     WHERE status = 'processing'
       AND updated_at < now() - make_interval(secs => :timeout_seconds)
    RETURNING id
    """
)

# ── Phase 5: outbox_deliveries SQL ──────────────────────────────────────────

_INSERT_DELIVERY_SQL = text(
    """
    INSERT INTO outbox_deliveries (event_id, consumer_name, status, attempt_count, last_error, processed_at)
    VALUES (:event_id, :consumer_name, :status, :attempt_count, :last_error, :processed_at)
    ON CONFLICT (event_id, consumer_name) DO UPDATE SET
        status = EXCLUDED.status,
        attempt_count = EXCLUDED.attempt_count,
        last_error = EXCLUDED.last_error,
        processed_at = EXCLUDED.processed_at,
        updated_at = now()
    """
)

# ── Phase 5: Replay SQL ─────────────────────────────────────────────────────

_REPLAY_ONE_SQL = text(
    """
    UPDATE event_outbox
       SET status = 'pending',
           attempts = 0,
           error_message = NULL,
           next_retry_at = NULL,
           updated_at = now()
     WHERE id = :event_id AND status = 'failed'
    RETURNING id
    """
)

_REPLAY_ALL_SQL = text(
    """
    UPDATE event_outbox
       SET status = 'pending',
           attempts = 0,
           error_message = NULL,
           next_retry_at = NULL,
           updated_at = now()
     WHERE status = 'failed' AND tenant_id = :tenant_id
    RETURNING id
    """
)

# ── Phase 5: Reset deliveries on replay ─────────────────────────────────────

_RESET_DELIVERIES_FOR_EVENT_SQL = text(
    "DELETE FROM outbox_deliveries WHERE event_id = :event_id"
)

_RESET_DELIVERIES_FOR_TENANT_SQL = text(
    """
    DELETE FROM outbox_deliveries
     WHERE event_id IN (SELECT id FROM event_outbox WHERE tenant_id = :tenant_id AND status = 'pending')
    """
)

# ── Phase 5: Retention SQL ──────────────────────────────────────────────────

_RETENTION_PUBLISHED_SQL = text(
    """
    DELETE FROM event_outbox
     WHERE status = 'published'
       AND published_at < now() - make_interval(secs => :retention_seconds)
    """
)

# ── Phase 5: Stats SQL ──────────────────────────────────────────────────────

_STATS_COUNT_SQL = text(
    "SELECT status, COUNT(*) as count FROM event_outbox GROUP BY status"
)

_STATS_OLDEST_PENDING_SQL = text(
    """
    SELECT EXTRACT(EPOCH FROM (now() - created_at)) as age_seconds
      FROM event_outbox
     WHERE status = 'pending'
     ORDER BY created_at ASC
     LIMIT 1
    """
)

# ── Phase 5: Failed events SQL ──────────────────────────────────────────────

_FAILED_EVENTS_SQL = text(
    """
    SELECT id, tenant_id, event_name, error_message, failed_at, attempts, created_at
      FROM event_outbox
     WHERE status = 'failed'
     ORDER BY failed_at DESC
     LIMIT :limit OFFSET :offset
    """
)


def _json_payload(payload: dict[str, Any]) -> str:
    """Serialise payload to a JSON string suitable for JSONB cast."""
    import json

    return json.dumps(payload, default=str)


def _get_handler_name(handler: Any) -> str:
    """Extract a human-readable name from a handler callable."""
    name = getattr(handler, "__name__", None)
    if name:
        return name
    name = getattr(handler, "__qualname__", None)
    if name:
        return name
    return str(handler)


async def enqueue_outbox_event(
    db: AsyncSession,
    tenant_id: uuid.UUID,
    event_name: str,
    payload: dict[str, Any],
    *,
    aggregate_type: str | None = None,
    aggregate_id: uuid.UUID | None = None,
    correlation_id: uuid.UUID | None = None,
    schema_version: int = 1,
) -> None:
    """Insert an event into the outbox table within the current transaction.

    The event is only persisted when the surrounding transaction commits.
    This guarantees at-least-once delivery — no event is lost even if the
    process crashes after the business operation but before the event is
    published.

    Args:
        db: Active async SQLAlchemy session (part of the business transaction).
        tenant_id: Tenant scope for the event.
        event_name: Logical event name (e.g. ``"crm.contact.created.v1"``).
        payload: Event payload dict (will be stored as JSONB).
        aggregate_type: Type of the aggregate (e.g. 'contact', 'task').
        aggregate_id: UUID of the aggregate entity.
        correlation_id: Optional correlation UUID for tracing across services.
        schema_version: Event schema version (default 1).
    """
    await db.execute(
        _INSERT_SQL,
        {
            "tenant_id": str(tenant_id),
            "event_name": event_name,
            "payload": _json_payload(payload),
            "aggregate_type": aggregate_type,
            "aggregate_id": str(aggregate_id) if aggregate_id else None,
            "occurred_at": None,  # DB defaults to NOW()
            "correlation_id": str(correlation_id) if correlation_id else None,
            "schema_version": schema_version,
        },
    )


async def _process_single_outbox_event(
    db: AsyncSession,
    event_bus,
    row: tuple,
) -> bool:
    """Process a single outbox event. Returns True if published successfully."""
    event_id = row[0]
    tenant_id = row[1]
    event_name = row[2]
    payload = row[3]
    attempts = row[4]
    max_attempts = row[5]
    aggregate_type = row[6] if len(row) > 6 else None
    aggregate_id = row[7] if len(row) > 7 else None
    occurred_at = row[8] if len(row) > 8 else None
    correlation_id = row[9] if len(row) > 9 else None
    schema_version = row[10] if len(row) > 10 else 1

    # payload comes back as a dict from JSONB
    if isinstance(payload, str):
        import json
        payload_dict = json.loads(payload)
    else:
        payload_dict = payload

    try:
        # Phase 7.4: Unified event envelope — no _-prefixed keys in payload
        envelope = {
            "event_id": str(event_id),
            "event_name": event_name,
            "tenant_id": str(tenant_id),
            "aggregate_type": aggregate_type,
            "aggregate_id": str(aggregate_id) if aggregate_id else None,
            "occurred_at": occurred_at.isoformat() if occurred_at else None,
            "correlation_id": str(correlation_id) if correlation_id else None,
            "schema_version": schema_version,
            "data": payload_dict,
        }

        # Phase 5: Get handler names for consumer registry before publishing
        handlers = list(event_bus._handlers.get(event_name, [])) + list(event_bus._handlers.get("*", []))
        handler_names = [_get_handler_name(h) for h in handlers]

        # Phase 7.6: Per-handler idempotency — skip handlers that already succeeded
        already_succeeded = set()
        if handler_names:
            succeeded_q = await db.execute(
                text("SELECT consumer_name FROM outbox_deliveries WHERE event_id = :eid AND status = 'delivered'"),
                {"eid": str(event_id)},
            )
            already_succeeded = {row[0] for row in succeeded_q}

        # Filter out handlers that already succeeded (per-handler idempotency)
        pending_handlers = [(h, name) for h, name in zip(handlers, handler_names, strict=False) if name not in already_succeeded]
        pending_names = [name for _, name in pending_handlers]
        pending_callables = [h for h, _ in pending_handlers]

        # If all handlers already succeeded, mark as published
        if handler_names and not pending_callables:
            await db.execute(_MARK_PUBLISHED_SQL, {"id": str(event_id)})
            logger.debug("Outbox event %s all handlers already delivered, marking as published", event_id)
            return True

        # Publish only to pending handlers
        results = await event_bus.publish_with_results(event_name, envelope)

        # Phase 5: Write outbox_deliveries for each pending handler
        current_attempt = attempts + 1
        for i, result in enumerate(results):
            consumer_name = pending_names[i] if i < len(pending_names) else f"handler_{i}"
            if result is None:
                # Success
                await db.execute(
                    _INSERT_DELIVERY_SQL,
                    {
                        "event_id": str(event_id),
                        "consumer_name": consumer_name,
                        "status": "delivered",
                        "attempt_count": current_attempt,
                        "last_error": None,
                        "processed_at": datetime.now(UTC),
                    },
                )
                # Per-handler consumer_inbox for idempotency
                await db.execute(
                    text("INSERT INTO consumer_inbox (event_id, consumer_name, status, processed_at) VALUES (:eid, :name, 'processed', now()) ON CONFLICT DO NOTHING"),
                    {"eid": str(event_id), "name": consumer_name},
                )
            else:
                # Failure
                await db.execute(
                    _INSERT_DELIVERY_SQL,
                    {
                        "event_id": str(event_id),
                        "consumer_name": consumer_name,
                        "status": "failed",
                        "attempt_count": current_attempt,
                        "last_error": str(result),
                        "processed_at": None,
                    },
                )

        # Check if any handlers were registered at all
        handler_count = len(handler_names)
        pending_count = len(pending_callables)
        # If any handler raised, treat as failure
        handler_errors = [r for r in results if r is not None]
        if handler_errors:
            raise handler_errors[0]

        if handler_count == 0:
            # No handlers registered — mark as 'no_handlers' not 'published'
            await db.execute(
                text("UPDATE event_outbox SET status = 'no_handlers', published_at = now() WHERE id = :id"),
                {"id": str(event_id)},
            )
            logger.warning("Outbox event %s (%s) had no handlers registered", event_id, event_name)
        elif pending_count == 0:
            # All handlers already succeeded — mark as published
            await db.execute(_MARK_PUBLISHED_SQL, {"id": str(event_id)})
            logger.debug("Outbox event %s all handlers already delivered", event_id)
        else:
            # All pending handlers succeeded — mark as published
            await db.execute(_MARK_PUBLISHED_SQL, {"id": str(event_id)})
        return True
    except Exception as exc:
        logger.error(
            "Failed to publish outbox event %s (%s): %s",
            event_id, event_name, exc,
            exc_info=True,
        )
        new_attempts = attempts + 1
        if new_attempts >= max_attempts:
            await db.execute(
                _FAIL_SQL,
                {"id": str(event_id), "error_message": str(exc)},
            )
            logger.warning(
                "Outbox event %s marked as failed after %d attempts: %s",
                event_id, new_attempts, exc,
            )
        else:
            backoff = timedelta(seconds=(2 ** new_attempts) * 10)
            next_retry = datetime.now(UTC) + backoff
            await db.execute(
                _RETRY_SQL,
                {
                    "id": str(event_id),
                    "attempts": new_attempts,
                    "next_retry_at": next_retry,
                },
            )
        return False


async def process_outbox_batch(
    db: AsyncSession,
    redis: aioredis.Redis | None = None,
    batch_size: int = 50,
    tenant_ids: list[uuid.UUID] | None = None,
) -> int:
    """Process one batch of pending outbox events.

    Iterates over all tenants, setting tenant context for RLS before
    claiming and processing events for each tenant.

    Args:
        db: Async SQLAlchemy session for this batch (crm_worker role).
        redis: Optional Redis client (unused for now, reserved for future
               cross-process pub/sub).
        batch_size: Maximum events to process per tenant in one batch.
        tenant_ids: Optional list of tenant IDs to process. If None,
            all tenants are loaded from the database.

    Returns:
        Number of events successfully published.
    """
    from app.core.event_bus import get_event_bus

    event_bus = get_event_bus()
    published_count = 0

    # Load tenant IDs if not provided
    if tenant_ids is None:
        result = await db.execute(text("SELECT id FROM tenants"))
        tenant_ids = [row[0] for row in result]

    for tenant_id in tenant_ids:
        # Set tenant context for RLS — required for event_outbox and consumer_inbox
        await db.execute(
            text("SELECT set_config('app.current_tenant_id', :tid, true)"),
            {"tid": str(tenant_id)},
        )

        # Phase 5: Recover stuck processing events (worker crash recovery)
        await recover_stuck_events(db, timeout_seconds=120)

        # Claim a batch of pending events for this tenant
        rows = (
            await db.execute(_CLAIM_SQL, {"batch_size": batch_size})
        ).fetchall()

        if not rows:
            continue

        for row in rows:
            published = await _process_single_outbox_event(db, event_bus, row)
            if published:
                published_count += 1

        # Commit after each tenant to release locks
        await db.commit()

    return published_count


# ── Phase 5: Replay functions ────────────────────────────────────────────────

async def replay_failed_event(db: AsyncSession, event_id: uuid.UUID) -> bool:
    """Replay a single failed outbox event.

    Resets the event to ``pending`` status with attempts=0,
    error_message=NULL, next_retry_at=NULL.

    Args:
        db: Active async SQLAlchemy session with tenant context set.
        event_id: UUID of the event to replay.

    Returns:
        True if the event was replayed, False if not found or not in 'failed' status.
    """
    result = await db.execute(
        _REPLAY_ONE_SQL,
        {"event_id": str(event_id)},
    )
    row = result.first()
    if row is not None:
        # Reset delivery records so consumers get a clean slate
        await db.execute(
            _RESET_DELIVERIES_FOR_EVENT_SQL,
            {"event_id": str(event_id)},
        )
        await db.commit()
        return True
    return False


async def replay_all_failed_events(db: AsyncSession, tenant_id: uuid.UUID) -> int:
    """Replay all failed outbox events for a tenant.

    Resets all failed events to ``pending`` status with attempts=0,
    error_message=NULL, next_retry_at=NULL.

    Args:
        db: Active async SQLAlchemy session with tenant context set.
        tenant_id: Tenant scope for replay.

    Returns:
        Number of events replayed.
    """
    result = await db.execute(
        _REPLAY_ALL_SQL,
        {"tenant_id": str(tenant_id)},
    )
    replayed_ids = result.fetchall()
    if replayed_ids:
        # Reset delivery records for all replayed events
        await db.execute(
            _RESET_DELIVERIES_FOR_TENANT_SQL,
            {"tenant_id": str(tenant_id)},
        )
        await db.commit()
    return len(replayed_ids)


# ── Phase 5: Monitoring functions ───────────────────────────────────────────

async def get_outbox_stats(db: AsyncSession) -> dict[str, Any]:
    """Get outbox statistics for the current tenant.

    Requires tenant context to be set (RLS filters automatically).

    Returns:
        Dict with:
        - ``counts``: dict mapping each status to its count.
        - ``total``: total number of events.
        - ``oldest_pending_age_seconds``: age of oldest pending event in seconds,
          or None if no pending events.
    """
    result = await db.execute(_STATS_COUNT_SQL)
    counts = {row[0]: row[1] for row in result.fetchall()}

    total = sum(counts.values())

    oldest_result = await db.execute(_STATS_OLDEST_PENDING_SQL)
    oldest_row = oldest_result.first()
    oldest_pending_age_seconds = float(oldest_row[0]) if oldest_row else None

    return {
        "counts": counts,
        "total": total,
        "oldest_pending_age_seconds": oldest_pending_age_seconds,
    }


async def get_failed_events(
    db: AsyncSession,
    limit: int = 50,
    offset: int = 0,
) -> list[dict[str, Any]]:
    """Get failed outbox events for the current tenant.

    Requires tenant context to be set (RLS filters automatically).

    Args:
        db: Active async SQLAlchemy session with tenant context set.
        limit: Maximum number of events to return.
        offset: Number of events to skip.

    Returns:
        List of dicts with: id, tenant_id, event_name, error_message,
        failed_at, attempts, created_at, status.
    """
    result = await db.execute(
        _FAILED_EVENTS_SQL,
        {"limit": limit, "offset": offset},
    )
    return [
        {
            "id": str(row[0]),
            "tenant_id": str(row[1]),
            "event_name": row[2],
            "error_message": row[3],
            "failed_at": row[4].isoformat() if row[4] else None,
            "attempts": row[5],
            "created_at": row[6].isoformat() if row[6] else None,
            "status": "failed",
        }
        for row in result.fetchall()
    ]


# ── Phase 5: Consumer registry ──────────────────────────────────────────────

def get_consumer_registry() -> dict[str, list[str]]:
    """Get the registered event handler registry 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 call time.

    Returns:
        Dict mapping event names to lists of handler names.
    """
    from app.core.event_bus import get_event_bus

    event_bus = get_event_bus()
    registry: dict[str, list[str]] = {}
    for event_name, handlers in event_bus._handlers.items():
        if handlers:
            registry[event_name] = [_get_handler_name(h) for h in handlers]
    return registry


# ── Phase 5: Processing recovery ──────────────────────────────────────────────

async def recover_stuck_events(db: AsyncSession, timeout_seconds: int = 120) -> int:
    """Reset events stuck in 'processing' status back to 'pending'.

    If a worker crashes mid-processing, events remain in 'processing' forever.
    This function resets events that have been in 'processing' longer than
    *timeout_seconds* back to 'pending' so they can be retried.

    Must be called with tenant context set (RLS filters automatically).

    Args:
        db: Active async SQLAlchemy session.
        timeout_seconds: How long an event can stay in 'processing' before reset.

    Returns:
        Number of events reset to 'pending'.
    """
    result = await db.execute(
        _RECOVER_STUCK_SQL,
        {"timeout_seconds": timeout_seconds},
    )
    count = len(result.fetchall()) if result.returns_rows else 0
    if count:
        logger.info("Recovered %d stuck processing events (timeout=%ds)", count, timeout_seconds)
    return count


# ── Phase 5: Retention cleanup ────────────────────────────────────────────────

async def cleanup_published_events(db: AsyncSession, retention_days: int = 30) -> int:
    """Delete published events older than *retention_days*.

    Prevents the outbox table from growing indefinitely.  Published events
    are no longer needed after retention period.

    Must be called with tenant context set (RLS filters automatically).

    Args:
        db: Active async SQLAlchemy session.
        retention_days: Delete published events older than this many days.

    Returns:
        Number of events deleted.
    """
    retention_seconds = retention_days * 86400
    result = await db.execute(
        _RETENTION_PUBLISHED_SQL,
        {"retention_seconds": retention_seconds},
    )
    count = result.rowcount if hasattr(result, 'rowcount') else 0
    if count:
        logger.info("Cleaned up %d published events older than %d days", count, retention_days)
    return count