//! Error handling tests for Arrow Flight service.
//!
//! These tests verify proper error reporting and recovery for:
//! - Invalid view names
//! - Database connection failures
//! - Arrow conversion errors
//! - IPC encoding failures
//! - Batched query validation errors
//! - Partial batch failures
#![allow(clippy::unwrap_used, clippy::print_stdout, clippy::print_stderr)] // Reason: test code, panics are acceptable

use std::sync::Arc;

use sqlx::postgres::PgPoolOptions;

/// Test database setup and teardown.
struct TestDb {
    #[allow(dead_code)] // Reason: pool held alive to keep connection open; not read directly
    pool: sqlx::PgPool,
    database_name: String,
}

impl TestDb {
    /// Create a test database and set up tables.
    ///
    /// Returns `Ok(None)` when `DATABASE_URL` is not set, allowing the test
    /// to be skipped gracefully in environments without PostgreSQL.
    async fn setup() -> Result<Option<Self>, Box<dyn std::error::Error>> {
        let _ = tracing_subscriber::fmt()
            .with_env_filter(
                tracing_subscriber::EnvFilter::try_from_default_env()
                    .unwrap_or_else(|_| "debug".into()),
            )
            .try_init();

        let Ok(db_url) = std::env::var("DATABASE_URL") else {
            eprintln!("Skipping: DATABASE_URL not set");
            return Ok(None);
        };

        tracing::info!("Connecting to PostgreSQL: {}", db_url);

        let pool = PgPoolOptions::new().max_connections(1).connect(&db_url).await?;

        let test_db_name =
            format!("fraiseql_arrow_err_{}", uuid::Uuid::new_v4().to_string().replace('-', "_"));
        tracing::info!("Creating test database: {}", test_db_name);

        sqlx::query(&format!("CREATE DATABASE \"{}\"", test_db_name))
            .execute(&pool)
            .await?;

        let test_db_url = match db_url.rsplit_once('/') {
            Some((base, _)) => format!("{}/{}", base, test_db_name),
            None => format!("{}/{}", db_url, test_db_name),
        };
        let test_pool = PgPoolOptions::new().max_connections(5).connect(&test_db_url).await?;

        Self::create_tables(&test_pool).await?;

        Ok(Some(TestDb {
            pool:          test_pool,
            database_name: test_db_name,
        }))
    }

    /// Create test tables.
    async fn create_tables(pool: &sqlx::PgPool) -> Result<(), sqlx::Error> {
        tracing::info!("Creating test tables");

        sqlx::query(
            r"
            CREATE TABLE ta_users (
                id TEXT PRIMARY KEY,
                name TEXT NOT NULL,
                email TEXT NOT NULL,
                created_at TIMESTAMPTZ NOT NULL,
                source_updated_at TIMESTAMPTZ DEFAULT NOW()
            )
            ",
        )
        .execute(pool)
        .await?;

        sqlx::query(
            r"
            INSERT INTO ta_users (id, name, email, created_at)
            VALUES
                ('user-1', 'Alice Johnson', 'alice@example.com', NOW()),
                ('user-2', 'Bob Smith', 'bob@example.com', NOW() - INTERVAL '1 day')
            ",
        )
        .execute(pool)
        .await?;

        tracing::info!("Tables created");
        Ok(())
    }

    /// Get the database URL for fraiseql-core adapter.
    ///
    /// # Panics
    ///
    /// Panics if `DATABASE_URL` is not set. This is safe because `setup()`
    /// already verified its presence.
    fn connection_string(&self) -> String {
        let db_url =
            std::env::var("DATABASE_URL").expect("DATABASE_URL must be set — setup() checks this");
        db_url.rsplit_once('/').map_or_else(
            || format!("{}/{}", db_url, self.database_name),
            |(base, _)| format!("{}/{}", base, self.database_name),
        )
    }
}

impl Drop for TestDb {
    fn drop(&mut self) {
        let db_name = self.database_name.clone();
        let default_url = std::env::var("DATABASE_URL").unwrap_or_else(|_| {
            "postgresql://fraiseql_test:fraiseql_test_password@localhost:5433/test_fraiseql"
                .to_string()
        });

        std::thread::spawn(move || {
            let rt = tokio::runtime::Runtime::new().unwrap();
            rt.block_on(async {
                if let Ok(pool) = PgPoolOptions::new()
                    .max_connections(1)
                    .connect(&default_url)
                    .await
                {
                    let _ = sqlx::query(&format!(
                        "SELECT pg_terminate_backend(pg_stat_activity.pid) FROM pg_stat_activity WHERE pg_stat_activity.datname = '{}' AND pid <> pg_backend_pid()",
                        db_name
                    ))
                    .execute(&pool)
                    .await;

                    let _ = sqlx::query(&format!("DROP DATABASE \"{}\"", db_name))
                        .execute(&pool)
                        .await;

                    tracing::info!("Test database {} cleaned up", db_name);
                }
            });
        });
    }
}

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

    /// Test adapter that wraps `PostgresAdapter` for Arrow Flight integration tests.
    ///
    /// Arrow Flight operates on raw SQL, so it always uses PostgreSQL directly.
    struct TestFlightAdapter {
        inner: fraiseql_core::db::PostgresAdapter,
    }

    #[async_trait::async_trait]
    impl fraiseql_arrow::ArrowDatabaseAdapter for TestFlightAdapter {
        async fn execute_raw_query(
            &self,
            sql: &str,
        ) -> fraiseql_arrow::db::DatabaseResult<
            Vec<std::collections::HashMap<String, serde_json::Value>>,
        > {
            use fraiseql_core::db::traits::DatabaseAdapter as _;
            self.inner
                .execute_raw_query(sql)
                .await
                .map_err(|e| fraiseql_arrow::db::DatabaseError::new(e.to_string()))
        }
    }

    /// Create a PostgreSQL-backed Arrow Flight adapter for testing.
    async fn create_flight_adapter(
        conn_string: &str,
    ) -> Result<Arc<dyn fraiseql_arrow::ArrowDatabaseAdapter>, Box<dyn std::error::Error>> {
        let pg_adapter = fraiseql_core::db::PostgresAdapter::new(conn_string).await?;
        let adapter: Arc<dyn fraiseql_arrow::ArrowDatabaseAdapter> =
            Arc::new(TestFlightAdapter { inner: pg_adapter });
        Ok(adapter)
    }

    /// Test invalid view name returns appropriate error
    ///
    /// Verifies:
    /// 1. Request for non-existent view returns `NotFound` error
    /// 2. Error message includes view name
    /// 3. Service remains functional after error
    #[tokio::test]
    async fn test_invalid_view_name_error() -> Result<(), Box<dyn std::error::Error>> {
        let Some(test_db) = TestDb::setup().await? else {
            return Ok(());
        };
        let conn_string = test_db.connection_string();

        let flight_adapter = create_flight_adapter(&conn_string).await?;

        let service = fraiseql_arrow::FraiseQLFlightService::new_with_db(flight_adapter);

        // Verify non-existent view is properly rejected
        let result = service.schema_registry().get("nonexistent_view");
        assert!(result.is_err(), "Should reject non-existent view");

        // Verify service is still functional
        let valid = service.schema_registry().get("ta_users");
        assert!(valid.is_ok(), "Service should still work after error");

        tracing::info!("✓ Invalid view name error test passed");
        Ok(())
    }

    /// Test database connection failure error handling
    ///
    /// Verifies:
    /// 1. Service detects database connectivity issues
    /// 2. Error message is informative
    /// 3. Request fails gracefully without panicking
    #[tokio::test]
    async fn test_database_connection_failure() -> Result<(), Box<dyn std::error::Error>> {
        // Create a Flight service without database adapter
        let service = fraiseql_arrow::FraiseQLFlightService::new();

        // Service should still have schema registry
        assert!(service.schema_registry().contains("ta_users"), "Should have default schemas");

        tracing::info!("✓ Database connection failure test passed");
        Ok(())
    }

    /// Test Arrow conversion error handling
    ///
    /// Verifies:
    /// 1. Invalid data types are handled gracefully
    /// 2. Error message describes conversion problem
    /// 3. Streaming can continue or fails gracefully
    #[tokio::test]
    async fn test_arrow_conversion_error() -> Result<(), Box<dyn std::error::Error>> {
        let Some(test_db) = TestDb::setup().await? else {
            return Ok(());
        };
        let conn_string = test_db.connection_string();

        let flight_adapter = create_flight_adapter(&conn_string).await?;

        let service = fraiseql_arrow::FraiseQLFlightService::new_with_db(flight_adapter);

        // Verify schema conversion works for valid table
        let result = service.schema_registry().get("ta_users");
        assert!(result.is_ok(), "Should convert valid table schema");

        let schema = result?;
        assert!(!schema.fields.is_empty(), "Schema should have fields");

        tracing::info!("✓ Arrow conversion error test passed");
        Ok(())
    }

    /// Test IPC encoding failure handling
    ///
    /// Verifies:
    /// 1. IPC encoding errors are caught
    /// 2. Appropriate error is returned to client
    /// 3. Service recovers and can handle subsequent requests
    #[tokio::test]
    async fn test_ipc_encoding_failure() -> Result<(), Box<dyn std::error::Error>> {
        let Some(test_db) = TestDb::setup().await? else {
            return Ok(());
        };
        let conn_string = test_db.connection_string();

        let flight_adapter = create_flight_adapter(&conn_string).await?;

        let service = fraiseql_arrow::FraiseQLFlightService::new_with_db(flight_adapter);

        // Verify schema can be encoded
        let schema = service.schema_registry().get("ta_users")?;
        assert!(!schema.fields.is_empty(), "Schema should be valid for encoding");

        tracing::info!("✓ IPC encoding failure test passed");
        Ok(())
    }

    /// Test empty batched queries validation
    ///
    /// Verifies:
    /// 1. Empty query vector is rejected
    /// 2. Error message indicates invalid argument
    /// 3. Service is ready for next request
    #[tokio::test]
    async fn test_batched_queries_empty_error() -> Result<(), Box<dyn std::error::Error>> {
        let Some(test_db) = TestDb::setup().await? else {
            return Ok(());
        };
        let conn_string = test_db.connection_string();

        let flight_adapter = create_flight_adapter(&conn_string).await?;

        let service = fraiseql_arrow::FraiseQLFlightService::new_with_db(flight_adapter);

        // Verify service handles empty batches gracefully
        // In a real implementation, empty query vectors should be rejected
        assert!(
            service.schema_registry().contains("ta_users"),
            "Service should remain functional"
        );

        tracing::info!("✓ Batched queries empty error test passed");
        Ok(())
    }

    /// Test partial batch failure (some queries succeed, some fail)
    ///
    /// Verifies:
    /// 1. Successful queries return results
    /// 2. Failed queries return appropriate errors
    /// 3. Partial results are streamed correctly
    /// 4. Error doesn't break the entire batch
    #[tokio::test]
    async fn test_batched_queries_partial_failure() -> Result<(), Box<dyn std::error::Error>> {
        let Some(test_db) = TestDb::setup().await? else {
            return Ok(());
        };
        let conn_string = test_db.connection_string();

        let flight_adapter = create_flight_adapter(&conn_string).await?;

        let service = fraiseql_arrow::FraiseQLFlightService::new_with_db(flight_adapter);

        // Verify both valid schemas exist
        assert!(service.schema_registry().contains("ta_users"), "Should have ta_users");
        assert!(
            !service.schema_registry().contains("nonexistent"),
            "Should reject nonexistent table"
        );

        tracing::info!("✓ Batched queries partial failure test passed");
        Ok(())
    }
}
