macltok

mod support;

use std::time::Duration;

use macltok::{

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

    HistoryRepairPhase, HistoryRepairPolicy, RecentHistoryState, Termination, TimelineItem,

};

#[cfg(feature = "mirror")]

use macltok::{Mirror, TimelineId};

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::{event_messages, receive_until, shut_down};

use wiremock::ResponseTemplate;

struct TestRoom {

    server: MatrixMockServer,

    sync_client: Client,

    handle: ClientHandle,

    events: EventStream,

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

}

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

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

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

    TestRoom {

        server,

        sync_client,

        handle,

        events,

        running,

    }

}

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

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

    receive_until(&mut room.events, |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 = receive_until(&mut room.events, |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_report(room: &mut TestRoom, room_id: &RoomId) -> macltok::HistoryRepairReport {

    match receive_until(

        &mut room.events,

        |event| matches!(event, Event::HistoryRepairChanged(report) if &report.room_id == room_id),

    )

    .await

    {

        Event::HistoryRepairChanged(report) => report,

        _ => unreachable!(),

    }

}

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

    room.handle

        .set_history_repair_policy(HistoryRepairPolicy::Bridge)

        .await

        .unwrap();

    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;

    let reset = receive_until(&mut room.events, |event| {

        let items = match event {

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

                items

            }

            _ => return false,

        };

        gap_in(items).is_some()

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

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

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

            })

            && new_ids.iter().all(|id| {

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

                    matches!(item,

                TimelineItem::Event(event) if event.event_id.as_deref() == Some(id))

                })

            })

    })

    .await;

    let Event::TimelineReset { items, .. } = reset else {

        unreachable!();

    };

    assert!(gap_in(&items).is_some());

    assert_eq!(

        wait_for_recent(&mut room, room_id).await,

        RecentHistoryState::Loaded

    );

    let report = wait_for_report(&mut room, room_id).await;

    assert_eq!(report.requested_policy, HistoryRepairPolicy::StartRecent);

    assert_eq!(report.phase, HistoryRepairPhase::Ready);

    assert_eq!(report.pages, 4);

    assert_eq!(report.events, 4);

    assert!(report.recent_available);

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

    let report = room

        .handle

        .start_recent_history(room_id.to_owned())

        .await

        .unwrap();

    assert_eq!(report.pages, 1);

    assert_eq!(report.events, 1);

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

    )

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

    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.as_ref() == Some(&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;

    let reset = receive_until(&mut room.events, |event| {

        matches!(event, Event::TimelineReset { timeline_id, items }

            if timeline_id.room_id() == room_id && gap_in(items).is_some())

    })

    .await;

    let gap = match reset {

        Event::TimelineReset { items, .. } => gap_in(&items).unwrap(),

        _ => unreachable!(),

    };

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

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

    let final_reset = receive_until(&mut room.events, |event| {

        matches!(event, Event::TimelineReset { timeline_id, items }

            if timeline_id.room_id() == room_id

                && items.iter().any(|item| matches!(item, TimelineItem::Event(event)

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

                && items.iter().any(|item| matches!(item, TimelineItem::Event(event)

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

    })

    .await;

    let ids = event_messages(&final_reset)

        .iter()

        .filter_map(|event| event.event_id.as_deref())

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

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

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

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

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

}

#[tokio::test]

async fn failed_automatic_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;

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

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

    let report = wait_for_report(&mut room, room_id).await;

    assert_eq!(report.phase, HistoryRepairPhase::Paused);

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

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

}