macltok

mod support;

use std::time::Duration;

use macltok::{

    ConnectionState, Engine, EngineConfig, EngineError, Event, PackUsage, Retain, RoomMembership,

    Termination, ThreadListEntry, ThreadUnread,

};

use matrix_sdk::{

    ThreadingSupport,

    ruma::{

        EventId, MilliSecondsSinceUnixEpoch, RoomId,

        api::client::sync::sync_events::v5,

        event_id,

        events::{

            AnySyncStateEvent, AnyTimelineEvent, Mentions,

            receipt::{ReceiptThread, ReceiptType},

        },

        room_id,

        serde::Raw,

    },

    test_utils::mocks::MatrixMockServer,

};

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

use serde_json::json;

use tokio::sync::watch;

#[tokio::test]

async fn rooms_changed_invalidates_room_summaries() {

    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!("!state: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());

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

    let rooms = handle.rooms().await;

    let summary = rooms

        .iter()

        .find(|summary| summary.room_id == room_id)

        .expect("synced room was absent from the room summaries");

    assert_eq!(summary.membership, RoomMembership::Joined);

    assert!(!summary.name.is_empty());

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

}

#[tokio::test]

async fn connection_watch_tracks_the_engine_lifecycle() {

    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 config = EngineConfig::default().with_initial_history_limit(0);

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

    let mut connection = handle.connection_state();

    let connection_peer = connection.clone();

    assert_eq!(*connection.borrow(), ConnectionState::Idle);

    assert_eq!(*connection_peer.borrow(), ConnectionState::Idle);

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

    support::receive_until(&mut events, |event| {

        matches!(event, Event::ConnectionChanged(ConnectionState::Connecting))

    })

    .await;

    assert_ne!(*connection.borrow(), ConnectionState::Idle);

    support::receive_until(&mut events, |event| {

        matches!(event, Event::ConnectionChanged(ConnectionState::Online))

    })

    .await;

    wait_for_connection(&mut connection, ConnectionState::Online).await;

    assert_eq!(*connection_peer.borrow(), ConnectionState::Online);

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

    wait_for_connection(&mut connection, ConnectionState::Stopped).await;

    assert_eq!(*connection_peer.borrow(), ConnectionState::Stopped);

}

#[tokio::test]

async fn soft_logout_says_so_and_stops_the_engine() {

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

    server

        .mock_sliding_sync()

        .error_unknown_token(true)

        .mount()

        .await;

    let client = server.client_builder().build().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());

    support::receive_until(&mut events, |event| {

        matches!(

            event,

            Event::SessionAccessRejected {

                soft_logout: Some(_)

            }

        )

    })

    .await;

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

    let termination = running.await.unwrap().unwrap();

    assert!(matches!(termination, Termination::SoftLoggedOut { .. }));

    assert_eq!(termination.retain(), Retain::Store);

}

#[tokio::test]

async fn closing_the_event_stream_stops_an_idle_engine() {

    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 (engine, handle, mut events) = Engine::from_client(

        client,

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

    )

    .await

    .unwrap();

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

    support::receive_until(&mut events, |event| {

        matches!(event, Event::ConnectionChanged(ConnectionState::Online))

    })

    .await;

    drop(events);

    let result = tokio::time::timeout(Duration::from_secs(2), running)

        .await

        .expect("engine did not stop after its event receiver closed")

        .unwrap();

    assert!(matches!(result, Err(EngineError::EventReceiverClosed)));

    drop(handle);

}

async fn wait_for_connection(

    connection: &mut watch::Receiver<ConnectionState>,

    expected: ConnectionState,

) {

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

        loop {

            if *connection.borrow_and_update() == expected {

                return;

            }

            connection

                .changed()

                .await

                .expect("connection watch closed before the expected state");

        }

    })

    .await

    .expect("expected connection state did not arrive");

}

#[tokio::test]

async fn image_packs_follow_room_state_and_account_data() {

    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!("!packs: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.clone(), config).await.unwrap();

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

    let pack: Raw<AnySyncStateEvent> = serde_json::from_value(json!({

        "type": "im.ponies.room_emotes",

        "state_key": "cats",

        "event_id": "$pack",

        "sender": "@example:localhost",

        "origin_server_ts": 1,

        "content": {

            "pack": { "display_name": "cats", "usage": ["emoticon"] },

            "images": {

                "blep": { "url": "mxc://localhost/blep" },

                "broken": { "url": "not an mxc" },

                "sticker": { "url": "mxc://localhost/sticker", "usage": ["sticker"] },

            },

        },

    }))

    .unwrap();

    server

        .mock_sync()

        .ok_and_run(&client, |builder| {

            builder.add_joined_room(JoinedRoomBuilder::new(room_id).add_state_event(pack));

            builder.add_custom_global_account_data(json!({

                "type": "im.ponies.emote_rooms",

                "content": { "rooms": { "!packs:localhost": { "cats": {} } } },

            }));

        })

        .await;

    support::receive_until(&mut events, |event| {

        matches!(event, Event::ImagePacksChanged)

    })

    .await;

    let packs = handle.image_packs().await.unwrap();

    let offered = packs.for_room(Some(room_id));

    assert_eq!(offered.len(), 1, "an enabled room's pack was offered twice");

    let pack = offered[0];

    assert_eq!(pack.id, "room:!packs:localhost:cats");

    assert_eq!(packs.global, vec![pack.clone()]);

    let images: Vec<_> = pack

        .images

        .iter()

        .map(|image| {

            (

                image.shortcode.as_str(),

                image.reaction_key.as_str(),

                image.usage.clone(),

            )

        })

        .collect();

    assert_eq!(

        images,

        [

            ("blep", "mxc://localhost/blep", vec![PackUsage::Emoticon]),

            (

                "sticker",

                "mxc://localhost/sticker",

                vec![PackUsage::Sticker]

            ),

        ]

    );

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

}

/// the sdk counts none of this once threads are on, and a threaded receipt never reaches

/// room info, so only our own recount notices it

#[tokio::test]

async fn thread_unreads_follow_threaded_receipts() {

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

    server

        .mock_sliding_sync()

        .ok(v5::Response::new("1".to_owned()))

        .mount()

        .await;

    let client = server

        .client_builder()

        .on_builder(|builder| {

            builder.with_threading_support(ThreadingSupport::Enabled {

                with_subscriptions: false,

            })

        })

        .build()

        .await;

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

    let own = client.user_id().unwrap().to_owned();

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

    client.event_cache().subscribe().unwrap();

    // without our own member event there's no push context, so no push actions to count

    server

        .sync_room(

            &client,

            JoinedRoomBuilder::new(room_id)

                .add_state_event(EventFactory::new().room(room_id).member(&own)),

        )

        .await;

    let config = EngineConfig::default().with_initial_history_limit(0);

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

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

    let (root, first, second) = (

        event_id!("$root"),

        event_id!("$first"),

        event_id!("$second"),

    );

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

    server

        .sync_room(

            &client,

            JoinedRoomBuilder::new(room_id)

                .add_timeline_event(factory.text_msg("root").event_id(root))

                .add_timeline_event(

                    factory

                        .text_msg("you there?")

                        .in_thread(root, root)

                        .mentions(Mentions::with_user_ids([own.clone()]))

                        .event_id(first),

                )

                .add_timeline_event(

                    factory

                        .text_msg("hello?")

                        .in_thread(root, first)

                        .event_id(second),

                ),

        )

        .await;

    let unread = thread_unread_once(&handle, &mut events, room_id, |unread| {

        unread.is_some_and(|unread| unread.messages == 2)

    })

    .await;

    assert_eq!(unread.mentions, 1);

    let summary = room(&handle, room_id).await;

    assert!(

        summary.unread_mentions >= 1,

        "the thread mention left the room's count"

    );

    server

        .sync_room(

            &client,

            JoinedRoomBuilder::new(room_id).add_receipt(

                factory

                    .read_receipts()

                    .add(

                        first,

                        &own,

                        ReceiptType::Read,

                        ReceiptThread::Thread(root.to_owned()),

                    )

                    .into_event(),

            ),

        )

        .await;

    let unread = thread_unread_once(&handle, &mut events, room_id, |unread| {

        unread.is_some_and(|unread| unread.messages == 1)

    })

    .await;

    assert_eq!(unread.mentions, 0);

    server

        .sync_room(

            &client,

            JoinedRoomBuilder::new(room_id).add_receipt(

                factory

                    .read_receipts()

                    .add(second, &own, ReceiptType::Read, ReceiptThread::Unthreaded)

                    .into_event(),

            ),

        )

        .await;

    thread_unread_once(&handle, &mut events, room_id, |unread| unread.is_none()).await;

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

}

async fn room(handle: &macltok::ClientHandle, room_id: &RoomId) -> macltok::RoomSummary {

    handle

        .rooms()

        .await

        .into_iter()

        .find(|summary| summary.room_id == room_id)

        .expect("synced room was absent from the room summaries")

}

/// waits out `RoomsChanged` until the root's counts satisfy `done`

async fn thread_unread_once(

    handle: &macltok::ClientHandle,

    events: &mut macltok::EventStream,

    room_id: &RoomId,

    done: impl Fn(Option<&ThreadUnread>) -> bool,

) -> ThreadUnread {

    let root = event_id!("$root");

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

        loop {

            let summary = room(handle, room_id).await;

            let unread = summary.thread_unreads.get(root);

            if done(unread) {

                return unread.copied().unwrap_or_default();

            }

            support::receive_until(events, |event| matches!(event, Event::RoomsChanged)).await;

        }

    })

    .await

    .expect("thread unreads never settled")

}

/// continuwuity 26.9 bundles a thread's latest reply as bare content, so nothing dates the

/// page's count, and the cache handing back replies the server already counted counted them

/// again

#[tokio::test]

async fn a_thread_list_page_keeps_the_servers_count() {

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

    server

        .mock_sliding_sync()

        .ok(v5::Response::new("1".to_owned()))

        .mount()

        .await;

    let client = server

        .client_builder()

        .on_builder(|builder| {

            builder.with_threading_support(ThreadingSupport::Enabled {

                with_subscriptions: false,

            })

        })

        .build()

        .await;

    let room_id = room_id!("!thread-list:localhost");

    let own = client.user_id().unwrap().to_owned();

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

    client.event_cache().subscribe().unwrap();

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

    let (root, first, second, third) = (

        event_id!("$list-root"),

        event_id!("$list-first"),

        event_id!("$list-second"),

        event_id!("$list-third"),

    );

    let reply = |event_id, previous, body| {

        factory

            .text_msg(body)

            .in_thread(root, previous)

            .event_id(event_id)

    };

    server

        .sync_room(

            &client,

            JoinedRoomBuilder::new(room_id)

                .add_state_event(EventFactory::new().room(room_id).member(&own))

                .add_timeline_event(factory.text_msg("root").event_id(root))

                .add_timeline_event(reply(first, root, "one"))

                .add_timeline_event(reply(second, first, "two")),

        )

        .await;

    let listed: Raw<AnyTimelineEvent> = serde_json::from_value(json!({

        "type": "m.room.message",

        "event_id": root,

        "room_id": room_id,

        "sender": &*ALICE,

        "origin_server_ts": 1,

        "content": { "msgtype": "m.text", "body": "root" },

        "unsigned": { "m.relations": { "m.thread": {

            "latest_event": { "msgtype": "m.text", "body": "two" },

            "count": 2,

            "current_user_participated": false,

        } } },

    }))

    .unwrap();

    server

        .mock_room_threads()

        .ok(vec![listed], None)

        .mount()

        .await;

    let config = EngineConfig::default().with_initial_history_limit(0);

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

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

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

    handle

        .paginate_thread_list(room_id.to_owned(), 20)

        .await

        .unwrap();

    tokio::time::sleep(Duration::from_millis(5)).await;

    server

        .sync_room(

            &client,

            JoinedRoomBuilder::new(room_id)

                .add_timeline_event(reply(first, root, "one"))

                .add_timeline_event(reply(second, first, "two"))

                .add_timeline_event(

                    reply(third, second, "three").server_ts(MilliSecondsSinceUnixEpoch::now()),

                ),

        )

        .await;

    let changed = tokio::time::timeout(

        Duration::from_secs(10),

        support::receive_until(&mut events, |event| {

            matches!(event, Event::ThreadListChanged { upserts, .. }

                if upserts.iter().any(|entry| entry.summary.latest_event_id.as_deref() == Some(third)))

        }),

    )

    .await

    .expect("the new reply never reached the thread list");

    let Event::ThreadListChanged { upserts, .. } = changed else {

        unreachable!()

    };

    let entry = upserts.last().unwrap();

    assert_eq!(

        entry.summary.reply_count, 3,

        "replies the server had counted were counted again"

    );

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

}

/// with continuwuity's undated latest, the page has to date each thread from the replies the

/// cache already holds, or it sorts by the root

#[tokio::test]

async fn a_thread_list_page_dates_threads_from_cached_replies() {

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

    server

        .mock_sliding_sync()

        .ok(v5::Response::new("1".to_owned()))

        .mount()

        .await;

    let client = server

        .client_builder()

        .on_builder(|builder| {

            builder.with_threading_support(ThreadingSupport::Enabled {

                with_subscriptions: false,

            })

        })

        .build()

        .await;

    let room_id = room_id!("!thread-dates:localhost");

    let own = client.user_id().unwrap().to_owned();

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

    client.event_cache().subscribe().unwrap();

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

    let (early, first, second, late) = (

        event_id!("$dates-early"),

        event_id!("$dates-first"),

        event_id!("$dates-second"),

        event_id!("$dates-late"),

    );

    server

        .sync_room(

            &client,

            JoinedRoomBuilder::new(room_id)

                .add_state_event(EventFactory::new().room(room_id).member(&own))

                .add_timeline_event(factory.text_msg("early").event_id(early).server_ts(1_000))

                .add_timeline_event(

                    factory

                        .text_msg("one")

                        .in_thread(early, early)

                        .event_id(first)

                        .server_ts(5_000),

                )

                .add_timeline_event(factory.text_msg("late").event_id(late).server_ts(7_000))

                .add_timeline_event(

                    factory

                        .text_msg("two")

                        .in_thread(early, first)

                        .event_id(second)

                        .server_ts(9_000),

                ),

        )

        .await;

    let listed = |root: &EventId, timestamp: u64, count: u64| -> Raw<AnyTimelineEvent> {

        serde_json::from_value(json!({

            "type": "m.room.message",

            "event_id": root,

            "room_id": room_id,

            "sender": &*ALICE,

            "origin_server_ts": timestamp,

            "content": { "msgtype": "m.text", "body": "root" },

            "unsigned": { "m.relations": { "m.thread": {

                "latest_event": { "msgtype": "m.text", "body": "latest" },

                "count": count,

                "current_user_participated": false,

            } } },

        }))

        .unwrap()

    };

    server

        .mock_room_threads()

        .ok(vec![listed(late, 7_000, 1), listed(early, 1_000, 2)], None)

        .mount()

        .await;

    let config = EngineConfig::default().with_initial_history_limit(0);

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

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

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

    handle

        .paginate_thread_list(room_id.to_owned(), 20)

        .await

        .unwrap();

    let changed = support::receive_until(&mut events, |event| {

        matches!(event, Event::ThreadListChanged { .. })

    })

    .await;

    let Event::ThreadListChanged { upserts, .. } = changed else {

        unreachable!()

    };

    let mut entries = macltok::imbl::Vector::new();

    for entry in upserts {

        ThreadListEntry::upsert(&mut entries, entry);

    }

    let order: Vec<_> = entries

        .iter()

        .map(|entry| (entry.root_event_id.as_str(), entry.latest_activity_ms))

        .collect();

    assert_eq!(

        order,

        [(early.as_str(), 9_000), (late.as_str(), 7_000)],

        "a thread was dated by its root rather than its newest cached reply"

    );

    let dated = &entries[0];

    assert_eq!(dated.summary.latest_event_id.as_deref(), Some(second));

    assert_eq!(dated.summary.latest_event_timestamp_ms, Some(9_000));

    assert_eq!(dated.summary.reply_count, 2);

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

}