macltok

use std::time::Duration;

use macltok::{

    Command, ConnectionState, Engine, EngineConfig, EngineError, Event, LifecycleCommand,

    RoomMembership,

};

use matrix_sdk::{

    ruma::{api::client::sync::sync_events::v5, room_id},

    test_utils::mocks::MatrixMockServer,

};

use tokio::sync::watch;

#[tokio::test]

async fn rooms_changed_invalidates_room_summaries() {

    let server = MatrixMockServer::new().await;

    server

        .mock_sliding_sync()

        .ok(v5::Response::new("1".to_owned()))

        .mount()

        .await;

    let client = server.client_builder().build().await;

    let room_id = room_id!("!state:localhost");

    server.mock_room_state_encryption().plain().mount().await;

    server.sync_joined_room(&client, room_id).await;

    let config = EngineConfig::default().with_initial_history_limit(0);

    let (engine, handle, mut events) = Engine::from_client(client, config).await.unwrap();

    let running = tokio::spawn(engine.run());

    receive_until(&mut events, |event| matches!(event, Event::RoomsChanged)).await;

    let rooms = handle.rooms().await;

    let summary = rooms

        .iter()

        .find(|summary| summary.room_id == room_id)

        .expect("synced room was absent from the room summaries");

    assert_eq!(summary.membership, RoomMembership::Joined);

    assert!(!summary.name.is_empty());

    handle

        .send(Command::Lifecycle(LifecycleCommand::Shutdown))

        .await

        .unwrap();

    receive_until(&mut events, |event| matches!(event, Event::Stopped)).await;

    running.await.unwrap().unwrap();

}

#[tokio::test]

async fn connection_watch_tracks_the_engine_lifecycle() {

    let server = MatrixMockServer::new().await;

    server

        .mock_sliding_sync()

        .ok(v5::Response::new("1".to_owned()))

        .mount()

        .await;

    let client = server.client_builder().build().await;

    let config = EngineConfig::default().with_initial_history_limit(0);

    let (engine, handle, mut events) = Engine::from_client(client, config).await.unwrap();

    let mut connection = handle.connection_state();

    let connection_peer = connection.clone();

    assert_eq!(*connection.borrow(), ConnectionState::Idle);

    assert_eq!(*connection_peer.borrow(), ConnectionState::Idle);

    let running = tokio::spawn(engine.run());

    receive_until(&mut events, |event| {

        matches!(event, Event::ConnectionChanged(ConnectionState::Connecting))

    })

    .await;

    assert_ne!(*connection.borrow(), ConnectionState::Idle);

    receive_until(&mut events, |event| {

        matches!(event, Event::ConnectionChanged(ConnectionState::Online))

    })

    .await;

    wait_for_connection(&mut connection, ConnectionState::Online).await;

    assert_eq!(*connection_peer.borrow(), ConnectionState::Online);

    handle

        .send(Command::Lifecycle(LifecycleCommand::Shutdown))

        .await

        .unwrap();

    receive_until(&mut events, |event| matches!(event, Event::Stopped)).await;

    running.await.unwrap().unwrap();

    wait_for_connection(&mut connection, ConnectionState::Stopped).await;

    assert_eq!(*connection_peer.borrow(), ConnectionState::Stopped);

}

#[tokio::test]

async fn closing_the_event_stream_stops_an_idle_engine() {

    let server = MatrixMockServer::new().await;

    server

        .mock_sliding_sync()

        .ok(v5::Response::new("1".to_owned()))

        .mount()

        .await;

    let client = server.client_builder().build().await;

    let (engine, handle, mut events) = Engine::from_client(

        client,

        EngineConfig::default().with_initial_history_limit(0),

    )

    .await

    .unwrap();

    let running = tokio::spawn(engine.run());

    receive_until(&mut events, |event| {

        matches!(event, Event::ConnectionChanged(ConnectionState::Online))

    })

    .await;

    drop(events);

    let result = tokio::time::timeout(Duration::from_secs(2), running)

        .await

        .expect("engine did not stop after its event receiver closed")

        .unwrap();

    assert!(matches!(result, Err(EngineError::EventReceiverClosed)));

    drop(handle);

}

async fn wait_for_connection(

    connection: &mut watch::Receiver<ConnectionState>,

    expected: ConnectionState,

) {

    tokio::time::timeout(Duration::from_secs(2), async {

        loop {

            if *connection.borrow_and_update() == expected {

                return;

            }

            connection

                .changed()

                .await

                .expect("connection watch closed before the expected state");

        }

    })

    .await

    .expect("expected connection state did not arrive");

}

async fn receive_until(

    events: &mut macltok::EventStream,

    predicate: impl Fn(&Event) -> bool,

) -> Event {

    tokio::time::timeout(Duration::from_secs(2), async {

        while let Some(event) = events.recv().await {

            if predicate(&event) {

                return event;

            }

        }

        panic!("event stream closed before the expected event");

    })

    .await

    .expect("expected event did not arrive")

}