cratestack_sqlx/audit/
schema.rs

1//! Audit-log table DDL + idempotent bootstrap.
2
3use std::sync::atomic::Ordering;
4
5use cratestack_core::CratestackError;
6
7use crate::SqlxRuntime;
8use crate::sqlx;
9
10/// DDL for the audit log table. Banks typically run migrations
11/// through their own tooling — this DDL is exposed so the
12/// [`crate::SqlxRuntime`] can idempotently ensure the table exists
13/// during bootstrap.
14pub const AUDIT_TABLE_DDL: &str = r#"
15CREATE TABLE IF NOT EXISTS cratestack_audit (
16    event_id UUID PRIMARY KEY,
17    schema_name TEXT NOT NULL,
18    model TEXT NOT NULL,
19    operation TEXT NOT NULL,
20    primary_key JSONB NOT NULL,
21    actor JSONB NOT NULL,
22    tenant TEXT,
23    before JSONB,
24    after JSONB,
25    request_id TEXT,
26    occurred_at TIMESTAMPTZ NOT NULL,
27    delivered_at TIMESTAMPTZ,
28    attempts BIGINT NOT NULL DEFAULT 0,
29    last_error TEXT
30);
31
32CREATE INDEX IF NOT EXISTS cratestack_audit_model_idx
33    ON cratestack_audit (schema_name, model, occurred_at DESC);
34
35CREATE INDEX IF NOT EXISTS cratestack_audit_tenant_idx
36    ON cratestack_audit (tenant, occurred_at DESC)
37    WHERE tenant IS NOT NULL;
38
39CREATE INDEX IF NOT EXISTS cratestack_audit_undelivered_idx
40    ON cratestack_audit (occurred_at)
41    WHERE delivered_at IS NULL;
42"#;
43
44/// Whether everything [`AUDIT_TABLE_DDL`] creates already exists. Keep the
45/// names in step with the DDL above.
46const AUDIT_OBJECTS_EXIST: &str = "SELECT to_regclass('cratestack_audit') IS NOT NULL \
47     AND to_regclass('cratestack_audit_model_idx') IS NOT NULL \
48     AND to_regclass('cratestack_audit_tenant_idx') IS NOT NULL \
49     AND to_regclass('cratestack_audit_undelivered_idx') IS NOT NULL";
50
51/// Idempotently bootstraps `cratestack_audit`, but only actually runs
52/// the DDL once per [`SqlxRuntime`] (cached on a shared flag, so every
53/// clone of the same runtime agrees). This is load-bearing, not just
54/// an optimization: `CREATE INDEX IF NOT EXISTS` still takes a
55/// `ShareLock` on the table even when it's a no-op, which self-
56/// deadlocks against a `RowExclusiveLock` a prior audited write in the
57/// same caller-managed transaction is already holding. Skipping the
58/// DDL entirely after the first successful run avoids taking that
59/// lock at all on every subsequent call.
60///
61/// `probe` is the transaction the audited write runs in, which already
62/// holds one pooled connection for its whole life. For every caller the
63/// bootstrap first asks that transaction whether the table *and every index
64/// the DDL creates* already exist — the normal case wherever migrations
65/// created them — so it does not need a second connection just to learn
66/// that (docs/design/procedure-isolation.md §4.1, cratestack#1117). A
67/// table created without those indexes still gets them from the DDL, which
68/// runs on the pool.
69pub(crate) async fn ensure_audit_table<'e, E>(
70    runtime: &SqlxRuntime,
71    probe: E,
72) -> Result<(), CratestackError>
73where
74    E: sqlx::Executor<'e, Database = sqlx::Postgres>,
75{
76    if runtime.audit_table_ensured().load(Ordering::Acquire) {
77        return Ok(());
78    }
79    let exists: bool = sqlx::query_scalar(AUDIT_OBJECTS_EXIST)
80        .fetch_one(probe)
81        .await
82        .map_err(|error| CratestackError::Database(error.to_string()))?;
83    if exists {
84        runtime.audit_table_ensured().store(true, Ordering::Release);
85        return Ok(());
86    }
87
88    // `raw_sql` sends the whole DDL block as one batch over PG's
89    // simple-query protocol instead of splitting on `;` client-side
90    // (which would corrupt any dollar-quoted body). Sub-statements are
91    // idempotent (`CREATE ... IF NOT EXISTS`), so this stays safe under
92    // concurrent first-runs.
93    sqlx::raw_sql(AUDIT_TABLE_DDL)
94        .execute(runtime.pool())
95        .await
96        .map_err(|error| CratestackError::Database(error.to_string()))?;
97
98    runtime.audit_table_ensured().store(true, Ordering::Release);
99    Ok(())
100}
cratestack_sqlx/audit/schema.rs

cratestack_sqlx/audit/
schema.rs
