ruststream-sqlx 0.7.1

//! A table in row mode, in the lease form, with a partition key, an `ack` of the service's own and
//! the insert a producer writes a task with: the handler takes the row itself.

use std::pin::pin;

use futures::StreamExt;
use ruststream::{
    Broker, Carries, ConnectedBroker, IncomingMessage, Subscriber, SubscriptionSource,
};
use ruststream_sqlx::__private::Events;
use ruststream_sqlx::dialect::{Dialect, Sqlite as SqliteDialect};
use ruststream_sqlx::prelude::*;
use ruststream_sqlx::{InboxRow, InboxTable, Insert};
use sqlx::{Sqlite, SqliteConnection, SqlitePool};

use super::{broker, database};

const SCHEMA: &str = "
CREATE TABLE mail_tasks (
    job_id       INTEGER PRIMARY KEY AUTOINCREMENT,
    name         TEXT NOT NULL,
    tenant       TEXT NOT NULL,
    attempt      INTEGER NOT NULL DEFAULT 1,
    locked_until TEXT,
    recipient    TEXT NOT NULL
);
CREATE TABLE mail_done (job_id INTEGER NOT NULL, recipient TEXT NOT NULL);
";

/// The service's own `ack`: the finished mail moves to `mail_done`.
async fn move_done(conn: &mut SqliteConnection, id: i64) -> Result<(), sqlx::Error> {
    sqlx::query(
        "INSERT INTO mail_done (job_id, recipient) SELECT job_id, recipient FROM mail_tasks \
         WHERE job_id = ?",
    )
    .bind(id)
    .execute(&mut *conn)
    .await?;
    sqlx::query("DELETE FROM mail_tasks WHERE job_id = ?")
        .bind(id)
        .execute(conn)
        .await?;
    Ok(())
}

mod derived {
    use chrono::{DateTime, Utc};
    use ruststream_sqlx::{Ack, Inbox};
    use sqlx::{Sqlite, SqliteConnection};

    #[derive(Debug, Clone, PartialEq, Inbox, sqlx::FromRow)]
    #[inbox(table = "mail_tasks", custom(ack))]
    pub(super) struct Mail {
        #[field(id, generated)]
        pub(super) job_id: i64,
        #[field(group)]
        pub(super) name: String,
        #[field(partition_key)]
        pub(super) tenant: String,
        #[field(attempt, generated)]
        pub(super) attempt: i16,
        #[field(locked_until)]
        pub(super) locked_until: Option<DateTime<Utc>>,
        pub(super) recipient: String,
    }

    impl Ack<Sqlite> for Mail {
        async fn ack(conn: &mut SqliteConnection, id: &i64) -> Result<(), sqlx::Error> {
            super::move_done(conn, *id).await
        }
    }
}

mod manual {
    use chrono::{DateTime, Utc};
    use ruststream::runtime::{Input, SoloCarried};
    use ruststream_sqlx::dialect::Column;
    use ruststream_sqlx::dialect::insert::{self, Sql};
    use ruststream_sqlx::spec::{Attempt, Key, Lease, own};
    use ruststream_sqlx::{Ack, AttemptRow, InboxSpec, InboxTable, Insert, KeyRow};
    use sqlx::{Sqlite, SqliteConnection};

    #[derive(Debug, Clone, PartialEq, sqlx::FromRow)]
    pub(super) struct Mail {
        pub(super) job_id: i64,
        pub(super) name: String,
        pub(super) tenant: String,
        pub(super) attempt: i16,
        pub(super) locked_until: Option<DateTime<Utc>>,
        pub(super) recipient: String,
    }

    impl InboxTable for Mail {
        type Id = i64;
        type Table = InboxSpec<(Lease<DateTime<Utc>>, Key, Attempt, own::Ack)>;
        const TABLE: Self::Table = InboxSpec::new("mail_tasks", Column::new("job_id").generated())
            .lease(Column::new("locked_until"))
            .group(Column::new("name"))
            .partition_key(Column::new("tenant"))
            .attempt(Column::new("attempt").generated())
            .data(&[Column::new("recipient")])
            .own::<own::Ack>();

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

    // Row mode: the handler takes the row itself, on the core's carried lane.
    impl Input for Mail {
        type Axis = SoloCarried<Self>;
    }

    impl KeyRow for Mail {
        type Key = String;

        fn partition_key(&self) -> &String {
            &self.tenant
        }
    }

    impl AttemptRow for Mail {
        type Attempt = i16;

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

    impl Ack<Sqlite> for Mail {
        async fn ack(conn: &mut SqliteConnection, id: &i64) -> Result<(), sqlx::Error> {
            super::move_done(conn, *id).await
        }
    }

    /// The insert, written while the service compiles from the description the subscription
    /// reads.
    pub(super) const INSERT: Sql<256> = insert::sqlite(&Mail::TABLE.spec());

    impl Insert<SqliteConnection> for Mail {
        // The columns bind in the description's order: the roles, then the data columns.
        async fn insert(&self, conn: &mut SqliteConnection) -> Result<(), sqlx::Error> {
            sqlx::query(INSERT.as_str())
                .bind(&self.name)
                .bind(&self.tenant)
                .bind(self.locked_until)
                .bind(&self.recipient)
                .execute(conn)
                .await?;
            Ok(())
        }
    }
}

async fn done(pool: &SqlitePool) -> Vec<String> {
    sqlx::query_scalar("SELECT recipient FROM mail_done")
        .fetch_all(pool)
        .await
        .expect("the table reads")
}

async fn write(pool: &SqlitePool, row: &impl Insert<SqliteConnection>) {
    let mut conn = pool.acquire().await.expect("a connection");
    row.insert(&mut conn).await.expect("the row writes");
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn the_derived_row_runs_its_own_ack() {
    let db = database(SCHEMA).await;
    write(
        &db.pool,
        &derived::Mail {
            job_id: 0,
            name: "mail".into(),
            tenant: "acme".into(),
            attempt: 0,
            locked_until: None,
            recipient: "ops@example.com".into(),
        },
    )
    .await;
    // The subject is the row a delivery lends and the ack it runs, read off the subscription.
    let connected = broker(&db.pool)
        .connect()
        .await
        .expect("the broker connects");
    let mut subscriber = InboxQueue::<derived::Mail>::new("mail")
        .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 lent = delivery.carried().expect("the row").clone();
        assert_eq!(
            (lent.recipient.as_str(), lent.tenant.as_str(), lent.attempt),
            ("ops@example.com", "acme", 1)
        );
        delivery.ack().await.expect("the delivery acknowledges");
    }
    assert_eq!(
        done(&db.pool).await,
        ["ops@example.com"],
        "the own ack moved the mail"
    );
    drop(subscriber);
    connected.shutdown().await.expect("the broker shuts down");
    db.finish().await;
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn the_manual_row_runs_its_own_ack() {
    let db = database(SCHEMA).await;
    write(
        &db.pool,
        &manual::Mail {
            job_id: 0,
            name: "mail".into(),
            tenant: "acme".into(),
            attempt: 0,
            locked_until: None,
            recipient: "ops@example.com".into(),
        },
    )
    .await;
    // The subject is the row a delivery lends and the ack it runs, read off the subscription.
    let connected = broker(&db.pool)
        .connect()
        .await
        .expect("the broker connects");
    let mut subscriber = InboxQueue::<manual::Mail>::new("mail")
        .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 lent = delivery.carried().expect("the row").clone();
        assert_eq!(
            (lent.recipient.as_str(), lent.tenant.as_str(), lent.attempt),
            ("ops@example.com", "acme", 1)
        );
        delivery.ack().await.expect("the delivery acknowledges");
    }
    assert_eq!(
        done(&db.pool).await,
        ["ops@example.com"],
        "the own ack moved the mail"
    );
    drop(subscriber);
    connected.shutdown().await.expect("the broker shuts down");
    db.finish().await;
}

#[test]
fn both_forms_describe_the_same_table_events_and_insert() {
    assert_eq!(
        <derived::Mail as InboxRow>::SPEC,
        <manual::Mail as InboxRow>::SPEC
    );
    assert_eq!(
        <derived::Mail as Events<Sqlite>>::SHAPE,
        <manual::Mail as Events<Sqlite>>::SHAPE
    );
    let written = SqliteDialect
        .insert(&<manual::Mail as InboxTable>::TABLE.spec())
        .expect("the dialect writes the insert");
    assert_eq!(
        manual::INSERT.as_str(),
        written.sql(),
        "the const text is the dialect's"
    );
    let row = manual::Mail {
        job_id: 1,
        name: "mail".into(),
        tenant: "acme".into(),
        attempt: 1,
        locked_until: None,
        recipient: "ops@example.com".into(),
    };
    assert_eq!(
        <manual::Mail as Events<Sqlite>>::partition_key(&row),
        Some(b"acme".as_slice())
    );
    // Row mode: a by-name subscription reads no table in it, in either form.
    assert_eq!(<derived::Mail as Events<Sqlite>>::kinds(), None);
    assert_eq!(<manual::Mail as Events<Sqlite>>::kinds(), None);
}