macltok

mod support;

use std::sync::{

    Arc,

    atomic::{AtomicUsize, Ordering},

};

use std::time::Duration;

use macltok::{Engine, EngineConfig, Event, EventStream, MentionsPage};

use matrix_sdk::{

    ruma::{RoomVersionId, device_id, event_id, room_id, user_id},

    test_utils::mocks::MatrixMockServer,

};

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

use serde_json::{Value, json};

use support::shut_down;

use wiremock::{

    Mock, Request, Respond, ResponseTemplate,

    matchers::{method, path},

};

const SLIDING_SYNC_PATH: &str = "/_matrix/client/unstable/org.matrix.simplified_msc3575/sync";

struct EncryptedSlidingFeed {

    room_id: String,

    event: Value,

    requests: Arc<AtomicUsize>,

}

impl Clone for EncryptedSlidingFeed {

    fn clone(&self) -> Self {

        Self {

            room_id: self.room_id.clone(),

            event: self.event.clone(),

            requests: self.requests.clone(),

        }

    }

}

impl EncryptedSlidingFeed {

    async fn mount(&self, server: &MatrixMockServer) {

        Mock::given(method("POST"))

            .and(path(SLIDING_SYNC_PATH))

            .respond_with(self.clone())

            .mount(server.server())

            .await;

    }

}

impl Respond for EncryptedSlidingFeed {

    fn respond(&self, _request: &Request) -> ResponseTemplate {

        let request = self.requests.fetch_add(1, Ordering::AcqRel);

        ResponseTemplate::new(200)

            .set_delay(Duration::from_millis(20))

            .set_body_json(json!({

                "txn_id": null,

                "pos": format!("late-decryption-{request}"),

                "lists": {},

                "rooms": {

                    (self.room_id.clone()): {

                        "initial": request == 0,

                        "required_state": [],

                        "timeline": [self.event.clone()],

                    }

                },

                "extensions": {},

            }))

    }

}

async fn mentions_page(

    events: &mut EventStream,

    room_id: &ruma::RoomId,

    predicate: impl Fn(&MentionsPage) -> bool,

) -> MentionsPage {

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

        while let Some(event) = events.recv().await {

            match event {

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

                    panic!("the unopened room produced a timeline reset");

                }

                Event::MentionsChanged(update) if predicate(&update.snapshot) => {

                    return update.snapshot;

                }

                _ => {}

            }

        }

        panic!("event stream closed before the expected mentions update");

    })

    .await

    .expect("expected mentions update did not arrive")

}

async fn wait_for_requests(feed: &EncryptedSlidingFeed, minimum: usize) {

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

        while feed.requests.load(Ordering::Acquire) < minimum {

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

        }

    })

    .await

    .expect("the sliding sync feed did not receive another request");

}

#[tokio::test]

async fn late_room_key_updates_the_production_mentions_inbox_once() {

    let server = MatrixMockServer::new().await;

    server.mock_crypto_endpoints_preset().await;

    let alice_id = user_id!("@alice:localhost");

    let bob_id = user_id!("@bob:localhost");

    let room_id = room_id!("!mentions-late-decryption:localhost");

    let event_id = event_id!("$late-production-mention");

    let alice = server

        .client_builder_for_crypto_end_to_end(alice_id, device_id!("ALICEDEVICE"))

        .build()

        .await;

    let bob = server

        .client_builder_for_crypto_end_to_end(bob_id, device_id!("BOBDEVICE"))

        .build()

        .await;

    bob.event_cache().subscribe().unwrap();

    server.exchange_e2ee_identities(&alice, &bob).await;

    let factory = EventFactory::new().room(room_id).sender(alice_id);

    let joined = JoinedRoomBuilder::new(room_id)

        .add_state_event(factory.create(alice_id, RoomVersionId::V1))

        .add_state_event(factory.room_encryption())

        .add_state_event(factory.member(alice_id))

        .add_state_event(factory.member(bob_id));

    server.sync_room(&alice, joined.clone()).await;

    server.sync_room(&bob, joined).await;

    let room = alice.get_room(room_id).unwrap();

    let (encrypted_event, send) = server.mock_room_send().ok_with_capture(event_id, alice_id);

    let _send = send.mock_once().mount_as_scoped().await;

    let (_keys, room_key) = server.mock_capture_put_to_device(alice_id).await;

    server

        .mock_get_members()

        .ok(vec![

            factory.member(alice_id).into_raw(),

            factory.member(bob_id).into_raw(),

        ])

        .mock_once()

        .mount()

        .await;

    room.send_raw(

        "m.room.message",

        json!({

            "msgtype": "m.text",

            "body": "a private mention",

            "m.mentions": { "user_ids": [bob_id] }

        }),

    )

    .await

    .unwrap();

    let encrypted_event = encrypted_event.await.unwrap();

    let room_key = room_key.await;

    let encrypted_event: Value = serde_json::from_str(encrypted_event.json().get()).unwrap();

    let feed = EncryptedSlidingFeed {

        room_id: room_id.to_string(),

        event: encrypted_event,

        requests: Arc::new(AtomicUsize::new(0)),

    };

    feed.mount(&server).await;

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

        bob.clone(),

        EngineConfig::default().with_initial_history_limit(0),

    )

    .await

    .unwrap();

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

    let unresolved = mentions_page(&mut events, room_id, |page| {

        page.coverage.unresolved_candidates == 1

    })

    .await;

    assert!(unresolved.entries.is_empty());

    server

        .mock_sync()

        .ok_and_run(&bob, |sync| {

            sync.add_to_device_event(room_key.deserialize_as().unwrap());

        })

        .await;

    let decrypted = mentions_page(&mut events, room_id, |page| {

        page.entries.len() == 1 && page.entries[0].key.event_id == event_id

    })

    .await;

    assert_eq!(decrypted.entries.len(), 1);

    assert_eq!(decrypted.entries[0].key.event_id, event_id);

    assert_eq!(decrypted.coverage.unresolved_candidates, 0);

    let replay_start = feed.requests.load(Ordering::Acquire);

    wait_for_requests(&feed, replay_start + 3).await;

    let replayed = handle.mentions_page(None, 50).await.unwrap();

    assert_eq!(replayed.entries.len(), 1);

    assert_eq!(replayed.entries[0].key.event_id, event_id);

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

        if let Event::TimelineReset { timeline_id, .. } = event {

            assert_ne!(timeline_id.room_id(), room_id);

        }

    }

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

}