//! Fieldtest c0010 #2 — `aura_std::Recorder` as a reusable sink block (axis b).
//!
//! Question under test: a downstream project (C16) wires the SHIPPED
//! `aura_std::Recorder` onto a node's output via the raw `Harness::bootstrap`
//! API, runs the harness, drains the `mpsc::Receiver`, and the recorded
//! `(Timestamp, Vec<Scalar>)` rows match what the public rustdoc led us to
//! expect. The rustdoc says (struct.Recorder):
//!   - `Recorder::new(kinds, firing, tx)` — one input column per `kinds` entry;
//!   - "Each fired cycle it reads the newest value of every column and sends the
//!     row to tx; it returns None during warm-up until all columns have a value."
//!
//! Topology (single f64 source -> SMA(3) -> Recorder[F64]):
//!
//!   price --> SMA(3) --> Recorder(tx)   (Recorder is a pure consumer, C8)
//!
//! Public-surface facts used (rustdoc only):
//!  - aura_std::Sma::new(length): SMA over the last `length` f64 values, None
//!    until warm (warms once `length` values have arrived).
//!  - aura_std::Recorder::new(&[ScalarKind::F64], Firing::Any, tx).
//!  - aura_engine::{Harness, Edge, SourceSpec, Target}.

use std::sync::mpsc;

use aura_core::{Firing, Scalar, ScalarKind, Timestamp};
use aura_engine::{Edge, Harness, SourceSpec, Target};
use aura_std::{Recorder, Sma};

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::<(Timestamp, Vec<Scalar>)>();

    // nodes: 0 = SMA(3), 1 = Recorder over one F64 column.
    let mut h = Harness::bootstrap(
        vec![
            Box::new(Sma::new(3)),
            Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx)),
        ],
        vec![SourceSpec {
            kind: ScalarKind::F64,
            targets: vec![Target { node: 0, slot: 0 }], // price -> SMA
        }],
        vec![
            Edge { from: 0, to: 1, slot: 0, from_field: 0 }, // SMA -> Recorder.in0
        ],
    )
    .expect("valid SMA -> Recorder DAG");

    // 5 ticks. SMA(3) warms at the 3rd tick; the Recorder then records once per
    // SMA-fired cycle (Firing::Any over one column already warm).
    let prices: &[(i64, f64)] = &[(1, 10.0), (2, 20.0), (3, 30.0), (4, 40.0), (5, 50.0)];
    h.run(vec![f64_stream(prices)]);

    let rows: Vec<(Timestamp, Vec<Scalar>)> = rx.try_iter().collect();
    println!("recorded rows = {rows:?}");

    // Hand model from rustdoc alone:
    //   SMA(3) emits None for t=1,2 (warm-up), then:
    //     t=3 -> mean(10,20,30) = 20.0
    //     t=4 -> mean(20,30,40) = 30.0
    //     t=5 -> mean(30,40,50) = 40.0
    //   The Recorder records one row per fired (warm) cycle, tagged ctx.now(),
    //   each row a single-element Vec<Scalar> with the F64 column's newest value.
    let expected: Vec<(Timestamp, Vec<Scalar>)> = vec![
        (Timestamp(3), vec![Scalar::F64(20.0)]),
        (Timestamp(4), vec![Scalar::F64(30.0)]),
        (Timestamp(5), vec![Scalar::F64(40.0)]),
    ];
    assert_eq!(rows.len(), expected.len(), "one row per warm SMA cycle (no warm-up rows)");
    for (got, want) in rows.iter().zip(expected.iter()) {
        assert_eq!(got.0, want.0, "row timestamp = ctx.now()");
        assert_eq!(got.1.len(), 1, "single F64 column => one-element row");
        match (got.1[0], want.1[0]) {
            (Scalar::F64(g), Scalar::F64(w)) => {
                assert!((g - w).abs() < 1e-9, "recorded f64 {g} ~= {w}");
            }
            other => panic!("expected F64 scalar, got {other:?}"),
        }
    }

    println!("c0010_2 OK: Recorder[F64] drained from mpsc matches the rustdoc model");
}
