systemprompt_database/services/postgres/
transaction.rs

1//! Concrete [`DatabaseTransaction`] for `PostgreSQL`.
2//!
3//! Part of the documented sqlx allowlist — the SQL strings here come from
4//! runtime-supplied [`QuerySelector`] values.
5//!
6//! Copyright (c) systemprompt.io — Business Source License 1.1.
7//! See <https://systemprompt.io> for licensing details.
8
9use crate::error::{DatabaseResult, RepositoryError};
10use async_trait::async_trait;
11
12use super::conversion::{bind_params, row_to_json};
13use crate::models::{DatabaseTransaction, JsonRow, QuerySelector, ToDbValue};
14
15pub struct PostgresTransaction {
16    tx: Option<sqlx::Transaction<'static, sqlx::Postgres>>,
17}
18
19impl std::fmt::Debug for PostgresTransaction {
20    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
21        f.debug_struct("PostgresTransaction")
22            .field("tx", &self.tx.is_some())
23            .finish()
24    }
25}
26
27impl PostgresTransaction {
28    #[must_use]
29    pub const fn new(tx: sqlx::Transaction<'static, sqlx::Postgres>) -> Self {
30        Self { tx: Some(tx) }
31    }
32}
33
34#[async_trait]
35impl DatabaseTransaction for PostgresTransaction {
36    async fn execute(
37        &mut self,
38        query: &dyn QuerySelector,
39        params: &[&dyn ToDbValue],
40    ) -> DatabaseResult<u64> {
41        let sql = query.select_query();
42        let tx = self
43            .tx
44            .as_mut()
45            .ok_or_else(|| RepositoryError::invalid_state("Transaction already consumed"))?;
46
47        let query_obj = sqlx::query(sqlx::AssertSqlSafe(sql));
48        let query_obj = bind_params(query_obj, params);
49
50        let result = query_obj.execute(&mut **tx).await?;
51
52        Ok(result.rows_affected())
53    }
54
55    async fn fetch_all(
56        &mut self,
57        query: &dyn QuerySelector,
58        params: &[&dyn ToDbValue],
59    ) -> DatabaseResult<Vec<JsonRow>> {
60        let sql = query.select_query();
61        let tx = self
62            .tx
63            .as_mut()
64            .ok_or_else(|| RepositoryError::invalid_state("Transaction already consumed"))?;
65
66        let query_obj = sqlx::query(sqlx::AssertSqlSafe(sql));
67        let query_obj = bind_params(query_obj, params);
68
69        let rows = query_obj.fetch_all(&mut **tx).await?;
70
71        Ok(rows.iter().map(row_to_json).collect())
72    }
73
74    async fn fetch_one(
75        &mut self,
76        query: &dyn QuerySelector,
77        params: &[&dyn ToDbValue],
78    ) -> DatabaseResult<JsonRow> {
79        let sql = query.select_query();
80        let tx = self
81            .tx
82            .as_mut()
83            .ok_or_else(|| RepositoryError::invalid_state("Transaction already consumed"))?;
84
85        let query_obj = sqlx::query(sqlx::AssertSqlSafe(sql));
86        let query_obj = bind_params(query_obj, params);
87
88        let row = query_obj.fetch_one(&mut **tx).await?;
89
90        Ok(row_to_json(&row))
91    }
92
93    async fn fetch_optional(
94        &mut self,
95        query: &dyn QuerySelector,
96        params: &[&dyn ToDbValue],
97    ) -> DatabaseResult<Option<JsonRow>> {
98        let sql = query.select_query();
99        let tx = self
100            .tx
101            .as_mut()
102            .ok_or_else(|| RepositoryError::invalid_state("Transaction already consumed"))?;
103
104        let query_obj = sqlx::query(sqlx::AssertSqlSafe(sql));
105        let query_obj = bind_params(query_obj, params);
106
107        let row = query_obj.fetch_optional(&mut **tx).await?;
108
109        Ok(row.map(|r| row_to_json(&r)))
110    }
111
112    async fn commit(mut self: Box<Self>) -> DatabaseResult<()> {
113        let tx = self
114            .tx
115            .take()
116            .ok_or_else(|| RepositoryError::invalid_state("Transaction already consumed"))?;
117
118        tx.commit().await?;
119
120        Ok(())
121    }
122
123    async fn rollback(mut self: Box<Self>) -> DatabaseResult<()> {
124        let tx = self
125            .tx
126            .take()
127            .ok_or_else(|| RepositoryError::invalid_state("Transaction already consumed"))?;
128
129        tx.rollback().await?;
130
131        Ok(())
132    }
133}
systemprompt_database/services/postgres/transaction.rs

systemprompt_database/services/postgres/
transaction.rs
