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 repair_policy_and_coverage_roundtrip_through_the_inbox() {

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

    let room_id = room_id!("!repair-coverage:localhost");

    let (client, _) = joined_client(&server, room_id).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, mut service, _) = Service::new(

        client.clone(),

        media,

        Some(Ok(store)),

        EngineConfig::default(),

    )

    .await

    .unwrap();

    service.recovery_budget = 0;

    service.latest_scan = false;

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

    let default_policy = HistoryRepairPolicy::default();

    assert_eq!(handle.repair_policy().await.unwrap(), default_policy);

    let stored_policy = HistoryRepairPolicy::Bridge;

    handle

        .set_repair_policy(stored_policy.clone())

        .await

        .unwrap();

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

    let coverage = handle.repair_coverage(room_id.to_owned()).await.unwrap();

    assert!(coverage.last_confirmed_ms.is_some());

    assert!(!coverage.reliable);

    assert_eq!(coverage.duration_ms(), None);

    handle.stop().await;

    service_task.await.unwrap();

    drop(handle);

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

        .await

        .unwrap();

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

    assert_eq!(persisted.repair_policy, stored_policy);

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

    let coverage = sync.repair_coverage.get(room_id).unwrap();

    assert!(coverage.last_confirmed_ms.is_some());

    assert!(coverage.boundary_ms.is_none());

    assert!(!coverage.unresolved);

    assert!(!coverage.reliable);

    drop(reopened);

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

}

#[tokio::test]

async fn repair_limited_sync_freezes_the_last_room_checkpoint() {

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

    let room_id = room_id!("!repair-boundary:localhost");

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

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

        client.clone(),

        EngineConfig::default().max_media_bytes,

        EngineConfig::default().max_media_sources,

    ));

    let (_handle, mut service, _) = Service::new(client, media, None, EngineConfig::default())

        .await

        .unwrap();

    service.sync.checkpoints.insert(

        "mentions-connection".to_owned(),

        Checkpoint {

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

            previous: None,

        },

    );

    service.sync.observed_rooms.insert(room_id.to_owned());

    service

        .sync

        .repair_coverage

        .entry(room_id.to_owned())

        .or_default()

        .last_confirmed_ms = Some(1_000);

    let events = [

        EventFactory::new()

            .room(room_id)

            .text_msg("older event")

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

            .event_id(event_id!("$repair-older:localhost"))

            .into_raw_sync(),

        EventFactory::new()

            .room(room_id)

            .text_msg("latest event")

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

            .event_id(event_id!("$repair-latest:localhost"))

            .into_raw_sync(),

    ];

    let timeline: Vec<_> = events

        .iter()

        .map(|event| serde_json::to_value(event).unwrap())

        .collect();

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

        "limited": true,

        "prev_batch": "repair-gap-token",

        "timeline": timeline,

    }))

    .unwrap();

    let batch = ResponseCommitBatch {

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

        previous_pos: Some("position-before".to_owned()),

        next_pos: "position-after".to_owned(),

        response: Default::default(),

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

    };

    service.batch(batch).await;

    let coverage = service.sync.repair_coverage.get(room_id).unwrap();

    assert_eq!(coverage.last_confirmed_ms, Some(1_000));

    assert!(