systemprompt_database/lifecycle/
validation.rs

1//! Pre-flight validation helpers used by the boot path and tests.
2//!
3//! Copyright (c) systemprompt.io — Business Source License 1.1.
4//! See <https://systemprompt.io> for licensing details.
5
6use crate::error::{DatabaseResult, RepositoryError};
7use crate::services::{Database, DatabaseProvider};
8
9pub async fn validate_database_connection(db: &dyn DatabaseProvider) -> DatabaseResult<()> {
10    db.test_connection().await.map_err(|e| {
11        RepositoryError::Internal(format!("Failed to establish database connection: {e}"))
12    })
13}
14
15/// Rejects a write pool that resolves to a streaming-replication standby.
16///
17/// Every write path — schema installation, extension seeds, the job scheduler,
18/// log persistence, and `LISTEN`/`NOTIFY` on the event bridge — goes through
19/// [`Database::write`], which falls back to the read pool when no separate
20/// write URL is configured. A read URL aimed at a replica therefore turns into
21/// a slow, opaque boot failure: DDL stalls, then jobs die on `25006`. Failing
22/// here names the cause instead.
23pub async fn validate_write_pool_is_primary(db: &Database) -> DatabaseResult<()> {
24    if !db.write().is_postgres() {
25        return Ok(());
26    }
27
28    let result = db
29        .write()
30        .query_raw(&"SELECT pg_is_in_recovery() as in_recovery")
31        .await?;
32
33    let in_recovery = result
34        .first()
35        .and_then(|row| row.get("in_recovery"))
36        .and_then(serde_json::Value::as_bool)
37        .ok_or_else(|| {
38            RepositoryError::Internal(
39                "Failed to determine whether the write pool is a primary".to_owned(),
40            )
41        })?;
42
43    if !in_recovery {
44        return Ok(());
45    }
46
47    Err(RepositoryError::invalid_state(if db.has_write_pool() {
48        "`database_write_url` points at a read-only standby. Writes, migrations and \
49         LISTEN/NOTIFY all require the primary — point it at the primary and restart"
50    } else {
51        "`database_url` points at a read-only standby and no `database_write_url` is set, so \
52         the write pool falls back to it. Set `database_write_url` (or `DATABASE_WRITE_URL` \
53         with the env secrets source) to the primary and restart"
54    }))
55}
56
57pub async fn validate_table_exists(
58    db: &dyn DatabaseProvider,
59    table_name: &str,
60) -> DatabaseResult<bool> {
61    let result = db
62        .query_raw_with(
63            &"SELECT EXISTS (SELECT 1 FROM information_schema.tables WHERE table_schema = \
64              'public' AND table_name = $1) as exists",
65            &[&table_name],
66        )
67        .await?;
68
69    result
70        .first()
71        .and_then(|row| row.get("exists"))
72        .and_then(serde_json::Value::as_bool)
73        .ok_or_else(|| {
74            RepositoryError::Internal(format!(
75                "Failed to check table existence for '{table_name}'"
76            ))
77        })
78}
79
80pub async fn validate_column_exists(
81    db: &dyn DatabaseProvider,
82    table_name: &str,
83    column_name: &str,
84) -> DatabaseResult<bool> {
85    let result = db
86        .query_raw_with(
87            &"SELECT EXISTS (SELECT 1 FROM information_schema.columns WHERE table_schema = \
88              'public' AND table_name = $1 AND column_name = $2) as exists",
89            &[&table_name, &column_name],
90        )
91        .await?;
92
93    result
94        .first()
95        .and_then(|row| row.get("exists"))
96        .and_then(serde_json::Value::as_bool)
97        .ok_or_else(|| {
98            RepositoryError::Internal(format!(
99                "Failed to check column existence for '{table_name}.{column_name}'"
100            ))
101        })
102}
systemprompt_database/lifecycle/validation.rs

systemprompt_database/lifecycle/
validation.rs
