reliar-outbox 0.9.0

//! The enqueue capability an application calls directly, in its own transaction (ADR 0036
//! amendment B): there is no facade type between the caller and the store.

use reliar_core::{Classify, Envelope, Message, MessageId};

/// Enqueuing a typed message in the caller's own transaction. A provider (`PostgresOutboxStore`)
/// implements this directly, alongside [`crate::OutboxStore`]; there is no separate handle type
/// to construct.
///
/// `Tx` is the provider's transaction type: `sqlx::Transaction<'_, Postgres>` for
/// `reliar-store-postgres`. It is a **type parameter** precisely so this crate names no storage
/// type, and one implementor may support several.
///
/// Deliberately **not** a method on [`crate::OutboxStore`]: enqueuing takes a transaction handle
/// the claim side never sees, `OutboxStore` is already published, and a GAT `type Tx<'a>` would
/// have to spell `&'a mut Transaction<'c, _>` and reintroduce an invariance problem.
///
/// **Two methods, one call each** (ADR 0037 amendment A): a required
/// [`Self::enqueue_envelope`] that every implementor writes, and a provided [`Self::enqueue`] —
/// the fire-and-forget spelling — built on top of it. An earlier shape tried an `impl
/// Into<Envelope<T>>` parameter so one method covered both a bare `T: Message` and an
/// already-built `Envelope<T>`; that was rejected in favor of a second, explicit method instead
/// — no inference trick, no `Envelope<T>: !Message` invariant to guard.
///
/// **One serialized twin, deliberately absent** (ADR 0036 amendment B.10): a
/// `enqueue_serialized(&mut tx, &SerializedEnvelope)` existed briefly and was cut for having no
/// production caller; re-adding it is additive.
///
/// Renamed from `OutboxStaging` in 0.4.0; `stage` became `enqueue`, and the facade
/// `OutboxPublisher` that briefly wrapped it (0.4.0, never released) was withdrawn before
/// shipping in favor of calling this trait directly.
///
/// A provider implements the trait, then a caller enqueues in its own transaction. This crate
/// constructs no store of its own (ADR 0043), so the shape below is a compiled generic call,
/// never invoked — `reliar-store-postgres`'s own rustdoc has the runnable version over
/// `PgPool::connect`.
///
/// ```
/// use reliar_core::Message;
/// use reliar_outbox::OutboxEnqueue;
///
/// # #[derive(serde::Serialize, serde::Deserialize)]
/// # struct OrderCreated;
/// # impl Message for OrderCreated {
/// #     const TYPE: &'static str = "orders.created";
/// #     const VERSION: u16 = 1;
/// # }
/// async fn enqueue_order<S, Tx>(store: &S, tx: &mut Tx) -> Result<(), S::Error>
/// where
///     S: OutboxEnqueue<Tx>,
/// {
///     store.enqueue(tx, OrderCreated).await?;
///     Ok(())
/// }
/// ```
pub trait OutboxEnqueue<Tx>: Send + Sync {
    /// What enqueuing fails with.
    type Error: std::error::Error + Send + Sync + 'static + Classify;

    /// Serializes `envelope`'s body with **this implementor's own** configured `Serializer` and
    /// enqueues it in `tx`. Writes the serializer's own `content_type`.
    ///
    /// The **propagating** spelling: use it when an id must carry over from an inbound request
    /// — conversation, correlation, causation, tenant, trace, headers — via
    /// [`Envelope::builder`](reliar_core::Envelope::builder).
    ///
    /// ```
    /// use reliar_core::{ConversationId, Envelope, Message};
    /// use reliar_outbox::OutboxEnqueue;
    ///
    /// # #[derive(serde::Serialize, serde::Deserialize)]
    /// # struct OrderCreated;
    /// # impl Message for OrderCreated {
    /// #     const TYPE: &'static str = "orders.created";
    /// #     const VERSION: u16 = 1;
    /// # }
    /// async fn propagate_conversation<S, Tx>(
    ///     store: &S,
    ///     tx: &mut Tx,
    ///     conversation: ConversationId,
    /// ) -> Result<(), S::Error>
    /// where
    ///     S: OutboxEnqueue<Tx>,
    /// {
    ///     let envelope = Envelope::builder(OrderCreated).conversation(conversation).build();
    ///     store.enqueue_envelope(tx, envelope).await?;
    ///     Ok(())
    /// }
    /// ```
    ///
    /// Returns the id written, for the caller's own use — e.g. as a *next* message's
    /// `causation_id` in the same transaction. The envelope already carries its `id`; this is not
    /// how a caller learns it, only a convenience.
    ///
    /// **Implementors:** the trait bounds neither `T` nor `Tx` on `Send`, so a plain `async fn`
    /// that carries `envelope: Envelope<T>` (or `tx`) across an `.await` will not satisfy this
    /// method's `+ Send` return bound. Serialize (or otherwise consume) `T` synchronously, before
    /// the async block is built — see `PostgresOutboxStore::enqueue_envelope`'s `//` comment in
    /// `reliar-store-postgres` for the reference shape and the full argument.
    ///
    /// The implementation SHALL issue no network I/O other than the statement itself, and SHALL
    /// NOT commit, roll back or otherwise consume `tx` — the caller owns it.
    ///
    /// # Errors
    ///
    /// Provider-defined. An `Err` **MAY** leave `tx` unusable, and whether it does is the
    /// provider's contract — every implementor documents which. The portable rule a caller can
    /// rely on is therefore: treat any enqueue error as *abort this transaction* — issue no
    /// further statement on `tx`, roll it back, and consider every earlier write in it lost. With
    /// `reliar-store-postgres` the transaction **is** aborted: PostgreSQL rejects every
    /// subsequent statement on it, so no earlier write in that transaction can still be committed.
    fn enqueue_envelope<T: Message + Sync>(
        &self,
        tx: &mut Tx,
        envelope: Envelope<T>,
    ) -> impl Future<Output = Result<MessageId, Self::Error>> + Send;

    /// The **fire-and-forget** spelling: builds `body` into an envelope with default metadata
    /// and a freshly rooted conversation — exactly `Envelope::builder(body).build()` — then
    /// enqueues it via [`Self::enqueue_envelope`].
    ///
    /// ```
    /// use reliar_core::Message;
    /// use reliar_outbox::OutboxEnqueue;
    ///
    /// # #[derive(serde::Serialize, serde::Deserialize)]
    /// # struct OrderCreated {
    /// #     order_id: u64,
    /// # }
    /// # impl Message for OrderCreated {
    /// #     const TYPE: &'static str = "orders.created";
    /// #     const VERSION: u16 = 1;
    /// # }
    /// async fn enqueue_order<S, Tx>(store: &S, tx: &mut Tx, order_id: u64) -> Result<(), S::Error>
    /// where
    ///     S: OutboxEnqueue<Tx>,
    /// {
    ///     store.enqueue(tx, OrderCreated { order_id }).await?;
    ///     Ok(())
    /// }
    /// ```
    ///
    /// Use [`Self::enqueue_envelope`] instead when an id must propagate from an inbound request;
    /// this spelling never has anything to propagate from.
    ///
    /// Implementors **SHALL NOT** override this method — it is a fixed, provided spelling of
    /// [`Self::enqueue_envelope`], not an extension point. `Envelope::builder(body).build()` mints
    /// the envelope and its id eagerly, at call time, not on the returned future's first poll.
    ///
    /// # Errors
    ///
    /// Same as [`Self::enqueue_envelope`].
    // Direct delegation, not `async fn` and no `async move` block either — returns
    // `enqueue_envelope`'s own future unchanged. See `PostgresOutboxStore::enqueue_envelope`'s
    // comment in `reliar-store-postgres/src/outbox/enqueue.rs` for the full `Send`/capture argument.
    fn enqueue<T: Message + Sync>(
        &self,
        tx: &mut Tx,
        body: T,
    ) -> impl Future<Output = Result<MessageId, Self::Error>> + Send {
        self.enqueue_envelope(tx, Envelope::builder(body).build())
    }
}