macltok

mod support;

use std::time::Duration;

use macltok::{

    AttachmentUpload, ClientHandle, Engine, EngineConfig, ErrorKind, EventStream, MessageContent,

};

use matrix_sdk::{

    Client,

    ruma::{OwnedTransactionId, RoomId, api::client::sync::sync_events::v5, event_id, room_id},

    test_utils::mocks::MatrixMockServer,

};

#[tokio::test]

async fn cancel_send_uses_transaction_id_from_public_timeline() {

    let (server, client, handle, mut events, running) = setup().await;

    let room_id = room_id!("!queue-controls:localhost");

    client.send_queue().set_enabled(false).await;

    let transaction_id = queue_message(&handle, &mut events, room_id, "discard me").await;

    assert!(

        handle

            .pending_send_ids(room_id)

            .await

            .unwrap()

            .contains(&transaction_id)

    );

    handle

        .cancel_send(room_id.to_owned(), transaction_id.clone())

        .await

        .unwrap();

    let (echoes, _) = client

        .get_room(room_id)

        .unwrap()

        .send_queue()

        .subscribe()

        .await

        .unwrap();

    assert!(

        !echoes

            .iter()

            .any(|echo| echo.transaction_id == transaction_id)

    );

    let error = handle

        .cancel_send(room_id.to_owned(), transaction_id)

        .await

        .unwrap_err();

    assert_eq!(error.kind, ErrorKind::Invalid);

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

    drop(server);

}

#[tokio::test]

async fn cancel_queued_attachment_by_pending_send_id() {

    let (server, client, handle, mut events, running) = setup().await;

    let room_id = room_id!("!queue-controls:localhost");

    client.send_queue().set_enabled(false).await;

    handle

        .upload(AttachmentUpload::new(

            room_id.to_owned(),

            "note.txt",

            "text/plain",

            b"attachment".to_vec(),

        ))

        .await

        .unwrap();

    let pending = handle.pending_send_ids(room_id).await.unwrap();

    assert_eq!(pending.len(), 1);

    handle

        .cancel_send(room_id.to_owned(), pending[0].clone())

        .await

        .unwrap();

    assert!(handle.pending_send_ids(room_id).await.unwrap().is_empty());

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

    drop(server);

}

#[tokio::test]

async fn retry_send_wakes_paused_queue() {

    let (server, client, handle, mut events, running) = setup().await;

    let room_id = room_id!("!queue-controls:localhost");

    client

        .get_room(room_id)

        .unwrap()

        .send_queue()

        .set_enabled(false);

    server

        .mock_room_send()

        .body_matches_partial_json(serde_json::json!({ "body": "retry me" }))

        .ok(event_id!("$retried:localhost"))

        .expect(1)

        .mount()

        .await;

    let transaction_id = queue_message(&handle, &mut events, room_id, "retry me").await;

    handle

        .retry_send(room_id.to_owned(), transaction_id)

        .await

        .unwrap();

    assert!(client.get_room(room_id).unwrap().send_queue().is_enabled());

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

        loop {

            let (echoes, _) = client

                .get_room(room_id)

                .unwrap()

                .send_queue()

                .subscribe()

                .await

                .unwrap();

            if echoes.is_empty() {

                break;

            }

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

        }

    })

    .await

    .expect("retried message was not sent");

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

    drop(server);

}

async fn setup() -> (

    MatrixMockServer,

    Client,

    ClientHandle,

    EventStream,

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

) {

    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!("!queue-controls:localhost");

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

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

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

        client.clone(),

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

    )

    .await

    .unwrap();

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

    let (opened, _) = tokio::join!(handle.open_room(room_id.to_owned()), async {

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

            matches!(event, macltok::Event::TimelineReset { .. })

        })

        .await

    });

    opened.unwrap();

    (server, client, handle, events, running)

}

/// sends a message and waits for its local echo, since only the timeline reveals the transaction id

async fn queue_message(

    handle: &ClientHandle,

    events: &mut EventStream,

    room_id: &RoomId,

    body: &str,

) -> OwnedTransactionId {

    let send = handle.send_message(room_id.to_owned().into(), MessageContent::text(body));

    let wait_for_echo = async {

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

            loop {

                let event = events.recv().await.expect("event stream closed");

                let transaction = support::event_messages(&event)

                    .into_iter()

                    .find(|event| {

                        support::message(event).is_some_and(|message| message.body == body)

                    })

                    .and_then(|event| event.transaction_id.clone());

                if let Some(transaction) = transaction {

                    return transaction;

                }

            }

        })

        .await

        .expect("local echo did not appear")

    };

    let (result, transaction_id) = tokio::join!(send, wait_for_echo);

    result.unwrap();

    transaction_id

}