adaptive_pipeline/infrastructure/repositories/

sqlite_pipeline.rs

// /////////////////////////////////////////////////////////////////////////////
// Adaptive Pipeline
// Copyright (c) 2025 Michael Gardner, A Bit of Help, Inc.
// SPDX-License-Identifier: BSD-3-Clause
// See LICENSE file in the project root.
// /////////////////////////////////////////////////////////////////////////////

//! # SQLite Pipeline Repository Adapter
//!
//! Implements the pipeline repository interface using SQLite for persistence.
//! Follows Hexagonal Architecture with proper relational schema, ACID
//! transactions, connection pooling, and parameterized queries for security.
//! See mdBook for detailed schema documentation and usage examples.

use adaptive_pipeline_domain::entities::pipeline_stage::{StageConfiguration, StageType};
use adaptive_pipeline_domain::value_objects::PipelineId;
use adaptive_pipeline_domain::{Pipeline, PipelineError, PipelineStage, ProcessingMetrics};
use sqlx::{Row, SqlitePool};
use std::collections::HashMap;
use tracing::debug;
// REMOVED: Generic Repository import - violates DIP
// DDD Principle: Use only domain-specific repository interfaces

/// Structured SQLite pipeline repository using proper database columns
///
/// This implementation provides a concrete SQLite-based implementation of the
/// pipeline repository interface, following Domain-Driven Design principles
/// and proper relational database design patterns.
///
/// # Key Features
///
/// - **Relational Design**: Proper normalized database schema with separate
///   tables
/// - **Type Safety**: Strong typing with compile-time query validation using
///   sqlx
/// - **Transaction Support**: ACID transactions for data consistency
/// - **Performance Optimization**: Efficient queries with proper indexing
/// - **Error Handling**: Comprehensive error handling and recovery
///
/// # Database Schema
///
/// The repository uses a normalized relational schema:
/// - **pipelines**: Main pipeline entity data
/// - **pipeline_stages**: Pipeline stage configurations
/// - **pipeline_metrics**: Performance and execution metrics
///
/// # Architecture
///
/// This implementation avoids JSON serialization issues by using proper
/// relational database design with separate tables for pipeline data,
/// configuration, stages, and metrics.
///
/// # Examples
///
///
/// # Visibility
///
/// - **Public**: For dependency injection and external usage
/// - **Private Fields**: Database connection pool is encapsulated
pub struct SqlitePipelineRepository {
    // PRIVATE: Database connection pool - internal implementation detail
    pool: SqlitePool,
}

impl SqlitePipelineRepository {
    /// Creates a new structured pipeline repository with database connection
    ///
    /// This constructor establishes a connection pool to the SQLite database,
    /// which will be used for all subsequent repository operations. The
    /// connection pool provides efficient resource management and supports
    /// concurrent access.
    ///
    /// # Why Connection Pooling?
    ///
    /// Connection pooling is used because:
    /// 1. **Performance**: Reusing connections is faster than creating new ones
    /// 2. **Resource Management**: Limits the number of open database
    ///    connections
    /// 3. **Concurrency**: Allows multiple operations to share connections
    ///    safely
    /// 4. **Reliability**: Automatically handles connection failures and
    ///    retries
    ///
    /// # Arguments
    ///
    /// * `database_path` - Path to the SQLite database file, or special values:
    ///   - `:memory:` or `sqlite::memory:` for in-memory database (useful for
    ///     testing)
    ///   - Any file path like `"data/pipelines.db"` for persistent storage
    ///
    /// # Returns
    ///
    /// * `Ok(SqlitePipelineRepository)` - Successfully connected repository
    /// * `Err(PipelineError)` - Connection failed with error details
    ///
    /// # Errors
    ///
    /// This function returns an error if:
    /// - The database file cannot be opened or created
    /// - File permissions prevent access
    /// - The database format is incompatible
    /// - The connection URL is malformed
    ///
    /// # Examples
    ///
    ///
    /// # Implementation Notes
    ///
    /// The function normalizes the database path to sqlx's expected format:
    /// - `:memory:` → `sqlite::memory:`
    /// - File paths → `sqlite://<path>`
    pub async fn new(database_path: &str) -> Result<Self, PipelineError> {
        debug!("Creating SqlitePipelineRepository with database: {}", database_path);

        // Build a proper SQLite connection URL. sqlx expects either:
        // - "sqlite::memory:" for in-memory DB, or
        // - "sqlite://<path>" for file-backed DB
        let database_url = if database_path == ":memory:" || database_path == "sqlite::memory:" {
            "sqlite::memory:".to_string()
        } else {
            format!("sqlite://{}", database_path)
        };

        // Use schema initialization which handles database creation and migrations
        let pool = crate::infrastructure::repositories::schema::initialize_database(&database_url)
            .await
            .map_err(|e| {
                PipelineError::database_error(format!("Failed to initialize database '{}': {}", database_path, e))
            })?;

        debug!("Successfully connected to structured SQLite database");
        Ok(Self { pool })
    }

    /// Saves a pipeline to the database with ACID transaction guarantees
    ///
    /// This method persists a complete pipeline entity to the database,
    /// including all associated data: configuration parameters, stages, and
    /// stage parameters. The entire operation is wrapped in a database
    /// transaction to ensure atomicity - either all data is saved
    /// successfully, or none of it is.
    ///
    /// # Why ACID Transactions?
    ///
    /// ACID (Atomicity, Consistency, Isolation, Durability) transactions
    /// ensure:
    /// 1. **Atomicity**: All-or-nothing - if any part fails, everything rolls
    ///    back
    /// 2. **Consistency**: Database constraints are always maintained
    /// 3. **Isolation**: Concurrent operations don't interfere with each other
    /// 4. **Durability**: Once committed, data survives system crashes
    ///
    /// # What Gets Saved?
    ///
    /// The method saves to multiple related tables:
    /// - **pipelines**: Main pipeline record (id, name, archived status,
    ///   timestamps)
    /// - **pipeline_configuration**: Key-value configuration parameters
    /// - **pipeline_stages**: Processing stages with their configurations
    /// - **stage_parameters**: Parameters for each stage
    ///
    /// # Arguments
    ///
    /// * `entity` - The pipeline entity to save. Must be a complete, valid
    ///   pipeline with all required fields populated.
    ///
    /// # Returns
    ///
    /// * `Ok(())` - Pipeline saved successfully
    /// * `Err(PipelineError)` - Save operation failed, transaction rolled back
    ///
    /// # Errors
    ///
    /// This function returns an error if:
    /// - A pipeline with the same ID already exists (unique constraint
    ///   violation)
    /// - Database connection is lost during the operation
    /// - Any SQL query fails (syntax error, constraint violation, etc.)
    /// - Transaction cannot be started or committed
    ///
    /// Note: If an error occurs, the transaction is automatically rolled back,
    /// leaving the database in its original state.
    ///
    /// # Examples
    ///
    ///
    /// # Thread Safety
    ///
    /// This method is safe to call concurrently from multiple tasks. The
    /// database connection pool handles concurrent access, and transaction
    /// isolation prevents interference between concurrent saves.
    ///
    /// # Performance
    ///
    /// - **Complexity**: O(n + m) where n = number of config entries, m =
    ///   number of stages
    /// - **Database Writes**: Multiple INSERT statements within one transaction
    /// - **Network**: Single round-trip for transaction commit
    /// - **Locking**: Row-level locks acquired during transaction
    pub async fn save(&self, entity: &Pipeline) -> Result<(), PipelineError> {
        debug!(
            pipeline_name = %entity.name(),
            "SqlitePipelineRepository::save called"
        );

        // Start database transaction for ACID compliance
        let mut tx = self
            .pool
            .begin()
            .await
            .map_err(|e| PipelineError::database_error(format!("Failed to start transaction: {}", e)))?;

        // Insert main pipeline record
        let pipeline_query = r#"
            INSERT INTO pipelines (id, name, archived, created_at, updated_at)
            VALUES (?, ?, ?, ?, ?)
        "#;

        sqlx::query(pipeline_query)
            .bind(entity.id().to_string())
            .bind(entity.name())
            .bind(entity.archived())
            .bind(entity.created_at().to_rfc3339())
            .bind(entity.updated_at().to_rfc3339())
            .execute(&mut *tx)
            .await
            .map_err(|e| PipelineError::database_error(format!("Failed to insert pipeline: {}", e)))?;

        // Insert pipeline configuration
        for (key, value) in entity.configuration() {
            let config_query = r#"
                INSERT INTO pipeline_configuration (pipeline_id, key, value, created_at, updated_at)
                VALUES (?, ?, ?, ?, ?)
            "#;

            sqlx::query(config_query)
                .bind(entity.id().to_string())
                .bind(key)
                .bind(value)
                .bind(entity.created_at().to_rfc3339())
                .bind(entity.updated_at().to_rfc3339())
                .execute(&mut *tx)
                .await
                .map_err(|e| PipelineError::database_error(format!("Failed to insert configuration: {}", e)))?;
        }

        // Insert pipeline stages
        for (index, stage) in entity.stages().iter().enumerate() {
            let stage_query = r#"
                INSERT INTO pipeline_stages (
                    id, pipeline_id, name, stage_type, enabled, stage_order, 
                    algorithm, parallel_processing, chunk_size, created_at, updated_at
                )
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            "#;

            sqlx::query(stage_query)
                .bind(stage.id().to_string())
                .bind(entity.id().to_string())
                .bind(stage.name())
                .bind(stage.stage_type().to_string())
                .bind(stage.is_enabled())
                .bind(index as i32)
                .bind(&stage.configuration().algorithm)
                .bind(stage.configuration().parallel_processing)
                .bind(stage.configuration().chunk_size.map(|s| s as i64))
                .bind(stage.created_at().to_rfc3339())
                .bind(stage.updated_at().to_rfc3339())
                .execute(&mut *tx)
                .await
                .map_err(|e| PipelineError::database_error(format!("Failed to insert stage: {}", e)))?;

            // Insert stage parameters
            for (param_key, param_value) in &stage.configuration().parameters {
                let param_query = r#"
                    INSERT INTO stage_parameters (stage_id, key, value, created_at, updated_at)
                    VALUES (?, ?, ?, ?, ?)
                "#;

                sqlx::query(param_query)
                    .bind(stage.id().to_string())
                    .bind(param_key)
                    .bind(param_value)
                    .bind(stage.created_at().to_rfc3339())
                    .bind(stage.updated_at().to_rfc3339())
                    .execute(&mut *tx)
                    .await
                    .map_err(|e| PipelineError::database_error(format!("Failed to insert stage parameter: {}", e)))?;
            }
        }

        // NOTE: Metrics are handled by Prometheus (per SRS requirements), not stored in
        // database Skip metrics insertion - observability is handled externally
        // This keeps the database focused on core pipeline data only

        // Commit transaction - ensures ACID compliance
        tx.commit()
            .await
            .map_err(|e| PipelineError::database_error(format!("Failed to commit transaction: {}", e)))?;

        debug!(
            pipeline_name = %entity.name(),
            "Successfully saved pipeline with ACID transaction"
        );
        Ok(())
    }

    /// PUBLIC: Domain interface - Find pipeline by ID
    pub async fn find_by_id(&self, id: PipelineId) -> Result<Option<Pipeline>, PipelineError> {
        self.load_pipeline_from_db(id).await
    }

    /// PUBLIC: Domain interface - Update a pipeline
    pub async fn update(&self, pipeline: &Pipeline) -> Result<(), PipelineError> {
        // Implementation simplified for now
        debug!(
            pipeline_name = %pipeline.name(),
            "SqlitePipelineRepository::update called"
        );
        Ok(())
    }

    /// PUBLIC: Domain interface - Soft delete a pipeline with cascading archive
    pub async fn delete(&self, id: PipelineId) -> Result<bool, PipelineError> {
        debug!(pipeline_id = %id, "Starting delete for pipeline");

        let mut tx = self
            .pool
            .begin()
            .await
            .map_err(|e| PipelineError::database_error(format!("Failed to begin transaction: {}", e)))?;

        let now = chrono::Utc::now().to_rfc3339();
        let id_str = id.to_string();

        debug!("Archiving pipeline stages...");
        // Archive pipeline stages first
        let stages_query = r#"
            UPDATE pipeline_stages 
            SET archived = true, updated_at = ?
            WHERE pipeline_id = ? AND archived = false
        "#;

        let stages_result = sqlx::query(stages_query)
            .bind(&now)
            .bind(&id_str)
            .execute(&mut *tx)
            .await
            .map_err(|e| PipelineError::database_error(format!("Failed to archive pipeline stages: {}", e)))?;

        debug!(
            stages_archived = stages_result.rows_affected(),
            "Archived pipeline stages"
        );

        debug!("Archiving stage parameters...");
        // Archive stage parameters
        let params_query = r#"
            UPDATE stage_parameters 
            SET archived = true, updated_at = ?
            WHERE stage_id IN (
                SELECT id FROM pipeline_stages 
                WHERE pipeline_id = ?
            ) AND archived = false
        "#;

        let params_result = sqlx::query(params_query)
            .bind(&now)
            .bind(&id_str)
            .execute(&mut *tx)
            .await
            .map_err(|e| PipelineError::database_error(format!("Failed to archive stage parameters: {}", e)))?;

        debug!(
            parameters_archived = params_result.rows_affected(),
            "Archived stage parameters"
        );

        debug!("Archiving pipeline configuration...");
        // Archive pipeline configuration
        let config_query = r#"
            UPDATE pipeline_configuration 
            SET archived = true, updated_at = ?
            WHERE pipeline_id = ? AND archived = false
        "#;

        let config_result = sqlx::query(config_query)
            .bind(&now)
            .bind(&id_str)
            .execute(&mut *tx)
            .await
            .map_err(|e| PipelineError::database_error(format!("Failed to archive pipeline configuration: {}", e)))?;

        debug!(
            config_entries_archived = config_result.rows_affected(),
            "Archived config entries"
        );

        debug!("Archiving main pipeline...");
        // Finally, archive the main pipeline record
        let pipeline_query = r#"
            UPDATE pipelines 
            SET archived = true, updated_at = ?
            WHERE id = ? AND archived = false
        "#;

        let result = sqlx::query(pipeline_query)
            .bind(&now)
            .bind(&id_str)
            .execute(&mut *tx)
            .await
            .map_err(|e| PipelineError::database_error(format!("Failed to archive pipeline: {}", e)))?;

        let success = result.rows_affected() > 0;
        debug!(
            success = success,
            rows_affected = result.rows_affected(),
            "Pipeline archive result"
        );

        if success {
            tx.commit()
                .await
                .map_err(|e| PipelineError::database_error(format!("Failed to commit archive transaction: {}", e)))?;
            debug!("Transaction committed successfully");
        } else {
            tx.rollback()
                .await
                .map_err(|e| PipelineError::database_error(format!("Failed to rollback archive transaction: {}", e)))?;
            debug!("Transaction rolled back");
        }

        Ok(success)
    }

    /// PUBLIC: Domain interface - List all active pipelines
    pub async fn list_all(&self) -> Result<Vec<Pipeline>, PipelineError> {
        debug!("SqlitePipelineRepository::list_all called (excluding archived)");

        // Get all non-archived pipelines
        let query = "SELECT id FROM pipelines WHERE archived = false ORDER BY name";
        let rows = sqlx::query(query)
            .fetch_all(&self.pool)
            .await
            .map_err(|e| PipelineError::database_error(format!("Failed to query pipelines: {}", e)))?;

        let mut pipelines = Vec::new();
        for row in rows {
            let id_str: String = row.get("id");
            let pipeline_id = PipelineId::from_string(&id_str)?;

            if let Some(pipeline) = self.load_pipeline_from_db(pipeline_id).await? {
                pipelines.push(pipeline);
            }
        }

        debug!(pipeline_count = pipelines.len(), "Found active pipelines");
        Ok(pipelines)
    }

    /// PUBLIC: Domain interface - Find all active pipelines (alias for
    /// list_all)
    pub async fn find_all(&self) -> Result<Vec<Pipeline>, PipelineError> {
        self.list_all().await
    }

    /// PUBLIC: Domain interface - List archived pipelines
    pub async fn list_archived(&self) -> Result<Vec<Pipeline>, PipelineError> {
        debug!("SqlitePipelineRepository::list_archived called");

        // Get all archived pipelines
        let query = "SELECT id FROM pipelines WHERE archived = true ORDER BY name";
        let rows = sqlx::query(query)
            .fetch_all(&self.pool)
            .await
            .map_err(|e| PipelineError::database_error(format!("Failed to query pipelines: {}", e)))?;

        let mut pipelines = Vec::new();
        for row in rows {
            let id_str: String = row.get("id");
            let pipeline_id = PipelineId::from_string(&id_str)?;

            if let Some(pipeline) = self.load_pipeline_from_db_with_archived(pipeline_id, true).await? {
                pipelines.push(pipeline);
            }
        }

        debug!(pipeline_count = pipelines.len(), "Found archived pipelines");
        Ok(pipelines)
    }

    /// PUBLIC: Domain interface - Check if pipeline exists
    pub async fn exists(&self, id: PipelineId) -> Result<bool, PipelineError> {
        let query = "SELECT 1 FROM pipelines WHERE id = ? AND archived = false";
        let result = sqlx::query(query)
            .bind(id.to_string())
            .fetch_optional(&self.pool)
            .await
            .map_err(|e| PipelineError::database_error(format!("Failed to check pipeline existence: {}", e)))?;

        Ok(result.is_some())
    }

    /// PUBLIC: Domain interface - Find pipeline by name
    pub async fn find_by_name(&self, name: &str) -> Result<Option<Pipeline>, PipelineError> {
        debug!("SqlitePipelineRepository::find_by_name called for: {}", name);

        let query = "SELECT id FROM pipelines WHERE name = ? AND archived = false";
        let row = sqlx::query(query)
            .bind(name)
            .fetch_optional(&self.pool)
            .await
            .map_err(|e| PipelineError::database_error(format!("Failed to find pipeline by name: {}", e)))?;

        if let Some(row) = row {
            let id_str: String = row.get("id");
            let pipeline_id = PipelineId::from_string(&id_str)?;
            self.load_pipeline_from_db(pipeline_id).await
        } else {
            debug!(pipeline_name = name, "No pipeline found with name");
            Ok(None)
        }
    }

    /// PUBLIC: Domain interface - List pipelines with pagination
    pub async fn list_paginated(&self, offset: usize, limit: usize) -> Result<Vec<Pipeline>, PipelineError> {
        let query = "SELECT id FROM pipelines WHERE archived = false ORDER BY name LIMIT ? OFFSET ?";
        let rows = sqlx::query(query)
            .bind(limit as i64)
            .bind(offset as i64)
            .fetch_all(&self.pool)
            .await
            .map_err(|e| PipelineError::database_error(format!("Failed to query paginated pipelines: {}", e)))?;

        let mut pipelines = Vec::new();
        for row in rows {
            let id_str: String = row.get("id");
            let pipeline_id = PipelineId::from_string(&id_str)?;

            if let Some(pipeline) = self.load_pipeline_from_db(pipeline_id).await? {
                pipelines.push(pipeline);
            }
        }

        Ok(pipelines)
    }

    /// PUBLIC: Domain interface - Count active pipelines
    pub async fn count(&self) -> Result<usize, PipelineError> {
        let count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM pipelines WHERE archived = false")
            .fetch_one(&self.pool)
            .await
            .map_err(|e| PipelineError::database_error(format!("Failed to count pipelines: {}", e)))?;

        Ok(count as usize)
    }

    /// PUBLIC: Domain interface - Find pipelines by configuration parameter
    pub async fn find_by_config(&self, key: &str, value: &str) -> Result<Vec<Pipeline>, PipelineError> {
        debug!(
            config_key = key,
            config_value = value,
            "SqlitePipelineRepository::find_by_config called"
        );

        let query = r#"
            SELECT DISTINCT p.id 
            FROM pipelines p 
            JOIN pipeline_configuration pc ON p.id = pc.pipeline_id 
            WHERE pc.key = ? AND pc.value = ? AND p.archived = false AND pc.archived = false
        "#;

        let rows = sqlx::query(query)
            .bind(key)
            .bind(value)
            .fetch_all(&self.pool)
            .await
            .map_err(|e| PipelineError::database_error(format!("Failed to find pipelines by config: {}", e)))?;

        let mut pipelines = Vec::new();
        for row in rows {
            let id_str: String = row.get("id");
            let pipeline_id = PipelineId::from_string(&id_str)?;

            if let Some(pipeline) = self.load_pipeline_from_db(pipeline_id).await? {
                pipelines.push(pipeline);
            }
        }

        debug!(
            pipeline_count = pipelines.len(),
            config_key = key,
            config_value = value,
            "Found pipelines with config"
        );
        Ok(pipelines)
    }

    /// PUBLIC: Domain interface - Archive a pipeline (soft delete)
    pub async fn archive(&self, id: PipelineId) -> Result<bool, PipelineError> {
        self.delete(id).await
    }

    /// PUBLIC: Domain interface - Restore an archived pipeline
    pub async fn restore(&self, id: PipelineId) -> Result<bool, PipelineError> {
        let query = r#"
            UPDATE pipelines 
            SET archived = false, updated_at = ?
            WHERE id = ? AND archived = true
        "#;

        let now = chrono::Utc::now().to_rfc3339();
        let result = sqlx::query(query)
            .bind(now)
            .bind(id.to_string())
            .execute(&self.pool)
            .await
            .map_err(|e| PipelineError::database_error(format!("Failed to restore pipeline: {}", e)))?;

        Ok(result.rows_affected() > 0)
    }

    // PRIVATE: Internal helper methods
    async fn load_pipeline_from_db(&self, id: PipelineId) -> Result<Option<Pipeline>, PipelineError> {
        self.load_pipeline_from_db_with_archived(id, false).await
    }

    async fn load_pipeline_from_db_with_archived(
        &self,
        id: PipelineId,
        include_archived: bool,
    ) -> Result<Option<Pipeline>, PipelineError> {
        debug!("Loading pipeline from structured DB: {}", id);

        // Load main pipeline record
        let pipeline_query = if include_archived {
            "SELECT id, name, archived, created_at, updated_at FROM pipelines WHERE id = ?"
        } else {
            "SELECT id, name, archived, created_at, updated_at FROM pipelines WHERE id = ? AND archived = false"
        };
        let pipeline_row = sqlx::query(pipeline_query)
            .bind(id.to_string())
            .fetch_optional(&self.pool)
            .await
            .map_err(|e| PipelineError::database_error(format!("Failed to load pipeline: {}", e)))?;

        let pipeline_row = match pipeline_row {
            Some(row) => row,
            None => {
                return Ok(None);
            }
        };

        // Parse pipeline data
        let name: String = pipeline_row.get("name");
        let archived: bool = pipeline_row.get("archived");
        let created_at_str: String = pipeline_row.get("created_at");
        let updated_at_str: String = pipeline_row.get("updated_at");

        let created_at = chrono::DateTime::parse_from_rfc3339(&created_at_str)
            .map_err(|e| PipelineError::SerializationError(format!("Invalid created_at format: {}", e)))?
            .with_timezone(&chrono::Utc);
        let updated_at = chrono::DateTime::parse_from_rfc3339(&updated_at_str)
            .map_err(|e| PipelineError::SerializationError(format!("Invalid updated_at format: {}", e)))?
            .with_timezone(&chrono::Utc);

        // Load configuration
        let config_query = "SELECT key, value FROM pipeline_configuration WHERE pipeline_id = ?";
        let config_rows = sqlx::query(config_query)
            .bind(id.to_string())
            .fetch_all(&self.pool)
            .await
            .map_err(|e| PipelineError::database_error(format!("Failed to load configuration: {}", e)))?;

        let mut configuration = HashMap::new();
        for row in config_rows {
            let key: String = row.get("key");
            let value: String = row.get("value");
            configuration.insert(key, value);
        }

        // Load stages
        let stage_query = r#"
            SELECT id, name, stage_type, enabled, stage_order, algorithm, 
                   parallel_processing, chunk_size, created_at, updated_at 
            FROM pipeline_stages 
            WHERE pipeline_id = ?
            ORDER BY stage_order
        "#;
        let stage_rows = sqlx::query(stage_query)
            .bind(id.to_string())
            .fetch_all(&self.pool)
            .await
            .map_err(|e| PipelineError::database_error(format!("Failed to load stages: {}", e)))?;

        let mut stages = Vec::new();
        for row in stage_rows {
            let stage_id_str: String = row.get("id");
            let stage_name: String = row.get("name");
            let stage_type_str: String = row.get("stage_type");
            let _enabled: bool = row.get("enabled");
            let stage_order: i32 = row.get("stage_order");
            let algorithm: String = row.get("algorithm");
            let parallel_processing: bool = row.get("parallel_processing");
            let chunk_size: Option<i64> = row.get("chunk_size");
            let created_at_str: String = row.get("created_at");
            let updated_at_str: String = row.get("updated_at");

            // Parse stage type
            let stage_type = stage_type_str
                .parse::<StageType>()
                .map_err(|e| PipelineError::SerializationError(format!("Invalid stage type: {}", e)))?;

            // Load stage parameters from stage_parameters table
            let params_query = "SELECT key, value FROM stage_parameters WHERE stage_id = ?";
            let params_rows = sqlx::query(params_query)
                .bind(&stage_id_str)
                .fetch_all(&self.pool)
                .await
                .map_err(|e| PipelineError::database_error(format!("Failed to load stage parameters: {}", e)))?;

            let mut parameters = std::collections::HashMap::new();
            for param_row in params_rows {
                let key: String = param_row.get("key");
                let value: String = param_row.get("value");
                parameters.insert(key, value);
            }

            // Build stage configuration
            let stage_config = StageConfiguration {
                algorithm,
                operation: adaptive_pipeline_domain::entities::Operation::Forward, // Default to Forward
                parameters,
                parallel_processing,
                chunk_size: chunk_size.map(|s| s as usize),
            };

            // Parse timestamps
            let _created_at = chrono::DateTime::parse_from_rfc3339(&created_at_str)
                .map_err(|e| PipelineError::SerializationError(format!("Invalid stage created_at format: {}", e)))?
                .with_timezone(&chrono::Utc);
            let _updated_at = chrono::DateTime::parse_from_rfc3339(&updated_at_str)
                .map_err(|e| PipelineError::SerializationError(format!("Invalid stage updated_at format: {}", e)))?
                .with_timezone(&chrono::Utc);

            // Create stage with proper arguments: name, stage_type, configuration, order
            let stage = PipelineStage::new(stage_name, stage_type, stage_config, stage_order as u32)?;

            // Set additional properties that can't be set via constructor
            // Note: We'd need setters or a from_database constructor for this
            // For now, we'll use the new constructor which creates new timestamps

            stages.push(stage);
        }

        // Load metrics (simplified for now)
        let metrics = ProcessingMetrics::new(0, 0); // TODO: Implement metrics loading

        // Construct DTO and reconstruct pipeline
        let data = adaptive_pipeline_domain::entities::pipeline::PipelineData {
            id,
            name,
            archived,
            configuration,
            metrics,
            stages,
            created_at,
            updated_at,
        };

        let pipeline = Pipeline::from_database(data)?;

        debug!("Successfully loaded pipeline: {}", pipeline.name());
        Ok(Some(pipeline))
    }
}

// Clean trait implementation that delegates to public methods
#[async_trait::async_trait]
impl adaptive_pipeline_domain::repositories::pipeline_repository::PipelineRepository for SqlitePipelineRepository {
    async fn save(&self, entity: &Pipeline) -> Result<(), PipelineError> {
        self.save(entity).await
    }

    async fn find_by_id(&self, id: PipelineId) -> Result<Option<Pipeline>, PipelineError> {
        self.find_by_id(id).await
    }

    async fn update(&self, pipeline: &Pipeline) -> Result<(), PipelineError> {
        self.update(pipeline).await
    }

    async fn delete(&self, id: PipelineId) -> Result<bool, PipelineError> {
        self.delete(id).await
    }

    async fn list_all(&self) -> Result<Vec<Pipeline>, PipelineError> {
        self.list_all().await
    }

    async fn find_all(&self) -> Result<Vec<Pipeline>, PipelineError> {
        self.find_all().await
    }

    async fn list_archived(&self) -> Result<Vec<Pipeline>, PipelineError> {
        self.list_archived().await
    }

    async fn exists(&self, id: PipelineId) -> Result<bool, PipelineError> {
        self.exists(id).await
    }

    async fn find_by_name(&self, name: &str) -> Result<Option<Pipeline>, PipelineError> {
        self.find_by_name(name).await
    }

    async fn list_paginated(&self, offset: usize, limit: usize) -> Result<Vec<Pipeline>, PipelineError> {
        self.list_paginated(offset, limit).await
    }

    async fn count(&self) -> Result<usize, PipelineError> {
        self.count().await
    }

    async fn find_by_config(&self, key: &str, value: &str) -> Result<Vec<Pipeline>, PipelineError> {
        self.find_by_config(key, value).await
    }

    async fn archive(&self, id: PipelineId) -> Result<bool, PipelineError> {
        self.archive(id).await
    }

    async fn restore(&self, id: PipelineId) -> Result<bool, PipelineError> {
        self.restore(id).await
    }
}

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

    /// Tests database URL formatting for various SQLite connection paths.
    #[test]
    fn test_database_url_formatting() {
        // Unit test: Test the database URL formatting logic
        // This tests the internal logic without requiring actual database connectivity
        // INFRASTRUCTURE CONCERN: Testing repository URL generation

        let test_cases = vec![
            ("/path/to/database.db", "sqlite:///path/to/database.db"),
            ("./local.db", "sqlite://./local.db"),
            (":memory:", "sqlite::memory:"),
            ("/tmp/test.db", "sqlite:///tmp/test.db"),
        ];

        for (input_path, expected_url) in test_cases {
            let formatted_url = if input_path == ":memory:" {
                "sqlite::memory:".to_string()
            } else {
                format!("sqlite://{}", input_path)
            };
            assert_eq!(
                formatted_url, expected_url,
                "Database URL formatting failed for path: {}",
                input_path
            );
        }
    }

    /// Tests repository error handling and PipelineError type mapping.
    #[test]
    fn test_error_handling() {
        // Unit test: Test repository error handling
        // INFRASTRUCTURE CONCERN: Testing error mapping and handling

        let db_error = PipelineError::DatabaseError("Connection failed".to_string());
        match db_error {
            PipelineError::DatabaseError(msg) => {
                assert_eq!(msg, "Connection failed");
            }
            _ => panic!("Expected DatabaseError"),
        }

        // Test using the helper method
        let db_error_helper = PipelineError::database_error("SQL syntax error");
        match db_error_helper {
            PipelineError::DatabaseError(msg) => {
                assert_eq!(msg, "SQL syntax error");
            }
            _ => panic!("Expected DatabaseError from helper"),
        }
    }

    // NOTE: Domain logic tests (Pipeline creation, Stage configuration, etc.)
    // have been moved to their proper domain entity files following DDD
    // principles. Repository tests should focus on infrastructure concerns
    // only.
    //
    // Repository functionality is tested through:
    // 1. Application layer integration tests (respecting DIP)
    // 2. The create_test_database utility (validates schema and operations)
    // 3. These focused infrastructure unit tests
}