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:?}");

                }

                _ => {}

            }

        }

    })

    .await

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

    let mut reset_seen = false;

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

        loop {

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

                Event::TimelineReset { timeline_id, .. }

                    if timeline_id.thread_root().is_some_and(|event| event == root) =>

                {

                    reset_seen = true

                }

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

                    assert!(reset_seen, "thread resync completed before its reset");

                    break;

                }

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

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

                }

                _ => {}

            }

        }

    })

    .await

    .expect("thread resync 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 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

        .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;

    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).any(|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

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

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

                limit: 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

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

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

            limit: 10,

        }))

        .await

        .unwrap();

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

    handle

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

        .await

        .unwrap();

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

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

}

async fn receive_history(

    events: &mut macltok::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;

}

fn event_messages(event: &Event) -> std::vec::IntoIter<&macltok::TimelineMessage> {

    let items = match event {

        Event::TimelineReset { items, .. } => items.iter().collect::<Vec<_>>(),

        Event::TimelineChanged { diffs, .. } => {

            diffs.iter().flat_map(diff_items).collect::<Vec<_>>()

        }

        _ => Vec::new(),

    };

    items

        .into_iter()

        .filter_map(|item| match item {

            TimelineItem::Message(message) => Some(message.as_ref()),

            _ => None,

        })

        .collect::<Vec<_>>()

        .into_iter()

}

fn diff_items(diff: &TimelineDiff) -> Vec<&TimelineItem> {

    match diff {

        TimelineDiff::Append { values } | TimelineDiff::Reset { values } => values.iter().collect(),

        TimelineDiff::PushFront { value }

        | TimelineDiff::PushBack { value }

        | TimelineDiff::Insert { value, .. }

        | TimelineDiff::Set { value, .. } => vec![value],

        TimelineDiff::Clear

        | TimelineDiff::PopFront

        | TimelineDiff::PopBack

        | TimelineDiff::Remove { .. }

        | TimelineDiff::Truncate { .. } => Vec::new(),

        _ => Vec::new(),

    }

}

async fn receive_history_terminal(

    events: &mut macltok::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");

}

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")

}