heliosdb-nano 3.22.2

//! Branch and Time-Travel Diffing Engine
//!
//! Provides schema and data comparison between:
//! - Different database branches (Branch A vs Branch B)
//! - Different points in time within the same branch (LSN X vs LSN Y)
//! - Cross-branch and cross-time comparisons (Branch A @ LSN X vs Branch B @ LSN Y)
//!
//! ## Diff Levels
//!
//! - **Schema-only**: Fast catalog comparison (<100ms)
//! - **Schema + Sampled Data**: Moderate speed with row sampling (~1-5s)
//! - **Full Data Diff**: Comprehensive row-level comparison (scales with data)
//!
//! ## Implementation
//!
//! Uses query-time comparison via SQL EXCEPT/INTERSECT, not structural sharing.
//! This approach leverages HeliosDB-Lite's existing MVCC and time-travel capabilities.

#![allow(dead_code)]
#![allow(unused_variables)]

use crate::storage::{StorageEngine, BranchManager};
use crate::Result;
use serde::{Deserialize, Serialize};
use std::sync::Arc;
use std::collections::HashMap;

// ============================================================================
// Time-Travel Diff Target Types
// ============================================================================

/// Specifies a point in the database history for diffing
///
/// A DiffTarget can refer to:
/// - A branch at its current state
/// - A specific LSN (Log Sequence Number / Transaction ID)
/// - A specific SCN (System Change Number)
/// - A combination of branch + LSN/SCN for cross-branch time-travel
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub enum DiffTarget {
    /// Current state of a branch
    Branch(String),
    /// Branch at a specific LSN (Transaction ID)
    BranchAtLsn {
        branch: String,
        lsn: u64,
    },
    /// Branch at a specific SCN (System Change Number)
    BranchAtScn {
        branch: String,
        scn: u64,
    },
    /// Specific LSN on the current/default branch
    Lsn(u64),
    /// Specific SCN on the current/default branch
    Scn(u64),
    /// Current state (latest snapshot)
    Current,
}

impl DiffTarget {
    /// Parse a diff target from string format
    ///
    /// Supported formats:
    /// - `branch_name` - Current state of branch
    /// - `branch_name@lsn:123` - Branch at LSN 123
    /// - `branch_name@scn:456` - Branch at SCN 456
    /// - `@lsn:123` - Current branch at LSN 123
    /// - `@scn:456` - Current branch at SCN 456
    /// - `@current` or `HEAD` - Current state
    pub fn parse(input: &str) -> Result<Self> {
        let input = input.trim();

        if input.eq_ignore_ascii_case("@current") || input.eq_ignore_ascii_case("HEAD") {
            return Ok(DiffTarget::Current);
        }

        // Check for @lsn: or @scn: prefix (no branch specified)
        if let Some(rest) = input.strip_prefix("@lsn:") {
            let lsn = rest.parse::<u64>()
                .map_err(|_| crate::Error::query_execution(format!("Invalid LSN: {}", rest)))?;
            return Ok(DiffTarget::Lsn(lsn));
        }

        if let Some(rest) = input.strip_prefix("@scn:") {
            let scn = rest.parse::<u64>()
                .map_err(|_| crate::Error::query_execution(format!("Invalid SCN: {}", rest)))?;
            return Ok(DiffTarget::Scn(scn));
        }

        // Check for branch@qualifier format
        if let Some(at_pos) = input.find('@') {
            let branch = input[..at_pos].to_string();
            let qualifier = &input[at_pos + 1..];

            if let Some(lsn_str) = qualifier.strip_prefix("lsn:") {
                let lsn = lsn_str.parse::<u64>()
                    .map_err(|_| crate::Error::query_execution(format!("Invalid LSN: {}", lsn_str)))?;
                return Ok(DiffTarget::BranchAtLsn { branch, lsn });
            }

            if let Some(scn_str) = qualifier.strip_prefix("scn:") {
                let scn = scn_str.parse::<u64>()
                    .map_err(|_| crate::Error::query_execution(format!("Invalid SCN: {}", scn_str)))?;
                return Ok(DiffTarget::BranchAtScn { branch, scn });
            }

            return Err(crate::Error::query_execution(
                format!("Invalid diff target qualifier: {}. Use @lsn:N or @scn:N", qualifier)
            ));
        }

        // Plain branch name
        Ok(DiffTarget::Branch(input.to_string()))
    }

    /// Get the branch name if specified
    pub fn branch_name(&self) -> Option<&str> {
        match self {
            DiffTarget::Branch(name) => Some(name),
            DiffTarget::BranchAtLsn { branch, .. } => Some(branch),
            DiffTarget::BranchAtScn { branch, .. } => Some(branch),
            DiffTarget::Lsn(_) | DiffTarget::Scn(_) | DiffTarget::Current => None,
        }
    }

    /// Get the LSN if specified
    pub fn lsn(&self) -> Option<u64> {
        match self {
            DiffTarget::BranchAtLsn { lsn, .. } => Some(*lsn),
            DiffTarget::Lsn(lsn) => Some(*lsn),
            _ => None,
        }
    }

    /// Get the SCN if specified
    pub fn scn(&self) -> Option<u64> {
        match self {
            DiffTarget::BranchAtScn { scn, .. } => Some(*scn),
            DiffTarget::Scn(scn) => Some(*scn),
            _ => None,
        }
    }

    /// Display format for diff headers
    pub fn display(&self) -> String {
        match self {
            DiffTarget::Branch(name) => name.clone(),
            DiffTarget::BranchAtLsn { branch, lsn } => format!("{}@lsn:{}", branch, lsn),
            DiffTarget::BranchAtScn { branch, scn } => format!("{}@scn:{}", branch, scn),
            DiffTarget::Lsn(lsn) => format!("@lsn:{}", lsn),
            DiffTarget::Scn(scn) => format!("@scn:{}", scn),
            DiffTarget::Current => "HEAD".to_string(),
        }
    }
}

impl std::fmt::Display for DiffTarget {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "{}", self.display())
    }
}

/// Time-travel diff specification
///
/// Defines the source and target points for comparison, supporting:
/// - Branch-to-branch diffs
/// - LSN-to-LSN diffs (within same or different branches)
/// - SCN-to-SCN diffs (within same or different branches)
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DiffSpec {
    /// Source (left side) of the diff
    pub source: DiffTarget,
    /// Target (right side) of the diff
    pub target: DiffTarget,
    /// Optional list of tables to diff (None = all tables)
    pub tables: Option<Vec<String>>,
}

impl DiffSpec {
    /// Create a new diff spec between two targets
    pub fn new(source: DiffTarget, target: DiffTarget) -> Self {
        Self {
            source,
            target,
            tables: None,
        }
    }

    /// Create a branch-to-branch diff
    pub fn branches(source: &str, target: &str) -> Self {
        Self::new(
            DiffTarget::Branch(source.to_string()),
            DiffTarget::Branch(target.to_string()),
        )
    }

    /// Create an LSN-to-LSN diff within the same branch
    pub fn lsn_range(branch: &str, from_lsn: u64, to_lsn: u64) -> Self {
        Self::new(
            DiffTarget::BranchAtLsn { branch: branch.to_string(), lsn: from_lsn },
            DiffTarget::BranchAtLsn { branch: branch.to_string(), lsn: to_lsn },
        )
    }

    /// Create an SCN-to-SCN diff within the same branch
    pub fn scn_range(branch: &str, from_scn: u64, to_scn: u64) -> Self {
        Self::new(
            DiffTarget::BranchAtScn { branch: branch.to_string(), scn: from_scn },
            DiffTarget::BranchAtScn { branch: branch.to_string(), scn: to_scn },
        )
    }

    /// Limit diff to specific tables
    pub fn with_tables(mut self, tables: Vec<String>) -> Self {
        self.tables = Some(tables);
        self
    }

    /// Parse diff spec from string format (e.g., "main..feature" or "main@lsn:10..main@lsn:20")
    pub fn parse(input: &str) -> Result<Self> {
        // Look for ".." separator
        if let Some(sep_pos) = input.find("..") {
            let source_str = &input[..sep_pos];
            let target_str = &input[sep_pos + 2..];

            let source = DiffTarget::parse(source_str)?;
            let target = DiffTarget::parse(target_str)?;

            return Ok(Self::new(source, target));
        }

        Err(crate::Error::query_execution(
            format!("Invalid diff spec: {}. Use format 'source..target'", input)
        ))
    }

    /// Check if this is a same-branch time-travel diff
    pub fn is_same_branch_diff(&self) -> bool {
        match (&self.source, &self.target) {
            (DiffTarget::BranchAtLsn { branch: b1, .. }, DiffTarget::BranchAtLsn { branch: b2, .. }) => b1 == b2,
            (DiffTarget::BranchAtScn { branch: b1, .. }, DiffTarget::BranchAtScn { branch: b2, .. }) => b1 == b2,
            (DiffTarget::Lsn(_), DiffTarget::Lsn(_)) => true,
            (DiffTarget::Scn(_), DiffTarget::Scn(_)) => true,
            _ => false,
        }
    }
}

/// Result of a time-travel diff operation
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TimeTravelDiff {
    /// Source target specification
    pub source: DiffTarget,
    /// Target target specification
    pub target: DiffTarget,
    /// Schema changes
    pub schema_diff: SchemaDiff,
    /// Data changes (if full diff was performed)
    pub data_changes: Option<Vec<TableDataDiff>>,
    /// Row counts per table
    pub row_counts: Option<Vec<RowCountDiff>>,
    /// Statistics
    pub stats: DiffStats,
}

/// Diff statistics
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct DiffStats {
    /// Total rows added
    pub rows_added: u64,
    /// Total rows removed
    pub rows_removed: u64,
    /// Total rows modified
    pub rows_modified: u64,
    /// Tables with changes
    pub tables_changed: usize,
    /// Duration of diff operation in milliseconds
    pub duration_ms: u64,
}

/// Diff level configuration
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum DiffLevel {
    /// Schema-only comparison (tables, columns, indexes)
    SchemaOnly,
    /// Schema plus sampled data comparison
    Sampled { sample_size: usize },
    /// Full data comparison for all rows
    Full,
}

impl Default for DiffLevel {
    fn default() -> Self {
        DiffLevel::SchemaOnly
    }
}

/// Output format for diff results
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum DiffFormat {
    /// Git-style unified diff with +/- prefixes
    Unified,
    /// DDL/DML statements to transform source → target
    Sql,
    /// Machine-readable JSON format
    Json,
}

impl Default for DiffFormat {
    fn default() -> Self {
        DiffFormat::Unified
    }
}

/// Schema difference for a single table
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TableSchemaDiff {
    /// Table name
    pub table_name: String,
    /// Change type
    pub change_type: SchemaChangeType,
    /// Column changes (for modified tables)
    pub column_changes: Vec<ColumnChange>,
    /// Index changes
    pub index_changes: Vec<IndexChange>,
}

/// Type of schema change
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub enum SchemaChangeType {
    /// Table added in target
    Added,
    /// Table removed in target
    Removed,
    /// Table modified (columns/indexes changed)
    Modified,
    /// No change
    Unchanged,
}

/// Column-level change
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ColumnChange {
    pub column_name: String,
    pub change_type: SchemaChangeType,
    pub old_type: Option<String>,
    pub new_type: Option<String>,
    pub old_nullable: Option<bool>,
    pub new_nullable: Option<bool>,
}

/// Index-level change
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct IndexChange {
    pub index_name: String,
    pub change_type: SchemaChangeType,
    pub old_columns: Option<Vec<String>>,
    pub new_columns: Option<Vec<String>>,
}

/// Schema diff result
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SchemaDiff {
    /// Source branch
    pub source_branch: String,
    /// Target branch
    pub target_branch: String,
    /// Tables with changes
    pub table_diffs: Vec<TableSchemaDiff>,
    /// Total tables added
    pub tables_added: usize,
    /// Total tables removed
    pub tables_removed: usize,
    /// Total tables modified
    pub tables_modified: usize,
}

/// Row count comparison per table
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RowCountDiff {
    pub table_name: String,
    pub source_count: u64,
    pub target_count: u64,
    pub difference: i64,
}

/// Sampled data diff result
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SampledDiff {
    /// Schema diff
    pub schema: SchemaDiff,
    /// Row count differences
    pub row_counts: Vec<RowCountDiff>,
    /// Sample of changed rows per table
    pub samples: Vec<TableSample>,
    /// Sample size used
    pub sample_size: usize,
}

/// Sample of changed rows for a table
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TableSample {
    pub table_name: String,
    pub added_rows: Vec<serde_json::Value>,
    pub removed_rows: Vec<serde_json::Value>,
    pub modified_rows: Vec<(serde_json::Value, serde_json::Value)>,
}

/// Full data diff result
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct FullDiff {
    /// Schema diff
    pub schema: SchemaDiff,
    /// Full row-level changes per table
    pub data_changes: Vec<TableDataDiff>,
    /// Total rows added
    pub total_rows_added: u64,
    /// Total rows removed
    pub total_rows_removed: u64,
}

/// Full data diff for a single table
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TableDataDiff {
    pub table_name: String,
    pub added_rows: Vec<serde_json::Value>,
    pub removed_rows: Vec<serde_json::Value>,
}

/// Branch diff engine
pub struct DiffEngine<'a> {
    /// Reference to storage engine for catalog and data access
    storage: Option<&'a StorageEngine>,
    /// Branch manager for branch resolution
    branch_manager: Option<Arc<BranchManager>>,
}

impl Default for DiffEngine<'_> {
    fn default() -> Self {
        Self::new()
    }
}

impl<'a> DiffEngine<'a> {
    /// Create a new diff engine without storage (for testing/standalone)
    pub fn new() -> Self {
        Self {
            storage: None,
            branch_manager: None,
        }
    }

    /// Create a diff engine with storage engine access
    pub fn with_storage(storage: &'a StorageEngine) -> Self {
        let branch_manager = storage.branch_manager();
        Self {
            storage: Some(storage),
            branch_manager,
        }
    }

    /// Perform schema-only diff between two branches
    ///
    /// Compares catalog entries at each branch to find:
    /// - Tables added in target
    /// - Tables removed from source
    /// - Tables modified (columns/indexes changed)
    pub fn diff_schema(&self, source: &str, target: &str) -> Result<SchemaDiff> {
        let storage = match self.storage {
            Some(s) => s,
            None => {
                // Return empty diff when no storage available
                return Ok(SchemaDiff {
                    source_branch: source.to_string(),
                    target_branch: target.to_string(),
                    table_diffs: Vec::new(),
                    tables_added: 0,
                    tables_removed: 0,
                    tables_modified: 0,
                });
            }
        };

        let catalog = storage.catalog();

        // Get all tables from catalog (current snapshot)
        // In a full implementation, we'd query at specific branch snapshots
        let all_tables = catalog.list_tables().unwrap_or_default();

        // For now, compare table existence and schemas
        // In full implementation, would use time-travel to get schemas at branch points
        let mut table_diffs = Vec::new();
        let tables_added = 0;
        let tables_removed = 0;
        let tables_modified = 0;

        // Get source and target branch metadata
        if let Some(ref bm) = self.branch_manager {
            let _source_branch = bm.get_branch_by_name(source).ok();
            let _target_branch = bm.get_branch_by_name(target).ok();

            // In full implementation:
            // 1. Get catalog snapshot at source branch creation point
            // 2. Get catalog snapshot at target branch creation point
            // 3. Compare table schemas
        }

        // For demonstration, check each table's schema
        for table_name in &all_tables {
            if let Ok(schema) = catalog.get_table_schema(table_name) {
                // In full impl, compare source vs target schemas
                // For now, just include as unchanged
                table_diffs.push(TableSchemaDiff {
                    table_name: table_name.clone(),
                    change_type: SchemaChangeType::Unchanged,
                    column_changes: Vec::new(),
                    index_changes: Vec::new(),
                });
            }
        }

        Ok(SchemaDiff {
            source_branch: source.to_string(),
            target_branch: target.to_string(),
            table_diffs,
            tables_added,
            tables_removed,
            tables_modified,
        })
    }

    /// Perform schema + sampled data diff
    ///
    /// Combines schema comparison with row count analysis and
    /// random sampling of changed rows.
    pub fn diff_sampled(&self, source: &str, target: &str, sample_size: usize) -> Result<SampledDiff> {
        // 1. Schema diff first
        let schema = self.diff_schema(source, target)?;

        // 2. Row counts per table
        let mut row_counts = Vec::new();
        if let Some(storage) = self.storage {
            let catalog = storage.catalog();
            for table_diff in &schema.table_diffs {
                if table_diff.change_type != SchemaChangeType::Removed {
                    let count = catalog.get_table_statistics(&table_diff.table_name)
                        .ok()
                        .flatten()
                        .map(|s| s.row_count)
                        .unwrap_or(0);
                    row_counts.push(RowCountDiff {
                        table_name: table_diff.table_name.clone(),
                        source_count: count, // In full impl, would be from source branch
                        target_count: count, // In full impl, would be from target branch
                        difference: 0,
                    });
                }
            }
        }

        // 3. Sample comparison (placeholder - full impl would use TABLESAMPLE)
        let samples = Vec::new();

        Ok(SampledDiff {
            schema,
            row_counts,
            samples,
            sample_size,
        })
    }

    /// Perform full data diff
    ///
    /// Uses time-travel queries with EXCEPT/INTERSECT pattern:
    /// ```sql
    /// -- Rows added in target
    /// SELECT * FROM {table} AS OF BRANCH target
    /// EXCEPT
    /// SELECT * FROM {table} AS OF BRANCH source
    ///
    /// -- Rows removed from source
    /// SELECT * FROM {table} AS OF BRANCH source
    /// EXCEPT
    /// SELECT * FROM {table} AS OF BRANCH target
    /// ```
    pub fn diff_full(&self, source: &str, target: &str, tables: Option<&[String]>) -> Result<FullDiff> {
        let schema = self.diff_schema(source, target)?;

        let mut data_changes = Vec::new();
        let total_rows_added = 0u64;
        let total_rows_removed = 0u64;

        // Determine which tables to compare
        let tables_to_diff: Vec<String> = if let Some(specific_tables) = tables {
            specific_tables.to_vec()
        } else {
            schema.table_diffs.iter()
                .filter(|d| d.change_type != SchemaChangeType::Removed)
                .map(|d| d.table_name.clone())
                .collect()
        };

        // For each table, compute row-level diff
        // Full implementation would use branched time-travel queries
        for table_name in tables_to_diff {
            data_changes.push(TableDataDiff {
                table_name,
                added_rows: Vec::new(),
                removed_rows: Vec::new(),
            });
        }

        Ok(FullDiff {
            schema,
            data_changes,
            total_rows_added,
            total_rows_removed,
        })
    }

    // ========================================================================
    // Time-Travel Diff Methods (LSN/SCN Support)
    // ========================================================================

    /// Perform a time-travel diff using a DiffSpec
    ///
    /// This is the main entry point for LSN/SCN-based diffing. It supports:
    /// - Branch-to-branch comparisons
    /// - Same-branch LSN-to-LSN comparisons
    /// - Same-branch SCN-to-SCN comparisons
    /// - Cross-branch time-travel comparisons
    pub fn diff_with_spec(&self, spec: &DiffSpec, level: DiffLevel) -> Result<TimeTravelDiff> {
        let start_time = std::time::Instant::now();

        // Get schema diff
        let schema_diff = self.diff_schema_for_targets(&spec.source, &spec.target)?;

        // Perform data diff based on level
        let (data_changes, row_counts) = match level {
            DiffLevel::SchemaOnly => (None, None),
            DiffLevel::Sampled { sample_size } => {
                let counts = self.get_row_counts_for_targets(&spec.source, &spec.target, &schema_diff)?;
                (None, Some(counts))
            }
            DiffLevel::Full => {
                let (changes, counts) = self.get_full_diff_for_targets(
                    &spec.source,
                    &spec.target,
                    spec.tables.as_deref(),
                    &schema_diff,
                )?;
                (Some(changes), Some(counts))
            }
        };

        // Calculate stats
        let mut stats = DiffStats::default();
        stats.tables_changed = schema_diff.table_diffs.iter()
            .filter(|t| t.change_type != SchemaChangeType::Unchanged)
            .count();

        if let Some(ref changes) = data_changes {
            for change in changes {
                stats.rows_added += change.added_rows.len() as u64;
                stats.rows_removed += change.removed_rows.len() as u64;
            }
        }

        stats.duration_ms = start_time.elapsed().as_millis() as u64;

        Ok(TimeTravelDiff {
            source: spec.source.clone(),
            target: spec.target.clone(),
            schema_diff,
            data_changes,
            row_counts,
            stats,
        })
    }

    /// Diff between two LSN points on the same branch (convenience method)
    ///
    /// This is a shorthand for comparing data at different points in time
    /// within the same branch.
    ///
    /// # Example
    /// ```ignore
    /// // Compare branch 'main' at LSN 100 vs LSN 200
    /// let diff = engine.diff_lsn("main", 100, 200, DiffLevel::Full)?;
    /// ```
    pub fn diff_lsn(
        &self,
        branch: &str,
        from_lsn: u64,
        to_lsn: u64,
        level: DiffLevel,
    ) -> Result<TimeTravelDiff> {
        let spec = DiffSpec::lsn_range(branch, from_lsn, to_lsn);
        self.diff_with_spec(&spec, level)
    }

    /// Diff between two SCN points on the same branch (convenience method)
    ///
    /// This is a shorthand for comparing data at different System Change Numbers
    /// within the same branch.
    ///
    /// # Example
    /// ```ignore
    /// // Compare branch 'main' at SCN 10 vs SCN 20
    /// let diff = engine.diff_scn("main", 10, 20, DiffLevel::Full)?;
    /// ```
    pub fn diff_scn(
        &self,
        branch: &str,
        from_scn: u64,
        to_scn: u64,
        level: DiffLevel,
    ) -> Result<TimeTravelDiff> {
        let spec = DiffSpec::scn_range(branch, from_scn, to_scn);
        self.diff_with_spec(&spec, level)
    }

    /// Diff using arbitrary targets (most flexible method)
    ///
    /// # Example
    /// ```ignore
    /// // Compare main@lsn:100 with feature@scn:50
    /// let source = DiffTarget::BranchAtLsn { branch: "main".to_string(), lsn: 100 };
    /// let target = DiffTarget::BranchAtScn { branch: "feature".to_string(), scn: 50 };
    /// let diff = engine.diff_targets(source, target, DiffLevel::Full)?;
    /// ```
    pub fn diff_targets(
        &self,
        source: DiffTarget,
        target: DiffTarget,
        level: DiffLevel,
    ) -> Result<TimeTravelDiff> {
        let spec = DiffSpec::new(source, target);
        self.diff_with_spec(&spec, level)
    }

    /// Internal: Get schema diff between two targets
    fn diff_schema_for_targets(&self, source: &DiffTarget, target: &DiffTarget) -> Result<SchemaDiff> {
        // Get branch names (defaulting to "main" for LSN/SCN-only targets)
        let source_name = source.branch_name().unwrap_or("main");
        let target_name = target.branch_name().unwrap_or("main");

        // For same-branch time-travel diffs, schema comparison uses current schema
        // (schema changes are captured via DDL history, not snapshots)
        let mut schema = self.diff_schema(source_name, target_name)?;

        // Update the branch names to include LSN/SCN info
        schema.source_branch = source.display();
        schema.target_branch = target.display();

        Ok(schema)
    }

    /// Internal: Get row counts for targets
    fn get_row_counts_for_targets(
        &self,
        _source: &DiffTarget,
        _target: &DiffTarget,
        schema: &SchemaDiff,
    ) -> Result<Vec<RowCountDiff>> {
        let mut counts = Vec::new();

        if let Some(storage) = self.storage {
            let catalog = storage.catalog();
            for table_diff in &schema.table_diffs {
                if table_diff.change_type != SchemaChangeType::Removed {
                    let count = catalog.get_table_statistics(&table_diff.table_name)
                        .ok()
                        .flatten()
                        .map(|s| s.row_count)
                        .unwrap_or(0);
                    // In full implementation, would query at specific LSN/SCN
                    counts.push(RowCountDiff {
                        table_name: table_diff.table_name.clone(),
                        source_count: count,
                        target_count: count,
                        difference: 0,
                    });
                }
            }
        }

        Ok(counts)
    }

    /// Internal: Get full data diff for targets
    fn get_full_diff_for_targets(
        &self,
        source: &DiffTarget,
        target: &DiffTarget,
        tables: Option<&[String]>,
        schema: &SchemaDiff,
    ) -> Result<(Vec<TableDataDiff>, Vec<RowCountDiff>)> {
        let mut data_changes = Vec::new();
        let mut row_counts = Vec::new();

        // Determine which tables to diff
        let tables_to_diff: Vec<String> = if let Some(specific) = tables {
            specific.to_vec()
        } else {
            schema.table_diffs.iter()
                .filter(|d| d.change_type != SchemaChangeType::Removed)
                .map(|d| d.table_name.clone())
                .collect()
        };

        if let Some(storage) = self.storage {
            let catalog = storage.catalog();

            // For time-travel diffs, we need to compare data at different timestamps
            // The timestamp is derived from LSN/SCN via the snapshot manager
            let source_ts = self.resolve_target_timestamp(source)?;
            let target_ts = self.resolve_target_timestamp(target)?;

            for table_name in tables_to_diff {
                let count = catalog.get_table_statistics(&table_name)
                    .ok()
                    .flatten()
                    .map(|s| s.row_count)
                    .unwrap_or(0);
                row_counts.push(RowCountDiff {
                    table_name: table_name.clone(),
                    source_count: count,
                    target_count: count,
                    difference: 0,
                });

                // In full implementation, compare rows at source_ts vs target_ts
                // using scan_versions_between or similar methods
                let table_diff = self.diff_table_data(&table_name, source_ts, target_ts)?;
                data_changes.push(table_diff);
            }
        }

        Ok((data_changes, row_counts))
    }

    /// Internal: Resolve a DiffTarget to a timestamp
    fn resolve_target_timestamp(&self, target: &DiffTarget) -> Result<u64> {
        match target {
            DiffTarget::Current => {
                // Current timestamp
                Ok(u64::MAX)
            }
            DiffTarget::Branch(_) => {
                // Current state of branch
                Ok(u64::MAX)
            }
            DiffTarget::Lsn(lsn) | DiffTarget::BranchAtLsn { lsn, .. } => {
                // LSN is directly usable as timestamp (they're equivalent in our system)
                Ok(*lsn)
            }
            DiffTarget::Scn(scn) | DiffTarget::BranchAtScn { scn, .. } => {
                // SCN needs to be resolved via snapshot manager
                if let Some(storage) = self.storage {
                    let sm = storage.snapshot_manager();
                    return sm.resolve_scn(*scn);
                }
                // If no storage, use SCN directly as approximation
                Ok(*scn)
            }
        }
    }

    /// Internal: Diff table data between two timestamps
    fn diff_table_data(
        &self,
        table_name: &str,
        source_ts: u64,
        target_ts: u64,
    ) -> Result<TableDataDiff> {
        let mut added_rows = Vec::new();
        let mut removed_rows = Vec::new();

        // In full implementation, this would:
        // 1. Get all rows visible at source_ts
        // 2. Get all rows visible at target_ts
        // 3. Compare to find added/removed rows
        //
        // Using scan_versions_between for efficiency:
        // - Rows in target but not source = added
        // - Rows in source but not target = removed

        if let Some(storage) = self.storage {
            let sm = storage.snapshot_manager();
            // Get versions in the range
            let (start_ts, end_ts) = if source_ts < target_ts {
                (source_ts, target_ts)
            } else {
                (target_ts, source_ts)
            };

            let versions = sm.scan_versions_between(table_name, start_ts, end_ts)?;

            // Group by row_id and determine changes
            let mut row_versions: HashMap<u64, Vec<(u64, Vec<u8>)>> = HashMap::new();
            for (row_id, ts, data) in versions {
                row_versions.entry(row_id)
                    .or_default()
                    .push((ts, data));
            }

            // For each row, determine if it was added or removed
            for (row_id, versions) in row_versions {
                let has_source = versions.iter().any(|(ts, _)| *ts <= source_ts);
                let has_target = versions.iter().any(|(ts, _)| *ts <= target_ts);

                // Get latest version data for JSON output
                let latest_data = versions.iter()
                    .max_by_key(|(ts, _)| *ts)
                    .map(|(_, data)| data);

                if let Some(data) = latest_data {
                    let json_value = serde_json::from_slice::<serde_json::Value>(data)
                        .unwrap_or_else(|_| serde_json::json!({
                            "row_id": row_id,
                            "data": "<binary>"
                        }));

                    if has_target && !has_source {
                        added_rows.push(json_value);
                    } else if has_source && !has_target {
                        removed_rows.push(json_value);
                    }
                }
            }
        }

        Ok(TableDataDiff {
            table_name: table_name.to_string(),
            added_rows,
            removed_rows,
        })
    }

    /// Format a TimeTravelDiff result
    pub fn format_time_travel(&self, diff: &TimeTravelDiff, format: DiffFormat) -> String {
        match format {
            DiffFormat::Unified => self.format_time_travel_unified(diff),
            DiffFormat::Sql => self.format_time_travel_sql(diff),
            DiffFormat::Json => serde_json::to_string_pretty(diff).unwrap_or_default(),
        }
    }

    fn format_time_travel_unified(&self, diff: &TimeTravelDiff) -> String {
        let mut output = String::new();

        // Header
        output.push_str(&format!(
            "--- {}\n+++ {}\n",
            diff.source.display(),
            diff.target.display()
        ));

        // Schema changes
        if !diff.schema_diff.table_diffs.is_empty() {
            output.push_str("\n@@ Schema Changes @@\n");
            output.push_str(&self.format_unified(&diff.schema_diff));
        }

        // Data changes
        if let Some(ref changes) = diff.data_changes {
            for table_change in changes {
                if !table_change.added_rows.is_empty() || !table_change.removed_rows.is_empty() {
                    output.push_str(&format!("\n@@ Table: {} @@\n", table_change.table_name));

                    for row in &table_change.removed_rows {
                        output.push_str(&format!("- {}\n", row));
                    }
                    for row in &table_change.added_rows {
                        output.push_str(&format!("+ {}\n", row));
                    }
                }
            }
        }

        // Stats
        output.push_str(&format!(
            "\n-- Stats: {} tables changed, +{} rows, -{} rows ({} ms)\n",
            diff.stats.tables_changed,
            diff.stats.rows_added,
            diff.stats.rows_removed,
            diff.stats.duration_ms
        ));

        output
    }

    fn format_time_travel_sql(&self, diff: &TimeTravelDiff) -> String {
        let mut output = String::new();

        output.push_str(&format!(
            "-- Diff: {} -> {}\n",
            diff.source.display(),
            diff.target.display()
        ));
        output.push_str("-- Generated SQL to transform source to target state\n\n");

        // Schema DDL
        output.push_str(&self.format_sql(&diff.schema_diff));

        // Data DML
        if let Some(ref changes) = diff.data_changes {
            for table_change in changes {
                if !table_change.removed_rows.is_empty() {
                    output.push_str(&format!(
                        "\n-- Delete {} rows from {}\n",
                        table_change.removed_rows.len(),
                        table_change.table_name
                    ));
                    output.push_str(&format!(
                        "-- DELETE FROM {} WHERE <pk> IN (...);\n",
                        table_change.table_name
                    ));
                }
                if !table_change.added_rows.is_empty() {
                    output.push_str(&format!(
                        "\n-- Insert {} rows into {}\n",
                        table_change.added_rows.len(),
                        table_change.table_name
                    ));
                    output.push_str(&format!(
                        "-- INSERT INTO {} VALUES (...);\n",
                        table_change.table_name
                    ));
                }
            }
        }

        output
    }

    /// Format diff result as string
    pub fn format(&self, diff: &SchemaDiff, format: DiffFormat) -> String {
        match format {
            DiffFormat::Unified => self.format_unified(diff),
            DiffFormat::Sql => self.format_sql(diff),
            DiffFormat::Json => serde_json::to_string_pretty(diff).unwrap_or_default(),
        }
    }

    fn format_unified(&self, diff: &SchemaDiff) -> String {
        let mut output = String::new();
        output.push_str(&format!(
            "--- {}\n+++ {}\n",
            diff.source_branch, diff.target_branch
        ));

        for table_diff in &diff.table_diffs {
            match table_diff.change_type {
                SchemaChangeType::Added => {
                    output.push_str(&format!("+ TABLE {}\n", table_diff.table_name));
                }
                SchemaChangeType::Removed => {
                    output.push_str(&format!("- TABLE {}\n", table_diff.table_name));
                }
                SchemaChangeType::Modified => {
                    output.push_str(&format!("~ TABLE {} (modified)\n", table_diff.table_name));
                    for col in &table_diff.column_changes {
                        match col.change_type {
                            SchemaChangeType::Added => {
                                output.push_str(&format!("  + COLUMN {}\n", col.column_name));
                            }
                            SchemaChangeType::Removed => {
                                output.push_str(&format!("  - COLUMN {}\n", col.column_name));
                            }
                            SchemaChangeType::Modified => {
                                output.push_str(&format!(
                                    "  ~ COLUMN {} ({:?} -> {:?})\n",
                                    col.column_name, col.old_type, col.new_type
                                ));
                            }
                            _ => {}
                        }
                    }
                }
                _ => {}
            }
        }

        output
    }

    fn format_sql(&self, diff: &SchemaDiff) -> String {
        let mut output = String::new();
        output.push_str(&format!(
            "-- Transform {} -> {}\n\n",
            diff.source_branch, diff.target_branch
        ));

        for table_diff in &diff.table_diffs {
            match table_diff.change_type {
                SchemaChangeType::Added => {
                    // Generate CREATE TABLE statement from column changes
                    output.push_str(&format!("CREATE TABLE {} (\n", table_diff.table_name));
                    let columns: Vec<String> = table_diff.column_changes.iter()
                        .filter(|c| matches!(c.change_type, SchemaChangeType::Added))
                        .map(|c| {
                            let col_type = c.new_type.as_deref().unwrap_or("TEXT");
                            let nullable = if c.new_nullable == Some(false) { " NOT NULL" } else { "" };
                            format!("    {} {}{}", c.column_name, col_type, nullable)
                        })
                        .collect();
                    output.push_str(&columns.join(",\n"));
                    output.push_str("\n);\n");
                }
                SchemaChangeType::Removed => {
                    output.push_str(&format!("DROP TABLE IF EXISTS {};\n", table_diff.table_name));
                }
                SchemaChangeType::Modified => {
                    for col in &table_diff.column_changes {
                        match col.change_type {
                            SchemaChangeType::Added => {
                                output.push_str(&format!(
                                    "ALTER TABLE {} ADD COLUMN {} {};\n",
                                    table_diff.table_name,
                                    col.column_name,
                                    col.new_type.as_deref().unwrap_or("TEXT")
                                ));
                            }
                            SchemaChangeType::Removed => {
                                output.push_str(&format!(
                                    "ALTER TABLE {} DROP COLUMN {};\n",
                                    table_diff.table_name, col.column_name
                                ));
                            }
                            _ => {}
                        }
                    }
                }
                _ => {}
            }
        }

        output
    }
}

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

    #[test]
    fn test_diff_level_default() {
        assert_eq!(DiffLevel::default(), DiffLevel::SchemaOnly);
    }

    #[test]
    fn test_diff_format_default() {
        assert_eq!(DiffFormat::default(), DiffFormat::Unified);
    }

    #[test]
    fn test_diff_engine_creation() {
        let engine = DiffEngine::new();
        let diff = engine.diff_schema("main", "feature").unwrap();
        assert_eq!(diff.source_branch, "main");
        assert_eq!(diff.target_branch, "feature");
    }

    // ========================================================================
    // DiffTarget Tests
    // ========================================================================

    #[test]
    fn test_diff_target_parse_branch() {
        let target = DiffTarget::parse("main").unwrap();
        assert_eq!(target, DiffTarget::Branch("main".to_string()));
    }

    #[test]
    fn test_diff_target_parse_branch_at_lsn() {
        let target = DiffTarget::parse("feature@lsn:100").unwrap();
        assert_eq!(target, DiffTarget::BranchAtLsn {
            branch: "feature".to_string(),
            lsn: 100,
        });
    }

    #[test]
    fn test_diff_target_parse_branch_at_scn() {
        let target = DiffTarget::parse("main@scn:50").unwrap();
        assert_eq!(target, DiffTarget::BranchAtScn {
            branch: "main".to_string(),
            scn: 50,
        });
    }

    #[test]
    fn test_diff_target_parse_lsn_only() {
        let target = DiffTarget::parse("@lsn:200").unwrap();
        assert_eq!(target, DiffTarget::Lsn(200));
    }

    #[test]
    fn test_diff_target_parse_scn_only() {
        let target = DiffTarget::parse("@scn:300").unwrap();
        assert_eq!(target, DiffTarget::Scn(300));
    }

    #[test]
    fn test_diff_target_parse_current() {
        let target = DiffTarget::parse("@current").unwrap();
        assert_eq!(target, DiffTarget::Current);

        let target2 = DiffTarget::parse("HEAD").unwrap();
        assert_eq!(target2, DiffTarget::Current);
    }

    #[test]
    fn test_diff_target_display() {
        assert_eq!(DiffTarget::Branch("main".to_string()).display(), "main");
        assert_eq!(
            DiffTarget::BranchAtLsn { branch: "dev".to_string(), lsn: 100 }.display(),
            "dev@lsn:100"
        );
        assert_eq!(
            DiffTarget::BranchAtScn { branch: "prod".to_string(), scn: 50 }.display(),
            "prod@scn:50"
        );
        assert_eq!(DiffTarget::Lsn(200).display(), "@lsn:200");
        assert_eq!(DiffTarget::Scn(300).display(), "@scn:300");
        assert_eq!(DiffTarget::Current.display(), "HEAD");
    }

    // ========================================================================
    // DiffSpec Tests
    // ========================================================================

    #[test]
    fn test_diff_spec_parse() {
        let spec = DiffSpec::parse("main..feature").unwrap();
        assert_eq!(spec.source, DiffTarget::Branch("main".to_string()));
        assert_eq!(spec.target, DiffTarget::Branch("feature".to_string()));
    }

    #[test]
    fn test_diff_spec_parse_with_lsn() {
        let spec = DiffSpec::parse("main@lsn:10..main@lsn:20").unwrap();
        assert!(spec.is_same_branch_diff());
        assert_eq!(spec.source, DiffTarget::BranchAtLsn {
            branch: "main".to_string(),
            lsn: 10,
        });
        assert_eq!(spec.target, DiffTarget::BranchAtLsn {
            branch: "main".to_string(),
            lsn: 20,
        });
    }

    #[test]
    fn test_diff_spec_parse_lsn_only() {
        let spec = DiffSpec::parse("@lsn:100..@lsn:200").unwrap();
        assert!(spec.is_same_branch_diff());
    }

    #[test]
    fn test_diff_spec_lsn_range() {
        let spec = DiffSpec::lsn_range("main", 100, 200);
        assert!(spec.is_same_branch_diff());
        assert_eq!(spec.source.lsn(), Some(100));
        assert_eq!(spec.target.lsn(), Some(200));
    }

    #[test]
    fn test_diff_spec_scn_range() {
        let spec = DiffSpec::scn_range("main", 10, 20);
        assert!(spec.is_same_branch_diff());
        assert_eq!(spec.source.scn(), Some(10));
        assert_eq!(spec.target.scn(), Some(20));
    }

    #[test]
    fn test_diff_spec_branches() {
        let spec = DiffSpec::branches("main", "feature");
        assert!(!spec.is_same_branch_diff());
    }

    // ========================================================================
    // Time-Travel Diff Tests
    // ========================================================================

    #[test]
    fn test_time_travel_diff_lsn() {
        let engine = DiffEngine::new();
        let diff = engine.diff_lsn("main", 100, 200, DiffLevel::SchemaOnly).unwrap();

        assert_eq!(diff.source.display(), "main@lsn:100");
        assert_eq!(diff.target.display(), "main@lsn:200");
    }

    #[test]
    fn test_time_travel_diff_scn() {
        let engine = DiffEngine::new();
        let diff = engine.diff_scn("main", 10, 20, DiffLevel::SchemaOnly).unwrap();

        assert_eq!(diff.source.display(), "main@scn:10");
        assert_eq!(diff.target.display(), "main@scn:20");
    }

    #[test]
    fn test_time_travel_diff_with_spec() {
        let engine = DiffEngine::new();
        let spec = DiffSpec::parse("main@lsn:50..feature@scn:100").unwrap();
        let diff = engine.diff_with_spec(&spec, DiffLevel::SchemaOnly).unwrap();

        assert_eq!(diff.source.display(), "main@lsn:50");
        assert_eq!(diff.target.display(), "feature@scn:100");
    }

    #[test]
    fn test_time_travel_diff_format() {
        let engine = DiffEngine::new();
        let diff = engine.diff_lsn("main", 100, 200, DiffLevel::SchemaOnly).unwrap();

        let unified = engine.format_time_travel(&diff, DiffFormat::Unified);
        assert!(unified.contains("--- main@lsn:100"));
        assert!(unified.contains("+++ main@lsn:200"));

        let json = engine.format_time_travel(&diff, DiffFormat::Json);
        assert!(json.contains("main@lsn:100"));
    }
}