//! `TapLive` — the live-subscription sink (C8) for one declared tap: calls a
//! consumer-owned closure `(Timestamp, Cell)` inline per fired warm cycle, on
//! the sim thread, zero alloc. Loss policy is the closure's own (a UI
//! consumer typically `try_send`s into its own channel and drops when behind)
//! — per #283: consumer-owned logic and state, no registry ceremony.

use aura_core::{Cell, Ctx, Node, ScalarKind, Timestamp};

use crate::tap_cell::newest_cell;

/// A single-column live sink around a consumer closure.
pub struct TapLive {
    kind: ScalarKind,
    consumer: Box<dyn FnMut(Timestamp, Cell) + Send>,
}

impl TapLive {
    pub fn new(kind: ScalarKind, consumer: impl FnMut(Timestamp, Cell) + Send + 'static) -> Self {
        Self { kind, consumer: Box::new(consumer) }
    }
}

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

    fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Cell]> {
        let cell = newest_cell(self.kind, &ctx)?; // None until warm
        (self.consumer)(ctx.now(), cell);
        None
    }

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

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

    #[test]
    fn calls_the_consumer_per_warm_cycle_in_order_and_not_before() {
        let (tx, rx) = mpsc::channel();
        let mut sink = TapLive::new(ScalarKind::F64, move |ts, cell| {
            let _ = tx.send((ts, cell.f64()));
        });
        let mut inputs = vec![AnyColumn::with_capacity(ScalarKind::F64, 1)];

        // cold: consumer not called.
        assert_eq!(sink.eval(Ctx::new(&inputs, Timestamp(1))), None);
        assert!(rx.try_recv().is_err(), "no call before warm-up");

        for (t, v) in [(2_i64, 1.5_f64), (3, 2.5)] {
            inputs[0].push(Scalar::f64(v)).unwrap();
            assert_eq!(sink.eval(Ctx::new(&inputs, Timestamp(t))), None);
        }
        let got: Vec<(Timestamp, f64)> = rx.try_iter().collect();
        assert_eq!(got, vec![(Timestamp(2), 1.5), (Timestamp(3), 2.5)]);
    }

    /// `newest_cell`'s four-kind dispatch (shared with `TapFold`) is
    /// otherwise only exercised at F64/I64; the Bool and Timestamp arms
    /// need a witness too, so a regression in either goes caught.
    #[test]
    fn calls_the_consumer_at_bool_and_timestamp_kinds_too() {
        let (tx, rx) = mpsc::channel();
        let mut sink = TapLive::new(ScalarKind::Bool, move |ts, cell| {
            let _ = tx.send((ts, cell.bool()));
        });
        let mut inputs = vec![AnyColumn::with_capacity(ScalarKind::Bool, 1)];
        inputs[0].push(Scalar::bool(true)).unwrap();
        assert_eq!(sink.eval(Ctx::new(&inputs, Timestamp(1))), None);
        assert_eq!(rx.try_recv(), Ok((Timestamp(1), true)));

        let (tx2, rx2) = mpsc::channel();
        let mut sink2 = TapLive::new(ScalarKind::Timestamp, move |ts, cell| {
            let _ = tx2.send((ts, cell.ts()));
        });
        let mut inputs2 = vec![AnyColumn::with_capacity(ScalarKind::Timestamp, 1)];
        inputs2[0].push(Scalar::ts(Timestamp(99))).unwrap();
        assert_eq!(sink2.eval(Ctx::new(&inputs2, Timestamp(1))), None);
        assert_eq!(rx2.try_recv(), Ok((Timestamp(1), Timestamp(99))));
    }
}