//! Fieldtest c0006 #2 — multi-stage fan-out / fan-in DAG, bootstrapped + run.
//!
//! Axes: "Build a multi-stage fan-out / fan-in DAG (one source fanning into
//! SMA(2) and SMA(4), then a combiner)" and "Bootstrap and run a Harness (note
//! the new bootstrap arity and that run returns ())."
//!
//! source --> SMA(2) --\
//!        \             Sub(fast - slow) --> Recorder
//!         \-> SMA(4) --/
//!
//! A downstream author wires the classic spread DAG and records the combiner's
//! output. This fixture leans on the post-0006 surface: bootstrap takes exactly
//! THREE positional args (no observe), and run returns () — retention is the
//! recorder's channel, owned by the caller.

use std::sync::mpsc::{self, Sender};

use aura_core::{Ctx, Firing, InputSpec, Node, NodeSchema, Scalar, ScalarKind, Timestamp};
use aura_engine::{Edge, Harness, SourceSpec, Target};
use aura_std::{Sma, Sub};

struct Recorder {
    tx: Sender<(Timestamp, f64)>,
}
impl Recorder {
    fn new(tx: Sender<(Timestamp, f64)>) -> Self {
        Self { tx }
    }
}
impl Node for Recorder {
    fn schema(&self) -> NodeSchema {
        NodeSchema {
            inputs: vec![InputSpec {
                kind: ScalarKind::F64,
                lookback: 1,
                firing: Firing::Any,
            }],
            output: vec![],
        }
    }
    fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Scalar]> {
        let w = ctx.f64_in(0);
        if w.is_empty() {
            return None;
        }
        let _ = self.tx.send((ctx.now(), w[0]));
        None
    }
}

fn f64_stream(pairs: &[(i64, f64)]) -> Vec<(Timestamp, Scalar)> {
    pairs
        .iter()
        .map(|&(t, v)| (Timestamp(t), Scalar::F64(v)))
        .collect()
}

fn main() {
    let (tx, rx) = mpsc::channel();

    // nodes: 0 = SMA(2), 1 = SMA(4), 2 = Sub(0 - 1), 3 = Recorder taps Sub.
    let mut h = Harness::bootstrap(
        vec![
            Box::new(Sma::new(2)),
            Box::new(Sma::new(4)),
            Box::new(Sub::new()),
            Box::new(Recorder::new(tx)),
        ],
        vec![SourceSpec {
            kind: ScalarKind::F64,
            // source fans out into both SMAs
            targets: vec![Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }],
        }],
        vec![
            Edge { from: 0, to: 2, slot: 0, from_field: 0 }, // SMA(2) -> Sub.in0
            Edge { from: 1, to: 2, slot: 1, from_field: 0 }, // SMA(4) -> Sub.in1
            Edge { from: 2, to: 3, slot: 0, from_field: 0 }, // Sub -> Recorder
        ],
    )
    .expect("valid DAG");

    // ascending-in-price stream: SMA(2) leads SMA(4), spread stays constant once warm.
    h.run(vec![f64_stream(&[
        (1, 10.0),
        (2, 12.0),
        (3, 14.0),
        (4, 16.0),
        (5, 18.0),
        (6, 20.0),
    ])]);

    let recorded: Vec<(Timestamp, f64)> = rx.try_iter().collect();

    // Sub fires once both SMAs are warm. SMA(4) warms at the 4th sample (t=4).
    //   t=4: SMA2=mean(16,14)=15, SMA4=mean(16,14,12,10)=13 -> 2
    //   t=5: SMA2=mean(18,16)=17, SMA4=mean(18,16,14,12)=15 -> 2
    //   t=6: SMA2=mean(20,18)=19, SMA4=mean(20,18,16,14)=17 -> 2
    let expected = vec![
        (Timestamp(4), 2.0),
        (Timestamp(5), 2.0),
        (Timestamp(6), 2.0),
    ];

    println!("recorded = {recorded:?}");
    println!("expected = {expected:?}");
    assert_eq!(recorded, expected, "fan-out/fan-in spread DAG recorded output");
    println!("c0006_2 OK: fan-out/fan-in DAG bootstrapped (3-arg) + run () + recorded");
}