use async_trait::async_trait;
use ducklake_catalog::{CatalogMetadata, MetadataKv};

use crate::{
    DuckLakeCatalogConnectionOptions, MetadataOptionScope, Result,
    catalog_async_query_methods::AsyncCatalogQueryExecutor, catalog_metadata,
};

#[async_trait]
pub(crate) trait AsyncCatalogMetadataExecutor: AsyncCatalogQueryExecutor + Sync {
    fn connection_options(&self) -> &DuckLakeCatalogConnectionOptions;

    async fn load_metadata_rows(&self) -> Result<Vec<MetadataKv>> {
        self.run_catalog_query(
            &crate::catalog_queries::metadata_rows(),
            catalog_metadata::metadata_kv_row,
        )
        .await
    }

    async fn replace_metadata_option(
        &self,
        key: String,
        value: String,
        scope: MetadataOptionScope,
    ) -> Result<()>;
}

pub(crate) async fn metadata<E>(executor: &E) -> Result<CatalogMetadata>
where
    E: AsyncCatalogMetadataExecutor + Sync,
{
    let metadata =
        catalog_metadata::catalog_metadata_from_rows(executor.load_metadata_rows().await?)?;
    Ok(executor.connection_options().apply_to_metadata(metadata))
}

pub(crate) async fn load_catalog_metadata<E>(executor: &E) -> Result<CatalogMetadata>
where
    E: AsyncCatalogQueryExecutor + Sync,
{
    catalog_metadata::catalog_metadata_from_rows(
        executor
            .run_catalog_query(
                &crate::catalog_queries::metadata_rows(),
                catalog_metadata::metadata_kv_row,
            )
            .await?,
    )
}

pub(crate) async fn set_metadata_option<E>(
    executor: &E,
    key: String,
    value: String,
    scope: MetadataOptionScope,
) -> Result<()>
where
    E: AsyncCatalogMetadataExecutor + Sync,
{
    executor.replace_metadata_option(key, value, scope).await
}

macro_rules! impl_async_catalog_metadata_executor {
    (
        impl $ty:ty {
            connection_options = |$options_self:ident| $connection_options:expr,
            replace_metadata_option = |$replace_self:ident, $key:ident, $value:ident, $scope:ident| $replace:expr $(,)?
        }
    ) => {
        #[async_trait]
        impl $crate::catalog_async_metadata_methods::AsyncCatalogMetadataExecutor for $ty {
            fn connection_options(&self) -> &$crate::DuckLakeCatalogConnectionOptions {
                let $options_self = self;
                $connection_options
            }

            async fn replace_metadata_option(
                &self,
                $key: String,
                $value: String,
                $scope: $crate::MetadataOptionScope,
            ) -> $crate::Result<()> {
                let $replace_self = self;
                $replace.await
            }
        }
    };
}

macro_rules! impl_async_catalog_metadata_methods {
    () => {
        fn metadata<'life0, 'async_trait>(
            &'life0 self,
        ) -> std::pin::Pin<
            Box<
                dyn std::future::Future<Output = $crate::Result<ducklake_catalog::CatalogMetadata>>
                    + Send
                    + 'async_trait,
            >,
        >
        where
            'life0: 'async_trait,
            Self: Sync + 'async_trait,
        {
            let this = self;
            Box::pin(async move { $crate::catalog_async_metadata_methods::metadata(this).await })
        }

        fn set_metadata_option<'life0, 'async_trait>(
            &'life0 self,
            key: String,
            value: String,
            scope: $crate::MetadataOptionScope,
        ) -> std::pin::Pin<
            Box<dyn std::future::Future<Output = $crate::Result<()>> + Send + 'async_trait>,
        >
        where
            'life0: 'async_trait,
            Self: Sync + 'async_trait,
        {
            let this = self;
            Box::pin(async move {
                $crate::catalog_async_metadata_methods::set_metadata_option(this, key, value, scope)
                    .await
            })
        }
    };
}
