#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
use std::collections::HashSet;

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
use ducklake_catalog::{CatalogError, SortExpression};

use crate::InlinedDataDeletionCommit;
#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
use crate::{
    CatalogSnapshotIds, CreateTableColumn, MetadataMutationPlan, Result, StorageError,
    sort_expression_unqualified_column_refs,
};

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PartitionColumnCommit {
    pub column_id: ducklake_catalog::ColumnId,
    pub transform: ducklake_catalog::PartitionTransform,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PartitioningCommit {
    pub table_id: ducklake_catalog::TableId,
    pub expected_snapshot: ducklake_catalog::SnapshotId,
    pub columns: Vec<PartitionColumnCommit>,
    pub commit_message: String,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PartitioningTableCommit {
    pub table_id: ducklake_catalog::TableId,
    pub columns: Vec<PartitionColumnCommit>,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PartitioningBatchCommit {
    pub expected_snapshot: ducklake_catalog::SnapshotId,
    pub changes: Vec<PartitioningTableCommit>,
    pub commit_message: String,
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct ActivePartitionSpec {
    pub partition_id: ducklake_catalog::PartitionId,
    pub columns: Vec<PartitionColumnCommit>,
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct PartitioningTableFacts {
    pub change: PartitioningTableCommit,
    pub active: Vec<ActivePartitionSpec>,
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct PlannedPartitioningTableChange {
    pub table_id: ducklake_catalog::TableId,
    pub columns: Vec<PartitionColumnCommit>,
    pub close_partition_id: Option<ducklake_catalog::PartitionId>,
    pub insert_partition_id: Option<ducklake_catalog::PartitionId>,
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct PartitioningBatchPlan {
    pub new_snapshot: i64,
    pub new_schema_version: i64,
    pub next_catalog_id: i64,
    pub next_file_id: i64,
    pub changes_made: String,
    pub commit_message: String,
    pub table_changes: Vec<PlannedPartitioningTableChange>,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SortExpressionCommit {
    pub expression: String,
    pub dialect: String,
    pub sort_direction: String,
    pub null_order: String,
}

#[derive(Debug, Clone)]
pub struct SortedByCommit {
    pub table_id: ducklake_catalog::TableId,
    pub expected_snapshot: ducklake_catalog::SnapshotId,
    pub expressions: Vec<SortExpressionCommit>,
    pub commit_message: String,
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct ActiveSortSpec {
    pub sort_id: ducklake_catalog::SortId,
    pub expressions: Vec<SortExpressionCommit>,
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct SortedByChangePlan {
    pub metadata: MetadataMutationPlan,
    pub close_sort_id: Option<ducklake_catalog::SortId>,
    pub insert_sort_id: Option<ducklake_catalog::SortId>,
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) enum SortedByCommitPlan {
    NoOp(ducklake_catalog::SnapshotId),
    Change(SortedByChangePlan),
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
pub(crate) fn ensure_expected_snapshot(
    operation: &'static str,
    expected_snapshot: ducklake_catalog::SnapshotId,
    latest_snapshot: i64,
) -> Result<()> {
    if latest_snapshot != expected_snapshot.0 {
        return Err(StorageError::conflict(
            operation,
            expected_snapshot,
            latest_snapshot,
        ));
    }
    Ok(())
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
pub(crate) fn sort_expression_commit(expression: SortExpression) -> SortExpressionCommit {
    SortExpressionCommit {
        expression: expression.expression,
        dialect: expression.dialect.unwrap_or_else(|| "duckdb".to_string()),
        sort_direction: expression
            .sort_direction
            .unwrap_or_else(|| "ASC".to_string()),
        null_order: expression
            .null_order
            .unwrap_or_else(|| "NULLS_LAST".to_string()),
    }
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
pub(crate) fn validate_sort_expression_columns(
    table_id: ducklake_catalog::TableId,
    expressions: &[SortExpressionCommit],
    column_names: &[String],
) -> Result<()> {
    let mut unmatched = Vec::new();
    for expression in expressions {
        for column_name in sort_expression_unqualified_column_refs(&expression.expression)? {
            if !column_names
                .iter()
                .any(|candidate| candidate.eq_ignore_ascii_case(&column_name))
                && !unmatched
                    .iter()
                    .any(|candidate: &String| candidate.eq_ignore_ascii_case(&column_name))
            {
                unmatched.push(column_name);
            }
        }
    }
    if !unmatched.is_empty() {
        return Err(StorageError::Catalog(CatalogError::NotFound(format!(
            "sort columns {} in table {}",
            unmatched.join(", "),
            table_id.0
        ))));
    }
    Ok(())
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
pub(crate) fn plan_sorted_by_commit(
    table_id: ducklake_catalog::TableId,
    expected_snapshot: ducklake_catalog::SnapshotId,
    snapshot: CatalogSnapshotIds,
    active: &[ActiveSortSpec],
    requested: &[SortExpressionCommit],
    commit_message: String,
) -> Result<SortedByCommitPlan> {
    ensure_expected_snapshot("alter-sorted-by", expected_snapshot, snapshot.snapshot_id)?;
    if active.len() > 1 {
        return Err(StorageError::Backend(format!(
            "DuckLake table {} has {} active sort definitions at snapshot {}",
            table_id.0,
            active.len(),
            snapshot.snapshot_id
        )));
    }

    let active_expressions = active
        .first()
        .map(|spec| spec.expressions.clone())
        .unwrap_or_default();
    if active_expressions == requested {
        return Ok(SortedByCommitPlan::NoOp(ducklake_catalog::SnapshotId(
            snapshot.snapshot_id,
        )));
    }

    let insert_sort_id =
        (!requested.is_empty()).then_some(ducklake_catalog::SortId(snapshot.next_catalog_id));
    let allocated_catalog_ids = if insert_sort_id.is_some() { 1 } else { 0 };
    let next_catalog_id = snapshot
        .next_catalog_id
        .checked_add(allocated_catalog_ids)
        .ok_or_else(|| {
            StorageError::Backend("DuckLake sorted metadata catalog id overflow".into())
        })?;
    let metadata = snapshot.table_metadata_mutation(
        "alter-sorted-by",
        expected_snapshot,
        table_id,
        next_catalog_id,
        commit_message,
    )?;
    Ok(SortedByCommitPlan::Change(SortedByChangePlan {
        metadata,
        close_sort_id: active.first().map(|spec| spec.sort_id),
        insert_sort_id,
    }))
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
pub(crate) fn sort_key_index(idx: usize) -> Result<i64> {
    i64::try_from(idx)
        .map_err(|_| StorageError::Backend("DuckLake sort key index is out of range".into()))
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
pub(crate) fn final_partitioning_changes(
    changes: &[PartitioningTableCommit],
) -> Vec<&PartitioningTableCommit> {
    let mut seen = HashSet::with_capacity(changes.len());
    let mut final_changes = Vec::new();
    for change in changes.iter().rev() {
        if seen.insert(change.table_id) {
            final_changes.push(change);
        }
    }
    final_changes
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
pub(crate) fn partitioning_batch_allocated_catalog_ids(
    changes: &[PartitioningTableCommit],
) -> Result<i64> {
    i64::try_from(changes.len()).map_err(|_| {
        StorageError::Backend("DuckLake partitioning change count is out of range".into())
    })
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
pub(crate) fn partitioning_batch_changed_table_ids(
    changes: &[PartitioningTableCommit],
) -> Vec<i64> {
    let mut changed_table_ids = changes
        .iter()
        .map(|change| change.table_id.0)
        .collect::<Vec<_>>();
    changed_table_ids.sort_unstable();
    changed_table_ids.dedup();
    changed_table_ids
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
pub(crate) fn plan_partitioning_batch_commit(
    partitioning: &PartitioningBatchCommit,
    snapshot: CatalogSnapshotIds,
    facts: Vec<PartitioningTableFacts>,
) -> Result<PartitioningBatchPlan> {
    if partitioning.changes.is_empty() {
        return Err(StorageError::Backend(
            "DuckLake partitioning batch has no table changes".into(),
        ));
    }
    ensure_expected_snapshot(
        "alter-partitioning",
        partitioning.expected_snapshot,
        snapshot.snapshot_id,
    )?;
    let allocated_catalog_ids = partitioning_batch_allocated_catalog_ids(&partitioning.changes)?;
    let new_snapshot = snapshot.snapshot_id.checked_add(1).ok_or_else(|| {
        StorageError::Backend("DuckLake partitioning snapshot id overflow".into())
    })?;
    let new_schema_version = snapshot.schema_version.checked_add(1).ok_or_else(|| {
        StorageError::Backend("DuckLake partitioning schema version overflow".into())
    })?;
    let next_catalog_id = snapshot
        .next_catalog_id
        .checked_add(allocated_catalog_ids)
        .ok_or_else(|| StorageError::Backend("DuckLake partitioning catalog id overflow".into()))?;
    let changed_table_ids = partitioning_batch_changed_table_ids(&partitioning.changes);
    let changes_made = changed_table_ids
        .into_iter()
        .map(|table_id| format!("altered_table:{table_id}"))
        .collect::<Vec<_>>()
        .join(",");
    let final_changes = final_partitioning_changes(&partitioning.changes);
    if facts.len() != final_changes.len() {
        return Err(StorageError::Backend(format!(
            "DuckLake partitioning planner expected {} final table changes, found {} fact rows",
            final_changes.len(),
            facts.len()
        )));
    }
    for (fact, change) in facts.iter().zip(final_changes) {
        if fact.change != *change {
            return Err(StorageError::Backend(
                "DuckLake partitioning planner facts do not match final table changes".into(),
            ));
        }
    }

    let mut inserted_partition_rows = 0_i64;
    let mut table_changes = Vec::new();
    for fact in facts {
        if fact.active.len() > 1 {
            return Err(StorageError::Backend(format!(
                "DuckLake table {} has {} active partition specs at snapshot {}",
                fact.change.table_id.0,
                fact.active.len(),
                snapshot.snapshot_id
            )));
        }
        let active_spec = fact
            .active
            .first()
            .map(|spec| spec.columns.clone())
            .unwrap_or_default();
        if active_spec == fact.change.columns {
            continue;
        }
        let partition_id = snapshot
            .next_catalog_id
            .checked_add(inserted_partition_rows)
            .ok_or_else(|| {
                StorageError::Backend("DuckLake partitioning partition id overflow".into())
            })?;
        inserted_partition_rows = inserted_partition_rows.checked_add(1).ok_or_else(|| {
            StorageError::Backend("DuckLake partitioning inserted partition count overflow".into())
        })?;
        table_changes.push(PlannedPartitioningTableChange {
            table_id: fact.change.table_id,
            columns: fact.change.columns,
            close_partition_id: fact.active.first().map(|spec| spec.partition_id),
            insert_partition_id: Some(ducklake_catalog::PartitionId(partition_id)),
        });
    }

    Ok(PartitioningBatchPlan {
        new_snapshot,
        new_schema_version,
        next_catalog_id,
        next_file_id: snapshot.next_file_id,
        changes_made,
        commit_message: partitioning.commit_message.clone(),
        table_changes,
    })
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
pub(crate) fn validate_create_table_columns(columns: &[CreateTableColumn]) -> Result<()> {
    let mut ids = HashSet::with_capacity(columns.len());
    let mut sibling_names: HashSet<(Option<i64>, String)> = HashSet::with_capacity(columns.len());
    for column in columns {
        if column.column_id.0 <= 0 {
            return Err(StorageError::Backend(format!(
                "DuckLake create-table column id {} is not positive",
                column.column_id.0
            )));
        }
        if !ids.insert(column.column_id) {
            return Err(StorageError::Backend(format!(
                "DuckLake create-table duplicate column id {}",
                column.column_id.0
            )));
        }
        if column.column_order <= 0 {
            return Err(StorageError::Backend(format!(
                "DuckLake create-table column {} has non-positive column_order {}",
                column.column_name, column.column_order
            )));
        }
        if column.column_name.trim().is_empty() {
            return Err(StorageError::Backend(
                "DuckLake create-table column name is empty".into(),
            ));
        }
        if column.column_type.trim().is_empty() {
            return Err(StorageError::Backend(format!(
                "DuckLake create-table column {} has empty type",
                column.column_name
            )));
        }
        if !sibling_names.insert((
            column.parent_column.map(|parent| parent.0),
            column.column_name.to_ascii_lowercase(),
        )) {
            return Err(StorageError::Backend(format!(
                "DuckLake create-table duplicate sibling column name {}",
                column.column_name
            )));
        }
    }
    for column in columns {
        if let Some(parent) = column.parent_column
            && !ids.contains(&parent)
        {
            return Err(StorageError::Backend(format!(
                "DuckLake create-table column {} references missing parent column {}",
                column.column_name, parent.0
            )));
        }
    }
    Ok(())
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
pub(crate) fn deterministic_catalog_uuid(id: i64) -> String {
    let low = (id as u64) & 0x0000_ffff_ffff_ffff;
    format!("00000000-0000-0000-0000-{low:012x}")
}

#[derive(Debug, Clone)]
pub struct TruncateTableCommit {
    pub table_id: ducklake_catalog::TableId,
    pub expected_snapshot: ducklake_catalog::SnapshotId,
    pub rows_deleted: i64,
    pub inlined_data_deletions: Vec<InlinedDataDeletionCommit>,
    pub changes_made: String,
    pub commit_message: String,
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) struct TruncateTableCommitPlan {
    pub(crate) table_id: ducklake_catalog::TableId,
    pub(crate) expected_snapshot: ducklake_catalog::SnapshotId,
    pub(crate) rows_deleted: i64,
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) struct TruncateTableSnapshotPlan {
    pub(crate) new_snapshot: ducklake_catalog::SnapshotId,
    pub(crate) next_file_id: i64,
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
pub(crate) fn plan_truncate_table_commit(
    truncate: &TruncateTableCommit,
) -> Result<TruncateTableCommitPlan> {
    if truncate.rows_deleted < 0 {
        return Err(StorageError::Backend(
            "DuckLake truncate row count is negative".into(),
        ));
    }
    Ok(TruncateTableCommitPlan {
        table_id: truncate.table_id,
        expected_snapshot: truncate.expected_snapshot,
        rows_deleted: truncate.rows_deleted,
    })
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
impl TruncateTableCommitPlan {
    pub(crate) fn plan_snapshot(
        &self,
        latest_snapshot: i64,
        next_file_id: i64,
    ) -> Result<TruncateTableSnapshotPlan> {
        if latest_snapshot != self.expected_snapshot.0 {
            return Err(StorageError::conflict(
                "truncate",
                self.expected_snapshot,
                latest_snapshot,
            ));
        }
        Ok(TruncateTableSnapshotPlan {
            new_snapshot: ducklake_catalog::SnapshotId(latest_snapshot + 1),
            next_file_id,
        })
    }

    pub(crate) fn validate_inlined_rows_deleted(&self, inlined_rows_deleted: i64) -> Result<()> {
        if inlined_rows_deleted > self.rows_deleted {
            return Err(StorageError::Backend(
                "DuckLake truncate inlined row count exceeds total row count".into(),
            ));
        }
        Ok(())
    }
}
