platform_core/

outbox.rs

use crate::db::{DbPool, DbTransaction};
use crate::error::{AppError, AppResult, ErrorCode};
use crate::events::EventEnvelope;
use crate::execution_logs::{
    ExecutionLogRecord, ExecutionLogSeverity, insert_execution_log_projection,
};
use crate::{RuntimeSpanAttributes, record_runtime_span_attributes, trace_context_from_headers};
use async_trait::async_trait;
use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
use serde_json::{Value, json};
use std::collections::BTreeMap;
use std::fmt::Debug;
use std::sync::Arc;
use tracing::Instrument;

const OUTBOX_RETRY_DELAY_SECONDS: i64 = 5;
const STALE_PROCESSING_LOCK_SECONDS: i64 = 300;

#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum OutboxStatus {
    Pending,
    Processing,
    Published,
    Failed,
    Dead,
}

impl OutboxStatus {
    fn as_str(self) -> &'static str {
        match self {
            Self::Pending => "pending",
            Self::Processing => "processing",
            Self::Published => "published",
            Self::Failed => "failed",
            Self::Dead => "dead",
        }
    }
}

#[derive(Debug, Clone, Deserialize, Serialize)]
pub struct OutboxEvent {
    pub id: String,
    pub event_name: String,
    pub event_version: u16,
    pub source_module: String,
    pub aggregate_type: String,
    pub aggregate_id: String,
    pub correlation_id: String,
    pub causation_id: Option<String>,
    pub occurred_at: DateTime<Utc>,
    pub payload: Value,
    pub headers: Value,
}

impl OutboxEvent {
    pub fn from_envelope(aggregate_type: impl Into<String>, event: &EventEnvelope) -> Self {
        Self {
            id: event.event_id.clone(),
            event_name: event.event_name.clone(),
            event_version: event.event_version,
            source_module: event.source_module.clone(),
            aggregate_type: aggregate_type.into(),
            aggregate_id: event.subject.clone(),
            correlation_id: event.correlation_id.0.clone(),
            causation_id: event.causation_id.clone(),
            occurred_at: event.occurred_at,
            payload: event.payload.clone(),
            headers: json!({
                "actor": event.actor,
                "tenant_id": event.tenant_id,
                "schema_ref": event.schema_ref,
                "trace": event.trace,
            }),
        }
    }
}

#[derive(Debug, Clone, Deserialize, Serialize)]
pub struct ClaimedOutboxEvent {
    pub id: String,
    pub event_name: String,
    pub event_version: u16,
    pub source_module: String,
    pub aggregate_type: String,
    pub aggregate_id: String,
    pub correlation_id: String,
    pub causation_id: Option<String>,
    pub occurred_at: DateTime<Utc>,
    pub payload: Value,
    pub headers: Value,
    pub attempts: i32,
    pub max_attempts: i32,
}

#[derive(Debug, Clone, Default)]
pub struct OutboxPublisher;

impl OutboxPublisher {
    pub async fn publish_in_tx(
        &self,
        tx: &mut DbTransaction<'_>,
        event: &OutboxEvent,
    ) -> AppResult<()> {
        let span = tracing::info_span!(
            "outbox_publish",
            lenso.correlation_id = tracing::field::Empty,
            lenso.story_id = tracing::field::Empty,
            lenso.outbox_event_id = tracing::field::Empty,
            lenso.execution.kind = tracing::field::Empty,
            lenso.execution.name = tracing::field::Empty,
        );
        record_runtime_span_attributes(
            &span,
            &RuntimeSpanAttributes::outbox(
                event.correlation_id.clone(),
                event.id.clone(),
                event.event_name.clone(),
            ),
        );

        async {
            sqlx::query(
                r#"
            insert into platform.outbox (
                id,
                event_name,
                event_version,
                source_module,
                aggregate_type,
                aggregate_id,
                correlation_id,
                causation_id,
                occurred_at,
                payload,
                headers
            )
            values ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)
            "#,
            )
            .bind(&event.id)
            .bind(&event.event_name)
            .bind(i32::from(event.event_version))
            .bind(&event.source_module)
            .bind(&event.aggregate_type)
            .bind(&event.aggregate_id)
            .bind(&event.correlation_id)
            .bind(&event.causation_id)
            .bind(event.occurred_at)
            .bind(&event.payload)
            .bind(&event.headers)
            .execute(&mut **tx)
            .await
            .map(|_| ())
            .map_err(map_outbox_error)
        }
        .instrument(span)
        .await
    }

    pub async fn pending_count(&self, pool: &DbPool) -> AppResult<i64> {
        sqlx::query_scalar(
            r#"
            select count(*)
            from platform.outbox
            where status = 'pending'
            "#,
        )
        .fetch_one(pool)
        .await
        .map_err(map_outbox_error)
    }
}

#[async_trait]
pub trait EventDispatcher: Debug + Send + Sync {
    async fn dispatch(&self, event: &ClaimedOutboxEvent) -> AppResult<()>;
}

#[async_trait]
pub trait EventHandler: Debug + Send + Sync {
    fn event_name(&self) -> &str;
    async fn handle(&self, event: &ClaimedOutboxEvent) -> AppResult<()>;
}

#[derive(Debug, Clone, Default)]
pub struct EventHandlerRegistry {
    handlers: BTreeMap<String, Vec<Arc<dyn EventHandler>>>,
}

impl EventHandlerRegistry {
    pub fn new() -> Self {
        Self::default()
    }

    pub fn register(&mut self, handler: Arc<dyn EventHandler>) {
        self.handlers
            .entry(handler.event_name().to_owned())
            .or_default()
            .push(handler);
    }

    pub fn register_all(&mut self, handlers: impl IntoIterator<Item = Arc<dyn EventHandler>>) {
        for handler in handlers {
            self.register(handler);
        }
    }

    pub fn handler_count(&self, event_name: &str) -> usize {
        self.handlers.get(event_name).map_or(0, Vec::len)
    }
}

#[async_trait]
impl EventDispatcher for EventHandlerRegistry {
    async fn dispatch(&self, event: &ClaimedOutboxEvent) -> AppResult<()> {
        let Some(handlers) = self.handlers.get(event.event_name.as_str()) else {
            tracing::debug!(
                event_name = %event.event_name,
                outbox_id = %event.id,
                "no in-process event handlers registered"
            );
            return Ok(());
        };

        for handler in handlers {
            handler.handle(event).await?;
        }

        Ok(())
    }
}

#[derive(Debug, Default)]
pub struct LoggingEventDispatcher;

#[async_trait]
impl EventDispatcher for LoggingEventDispatcher {
    async fn dispatch(&self, event: &ClaimedOutboxEvent) -> AppResult<()> {
        tracing::info!(
            outbox_id = %event.id,
            event_name = %event.event_name,
            event_version = event.event_version,
            aggregate_id = %event.aggregate_id,
            correlation_id = %event.correlation_id,
            "outbox event dispatched"
        );
        Ok(())
    }
}

#[derive(Debug, Clone)]
pub struct OutboxRelay {
    pool: DbPool,
    worker_id: String,
}

impl OutboxRelay {
    pub fn new(pool: DbPool, worker_id: impl Into<String>) -> Self {
        Self {
            pool,
            worker_id: worker_id.into(),
        }
    }

    pub async fn claim_batch(&self, batch_size: i64) -> AppResult<Vec<ClaimedOutboxEvent>> {
        let span = tracing::info_span!(
            "outbox_claim_batch",
            worker_id = %self.worker_id,
            lenso.execution.kind = "outbox_claim",
            lenso.execution.name = "outbox.claim_batch",
        );

        async {
            let events = sqlx::query_as::<_, OutboxRow>(
                r#"
            with claimed as (
                select id
                from platform.outbox
                where (
                    status in ('pending', 'failed')
                    and available_at <= now()
                )
                or (
                    status = 'processing'
                    and locked_at <= now() - ($1::double precision * interval '1 second')
                )
                order by available_at asc, created_at asc
                limit $2
                for update skip locked
            )
            update platform.outbox outbox
            set status = 'processing',
                locked_at = now(),
                locked_by = $3,
                last_error = null
            from claimed
            where outbox.id = claimed.id
            returning
                outbox.id,
                outbox.event_name,
                outbox.event_version,
                outbox.source_module,
                outbox.aggregate_type,
                outbox.aggregate_id,
                outbox.correlation_id,
                outbox.causation_id,
                outbox.occurred_at,
                outbox.payload,
                outbox.headers,
                outbox.attempts,
                outbox.max_attempts
            "#,
            )
            .bind(stale_processing_lock_seconds())
            .bind(batch_size)
            .bind(&self.worker_id)
            .fetch_all(&self.pool)
            .await
            .map(|rows| rows.into_iter().map(Into::into).collect())
            .map_err(map_outbox_error)?;

            for event in &events {
                self.record_outbox_execution_log(
                    event,
                    ExecutionLogSeverity::Info,
                    "Outbox event claimed",
                    json!({
                        "attempt": event.attempts + 1,
                        "max_attempts": event.max_attempts,
                        "worker_id": self.worker_id,
                    }),
                )
                .await;
            }

            Ok(events)
        }
        .instrument(span)
        .await
    }

    pub async fn relay_once(
        &self,
        dispatcher: &dyn EventDispatcher,
        batch_size: i64,
    ) -> AppResult<usize> {
        let span = tracing::info_span!(
            "outbox_relay_once",
            worker_id = %self.worker_id,
            lenso.execution.kind = "outbox_relay",
            lenso.execution.name = "outbox.relay_once",
        );

        async {
            let events = self.claim_batch(batch_size).await?;
            let count = events.len();

            for event in events {
                let event_span = tracing::info_span!(
                    "outbox_dispatch",
                    lenso.correlation_id = tracing::field::Empty,
                    lenso.story_id = tracing::field::Empty,
                    lenso.outbox_event_id = tracing::field::Empty,
                    lenso.execution.kind = tracing::field::Empty,
                    lenso.execution.name = tracing::field::Empty,
                );
                record_runtime_span_attributes(
                    &event_span,
                    &RuntimeSpanAttributes::outbox(
                        event.correlation_id.clone(),
                        event.id.clone(),
                        event.event_name.clone(),
                    ),
                );

                async {
                    self.record_outbox_execution_log(
                        &event,
                        ExecutionLogSeverity::Info,
                        "Outbox event dispatch started",
                        json!({
                            "event_name": event.event_name,
                            "attempt": event.attempts + 1,
                            "worker_id": self.worker_id,
                        }),
                    )
                    .await;
                    match dispatcher.dispatch(&event).await {
                        Ok(()) => self.mark_published(&event).await?,
                        Err(error) => self.mark_dispatch_failed(&event, &error).await?,
                    }

                    Ok::<(), AppError>(())
                }
                .instrument(event_span)
                .await?;
            }

            Ok(count)
        }
        .instrument(span)
        .await
    }

    pub async fn mark_published(&self, event: &ClaimedOutboxEvent) -> AppResult<()> {
        sqlx::query(
            r#"
            update platform.outbox
            set status = 'published',
                published_at = now(),
                locked_at = null,
                locked_by = null,
                last_error = null
            where id = $1
            "#,
        )
        .bind(&event.id)
        .execute(&self.pool)
        .await
        .map(|_| ())
        .map_err(map_outbox_error)?;

        self.record_outbox_execution_log(
            event,
            ExecutionLogSeverity::Info,
            "Outbox event published",
            json!({
                "event_name": event.event_name,
                "attempt": event.attempts + 1,
                "worker_id": self.worker_id,
            }),
        )
        .await;

        Ok(())
    }

    pub async fn mark_dispatch_failed(
        &self,
        event: &ClaimedOutboxEvent,
        error: &AppError,
    ) -> AppResult<()> {
        let next_attempt = event.attempts + 1;
        let status = if next_attempt >= event.max_attempts {
            OutboxStatus::Dead
        } else if error.retryable {
            OutboxStatus::Failed
        } else {
            OutboxStatus::Dead
        };

        let span = tracing::info_span!(
            "outbox_retry",
            lenso.correlation_id = tracing::field::Empty,
            lenso.story_id = tracing::field::Empty,
            lenso.outbox_event_id = tracing::field::Empty,
            lenso.execution.kind = tracing::field::Empty,
            lenso.execution.name = tracing::field::Empty,
        );
        record_runtime_span_attributes(
            &span,
            &RuntimeSpanAttributes::outbox(
                event.correlation_id.clone(),
                event.id.clone(),
                event.event_name.clone(),
            ),
        );

        async {
            sqlx::query(
                r#"
            update platform.outbox
            set status = $2,
                attempts = attempts + 1,
                available_at = case
                    when $2 = 'failed' then now() + ($4::double precision * interval '1 second')
                    else available_at
                end,
                locked_at = null,
                locked_by = null,
                last_error = $3
            where id = $1
            "#,
            )
            .bind(&event.id)
            .bind(status.as_str())
            .bind(error.public_message.as_str())
            .bind(outbox_retry_delay_seconds())
            .execute(&self.pool)
            .await
            .map(|_| ())
            .map_err(map_outbox_error)?;

            self.record_outbox_execution_log(
                event,
                ExecutionLogSeverity::Error,
                if status == OutboxStatus::Dead {
                    "Outbox event marked dead"
                } else {
                    "Outbox event failed"
                },
                json!({
                    "attempt": next_attempt,
                    "max_attempts": event.max_attempts,
                    "status": status.as_str(),
                    "retryable": error.retryable,
                    "error": error.public_message,
                    "worker_id": self.worker_id,
                }),
            )
            .await;

            Ok(())
        }
        .instrument(span)
        .await
    }

    async fn record_outbox_execution_log(
        &self,
        event: &ClaimedOutboxEvent,
        severity: ExecutionLogSeverity,
        body: &'static str,
        attributes: Value,
    ) {
        emit_outbox_lifecycle_event(event, severity, body, &attributes, Some(&self.worker_id));
        if let Err(error) = insert_execution_log_projection(
            &self.pool,
            outbox_log_record(event, severity, body, attributes),
        )
        .await
        {
            tracing::warn!(
                error = ?error,
                outbox_id = %event.id,
                "failed to write outbox execution log"
            );
        }
    }
}

type OutboxRow = (
    String,
    String,
    i32,
    String,
    String,
    String,
    String,
    Option<String>,
    DateTime<Utc>,
    Value,
    Value,
    i32,
    i32,
);

impl From<OutboxRow> for ClaimedOutboxEvent {
    fn from(row: OutboxRow) -> Self {
        let (
            id,
            event_name,
            event_version,
            source_module,
            aggregate_type,
            aggregate_id,
            correlation_id,
            causation_id,
            occurred_at,
            payload,
            headers,
            attempts,
            max_attempts,
        ) = row;

        Self {
            id,
            event_name,
            event_version: event_version
                .try_into()
                .expect("event_version should fit into u16"),
            source_module,
            aggregate_type,
            aggregate_id,
            correlation_id,
            causation_id,
            occurred_at,
            payload,
            headers,
            attempts,
            max_attempts,
        }
    }
}

fn map_outbox_error(source: sqlx::Error) -> AppError {
    AppError::new(ErrorCode::Internal, "Outbox operation failed").with_source(source)
}

fn outbox_retry_delay_seconds() -> f64 {
    OUTBOX_RETRY_DELAY_SECONDS as f64
}

fn stale_processing_lock_seconds() -> f64 {
    STALE_PROCESSING_LOCK_SECONDS as f64
}

fn emit_outbox_lifecycle_event(
    event: &ClaimedOutboxEvent,
    severity: ExecutionLogSeverity,
    body: &'static str,
    attributes: &Value,
    worker_id: Option<&str>,
) {
    match severity {
        ExecutionLogSeverity::Error => {
            tracing::error!(
                outbox_id = %event.id,
                event_name = %event.event_name,
                correlation_id = %event.correlation_id,
                worker_id = worker_id.unwrap_or(""),
                attributes = %attributes,
                "{body}"
            );
        }
        ExecutionLogSeverity::Warn => {
            tracing::warn!(
                outbox_id = %event.id,
                event_name = %event.event_name,
                correlation_id = %event.correlation_id,
                worker_id = worker_id.unwrap_or(""),
                attributes = %attributes,
                "{body}"
            );
        }
        _ => {
            tracing::info!(
                outbox_id = %event.id,
                event_name = %event.event_name,
                correlation_id = %event.correlation_id,
                worker_id = worker_id.unwrap_or(""),
                attributes = %attributes,
                "{body}"
            );
        }
    }
}

fn outbox_log_record(
    event: &impl OutboxLogSource,
    severity: ExecutionLogSeverity,
    body: impl Into<String>,
    attributes: Value,
) -> ExecutionLogRecord {
    ExecutionLogRecord::from_runtime_attrs(
        RuntimeSpanAttributes::outbox(event.correlation_id(), event.id(), event.execution_name()),
        severity,
        body,
    )
    .with_attributes(attributes)
    .with_trace(trace_context_from_headers(event.headers()))
}

trait OutboxLogSource {
    fn id(&self) -> String;
    fn correlation_id(&self) -> String;
    fn execution_name(&self) -> String;
    fn headers(&self) -> &Value;
}

impl OutboxLogSource for OutboxEvent {
    fn id(&self) -> String {
        self.id.clone()
    }

    fn correlation_id(&self) -> String {
        self.correlation_id.clone()
    }

    fn execution_name(&self) -> String {
        self.event_name.clone()
    }

    fn headers(&self) -> &Value {
        &self.headers
    }
}

impl OutboxLogSource for ClaimedOutboxEvent {
    fn id(&self) -> String {
        self.id.clone()
    }

    fn correlation_id(&self) -> String {
        self.correlation_id.clone()
    }

    fn execution_name(&self) -> String {
        self.event_name.clone()
    }

    fn headers(&self) -> &Value {
        &self.headers
    }
}