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

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

#[derive(Debug, Clone)]
pub struct MergeAdjacentCompactionCommit {
    pub table_id: ducklake_catalog::TableId,
    pub expected_snapshot: ducklake_catalog::SnapshotId,
    pub source_data_file_ids: Vec<ducklake_catalog::DataFileId>,
    pub compacted_files: Vec<AppendFile>,
}

impl MergeAdjacentCompactionCommit {
    #[cfg(any(
        feature = "sqlite",
        feature = "duckdb",
        feature = "postgres",
        feature = "quack"
    ))]
    fn validate_compacted_file_metadata(&self, source_ids: &[i64]) -> Result<()> {
        for file in &self.compacted_files {
            if file.table_id != self.table_id {
                return Err(StorageError::Backend(
                    "DuckLake merge-adjacent compacted file belongs to a different table".into(),
                ));
            }
            if file.expected_snapshot != self.expected_snapshot {
                return Err(StorageError::Backend(
                    "DuckLake merge-adjacent compacted file has inconsistent expected snapshot"
                        .into(),
                ));
            }
            if source_ids.contains(&file.data_file_id.0) {
                return Err(StorageError::Backend(
                    "DuckLake merge-adjacent compacted file id overlaps a source file".into(),
                ));
            }
        }
        Ok(())
    }

    pub fn validate_compacted_file_sequence(&self, next_file_id: i64) -> Result<()> {
        for (idx, file) in self.compacted_files.iter().enumerate() {
            let offset = i64::try_from(idx).map_err(|_| {
                StorageError::Backend(
                    "DuckLake merge-adjacent compacted file count overflow".into(),
                )
            })?;
            let expected_file_id = next_file_id.checked_add(offset).ok_or_else(|| {
                StorageError::Backend("DuckLake next_file_id sequence overflow".into())
            })?;
            if file.data_file_id.0 != expected_file_id {
                return Err(StorageError::Backend(format!(
                    "DuckLake next_file_id sequence expected compaction file id {}, found {}",
                    expected_file_id, file.data_file_id.0
                )));
            }
        }
        Ok(())
    }

    pub fn compacted_file_count(&self) -> Result<i64> {
        i64::try_from(self.compacted_files.len()).map_err(|_| {
            StorageError::Backend("DuckLake merge-adjacent compacted file count overflow".into())
        })
    }

    pub fn compacted_file_size_bytes(&self) -> Result<i64> {
        self.compacted_files
            .iter()
            .map(|file| file.file_size_bytes)
            .try_fold(0_i64, |acc, value| {
                acc.checked_add(value).ok_or_else(|| {
                    StorageError::Backend(
                        "DuckLake merge-adjacent compacted file size overflow".into(),
                    )
                })
            })
    }

    pub fn has_partition_values(&self) -> bool {
        self.compacted_files
            .iter()
            .any(|file| !file.partition_values.is_empty())
    }
}

#[derive(Debug, Clone)]
pub struct RewriteDataFilesCommit {
    pub table_id: ducklake_catalog::TableId,
    pub expected_snapshot: ducklake_catalog::SnapshotId,
    pub source_data_file_ids: Vec<ducklake_catalog::DataFileId>,
    pub rewritten_files: Vec<AppendFile>,
}

impl RewriteDataFilesCommit {
    #[cfg(any(
        feature = "sqlite",
        feature = "duckdb",
        feature = "postgres",
        feature = "quack"
    ))]
    fn validate_rewritten_file_metadata(&self, source_ids: &[i64]) -> Result<()> {
        for file in &self.rewritten_files {
            if file.table_id != self.table_id {
                return Err(StorageError::Backend(
                    "DuckLake rewrite-data-files output belongs to a different table".into(),
                ));
            }
            if file.expected_snapshot != self.expected_snapshot {
                return Err(StorageError::Backend(
                    "DuckLake rewrite-data-files output has inconsistent expected snapshot".into(),
                ));
            }
            if source_ids.contains(&file.data_file_id.0) {
                return Err(StorageError::Backend(
                    "DuckLake rewrite-data-files output file id overlaps a source file".into(),
                ));
            }
        }
        Ok(())
    }

    pub fn validate_rewritten_file_sequence(&self, next_file_id: i64) -> Result<()> {
        for (idx, file) in self.rewritten_files.iter().enumerate() {
            let offset = i64::try_from(idx).map_err(|_| {
                StorageError::Backend(
                    "DuckLake rewrite-data-files output file count overflow".into(),
                )
            })?;
            let expected_file_id = next_file_id.checked_add(offset).ok_or_else(|| {
                StorageError::Backend("DuckLake next_file_id sequence overflow".into())
            })?;
            if file.data_file_id.0 != expected_file_id {
                return Err(StorageError::Backend(format!(
                    "DuckLake next_file_id sequence expected rewrite file id {}, found {}",
                    expected_file_id, file.data_file_id.0
                )));
            }
        }
        Ok(())
    }

    pub fn rewritten_file_count(&self) -> Result<i64> {
        i64::try_from(self.rewritten_files.len()).map_err(|_| {
            StorageError::Backend("DuckLake rewrite-data-files output file count overflow".into())
        })
    }

    pub fn rewritten_file_size_bytes(&self) -> Result<i64> {
        self.rewritten_files
            .iter()
            .map(|file| file.file_size_bytes)
            .try_fold(0_i64, |acc, value| {
                acc.checked_add(value).ok_or_else(|| {
                    StorageError::Backend(
                        "DuckLake rewrite-data-files output file size overflow".into(),
                    )
                })
            })
    }

    pub fn has_partition_values(&self) -> bool {
        self.rewritten_files
            .iter()
            .any(|file| !file.partition_values.is_empty())
    }
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct MaintenanceCommitPlan {
    pub source_ids: Vec<i64>,
    pub new_snapshot: i64,
    pub schema_version: i64,
    pub next_catalog_id: i64,
    pub new_next_file_id: i64,
    pub source_file_size_bytes: i64,
    pub output_file_size_bytes: i64,
    pub changes_made: String,
    pub commit_message: String,
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
fn plan_maintenance_source_ids(
    operation: &str,
    source_data_file_ids: &[DataFileId],
) -> Result<Vec<i64>> {
    if source_data_file_ids.is_empty() {
        return Err(StorageError::Backend(format!(
            "DuckLake {operation} received no source files"
        )));
    }
    let mut source_ids = source_data_file_ids
        .iter()
        .map(|id| id.0)
        .collect::<Vec<_>>();
    source_ids.sort_unstable();
    let source_len = source_ids.len();
    source_ids.dedup();
    if source_ids.len() != source_len {
        return Err(StorageError::Backend(format!(
            "DuckLake {operation} source files contain duplicates"
        )));
    }
    Ok(source_ids)
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
pub(crate) fn plan_merge_adjacent_compaction_sources(
    compaction: &MergeAdjacentCompactionCommit,
) -> Result<Vec<i64>> {
    let source_ids = plan_maintenance_source_ids(
        "merge-adjacent compaction",
        &compaction.source_data_file_ids,
    )?;
    compaction.validate_compacted_file_metadata(&source_ids)?;
    Ok(source_ids)
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
pub(crate) fn plan_rewrite_data_files_sources(
    rewrite: &RewriteDataFilesCommit,
) -> Result<Vec<i64>> {
    let source_ids =
        plan_maintenance_source_ids("rewrite-data-files", &rewrite.source_data_file_ids)?;
    rewrite.validate_rewritten_file_metadata(&source_ids)?;
    Ok(source_ids)
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
fn plan_maintenance_snapshot(
    operation: &str,
    expected_snapshot: ducklake_catalog::SnapshotId,
    snapshot: CatalogSnapshotIds,
    output_file_count: i64,
) -> Result<(i64, i64)> {
    if snapshot.snapshot_id != expected_snapshot.0 {
        return Err(StorageError::conflict(
            operation,
            expected_snapshot,
            snapshot.snapshot_id,
        ));
    }
    let new_snapshot = snapshot.snapshot_id.checked_add(1).ok_or_else(|| {
        StorageError::Backend(format!("DuckLake {operation} snapshot id overflow"))
    })?;
    let new_next_file_id = snapshot
        .next_file_id
        .checked_add(output_file_count)
        .ok_or_else(|| StorageError::Backend("DuckLake next_file_id sequence overflow".into()))?;
    Ok((new_snapshot, new_next_file_id))
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
fn checked_maintenance_file_size_sum(
    values: impl IntoIterator<Item = i64>,
    overflow_message: &'static str,
) -> Result<i64> {
    values.into_iter().try_fold(0_i64, |acc, value| {
        acc.checked_add(value)
            .ok_or_else(|| StorageError::Backend(overflow_message.into()))
    })
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
pub(crate) fn plan_merge_adjacent_compaction_commit(
    compaction: &MergeAdjacentCompactionCommit,
    snapshot: CatalogSnapshotIds,
    source_ids: &[i64],
    active_source_file_count: usize,
    source_file_sizes: impl IntoIterator<Item = i64>,
) -> Result<MaintenanceCommitPlan> {
    let output_file_count = compaction.compacted_file_count()?;
    let (new_snapshot, new_next_file_id) = plan_maintenance_snapshot(
        "merge-adjacent",
        compaction.expected_snapshot,
        snapshot,
        output_file_count,
    )?;
    compaction.validate_compacted_file_sequence(snapshot.next_file_id)?;
    if active_source_file_count != source_ids.len() {
        return Err(StorageError::Backend(
            "DuckLake merge-adjacent source files must all be active data files for the table"
                .into(),
        ));
    }
    Ok(MaintenanceCommitPlan {
        source_ids: source_ids.to_vec(),
        new_snapshot,
        schema_version: snapshot.schema_version,
        next_catalog_id: snapshot.next_catalog_id,
        new_next_file_id,
        source_file_size_bytes: checked_maintenance_file_size_sum(
            source_file_sizes,
            "DuckLake merge-adjacent source file size overflow",
        )?,
        output_file_size_bytes: compaction.compacted_file_size_bytes()?,
        changes_made: format!("merge_adjacent:{}", compaction.table_id.0),
        commit_message: "datafusion merge adjacent".into(),
    })
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
pub(crate) fn plan_rewrite_data_files_commit(
    rewrite: &RewriteDataFilesCommit,
    snapshot: CatalogSnapshotIds,
    source_ids: &[i64],
    active_source_file_count: usize,
    source_file_sizes: impl IntoIterator<Item = i64>,
) -> Result<MaintenanceCommitPlan> {
    let output_file_count = rewrite.rewritten_file_count()?;
    let (new_snapshot, new_next_file_id) = plan_maintenance_snapshot(
        "rewrite-data-files",
        rewrite.expected_snapshot,
        snapshot,
        output_file_count,
    )?;
    rewrite.validate_rewritten_file_sequence(snapshot.next_file_id)?;
    if active_source_file_count != source_ids.len() {
        return Err(StorageError::Backend(
            "DuckLake rewrite-data-files source files must all be active data files for the table"
                .into(),
        ));
    }
    Ok(MaintenanceCommitPlan {
        source_ids: source_ids.to_vec(),
        new_snapshot,
        schema_version: snapshot.schema_version,
        next_catalog_id: snapshot.next_catalog_id,
        new_next_file_id,
        source_file_size_bytes: checked_maintenance_file_size_sum(
            source_file_sizes,
            "DuckLake rewrite-data-files source file size overflow",
        )?,
        output_file_size_bytes: rewrite.rewritten_file_size_bytes()?,
        changes_made: format!("rewrite_data_files:{}", rewrite.table_id.0),
        commit_message: "datafusion rewrite data files".into(),
    })
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct CleanupOldFilesOptions {
    pub dry_run: bool,
    pub cleanup_all: bool,
    pub older_than_micros: Option<i64>,
}

impl CleanupOldFilesOptions {
    pub fn dry_run_all() -> Self {
        Self {
            dry_run: true,
            cleanup_all: true,
            older_than_micros: None,
        }
    }

    pub fn cleanup_all() -> Self {
        Self {
            dry_run: false,
            cleanup_all: true,
            older_than_micros: None,
        }
    }

    pub fn dry_run_older_than(older_than_micros: i64) -> Self {
        Self {
            dry_run: true,
            cleanup_all: false,
            older_than_micros: Some(older_than_micros),
        }
    }

    pub fn older_than(older_than_micros: i64) -> Self {
        Self {
            dry_run: false,
            cleanup_all: false,
            older_than_micros: Some(older_than_micros),
        }
    }

    pub fn validate(self) -> Result<()> {
        if self.cleanup_all == self.older_than_micros.is_some() {
            return Err(StorageError::Backend(
                "DuckLake cleanup_old_files requires exactly one of cleanup_all or older_than"
                    .into(),
            ));
        }
        Ok(())
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CleanupFile {
    pub data_file_id: ducklake_catalog::DataFileId,
    pub path: String,
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
pub(crate) fn plan_cleanup_old_files(
    options: CleanupOldFilesOptions,
    data_path: &str,
    rows: &[FileScheduledForDeletion],
) -> Result<Vec<CleanupFile>> {
    options.validate()?;
    let mut files = Vec::new();
    for row in rows {
        if !options.cleanup_all {
            let older_than = options.older_than_micros.ok_or_else(|| {
                StorageError::Backend("DuckLake cleanup_old_files requires older_than".into())
            })?;
            if row.schedule_start >= older_than {
                continue;
            }
        }
        files.push(CleanupFile {
            data_file_id: row.data_file_id,
            path: resolve_cleanup_file_path(data_path, row),
        });
    }
    Ok(files)
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ExpireSnapshotsOptions {
    pub dry_run: bool,
    pub versions: Vec<ducklake_catalog::SnapshotId>,
    pub older_than_micros: Option<i64>,
}

impl ExpireSnapshotsOptions {
    pub fn versions(versions: Vec<ducklake_catalog::SnapshotId>) -> Self {
        Self {
            dry_run: false,
            versions,
            older_than_micros: None,
        }
    }

    pub fn dry_run_versions(versions: Vec<ducklake_catalog::SnapshotId>) -> Self {
        Self {
            dry_run: true,
            versions,
            older_than_micros: None,
        }
    }

    pub fn older_than(older_than_micros: i64) -> Self {
        Self {
            dry_run: false,
            versions: Vec::new(),
            older_than_micros: Some(older_than_micros),
        }
    }

    pub fn dry_run_older_than(older_than_micros: i64) -> Self {
        Self {
            dry_run: true,
            versions: Vec::new(),
            older_than_micros: Some(older_than_micros),
        }
    }

    pub fn validate(&self) -> Result<()> {
        if self.versions.is_empty() == self.older_than_micros.is_none() {
            return Err(StorageError::Backend(
                "DuckLake expire_snapshots requires exactly one of versions or older_than".into(),
            ));
        }
        Ok(())
    }
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
pub(crate) fn plan_expirable_snapshots(
    options: &ExpireSnapshotsOptions,
    latest_snapshot: i64,
    snapshots: Vec<Snapshot>,
) -> Result<Vec<Snapshot>> {
    options.validate()?;
    Ok(snapshots
        .into_iter()
        .filter(|snapshot| snapshot.snapshot_id.0 != latest_snapshot)
        .filter(|snapshot| {
            if let Some(older_than) = options.older_than_micros {
                snapshot.snapshot_time < older_than
            } else {
                options.versions.contains(&snapshot.snapshot_id)
            }
        })
        .collect())
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct CleanupMetadataFile {
    pub id: i64,
    pub path: String,
    pub path_is_relative: bool,
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct ExpireSnapshotsPrunePlan {
    pub snapshot_ids: Vec<i64>,
    pub deleted_table_ids: Vec<i64>,
    pub deleted_data_file_ids: Vec<i64>,
    pub deleted_delete_file_ids: Vec<i64>,
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
pub(crate) fn plan_expire_snapshot_ids(snapshots: &[Snapshot]) -> Vec<i64> {
    snapshots
        .iter()
        .map(|snapshot| snapshot.snapshot_id.0)
        .collect()
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
pub(crate) fn plan_cleanup_metadata_file_ids(files: &[CleanupMetadataFile]) -> Vec<i64> {
    files.iter().map(|file| file.id).collect()
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
pub(crate) fn plan_snapshot_expiration_prune(
    snapshots: &[Snapshot],
    deleted_table_ids: Vec<i64>,
    deleted_data_file_rows: &[CleanupMetadataFile],
    deleted_delete_file_rows: &[CleanupMetadataFile],
) -> ExpireSnapshotsPrunePlan {
    ExpireSnapshotsPrunePlan {
        snapshot_ids: plan_expire_snapshot_ids(snapshots),
        deleted_table_ids,
        deleted_data_file_ids: plan_cleanup_metadata_file_ids(deleted_data_file_rows),
        deleted_delete_file_ids: plan_cleanup_metadata_file_ids(deleted_delete_file_rows),
    }
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
pub(crate) fn plan_remove_files_scheduled_for_deletion(data_file_ids: Vec<DataFileId>) -> Vec<i64> {
    let mut ids = data_file_ids
        .into_iter()
        .map(|data_file_id| data_file_id.0)
        .collect::<Vec<_>>();
    ids.sort_unstable();
    ids.dedup();
    ids
}
