macltok

use std::time::Duration;

use macltok::{

    Command, CommandErrorKind, CommandKind, Engine, EngineConfig, Event, LifecycleCommand,

    MediaCommand, MediaSize, MessageContent, RoomCommand, TimelineCommand, TimelineDiff,

    TimelineItem,

};

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 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

        .send(Command::Room(RoomCommand::Open(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).any(|message| message.media.is_some())

    })

    .await;

    let media_id = event_messages(&media_event)

        .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;

    let media_command_id = handle

        .send(Command::Media(MediaCommand::Fetch {

            id: media_id.clone(),

            size: MediaSize::Full,

        }))

        .await

        .unwrap();

    wait_for_media_request(&server).await;

    let timeline_command_id = handle

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

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

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

        }))

        .await

        .unwrap();

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

        loop {

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

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

                    assert_eq!(failure.command, CommandKind::Timeline);

                    assert_eq!(failure.kind, CommandErrorKind::EmptyBody);

                    break;

                }

                Event::CommandSucceeded { id, .. } if id == media_command_id => {

                    panic!("media command completed before the timeline command")

                }

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

                    panic!("media command failed before the timeline command {failure:?}")

                }

                Event::MediaReady {

                    media_id: ready_id, ..

                } if ready_id == media_id => {

                    panic!("media became ready before the timeline command completed")

                }

                _ => {}

            }

        }

    })

    .await

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

    receive_until(&mut events, |event| {

        matches!(

            event,

            Event::CommandSucceeded {

                id,

                command: CommandKind::Media,

            } if id == &media_command_id

        )

    })

    .await;

    handle

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

        .await

        .unwrap();

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

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

}

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");

}

fn event_messages(event: &Event) -> std::vec::IntoIter<&macltok::TimelineMessage> {

    let items = match event {

        Event::TimelineReset { items, .. } => items.iter().collect::<Vec<_>>(),

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

            diffs.iter().flat_map(diff_items).collect::<Vec<_>>()

        }

        _ => Vec::new(),

    };

    items

        .into_iter()

        .filter_map(|item| match item {

            TimelineItem::Message(message) => Some(message.as_ref()),

            _ => None,

        })

        .collect::<Vec<_>>()

        .into_iter()

}

fn diff_items(diff: &TimelineDiff) -> Vec<&TimelineItem> {

    match diff {

        TimelineDiff::Append { values } | TimelineDiff::Reset { values } => values.iter().collect(),

        TimelineDiff::PushFront { value }

        | TimelineDiff::PushBack { value }

        | TimelineDiff::Insert { value, .. }

        | TimelineDiff::Set { value, .. } => vec![value],

        TimelineDiff::Clear

        | TimelineDiff::PopFront

        | TimelineDiff::PopBack

        | TimelineDiff::Remove { .. }

        | TimelineDiff::Truncate { .. } => Vec::new(),

        _ => Vec::new(),

    }

}

async fn receive_until(

    events: &mut macltok::EventStream,

    predicate: impl Fn(&Event) -> bool,

) -> Event {

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

        while let Some(event) = events.recv().await {

            if predicate(&event) {

                return event;

            }

        }

        panic!("event stream closed before the expected event")

    })

    .await

    .expect("expected event did not arrive")

}