macltok

use std::{

    sync::{

        Arc,

        atomic::{AtomicUsize, Ordering},

    },

    time::Duration,

};

use matrix_sdk::ruma::events::room::MediaSource;

use matrix_sdk::{

    ruma::{event_id, events::Mentions, push::Ruleset, room_id, user_id},

    test_utils::mocks::MatrixMockServer,

};

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

use rusqlite::Connection;

use serde_json::json;

use wiremock::{

    Mock, ResponseTemplate,

    matchers::{method, path, query_param, query_param_is_missing},

};

use super::super::store::{Account, MentionsStore};

use super::*;

use crate::{

    MediaId, MediaKind, StoreConfig, TimelineContent, TimelineItem,

    normalise::{MediaRegistration, MediaSourceInfo},

};

static STORES: AtomicUsize = AtomicUsize::new(0);

const MAX_BYTES: usize = 64 * 1024 * 1024;

fn config() -> StoreConfig {

    let number = STORES.fetch_add(1, Ordering::Relaxed);

    StoreConfig::sqlite(

        std::env::temp_dir().join(format!(

            "macltok-mentions-service-{}-{number}",

            std::process::id()

        )),

        "service commit test passphrase",

    )

}

fn account(client: &Client) -> Account {

    Account {

        homeserver: client.homeserver().to_string(),

        user_id: client.user_id().unwrap().to_owned(),

    }

}

async fn joined_client(

    server: &MatrixMockServer,

    room_id: &ruma::RoomId,

) -> (Client, ruma::OwnedUserId) {

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

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

    let room_id = room_id.to_owned();

    server

        .mock_sync()

        .ok_and_run(&client, {

            let own_user = own_user.clone();

            move |builder| {

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

                builder.add_global_account_data(

                    factory.push_rules(Ruleset::server_default(&own_user)),

                );

                builder.add_joined_room(

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

                );

            }

        })

        .await;

    (client, own_user)

}

fn mention_batch(

    room_id: &ruma::RoomId,

    own_user: &ruma::UserId,

    event: &str,

    previous_pos: Option<&str>,

    next_pos: &str,

) -> ResponseCommitBatch {

    let event = EventFactory::new()

        .room(room_id)

        .text_msg(event)

        .mentions(Mentions::with_user_ids([own_user.to_owned()]))

        .event_id(&ruma::EventId::parse(format!("${event}:localhost")).unwrap())

        .sender(user_id!("@bob:localhost"))

        .into_raw_sync();

    let raw_room = serde_json::from_value(json!({

        "timeline": [serde_json::to_value(event).unwrap()],

    }))

    .unwrap();

    ResponseCommitBatch {

        connection_id: "mentions-connection".to_owned(),

        previous_pos: previous_pos.map(str::to_owned),

        next_pos: next_pos.to_owned(),

        response: Default::default(),

        raw_rooms: [(room_id.to_owned(), raw_room)].into(),

    }

}

fn quiet_batch(previous_pos: &str, next_pos: &str) -> ResponseCommitBatch {

    ResponseCommitBatch {

        connection_id: "mentions-connection".to_owned(),

        previous_pos: Some(previous_pos.to_owned()),

        next_pos: next_pos.to_owned(),

        response: Default::default(),

        raw_rooms: Default::default(),

    }

}

async fn service(client: &Client, store: Option<Result<MentionsStore, Error>>) -> Service {

    let media = Arc::new(MediaService::new(

        client.clone(),

        EngineConfig::default().max_media_bytes,

        EngineConfig::default().max_media_sources,

    ));

    let (_, mut service, _) = Service::new(client.clone(), media, store, EngineConfig::default())

        .await

        .unwrap();

    service.recovery_budget = 0;

    service.latest_scan = false;

    service

}

fn rows(connection: &Connection) -> BTreeMap<String, Vec<u8>> {

    let mut statement = connection

        .prepare("SELECT key, value FROM records WHERE key NOT IN ('!cipher', '!head')")

        .unwrap();

    statement

        .query_map([], |row| Ok((row.get(0)?, row.get(1)?)))

        .unwrap()

        .map(Result::unwrap)

        .collect()

}

fn discontinuities(service: &Service) -> usize {

    service

        .data

        .coverage

        .gaps

        .iter()

        .filter(|gap| gap.reason == MentionGapReason::SyncDiscontinuity)

        .count()

}

#[tokio::test]

async fn failed_write_acks_and_keeps_the_inbox_in_memory_until_a_later_write() {

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

    let room_id = room_id!("!mentions:localhost");

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

    let config = config();

    let store = MentionsStore::open(&config, account(&client), MAX_BYTES)

        .await

        .unwrap();

    let mut service = service(&client, Some(Ok(store))).await;

    let connection = Connection::open(config.mentions_location()).unwrap();

    connection

        .execute_batch(

            "CREATE TRIGGER fail_mentions_inbox_insert BEFORE INSERT ON records WHEN NEW.key NOT IN ('!cipher', '!head') BEGIN SELECT RAISE(ABORT, 'injected inbox write failure'); END;

             CREATE TRIGGER fail_mentions_inbox_update BEFORE UPDATE ON records WHEN OLD.key NOT IN ('!cipher', '!head') BEGIN SELECT RAISE(ABORT, 'injected inbox write failure'); END;",

        )

        .unwrap();

    service

        .batch(mention_batch(

            room_id,

            &own_user,

            "first",

            None,

            "position-1",

        ))

        .await;

    assert!(service.dirty);

    assert_eq!(service.data.entries.len(), 1);

    assert!(

        service

            .data

            .coverage

            .gaps

            .iter()

            .any(|gap| gap.reason == MentionGapReason::StoreFailed)

    );

    assert_eq!(service.updates.borrow().entries.len(), 1);

    assert!(rows(&connection).is_empty());

    connection

        .execute_batch(

            "DROP TRIGGER fail_mentions_inbox_insert;

             DROP TRIGGER fail_mentions_inbox_update;",

        )

        .unwrap();

    service.flush_after = 0;

    service.batch(quiet_batch("position-1", "position-2")).await;

    assert!(!service.dirty);

    drop(service);

    let reopened = MentionsStore::open(&config, account(&client), MAX_BYTES)

        .await

        .unwrap();

    let persisted: Data = reopened.read("inbox").unwrap().unwrap();

    let sync: SyncState = reopened.read("sync").unwrap().unwrap();

    assert_eq!(persisted.entries.len(), 1);

    assert_eq!(sync.inbox_revision, persisted.revision);

    assert_eq!(sync.checkpoints["mentions-connection"].pos, "position-2");

    drop(reopened);

    drop(connection);

    config.delete().await.unwrap();

}

#[tokio::test]

async fn a_quiet_response_writes_only_the_checkpoint() {

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

    let room_id = room_id!("!mentions:localhost");

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

    let config = config();

    let store = MentionsStore::open(&config, account(&client), MAX_BYTES)

        .await

        .unwrap();

    let mut service = service(&client, Some(Ok(store))).await;

    service

        .batch(mention_batch(

            room_id,

            &own_user,

            "first",

            None,

            "position-1",

        ))

        .await;

    let connection = Connection::open(config.mentions_location()).unwrap();

    let before = rows(&connection);

    assert_eq!(before.len(), 2);

    service.batch(quiet_batch("position-1", "position-2")).await;

    let after = rows(&connection);

    assert_eq!(after.len(), 2);

    assert_eq!(

        before

            .iter()

            .filter(|(key, value)| after.get(*key) != Some(value))

            .count(),

        1

    );

    assert_eq!(service.data.revision, 1);

    assert_eq!(

        service.sync.checkpoints["mentions-connection"].pos,

        "position-2"

    );

    drop(service);

    drop(connection);

    config.delete().await.unwrap();

}

#[tokio::test]

async fn a_replay_from_the_previous_checkpoint_is_not_a_discontinuity() {

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

    let room_id = room_id!("!mentions:localhost");

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

    let config = config();

    let store = MentionsStore::open(&config, account(&client), MAX_BYTES)

        .await

        .unwrap();

    let mut first = service(&client, Some(Ok(store))).await;

    first

        .batch(mention_batch(room_id, &own_user, "one", None, "position-1"))

        .await;

    first

        .batch(mention_batch(

            room_id,

            &own_user,

            "two",

            Some("position-1"),

            "position-2",

        ))

        .await;

    drop(first);

    // the sdk never saved position-2, so it asks again from position-1

    let store = MentionsStore::open(&config, account(&client), MAX_BYTES)

        .await

        .unwrap();

    let mut restarted = service(&client, Some(Ok(store))).await;

    restarted

        .batch(mention_batch(

            room_id,

            &own_user,

            "three",

            Some("position-1"),

            "position-3",

        ))

        .await;

    assert_eq!(discontinuities(&restarted), 0);

    assert_eq!(restarted.data.entries.len(), 3);

    assert_eq!(

        restarted.sync.checkpoints["mentions-connection"],

        Checkpoint {

            pos: "position-3".to_owned(),

            previous: Some("position-1".to_owned()),

        }

    );

    restarted

        .batch(quiet_batch("position-3", "position-4"))

        .await;

    assert_eq!(discontinuities(&restarted), 0);

    restarted

        .batch(quiet_batch("position-1", "position-5"))

        .await;

    assert_eq!(discontinuities(&restarted), 1);

    drop(restarted);

    config.delete().await.unwrap();

}

#[tokio::test]

async fn an_unreadable_store_resets_behind_a_gap() {

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

    let room_id = room_id!("!mentions:localhost");

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

    let config = config();

    let store = MentionsStore::open(&config, account(&client), MAX_BYTES)

        .await

        .unwrap();

    let mut first = service(&client, Some(Ok(store))).await;

    first

        .batch(mention_batch(room_id, &own_user, "one", None, "position-1"))

        .await;

    drop(first);

    let connection = Connection::open(config.mentions_location()).unwrap();

    connection

        .execute(

            "UPDATE records SET value = x'00' WHERE key NOT IN ('!cipher', '!head')",

            [],

        )

        .unwrap();

    drop(connection);

    let store = MentionsStore::open(&config, account(&client), MAX_BYTES)

        .await

        .unwrap();

    let restarted = service(&client, Some(Ok(store))).await;

    assert!(restarted.data.entries.is_empty());

    assert!(

        restarted

            .data

            .coverage

            .gaps

            .iter()

            .any(|gap| gap.reason == MentionGapReason::StoreReset)

    );

    drop(restarted);

    config.delete().await.unwrap();

}

#[tokio::test]

async fn service_recovers_notification_history_into_durable_unread_inbox() {

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

    let room_id = room_id!("!mentions:localhost");

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

    let first = EventFactory::new()

        .room(room_id)

        .text_msg("history mention one")

        .mentions(Mentions::with_user_ids([own_user.clone()]))

        .event_id(event_id!("$history-one:localhost"))

        .sender(user_id!("@bob:localhost"))

        .into_raw_sync();

    let second = EventFactory::new()

        .room(room_id)

        .text_msg("history mention two")

        .mentions(Mentions::with_user_ids([own_user]))

        .event_id(event_id!("$history-two:localhost"))

        .sender(user_id!("@bob:localhost"))

        .into_raw_sync();

    let first = serde_json::from_str::<serde_json::Value>(first.json().get()).unwrap();

    let second = serde_json::from_str::<serde_json::Value>(second.json().get()).unwrap();

    Mock::given(method("GET"))

        .and(path("/_matrix/client/v3/notifications"))

        .and(query_param_is_missing("from"))

        .and(query_param("limit", "50"))

        .respond_with(ResponseTemplate::new(200).set_body_json(json!({

            "notifications": [{

                "actions": ["notify"],

                "event": first,

                "read": true,

                "room_id": room_id.as_str(),

                "ts": 1,

            }],

            "next_token": "history-page-two",

        })))

        .mount(server.server())

        .await;

    Mock::given(method("GET"))

        .and(path("/_matrix/client/v3/notifications"))

        .and(query_param("from", "history-page-two"))

        .and(query_param("limit", "50"))

        .respond_with(ResponseTemplate::new(200).set_body_json(json!({

            "notifications": [{

                "actions": ["notify"],

                "event": second,

                "read": true,

                "room_id": room_id.as_str(),

                "ts": 2,

            }],

        })))

        .mount(server.server())

        .await;

    let config = config();

    let store = MentionsStore::open(&config, account(&client), MAX_BYTES)

        .await

        .unwrap();

    let media = Arc::new(MediaService::new(

        client.clone(),

        EngineConfig::default().max_media_bytes,

        EngineConfig::default().max_media_sources,

    ));

    let (handle, service, mut updates) = Service::new(

        client.clone(),

        media,

        Some(Ok(store)),

        EngineConfig::default(),

    )

    .await

    .unwrap();

    let service_task = tokio::spawn(service.run());

    let discovered = tokio::time::timeout(Duration::from_secs(10), async {

        loop {

            let page = updates.borrow_and_update().clone();

            if page.entries.len() == 2 && !page.coverage.recovering {

                break page;

            }

            updates.changed().await.unwrap();

        }

    })

    .await;

    handle.stop().await;

    service_task.await.unwrap();

    let discovered = discovered.expect("notification history should be recovered");

    assert_eq!(discovered.entries.len(), 2);

    assert!(discovered.entries.iter().all(|entry| !entry.read));

    let reopened = MentionsStore::open(&config, account(&client), MAX_BYTES)

        .await

        .unwrap();

    let persisted: Data = reopened.read("inbox").unwrap().unwrap();

    let sync: SyncState = reopened.read("sync").unwrap().unwrap();

    assert_eq!(sync.inbox_revision, persisted.revision);

    assert_eq!(persisted.entries.len(), 2);

    assert!(persisted.entries.values().all(|entry| !entry.read));

    assert!(persisted.latest_cursor.is_none());

    assert!(persisted.history_cursor.is_none());

    drop(reopened);

    config.delete().await.unwrap();

}

fn message_body(item: &TimelineItem) -> Option<&str> {

    match item {

        TimelineItem::Event(event) => match &event.content {

            TimelineContent::Message(message) => Some(&message.body),

            _ => None,

        },

        _ => None,

    }

}

fn media_registration() -> MediaRegistration {

    MediaRegistration {

        id: MediaId::new("test-scope#image".to_owned()),

        kind: MediaKind::Image,

        full: MediaSourceInfo {

            source: MediaSource::Plain(ruma::mxc_uri!("mxc://mentions.example/image").to_owned()),