API Reference¶

class AckHandle(ABC):
    """Acknowledgement handle for broker commit/ack operations."""

    @abstractmethod
    async def commit(self) -> None:
        """Commit/acknowledge the consumed message."""
        ...

class AckStrategy(StrEnum):
    """Commit semantics for consumed broker messages."""

    AT_MOST_ONCE = "at_most_once"
    AT_LEAST_ONCE = "at_least_once"
    EXACTLY_ONCE_INBOX = "exactly_once_inbox"

class BaseEvent(BaseModel):
    """Base generic event domain entity for Transactional Outbox and Inbox.

    Provides core fields for event identification, status tracking, retry
    orchestration, and metadata. All events are frozen (immutable) to
    ensure data integrity during processing.
    """

    model_config = ConfigDict(frozen=True)

    # Identifiers
    id: UUID = Field(default_factory=uuid4)
    event_type: StrippedNonEmptyStr = Field(max_length=EVENT_TYPE_MAX_LENGTH)

    # Payload & Metadata
    payload: dict[str, JsonValue] = Field(..., description="Event payload as a JSON-compatible dictionary.")
    headers: dict[str, str] | None = Field(default=None, description="Optional message headers.")
    trace_id: StrippedNonEmptyStr | None = Field(default=None, max_length=TRACE_ID_MAX_LENGTH)
    idempotency_key: StrippedNonEmptyStr | None = Field(default=None, max_length=IDEMPOTENCY_KEY_MAX_LENGTH)
    correlation_id: StrippedNonEmptyStr | None = Field(default=None, max_length=CORRELATION_ID_MAX_LENGTH)
    causation_id: StrippedNonEmptyStr | None = Field(default=None, max_length=CAUSATION_ID_MAX_LENGTH)
    schema_version: StrippedNonEmptyStr | None = Field(default=None, max_length=SCHEMA_VERSION_MAX_LENGTH)

    # Status & Retries
    status: EventStatus = EventStatus.PENDING
    attempts_made: int = Field(
        default=0,
        ge=0,
        description="Number of failed processing attempts recorded (does not include a successful attempt).",
    )
    max_attempts: int = Field(
        default=DEFAULT_MAX_ATTEMPTS,
        ge=1,
        description=(
            "Maximum number of processing failures allowed. "
            "Event transitions to FAILED when attempts_made == max_attempts."
        ),
    )
    last_error: str | None = Field(default=None, max_length=LAST_ERROR_MAX_LENGTH)

    # Timing
    created_at: datetime = Field(default_factory=utc_now)
    scheduled_at: datetime = Field(default_factory=utc_now)
    completed_at: datetime | None = Field(default=None)
    locked_at: datetime | None = Field(default=None)
    locked_by: StrippedNonEmptyStr | None = Field(default=None, max_length=WORKER_ID_MAX_LENGTH)

    @field_validator("created_at", "scheduled_at", "completed_at", "locked_at", mode="after")
    @classmethod
    def validate_timezone(cls, v: datetime | None) -> datetime | None:
        """Ensure all datetimes are timezone-aware and normalize to UTC."""
        if v is None:
            return v

        if v.tzinfo is None:
            raise ValueError("Datetime must be timezone-aware")

        # Normalize any timezone to UTC
        return v.astimezone(UTC)

    @field_validator("scheduled_at")
    @classmethod
    def validate_scheduled_at(cls, v: datetime, info: ValidationInfo) -> datetime:
        """Ensure scheduled_at is within reasonable range."""
        created_at = info.data.get("created_at")
        if created_at is not None:
            skew_limit = DEFAULT_SCHEDULE_AT_SKEW_SECONDS
            max_future_seconds = DEFAULT_SCHEDULE_AT_MAX_FUTURE_SECONDS
            if info.context:
                skew_limit = info.context.get("scheduled_at_skew_seconds", DEFAULT_SCHEDULE_AT_SKEW_SECONDS)
                max_future_seconds = info.context.get(
                    "scheduled_at_max_future_seconds", DEFAULT_SCHEDULE_AT_MAX_FUTURE_SECONDS
                )

            if skew_limit < 0:
                raise ValueError(f"scheduled_at_skew_seconds must be >= 0, got {skew_limit}")
            if max_future_seconds < 1:
                raise ValueError(f"scheduled_at_max_future_seconds must be >= 1, got {max_future_seconds}")

            if v < created_at and (created_at - v).total_seconds() > skew_limit:
                raise ValueError(f"scheduled_at {v} cannot be significantly before created_at {created_at}")

            if (v - created_at).total_seconds() > max_future_seconds:
                raise ValueError(
                    f"scheduled_at {v} is too far in the future (max {max_future_seconds} seconds from creation)"
                )

        return v

    @field_validator("payload")
    @classmethod
    def validate_payload_content(cls, v: dict[str, JsonValue], info: ValidationInfo) -> dict[str, JsonValue]:
        return validate_payload(v, info)

    @field_validator("headers")
    @classmethod
    def validate_headers_content(cls, v: dict[str, str] | None, info: ValidationInfo) -> dict[str, str] | None:
        return validate_headers(v, info)

    @model_validator(mode="after")
    def _pydantic_validate_invariants(self) -> Self:
        """Internal Pydantic model validator hook."""
        return self.validate_invariants()

    def validate_invariants(self) -> Self:
        """Enforce business invariants across all fields."""
        self._validate_status_timing()
        self._validate_attempts()
        self._validate_lock()
        return self

    def _validate_status_timing(self) -> None:
        """Validate status vs completed timing."""
        if self.status == EventStatus.COMPLETED:
            if self.completed_at is None:
                raise ValueError("completed_at must be set when status is COMPLETED")

            # Allow small clock skew (e.g. 1 second)
            skew_limit = 1.0
            if (
                self.completed_at < self.created_at
                and (self.created_at - self.completed_at).total_seconds() > skew_limit
            ):
                raise ValueError(f"completed_at {self.completed_at} cannot be before created_at {self.created_at}")
            if (
                self.completed_at < self.scheduled_at
                and (self.scheduled_at - self.completed_at).total_seconds() > skew_limit
            ):
                raise ValueError(f"completed_at {self.completed_at} cannot be before scheduled_at {self.scheduled_at}")
        elif self.completed_at is not None:
            raise ValueError(f"completed_at must be None when status is {self.status}")

    def _validate_attempts(self) -> None:
        """Validate attempts consistency."""
        if self.attempts_made > self.max_attempts:
            raise ValueError(f"attempts_made ({self.attempts_made}) cannot exceed max_attempts ({self.max_attempts})")

        if self.status == EventStatus.FAILED and self.attempts_made != self.max_attempts:
            raise ValueError(
                f"status is FAILED, but attempts_made ({self.attempts_made}) "
                f"must equal max_attempts ({self.max_attempts})"
            )
        if self.status == EventStatus.PENDING and self.attempts_made >= self.max_attempts:
            raise ValueError(
                f"status is PENDING, but attempts_made ({self.attempts_made}) "
                f"has reached max_attempts ({self.max_attempts})"
            )

    def _validate_lock(self) -> None:
        """Validate lock consistency."""
        if (self.locked_at is not None) != (self.locked_by is not None):
            raise ValueError("locked_at and locked_by must be both set or both None")

        # Locked events must always be PENDING.
        if self.is_locked and self.status != EventStatus.PENDING:
            raise ValueError(f"locked event must be in PENDING status, got {self.status}")

    @property
    def is_locked(self) -> bool:
        """Check if the event is currently locked."""
        return self.locked_at is not None

    @property
    def can_retry(self) -> bool:
        """Check if the event can be retried."""
        return self.status == EventStatus.PENDING and self.attempts_made < self.max_attempts

    @property
    def attempts_left(self) -> int:
        """Number of attempts remaining before FAILED status."""
        if self.status != EventStatus.PENDING:
            return 0
        return self.max_attempts - self.attempts_made

    @property
    def failure_count(self) -> int:
        """Number of attempts made when in FAILED status."""
        return self.attempts_made if self.status == EventStatus.FAILED else 0

    @staticmethod
    def truncate_error(error: str, max_bytes: int, suffix: str) -> str:
        """Truncate error message to database limit (in bytes)."""
        if max_bytes < 1:
            raise ValueError(f"max_bytes must be >= 1, got {max_bytes}")

        stripped = error.strip()
        if not stripped:
            raise ValueError("Error message cannot be empty or whitespace")

        encoded = stripped.encode("utf-8")
        if len(encoded) <= max_bytes:
            return stripped

        suffix_encoded = suffix.encode("utf-8")
        if len(suffix_encoded) >= max_bytes:
            return encoded[:max_bytes].decode("utf-8", errors="ignore")

        keep_bytes = max_bytes - len(suffix_encoded)
        return encoded[:keep_bytes].decode("utf-8", errors="ignore") + suffix

class BaseEventHandler(ABC):
    """Base class for event handlers.

    Subclasses should define topic at class level and use @event_handler
    decorator on methods to register them.

    Example:
        class UserEventHandlers(BaseEventHandler):
            topic = "users"

            @event_handler(UserEventType.CREATED)
            async def on_user_created(self, event: InboxEvent, uow: AsyncUnitOfWork):
                # Handle event
                pass
    """

    topic: ClassVar[str | StrEnum]  # Must be defined in subclass

class BaseEventSchema(BaseModel, ABC):
    """Base class for all event payload schemas with self-registration and versioning support."""

    _registry: ClassVar[dict[str, dict[str, type["BaseEventSchema"]]]] = {}
    _migrations: ClassVar[dict[str, dict[tuple[str, str], Callable[[dict[str, Any]], dict[str, Any]]]]] = {}

    @classmethod
    def __init_subclass__(cls, event_type: str | None = None, version_prefix: str | None = None, **kwargs: Any) -> None:
        super().__init_subclass__(**kwargs)
        if event_type:
            if event_type not in cls._registry:
                cls._registry[event_type] = {}

            # If a version_prefix is provided (e.g. "1" or "1.2"), use it as the registration key.
            # Otherwise, use the full version from schema_version().
            key = version_prefix or cls.schema_version()
            cls._registry[event_type][key] = cls

    @classmethod
    @abstractmethod
    def schema_version(cls) -> str:
        """Return full Semantic Version (e.g., '1.0.0')."""

    @classmethod
    def register_migration(
        cls,
        event_type: str,
        from_version: str,
        to_version: str,
        func: Callable[[dict[str, Any]], dict[str, Any]],
    ) -> None:
        """Register a migration function between two versions of an event type."""
        if event_type not in cls._migrations:
            cls._migrations[event_type] = {}
        cls._migrations[event_type][(from_version, to_version)] = func

    @classmethod
    def migrate_payload(
        cls,
        event_type: str,
        payload: dict[str, Any],
        from_version: str | None,
        to_version: str,
    ) -> dict[str, Any]:
        """Migrate payload to the target version if migrations are available.

        If from_version is None, it assumes the earliest known version or skips.
        Currently supports single-step migrations.
        """
        if from_version == to_version:
            return payload

        if not from_version:
            return payload

        type_migrations = cls._migrations.get(event_type)
        if not type_migrations:
            return payload

        # Simple path: try to find a direct migration
        func = type_migrations.get((from_version, to_version))
        if func:
            return func(payload)

        # Multi-step migration pathfinding using BFS
        queue: deque[tuple[str, dict[str, Any]]] = deque([(from_version, payload)])
        visited: set[str] = {from_version}

        while queue:
            current_v, current_p = queue.popleft()

            for (v_start, v_end), migrate_func in type_migrations.items():
                if v_start == current_v and v_end not in visited:
                    new_p = migrate_func(current_p)
                    if v_end == to_version:
                        return new_p

                    visited.add(v_end)
                    queue.append((v_end, new_p))

        return payload

    @classmethod
    def from_payload(cls, payload: dict[str, Any]) -> Self:
        """Create schema instance from raw payload dictionary."""
        return cls.model_validate(payload)

    @classmethod
    def resolve(cls, event_type: str, version: str | None) -> type["BaseEventSchema"]:
        """Resolve the best matching schema class based on event type and version prefix.

        Raises:
            SchemaResolutionError: if no schemas are registered for the event
                type, the version is unknown, or no prefix match exists.
        """
        type_map = cls._registry.get(event_type)
        if not type_map:
            raise SchemaResolutionError(f"No schemas registered for event type: {event_type}")

        if version is None:
            raise SchemaResolutionError(f"No schema_version provided for event type: {event_type}")

        # Longest Prefix Match for SemVer (e.g., for "1.2.3" try "1.2.3", then "1.2", then "1")
        parts = version.split(".")
        prefixes = [".".join(parts[:i]) for i in range(len(parts), 0, -1)]

        for prefix in prefixes:
            if prefix in type_map:
                return type_map[prefix]

        raise SchemaResolutionError(f"Unsupported version {version} for event type {event_type}")

    def to_payload(self) -> dict[str, object]:
        """Convert to dict suitable for OutboxEvent.payload."""
        return self.model_dump(mode="json")

@dataclass(frozen=True, slots=True)
class BatchProcessingResult[T: BaseEvent]:
    """Results of batch event processing."""

    processed_event_ids: list[UUID] = field(default_factory=list)
    failed_counted: list[EventFailureUpdate] = field(default_factory=list)
    failed_noncounted: list[EventFailureUpdate] = field(default_factory=list)
    remaining_event_ids: set[UUID] = field(default_factory=set)
    commit_failed: bool = False

class CommitOffsetPolicy(StrEnum):
    """When to commit (ACK) the broker message in AT_LEAST_ONCE mode."""

    ON_PERSIST = "on_persist"
    ON_SUCCESS = "on_success"

@dataclass(frozen=True, slots=True)
class ConsumedMessage:
    """Normalized consumed message passed to Inbox runner."""

    message_id: str
    source: str
    event_type: str
    payload: dict[str, JsonValue]
    headers: dict[str, str] | None = None
    trace_id: str | None = None
    correlation_id: str | None = None
    causation_id: str | None = None
    schema_version: str | None = None
    ack_handle: AckHandle | None = None
    raw_message: object = None

@dataclass(frozen=True, slots=True)
class EnvelopeData:
    """Unwrapped message payload and extracted metadata from an envelope."""

    payload: dict[str, JsonValue]
    trace_id: str | None = None
    correlation_id: str | None = None
    causation_id: str | None = None
    schema_version: str | None = None

class EnvelopeEventConverter:
    """Full envelope: schema_version, event_type, aggregate info, payload, tracing."""

    def __init__(self, default_schema_version: str = "1.0.0") -> None:
        self._default_schema_version = default_schema_version

    def convert(self, event: OutboxEvent) -> dict[str, JsonValue]:
        result: dict[str, JsonValue] = {
            "schema_version": event.schema_version or self._default_schema_version,
            "event_type": event.event_type,
            "aggregate_type": event.aggregate_type,
            "aggregate_id": str(event.aggregate_id),
            "payload": event.payload,
            "timestamp": event.created_at.isoformat(),
        }
        if event.trace_id is not None:
            result["trace_id"] = event.trace_id
        if event.correlation_id is not None:
            result["correlation_id"] = event.correlation_id
        if event.causation_id is not None:
            result["causation_id"] = event.causation_id
        return result

class EnvelopeParser(Protocol):
    """Protocol for unwrapping nested payloads and extracting metadata."""

    def parse(self, raw_payload: dict[str, JsonValue], headers: dict[str, str] | None) -> EnvelopeData:
        """Unwrap payload and extract metadata."""
        ...

class EventAlreadyLockedError(OmniBoxError):
    """Raised when an event is already locked by another worker."""

    def __init__(self, event_id: UUID, locked_by: str | None) -> None:
        self.event_id = event_id
        self.locked_by = locked_by
        super().__init__(f"Event {event_id} is already locked by {locked_by}")

class EventBatchProcessor[T: BaseEvent]:
    """Unified batch processor powered by pipeline architecture."""

    def __init__(
        self,
        repo: EventRepository[T],
        pipeline: ProcessingPipeline[T],
        fetch_strategy: FetchStrategy[T],
        commit_strategy: CommitStrategy[T],
        *,
        job_name: str = "event_processor",
        metrics: ProcessingMetrics | None = None,
    ) -> None:
        self._repo = repo
        self._pipeline = pipeline
        self._fetch_strategy = fetch_strategy
        self._commit_strategy = commit_strategy
        self._job_name = job_name
        self._metrics = metrics

    async def process_batch(
        self,
        worker_id: str,
        batch_size: int,
        shutdown_requested_func: Callable[[], bool] | None = None,
        **fetch_filters: Unpack[FetchFilters],
    ) -> BatchProcessingResult:
        """Fetch and process a batch of events using the configured pipeline."""
        try:
            # 1. Fetch
            events = await self._fetch_strategy.fetch(
                self._repo,
                batch_size=batch_size,
                worker_id=worker_id,
                **fetch_filters,
            )

            if not events:
                return BatchProcessingResult()

            # 2. Process
            context = ProcessingContext[T](
                repo=self._repo,
                worker_id=worker_id,
                metrics=self._metrics,
            )

            await self._pipeline.process_batch(events, context)

            # 3. Commit
            success = await self._commit_strategy.commit(context)
            if not success:
                logger.error(
                    "Failed to commit batch processing results",
                    worker_id=worker_id,
                    job_name=self._job_name,
                )

            # 4. Result
            return BatchProcessingResult(
                processed_event_ids=context.completed_ids,
                failed_counted=context.failed_counted,
                failed_noncounted=context.failed_noncounted,
                remaining_event_ids=context.skipped_ids,
                commit_failed=not success,
            )
        except (Exception, asyncio.CancelledError) as e:
            if not isinstance(e, asyncio.CancelledError):
                logger.exception(
                    f"{self._job_name} batch failed due to unexpected error",
                    error=str(e),
                    worker_id=worker_id,
                    error_type=type(e).__name__,
                )
            raise

class EventConcurrentUpdateError(OmniBoxError):
    """Raised when a repository update affects fewer rows than expected.

    Indicates that some events might have been modified by another worker,
    unlocked by an admin, or already reached the target state.
    """

    def __init__(
        self,
        expected: int,
        actual: int,
        message: str | None = None,
        missing_ids: list[UUID] | None = None,
    ) -> None:
        self.expected = expected
        self.actual = actual
        self.missing_ids = missing_ids or []

        msg = message or f"Concurrent update detected: expected {expected} rows, but updated {actual}"
        if self.missing_ids:
            # Show up to first 10 missing IDs for diagnostics
            ids_str = ", ".join(str(i) for i in self.missing_ids[:10])
            if len(self.missing_ids) > 10:
                ids_str += ", ..."
            msg += f". Missing IDs: [{ids_str}]"

        super().__init__(msg)

class EventConsumer(ABC):
    """Abstract event consumer."""

    @abstractmethod
    async def start(self) -> None:
        """Start consumer lifecycle resources."""
        ...

    @abstractmethod
    async def stop(self) -> None:
        """Stop consumer lifecycle resources."""
        ...

    @abstractmethod
    async def getone(self) -> ConsumedMessage:
        """Fetch one message from broker."""
        ...