macltok

use matrix_sdk::{

    deserialized_responses::TimelineEvent as SdkTimelineEvent,

    ruma::{events::AnySyncTimelineEvent, room_id},

    test_utils::mocks::MatrixMockServer,

};

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

use serde_json::json;

use super::*;

async fn joined(server: &MatrixMockServer, room_id: &ruma::RoomId) -> Client {

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

    client.event_cache().subscribe().unwrap();

    let own_user = client.user_id().unwrap().to_owned();

    let room_id = room_id.to_owned();

    server

        .mock_sync()

        .ok_and_run(&client, move |builder| {

            let factory = EventFactory::new().room(&room_id);

            builder.add_joined_room(

                JoinedRoomBuilder::new(&room_id).add_state_event(factory.member(&own_user)),

            );

        })

        .await;

    client

}

fn service(client: &Client) -> Service {

    let media = Arc::new(MediaService::new(client.clone(), 1 << 20, 64));

    let (events, _) = EventSender::detached(16);

    Service::new(

        client.clone(),

        media,

        None,

        events,

        &EngineConfig::default(),

    )

    .1

}

fn message(id: &str, ts: u64, body: &str) -> Value {

    json!({

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

        "event_id": id,

        "sender": "@a:x",

        "origin_server_ts": ts,

        "content": { "msgtype": "m.text", "body": body },

    })

}

fn page(events: &[Value], end: Option<&str>) -> Result<Messages, Error> {

    Ok(Messages {

        start: "start".to_owned(),

        end: end.map(ToOwned::to_owned),

        chunk: events

            .iter()

            .map(|value| {

                SdkTimelineEvent::from_plaintext(

                    Raw::<AnySyncTimelineEvent>::from_json_string(value.to_string()).unwrap(),

                )

            })

            .collect(),

        state: Vec::new(),

    })

}

/// live events indexed this run used to count as overlap, so the first head page always

/// ended the pass and an offline gap past one page was never filled

#[tokio::test]

async fn a_head_pass_walks_past_what_this_run_indexed_live() {

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

    let room_id = room_id!("!head:x");

    let client = joined(&server, room_id).await;

    let mut service = service(&client);

    let room = room_id.to_owned();

    service.rooms.insert(

        room.to_string(),

        RoomProgress {

            newest: Some(100),

            started: true,

            ..RoomProgress::default()

        },

    );

    service.baselines.insert(room.to_string(), 100);

    service.observe(&room, &message("$old", 100, "old"));

    service.observe(&room, &message("$live", 1_000, "live"));

    service.backfilled(

        room.clone(),

        true,

        None,

        page(

            &[message("$live", 1_000, "live"), message("$gap", 500, "gap")],

            Some("t1"),

        ),

    );

    assert_eq!(

        service.head_tokens.get(&room).map(String::as_str),

        Some("t1")

    );

    assert_eq!(service.heads.front(), Some(&room));

    service.backfilled(

        room.clone(),

        true,

        Some("t1".to_owned()),

        page(&[message("$old", 100, "old")], Some("t2")),

    );

    assert!(!service.head_tokens.contains_key(&room));

    assert!(service.index.get("$gap").is_some());

}

/// the generic channel is bounded, and a room whose only update was lost was never followed

#[tokio::test]

async fn a_lagged_generic_channel_follows_every_joined_room() {

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

    let room_id = room_id!("!lag:x");

    let client = joined(&server, room_id).await;

    let mut service = service(&client);

    assert!(

        service

            .generic(Err(broadcast::error::RecvError::Lagged(3)))

            .await

    );

    assert!(service.subscriptions.contains_key(room_id));

}

/// a closed stream left its entry behind, and the room was never followed again

#[tokio::test]

async fn a_closed_room_stream_can_be_followed_again() {

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

    let room_id = room_id!("!closed:x");

    let client = joined(&server, room_id).await;

    let mut service = service(&client);

    service.subscribe(room_id.to_owned()).await;

    assert!(service.subscriptions.contains_key(room_id));

    service

        .room_update(room_id.to_owned(), RoomUpdate::Closed)

        .await;

    assert!(!service.subscriptions.contains_key(room_id));

    assert!(service.generic(Ok(room_id.to_owned())).await);

    assert!(service.subscriptions.contains_key(room_id));

}