//! Integration tests for `StreamRegistry`: scope-based filtering, health
//! drift, and `detach()` shutdown semantics.
//!
//! Uses the `MockStream` from `dirigent_core::sharing` (enabled via the
//! `test-utils` feature; see `required-features` in `Cargo.toml`).

use std::sync::Arc;
use std::time::Duration;

use uuid::Uuid;

use dirigent_core::sharing::MockStream;
use dirigent_core::sharing::bus::SharingBus;
use dirigent_core::sharing::registry::StreamRegistry;
use dirigent_protocol::{
    Event,
    streaming::{
        BusEvent, EventKind, EventOrigin, EventRouting, StreamScope,
    },
};

// ─── Helpers ────────────────────────────────────────────────────────────────

/// Build a `BusEvent` with an explicit `scroll_id` in the routing so the
/// bus does not need to consult its late-bind cache. `connector_uid` is
/// populated too so tests can mix-and-match filters.
fn make_scoped_event(scroll_id: Uuid, connector_uid: Uuid) -> BusEvent {
    BusEvent {
        routing: EventRouting {
            scroll_id: Some(scroll_id),
            connector_uid: Some(connector_uid),
            connector_id: Some("conn-test".to_string()),
            native_session_id: None,
            kind: EventKind::System,
        },
        origin: EventOrigin::Runtime,
        event: Arc::new(Event::Connected),
    }
}

// ─── Tests ───────────────────────────────────────────────────────────────────

#[tokio::test]
async fn mock_stream_receives_only_session_scoped_events() {
    let bus = SharingBus::new();
    let registry = StreamRegistry::new(Arc::clone(&bus));

    let scroll_x = Uuid::now_v7();
    let scroll_y = Uuid::now_v7();
    let connector_uid = Uuid::now_v7();

    let mock = MockStream::new("mock", StreamScope::Session {
        scroll_id: scroll_x,
    });
    let _id = registry.attach("mock".to_string(), mock.clone()).await;

    // One event in-scope (scroll_x), two out-of-scope (scroll_y).
    bus.publish(make_scoped_event(scroll_x, connector_uid))
        .await;
    bus.publish(make_scoped_event(scroll_y, connector_uid))
        .await;
    bus.publish(make_scoped_event(scroll_y, connector_uid))
        .await;

    // Give the bus worker + stream worker a chance to process.
    tokio::time::sleep(Duration::from_millis(50)).await;

    let count = mock.received_count();
    assert_eq!(
        count,
        1,
        "expected 1 in-scope event, got {count}",
    );
    assert_eq!(
        mock.received.lock().unwrap()[0].routing.scroll_id,
        Some(scroll_x)
    );
}

#[tokio::test]
async fn health_drifts_to_unavailable_after_five_failures() {
    use dirigent_core::sharing::HealthStatus;

    let bus = SharingBus::new();
    let registry = StreamRegistry::new(Arc::clone(&bus));

    let scroll_id = Uuid::now_v7();
    let connector_uid = Uuid::now_v7();

    let mock = MockStream::new("mock", StreamScope::Session { scroll_id });
    mock.fail_next(5);
    let _id = registry.attach("mock".to_string(), mock.clone()).await;

    // Five consecutive failures should drift Healthy → Unavailable.
    for _ in 0..5 {
        bus.publish(make_scoped_event(scroll_id, connector_uid))
            .await;
    }

    tokio::time::sleep(Duration::from_millis(100)).await;

    let infos = registry.list().await;
    assert_eq!(infos.len(), 1);
    match &infos[0].health {
        HealthStatus::Unavailable { reason } => {
            assert!(
                reason.contains("5 failures"),
                "expected reason to mention the failure count, got: {reason}"
            );
        }
        other => panic!("expected Unavailable after 5 failures, got {:?}", other),
    }
    assert_eq!(infos[0].lagged_count, 5);
}

#[tokio::test]
async fn detach_invokes_shutdown_and_stops_delivery() {
    let bus = SharingBus::new();
    let registry = StreamRegistry::new(Arc::clone(&bus));

    let scroll_id = Uuid::now_v7();
    let connector_uid = Uuid::now_v7();

    let mock = MockStream::new("mock", StreamScope::Session { scroll_id });
    let id = registry.attach("mock".to_string(), mock.clone()).await;

    // One in-scope event before detach to prove delivery works at all.
    bus.publish(make_scoped_event(scroll_id, connector_uid))
        .await;
    tokio::time::sleep(Duration::from_millis(50)).await;
    assert_eq!(mock.received_count(), 1, "pre-detach delivery failed");

    // Detach — worker should stop, shutdown should run.
    let reg = registry.detach(id).await.expect("stream should exist");
    // Wait for the worker to finish so we know the post-detach publish
    // cannot possibly reach the stream.
    let _ = tokio::time::timeout(Duration::from_millis(500), async {
        let handle = &reg.worker;
        while !handle.is_finished() {
            tokio::task::yield_now().await;
        }
    })
    .await;

    // Publish post-detach; the stream must not receive it.
    bus.publish(make_scoped_event(scroll_id, connector_uid))
        .await;
    tokio::time::sleep(Duration::from_millis(50)).await;

    assert_eq!(
        mock.received_count(),
        1,
        "stream received an event after detach"
    );
    assert!(registry.list().await.is_empty());
}
