macro_rules! catalog_commit_method {
    (fn $name:ident (&self, $arg:ident : $arg_ty:ty) => $request:expr) => {
        fn $name<'life0, 'async_trait>(
            &'life0 self,
            $arg: $arg_ty,
        ) -> std::pin::Pin<
            Box<
                dyn std::future::Future<Output = $crate::Result<ducklake_catalog::SnapshotId>>
                    + Send
                    + 'async_trait,
            >,
        >
        where
            'life0: 'async_trait,
            Self: Sync + 'async_trait,
        {
            Box::pin(async move { self.run_commit_request($request).await })
        }
    };
}

macro_rules! impl_connection_options_builder_method {
    () => {
        pub fn with_connection_options(
            mut self,
            options: $crate::DuckLakeCatalogConnectionOptions,
        ) -> Self {
            self.connection_options = options;
            self
        }
    };
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
macro_rules! impl_path_accessor_method {
    () => {
        pub fn path(&self) -> &std::path::Path {
            &self.path
        }
    };
}

#[cfg(any(feature = "duckdb", feature = "postgres", test))]
macro_rules! impl_metadata_schema_accessor_method {
    ($return_ty:ty, |$this:ident| $body:expr) => {
        pub fn metadata_schema(&self) -> $return_ty {
            let $this = self;
            $body
        }
    };
}

macro_rules! impl_catalog_type_method {
    ($catalog_type:expr) => {
        fn catalog_type(&self) -> &'static str {
            $catalog_type
        }
    };
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
macro_rules! impl_embedded_set_metadata_option_method {
    ($with_conn:ident, |$tx_conn:ident| $begin_tx:expr) => {
        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 {
                let sql = $crate::metadata_option::render_set_metadata_option(
                    "ducklake_metadata",
                    $crate::metadata_option::MetadataPlaceholderStyle::Numbered,
                    $crate::metadata_option::NullSafeEquality::OrIsNull,
                );
                this.$with_conn(move |mut conn| {
                    let tx = {
                        let $tx_conn = &mut conn;
                        $begin_tx
                    }?;
                    tx.execute(&sql.delete, params![&key, &scope.scope, &scope.scope_id])
                        .map_err(sql_err)?;
                    tx.execute(
                        &sql.insert,
                        params![
                            $crate::metadata_option::normalized_metadata_option_key(&key),
                            value,
                            &scope.scope,
                            &scope.scope_id
                        ],
                    )
                    .map_err(sql_err)?;
                    tx.commit().map_err(sql_err)
                })
                .await
            })
        }
    };
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
macro_rules! impl_embedded_run_commit_request_method {
    ($ops:expr, $with_conn:ident, |$tx_conn:ident| $begin_tx:expr) => {
        async fn run_commit_request(
            &self,
            request: $crate::commit_engine::CommitRequest,
        ) -> $crate::Result<ducklake_catalog::SnapshotId> {
            request.validate_preflight()?;
            self.$with_conn(move |mut conn| {
                let tx = {
                    let $tx_conn = &mut conn;
                    $begin_tx
                }?;
                let snapshot =
                    $crate::commit_engine::apply_sync_commit_request(&$ops, &tx, &request)?;
                tx.commit().map_err(sql_err)?;
                Ok(snapshot)
            })
            .await
        }
    };
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
macro_rules! impl_embedded_metadata_method {
    ($with_conn:ident) => {
        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;
            let connection_options = self.connection_options.clone();
            Box::pin(async move {
                this.$with_conn(move |conn| {
                    let metadata =
                        $crate::catalog_metadata::load_catalog_metadata_sync(|query, mapper| {
                            run_catalog_query(&conn, query, mapper)
                        })?;
                    Ok(connection_options.apply_to_metadata(metadata))
                })
                .await
            })
        }
    };
}

macro_rules! impl_catalog_backend_commit_methods {
    () => {
        catalog_commit_method! {
            fn commit_append_files(&self, files: Vec<$crate::AppendFile>) =>
                $crate::commit_engine::CommitRequest::AppendFiles(files)
        }

        fn supports_inlined_appends(&self) -> bool {
            true
        }

        catalog_commit_method! {
            fn commit_append_inlined_data(&self, append: $crate::InlinedDataAppendCommit) =>
                $crate::commit_engine::CommitRequest::AppendInlinedData(append)
        }

        catalog_commit_method! {
            fn commit_delete_files(&self, files: Vec<$crate::DeleteFileCommit>) =>
                $crate::commit_engine::CommitRequest::DeleteFiles(files)
        }

        catalog_commit_method! {
            fn commit_table_changes(&self, changes: $crate::TableChangeCommit) =>
                $crate::commit_engine::CommitRequest::TableChanges(changes)
        }

        catalog_commit_method! {
            fn commit_create_schema(&self, create: $crate::CreateSchemaCommit) =>
                $crate::commit_engine::CommitRequest::CreateSchema(create)
        }

        catalog_commit_method! {
            fn commit_create_table(&self, create: $crate::CreateTableCommit) =>
                $crate::commit_engine::CommitRequest::CreateTable(create)
        }

        catalog_commit_method! {
            fn commit_create_table_with_files(&self, create: $crate::CreateTableWithFilesCommit) =>
                $crate::commit_engine::CommitRequest::CreateTableWithFiles(create)
        }

        catalog_commit_method! {
            fn commit_create_table_with_inlined_data(
                &self,
                create: $crate::CreateTableWithInlinedDataCommit
            ) => $crate::commit_engine::CommitRequest::CreateTableWithInlinedData(create)
        }

        catalog_commit_method! {
            fn commit_create_view(&self, create: $crate::CreateViewCommit) =>
                $crate::commit_engine::CommitRequest::CreateView(create)
        }

        catalog_commit_method! {
            fn commit_drop_table(&self, drop: $crate::DropTableCommit) =>
                $crate::commit_engine::CommitRequest::DropTable(drop)
        }

        catalog_commit_method! {
            fn commit_rename_table(&self, rename: $crate::RenameTableCommit) =>
                $crate::commit_engine::CommitRequest::RenameTable(rename)
        }

        catalog_commit_method! {
            fn commit_drop_view(&self, drop: $crate::DropViewCommit) =>
                $crate::commit_engine::CommitRequest::DropView(drop)
        }

        catalog_commit_method! {
            fn commit_drop_schema(&self, drop: $crate::DropSchemaCommit) =>
                $crate::commit_engine::CommitRequest::DropSchema(drop)
        }

        catalog_commit_method! {
            fn commit_column_nullability(&self, change: $crate::ColumnNullabilityCommit) =>
                $crate::commit_engine::CommitRequest::ColumnNullability(change)
        }

        catalog_commit_method! {
            fn commit_column_default(&self, change: $crate::ColumnDefaultCommit) =>
                $crate::commit_engine::CommitRequest::ColumnDefault(change)
        }

        catalog_commit_method! {
            fn commit_column_type(&self, change: $crate::ColumnTypeCommit) =>
                $crate::commit_engine::CommitRequest::ColumnType(change)
        }

        catalog_commit_method! {
            fn commit_column_type_tree(&self, change: $crate::ColumnTypeTreeCommit) =>
                $crate::commit_engine::CommitRequest::ColumnTypeTree(change)
        }

        catalog_commit_method! {
            fn commit_rename_column(&self, rename: $crate::RenameColumnCommit) =>
                $crate::commit_engine::CommitRequest::RenameColumn(rename)
        }

        catalog_commit_method! {
            fn commit_drop_column(&self, drop: $crate::DropColumnCommit) =>
                $crate::commit_engine::CommitRequest::DropColumn(drop)
        }

        catalog_commit_method! {
            fn commit_add_columns(&self, add: $crate::AddColumnCommit) =>
                $crate::commit_engine::CommitRequest::AddColumns(add)
        }

        catalog_commit_method! {
            fn commit_partitioning_batch(&self, partitioning: $crate::PartitioningBatchCommit) =>
                $crate::commit_engine::CommitRequest::PartitioningBatch(partitioning)
        }

        catalog_commit_method! {
            fn commit_sorted_by(&self, sorted_by: $crate::SortedByCommit) =>
                $crate::commit_engine::CommitRequest::SortedBy(sorted_by)
        }

        catalog_commit_method! {
            fn commit_truncate_table(&self, truncate: $crate::TruncateTableCommit) =>
                $crate::commit_engine::CommitRequest::TruncateTable(truncate)
        }

        catalog_commit_method! {
            fn commit_merge_adjacent_compaction(
                &self,
                compaction: $crate::MergeAdjacentCompactionCommit
            ) => $crate::commit_engine::CommitRequest::MergeAdjacentCompaction(compaction)
        }

        catalog_commit_method! {
            fn commit_rewrite_data_files(&self, rewrite: $crate::RewriteDataFilesCommit) =>
                $crate::commit_engine::CommitRequest::RewriteDataFiles(rewrite)
        }
    };
}
