use async_trait::async_trait;
use ducklake_catalog::TableId;

use crate::catalog_async_query_methods::AsyncCatalogQueryExecutor;
use crate::{
    FlushedInlinedDataCommit, FlushedInlinedFileDeletionCommit, InlinedDataDeletionCommit,
    InlinedFileDeletionCommit, Result,
};

#[async_trait]
pub(crate) trait AsyncInlinedMutationExecutor: Sync {
    fn inlined_sql_dialect(&self) -> crate::inlined_values::InlinedSqlDialect;

    fn quote_inlined_identifier(&self, value: &str) -> String {
        crate::ducklake_quote_identifier(value)
    }

    fn inlined_metadata_table_identifier(&self, table_name: &str) -> String;

    fn inlined_data_table_identifier(&self, table_name: &str) -> String {
        self.inlined_metadata_table_identifier(table_name)
    }

    fn inlined_file_deletion_table_identifier(&self, table_name: &str) -> String {
        self.inlined_metadata_table_identifier(table_name)
    }

    fn inlined_not_equal_op(&self) -> &'static str {
        match self.inlined_sql_dialect() {
            crate::inlined_values::InlinedSqlDialect::Postgres => "<>",
            crate::inlined_values::InlinedSqlDialect::Quack => "!=",
            crate::inlined_values::InlinedSqlDialect::Sqlite
            | crate::inlined_values::InlinedSqlDialect::DuckDbFile => "!=",
        }
    }

    async fn inlined_table_exists(&self, table_name: String) -> Result<bool>;

    async fn inlined_data_table_names(&self, table_id: TableId) -> Result<Vec<String>>
    where
        Self: AsyncCatalogQueryExecutor,
    {
        self.run_catalog_query(
            &crate::catalog_queries::inlined_data_table_names(table_id),
            crate::catalog_queries::string_value,
        )
        .await
    }

    async fn execute_inlined_sql(&self, sql: String) -> Result<()>;

    async fn query_inlined_i64(&self, sql: String) -> Result<i64>;

    async fn query_optional_inlined_string(&self, sql: String) -> Result<Option<String>>;
}

#[async_trait]
pub(crate) trait AsyncInlinedBatchInsertExecutor: AsyncInlinedMutationExecutor {
    type Value: Send;
    type InsertCommand: Send + Sync;

    fn prepare_inlined_insert(
        &self,
        table_name: &str,
        schema: &ducklake_catalog::TableSchema,
    ) -> Result<Self::InsertCommand>;

    fn inlined_insert_i64_value(&self, value: i64) -> Self::Value;

    fn inlined_insert_null_value(&self) -> Self::Value;

    fn inlined_insert_metadata_values(&self, row_id: i64, begin_snapshot: i64) -> Vec<Self::Value> {
        crate::inlined_values::inlined_insert_metadata_values(
            row_id,
            begin_snapshot,
            |value| self.inlined_insert_i64_value(value),
            || self.inlined_insert_null_value(),
        )
    }

    fn inlined_insert_value(
        &self,
        field: &ducklake_catalog::SchemaField,
        array: &dyn arrow::array::Array,
        row: usize,
    ) -> Result<Self::Value>;

    async fn execute_prepared_inlined_insert(
        &self,
        command: &Self::InsertCommand,
        values: Vec<Self::Value>,
    ) -> Result<()>;
}

#[async_trait]
impl<T> AsyncInlinedMutationExecutor for &T
where
    T: AsyncInlinedMutationExecutor + Sync + ?Sized,
{
    fn inlined_sql_dialect(&self) -> crate::inlined_values::InlinedSqlDialect {
        (**self).inlined_sql_dialect()
    }

    fn quote_inlined_identifier(&self, value: &str) -> String {
        (**self).quote_inlined_identifier(value)
    }

    fn inlined_metadata_table_identifier(&self, table_name: &str) -> String {
        (**self).inlined_metadata_table_identifier(table_name)
    }

    fn inlined_data_table_identifier(&self, table_name: &str) -> String {
        (**self).inlined_data_table_identifier(table_name)
    }

    fn inlined_file_deletion_table_identifier(&self, table_name: &str) -> String {
        (**self).inlined_file_deletion_table_identifier(table_name)
    }

    fn inlined_not_equal_op(&self) -> &'static str {
        (**self).inlined_not_equal_op()
    }

    async fn inlined_table_exists(&self, table_name: String) -> Result<bool> {
        (**self).inlined_table_exists(table_name).await
    }

    async fn execute_inlined_sql(&self, sql: String) -> Result<()> {
        (**self).execute_inlined_sql(sql).await
    }

    async fn query_inlined_i64(&self, sql: String) -> Result<i64> {
        (**self).query_inlined_i64(sql).await
    }

    async fn query_optional_inlined_string(&self, sql: String) -> Result<Option<String>> {
        (**self).query_optional_inlined_string(sql).await
    }
}

#[async_trait]
impl<T> AsyncInlinedBatchInsertExecutor for &T
where
    T: AsyncInlinedBatchInsertExecutor + Sync + ?Sized,
{
    type Value = T::Value;
    type InsertCommand = T::InsertCommand;

    fn prepare_inlined_insert(
        &self,
        table_name: &str,
        schema: &ducklake_catalog::TableSchema,
    ) -> Result<Self::InsertCommand> {
        (**self).prepare_inlined_insert(table_name, schema)
    }

    fn inlined_insert_i64_value(&self, value: i64) -> Self::Value {
        (**self).inlined_insert_i64_value(value)
    }

    fn inlined_insert_null_value(&self) -> Self::Value {
        (**self).inlined_insert_null_value()
    }

    fn inlined_insert_metadata_values(&self, row_id: i64, begin_snapshot: i64) -> Vec<Self::Value> {
        (**self).inlined_insert_metadata_values(row_id, begin_snapshot)
    }

    fn inlined_insert_value(
        &self,
        field: &ducklake_catalog::SchemaField,
        array: &dyn arrow::array::Array,
        row: usize,
    ) -> Result<Self::Value> {
        (**self).inlined_insert_value(field, array, row)
    }

    async fn execute_prepared_inlined_insert(
        &self,
        command: &Self::InsertCommand,
        values: Vec<Self::Value>,
    ) -> Result<()> {
        (**self)
            .execute_prepared_inlined_insert(command, values)
            .await
    }
}

pub(crate) async fn insert_inlined_batches<E>(
    executor: &E,
    table_name: &str,
    schema: &ducklake_catalog::TableSchema,
    batches: &[arrow::record_batch::RecordBatch],
    preserved_row_ids: Option<&[i64]>,
    row_id_start: i64,
    begin_snapshot: i64,
) -> Result<()>
where
    E: AsyncInlinedBatchInsertExecutor + Sync,
{
    let command = executor.prepare_inlined_insert(table_name, schema)?;
    crate::inlined_values::write_inlined_batch_rows_async(
        schema,
        batches,
        preserved_row_ids,
        row_id_start,
        begin_snapshot,
        |row_id, begin_snapshot| executor.inlined_insert_metadata_values(row_id, begin_snapshot),
        |field, array, row| executor.inlined_insert_value(field, array, row),
        |values| {
            let command = &command;
            async move {
                executor
                    .execute_prepared_inlined_insert(command, values)
                    .await
            }
        },
    )
    .await
}

pub(crate) async fn apply_inlined_data_deletions<E>(
    executor: &E,
    deletions: &[InlinedDataDeletionCommit],
    new_snapshot: i64,
) -> Result<i64>
where
    E: AsyncInlinedMutationExecutor + Sync,
{
    crate::inlined_values::apply_inlined_data_deletions_bulk_sql_async(
        deletions,
        new_snapshot,
        |table_name| executor.inlined_data_table_identifier(table_name),
        executor.inlined_not_equal_op(),
        |sql| async move { executor.execute_inlined_sql(sql).await },
        |sql| async move { executor.query_inlined_i64(sql).await },
    )
    .await
}

pub(crate) async fn ensure_inlined_data_table<E>(
    executor: &E,
    table_id: TableId,
    schema_version: i64,
    schema: &ducklake_catalog::TableSchema,
) -> Result<String>
where
    E: AsyncInlinedMutationExecutor + AsyncCatalogQueryExecutor + Sync,
{
    crate::inlined_values::ensure_inlined_data_table_async(
        table_id,
        schema_version,
        schema,
        executor.inlined_sql_dialect(),
        |value| executor.quote_inlined_identifier(value),
        |table_name| executor.inlined_data_table_identifier(table_name),
        |table_id, schema_version| async move {
            executor
                .query_optional_inlined_string(format!(
                    "SELECT table_name \
                     FROM {} \
                     WHERE table_id = {} AND schema_version = {}",
                    executor.inlined_metadata_table_identifier("ducklake_inlined_data_tables"),
                    table_id.0,
                    schema_version
                ))
                .await
        },
        |table_id, table_name, schema_version| async move {
            executor
                .execute_inlined_sql(format!(
                    "INSERT INTO {} VALUES ({}, {}, {})",
                    executor.inlined_metadata_table_identifier("ducklake_inlined_data_tables"),
                    table_id.0,
                    crate::catalog_ddl::sql_string(&table_name),
                    schema_version
                ))
                .await
        },
        |create_sql| async move { executor.execute_inlined_sql(create_sql).await },
    )
    .await
}

pub(crate) async fn delete_flushed_inlined_data<E>(
    executor: &E,
    flushed_tables: &[FlushedInlinedDataCommit],
    flush_snapshot: i64,
) -> Result<i64>
where
    E: AsyncInlinedMutationExecutor + Sync,
{
    crate::inlined_values::delete_flushed_inlined_data_bulk_sql_async(
        flushed_tables,
        flush_snapshot,
        |table_name| executor.inlined_data_table_identifier(table_name),
        |sql| async move { executor.execute_inlined_sql(sql).await },
        |sql| async move { executor.query_inlined_i64(sql).await },
    )
    .await
}

pub(crate) async fn delete_flushed_inlined_file_deletions<E>(
    executor: &E,
    table_id: TableId,
    flushed_deletions: &[FlushedInlinedFileDeletionCommit],
) -> Result<()>
where
    E: AsyncInlinedMutationExecutor + Sync,
{
    crate::inlined_values::delete_flushed_inlined_file_deletions_by_predicate_sql_async(
        table_id,
        flushed_deletions,
        |table_name| executor.inlined_file_deletion_table_identifier(table_name),
        |table_name| async move { executor.inlined_table_exists(table_name).await },
        |sql| async move { executor.query_inlined_i64(sql).await },
        |sql| async move { executor.execute_inlined_sql(sql).await },
    )
    .await
}

pub(crate) async fn close_inlined_data_rows_for_table<E>(
    executor: &E,
    table_id: TableId,
    new_snapshot: i64,
) -> Result<()>
where
    E: AsyncInlinedMutationExecutor + AsyncCatalogQueryExecutor + Sync,
{
    for table_name in executor.inlined_data_table_names(table_id).await? {
        if executor.inlined_table_exists(table_name.clone()).await? {
            executor
                .execute_inlined_sql(crate::inlined_values::close_all_inlined_data_rows_sql(
                    &executor.inlined_data_table_identifier(&table_name),
                    &new_snapshot.to_string(),
                ))
                .await?;
        }
    }
    Ok(())
}

pub(crate) async fn write_inlined_file_deletions<E>(
    executor: &E,
    table_id: TableId,
    deletions: &[InlinedFileDeletionCommit],
    new_snapshot: i64,
) -> Result<()>
where
    E: AsyncInlinedMutationExecutor + Sync,
{
    crate::inlined_values::write_inlined_file_deletions_by_row_async(
        table_id,
        deletions,
        new_snapshot,
        |table_name| executor.inlined_file_deletion_table_identifier(table_name),
        |create_sql| async move { executor.execute_inlined_sql(create_sql).await },
        |quoted_table, row, new_snapshot| async move {
            executor
                .execute_inlined_sql(crate::inlined_values::insert_inlined_file_deletion_sql(
                    &quoted_table,
                    &format!("{}, {}, {new_snapshot}", row.data_file_id, row.row_id),
                ))
                .await
        },
    )
    .await
}

pub(crate) async fn delete_inlined_file_deletions_for_files<E>(
    executor: &E,
    table_id: TableId,
    source_ids: &[i64],
) -> Result<()>
where
    E: AsyncInlinedMutationExecutor + Sync,
{
    crate::inlined_values::delete_inlined_file_deletions_for_files_async(
        table_id,
        source_ids,
        |table_name| executor.inlined_file_deletion_table_identifier(table_name),
        |table_name| async move { executor.inlined_table_exists(table_name).await },
        |sql| async move { executor.execute_inlined_sql(sql).await },
    )
    .await
}
