//! `SeriesReducer` — a folding sink (C8) over one `f64` column. It folds each
//! warm value into a `SeriesFold` and, on `finalize`, emits a single summary row
//! `[last, max_drawdown, sign_flips]`, so a downstream reduction reads O(1) per
//! member instead of the whole per-cycle series. C7-pure: owned fold + channel.

use aura_core::{Cell, Ctx, Firing, Node, NodeSchema, PortSpec, PrimitiveBuilder, Scalar, ScalarKind, SeriesFold, Timestamp};
use std::sync::mpsc::Sender;

/// A single-`f64`-column folding sink. Emits exactly one row on `finalize`:
/// `[Scalar::f64(last), Scalar::f64(max_drawdown), Scalar::i64(sign_flips)]`.
pub struct SeriesReducer {
    fold: SeriesFold,
    last_ts: Timestamp,
    tx: Sender<(Timestamp, Vec<Scalar>)>,
}

impl SeriesReducer {
    pub fn new(tx: Sender<(Timestamp, Vec<Scalar>)>) -> Self {
        Self { fold: SeriesFold::new(), last_ts: Timestamp(0), tx }
    }

    pub fn builder(firing: Firing, tx: Sender<(Timestamp, Vec<Scalar>)>) -> PrimitiveBuilder {
        PrimitiveBuilder::new(
            "SeriesReducer",
            NodeSchema {
                inputs: vec![PortSpec { kind: ScalarKind::F64, firing, name: "col[0]".to_string() }],
                output: vec![], // sink: empty output (C8)
                params: vec![],
            },
            move |_| Box::new(SeriesReducer::new(tx.clone())),
        )
    }
}

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

    fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Cell]> {
        let v = ctx.f64_in(0).get(0)?; // None during warm-up
        self.fold.fold(v);
        self.last_ts = ctx.now();
        None
    }

    fn finalize(&mut self) {
        let row = vec![
            Scalar::f64(self.fold.last),
            Scalar::f64(self.fold.max_drawdown),
            Scalar::i64(self.fold.sign_flips as i64),
        ];
        let _ = self.tx.send((self.last_ts, row));
    }

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

#[cfg(test)]
mod tests {
    use super::*;
    use aura_core::AnyColumn;
    use std::sync::mpsc;

    #[test]
    fn folds_and_emits_one_summary_row_on_finalize() {
        let (tx, rx) = mpsc::channel();
        let mut r = SeriesReducer::new(tx);
        let mut inputs = vec![AnyColumn::with_capacity(ScalarKind::F64, 1)];
        for (t, v) in [(1_i64, 10.0_f64), (2, 12.0), (3, 8.0)] {
            inputs[0].push(Scalar::f64(v)).unwrap();
            assert_eq!(r.eval(Ctx::new(&inputs, Timestamp(t))), None);
        }
        // nothing emitted per-eval.
        assert!(rx.try_recv().is_err());
        r.finalize();
        let out: Vec<(Timestamp, Vec<Scalar>)> = rx.try_iter().collect();
        assert_eq!(out.len(), 1);
        assert_eq!(out[0].1[0], Scalar::f64(8.0)); // last
        assert_eq!(out[0].1[1], Scalar::f64(4.0)); // max_drawdown 12 - 8
        assert_eq!(out[0].1[2], Scalar::i64(0)); // sign_flips (all positive)
    }

    #[test]
    fn empty_finalize_emits_zero_summary() {
        let (tx, rx) = mpsc::channel();
        let mut r = SeriesReducer::new(tx);
        r.finalize();
        let out: Vec<(Timestamp, Vec<Scalar>)> = rx.try_iter().collect();
        assert_eq!(out, vec![(Timestamp(0), vec![Scalar::f64(0.0), Scalar::f64(0.0), Scalar::i64(0)])]);
    }
}