macltok

use std::time::Duration;

use macltok::{

    Command, CommandErrorKind, Engine, EngineConfig, Event, HistoryState, LifecycleCommand,

    MediaCommand, MessageContent, RoomCommand, TimelineCommand, TimelineDiff, TimelineId,

    TimelineItem,

};

use matrix_sdk::{

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

    test_utils::mocks::{MatrixMockServer, RoomMessagesResponseTemplate},

};

use matrix_sdk_test::{ALICE, JoinedRoomBuilder, event_factory::EventFactory};

#[tokio::test]

async fn opens_a_room_and_sends_messages() {

    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!("!toolkit:localhost");

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

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

    server

        .mock_room_send()

        .body_matches_partial_json(serde_json::json!({ "body": "hello from macltok" }))

        .ok(event_id!("$sent:localhost"))

        .expect(1)

        .mount()

        .await;

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

    let sync_client = client.clone();

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

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

    handle

        .send(Command::Room(RoomCommand::Open(room_id.to_owned())))

        .await

        .unwrap();

    receive_until(

        &mut events,

        |event| matches!(event, Event::TimelineReset { timeline_id: reset, .. } if reset.room_id() == room_id),

    )

    .await;

    handle

        .send(Command::Timeline(TimelineCommand::Send {

            timeline_id: room_id.to_owned().into(),

            content: MessageContent::text("hello from macltok"),

        }))

        .await

        .unwrap();

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

        loop {

            let sent = server

                .received_requests()

                .await

                .unwrap()

                .iter()

                .any(|request| {

                    request.url.path().contains("/send/m.room.message/")

                        && serde_json::from_slice::<serde_json::Value>(&request.body)

                            .is_ok_and(|body| body["body"] == "hello from macltok")

                });

            if sent {

                break;

            }

            tokio::task::yield_now().await;

        }

    })

    .await

    .expect("message did not reach the homeserver");

    server.mock_media_download().ok_image().mount().await;

    server.mock_authed_media_download().ok_image().mount().await;

    server

        .sync_room(

            &sync_client,

            JoinedRoomBuilder::new(room_id).add_timeline_event(

                EventFactory::new()

                    .image(

                        "image.jpg".into(),

                        mxc_uri!("mxc://localhost/image").to_owned(),

                    )

                    .sender(&ALICE),

            ),

        )

        .await;

    let media_event = receive_until(&mut events, |event| {

        event_messages(event).any(|message| message.media.is_some())

    })

    .await;

    let media_id = event_messages(&media_event)

        .find_map(|message| message.media.as_ref().map(|media| media.id.clone()))

        .unwrap();

    handle

        .send(Command::Media(MediaCommand::Fetch {

            id: media_id.clone(),

            size: macltok::MediaSize::Full,

        }))

        .await

        .unwrap();

    let media = receive_until(&mut events, |event| {

        matches!(event, Event::MediaReady { media_id: received, .. } if received == &media_id)

    })

    .await;

    assert!(matches!(media, Event::MediaReady { bytes, .. } if !bytes.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 keeps_open_room_timelines_independent() {

    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 sync_client = client.clone();

    let first_id = room_id!("!first:localhost");

    let second_id = room_id!("!second:localhost");

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

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

    server.sync_joined_room(&client, second_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());

    handle

        .send(Command::Room(RoomCommand::Open(first_id.to_owned())))

        .await

        .unwrap();

    handle

        .send(Command::Room(RoomCommand::Open(second_id.to_owned())))

        .await

        .unwrap();

    receive_until(

        &mut events,

        |event| matches!(event, Event::TimelineReset { timeline_id, .. } if timeline_id.room_id() == first_id),

    )

    .await;

    receive_until(

        &mut events,

        |event| matches!(event, Event::TimelineReset { timeline_id, .. } if timeline_id.room_id() == second_id),

    )

    .await;

    wait_for_room_subscriptions(&server, &[first_id, second_id]).await;

    let factory = EventFactory::new();

    server

        .sync_room(

            &sync_client,

            JoinedRoomBuilder::new(first_id)

                .add_timeline_event(factory.text_msg("first update").sender(&ALICE)),

        )

        .await;

    server

        .sync_room(

            &sync_client,

            JoinedRoomBuilder::new(second_id)

                .add_timeline_event(factory.text_msg("second update").sender(&ALICE)),

        )

        .await;

    let mut first_updated = false;

    let mut second_updated = false;

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

        while !(first_updated && second_updated) {

            let event = events.recv().await.expect("event stream closed");

            if let Event::TimelineChanged { timeline_id, .. } = &event {

                first_updated |= timeline_id.room_id() == first_id

                    && event_messages(&event).any(|message| message.body == "first update");

                second_updated |= timeline_id.room_id() == second_id

                    && event_messages(&event).any(|message| message.body == "second update");

            }

        }

    })

    .await

    .expect("both retained rooms did not update");

    handle

        .send(Command::Room(RoomCommand::Close(first_id.to_owned())))

        .await

        .unwrap();

    server

        .sync_room(

            &sync_client,

            JoinedRoomBuilder::new(second_id)

                .add_timeline_event(factory.text_msg("still live").sender(&ALICE)),

        )

        .await;

    receive_until(&mut events, |event| {

        matches!(event, Event::TimelineChanged { timeline_id, .. } if timeline_id.room_id() == second_id)

            && event_messages(event).any(|message| message.body == "still live")

    })

    .await;

    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 resync_replaces_timeline_before_ack_and_preserves_following_diffs() {

    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 sync_client = client.clone();

    let room_id = room_id!("!resync: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());

    handle

        .send(Command::Room(RoomCommand::Open(room_id.to_owned())))

        .await

        .unwrap();

    receive_until(&mut events, |event| {

        matches!(event,

            Event::TimelineReset { timeline_id, .. } if timeline_id.room_id() == room_id

        )

    })

    .await;

    server

        .sync_room(

            &sync_client,

            JoinedRoomBuilder::new(room_id)

                .add_timeline_event(EventFactory::new().text_msg("before reset").sender(&ALICE)),

        )

        .await;

    receive_until(&mut events, |event| {

        event_messages(event).any(|message| message.body == "before reset")

    })

    .await;

    let id = handle

        .send(Command::Timeline(TimelineCommand::Resync {

            timeline_id: room_id.to_owned().into(),

        }))

        .await

        .unwrap();

    let mut reset = None;

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

        loop {

            match events.recv().await.expect("event stream closed") {

                Event::TimelineReset { timeline_id, items } if timeline_id.room_id() == room_id => {

                    reset = Some(items);

                }

                Event::CommandSucceeded { id: completed, .. } if completed == id => {

                    assert!(reset.is_some(), "resync completed before its reset");

                    break;

                }

                Event::CommandFailed(failure) if failure.id == id => {

                    panic!("resync failed: {failure:?}")

                }

                _ => {}

            }

        }

    })

    .await

    .expect("resync did not complete");

    let mut items = reset.unwrap();

    assert!(items.iter().any(|item| matches!(item,

        TimelineItem::Message(message) if message.body == "before reset"

    )));

    server

        .sync_room(

            &sync_client,

            JoinedRoomBuilder::new(room_id)

                .add_timeline_event(EventFactory::new().text_msg("after reset").sender(&ALICE)),

        )

        .await;

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

        loop {

            if let Event::TimelineChanged { timeline_id, diffs } =

                events.recv().await.expect("event stream closed")

                && timeline_id.room_id() == room_id

            {

                for diff in diffs {

                    diff.try_apply(&mut items)

                        .expect("resynced snapshot accepts following diffs");

                }

                if items.iter().any(|item| {

                    matches!(item,

                        TimelineItem::Message(message) if message.body == "after reset"

                    )

                }) {

                    break;

                }

            }

        }

    })

    .await

    .expect("resynced timeline did not update");

    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 closes_and_reopens_a_room_in_command_order() {

    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!("!reopen: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());

    let initial_open = handle

        .send(Command::Room(RoomCommand::Open(room_id.to_owned())))

        .await

        .unwrap();

    receive_until(

        &mut events,

        |event| matches!(event, Event::TimelineReset { timeline_id, .. } if timeline_id.room_id() == room_id),

    )

    .await;

    receive_until(

        &mut events,

        |event| matches!(event, Event::CommandSucceeded { id, .. } if id == &initial_open),

    )

    .await;

    let close = handle

        .send(Command::Room(RoomCommand::Close(room_id.to_owned())))

        .await

        .unwrap();

    let replacement_open = handle

        .send(Command::Room(RoomCommand::Open(room_id.to_owned())))

        .await

        .unwrap();

    let second_close = handle

        .send(Command::Room(RoomCommand::Close(room_id.to_owned())))

        .await

        .unwrap();

    let final_open = handle

        .send(Command::Room(RoomCommand::Open(room_id.to_owned())))

        .await

        .unwrap();

    let expected = [close, replacement_open, second_close, final_open];

    let mut completed = 0;

    let mut resets = 0;

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

        while completed < expected.len() {

            match events.recv().await.expect("event stream closed") {

                Event::TimelineReset { timeline_id, .. } if timeline_id.room_id() == room_id => {

                    resets += 1;

                    assert!(resets <= 2, "a room generation reset more than once");

                }

                Event::CommandSucceeded { id, .. } if expected.contains(&id) => {

                    assert_eq!(

                        id, expected[completed],

                        "room lifecycle commands completed out of order"

                    );

                    if completed == 1 || completed == 3 {

                        assert_eq!(

                            resets,

                            completed / 2 + 1,

                            "room open completed before timeline reset"

                        );

                    }

                    completed += 1;

                }

                Event::CommandFailed(failure) if expected.contains(&failure.id) => {

                    panic!("room lifecycle command failed: {failure:?}");

                }

                _ => {}

            }

        }

    })

    .await

    .expect("room lifecycle commands did not complete");

    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 thread_command_waits_for_opening_timeline() {

    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!("!thread:localhost");

    let root = event_id!("$thread-root:localhost");

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

    server.sync_joined_room(&client, room_id).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());

    let room_open = handle

        .send(Command::Room(RoomCommand::Open(room_id.to_owned())))

        .await

        .unwrap();

    receive_until(

        &mut events,

        |event| matches!(event, Event::CommandSucceeded { id, .. } if *id == room_open),

    )

    .await;

    let open = handle

        .send(Command::Timeline(TimelineCommand::OpenThread {

            room_id: room_id.to_owned(),

            root_event_id: root.to_owned(),

        }))

        .await

        .unwrap();

    let send = handle

        .send(Command::Timeline(TimelineCommand::Send {

            timeline_id: TimelineId::thread(room_id.to_owned(), root.to_owned()),

            content: MessageContent::text(""),

        }))

        .await

        .unwrap();

    let resync = handle

        .send(Command::Timeline(TimelineCommand::Resync {

            timeline_id: TimelineId::thread(room_id.to_owned(), root.to_owned()),

        }))

        .await

        .unwrap();

    let mut open_succeeded = false;

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

        loop {

            match events.recv().await.expect("event stream closed") {

                Event::CommandSucceeded { id, .. } if id == open => open_succeeded = true,

                Event::CommandFailed(failure) if failure.id == send => {

                    assert!(open_succeeded, "thread command completed before opening");

                    assert_eq!(failure.kind, CommandErrorKind::EmptyBody);

                    break;

                }

                Event::CommandFailed(failure) if failure.id == open => {

                    panic!("thread open failed: {failure:?}");

                }