reliar-store-postgres 0.9.0

//! Inbox contract §5.2 I-P16 — `migrate()` is idempotent, `0004_inbox.sql` applies cleanly to a
//! database that already has `0001`–`0003` applied (including a populated `outbox`, whose rows
//! later migrations must preserve), and the resulting constraint/index names are exactly `pk_inbox`
//! (on `id`), `ck_inbox_attempts`, `ck_inbox_message_version`, `ck_inbox_scope_len`,
//! `ck_inbox_terminal`, `ix_inbox_scope_message_id` (**unique**), `ix_inbox_completed`,
//! `ix_inbox_dead` (ADR 0042 Amendment A) — and there is no `uq_`-prefixed object and no
//! `ix_inbox_conversation`.

use crate::common;

use std::collections::HashSet;

use crate::common::OrderCreated;
use reliar_core::Envelope;
use reliar_outbox::OutboxEnqueue;
use sqlx::PgPool;
use sqlx::postgres::PgConnectOptions;

const EXPECTED_CONSTRAINTS: &[&str] = &[
    "pk_inbox",
    "ck_inbox_attempts",
    "ck_inbox_message_version",
    "ck_inbox_scope_len",
    "ck_inbox_terminal",
];
const EXPECTED_INDEXES: &[&str] = &[
    "ix_inbox_scope_message_id",
    "ix_inbox_completed",
    "ix_inbox_dead",
];

async fn migrate_is_idempotent_and_preserves_existing_outbox_rows() {
    let base = common::fresh_unmigrated_db().await;

    // First migrate: applies the complete migration set.
    reliar_store_postgres::migrate(&base, reliar_store_postgres::MigrateOptions::default())
        .await
        .unwrap();

    // `fresh_unmigrated_db` documents "no search_path set yet" (`common/mod.rs`) — reconnect with
    // it set, the same way `common::fresh_db` does, before constructing a store that verifies
    // `search_path` at construction.
    let options: PgConnectOptions = base
        .connect_options()
        .as_ref()
        .clone()
        .options([("search_path", "reliar,public")]);
    let base = PgPool::connect_with(options).await.unwrap();

    // Populate `outbox` before re-running migrate(), so a regression that re-applies 0001 or
    // otherwise damages existing `outbox` rows fails this assertion rather than passing on an
    // empty table. Migrations 0005–0006 deliberately replace only its dead indexes.
    let store = reliar_store_postgres::PostgresOutboxStore::new(base.clone());
    let envelope = Envelope::builder(OrderCreated { order_id: 1 }).build();
    let mut tx = base.begin().await.unwrap();
    store.enqueue_envelope(&mut tx, envelope).await.unwrap();
    tx.commit().await.unwrap();

    // Second migrate: idempotent — no error, and the outbox row survives untouched.
    reliar_store_postgres::migrate(&base, reliar_store_postgres::MigrateOptions::default())
        .await
        .unwrap();

    let outbox_count = sqlx::query_scalar!(r#"SELECT count(*) AS "count!" FROM outbox"#)
        .fetch_one(&base)
        .await
        .unwrap();
    assert_eq!(
        outbox_count, 1,
        "a re-run of migrate() must preserve existing outbox rows"
    );

    let inbox_count = sqlx::query_scalar!(r#"SELECT count(*) AS "count!" FROM inbox"#)
        .fetch_one(&base)
        .await
        .unwrap();
    assert_eq!(inbox_count, 0, "0004 creates an empty table");
}

async fn every_constraint_and_index_is_named_as_specified() {
    let pool = common::fresh_db().await;

    let constraints: Vec<String> = sqlx::query_scalar(
        "SELECT conname FROM pg_constraint c \
           JOIN pg_class t ON t.oid = c.conrelid \
           JOIN pg_namespace n ON n.oid = t.relnamespace \
          WHERE t.relname = 'inbox' AND n.nspname = 'reliar'",
    )
    .fetch_all(&pool)
    .await
    .unwrap();
    let constraints: HashSet<&str> = constraints.iter().map(String::as_str).collect();

    for expected in EXPECTED_CONSTRAINTS {
        assert!(
            constraints.contains(expected),
            "missing constraint {expected}; found {constraints:?}"
        );
    }

    let indexes: Vec<String> = sqlx::query_scalar(
        "SELECT indexname FROM pg_indexes WHERE tablename = 'inbox' AND schemaname = 'reliar'",
    )
    .fetch_all(&pool)
    .await
    .unwrap();
    let indexes: HashSet<&str> = indexes.iter().map(String::as_str).collect();

    for expected in EXPECTED_INDEXES {
        assert!(
            indexes.contains(expected),
            "missing index {expected}; found {indexes:?}"
        );
    }

    assert!(
        !indexes.iter().any(|name| name.starts_with("uq_")),
        "no index may be named uq_ — every one is ix_ regardless of uniqueness"
    );
    assert!(
        !indexes.contains("ix_inbox_conversation"),
        "no index on conversation_id — it is written on every claim and the query is documented, \
         not indexed"
    );

    let is_unique: bool = sqlx::query_scalar(
        "SELECT indisunique FROM pg_index i \
           JOIN pg_class c ON c.oid = i.indexrelid \
          WHERE c.relname = 'ix_inbox_scope_message_id'",
    )
    .fetch_one(&pool)
    .await
    .unwrap();
    assert!(
        is_unique,
        "ix_inbox_scope_message_id must be a unique index — ON CONFLICT infers from it"
    );

    // ADR 0042 Amendment E: asserted by **key columns** (`pg_get_indexdef`), not by name alone.
    let dead_index_def: String = sqlx::query_scalar(
        "SELECT pg_get_indexdef(indexrelid) FROM pg_index i \
           JOIN pg_class c ON c.oid = i.indexrelid \
          WHERE c.relname = 'ix_inbox_dead'",
    )
    .fetch_one(&pool)
    .await
    .unwrap();
    assert!(
        dead_index_def.contains("(dead_at, id)"),
        "ix_inbox_dead must key on (dead_at, id) — got: {dead_index_def}"
    );
    assert!(
        dead_index_def.contains("WHERE (dead_at IS NOT NULL)"),
        "ix_inbox_dead must stay partial on dead_at IS NOT NULL — got: {dead_index_def}"
    );
}

pub(crate) fn trials(rt: &'static tokio::runtime::Runtime) -> Vec<libtest_mimic::Trial> {
    vec![
        libtest_mimic::Trial::test(
            "inbox_migration::migrate_is_idempotent_and_preserves_existing_outbox_rows",
            move || {
                rt.block_on(migrate_is_idempotent_and_preserves_existing_outbox_rows());
                Ok(())
            },
        ),
        libtest_mimic::Trial::test(
            "inbox_migration::every_constraint_and_index_is_named_as_specified",
            move || {
                rt.block_on(every_constraint_and_index_is_named_as_specified());
                Ok(())
            },
        ),
    ]
}