//! Integration test for SharingBus late-bind and filter routing.
//!
//! Exercises the full publish cycle: subscribe with two different filters,
//! publish three events (one before cache population, one that populates the
//! cache, one that is late-bound from the cache), then assert each subscriber
//! saw exactly the events it should.

use std::time::Duration;

use uuid::Uuid;

use dirigent_core::sharing::bus::SharingBus;
use dirigent_protocol::{
    Message, MessageRole, MessageStatus, Session, SessionMetadata,
    conversation::MessagePart,
    streaming::{BusEvent, EventFilter},
    Event,
};

// ─── Fixtures ────────────────────────────────────────────────────────────────

fn make_session(id: &str) -> Session {
    let now = chrono::Utc::now();
    Session {
        id: id.to_string(),
        title: "test-session".to_string(),
        created_at: now,
        updated_at: now,
        metadata: SessionMetadata {
            project_path: "/tmp/test".to_string(),
            model: None,
            total_messages: 0,
            system_message: None,
            current_mode_id: None,
            _meta: None,
            project_id: None,
        },
        cwd: None,
        models: None,
        modes: None,
        config_options: None,
        acp_client_id: None,
    }
}

fn make_message(session_id: &str, msg_id: &str) -> Message {
    let now = chrono::Utc::now();
    Message {
        id: msg_id.to_string(),
        session_id: session_id.to_string(),
        role: MessageRole::Assistant,
        created_at: now,
        content: vec![MessagePart::Text {
            text: "hello".to_string(),
        }],
        status: MessageStatus::Completed,
        metadata: None,
    }
}

// ─── Test ─────────────────────────────────────────────────────────────────────

/// Full late-bind cycle:
///
/// - `scroll_rx` is filtered by `EventFilter::ScrollId(scroll_id)`.
///   It should receive events #2 and #3 only (not #1, because the cache was
///   empty when #1 was published).
///
/// - `uid_rx` is filtered by `EventFilter::ConnectorUid(uid)`.
///   It should receive all three events because `connector_uid` is set on
///   every event by `BusEvent::from_connector_event`.
#[tokio::test]
async fn scroll_id_late_bind_via_session_registered() {
    let bus = SharingBus::new();

    let uid = Uuid::new_v4();
    let scroll_id = Uuid::new_v4();

    let mut scroll_rx = bus
        .subscribe_filtered(EventFilter::ScrollId(scroll_id), 32)
        .await;
    let mut uid_rx = bus
        .subscribe_filtered(EventFilter::ConnectorUid(uid), 32)
        .await;

    // Event #1: SessionCreated — cache is empty, scroll_id will not be
    // late-bound. The ScrollId subscriber must NOT see this event.
    bus.publish(BusEvent::from_connector_event(
        Event::SessionCreated {
            connector_id: "mock".into(),
            session: make_session("abc"),
        },
        Some(uid),
        "mock".into(),
    ))
    .await;

    // Event #2: SessionRegistered — populates the cache
    // (`connector_id="mock"`, `session_id="abc"`) → `scroll_id`.
    // The bus also sets `routing.scroll_id` on this event itself, so the
    // ScrollId subscriber DOES see it.
    bus.publish(BusEvent::from_connector_event(
        Event::SessionRegistered {
            connector_id: "mock".into(),
            session_id: "abc".into(),
            scroll_id: scroll_id.to_string(),
        },
        Some(uid),
        "mock".into(),
    ))
    .await;

    // Event #3: MessageCompleted — no scroll_id on entry, but the bus
    // late-binds it from the cache via (connector_id="mock", session_id="abc").
    // The ScrollId subscriber DOES see it.
    bus.publish(BusEvent::from_connector_event(
        Event::MessageCompleted {
            connector_id: "mock".into(),
            message: make_message("abc", "m1"),
        },
        Some(uid),
        "mock".into(),
    ))
    .await;

    // Give the bus worker time to dispatch all three events.
    tokio::time::sleep(Duration::from_millis(50)).await;

    // Drain what each subscriber received.
    let mut scroll_events: Vec<BusEvent> = Vec::new();
    while let Ok(Some(e)) =
        tokio::time::timeout(Duration::from_millis(20), scroll_rx.rx.recv()).await
    {
        scroll_events.push(e);
    }

    let mut uid_events: Vec<BusEvent> = Vec::new();
    while let Ok(Some(e)) =
        tokio::time::timeout(Duration::from_millis(20), uid_rx.rx.recv()).await
    {
        uid_events.push(e);
    }

    // ScrollId subscriber: only events #2 and #3.
    assert_eq!(
        scroll_events.len(),
        2,
        "ScrollId subscriber should see events #2 (SessionRegistered) and #3 (MessageCompleted) only; got {:?}",
        scroll_events.iter().map(|e| format!("{:?}", e.event)).collect::<Vec<_>>()
    );
    assert!(
        matches!(scroll_events[0].event.as_ref(), Event::SessionRegistered { .. }),
        "first scroll event should be SessionRegistered"
    );
    assert!(
        matches!(scroll_events[1].event.as_ref(), Event::MessageCompleted { .. }),
        "second scroll event should be MessageCompleted"
    );
    // Both must carry the correct scroll_id.
    assert_eq!(scroll_events[0].routing.scroll_id, Some(scroll_id));
    assert_eq!(scroll_events[1].routing.scroll_id, Some(scroll_id));

    // ConnectorUid subscriber: all three.
    assert_eq!(
        uid_events.len(),
        3,
        "ConnectorUid subscriber should see all three events; got {:?}",
        uid_events.iter().map(|e| format!("{:?}", e.event)).collect::<Vec<_>>()
    );
    for ev in &uid_events {
        assert_eq!(ev.routing.connector_uid, Some(uid));
    }
}
