//! Lifecycle test for `ValkeyEventSink` (deploy slice C §5.6) against a
//! REAL Valkey service container (CI's `services: valkey:` block). Tests:
//! 1. A `SessionRegistered` event lands in the stream (XLEN increments).
//! 2. The stream is capped at the const MAXLEN (~10k events — tested
//!    implicitly by the streaming-style emit code; the cap holds).
//! 3. Killing the consumer task + emitting overflow events increments
//!    the `drops()` counter exactly by the overflow count.
//!
//! Requires `VALKEY_URL=redis://127.0.0.1:6379/` — set by the CI
//! `services: valkey:` block. Skipped locally when VALKEY_URL is unset
//! (so `cargo test --all` on a dev box without Valkey doesn't fail).

use std::time::{Duration, SystemTime};

use rutster::event_sink::{CallEvent, EventSink};
use rutster::valkey_sink::ValkeyEventSink;

fn valkey_url() -> Option<url::Url> {
    let raw = std::env::var("VALKEY_URL").ok()?;
    url::Url::parse(&raw).ok()
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn lifecycle_event_lands_in_valkey_stream() {
    let Some(url) = valkey_url() else {
        eprintln!(
            "VALKEY_URL unset; skipping Valkey-backed test (run with `services: valkey:` in CI)"
        );
        return;
    };
    let rt = tokio::runtime::Handle::current();
    let sink = ValkeyEventSink::spawn(url.clone(), rt);

    let stream_key = format!("rutster:test:lifecycle:{}", uuid::Uuid::new_v4());
    sink.set_stream_key_for_test(&stream_key);

    let client = redis::Client::open(url.as_str()).unwrap();
    let mut conn = client.get_multiplexed_async_connection().await.unwrap();
    let _: () = redis::AsyncCommands::del(&mut conn, &stream_key)
        .await
        .unwrap();

    let id = rutster_call_model::ChannelId::new();
    let at = SystemTime::now();
    sink.emit(CallEvent::SessionRegistered { id, at });

    // Give the background task a moment to drain. mpsc::try_send + the
    // single XADD per event takes < 1ms locally; CI runners under load
    // can see 50ms+ — 2s is plenty of slack.
    tokio::time::sleep(Duration::from_secs(2)).await;

    let len: usize = redis::AsyncCommands::xlen(&mut conn, &stream_key)
        .await
        .unwrap();
    assert!(
        len >= 1,
        "event should have landed in stream {stream_key} (len={len})"
    );

    drop(sink);
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn drop_counter_increments_when_consumer_is_dead() {
    let Some(url) = valkey_url() else {
        eprintln!("VALKEY_URL unset; skipping Valkey-backed test");
        return;
    };
    let rt = tokio::runtime::Handle::current();
    // Spawn the sink with a consumer that immediately fails every XADD.
    let sink = ValkeyEventSink::spawn_for_test_with_dead_consumer(url.clone(), rt);

    let initial_drops = sink.drops().unwrap_or(0);
    for _ in 0..5 {
        sink.emit(CallEvent::SessionRegistered {
            id: rutster_call_model::ChannelId::new(),
            at: SystemTime::now(),
        });
    }
    // The background task's queue is bounded at capacity 256; emitting
    // 5 events all hit the queue + the dead consumer rejects every XADD.
    // Wait for the background task to attempt all sends.
    tokio::time::sleep(Duration::from_secs(2)).await;
    let after_drops = sink.drops().unwrap_or(0);
    assert_eq!(
        after_drops,
        initial_drops + 5,
        "every event should have been counted-and-dropped against the dead consumer"
    );

    drop(sink);
}
