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
15pub async fn validate_write_pool_is_primary(db: &Database) -> DatabaseResult<()> {
16    if !db.write().is_postgres() {
17        return Ok(());
18    }
19
20    let result = db
21        .write()
22        .query_raw(&"SELECT pg_is_in_recovery() as in_recovery")
23        .await?;
24
25    let in_recovery = result
26        .first()
27        .and_then(|row| row.get("in_recovery"))
28        .and_then(serde_json::Value::as_bool)
29        .ok_or_else(|| {
30            RepositoryError::Internal(
31                "Failed to determine whether the write pool is a primary".to_owned(),
32            )
33        })?;
34
35    if !in_recovery {
36        return Ok(());
37    }
38
39    Err(RepositoryError::invalid_state(if db.has_write_pool() {
40        "`database_write_url` points at a read-only standby. Writes, migrations and \
41         LISTEN/NOTIFY all require the primary — point it at the primary and restart"
42    } else {
43        "`database_url` points at a read-only standby and no `database_write_url` is set, so \
44         the write pool falls back to it. Set `database_write_url` (or `DATABASE_WRITE_URL` \
45         with the env secrets source) to the primary and restart"
46    }))
47}
48
49pub async fn validate_table_exists(
50    db: &dyn DatabaseProvider,
51    table_name: &str,
52) -> DatabaseResult<bool> {
53    let result = db
54        .query_raw_with(
55            &"SELECT EXISTS (SELECT 1 FROM information_schema.tables WHERE table_schema = \
56              'public' AND table_name = $1) as exists",
57            &[&table_name],
58        )
59        .await?;
60
61    result
62        .first()
63        .and_then(|row| row.get("exists"))
64        .and_then(serde_json::Value::as_bool)
65        .ok_or_else(|| {
66            RepositoryError::Internal(format!(
67                "Failed to check table existence for '{table_name}'"
68            ))
69        })
70}
71
72pub async fn validate_column_exists(
73    db: &dyn DatabaseProvider,
74    table_name: &str,
75    column_name: &str,
76) -> DatabaseResult<bool> {
77    let result = db
78        .query_raw_with(
79            &"SELECT EXISTS (SELECT 1 FROM information_schema.columns WHERE table_schema = \
80              'public' AND table_name = $1 AND column_name = $2) as exists",
81            &[&table_name, &column_name],
82        )
83        .await?;
84
85    result
86        .first()
87        .and_then(|row| row.get("exists"))
88        .and_then(serde_json::Value::as_bool)
89        .ok_or_else(|| {
90            RepositoryError::Internal(format!(
91                "Failed to check column existence for '{table_name}.{column_name}'"
92            ))
93        })
94}
systemprompt_database/lifecycle/validation.rs

systemprompt_database/lifecycle/
validation.rs
