macltok

use std::{sync::Arc, time::Duration};

use matrix_sdk::{

    Client, Room,

    ruma::{api::client::sync::sync_events::v5, events::AnySyncTimelineEvent, room_id, serde::Raw},

    sliding_sync::{SlidingSyncList, SlidingSyncMode},

    test_utils::mocks::MatrixMockServer,

};

use serde_json::{Value, json};

use tokio::sync::{Notify, Semaphore};

async fn saved_position(client: &Client) -> Option<String> {

    let machine = client.olm_machine_for_testing().await;

    let key = format!(

        "sliding_sync_store::mentions-spike::{}::instance",

        client.user_id().unwrap()

    );

    let saved = machine

        .as_ref()

        .unwrap()

        .store()

        .get_custom_value(&key)

        .await

        .unwrap()?;

    serde_json::from_slice::<Value>(&saved).unwrap()["pos"]

        .as_str()

        .map(ToOwned::to_owned)

}

async fn server_with_mention() -> (MatrixMockServer, Client) {

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

    let client = server.client_builder().build().await;

    let mut response = v5::Response::new("position-1".to_owned());

    response.rooms.insert(

        room_id!("!checkpoint:localhost").to_owned(),

        serde_json::from_value(json!({

                "initial": true,

                "required_state": [],

                "timeline": [{

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

                    "event_id": "$checkpoint-mention",

                    "sender": "@sender:localhost",

                    "origin_server_ts": 1,

                    "content": {

                        "msgtype": "m.text",

                        "body": "checkpoint candidate",

                        "m.mentions": { "user_ids": [client.user_id().unwrap()] }

                    }

                }]

        }))

        .unwrap(),

    );

    server.mock_sliding_sync().ok(response).mount().await;

    (server, client)

}

async fn new_sync(client: &Client) -> matrix_sdk::sliding_sync::SlidingSync {

    client

        .sliding_sync("mentions-spike")

        .unwrap()

        .share_pos()

        .add_list(

            SlidingSyncList::builder("mentions")

                .sync_mode(SlidingSyncMode::new_selective().add_range(0..=0)),

        )

        .build()

        .await

        .unwrap()

}

#[tokio::test]

async fn awaited_event_handler_fences_position_even_after_caller_cancellation() {

    let (server, client) = server_with_mention().await;

    let entered = Arc::new(Notify::new());

    let committed = Arc::new(Semaphore::new(0));

    let handler = client.add_event_handler({

        let entered = entered.clone();

        let committed = committed.clone();

        move |event: Raw<AnySyncTimelineEvent>, room: Room| {

            let entered = entered.clone();

            let committed = committed.clone();

            async move {

                assert_eq!(room.room_id().as_str(), "!checkpoint:localhost");

                assert_eq!(

                    event.get_field::<String>("event_id").unwrap().as_deref(),

                    Some("$checkpoint-mention")

                );

                entered.notify_one();

                committed.acquire().await.unwrap().forget();

            }

        }

    });

    let sync = new_sync(&client).await;

    let processing = tokio::spawn(async move { sync.sync_once().await });

    tokio::time::timeout(Duration::from_secs(5), entered.notified())

        .await

        .unwrap();

    assert!(!processing.is_finished());

    assert!(saved_position(&client).await.is_none());

    processing.abort();

    assert!(processing.await.unwrap_err().is_cancelled());

    assert!(saved_position(&client).await.is_none());

    committed.add_permits(1);

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

        while saved_position(&client).await.as_deref() != Some("position-1") {

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

        }

    })

    .await

    .unwrap();

    client.remove_event_handler(handler);

    let restored = new_sync(&client).await;

    tokio::time::timeout(Duration::from_secs(5), restored.sync_once())

        .await

        .unwrap()

        .unwrap();

    let requests: Vec<_> = server

        .received_requests()

        .await

        .unwrap()

        .into_iter()

        .filter_map(|request| {

            let body = serde_json::from_slice::<Value>(&request.body).ok()?;

            let pos = request

                .url

                .query_pairs()

                .find(|(key, _)| key == "pos")

                .map(|(_, value)| value.into_owned());

            Some((body, pos))

        })

        .filter(|(body, _)| body["conn_id"] == "mentions-spike")

        .collect();

    assert_eq!(requests.len(), 2);

    assert!(requests[0].1.is_none());

    assert_eq!(requests[1].1.as_deref(), Some("position-1"));

}

#[tokio::test]

async fn event_handler_error_does_not_prevent_position_persistence() {

    let (_server, client) = server_with_mention().await;

    let handler =

        client.add_event_handler(|_event: Raw<AnySyncTimelineEvent>, _room: Room| async {

            Err::<(), _>("the candidate transaction failed")

        });

    let sync = new_sync(&client).await;

    tokio::time::timeout(Duration::from_secs(5), sync.sync_once())

        .await

        .unwrap()

        .unwrap();

    assert_eq!(saved_position(&client).await.as_deref(), Some("position-1"));

    client.remove_event_handler(handler);

}