spring_batch_rs/item/orm/

orm_reader.rs

use std::cell::{Cell, RefCell};

use sea_orm::{DatabaseConnection, DbErr, EntityTrait, FromQueryResult, PaginatorTrait, Select};

use crate::{
    core::item::{ItemReader, ItemReaderResult},
    BatchError,
};

/// A reader for reading entities from a database using SeaORM.
///
/// This reader provides an implementation of the `ItemReader` trait for SeaORM-based
/// database operations. It supports reading entities from any SeaORM-compatible database
/// with optional pagination for efficient memory usage when dealing with large datasets.
///
/// # Generic Parameters
///
/// - `I`: The SeaORM entity type that implements `EntityTrait`
///
/// # Design Philosophy
///
/// This reader follows SeaORM's conventions and leverages its built-in pagination
/// mechanisms for optimal performance. It provides a bridge between SeaORM's async
/// API and the batch framework's synchronous `ItemReader` trait.
///
/// # Memory Management
///
/// The reader uses an internal buffer to store entity models from the current page,
/// which helps balance memory usage with query performance. When pagination is enabled,
/// only one page of data is kept in memory at any time.
///
/// # Query Flexibility
///
/// Accepts any SeaORM `Select` query, allowing for complex filtering, joins, and
/// ordering. The query is executed as-is, with pagination parameters added when configured.
///
/// # State Management
///
/// - `offset`: Tracks the absolute position across all pages
/// - `current_page`: Tracks which page is currently loaded in the buffer
/// - `buffer`: Holds the current page's entity models
pub struct OrmItemReader<'a, I>
where
    I: EntityTrait,
{
    /// Database connection reference - borrowed for the lifetime of the reader
    /// This ensures the connection remains valid throughout the reader's lifecycle
    connection: Option<&'a DatabaseConnection>,

    /// SeaORM select query to execute
    /// This query is cloned for each page fetch to maintain immutability
    query: Option<Select<I>>,

    /// Optional page size for pagination
    /// When Some(size), data is loaded in chunks of this size
    /// When None, all data is loaded at once
    page_size: Option<u64>,

    /// Current offset for tracking position across all pages
    /// Uses Cell for interior mutability in single-threaded context
    /// This tracks the absolute position, not just within the current page
    offset: Cell<u64>,

    /// Internal buffer for storing fetched items from the current page
    /// Uses RefCell for interior mutability to allow borrowing during reads
    /// The buffer is cleared and refilled when a new page is loaded
    buffer: RefCell<Vec<I::Model>>,

    /// Current page number (used with pagination)
    /// Starts at 0 and increments when moving to the next page
    /// Only relevant when page_size is Some(_)
    current_page: Cell<u64>,
}

impl<'a, I> Default for OrmItemReader<'a, I>
where
    I: EntityTrait,
    I::Model: FromQueryResult + Send + Sync + Clone,
{
    fn default() -> Self {
        Self::new()
    }
}

impl<'a, I> OrmItemReader<'a, I>
where
    I: EntityTrait,
    I::Model: FromQueryResult + Send + Sync + Clone,
{
    /// Creates a new `OrmItemReader` with default configuration.
    ///
    /// All parameters must be set using the builder methods before use.
    /// Use the builder pattern for a more convenient API.
    ///
    /// # Returns
    ///
    /// A new `OrmItemReader` instance with default settings.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// use spring_batch_rs::item::orm::OrmItemReader;
    /// use sea_orm::{Database, EntityTrait};
    ///
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// let db = Database::connect("sqlite::memory:").await?;
    ///
    /// // Using direct instantiation (less recommended)
    /// // let reader = OrmItemReader::<MyEntity>::new()
    /// //     .connection(&db)
    /// //     .query(MyEntity::find())
    /// //     .page_size(100);
    ///
    /// // Using builder (recommended)
    /// // let reader = OrmItemReaderBuilder::new()
    /// //     .connection(&db)
    /// //     .query(MyEntity::find())
    /// //     .page_size(100)
    /// //     .build();
    /// # Ok(())
    /// # }
    /// ```
    pub fn new() -> Self {
        Self {
            connection: None,
            query: None,
            page_size: None,
            offset: Cell::new(0),
            buffer: RefCell::new(Vec::new()),
            current_page: Cell::new(0),
        }
    }

    /// Sets the database connection for the reader.
    ///
    /// This is a required parameter that must be set before using the reader.
    ///
    /// # Arguments
    ///
    /// * `connection` - Reference to the SeaORM database connection
    ///
    /// # Returns
    ///
    /// The updated `OrmItemReader` instance.
    ///
    /// # Examples
    ///
    /// ```compile_fail
    /// use spring_batch_rs::item::orm::OrmItemReader;
    /// use sea_orm::Database;
    ///
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// let db = Database::connect("sqlite::memory:").await?;
    /// let reader = OrmItemReader::<String>::new()
    ///     .connection(&db);
    /// # Ok(())
    /// # }
    /// ```
    pub fn connection(mut self, connection: &'a DatabaseConnection) -> Self {
        self.connection = Some(connection);
        self
    }

    /// Sets the SeaORM select query for the reader.
    ///
    /// This is a required parameter that must be set before using the reader.
    /// The query can include filtering, ordering, joins, and other SeaORM operations.
    ///
    /// # Arguments
    ///
    /// * `query` - The SeaORM select query to execute
    ///
    /// # Returns
    ///
    /// The updated `OrmItemReader` instance.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// use spring_batch_rs::item::orm::OrmItemReader;
    /// use sea_orm::EntityTrait;
    ///
    /// // Basic query
    /// // let reader = OrmItemReader::<MyEntity>::new()
    /// //     .query(MyEntity::find());
    ///
    /// // Filtered query
    /// // let reader = OrmItemReader::<MyEntity>::new()
    /// //     .query(MyEntity::find().filter(my_entity::Column::Active.eq(true)));
    ///
    /// // Ordered query
    /// // let reader = OrmItemReader::<MyEntity>::new()
    /// //     .query(MyEntity::find().order_by_asc(my_entity::Column::Id));
    /// ```
    pub fn query(mut self, query: Select<I>) -> Self {
        self.query = Some(query);
        // Pre-allocate buffer capacity based on page size if available
        if let Some(page_size) = self.page_size {
            self.buffer = RefCell::new(Vec::with_capacity(page_size as usize));
        }
        self
    }

    /// Sets the page size for the reader.
    ///
    /// When set, the reader will use pagination to limit memory usage.
    /// When not set, all data will be loaded at once.
    ///
    /// # Arguments
    ///
    /// * `page_size` - The number of items to read per page
    ///
    /// # Returns
    ///
    /// The updated `OrmItemReader` instance.
    ///
    /// # Performance Considerations
    ///
    /// - Larger page sizes reduce the number of database round trips but use more memory
    /// - Smaller page sizes use less memory but may result in more database queries
    /// - Consider your dataset size and available memory when choosing a page size
    /// - Recommended range: 50-1000 items per page for most use cases
    ///
    /// # Examples
    ///
    /// ```compile_fail
    /// use spring_batch_rs::item::orm::OrmItemReader;
    ///
    /// // Small page size for memory-constrained environments
    /// let reader = OrmItemReader::<String>::new()
    ///     .page_size(50);
    ///
    /// // Larger page size for better performance with ample memory
    /// let reader = OrmItemReader::<String>::new()
    ///     .page_size(1000);
    /// ```
    pub fn page_size(mut self, page_size: u64) -> Self {
        self.page_size = Some(page_size);
        // Update buffer capacity if available
        self.buffer = RefCell::new(Vec::with_capacity(page_size as usize));
        self
    }

    /// Reads a page of items from the database using SeaORM.
    ///
    /// This method executes the SeaORM query with pagination parameters (if page_size is set),
    /// and fills the internal buffer with the results. It uses SeaORM's async methods
    /// within a blocking context to maintain compatibility with the synchronous ItemReader trait.
    ///
    /// # Pagination Logic
    ///
    /// - **With pagination**: Uses SeaORM's `paginate()` method to fetch a specific page
    /// - **Without pagination**: Uses SeaORM's `all()` method to fetch all results
    ///
    /// # Buffer Management
    ///
    /// - Clears the existing buffer before loading new data
    /// - Stores all entity models directly in the buffer for sequential access
    ///
    /// # Error Handling
    ///
    /// Returns a `DbErr` if the database query fails. This error will be converted
    /// to a `BatchError` by the calling method.
    async fn read_page_async(&self) -> Result<(), DbErr> {
        let results = if let Some(page_size) = self.page_size {
            // Use SeaORM's paginator for efficient pagination
            // This creates a paginator that can fetch specific pages
            let paginator = self
                .query
                .as_ref()
                .unwrap()
                .clone()
                .paginate(self.connection.unwrap(), page_size);
            let current_page = self.current_page.get();

            // Fetch the specific page we're currently on
            // Pages are 0-indexed in SeaORM's paginator
            paginator.fetch_page(current_page).await?
        } else {
            // Load all results at once - suitable for smaller datasets
            // This executes the query and returns all matching rows
            self.query
                .as_ref()
                .unwrap()
                .clone()
                .all(self.connection.unwrap())
                .await?
        };

        // Clear the buffer and fill it with entity models
        let mut buffer = self.buffer.borrow_mut();
        buffer.clear(); // Remove any existing items from the previous page

        // Store entity models directly in the buffer
        buffer.extend(results);

        Ok(())
    }

    /// Reads a page of items from the database.
    ///
    /// This is a synchronous wrapper around the async `read_page_async` method.
    /// It uses tokio's `block_in_place` to run the async operation in a blocking context,
    /// which is necessary because the `ItemReader` trait is synchronous.
    ///
    /// # Async-to-Sync Bridge
    ///
    /// SeaORM is inherently async, but the batch framework uses synchronous traits
    /// for simplicity. This method bridges that gap by:
    /// 1. Using `block_in_place` to avoid blocking the tokio runtime
    /// 2. Getting the current runtime handle to execute the async operation
    /// 3. Converting SeaORM's `DbErr` to the batch framework's `BatchError`
    ///
    /// # Error Conversion
    ///
    /// Converts SeaORM's `DbErr` to `BatchError::ItemReader` with context information
    /// to help with debugging database-related issues.
    fn read_page(&self) -> Result<(), BatchError> {
        tokio::task::block_in_place(|| {
            tokio::runtime::Handle::current().block_on(async {
                self.read_page_async()
                    .await
                    .map_err(|e| BatchError::ItemReader(format!("SeaORM query failed: {}", e)))
            })
        })
    }
}

/// Implementation of ItemReader trait for OrmItemReader.
///
/// This implementation provides a way to read items from a database using SeaORM
/// with support for pagination. It uses an internal buffer to store the results
/// of database queries and keeps track of the current offset to determine when
/// a new page of data needs to be fetched.
///
/// # Reading Strategy
///
/// The reader implements a lazy-loading strategy:
/// 1. **First read**: Loads the first page of data
/// 2. **Subsequent reads**: Returns items from the buffer
/// 3. **Page boundary**: When the buffer is exhausted, loads the next page
/// 4. **End of data**: Returns None when no more data is available
///
/// # State Management
///
/// - `offset`: Tracks the absolute position across all pages
/// - `current_page`: Tracks which page we're currently reading from
/// - `buffer`: Holds the current page's data
///
/// The reader maintains these invariants:
/// - `offset` always represents the next item to be read
/// - `current_page` represents the page currently loaded in the buffer
/// - The buffer contains items for the current page only
impl<I> ItemReader<I::Model> for OrmItemReader<'_, I>
where
    I: EntityTrait,
    I::Model: FromQueryResult + Send + Sync + Clone,
{
    /// Reads the next item from the reader.
    ///
    /// This method manages pagination internally and provides a simple interface
    /// for sequential reading. The complexity of pagination is hidden from the caller.
    ///
    /// # Reading Algorithm
    ///
    /// 1. **Calculate buffer index**: Determine where we are within the current page
    /// 2. **Check if new page needed**: If at the start of a page, load new data
    /// 3. **Fetch from buffer**: Get the item at the current index
    /// 4. **Update counters**: Increment offset and page number as needed
    /// 5. **Return result**: Clone the item or return None if exhausted
    ///
    /// # Pagination Logic
    ///
    /// - **With pagination**: `index = offset % page_size`
    /// - **Without pagination**: `index = offset` (all data in one "page")
    /// - **New page trigger**: When `index == 0`, we need to load a new page
    /// - **Page advancement**: When we reach the end of a page, increment `current_page`
    ///
    /// # Memory Management
    ///
    /// Items are cloned when returned to ensure the caller owns the data.
    /// This prevents borrowing issues and allows the buffer to be modified
    /// for the next page load.
    ///
    /// # Returns
    ///
    /// - `Ok(Some(item))` if an item was successfully read
    /// - `Ok(None)` if there are no more items to read
    /// - `Err(BatchError)` if an error occurred during reading (e.g., database error)
    ///
    /// # Examples
    ///
    /// ```
    /// use spring_batch_rs::item::orm::OrmItemReaderBuilder;
    /// use spring_batch_rs::core::item::ItemReader;
    /// use sea_orm::FromQueryResult;
    /// use serde::{Deserialize, Serialize};
    ///
    /// #[derive(Debug, Clone, FromQueryResult, Deserialize, Serialize)]
    /// struct Product {
    ///     id: i32,
    ///     name: String,
    ///     price: f64,
    /// }
    ///
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// // Assuming you have a database connection and query set up
    /// // let reader = OrmItemReaderBuilder::new()
    /// //     .connection(&db)
    /// //     .query(products::Entity::find())
    /// //     .page_size(50)
    /// //     .build();
    ///
    /// // Read all products
    /// // let mut products = Vec::new();
    /// // while let Some(product) = reader.read()? {
    /// //     products.push(product);
    /// // }
    /// # Ok(())
    /// # }
    /// ```
    fn read(&self) -> ItemReaderResult<I::Model> {
        // Calculate the index within the current page
        // This determines where we are in the current buffer
        let index = if let Some(page_size) = self.page_size {
            // With pagination: index is position within the current page
            self.offset.get() % page_size
        } else {
            // Without pagination: index is the absolute position
            self.offset.get()
        };

        // When index is 0, we've reached the start of a new page
        // or this is the first read operation, so we need to fetch data
        if index == 0 {
            self.read_page()?
        }

        // Retrieve the item at the current index from the buffer
        let buffer = self.buffer.borrow();
        let result = buffer.get(index as usize);

        match result {
            Some(item) => {
                // We found an item at the current position

                // Increment the offset for the next read operation
                // This moves us to the next item in the sequence
                self.offset.set(self.offset.get() + 1);

                // If we're using pagination and have reached the end of the current page,
                // increment the page number for the next page load
                if let Some(page_size) = self.page_size {
                    if self.offset.get().is_multiple_of(page_size) {
                        // We've read all items in the current page
                        // Move to the next page for the next read cycle
                        self.current_page.set(self.current_page.get() + 1);
                    }
                }

                // Clone the item to give ownership to the caller
                // This prevents borrowing conflicts with the buffer
                Ok(Some(item.clone()))
            }
            None => {
                // No more items in the current buffer
                // This means we've reached the end of the data

                // If we're using pagination, this might mean we've reached the end of all data
                // If not using pagination, this definitely means we're done
                Ok(None)
            }
        }
    }
}

/// Builder for creating a `OrmItemReader`.
///
/// This builder provides a convenient way to configure and create a `OrmItemReader`
/// with custom parameters like page size and database connection.
/// It follows the builder pattern to ensure all required components are provided
/// before creating the reader instance.
///
/// # Builder Pattern Benefits
///
/// - **Fluent API**: Method chaining for readable configuration
/// - **Compile-time Safety**: Required fields are enforced at build time
/// - **Flexibility**: Optional parameters can be omitted
/// - **Validation**: Build method validates all required components are present
///
/// # Required Components
///
/// The following components must be set before calling `build()`:
/// - **Connection**: Database connection for executing queries
/// - **Query**: SeaORM select query to execute
///
/// # Optional Components
///
/// - **Page Size**: When set, enables pagination for memory-efficient reading
///
/// # Usage Pattern
///
/// ```text
/// let reader = OrmItemReaderBuilder::new()
///     .connection(&db)           // Required
///     .query(entity_query)       // Required  
///     .page_size(100)           // Optional
///     .build();                 // Creates the reader
/// ```
///
/// # Examples
///
/// ```
/// use spring_batch_rs::item::orm::OrmItemReaderBuilder;
/// use sea_orm::{Database, EntityTrait, FromQueryResult};
/// use serde::{Deserialize, Serialize};
///
/// #[derive(Debug, Clone, FromQueryResult, Deserialize, Serialize)]
/// struct Order {
///     id: i32,
///     customer_name: String,
///     total_amount: f64,
///     status: String,
/// }
///
/// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
/// // Create database connection
/// let db = Database::connect("postgresql://user:pass@localhost/db").await?;
///
/// // Create a query (assuming you have an orders entity)
/// // let query = orders::Entity::find()
/// //     .filter(orders::Column::Status.eq("pending"));
///
/// // Create the reader with explicit type annotations
/// // let reader: OrmItemReader<orders::Entity> = OrmItemReaderBuilder::new()
/// //     .connection(&db)
/// //     .query(query)
/// //     .page_size(100)  // Process 100 orders at a time
/// //     .build();
/// # Ok(())
/// # }
/// ```
pub struct OrmItemReaderBuilder<'a, I>
where
    I: EntityTrait,
{
    /// Database connection - None until set by the user
    /// This will be validated as required during build()
    connection: Option<&'a DatabaseConnection>,

    /// SeaORM select query - None until set by the user
    /// This will be validated as required during build()
    query: Option<Select<I>>,

    /// Optional page size for pagination
    /// When None, all data will be loaded at once
    /// When Some(size), data will be loaded in chunks
    page_size: Option<u64>,
}

impl<I> Default for OrmItemReaderBuilder<'_, I>
where
    I: EntityTrait,
{
    /// Creates a new builder with all fields set to None/default values.
    ///
    /// This is the starting point for the builder pattern. All required
    /// fields must be set before calling `build()`.
    fn default() -> Self {
        Self {
            connection: None,
            query: None,
            page_size: None,
        }
    }
}

impl<'a, I> OrmItemReaderBuilder<'a, I>
where
    I: EntityTrait,
    I::Model: FromQueryResult + Send + Sync + Clone,
{
    /// Sets the page size for the reader.
    ///
    /// When set, the reader will use pagination to load data in chunks of the specified size.
    /// This is useful for processing large datasets without loading everything into memory.
    ///
    /// # Memory Management
    ///
    /// - **Small page sizes** (e.g., 10-100): Lower memory usage, more database round trips
    /// - **Large page sizes** (e.g., 1000-10000): Higher memory usage, fewer database round trips
    /// - **No pagination** (None): All data loaded at once, highest memory usage
    ///
    /// # Performance Considerations
    ///
    /// - Choose page size based on available memory and network latency
    /// - Consider the size of your data rows when setting page size
    /// - Monitor memory usage and adjust as needed
    ///
    /// # Arguments
    ///
    /// * `page_size` - The number of items to read per page (must be > 0 for meaningful pagination)
    ///
    /// # Returns
    ///
    /// The updated `OrmItemReaderBuilder` instance for method chaining.
    ///
    /// # Examples
    ///
    /// ```
    /// use spring_batch_rs::item::orm::OrmItemReaderBuilder;
    /// use sea_orm::FromQueryResult;
    /// use serde::{Deserialize, Serialize};
    ///
    /// // In practice, you would use actual SeaORM entities
    /// // #[derive(Debug, Clone, FromQueryResult)]
    /// // struct Record {
    /// //     id: i32,
    /// //     data: String,
    /// // }
    /// //
    /// // let builder = OrmItemReaderBuilder::<record::Entity>::new()
    /// //     .page_size(50);  // Process 50 records at a time
    /// ```
    pub fn page_size(mut self, page_size: u64) -> Self {
        self.page_size = Some(page_size);
        self
    }

    /// Sets the SeaORM select query for the reader.
    ///
    /// This query will be executed to fetch data from the database. The query can include
    /// filters, joins, ordering, and other SeaORM query operations. The query will be
    /// cloned for each page fetch, so it should be relatively lightweight to clone.
    ///
    /// # Query Design Considerations
    ///
    /// - **Ordering**: Include `ORDER BY` clauses for consistent pagination
    /// - **Filtering**: Apply filters to reduce the dataset size
    /// - **Joins**: Use joins to fetch related data in a single query
    /// - **Indexing**: Ensure proper database indexes for query performance
    ///
    /// # Pagination Compatibility
    ///
    /// When using pagination, the query should:
    /// - Have a deterministic order (use ORDER BY)
    /// - Not use LIMIT/OFFSET (handled by the paginator)
    /// - Be compatible with SeaORM's paginator
    ///
    /// # Arguments
    ///
    /// * `query` - The SeaORM select query to execute
    ///
    /// # Returns
    ///
    /// The updated `OrmItemReaderBuilder` instance for method chaining.
    ///
    /// # Examples
    ///
    /// ```
    /// use spring_batch_rs::item::orm::OrmItemReaderBuilder;
    /// use sea_orm::{EntityTrait, QueryFilter};
    /// use serde::{Deserialize, Serialize};
    ///
    /// // In practice, you would use actual SeaORM entities
    /// // let query = user::Entity::find()
    /// //     .filter(user::Column::Active.eq(true))
    /// //     .order_by_asc(user::Column::Id);
    /// //
    /// // let builder = OrmItemReaderBuilder::<user::Entity>::new()
    /// //     .query(query);
    /// ```
    pub fn query(mut self, query: Select<I>) -> Self {
        self.query = Some(query);
        self
    }

    /// Sets the database connection for the reader.
    ///
    /// This connection will be used to execute the SeaORM queries. The connection
    /// must remain valid for the entire lifetime of the reader, which is enforced
    /// by the lifetime parameter 'a.
    ///
    /// # Connection Management
    ///
    /// - The connection is borrowed, not owned by the reader
    /// - Ensure the connection pool/manager outlives the reader
    /// - The connection should be properly configured for your database
    ///
    /// # Database Compatibility
    ///
    /// SeaORM supports multiple databases:
    /// - PostgreSQL
    /// - MySQL
    /// - SQLite
    /// - SQL Server (via sqlx)
    ///
    /// # Arguments
    ///
    /// * `connection` - The SeaORM database connection (must outlive the reader)
    ///
    /// # Returns
    ///
    /// The updated `OrmItemReaderBuilder` instance for method chaining.
    ///
    /// # Examples
    ///
    /// ```
    /// use spring_batch_rs::item::orm::OrmItemReaderBuilder;
    /// use sea_orm::{Database, EntityTrait};
    /// use serde::{Deserialize, Serialize};
    ///
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// let db = Database::connect("sqlite::memory:").await?;
    ///
    /// // In practice, you would use actual SeaORM entities
    /// // let builder = OrmItemReaderBuilder::<product::Entity>::new()
    /// //     .connection(&db);
    /// # Ok(())
    /// # }
    /// ```
    pub fn connection(mut self, connection: &'a DatabaseConnection) -> Self {
        self.connection = Some(connection);
        self
    }

    /// Builds the ORM item reader with the configured parameters.
    ///
    /// This method validates that all required parameters have been set and creates
    /// a new `OrmItemReader` instance.
    ///
    /// # Returns
    /// A configured `OrmItemReader` instance
    ///
    /// # Panics
    /// Panics if required parameters (connection and query) are missing.
    ///
    /// # Validation
    ///
    /// The builder performs the following validation:
    /// - Ensures a database connection has been provided
    /// - Ensures a SeaORM query has been provided
    /// - Page size is optional and defaults to loading all data at once
    ///
    /// If any required parameter is missing, the method will panic with a descriptive error message.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// use spring_batch_rs::item::orm::OrmItemReaderBuilder;
    /// use sea_orm::{Database, EntityTrait};
    ///
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// let db = Database::connect("sqlite::memory:").await?;
    ///
    /// // With pagination
    /// // let reader = OrmItemReaderBuilder::new()
    /// //     .connection(&db)
    /// //     .query(MyEntity::find())
    /// //     .page_size(100)
    /// //     .build();
    ///
    /// // Without pagination (load all at once)
    /// // let reader = OrmItemReaderBuilder::new()
    /// //     .connection(&db)
    /// //     .query(MyEntity::find())
    /// //     .build();
    /// # Ok(())
    /// # }
    /// ```
    pub fn build(self) -> OrmItemReader<'a, I> {
        let mut reader = OrmItemReader::new()
            .connection(self.connection.expect("Database connection is required"))
            .query(self.query.expect("Query is required"));

        if let Some(page_size) = self.page_size {
            reader = reader.page_size(page_size);
        }

        reader
    }

    /// Creates a new `OrmItemReaderBuilder`.
    ///
    /// This is the entry point for the builder pattern. It creates a new builder
    /// instance with all fields set to their default values (None for optional fields).
    ///
    /// # Builder Lifecycle
    ///
    /// 1. **Creation**: `new()` creates an empty builder
    /// 2. **Configuration**: Use setter methods to configure the reader
    /// 3. **Validation**: `build()` validates and creates the final reader
    ///
    /// # Examples
    ///
    /// ```
    /// use spring_batch_rs::item::orm::OrmItemReaderBuilder;
    /// use sea_orm::EntityTrait;
    /// use serde::{Deserialize, Serialize};
    ///
    /// // In practice, you would use actual SeaORM entities
    /// // let builder = OrmItemReaderBuilder::<record::Entity>::new()
    /// //     .page_size(50);  // Process 50 records at a time
    /// ```
    pub fn new() -> Self {
        Self::default()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::core::item::ItemReader;
    use sea_orm::{
        entity::prelude::*, ActiveValue::Set, DatabaseBackend, MockDatabase, MockExecResult,
    };

    // Minimal entity definition for testing
    #[derive(Clone, Debug, PartialEq, Eq, DeriveEntityModel)]
    #[sea_orm(table_name = "record")]
    pub struct Model {
        #[sea_orm(primary_key)]
        pub id: i32,
        pub name: String,
    }

    #[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)]
    pub enum Relation {}

    impl ActiveModelBehavior for ActiveModel {}

    // --- OrmItemReader unit tests ---

    #[test]
    fn should_create_reader_with_default_state() {
        let reader = OrmItemReader::<Entity>::new();
        assert!(
            reader.connection.is_none(),
            "connection should start as None"
        );
        assert_eq!(reader.page_size, None);
        assert_eq!(reader.offset.get(), 0);
        assert_eq!(reader.current_page.get(), 0);
        assert!(reader.buffer.borrow().is_empty());
    }

    #[test]
    fn should_set_page_size_via_method() {
        let reader = OrmItemReader::<Entity>::new().page_size(50);
        assert_eq!(reader.page_size, Some(50));
        // buffer capacity should be pre-allocated
        assert_eq!(reader.buffer.borrow().capacity(), 50);
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn should_return_none_when_database_is_empty() {
        let db = MockDatabase::new(DatabaseBackend::Sqlite)
            .append_query_results([Vec::<Model>::new()])
            .into_connection();

        let reader = OrmItemReader::<Entity>::new()
            .connection(&db)
            .query(Entity::find());

        let result = reader.read().unwrap();
        assert!(result.is_none(), "empty DB should yield None");
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn should_read_single_item_then_return_none() {
        let db = MockDatabase::new(DatabaseBackend::Sqlite)
            .append_query_results([vec![Model {
                id: 1,
                name: "Alice".to_string(),
            }]])
            .into_connection();

        let reader = OrmItemReader::<Entity>::new()
            .connection(&db)
            .query(Entity::find());

        let first = reader.read().unwrap().expect("first item should exist");
        assert_eq!(first.name, "Alice");
        assert_eq!(reader.offset.get(), 1);

        let second = reader.read().unwrap();
        assert!(second.is_none(), "should return None after the only item");
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn should_read_multiple_items_without_pagination() {
        let db = MockDatabase::new(DatabaseBackend::Sqlite)
            .append_query_results([vec![
                Model {
                    id: 1,
                    name: "Alice".to_string(),
                },
                Model {
                    id: 2,
                    name: "Bob".to_string(),
                },
            ]])
            .into_connection();

        let reader = OrmItemReader::<Entity>::new()
            .connection(&db)
            .query(Entity::find());

        let a = reader.read().unwrap().unwrap();
        assert_eq!(a.name, "Alice");
        let b = reader.read().unwrap().unwrap();
        assert_eq!(b.name, "Bob");
        assert!(reader.read().unwrap().is_none());
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn should_paginate_across_multiple_pages() {
        // page_size=1: two DB calls, each returning one item, then empty
        let db = MockDatabase::new(DatabaseBackend::Sqlite)
            .append_query_results([
                vec![Model {
                    id: 1,
                    name: "P1".to_string(),
                }],
                vec![Model {
                    id: 2,
                    name: "P2".to_string(),
                }],
                vec![], // signals end of data
            ])
            .into_connection();

        let reader = OrmItemReader::<Entity>::new()
            .connection(&db)
            .query(Entity::find())
            .page_size(1);

        let first = reader.read().unwrap().unwrap();
        assert_eq!(first.name, "P1");
        assert_eq!(
            reader.current_page.get(),
            1,
            "page should advance after full page"
        );

        let second = reader.read().unwrap().unwrap();
        assert_eq!(second.name, "P2");

        assert!(
            reader.read().unwrap().is_none(),
            "should stop after all pages"
        );
    }

    // --- OrmItemReaderBuilder unit tests ---

    #[test]
    fn should_create_builder_with_default_state() {
        let builder = OrmItemReaderBuilder::<Entity>::new();
        assert!(builder.connection.is_none());
        assert!(builder.query.is_none());
        assert_eq!(builder.page_size, None);
    }

    #[test]
    fn should_set_page_size_on_builder() {
        let builder = OrmItemReaderBuilder::<Entity>::new().page_size(100);
        assert_eq!(builder.page_size, Some(100));
    }

    #[test]
    #[should_panic(expected = "Database connection is required")]
    fn should_panic_when_building_without_connection() {
        OrmItemReaderBuilder::<Entity>::new()
            .query(Entity::find())
            .build();
    }

    #[test]
    #[should_panic(expected = "Query is required")]
    fn should_panic_when_building_without_query() {
        let db = MockDatabase::new(DatabaseBackend::Sqlite).into_connection();
        OrmItemReaderBuilder::<Entity>::new()
            .connection(&db)
            .build();
    }

    #[test]
    fn should_build_reader_with_connection_and_query() {
        let db = MockDatabase::new(DatabaseBackend::Sqlite).into_connection();
        let reader = OrmItemReaderBuilder::<Entity>::new()
            .connection(&db)
            .query(Entity::find())
            .build();
        assert!(reader.connection.is_some());
        assert_eq!(reader.page_size, None);
    }

    #[test]
    fn should_build_reader_with_page_size() {
        let db = MockDatabase::new(DatabaseBackend::Sqlite).into_connection();
        let reader = OrmItemReaderBuilder::<Entity>::new()
            .connection(&db)
            .query(Entity::find())
            .page_size(50)
            .build();
        assert_eq!(reader.page_size, Some(50));
    }
}