hamelin_sql 0.4.1

use anyhow::bail;

use hamelin_lib::func::def::ParameterBinding;
use hamelin_lib::sql::expression::apply::{
    BinaryOperatorApply, FunctionCallApply, Lambda, UnaryOperatorApply,
};
use hamelin_lib::sql::expression::literal::{BooleanLiteral, IntegerLiteral, StringLiteral};
use hamelin_lib::sql::expression::operator::Operator;
use hamelin_lib::sql::expression::IndexLookup;
use hamelin_lib::sql::expression::{Leaf, SQLExpression};
use hamelin_lib::translation::ExpressionTranslation;
use hamelin_lib::types::TIMESTAMP;

use crate::range_builder::RangeBuilder;

/// Default translation for a function call. It just calls a SQL function with the same name.
pub fn direct_function_translation(
    func: &str,
    bindings: ParameterBinding<ExpressionTranslation>,
) -> anyhow::Result<SQLExpression> {
    let all_args = bindings.into_iter().map(|t| t.sql).collect();
    let fca: SQLExpression = FunctionCallApply::with_positional(func, all_args).into();

    Ok(fca)
}

pub fn function_translation_respecting_ignore_nulls(
    func: &str,
    mut bindings: ParameterBinding<ExpressionTranslation>,
) -> anyhow::Result<SQLExpression> {
    let ignore_nulls = bindings.take_by_name("ignore_nulls")?;
    let all_args = bindings.into_iter().map(|t| t.sql).collect();
    let mut fca = FunctionCallApply::with_positional(func, all_args);

    match ignore_nulls.sql {
        SQLExpression::Leaf(Leaf::BooleanLiteral(b)) => {
            if b.value {
                fca = fca.with_ignore_nulls();
            }
        }
        _ => bail!("ignore_nulls must be boolean literal"),
    }

    Ok(fca.into())
}

/// Default translation for a unary operator.
pub fn direct_unary_operator_translation(
    func: &str,
    mut bindings: ParameterBinding<ExpressionTranslation>,
) -> anyhow::Result<SQLExpression> {
    let sql_op: Operator = hamelin_lib::operator::Operator::of(func)?.try_into()?;
    Ok(UnaryOperatorApply::new(sql_op, bindings.take()?.sql).into())
}

/// Default translation for a binary operator.
pub fn direct_binary_operator_translation(
    func: &str,
    mut bindings: ParameterBinding<ExpressionTranslation>,
) -> anyhow::Result<SQLExpression> {
    let left = bindings.take()?.sql;
    let right = bindings.take()?.sql;
    let sql_op: Operator = hamelin_lib::operator::Operator::of(func)?.try_into()?;
    Ok(BinaryOperatorApply::new(sql_op, left, right).into())
}

/// Default translation for a binary operator that calls SQL's concatenation operator.
pub fn to_sql_concat(
    _: &str,
    mut bindings: ParameterBinding<ExpressionTranslation>,
) -> anyhow::Result<SQLExpression> {
    let p0 = bindings.take()?;
    let p1 = bindings.take()?;
    Ok(BinaryOperatorApply::new(Operator::Concat, p0.sql, p1.sql).into())
}

/// Converts an interval expression to a timestamp range from now() + interval to now().
pub fn interval_to_range(interval: SQLExpression) -> RangeBuilder {
    RangeBuilder::default()
        .with_begin(
            interval_to_timestamp(interval),
            TIMESTAMP.to_sql().unwrap().into(),
        )
        .with_end(
            FunctionCallApply::with_no_arguments("now").into(),
            TIMESTAMP.to_sql().unwrap().into(),
        )
}

/// Converts a timestamp expression to a timestamp range from timestamp to now().
pub fn timestamp_to_range(timestamp: SQLExpression) -> RangeBuilder {
    RangeBuilder::default()
        .with_begin(timestamp, TIMESTAMP.to_sql().unwrap().into())
        .with_end(
            FunctionCallApply::with_no_arguments("now").into(),
            TIMESTAMP.to_sql().unwrap().into(),
        )
}

/// Converts an interval expression to a timestamp by calculating now() + interval.
pub fn interval_to_timestamp(interval: SQLExpression) -> SQLExpression {
    BinaryOperatorApply::new(
        Operator::Plus,
        FunctionCallApply::with_no_arguments("now").into(),
        interval,
    )
    .into()
}

/// Converts a timestamp range from an interval range by converting each interval to a timestamp by
/// adding it to now().
pub fn interval_range_to_timestamp_range(range_expr: SQLExpression) -> RangeBuilder {
    RangeBuilder::from_literal(range_expr.clone())
        .map(|rb| {
            let mut rb = rb;
            if let SQLExpression::Leaf(Leaf::NullLiteral(_)) = &(rb.begin) {
            } else {
                rb = rb.clone().with_begin(
                    interval_to_timestamp(rb.begin.clone()),
                    TIMESTAMP.to_sql().unwrap(),
                );
            }

            if let SQLExpression::Leaf(Leaf::NullLiteral(_)) = rb.end {
            } else {
                rb = rb.clone().with_end(
                    interval_to_timestamp(rb.end.clone()),
                    TIMESTAMP.to_sql().unwrap(),
                );
            }

            rb
        })
        .unwrap_or(
            RangeBuilder::default()
                .with_begin(
                    interval_to_timestamp(
                        range_expr.clone().index(IntegerLiteral::new("1").into()),
                    ),
                    TIMESTAMP.to_sql().unwrap().into(),
                )
                .with_end(
                    interval_to_timestamp(
                        range_expr.clone().index(IntegerLiteral::new("2").into()),
                    ),
                    TIMESTAMP.to_sql().unwrap().into(),
                ),
        )
}

/// A helper that implements WITHIN (range) and also (exp) IN (range) where range is a literal
pub fn within_range(left: SQLExpression, range: RangeBuilder) -> SQLExpression {
    BinaryOperatorApply::new(
        Operator::And,
        if let SQLExpression::Leaf(Leaf::NullLiteral(_)) = range.begin {
            BooleanLiteral::new(true).into()
        } else {
            BinaryOperatorApply::new(Operator::Gte, left.clone(), range.begin).into()
        },
        if let SQLExpression::Leaf(Leaf::NullLiteral(_)) = range.end {
            BooleanLiteral::new(true).into()
        } else {
            BinaryOperatorApply::new(Operator::Lte, left, range.end).into()
        },
    )
    .into()
}

/// A helper that implements WITHIN (range) and also (exp) IN (range) where range is an expression
pub fn within_range_expr(left: SQLExpression, right: SQLExpression) -> SQLExpression {
    BinaryOperatorApply::new(
        Operator::And,
        FunctionCallApply::with_two(
            "coalesce",
            BinaryOperatorApply::new(
                Operator::Gte,
                left.clone(),
                IndexLookup::new(right.clone(), IntegerLiteral::new("1").into()).into(),
            )
            .into(),
            BooleanLiteral::new(true).into(),
        )
        .into(),
        FunctionCallApply::with_two(
            "coalesce",
            BinaryOperatorApply::new(
                Operator::Lt,
                left,
                IndexLookup::new(right.clone(), IntegerLiteral::new("2").into()).into(),
            )
            .into(),
            BooleanLiteral::new(true).into(),
        )
        .into(),
    )
    .into()
}

/// Converts nested transform calls to a different function.
///
/// For example, if the original expression is `transform(a, x -> transform(b, y -> f(x, y)))`
/// and `toFunction` is `g`, this function will return `g(a, x -> g(b, y -> f(x, y)))`.
///
/// # Arguments
/// - `original` - the original expression
/// - `to_function` - the function to convert the nested transform calls to
///
/// # Returns
/// The converted expression
pub fn convert_nested_transforms(original: SQLExpression, to_function: &str) -> SQLExpression {
    if let SQLExpression::FunctionCallApply(fca) = &original {
        if fca.function_name == "transform" {
            if let Some(SQLExpression::Lambda(lmbda)) = fca.arguments.get(1) {
                return FunctionCallApply::with_two(
                    to_function,
                    fca.arguments.first().unwrap().clone(),
                    Lambda::new(
                        lmbda.arguments.clone(),
                        convert_nested_transforms(lmbda.body.as_ref().clone(), to_function),
                    )
                    .into(),
                )
                .into();
            }
        }
    }

    original
}

pub fn wrap_timestamp(expr: SQLExpression) -> SQLExpression {
    FunctionCallApply::with_one(
        "try",
        FunctionCallApply::with_one(
            "from_iso8601_timestamp",
            FunctionCallApply::with_three(
                "replace",
                expr,
                StringLiteral::new(" ").into(),
                StringLiteral::new("T").into(),
            )
            .into(),
        )
        .into(),
    )
    .into()
}

/// Get the length (cardinality) of an array expression
pub fn array_length(array_expr: SQLExpression) -> SQLExpression {
    FunctionCallApply::with_one("cardinality", array_expr).into()
}

/// Get the length of a string expression
pub fn string_length(string_expr: SQLExpression) -> SQLExpression {
    FunctionCallApply::with_one("length", string_expr).into()
}

/// Convert a negative index to positive based on the provided length expression
/// This maintains Hamelin's zero-based indexing semantics
fn negative_index_to_positive(
    index_expr: SQLExpression,
    length_expr: SQLExpression,
) -> SQLExpression {
    FunctionCallApply::with_three(
        "if",
        BinaryOperatorApply::new(
            Operator::Lt,
            index_expr.clone(),
            IntegerLiteral::new("0").into(),
        )
        .into(),
        BinaryOperatorApply::new(Operator::Plus, length_expr, index_expr.clone()).into(),
        index_expr,
    )
    .into()
}

pub fn array_negative_index_to_positive(
    index_expr: SQLExpression,
    array_expr: SQLExpression,
) -> SQLExpression {
    negative_index_to_positive(index_expr.clone(), array_length(array_expr))
}

pub fn string_negative_index_to_positive(
    index_expr: SQLExpression,
    string_expr: SQLExpression,
) -> SQLExpression {
    negative_index_to_positive(index_expr.clone(), string_length(string_expr))
}

/// Convert a zero-based Hamelin index to one-based SQL index, handling negative indices
/// For array slicing operations where negative indices need special handling
pub fn hamelin_array_index_to_sql_with_negatives(
    index_expr: SQLExpression,
    array_expr: SQLExpression,
) -> SQLExpression {
    let positive_index = array_negative_index_to_positive(index_expr.clone(), array_expr);

    // Convert to SQL space: always add 1 to convert from zero-based to one-based indexing
    // This applies to both originally positive and originally negative indices after conversion
    BinaryOperatorApply::new(
        Operator::Plus,
        positive_index,
        IntegerLiteral::new("1").into(),
    )
    .into()
}

/// Convert a zero-based Hamelin string index to one-based SQL index, handling negative indices
/// For string operations like substr where negative indices need special handling
pub fn hamelin_string_index_to_sql_with_negatives(
    index_expr: SQLExpression,
    string_expr: SQLExpression,
) -> SQLExpression {
    let positive_index = string_negative_index_to_positive(index_expr.clone(), string_expr);

    // Convert to SQL space: always add 1 to convert from zero-based to one-based indexing
    // This applies to both originally positive and originally negative indices after conversion
    BinaryOperatorApply::new(
        Operator::Plus,
        positive_index,
        IntegerLiteral::new("1").into(),
    )
    .into()
}