hamelin_lib 0.9.3

//! Membership operators: In, NotIn for arrays, maps, ranges, tuples, timestamps, intervals

use std::any::TypeId;
use std::iter::once;

use anyhow::bail;

use crate::func::def::{FunctionDef, ParameterBindingProvider, Parameters};
use crate::func::utils::must_be_same_type;
use crate::operator::Operator;
use crate::types::matcher::{
    AnyMatcher, ArrayMatcher, ExactMatcher, IntervalMatcher, MapMatcher, NumericMatcher, OrMatcher,
    RangeMatcher, UnboundedTupleMatcher,
};
use crate::types::range::Range;
use crate::types::{Type, BOOLEAN, CALENDAR_INTERVAL, INTERVAL, TIMESTAMP};

/// IN operator for array membership: `left IN right` where right is an array.
#[derive(Default, Clone)]
pub struct InArray;

impl FunctionDef for InArray {
    fn name(&self) -> &'static str {
        Operator::In.str()
    }

    fn parameters(&self) -> Parameters {
        Parameters::new()
            .with("left", AnyMatcher::default())
            .with("right", ArrayMatcher::default())
    }

    fn return_type(&self, bindings: &dyn ParameterBindingProvider) -> anyhow::Result<Type> {
        let left = bindings.get_by_index(0)?.typ();
        let inner = match bindings.get_by_index(1)?.typ() {
            Type::Array(a) => a.element_type.as_ref().clone(),
            _ => bail!("right should be an array, checked by matcher"),
        };

        must_be_same_type(vec![left, &inner].into_iter()).map(|_| BOOLEAN)
    }

    fn type_id(&self) -> TypeId {
        TypeId::of::<Self>()
    }
}

/// NOT IN operator for array membership: `left NOT IN right` where right is an array.
#[derive(Default, Clone)]
pub struct NotInArray;

impl FunctionDef for NotInArray {
    fn name(&self) -> &'static str {
        Operator::NotIn.str()
    }

    fn parameters(&self) -> Parameters {
        Parameters::new()
            .with("left", AnyMatcher::default())
            .with("right", ArrayMatcher::default())
    }

    fn return_type(&self, bindings: &dyn ParameterBindingProvider) -> anyhow::Result<Type> {
        let left = bindings.get_by_index(0)?.typ();
        let inner = match bindings.get_by_index(1)?.typ() {
            Type::Array(a) => a.element_type.as_ref().clone(),
            _ => bail!("right should be an array, checked by matcher"),
        };

        must_be_same_type(vec![left, &inner].into_iter()).map(|_| BOOLEAN)
    }

    fn type_id(&self) -> TypeId {
        TypeId::of::<Self>()
    }
}

/// IN operator for map key membership: `left IN right` where right is a map.
#[derive(Default, Clone)]
pub struct InMap;

impl FunctionDef for InMap {
    fn name(&self) -> &'static str {
        Operator::In.str()
    }

    fn parameters(&self) -> Parameters {
        Parameters::new()
            .with("left", AnyMatcher::default())
            .with("right", MapMatcher::default())
    }

    fn return_type(&self, bindings: &dyn ParameterBindingProvider) -> anyhow::Result<Type> {
        let left = bindings.get_by_index(0)?.typ();
        let key_type = match bindings.get_by_index(1)?.typ() {
            Type::Map(map) => map.key_type.as_ref().clone(),
            _ => bail!("right should be a map, checked by matcher"),
        };

        must_be_same_type(vec![left, &key_type].into_iter()).map(|_| BOOLEAN)
    }

    fn type_id(&self) -> TypeId {
        TypeId::of::<Self>()
    }
}

/// NOT IN operator for map key membership: `left NOT IN right` where right is a map.
#[derive(Default, Clone)]
pub struct NotInMap;

impl FunctionDef for NotInMap {
    fn name(&self) -> &'static str {
        Operator::NotIn.str()
    }

    fn parameters(&self) -> Parameters {
        Parameters::new()
            .with("left", AnyMatcher::default())
            .with("right", MapMatcher::default())
    }

    fn return_type(&self, bindings: &dyn ParameterBindingProvider) -> anyhow::Result<Type> {
        let left = bindings.get_by_index(0)?.typ();
        let key_type = match bindings.get_by_index(1)?.typ() {
            Type::Map(map) => map.key_type.as_ref().clone(),
            _ => bail!("right should be a map, checked by matcher"),
        };

        must_be_same_type(vec![left, &key_type].into_iter()).map(|_| BOOLEAN)
    }

    fn type_id(&self) -> TypeId {
        TypeId::of::<Self>()
    }
}

/// Range operator for numeric types: `left..right` creates a numeric range.
#[derive(Default, Clone)]
pub struct NumericRange;

impl FunctionDef for NumericRange {
    fn name(&self) -> &'static str {
        Operator::Range.str()
    }

    fn parameters(&self) -> Parameters {
        Parameters::new()
            .with("left", NumericMatcher::default())
            .with("right", NumericMatcher::default())
    }

    fn return_type(&self, bindings: &dyn ParameterBindingProvider) -> anyhow::Result<Type> {
        let left = bindings.get_by_index(0)?.typ().clone();
        let right = bindings.get_by_index(1)?.typ().clone();
        let inner = left.merge(right)?;
        Ok(Range::new(inner).into())
    }

    fn type_id(&self) -> TypeId {
        TypeId::of::<Self>()
    }
}

/// Range operator for timestamp/interval types: `left..right` creates a timestamp range.
#[derive(Default, Clone)]
pub struct TimestampRange;

impl FunctionDef for TimestampRange {
    fn name(&self) -> &'static str {
        Operator::Range.str()
    }

    fn parameters(&self) -> Parameters {
        Parameters::new()
            .with(
                "left",
                OrMatcher::default()
                    .with(ExactMatcher::of(INTERVAL))
                    .with(ExactMatcher::of(CALENDAR_INTERVAL))
                    .with(ExactMatcher::of(TIMESTAMP)),
            )
            .with(
                "right",
                OrMatcher::default()
                    .with(ExactMatcher::of(INTERVAL))
                    .with(ExactMatcher::of(CALENDAR_INTERVAL))
                    .with(ExactMatcher::of(TIMESTAMP)),
            )
    }

    fn return_type(&self, bindings: &dyn ParameterBindingProvider) -> anyhow::Result<Type> {
        let left = bindings.get_by_index(0)?.typ();
        let right = bindings.get_by_index(1)?.typ();

        if *left == TIMESTAMP || *right == TIMESTAMP {
            Ok(Range::new(TIMESTAMP).into())
        } else {
            must_be_same_type(vec![left, right].into_iter()).map(|inner| Range::new(inner).into())
        }
    }

    fn type_id(&self) -> TypeId {
        TypeId::of::<Self>()
    }
}

/// Inclusive range operator for numeric types: `left..=right` creates an inclusive numeric range.
#[derive(Default, Clone)]
pub struct NumericRangeInclusive;

impl FunctionDef for NumericRangeInclusive {
    fn name(&self) -> &'static str {
        Operator::RangeInclusive.str()
    }

    fn parameters(&self) -> Parameters {
        Parameters::new()
            .with("left", NumericMatcher::default())
            .with("right", NumericMatcher::default())
    }

    fn return_type(&self, bindings: &dyn ParameterBindingProvider) -> anyhow::Result<Type> {
        let left = bindings.get_by_index(0)?.typ().clone();
        let right = bindings.get_by_index(1)?.typ().clone();
        let inner = left.merge(right)?;
        Ok(Type::RangeInclusive(Range::new(inner)))
    }

    fn type_id(&self) -> TypeId {
        TypeId::of::<Self>()
    }
}

/// Inclusive range operator for timestamp/interval types: `left..=right` creates an inclusive timestamp range.
#[derive(Default, Clone)]
pub struct TimestampRangeInclusive;

impl FunctionDef for TimestampRangeInclusive {
    fn name(&self) -> &'static str {
        Operator::RangeInclusive.str()
    }

    fn parameters(&self) -> Parameters {
        Parameters::new()
            .with(
                "left",
                OrMatcher::default()
                    .with(ExactMatcher::of(INTERVAL))
                    .with(ExactMatcher::of(CALENDAR_INTERVAL))
                    .with(ExactMatcher::of(TIMESTAMP)),
            )
            .with(
                "right",
                OrMatcher::default()
                    .with(ExactMatcher::of(INTERVAL))
                    .with(ExactMatcher::of(CALENDAR_INTERVAL))
                    .with(ExactMatcher::of(TIMESTAMP)),
            )
    }

    fn return_type(&self, bindings: &dyn ParameterBindingProvider) -> anyhow::Result<Type> {
        let left = bindings.get_by_index(0)?.typ();
        let right = bindings.get_by_index(1)?.typ();

        if *left == TIMESTAMP || *right == TIMESTAMP {
            Ok(Type::RangeInclusive(Range::new(TIMESTAMP)))
        } else {
            must_be_same_type(vec![left, right].into_iter())
                .map(|inner| Type::RangeInclusive(Range::new(inner)))
        }
    }

    fn type_id(&self) -> TypeId {
        TypeId::of::<Self>()
    }
}

/// IN operator for range membership: `left IN right` where right is a range.
#[derive(Default, Clone)]
pub struct InRange;

impl FunctionDef for InRange {
    fn name(&self) -> &'static str {
        Operator::In.str()
    }

    fn parameters(&self) -> Parameters {
        Parameters::new()
            .with("left", AnyMatcher::default())
            .with("right", RangeMatcher::default())
    }

    fn return_type(&self, bindings: &dyn ParameterBindingProvider) -> anyhow::Result<Type> {
        let left = bindings.get_by_index(0)?.typ();
        let right = match bindings.get_by_index(1)?.typ() {
            Type::Range(range) | Type::RangeInclusive(range) => range.of.as_ref().clone(),
            _ => bail!("right should be a range, checked by matcher"),
        };
        match (left, &right) {
            (Type::Timestamp, Type::Interval) => Ok(BOOLEAN),
            (left, right) => must_be_same_type(vec![left, right].into_iter()).map(|_| BOOLEAN),
        }
    }

    fn type_id(&self) -> TypeId {
        TypeId::of::<Self>()
    }
}

/// NOT IN operator for range membership: `left NOT IN right` where right is a range.
#[derive(Default, Clone)]
pub struct NotInRange;

impl FunctionDef for NotInRange {
    fn name(&self) -> &'static str {
        Operator::NotIn.str()
    }

    fn parameters(&self) -> Parameters {
        Parameters::new()
            .with("left", AnyMatcher::default())
            .with("right", RangeMatcher::default())
    }

    fn return_type(&self, bindings: &dyn ParameterBindingProvider) -> anyhow::Result<Type> {
        let left = bindings.get_by_index(0)?.typ();
        let right = match bindings.get_by_index(1)?.typ() {
            Type::Range(range) | Type::RangeInclusive(range) => range.of.as_ref().clone(),
            _ => bail!("right should be a range, checked by matcher"),
        };
        match (left, &right) {
            (Type::Timestamp, Type::Interval) => Ok(BOOLEAN),
            (left, right) => must_be_same_type(vec![left, right].into_iter()).map(|_| BOOLEAN),
        }
    }

    fn type_id(&self) -> TypeId {
        TypeId::of::<Self>()
    }
}

/// IN operator for timestamp in interval: `timestamp IN interval`.
#[derive(Default, Clone)]
pub struct InTimestampInterval;

impl FunctionDef for InTimestampInterval {
    fn name(&self) -> &'static str {
        Operator::In.str()
    }

    fn parameters(&self) -> Parameters {
        Parameters::new()
            .with("left", ExactMatcher::of(TIMESTAMP))
            .with("right", IntervalMatcher::default())
    }

    fn return_type(&self, _bindings: &dyn ParameterBindingProvider) -> anyhow::Result<Type> {
        Ok(BOOLEAN)
    }

    fn type_id(&self) -> TypeId {
        TypeId::of::<Self>()
    }
}

/// NOT IN operator for timestamp in interval: `timestamp NOT IN interval`.
#[derive(Default, Clone)]
pub struct NotInTimestampInterval;

impl FunctionDef for NotInTimestampInterval {
    fn name(&self) -> &'static str {
        Operator::NotIn.str()
    }

    fn parameters(&self) -> Parameters {
        Parameters::new()
            .with("left", ExactMatcher::of(TIMESTAMP))
            .with("right", IntervalMatcher::default())
    }

    fn return_type(&self, _bindings: &dyn ParameterBindingProvider) -> anyhow::Result<Type> {
        Ok(BOOLEAN)
    }

    fn type_id(&self) -> TypeId {
        TypeId::of::<Self>()
    }
}

/// IN operator for timestamp in timestamp (day): `timestamp IN timestamp`.
#[derive(Default, Clone)]
pub struct InTimestampTimestamp;

impl FunctionDef for InTimestampTimestamp {
    fn name(&self) -> &'static str {
        Operator::In.str()
    }

    fn parameters(&self) -> Parameters {
        Parameters::new()
            .with("left", ExactMatcher::of(TIMESTAMP))
            .with("right", ExactMatcher::of(TIMESTAMP))
    }

    fn return_type(&self, _bindings: &dyn ParameterBindingProvider) -> anyhow::Result<Type> {
        Ok(BOOLEAN)
    }

    fn type_id(&self) -> TypeId {
        TypeId::of::<Self>()
    }
}

/// NOT IN operator for timestamp in timestamp (day): `timestamp NOT IN timestamp`.
#[derive(Default, Clone)]
pub struct NotInTimestampTimestamp;

impl FunctionDef for NotInTimestampTimestamp {
    fn name(&self) -> &'static str {
        Operator::NotIn.str()
    }

    fn parameters(&self) -> Parameters {
        Parameters::new()
            .with("left", ExactMatcher::of(TIMESTAMP))
            .with("right", ExactMatcher::of(TIMESTAMP))
    }

    fn return_type(&self, _bindings: &dyn ParameterBindingProvider) -> anyhow::Result<Type> {
        Ok(BOOLEAN)
    }

    fn type_id(&self) -> TypeId {
        TypeId::of::<Self>()
    }
}

/// IN operator for tuple membership: `left IN (a, b, c)`.
#[derive(Default, Clone)]
pub struct InTuple;

impl FunctionDef for InTuple {
    fn name(&self) -> &'static str {
        Operator::In.str()
    }

    fn parameters(&self) -> Parameters {
        Parameters::new()
            .with("left", AnyMatcher::default())
            .with("right", UnboundedTupleMatcher::default())
    }

    fn return_type(&self, bindings: &dyn ParameterBindingProvider) -> anyhow::Result<Type> {
        let left = bindings.get_by_index(0)?.typ();
        match bindings.get_by_index(1)?.typ() {
            Type::Tuple(tuple) => {
                if tuple.elements.is_empty() {
                    bail!("right should be a tuple with at least one element")
                }
                // Validate left and ALL tuple elements are type-compatible
                must_be_same_type(once(left).chain(tuple.elements.iter().map(|t| t.as_ref())))
                    .map(|_| BOOLEAN)
            }
            _ => bail!("right should be a tuple, checked by matcher"),
        }
    }

    fn type_id(&self) -> TypeId {
        TypeId::of::<Self>()
    }
}

/// NOT IN operator for tuple membership: `left NOT IN (a, b, c)`.
#[derive(Default, Clone)]
pub struct NotInTuple;

impl FunctionDef for NotInTuple {
    fn name(&self) -> &'static str {
        Operator::NotIn.str()
    }

    fn parameters(&self) -> Parameters {
        Parameters::new()
            .with("left", AnyMatcher::default())
            .with("right", UnboundedTupleMatcher::default())
    }

    fn return_type(&self, bindings: &dyn ParameterBindingProvider) -> anyhow::Result<Type> {
        let left = bindings.get_by_index(0)?.typ();
        match bindings.get_by_index(1)?.typ() {
            Type::Tuple(tuple) => {
                if tuple.elements.is_empty() {
                    bail!("right should be a tuple with at least one element")
                }
                // Validate left and ALL tuple elements are type-compatible
                must_be_same_type(once(left).chain(tuple.elements.iter().map(|t| t.as_ref())))
                    .map(|_| BOOLEAN)
            }
            _ => bail!("right should be a tuple, checked by matcher"),
        }
    }

    fn type_id(&self) -> TypeId {
        TypeId::of::<Self>()
    }
}