visual-cortex 0.4.0

use std::sync::{Arc, Mutex};
use std::thread::ThreadId;
use std::time::Duration;

use visual_cortex::{
    Detector, DetectorError, DetectorOutput, EventKind, FakeSource, Frame, FrameView, Rate, Region,
    Session,
};

/// Heavy detector that returns an incrementing number each evaluation and
/// records which thread ran it: output changes every tick, so Change mode
/// emits every tick it runs.
struct HeavyCounter {
    count: u64,
    seen_thread: Arc<Mutex<Option<ThreadId>>>,
}

impl Detector for HeavyCounter {
    fn evaluate(&mut self, _view: &FrameView<'_>) -> Result<DetectorOutput, DetectorError> {
        self.count += 1;
        *self.seen_thread.lock().unwrap() = Some(std::thread::current().id());
        Ok(DetectorOutput::Number(self.count as f64))
    }
    fn is_heavy(&self) -> bool {
        true
    }
}

/// Heavy detector that always panics; the loop must survive and degrade.
struct Panicky;

impl Detector for Panicky {
    fn evaluate(&mut self, _view: &FrameView<'_>) -> Result<DetectorOutput, DetectorError> {
        panic!("detector exploded");
    }
    fn is_heavy(&self) -> bool {
        true
    }
}

async fn next_event(stream: &mut visual_cortex::EventStream) -> visual_cortex::Event {
    tokio::time::timeout(Duration::from_secs(30), stream.next())
        .await
        .expect("timed out waiting for event")
        .expect("stream closed unexpectedly")
}

#[tokio::test(start_paused = true)]
async fn heavy_detector_runs_off_loop_and_keeps_state() {
    let session = Session::builder()
        .source(FakeSource::new([Frame::solid(8, 8, [0, 0, 0, 255])]))
        .build()
        .await
        .unwrap();

    let seen = Arc::new(Mutex::new(None));
    let mut stream = session
        .watch("counter")
        .region(Region::Full)
        .rate(Rate::hz(10.0))
        .frame_diff(false) // static frame; the detector output is what changes
        .detector(HeavyCounter {
            count: 0,
            seen_thread: seen.clone(),
        })
        .subscribe()
        .unwrap();

    // Consecutive events prove the boxed detector survives the round-trip
    // onto the blocking pool and back with its state intact.
    let a = next_event(&mut stream).await;
    let b = next_event(&mut stream).await;
    let (EventKind::Changed { new: na, .. }, EventKind::Changed { new: nb, .. }) = (a.kind, b.kind)
    else {
        panic!("expected Changed events");
    };
    assert_eq!(na.as_number().unwrap() + 1.0, nb.as_number().unwrap());

    // Under start_paused the runtime is current-thread, so the async runtime
    // runs on this test thread; a differing thread id pins off-loop execution.
    let evaluated_on = seen.lock().unwrap().expect("detector ran");
    assert_ne!(
        evaluated_on,
        std::thread::current().id(),
        "heavy detector must run off the async runtime thread"
    );
}

#[tokio::test(start_paused = true)]
async fn panicking_heavy_detector_degrades_but_does_not_kill_the_loop() {
    let session = Session::builder()
        .source(FakeSource::new([Frame::solid(8, 8, [0, 0, 0, 255])]))
        .build()
        .await
        .unwrap();

    let mut stream = session
        .watch("boom")
        .region(Region::Full)
        .rate(Rate::hz(10.0))
        .frame_diff(false)
        .detector(Panicky)
        .subscribe()
        .unwrap();

    let ev = next_event(&mut stream).await;
    assert!(
        matches!(
            ev.kind,
            EventKind::WatcherDegraded {
                consecutive_errors: 5
            }
        ),
        "expected WatcherDegraded after 5 panics, got {:?}",
        ev.kind
    );
}