macltok

mod support;

use std::time::Duration;

use macltok::{

    Command, Engine, EngineConfig, ErrorKind, Event, MediaSize, MessageContent, TimelineCommand,

};

use matrix_sdk::{

    ruma::{api::client::sync::sync_events::v5, mxc_uri, room_id},

    test_utils::mocks::MatrixMockServer,

};

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

use support::{event_messages, receive_until, shut_down};

use wiremock::ResponseTemplate;

#[tokio::test]

async fn delayed_media_fetch_does_not_block_timeline_command() {

    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 sync_client = client.clone();

    let room_id = room_id!("!nonblocking: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());

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

    receive_until(

        &mut events,

        |event| matches!(event, Event::TimelineReset { timeline_id: reset, .. } if reset.room_id() == room_id),

    )

    .await;

    server

        .sync_room(

            &sync_client,

            JoinedRoomBuilder::new(room_id).add_timeline_event(

                EventFactory::new()

                    .image(

                        "image.jpg".into(),

                        mxc_uri!("mxc://localhost/image").to_owned(),

                    )

                    .sender(&ALICE),

            ),

        )

        .await;

    let media_event = receive_until(&mut events, |event| {

        event_messages(event)

            .into_iter()

            .any(|message| message.media.is_some())

    })

    .await;

    let media_id = event_messages(&media_event)

        .into_iter()

        .find_map(|message| message.media.as_ref().map(|media| media.id.clone()))

        .unwrap();

    let delayed_image = || {

        ResponseTemplate::new(200)

            .set_body_raw(b"binaryjpegfullimagedata", "image/jpeg")

            .set_delay(Duration::from_millis(500))

    };

    server

        .mock_media_download()

        .respond_with(delayed_image())

        .mount()

        .await;

    server

        .mock_authed_media_download()

        .respond_with(delayed_image())

        .mount()

        .await;

    // media no longer travels through the command queue, so driving it on its

    // own task is what puts the request in flight concurrently with the send below

    let media_task = tokio::spawn({

        let handle = handle.clone();

        async move { handle.media(&media_id, MediaSize::Full).await }

    });

    wait_for_media_request(&server).await;

    let error = tokio::time::timeout(

        Duration::from_millis(300),

        handle.execute(Command::Timeline(TimelineCommand::Send {

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

            content: MessageContent::text(String::new()),

        })),

    )

    .await

    .expect("timeline command was blocked by media fetch")

    .expect_err("empty body command unexpectedly succeeded");

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

    let media = tokio::time::timeout(Duration::from_secs(2), media_task)

        .await

        .expect("media fetch did not complete")

        .unwrap()

        .unwrap();

    assert!(!media.bytes.is_empty());

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

}

async fn wait_for_media_request(server: &MatrixMockServer) {

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

        loop {

            if server

                .received_requests()

                .await

                .unwrap()

                .iter()

                .any(|request| request.url.path().ends_with("/download/localhost/image"))

            {

                break;

            }

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

        }

    })

    .await

    .expect("media request did not reach the homeserver");

}