reliar-store-postgres 0.6.0

//! `search_path` verification and the transaction-local `search_path` wrap `enqueue` opts into
//! (ADR 0017). [`verify_schema`]/[`other_outbox_schemas`] run once at
//! [`super::PostgresOutboxStore::connect`]; [`set_search_path`]/[`restore_search_path`] run on
//! every `enqueue` when `PostgresOutboxSettings::enqueue_sets_search_path` is set.

use sqlx::{PgPool, Postgres, Transaction};

use crate::error::{PostgresStoreError, is_undefined_table};

/// One row of the startup `search_path` verification query (ADR 0017).
pub(super) struct SchemaCheck {
    pub(super) resolved_schema: Option<String>,

    pub(super) configured_exists: bool,

    pub(super) search_path: String,
}

pub(super) async fn verify_schema(
    pool: &PgPool,
    schema: &str,
) -> Result<SchemaCheck, PostgresStoreError> {
    let qualified = format!("{schema}.outbox");
    let row = sqlx::query!(
        r#"SELECT
             current_setting('search_path') AS "search_path!",
             (SELECT n.nspname
                FROM pg_class c JOIN pg_namespace n ON n.oid = c.relnamespace
               WHERE c.oid = to_regclass('outbox')) AS resolved_schema,
             (to_regclass($1) IS NOT NULL) AS "configured_exists!""#,
        qualified,
    )
    .fetch_one(pool)
    .await
    .map_err(|err| {
        if is_undefined_table(&err) {
            PostgresStoreError::NotMigrated {
                schema: schema.to_owned(),
            }
        } else {
            PostgresStoreError::from(err)
        }
    })?;

    Ok(SchemaCheck {
        resolved_schema: row.resolved_schema,
        configured_exists: row.configured_exists,
        search_path: row.search_path,
    })
}

/// Rows with a `relname = 'outbox'` outside `schema`, for the same-named-table warning
/// (ADR 0017). Empty when there is no such duplicate.
pub(super) async fn other_outbox_schemas(
    pool: &PgPool,
    schema: &str,
) -> Result<Vec<String>, PostgresStoreError> {
    let schemas = sqlx::query_scalar!(
        r#"SELECT n.nspname
             FROM pg_class c JOIN pg_namespace n ON n.oid = c.relnamespace
            WHERE c.relname = 'outbox' AND n.nspname <> $1"#,
        schema,
    )
    .fetch_all(pool)
    .await?;

    Ok(schemas)
}

/// Reads the caller's current `search_path`, sets it transaction-locally
/// (`set_config(.., true)` — dies with the caller's `COMMIT`/`ROLLBACK`) to put `schema` first,
/// and returns the previous value so it can be restored. Returns the raw
/// [`sqlx::Error`] — not [`crate::error::EnqueueError`] — so `enqueue::insert_enqueued`, its only
/// caller, is the one place that maps it.
pub(super) async fn set_search_path(
    tx: &mut Transaction<'_, Postgres>,
    schema: &str,
) -> Result<String, sqlx::Error> {
    let previous: String = sqlx::query_scalar!("SELECT current_setting('search_path')")
        .fetch_one(&mut **tx)
        .await?
        .unwrap_or_default();
    let wanted = format!("{schema},public");

    sqlx::query_scalar!("SELECT set_config('search_path', $1, true)", wanted)
        .fetch_one(&mut **tx)
        .await?;

    Ok(previous)
}

pub(super) async fn restore_search_path(
    tx: &mut Transaction<'_, Postgres>,
    previous: &str,
) -> Result<(), sqlx::Error> {
    sqlx::query_scalar!("SELECT set_config('search_path', $1, true)", previous)
        .fetch_one(&mut **tx)
        .await?;

    Ok(())
}