ironflow_ops_postgres/
client.rs

1//! [`PostgresClient`] -- connection pool wrapper for PostgreSQL operations.
2
3use std::fmt;
4
5use ironflow_core::error::OperationError;
6use ironflow_core::operation::OperationContext;
7use sqlx::PgPool;
8use sqlx::postgres::PgPoolOptions;
9
10/// A PostgreSQL client wrapping a [`PgPool`].
11///
12/// Holds a connection pool that is shared across all operations created from
13/// this client. The pool is created once and reused for the lifetime of the
14/// client.
15///
16/// # Examples
17///
18/// ```no_run
19/// use ironflow_ops_postgres::PostgresClient;
20///
21/// # async fn example() -> Result<(), ironflow_core::error::OperationError> {
22/// let client = PostgresClient::connect("postgres://localhost/mydb").await?;
23/// let pool = client.pool();
24/// # Ok(())
25/// # }
26/// ```
27pub struct PostgresClient {
28    pool: PgPool,
29}
30
31impl PostgresClient {
32    /// Connect to a PostgreSQL database using the given connection URL.
33    ///
34    /// # Errors
35    ///
36    /// Returns [`OperationError::External`] if the connection pool cannot be
37    /// created (invalid URL, unreachable server, authentication failure).
38    ///
39    /// # Examples
40    ///
41    /// ```no_run
42    /// use ironflow_ops_postgres::PostgresClient;
43    ///
44    /// # async fn example() -> Result<(), ironflow_core::error::OperationError> {
45    /// let client = PostgresClient::connect("postgres://localhost/mydb").await?;
46    /// # Ok(())
47    /// # }
48    /// ```
49    pub async fn connect(url: &str) -> Result<Self, OperationError> {
50        if url.is_empty() {
51            return Err(OperationError::External {
52                origin: "postgres".to_string(),
53                message: "connection URL must not be empty".to_string(),
54            });
55        }
56        let pool = PgPoolOptions::new()
57            .max_connections(5)
58            .connect(url)
59            .await
60            .map_err(|e| OperationError::External {
61                origin: "postgres".to_string(),
62                message: e.to_string(),
63            })?;
64        Ok(Self { pool })
65    }
66
67    /// Create a client from an existing [`PgPool`].
68    ///
69    /// # Examples
70    ///
71    /// ```no_run
72    /// use ironflow_ops_postgres::PostgresClient;
73    /// use sqlx::PgPool;
74    ///
75    /// # async fn example() -> Result<(), sqlx::Error> {
76    /// let pool = PgPool::connect("postgres://localhost/mydb").await?;
77    /// let client = PostgresClient::from_pool(pool);
78    /// # Ok(())
79    /// # }
80    /// ```
81    pub fn from_pool(pool: PgPool) -> Self {
82        Self { pool }
83    }
84
85    /// Create a client by reading `postgres_url` from the secret store.
86    ///
87    /// # Errors
88    ///
89    /// Returns [`OperationError::Secret`] if the secret store fails, or
90    /// [`OperationError::External`] if the secret is missing or the connection
91    /// cannot be established.
92    ///
93    /// # Examples
94    ///
95    /// ```no_run
96    /// use ironflow_ops_postgres::PostgresClient;
97    /// use ironflow_core::operation::{OperationContext, NoopSecretResolver};
98    /// use std::sync::Arc;
99    ///
100    /// # async fn example() -> Result<(), ironflow_core::error::OperationError> {
101    /// let ctx = OperationContext::new(Arc::new(NoopSecretResolver));
102    /// let client = PostgresClient::from_context(&ctx).await?;
103    /// # Ok(())
104    /// # }
105    /// ```
106    pub async fn from_context(ctx: &OperationContext) -> Result<Self, OperationError> {
107        let secret = ctx.secrets().get("postgres_url").await?;
108        let url = secret
109            .ok_or_else(|| OperationError::External {
110                origin: "postgres".to_string(),
111                message: "secret 'postgres_url' not found".to_string(),
112            })?
113            .value;
114        Self::connect(&url).await
115    }
116
117    /// Returns a reference to the underlying connection pool.
118    pub fn pool(&self) -> &PgPool {
119        &self.pool
120    }
121}
122
123impl fmt::Debug for PostgresClient {
124    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
125        f.debug_struct("PostgresClient")
126            .field("url", &"[REDACTED]")
127            .finish()
128    }
129}
130
131#[cfg(test)]
132mod tests {
133    use std::sync::Arc;
134
135    use ironflow_core::operation::{NoopSecretResolver, OperationContext};
136
137    use super::*;
138
139    #[tokio::test]
140    async fn debug_does_not_leak() {
141        let client = PostgresClient {
142            pool: PgPool::connect_lazy("postgres://user:pass@localhost/db").unwrap(),
143        };
144        let debug = format!("{client:?}");
145        assert!(!debug.contains("user"), "leaked user: {debug}");
146        assert!(!debug.contains("pass"), "leaked password: {debug}");
147        assert!(!debug.contains("localhost"), "leaked host: {debug}");
148        assert!(debug.contains("REDACTED"), "missing redaction: {debug}");
149    }
150
151    #[tokio::test]
152    async fn connect_empty_url() {
153        let err = PostgresClient::connect("").await.unwrap_err();
154        let msg = err.to_string();
155        assert!(msg.contains("must not be empty"), "unexpected error: {msg}");
156    }
157
158    #[tokio::test]
159    async fn from_context_missing_secret() {
160        let ctx = OperationContext::new(Arc::new(NoopSecretResolver));
161        let err = PostgresClient::from_context(&ctx).await.unwrap_err();
162        let msg = err.to_string();
163        assert!(msg.contains("postgres_url"), "unexpected error: {msg}");
164    }
165}
ironflow_ops_postgres/client.rs

ironflow_ops_postgres/
client.rs
