use async_trait::async_trait;
use ducklake_catalog::{
    InlinedDataTableRef, InlinedFileDeletion, SnapshotId, TableId, TableSchema,
};

use crate::{InlinedDataBatch, Result};

#[async_trait]
pub(crate) trait AsyncInlinedReadExecutor: Sync {
    async fn inlined_file_deletion_table_exists(&self, table_name: &str) -> Result<bool>;

    async fn load_inlined_file_deletions(
        &self,
        table_name: &str,
        snapshot: SnapshotId,
    ) -> Result<Vec<InlinedFileDeletion>>;

    async fn load_inlined_data(
        &self,
        table: &InlinedDataTableRef,
        schema: &TableSchema,
        snapshot: SnapshotId,
    ) -> Result<Vec<arrow::record_batch::RecordBatch>>;

    async fn load_inlined_data_with_metadata(
        &self,
        table: &InlinedDataTableRef,
        schema: &TableSchema,
        snapshot: SnapshotId,
    ) -> Result<Vec<InlinedDataBatch>>;
}

pub(crate) async fn inlined_file_deletions_at<E>(
    executor: &E,
    table: TableId,
    snapshot: SnapshotId,
) -> Result<Vec<InlinedFileDeletion>>
where
    E: AsyncInlinedReadExecutor + Sync,
{
    let table_name = crate::inlined_values::inlined_file_deletion_table_name(table);
    if !executor
        .inlined_file_deletion_table_exists(&table_name)
        .await?
    {
        return Ok(Vec::new());
    }
    executor
        .load_inlined_file_deletions(&table_name, snapshot)
        .await
}

pub(crate) async fn inlined_data<E>(
    executor: &E,
    table: &InlinedDataTableRef,
    schema: &TableSchema,
    snapshot: SnapshotId,
) -> Result<Vec<arrow::record_batch::RecordBatch>>
where
    E: AsyncInlinedReadExecutor + Sync,
{
    executor.load_inlined_data(table, schema, snapshot).await
}

pub(crate) async fn inlined_data_with_metadata<E>(
    executor: &E,
    table: &InlinedDataTableRef,
    schema: &TableSchema,
    snapshot: SnapshotId,
) -> Result<Vec<InlinedDataBatch>>
where
    E: AsyncInlinedReadExecutor + Sync,
{
    executor
        .load_inlined_data_with_metadata(table, schema, snapshot)
        .await
}

macro_rules! impl_async_inlined_read_executor {
    (
        impl $ty:ty {
            file_deletion_table_exists = |$exists_self:ident, $exists_table_name:ident| $exists:expr,
            load_file_deletions = |$delete_self:ident, $delete_table_name:ident, $delete_snapshot:ident| $load_deletions:expr,
            load_data = |$data_self:ident, $table:ident, $schema:ident, $snapshot:ident| $load_data:expr,
            load_data_with_metadata = |$metadata_self:ident, $metadata_table:ident, $metadata_schema:ident, $metadata_snapshot:ident| $load_data_with_metadata:expr $(,)?
        }
    ) => {
        #[async_trait]
        impl $crate::catalog_async_inlined_read_methods::AsyncInlinedReadExecutor for $ty {
            async fn inlined_file_deletion_table_exists(
                &self,
                $exists_table_name: &str,
            ) -> $crate::Result<bool> {
                let $exists_self = self;
                $exists.await
            }

            async fn load_inlined_file_deletions(
                &self,
                $delete_table_name: &str,
                $delete_snapshot: ducklake_catalog::SnapshotId,
            ) -> $crate::Result<Vec<ducklake_catalog::InlinedFileDeletion>> {
                let $delete_self = self;
                $load_deletions.await
            }

            async fn load_inlined_data(
                &self,
                $table: &ducklake_catalog::InlinedDataTableRef,
                $schema: &ducklake_catalog::TableSchema,
                $snapshot: ducklake_catalog::SnapshotId,
            ) -> $crate::Result<Vec<arrow::record_batch::RecordBatch>> {
                let $data_self = self;
                $load_data.await
            }

            async fn load_inlined_data_with_metadata(
                &self,
                $metadata_table: &ducklake_catalog::InlinedDataTableRef,
                $metadata_schema: &ducklake_catalog::TableSchema,
                $metadata_snapshot: ducklake_catalog::SnapshotId,
            ) -> $crate::Result<Vec<$crate::InlinedDataBatch>> {
                let $metadata_self = self;
                $load_data_with_metadata.await
            }
        }
    };
}

macro_rules! impl_async_inlined_read_methods {
    () => {
        fn inlined_file_deletions_at<'life0, 'async_trait>(
            &'life0 self,
            table: ducklake_catalog::TableId,
            snapshot: ducklake_catalog::SnapshotId,
        ) -> std::pin::Pin<
            Box<
                dyn std::future::Future<
                        Output = $crate::Result<Vec<ducklake_catalog::InlinedFileDeletion>>,
                    > + Send
                    + 'async_trait,
            >,
        >
        where
            'life0: 'async_trait,
            Self: Sync + 'async_trait,
        {
            let this = self;
            Box::pin(async move {
                $crate::catalog_async_inlined_read_methods::inlined_file_deletions_at(
                    this, table, snapshot,
                )
                .await
            })
        }

        fn inlined_data<'life0, 'life1, 'life2, 'async_trait>(
            &'life0 self,
            table: &'life1 ducklake_catalog::InlinedDataTableRef,
            schema: &'life2 ducklake_catalog::TableSchema,
            snapshot: ducklake_catalog::SnapshotId,
        ) -> std::pin::Pin<
            Box<
                dyn std::future::Future<
                        Output = $crate::Result<Vec<arrow::record_batch::RecordBatch>>,
                    > + Send
                    + 'async_trait,
            >,
        >
        where
            'life0: 'async_trait,
            'life1: 'async_trait,
            'life2: 'async_trait,
            Self: Sync + 'async_trait,
        {
            let this = self;
            Box::pin(async move {
                $crate::catalog_async_inlined_read_methods::inlined_data(
                    this, table, schema, snapshot,
                )
                .await
            })
        }

        fn inlined_data_with_metadata<'life0, 'life1, 'life2, 'async_trait>(
            &'life0 self,
            table: &'life1 ducklake_catalog::InlinedDataTableRef,
            schema: &'life2 ducklake_catalog::TableSchema,
            snapshot: ducklake_catalog::SnapshotId,
        ) -> std::pin::Pin<
            Box<
                dyn std::future::Future<Output = $crate::Result<Vec<$crate::InlinedDataBatch>>>
                    + Send
                    + 'async_trait,
            >,
        >
        where
            'life0: 'async_trait,
            'life1: 'async_trait,
            'life2: 'async_trait,
            Self: Sync + 'async_trait,
        {
            let this = self;
            Box::pin(async move {
                $crate::catalog_async_inlined_read_methods::inlined_data_with_metadata(
                    this, table, schema, snapshot,
                )
                .await
            })
        }
    };
}
