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.
"""

from __future__ import annotations

import logging
import uuid
from datetime import datetime, timedelta, timezone
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)
    VALUES (:tenant_id, :event_name, CAST(:payload AS JSONB))
    """
)

_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
    """
)

_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',
           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
    """
)


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)


async def enqueue_outbox_event(
    db: AsyncSession,
    tenant_id: uuid.UUID,
    event_name: str,
    payload: dict[str, Any],
) -> 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. ``"contact.created"``).
        payload: Event payload dict (will be stored as JSONB).
    """
    await db.execute(
        _INSERT_SQL,
        {
            "tenant_id": str(tenant_id),
            "event_name": event_name,
            "payload": _json_payload(payload),
        },
    )


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

    1. Claim up to *batch_size* pending events using ``FOR UPDATE SKIP LOCKED``
       so multiple workers don't interfere.
    2. Publish each event to the in-process event bus (for local handlers).
    3. On success: mark as ``published``.
    4. On failure: increment attempts, schedule retry with exponential
       backoff, or mark as ``failed`` if max attempts exceeded.

    Args:
        db: Async SQLAlchemy session for this batch.
        redis: Optional Redis client (unused for now, reserved for future
               cross-process pub/sub).
        batch_size: Maximum events to process in one batch.

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

    event_bus = get_event_bus()
    published_count = 0

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

    if not rows:
        return 0

    for row in rows:
        event_id = row[0]
        event_name = row[2]
        payload = row[3]
        attempts = row[4]
        max_attempts = row[5]

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

        try:
            results = await event_bus.publish_with_results(event_name, payload_dict)
            # 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]
            await db.execute(_MARK_PUBLISHED_SQL, {"id": str(event_id)})
            published_count += 1
        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)})
                logger.warning(
                    "Outbox event %s marked as failed after %d attempts",
                    event_id, new_attempts,
                )
            else:
                backoff = timedelta(seconds=(2 ** new_attempts) * 10)
                next_retry = datetime.now(timezone.utc) + backoff
                await db.execute(
                    _RETRY_SQL,
                    {
                        "id": str(event_id),
                        "attempts": new_attempts,
                        "next_retry_at": next_retry,
                    },
                )

    await db.commit()
    return published_count