synctv/synctv-realtime/tests/connection_manager_tests.rs

//! `ConnectionManager` integration tests (no Redis required)
//!
//! Tests for connection lifecycle, room joins, limits, disconnect signals,
//! and RTC filtering.

#![allow(clippy::unwrap_used)]
use std::time::Duration;
use synctv_core::models::{RealtimeActor, RoomId, UserId};
use synctv_realtime::sync::ConnectionManager;

fn stable_id(s: &str) -> i64 {
    i64::from(
        s.bytes()
            .fold(1u16, |acc, byte| acc.wrapping_add(u16::from(byte))),
    )
}

fn uid(s: &str) -> UserId {
    UserId::expect_positive(stable_id(s))
}

fn actor(user_id: UserId) -> RealtimeActor {
    RealtimeActor::user(user_id, user_id.to_string())
}

fn rid(s: &str) -> RoomId {
    RoomId::expect_positive(stable_id(s))
}

#[tokio::test]
async fn test_disconnect_signal_with_no_receiver_is_ignored() {
    let mgr = ConnectionManager::default();
    let user = uid("u1");

    mgr.register("c1".to_string(), user).await.unwrap();

    mgr.disconnect_connection("c1");
}

#[tokio::test]
async fn test_disconnect_user_from_room_signal() {
    use synctv_realtime::sync::DisconnectSignal;

    let mgr = ConnectionManager::default();
    let user = uid("u1");
    let room = rid("r1");

    mgr.register("c1".to_string(), user).await.unwrap();
    mgr.join_room("c1", room).await.unwrap();

    // Subscribe before sending signal
    let mut rx = mgr.subscribe_disconnect();

    // Disconnect user from room
    mgr.disconnect_user_from_room(&user, &room);

    let sig = rx.recv().await.unwrap();
    assert!(
        matches!(sig, DisconnectSignal::UserFromRoom { ref user_id, ref room_id }
            if user_id == &user && room_id == &room),
        "Expected UserFromRoom signal"
    );
}

#[tokio::test]
async fn test_disconnect_signal_reliability_under_load() {
    let mgr = ConnectionManager::default();

    // Register multiple connections
    for i in 0..10 {
        let user = uid(&format!("u{i}"));
        mgr.register(format!("c{i}"), user).await.unwrap();
    }

    // Subscribe to receive signals
    let mut rx = mgr.subscribe_disconnect();

    for i in 0..10 {
        mgr.disconnect_connection(&format!("c{i}"));
    }

    // All signals should be received (broadcast channel should handle this)
    let mut received_count = 0;
    for _ in 0..10 {
        match rx.recv().await {
            Ok(_) => received_count += 1,
            Err(_) => break,
        }
    }

    assert_eq!(
        received_count, 10,
        "All disconnect signals should be received"
    );
}

// Test 1: join_room is idempotent

#[tokio::test]
async fn test_join_room_idempotent() {
    let mgr = ConnectionManager::default();
    let user = uid("u1");
    let room = rid("r1");

    mgr.register("c1".to_string(), user).await.unwrap();
    mgr.join_room("c1", room).await.unwrap();
    mgr.join_room("c1", room).await.unwrap(); // second join to same room

    assert_eq!(
        mgr.room_connection_count(&room),
        1,
        "Joining the same room twice should not double-count"
    );
}

// Test 2: join_room moves between rooms

#[tokio::test]
async fn test_join_room_moves_between_rooms() {
    let mgr = ConnectionManager::default();
    let user = uid("u1");
    let r1 = rid("r1");
    let r2 = rid("r2");

    mgr.register("c1".to_string(), user).await.unwrap();

    mgr.join_room("c1", r1).await.unwrap();
    assert_eq!(mgr.room_connection_count(&r1), 1);

    mgr.join_room("c1", r2).await.unwrap();
    assert_eq!(
        mgr.room_connection_count(&r1),
        0,
        "Old room should have 0 connections after moving"
    );
    assert_eq!(
        mgr.room_connection_count(&r2),
        1,
        "New room should have 1 connection after moving"
    );

    let conn = mgr.get_connection("c1").unwrap();
    assert_eq!(conn.room_id.unwrap(), r2);
}

#[tokio::test]
async fn test_join_room_rejection_preserves_previous_room_membership() {
    let limits = synctv_realtime::sync::ConnectionLimits {
        max_per_room: 1,
        ..Default::default()
    };
    let mgr = ConnectionManager::new(limits);
    let room_a = rid("room_a");
    let room_b = rid("room_b");

    mgr.register("conn_a".to_string(), uid("user_a"))
        .await
        .unwrap();
    mgr.register("conn_b".to_string(), uid("user_b"))
        .await
        .unwrap();

    mgr.join_room("conn_a", room_a).await.unwrap();
    mgr.join_room("conn_b", room_b).await.unwrap();

    let err = mgr.join_room("conn_a", room_b).await.unwrap_err();
    assert!(err.contains("Room at capacity"));

    assert_eq!(mgr.room_connection_count(&room_a), 1);
    assert_eq!(mgr.room_connection_count(&room_b), 1);

    let conn = mgr.get_connection("conn_a").unwrap();
    assert_eq!(conn.room_id, Some(room_a));
}

// Test 3: max_duration timeout

#[tokio::test]
async fn test_max_duration_timeout() {
    use synctv_realtime::sync::ConnectionManager;

    let limits = synctv_realtime::sync::ConnectionLimits {
        max_duration: Duration::from_millis(50),
        idle_timeout: Duration::from_hours(1), // effectively disabled
        ..Default::default()
    };
    let mgr = ConnectionManager::new(limits);
    let user = uid("u1");

    mgr.register("c1".to_string(), user).await.unwrap();

    // Not yet expired
    let timeouts = mgr.check_timeouts();
    assert!(
        timeouts.is_empty(),
        "Connection should not time out immediately"
    );

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

    let timeouts = mgr.check_timeouts();
    assert_eq!(timeouts.len(), 1, "Connection should have timed out");
    assert_eq!(timeouts[0], "c1");
}

// Test 4: total connection limit

#[tokio::test]
async fn test_total_connection_limit() {
    let limits = synctv_realtime::sync::ConnectionLimits {
        max_total: 2,
        ..Default::default()
    };
    let mgr = ConnectionManager::new(limits);

    assert!(mgr.register("c1".to_string(), uid("u1")).await.is_ok());
    assert!(mgr.register("c2".to_string(), uid("u2")).await.is_ok());

    let result = mgr.register("c3".to_string(), uid("u3")).await;
    assert!(
        result.is_err(),
        "Third registration should fail when max_total=2"
    );
    assert!(result.unwrap_err().contains("capacity"));

    assert_eq!(mgr.connection_count(), 2);
}

// Test 5: disconnect signals

#[tokio::test]
async fn test_disconnect_signals() {
    use synctv_realtime::sync::{DisconnectSignal, RoomDisconnectReason};

    let mgr = ConnectionManager::default();
    let user = uid("u1");
    let room = rid("r1");

    mgr.register("c1".to_string(), user).await.unwrap();
    mgr.join_room("c1", room).await.unwrap();

    // Subscribe before sending signals
    let mut rx = mgr.subscribe_disconnect();

    // Connection disconnect
    mgr.disconnect_connection("c1");
    let sig = rx.recv().await.unwrap();
    assert!(
        matches!(sig, DisconnectSignal::Connection(ref id) if id == "c1"),
        "Expected Connection signal"
    );

    // User disconnect
    mgr.disconnect_user(&user);
    let sig = rx.recv().await.unwrap();
    assert!(
        matches!(sig, DisconnectSignal::User(ref id) if id == &user),
        "Expected User signal"
    );

    // Room disconnect
    mgr.disconnect_room(&room, RoomDisconnectReason::AccessRevoked);
    let sig = rx.recv().await.unwrap();
    assert!(
        matches!(
            sig,
            DisconnectSignal::Room {
                room_id,
                reason: RoomDisconnectReason::AccessRevoked,
            } if room_id == room
        ),
        "Expected Room signal"
    );
}

// Test 6: RTC connections filter

#[tokio::test]
async fn test_rtc_connections_filter() {
    let mgr = ConnectionManager::default();
    let u1 = uid("u1");
    let u2 = uid("u2");
    let room = rid("r1");

    mgr.register("c1".to_string(), u1).await.unwrap();
    mgr.register("c2".to_string(), u2).await.unwrap();
    mgr.join_room("c1", room).await.unwrap();
    mgr.join_room("c2", room).await.unwrap();

    // Mark only c1 as RTC-joined
    mgr.mark_voice_rtc_joined(&room, &actor(u1), "c1", true);

    let rtc = mgr.get_voice_rtc_connections(&room);
    assert_eq!(rtc.len(), 1, "Only 1 connection should be RTC-joined");
    assert_eq!(rtc[0].connection_id, "c1");

    // Mark c1 as not RTC-joined
    mgr.mark_voice_rtc_joined(&room, &actor(u1), "c1", false);
    let rtc = mgr.get_voice_rtc_connections(&room);
    assert_eq!(
        rtc.len(),
        0,
        "No connections should be RTC-joined after unmark"
    );
}

// Test 7: unregister nonexistent is a no-op