use uuid::Uuid;

pub(crate) use crate::ducklake_sql_string_literal as sql_string;
use crate::{
    DuckLakeCatalogInitOptions, Result, StorageError, ducklake_quote_identifier as quote_ident,
    ducklake_store_data_path,
};

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) struct ReferenceCatalogColumn {
    pub(crate) name: &'static str,
    pub(crate) sql_type: &'static str,
    pub(crate) primary_key: bool,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) struct ReferenceCatalogTable {
    pub(crate) name: &'static str,
    pub(crate) columns: &'static [ReferenceCatalogColumn],
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct CatalogBootstrap {
    pub(crate) snapshot_time_sql: String,
    pub(crate) schema_uuid: String,
    pub(crate) data_path: String,
    pub(crate) encrypted: bool,
    pub(crate) created_by: String,
}

impl CatalogBootstrap {
    pub(crate) fn new(
        data_path: &str,
        encrypted: bool,
        snapshot_time_sql: impl Into<String>,
    ) -> Self {
        Self {
            snapshot_time_sql: snapshot_time_sql.into(),
            schema_uuid: Uuid::new_v4().to_string(),
            data_path: ducklake_store_data_path(data_path),
            encrypted,
            created_by: format!("datafusion-ducklake-provider {}", env!("CARGO_PKG_VERSION")),
        }
    }
}

pub(crate) fn bootstrap_from_options(
    options: &DuckLakeCatalogInitOptions,
    default_data_path: Option<&str>,
    snapshot_time_sql: impl Into<String>,
) -> Result<CatalogBootstrap> {
    let data_path = options
        .data_path
        .as_deref()
        .or(default_data_path)
        .ok_or_else(|| {
            StorageError::Backend(
                "Attempting to create a new DuckLake catalog but DATA_PATH is not set".into(),
            )
        })?;
    Ok(CatalogBootstrap::new(
        data_path,
        options.encrypted.unwrap_or(false),
        snapshot_time_sql,
    ))
}

macro_rules! col {
    ($name:literal, $ty:literal) => {
        ReferenceCatalogColumn {
            name: $name,
            sql_type: $ty,
            primary_key: false,
        }
    };
    ($name:literal, $ty:literal, pk) => {
        ReferenceCatalogColumn {
            name: $name,
            sql_type: $ty,
            primary_key: true,
        }
    };
}

// Source of truth: external-projects/ducklake/src/storage/ducklake_metadata_manager.cpp,
// DuckLakeMetadataManager::InitializeDuckLake.
pub(crate) const REFERENCE_CATALOG_TABLES: &[ReferenceCatalogTable] = &[
    ReferenceCatalogTable {
        name: "ducklake_metadata",
        columns: &[
            col!("key", "VARCHAR NOT NULL"),
            col!("value", "VARCHAR NOT NULL"),
            col!("scope", "VARCHAR"),
            col!("scope_id", "BIGINT"),
        ],
    },
    ReferenceCatalogTable {
        name: "ducklake_snapshot",
        columns: &[
            col!("snapshot_id", "BIGINT", pk),
            col!("snapshot_time", "TIMESTAMPTZ"),
            col!("schema_version", "BIGINT"),
            col!("next_catalog_id", "BIGINT"),
            col!("next_file_id", "BIGINT"),
        ],
    },
    ReferenceCatalogTable {
        name: "ducklake_snapshot_changes",
        columns: &[
            col!("snapshot_id", "BIGINT", pk),
            col!("changes_made", "VARCHAR"),
            col!("author", "VARCHAR"),
            col!("commit_message", "VARCHAR"),
            col!("commit_extra_info", "VARCHAR"),
        ],
    },
    ReferenceCatalogTable {
        name: "ducklake_schema",
        columns: &[
            col!("schema_id", "BIGINT", pk),
            col!("schema_uuid", "UUID"),
            col!("begin_snapshot", "BIGINT"),
            col!("end_snapshot", "BIGINT"),
            col!("schema_name", "VARCHAR"),
            col!("path", "VARCHAR"),
            col!("path_is_relative", "BOOLEAN"),
        ],
    },
    ReferenceCatalogTable {
        name: "ducklake_table",
        columns: &[
            col!("table_id", "BIGINT"),
            col!("table_uuid", "UUID"),
            col!("begin_snapshot", "BIGINT"),
            col!("end_snapshot", "BIGINT"),
            col!("schema_id", "BIGINT"),
            col!("table_name", "VARCHAR"),
            col!("path", "VARCHAR"),
            col!("path_is_relative", "BOOLEAN"),
        ],
    },
    ReferenceCatalogTable {
        name: "ducklake_view",
        columns: &[
            col!("view_id", "BIGINT"),
            col!("view_uuid", "UUID"),
            col!("begin_snapshot", "BIGINT"),
            col!("end_snapshot", "BIGINT"),
            col!("schema_id", "BIGINT"),
            col!("view_name", "VARCHAR"),
            col!("dialect", "VARCHAR"),
            col!("sql", "VARCHAR"),
            col!("column_aliases", "VARCHAR"),
        ],
    },
    ReferenceCatalogTable {
        name: "ducklake_tag",
        columns: &[
            col!("object_id", "BIGINT"),
            col!("begin_snapshot", "BIGINT"),
            col!("end_snapshot", "BIGINT"),
            col!("key", "VARCHAR"),
            col!("value", "VARCHAR"),
        ],
    },
    ReferenceCatalogTable {
        name: "ducklake_column_tag",
        columns: &[
            col!("table_id", "BIGINT"),
            col!("column_id", "BIGINT"),
            col!("begin_snapshot", "BIGINT"),
            col!("end_snapshot", "BIGINT"),
            col!("key", "VARCHAR"),
            col!("value", "VARCHAR"),
        ],
    },
    ReferenceCatalogTable {
        name: "ducklake_data_file",
        columns: &[
            col!("data_file_id", "BIGINT", pk),
            col!("table_id", "BIGINT"),
            col!("begin_snapshot", "BIGINT"),
            col!("end_snapshot", "BIGINT"),
            col!("file_order", "BIGINT"),
            col!("path", "VARCHAR"),
            col!("path_is_relative", "BOOLEAN"),
            col!("file_format", "VARCHAR"),
            col!("record_count", "BIGINT"),
            col!("file_size_bytes", "BIGINT"),
            col!("footer_size", "BIGINT"),
            col!("row_id_start", "BIGINT"),
            col!("partition_id", "BIGINT"),
            col!("encryption_key", "VARCHAR"),
            col!("mapping_id", "BIGINT"),
            col!("partial_max", "BIGINT"),
        ],
    },
    ReferenceCatalogTable {
        name: "ducklake_file_column_stats",
        columns: &[
            col!("data_file_id", "BIGINT"),
            col!("table_id", "BIGINT"),
            col!("column_id", "BIGINT"),
            col!("column_size_bytes", "BIGINT"),
            col!("value_count", "BIGINT"),
            col!("null_count", "BIGINT"),
            col!("min_value", "VARCHAR"),
            col!("max_value", "VARCHAR"),
            col!("contains_nan", "BOOLEAN"),
            col!("extra_stats", "VARCHAR"),
        ],
    },
    ReferenceCatalogTable {
        name: "ducklake_file_variant_stats",
        columns: &[
            col!("data_file_id", "BIGINT"),
            col!("table_id", "BIGINT"),
            col!("column_id", "BIGINT"),
            col!("variant_path", "VARCHAR"),
            col!("shredded_type", "VARCHAR"),
            col!("column_size_bytes", "BIGINT"),
            col!("value_count", "BIGINT"),
            col!("null_count", "BIGINT"),
            col!("min_value", "VARCHAR"),
            col!("max_value", "VARCHAR"),
            col!("contains_nan", "BOOLEAN"),
            col!("extra_stats", "VARCHAR"),
        ],
    },
    ReferenceCatalogTable {
        name: "ducklake_delete_file",
        columns: &[
            col!("delete_file_id", "BIGINT", pk),
            col!("table_id", "BIGINT"),
            col!("begin_snapshot", "BIGINT"),
            col!("end_snapshot", "BIGINT"),
            col!("data_file_id", "BIGINT"),
            col!("path", "VARCHAR"),
            col!("path_is_relative", "BOOLEAN"),
            col!("format", "VARCHAR"),
            col!("delete_count", "BIGINT"),
            col!("file_size_bytes", "BIGINT"),
            col!("footer_size", "BIGINT"),
            col!("encryption_key", "VARCHAR"),
            col!("partial_max", "BIGINT"),
        ],
    },
    ReferenceCatalogTable {
        name: "ducklake_column",
        columns: &[
            col!("column_id", "BIGINT"),
            col!("begin_snapshot", "BIGINT"),
            col!("end_snapshot", "BIGINT"),
            col!("table_id", "BIGINT"),
            col!("column_order", "BIGINT"),
            col!("column_name", "VARCHAR"),
            col!("column_type", "VARCHAR"),
            col!("initial_default", "VARCHAR"),
            col!("default_value", "VARCHAR"),
            col!("nulls_allowed", "BOOLEAN"),
            col!("parent_column", "BIGINT"),
            col!("default_value_type", "VARCHAR"),
            col!("default_value_dialect", "VARCHAR"),
        ],
    },
    ReferenceCatalogTable {
        name: "ducklake_table_stats",
        columns: &[
            col!("table_id", "BIGINT"),
            col!("record_count", "BIGINT"),
            col!("next_row_id", "BIGINT"),
            col!("file_size_bytes", "BIGINT"),
        ],
    },
    ReferenceCatalogTable {
        name: "ducklake_table_column_stats",
        columns: &[
            col!("table_id", "BIGINT"),
            col!("column_id", "BIGINT"),
            col!("contains_null", "BOOLEAN"),
            col!("contains_nan", "BOOLEAN"),
            col!("min_value", "VARCHAR"),
            col!("max_value", "VARCHAR"),
            col!("extra_stats", "VARCHAR"),
        ],
    },
    ReferenceCatalogTable {
        name: "ducklake_partition_info",
        columns: &[
            col!("partition_id", "BIGINT"),
            col!("table_id", "BIGINT"),
            col!("begin_snapshot", "BIGINT"),
            col!("end_snapshot", "BIGINT"),
        ],
    },
    ReferenceCatalogTable {
        name: "ducklake_partition_column",
        columns: &[
            col!("partition_id", "BIGINT"),
            col!("table_id", "BIGINT"),
            col!("partition_key_index", "BIGINT"),
            col!("column_id", "BIGINT"),
            col!("transform", "VARCHAR"),
        ],
    },
    ReferenceCatalogTable {
        name: "ducklake_file_partition_value",
        columns: &[
            col!("data_file_id", "BIGINT"),
            col!("table_id", "BIGINT"),
            col!("partition_key_index", "BIGINT"),
            col!("partition_value", "VARCHAR"),
        ],
    },
    ReferenceCatalogTable {
        name: "ducklake_files_scheduled_for_deletion",
        columns: &[
            col!("data_file_id", "BIGINT"),
            col!("path", "VARCHAR"),
            col!("path_is_relative", "BOOLEAN"),
            col!("schedule_start", "TIMESTAMPTZ"),
        ],
    },
    ReferenceCatalogTable {
        name: "ducklake_inlined_data_tables",
        columns: &[
            col!("table_id", "BIGINT"),
            col!("table_name", "VARCHAR"),
            col!("schema_version", "BIGINT"),
        ],
    },
    ReferenceCatalogTable {
        name: "ducklake_column_mapping",
        columns: &[
            col!("mapping_id", "BIGINT"),
            col!("table_id", "BIGINT"),
            col!("type", "VARCHAR"),
        ],
    },
    ReferenceCatalogTable {
        name: "ducklake_name_mapping",
        columns: &[
            col!("mapping_id", "BIGINT"),
            col!("column_id", "BIGINT"),
            col!("source_name", "VARCHAR"),
            col!("target_field_id", "BIGINT"),
            col!("parent_column", "BIGINT"),
            col!("is_partition", "BOOLEAN"),
        ],
    },
    ReferenceCatalogTable {
        name: "ducklake_schema_versions",
        columns: &[
            col!("begin_snapshot", "BIGINT"),
            col!("schema_version", "BIGINT"),
            col!("table_id", "BIGINT"),
        ],
    },
    ReferenceCatalogTable {
        name: "ducklake_macro",
        columns: &[
            col!("schema_id", "BIGINT"),
            col!("macro_id", "BIGINT"),
            col!("macro_name", "VARCHAR"),
            col!("begin_snapshot", "BIGINT"),
            col!("end_snapshot", "BIGINT"),
        ],
    },
    ReferenceCatalogTable {
        name: "ducklake_macro_impl",
        columns: &[
            col!("macro_id", "BIGINT"),
            col!("impl_id", "BIGINT"),
            col!("dialect", "VARCHAR"),
            col!("sql", "VARCHAR"),
            col!("type", "VARCHAR"),
        ],
    },
    ReferenceCatalogTable {
        name: "ducklake_macro_parameters",
        columns: &[
            col!("macro_id", "BIGINT"),
            col!("impl_id", "BIGINT"),
            col!("column_id", "BIGINT"),
            col!("parameter_name", "VARCHAR"),
            col!("parameter_type", "VARCHAR"),
            col!("default_value", "VARCHAR"),
            col!("default_value_type", "VARCHAR"),
        ],
    },
    ReferenceCatalogTable {
        name: "ducklake_sort_info",
        columns: &[
            col!("sort_id", "BIGINT"),
            col!("table_id", "BIGINT"),
            col!("begin_snapshot", "BIGINT"),
            col!("end_snapshot", "BIGINT"),
        ],
    },
    ReferenceCatalogTable {
        name: "ducklake_sort_expression",
        columns: &[
            col!("sort_id", "BIGINT"),
            col!("table_id", "BIGINT"),
            col!("sort_key_index", "BIGINT"),
            col!("expression", "VARCHAR"),
            col!("dialect", "VARCHAR"),
            col!("sort_direction", "VARCHAR"),
            col!("null_order", "VARCHAR"),
        ],
    },
];

#[cfg(feature = "postgres")]
pub(crate) fn render_create_schema(schema: &str) -> String {
    format!("CREATE SCHEMA IF NOT EXISTS {}", quote_ident(schema))
}

pub(crate) fn render_create_table_statements(qualifier: Option<&str>) -> Vec<String> {
    REFERENCE_CATALOG_TABLES
        .iter()
        .map(|table| render_create_table(table, qualifier))
        .collect()
}

pub(crate) fn render_bootstrap_statements(
    qualifier: Option<&str>,
    bootstrap: &CatalogBootstrap,
) -> Vec<String> {
    let snapshot = qualified_table("ducklake_snapshot", qualifier);
    let snapshot_changes = qualified_table("ducklake_snapshot_changes", qualifier);
    let metadata = qualified_table("ducklake_metadata", qualifier);
    let schema = qualified_table("ducklake_schema", qualifier);
    let created_schema = sql_string("created_schema:\"main\"");
    let version = sql_string(ducklake_catalog::SUPPORTED_VERSION);
    let created_by = sql_string(&bootstrap.created_by);
    let data_path = sql_string(&bootstrap.data_path);
    let encrypted = sql_string(if bootstrap.encrypted { "true" } else { "false" });
    let schema_uuid = sql_string(&bootstrap.schema_uuid);

    vec![
        format!(
            "INSERT INTO {snapshot} VALUES (0, {}, 0, 1, 0)",
            bootstrap.snapshot_time_sql
        ),
        format!("INSERT INTO {snapshot_changes} VALUES (0, {created_schema}, NULL, NULL, NULL)"),
        format!(
            "INSERT INTO {metadata} (key, value) VALUES \
             ('version', {version}), \
             ('created_by', {created_by}), \
             ('data_path', {data_path}), \
             ('encrypted', {encrypted})"
        ),
        format!("INSERT INTO {schema} VALUES (0, {schema_uuid}, 0, NULL, 'main', 'main/', true)"),
    ]
}

pub(crate) fn render_initialization_statements(
    qualifier: Option<&str>,
    bootstrap: &CatalogBootstrap,
) -> Vec<String> {
    let mut statements = render_create_table_statements(qualifier);
    statements.extend(render_bootstrap_statements(qualifier, bootstrap));
    statements
}

pub(crate) fn render_initialization_batch(
    qualifier: Option<&str>,
    bootstrap: &CatalogBootstrap,
) -> String {
    let statements = render_initialization_statements(qualifier, bootstrap);
    statements
        .into_iter()
        .map(|statement| format!("{statement};"))
        .collect::<Vec<_>>()
        .join("\n")
}

fn render_create_table(table: &ReferenceCatalogTable, qualifier: Option<&str>) -> String {
    let columns = table
        .columns
        .iter()
        .map(|column| {
            let primary_key = if column.primary_key {
                " PRIMARY KEY"
            } else {
                ""
            };
            format!("{} {}{primary_key}", column.name, column.sql_type)
        })
        .collect::<Vec<_>>()
        .join(", ");
    format!(
        "CREATE TABLE {}({columns})",
        qualified_table(table.name, qualifier)
    )
}

fn qualified_table(table: &str, qualifier: Option<&str>) -> String {
    qualifier
        .map(|qualifier| format!("{}.{}", quote_ident(qualifier), table))
        .unwrap_or_else(|| table.to_string())
}

#[cfg(test)]
#[path = "ddl/tests.rs"]
mod tests;
