fraiseql_db/

fraiseql_wire_adapter.rs

//! FraiseQL-Wire adapter implementation.
//!
//! This adapter integrates fraiseql-wire as an alternative database backend,
//! providing streaming JSON queries with low memory overhead.

use std::collections::HashMap;

use async_trait::async_trait;
use fraiseql_error::{FraiseQLError, Result};
use futures::stream::StreamExt;

use super::{
    order_by::{Tiebreak, render_order_by_columns},
    traits::DatabaseAdapter,
    types::{DatabaseType, JsonbValue, PoolMetrics, sql_hints::OrderByClause},
    where_clause::WhereClause,
    where_sql_generator::WhereSqlGenerator,
    wire_pool::WireClientFactory,
};

/// The bare `ORDER BY` column list for the wire builder, or a named refusal.
///
/// This adapter assembles raw SQL — `WhereSqlGenerator` escapes values into the
/// string because the wire protocol path has no bind parameters, and that
/// generator's own documentation says not to use it in new code for exactly
/// that reason. So an ordering that binds a value has nowhere to put it here.
///
/// Today that is full-text relevance (#1284), and the refusal is consistent
/// with the WHERE side: this backend already refuses every full-text *operator*
/// (`where_sql_generator.rs`), so a `?search=` request cannot reach this
/// function in the first place. It is refused rather than escaped because the
/// alternative — interpolating a client's search string into an `ORDER BY` —
/// is the one thing the parameter channel exists to prevent.
///
/// # Errors
///
/// Returns `FraiseQLError::Validation` when a clause binds a parameter, or when
/// field-name validation fails.
fn order_by_columns(order_by: Option<&[OrderByClause]>) -> Result<Option<String>> {
    if order_by.is_some_and(|cs| cs.iter().any(OrderByClause::binds_parameter)) {
        return Err(FraiseQLError::validation(
            "full-text relevance ordering binds the search text as a parameter, which the              fraiseql-wire adapter cannot supply: it assembles raw SQL. This backend already              refuses full-text search predicates for the same reason.",
        ));
    }
    // `Tiebreak::Identity`: this adapter slices LIMIT/OFFSET in memory, but each
    // call re-runs the query, so consecutive pages are two orderings of tied rows
    // exactly as on the SQL path (#1287).
    Ok(
        render_order_by_columns(order_by, DatabaseType::PostgreSQL, 1, Tiebreak::Identity)?
            .map(|r| r.columns),
    )
}

/// FraiseQL-Wire database adapter.
///
/// Uses fraiseql-wire for streaming JSON queries with bounded memory usage.
/// This adapter is optimized for read-heavy workloads with large result sets.
///
/// # Streaming reads
///
/// This adapter inherits the trait's *buffering* default for
/// [`stream_with_projection`](DatabaseAdapter::stream_with_projection), so its rows
/// are correct but materialised — the same shape the collecting methods below already
/// have. That is a deliberate non-implementation, not an oversight: the underlying
/// `QueryStream` really does stream, and threading it through would be a small change,
/// but no CI leg executes this adapter against a real database, so the result could
/// not be proven — only compiled. See #1115.
///
/// # Architecture
///
/// - Connection Factory: Creates fresh clients on demand
/// - Streaming: Results are streamed incrementally (O(chunk_size) memory)
/// - WHERE Translation: AST → SQL via `WhereSqlGenerator`
///
/// # Example
///
/// ```ignore
/// use fraiseql_core::db::{FraiseWireAdapter, WhereClause, WhereOperator, DatabaseAdapter};
/// use serde_json::json;
///
/// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
/// // Create adapter
/// let adapter = FraiseWireAdapter::new("postgres://localhost/mydb");
///
/// // Build WHERE clause
/// let where_clause = WhereClause::Field {
///     path: vec!["status".to_string()],
///     operator: WhereOperator::Eq,
///     value: json!("active"),
/// };
///
/// // Execute query
/// let results = adapter
///     .execute_where_query("v_user", Some(&where_clause), Some(10), None, None)
///     .await?;
///
/// println!("Found {} users", results.len());
/// # Ok(())
/// # }
/// ```
#[derive(Debug, Clone)]
pub struct FraiseWireAdapter {
    factory:    WireClientFactory,
    chunk_size: usize,
}

impl FraiseWireAdapter {
    /// Create a new FraiseWire adapter.
    ///
    /// # Arguments
    ///
    /// * `connection_string` - PostgreSQL connection string
    ///
    /// # Example
    ///
    /// ```ignore
    /// use fraiseql_core::db::FraiseWireAdapter;
    ///
    /// let adapter = FraiseWireAdapter::new("postgres://localhost/fraiseql");
    /// ```
    #[must_use]
    pub fn new(connection_string: impl Into<String>) -> Self {
        Self {
            factory:    WireClientFactory::new(connection_string),
            chunk_size: 1024, // Default chunk size
        }
    }

    /// Set chunk size for streaming queries.
    ///
    /// Larger chunk sizes increase throughput but use more memory.
    /// Smaller chunk sizes reduce memory usage but may decrease throughput.
    ///
    /// # Arguments
    ///
    /// * `chunk_size` - Number of rows per chunk (default: 1024)
    ///
    /// # Example
    ///
    /// ```ignore
    /// use fraiseql_core::db::FraiseWireAdapter;
    ///
    /// let adapter = FraiseWireAdapter::new("postgres://localhost/fraiseql")
    ///     .with_chunk_size(512);
    /// ```
    #[must_use = "builder method returns modified builder"]
    pub const fn with_chunk_size(mut self, chunk_size: usize) -> Self {
        self.chunk_size = chunk_size;
        self
    }

    /// Execute manual query with in-memory LIMIT/OFFSET slicing.
    ///
    /// fraiseql-wire's `QueryBuilder` does not expose LIMIT/OFFSET, so this collects the
    /// (optionally ordered + filtered) stream and applies the window in memory. `order_by`
    /// is pushed down to the wire builder so the slice is taken over correctly-sorted rows.
    ///
    /// # Errors
    ///
    /// Returns [`FraiseQLError::Validation`] if SQL generation fails, or
    /// [`FraiseQLError::Database`] if the wire connection or query execution fails.
    async fn execute_manual_query(
        &self,
        view: &str,
        where_clause: Option<&WhereClause>,
        limit: Option<u32>,
        offset: Option<u32>,
        order_by: Option<&[OrderByClause]>,
    ) -> Result<Vec<JsonbValue>> {
        // Create fresh client
        let client = self.factory.create_client().await?;

        // fraiseql-wire doesn't expose LIMIT/OFFSET on its QueryBuilder, so we collect the
        // full (filtered + ordered) stream and slice the window in memory below. ORDER BY is
        // pushed down so the slice is taken over sorted rows, not DB-native order.

        // Pass view name directly - fraiseql-wire now uses entity names as-is (fixed in commit
        // 6c78e30)
        let mut builder = client.query::<serde_json::Value>(view).chunk_size(self.chunk_size);

        if let Some(clause) = where_clause {
            let where_sql = WhereSqlGenerator::to_sql(clause)?;
            builder = builder.where_sql(where_sql);
        }

        // FraiseWire is PostgreSQL-only (see `database_type`); render the bare ORDER BY
        // column list (the wire builder supplies the `ORDER BY` keyword itself).
        if let Some(columns) = order_by_columns(order_by)? {
            builder = builder.order_by(columns);
        }

        let mut stream = builder.execute().await.map_err(|e| FraiseQLError::Database {
            message:   format!("fraiseql-wire query failed: {e}"),
            sql_state: None,
        })?;

        // Collect all results
        let mut results = Vec::new();
        let offset_usize = offset.unwrap_or(0) as usize;
        let limit_usize = limit.map(|l| l as usize);

        let mut count = 0;
        while let Some(item) = stream.next().await {
            let json = item.map_err(|e| FraiseQLError::Database {
                message:   format!("Stream error: {e}"),
                sql_state: None,
            })?;

            // Apply offset and limit manually
            if count >= offset_usize {
                results.push(JsonbValue::new(json));

                if let Some(lim) = limit_usize {
                    if results.len() >= lim {
                        break;
                    }
                }
            }
            count += 1;
        }

        Ok(results)
    }
}

// Reason: DatabaseAdapter is defined with #[async_trait]; all implementations must match
// its transformed method signatures to satisfy the trait contract
// async_trait: dyn-dispatch required; remove when RTN + Send is stable (RFC 3425)
#[async_trait]
impl DatabaseAdapter for FraiseWireAdapter {
    /// # Errors
    ///
    /// Returns [`FraiseQLError::Database`] if the wire connection or query execution fails.
    async fn execute_with_projection(
        &self,
        view: &str,
        _projection: Option<&crate::types::SqlProjectionHint>,
        where_clause: Option<&WhereClause>,
        limit: Option<u32>,
        offset: Option<u32>,
        order_by: Option<&[OrderByClause]>,
    ) -> Result<Vec<JsonbValue>> {
        // FraiseWire uses streaming, so projection is handled by field selection
        // Fall back to standard query for now
        self.execute_where_query(view, where_clause, limit, offset, order_by).await
    }

    /// # Errors
    ///
    /// Returns [`FraiseQLError::Validation`] if SQL generation fails, or
    /// [`FraiseQLError::Database`] if the wire connection or query execution fails.
    async fn execute_where_query(
        &self,
        view: &str,
        where_clause: Option<&WhereClause>,
        limit: Option<u32>,
        offset: Option<u32>,
        order_by: Option<&[OrderByClause]>,
    ) -> Result<Vec<JsonbValue>> {
        // fraiseql-wire generates SQL as: SELECT data FROM {entity}
        // where entity is used exactly as provided (no prefix modifications)
        let entity = view;

        // Add LIMIT and OFFSET if provided
        // Note: fraiseql-wire QueryBuilder doesn't expose limit/offset, so these cases
        // collect-and-slice in memory (order_by is pushed down there too).
        if limit.is_some() || offset.is_some() {
            return self.execute_manual_query(view, where_clause, limit, offset, order_by).await;
        }

        // Create fresh client
        let client = self.factory.create_client().await?;

        // Start building query
        let mut builder = client.query::<serde_json::Value>(entity).chunk_size(self.chunk_size);

        // Add WHERE clause if provided
        if let Some(clause) = where_clause {
            let where_sql = WhereSqlGenerator::to_sql(clause)?;
            builder = builder.where_sql(where_sql);
        }

        // Honor ORDER BY (audit #442: previously dropped — relay/keyset pagination over the
        // wire backend silently returned DB-native order). FraiseWire is PostgreSQL-only.
        if let Some(columns) = order_by_columns(order_by)? {
            builder = builder.order_by(columns);
        }

        // Execute streaming query
        let mut stream = builder.execute().await.map_err(|e| FraiseQLError::Database {
            message:   format!("fraiseql-wire query failed: {e}"),
            sql_state: None,
        })?;

        // Collect results
        let mut results = Vec::new();
        while let Some(item) = stream.next().await {
            let json = item.map_err(|e| FraiseQLError::Database {
                message:   format!("Stream error: {e}"),
                sql_state: None,
            })?;
            results.push(JsonbValue::new(json));
        }

        Ok(results)
    }

    fn database_type(&self) -> DatabaseType {
        DatabaseType::PostgreSQL
    }

    /// Stated rather than inherited, with the reason: a composed read carries every
    /// embedded level's RLS predicate in its `WHERE`, and this adapter binds no parameters —
    /// it escapes values into literal SQL. Composing here would run authorization
    /// predicates through string escaping; binding parameters comes first.
    fn supports_composed_reads(&self) -> bool {
        false
    }

    async fn health_check(&self) -> Result<()> {
        // fraiseql-wire's FraiseClient contains non-Send types (raw pointers in TLS),
        // which makes it incompatible with the async_trait Send requirement.
        // As a workaround, we just verify the connection string is non-empty.
        // Actual connectivity is verified when queries are executed.
        if self.factory.connection_string().is_empty() {
            return Err(FraiseQLError::Database {
                message:   "Connection string is empty".to_string(),
                sql_state: None,
            });
        }
        Ok(())
    }

    fn pool_metrics(&self) -> PoolMetrics {
        // fraiseql-wire doesn't pool connections, so metrics are not applicable
        PoolMetrics {
            total_connections:  0,
            idle_connections:   0,
            active_connections: 0,
            waiting_requests:   0,
        }
    }

    /// # Errors
    ///
    /// Always returns [`FraiseQLError::Database`] — arbitrary SQL is not supported
    /// by fraiseql-wire; use [`Self::execute_where_query`] instead.
    async fn execute_raw_query(
        &self,
        _sql: &str,
    ) -> Result<Vec<HashMap<String, serde_json::Value>>> {
        // fraiseql-wire doesn't support arbitrary SQL queries
        // It only supports SELECT data FROM v_{entity} WHERE ... queries
        // This limitation is intentional as per fraiseql-wire's design
        Err(FraiseQLError::Database {
            message: "fraiseql-wire does not support arbitrary SQL queries. Use execute_where_query instead.".to_string(),
            sql_state: None,
        })
    }

    /// # Errors
    ///
    /// Always returns [`FraiseQLError::Database`] — aggregate SQL is not supported
    /// by fraiseql-wire.
    async fn execute_parameterized_aggregate(
        &self,
        _sql: &str,
        _params: &[serde_json::Value],
    ) -> Result<Vec<HashMap<String, serde_json::Value>>> {
        // fraiseql-wire does not support aggregate queries with arbitrary SQL.
        Err(FraiseQLError::Database {
            message:   "fraiseql-wire does not support aggregate SQL queries.".to_string(),
            sql_state: None,
        })
    }

    async fn query_stats(&self, _limit: u32) -> Result<Vec<crate::types::QueryStatEntry>> {
        // fraiseql-wire does not have access to database-level query statistics.
        Ok(vec![])
    }

    async fn query_stats_by_id(&self, _id: &str) -> Result<Option<crate::types::QueryStatEntry>> {
        Ok(None)
    }
}

#[cfg(test)]
mod tests;