visual-cortex 0.6.0

use std::collections::{HashSet, VecDeque};
use std::time::Duration;

use visual_cortex::{
    from_fn, DetectorError, DetectorOutput, Error, EventKind, FakeSource, Frame, Rate, Region,
    Session, ThresholdDirection,
};

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 every_match_mode_emits_per_tick() {
    let session = Session::builder()
        .source(FakeSource::new([Frame::solid(10, 10, [0, 0, 200, 255])]))
        .build()
        .await
        .unwrap();

    let mut stream = session
        .watch("always")
        .region(Region::Full)
        .rate(Rate::hz(20.0))
        .frame_diff(false) // identical frames must still evaluate in this test
        .detector(from_fn(|_| Ok(DetectorOutput::Bool(true))))
        .on_every_match()
        .subscribe()
        .unwrap();

    for _ in 0..3 {
        let ev = next_event(&mut stream).await;
        assert_eq!(&*ev.watcher, "always");
        assert!(matches!(
            ev.kind,
            EventKind::Matched {
                output: DetectorOutput::Bool(true)
            }
        ));
    }
}

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

    let mut stream = session
        .watch("gated")
        .rate(Rate::hz(20.0))
        .frame_diff(false)
        .gate(from_fn(|_| Ok(DetectorOutput::Bool(false)))) // always closed
        .detector(from_fn(|_| Ok(DetectorOutput::Bool(true))))
        .on_every_match()
        .subscribe()
        .unwrap();

    let got = tokio::time::timeout(Duration::from_millis(500), stream.next()).await;
    assert!(got.is_err(), "closed gate must suppress all events");
}

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

    let _first = session
        .watch("hp")
        .detector(from_fn(|_| Ok(DetectorOutput::None)))
        .subscribe()
        .unwrap();

    let err = session
        .watch("hp")
        .detector(from_fn(|_| Ok(DetectorOutput::None)))
        .subscribe()
        .unwrap_err();
    assert!(matches!(err, Error::DuplicateWatcher(name) if name == "hp"));
}

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

    let err = session.watch("empty").subscribe().unwrap_err();
    assert!(matches!(err, Error::MissingDetector(name) if name == "empty"));
}

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

    let mut stream = session
        .watch("doomed")
        .rate(Rate::hz(20.0))
        .frame_diff(false)
        .detector(from_fn(|_| Ok(DetectorOutput::Bool(true))))
        .on_every_match()
        .subscribe()
        .unwrap();

    // Drain one event so the watcher is demonstrably live first.
    tokio::time::timeout(Duration::from_secs(30), stream.next())
        .await
        .expect("watcher should emit while session is alive")
        .expect("stream closed prematurely");

    drop(session);

    // Aborting tasks drops their broadcast Sender clones; the stream must end.
    let end = tokio::time::timeout(Duration::from_secs(30), stream.next())
        .await
        .expect("stream did not end after session drop");
    assert!(
        end.is_none(),
        "expected None after session drop, got {end:?}"
    );
}

#[tokio::test(start_paused = true)]
async fn change_mode_emits_only_on_transitions() {
    // 5 dark frames, then bright red forever (FakeSource repeats its last frame).
    let mut frames = vec![Frame::solid(10, 10, [10, 10, 10, 255]); 5];
    frames.push(Frame::solid(10, 10, [10, 10, 250, 255])); // BGRA: R = 250

    let session = Session::builder()
        .source(FakeSource::new(frames))
        .build()
        .await
        .unwrap();

    // Default mode is Change; default frame_diff(true) suppresses re-evaluating
    // identical frames, which is exactly what this scenario exercises.
    let mut stream = session
        .watch("red_banner")
        .region(Region::Full)
        .rate(Rate::hz(20.0))
        .detector(from_fn(|view: &visual_cortex::FrameView<'_>| {
            let first_row = view.rows().next().expect("view is non-empty");
            Ok(DetectorOutput::Bool(first_row[2] > 128)) // red channel of first pixel
        }))
        .subscribe()
        .unwrap();

    let first = next_event(&mut stream).await;
    assert!(
        matches!(
            &first.kind,
            EventKind::Changed {
                old: None,
                new: DetectorOutput::Bool(false)
            }
        ),
        "first evaluation reports the initial state, got {:?}",
        first.kind
    );

    let second = next_event(&mut stream).await;
    assert!(
        matches!(
            &second.kind,
            EventKind::Changed {
                old: Some(DetectorOutput::Bool(false)),
                new: DetectorOutput::Bool(true)
            }
        ),
        "transition to red fires exactly one Changed event, got {:?}",
        second.kind
    );

    // Once the source settles on repeating red, no further transitions may fire.
    let more = tokio::time::timeout(Duration::from_millis(500), stream.next()).await;
    assert!(
        more.is_err(),
        "Change mode must not emit without a transition"
    );
}

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

    // Detector ignores pixels and replays a scripted value sequence — this keeps
    // the test independent of capture/watcher tick interleaving.
    let mut values = VecDeque::from([200.0, 150.0, 80.0, 60.0, 120.0]);
    let mut stream = session
        .watch("hp")
        .rate(Rate::hz(20.0))
        .frame_diff(false) // the frame never changes; evaluate every tick anyway
        .detector(from_fn(move |_| {
            let v = values.pop_front().unwrap_or(120.0);
            Ok(DetectorOutput::Number(v))
        }))
        .on_threshold(|hp| hp < 100.0)
        .subscribe()
        .unwrap();

    let entered = next_event(&mut stream).await;
    assert!(
        matches!(
            entered.kind,
            EventKind::ThresholdCrossed { value, direction: ThresholdDirection::Entered } if value == 80.0
        ),
        "expected Entered at 80, got {:?}",
        entered.kind
    );

    let exited = next_event(&mut stream).await;
    assert!(
        matches!(
            exited.kind,
            EventKind::ThresholdCrossed { value, direction: ThresholdDirection::Exited } if value == 120.0
        ),
        "expected Exited at 120, got {:?}",
        exited.kind
    );

    // The script settles on 120.0 (outside); no further crossings may fire.
    let more = tokio::time::timeout(Duration::from_millis(500), stream.next()).await;
    assert!(
        more.is_err(),
        "threshold mode must not emit without a predicate flip"
    );
}

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

    let mut stream = session
        .watch("broken")
        .rate(Rate::hz(20.0))
        .frame_diff(false) // static frame; must still tick to accumulate errors
        .detector(from_fn(|_| Err(DetectorError::Other("boom".into()))))
        .on_every_match()
        .subscribe()
        .unwrap();

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

    // It must fire once per degradation, not on every subsequent error tick.
    let more = tokio::time::timeout(Duration::from_millis(500), stream.next()).await;
    assert!(
        more.is_err(),
        "degraded event must not repeat while errors continue"
    );
}

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

    let a = session
        .watch("a")
        .rate(Rate::hz(20.0))
        .frame_diff(false)
        .detector(from_fn(|_| Ok(DetectorOutput::Bool(true))))
        .on_every_match()
        .subscribe()
        .unwrap();

    let b = session
        .watch("b")
        .rate(Rate::hz(20.0))
        .frame_diff(false)
        .detector(from_fn(|_| Ok(DetectorOutput::Bool(true))))
        .on_every_match()
        .subscribe()
        .unwrap();

    let mut merged = a.merge(b);
    let mut seen = HashSet::new();
    // Bounded: both names arrive within the first tick or two. If a watcher
    // dies, the loop must run out and fail the assertion, not hang forever.
    for _ in 0..10 {
        if seen.len() == 2 {
            break;
        }
        let ev = next_event(&mut merged).await;
        assert!(matches!(ev.kind, EventKind::Matched { .. }));
        seen.insert(ev.watcher.to_string());
    }
    assert_eq!(seen, HashSet::from(["a".to_string(), "b".to_string()]));
}