//! `Min` — two-input f64 pairwise minimum (input 0 vs input 1), the companion
//! to `Max`, e.g. the negative-part split of a price change:
//! `loss = Abs(Min(delta, 0))`. Distinct from `RollingMin`, which reduces a
//! *window* of one input; `Min` combines two co-present inputs, mirroring
//! `Add`/`Sub`.

use aura_core::{Cell, Ctx, FieldSpec, Firing, Node, NodeSchema, PortSpec, PrimitiveBuilder, ScalarKind};

/// Two-input f64 pairwise minimum: `input 0 .min(input 1)`. Emits `None` until
/// both inputs have a value.
pub struct Min {
    out: [Cell; 1],
}

impl Min {
    pub fn new() -> Self {
        Self { out: [Cell::from_f64(0.0)] }
    }

    pub fn builder() -> PrimitiveBuilder {
        PrimitiveBuilder::new(
            "Min",
            NodeSchema {
                inputs: vec![
                    PortSpec { kind: ScalarKind::F64, firing: Firing::Any, name: "lhs".into() },
                    PortSpec { kind: ScalarKind::F64, firing: Firing::Any, name: "rhs".into() },
                ],
                output: vec![FieldSpec { name: "value".into(), kind: ScalarKind::F64 }],
                params: vec![],
                doc: "elementwise minimum of two input series",
            },
            |_| Box::new(Min::new()),
        )
    }
}

impl Default for Min {
    fn default() -> Self {
        Self::new()
    }
}

impl Node for Min {
    fn lookbacks(&self) -> Vec<usize> {
        vec![1, 1]
    }

    fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Cell]> {
        let a = ctx.f64_in(0);
        let b = ctx.f64_in(1);
        if a.is_empty() || b.is_empty() {
            return None;
        }
        self.out[0] = Cell::from_f64(a[0].min(b[0]));
        Some(&self.out)
    }

    fn label(&self) -> String {
        "Min".to_string()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use aura_core::{AnyColumn, Scalar, Timestamp};

    #[test]
    fn min_is_the_pairwise_smaller_once_both_inputs_present() {
        let mut m = Min::new();
        let mut inputs = vec![
            AnyColumn::with_capacity(ScalarKind::F64, 1),
            AnyColumn::with_capacity(ScalarKind::F64, 1),
        ];

        // only input 0 present -> None
        inputs[0].push(Scalar::f64(2.0)).unwrap();
        assert_eq!(m.eval(Ctx::new(&inputs, Timestamp(0))), None);

        // both present -> min(a, b), either order
        inputs[1].push(Scalar::f64(-1.0)).unwrap();
        assert_eq!(m.eval(Ctx::new(&inputs, Timestamp(0))), Some([Cell::from_f64(-1.0)].as_slice()));
    }

    #[test]
    fn input_slots_are_named_lhs_rhs() {
        let names: Vec<String> = Min::builder().schema().inputs.iter().map(|p| p.name.clone()).collect();
        assert_eq!(names, ["lhs", "rhs"]);
    }
}