reliar-store-postgres 0.7.0

//! `search_path` verification and the transaction-local `search_path` wrap the outbox's `enqueue`
//! and the inbox's `claim`/`complete` opt into (ADR 0017). [`verify_table_schema`]/
//! [`other_table_schemas`] run once at each store's `connect`; [`set_search_path`]/
//! [`restore_search_path`] run per operation when the owning `*Settings`'s
//! `*_sets_search_path` knob is set.
//!
//! Table-agnostic throughout — the table name is always a bind parameter, never interpolated
//! into the query text — so `crate::outbox` and `crate::inbox` share this one query shape instead
//! of each duplicating it (inbox contract §3).

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

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

    pub(crate) configured_exists: bool,

    pub(crate) search_path: String,

    /// The subset of the caller's `required_columns` that exist on the resolved relation **and**
    /// are `NOT NULL` (excludes dropped columns, system columns, and a column that exists but is
    /// still nullable) — empty when `required_columns` itself was empty. The caller diffs this
    /// against its own required list to name the first unsatisfied one (ADR 0044 A.4, Amendment
    /// A.5): a required column is a **completion marker** for a migration sequence, not a bare
    /// inventory check, so a column that exists but has not yet been backfilled and `SET NOT
    /// NULL`ed must count the same as one that does not exist yet.
    pub(crate) satisfied_required_columns: Vec<String>,
}

/// The raw check, parameterized by `table` and `required_columns` — bind parameters, never
/// interpolated into the query text, so this stays a single `sqlx::query!` invocation for every
/// table this crate verifies at startup (`outbox`, `inbox`; ADR 0044 A.4 folded the required-column
/// check into the same round trip rather than a second query). Returns the raw [`sqlx::Error`];
/// callers map it into their own typed error (`crate::error::is_undefined_table` is already
/// table-agnostic).
pub(crate) async fn verify_table_schema(
    pool: &PgPool,
    schema: &str,
    table: &str,
    required_columns: &[&str],
) -> Result<SchemaCheck, sqlx::Error> {
    let qualified = format!("{schema}.{table}");
    let required: Vec<String> = required_columns.iter().map(|&c| c.to_owned()).collect();
    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($2)) AS resolved_schema,
             (to_regclass($1) IS NOT NULL) AS "configured_exists!",
             ARRAY(
               SELECT a.attname FROM pg_attribute a
                WHERE a.attrelid = to_regclass($2) AND a.attname = ANY($3::text[])
                  AND NOT a.attisdropped AND a.attnum > 0 AND a.attnotnull
             ) AS "satisfied_required_columns!: Vec<String>""#,
        qualified,
        table,
        &required,
    )
    .fetch_one(pool)
    .await?;

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

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

/// 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`] so each caller (`outbox::enqueue`, `inbox::claim`) maps it into its own typed
/// error.
pub(crate) 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(crate) 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(())
}

/// Validates a schema name against PostgreSQL's unquoted-identifier grammar, restricted to
/// **lowercase** (`[a-z_][a-z0-9_$]*`, at most 63 bytes — Postgres's own `NAMEDATALEN` limit)
/// **before** it is ever interpolated into `SET search_path`/`dangerous_set_table_name`, both of
/// which build SQL text from this value rather than binding it as data. Used by
/// both `PostgresOutboxSettings::schema` (at `connect`) and `MigrateOptions::schema` (at
/// `migrate`), so the two validate identically.
///
/// **Lowercase only, not merely case-insensitive (ADR 0040 §5).** PostgreSQL folds an *unquoted*
/// identifier to lowercase, so `schema = "Foo"` would migrate into a schema literally named
/// `"Foo"` (quoted) while every unqualified reference — the claim, `stats()`, the host's own
/// `search_path` — resolves the unquoted, lowercase-folded `foo` instead: a mismatch this crate
/// cannot detect from inside a single connection's `search_path`, since the host's own connection
/// string or `ALTER ROLE` also has to agree, and cannot be fixed here. Rejecting every uppercase
/// character removes the class of mismatch instead of chasing it through four call sites.
pub(crate) fn is_valid_schema_name(schema: &str) -> bool {
    let mut chars = schema.chars();
    let Some(first) = chars.next() else {
        return false;
    };

    if !(first.is_ascii_lowercase() || first == '_') {
        return false;
    }

    schema.len() <= 63
        && chars.all(|c| c.is_ascii_lowercase() || c.is_ascii_digit() || c == '_' || c == '$')
}