systemprompt_database/repository/
change_context.rs

1//! The change context every config plane writer sets before it writes, and
2//! the cluster-wide config version a watcher polls.
3//!
4//! [`with_change_context`] sets `systemprompt.actor`, `systemprompt.reason`
5//! and `systemprompt.source` transaction-locally (`set_config(.., true)`), so
6//! they vanish at commit or rollback and can never leak to the next user of a
7//! pooled connection. The plane tables' `sp_stamp_change()` trigger copies
8//! them into `changed_by`, `reason` and `source`; a write made without them is
9//! recorded as unattributed. Call it on the transaction, never on a bare pool
10//! connection: outside a transaction the settings last one statement.
11//!
12//! [`PgConfigVersion`] reads the one `config_version` row that
13//! `sp_bump_config_version()` bumps on every committed plane write.
14//!
15//! [`PlaneTransaction`] names the transaction the host hands every config
16//! plane through `PlaneTx`, so a plane in a crate without a direct `sqlx`
17//! dependency can recover it.
18//!
19//! Copyright (c) systemprompt.io — Business Source License 1.1.
20//! See <https://systemprompt.io> for licensing details.
21
22use std::sync::Arc;
23
24use async_trait::async_trait;
25use sqlx::{PgConnection, PgPool};
26use systemprompt_traits::{ChangeActor, ChangeSource, ConfigVersionSource, RepositoryError};
27
28use crate::DbPool;
29
30pub type PlaneTransaction = sqlx::Transaction<'static, sqlx::Postgres>;
31
32pub async fn with_change_context(
33    conn: &mut PgConnection,
34    actor: &ChangeActor,
35    reason: Option<&str>,
36    source: &ChangeSource,
37) -> Result<(), RepositoryError> {
38    let source = source.to_string();
39    sqlx::query!(
40        "SELECT set_config('systemprompt.actor', $1, true) AS actor, \
41         set_config('systemprompt.reason', $2, true) AS reason, \
42         set_config('systemprompt.source', $3, true) AS source",
43        actor.label(),
44        reason.unwrap_or_default(),
45        source,
46    )
47    .fetch_one(conn)
48    .await?;
49    Ok(())
50}
51
52#[derive(Debug, Clone)]
53pub struct PgConfigVersion {
54    pool: Arc<PgPool>,
55}
56
57impl PgConfigVersion {
58    #[must_use]
59    pub fn new(db: &DbPool) -> Self {
60        Self {
61            pool: db.write_pool(),
62        }
63    }
64}
65
66#[async_trait]
67impl ConfigVersionSource for PgConfigVersion {
68    async fn current_version(&self) -> Result<i64, RepositoryError> {
69        let version = sqlx::query_scalar!("SELECT version FROM config_version WHERE id = 1")
70            .fetch_optional(&*self.pool)
71            .await?;
72        Ok(version.unwrap_or(0))
73    }
74}
systemprompt_database/repository/change_context.rs

systemprompt_database/repository/
change_context.rs
