athena_rs 3.6.1

//! SqlX error parser for converting database errors into user-friendly ProcessedError responses.
//!
//! This module handles:
//! - PostgreSQL-specific error codes
//! - Connection and pool errors
//! - Query execution errors
//! - Type conversion errors

use super::sql_sanitizer::ExtractedInfo;
use actix_web::http::StatusCode;
use serde_json::{Map, json};
use sqlx::error::DatabaseError;
use sqlx::postgres::{PgDatabaseError, PgErrorPosition};

use super::hints;
use super::sql_sanitizer::{extract_metadata, sanitize_error_message};
use super::{ErrorCategory, ProcessedError, generate_trace_id};

/// Converts a sqlx::Error into a ProcessedError with sanitized messages and metadata.
///
/// This function:
/// - Categorizes the error type
/// - Extracts safe metadata (constraint names, SQL state codes)
/// - Sanitizes error messages to remove SQL queries
/// - Generates user-friendly messages and hints
/// - Assigns appropriate HTTP status codes
pub fn process_sqlx_error(err: &sqlx::Error) -> ProcessedError {
    process_sqlx_error_with_context_and_columns(err, None, None)
}

/// Converts a sqlx::Error with additional context into a ProcessedError.
///
/// The context parameter can provide table names or other safe identifiers
/// that help make the error message more specific.
pub fn process_sqlx_error_with_context(err: &sqlx::Error, context: Option<&str>) -> ProcessedError {
    process_sqlx_error_with_context_and_columns(err, context, None)
}

/// Converts a sqlx::Error with optional context and known table columns.
pub fn process_sqlx_error_with_context_and_columns(
    err: &sqlx::Error,
    context: Option<&str>,
    available_columns: Option<&[String]>,
) -> ProcessedError {
    let trace_id: String = generate_trace_id();

    match err {
        sqlx::Error::Database(db_err) => {
            process_database_error(db_err.as_ref(), &trace_id, context, available_columns)
        }
        sqlx::Error::PoolTimedOut => create_connection_error(
            "pool_timeout",
            "Database connection pool timed out",
            hints::POOL_TIMEOUT,
            &trace_id,
        ),
        sqlx::Error::PoolClosed => create_connection_error(
            "pool_closed",
            "Database connection pool is closed",
            hints::POOL_CLOSED,
            &trace_id,
        ),
        sqlx::Error::WorkerCrashed => create_internal_error(
            "worker_crashed",
            "Database worker thread crashed",
            &trace_id,
        ),
        sqlx::Error::Io(io_err) => create_connection_error(
            "io_error",
            "Database I/O error",
            &format!("Network or I/O issue occurred: {}", io_err.kind()),
            &trace_id,
        ),
        sqlx::Error::Tls(tls_err) => create_connection_error(
            "tls_error",
            "Database TLS connection error",
            &format!(
                "Secure connection failed: {}",
                sanitize_error_message(&tls_err.to_string())
            ),
            &trace_id,
        ),
        sqlx::Error::ColumnNotFound(column) => {
            let mut error: ProcessedError = ProcessedError::new(
                ErrorCategory::QuerySyntax,
                StatusCode::INTERNAL_SERVER_ERROR,
                "column_not_found",
                format!("Column '{}' not found in query results", column),
                trace_id,
            )
            .with_hint(hints::COLUMN_NOT_FOUND)
            .with_metadata("column", json!(column));

            if let Some(ctx) = context {
                error = error.with_metadata("context", json!(ctx));
            }

            error
        }
        sqlx::Error::ColumnIndexOutOfBounds { index, len } => ProcessedError::new(
            ErrorCategory::Internal,
            StatusCode::INTERNAL_SERVER_ERROR,
            "column_index_out_of_bounds",
            format!(
                "Column index {} is out of bounds (total columns: {})",
                index, len
            ),
            trace_id,
        )
        .with_metadata("index", json!(index))
        .with_metadata("column_count", json!(len)),
        sqlx::Error::TypeNotFound { type_name } => ProcessedError::new(
            ErrorCategory::Internal,
            StatusCode::INTERNAL_SERVER_ERROR,
            "type_not_found",
            format!("Database type '{}' not recognized", type_name),
            trace_id,
        )
        .with_metadata("type_name", json!(type_name)),
        sqlx::Error::ColumnDecode { index, source } => {
            let sanitized_msg: String = sanitize_error_message(&source.to_string());
            ProcessedError::new(
                ErrorCategory::Internal,
                StatusCode::INTERNAL_SERVER_ERROR,
                "column_decode_error",
                format!(
                    "Failed to decode column at index {}: {}",
                    index, sanitized_msg
                ),
                trace_id,
            )
            .with_metadata("column_index", json!(index))
        }
        sqlx::Error::RowNotFound => ProcessedError::new(
            ErrorCategory::NotFound,
            StatusCode::NOT_FOUND,
            "row_not_found",
            "The requested record was not found",
            trace_id,
        )
        .with_hint(hints::ROW_NOT_FOUND),
        sqlx::Error::Migrate(_) => ProcessedError::new(
            ErrorCategory::Internal,
            StatusCode::INTERNAL_SERVER_ERROR,
            "migration_error",
            "Database migration error",
            trace_id,
        ),
        sqlx::Error::Configuration(config_err) => {
            let sanitized_msg = sanitize_error_message(&config_err.to_string());
            ProcessedError::new(
                ErrorCategory::Internal,
                StatusCode::INTERNAL_SERVER_ERROR,
                "configuration_error",
                format!("Database configuration error: {}", sanitized_msg),
                trace_id,
            )
        }
        _ => {
            // Fallback for any unhandled error types
            let sanitized_msg: String = sanitize_error_message(&err.to_string());
            ProcessedError::new(
                ErrorCategory::Internal,
                StatusCode::INTERNAL_SERVER_ERROR,
                "database_error",
                format!("Database operation failed: {}", sanitized_msg),
                trace_id,
            )
        }
    }
}

/// Processes database-specific errors (PostgreSQL errors with codes).
fn process_database_error(
    db_err: &dyn DatabaseError,
    trace_id: &str,
    context: Option<&str>,
    available_columns: Option<&[String]>,
) -> ProcessedError {
    let code: Option<String> = db_err.code().map(|c| c.to_string());
    let message: &str = db_err.message();
    let constraint: Option<&str> = db_err.constraint();
    let pg_err: Option<&PgDatabaseError> = db_err.try_downcast_ref::<PgDatabaseError>();

    // Extract metadata from the error message
    let extracted: ExtractedInfo = extract_metadata(message);
    let mut metadata: Map<String, serde_json::Value> = Map::new();

    if let Some(sql_code) = &code {
        metadata.insert("sql_state".to_string(), json!(sql_code));
    }

    if let Some(constraint_name) = constraint.or(extracted.constraint_name.as_deref()) {
        metadata.insert("constraint".to_string(), json!(constraint_name));
    }

    if let Some(column_name) = pg_err
        .and_then(PgDatabaseError::column)
        .or(extracted.column_name.as_deref())
    {
        metadata.insert("column".to_string(), json!(column_name));
    }

    if let Some(table_name) = pg_err
        .and_then(PgDatabaseError::table)
        .or(db_err.table())
        .or(extracted.table_name.as_deref())
    {
        metadata.insert("table".to_string(), json!(table_name));
    }

    if let Some(schema_name) = pg_err.and_then(PgDatabaseError::schema) {
        metadata.insert("schema".to_string(), json!(schema_name));
    }

    if let Some(detail) = pg_err
        .and_then(PgDatabaseError::detail)
        .map(sanitize_error_message)
    {
        metadata.insert("detail".to_string(), json!(detail));
    }

    if let Some(position) = pg_err.and_then(PgDatabaseError::position) {
        match position {
            PgErrorPosition::Original(position) => {
                metadata.insert("position".to_string(), json!(position));
            }
            PgErrorPosition::Internal { position, .. } => {
                metadata.insert("position".to_string(), json!(position));
            }
        }
    }

    if let Some(ctx) = context {
        metadata.insert("context".to_string(), json!(ctx));
    }

    if let Some(columns) = available_columns {
        metadata.insert("available_columns".to_string(), json!(columns));
    }

    // Map SQL state codes to user-friendly errors
    match code.as_deref() {
        // Unique violation
        Some("23505") => ProcessedError::new(
            ErrorCategory::DatabaseConstraint,
            StatusCode::CONFLICT,
            "unique_violation",
            "A record with this value already exists",
            trace_id,
        )
        .with_hint(hints::UNIQUE_VIOLATION)
        .with_metadata_map(metadata),

        // Foreign key violation
        Some("23503") => ProcessedError::new(
            ErrorCategory::DatabaseConstraint,
            StatusCode::UNPROCESSABLE_ENTITY,
            "foreign_key_violation",
            "The operation references a non-existent or inaccessible resource",
            trace_id,
        )
        .with_hint(hints::FOREIGN_KEY_VIOLATION)
        .with_metadata_map(metadata),

        // Not null violation
        Some("23502") => ProcessedError::new(
            ErrorCategory::Validation,
            StatusCode::BAD_REQUEST,
            "not_null_violation",
            "A required field is missing",
            trace_id,
        )
        .with_hint(hints::NOT_NULL_VIOLATION)
        .with_metadata_map(metadata),

        // Check constraint violation
        Some("23514") => ProcessedError::new(
            ErrorCategory::DatabaseConstraint,
            StatusCode::BAD_REQUEST,
            "check_constraint_violation",
            "The provided value does not meet the required constraints",
            trace_id,
        )
        .with_hint(hints::CHECK_CONSTRAINT_VIOLATION)
        .with_metadata_map(metadata),

        // Invalid text encoding (for example, non-UTF-8 payload bytes)
        Some("22021") => ProcessedError::new(
            ErrorCategory::Validation,
            StatusCode::BAD_REQUEST,
            "invalid_text_encoding",
            "The request contains text that is not valid UTF-8.",
            trace_id,
        )
        .with_hint(hints::INVALID_TEXT_ENCODING)
        .with_metadata_map(metadata),

        // Undefined column
        Some("42703") => ProcessedError::new(
            ErrorCategory::QuerySyntax,
            StatusCode::BAD_REQUEST,
            "undefined_column",
            undefined_column_message(
                metadata.get("column").and_then(|v| v.as_str()),
                available_columns,
            ),
            trace_id,
        )
        .with_hint(hints::UNDEFINED_COLUMN)
        .with_metadata_map(metadata),

        // Undefined table
        Some("42P01") => {
            let (message, hint) = hints::undefined_table(extracted.table_name.as_deref());
            ProcessedError::new(
                ErrorCategory::QuerySyntax,
                StatusCode::BAD_REQUEST,
                "undefined_table",
                message,
                trace_id,
            )
            .with_hint(hint)
            .with_metadata_map(metadata)
        }

        // Syntax error
        Some("42601") => ProcessedError::new(
            ErrorCategory::QuerySyntax,
            StatusCode::BAD_REQUEST,
            "syntax_error",
            syntax_error_message(message),
            trace_id,
        )
        .with_hint(
            pg_err
                .and_then(PgDatabaseError::hint)
                .map(sanitize_error_message)
                .unwrap_or_else(|| hints::SYNTAX_ERROR.to_string()),
        )
        .with_metadata_map(metadata),

        // Insufficient privilege
        Some("42501") => ProcessedError::new(
            ErrorCategory::Authorization,
            StatusCode::FORBIDDEN,
            "insufficient_privilege",
            "You do not have permission to perform this operation",
            trace_id,
        )
        .with_hint(hints::INSUFFICIENT_PRIVILEGE)
        .with_metadata_map(metadata),

        // Connection exceptions (08xxx)
        Some(code) if code.starts_with("08") => ProcessedError::new(
            ErrorCategory::DatabaseConnection,
            StatusCode::SERVICE_UNAVAILABLE,
            "connection_error",
            "Database connection issue",
            trace_id,
        )
        .with_hint(hints::CONNECTION_ERROR)
        .with_metadata_map(metadata),

        // Undefined function / operator mismatch
        Some("42883") => {
            if let Some(parts) = parse_operator_mismatch(message) {
                let (lhs_type, operator, rhs_type) = parts;
                metadata.insert("lhs_type".to_string(), json!(lhs_type));
                metadata.insert("rhs_type".to_string(), json!(rhs_type));
                metadata.insert("operator".to_string(), json!(operator));

                // Preserve the historical UUID-specific code so existing
                // clients / dashboards keep working, while still reporting
                // structured lhs/rhs/operator metadata.
                let is_text_uuid: bool = (lhs_type == "text" && rhs_type == "uuid")
                    || (lhs_type == "uuid" && rhs_type == "text");
                let (code, user_message, hint): (&str, String, &str) = if is_text_uuid {
                    (
                        "text_uuid_operator_mismatch",
                        "A text value was compared to a UUID value without a compatible cast."
                            .to_string(),
                        hints::TEXT_UUID_OPERATOR_MISMATCH,
                    )
                } else {
                    (
                        "type_operator_mismatch",
                        format!(
                            "Cannot compare a column of type `{lhs}` to a value of type `{rhs}` using `{op}`. The value could not be coerced to the column's type.",
                            lhs = lhs_type,
                            rhs = rhs_type,
                            op = operator,
                        ),
                        hints::TYPE_OPERATOR_MISMATCH,
                    )
                };

                return ProcessedError::new(
                    ErrorCategory::Validation,
                    StatusCode::BAD_REQUEST,
                    code,
                    user_message,
                    trace_id,
                )
                .with_hint(hint)
                .with_metadata_map(metadata);
            }

            let sanitized_msg: String = sanitize_error_message(message);
            ProcessedError::new(
                ErrorCategory::Validation,
                StatusCode::BAD_REQUEST,
                "undefined_function",
                format!("Database operation failed: {}", sanitized_msg),
                trace_id,
            )
            .with_hint(hints::TYPE_OPERATOR_MISMATCH)
            .with_metadata_map(metadata)
        }

        // Fallback for other database errors
        _ => {
            if message.contains("invalid byte sequence for encoding \"UTF8\"") {
                return ProcessedError::new(
                    ErrorCategory::Validation,
                    StatusCode::BAD_REQUEST,
                    "invalid_text_encoding",
                    "The request contains text that is not valid UTF-8.",
                    trace_id,
                )
                .with_hint(hints::INVALID_TEXT_ENCODING)
                .with_metadata_map(metadata);
            }

            let sanitized_msg: String = sanitize_error_message(message);
            ProcessedError::new(
                ErrorCategory::Internal,
                StatusCode::INTERNAL_SERVER_ERROR,
                "database_error",
                format!("Database operation failed: {}", sanitized_msg),
                trace_id,
            )
            .with_metadata_map(metadata)
        }
    }
}

/// Parses `operator does not exist: <LHS_TYPE> <OP> <RHS_TYPE>` messages emitted
/// by PostgreSQL SQLSTATE `42883`. Returns `(lhs_type, operator, rhs_type)`
/// when the message matches, or `None` for the non-operator variants of
/// `42883` (undefined function call).
///
/// Handles multi-word type names (`double precision`, `timestamp with time zone`,
/// `character varying`) by treating the operator token as any run of
/// non-alphanumeric/non-whitespace characters (e.g. `=`, `<=`, `<>`, `@>`).
fn parse_operator_mismatch(message: &str) -> Option<(String, String, String)> {
    const PREFIX: &str = "operator does not exist: ";
    let start: usize = message.find(PREFIX)?;
    let remainder: &str = &message[start + PREFIX.len()..];
    let end: usize = remainder
        .find('\n')
        .unwrap_or_else(|| remainder.find('.').unwrap_or(remainder.len()));
    let body: &str = remainder[..end].trim();

    let op_start: usize = body
        .char_indices()
        .find(|(_, c)| !c.is_ascii_alphanumeric() && !c.is_whitespace() && *c != '_')?
        .0;
    let op_end_rel: usize = body[op_start..]
        .char_indices()
        .find(|(_, c)| c.is_ascii_alphanumeric() || c.is_whitespace() || *c == '_')
        .map(|(i, _)| i)
        .unwrap_or(body.len() - op_start);
    let op_end: usize = op_start + op_end_rel;

    let lhs: &str = body[..op_start].trim();
    let operator: &str = body[op_start..op_end].trim();
    let rhs: &str = body[op_end..].trim();

    if lhs.is_empty() || operator.is_empty() || rhs.is_empty() {
        return None;
    }

    Some((lhs.to_string(), operator.to_string(), rhs.to_string()))
}

fn syntax_error_message(message: &str) -> String {
    let sanitized: String = sanitize_error_message(message).trim().to_string();
    if sanitized.is_empty() || sanitized.eq_ignore_ascii_case("syntax error") {
        "The query contains a syntax error".to_string()
    } else {
        sanitized
    }
}

fn undefined_column_message(
    tried_column: Option<&str>,
    available_columns: Option<&[String]>,
) -> String {
    let prefix = match tried_column {
        Some(column) => format!("The specified column '{}' does not exist", column),
        None => "The specified column does not exist".to_string(),
    };

    match available_columns {
        Some(columns) if !columns.is_empty() => {
            format!("{}. Available columns: {}", prefix, columns.join(", "))
        }
        _ => prefix,
    }
}

/// Creates a connection error ProcessedError.
fn create_connection_error(
    code: &'static str,
    message: &str,
    hint: &str,
    trace_id: &str,
) -> ProcessedError {
    ProcessedError::new(
        ErrorCategory::DatabaseConnection,
        StatusCode::SERVICE_UNAVAILABLE,
        code,
        message,
        trace_id,
    )
    .with_hint(hint)
}

/// Creates an internal error ProcessedError.
fn create_internal_error(code: &'static str, message: &str, trace_id: &str) -> ProcessedError {
    ProcessedError::new(
        ErrorCategory::Internal,
        StatusCode::INTERNAL_SERVER_ERROR,
        code,
        message,
        trace_id,
    )
    .with_hint(hints::INTERNAL_ERROR)
}

#[cfg(test)]
mod tests {
    use super::{parse_operator_mismatch, syntax_error_message, undefined_column_message};
    use crate::error::sql_sanitizer::extract_metadata;

    #[test]
    fn syntax_error_message_preserves_useful_postgres_text() {
        assert_eq!(
            syntax_error_message("syntax error at or near \";\""),
            "syntax error at or near \";\""
        );
    }

    #[test]
    fn syntax_error_message_falls_back_for_generic_messages() {
        assert_eq!(
            syntax_error_message("syntax error"),
            "The query contains a syntax error"
        );
    }

    #[test]
    fn undefined_column_message_includes_column_and_available_columns() {
        let available = vec![
            "id".to_string(),
            "email".to_string(),
            "created_at".to_string(),
        ];
        assert_eq!(
            undefined_column_message(Some("emali"), Some(&available)),
            "The specified column 'emali' does not exist. Available columns: id, email, created_at"
        );
    }

    #[test]
    fn undefined_column_message_without_known_columns_uses_fallback() {
        assert_eq!(
            undefined_column_message(Some("emali"), None),
            "The specified column 'emali' does not exist"
        );
    }

    #[test]
    fn parse_operator_mismatch_extracts_lhs_rhs_and_operator() {
        let (lhs, op, rhs) =
            parse_operator_mismatch("operator does not exist: double precision = text")
                .expect("should parse");
        assert_eq!(lhs, "double precision");
        assert_eq!(op, "=");
        assert_eq!(rhs, "text");
    }

    #[test]
    fn parse_operator_mismatch_handles_bigint_and_boolean_and_multiword_types() {
        let (lhs, op, rhs) = parse_operator_mismatch("operator does not exist: bigint = text")
            .expect("should parse");
        assert_eq!(lhs, "bigint");
        assert_eq!(op, "=");
        assert_eq!(rhs, "text");

        let (lhs, op, rhs) = parse_operator_mismatch("operator does not exist: boolean = text")
            .expect("should parse");
        assert_eq!(lhs, "boolean");
        assert_eq!(op, "=");
        assert_eq!(rhs, "text");

        let (lhs, op, rhs) = parse_operator_mismatch(
            "operator does not exist: timestamp with time zone = text",
        )
        .expect("should parse");
        assert_eq!(lhs, "timestamp with time zone");
        assert_eq!(op, "=");
        assert_eq!(rhs, "text");
    }

    #[test]
    fn parse_operator_mismatch_handles_non_eq_operators() {
        let (lhs, op, rhs) = parse_operator_mismatch("operator does not exist: jsonb @> text")
            .expect("should parse");
        assert_eq!(lhs, "jsonb");
        assert_eq!(op, "@>");
        assert_eq!(rhs, "text");
    }

    #[test]
    fn parse_operator_mismatch_returns_none_for_non_operator_42883_messages() {
        assert!(parse_operator_mismatch("function foo(text) does not exist").is_none());
    }

    /// Regression: the metadata extractor used to capture `t` as the column
    /// name from messages like `column "t"."cache_hit_ratio"`, leaking the
    /// internal table alias. The tightened regex skips alias captures.
    #[test]
    fn sanitizer_does_not_report_single_letter_alias_as_column() {
        let info = extract_metadata(
            "column \"t\".\"cache_hit_ratio\" does not exist (this is the real message)",
        );
        assert_ne!(info.column_name.as_deref(), Some("t"));
    }

    #[test]
    fn sanitizer_still_reports_real_column_names() {
        let info = extract_metadata("column \"cache_hit_ratio\" does not exist");
        assert_eq!(info.column_name.as_deref(), Some("cache_hit_ratio"));
    }
}