use async_trait::async_trait;
use ducklake_catalog::CatalogMetadata;
#[cfg(any(feature = "postgres", feature = "quack"))]
use std::future::Future;

use crate::{
    DuckLakeCatalogConnectionOptions, DuckLakeCatalogInitOptions, Result, StorageError,
    catalog_ddl::{self, CatalogBootstrap},
};

#[async_trait]
pub(crate) trait CatalogInitializationExecutor: Sync {
    fn initialization_catalog_type(&self) -> &'static str;

    fn initialization_connection_options(&self) -> &DuckLakeCatalogConnectionOptions;

    fn default_bootstrap_data_path(&self) -> Option<String>;

    fn bootstrap_snapshot_time_sql(&self) -> Result<String>;

    async fn run_initialization_request(
        &self,
        request: CatalogInitializationRequest,
    ) -> Result<CatalogInitializationResponse>;

    async fn load_existing_metadata_for_init(&self) -> Result<Option<CatalogMetadata>> {
        self.run_initialization_request(CatalogInitializationRequest::LoadExisting)
            .await?
            .into_optional_metadata()
    }

    async fn create_metadata_or_load_existing(
        &self,
        bootstrap: CatalogBootstrap,
    ) -> Result<CatalogMetadata> {
        self.run_initialization_request(CatalogInitializationRequest::CreateOrLoad(bootstrap))
            .await?
            .into_metadata()
    }
}

pub(crate) enum CatalogInitializationRequest {
    LoadExisting,
    CreateOrLoad(CatalogBootstrap),
}

pub(crate) enum CatalogInitializationResponse {
    OptionalMetadata(Option<CatalogMetadata>),
    Metadata(CatalogMetadata),
}

impl CatalogInitializationResponse {
    fn into_optional_metadata(self) -> Result<Option<CatalogMetadata>> {
        match self {
            Self::OptionalMetadata(metadata) => Ok(metadata),
            Self::Metadata(_) => Err(unexpected_initialization_response("optional metadata")),
        }
    }

    fn into_metadata(self) -> Result<CatalogMetadata> {
        match self {
            Self::Metadata(metadata) => Ok(metadata),
            Self::OptionalMetadata(_) => Err(unexpected_initialization_response("metadata")),
        }
    }
}

impl CatalogInitializationRequest {
    pub(crate) fn execute_sync_session<S, Exists, Load, Create>(
        self,
        session: &mut S,
        exists: Exists,
        mut load: Load,
        create: Create,
    ) -> Result<CatalogInitializationResponse>
    where
        Exists: FnOnce(&mut S) -> Result<bool>,
        Load: FnMut(&mut S) -> Result<CatalogMetadata>,
        Create: FnOnce(&mut S, CatalogBootstrap) -> Result<()>,
    {
        match self {
            Self::LoadExisting => {
                if exists(session)? {
                    Ok(CatalogInitializationResponse::OptionalMetadata(Some(load(
                        session,
                    )?)))
                } else {
                    Ok(CatalogInitializationResponse::OptionalMetadata(None))
                }
            }
            Self::CreateOrLoad(bootstrap) => {
                if exists(session)? {
                    return Ok(CatalogInitializationResponse::Metadata(load(session)?));
                }
                create(session, bootstrap)?;
                Ok(CatalogInitializationResponse::Metadata(load(session)?))
            }
        }
    }

    #[cfg(any(feature = "postgres", feature = "quack"))]
    pub(crate) async fn execute_async<
        Exists,
        ExistsFuture,
        Load,
        LoadFuture,
        Create,
        CreateFuture,
    >(
        self,
        exists: Exists,
        load: Load,
        create: Create,
    ) -> Result<CatalogInitializationResponse>
    where
        Exists: FnOnce() -> ExistsFuture,
        ExistsFuture: Future<Output = Result<bool>>,
        Load: Fn() -> LoadFuture,
        LoadFuture: Future<Output = Result<CatalogMetadata>>,
        Create: FnOnce(CatalogBootstrap) -> CreateFuture,
        CreateFuture: Future<Output = Result<()>>,
    {
        match self {
            Self::LoadExisting => {
                if exists().await? {
                    Ok(CatalogInitializationResponse::OptionalMetadata(Some(
                        load().await?,
                    )))
                } else {
                    Ok(CatalogInitializationResponse::OptionalMetadata(None))
                }
            }
            Self::CreateOrLoad(bootstrap) => {
                if exists().await? {
                    return Ok(CatalogInitializationResponse::Metadata(load().await?));
                }
                create(bootstrap).await?;
                Ok(CatalogInitializationResponse::Metadata(load().await?))
            }
        }
    }
}

pub(crate) async fn initialize_ducklake<E>(
    executor: &E,
    options: DuckLakeCatalogInitOptions,
) -> Result<CatalogMetadata>
where
    E: CatalogInitializationExecutor + Sync,
{
    if let Some(metadata) = executor.load_existing_metadata_for_init().await? {
        return validate_and_apply(executor, metadata, &options);
    }

    if !options.create_if_not_exists {
        return Err(missing_catalog_error());
    }

    let default_data_path = executor.default_bootstrap_data_path();
    let bootstrap = catalog_ddl::bootstrap_from_options(
        &options,
        default_data_path.as_deref(),
        executor.bootstrap_snapshot_time_sql()?,
    )?;
    let metadata = executor.create_metadata_or_load_existing(bootstrap).await?;
    validate_and_apply(executor, metadata, &options)
}

fn validate_and_apply<E>(
    executor: &E,
    metadata: CatalogMetadata,
    options: &DuckLakeCatalogInitOptions,
) -> Result<CatalogMetadata>
where
    E: CatalogInitializationExecutor + Sync,
{
    let metadata = crate::validate_loaded_ducklake_metadata(
        executor.initialization_catalog_type(),
        &metadata,
        options,
    )?;
    Ok(executor
        .initialization_connection_options()
        .apply_to_metadata(metadata))
}

pub(crate) fn missing_catalog_error() -> StorageError {
    StorageError::Backend(
        "Existing DuckLake catalog does not exist and CREATE_IF_NOT_EXISTS is false".into(),
    )
}

fn unexpected_initialization_response(expected: &str) -> StorageError {
    StorageError::Backend(format!(
        "internal catalog initialization returned a response that did not contain {expected}"
    ))
}

macro_rules! impl_catalog_initialization_info_methods {
    (
        catalog_type = $catalog_type:expr,
        default_data_path = |$default_self:ident| $default_data_path:expr,
        snapshot_time_sql = |$snapshot_self:ident| $snapshot_time_sql:expr $(,)?
    ) => {
        fn initialization_catalog_type(&self) -> &'static str {
            $catalog_type
        }

        fn initialization_connection_options(&self) -> &$crate::DuckLakeCatalogConnectionOptions {
            &self.connection_options
        }

        fn default_bootstrap_data_path(&self) -> Option<String> {
            let $default_self = self;
            $default_data_path
        }

        fn bootstrap_snapshot_time_sql(&self) -> $crate::Result<String> {
            let $snapshot_self = self;
            $snapshot_time_sql
        }
    };
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
macro_rules! impl_sync_catalog_initialization_request {
    (
        with_session = $with_session:ident,
        capture = |$this:ident| { $($capture:stmt)* },
        exists = |$exists_session:ident| $exists:expr,
        load = |$load_session:ident| $load:expr,
        create = |$create_session:ident, $bootstrap:ident| $create:expr $(,)?
    ) => {
        fn run_initialization_request<'life0, 'async_trait>(
            &'life0 self,
            request: $crate::catalog_initialization_methods::CatalogInitializationRequest,
        ) -> std::pin::Pin<
            Box<
                dyn std::future::Future<
                        Output = $crate::Result<
                            $crate::catalog_initialization_methods::CatalogInitializationResponse,
                        >,
                    > + Send
                    + 'async_trait,
            >,
        >
        where
            'life0: 'async_trait,
            Self: Sync + 'async_trait,
        {
            let $this = self;
            Box::pin(async move {
                $($capture)*
                $this
                    .$with_session(move |mut session| {
                        request.execute_sync_session(
                            &mut session,
                            |$exists_session| $exists,
                            |$load_session| $load,
                            |$create_session, $bootstrap| $create,
                        )
                    })
                    .await
            })
        }
    };
}

macro_rules! impl_catalog_initialization_method {
    () => {
        fn initialize_ducklake<'life0, 'async_trait>(
            &'life0 self,
            options: $crate::DuckLakeCatalogInitOptions,
        ) -> 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_initialization_methods::initialize_ducklake(this, options).await
            })
        }
    };
}
