use arrow::record_batch::RecordBatch;

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

#[derive(Debug, Clone)]
pub struct CreateSchemaCommit {
    pub schema_name: String,
    pub commit_message: String,
}

#[derive(Debug, Clone)]
pub struct CreateTableCommit {
    pub schema_name: String,
    pub table_name: String,
    pub columns: Vec<CreateTableColumn>,
    pub commit_message: String,
}

#[derive(Debug, Clone)]
pub struct DropSchemaCommit {
    pub schema_name: String,
    pub cascade: bool,
    pub commit_message: String,
}

#[derive(Debug, Clone)]
pub struct CreateTableWithFilesCommit {
    pub create: CreateTableCommit,
    pub table_id: ducklake_catalog::TableId,
    pub expected_snapshot: ducklake_catalog::SnapshotId,
    pub files: Vec<AppendFile>,
}

#[derive(Debug, Clone)]
pub struct CreateTableWithInlinedDataCommit {
    pub create: CreateTableCommit,
    pub table_id: ducklake_catalog::TableId,
    pub expected_snapshot: ducklake_catalog::SnapshotId,
    pub batches: Vec<RecordBatch>,
    pub column_stats: Vec<ducklake_catalog::FileColumnStats>,
    pub rows_inserted: i64,
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct PlannedCreateTableWithFilesFile {
    pub row_id_start: Option<i64>,
    pub begin_snapshot: i64,
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct CreateTableWithFilesCommitPlan {
    pub catalog: CatalogObjectMutationPlan,
    pub new_next_file_id: i64,
    pub files: Vec<PlannedCreateTableWithFilesFile>,
    pub total_record_count: i64,
    pub total_file_size_bytes: i64,
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct CreateTableWithInlinedDataCommitPlan {
    pub catalog: CatalogObjectMutationPlan,
    pub total_record_count: i64,
}

#[derive(Debug, Clone)]
pub struct CreateViewCommit {
    pub schema_name: String,
    pub view_name: String,
    pub dialect: String,
    pub sql: String,
    pub column_aliases: Option<String>,
    pub commit_message: String,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CreateTableColumn {
    pub column_id: ducklake_catalog::ColumnId,
    pub column_order: i64,
    pub column_name: String,
    pub column_type: String,
    pub nulls_allowed: bool,
    pub parent_column: Option<ducklake_catalog::ColumnId>,
}

#[derive(Debug, Clone)]
pub struct DropTableCommit {
    pub schema_name: String,
    pub table_name: String,
    pub commit_message: String,
}

#[derive(Debug, Clone)]
pub struct RenameTableCommit {
    pub schema_name: String,
    pub old_table_name: String,
    pub new_schema_name: Option<String>,
    pub new_table_name: String,
    pub commit_message: String,
}

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

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
impl CatalogSnapshotIds {
    pub(crate) fn catalog_object_mutation(
        &self,
        allocated_catalog_ids: i64,
        changes_made: String,
        commit_message: String,
    ) -> Result<CatalogObjectMutationPlan> {
        if allocated_catalog_ids < 0 {
            return Err(StorageError::Backend(
                "DuckLake catalog mutation allocated catalog ids is negative".into(),
            ));
        }
        let new_snapshot = self.snapshot_id.checked_add(1).ok_or_else(|| {
            StorageError::Backend("DuckLake catalog mutation snapshot id overflow".into())
        })?;
        let new_schema_version = self.schema_version.checked_add(1).ok_or_else(|| {
            StorageError::Backend("DuckLake catalog mutation schema version overflow".into())
        })?;
        let next_catalog_id = self
            .next_catalog_id
            .checked_add(allocated_catalog_ids)
            .ok_or_else(|| {
                StorageError::Backend("DuckLake catalog mutation catalog id overflow".into())
            })?;
        Ok(CatalogObjectMutationPlan {
            new_snapshot,
            new_schema_version,
            next_catalog_id,
            next_file_id: self.next_file_id,
            allocated_catalog_id: (allocated_catalog_ids > 0).then_some(self.next_catalog_id),
            changes_made,
            commit_message,
        })
    }
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
pub(crate) fn ducklake_quote_identifier(identifier: &str) -> String {
    format!("\"{}\"", identifier.replace('"', "\"\""))
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
pub(crate) fn ducklake_sql_string_literal(value: &str) -> String {
    format!("'{}'", value.replace('\'', "''"))
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
fn require_non_empty_catalog_name(kind: &str, value: &str) -> Result<()> {
    if value.trim().is_empty() {
        return Err(StorageError::Backend(format!("DuckLake {kind} is empty")));
    }
    Ok(())
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
pub(crate) fn plan_create_schema_catalog_mutation(
    create: &CreateSchemaCommit,
    snapshot: CatalogSnapshotIds,
) -> Result<CatalogObjectMutationPlan> {
    require_non_empty_catalog_name("create-schema schema name", &create.schema_name)?;
    snapshot.catalog_object_mutation(
        1,
        format!(
            "created_schema:{}",
            ducklake_quote_identifier(&create.schema_name)
        ),
        create.commit_message.clone(),
    )
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
pub(crate) fn plan_create_table_catalog_mutation(
    create: &CreateTableCommit,
    snapshot: CatalogSnapshotIds,
) -> Result<CatalogObjectMutationPlan> {
    require_non_empty_catalog_name("create-table schema name", &create.schema_name)?;
    require_non_empty_catalog_name("create-table table name", &create.table_name)?;
    if create.columns.is_empty() {
        return Err(StorageError::Backend(
            "DuckLake create-table received no columns".into(),
        ));
    }
    validate_create_table_columns(&create.columns)?;
    snapshot.catalog_object_mutation(
        1,
        format!(
            "created_table:{}.{}",
            ducklake_quote_identifier(&create.schema_name),
            ducklake_quote_identifier(&create.table_name)
        ),
        create.commit_message.clone(),
    )
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
pub(crate) fn plan_create_table_with_files_commit(
    create: &CreateTableWithFilesCommit,
    snapshot: CatalogSnapshotIds,
    has_table_stats: bool,
) -> Result<CreateTableWithFilesCommitPlan> {
    if create.files.is_empty() {
        return Err(StorageError::Backend(
            "DuckLake CREATE TABLE AS commit received no files".into(),
        ));
    }
    validate_create_table_columns(&create.create.columns)?;
    ensure_expected_snapshot(
        "create-table-as",
        create.expected_snapshot,
        snapshot.snapshot_id,
    )?;
    if create.table_id.0 != snapshot.next_catalog_id {
        return Err(StorageError::Backend(format!(
            "DuckLake CREATE TABLE AS expected table id {}, found {}",
            snapshot.next_catalog_id, create.table_id.0
        )));
    }

    for file in &create.files {
        if file.table_id != create.table_id {
            return Err(StorageError::Backend(
                "DuckLake CREATE TABLE AS files span multiple tables".into(),
            ));
        }
        if file.expected_snapshot != create.expected_snapshot {
            return Err(StorageError::Backend(
                "DuckLake CREATE TABLE AS files have inconsistent expected snapshots".into(),
            ));
        }
    }

    for (idx, file) in create.files.iter().enumerate() {
        let expected_file_id = snapshot
            .next_file_id
            .checked_add(i64::try_from(idx).map_err(|_| {
                StorageError::Backend("DuckLake CREATE TABLE AS file index is out of range".into())
            })?)
            .ok_or_else(|| {
                StorageError::Backend("DuckLake CREATE TABLE AS file id overflow".into())
            })?;
        if expected_file_id != file.data_file_id.0 {
            return Err(StorageError::Backend(format!(
                "DuckLake next_file_id sequence expected CREATE TABLE AS file id {}, found {}",
                expected_file_id, file.data_file_id.0
            )));
        }
    }

    let catalog = plan_create_table_catalog_mutation(&create.create, snapshot)?;
    let file_count = i64::try_from(create.files.len()).map_err(|_| {
        StorageError::Backend("DuckLake CREATE TABLE AS file count is out of range".into())
    })?;
    let new_next_file_id = snapshot
        .next_file_id
        .checked_add(file_count)
        .ok_or_else(|| {
            StorageError::Backend("DuckLake CREATE TABLE AS next_file_id overflow".into())
        })?;

    let mut next_row_id = has_table_stats.then_some(0_i64);
    let mut files = Vec::with_capacity(create.files.len());
    let mut total_record_count = 0_i64;
    let mut total_file_size_bytes = 0_i64;
    for file in &create.files {
        let row_id_start = next_row_id;
        if let Some(next_row_id) = &mut next_row_id {
            *next_row_id = next_row_id.checked_add(file.record_count).ok_or_else(|| {
                StorageError::Backend("DuckLake CREATE TABLE AS row id overflow".into())
            })?;
        }
        total_record_count = total_record_count
            .checked_add(file.record_count)
            .ok_or_else(|| {
                StorageError::Backend("DuckLake CREATE TABLE AS row count overflow".into())
            })?;
        total_file_size_bytes = total_file_size_bytes
            .checked_add(file.file_size_bytes)
            .ok_or_else(|| {
                StorageError::Backend("DuckLake CREATE TABLE AS file size overflow".into())
            })?;
        files.push(PlannedCreateTableWithFilesFile {
            row_id_start,
            begin_snapshot: catalog.new_snapshot,
        });
    }

    Ok(CreateTableWithFilesCommitPlan {
        catalog,
        new_next_file_id,
        files,
        total_record_count,
        total_file_size_bytes,
    })
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
pub(crate) fn plan_create_table_with_inlined_data_commit(
    create: &CreateTableWithInlinedDataCommit,
    snapshot: CatalogSnapshotIds,
    has_inlined_data_tables: bool,
) -> Result<CreateTableWithInlinedDataCommitPlan> {
    if create.rows_inserted <= 0 {
        return Err(StorageError::Backend(
            "DuckLake CREATE TABLE AS inlined commit received no rows".into(),
        ));
    }
    validate_create_table_columns(&create.create.columns)?;
    ensure_expected_snapshot(
        "create-table-as",
        create.expected_snapshot,
        snapshot.snapshot_id,
    )?;
    if create.table_id.0 != snapshot.next_catalog_id {
        return Err(StorageError::Backend(format!(
            "DuckLake CREATE TABLE AS expected table id {}, found {}",
            snapshot.next_catalog_id, create.table_id.0
        )));
    }
    if !has_inlined_data_tables {
        return Err(StorageError::Backend(
            "DuckLake CREATE TABLE AS inlined commit requires ducklake_inlined_data_tables".into(),
        ));
    }

    let catalog = plan_create_table_catalog_mutation(&create.create, snapshot)?;
    Ok(CreateTableWithInlinedDataCommitPlan {
        catalog,
        total_record_count: create.rows_inserted,
    })
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
pub(crate) fn plan_create_view_catalog_mutation(
    create: &CreateViewCommit,
    snapshot: CatalogSnapshotIds,
) -> Result<CatalogObjectMutationPlan> {
    require_non_empty_catalog_name("create-view schema name", &create.schema_name)?;
    require_non_empty_catalog_name("create-view view name", &create.view_name)?;
    require_non_empty_catalog_name("create-view dialect", &create.dialect)?;
    require_non_empty_catalog_name("create-view SQL", &create.sql)?;
    snapshot.catalog_object_mutation(
        1,
        format!(
            "created_view:{}.{}",
            ducklake_quote_identifier(&create.schema_name),
            ducklake_quote_identifier(&create.view_name)
        ),
        create.commit_message.clone(),
    )
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
pub(crate) fn plan_drop_table_catalog_mutation(
    drop: &DropTableCommit,
    snapshot: CatalogSnapshotIds,
    table_id: ducklake_catalog::TableId,
) -> Result<CatalogObjectMutationPlan> {
    require_non_empty_catalog_name("drop-table schema name", &drop.schema_name)?;
    require_non_empty_catalog_name("drop-table table name", &drop.table_name)?;
    snapshot.catalog_object_mutation(
        0,
        format!("dropped_table:{}", table_id.0),
        drop.commit_message.clone(),
    )
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
pub(crate) fn plan_rename_table_catalog_mutation(
    rename: &RenameTableCommit,
    snapshot: CatalogSnapshotIds,
    target_schema_name: &str,
) -> Result<CatalogObjectMutationPlan> {
    require_non_empty_catalog_name("rename-table schema name", &rename.schema_name)?;
    require_non_empty_catalog_name("rename-table source table name", &rename.old_table_name)?;
    require_non_empty_catalog_name("rename-table target table name", &rename.new_table_name)?;
    if rename
        .new_schema_name
        .as_deref()
        .is_some_and(|schema| schema.trim().is_empty())
    {
        return Err(StorageError::Backend(
            "DuckLake rename-table target schema name is empty".into(),
        ));
    }
    snapshot.catalog_object_mutation(
        0,
        format!(
            "created_table:{}.{}",
            ducklake_quote_identifier(target_schema_name),
            ducklake_quote_identifier(&rename.new_table_name)
        ),
        rename.commit_message.clone(),
    )
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
pub(crate) fn plan_drop_view_catalog_mutation(
    drop: &DropViewCommit,
    snapshot: CatalogSnapshotIds,
    view_id: ducklake_catalog::ViewId,
) -> Result<CatalogObjectMutationPlan> {
    require_non_empty_catalog_name("drop-view schema name", &drop.schema_name)?;
    require_non_empty_catalog_name("drop-view view name", &drop.view_name)?;
    snapshot.catalog_object_mutation(
        0,
        format!("dropped_view:{}", view_id.0),
        drop.commit_message.clone(),
    )
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
pub(crate) fn plan_drop_schema_catalog_mutation(
    drop: &DropSchemaCommit,
    snapshot: CatalogSnapshotIds,
    schema_id: ducklake_catalog::SchemaId,
    table_ids: &[i64],
    view_ids: &[i64],
) -> Result<CatalogObjectMutationPlan> {
    require_non_empty_catalog_name("drop-schema schema name", &drop.schema_name)?;
    if !drop.cascade && (!table_ids.is_empty() || !view_ids.is_empty()) {
        return Err(StorageError::Backend(format!(
            "DuckLake schema {} is not empty",
            drop.schema_name
        )));
    }
    let mut changes = vec![format!("dropped_schema:{}", schema_id.0)];
    changes.extend(
        table_ids
            .iter()
            .map(|table_id| format!("dropped_table:{table_id}")),
    );
    changes.extend(
        view_ids
            .iter()
            .map(|view_id| format!("dropped_view:{view_id}")),
    );
    snapshot.catalog_object_mutation(0, changes.join(","), drop.commit_message.clone())
}

#[derive(Debug, Clone)]
pub struct DropViewCommit {
    pub schema_name: String,
    pub view_name: String,
    pub commit_message: String,
}
