use ducklake_catalog::{CatalogMetadata, MetadataKv};

use crate::{Result, StorageError};

pub(crate) fn metadata_kv_row(row: &dyn crate::embedded_rows::EmbeddedRow) -> Result<MetadataKv> {
    Ok(MetadataKv {
        key: row.get_string(0)?,
        value: row.get_string(1)?,
        scope: row.get_opt_string(2)?,
        scope_id: row.get_opt_i64(3)?,
    })
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
pub(crate) fn load_catalog_metadata_sync<Query>(query: Query) -> Result<CatalogMetadata>
where
    Query: FnOnce(
        &crate::catalog_queries::CatalogQuery,
        fn(&dyn crate::embedded_rows::EmbeddedRow) -> Result<MetadataKv>,
    ) -> Result<Vec<MetadataKv>>,
{
    catalog_metadata_from_rows(query(
        &crate::catalog_queries::metadata_rows(),
        metadata_kv_row,
    )?)
}

pub(crate) fn catalog_metadata_from_rows(raw: Vec<MetadataKv>) -> Result<CatalogMetadata> {
    let version = raw
        .iter()
        .find(|kv| kv.key == "version" && kv.scope.is_none())
        .map(|kv| kv.value.clone())
        .ok_or_else(|| StorageError::Backend("ducklake_metadata.version missing".into()))?;
    let data_path = raw
        .iter()
        .find(|kv| kv.key == "data_path" && kv.scope.is_none())
        .map(|kv| kv.value.clone())
        .unwrap_or_default();
    let encrypted = raw
        .iter()
        .find(|kv| kv.key == "encrypted" && kv.scope.is_none())
        .map(|kv| kv.value.eq_ignore_ascii_case("true"))
        .unwrap_or(false);

    Ok(CatalogMetadata {
        version,
        data_path,
        encrypted,
        raw,
    })
}
