cratestack-sqlx 0.15.3

//! Audit-log table DDL + idempotent bootstrap.

use std::sync::atomic::Ordering;

use cratestack_core::CratestackError;

use crate::SqlxRuntime;
use crate::sqlx;

/// DDL for the audit log table. Banks typically run migrations
/// through their own tooling — this DDL is exposed so the
/// [`crate::SqlxRuntime`] can idempotently ensure the table exists
/// during bootstrap.
pub const AUDIT_TABLE_DDL: &str = r#"
CREATE TABLE IF NOT EXISTS cratestack_audit (
    event_id UUID PRIMARY KEY,
    schema_name TEXT NOT NULL,
    model TEXT NOT NULL,
    operation TEXT NOT NULL,
    primary_key JSONB NOT NULL,
    actor JSONB NOT NULL,
    tenant TEXT,
    before JSONB,
    after JSONB,
    request_id TEXT,
    occurred_at TIMESTAMPTZ NOT NULL,
    delivered_at TIMESTAMPTZ,
    attempts BIGINT NOT NULL DEFAULT 0,
    last_error TEXT
);

CREATE INDEX IF NOT EXISTS cratestack_audit_model_idx
    ON cratestack_audit (schema_name, model, occurred_at DESC);

CREATE INDEX IF NOT EXISTS cratestack_audit_tenant_idx
    ON cratestack_audit (tenant, occurred_at DESC)
    WHERE tenant IS NOT NULL;

CREATE INDEX IF NOT EXISTS cratestack_audit_undelivered_idx
    ON cratestack_audit (occurred_at)
    WHERE delivered_at IS NULL;
"#;

/// Whether everything [`AUDIT_TABLE_DDL`] creates already exists. Keep the
/// names in step with the DDL above.
const AUDIT_OBJECTS_EXIST: &str = "SELECT to_regclass('cratestack_audit') IS NOT NULL \
     AND to_regclass('cratestack_audit_model_idx') IS NOT NULL \
     AND to_regclass('cratestack_audit_tenant_idx') IS NOT NULL \
     AND to_regclass('cratestack_audit_undelivered_idx') IS NOT NULL";

/// Idempotently bootstraps `cratestack_audit`, but only actually runs
/// the DDL once per [`SqlxRuntime`] (cached on a shared flag, so every
/// clone of the same runtime agrees). This is load-bearing, not just
/// an optimization: `CREATE INDEX IF NOT EXISTS` still takes a
/// `ShareLock` on the table even when it's a no-op, which self-
/// deadlocks against a `RowExclusiveLock` a prior audited write in the
/// same caller-managed transaction is already holding. Skipping the
/// DDL entirely after the first successful run avoids taking that
/// lock at all on every subsequent call.
///
/// `probe` is the transaction the audited write runs in, which already
/// holds one pooled connection for its whole life. For every caller the
/// bootstrap first asks that transaction whether the table *and every index
/// the DDL creates* already exist — the normal case wherever migrations
/// created them — so it does not need a second connection just to learn
/// that (docs/design/procedure-isolation.md §4.1, cratestack#1117). A
/// table created without those indexes still gets them from the DDL, which
/// runs on the pool.
pub(crate) async fn ensure_audit_table<'e, E>(
    runtime: &SqlxRuntime,
    probe: E,
) -> Result<(), CratestackError>
where
    E: sqlx::Executor<'e, Database = sqlx::Postgres>,
{
    if runtime.audit_table_ensured().load(Ordering::Acquire) {
        return Ok(());
    }
    let exists: bool = sqlx::query_scalar(AUDIT_OBJECTS_EXIST)
        .fetch_one(probe)
        .await
        .map_err(|error| CratestackError::Database(error.to_string()))?;
    if exists {
        runtime.audit_table_ensured().store(true, Ordering::Release);
        return Ok(());
    }

    // `raw_sql` sends the whole DDL block as one batch over PG's
    // simple-query protocol instead of splitting on `;` client-side
    // (which would corrupt any dollar-quoted body). Sub-statements are
    // idempotent (`CREATE ... IF NOT EXISTS`), so this stays safe under
    // concurrent first-runs.
    sqlx::raw_sql(AUDIT_TABLE_DDL)
        .execute(runtime.pool())
        .await
        .map_err(|error| CratestackError::Database(error.to_string()))?;

    runtime.audit_table_ensured().store(true, Ordering::Release);
    Ok(())
}