obix 0.12.0

mod helpers;

use futures::stream::StreamExt;
use obix::{EventSequence, MailboxConfig, PartitionMaintainerConfig, out::Outbox};
use serde::{Deserialize, Serialize};
use serial_test::file_serial;
use sqlx::Row;

use helpers::{
    TestTables, init_outbox, init_pool, wipeout_outbox_job_tables, wipeout_outbox_tables,
};

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
enum TestEvent {
    Ping(u64),
}

const WIDTH: u64 = obix::DEFAULT_PARTITION_WIDTH; // 2_000_000 — the p0/p1 boundary
const PREMAKE: u64 = obix::DEFAULT_PARTITION_PREMAKE;
const BOUNDARY: i64 = WIDTH as i64;

// Driving the sequence to the p0 boundary uses `setval` rather than millions of
// rows; created partitions persist across serial tests, hence the reset below.

async fn reset_partitions_to_baseline(pool: &sqlx::PgPool) -> anyhow::Result<()> {
    for table in ["persistent_outbox_events"] {
        // A prior recovery may have left a detached DEFAULT copy.
        sqlx::query(&format!("DROP TABLE IF EXISTS {table}_default_old"))
            .execute(pool)
            .await?;
        // Their count varies with `premake`, so enumerate them from the catalog
        // rather than hard-coding names.
        let children = sqlx::query(
            "SELECT c.relname FROM pg_inherits i \
             JOIN pg_class c ON c.oid = i.inhrelid \
             WHERE i.inhparent = $1::regclass \
               AND c.relname NOT IN ($2, $3)",
        )
        .bind(table)
        .bind(format!("{table}_p0"))
        .bind(format!("{table}_default"))
        .fetch_all(pool)
        .await?;
        for row in children {
            let name: String = row.get("relname");
            sqlx::query(&format!("DROP TABLE IF EXISTS {name}"))
                .execute(pool)
                .await?;
        }
        // Recovery recreates DEFAULT, but guard against a half-run leaving it
        // gone.
        sqlx::query(&format!(
            "CREATE TABLE IF NOT EXISTS {table}_default PARTITION OF {table} DEFAULT"
        ))
        .execute(pool)
        .await?;
    }
    Ok(())
}

async fn set_sequence(pool: &sqlx::PgPool, value: i64) -> anyhow::Result<()> {
    sqlx::query("SELECT setval('persistent_outbox_events_sequence_seq', $1)")
        .bind(value)
        .execute(pool)
        .await?;
    Ok(())
}

async fn relation_exists(pool: &sqlx::PgPool, name: &str) -> anyhow::Result<bool> {
    let row = sqlx::query("SELECT to_regclass($1) IS NOT NULL AS present")
        .bind(name)
        .fetch_one(pool)
        .await?;
    Ok(row.get::<bool, _>("present"))
}

async fn default_row_count(pool: &sqlx::PgPool) -> anyhow::Result<i64> {
    let row = sqlx::query("SELECT COUNT(*) AS n FROM persistent_outbox_events_default")
        .fetch_one(pool)
        .await?;
    Ok(row.get::<i64, _>("n"))
}

async fn max_sequence(pool: &sqlx::PgPool) -> anyhow::Result<Option<i64>> {
    let row = sqlx::query("SELECT MAX(sequence) AS m FROM persistent_outbox_events")
        .fetch_one(pool)
        .await?;
    Ok(row.get::<Option<i64>, _>("m"))
}

async fn reloptions(pool: &sqlx::PgPool, relname: &str) -> anyhow::Result<String> {
    let row = sqlx::query(
        "SELECT COALESCE(array_to_string(reloptions, ','), '') AS opts \
         FROM pg_class WHERE relname = $1",
    )
    .bind(relname)
    .fetch_one(pool)
    .await?;
    Ok(row.get::<String, _>("opts"))
}

/// Wipe, reset to baseline partitions, and open an outbox whose cache starts at
/// `head`, so a `None` listener never tries to backfill from 0.
async fn prepare_outbox(
    pool: &sqlx::PgPool,
    head: i64,
) -> anyhow::Result<Outbox<TestEvent, TestTables>> {
    wipeout_outbox_tables(pool).await?;
    reset_partitions_to_baseline(pool).await?;
    set_sequence(pool, head).await?;
    let outbox = Outbox::<TestEvent, TestTables>::init(
        pool,
        MailboxConfig::builder()
            .build()
            .expect("Couldn't build MailboxConfig"),
    )
    .await?;
    Ok(outbox)
}

#[tokio::test]
#[file_serial]
async fn maintainer_premakes_partitions_ahead() -> anyhow::Result<()> {
    let pool = init_pool().await?;
    let job_type = "test-partition-maintainer";
    wipeout_outbox_tables(&pool).await?;
    wipeout_outbox_job_tables(&pool, job_type).await?;
    reset_partitions_to_baseline(&pool).await?;

    // Head approaching the p0 boundary, with no explicit partition beyond p0.
    set_sequence(&pool, BOUNDARY - 10).await?;
    assert!(
        !relation_exists(&pool, "persistent_outbox_events_p1").await?,
        "p1 must not exist before registration"
    );

    let job_config = job::JobSvcConfig::builder()
        .pool(pool.clone())
        .build()
        .unwrap();
    let mut jobs = job::Jobs::init(job_config).await?;

    let outbox = Outbox::<TestEvent, TestTables>::init(
        &pool,
        MailboxConfig::builder().build().expect("config"),
    )
    .await?;

    // Registration runs one premake pass synchronously before returning.
    outbox
        .register_partition_maintainer(
            &mut jobs,
            PartitionMaintainerConfig::new(job::JobType::new(job_type)),
        )
        .await
        .map_err(|e| anyhow::anyhow!("{e}"))?;

    // Both the next partition AND the full premake runway exist.
    for table in ["persistent_outbox_events"] {
        assert!(
            relation_exists(&pool, &format!("{table}_p1")).await?,
            "{table}_p1 pre-created before the head reaches the boundary"
        );
        let deepest = format!("{table}_p{PREMAKE}");
        assert!(
            relation_exists(&pool, &deepest).await?,
            "premake keeps {PREMAKE} partitions ahead ({deepest})"
        );

        let opts = reloptions(&pool, &format!("{table}_p1")).await?;
        assert!(
            opts.contains("autovacuum_freeze_min_age=0"),
            "freeze param present on {table}_p1: {opts}"
        );
        assert!(
            opts.contains("autovacuum_vacuum_insert_scale_factor=0"),
            "insert-vacuum param present on {table}_p1: {opts}"
        );
    }

    let _ = jobs.shutdown().await;
    Ok(())
}

#[tokio::test]
#[file_serial]
async fn default_fill_then_recover() -> anyhow::Result<()> {
    let pool = init_pool().await?;
    // No explicit partition beyond p0, so rows past the boundary land in DEFAULT.
    let outbox = prepare_outbox(&pool, BOUNDARY - 2).await?;

    let mut op = outbox.begin_op().await?;
    outbox
        .publish_all_persisted(
            &mut op,
            [
                TestEvent::Ping(0), // BOUNDARY-1, p0
                TestEvent::Ping(1), // BOUNDARY,   DEFAULT
                TestEvent::Ping(2), // BOUNDARY+1, DEFAULT
            ],
        )
        .await?;
    op.commit().await?;

    assert_eq!(
        default_row_count(&pool).await?,
        2,
        "two rows stranded in DEFAULT"
    );
    let max_before = max_sequence(&pool).await?;
    assert_eq!(max_before, Some(BOUNDARY + 1));

    // The stranded rows are readable before recovery.
    assert_replayable(&pool).await?;

    obix::out::Partitions::<TestTables>::new(&pool, PREMAKE)
        .recover_default()
        .await?;

    assert_eq!(default_row_count(&pool).await?, 0, "DEFAULT drained");
    assert!(
        relation_exists(&pool, "persistent_outbox_events_p1").await?,
        "stranded rows moved into explicit p1"
    );
    assert_eq!(
        max_sequence(&pool).await?,
        max_before,
        "MAX(sequence) never regressed"
    );
    // Recovery left no artifact behind.
    assert!(!relation_exists(&pool, "persistent_outbox_events_default_old").await?);

    assert_replayable(&pool).await?;
    Ok(())
}

/// Replay the three events published around the boundary from a fresh outbox
/// (empty cache → DB backfill).
async fn assert_replayable(pool: &sqlx::PgPool) -> anyhow::Result<()> {
    let replay = Outbox::<TestEvent, TestTables>::init(
        pool,
        MailboxConfig::builder().build().expect("config"),
    )
    .await?;
    let mut listener = replay.listen_persisted(Some(EventSequence::from((BOUNDARY - 2) as u64)));
    for (i, expected_seq) in [BOUNDARY - 1, BOUNDARY, BOUNDARY + 1]
        .into_iter()
        .enumerate()
    {
        let ev = tokio::time::timeout(std::time::Duration::from_secs(3), listener.next())
            .await?
            .expect("replayed event")?;
        assert_eq!(ev.sequence, EventSequence::from(expected_seq as u64));
        assert_eq!(ev.payload, Some(TestEvent::Ping(i as u64)));
    }
    Ok(())
}

#[tokio::test]
#[file_serial]
async fn concurrent_ensure_does_not_race() -> anyhow::Result<()> {
    let pool = init_pool().await?;
    wipeout_outbox_tables(&pool).await?;
    reset_partitions_to_baseline(&pool).await?;
    set_sequence(&pool, BOUNDARY - 10).await?;

    let results = futures::future::join_all((0..8).map(|_| {
        let pool = pool.clone();
        async move {
            obix::out::Partitions::<TestTables>::new(&pool, PREMAKE)
                .ensure()
                .await
        }
    }))
    .await;
    for r in results {
        r?;
    }
    let deepest = format!("persistent_outbox_events_p{PREMAKE}");
    assert!(relation_exists(&pool, &deepest).await?);
    Ok(())
}

#[tokio::test]
#[file_serial]
async fn replay_from_begin_still_contiguous() -> anyhow::Result<()> {
    let pool = init_pool().await?;
    let outbox =
        init_outbox::<TestEvent>(&pool, MailboxConfig::builder().build().expect("config")).await?;
    reset_partitions_to_baseline(&pool).await?;

    let mut op = outbox.begin_op().await?;
    outbox
        .publish_all_persisted(
            &mut op,
            [TestEvent::Ping(0), TestEvent::Ping(1), TestEvent::Ping(2)],
        )
        .await?;
    op.commit().await?;

    let replay = Outbox::<TestEvent, TestTables>::init(
        &pool,
        MailboxConfig::builder().build().expect("config"),
    )
    .await?;
    let mut listener = replay.listen_persisted(EventSequence::BEGIN);
    for i in 0..3u64 {
        let ev = tokio::time::timeout(std::time::Duration::from_secs(3), listener.next())
            .await?
            .expect("replayed event")?;
        assert_eq!(ev.sequence, EventSequence::from(i + 1));
        assert_eq!(ev.payload, Some(TestEvent::Ping(i)));
    }
    Ok(())
}