reliar-store-postgres 0.7.0

//! [`PostgresOutboxStore`]'s [`OutboxEnqueue`] implementation.

use bytes::Bytes;
use reliar_core::{ContentType, Message, MessageId, Serializer};
use reliar_outbox::OutboxEnqueue;
use sqlx::{Postgres, Transaction};
use tracing::Instrument as _;

use super::error::{EnqueueError, map_enqueue_error};
use crate::settings::PostgresOutboxSettings;

use super::PostgresOutboxStore;
use crate::connection::schema::{restore_search_path, set_search_path};

/// Writes `payload` under `content_type`, running the schema's own `search_path` handling first
/// (set it, when configured; restore only on success). A free function, not a method: it never
/// touches `self.serializer` — the caller, [`OutboxEnqueue::enqueue_envelope`], has already
/// serialized `envelope.body` into `payload` by the time it calls here.
async fn insert_enqueued<T>(
    tx: &mut Transaction<'_, Postgres>,
    settings: &PostgresOutboxSettings,
    envelope: &reliar_core::Envelope<T>,
    payload: &Bytes,
    content_type: &ContentType,
) -> Result<(), sqlx::Error> {
    let restore = if settings.enqueue_sets_search_path {
        Some(set_search_path(tx, &settings.schema).await?)
    } else {
        None
    };

    let result = insert_row(tx, envelope, payload, content_type).await;

    // Only restore on success: a failed INSERT already aborts the transaction (25P02), so
    // issuing another statement on it would mask the real error behind "current transaction is
    // aborted" instead. The transaction-local scope makes skipping the restore safe — the
    // caller's own rollback/abandonment is what actually undoes it.
    if result.is_ok()
        && let Some(previous) = restore
    {
        restore_search_path(tx, &previous).await?;
    }

    result
}

/// Generic over the envelope body `T` — `T: Message` is never needed here, only
/// `envelope.message_type` (the promoted `message_type`/`message_version` columns), which every
/// `Envelope<T>` carries regardless of `T`.
async fn insert_row<T>(
    tx: &mut Transaction<'_, Postgres>,
    envelope: &reliar_core::Envelope<T>,
    payload: &Bytes,
    content_type: &ContentType,
) -> Result<(), sqlx::Error> {
    let corr = &envelope.metadata.correlation;
    let sent_at_ms = envelope
        .metadata
        .delivery
        .sent_at
        .map(crate::records::encode_epoch_millis);
    let rest = crate::records::MetadataRest {
        trace: crate::records::TraceRest {
            traceparent: envelope.metadata.trace.traceparent.clone(),
            tracestate: envelope.metadata.trace.tracestate.clone(),
        },
        routing: crate::records::RoutingRest {
            source: envelope
                .metadata
                .routing
                .source
                .as_ref()
                .map(|v| v.as_str().to_owned()),
            destination: envelope
                .metadata
                .routing
                .destination
                .as_ref()
                .map(|v| v.as_str().to_owned()),
            reply_to: envelope
                .metadata
                .routing
                .reply_to
                .as_ref()
                .map(|v| v.as_str().to_owned()),
        },
        delivery: crate::records::DeliveryRest {
            sent_at_ms,
            deduplication_id: envelope.metadata.delivery.deduplication_id.clone(),
        },
    };
    // An empty remainder is written as SQL NULL, not '{}', so pending rows stay small.
    let metadata_json = if rest.trace.traceparent.is_none()
        && rest.trace.tracestate.is_none()
        && rest.routing.source.is_none()
        && rest.routing.destination.is_none()
        && rest.routing.reply_to.is_none()
        && rest.delivery.sent_at_ms.is_none()
        && rest.delivery.deduplication_id.is_none()
    {
        None
    } else {
        // `MetadataRest`'s fields are now all plain owned `String`/`i64`/`Option` values (no
        // RFC3339 formatting) — `serde_json::to_value` is total over this
        // shape. The fallback is unreachable in practice; kept non-panicking rather than
        // `.expect()`'d away, since a panic on the enqueue path is never acceptable. `.ok()` rather than a
        // `Value::Null` fallback: on the unreachable error branch this writes SQL `NULL` — the
        // same "no remainder" shape as the empty-check above — rather than a JSON `null` a reader
        // would then have to treat as yet another poison case.
        serde_json::to_value(&rest).ok()
    };

    let headers_json = envelope.headers().filter(|h| !h.is_empty()).map(|h| {
        let map: serde_json::Map<String, serde_json::Value> = h
            .iter()
            .map(|(k, v)| (k.to_owned(), serde_json::Value::String(v.to_owned())))
            .collect();

        serde_json::Value::Object(map)
    });

    // Omits `id`: the surrogate row identity fires from `DEFAULT uuidv7()` (ADR 0044 §1) — no
    // caller needs it before the row is read back, so nothing here mints or binds one.
    sqlx::query!(
        r#"INSERT INTO outbox (
             message_id, message_type, message_version,
             correlation_id, conversation_id, causation_id, request_id,
             content_type, payload, tenant_id, expires_at, ordering_key,
             metadata, headers, available_at
           ) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14, now())"#,
        envelope.id.as_uuid(),
        envelope.message_type.name(),
        i32::from(envelope.message_type.version()),
        corr.correlation_id
            .as_ref()
            .map(reliar_core::CorrelationId::as_str),
        corr.conversation_id.as_uuid(),
        corr.causation_id.map(|id| id.as_uuid()),
        corr.request_id.map(|id| id.as_uuid()),
        content_type.as_str(),
        &payload[..],
        envelope.metadata.tenant_id.as_deref(),
        envelope.metadata.delivery.expires_at,
        // No writer sets `ordering_key` in this release — `Ordering::PerKey` is a configuration
        // error — but the column and its bind stay so a later `PerKey` writer needs no
        // migration. The statement text is unchanged from when this bound `options.ordering_key`,
        // so `.sqlx/` needs no regeneration.
        None::<&str>,
        metadata_json,
        headers_json,
    )
    .execute(&mut **tx)
    .await?;

    Ok(())
}

/// [`PostgresOutboxStore`]'s [`OutboxEnqueue`] implementation: reuses `insert_enqueued`'s
/// `search_path` handling and its `insert_row` helper. Implements only
/// [`OutboxEnqueue::enqueue_envelope`] — the provided `enqueue` (bare `T: Message`) calls back
/// into it, so the `reliar.outbox.enqueue` span fires exactly once per row for either spelling
/// (ADR 0037 amendment A).
///
/// **`Ser: 'static`** — needed because the method reaches `self.serializer` (held as `Arc<Ser>`)
/// across the `.await` in `insert_enqueued`; without it the future fails to type-check (a
/// borrowed type must outlive the generic parameters it references).
/// Every concrete serializer (`JsonSerializer` or any owned one) is `'static`, so no real host is
/// excluded.
///
/// **A single lifetime, `'c`, quantified by the impl.** With `&mut Tx` in the trait's own method
/// signature the reborrow lifetime is quantified by the method itself, so the higher-ranked
/// "implementation is not general enough" trap an earlier `OutboxEnqueueIn<&'a mut
/// Transaction<'c, _>>` shape had — where an explicit, implied-looking `where 'c: 'a` bound broke
/// every `tokio::spawn`/Axum call site — cannot arise here; there is no second lifetime to
/// accidentally bound. That regression guard lives as
/// `outbox_enqueue::enqueue_is_send_through_tokio_spawn` in this crate's Postgres suite.
///
/// Renamed from `OutboxStaging`/`stage` in 0.4.0; the store's own typed `enqueue`/`enqueue_with`
/// were folded into this impl since no inherent method may share a name with a trait method — a
/// caller now needs `use reliar_outbox::OutboxEnqueue;` in scope to call
/// `store.enqueue(..)`/`store.enqueue_envelope(..)`. The trait's serialized twin,
/// `enqueue_serialized`, was cut before shipping.
impl<'c, Ser> OutboxEnqueue<Transaction<'c, Postgres>> for PostgresOutboxStore<Ser>
where
    Ser: Serializer + Send + Sync + 'static,
{
    type Error = EnqueueError<Ser::Error>;

    /// Serializes `envelope.body` with this store's configured `Serializer` and writes the
    /// serializer's own `content_type`. Plain `INSERT`, **no `ON CONFLICT`**: a reused
    /// `MessageId` aborts the caller's transaction rather than silently losing a message.
    ///
    /// # Errors
    ///
    /// [`EnqueueError::Serialize`] if the configured `Serializer` rejects the body,
    /// [`EnqueueError::Duplicate`] for a reused [`MessageId`] (`ix_outbox_message_id` violation,
    /// ADR 0044 §1), or [`EnqueueError::Database`] for any other `sqlx` failure.
    ///
    /// [`EnqueueError::Duplicate`]/[`EnqueueError::Database`] leave `tx` aborted: the failed
    /// `INSERT` puts the PostgreSQL transaction in the aborted state, so PostgreSQL rejects every
    /// subsequent statement on it, and every earlier write in that transaction is rolled back at
    /// commit. [`EnqueueError::Serialize`] is returned before any statement runs, so `tx` is
    /// untouched and stays usable.
    // Block form, not `async fn` (conventions §3(b)): the trait bounds neither `T: Send` nor
    // `Tx: Send`, and an `async fn`'s parameters live in its own generator's *unstarted* state
    // regardless of when the body consumes them, so it would need both. Serializing and dropping
    // the typed body (`map_body(|_| ())`) happen synchronously, before any future is constructed,
    // so the `async` block below is built only after `T` is gone — its captured state (`tx`,
    // `envelope: Envelope<()>`, `payload`) is `Send` independently of `T`.
    fn enqueue_envelope<T: Message + Sync>(
        &self,
        tx: &mut Transaction<'c, Postgres>,
        typed_envelope: reliar_core::Envelope<T>,
    ) -> impl Future<Output = Result<MessageId, Self::Error>> + Send {
        let span = tracing::debug_span!(
            "reliar.outbox.enqueue",
            message.id = %typed_envelope.id,
            message.type = %typed_envelope.message_type,
        );
        // Entered, not `.instrument()`ed: serialization is synchronous and must run before the
        // async block below exists, but a serializer's own `tracing` events still belong inside
        // this span. The guard is dropped before the block is built.
        let payload = {
            let _guard = span.enter();

            self.serializer
                .serialize(&typed_envelope.body)
                .map_err(|source| EnqueueError::Serialize { source })
        };
        // `insert_enqueued` never reads `envelope.body` anyway.
        let envelope = typed_envelope.map_body(|_| ());

        async move {
            let payload = payload?;

            insert_enqueued(tx, &self.settings, &envelope, &payload, self.content_type())
                .await
                .map_err(|source| map_enqueue_error(envelope.id, source))?;

            Ok(envelope.id)
        }
        .instrument(span)
    }
}