//! Fieldtest c0006 #3 — record MANY distinct interior streams of ONE run.
//!
//! Axis: "Record multiple distinct interior streams of one run via
//! recording-nodes (sink-as-role)."
//!
//! This is the cycle headline: impossible before 0006 (one `observe` index = one
//! recorded row). Two independent Recorder nodes tap two interior indicators —
//! SMA(2) and SMA(4) — in the SAME harness, each draining to its own channel.
//! Mirrors the 0006 spec's worked "User-facing program" example.
//!
//! It also records `(Timestamp, Vec<Scalar>)` rows (the exact type the spec's
//! worked example drains into: `Vec<(Timestamp, Vec<Scalar>)>`) to fieldtest
//! that the spec's stated drain shape is reachable.

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;

/// Recorder declared over arbitrary input kinds; records the full input row
/// (one value per declared slot) tagged with now(). Matches the spec's
/// `Recorder::new(&[ScalarKind], tx)` shape and `(Timestamp, Vec<Scalar>)` drain.
struct Recorder {
    kinds: Vec<ScalarKind>,
    tx: Sender<(Timestamp, Vec<Scalar>)>,
}
impl Recorder {
    fn new(kinds: &[ScalarKind], tx: Sender<(Timestamp, Vec<Scalar>)>) -> Self {
        Self {
            kinds: kinds.to_vec(),
            tx,
        }
    }
}
impl Node for Recorder {
    fn schema(&self) -> NodeSchema {
        NodeSchema {
            inputs: self
                .kinds
                .iter()
                .map(|&kind| InputSpec {
                    kind,
                    lookback: 1,
                    firing: Firing::Any,
                })
                .collect(),
            output: vec![],
        }
    }
    fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Scalar]> {
        let mut row = Vec::with_capacity(self.kinds.len());
        for (i, &kind) in self.kinds.iter().enumerate() {
            // newest of each input, by declared kind
            match kind {
                ScalarKind::F64 => {
                    let w = ctx.f64_in(i);
                    if w.is_empty() {
                        return None;
                    }
                    row.push(Scalar::F64(w[0]));
                }
                ScalarKind::I64 => {
                    let w = ctx.i64_in(i);
                    if w.is_empty() {
                        return None;
                    }
                    row.push(Scalar::I64(w[0]));
                }
                ScalarKind::Bool => {
                    let w = ctx.bool_in(i);
                    if w.is_empty() {
                        return None;
                    }
                    row.push(Scalar::Bool(w[0]));
                }
                ScalarKind::Timestamp => {
                    let w = ctx.ts_in(i);
                    if w.is_empty() {
                        return None;
                    }
                    row.push(Scalar::Ts(w[0]));
                }
            }
        }
        let _ = self.tx.send((ctx.now(), row));
        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_fast, rx_fast) = mpsc::channel();
    let (tx_slow, rx_slow) = mpsc::channel();

    let mut h = Harness::bootstrap(
        vec![
            Box::new(Sma::new(2)),                                  // 0
            Box::new(Sma::new(4)),                                  // 1
            Box::new(Recorder::new(&[ScalarKind::F64], tx_fast)),   // 2: taps node 0
            Box::new(Recorder::new(&[ScalarKind::F64], tx_slow)),   // 3: taps node 1
        ],
        vec![SourceSpec {
            kind: ScalarKind::F64,
            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) -> recorder fast
            Edge { from: 1, to: 3, slot: 0, from_field: 0 }, // SMA(4) -> recorder slow
        ],
    )
    .expect("valid DAG");

    h.run(vec![f64_stream(&[
        (1, 10.0),
        (2, 12.0),
        (3, 14.0),
        (4, 16.0),
        (5, 18.0),
    ])]);

    let fast: Vec<(Timestamp, Vec<Scalar>)> = rx_fast.try_iter().collect();
    let slow: Vec<(Timestamp, Vec<Scalar>)> = rx_slow.try_iter().collect();

    // SMA(2): warm at t=2 -> means 11,13,15,17 at t=2,3,4,5.
    let fast_expected = vec![
        (Timestamp(2), vec![Scalar::F64(11.0)]),
        (Timestamp(3), vec![Scalar::F64(13.0)]),
        (Timestamp(4), vec![Scalar::F64(15.0)]),
        (Timestamp(5), vec![Scalar::F64(17.0)]),
    ];
    // SMA(4): warm at t=4 -> means 13,15 at t=4,5.
    let slow_expected = vec![
        (Timestamp(4), vec![Scalar::F64(13.0)]),
        (Timestamp(5), vec![Scalar::F64(15.0)]),
    ];

    println!("fast = {fast:?}");
    println!("slow = {slow:?}");
    assert_eq!(fast, fast_expected, "SMA(2) interior stream");
    assert_eq!(slow, slow_expected, "SMA(4) interior stream");
    // The headline: two DIFFERENT-RATE streams recorded from ONE run, no shared
    // cycle index — each sparse and timestamped (4 vs 2 records).
    assert_ne!(fast.len(), slow.len(), "different-rate recorders are independent");
    println!("c0006_3 OK: two interior streams recorded from one run (sink-as-role)");
}
