#![allow(dead_code)]
// Metadata pruning has shared sync and async surfaces. Narrow backend builds
// intentionally compile only the subset they execute.

#[cfg(any(feature = "postgres", feature = "quack"))]
use async_trait::async_trait;

use crate::Result;
use crate::sql_utils::join_i64s;

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) struct DeleteTarget {
    pub(crate) table: &'static str,
    pub(crate) column: &'static str,
}

pub(crate) const FILE_METADATA_DELETE_TARGETS: &[DeleteTarget] = &[
    DeleteTarget {
        table: "ducklake_data_file",
        column: "data_file_id",
    },
    DeleteTarget {
        table: "ducklake_file_column_stats",
        column: "data_file_id",
    },
    DeleteTarget {
        table: "ducklake_file_variant_stats",
        column: "data_file_id",
    },
    DeleteTarget {
        table: "ducklake_file_partition_value",
        column: "data_file_id",
    },
];

pub(crate) const TABLE_METADATA_DELETE_TARGETS: &[DeleteTarget] = &[
    DeleteTarget {
        table: "ducklake_table",
        column: "table_id",
    },
    DeleteTarget {
        table: "ducklake_table_stats",
        column: "table_id",
    },
    DeleteTarget {
        table: "ducklake_table_column_stats",
        column: "table_id",
    },
    DeleteTarget {
        table: "ducklake_partition_info",
        column: "table_id",
    },
    DeleteTarget {
        table: "ducklake_partition_column",
        column: "table_id",
    },
    DeleteTarget {
        table: "ducklake_column",
        column: "table_id",
    },
    DeleteTarget {
        table: "ducklake_column_tag",
        column: "table_id",
    },
    DeleteTarget {
        table: "ducklake_sort_info",
        column: "table_id",
    },
    DeleteTarget {
        table: "ducklake_sort_expression",
        column: "table_id",
    },
    DeleteTarget {
        table: "ducklake_schema_versions",
        column: "table_id",
    },
    DeleteTarget {
        table: "ducklake_inlined_data_tables",
        column: "table_id",
    },
    DeleteTarget {
        table: "ducklake_column_mapping",
        column: "table_id",
    },
];

pub(crate) const VERSIONED_METADATA_TABLES: &[&str] = &[
    "ducklake_schema",
    "ducklake_view",
    "ducklake_tag",
    "ducklake_macro",
];

pub(crate) const MACRO_DEPENDENT_METADATA_TABLES: &[&str] =
    &["ducklake_macro_impl", "ducklake_macro_parameters"];

pub(crate) fn unreferenced_versioned_metadata_predicate(snapshot_table_ref: &str) -> String {
    format!(
        "end_snapshot IS NOT NULL AND NOT EXISTS ( \
         SELECT 1 FROM {snapshot_table_ref} s \
         WHERE s.snapshot_id >= begin_snapshot AND s.snapshot_id < end_snapshot \
         )"
    )
}

pub(crate) fn unreferenced_macro_metadata_predicate(
    macro_table_ref: &str,
    target_table_ref: &str,
) -> String {
    format!(
        "NOT EXISTS ( \
         SELECT 1 FROM {macro_table_ref} m WHERE m.macro_id = {target_table_ref}.macro_id \
         )"
    )
}

pub(crate) fn unreferenced_name_mapping_predicate(
    column_mapping_table_ref: &str,
    name_mapping_table_ref: &str,
) -> String {
    format!(
        "NOT EXISTS ( \
         SELECT 1 FROM {column_mapping_table_ref} m \
         WHERE m.mapping_id = {name_mapping_table_ref}.mapping_id \
         )"
    )
}

#[cfg(any(feature = "postgres", feature = "quack", test))]
pub(crate) fn inlined_table_names_for_table_ids_sql(table_ref: &str, table_ids: &[i64]) -> String {
    format!(
        "SELECT table_name FROM {table_ref} WHERE table_id IN ({})",
        join_i64s(table_ids)
    )
}

#[cfg(any(feature = "postgres", feature = "quack", test))]
pub(crate) fn drop_table_if_exists_sql(table_ref: &str) -> String {
    format!("DROP TABLE IF EXISTS {table_ref}")
}

#[cfg(any(feature = "postgres", feature = "quack", test))]
pub(crate) fn delete_where_sql(
    table_ref: &str,
    target_ref: Option<&str>,
    predicate: &str,
) -> String {
    let target_ref = target_ref
        .map(|target_ref| format!(" {target_ref}"))
        .unwrap_or_default();
    format!("DELETE FROM {table_ref}{target_ref} WHERE {predicate}")
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
pub(crate) trait SyncMetadataPruneExecutor {
    fn table_exists(&self, table: &str) -> Result<bool>;

    fn inlined_table_names_for_table_ids(&self, table_ids: &[i64]) -> Result<Vec<String>>;

    fn drop_inlined_table(&self, table_name: &str) -> Result<()>;

    fn snapshot_table_ref(&self) -> String;

    fn macro_table_ref(&self) -> String;

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

    fn column_mapping_table_ref(&self) -> String {
        "ducklake_column_mapping".to_string()
    }

    fn name_mapping_target_ref(&self) -> String {
        "ducklake_name_mapping".to_string()
    }

    fn delete_where(&self, table: &str, target_ref: Option<&str>, predicate: &str) -> Result<()>;
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
pub(crate) fn delete_file_metadata_sync(
    executor: &impl SyncMetadataPruneExecutor,
    data_file_ids: &[i64],
) -> Result<()> {
    for target in FILE_METADATA_DELETE_TARGETS {
        delete_by_ids_sync(executor, target.table, target.column, data_file_ids)?;
    }
    Ok(())
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
pub(crate) fn delete_table_metadata_sync(
    executor: &impl SyncMetadataPruneExecutor,
    table_ids: &[i64],
) -> Result<()> {
    if table_ids.is_empty() {
        return Ok(());
    }
    if executor.table_exists("ducklake_inlined_data_tables")? {
        for table_name in executor.inlined_table_names_for_table_ids(table_ids)? {
            executor.drop_inlined_table(&table_name)?;
        }
    }
    for target in TABLE_METADATA_DELETE_TARGETS {
        delete_by_ids_sync(executor, target.table, target.column, table_ids)?;
    }
    Ok(())
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
pub(crate) fn delete_unreferenced_versioned_metadata_sync(
    executor: &impl SyncMetadataPruneExecutor,
) -> Result<()> {
    let snapshot = executor.snapshot_table_ref();
    let predicate = unreferenced_versioned_metadata_predicate(&snapshot);
    for table in VERSIONED_METADATA_TABLES {
        executor.delete_where(table, None, &predicate)?;
    }
    Ok(())
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
pub(crate) fn delete_unreferenced_dependent_metadata_sync(
    executor: &impl SyncMetadataPruneExecutor,
) -> Result<()> {
    if executor.table_exists("ducklake_macro")? {
        let macro_table = executor.macro_table_ref();
        for table in MACRO_DEPENDENT_METADATA_TABLES {
            if !executor.table_exists(table)? {
                continue;
            }
            let target_ref = executor.macro_dependent_target_ref(table);
            let predicate = unreferenced_macro_metadata_predicate(&macro_table, &target_ref);
            executor.delete_where(table, Some(target_ref.as_str()), &predicate)?;
        }
    }
    if executor.table_exists("ducklake_column_mapping")?
        && executor.table_exists("ducklake_name_mapping")?
    {
        let column_mapping = executor.column_mapping_table_ref();
        let name_mapping = executor.name_mapping_target_ref();
        let predicate = unreferenced_name_mapping_predicate(&column_mapping, &name_mapping);
        executor.delete_where(
            "ducklake_name_mapping",
            Some(name_mapping.as_str()),
            &predicate,
        )?;
    }
    Ok(())
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
pub(crate) fn delete_by_ids_sync(
    executor: &impl SyncMetadataPruneExecutor,
    table: &str,
    column: &str,
    ids: &[i64],
) -> Result<()> {
    if ids.is_empty() {
        return Ok(());
    }
    executor.delete_where(table, None, &format!("{column} IN ({})", join_i64s(ids)))
}

#[cfg(any(feature = "postgres", feature = "quack"))]
#[async_trait]
pub(crate) trait AsyncMetadataPruneExecutor {
    async fn table_exists(&self, table: &str) -> Result<bool>;

    async fn inlined_table_names_for_table_ids(&self, table_ids: &[i64]) -> Result<Vec<String>>;

    async fn drop_inlined_table(&self, table_name: &str) -> Result<()>;

    fn snapshot_table_ref(&self) -> String;

    fn macro_table_ref(&self) -> String;

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

    fn column_mapping_table_ref(&self) -> String {
        "ducklake_column_mapping".to_string()
    }

    fn name_mapping_target_ref(&self) -> String {
        "ducklake_name_mapping".to_string()
    }

    async fn delete_where(
        &self,
        table: &str,
        target_ref: Option<&str>,
        predicate: &str,
    ) -> Result<()>;
}

#[cfg(any(feature = "postgres", feature = "quack"))]
macro_rules! impl_async_metadata_prune_executor {
    (
        impl $ty:ty {
            table_exists = |$exists_self:ident, $exists_table:ident| $exists:expr,
            inlined_table_names = |$names_self:ident, $table_ids:ident| $names:expr,
            drop_inlined_table = |$drop_self:ident, $drop_table_name:ident| $drop:expr,
            snapshot_table_ref = |$snapshot_self:ident| $snapshot_ref:expr,
            macro_table_ref = |$macro_self:ident| $macro_ref:expr,
            macro_dependent_target_ref = |$macro_dep_self:ident, $macro_dep_table:ident| $macro_dep_ref:expr,
            column_mapping_table_ref = |$mapping_self:ident| $mapping_ref:expr,
            name_mapping_target_ref = |$name_mapping_self:ident| $name_mapping_ref:expr,
            delete_where = |$delete_self:ident, $delete_table:ident, $target_ref:ident, $predicate:ident| $delete:expr $(,)?
        }
    ) => {
        #[async_trait]
        impl $crate::metadata_prune_engine::AsyncMetadataPruneExecutor for $ty {
            async fn table_exists(&self, $exists_table: &str) -> $crate::Result<bool> {
                let $exists_self = self;
                $exists.await
            }

            async fn inlined_table_names_for_table_ids(
                &self,
                $table_ids: &[i64],
            ) -> $crate::Result<Vec<String>> {
                let $names_self = self;
                $names.await
            }

            async fn drop_inlined_table(&self, $drop_table_name: &str) -> $crate::Result<()> {
                let $drop_self = self;
                $drop.await
            }

            fn snapshot_table_ref(&self) -> String {
                let $snapshot_self = self;
                $snapshot_ref
            }

            fn macro_table_ref(&self) -> String {
                let $macro_self = self;
                $macro_ref
            }

            fn macro_dependent_target_ref(&self, $macro_dep_table: &str) -> String {
                let $macro_dep_self = self;
                $macro_dep_ref
            }

            fn column_mapping_table_ref(&self) -> String {
                let $mapping_self = self;
                $mapping_ref
            }

            fn name_mapping_target_ref(&self) -> String {
                let $name_mapping_self = self;
                $name_mapping_ref
            }

            async fn delete_where(
                &self,
                $delete_table: &str,
                $target_ref: Option<&str>,
                $predicate: &str,
            ) -> $crate::Result<()> {
                let $delete_self = self;
                $delete.await
            }
        }
    };
}

#[cfg(any(feature = "postgres", feature = "quack"))]
pub(crate) async fn delete_file_metadata_async(
    executor: &impl AsyncMetadataPruneExecutor,
    data_file_ids: &[i64],
) -> Result<()> {
    for target in FILE_METADATA_DELETE_TARGETS {
        delete_by_ids_async(executor, target.table, target.column, data_file_ids).await?;
    }
    Ok(())
}

#[cfg(any(feature = "postgres", feature = "quack"))]
pub(crate) async fn delete_table_metadata_async(
    executor: &impl AsyncMetadataPruneExecutor,
    table_ids: &[i64],
) -> Result<()> {
    if table_ids.is_empty() {
        return Ok(());
    }
    if executor
        .table_exists("ducklake_inlined_data_tables")
        .await?
    {
        for table_name in executor
            .inlined_table_names_for_table_ids(table_ids)
            .await?
        {
            executor.drop_inlined_table(&table_name).await?;
        }
    }
    for target in TABLE_METADATA_DELETE_TARGETS {
        delete_by_ids_async(executor, target.table, target.column, table_ids).await?;
    }
    Ok(())
}

#[cfg(any(feature = "postgres", feature = "quack"))]
pub(crate) async fn delete_unreferenced_versioned_metadata_async(
    executor: &impl AsyncMetadataPruneExecutor,
) -> Result<()> {
    let snapshot = executor.snapshot_table_ref();
    let predicate = unreferenced_versioned_metadata_predicate(&snapshot);
    for table in VERSIONED_METADATA_TABLES {
        executor.delete_where(table, None, &predicate).await?;
    }
    Ok(())
}

#[cfg(any(feature = "postgres", feature = "quack"))]
pub(crate) async fn delete_unreferenced_dependent_metadata_async(
    executor: &impl AsyncMetadataPruneExecutor,
) -> Result<()> {
    if executor.table_exists("ducklake_macro").await? {
        let macro_table = executor.macro_table_ref();
        for table in MACRO_DEPENDENT_METADATA_TABLES {
            if !executor.table_exists(table).await? {
                continue;
            }
            let target_ref = executor.macro_dependent_target_ref(table);
            let predicate = unreferenced_macro_metadata_predicate(&macro_table, &target_ref);
            executor
                .delete_where(table, Some(target_ref.as_str()), &predicate)
                .await?;
        }
    }
    if executor.table_exists("ducklake_column_mapping").await?
        && executor.table_exists("ducklake_name_mapping").await?
    {
        let column_mapping = executor.column_mapping_table_ref();
        let name_mapping = executor.name_mapping_target_ref();
        let predicate = unreferenced_name_mapping_predicate(&column_mapping, &name_mapping);
        executor
            .delete_where(
                "ducklake_name_mapping",
                Some(name_mapping.as_str()),
                &predicate,
            )
            .await?;
    }
    Ok(())
}

#[cfg(any(feature = "postgres", feature = "quack"))]
pub(crate) async fn delete_by_ids_async(
    executor: &impl AsyncMetadataPruneExecutor,
    table: &str,
    column: &str,
    ids: &[i64],
) -> Result<()> {
    if ids.is_empty() {
        return Ok(());
    }
    executor
        .delete_where(table, None, &format!("{column} IN ({})", join_i64s(ids)))
        .await
}
