macltok

mod support;

use std::time::Duration;

use macltok::{

    ClientHandle, Engine, EngineConfig, EngineError, Event, EventStream, HistoryContinuity,

    HistoryRepairBudget, HistoryRepairPhase, HistoryRepairPolicy, HistoryRepairStop,

    HistoryRepairStrategy, 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 open_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, 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;

    TestRoom {

        server,

        sync_client,

        handle,

        events,

        running,

    }

}

async fn sync_cached_event(room: &mut 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;

    receive_until(&mut room.events, |event| {

        event_messages(event)

            .into_iter()

            .any(|event| event.event_id.as_deref() == Some(id))

    })

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

}

fn report_budget(max_pages: u16) -> HistoryRepairBudget {

    HistoryRepairBudget {

        max_pages,

        ..HistoryRepairBudget::default()

    }

}

#[tokio::test]

async fn recent_repair_starts_without_a_cursor_and_keeps_its_end_for_pagination() {

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

    let stale_id = event_id!("$stale:localhost");

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

    let older_id = event_id!("$older:localhost");

    let mut room = open_room(room_id).await;

    sync_cached_event(&mut room, room_id, stale_id, "cached before repair").await;

    room.server

        .mock_room_messages()

        .ok(RoomMessagesResponseTemplate::default()

            .end_token("recent-prev")

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

        .mock_once()

        .mount()

        .await;

    let report = room

        .handle

        .history_repair(

            room_id.to_owned(),

            Some(HistoryRepairPolicy::StartRecent),

            HistoryRepairBudget::default(),

        )

        .await

        .unwrap();

    assert_eq!(report.strategy, HistoryRepairStrategy::Recent);

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

    assert!(report.recent_available);

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

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

            && event_messages(event)

                .into_iter()

                .any(|event| event.event_id.as_deref() == Some(fresh_id))

    })

    .await;

    let reset_ids = event_messages(&reset)

        .into_iter()

        .filter_map(|event| event.event_id.as_ref().map(ToString::to_string))

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

    assert_eq!(reset_ids, vec![fresh_id.to_string()]);

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

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

    room.server

        .mock_room_messages()

        .match_from("recent-prev")

        .ok(

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

                room_id,

                older_id,

                "older page",

            )]),

        )

        .mock_once()

        .mount()

        .await;

    room.handle

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

        .await

        .unwrap();

    receive_until(&mut room.events, |event| {

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

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

        })

    })

    .await;

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

}

#[cfg(feature = "mirror")]

#[tokio::test]

async fn repair_ack_waits_until_the_mirror_has_applied_the_reset_and_report() {

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

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

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

    let sync_client = client.clone();

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

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

    let stale_id = event_id!("$mirror-stale:localhost");

    server

        .sync_room(

            &sync_client,

            JoinedRoomBuilder::new(room_id).add_timeline_event(

                EventFactory::new()

                    .text_msg("cached event")

                    .event_id(stale_id)

                    .sender(&ALICE),

            ),

        )

        .await;

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

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

        loop {

            let has_cached_event = mirror_handle

                .snapshot()

                .timelines

                .get(&timeline_id)

                .is_some_and(|timeline| {

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

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

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

                    })

                });

            if has_cached_event {

                break;

            }

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

        }

    })

    .await

    .expect("the cached timeline event did not reach the mirror");

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

    server

        .mock_room_messages()

        .ok(

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

                room_id,

                fresh_id,

                "fresh event",

            )]),

        )

        .mock_once()

        .mount()

        .await;

    let report = handle

        .history_repair(

            room_id.to_owned(),

            Some(HistoryRepairPolicy::StartRecent),

            HistoryRepairBudget::default(),

        )

        .await

        .unwrap();

    let state = mirror_handle.snapshot();

    assert_eq!(state.repairs.get(room_id), Some(&report));

    let timeline = state

        .timelines

        .get(&timeline_id)

        .expect("the repaired timeline was not applied before acknowledgement");

    let ids = timeline

        .items

        .iter()

        .filter_map(|item| match item {

            TimelineItem::Event(event) => event.event_id.as_ref().map(ToString::to_string),

            _ => None,

        })

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

    assert_eq!(ids, vec![fresh_id.to_string()]);

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

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

    mirror_running.await.unwrap();

}

#[tokio::test]

async fn failed_recent_fetch_leaves_the_existing_timeline_in_place() {

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

    let stale_id = event_id!("$kept:localhost");

    let mut room = open_room(room_id).await;

    sync_cached_event(&mut room, room_id, stale_id, "keep this event").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;

    let report = room

        .handle

        .history_repair(

            room_id.to_owned(),

            Some(HistoryRepairPolicy::StartRecent),

            HistoryRepairBudget::default(),

        )

        .await

        .unwrap();

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

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

    let mut reset_after_failure = false;

    while let Ok(event) = room.events.try_recv() {

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

    }

    assert!(!reset_after_failure);

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

}

#[tokio::test]

async fn cancelling_a_delayed_bridge_page_preserves_the_recent_stage() {

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

    let anchor_id = event_id!("$cancel-anchor:localhost");

    let recent_id = event_id!("$cancel-recent:localhost");

    let mut room = open_room(room_id).await;

    sync_cached_event(&mut room, room_id, anchor_id, "cached anchor").await;

    room.server

        .mock_room_messages()

        .ok(RoomMessagesResponseTemplate::default()

            .end_token("cancel-prev")

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

        .mock_once()

        .mount()

        .await;

    let initial = room

        .handle

        .history_repair(

            room_id.to_owned(),

            Some(HistoryRepairPolicy::Bridge),

            report_budget(1),

        )

        .await

        .unwrap();

    assert_eq!(initial.stop, Some(HistoryRepairStop::Budget));

    assert!(initial.recent_available);

    receive_until(&mut room.events, |event| {

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

            && event_messages(event)

                .into_iter()

                .any(|event| event.event_id.as_deref() == Some(recent_id))

    })

    .await;

    room.server

        .mock_room_messages()

        .match_from("cancel-prev")

        .ok(RoomMessagesResponseTemplate::default()

            .with_delay(Duration::from_secs(5))

            .events(vec![history_event(room_id, anchor_id, "cached anchor")]))

        .mock_once()

        .mount()

        .await;

    let repair_handle = room.handle.clone();

    let repair = tokio::spawn(async move {

        repair_handle

            .continue_history_repair(room_id.to_owned(), HistoryRepairBudget::default())

            .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 == "cancel-prev")

                });

            if requested {

                break;

            }

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

        }

    })

    .await

    .expect("bridge request did not start");

    room.handle

        .cancel_history_repair(room_id.to_owned())

        .await

        .unwrap();

    let report = repair.await.unwrap().unwrap();

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

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

    assert!(report.recent_available);

    assert_eq!(report.pages, 1);

    let mut reset_after_cancel = false;

    while let Ok(event) = room.events.try_recv() {

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

    }

    assert!(!reset_after_cancel);

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

}

#[tokio::test]

async fn bridge_reaches_the_cached_anchor_deduplicates_overlap_and_can_continue_after_budget() {

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

    let anchor_id = event_id!("$anchor:localhost");

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

    let recent_id = event_id!("$recent:localhost");

    let latest_id = event_id!("$latest:localhost");

    let mut room = open_room(room_id).await;

    sync_cached_event(&mut room, room_id, anchor_id, "cached anchor").await;

    room.server

        .mock_room_messages()

        .ok(RoomMessagesResponseTemplate::default()

            .end_token("recent-prev")

            .events(vec![

                history_event(room_id, recent_id, "recent"),

                history_event(room_id, middle_id, "middle"),

            ]))

        .mock_once()

        .mount()

        .await;

    let initial = room

        .handle

        .history_repair(

            room_id.to_owned(),

            Some(HistoryRepairPolicy::Bridge),

            report_budget(1),

        )

        .await

        .unwrap();

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

    assert_eq!(initial.stop, Some(HistoryRepairStop::Budget));

    assert!(initial.recent_available);

    wiremock::Mock::given(|request: &wiremock::Request| {

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

            && !request.url.query_pairs().any(|(key, _)| key == "from")

    })

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

        "start": "head-start",

        "chunk": [

            history_event(room_id, latest_id, "latest"),

            history_event(room_id, recent_id, "recent overlap"),

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

        ],

        "state": [],

    })))

    .up_to_n_times(1)

    .expect(1)

    .mount(room.server.server())

    .await;

    room.server

        .mock_room_messages()

        .match_from("recent-prev")

        .ok(RoomMessagesResponseTemplate::default()

            .end_token("older-prev")

            .events(vec![

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

                history_event(room_id, anchor_id, "cached anchor overlap"),

            ]))

        .mock_once()

        .mount()

        .await;

    let report = room

        .handle

        .continue_history_repair(room_id.to_owned(), HistoryRepairBudget::default())

        .await

        .unwrap();

    assert_eq!(report.strategy, HistoryRepairStrategy::Bridge);

    assert_eq!(report.phase, HistoryRepairPhase::Ready, "{report:?}");

    assert_eq!(report.continuity, HistoryContinuity::Verified);

    assert_eq!(report.pages, 3);

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

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

            && event_messages(event)

                .into_iter()

                .any(|event| event.event_id.as_deref() == Some(latest_id))

    })

    .await;

    let reset_ids = event_messages(&reset)

        .into_iter()

        .filter_map(|event| event.event_id.as_ref().map(ToString::to_string))

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

    assert_eq!(

        reset_ids,

        vec![

            anchor_id.to_string(),

            middle_id.to_string(),

            recent_id.to_string(),

            latest_id.to_string(),

        ]

    );

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

}

#[tokio::test]

async fn a_recent_window_containing_the_cached_anchor_overlaps_under_every_policy() {

    let cases = [

        (

            room_id!("!policy-recent:localhost"),

            event_id!("$policy-recent-anchor:localhost"),

            HistoryRepairPolicy::StartRecent,

        ),

        (

            room_id!("!policy-bridge:localhost"),

            event_id!("$policy-bridge-anchor:localhost"),

            HistoryRepairPolicy::Bridge,

        ),

        (

            room_id!("!policy-short-gap:localhost"),

            event_id!("$policy-short-anchor:localhost"),

            HistoryRepairPolicy::BridgeShortGaps { threshold_ms: 0 },

        ),

    ];

    for (room_id, anchor_id, policy) in cases {

        let mut room = open_room(room_id).await;

        sync_cached_event(&mut room, room_id, anchor_id, "cached anchor").await;

        room.server

            .mock_room_messages()

            .ok(

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

                    room_id,

                    anchor_id,

                    "same anchor in recent history",

                )]),

            )

            .mock_once()

            .mount()

            .await;

        let report = room

            .handle

            .history_repair(

                room_id.to_owned(),

                Some(policy.clone()),

                HistoryRepairBudget::default(),

            )

            .await

            .unwrap();

        assert_eq!(report.requested_policy, policy);

        assert_eq!(report.strategy, HistoryRepairStrategy::Overlap);

        assert_eq!(report.continuity, HistoryContinuity::Verified);

        assert_eq!(report.pages, 1);

        assert!(report.recent_available);

        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;

    }

}

#[tokio::test]

async fn short_gap_policy_stays_recent_when_coverage_is_unknown() {

    let room_id = room_id!("!unknown-coverage:localhost");

    let anchor_id = event_id!("$unknown-anchor:localhost");

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

    let mut room = open_room(room_id).await;

    sync_cached_event(&mut room, room_id, anchor_id, "cached anchor").await;

    room.server

        .mock_room_messages()

        .ok(RoomMessagesResponseTemplate::default()

            .end_token("recent-prev")

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

        .mock_once()

        .mount()

        .await;

    let report = room

        .handle

        .history_repair(

            room_id.to_owned(),

            Some(HistoryRepairPolicy::BridgeShortGaps {

                threshold_ms: 60_000,

            }),

            HistoryRepairBudget::default(),

        )

        .await

        .unwrap();

    assert_eq!(report.coverage_duration_ms, None);

    assert!(!report.threshold_evaluated);

    assert_eq!(report.strategy, HistoryRepairStrategy::Recent);

    assert_eq!(report.continuity, HistoryContinuity::Unresolved);

    assert_eq!(report.pages, 1);

    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;

}