fraiseql_wire/client/
fraise_client.rs

1//! `FraiseClient` implementation
2
3use super::connection_string::{ConnectionInfo, TransportType};
4use super::query_builder::QueryBuilder;
5use crate::connection::{Connection, ConnectionConfig, Transport};
6#[allow(unused_imports)] // Reason: used only in doc links for `# Errors` sections
7use crate::error::WireError;
8use crate::stream::JsonStream;
9use crate::Result;
10use serde::de::DeserializeOwned;
11
12/// FraiseQL wire protocol client
13pub struct FraiseClient {
14    conn: Connection,
15}
16
17impl FraiseClient {
18    /// Connect to Postgres using connection string
19    ///
20    /// # Errors
21    ///
22    /// Returns [`WireError::Config`] if the connection string is invalid or missing required
23    /// fields. Returns [`WireError`] if the TCP or Unix socket connection fails, or if
24    /// startup/authentication is rejected by the server.
25    ///
26    /// # Examples
27    ///
28    /// ```no_run
29    /// // Requires: live Postgres server.
30    /// # async fn example() -> fraiseql_wire::Result<()> {
31    /// use fraiseql_wire::FraiseClient;
32    ///
33    /// // TCP connection
34    /// let client = FraiseClient::connect("postgres://localhost/mydb").await?;
35    ///
36    /// // Unix socket
37    /// let client = FraiseClient::connect("postgres:///mydb").await?;
38    /// # Ok(())
39    /// # }
40    /// ```
41    ///
42    /// # Errors
43    ///
44    /// Returns [`WireError::Config`] if the connection string is malformed or missing
45    /// required fields (host/port for TCP, path for Unix sockets).
46    /// Returns [`WireError::Io`] if the underlying TCP or Unix socket connection fails.
47    pub async fn connect(connection_string: &str) -> Result<Self> {
48        let info = ConnectionInfo::parse(connection_string)?;
49
50        let transport = match info.transport {
51            TransportType::Tcp => {
52                let host = info.host.as_ref().ok_or_else(|| {
53                    crate::WireError::Config("TCP transport requires a host".into())
54                })?;
55                let port = info.port.ok_or_else(|| {
56                    crate::WireError::Config("TCP transport requires a port".into())
57                })?;
58                Transport::connect_tcp(host, port).await?
59            }
60            TransportType::Unix => {
61                let path = info.unix_socket.as_ref().ok_or_else(|| {
62                    crate::WireError::Config("Unix transport requires a socket path".into())
63                })?;
64                Transport::connect_unix(path).await?
65            }
66        };
67
68        let mut conn = Connection::new(transport);
69        let config = info.to_config();
70        conn.startup(&config).await?;
71
72        Ok(Self { conn })
73    }
74
75    /// Connect to Postgres with TLS encryption
76    ///
77    /// TLS is configured independently from the connection string. The connection string
78    /// should contain the hostname and credentials (user/password), while TLS configuration
79    /// is provided separately via `TlsConfig`.
80    ///
81    /// # Errors
82    ///
83    /// Returns [`WireError::Config`] if the connection string is invalid, TLS is requested
84    /// over a Unix socket, or required fields are missing. Returns [`WireError::Io`] if the
85    /// TLS handshake or TCP connection fails.
86    ///
87    /// # Examples
88    ///
89    /// ```no_run
90    /// // Requires: live Postgres server with TLS.
91    /// # async fn example() -> fraiseql_wire::Result<()> {
92    /// use fraiseql_wire::{FraiseClient, connection::TlsConfig};
93    ///
94    /// // Configure TLS with system root certificates
95    /// let tls = TlsConfig::builder()
96    ///     .verify_hostname(true)
97    ///     .build()?;
98    ///
99    /// // Connect with TLS
100    /// let client = FraiseClient::connect_tls("postgres://secure.db.example.com/mydb", tls).await?;
101    /// # Ok(())
102    /// # }
103    /// ```
104    pub async fn connect_tls(
105        connection_string: &str,
106        tls_config: crate::connection::TlsConfig,
107    ) -> Result<Self> {
108        let info = ConnectionInfo::parse(connection_string)?;
109
110        let transport = match info.transport {
111            TransportType::Tcp => {
112                let host = info.host.as_ref().ok_or_else(|| {
113                    crate::WireError::Config("TCP transport requires a host".into())
114                })?;
115                let port = info.port.ok_or_else(|| {
116                    crate::WireError::Config("TCP transport requires a port".into())
117                })?;
118                Transport::connect_tcp_tls(host, port, &tls_config).await?
119            }
120            TransportType::Unix => {
121                return Err(crate::WireError::Config(
122                    "TLS is only supported for TCP connections".into(),
123                ));
124            }
125        };
126
127        let mut conn = Connection::new(transport);
128        let config = info.to_config();
129        conn.startup(&config).await?;
130
131        Ok(Self { conn })
132    }
133
134    /// Connect to Postgres with custom connection configuration
135    ///
136    /// This method allows you to configure timeouts, keepalive intervals, and other
137    /// connection options. The connection configuration is merged with parameters from
138    /// the connection string.
139    ///
140    /// # Errors
141    ///
142    /// Returns [`WireError::Config`] if the connection string is invalid or missing required
143    /// fields. Returns [`WireError::Io`] if the TCP or Unix socket connection fails, or if
144    /// startup/authentication is rejected by the server.
145    ///
146    /// # Examples
147    ///
148    /// ```no_run
149    /// // Requires: live Postgres server.
150    /// # async fn example() -> fraiseql_wire::Result<()> {
151    /// use fraiseql_wire::{FraiseClient, connection::ConnectionConfig};
152    /// use std::time::Duration;
153    ///
154    /// // Build connection configuration with timeouts
155    /// let config = ConnectionConfig::builder("localhost", "mydb")
156    ///     .password("secret")
157    ///     .statement_timeout(Duration::from_secs(30))
158    ///     .keepalive_idle(Duration::from_secs(300))
159    ///     .application_name("my_app")
160    ///     .build();
161    ///
162    /// // Connect with configuration
163    /// let client = FraiseClient::connect_with_config("postgres://localhost:5432/mydb", config).await?;
164    /// # Ok(())
165    /// # }
166    /// ```
167    pub async fn connect_with_config(
168        connection_string: &str,
169        config: ConnectionConfig,
170    ) -> Result<Self> {
171        let info = ConnectionInfo::parse(connection_string)?;
172
173        let transport = match info.transport {
174            TransportType::Tcp => {
175                let host = info.host.as_ref().ok_or_else(|| {
176                    crate::WireError::Config("TCP transport requires a host".into())
177                })?;
178                let port = info.port.ok_or_else(|| {
179                    crate::WireError::Config("TCP transport requires a port".into())
180                })?;
181                with_connect_timeout(config.connect_timeout, Transport::connect_tcp(host, port))
182                    .await?
183            }
184            TransportType::Unix => {
185                let path = info.unix_socket.as_ref().ok_or_else(|| {
186                    crate::WireError::Config("Unix transport requires a socket path".into())
187                })?;
188                with_connect_timeout(config.connect_timeout, Transport::connect_unix(path)).await?
189            }
190        };
191
192        // Apply TCP keepalive when configured.
193        if let Some(idle) = config.keepalive_idle {
194            if let Err(e) = transport.apply_keepalive(idle) {
195                tracing::warn!("Failed to apply TCP keepalive (idle={idle:?}): {e}");
196            }
197        }
198
199        let mut conn = Connection::new(transport);
200        conn.startup(&config).await?;
201
202        Ok(Self { conn })
203    }
204
205    /// Connect to Postgres with both custom configuration and TLS encryption
206    ///
207    /// This method combines connection configuration (timeouts, keepalive, etc.)
208    /// with TLS encryption for secure connections with advanced options.
209    ///
210    /// # Errors
211    ///
212    /// Returns [`WireError::Config`] if the connection string is invalid, TLS is requested
213    /// over a Unix socket, or required fields are missing. Returns [`WireError::Io`] if the
214    /// TLS handshake or TCP connection fails.
215    ///
216    /// # Examples
217    ///
218    /// ```no_run
219    /// // Requires: live Postgres server with TLS.
220    /// # async fn example() -> fraiseql_wire::Result<()> {
221    /// use fraiseql_wire::{FraiseClient, connection::{ConnectionConfig, TlsConfig}};
222    /// use std::time::Duration;
223    ///
224    /// // Configure connection with timeouts
225    /// let config = ConnectionConfig::builder("localhost", "mydb")
226    ///     .password("secret")
227    ///     .statement_timeout(Duration::from_secs(30))
228    ///     .build();
229    ///
230    /// // Configure TLS
231    /// let tls = TlsConfig::builder()
232    ///     .verify_hostname(true)
233    ///     .build()?;
234    ///
235    /// // Connect with both configuration and TLS
236    /// let client = FraiseClient::connect_with_config_and_tls(
237    ///     "postgres://secure.db.example.com/mydb",
238    ///     config,
239    ///     tls
240    /// ).await?;
241    /// # Ok(())
242    /// # }
243    /// ```
244    pub async fn connect_with_config_and_tls(
245        connection_string: &str,
246        config: ConnectionConfig,
247        tls_config: crate::connection::TlsConfig,
248    ) -> Result<Self> {
249        let info = ConnectionInfo::parse(connection_string)?;
250
251        let transport = match info.transport {
252            TransportType::Tcp => {
253                let host = info.host.as_ref().ok_or_else(|| {
254                    crate::WireError::Config("TCP transport requires a host".into())
255                })?;
256                let port = info.port.ok_or_else(|| {
257                    crate::WireError::Config("TCP transport requires a port".into())
258                })?;
259                with_connect_timeout(
260                    config.connect_timeout,
261                    Transport::connect_tcp_tls(host, port, &tls_config),
262                )
263                .await?
264            }
265            TransportType::Unix => {
266                return Err(crate::WireError::Config(
267                    "TLS is only supported for TCP connections".into(),
268                ));
269            }
270        };
271
272        // Apply TCP keepalive when configured.
273        if let Some(idle) = config.keepalive_idle {
274            if let Err(e) = transport.apply_keepalive(idle) {
275                tracing::warn!("Failed to apply TCP keepalive (idle={idle:?}): {e}");
276            }
277        }
278
279        let mut conn = Connection::new(transport);
280        conn.startup(&config).await?;
281
282        Ok(Self { conn })
283    }
284
285    /// Start building a query for an entity with automatic deserialization
286    ///
287    /// The type parameter T controls consumer-side deserialization only.
288    /// Type T does NOT affect SQL generation, filtering, ordering, or wire protocol.
289    ///
290    /// # Examples
291    ///
292    /// Type-safe query (recommended):
293    /// ```no_run
294    /// // Requires: live Postgres server.
295    /// # async fn example(client: fraiseql_wire::FraiseClient) -> fraiseql_wire::Result<()> {
296    /// use serde::Deserialize;
297    /// use futures::stream::StreamExt;
298    ///
299    /// #[derive(Deserialize)]
300    /// struct User {
301    ///     id: String,
302    ///     name: String,
303    /// }
304    ///
305    /// let mut stream = client
306    ///     .query::<User>("user")
307    ///     .where_sql("data->>'type' = 'customer'")  // SQL predicate
308    ///     .where_rust(|json| {
309    ///         // Rust predicate (applied client-side, on JSON)
310    ///         json["estimated_value"].as_f64().unwrap_or(0.0) > 1000.0
311    ///     })
312    ///     .order_by("data->>'name' ASC")
313    ///     .execute()
314    ///     .await?;
315    ///
316    /// while let Some(result) = stream.next().await {
317    ///     let user: User = result?;
318    ///     println!("User: {}", user.name);
319    /// }
320    /// # Ok(())
321    /// # }
322    /// ```
323    ///
324    /// Raw JSON query (debugging, forward compatibility):
325    /// ```no_run
326    /// // Requires: live Postgres server.
327    /// # async fn example(client: fraiseql_wire::FraiseClient) -> fraiseql_wire::Result<()> {
328    /// use futures::stream::StreamExt;
329    ///
330    /// let mut stream = client
331    ///     .query::<serde_json::Value>("user")  // Escape hatch
332    ///     .execute()
333    ///     .await?;
334    ///
335    /// while let Some(result) = stream.next().await {
336    ///     let json = result?;
337    ///     println!("JSON: {:?}", json);
338    /// }
339    /// # Ok(())
340    /// # }
341    /// ```
342    pub fn query<T: DeserializeOwned + std::marker::Unpin + 'static>(
343        self,
344        entity: impl Into<String>,
345    ) -> QueryBuilder<T> {
346        QueryBuilder::new(self, entity)
347    }
348
349    /// Execute a raw SQL query (must match fraiseql-wire constraints)
350    ///
351    /// The adaptive-chunking options are threaded through from the query builder
352    /// instead of being hardcoded off, so `adaptive_chunking`/`adaptive_min_size`/
353    /// `adaptive_max_size` actually take effect (audit L-wire-builder).
354    #[allow(clippy::too_many_arguments)] // Reason: mirrors streaming_query's chunking parameters; a struct would add allocation in the hot path
355    pub(crate) async fn execute_query(
356        self,
357        sql: &str,
358        chunk_size: usize,
359        max_memory: Option<usize>,
360        soft_limit_warn_threshold: Option<f32>,
361        soft_limit_fail_threshold: Option<f32>,
362        enable_adaptive_chunking: bool,
363        adaptive_min_chunk_size: Option<usize>,
364        adaptive_max_chunk_size: Option<usize>,
365    ) -> Result<JsonStream> {
366        self.conn
367            .streaming_query(
368                sql,
369                chunk_size,
370                max_memory,
371                soft_limit_warn_threshold,
372                soft_limit_fail_threshold,
373                enable_adaptive_chunking,
374                adaptive_min_chunk_size,
375                adaptive_max_chunk_size,
376            )
377            .await
378    }
379}
380
381/// Apply an optional connect timeout to a transport-connect future.
382///
383/// When `timeout` is `Some`, the future is bounded by [`tokio::time::timeout`] and a
384/// lapse surfaces as [`crate::WireError::Connection`]; when `None` the future runs
385/// to completion unbounded. The `connect_timeout` config field was parsed but never
386/// applied to the connect path (audit L-wire-timeout); the `connect_with_config*`
387/// methods now route their transport setup through this helper.
388async fn with_connect_timeout<F, T>(timeout: Option<std::time::Duration>, fut: F) -> Result<T>
389where
390    F: std::future::Future<Output = Result<T>>,
391{
392    match timeout {
393        Some(d) => match tokio::time::timeout(d, fut).await {
394            Ok(result) => result,
395            Err(_) => Err(crate::WireError::Connection(format!(
396                "connection timed out after {d:?}"
397            ))),
398        },
399        None => fut.await,
400    }
401}
402
403#[cfg(test)]
404mod tests;
fraiseql_wire/client/fraise_client.rs

fraiseql_wire/client/
fraise_client.rs
