macltok

use std::time::Duration;

use macltok::{

    ClientHandle, Command, CommandErrorKind, CommandId, CommandKind, Engine, EngineConfig, Event,

    EventStream, LifecycleCommand, MessageContent, RoomCommand, TimelineCommand, TimelineItem,

};

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)

    );

    let id = handle

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

        .await

        .unwrap();

    expect_success(&mut events, id, CommandKind::Room).await;

    let (echoes, _) = client

        .get_room(room_id)

        .unwrap()

        .send_queue()

        .subscribe()

        .await

        .unwrap();

    assert!(

        !echoes

            .iter()

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

    );

    let id = handle

        .cancel_send(room_id.to_owned(), transaction_id)

        .await

        .unwrap();

    let failure = receive_until(

        &mut events,

        |event| matches!(event, Event::CommandFailed(failure) if failure.id == id),

    )

    .await;

    assert!(

        matches!(failure, Event::CommandFailed(failure) if failure.kind == CommandErrorKind::Other)

    );

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

    let id = handle

        .upload_media(

            room_id.to_owned().into(),

            "note.txt".to_owned(),

            "text/plain".to_owned(),

            b"attachment".to_vec(),

            None,

        )

        .await

        .unwrap();

    expect_success(&mut events, id, CommandKind::Media).await;

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

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

    let id = handle

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

        .await

        .unwrap();

    expect_success(&mut events, id, CommandKind::Room).await;

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

    stop(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.send_queue().set_enabled(false).await;

    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;

    let id = handle

        .retry_send(room_id.to_owned(), transaction_id)

        .await

        .unwrap();

    expect_success(&mut events, id, CommandKind::Room).await;

    assert!(client.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");

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

    drop(server);

}

async fn setup() -> (

    MatrixMockServer,

    Client,

    ClientHandle,

    EventStream,

    tokio::task::JoinHandle<Result<(), 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, events) = Engine::from_client(

        client.clone(),

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

    )

    .await

    .unwrap();

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

    let mut events = events;

    let id = handle

        .send(Command::Room(RoomCommand::Open(room_id.to_owned())))

        .await

        .unwrap();

    expect_success(&mut events, id, CommandKind::Room).await;

    (server, client, handle, events, running)

}

async fn queue_message(

    handle: &ClientHandle,

    events: &mut EventStream,

    room_id: &RoomId,

    body: &str,

) -> OwnedTransactionId {

    let id = handle

        .send(Command::Timeline(TimelineCommand::Send {

            timeline_id: room_id.to_owned().into(),

            content: MessageContent::text(body),

        }))

        .await

        .unwrap();

    let mut succeeded = false;

    let mut items = Vec::new();

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

        loop {

            match events.recv().await.expect("event stream closed") {

                Event::TimelineChanged { diffs, .. } => {

                    for diff in diffs {

                        diff.try_apply(&mut items).unwrap();

                    }

                    let transaction = items.iter().find_map(|item| match item {

                        TimelineItem::Message(message) if message.body == body => {

                            message.transaction_id.clone()

                        }

                        _ => None,

                    });

                    if let Some(transaction) = transaction {

                        if !succeeded {

                            expect_success(events, id, CommandKind::Timeline).await;

                        }

                        return transaction;

                    }

                }

                Event::CommandSucceeded {

                    id: completed,

                    command: CommandKind::Timeline,

                } if completed == id => succeeded = true,

                Event::CommandFailed(failure) if failure.id == id => {

                    panic!("send failed {failure:?}")

                }

                _ => {}

            }

        }

    })

    .await

    .expect("local echo did not appear")

}

async fn expect_success(events: &mut EventStream, id: CommandId, command: CommandKind) {

    receive_until(events, |event| {

        matches!(event, Event::CommandSucceeded { id: completed, command: kind } if *completed == id && *kind == command)

    })

    .await;

}

async fn receive_until(events: &mut EventStream, predicate: impl Fn(&Event) -> bool) -> Event {

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

        loop {

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

            if predicate(&event) {

                return event;

            }

        }

    })

    .await

    .expect("expected event did not arrive")

}

async fn stop(

    handle: ClientHandle,

    events: &mut EventStream,

    running: tokio::task::JoinHandle<Result<(), macltok::EngineError>>,

) {

    let id = handle

        .send(Command::Lifecycle(LifecycleCommand::Shutdown))

        .await

        .unwrap();

    expect_success(events, id, CommandKind::Lifecycle).await;

    running.await.unwrap().unwrap();

}