fraiseql-db 2.2.0

//! SQL Server database adapter implementation.
use std::fmt::Write;

use async_trait::async_trait;
use bb8::Pool;
use bb8_tiberius::ConnectionManager;
use fraiseql_error::{FraiseQLError, Result};
use tiberius::Config;
use tracing;

use super::where_generator::SqlServerWhereGenerator;
use crate::{
    dialect::SqlServerDialect,
    identifier::quote_sqlserver_identifier,
    order_by::append_order_by,
    traits::{
        CursorValue, DatabaseAdapter, RelayDatabaseAdapter, RelayPageResult, SupportsMutations,
    },
    types::{
        DatabaseType, JsonbValue, PoolMetrics,
        sql_hints::{OrderByClause, OrderDirection},
    },
    where_clause::WhereClause,
};

/// Map an MSSQL server error code to the closest ANSI SQLSTATE string.
///
/// Returns `None` for codes that don't have a clear ANSI SQLSTATE mapping (e.g. MSSQL 701
/// "out of memory"), which is preferable to returning a vendor-namespaced or incorrect code.
///
/// | MSSQL Code | Meaning                         | SQLSTATE  |
/// |------------|---------------------------------|-----------|
/// | 2627       | Unique constraint violation     | 23505     |
/// | 2601       | Duplicate key (unique index)    | 23505     |
/// | 547        | Foreign key violation           | 23503     |
/// | 515        | NOT NULL violation              | 23502     |
/// | 1205       | Deadlock victim                 | 40001     |
/// | 8152       | String or binary data truncation| 22001     |
/// | 701        | Insufficient memory             | (unmapped)|
pub(super) fn map_mssql_error_code(code: u32) -> Option<String> {
    let sqlstate = match code {
        // Unique constraint / duplicate key → ANSI unique violation
        2627 | 2601 => "23505",
        // NOT NULL violation → ANSI not-null violation
        515 => "23502",
        // Foreign key violation → ANSI FK violation (unchanged)
        547 => "23503",
        // Deadlock → ANSI serialization failure (NOT PostgreSQL-vendor 40P01)
        1205 => "40001",
        // String truncation (unchanged)
        8152 => "22001",
        // 701 (out of memory) has no ANSI equivalent; emit None rather than a PostgreSQL code
        _ => return None,
    };
    Some(sqlstate.to_string())
}

/// SQL Server database adapter with connection pooling.
///
/// Uses `tiberius` for native TDS protocol support and `bb8` for connection pooling.
///
/// # Example
///
/// ```no_run
/// use fraiseql_db::sqlserver::SqlServerAdapter;
/// use fraiseql_db::{DatabaseAdapter, WhereClause, WhereOperator};
/// use serde_json::json;
///
/// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
/// // Create adapter with connection string
/// let adapter = SqlServerAdapter::new("server=localhost;database=mydb;user=sa;password=Password123").await?;
///
/// // Execute query
/// let where_clause = WhereClause::Field {
///     path: vec!["email".to_string()],
///     operator: WhereOperator::Icontains,
///     value: json!("example.com"),
/// };
///
/// let results = adapter
///     .execute_where_query("v_user", Some(&where_clause), Some(10), None, None)
///     .await?;
///
/// println!("Found {} users", results.len());
/// # Ok(())
/// # }
/// ```
#[derive(Clone)]
pub struct SqlServerAdapter {
    pool: Pool<ConnectionManager>,
}

impl SqlServerAdapter {
    /// Create new SQL Server adapter with default pool configuration.
    ///
    /// # Arguments
    ///
    /// * `connection_string` - SQL Server connection string (ADO.NET format)
    ///
    /// # Connection String Format
    ///
    /// ```text
    /// server=localhost;database=mydb;user=sa;password=Password123
    /// server=localhost,1433;database=mydb;integratedSecurity=true
    /// ```
    ///
    /// # Errors
    ///
    /// Returns `FraiseQLError::ConnectionPool` if pool creation fails.
    pub async fn new(connection_string: &str) -> Result<Self> {
        Self::with_pool_size(connection_string, 10).await
    }

    /// Create new SQL Server adapter with custom pool configuration.
    ///
    /// # Arguments
    ///
    /// * `connection_string` - SQL Server connection string
    /// * `min_size` - Minimum pool size (bb8 doesn't support this; connections are created
    ///   on-demand)
    /// * `max_size` - Maximum number of connections in pool
    ///
    /// # Note on min_size
    ///
    /// The bb8 connection pool used by this adapter creates connections on-demand
    /// rather than pre-creating a minimum pool size. The `min_size` parameter is accepted
    /// for API compatibility with other database adapters but does not affect behavior.
    /// If you need a minimum number of pre-created connections, consider:
    /// 1. Calling `warmup_connections()` after creating the pool
    /// 2. Increasing `max_size` to ensure sufficient capacity
    /// 3. Using a different connection pool strategy for your use case
    ///
    /// # Errors
    ///
    /// Returns `FraiseQLError::ConnectionPool` if pool creation fails.
    pub async fn with_pool_config(
        connection_string: &str,
        min_size: u32,
        max_size: u32,
    ) -> Result<Self> {
        if min_size > 0 {
            tracing::warn!(
                min_size,
                "SQL Server adapter does not support min_size parameter - connections are created \
                 on-demand. Consider warmup_connections() if pre-allocation is needed."
            );
        }
        Self::with_pool_size(connection_string, max_size).await
    }

    /// Create new SQL Server adapter with custom pool size.
    ///
    /// # Arguments
    ///
    /// * `connection_string` - SQL Server connection string
    /// * `max_size` - Maximum number of connections in pool
    ///
    /// # Errors
    ///
    /// Returns `FraiseQLError::ConnectionPool` if pool creation fails.
    pub async fn with_pool_size(connection_string: &str, max_size: u32) -> Result<Self> {
        // Parse connection string into tiberius Config
        let config = Config::from_ado_string(connection_string).map_err(|e| {
            FraiseQLError::ConnectionPool {
                message: format!("Invalid SQL Server connection string: {e}"),
            }
        })?;

        let manager = ConnectionManager::new(config);

        let pool = Pool::builder().max_size(max_size).build(manager).await.map_err(|e| {
            FraiseQLError::ConnectionPool {
                message: format!("Failed to create SQL Server connection pool: {e}"),
            }
        })?;

        // Test connection
        {
            let mut conn = pool.get().await.map_err(|e| FraiseQLError::ConnectionPool {
                message: format!("Failed to acquire connection: {e}"),
            })?;

            conn.simple_query("SELECT 1").await.map_err(|e| FraiseQLError::Database {
                message:   format!("Failed to connect to SQL Server database: {e}"),
                sql_state: None,
            })?;
        }

        Ok(Self { pool })
    }

    /// Execute raw SQL query and return JSONB rows.
    async fn execute_raw(
        &self,
        sql: &str,
        params: Vec<serde_json::Value>,
    ) -> Result<Vec<JsonbValue>> {
        let mut conn = self.pool.get().await.map_err(|e| FraiseQLError::ConnectionPool {
            message: format!("Failed to acquire connection: {e}"),
        })?;

        // Build and execute query
        // Note: tiberius doesn't support dynamic parameter binding like sqlx
        // We need to use simple_query for dynamic SQL or build the query differently
        let rows = if params.is_empty() {
            let result = conn.simple_query(sql).await.map_err(|e| FraiseQLError::Database {
                message:   format!("SQL Server query execution failed: {e}"),
                sql_state: e.code().and_then(map_mssql_error_code),
            })?;
            result.into_first_result().await.map_err(|e| FraiseQLError::Database {
                message:   format!("Failed to get result set: {e}"),
                sql_state: e.code().and_then(map_mssql_error_code),
            })?
        } else {
            // For parameterized queries, use typed parameter binding.
            let string_params = serialise_complex_params(&params);
            let mut query = tiberius::Query::new(sql);
            bind_json_params(&mut query, &params, &string_params)?;

            let result = query.query(&mut *conn).await.map_err(|e| FraiseQLError::Database {
                message:   format!("SQL Server query execution failed: {e}"),
                sql_state: e.code().and_then(map_mssql_error_code),
            })?;
            result.into_first_result().await.map_err(|e| FraiseQLError::Database {
                message:   format!("Failed to get result set: {e}"),
                sql_state: e.code().and_then(map_mssql_error_code),
            })?
        };

        // Process result set
        let mut results = Vec::new();

        for row in rows {
            // Try to get 'data' column as string and parse as JSON
            if let Some(data_str) = row.try_get::<&str, _>("data").ok().flatten() {
                let data: serde_json::Value =
                    serde_json::from_str(data_str).unwrap_or(serde_json::Value::Null);
                results.push(JsonbValue::new(data));
            } else {
                results.push(JsonbValue::new(serde_json::Value::Null));
            }
        }

        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 SqlServerAdapter {
    /// # Errors
    ///
    /// Returns [`FraiseQLError::Database`] if the SQL Server query fails or the
    /// result cannot be deserialized as JSONB.
    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>> {
        // If no projection provided, fall back to standard query
        if projection.is_none() {
            return self.execute_where_query(view, where_clause, limit, offset, order_by).await;
        }

        let Some(projection) = projection else {
            // Reason: unreachable — `is_none()` check above returns early
            unreachable!("projection is Some; None case returned above");
        };

        // Build SQL with SQL Server-specific JSON projection.
        // TOP and OFFSET...FETCH are mutually exclusive in T-SQL, so only use
        // TOP when there is no OFFSET and no ORDER BY (otherwise OFFSET...FETCH
        // handles both, and ORDER BY requires the OFFSET...FETCH form).
        let needs_offset_fetch = offset.is_some() || order_by.is_some_and(|c| !c.is_empty());
        let mut sql = if let Some(lim) = limit {
            if needs_offset_fetch {
                format!(
                    "SELECT {} FROM {}",
                    projection.projection_template,
                    quote_sqlserver_identifier(view)
                )
            } else {
                format!(
                    "SELECT TOP {} {} FROM {}",
                    lim,
                    projection.projection_template,
                    quote_sqlserver_identifier(view)
                )
            }
        } else {
            format!(
                "SELECT {} FROM {}",
                projection.projection_template,
                quote_sqlserver_identifier(view)
            )
        };

        // Add WHERE clause if present
        let params: Vec<serde_json::Value> = if let Some(clause) = where_clause {
            let generator = super::where_generator::SqlServerWhereGenerator::new(SqlServerDialect);
            let (where_sql, where_params) = generator.generate(clause)?;
            sql.push_str(" WHERE ");
            sql.push_str(&where_sql);
            where_params
        } else {
            Vec::new()
        };

        // ORDER BY + pagination for SQL Server.
        // T-SQL requires ORDER BY for OFFSET...FETCH. When the user provides
        // order_by, use it; otherwise fall back to `ORDER BY (SELECT NULL)`.
        // Reason (expect below): fmt::Write for String is infallible.
        let has_order = append_order_by(&mut sql, order_by, DatabaseType::SQLServer)?;
        if needs_offset_fetch {
            if !has_order {
                sql.push_str(" ORDER BY (SELECT NULL)");
            }
            let off = offset.unwrap_or(0);
            write!(sql, " OFFSET {off} ROWS").expect("write to String");
            if let Some(lim) = limit {
                write!(sql, " FETCH NEXT {lim} ROWS ONLY").expect("write to String");
            }
        }

        // Execute the query
        self.execute_raw(&sql, params).await
    }

    /// # Errors
    ///
    /// Returns [`FraiseQLError::Validation`] if WHERE clause generation fails, or
    /// [`FraiseQLError::Database`] if the SQL Server query 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>> {
        // Build base query — SQL Server uses square brackets for identifiers.
        // TOP and OFFSET...FETCH are mutually exclusive in T-SQL.
        // When ORDER BY is provided, always use OFFSET...FETCH (even without offset)
        // so the ordering is honoured with LIMIT.
        let needs_offset_fetch = offset.is_some() || order_by.is_some_and(|c| !c.is_empty());
        let mut sql = if let Some(lim) = limit {
            if needs_offset_fetch {
                format!("SELECT data FROM {}", quote_sqlserver_identifier(view))
            } else {
                format!("SELECT TOP {lim} data FROM {}", quote_sqlserver_identifier(view))
            }
        } else {
            format!("SELECT data FROM {}", quote_sqlserver_identifier(view))
        };

        // Add WHERE clause if present
        let (mut params, mut param_count): (Vec<serde_json::Value>, usize) =
            if let Some(clause) = where_clause {
                let generator = SqlServerWhereGenerator::new(SqlServerDialect);
                let (where_sql, where_params) = generator.generate(clause)?;
                sql.push_str(" WHERE ");
                sql.push_str(&where_sql);
                let len = where_params.len();
                (where_params, len)
            } else {
                (Vec::new(), 0)
            };

        // ORDER BY + pagination for SQL Server.
        // T-SQL requires ORDER BY for OFFSET...FETCH. When the user provides
        // order_by, use it; otherwise fall back to `ORDER BY (SELECT NULL)`.
        // Reason (expect below): fmt::Write for String is infallible.
        let has_order = append_order_by(&mut sql, order_by, DatabaseType::SQLServer)?;
        if needs_offset_fetch {
            if !has_order {
                sql.push_str(" ORDER BY (SELECT NULL)");
            }
            let off = offset.unwrap_or(0);
            param_count += 1;
            write!(sql, " OFFSET @p{param_count} ROWS").expect("write to String");
            params.push(serde_json::Value::Number(off.into()));
            if let Some(lim) = limit {
                param_count += 1;
                write!(sql, " FETCH NEXT @p{param_count} ROWS ONLY").expect("write to String");
                params.push(serde_json::Value::Number(lim.into()));
            }
        } else if has_order && limit.is_some() {
            // ORDER BY without OFFSET — SQL Server needs OFFSET 0 for FETCH
            param_count += 1;
            write!(sql, " OFFSET 0 ROWS FETCH NEXT @p{param_count} ROWS ONLY")
                .expect("write to String");
            params.push(serde_json::Value::Number(limit.expect("checked above").into()));
        }

        self.execute_raw(&sql, params).await
    }

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

    async fn health_check(&self) -> Result<()> {
        let mut conn = self.pool.get().await.map_err(|e| FraiseQLError::ConnectionPool {
            message: format!("Failed to acquire connection: {e}"),
        })?;

        conn.simple_query("SELECT 1").await.map_err(|e| FraiseQLError::Database {
            message:   format!("SQL Server health check failed: {e}"),
            sql_state: None,
        })?;

        Ok(())
    }

    fn pool_metrics(&self) -> PoolMetrics {
        let state = self.pool.state();

        PoolMetrics {
            total_connections:  state.connections,
            idle_connections:   state.idle_connections,
            active_connections: state.connections - state.idle_connections,
            waiting_requests:   0, // bb8 doesn't expose waiting count directly
        }
    }

    /// # Security
    ///
    /// `sql` **must** be compiler-generated. Never pass user-supplied strings
    /// directly — doing so would open SQL-injection vulnerabilities.
    async fn execute_raw_query(
        &self,
        sql: &str,
    ) -> Result<Vec<std::collections::HashMap<String, serde_json::Value>>> {
        let mut conn = self.pool.get().await.map_err(|e| FraiseQLError::ConnectionPool {
            message: format!("Failed to acquire connection: {e}"),
        })?;

        let result = conn.simple_query(sql).await.map_err(|e| FraiseQLError::Database {
            message:   format!("SQL Server query execution failed: {e}"),
            sql_state: e.code().and_then(map_mssql_error_code),
        })?;

        let rows = result.into_first_result().await.map_err(|e| FraiseQLError::Database {
            message:   format!("Failed to get result set: {e}"),
            sql_state: e.code().and_then(map_mssql_error_code),
        })?;

        // Convert each row to HashMap<String, Value>
        let results: Vec<std::collections::HashMap<String, serde_json::Value>> = rows
            .into_iter()
            .map(|row| {
                let mut map = std::collections::HashMap::new();

                // Iterate over all columns in the row
                for (idx, column) in row.columns().iter().enumerate() {
                    let column_name = column.name().to_string();

                    // Try to extract value based on SQL Server type
                    let value: serde_json::Value =
                        if let Some(v) = row.try_get::<i32, _>(idx).ok().flatten() {
                            serde_json::json!(v)
                        } else if let Some(v) = row.try_get::<i64, _>(idx).ok().flatten() {
                            serde_json::json!(v)
                        } else if let Some(v) = row.try_get::<f64, _>(idx).ok().flatten() {
                            serde_json::json!(v)
                        } else if let Some(v) = row.try_get::<&str, _>(idx).ok().flatten() {
                            // Try to parse as JSON first
                            if let Ok(json_val) = serde_json::from_str::<serde_json::Value>(v) {
                                json_val
                            } else {
                                serde_json::json!(v)
                            }
                        } else if let Some(v) = row.try_get::<bool, _>(idx).ok().flatten() {
                            serde_json::json!(v)
                        } else {
                            // Fallback: NULL
                            serde_json::Value::Null
                        };

                    map.insert(column_name, value);
                }

                map
            })
            .collect();

        Ok(results)
    }

    async fn execute_parameterized_aggregate(
        &self,
        sql: &str,
        params: &[serde_json::Value],
    ) -> Result<Vec<std::collections::HashMap<String, serde_json::Value>>> {
        let mut conn = self.pool.get().await.map_err(|e| FraiseQLError::ConnectionPool {
            message: format!("Failed to acquire connection: {e}"),
        })?;

        let string_params = serialise_complex_params(params);
        let mut query = tiberius::Query::new(sql);
        bind_json_params(&mut query, params, &string_params)?;

        let result = query.query(&mut *conn).await.map_err(|e| FraiseQLError::Database {
            message:   format!("SQL Server parameterized aggregate query failed: {e}"),
            sql_state: e.code().and_then(map_mssql_error_code),
        })?;

        let rows = result.into_first_result().await.map_err(|e| FraiseQLError::Database {
            message:   format!("Failed to get aggregate result set: {e}"),
            sql_state: e.code().and_then(map_mssql_error_code),
        })?;

        let results = rows
            .into_iter()
            .map(|row| {
                let mut map = std::collections::HashMap::new();
                for (idx, column) in row.columns().iter().enumerate() {
                    let col = column.name().to_string();
                    let value: serde_json::Value =
                        if let Some(v) = row.try_get::<i32, _>(idx).ok().flatten() {
                            serde_json::json!(v)
                        } else if let Some(v) = row.try_get::<i64, _>(idx).ok().flatten() {
                            serde_json::json!(v)
                        } else if let Some(v) = row.try_get::<f64, _>(idx).ok().flatten() {
                            serde_json::json!(v)
                        } else if let Some(v) = row.try_get::<bool, _>(idx).ok().flatten() {
                            serde_json::json!(v)
                        } else if let Some(s) = row.try_get::<&str, _>(idx).ok().flatten() {
                            serde_json::from_str(s).unwrap_or_else(|_| serde_json::json!(s))
                        } else {
                            serde_json::Value::Null
                        };
                    map.insert(col, value);
                }
                map
            })
            .collect();

        Ok(results)
    }

    async fn execute_function_call(
        &self,
        function_name: &str,
        args: &[serde_json::Value],
    ) -> Result<Vec<std::collections::HashMap<String, serde_json::Value>>> {
        // Build: EXEC [fn_name] @p1, @p2, ...
        // Stored procedures in SQL Server must be called with EXEC, not SELECT FROM.
        let placeholders: Vec<String> = (1..=args.len()).map(|i| format!("@p{i}")).collect();
        let sql = format!(
            "EXEC {}{}",
            quote_sqlserver_identifier(function_name),
            if placeholders.is_empty() {
                String::new()
            } else {
                format!(" {}", placeholders.join(", "))
            }
        );

        let mut conn = self.pool.get().await.map_err(|e| FraiseQLError::ConnectionPool {
            message: format!("Failed to acquire connection: {e}"),
        })?;

        // Bind args; non-primitive types are serialised to JSON strings.
        let string_params = serialise_complex_params(args);
        let mut query = tiberius::Query::new(sql);
        bind_json_params(&mut query, args, &string_params)?;

        let result = query.query(&mut *conn).await.map_err(|e| FraiseQLError::Database {
            message:   format!("SQL Server function call failed ({function_name}): {e}"),
            sql_state: e.code().and_then(map_mssql_error_code),
        })?;

        let rows = result.into_first_result().await.map_err(|e| FraiseQLError::Database {
            message:   format!("Failed to get result set from {function_name}: {e}"),
            sql_state: e.code().and_then(map_mssql_error_code),
        })?;

        let results = rows
            .into_iter()
            .map(|row| {
                let mut map = std::collections::HashMap::new();
                for (idx, column) in row.columns().iter().enumerate() {
                    let col = column.name().to_string();
                    let value: serde_json::Value =
                        if let Some(v) = row.try_get::<i32, _>(idx).ok().flatten() {
                            serde_json::json!(v)
                        } else if let Some(v) = row.try_get::<i64, _>(idx).ok().flatten() {
                            serde_json::json!(v)
                        } else if let Some(v) = row.try_get::<f64, _>(idx).ok().flatten() {
                            serde_json::json!(v)
                        } else if let Some(v) = row.try_get::<bool, _>(idx).ok().flatten() {
                            serde_json::json!(v)
                        } else if let Some(s) = row.try_get::<&str, _>(idx).ok().flatten() {
                            serde_json::from_str(s).unwrap_or_else(|_| serde_json::json!(s))
                        } else {
                            serde_json::Value::Null
                        };
                    map.insert(col, value);
                }
                map
            })
            .collect();

        Ok(results)
    }
}

// ============================================================================
// Parameter binding helpers
// ============================================================================

/// Pre-serialise `Array` and `Object` values so their string forms live long enough
/// to be referenced by `bind_json_params`.
fn serialise_complex_params(params: &[serde_json::Value]) -> Vec<String> {
    params
        .iter()
        .filter(|v| matches!(v, serde_json::Value::Array(_) | serde_json::Value::Object(_)))
        .map(|v| v.to_string())
        .collect()
}

/// Bind `serde_json::Value` parameters to a tiberius `Query`.
///
/// `string_params` must be the output of `serialise_complex_params` for the same
/// `params` slice.  The caller allocates this before creating the `Query` so the
/// strings live long enough for the lifetime `'a`.
///
/// # Errors
///
/// Returns `FraiseQLError::Validation` if a `Number` value cannot be represented
/// as either `i64` or `f64` (this is extremely rare and indicates an out-of-range
/// numeric literal in the GraphQL input).
fn bind_json_params<'a>(
    query: &mut tiberius::Query<'a>,
    params: &'a [serde_json::Value],
    string_params: &'a [String],
) -> Result<()> {
    let mut string_idx = 0usize;
    for param in params {
        match param {
            serde_json::Value::String(s) => query.bind(s.as_str()),
            serde_json::Value::Number(n) => {
                if let Some(i) = n.as_i64() {
                    query.bind(i);
                } else if let Some(f) = n.as_f64() {
                    query.bind(f);
                } else {
                    return Err(FraiseQLError::Validation {
                        message: format!("Cannot bind numeric value {n}: out of i64 and f64 range"),
                        path:    None,
                    });
                }
            },
            serde_json::Value::Bool(b) => query.bind(*b),
            serde_json::Value::Null => query.bind(Option::<&str>::None),
            serde_json::Value::Array(_) | serde_json::Value::Object(_) => {
                query.bind(string_params[string_idx].as_str());
                string_idx += 1;
            },
        }
    }
    Ok(())
}

// ============================================================================
// Relay cursor pagination
// ============================================================================

/// Custom sort columns come first, then the cursor column as tiebreaker.
/// For backward pagination every direction is flipped so the inner `FETCH NEXT` subquery
/// retrieves the correct `N` rows before the cursor; the outer re-sort in
/// `build_relay_backward_outer_order_sql` then restores the original order.
fn build_relay_order_sql(
    quoted_col: &str,
    order_by: Option<&[OrderByClause]>,
    forward: bool,
) -> String {
    if let Some(clauses) = order_by {
        let mut parts: Vec<String> = clauses
            .iter()
            .map(|c| {
                // Flip every custom sort direction for backward pages so the inner
                // DESC subquery fetches the correct N rows before the cursor.
                let dir = match (c.direction, forward) {
                    (OrderDirection::Asc, true) => "ASC",
                    (OrderDirection::Asc, false) => "DESC",
                    (OrderDirection::Desc, true) => "DESC",
                    (OrderDirection::Desc, false) => "ASC",
                };
                // Field names are validated as GraphQL identifiers (no single quotes).
                format!("JSON_VALUE(data, '$.{}') {dir}", c.field)
            })
            .collect();
        let primary_dir = if forward { "ASC" } else { "DESC" };
        parts.push(format!("{quoted_col} {primary_dir}"));
        format!(" ORDER BY {}", parts.join(", "))
    } else {
        let dir = if forward { "ASC" } else { "DESC" };
        format!(" ORDER BY {quoted_col} {dir}")
    }
}

/// Build the outer ORDER BY for the backward-page re-sort wrapper.
///
/// Uses the *original* (non-flipped) sort directions and references the aliased
/// sort columns (`_relay_sort_0`, `_relay_sort_1`, …) projected by the inner query.
/// The cursor column is always the final tiebreaker, re-sorted ASC to present rows
/// in ascending cursor order as required by the Relay spec.
fn build_relay_backward_outer_order_sql(order_by: Option<&[OrderByClause]>) -> String {
    if let Some(clauses) = order_by {
        let mut parts: Vec<String> = clauses
            .iter()
            .enumerate()
            .map(|(i, c)| {
                let dir = match c.direction {
                    OrderDirection::Asc => "ASC",
                    OrderDirection::Desc => "DESC",
                };
                format!("_relay_sort_{i} {dir}")
            })
            .collect();
        parts.push("_relay_cursor ASC".to_string());
        format!(" ORDER BY {}", parts.join(", "))
    } else {
        " ORDER BY _relay_cursor ASC".to_string()
    }
}

/// Validate that a string looks like a UUID without pulling in regex.
///
/// Accepts the canonical `xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx` form (36 chars).
/// Returns `false` for any other format so callers can emit a clean
/// `FraiseQLError::Validation` instead of letting SQL Server produce an opaque
/// type-conversion error (MSSQL 8169).
fn is_valid_uuid_format(uuid: &str) -> bool {
    let parts: Vec<&str> = uuid.split('-').collect();
    matches!(
        parts.as_slice(),
        [p0, p1, p2, p3, p4]
            if p0.len() == 8
            && p1.len() == 4
            && p2.len() == 4
            && p3.len() == 4
            && p4.len() == 12
            && uuid.chars().all(|ch| ch.is_ascii_hexdigit() || ch == '-')
    )
}

/// Build the WHERE portion of a relay page query.
fn build_relay_where_sql(cursor_part: Option<&str>, user_part: Option<&str>) -> String {
    match (cursor_part, user_part) {
        (None, None) => String::new(),
        (Some(c), None) => format!(" WHERE {c}"),
        (None, Some(u)) => format!(" WHERE ({u})"),
        (Some(c), Some(u)) => format!(" WHERE {c} AND ({u})"),
    }
}

impl SupportsMutations for SqlServerAdapter {}

impl RelayDatabaseAdapter for SqlServerAdapter {
    /// Execute keyset (cursor-based) pagination against a JSONB view.
    ///
    /// Uses `OFFSET 0 ROWS FETCH NEXT n ROWS ONLY` with mandatory `ORDER BY`.
    /// Backward pagination wraps a DESC inner query in an outer ASC sort.
    ///
    /// UUID cursors are compared via `CONVERT(UNIQUEIDENTIFIER, @pN)`.
    async fn execute_relay_page(
        &self,
        view: &str,
        cursor_column: &str,
        after: Option<CursorValue>,
        before: Option<CursorValue>,
        limit: u32,
        forward: bool,
        where_clause: Option<&WhereClause>,
        order_by: Option<&[OrderByClause]>,
        include_total_count: bool,
    ) -> Result<RelayPageResult> {
        let quoted_view = quote_sqlserver_identifier(view);
        let quoted_col = quote_sqlserver_identifier(cursor_column);

        // ── Cursor condition ─────────────────────────────────────────────────
        let active_cursor = if forward { after } else { before };
        let (cursor_param, cursor_where_part): (Option<serde_json::Value>, Option<String>) =
            match active_cursor {
                None => (None, None),
                Some(CursorValue::Int64(pk)) => {
                    let op = if forward { ">" } else { "<" };
                    (Some(serde_json::json!(pk)), Some(format!("{quoted_col} {op} @p1")))
                },
                Some(CursorValue::Uuid(uuid)) => {
                    // Validate format before sending to SQL Server; malformed UUIDs produce an
                    // opaque conversion error (MSSQL 8169) rather than a useful validation message.
                    if !is_valid_uuid_format(&uuid) {
                        return Err(FraiseQLError::Validation {
                            message: format!("Invalid UUID cursor value: '{uuid}'"),
                            path:    None,
                        });
                    }
                    let op = if forward { ">" } else { "<" };
                    (
                        Some(serde_json::json!(uuid)),
                        Some(format!("{quoted_col} {op} CONVERT(UNIQUEIDENTIFIER, @p1)")),
                    )
                },
            };
        let cursor_param_count: usize = usize::from(cursor_param.is_some());

        // ── User WHERE clause ────────────────────────────────────────────────
        let mut user_where_params: Vec<serde_json::Value> = Vec::new();
        let page_user_where_sql: Option<String> = if let Some(clause) = where_clause {
            let generator = SqlServerWhereGenerator::new(SqlServerDialect);
            let (sql, params) = generator.generate_with_param_offset(clause, cursor_param_count)?;
            user_where_params = params;
            Some(sql)
        } else {
            None
        };
        let user_param_count = user_where_params.len();

        // ── ORDER BY and WHERE strings ────────────────────────────────────────
        let order_sql = build_relay_order_sql(&quoted_col, order_by, forward);
        let page_where_sql =
            build_relay_where_sql(cursor_where_part.as_deref(), page_user_where_sql.as_deref());

        // ── LIMIT parameter index ─────────────────────────────────────────────
        let limit_idx = cursor_param_count + user_param_count + 1;

        // ── Page SQL ──────────────────────────────────────────────────────────
        //
        // SQL Server requires ORDER BY for OFFSET…FETCH. The mandatory
        // `OFFSET 0 ROWS` prelude cannot be elided.
        //
        // Backward pagination: the inner query uses flipped sort directions + FETCH so it
        // retrieves the correct N rows before the cursor.  The inner query also projects each
        // custom sort column under an alias (`_relay_sort_0`, …) so the outer re-sort can
        // reference them.  The outer query restores the original (non-flipped) sort order.
        let page_sql = if forward {
            format!(
                "SELECT data FROM {quoted_view}{page_where_sql}{order_sql} \
                 OFFSET 0 ROWS FETCH NEXT @p{limit_idx} ROWS ONLY"
            )
        } else {
            // Project each custom sort column under a stable alias for the outer re-sort.
            let sort_aliases: String = order_by.unwrap_or(&[]).iter().enumerate().fold(
                String::new(),
                |mut acc, (i, c)| {
                    use std::fmt::Write as _;
                    let _ = write!(acc, ", JSON_VALUE(data, '$.{}') AS _relay_sort_{i}", c.field);
                    acc
                },
            );

            let inner = format!(
                "SELECT data, {quoted_col} AS _relay_cursor{sort_aliases} \
                 FROM {quoted_view}{page_where_sql}{order_sql} \
                 OFFSET 0 ROWS FETCH NEXT @p{limit_idx} ROWS ONLY"
            );
            let outer_order = build_relay_backward_outer_order_sql(order_by);
            format!("SELECT data FROM ({inner}) AS _relay_page{outer_order}")
        };

        // ── Page params: [cursor?, user_where..., limit] ──────────────────────
        let mut page_params: Vec<serde_json::Value> = Vec::new();
        if let Some(cp) = cursor_param {
            page_params.push(cp);
        }
        page_params.extend_from_slice(&user_where_params);
        page_params.push(serde_json::json!(limit));

        // ── Execute page query ────────────────────────────────────────────────
        let rows = self.execute_raw(&page_sql, page_params).await?;

        // ── Count query (Relay spec: totalCount ignores cursor position) ──────
        //
        // Re-generates the WHERE clause with offset 0 (no cursor prefix) because
        // the count is a standalone query.
        let total_count = if include_total_count {
            let (count_sql, count_params) = if let Some(clause) = where_clause {
                let generator = SqlServerWhereGenerator::new(SqlServerDialect);
                let (where_sql, params) = generator.generate_with_param_offset(clause, 0)?;
                (
                    format!("SELECT COUNT_BIG(*) AS cnt FROM {quoted_view} WHERE ({where_sql})"),
                    params,
                )
            } else {
                (format!("SELECT COUNT_BIG(*) AS cnt FROM {quoted_view}"), vec![])
            };

            let mut conn = self.pool.get().await.map_err(|e| FraiseQLError::ConnectionPool {
                message: format!("Failed to acquire connection for relay count: {e}"),
            })?;

            // Serialise complex types up-front so their string refs live long enough.
            let count_string_params = serialise_complex_params(&count_params);
            let mut count_query = tiberius::Query::new(&count_sql);
            bind_json_params(&mut count_query, &count_params, &count_string_params)?;

            let count_result =
                count_query.query(&mut *conn).await.map_err(|e| FraiseQLError::Database {
                    message:   format!("SQL Server relay count query failed: {e}"),
                    sql_state: e.code().and_then(map_mssql_error_code),
                })?;

            let count_rows =
                count_result.into_first_result().await.map_err(|e| FraiseQLError::Database {
                    message:   format!("Failed to get relay count result set: {e}"),
                    sql_state: e.code().and_then(map_mssql_error_code),
                })?;

            // A missing or empty row must surface as an error rather than silently
            // reporting `totalCount: 0`, which would be a misleading data-loss trap.
            let n: i64 = count_rows
                .first()
                .and_then(|row| row.try_get::<i64, _>(0).ok().flatten())
                .ok_or_else(|| FraiseQLError::Database {
                    message:   format!("Relay count query returned no rows for view '{view}'"),
                    sql_state: None,
                })?;
            let count = u64::try_from(n).map_err(|_| FraiseQLError::Database {
                message:   format!(
                    "Relay count query returned negative value ({n}) for view '{view}'"
                ),
                sql_state: None,
            })?;

            Some(count)
        } else {
            None
        };

        Ok(RelayPageResult { rows, total_count })
    }
}

#[cfg(test)]
mod error_code_tests {
    use super::*;

    #[test]
    fn test_unique_constraint_violation_2627() {
        // 2627 = unique constraint → ANSI unique violation 23505
        assert_eq!(map_mssql_error_code(2627), Some("23505".to_string()));
    }

    #[test]
    fn test_duplicate_key_2601() {
        // 2601 = duplicate key in unique index → same ANSI code 23505
        assert_eq!(map_mssql_error_code(2601), Some("23505".to_string()));
    }

    #[test]
    fn test_not_null_violation_515() {
        // 515 = NOT NULL violation → ANSI not-null violation 23502
        assert_eq!(map_mssql_error_code(515), Some("23502".to_string()));
    }

    #[test]
    fn test_foreign_key_violation_547() {
        // 547 = FK violation → ANSI FK violation 23503 (unchanged)
        assert_eq!(map_mssql_error_code(547), Some("23503".to_string()));
    }

    #[test]
    fn test_deadlock_1205() {
        // 1205 = deadlock victim → ANSI serialization failure 40001 (NOT PostgreSQL-vendor 40P01)
        assert_eq!(map_mssql_error_code(1205), Some("40001".to_string()));
    }

    #[test]
    fn test_string_truncation_8152() {
        // 8152 = string truncation → ANSI value too long 22001 (unchanged)
        assert_eq!(map_mssql_error_code(8152), Some("22001".to_string()));
    }

    #[test]
    fn test_out_of_memory_701_returns_none() {
        // 701 = insufficient memory — no ANSI equivalent; must return None
        // (previously incorrectly returned the PostgreSQL-vendor code "53200")
        assert_eq!(map_mssql_error_code(701), None);
    }

    #[test]
    fn test_unknown_code_returns_none() {
        assert_eq!(map_mssql_error_code(9999), None);
        assert_eq!(map_mssql_error_code(0), None);
        assert_eq!(map_mssql_error_code(u32::MAX), None);
    }
}

#[cfg(test)]
mod relay_sql_tests {
    use super::*;
    use crate::types::sql_hints::{OrderByClause, OrderDirection};

    // ── build_relay_order_sql ──────────────────────────────────────────────

    #[test]
    fn test_build_relay_order_sql_forward_no_order_by() {
        let sql = build_relay_order_sql("[id]", None, true);
        assert_eq!(sql, " ORDER BY [id] ASC");
    }

    #[test]
    fn test_build_relay_order_sql_backward_no_order_by() {
        let sql = build_relay_order_sql("[id]", None, false);
        assert_eq!(sql, " ORDER BY [id] DESC");
    }

    #[test]
    fn test_build_relay_order_sql_forward_custom_order_by_asc() {
        let order_by = vec![OrderByClause::new("score".to_string(), OrderDirection::Asc)];
        let sql = build_relay_order_sql("[id]", Some(&order_by), true);
        assert_eq!(sql, " ORDER BY JSON_VALUE(data, '$.score') ASC, [id] ASC");
    }

    #[test]
    fn test_build_relay_order_sql_backward_custom_order_by_asc_flips_to_desc() {
        // KEY TEST: backward pagination must flip ASC → DESC so the inner
        // FETCH NEXT subquery retrieves the correct N rows before the cursor.
        let order_by = vec![OrderByClause::new("score".to_string(), OrderDirection::Asc)];
        let sql = build_relay_order_sql("[id]", Some(&order_by), false);
        assert_eq!(sql, " ORDER BY JSON_VALUE(data, '$.score') DESC, [id] DESC");
    }

    #[test]
    fn test_build_relay_order_sql_backward_custom_order_by_desc_flips_to_asc() {
        let order_by = vec![OrderByClause::new(
            "created_at".to_string(),
            OrderDirection::Desc,
        )];
        let sql = build_relay_order_sql("[id]", Some(&order_by), false);
        assert_eq!(sql, " ORDER BY JSON_VALUE(data, '$.created_at') ASC, [id] DESC");
    }

    #[test]
    fn test_build_relay_order_sql_multi_column_forward() {
        let order_by = vec![
            OrderByClause::new("a".to_string(), OrderDirection::Asc),
            OrderByClause::new("b".to_string(), OrderDirection::Desc),
        ];
        let sql = build_relay_order_sql("[id]", Some(&order_by), true);
        assert_eq!(
            sql,
            " ORDER BY JSON_VALUE(data, '$.a') ASC, JSON_VALUE(data, '$.b') DESC, [id] ASC"
        );
    }

    #[test]
    fn test_build_relay_order_sql_multi_column_backward_all_flipped() {
        let order_by = vec![
            OrderByClause::new("a".to_string(), OrderDirection::Asc),
            OrderByClause::new("b".to_string(), OrderDirection::Desc),
        ];
        let sql = build_relay_order_sql("[id]", Some(&order_by), false);
        assert_eq!(
            sql,
            " ORDER BY JSON_VALUE(data, '$.a') DESC, JSON_VALUE(data, '$.b') ASC, [id] DESC"
        );
    }

    // ── build_relay_backward_outer_order_sql ──────────────────────────────

    #[test]
    fn test_build_relay_backward_outer_order_sql_no_order_by() {
        let sql = build_relay_backward_outer_order_sql(None);
        assert_eq!(sql, " ORDER BY _relay_cursor ASC");
    }

    #[test]
    fn test_build_relay_backward_outer_order_sql_with_custom_asc() {
        let order_by = vec![OrderByClause::new("score".to_string(), OrderDirection::Asc)];
        let sql = build_relay_backward_outer_order_sql(Some(&order_by));
        assert_eq!(sql, " ORDER BY _relay_sort_0 ASC, _relay_cursor ASC");
    }

    #[test]
    fn test_build_relay_backward_outer_order_sql_desc_preserved() {
        let order_by = vec![OrderByClause::new(
            "score".to_string(),
            OrderDirection::Desc,
        )];
        let sql = build_relay_backward_outer_order_sql(Some(&order_by));
        assert_eq!(sql, " ORDER BY _relay_sort_0 DESC, _relay_cursor ASC");
    }

    // ── build_relay_where_sql ─────────────────────────────────────────────

    #[test]
    fn test_build_relay_where_sql_none_none() {
        let sql = build_relay_where_sql(None, None);
        assert_eq!(sql, "");
    }

    #[test]
    fn test_build_relay_where_sql_cursor_only() {
        let sql = build_relay_where_sql(Some("cur > @p1"), None);
        assert_eq!(sql, " WHERE cur > @p1");
    }

    #[test]
    fn test_build_relay_where_sql_user_only() {
        let sql = build_relay_where_sql(None, Some("user_filter"));
        assert_eq!(sql, " WHERE (user_filter)");
    }

    #[test]
    fn test_build_relay_where_sql_both() {
        let sql = build_relay_where_sql(Some("cur > @p1"), Some("user_filter"));
        assert_eq!(sql, " WHERE cur > @p1 AND (user_filter)");
    }

    // ── is_valid_uuid_format ──────────────────────────────────────────────

    #[test]
    fn test_is_valid_uuid_format_accepts_valid_uuid() {
        assert!(is_valid_uuid_format("550e8400-e29b-41d4-a716-446655440000"));
    }

    #[test]
    fn test_is_valid_uuid_format_rejects_malformed() {
        assert!(!is_valid_uuid_format("not-a-uuid"));
        assert!(!is_valid_uuid_format("550e8400-e29b-41d4-a716")); // too short
        assert!(!is_valid_uuid_format("550e8400-e29b-41d4-a716-44665544000Z")); // invalid char
    }

    #[test]
    fn test_is_valid_uuid_format_rejects_empty() {
        assert!(!is_valid_uuid_format(""));
    }
}

#[cfg(all(test, feature = "test-sqlserver"))]
mod tests {
    use super::*;

    // Note: These tests require a running SQL Server instance with test data.
    // Run with: cargo test --features test-sqlserver -p fraiseql-core db::sqlserver::adapter

    const TEST_DB_URL: &str = "server=localhost,1434;database=fraiseql_test;user=sa;password=FraiseQL_Test1234;TrustServerCertificate=true";

    #[tokio::test]
    async fn test_adapter_creation() {
        let adapter = SqlServerAdapter::new(TEST_DB_URL)
            .await
            .expect("Failed to create SQL Server adapter");

        let metrics = adapter.pool_metrics();
        assert!(metrics.total_connections > 0);
        assert_eq!(adapter.database_type(), DatabaseType::SQLServer);
    }

    #[tokio::test]
    async fn test_health_check() {
        let adapter = SqlServerAdapter::new(TEST_DB_URL)
            .await
            .expect("Failed to create SQL Server adapter");

        adapter.health_check().await.expect("Health check failed");
    }

    #[tokio::test]
    async fn test_parameterized_limit_and_offset() {
        let adapter = SqlServerAdapter::new(TEST_DB_URL)
            .await
            .expect("Failed to create SQL Server adapter");

        // SQL Server requires ORDER BY for OFFSET...FETCH
        // This test just ensures parameterization works
        let results = adapter
            .execute_where_query("v_user", None, Some(2), Some(1), None)
            .await
            .expect("Failed to execute query");

        assert!(results.len() <= 2);
    }
}