Record

#[derive(Outbox)] describes the outbox table: id, name and payload are required, headers and processed_at optional; the service writes the record.

#[outbox(table = "..")] names the table and is required; schema = ".." places it in a schema. #[field(..)] gives a field one role:

id: the record's identity. A tracked message carries it in the id header through its Display and FromStr, so an integer and a UUID both fit.
name: the name the message was published under.
payload: the published bytes.
headers: the published headers, a HeaderColumn type such as Option<Json<BTreeMap<String, String>>> (feature json). A republish sends them again.
processed_at: the mark of a processed record. The outbox writes it from the database's clock, so any time type fits. A record without it is deleted once processed.

Every other field is a plain column sqlx reads, and sqlx's rename, rename_all and skip name the columns.

The service writes the record itself, in outbox::Publish, because its statement decides how the name, the payload and the headers are laid out. publish runs on a connection of its own and returns the new record's id. The registry builds the other events from the description and the database's built-in dialect when the record type is registered; Statements replaces them with the service's own. The crate implements OutboxRow for every such record, and a service names it only as a bound.

A record described by hand implements OutboxTable with an OutboxSpec chain. OutboxSpec::new takes the table and its three required columns, and type Table lists the typed settings in the order the chain sets them: Headers, ProcessedAt and the events of spec::own. A record with Headers implements HeaderRow for the field that holds the column. Both forms describe one table, and the registry builds the same statements from either.

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, an unknown role or event, a dot in table or schema. A table whose identifier is longer than the database takes panics when the record type is registered.

Example

OrderOutbox keeps every role, and its Publish stores the headers the message carried; the service stamps each order with its tenant on the way out.

Macros
Manual

examples/sqlx_outbox_record.rs
use ruststream_sqlx::{Outbox, outbox};

/// The `headers` column: a JSON object of strings, `NULL` for none.
type Headers = Option<Json<BTreeMap<String, String>>>;

/// One record of `order_outbox`.
#[derive(Outbox, sqlx::FromRow)]
#[outbox(table = "order_outbox")]
struct OrderOutbox {
    #[field(id)]
    id: i64,
    #[field(name)]
    name: String,
    #[field(payload)]
    payload: Vec<u8>,
    #[field(headers)]
    headers: Headers,
    #[field(processed_at)]
    processed_at: Option<DateTime<Utc>>,
}

/// The record keeps what the message carried, so a republish sends the same headers again.
impl outbox::Publish<Sqlite> for OrderOutbox {
    async fn publish(conn: &mut SqliteConnection, msg: &OutgoingMessage<'_>) -> sqlx::Result<i64> {
        sqlx::query_scalar(
            "INSERT INTO order_outbox (name, payload, headers) VALUES (?, ?, ?) RETURNING id",
        )
        .bind(msg.name())
        .bind(msg.payload())
        .bind(Headers::from_headers(msg.headers()))
        .fetch_one(conn)
        .await
    }
}

examples/manual/sqlx_outbox_record.rs
use ruststream_sqlx::outbox::{self, Outbox, OutboxSpec, OutboxTable, TrackedName};

/// The `headers` column: a JSON object of strings, `NULL` for none.
type Headers = Option<Json<BTreeMap<String, String>>>;

/// One record of `order_outbox`: the struct holds what the outbox reads.
#[derive(sqlx::FromRow)]
struct OrderOutbox {
    id: i64,
    name: String,
    payload: Vec<u8>,
    headers: Headers,
}

impl OutboxTable for OrderOutbox {
    type Id = i64;
    type Table = OutboxSpec<(HeadersColumn, ProcessedAt)>;
    const TABLE: Self::Table = OutboxSpec::new(
        "order_outbox",
        Column::new("id"),
        Column::new("name"),
        Column::new("payload"),
    )
    .headers(Column::new("headers"))
    .processed_at(Column::new("processed_at"));

    fn id(&self) -> &i64 {
        &self.id
    }

    fn name(&self) -> &str {
        &self.name
    }

    fn payload(&self) -> &[u8] {
        &self.payload
    }
}

/// The field that holds the headers column.
impl HeaderRow for OrderOutbox {
    type Column = Headers;

    fn headers_mut(&mut self) -> &mut Headers {
        &mut self.headers
    }
}

/// The record keeps what the message carried, so a republish sends the same headers again.
impl outbox::Publish<Sqlite> for OrderOutbox {
    async fn publish(conn: &mut SqliteConnection, msg: &OutgoingMessage<'_>) -> sqlx::Result<i64> {
        sqlx::query_scalar(
            "INSERT INTO order_outbox (name, payload, headers) VALUES (?, ?, ?) RETURNING id",
        )
        .bind(msg.name())
        .bind(msg.payload())
        .bind(Headers::from_headers(msg.headers()))
        .fetch_one(conn)
        .await
    }
}

/// The name `OrderPlaced` is published under.
struct Orders;

impl TrackedName for Orders {
    const NAME: &'static str = "orders";
}

Example​

Example