reliar-store-postgres 0.9.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::enqueue::{InsertOutboxRowParams, insert_outbox_row};
use super::error::{EnqueueError, map_enqueue_error};

use super::PostgresOutboxStore;

/// 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`. Builds [`InsertOutboxRowParams`] (encoding policy —
/// the `MetadataRest`/`headers` JSON shape and the "empty remainder is SQL NULL" decision are the
/// store layer's, not the concern function's) and calls the one-statement concern function
/// directly on the caller's transaction.
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)
    });

    insert_outbox_row(
        &mut *tx,
        InsertOutboxRowParams {
            message_id: envelope.id.as_uuid(),
            message_type: envelope.message_type.name(),
            message_version: i32::from(envelope.message_type.version()),
            correlation_id: corr
                .correlation_id
                .as_ref()
                .map(reliar_core::CorrelationId::as_str),
            conversation_id: corr.conversation_id.as_uuid(),
            causation_id: corr.causation_id.map(|id| id.as_uuid()),
            request_id: corr.request_id.map(|id| id.as_uuid()),
            content_type: content_type.as_str(),
            payload: &payload[..],
            tenant_id: envelope.metadata.tenant_id.as_deref(),
            expires_at: 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.
            ordering_key: None,
            metadata: metadata_json,
            headers: headers_json,
        },
    )
    .await
}

/// [`PostgresOutboxStore`]'s [`OutboxEnqueue`] implementation: reuses `insert_row`. 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_row`; 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_row` never reads `envelope.body` anyway.
        let envelope = typed_envelope.map_body(|_| ());

        async move {
            let payload = payload?;

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

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