use std::fs;
use std::future::Future;
use std::io::ErrorKind;

use async_trait::async_trait;
use ducklake_catalog::{DataFileId, FileScheduledForDeletion, Snapshot, TableId};

use crate::catalog_async_query_methods::AsyncReadQueryExecutor;
use crate::{
    CleanupFile, CleanupMetadataFile, CleanupOldFilesOptions, ExpireSnapshotsOptions, Result,
    StorageError, local_filesystem_path, plan_cleanup_metadata_file_ids, plan_cleanup_old_files,
    plan_remove_files_scheduled_for_deletion, plan_snapshot_expiration_prune,
};

pub(crate) struct BorrowedCleanupSqlExecutor<'a, E: ?Sized> {
    executor: &'a E,
}

pub(crate) fn borrowed_cleanup_sql_executor<E: ?Sized>(
    executor: &E,
) -> BorrowedCleanupSqlExecutor<'_, E> {
    BorrowedCleanupSqlExecutor { executor }
}

#[async_trait]
pub(crate) trait AsyncCatalogMaintenanceExecutor: Sync {
    async fn load_cleanup_data_path(&self) -> Result<String>;

    async fn load_scheduled_cleanup_files(&self) -> Result<Vec<FileScheduledForDeletion>>;

    async fn delete_scheduled_cleanup_files(&self, data_file_ids: &[i64]) -> Result<()>;

    async fn load_expirable_snapshots(
        &self,
        options: &ExpireSnapshotsOptions,
    ) -> Result<Vec<Snapshot>>;

    async fn prune_expired_snapshots(&self, snapshots: &[Snapshot]) -> Result<()>;
}

#[async_trait]
pub(crate) trait AsyncSnapshotPruneExecutor:
    crate::metadata_prune_engine::AsyncMetadataPruneExecutor + Sync
{
    async fn tables_unreferenced_by_snapshots(&self) -> Result<Vec<i64>>;

    async fn data_files_unreferenced_by_snapshots(
        &self,
        deleted_table_ids: &[i64],
    ) -> Result<Vec<CleanupMetadataFile>>;

    async fn delete_files_unreferenced_by_snapshots(
        &self,
        deleted_table_ids: &[i64],
        deleted_data_file_ids: &[i64],
    ) -> Result<Vec<CleanupMetadataFile>>;

    async fn schedule_cleanup_files(&self, files: &[CleanupMetadataFile]) -> Result<()>;
}

#[async_trait]
pub(crate) trait AsyncCleanupSqlExecutor:
    crate::metadata_prune_engine::AsyncMetadataPruneExecutor + Sync
{
    fn metadata_table_ref(&self, table: &str) -> String;

    fn i64_placeholder(&self, index: usize) -> String;

    async fn cleanup_table_exists(&self, table: &str) -> Result<bool>;

    async fn cleanup_path_context_available(&self) -> Result<bool>;

    async fn cleanup_data_path(&self) -> Result<String>;

    async fn query_i64_rows<T, F>(
        &self,
        sql: String,
        params: Vec<i64>,
        mapper: F,
    ) -> Result<Vec<T>>
    where
        T: Send + 'static,
        F: FnMut(&dyn crate::embedded_rows::EmbeddedRow) -> Result<T> + Send;

    async fn insert_scheduled_cleanup_file(&self, file: CleanupMetadataFile) -> Result<()>;
}

#[async_trait]
impl<E> crate::metadata_prune_engine::AsyncMetadataPruneExecutor
    for BorrowedCleanupSqlExecutor<'_, E>
where
    E: crate::metadata_prune_engine::AsyncMetadataPruneExecutor + Sync + ?Sized,
{
    async fn table_exists(&self, table: &str) -> Result<bool> {
        self.executor.table_exists(table).await
    }

    async fn inlined_table_names_for_table_ids(&self, table_ids: &[i64]) -> Result<Vec<String>> {
        self.executor
            .inlined_table_names_for_table_ids(table_ids)
            .await
    }

    async fn drop_inlined_table(&self, table_name: &str) -> Result<()> {
        self.executor.drop_inlined_table(table_name).await
    }

    fn snapshot_table_ref(&self) -> String {
        self.executor.snapshot_table_ref()
    }

    fn macro_table_ref(&self) -> String {
        self.executor.macro_table_ref()
    }

    fn macro_dependent_target_ref(&self, table: &str) -> String {
        self.executor.macro_dependent_target_ref(table)
    }

    fn column_mapping_table_ref(&self) -> String {
        self.executor.column_mapping_table_ref()
    }

    fn name_mapping_target_ref(&self) -> String {
        self.executor.name_mapping_target_ref()
    }

    async fn delete_where(
        &self,
        table: &str,
        target_ref: Option<&str>,
        predicate: &str,
    ) -> Result<()> {
        self.executor
            .delete_where(table, target_ref, predicate)
            .await
    }
}

#[async_trait]
impl<E> AsyncCleanupSqlExecutor for BorrowedCleanupSqlExecutor<'_, E>
where
    E: AsyncCleanupSqlExecutor + Sync + ?Sized,
{
    fn metadata_table_ref(&self, table: &str) -> String {
        self.executor.metadata_table_ref(table)
    }

    fn i64_placeholder(&self, index: usize) -> String {
        self.executor.i64_placeholder(index)
    }

    async fn cleanup_table_exists(&self, table: &str) -> Result<bool> {
        self.executor.cleanup_table_exists(table).await
    }

    async fn cleanup_path_context_available(&self) -> Result<bool> {
        self.executor.cleanup_path_context_available().await
    }

    async fn cleanup_data_path(&self) -> Result<String> {
        self.executor.cleanup_data_path().await
    }

    async fn query_i64_rows<T, F>(&self, sql: String, params: Vec<i64>, mapper: F) -> Result<Vec<T>>
    where
        T: Send + 'static,
        F: FnMut(&dyn crate::embedded_rows::EmbeddedRow) -> Result<T> + Send,
    {
        self.executor.query_i64_rows(sql, params, mapper).await
    }

    async fn insert_scheduled_cleanup_file(&self, file: CleanupMetadataFile) -> Result<()> {
        self.executor.insert_scheduled_cleanup_file(file).await
    }
}

#[async_trait]
impl<E> AsyncSnapshotPruneExecutor for E
where
    E: AsyncCleanupSqlExecutor + Sync,
{
    async fn tables_unreferenced_by_snapshots(&self) -> Result<Vec<i64>> {
        tables_unreferenced_by_snapshots(self).await
    }

    async fn data_files_unreferenced_by_snapshots(
        &self,
        deleted_table_ids: &[i64],
    ) -> Result<Vec<CleanupMetadataFile>> {
        data_files_unreferenced_by_snapshots(self, deleted_table_ids).await
    }

    async fn delete_files_unreferenced_by_snapshots(
        &self,
        deleted_table_ids: &[i64],
        deleted_data_file_ids: &[i64],
    ) -> Result<Vec<CleanupMetadataFile>> {
        delete_files_unreferenced_by_snapshots(self, deleted_table_ids, deleted_data_file_ids).await
    }

    async fn schedule_cleanup_files(&self, files: &[CleanupMetadataFile]) -> Result<()> {
        schedule_cleanup_metadata_files(self, files).await
    }
}

pub(crate) async fn cleanup_old_files<E>(
    executor: &E,
    options: CleanupOldFilesOptions,
) -> Result<Vec<CleanupFile>>
where
    E: AsyncCatalogMaintenanceExecutor + Sync,
{
    let data_path = executor.load_cleanup_data_path().await?;
    let rows = executor.load_scheduled_cleanup_files().await?;
    let files = plan_cleanup_old_files(options, &data_path, &rows)?;

    if options.dry_run || files.is_empty() {
        return Ok(files);
    }

    remove_cleanup_files_from_disk(&files)?;
    let data_file_ids = files
        .iter()
        .map(|file| file.data_file_id.0)
        .collect::<Vec<_>>();
    executor
        .delete_scheduled_cleanup_files(&data_file_ids)
        .await?;
    Ok(files)
}

pub(crate) async fn remove_files_scheduled_for_deletion<E>(
    executor: &E,
    data_file_ids: Vec<DataFileId>,
) -> Result<()>
where
    E: AsyncCatalogMaintenanceExecutor + Sync,
{
    let data_file_ids = plan_remove_files_scheduled_for_deletion(data_file_ids);
    if data_file_ids.is_empty() {
        return Ok(());
    }
    executor
        .delete_scheduled_cleanup_files(&data_file_ids)
        .await
}

pub(crate) async fn expire_snapshots<E>(
    executor: &E,
    options: ExpireSnapshotsOptions,
) -> Result<Vec<Snapshot>>
where
    E: AsyncCatalogMaintenanceExecutor + Sync,
{
    options.validate()?;
    let snapshots = executor.load_expirable_snapshots(&options).await?;
    if options.dry_run || snapshots.is_empty() {
        return Ok(snapshots);
    }
    executor.prune_expired_snapshots(&snapshots).await?;
    Ok(snapshots)
}

pub(crate) async fn load_expirable_snapshots_from_read_executor<E>(
    executor: &E,
    options: &ExpireSnapshotsOptions,
) -> Result<Vec<Snapshot>>
where
    E: AsyncReadQueryExecutor + Sync,
{
    let latest = executor
        .run_read_query(&crate::embedded_reads::latest_snapshot_id(i64::MIN))
        .await?
        .into_iter()
        .next()
        .ok_or_else(|| {
            StorageError::Backend("DuckLake latest snapshot query returned no rows".into())
        })?;
    let snapshots = executor
        .run_read_query(&crate::embedded_reads::snapshots())
        .await?;
    crate::plan_expirable_snapshots(options, latest, snapshots)
}

pub(crate) async fn prune_expired_snapshots<E>(executor: &E, snapshots: &[Snapshot]) -> Result<()>
where
    E: AsyncSnapshotPruneExecutor + Sync,
{
    let snapshot_ids = plan_snapshot_expiration_prune(snapshots, Vec::new(), &[], &[]).snapshot_ids;
    crate::metadata_prune_engine::delete_by_ids_async(
        executor,
        "ducklake_snapshot",
        "snapshot_id",
        &snapshot_ids,
    )
    .await?;
    crate::metadata_prune_engine::delete_by_ids_async(
        executor,
        "ducklake_snapshot_changes",
        "snapshot_id",
        &snapshot_ids,
    )
    .await?;

    let deleted_table_ids = executor.tables_unreferenced_by_snapshots().await?;
    let deleted_file_rows = executor
        .data_files_unreferenced_by_snapshots(&deleted_table_ids)
        .await?;
    executor.schedule_cleanup_files(&deleted_file_rows).await?;
    let deleted_data_file_ids = plan_cleanup_metadata_file_ids(&deleted_file_rows);
    crate::metadata_prune_engine::delete_file_metadata_async(executor, &deleted_data_file_ids)
        .await?;

    let deleted_delete_file_rows = executor
        .delete_files_unreferenced_by_snapshots(&deleted_table_ids, &deleted_data_file_ids)
        .await?;
    executor
        .schedule_cleanup_files(&deleted_delete_file_rows)
        .await?;
    let prune_plan = plan_snapshot_expiration_prune(
        snapshots,
        deleted_table_ids,
        &deleted_file_rows,
        &deleted_delete_file_rows,
    );
    crate::metadata_prune_engine::delete_by_ids_async(
        executor,
        "ducklake_delete_file",
        "delete_file_id",
        &prune_plan.deleted_delete_file_ids,
    )
    .await?;

    crate::metadata_prune_engine::delete_table_metadata_async(
        executor,
        &prune_plan.deleted_table_ids,
    )
    .await?;
    crate::metadata_prune_engine::delete_unreferenced_versioned_metadata_async(executor).await?;
    crate::metadata_prune_engine::delete_unreferenced_dependent_metadata_async(executor).await
}

pub(crate) async fn compaction_source_rows<E>(
    executor: &E,
    table_id: TableId,
    snapshot: i64,
    source_ids: &[i64],
) -> Result<Vec<crate::commit::MaintenanceSourceFile>>
where
    E: AsyncCleanupSqlExecutor + Sync,
{
    let data_file = executor.metadata_table_ref("ducklake_data_file");
    let table = executor.metadata_table_ref("ducklake_table");
    let ducklake_schema = executor.metadata_table_ref("ducklake_schema");
    let table_id_expr = executor.i64_placeholder(1);
    let snapshot_expr = executor.i64_placeholder(2);

    crate::catalog_async_commit_methods::load_compaction_source_rows(
        table_id,
        snapshot,
        source_ids,
        crate::catalog_async_commit_methods::CompactionSourceSqlRefs {
            data_file_ref: &data_file,
            table_ref: &table,
            schema_ref: &ducklake_schema,
            table_id_expr: &table_id_expr,
            snapshot_expr: &snapshot_expr,
        },
        || async move { executor.cleanup_path_context_available().await },
        || async move { executor.cleanup_data_path().await },
        |sql, params| async move {
            compaction_source_rows_without_table_path(executor, sql, params).await
        },
        |sql, params, data_path| async move {
            compaction_source_rows_with_table_path(executor, sql, params, data_path).await
        },
    )
    .await
}

pub(crate) async fn compaction_delete_rows<E>(
    executor: &E,
    latest_snapshot: i64,
    source_ids: &[i64],
) -> Result<Vec<CleanupMetadataFile>>
where
    E: AsyncCleanupSqlExecutor + Sync,
{
    let delete_file = executor.metadata_table_ref("ducklake_delete_file");
    let table = executor.metadata_table_ref("ducklake_table");
    let ducklake_schema = executor.metadata_table_ref("ducklake_schema");
    let snapshot_expr = executor.i64_placeholder(1);

    crate::catalog_async_commit_methods::load_compaction_delete_rows(
        latest_snapshot,
        source_ids,
        crate::catalog_async_commit_methods::CompactionDeleteSqlRefs {
            delete_file_ref: &delete_file,
            table_ref: &table,
            schema_ref: &ducklake_schema,
            snapshot_expr: &snapshot_expr,
        },
        || async move { executor.cleanup_table_exists("ducklake_delete_file").await },
        || async move { executor.cleanup_path_context_available().await },
        || async move { executor.cleanup_data_path().await },
        |sql, params| async move { cleanup_rows_without_table_path(executor, sql, params).await },
        |sql, params, data_path| async move {
            cleanup_rows_with_table_path(executor, sql, params, data_path).await
        },
    )
    .await
}

pub(crate) async fn tables_unreferenced_by_snapshots<E>(executor: &E) -> Result<Vec<i64>>
where
    E: AsyncCleanupSqlExecutor + Sync,
{
    let table = executor.metadata_table_ref("ducklake_table");
    let snapshot = executor.metadata_table_ref("ducklake_snapshot");
    crate::catalog_async_commit_methods::load_tables_unreferenced_by_snapshots(
        crate::catalog_async_commit_methods::SnapshotTableSqlRefs {
            table_ref: &table,
            snapshot_ref: &snapshot,
        },
        || async move { executor.cleanup_table_exists("ducklake_table").await },
        |sql| async move {
            executor
                .query_i64_rows(sql, Vec::new(), |row| row.get_i64(0))
                .await
        },
    )
    .await
}

pub(crate) async fn data_files_unreferenced_by_snapshots<E>(
    executor: &E,
    deleted_table_ids: &[i64],
) -> Result<Vec<CleanupMetadataFile>>
where
    E: AsyncCleanupSqlExecutor + Sync,
{
    let data_file = executor.metadata_table_ref("ducklake_data_file");
    let snapshot = executor.metadata_table_ref("ducklake_snapshot");
    let table = executor.metadata_table_ref("ducklake_table");
    let ducklake_schema = executor.metadata_table_ref("ducklake_schema");
    crate::catalog_async_commit_methods::load_data_files_unreferenced_by_snapshots(
        deleted_table_ids,
        crate::catalog_async_commit_methods::SnapshotDataFileSqlRefs {
            data_file_ref: &data_file,
            table_ref: &table,
            schema_ref: &ducklake_schema,
            snapshot_ref: &snapshot,
        },
        || async move { executor.cleanup_table_exists("ducklake_data_file").await },
        || async move { executor.cleanup_path_context_available().await },
        || async move { executor.cleanup_data_path().await },
        |sql| async move { cleanup_rows_without_table_path(executor, sql, Vec::new()).await },
        |sql, data_path| async move {
            cleanup_rows_with_table_path(executor, sql, Vec::new(), data_path).await
        },
    )
    .await
}

pub(crate) async fn delete_files_unreferenced_by_snapshots<E>(
    executor: &E,
    deleted_table_ids: &[i64],
    deleted_data_file_ids: &[i64],
) -> Result<Vec<CleanupMetadataFile>>
where
    E: AsyncCleanupSqlExecutor + Sync,
{
    let delete_file = executor.metadata_table_ref("ducklake_delete_file");
    let snapshot = executor.metadata_table_ref("ducklake_snapshot");
    let table = executor.metadata_table_ref("ducklake_table");
    let ducklake_schema = executor.metadata_table_ref("ducklake_schema");
    crate::catalog_async_commit_methods::load_delete_files_unreferenced_by_snapshots(
        deleted_table_ids,
        deleted_data_file_ids,
        crate::catalog_async_commit_methods::SnapshotDeleteFileSqlRefs {
            delete_file_ref: &delete_file,
            table_ref: &table,
            schema_ref: &ducklake_schema,
            snapshot_ref: &snapshot,
        },
        || async move { executor.cleanup_table_exists("ducklake_delete_file").await },
        || async move { executor.cleanup_path_context_available().await },
        || async move { executor.cleanup_data_path().await },
        |sql| async move { cleanup_rows_without_table_path(executor, sql, Vec::new()).await },
        |sql, data_path| async move {
            cleanup_rows_with_table_path(executor, sql, Vec::new(), data_path).await
        },
    )
    .await
}

async fn compaction_source_rows_without_table_path<E>(
    executor: &E,
    sql: String,
    params: Vec<i64>,
) -> Result<Vec<crate::commit::MaintenanceSourceFile>>
where
    E: AsyncCleanupSqlExecutor + Sync,
{
    executor
        .query_i64_rows(
            sql,
            params,
            crate::embedded_maintenance::compaction_source_row_without_table_path,
        )
        .await
}

async fn compaction_source_rows_with_table_path<E>(
    executor: &E,
    sql: String,
    params: Vec<i64>,
    data_path: String,
) -> Result<Vec<crate::commit::MaintenanceSourceFile>>
where
    E: AsyncCleanupSqlExecutor + Sync,
{
    executor
        .query_i64_rows(sql, params, |row| {
            crate::embedded_maintenance::compaction_source_row_with_table_path(row, &data_path)
        })
        .await
}

async fn cleanup_rows_without_table_path<E>(
    executor: &E,
    sql: String,
    params: Vec<i64>,
) -> Result<Vec<CleanupMetadataFile>>
where
    E: AsyncCleanupSqlExecutor + Sync,
{
    executor
        .query_i64_rows(
            sql,
            params,
            crate::embedded_maintenance::cleanup_file_row_without_table_path,
        )
        .await
}

async fn cleanup_rows_with_table_path<E>(
    executor: &E,
    sql: String,
    params: Vec<i64>,
    data_path: String,
) -> Result<Vec<CleanupMetadataFile>>
where
    E: AsyncCleanupSqlExecutor + Sync,
{
    executor
        .query_i64_rows(sql, params, |row| {
            crate::embedded_maintenance::cleanup_file_row_with_table_path(row, &data_path)
        })
        .await
}

pub(crate) async fn schedule_cleanup_metadata_files<E>(
    executor: &E,
    files: &[CleanupMetadataFile],
) -> Result<()>
where
    E: AsyncCleanupSqlExecutor + Sync,
{
    schedule_cleanup_files(
        files,
        || async move {
            executor
                .cleanup_table_exists("ducklake_files_scheduled_for_deletion")
                .await
        },
        |file| async move { executor.insert_scheduled_cleanup_file(file).await },
    )
    .await
}

pub(crate) async fn schedule_cleanup_files<TableExists, TableExistsFut, Insert, InsertFut>(
    files: &[CleanupMetadataFile],
    table_exists: TableExists,
    mut insert: Insert,
) -> Result<()>
where
    TableExists: FnOnce() -> TableExistsFut,
    TableExistsFut: Future<Output = Result<bool>>,
    Insert: FnMut(CleanupMetadataFile) -> InsertFut,
    InsertFut: Future<Output = Result<()>>,
{
    if files.is_empty() || !table_exists().await? {
        return Ok(());
    }
    for file in files {
        insert(file.clone()).await?;
    }
    Ok(())
}

macro_rules! impl_async_catalog_maintenance_executor {
    (
        impl $ty:ty {
            load_cleanup_data_path = |$path_self:ident| $load_path:expr,
            load_scheduled_cleanup_files = |$files_self:ident| $load_files:expr,
            delete_scheduled_cleanup_files = |$delete_self:ident, $data_file_ids:ident| $delete_files:expr,
            load_expirable_snapshots = |$expirable_self:ident, $options:ident| $load_expirable:expr,
            prune_expired_snapshots = |$prune_self:ident, $snapshots:ident| $prune:expr $(,)?
        }
    ) => {
        #[async_trait]
        impl $crate::catalog_async_maintenance_methods::AsyncCatalogMaintenanceExecutor for $ty {
            async fn load_cleanup_data_path(&self) -> $crate::Result<String> {
                let $path_self = self;
                $load_path.await
            }

            async fn load_scheduled_cleanup_files(
                &self,
            ) -> $crate::Result<Vec<ducklake_catalog::FileScheduledForDeletion>> {
                let $files_self = self;
                $load_files.await
            }

            async fn delete_scheduled_cleanup_files(
                &self,
                $data_file_ids: &[i64],
            ) -> $crate::Result<()> {
                let $delete_self = self;
                $delete_files.await
            }

            async fn load_expirable_snapshots(
                &self,
                $options: &$crate::ExpireSnapshotsOptions,
            ) -> $crate::Result<Vec<ducklake_catalog::Snapshot>> {
                let $expirable_self = self;
                $load_expirable.await
            }

            async fn prune_expired_snapshots(
                &self,
                $snapshots: &[ducklake_catalog::Snapshot],
            ) -> $crate::Result<()> {
                let $prune_self = self;
                $prune.await
            }
        }
    };
}

fn remove_cleanup_files_from_disk(files: &[CleanupFile]) -> Result<()> {
    let local_paths = files
        .iter()
        .map(|file| local_filesystem_path(&file.path))
        .collect::<Result<Vec<_>>>()?;
    for (file, local_path) in files.iter().zip(local_paths) {
        match fs::remove_file(&local_path) {
            Ok(()) => {}
            Err(err) if err.kind() == ErrorKind::NotFound => {}
            Err(err) => {
                return Err(StorageError::Backend(format!(
                    "failed to remove DuckLake scheduled file {}: {err}",
                    file.path
                )));
            }
        }
    }
    Ok(())
}

macro_rules! impl_async_catalog_maintenance_methods {
    () => {
        fn cleanup_old_files<'life0, 'async_trait>(
            &'life0 self,
            options: $crate::CleanupOldFilesOptions,
        ) -> std::pin::Pin<
            Box<
                dyn std::future::Future<Output = $crate::Result<Vec<$crate::CleanupFile>>>
                    + Send
                    + 'async_trait,
            >,
        >
        where
            'life0: 'async_trait,
            Self: Sync + 'async_trait,
        {
            let this = self;
            Box::pin(async move {
                $crate::catalog_async_maintenance_methods::cleanup_old_files(this, options).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 {
                $crate::catalog_async_maintenance_methods::remove_files_scheduled_for_deletion(
                    this,
                    data_file_ids,
                )
                .await
            })
        }

        fn expire_snapshots<'life0, 'async_trait>(
            &'life0 self,
            options: $crate::ExpireSnapshotsOptions,
        ) -> std::pin::Pin<
            Box<
                dyn std::future::Future<Output = $crate::Result<Vec<ducklake_catalog::Snapshot>>>
                    + Send
                    + 'async_trait,
            >,
        >
        where
            'life0: 'async_trait,
            Self: Sync + 'async_trait,
        {
            let this = self;
            Box::pin(async move {
                $crate::catalog_async_maintenance_methods::expire_snapshots(this, options).await
            })
        }
    };
}
