macro_rules! embedded_maintenance_method {
    (|$this:ident| fn $name:ident (&self, $arg:ident : $arg_ty:ty) -> $ret:ty $body:block) => {
        fn $name<'life0, 'async_trait>(
            &'life0 self,
            $arg: $arg_ty,
        ) -> std::pin::Pin<
            Box<dyn std::future::Future<Output = $crate::Result<$ret>> + Send + 'async_trait>,
        >
        where
            'life0: 'async_trait,
            Self: Sync + 'async_trait,
        {
            let $this = self;
            Box::pin(async move $body)
        }
    };
}

macro_rules! impl_embedded_catalog_maintenance_methods {
    ($with_conn:ident, |$tx_conn:ident| $begin_tx:expr) => {
        embedded_maintenance_method! { |this|
            fn cleanup_old_files(
                &self,
                options: $crate::CleanupOldFilesOptions
            ) -> Vec<$crate::CleanupFile> {
                this.$with_conn(move |mut conn| {
                    let files = $crate::embedded_maintenance::planned_cleanup_files(&conn, options)?;
                    if options.dry_run || files.is_empty() {
                        return Ok(files);
                    }
                    $crate::embedded_maintenance::remove_scheduled_files_from_disk(&files)?;
                    let removed_ids: Vec<i64> =
                        files.iter().map(|file| file.data_file_id.0).collect();
                    let tx = {
                        let $tx_conn = &mut conn;
                        $begin_tx
                    }?;
                    $crate::embedded_maintenance::delete_by_ids(
                        &*tx,
                        "ducklake_files_scheduled_for_deletion",
                        "data_file_id",
                        &removed_ids,
                    )?;
                    tx.commit().map_err(sql_err)?;
                    Ok(files)
                })
                .await
            }
        }

        fn remove_files_scheduled_for_deletion<'life0, 'async_trait>(
            &'life0 self,
            data_file_ids: Vec<ducklake_catalog::DataFileId>,
        ) -> 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 data_file_ids =
                    $crate::plan_remove_files_scheduled_for_deletion(data_file_ids);
                if data_file_ids.is_empty() {
                    return Ok(());
                }
                this.$with_conn(move |mut conn| {
                    let tx = {
                        let $tx_conn = &mut conn;
                        $begin_tx
                    }?;
                    $crate::embedded_maintenance::delete_by_ids(
                        &*tx,
                        "ducklake_files_scheduled_for_deletion",
                        "data_file_id",
                        &data_file_ids,
                    )?;
                    tx.commit().map_err(sql_err)?;
                    Ok(())
                })
                .await
            })
        }

        embedded_maintenance_method! { |this|
            fn expire_snapshots(
                &self,
                options: $crate::ExpireSnapshotsOptions
            ) -> Vec<ducklake_catalog::Snapshot> {
                options.validate()?;
                this.$with_conn(move |mut conn| {
                    let snapshots =
                        $crate::embedded_maintenance::expirable_snapshots(&conn, &options)?;
                    if options.dry_run || snapshots.is_empty() {
                        return Ok(snapshots);
                    }
                    let tx = {
                        let $tx_conn = &mut conn;
                        $begin_tx
                    }?;
                    $crate::embedded_maintenance::prune_expired_snapshots(&*tx, &snapshots)?;
                    tx.commit().map_err(sql_err)?;
                    Ok(snapshots)
                })
                .await
            }
        }
    };
}

macro_rules! impl_embedded_maintenance_conn {
    (
        table_exists = $table_exists:path,
        catalog_params = $catalog_params:path,
        scheduled_cleanup_timestamp = bound_current_time $(,)?
    ) => {
        impl_embedded_maintenance_conn!(
            @impl
            table_exists = $table_exists,
            catalog_params = $catalog_params,
            insert_scheduled_cleanup_files = |conn, files| {
                let now = $crate::embedded_rows::format_timestamp_micros(
                    $crate::sql_utils::current_time_micros()?,
                )?;
                let mut stmt = conn
                    .prepare(
                        "INSERT INTO ducklake_files_scheduled_for_deletion \
                         VALUES (?1, ?2, ?3, ?4)",
                    )
                    .map_err(sql_err)?;
                for file in files {
                    stmt.execute(params![file.id, file.path, file.path_is_relative, now])
                        .map_err(sql_err)?;
                }
                Ok(())
            },
        );
    };
    (
        table_exists = $table_exists:path,
        catalog_params = $catalog_params:path,
        scheduled_cleanup_timestamp = current_timestamp $(,)?
    ) => {
        impl_embedded_maintenance_conn!(
            @impl
            table_exists = $table_exists,
            catalog_params = $catalog_params,
            insert_scheduled_cleanup_files = |conn, files| {
                let mut stmt = conn
                    .prepare(
                        "INSERT INTO ducklake_files_scheduled_for_deletion \
                         VALUES (?1, ?2, ?3, CURRENT_TIMESTAMP)",
                    )
                    .map_err(sql_err)?;
                for file in files {
                    stmt.execute(params![file.id, file.path, file.path_is_relative])
                        .map_err(sql_err)?;
                }
                Ok(())
            },
        );
    };
    (
        @impl
        table_exists = $table_exists:path,
        catalog_params = $catalog_params:path,
        insert_scheduled_cleanup_files = |$this:ident, $files:ident| $insert:block $(,)?
    ) => {
        impl $crate::embedded_maintenance::MaintenanceConn for Connection {
            fn table_exists(&self, table: &str) -> $crate::Result<bool> {
                $table_exists(self, table)
            }

            fn execute_sql(&self, sql: &str) -> $crate::Result<()> {
                self.execute(sql, []).map_err(sql_err)?;
                Ok(())
            }

            fn execute_catalog_query(
                &self,
                query: &$crate::catalog_queries::CatalogQuery,
            ) -> $crate::Result<usize> {
                let sql = query.render_embedded();
                let params = $catalog_params(query.params());
                self.execute(&sql, params_from_iter(params))
                    .map_err(sql_err)
            }

            fn query_rows<T, F>(&self, sql: &str, params: &[i64], map: F) -> $crate::Result<Vec<T>>
            where
                F: FnMut(&dyn $crate::embedded_rows::EmbeddedRow) -> $crate::Result<T>,
            {
                let mut stmt = self.prepare(sql).map_err(sql_err)?;
                collect_shared_rows(&mut stmt, params_from_iter(params.iter().copied()), map)
            }

            fn metadata_value(&self, key: &str) -> $crate::Result<Option<String>> {
                global_metadata_value(self, key)
            }

            fn insert_scheduled_cleanup_files(
                &self,
                files: &[$crate::CleanupMetadataFile],
            ) -> $crate::Result<()> {
                let $this = self;
                let $files = files;
                $insert
            }
        }
    };
}
