macltok

mod support;

use std::time::Duration;

#[cfg(feature = "mirror")]

use macltok::Mirror;

use macltok::{

    ClientHandle, Engine, EngineConfig, EngineError, Event, EventStream, HistoryGap,

    HistoryRepairPhase, HistoryRepairReport, HistoryRepairStop, RecentHistoryState, Termination,

    TimelineDiff, TimelineId, TimelineItem,

};

use matrix_sdk::{

    Client,

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

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

};

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

use support::shut_down;

use wiremock::ResponseTemplate;

struct TestRoom {

    server: MatrixMockServer,

    sync_client: Client,

    handle: ClientHandle,

    events: EventStream,

    running: tokio::task::JoinHandle<Result<Termination, EngineError>>,

    /// the open room's timeline, kept current by every helper that reads events

    items: imbl::Vector<TimelineItem>,

    opened: bool,

}

async fn start_room(room_id: &RoomId) -> TestRoom {

    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;

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

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

    let sync_client = client.clone();

    let config = EngineConfig::default()

        .with_initial_history_limit(0)

        .with_search_backfill_pages_per_room(0);

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

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

    TestRoom {

        server,

        sync_client,

        handle,

        events,

        running,

        items: imbl::Vector::new(),

        opened: false,

    }

}

/// the next event that satisfies `predicate`. only the open itself may reset the room

/// timeline, history after it has to arrive as diffs

async fn next_until(room: &mut TestRoom, mut predicate: impl FnMut(&Event) -> bool) -> Event {

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

        loop {

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

            match &event {

                Event::TimelineReset {

                    timeline_id: TimelineId::Room(_),

                    items,

                } => {

                    assert!(!room.opened, "history reset the room timeline");

                    room.opened = true;

                    room.items = items.clone();

                }

                Event::TimelineChanged {

                    timeline_id: TimelineId::Room(_),

                    diffs,

                } => {

                    for diff in diffs.clone() {

                        assert!(

                            !matches!(diff, TimelineDiff::Reset { .. }),

                            "history replaced the room timeline"

                        );

                        diff.try_apply(&mut room.items)

                            .expect("diff did not fit the timeline");

                    }

                }

                _ => {}

            }

            if predicate(&event) {

                return event;

            }

        }

    })

    .await

    .expect("expected event did not arrive")

}

async fn timeline_until(

    room: &mut TestRoom,

    predicate: impl Fn(&imbl::Vector<TimelineItem>) -> bool,

) {

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

        while !predicate(&room.items) {

            next_until(room, |_| true).await;

        }

    })

    .await

    .expect("the timeline never got there");

}

async fn open_room(room: &mut TestRoom, room_id: &RoomId) {

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

    next_until(room, |event| {

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

    })

    .await;

}

async fn cache_event_before_open(room: &TestRoom, room_id: &RoomId, id: &EventId, body: &str) {

    room.server

        .sync_room(

            &room.sync_client,

            JoinedRoomBuilder::new(room_id).add_timeline_event(

                EventFactory::new()

                    .text_msg(body)

                    .event_id(id)

                    .sender(&ALICE),

            ),

        )

        .await;

}

fn history_event(

    room_id: &RoomId,

    id: &EventId,

    body: &str,

) -> matrix_sdk::ruma::serde::Raw<matrix_sdk::ruma::events::AnyTimelineEvent> {

    EventFactory::new()

        .room(room_id)

        .text_msg(body)

        .event_id(id)

        .sender(&ALICE)

        .into_raw_timeline()

}

async fn wait_for_recent(room: &mut TestRoom, room_id: &RoomId) -> RecentHistoryState {

    let event = next_until(room, |event| {

        matches!(event, Event::RecentHistoryChanged { room_id: id, state, .. }

            if id == room_id && !matches!(state, RecentHistoryState::Loading))

    })

    .await;

    match event {

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

        _ => unreachable!(),

    }

}

async fn wait_for_settled_fill(room: &mut TestRoom) -> HistoryRepairReport {

    match next_until(room, |event| {

        matches!(event, Event::HistoryRepairChanged(report)

            if report.phase != HistoryRepairPhase::Bridging)

    })

    .await

    {

        Event::HistoryRepairChanged(report) => report,

        _ => unreachable!(),

    }

}

fn has_event(items: &imbl::Vector<TimelineItem>, id: &EventId) -> bool {

    items.iter().any(

        |item| matches!(item, TimelineItem::Event(event) if event.event_id.as_deref() == Some(id)),

    )

}

fn gap_in(items: &imbl::Vector<TimelineItem>) -> Option<HistoryGap> {

    items.iter().find_map(|item| match item {

        TimelineItem::HistoryGap(gap) => Some(gap.clone()),

        _ => None,

    })

}

#[tokio::test]

async fn a_recent_load_is_bounded_and_keeps_cached_history_behind_a_gap() {

    let room_id = room_id!("!recent-repair:localhost");

    let old_id = event_id!("$old:localhost");

    let new_ids = [

        event_id!("$fresh-a:localhost"),

        event_id!("$fresh-b:localhost"),

        event_id!("$fresh-c:localhost"),

        event_id!("$fresh-d:localhost"),

    ];

    let mut room = start_room(room_id).await;

    for (index, id) in new_ids.iter().enumerate() {

        let token = format!("recent-{index}");

        let response = RoomMessagesResponseTemplate::default()

            .end_token(token.clone())

            .events(vec![history_event(room_id, id, "recent event")]);

        if index > 0 {

            let previous = format!("recent-{}", index - 1);

            room.server

                .mock_room_messages()

                .match_from(&previous)

                .ok(response)

                .mock_once()

                .mount()

                .await;

        } else {

            room.server

                .mock_room_messages()

                .ok(response)

                .mock_once()

                .mount()

                .await;

        }

    }

    cache_event_before_open(&room, room_id, old_id, "retained old event").await;

    open_room(&mut room, room_id).await;

    room.handle

        .retry_recent_history(room_id.to_owned())

        .await

        .unwrap();

    timeline_until(&mut room, |items| {

        gap_in(items).is_some()

            && has_event(items, old_id)

            && new_ids.iter().all(|id| has_event(items, id))

    })

    .await;

    assert_eq!(

        wait_for_recent(&mut room, room_id).await,

        RecentHistoryState::Loaded

    );

    let requests = room.server.received_requests().await.unwrap();

    let messages = requests

        .iter()

        .filter(|request| request.url.path().ends_with("/messages"))

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

    assert_eq!(messages.len(), 4);

    assert!(!messages[0].url.query_pairs().any(|(key, _)| key == "from"));

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

}

#[tokio::test]

async fn explicit_recent_start_coalesces_with_the_automatic_load() {

    let room_id = room_id!("!recent-coalesce:localhost");

    let fresh_id = event_id!("$coalesced:localhost");

    let mut room = start_room(room_id).await;

    room.server

        .mock_room_messages()

        .ok(RoomMessagesResponseTemplate::default()

            .with_delay(Duration::from_millis(300))

            .events(vec![history_event(room_id, fresh_id, "shared result")]))

        .mock_once()

        .mount()

        .await;

    open_room(&mut room, room_id).await;

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

        loop {

            let requested = room

                .server

                .received_requests()

                .await

                .unwrap()

                .iter()

                .any(|request| request.url.path().ends_with("/messages"));

            if requested {

                break;

            }

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

        }

    })

    .await

    .expect("automatic recent request did not start");

    room.handle

        .retry_recent_history(room_id.to_owned())

        .await

        .unwrap();

    assert_eq!(

        wait_for_recent(&mut room, room_id).await,

        RecentHistoryState::Loaded

    );

    let requests = room.server.received_requests().await.unwrap();

    assert_eq!(

        requests

            .iter()

            .filter(|request| request.url.path().ends_with("/messages"))

            .count(),

        1

    );

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

}

#[cfg(feature = "mirror")]

#[tokio::test]

async fn fill_gap_returns_after_the_mirror_contains_the_committed_rows() {

    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!("!fill-gap-mirror:localhost");

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

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

    let sync_client = client.clone();

    let old_id = event_id!("$gap-old:localhost");

    let fresh_id = event_id!("$gap-fresh:localhost");

    let middle_id = event_id!("$gap-middle:localhost");

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

        client,

        EngineConfig::default()

            .with_initial_history_limit(0)

            .with_search_backfill_pages_per_room(0),

    )

    .await

    .unwrap();

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

    let (mirror, mirror_handle) = Mirror::new(handle.clone(), events);

    let mirror_running = tokio::spawn(mirror.run());

    server

        .mock_room_messages()

        .ok(RoomMessagesResponseTemplate::default()

            .end_token("recent-gap")

            .events(vec![history_event(room_id, fresh_id, "new window")]))

        .mock_once()

        .mount()

        .await;

    server

        .mock_room_messages()

        .match_from("recent-gap")

        .ok(RoomMessagesResponseTemplate::default().end_token("recent-gap"))

        .mock_once()

        .mount()

        .await;

    server

        .sync_room(

            &sync_client,

            JoinedRoomBuilder::new(room_id).add_timeline_event(

                EventFactory::new()

                    .text_msg("old retained")

                    .event_id(old_id)

                    .sender(&ALICE),

            ),

        )

        .await;

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

    handle

        .retry_recent_history(room_id.to_owned())

        .await

        .unwrap();

    let timeline_id = TimelineId::from(room_id.to_owned());

    let gap = tokio::time::timeout(Duration::from_secs(5), async {

        loop {

            if let Some(gap) = mirror_handle

                .snapshot()

                .timelines

                .get(&timeline_id)

                .and_then(|timeline| gap_in(&timeline.items))

            {

                break gap;

            }

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

        }

    })

    .await

    .expect("the recent history gap did not reach the mirror");

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

        loop {

            if mirror_handle.snapshot().recent_history.get(room_id)

                == Some(&RecentHistoryState::Loaded)

            {

                break;

            }

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

        }

    })

    .await

    .expect("automatic recent loading did not finish");

    server

        .mock_room_messages()

        .match_from("recent-gap")

        .ok(RoomMessagesResponseTemplate::default()

            .with_delay(Duration::from_millis(300))

            .events(vec![

                history_event(room_id, middle_id, "middle page"),

                history_event(room_id, old_id, "old retained overlap"),

            ]))

        .mock_once()

        .mount()

        .await;

    let fill_handle = handle.clone();

    let target_gap_id = gap.id.clone();

    let fill_gap_id = target_gap_id.clone();

    let fill = tokio::spawn(async move {

        fill_handle

            .fill_history_gap(room_id.to_owned(), fill_gap_id)

            .await

    });

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

        loop {

            let snapshot = mirror_handle.snapshot();

            let active_gap = snapshot.repairs.get(room_id).is_some_and(|report| {

                report.gap_id == target_gap_id && report.phase == HistoryRepairPhase::Bridging

            });

            if active_gap {

                break;

            }

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

        }

    })

    .await

    .expect("the active repair report did not identify its target gap");

    fill.await.unwrap().unwrap();

    let snapshot = mirror_handle.snapshot();

    let timeline = snapshot

        .timelines

        .get(&timeline_id)

        .expect("the room timeline disappeared");

    let event_ids = timeline

        .items

        .iter()

        .filter_map(|item| match item {

            TimelineItem::Event(event) => event.event_id.as_deref(),

            _ => None,

        })

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

    assert!(event_ids.contains(&fresh_id));

    assert!(event_ids.contains(&middle_id));

    assert!(event_ids.contains(&old_id));

    handle.shutdown().await.unwrap();

    assert_eq!(running.await.unwrap().unwrap(), Termination::ShutDown);

    mirror_running.await.unwrap();

}

#[tokio::test]

async fn cancelling_a_gap_fill_keeps_committed_progress_for_retry() {

    let room_id = room_id!("!cancel-gap-fill:localhost");

    let old_id = event_id!("$cancel-gap-old:localhost");

    let fresh_id = event_id!("$cancel-gap-fresh:localhost");

    let middle_id = event_id!("$cancel-gap-middle:localhost");

    let mut room = start_room(room_id).await;

    room.server

        .mock_room_messages()

        .ok(RoomMessagesResponseTemplate::default()

            .end_token("gap-cursor-1")

            .events(vec![history_event(room_id, fresh_id, "recent")]))

        .mock_once()

        .mount()

        .await;

    room.server

        .mock_room_messages()

        .match_from("gap-cursor-1")

        .ok(RoomMessagesResponseTemplate::default().end_token("gap-cursor-1"))

        .mock_once()

        .mount()

        .await;

    cache_event_before_open(&room, room_id, old_id, "old cached").await;

    open_room(&mut room, room_id).await;

    room.handle

        .retry_recent_history(room_id.to_owned())

        .await

        .unwrap();

    timeline_until(&mut room, |items| gap_in(items).is_some()).await;

    let gap = gap_in(&room.items).unwrap();

    assert_eq!(

        wait_for_recent(&mut room, room_id).await,

        RecentHistoryState::Loaded

    );

    room.server

        .mock_room_messages()

        .match_from("gap-cursor-1")

        .ok(RoomMessagesResponseTemplate::default()

            .end_token("gap-cursor-2")

            .events(vec![history_event(

                room_id,

                middle_id,

                "committed progress",

            )]))

        .mock_once()

        .mount()

        .await;

    room.server

        .mock_room_messages()

        .match_from("gap-cursor-2")

        .ok(RoomMessagesResponseTemplate::default()

            .with_delay(Duration::from_secs(5))

            .events(vec![history_event(room_id, old_id, "old overlap")]))

        .mock_once()

        .mount()

        .await;

    let fill_handle = room.handle.clone();

    let gap_id = gap.id.clone();

    let fill = tokio::spawn(async move {

        fill_handle

            .fill_history_gap(room_id.to_owned(), gap_id)

            .await

    });

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

        loop {

            let requested = room

                .server

                .received_requests()

                .await

                .unwrap()

                .iter()

                .any(|request| {

                    request.url.path().ends_with("/messages")

                        && request

                            .url

                            .query_pairs()

                            .any(|(key, value)| key == "from" && value == "gap-cursor-2")

                });

            if requested {

                break;

            }

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

        }

    })

    .await

    .expect("the second gap page did not start");

    room.handle

        .cancel_history_repair(room_id.to_owned())

        .await

        .unwrap();

    let error = fill.await.unwrap().unwrap_err();

    assert_eq!(error.kind, macltok::ErrorKind::Interrupted);

    let report = wait_for_settled_fill(&mut room).await;

    assert_eq!(report.stop, Some(HistoryRepairStop::Cancelled));

    assert!(report.failure.is_none());

    room.server

        .mock_room_messages()

        .match_from("gap-cursor-2")

        .ok(

            RoomMessagesResponseTemplate::default().events(vec![history_event(

                room_id,

                old_id,

                "old overlap",

            )]),

        )

        .mock_once()

        .mount()

        .await;

    room.handle

        .fill_history_gap(room_id.to_owned(), gap.id)

        .await

        .unwrap();

    timeline_until(&mut room, |items| {

        [fresh_id, middle_id, old_id]

            .iter()

            .all(|id| has_event(items, id))

            && gap_in(items).is_none()

    })

    .await;

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

}

#[tokio::test]

async fn failed_recent_load_is_public_and_keeps_the_cached_timeline() {

    let room_id = room_id!("!failed-recent:localhost");

    let old_id = event_id!("$failed-old:localhost");

    let mut room = start_room(room_id).await;

    room.server

        .mock_room_messages()

        .respond_with(ResponseTemplate::new(403).set_body_json(serde_json::json!({

            "errcode": "M_FORBIDDEN",

            "error": "history is unavailable",

        })))

        .mock_once()

        .mount()

        .await;

    cache_event_before_open(&room, room_id, old_id, "kept event").await;

    open_room(&mut room, room_id).await;

    assert!(

        room.handle

            .retry_recent_history(room_id.to_owned())

            .await

            .is_err()

    );

    let state = wait_for_recent(&mut room, room_id).await;

    assert!(matches!(state, RecentHistoryState::Failed(_)));

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

}

#[tokio::test]

async fn closing_during_a_recent_load_settles_idle_not_failed() {

    let room_id = room_id!("!close-recent:localhost");

    let mut room = start_room(room_id).await;

    room.server

        .mock_room_messages()

        .ok(RoomMessagesResponseTemplate::default()

            .with_delay(Duration::from_secs(5))

            .events(vec![history_event(

                room_id,

                event_id!("$never:localhost"),

                "too late",

            )]))

        .mock_once()

        .mount()

        .await;

    open_room(&mut room, room_id).await;

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

        while !room

            .server

            .received_requests()

            .await

            .unwrap()

            .iter()

            .any(|request| request.url.path().ends_with("/messages"))

        {

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

        }

    })

    .await

    .expect("the recent load never asked for messages");

    room.handle.close_room(room_id.to_owned()).await.unwrap();

    let mut states = Vec::new();

    next_until(&mut room, |event| {

        if let Event::RecentHistoryChanged { state, .. } = event {

            states.push(state.clone());

        }

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

    })

    .await;

    assert_eq!(

        states,

        [RecentHistoryState::Loading, RecentHistoryState::Idle]

    );

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

}

#[tokio::test]

async fn a_reply_whose_target_is_not_loaded_fetches_it() {

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

    let target_id = event_id!("$target:localhost");

    let reply_id = event_id!("$reply:localhost");

    let mut room = start_room(room_id).await;

    room.server

        .mock_room_messages()

        .ok(RoomMessagesResponseTemplate::default())

        .mount()

        .await;

    room.server

        .mock_room_event()

        .match_event_id()

        .ok(EventFactory::new()

            .room(room_id)

            .text_msg("target")

            .event_id(target_id)

            .sender(&ALICE)

            .into_event())

        .mount()

        .await;

    room.server

        .sync_room(

            &room.sync_client,

            JoinedRoomBuilder::new(room_id).add_timeline_event(

                EventFactory::new()

                    .text_msg("reply")

                    .reply_to(target_id)

                    .event_id(reply_id)

                    .sender(&ALICE),

            ),

        )

        .await;

    open_room(&mut room, room_id).await;

    timeline_until(&mut room, |items| {

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

            matches!(item, TimelineItem::Event(event)

            if event.event_id.as_deref() == Some(reply_id)

                && event.reply_to.as_ref().is_some_and(|reply| {

                    matches!(reply.details, macltok::ReplyDetails::Ready { .. })

                }))

        })

    })

    .await;

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

}

#[tokio::test]

async fn opening_a_retained_room_only_probes_its_newest_gap() {

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

    let old_id = event_id!("$probe-old:localhost");

    let live_id = event_id!("$probe-live:localhost");

    let mut room = start_room(room_id).await;

    cache_event_before_open(&room, room_id, old_id, "retained").await;

    room.server

        .sync_room(

            &room.sync_client,

            JoinedRoomBuilder::new(room_id)

                .add_timeline_event(

                    EventFactory::new()

                        .text_msg("live")

                        .event_id(live_id)

                        .sender(&ALICE),

                )

                .set_timeline_limited()

                .set_timeline_prev_batch("sync-edge"),

        )

        .await;

    room.server

        .mock_room_messages()

        .match_from("sync-edge")

        .ok(RoomMessagesResponseTemplate::default()

            .end_token("further")

            .events(vec![history_event(room_id, old_id, "retained")]))

        .mock_once()

        .mount()

        .await;

    open_room(&mut room, room_id).await;

    let report = wait_for_settled_fill(&mut room).await;

    assert_eq!(report.pages, 1);

    assert_eq!(report.stop, None);

    timeline_until(&mut room, |items| {

        gap_in(items).is_none() && has_event(items, old_id) && has_event(items, live_id)

    })

    .await;

    let requests = room.server.received_requests().await.unwrap();

    let messages = requests

        .iter()

        .filter(|request| request.url.path().ends_with("/messages"))

        .count();

    assert_eq!(messages, 1);

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

}