ruststream-sqlx 0.7.1

//! The headers layout: the queue's mechanics and the service's header fields sit in one struct,
//! the message flattens it, and the delivery's header map is built from the fields on its first
//! read.

use std::pin::pin;

use futures::StreamExt;
use ruststream::{
    Broker, Carries, ConnectedBroker, IncomingMessage, Subscriber, SubscriptionSource,
};
use ruststream_sqlx::__private::Events;
use ruststream_sqlx::InboxRow;
use ruststream_sqlx::prelude::*;
use sqlx::Sqlite;

use super::{broker, count, database};

const SCHEMA: &str = "
CREATE TABLE shipping_jobs (
    job_id       INTEGER PRIMARY KEY AUTOINCREMENT,
    name         TEXT NOT NULL,
    attempt      INTEGER NOT NULL DEFAULT 1,
    locked_until TEXT,
    tenant       TEXT NOT NULL,
    trace        TEXT,
    note         TEXT
);
INSERT INTO shipping_jobs (name, tenant, trace, note) VALUES ('orders', 'acme', NULL, 'fragile');
";

const LEFT: &str = "SELECT count(*) FROM shipping_jobs";

mod derived {
    use chrono::{DateTime, Utc};
    use ruststream_sqlx::{Inbox, InboxHeaders};

    #[derive(Debug, Clone, InboxHeaders, sqlx::FromRow)]
    #[inbox(table = "shipping_jobs")]
    pub(super) struct OrderHeaders {
        #[field(id, generated)]
        job_id: i64,
        #[field(group)]
        name: String,
        #[field(attempt, generated)]
        attempt: i16,
        #[field(locked_until)]
        locked_until: Option<DateTime<Utc>>,
        tenant: String,
        trace: Option<String>,
    }

    #[derive(Debug, Clone, Inbox, sqlx::FromRow)]
    pub(super) struct OrderJob {
        #[field(headers)]
        #[sqlx(flatten)]
        headers: OrderHeaders,
        pub(super) note: Option<String>,
    }
}

mod manual {
    use chrono::{DateTime, Utc};
    use ruststream::HeaderMap;
    use ruststream::runtime::{Input, SoloCarried};
    use ruststream_sqlx::dialect::Column;
    use ruststream_sqlx::spec::{Attempt, HeaderFields as Fields, Lease};
    use ruststream_sqlx::{AttemptRow, HeaderFields, InboxSpec, InboxTable, put_header};

    // The service keeps its header fields in a struct of their own; the crate needs neither
    // struct to know about the other.
    #[derive(Debug, Clone, sqlx::FromRow)]
    pub(super) struct OrderHeaders {
        job_id: i64,
        attempt: i16,
        tenant: String,
        trace: Option<String>,
    }

    #[derive(Debug, Clone, sqlx::FromRow)]
    pub(super) struct OrderJob {
        #[sqlx(flatten)]
        headers: OrderHeaders,
        pub(super) note: Option<String>,
    }

    impl InboxTable for OrderJob {
        type Id = i64;
        type Table = InboxSpec<(Lease<DateTime<Utc>>, Attempt, Fields)>;
        const TABLE: Self::Table =
            InboxSpec::new("shipping_jobs", Column::new("job_id").generated())
                .lease(Column::new("locked_until"))
                .group(Column::new("name"))
                .attempt(Column::new("attempt").generated())
                .data(&[Column::new("tenant"), Column::new("trace")])
                .fetching(&[Column::new("note")])
                .header_fields();

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

    impl Input for OrderJob {
        type Axis = SoloCarried<Self>;
    }

    impl AttemptRow for OrderJob {
        type Attempt = i16;

        fn attempt(&self) -> &i16 {
            &self.headers.attempt
        }
    }

    impl HeaderFields for OrderJob {
        const NAMES: &'static [&'static str] = &["tenant", "trace"];

        fn header_map(&self) -> HeaderMap {
            let mut headers = HeaderMap::with_capacity(Self::NAMES.len());
            put_header(&mut headers, "tenant", &self.headers.tenant);
            put_header(&mut headers, "trace", &self.headers.trace);
            headers
        }
    }
}

// The subject is the header map a delivery builds, read off the subscription itself.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn the_derived_layout_builds_its_headers() {
    let db = database(SCHEMA).await;
    let connected = broker(&db.pool)
        .connect()
        .await
        .expect("the broker connects");
    let mut subscriber = InboxQueue::<derived::OrderJob>::new("orders")
        .subscribe(&connected)
        .await
        .expect("the subscription opens");
    {
        let mut deliveries = pin!(subscriber.stream());
        let delivery = deliveries
            .next()
            .await
            .expect("a delivery")
            .expect("the claim");
        let note = delivery.carried().expect("the row").note.clone();
        let headers = delivery.headers();
        // A NULL leaves its header out.
        assert_eq!(
            (
                headers.get_str("tenant"),
                headers.get_str("trace"),
                note.as_deref()
            ),
            (Some("acme"), None, Some("fragile"))
        );
        delivery.ack().await.expect("the delivery acknowledges");
    }
    drop(subscriber);
    assert_eq!(count(&db.pool, LEFT).await, 0, "the ack deleted the row");
    connected.shutdown().await.expect("the broker shuts down");
    db.finish().await;
}

// The subject is the header map a delivery builds, read off the subscription itself.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn the_manual_layout_builds_its_headers() {
    let db = database(SCHEMA).await;
    let connected = broker(&db.pool)
        .connect()
        .await
        .expect("the broker connects");
    let mut subscriber = InboxQueue::<manual::OrderJob>::new("orders")
        .subscribe(&connected)
        .await
        .expect("the subscription opens");
    {
        let mut deliveries = pin!(subscriber.stream());
        let delivery = deliveries
            .next()
            .await
            .expect("a delivery")
            .expect("the claim");
        let note = delivery.carried().expect("the row").note.clone();
        let headers = delivery.headers();
        // A NULL leaves its header out.
        assert_eq!(
            (
                headers.get_str("tenant"),
                headers.get_str("trace"),
                note.as_deref()
            ),
            (Some("acme"), None, Some("fragile"))
        );
        delivery.ack().await.expect("the delivery acknowledges");
    }
    drop(subscriber);
    assert_eq!(count(&db.pool, LEFT).await, 0, "the ack deleted the row");
    connected.shutdown().await.expect("the broker shuts down");
    db.finish().await;
}

#[test]
fn both_forms_describe_the_same_table() {
    assert_eq!(
        <derived::OrderJob as InboxRow>::SPEC,
        <manual::OrderJob as InboxRow>::SPEC
    );
    assert_eq!(
        <derived::OrderJob as Events<Sqlite>>::SHAPE,
        <manual::OrderJob as Events<Sqlite>>::SHAPE
    );
    // A message assembled from fields is handed over in row mode, which no by-name subscription
    // reads, in either form.
    assert_eq!(<derived::OrderJob as Events<Sqlite>>::kinds(), None);
    assert_eq!(<manual::OrderJob as Events<Sqlite>>::kinds(), None);
}