macltok

mod support;

use std::time::Duration;

use macltok::{

    ClientHandle, Command, Engine, EngineConfig, ErrorKind, Event, EventStream, HistoryState,

    MediaSize, MessageContent, RoomCommand, TimelineCommand, 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};

use support::{event_messages, receive_until, shut_down};

#[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.open_room(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_message(

            room_id.to_owned().into(),

            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)

            .into_iter()

            .any(|event| support::media_of(event).is_some())

    })

    .await;

    let media_id = event_messages(&media_event)

        .into_iter()

        .find_map(|event| support::media_of(event).map(|media| media.id.clone()))

        .unwrap();

    let media = handle.media(&media_id, MediaSize::Full).await.unwrap();

    assert!(!media.bytes.is_empty());

    shut_down(&handle, &mut events, running).await;

}

#[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());

    let (first_open, second_open) = tokio::join!(

        handle.open_room(first_id.to_owned()),

        handle.open_room(second_id.to_owned())

    );

    first_open.unwrap();

    second_open.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).into_iter().any(|event| {

                        support::message(event)

                            .is_some_and(|message| message.body == "first update")

                    });

                second_updated |= timeline_id.room_id() == second_id

                    && event_messages(&event).into_iter().any(|event| {

                        support::message(event)

                            .is_some_and(|message| message.body == "second update")

                    });

            }

        }

    })

    .await

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

    handle.close_room(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).into_iter().any(|event| {

                support::message(event).is_some_and(|message| message.body == "still live")

            })

    })

    .await;

    shut_down(&handle, &mut events, running).await;

}

#[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.open_room(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).into_iter().any(|event| {

            support::message(event).is_some_and(|message| message.body == "before reset")

        })

    })

    .await;

    let mut items =

        resync_and_confirm_reset_first(&handle, &mut events, room_id.to_owned().into()).await;

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

        TimelineItem::Event(event) if support::message(event).is_some_and(|message| 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::Event(event) if support::message(event).is_some_and(|message| message.body == "after reset")

                    )

                }) {

                    break;

                }

            }

        }

    })

    .await

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

    shut_down(&handle, &mut events, running).await;

}

#[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());

    handle.open_room(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;

    let mut close = handle

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

        .unwrap();

    let mut replacement_open = handle

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

        .unwrap();

    let mut second_close = handle

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

        .unwrap();

    let mut final_open = handle

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

        .unwrap();

    // biased select drains every buffered reset before it ever reports a ready reply,

    // so the engine can race ahead and land both resets before either open is observed

    let mut resets = 0;

    let mut order = Vec::new();

    let (mut close_done, mut replacement_open_done, mut second_close_done, mut final_open_done) =

        (false, false, false, false);

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

        while order.len() < 4 {

            tokio::select! {

                biased;

                event = events.recv() => {

                    if let Event::TimelineReset { timeline_id, .. } = event.expect("event stream closed")

                        && timeline_id.room_id() == room_id

                    {

                        resets += 1;

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

                    }

                }

                result = &mut close, if !close_done => {

                    result.expect("close failed");

                    close_done = true;

                    order.push("close");

                }

                result = &mut replacement_open, if !replacement_open_done => {

                    result.expect("replacement open failed");

                    replacement_open_done = true;

                    order.push("replacement_open");

                }

                result = &mut second_close, if !second_close_done => {

                    result.expect("second close failed");

                    second_close_done = true;

                    order.push("second_close");

                }

                result = &mut final_open, if !final_open_done => {

                    result.expect("final open failed");

                    final_open_done = true;

                    order.push("final_open");

                }

            }

        }

    })

    .await

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

    assert_eq!(

        order,

        ["close", "replacement_open", "second_close", "final_open"]

    );

    assert_eq!(resets, 2, "each open should have reset the timeline once");

    shut_down(&handle, &mut events, running).await;

}

#[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());

    handle.open_room(room_id.to_owned()).await.unwrap();

    let mut open = handle

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

            room_id: room_id.to_owned(),

            root_event_id: root.to_owned(),

        }))

        .unwrap();

    let mut send = handle

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

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

            content: MessageContent::text(""),

        }))

        .unwrap();

    let mut open_succeeded = false;

    let mut send_done = false;

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

        while !send_done {

            tokio::select! {

                biased;

                result = &mut open, if !open_succeeded => {

                    result.expect("thread open failed");

                    open_succeeded = true;

                }

                result = &mut send, if !send_done => {

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

                    let error = result.expect_err("empty body send unexpectedly succeeded");

                    assert_eq!(error.kind, ErrorKind::EmptyBody);

                    send_done = true;

                }

            }

        }

    })

    .await

    .expect("thread commands did not complete");

    resync_and_confirm_reset_first(

        &handle,

        &mut events,

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

    )

    .await;

    shut_down(&handle, &mut events, running).await;

}

#[tokio::test]

async fn pagination_reports_and_remembers_start_of_history() {

    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!("!history: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.open_room(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;

    server

        .sync_room(

            &sync_client,

            JoinedRoomBuilder::new(room_id)

                .add_timeline_event(EventFactory::new().text_msg("latest").sender(&ALICE))

                .set_timeline_limited()

                .set_timeline_prev_batch("before-latest"),

        )

        .await;

    receive_until(&mut events, |event| {

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

            && event_messages(event).into_iter().any(|event| {

                support::message(event).is_some_and(|message| message.body == "latest")

            })

    })

    .await;

    server

        .mock_room_messages()

        .match_from("before-latest")

        .ok(RoomMessagesResponseTemplate::default())

        .mock_once()

        .mount()

        .await;

    let mut reached_start = false;

    for _ in 0..3 {

        handle

            .paginate(room_id.to_owned().into(), 10)

            .await

            .unwrap();

        receive_history(&mut events, room_id, HistoryState::Loading).await;

        match receive_history_terminal(&mut events, room_id).await {

            HistoryState::StartReached => {

                reached_start = true;

                break;

            }

            HistoryState::Idle => {}

            state => panic!("unexpected pagination state {state:?}"),

        }

    }

    assert!(

        reached_start,

        "pagination did not reach the mocked room start"

    );

    handle

        .paginate(room_id.to_owned().into(), 10)

        .await

        .unwrap();

    receive_history(&mut events, room_id, HistoryState::StartReached).await;

    shut_down(&handle, &mut events, running).await;

}

/// resyncs `timeline_id` and confirms its `TimelineReset` was already on the

/// stream by the time the resync command answered, racing the two so we

/// don't have to drain the whole stream first

async fn resync_and_confirm_reset_first(

    handle: &ClientHandle,

    events: &mut EventStream,

    timeline_id: TimelineId,

) -> Vec<TimelineItem> {

    let mut pending = handle

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

            timeline_id: timeline_id.clone(),

        }))

        .await

        .unwrap();

    let mut reset_items = None;

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

        loop {

            tokio::select! {

                biased;

                event = events.recv() => {

                    if let Event::TimelineReset { timeline_id: reset_id, items } = event.expect("event stream closed")

                        && reset_id == timeline_id

                    {

                        reset_items = Some(items);

                    }

                }

                result = &mut pending => {

                    result.expect("resync failed");

                    break;

                }

            }

        }

    })

    .await

    .expect("resync did not complete");

    reset_items.expect("resync completed before its reset")

}

async fn receive_history(

    events: &mut EventStream,

    room_id: &matrix_sdk::ruma::RoomId,

    expected: HistoryState,

) {

    receive_until(events, |event| {

        matches!(event, Event::HistoryChanged { timeline_id: changed, state } if changed.room_id() == room_id && state == &expected)

    })

    .await;

}

async fn receive_history_terminal(

    events: &mut EventStream,

    room_id: &matrix_sdk::ruma::RoomId,

) -> HistoryState {

    match receive_until(events, |event| {

        matches!(event, Event::HistoryChanged { timeline_id: changed, state } if changed.room_id() == room_id && !matches!(state, HistoryState::Loading))

    })

    .await

    {

        Event::HistoryChanged { state, .. } => state,

        _ => unreachable!(),

    }

}

async fn wait_for_room_subscriptions(

    server: &MatrixMockServer,

    room_ids: &[&matrix_sdk::ruma::RoomId],

) {

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

        loop {

            let found = server

                .received_requests()

                .await

                .unwrap()

                .iter()

                .filter_map(|request| {

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

                })

                .filter_map(|body| {

                    body.get("room_subscriptions")

                        .and_then(|value| value.as_object())

                        .cloned()

                })

                .any(|subscriptions| {

                    room_ids

                        .iter()

                        .all(|room_id| subscriptions.contains_key(room_id.as_str()))

                });

            if found {

                break;

            }

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

        }

    })

    .await

    .expect("Sliding Sync did not retain every open room subscription");

}