//! Top-level DataFusion translation API — async functions for translating typed ASTs
//! to DataFusion LogicalPlans.
//!
//! Callers supply a [`DatasetTableCatalog`] mapping resolved [`DatasetIdentifier`] keys
//! to table providers (built by Phantom resolvers after parsing query dependencies).

use std::sync::Arc;

use hamelin_lib::err::{Context, TranslationError, TranslationErrors};
use hamelin_lib::parse_and_typecheck_with_options;
use hamelin_lib::provider::EnvironmentProvider;
use hamelin_lib::tree::ast::expression::Expression;
use hamelin_lib::tree::ast::identifier::SimpleIdentifier;
use hamelin_lib::tree::ast::query::Query;
use hamelin_lib::tree::options::{TranslateOptions, TypeCheckOptions};
use hamelin_lib::tree::typed_ast::query::TypedStatement;
use hamelin_lib::type_check_with_options;
use hamelin_translation::{normalize_with, NormalizationOptions};

use crate::dataset_table_catalog::DatasetTableCatalog;
use crate::statement::{translate_statement, TranslatedStatement};

pub async fn translate_with_options(
    typed: impl Into<Arc<TypedStatement>>,
    catalog: &DatasetTableCatalog,
    opts: TranslateOptions,
) -> Result<TranslatedStatement, TranslationErrors> {
    let typed = typed.into();
    let mut norm_opts: NormalizationOptions = normalize_with()
        .with_lower_filter()
        .with_lower_transform()
        .with_registry(opts.registry)
        .with_provider(opts.provider);

    if let Some(ts_field) = opts.timestamp_field {
        norm_opts = norm_opts.with_timestamp_field(ts_field);
    }
    if let Some(msg_field) = opts.message_field {
        norm_opts = norm_opts.with_message_field(msg_field);
    }
    if let Some(space) = opts.default_space {
        norm_opts = norm_opts.with_default_space(space);
    }

    let ir = norm_opts.lower(typed)?;
    translate_statement(&ir, catalog)
        .await
        .map_err(|e| e.as_ref().clone().single())
}

pub async fn type_check_and_translate(
    ast: impl Into<Arc<Query>>,
    catalog: &DatasetTableCatalog,
    opts: TypeCheckOptions,
) -> Result<TranslatedStatement, TranslationErrors> {
    type_check_and_translate_with_options(ast, catalog, opts).await
}

pub async fn type_check_and_translate_with_options(
    ast: impl Into<Arc<Query>>,
    catalog: &DatasetTableCatalog,
    opts: TypeCheckOptions,
) -> Result<TranslatedStatement, TranslationErrors> {
    let translate_opts = TranslateOptions::builder()
        .registry(opts.registry.clone())
        .provider(schema_provider_from_catalog(catalog)?)
        .maybe_timestamp_field(opts.timestamp_field.clone())
        .maybe_message_field(opts.message_field.clone())
        .maybe_default_space(opts.default_space.clone())
        .build();

    let typed = type_check_with_options(ast, type_check_options_with_catalog(catalog, opts)?)
        .into_result()?;
    translate_with_options(Arc::new(typed), catalog, translate_opts).await
}

pub async fn parse_and_translate(
    input: impl Into<String>,
    catalog: &DatasetTableCatalog,
    default_space: Option<&SimpleIdentifier>,
) -> Result<TranslatedStatement, TranslationErrors> {
    parse_and_translate_with_options(
        input,
        catalog,
        TypeCheckOptions::builder()
            .maybe_default_space(default_space.cloned())
            .build(),
    )
    .await
}

pub async fn parse_and_translate_with_time_range(
    input: impl Into<String>,
    catalog: &DatasetTableCatalog,
    time_range: Option<Arc<Expression>>,
    default_space: Option<&SimpleIdentifier>,
) -> Result<TranslatedStatement, TranslationErrors> {
    parse_and_translate_with_options(
        input,
        catalog,
        TypeCheckOptions::builder()
            .maybe_time_range(time_range)
            .maybe_default_space(default_space.cloned())
            .build(),
    )
    .await
}

pub async fn parse_and_translate_with_options(
    input: impl Into<String>,
    catalog: &DatasetTableCatalog,
    opts: TypeCheckOptions,
) -> Result<TranslatedStatement, TranslationErrors> {
    let type_check_opts = type_check_options_with_catalog(catalog, opts)?;
    let translate_opts = TranslateOptions::builder()
        .registry(type_check_opts.registry.clone())
        .provider(schema_provider_from_catalog(catalog)?)
        .maybe_timestamp_field(type_check_opts.timestamp_field.clone())
        .maybe_message_field(type_check_opts.message_field.clone())
        .maybe_default_space(type_check_opts.default_space.clone())
        .build();

    let typed = parse_and_typecheck_with_options(input, type_check_opts).into_result()?;
    translate_with_options(Arc::new(typed), catalog, translate_opts).await
}

fn schema_provider_from_catalog(
    catalog: &DatasetTableCatalog,
) -> Result<Arc<dyn EnvironmentProvider>, TranslationErrors> {
    catalog
        .to_catalog_provider()
        .map(|provider| Arc::new(provider) as Arc<dyn EnvironmentProvider>)
        .map_err(|msg| {
            TranslationError::new(Context::new(0..=0, &msg))
                .single()
                .into()
        })
}

fn type_check_options_with_catalog(
    catalog: &DatasetTableCatalog,
    opts: TypeCheckOptions,
) -> Result<TypeCheckOptions, TranslationErrors> {
    Ok(TypeCheckOptions::builder()
        .registry(opts.registry)
        .provider(schema_provider_from_catalog(catalog)?)
        .maybe_interner(opts.interner)
        .maybe_time_range(opts.time_range)
        .maybe_timestamp_field(opts.timestamp_field)
        .maybe_message_field(opts.message_field)
        .maybe_template_parameters(opts.template_parameters)
        .maybe_default_space(opts.default_space)
        .build())
}
