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)

}

#[tokio::test]

async fn failed_batch_keeps_inbox_and_checkpoint_at_the_same_commit_boundary() {

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

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

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

    let event = EventFactory::new()

        .room(room_id)

        .text_msg("commit boundary")

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

        .event_id(event_id!("$commit-boundary:localhost"))

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

        .into_raw_sync();

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

        "initial": true,

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

    }))

    .unwrap();

    let batch = ResponseCommitBatch {

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

        previous_pos: None,

        next_pos: "mentions-position-1".to_owned(),

        response: Default::default(),

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

    };

    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, updates) =

        Service::new(client, media, Some(store), EngineConfig::default())

            .await

            .unwrap();

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

    assert!(service.batch(batch.clone()).await.is_err());

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

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

    assert!(service.data.checkpoints.is_empty());

    assert_eq!(updates.borrow().revision, 0);

    assert!(updates.borrow().entries.is_empty());

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

        .await

        .unwrap();

    assert_eq!(reopened.revision(), 0);

    assert!(reopened.read::<Data>("inbox").unwrap().is_none());

    drop(reopened);

    connection

        .execute_batch(

            "DROP TRIGGER fail_mentions_inbox_insert;

             DROP TRIGGER fail_mentions_inbox_update;",

        )

        .unwrap();

    drop(connection);

    service.batch(batch.clone()).await.unwrap();

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

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

    assert_eq!(

        service.data.checkpoints.get("mentions-connection"),

        Some(&batch.next_pos)

    );

    assert_eq!(updates.borrow().revision, 1);

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

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

        .await

        .unwrap();

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

    assert_eq!(reopened.revision(), 1);

    assert_eq!(persisted.revision, 1);

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

    assert_eq!(

        persisted.checkpoints.get("mentions-connection"),

        Some(&batch.next_pos)

    );

    drop(reopened);

    service.batch(batch).await.unwrap();

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

    assert_eq!(service.store.as_ref().unwrap().revision(), 1);

    drop(service);

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

}

#[test]

fn repair_fields_default_when_loading_an_older_inbox() {

    let mut inbox = serde_json::to_value(Data::default()).unwrap();

    inbox.as_object_mut().unwrap().remove("repair_policy");

    inbox.as_object_mut().unwrap().remove("repair_coverage");

    let restored: Data = serde_json::from_value(inbox).unwrap();

    assert_eq!(restored.repair_policy, HistoryRepairPolicy::default());

    assert!(restored.repair_coverage.is_empty());

}

#[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(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 coverage = persisted.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.data.checkpoints.insert(

        "mentions-connection".to_owned(),

        "position-before".to_owned(),

    );

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

    service

        .data

        .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.unwrap();

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

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

    assert!(

        coverage

            .boundary_ms

            .is_some_and(|boundary| boundary > 1_000)

    );

    assert!(coverage.unresolved);

    assert!(!coverage.reliable);

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

    assert_eq!(

        coverage.latest_anchor.as_deref(),

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

    );

}

#[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(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();

    assert_eq!(reopened.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()),

            mimetype: Some("image/png".to_owned()),

            declared_size: Some(1234),

        },

        thumbnail: None,

    }

}

#[tokio::test]

async fn live_context_survives_capture_then_edit_and_redaction() {

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

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

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

    let mention = EventFactory::new()

        .room(room_id)

        .text_msg("direct mention")

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

        .event_id(event_id!("$live-mention:localhost"))

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

        .into_raw_sync();

    let mut mention: serde_json::Value = serde_json::from_str(mention.json().get()).unwrap();

    mention["origin_server_ts"] = json!(10);

    let following = EventFactory::new()

        .room(room_id)

        .text_msg("following")

        .event_id(event_id!("$live-following:localhost"))

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

        .into_raw_sync();

    let mut following: serde_json::Value = serde_json::from_str(following.json().get()).unwrap();

    following["origin_server_ts"] = json!(11);

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

        client.clone(),

        EngineConfig::default().max_media_bytes,

        EngineConfig::default().max_media_sources,

    ));