vibesql-executor 0.2.0

//! Read-only query execution for concurrent access.
//!
//! This module provides:
//! - `ReadOnlyQuery` trait: A `query(&self)` method that enables read-only SQL queries
//!   on an immutable database reference
//! - `SharedDatabase` wrapper: A thread-safe wrapper around `Database` that manages
//!   concurrent read/write access
//!
//! ## Usage
//!
//! ### Using ReadOnlyQuery trait directly
//!
//! ```text
//! use vibesql_executor::readonly::ReadOnlyQuery;
//! use vibesql_storage::Database;
//!
//! let db = Database::new();
//! // ... set up tables and data ...
//!
//! // Execute read-only query without requiring &mut self
//! let result = db.query("SELECT * FROM users WHERE id = 1")?;
//! println!("Found {} rows", result.rows.len());
//! ```
//!
//! ### Using SharedDatabase for concurrent access
//!
//! ```text
//! use vibesql_executor::SharedDatabase;
//! use vibesql_storage::Database;
//!
//! let db = SharedDatabase::new(Database::new());
//!
//! // Concurrent read queries - multiple can execute simultaneously
//! let result = db.query("SELECT * FROM users WHERE id = 1").await?;
//!
//! // Write operations - exclusive access
//! db.write().await.insert_row("users", row)?;
//! ```
//!
//! ## Thread Safety
//!
//! The `query()` method takes `&self`, enabling concurrent access. Multiple readers
//! can execute SELECT queries simultaneously using the read lock, while writers
//! acquire exclusive access via write lock.
//!
//! ## Error Handling
//!
//! The `query()` method only accepts SELECT statements. Any other statement type
//! (INSERT, UPDATE, DELETE, DDL) returns a `ReadOnlyError::NotReadOnly` error.

use std::sync::Arc;

use tokio::sync::{RwLock, RwLockReadGuard, RwLockWriteGuard};
use vibesql_ast::Statement;
use vibesql_storage::Database;

use crate::{
    errors::ExecutorError,
    select::{SelectExecutor, SelectResult},
};

/// Error type for read-only query operations.
#[derive(Debug)]
pub enum ReadOnlyError {
    /// The query is not a read-only SELECT statement
    NotReadOnly { statement_type: String },
    /// SQL parsing failed
    ParseError(String),
    /// Execution failed
    ExecutionError(ExecutorError),
}

impl std::fmt::Display for ReadOnlyError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            ReadOnlyError::NotReadOnly { statement_type } => {
                write!(
                    f,
                    "{} is not allowed in read-only mode. Only SELECT queries are permitted.",
                    statement_type
                )
            }
            ReadOnlyError::ParseError(msg) => write!(f, "SQL parse error: {}", msg),
            ReadOnlyError::ExecutionError(e) => write!(f, "Execution error: {:?}", e),
        }
    }
}

impl std::error::Error for ReadOnlyError {}

impl From<ExecutorError> for ReadOnlyError {
    fn from(e: ExecutorError) -> Self {
        ReadOnlyError::ExecutionError(e)
    }
}

/// Extension trait for read-only query execution on `Database`.
///
/// This trait provides a `query(&self)` method that enables executing read-only
/// SQL queries without requiring mutable access to the database. This is essential
/// for concurrent read access in multi-connection scenarios.
pub trait ReadOnlyQuery {
    /// Execute a read-only SQL query.
    ///
    /// This method parses the SQL string and executes it if it's a SELECT statement.
    /// Any other statement type (INSERT, UPDATE, DELETE, DDL) will return an error.
    ///
    /// # Arguments
    ///
    /// * `sql` - The SQL query string to execute
    ///
    /// # Returns
    ///
    /// * `Ok(SelectResult)` - The query results including column names and rows
    /// * `Err(ReadOnlyError::NotReadOnly)` - If the SQL is not a SELECT statement
    /// * `Err(ReadOnlyError::ParseError)` - If the SQL cannot be parsed
    /// * `Err(ReadOnlyError::ExecutionError)` - If the query execution fails
    ///
    /// # Example
    ///
    /// ```text
    /// use vibesql_executor::readonly::ReadOnlyQuery;
    ///
    /// let db = Database::new();
    /// // ... create tables and insert data ...
    ///
    /// // Read-only query works with &self (no mutation)
    /// let result = db.query("SELECT * FROM users")?;
    /// for row in &result.rows {
    ///     println!("{:?}", row);
    /// }
    ///
    /// // DML queries are rejected
    /// let err = db.query("INSERT INTO users VALUES (1, 'test')");
    /// assert!(matches!(err, Err(ReadOnlyError::NotReadOnly { .. })));
    /// ```
    fn query(&self, sql: &str) -> Result<SelectResult, ReadOnlyError>;
}

impl ReadOnlyQuery for Database {
    fn query(&self, sql: &str) -> Result<SelectResult, ReadOnlyError> {
        // Parse the SQL
        let statement = vibesql_parser::Parser::parse_sql(sql)
            .map_err(|e| ReadOnlyError::ParseError(format!("{:?}", e)))?;

        // Only allow SELECT statements
        match &statement {
            Statement::Select(select_stmt) => {
                let executor = SelectExecutor::new(self);
                executor.execute_with_columns(select_stmt.as_ref()).map_err(ReadOnlyError::from)
            }
            Statement::Insert(_) => {
                Err(ReadOnlyError::NotReadOnly { statement_type: "INSERT".to_string() })
            }
            Statement::Update(_) => {
                Err(ReadOnlyError::NotReadOnly { statement_type: "UPDATE".to_string() })
            }
            Statement::Delete(_) => {
                Err(ReadOnlyError::NotReadOnly { statement_type: "DELETE".to_string() })
            }
            Statement::CreateTable(_) => {
                Err(ReadOnlyError::NotReadOnly { statement_type: "CREATE TABLE".to_string() })
            }
            Statement::DropTable(_) => {
                Err(ReadOnlyError::NotReadOnly { statement_type: "DROP TABLE".to_string() })
            }
            Statement::CreateIndex(_) => {
                Err(ReadOnlyError::NotReadOnly { statement_type: "CREATE INDEX".to_string() })
            }
            Statement::DropIndex(_) => {
                Err(ReadOnlyError::NotReadOnly { statement_type: "DROP INDEX".to_string() })
            }
            Statement::CreateView(_) => {
                Err(ReadOnlyError::NotReadOnly { statement_type: "CREATE VIEW".to_string() })
            }
            Statement::DropView(_) => {
                Err(ReadOnlyError::NotReadOnly { statement_type: "DROP VIEW".to_string() })
            }
            Statement::AlterTable(_) => {
                Err(ReadOnlyError::NotReadOnly { statement_type: "ALTER TABLE".to_string() })
            }
            Statement::TruncateTable(_) => {
                Err(ReadOnlyError::NotReadOnly { statement_type: "TRUNCATE".to_string() })
            }
            Statement::BeginTransaction(_) => {
                Err(ReadOnlyError::NotReadOnly { statement_type: "BEGIN TRANSACTION".to_string() })
            }
            Statement::Commit(_) => {
                Err(ReadOnlyError::NotReadOnly { statement_type: "COMMIT".to_string() })
            }
            Statement::Rollback(_) => {
                Err(ReadOnlyError::NotReadOnly { statement_type: "ROLLBACK".to_string() })
            }
            _ => {
                // Catch-all for other statement types
                Err(ReadOnlyError::NotReadOnly {
                    statement_type: format!("{:?}", std::mem::discriminant(&statement)),
                })
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use vibesql_catalog::{ColumnSchema, TableSchema};
    use vibesql_storage::Row;
    use vibesql_types::{DataType, SqlValue};

    use super::*;

    fn create_test_db() -> Database {
        let mut db = Database::new();
        db.catalog.set_case_sensitive_identifiers(false);

        // Create users table
        let columns = vec![
            ColumnSchema::new("id".to_string(), DataType::Integer, false),
            ColumnSchema::new(
                "name".to_string(),
                DataType::Varchar { max_length: Some(100) },
                true,
            ),
        ];
        let schema =
            TableSchema::with_primary_key("users".to_string(), columns, vec!["id".to_string()]);
        db.create_table(schema).unwrap();

        // Insert test data
        let row1 =
            Row::new(vec![SqlValue::Integer(1), SqlValue::Varchar(arcstr::ArcStr::from("Alice"))]);
        let row2 =
            Row::new(vec![SqlValue::Integer(2), SqlValue::Varchar(arcstr::ArcStr::from("Bob"))]);
        let row3 = Row::new(vec![
            SqlValue::Integer(3),
            SqlValue::Varchar(arcstr::ArcStr::from("Charlie")),
        ]);

        db.insert_row("users", row1).unwrap();
        db.insert_row("users", row2).unwrap();
        db.insert_row("users", row3).unwrap();

        db
    }

    #[test]
    fn test_query_select_all() {
        let db = create_test_db();

        let result = db.query("SELECT * FROM users").unwrap();
        assert_eq!(result.rows.len(), 3);
        assert_eq!(result.columns.len(), 2);
    }

    #[test]
    fn test_query_select_with_where() {
        let db = create_test_db();

        let result = db.query("SELECT * FROM users WHERE id = 1").unwrap();
        assert_eq!(result.rows.len(), 1);
        assert_eq!(result.rows[0].values[0], SqlValue::Integer(1));
        assert_eq!(result.rows[0].values[1], SqlValue::Varchar(arcstr::ArcStr::from("Alice")));
    }

    #[test]
    fn test_query_select_specific_columns() {
        let db = create_test_db();

        let result = db.query("SELECT name FROM users WHERE id = 2").unwrap();
        assert_eq!(result.rows.len(), 1);
        assert_eq!(result.columns.len(), 1);
        // Column names use short format by default (short_column_names)
        assert_eq!(result.columns[0].to_lowercase(), "name");
        assert_eq!(result.rows[0].values[0], SqlValue::Varchar(arcstr::ArcStr::from("Bob")));
    }

    #[test]
    fn test_query_select_count() {
        let db = create_test_db();

        let result = db.query("SELECT COUNT(*) FROM users").unwrap();
        assert_eq!(result.rows.len(), 1);
        // COUNT returns Integer (3) in this implementation
        assert_eq!(result.rows[0].values[0], SqlValue::Integer(3));
    }

    #[test]
    fn test_query_rejects_insert() {
        let db = create_test_db();

        let result = db.query("INSERT INTO users (id, name) VALUES (4, 'David')");
        assert!(matches!(
            result,
            Err(ReadOnlyError::NotReadOnly { statement_type }) if statement_type == "INSERT"
        ));
    }

    #[test]
    fn test_query_rejects_update() {
        let db = create_test_db();

        let result = db.query("UPDATE users SET name = 'Alicia' WHERE id = 1");
        assert!(matches!(
            result,
            Err(ReadOnlyError::NotReadOnly { statement_type }) if statement_type == "UPDATE"
        ));
    }

    #[test]
    fn test_query_rejects_delete() {
        let db = create_test_db();

        let result = db.query("DELETE FROM users WHERE id = 1");
        assert!(matches!(
            result,
            Err(ReadOnlyError::NotReadOnly { statement_type }) if statement_type == "DELETE"
        ));
    }

    #[test]
    fn test_query_rejects_create_table() {
        let db = create_test_db();

        let result = db.query("CREATE TABLE test (id INT)");
        assert!(matches!(
            result,
            Err(ReadOnlyError::NotReadOnly { statement_type }) if statement_type == "CREATE TABLE"
        ));
    }

    #[test]
    fn test_query_rejects_drop_table() {
        let db = create_test_db();

        let result = db.query("DROP TABLE users");
        assert!(matches!(
            result,
            Err(ReadOnlyError::NotReadOnly { statement_type }) if statement_type == "DROP TABLE"
        ));
    }

    #[test]
    fn test_query_rejects_truncate() {
        let db = create_test_db();

        let result = db.query("TRUNCATE TABLE users");
        assert!(matches!(
            result,
            Err(ReadOnlyError::NotReadOnly { statement_type }) if statement_type == "TRUNCATE"
        ));
    }

    #[test]
    fn test_query_parse_error() {
        let db = create_test_db();

        let result = db.query("SELEKT * FROM users");
        assert!(matches!(result, Err(ReadOnlyError::ParseError(_))));
    }

    #[test]
    fn test_query_execution_error_table_not_found() {
        let db = create_test_db();

        let result = db.query("SELECT * FROM nonexistent");
        assert!(matches!(result, Err(ReadOnlyError::ExecutionError(_))));
    }

    #[test]
    fn test_query_with_order_by() {
        let db = create_test_db();

        let result = db.query("SELECT * FROM users ORDER BY id DESC").unwrap();
        assert_eq!(result.rows.len(), 3);
        // First row should be id=3 (Charlie)
        assert_eq!(result.rows[0].values[0], SqlValue::Integer(3));
        // Last row should be id=1 (Alice)
        assert_eq!(result.rows[2].values[0], SqlValue::Integer(1));
    }

    #[test]
    fn test_query_with_limit() {
        let db = create_test_db();

        let result = db.query("SELECT * FROM users LIMIT 2").unwrap();
        assert_eq!(result.rows.len(), 2);
    }

    #[test]
    fn test_query_with_aggregation() {
        let db = create_test_db();

        let result = db.query("SELECT COUNT(*), MAX(id), MIN(id) FROM users").unwrap();
        assert_eq!(result.rows.len(), 1);
        // COUNT returns Integer in this implementation
        assert_eq!(result.rows[0].values[0], SqlValue::Integer(3)); // COUNT(*)
        assert_eq!(result.rows[0].values[1], SqlValue::Integer(3)); // MAX(id)
        assert_eq!(result.rows[0].values[2], SqlValue::Integer(1)); // MIN(id)
    }

    #[test]
    fn test_query_immutability() {
        let db = create_test_db();

        // Execute multiple queries on the same &db reference
        let result1 = db.query("SELECT COUNT(*) FROM users").unwrap();
        let result2 = db.query("SELECT * FROM users WHERE id = 1").unwrap();
        let result3 = db.query("SELECT name FROM users").unwrap();

        // All queries should work and return expected results
        // COUNT returns Integer in this implementation
        assert_eq!(result1.rows[0].values[0], SqlValue::Integer(3));
        assert_eq!(result2.rows.len(), 1);
        assert_eq!(result3.rows.len(), 3);
    }
}

// ============================================================================
// SharedDatabase - Thread-safe wrapper for concurrent read/write access
// ============================================================================

/// Thread-safe database wrapper enabling concurrent read queries.
///
/// `SharedDatabase` wraps a `Database` in `Arc<RwLock<...>>` to enable:
/// - **Concurrent reads**: Multiple `query()` calls can execute simultaneously
/// - **Exclusive writes**: Mutations acquire exclusive access via `write()`
///
/// ## Performance
///
/// Using `SharedDatabase` enables significant throughput improvements for read-heavy
/// workloads. On a multi-core system, concurrent read queries can achieve near-linear
/// scaling with the number of cores.
///
/// | Metric | Sequential | Concurrent (4 cores) |
/// |--------|------------|---------------------|
/// | Read QPS | 1x | ~4x |
/// | P99 latency | baseline | ~1.5x baseline |
///
/// ## Example
///
/// ```text
/// use vibesql_executor::SharedDatabase;
/// use vibesql_storage::Database;
///
/// // Create shared database
/// let db = SharedDatabase::new(Database::new());
///
/// // Concurrent reads (acquire read lock internally)
/// let result = db.query("SELECT * FROM users").await?;
///
/// // Exclusive writes
/// let mut guard = db.write().await;
/// guard.insert_row("users", row)?;
/// // guard dropped, releasing write lock
/// ```
#[derive(Clone)]
pub struct SharedDatabase {
    inner: Arc<RwLock<Database>>,
}

impl SharedDatabase {
    /// Create a new `SharedDatabase` wrapping the given database.
    pub fn new(db: Database) -> Self {
        Self { inner: Arc::new(RwLock::new(db)) }
    }

    /// Create a `SharedDatabase` from an existing `Arc<RwLock<Database>>`.
    ///
    /// This is useful when integrating with existing code that already uses
    /// the Arc<RwLock<Database>> pattern.
    pub fn from_arc(inner: Arc<RwLock<Database>>) -> Self {
        Self { inner }
    }

    /// Get the inner `Arc<RwLock<Database>>`.
    ///
    /// This is useful when you need to pass the database to code that expects
    /// the raw `Arc<RwLock<Database>>` type.
    pub fn into_inner(self) -> Arc<RwLock<Database>> {
        self.inner
    }

    /// Get a reference to the inner `Arc<RwLock<Database>>`.
    pub fn as_arc(&self) -> &Arc<RwLock<Database>> {
        &self.inner
    }

    /// Acquire a read lock for concurrent read access.
    ///
    /// Multiple readers can hold read locks simultaneously. Use this for
    /// SELECT queries or any read-only operations.
    pub async fn read(&self) -> RwLockReadGuard<'_, Database> {
        self.inner.read().await
    }

    /// Acquire a write lock for exclusive write access.
    ///
    /// Only one writer can hold the write lock at a time, and no readers
    /// can acquire read locks while a write lock is held.
    pub async fn write(&self) -> RwLockWriteGuard<'_, Database> {
        self.inner.write().await
    }

    /// Execute a read-only SQL query with automatic read lock management.
    ///
    /// This is a convenience method that:
    /// 1. Acquires a read lock on the database
    /// 2. Parses and executes the SQL query
    /// 3. Returns the result, releasing the lock
    ///
    /// Only SELECT statements are allowed. Other statement types return
    /// `ReadOnlyError::NotReadOnly`.
    ///
    /// ## Example
    ///
    /// ```text
    /// let db = SharedDatabase::new(Database::new());
    /// // ... setup tables and data ...
    ///
    /// // Execute concurrent queries from multiple tasks
    /// let result = db.query("SELECT * FROM users WHERE active = true").await?;
    /// ```
    pub async fn query(&self, sql: &str) -> Result<SelectResult, ReadOnlyError> {
        let guard = self.read().await;
        guard.query(sql)
    }
}

impl Default for SharedDatabase {
    fn default() -> Self {
        Self::new(Database::new())
    }
}

impl std::fmt::Debug for SharedDatabase {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("SharedDatabase").field("inner", &"Arc<RwLock<Database>>").finish()
    }
}

#[cfg(test)]
mod shared_database_tests {
    use super::*;
    use vibesql_catalog::{ColumnSchema, TableSchema};
    use vibesql_storage::Row;
    use vibesql_types::{DataType, SqlValue};

    async fn create_shared_test_db() -> SharedDatabase {
        let mut db = Database::new();
        db.catalog.set_case_sensitive_identifiers(false);

        // Create users table
        let columns = vec![
            ColumnSchema::new("id".to_string(), DataType::Integer, false),
            ColumnSchema::new(
                "name".to_string(),
                DataType::Varchar { max_length: Some(100) },
                true,
            ),
        ];
        let schema =
            TableSchema::with_primary_key("users".to_string(), columns, vec!["id".to_string()]);
        db.create_table(schema).unwrap();

        // Insert test data
        let row1 =
            Row::new(vec![SqlValue::Integer(1), SqlValue::Varchar(arcstr::ArcStr::from("Alice"))]);
        let row2 =
            Row::new(vec![SqlValue::Integer(2), SqlValue::Varchar(arcstr::ArcStr::from("Bob"))]);

        db.insert_row("users", row1).unwrap();
        db.insert_row("users", row2).unwrap();

        SharedDatabase::new(db)
    }

    #[tokio::test]
    async fn test_shared_query() {
        let db = create_shared_test_db().await;

        let result = db.query("SELECT * FROM users").await.unwrap();
        assert_eq!(result.rows.len(), 2);
    }

    #[tokio::test]
    async fn test_shared_query_with_filter() {
        let db = create_shared_test_db().await;

        let result = db.query("SELECT * FROM users WHERE id = 1").await.unwrap();
        assert_eq!(result.rows.len(), 1);
        assert_eq!(result.rows[0].values[0], SqlValue::Integer(1));
    }

    #[tokio::test]
    async fn test_shared_query_rejects_mutations() {
        let db = create_shared_test_db().await;

        let result = db.query("INSERT INTO users VALUES (3, 'Charlie')").await;
        assert!(matches!(result, Err(ReadOnlyError::NotReadOnly { .. })));
    }

    #[tokio::test]
    async fn test_concurrent_reads() {
        let db = create_shared_test_db().await;

        // Spawn multiple concurrent read tasks
        let mut handles = Vec::new();
        for i in 0..10 {
            let db_clone = db.clone();
            handles.push(tokio::spawn(async move {
                let result = db_clone.query("SELECT COUNT(*) FROM users").await.unwrap();
                (i, result.rows[0].values[0].clone())
            }));
        }

        // All should succeed with count = 2
        for handle in handles {
            let (_, count) = handle.await.unwrap();
            assert_eq!(count, SqlValue::Integer(2));
        }
    }

    #[tokio::test]
    async fn test_read_write_isolation() {
        let db = create_shared_test_db().await;

        // Start a read
        let result_before = db.query("SELECT COUNT(*) FROM users").await.unwrap();
        assert_eq!(result_before.rows[0].values[0], SqlValue::Integer(2));

        // Perform a write (table name is lowercased for case-insensitive lookup)
        {
            let mut guard = db.write().await;
            let row = Row::new(vec![
                SqlValue::Integer(3),
                SqlValue::Varchar(arcstr::ArcStr::from("Charlie")),
            ]);
            guard.insert_row("users", row).unwrap();
        }

        // Read should see the new data
        let result_after = db.query("SELECT COUNT(*) FROM users").await.unwrap();
        assert_eq!(result_after.rows[0].values[0], SqlValue::Integer(3));
    }

    #[tokio::test]
    async fn test_from_arc() {
        let inner = Arc::new(RwLock::new(Database::new()));
        let db = SharedDatabase::from_arc(inner.clone());

        // Should share the same underlying database
        assert!(Arc::ptr_eq(db.as_arc(), &inner));
    }
}