Derive Macro Outbox 

#[derive(Outbox)]
{
    // Attributes available to this derive:
    #[outbox]
    #[field]
    #[sqlx]
}

Describes a service’s outbox table with a struct: implements OutboxTable, the description a record by hand writes, and HeaderRow for a headers field.

The struct is an ordinary sqlx struct: #[outbox(..)] names the table, sqlx’s own attributes name the columns, and #[field(..)] marks the columns the outbox reads. The service writes the record of a published message itself, in outbox::Publish; the registry runs the other events on every OutboxDatabase, from statements it builds from the description when the record type is registered. The derive checks the description against each built-in dialect while the service compiles.

Examples

use ruststream::OutgoingMessage;
use ruststream_sqlx::{Outbox, outbox};
use sqlx::types::Json;
use sqlx::{PgConnection, Postgres};
use std::collections::BTreeMap;

// app.outbox: id BIGSERIAL PRIMARY KEY, name TEXT, payload BYTEA, headers JSONB,
// processed_at TIMESTAMPTZ
#[derive(Outbox, sqlx::FromRow)]
#[outbox(table = "outbox", schema = "app")]
pub struct OrderEvent {
    #[field(id)]
    id: i64,
    #[field(name)]
    name: String,
    #[field(payload)]
    payload: Vec<u8>,
    #[field(headers)]
    headers: Option<Json<BTreeMap<String, String>>>,
    #[field(processed_at)]
    processed_at: Option<chrono::DateTime<chrono::Utc>>,
}

impl outbox::Publish<Postgres> for OrderEvent {
    async fn publish(
        conn: &mut PgConnection,
        msg: &OutgoingMessage<'_>,
    ) -> Result<i64, sqlx::Error> {
        let headers: BTreeMap<String, String> = msg
            .headers()
            .iter()
            .map(|(name, value)| (name.to_owned(), String::from_utf8_lossy(value).into_owned()))
            .collect();
        sqlx::query_scalar(
            "INSERT INTO app.outbox (name, payload, headers) VALUES ($1, $2, $3) RETURNING id",
        )
        .bind(msg.name())
        .bind(msg.payload())
        .bind(Json(headers))
        .fetch_one(conn)
        .await
    }
}

The table

#[outbox(table = "..")] names the table and is required; schema = ".." places it in a schema. Each names one thing and holds no dot.

Roles

#[field(..)] gives a field one role:

• id, required: the record’s identity, which a tracked message carries in OUTBOX_ID_HEADER through its Display and FromStr, and which the default events bind as its own type.
• name, required: the name the record was published under, read through AsRef<str>.
• payload, required: the published bytes, read through AsRef<[u8]>.
• headers: the published headers, a HeaderColumn type.
• processed_at: the mark of a processed record, written from the database’s clock, so any time type fits. Without it a processed record is deleted.

Every other field is a plain column sqlx reads. Column names come from sqlx’s attributes: rename, rename_all, a raw identifier without r#, and skip for a field without a column; a flatten field makes the statements select *.

Events

The default Fetch reads the record while it is unprocessed, Ack and Discard mark it processed, Retry leaves it as it is and runs no statement, and Recover reads the unprocessed records of one name. #[outbox(custom(fetch, ack, retry, discard, recover))] lists the events the service implements itself instead; publish has no default and is never listed.

Compile errors

A struct the outbox cannot read does not compile, and the error points at the struct, the field or the word that causes it: no id, name or payload, a role played twice, a column named twice, a role on a field without a column, an unknown role or event, a dot in table or schema, an id read through #[sqlx(json)]. Implemented in ruststream-sqlx-macros and used through ruststream-sqlx.