use crate::{AppendFile, DeleteFileCommit, InlinedFileDeletionCommit, StorageError};
use ducklake_catalog::{
    ColumnId, DataFileId, DeleteFileId, DeleteFormat, DuckLakeType, FileColumnStats,
    FilePartitionValue, PartitionId, SchemaField, SnapshotId, TableId,
};

pub(crate) fn assert_snapshot_conflict(
    err: &StorageError,
    expected_operation: &str,
    expected_snapshot: i64,
    found_snapshot: i64,
) {
    assert!(err.is_conflict());
    match err {
        StorageError::Conflict {
            operation,
            expected_snapshot: actual_expected,
            found_snapshot: actual_found,
        } => {
            assert_eq!(operation, expected_operation);
            assert_eq!(*actual_expected, expected_snapshot);
            assert_eq!(*actual_found, found_snapshot);
        }
        other => panic!("expected conflict error, got {other:?}"),
    }
}

pub(crate) struct TestAppendFile {
    pub(crate) table_id: TableId,
    pub(crate) data_file_id: i64,
    pub(crate) expected_snapshot: SnapshotId,
    pub(crate) path: String,
    pub(crate) record_count: i64,
    pub(crate) file_size_bytes: i64,
    pub(crate) column_stats: Vec<FileColumnStats>,
    pub(crate) partition_id: Option<PartitionId>,
    pub(crate) partition_values: Vec<FilePartitionValue>,
}

impl TestAppendFile {
    pub(crate) fn into_commit(self) -> AppendFile {
        AppendFile {
            table_id: self.table_id,
            expected_snapshot: self.expected_snapshot,
            data_file_id: DataFileId(self.data_file_id),
            begin_snapshot: None,
            path: self.path,
            path_is_relative: true,
            record_count: self.record_count,
            file_size_bytes: self.file_size_bytes,
            footer_size: None,
            row_id_start: None,
            explicit_row_ids: false,
            column_stats: self.column_stats,
            variant_stats: Vec::new(),
            partition_id: self.partition_id,
            partition_values: self.partition_values,
            encryption_key: None,
            mapping_id: None,
            partial_max: None,
        }
    }
}

pub(crate) fn test_file_column_stats(
    data_file_id: i64,
    table_id: TableId,
    column_id: ColumnId,
    column_size_bytes: Option<i64>,
    value_count: i64,
    null_count: i64,
    min_value: &str,
    max_value: &str,
) -> FileColumnStats {
    FileColumnStats {
        data_file_id: DataFileId(data_file_id),
        table_id,
        column_id,
        column_size_bytes,
        value_count: Some(value_count),
        null_count: Some(null_count),
        min_value: Some(min_value.to_string()),
        max_value: Some(max_value.to_string()),
        contains_nan: Some(false),
        extra_stats: None,
    }
}

pub(crate) fn test_file_partition_value(
    data_file_id: i64,
    table_id: TableId,
    partition_key_index: i64,
    partition_value: &str,
) -> FilePartitionValue {
    FilePartitionValue {
        data_file_id: DataFileId(data_file_id),
        table_id,
        partition_key_index,
        partition_value: Some(partition_value.to_string()),
    }
}

pub(crate) struct TestDeleteFile {
    pub(crate) table_id: TableId,
    pub(crate) delete_file_id: i64,
    pub(crate) expected_snapshot: SnapshotId,
    pub(crate) data_file_id: i64,
    pub(crate) path: String,
    pub(crate) rows_deleted: i64,
    pub(crate) file_size_bytes: i64,
    pub(crate) footer_size: Option<i64>,
    pub(crate) overwritten_delete_file_id: Option<DeleteFileId>,
}

impl TestDeleteFile {
    pub(crate) fn into_commit(self) -> DeleteFileCommit {
        DeleteFileCommit {
            table_id: self.table_id,
            expected_snapshot: self.expected_snapshot,
            delete_file_id: DeleteFileId(self.delete_file_id),
            begin_snapshot: None,
            data_file_id: DataFileId(self.data_file_id),
            path: self.path,
            path_is_relative: true,
            format: DeleteFormat::Parquet,
            delete_count: self.rows_deleted,
            rows_deleted: self.rows_deleted,
            file_size_bytes: self.file_size_bytes,
            footer_size: self.footer_size,
            partial_max: None,
            overwritten_delete_file_id: self.overwritten_delete_file_id,
        }
    }
}

pub(crate) fn test_inlined_file_deletion(
    table_id: TableId,
    data_file_id: i64,
    expected_snapshot: SnapshotId,
    row_ids: Vec<i64>,
) -> InlinedFileDeletionCommit {
    InlinedFileDeletionCommit {
        table_id,
        expected_snapshot,
        data_file_id: DataFileId(data_file_id),
        rows_deleted: i64::try_from(row_ids.len()).unwrap(),
        row_ids,
    }
}

pub(crate) fn test_schema_field(
    id: i64,
    name: &str,
    type_name: &str,
    ducklake_type: DuckLakeType,
) -> SchemaField {
    test_schema_field_with_children(id, name, type_name, ducklake_type, Vec::new())
}

pub(crate) fn test_schema_field_with_children(
    id: i64,
    name: &str,
    type_name: &str,
    ducklake_type: DuckLakeType,
    children: Vec<SchemaField>,
) -> SchemaField {
    SchemaField {
        field_id: ColumnId(id),
        name: name.into(),
        type_name: type_name.into(),
        ducklake_type,
        nulls_allowed: true,
        initial_default: None,
        column_order: id,
        children,
    }
}
