ormkit/

batch.rs

//! Batch operations for efficient bulk database operations
//!
//! This module provides utilities for performing bulk insert, update, and delete
//! operations with automatic batching and transaction management.
//!
//! # Example
//!
//! ```rust,no_run
//! use ormkit::batch::{insert_many, update_many, BatchOptions};
//! # use ormkit::Entity;
//! # use sqlx::PgPool;
//! # use uuid::Uuid;
//! #
//! # #[derive(Debug, Clone)]
//! # struct User;
//! # impl Entity for User { type Id = Uuid; fn table_name() -> &'static str { "users" } fn id(&self) -> Self::Id { unimplemented!() } }
//!
//! # async fn test(pool: &PgPool) -> Result<(), Box<dyn std::error::Error>> {
//! // Bulk insert with automatic batching
//! let users = vec![/* ... */];
//! let inserted = insert_many(&pool, &users, None).await?;
//!
//! // Bulk insert with custom batch size
//! let options = BatchOptions::new().batch_size(100);
//! let inserted = insert_many(&pool, &users, Some(options)).await?;
//!
//! // Bulk update
//! let updates = vec![/* ... */];
//! let updated = update_many(&pool, &updates, None).await?;
//! # Ok(())
//! # }
//! ```

use crate::entity::Entity;
use sqlx::PgPool;
use std::fmt::Debug;

/// Default batch size for batch operations
pub const DEFAULT_BATCH_SIZE: usize = 100;

/// Maximum batch size to prevent statement overflow
pub const MAX_BATCH_SIZE: usize = 1000;

/// Errors that can occur during batch operations
#[derive(Debug, thiserror::Error)]
pub enum BatchError {
    /// Database error
    #[error("Database error: {0}")]
    Database(#[from] sqlx::Error),

    /// Empty batch provided
    #[error("Empty batch provided")]
    EmptyBatch,

    /// Batch size exceeds maximum
    #[error("Batch size {count} exceeds maximum {max}", count = count, max = max)]
    BatchSizeExceeded { count: usize, max: usize },

    /// Serialization error
    #[error("Serialization error: {0}")]
    Serialization(String),
}

/// Result type for batch operations
pub type BatchResult<T> = Result<T, BatchError>;

/// Options for batch operations
#[derive(Debug, Clone)]
pub struct BatchOptions {
    /// Number of items to process per batch
    batch_size: usize,
    /// Whether to wrap each batch in a transaction
    transaction_per_batch: bool,
    /// Whether to wrap the entire operation in a transaction
    wrap_in_transaction: bool,
    /// Whether to continue on error (only valid with transaction_per_batch)
    continue_on_error: bool,
}

impl BatchOptions {
    /// Create new batch options with defaults
    pub fn new() -> Self {
        Self {
            batch_size: DEFAULT_BATCH_SIZE,
            transaction_per_batch: true,
            wrap_in_transaction: true,
            continue_on_error: false,
        }
    }

    /// Set the batch size
    pub fn batch_size(mut self, size: usize) -> Self {
        let size = size.min(MAX_BATCH_SIZE).max(1);
        self.batch_size = size;
        self
    }

    /// Set whether to wrap each batch in a transaction
    pub fn transaction_per_batch(mut self, value: bool) -> Self {
        self.transaction_per_batch = value;
        self
    }

    /// Set whether to wrap the entire operation in a transaction
    pub fn wrap_in_transaction(mut self, value: bool) -> Self {
        self.wrap_in_transaction = value;
        self
    }

    /// Set whether to continue on error
    pub fn continue_on_error(mut self, value: bool) -> Self {
        self.continue_on_error = value;
        self
    }

    /// Get the batch size
    pub fn get_batch_size(&self) -> usize {
        self.batch_size
    }

    /// Check if transactions should be used per batch
    pub fn is_transaction_per_batch(&self) -> bool {
        self.transaction_per_batch
    }

    /// Check if the entire operation should be wrapped in a transaction
    pub fn is_wrap_in_transaction(&self) -> bool {
        self.wrap_in_transaction
    }

    /// Check if operation should continue on error
    pub fn is_continue_on_error(&self) -> bool {
        self.continue_on_error
    }
}

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

/// Result of a batch operation
#[derive(Debug)]
pub struct BatchResultInfo {
    /// Total number of items processed
    pub total_processed: usize,
    /// Number of items successfully processed
    pub successful: usize,
    /// Number of items that failed
    pub failed: usize,
    /// Number of batches processed
    pub batches: usize,
}

impl BatchResultInfo {
    /// Create a new batch result info
    pub fn new(total_processed: usize, successful: usize, failed: usize, batches: usize) -> Self {
        Self {
            total_processed,
            successful,
            failed,
            batches,
        }
    }

    /// Check if all items were processed successfully
    pub fn is_success(&self) -> bool {
        self.failed == 0
    }

    /// Get the success rate as a percentage
    pub fn success_rate(&self) -> f64 {
        if self.total_processed == 0 {
            return 100.0;
        }
        (self.successful as f64 / self.total_processed as f64) * 100.0
    }
}

/// Insert multiple entities in batches
///
/// This function inserts multiple entities in batches for better performance
/// and to avoid statement size limits.
///
/// # Arguments
///
/// * `pool` - The database connection pool
/// * `entities` - The entities to insert
/// * `options` - Optional batch operation settings
///
/// # Returns
///
/// Information about the batch operation
///
/// # Example
///
/// ```rust,no_run
/// use ormkit::batch::insert_many;
/// # use ormkit::Entity;
/// # use sqlx::PgPool;
/// # use uuid::Uuid;
/// # #[derive(Debug, Clone)]
/// # struct User;
/// # impl Entity for User { type Id = Uuid; fn table_name() -> &'static str { "users" } fn id(&self) -> Self::Id { unimplemented!() } }
///
/// # async fn test(pool: &PgPool) -> Result<(), Box<dyn std::error::Error>> {
/// let users = vec![/* ... */];
/// let result = insert_many(&pool, &users, None).await?;
/// println!("Inserted {} users", result.successful);
/// # Ok(())
/// # }
/// ```
pub async fn insert_many<E>(
    _pool: &PgPool,
    entities: &[E],
    options: Option<BatchOptions>,
) -> BatchResult<BatchResultInfo>
where
    E: Entity + Debug + Send + Sync,
{
    if entities.is_empty() {
        return Err(BatchError::EmptyBatch);
    }

    let options = options.unwrap_or_default();
    let _batch_size = options.get_batch_size();

    // TODO: Implement batch insert with proper reflection/macro codegen
    // This requires either:
    // 1. Reflection to extract field values from entities
    // 2. Procedural macro to generate insert code for each entity type
    //
    // For now, return an error indicating this is not yet implemented
    return Err(BatchError::Serialization(
        "Batch insert not yet implemented. Use individual inserts or implement reflection.".to_string()
    ));

}

/// Update multiple entities in batches
///
/// This function updates multiple entities in batches for better performance.
///
/// # Arguments
///
/// * `pool` - The database connection pool
/// * `updates` - Tuple of (column, value, condition) updates
/// * `options` - Optional batch operation settings
///
/// # Returns
///
/// Information about the batch operation
///
/// # Example
///
/// ```rust,no_run
/// use ormkit::batch::update_many;
/// # use sqlx::PgPool;
///
/// # async fn test(pool: &PgPool) -> Result<(), Box<dyn std::error::Error>> {
/// // Update all active users to have status='verified'
/// let result = update_many(
///     &pool,
///     "users",
///     &[("status", "verified")],
///     "status = $1",
///     &[&"active" as &(dyn sqlx::Encode<sqlx::Postgres> + sqlx::Type<sqlx::Postgres> + Sync + Send)],
///     None
/// ).await?;
/// # Ok(())
/// # }
/// ```
pub async fn update_many<P>(
    pool: &PgPool,
    table: &str,
    updates: &[(&str, &str)],
    condition: &str,
    params: &[P],
    _options: Option<BatchOptions>,
) -> BatchResult<u64>
where
    P: for<'r> sqlx::Encode<'r, sqlx::Postgres> + sqlx::Type<sqlx::Postgres> + Clone + Sync + Send,
{
    // Build SET clause
    let set_clause: String = updates
        .iter()
        .enumerate()
        .map(|(i, (col, _))| format!(r#""{}" = ${}"#, col, i + 1))
        .collect::<Vec<_>>()
        .join(", ");

    // Build the query
    let query = format!(
        r#"UPDATE "{}" SET {} WHERE {}"#,
        table, set_clause, condition
    );

    // Execute
    let mut query_builder = sqlx::query(&*query);

    // Bind update values
    for (_, val) in updates {
        query_builder = query_builder.bind(*val);
    }

    // Bind condition parameters
    for param in params {
        query_builder = query_builder.bind(param.clone());
    }

    let result = query_builder.execute(pool).await?;

    Ok(result.rows_affected())
}

/// Delete multiple entities in batches
///
/// This function deletes multiple entities in batches.
///
/// # Arguments
///
/// * `pool` - The database connection pool
/// * `ids` - The IDs of entities to delete
/// * `options` - Optional batch operation settings
///
/// # Returns
///
/// Information about the batch operation
pub async fn delete_many<E>(
    pool: &PgPool,
    ids: &[E::Id],
    options: Option<BatchOptions>,
) -> BatchResult<u64>
where
    E: Entity,
    for<'a> E::Id: sqlx::Encode<'a, sqlx::Postgres> + sqlx::Type<sqlx::Postgres> + Send + Sync,
{
    if ids.is_empty() {
        return Ok(0);
    }

    let options = options.unwrap_or_default();
    let batch_size = options.get_batch_size();

    let mut total_deleted = 0;

    for chunk in ids.chunks(batch_size) {
        let placeholders: Vec<String> = chunk
            .iter()
            .enumerate()
            .map(|(i, _)| format!("${}", i + 1))
            .collect();

        let query = format!(
            r#"DELETE FROM "{}" WHERE "id" IN ({})"#,
            E::table_name(),
            placeholders.join(", ")
        );

        let mut query_builder = sqlx::query(&query);

        for id in chunk {
            query_builder = query_builder.bind(id);
        }

        let result = query_builder.execute(pool).await?;

        total_deleted += result.rows_affected();
    }

    Ok(total_deleted)
}

/// Execute a batch of custom queries
///
/// This function executes multiple queries in batches with optional
/// transaction wrapping.
///
/// # Arguments
///
/// * `pool` - The database connection pool
/// * `queries` - The SQL queries to execute
/// * `options` - Optional batch operation settings
///
/// # Returns
///
/// Information about the batch operation
pub async fn execute_batch(
    pool: &PgPool,
    queries: &[&str],
    options: Option<BatchOptions>,
) -> BatchResult<BatchResultInfo> {
    if queries.is_empty() {
        return Err(BatchError::EmptyBatch);
    }

    let options = options.unwrap_or_default();
    let batch_size = options.get_batch_size();

    let mut total_processed = 0;
    let mut successful = 0;
    let mut failed = 0;
    let mut batches = 0;

    for chunk in queries.chunks(batch_size) {
        batches += 1;

        for query in chunk {
            total_processed += 1;

            let result = sqlx::query(query).execute(pool).await;

            match result {
                Ok(_) => successful += 1,
                Err(e) => {
                    failed += 1;
                    if !options.is_continue_on_error() {
                        return Err(BatchError::Database(e));
                    }
                }
            }
        }
    }

    Ok(BatchResultInfo::new(total_processed, successful, failed, batches))
}

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

    #[test]
    fn test_batch_options_new() {
        let options = BatchOptions::new();
        assert_eq!(options.get_batch_size(), DEFAULT_BATCH_SIZE);
        assert!(options.is_transaction_per_batch());
        assert!(options.is_wrap_in_transaction());
        assert!(!options.is_continue_on_error());
    }

    #[test]
    fn test_batch_options_batch_size() {
        let options = BatchOptions::new().batch_size(500);
        assert_eq!(options.get_batch_size(), 500);
    }

    #[test]
    fn test_batch_options_max_batch_size() {
        let options = BatchOptions::new().batch_size(10000);
        assert_eq!(options.get_batch_size(), MAX_BATCH_SIZE);
    }

    #[test]
    fn test_batch_options_min_batch_size() {
        let options = BatchOptions::new().batch_size(0);
        assert_eq!(options.get_batch_size(), 1);
    }

    #[test]
    fn test_batch_options_transaction_per_batch() {
        let options = BatchOptions::new().transaction_per_batch(false);
        assert!(!options.is_transaction_per_batch());
    }

    #[test]
    fn test_batch_options_continue_on_error() {
        let options = BatchOptions::new().continue_on_error(true);
        assert!(options.is_continue_on_error());
    }

    #[test]
    fn test_batch_options_default() {
        let options = BatchOptions::default();
        assert_eq!(options.get_batch_size(), DEFAULT_BATCH_SIZE);
    }

    #[test]
    fn test_batch_result_info_new() {
        let info = BatchResultInfo::new(100, 95, 5, 10);
        assert_eq!(info.total_processed, 100);
        assert_eq!(info.successful, 95);
        assert_eq!(info.failed, 5);
        assert_eq!(info.batches, 10);
    }

    #[test]
    fn test_batch_result_info_is_success() {
        let info = BatchResultInfo::new(100, 100, 0, 10);
        assert!(info.is_success());

        let info = BatchResultInfo::new(100, 95, 5, 10);
        assert!(!info.is_success());
    }

    #[test]
    fn test_batch_result_info_success_rate() {
        let info = BatchResultInfo::new(100, 75, 25, 10);
        assert_eq!(info.success_rate(), 75.0);
    }

    #[test]
    fn test_batch_result_info_success_rate_empty() {
        let info = BatchResultInfo::new(0, 0, 0, 0);
        assert_eq!(info.success_rate(), 100.0);
    }
}