atagen

macltok

public
Star0
Watch0
13da982fe7bb952b0fb1ae3efb8bd3ca4f39b204
3
ssh://brewery@pls.lobotomise.me:33/atagen/macltok.git
src/mentions/service_tests.rsRust780 lines25.7 KB
1
use std::{
2
    sync::{
3
        Arc,
4
        atomic::{AtomicUsize, Ordering},
5
    },
6
    time::Duration,
7
};
8
9
use matrix_sdk::ruma::events::room::MediaSource;
10
use matrix_sdk::{
11
    ruma::{event_id, events::Mentions, push::Ruleset, room_id, user_id},
12
    test_utils::mocks::MatrixMockServer,
13
};
14
use matrix_sdk_test::{JoinedRoomBuilder, event_factory::EventFactory};
15
use rusqlite::Connection;
16
use serde_json::json;
17
use wiremock::{
18
    Mock, ResponseTemplate,
19
    matchers::{method, path, query_param, query_param_is_missing},
20
};
21
22
use super::super::store::{Account, MentionsStore};
23
use super::*;
24
use crate::{
25
    MediaId, MediaKind, StoreConfig, TimelineContent, TimelineItem,
26
    normalise::{MediaRegistration, MediaSourceInfo},
27
};
28
29
static STORES: AtomicUsize = AtomicUsize::new(0);
30
const MAX_BYTES: usize = 64 * 1024 * 1024;
31
32
fn config() -> StoreConfig {
33
    let number = STORES.fetch_add(1, Ordering::Relaxed);
34
    StoreConfig::sqlite(
35
        std::env::temp_dir().join(format!(
36
            "macltok-mentions-service-{}-{number}",
37
            std::process::id()
38
        )),
39
        "service commit test passphrase",
40
    )
41
}
42
43
fn account(client: &Client) -> Account {
44
    Account {
45
        homeserver: client.homeserver().to_string(),
46
        user_id: client.user_id().unwrap().to_owned(),
47
    }
48
}
49
50
async fn joined_client(
51
    server: &MatrixMockServer,
52
    room_id: &ruma::RoomId,
53
) -> (Client, ruma::OwnedUserId) {
54
    let client = server.client_builder().build().await;
55
    let own_user = client.user_id().unwrap().to_owned();
56
    let room_id = room_id.to_owned();
57
    server
58
        .mock_sync()
59
        .ok_and_run(&client, {
60
            let own_user = own_user.clone();
61
            move |builder| {
62
                let factory = EventFactory::new().room(&room_id);
63
                builder.add_global_account_data(
64
                    factory.push_rules(Ruleset::server_default(&own_user)),
65
                );
66
                builder.add_joined_room(
67
                    JoinedRoomBuilder::new(&room_id).add_state_event(factory.member(&own_user)),
68
                );
69
            }
70
        })
71
        .await;
72
    (client, own_user)
73
}
74
75
fn mention_batch(
76
    room_id: &ruma::RoomId,
77
    own_user: &ruma::UserId,
78
    event: &str,
79
    previous_pos: Option<&str>,
80
    next_pos: &str,
81
) -> ResponseCommitBatch {
82
    let event = EventFactory::new()
83
        .room(room_id)
84
        .text_msg(event)
85
        .mentions(Mentions::with_user_ids([own_user.to_owned()]))
86
        .event_id(&ruma::EventId::parse(format!("${event}:localhost")).unwrap())
87
        .sender(user_id!("@bob:localhost"))
88
        .into_raw_sync();
89
    let raw_room = serde_json::from_value(json!({
90
        "timeline": [serde_json::to_value(event).unwrap()],
91
    }))
92
    .unwrap();
93
    ResponseCommitBatch {
94
        connection_id: "mentions-connection".to_owned(),
95
        previous_pos: previous_pos.map(str::to_owned),
96
        next_pos: next_pos.to_owned(),
97
        response: Default::default(),
98
        raw_rooms: [(room_id.to_owned(), raw_room)].into(),
99
    }
100
}
101
102
fn quiet_batch(previous_pos: &str, next_pos: &str) -> ResponseCommitBatch {
103
    ResponseCommitBatch {
104
        connection_id: "mentions-connection".to_owned(),
105
        previous_pos: Some(previous_pos.to_owned()),
106
        next_pos: next_pos.to_owned(),
107
        response: Default::default(),
108
        raw_rooms: Default::default(),
109
    }
110
}
111
112
async fn service(client: &Client, store: Option<Result<MentionsStore, Unopened>>) -> Service {
113
    let media = Arc::new(MediaService::new(
114
        client.clone(),
115
        EngineConfig::default().max_media_bytes,
116
        EngineConfig::default().max_context_media,
117
    ));
118
    let (_, mut service, _) = Service::new(client.clone(), media, store, EngineConfig::default())
119
        .await
120
        .unwrap();
121
    service.recovery_budget = 0;
122
    service.latest_scan = false;
123
    service
124
}
125
126
fn rows(connection: &Connection) -> BTreeMap<String, Vec<u8>> {
127
    let mut statement = connection
128
        .prepare("SELECT key, value FROM records WHERE key NOT IN ('!cipher', '!head')")
129
        .unwrap();
130
    statement
131
        .query_map([], |row| Ok((row.get(0)?, row.get(1)?)))
132
        .unwrap()
133
        .map(Result::unwrap)
134
        .collect()
135
}
136
137
fn discontinuities(service: &Service) -> usize {
138
    service
139
        .data
140
        .coverage
141
        .gaps
142
        .iter()
143
        .filter(|gap| gap.reason == MentionGapReason::SyncDiscontinuity)
144
        .count()
145
}
146
147
#[tokio::test]
148
async fn failed_write_acks_and_keeps_the_inbox_in_memory_until_a_later_write() {
149
    let server = MatrixMockServer::new().await;
150
    let room_id = room_id!("!mentions:localhost");
151
    let (client, own_user) = joined_client(&server, room_id).await;
152
    let config = config();
153
    let store = MentionsStore::open(&config, account(&client), MAX_BYTES)
154
        .await
155
        .unwrap();
156
    let mut service = service(&client, Some(Ok(store))).await;
157
158
    let connection = Connection::open(config.mentions_location()).unwrap();
159
    connection
160
        .execute_batch(
161
            "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;
162
             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;",
163
        )
164
        .unwrap();
165
166
    service
167
        .batch(mention_batch(
168
            room_id,
169
            &own_user,
170
            "first",
171
            None,
172
            "position-1",
173
        ))
174
        .await;
175
    assert!(service.dirty);
176
    assert_eq!(service.data.entries.len(), 1);
177
    assert!(
178
        service
179
            .data
180
            .coverage
181
            .gaps
182
            .iter()
183
            .any(|gap| gap.reason == MentionGapReason::StoreFailed)
184
    );
185
    assert_eq!(service.updates.borrow().entries.len(), 1);
186
    assert!(rows(&connection).is_empty());
187
188
    connection
189
        .execute_batch(
190
            "DROP TRIGGER fail_mentions_inbox_insert;
191
             DROP TRIGGER fail_mentions_inbox_update;",
192
        )
193
        .unwrap();
194
    service.flush_after = 0;
195
    service.batch(quiet_batch("position-1", "position-2")).await;
196
    assert!(!service.dirty);
197
    drop(service);
198
199
    let reopened = MentionsStore::open(&config, account(&client), MAX_BYTES)
200
        .await
201
        .unwrap();
202
    let persisted: Data = reopened.read("inbox").unwrap().unwrap();
203
    let sync: SyncState = reopened.read("sync").unwrap().unwrap();
204
    assert_eq!(persisted.entries.len(), 1);
205
    assert_eq!(sync.inbox_revision, persisted.revision);
206
    assert_eq!(sync.checkpoints["mentions-connection"].pos, "position-2");
207
    drop(reopened);
208
    drop(connection);
209
    config.delete().await.unwrap();
210
}
211
212
#[tokio::test]
213
async fn a_quiet_response_writes_only_the_checkpoint() {
214
    let server = MatrixMockServer::new().await;
215
    let room_id = room_id!("!mentions:localhost");
216
    let (client, own_user) = joined_client(&server, room_id).await;
217
    let config = config();
218
    let store = MentionsStore::open(&config, account(&client), MAX_BYTES)
219
        .await
220
        .unwrap();
221
    let mut service = service(&client, Some(Ok(store))).await;
222
    service
223
        .batch(mention_batch(
224
            room_id,
225
            &own_user,
226
            "first",
227
            None,
228
            "position-1",
229
        ))
230
        .await;
231
    let connection = Connection::open(config.mentions_location()).unwrap();
232
    let before = rows(&connection);
233
    assert_eq!(before.len(), 3);
234
235
    service.batch(quiet_batch("position-1", "position-2")).await;
236
    let after = rows(&connection);
237
    assert_eq!(after.len(), 3);
238
    assert_eq!(
239
        before
240
            .iter()
241
            .filter(|(key, value)| after.get(*key) != Some(value))
242
            .count(),
243
        1
244
    );
245
    assert_eq!(service.data.revision, 1);
246
    assert_eq!(
247
        service.sync.checkpoints["mentions-connection"].pos,
248
        "position-2"
249
    );
250
    drop(service);
251
    drop(connection);
252
    config.delete().await.unwrap();
253
}
254
255
#[tokio::test]
256
async fn a_failed_command_write_leaves_a_store_that_still_opens() {
257
    let server = MatrixMockServer::new().await;
258
    let room_id = room_id!("!mentions:localhost");
259
    let (client, own_user) = joined_client(&server, room_id).await;
260
    let config = config();
261
    let store = MentionsStore::open(&config, account(&client), MAX_BYTES)
262
        .await
263
        .unwrap();
264
    let mut first = service(&client, Some(Ok(store))).await;
265
    first
266
        .batch(mention_batch(
267
            room_id,
268
            &own_user,
269
            "first",
270
            None,
271
            "position-1",
272
        ))
273
        .await;
274
    let key = first.data.entries.values().next().unwrap().key.clone();
275
276
    let connection = Connection::open(config.mentions_location()).unwrap();
277
    connection
278
        .execute_batch(
279
            "CREATE TRIGGER fail_mentions_update BEFORE UPDATE ON records WHEN OLD.key NOT IN ('!cipher', '!head') BEGIN SELECT RAISE(ABORT, 'injected write failure'); END;",
280
        )
281
        .unwrap();
282
    assert!(
283
        first
284
            .command(MentionsCommand::SetRead {
285
                keys: vec![key],
286
                read: true,
287
            })
288
            .await
289
            .is_err()
290
    );
291
    connection
292
        .execute_batch("DROP TRIGGER fail_mentions_update;")
293
        .unwrap();
294
    drop(connection);
295
    first.batch(quiet_batch("position-1", "position-2")).await;
296
    drop(first);
297
298
    let store = MentionsStore::open(&config, account(&client), MAX_BYTES)
299
        .await
300
        .unwrap();
301
    let restarted = service(&client, Some(Ok(store))).await;
302
    assert_eq!(restarted.data.entries.len(), 1);
303
    assert!(
304
        restarted
305
            .data
306
            .coverage
307
            .gaps
308
            .iter()
309
            .all(|gap| gap.reason != MentionGapReason::StoreReset)
310
    );
311
    assert_eq!(
312
        restarted.sync.checkpoints["mentions-connection"].pos,
313
        "position-2"
314
    );
315
    drop(restarted);
316
    config.delete().await.unwrap();
317
}
318
319
#[tokio::test]
320
async fn a_replay_from_the_previous_checkpoint_is_not_a_discontinuity() {
321
    let server = MatrixMockServer::new().await;
322
    let room_id = room_id!("!mentions:localhost");
323
    let (client, own_user) = joined_client(&server, room_id).await;
324
    let config = config();
325
    let store = MentionsStore::open(&config, account(&client), MAX_BYTES)
326
        .await
327
        .unwrap();
328
    let mut first = service(&client, Some(Ok(store))).await;
329
    first
330
        .batch(mention_batch(room_id, &own_user, "one", None, "position-1"))
331
        .await;
332
    first
333
        .batch(mention_batch(
334
            room_id,
335
            &own_user,
336
            "two",
337
            Some("position-1"),
338
            "position-2",
339
        ))
340
        .await;
341
    drop(first);
342
343
    // the sdk never saved position-2, so it asks again from position-1
344
    let store = MentionsStore::open(&config, account(&client), MAX_BYTES)
345
        .await
346
        .unwrap();
347
    let mut restarted = service(&client, Some(Ok(store))).await;
348
    restarted
349
        .batch(mention_batch(
350
            room_id,
351
            &own_user,
352
            "three",
353
            Some("position-1"),
354
            "position-3",
355
        ))
356
        .await;
357
    assert_eq!(discontinuities(&restarted), 0);
358
    assert_eq!(restarted.data.entries.len(), 3);
359
    assert_eq!(
360
        restarted.sync.checkpoints["mentions-connection"],
361
        Checkpoint {
362
            pos: "position-3".to_owned(),
363
            previous: Some("position-1".to_owned()),
364
        }
365
    );
366
367
    restarted
368
        .batch(quiet_batch("position-3", "position-4"))
369
        .await;
370
    assert_eq!(discontinuities(&restarted), 0);
371
    restarted
372
        .batch(quiet_batch("position-1", "position-5"))
373
        .await;
374
    assert_eq!(discontinuities(&restarted), 1);
375
    drop(restarted);
376
    config.delete().await.unwrap();
377
}
378
379
#[tokio::test]
380
async fn an_unreadable_store_resets_behind_a_gap() {
381
    let server = MatrixMockServer::new().await;
382
    let room_id = room_id!("!mentions:localhost");
383
    let (client, own_user) = joined_client(&server, room_id).await;
384
    let config = config();
385
    let store = MentionsStore::open(&config, account(&client), MAX_BYTES)
386
        .await
387
        .unwrap();
388
    let mut first = service(&client, Some(Ok(store))).await;
389
    first
390
        .batch(mention_batch(room_id, &own_user, "one", None, "position-1"))
391
        .await;
392
    drop(first);
393
    let connection = Connection::open(config.mentions_location()).unwrap();
394
    connection
395
        .execute(
396
            "UPDATE records SET value = x'00' WHERE key NOT IN ('!cipher', '!head')",
397
            [],
398
        )
399
        .unwrap();
400
    drop(connection);
401
402
    let store = MentionsStore::open(&config, account(&client), MAX_BYTES)
403
        .await
404
        .unwrap();
405
    let restarted = service(&client, Some(Ok(store))).await;
406
    assert!(restarted.data.entries.is_empty());
407
    assert!(
408
        restarted
409
            .data
410
            .coverage
411
            .gaps
412
            .iter()
413
            .any(|gap| gap.reason == MentionGapReason::StoreReset)
414
    );
415
    drop(restarted);
416
    config.delete().await.unwrap();
417
}
418
419
#[tokio::test]
420
async fn service_recovers_notification_history_into_durable_unread_inbox() {
421
    let server = MatrixMockServer::new().await;
422
    let room_id = room_id!("!mentions:localhost");
423
    let (client, own_user) = joined_client(&server, room_id).await;
424
    let first = EventFactory::new()
425
        .room(room_id)
426
        .text_msg("history mention one")
427
        .mentions(Mentions::with_user_ids([own_user.clone()]))
428
        .event_id(event_id!("$history-one:localhost"))
429
        .sender(user_id!("@bob:localhost"))
430
        .into_raw_sync();
431
    let second = EventFactory::new()
432
        .room(room_id)
433
        .text_msg("history mention two")
434
        .mentions(Mentions::with_user_ids([own_user]))
435
        .event_id(event_id!("$history-two:localhost"))
436
        .sender(user_id!("@bob:localhost"))
437
        .into_raw_sync();
438
    let first = serde_json::from_str::<serde_json::Value>(first.json().get()).unwrap();
439
    let second = serde_json::from_str::<serde_json::Value>(second.json().get()).unwrap();
440
    Mock::given(method("GET"))
441
        .and(path("/_matrix/client/v3/notifications"))
442
        .and(query_param_is_missing("from"))
443
        .and(query_param("limit", "50"))
444
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({
445
            "notifications": [{
446
                "actions": ["notify"],
447
                "event": first,
448
                "read": true,
449
                "room_id": room_id.as_str(),
450
                "ts": 1,
451
            }],
452
            "next_token": "history-page-two",
453
        })))
454
        .mount(server.server())
455
        .await;
456
    Mock::given(method("GET"))
457
        .and(path("/_matrix/client/v3/notifications"))
458
        .and(query_param("from", "history-page-two"))
459
        .and(query_param("limit", "50"))
460
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({
461
            "notifications": [{
462
                "actions": ["notify"],
463
                "event": second,
464
                "read": true,
465
                "room_id": room_id.as_str(),
466
                "ts": 2,
467
            }],
468
        })))
469
        .mount(server.server())
470
        .await;
471
472
    let config = config();
473
    let store = MentionsStore::open(&config, account(&client), MAX_BYTES)
474
        .await
475
        .unwrap();
476
    let media = Arc::new(MediaService::new(
477
        client.clone(),
478
        EngineConfig::default().max_media_bytes,
479
        EngineConfig::default().max_context_media,
480
    ));
481
    let (handle, service, mut updates) = Service::new(
482
        client.clone(),
483
        media,
484
        Some(Ok(store)),
485
        EngineConfig::default(),
486
    )
487
    .await
488
    .unwrap();
489
    let service_task = tokio::spawn(service.run());
490
491
    let discovered = tokio::time::timeout(Duration::from_secs(10), async {
492
        loop {
493
            let page = updates.borrow_and_update().clone();
494
            if page.entries.len() == 2 && !page.coverage.recovering {
495
                break page;
496
            }
497
            updates.changed().await.unwrap();
498
        }
499
    })
500
    .await;
501
    handle.stop().await;
502
    service_task.await.unwrap();
503
    let discovered = discovered.expect("notification history should be recovered");
504
    assert_eq!(discovered.entries.len(), 2);
505
    assert!(discovered.entries.iter().all(|entry| !entry.read));
506
507
    let reopened = MentionsStore::open(&config, account(&client), MAX_BYTES)
508
        .await
509
        .unwrap();
510
    let persisted: Data = reopened.read("inbox").unwrap().unwrap();
511
    let sync: SyncState = reopened.read("sync").unwrap().unwrap();
512
    assert_eq!(sync.inbox_revision, persisted.revision);
513
    assert_eq!(persisted.entries.len(), 2);
514
    assert!(persisted.entries.values().all(|entry| !entry.read));
515
    assert!(persisted.latest_cursor.is_none());
516
    assert!(persisted.history_cursor.is_none());
517
    drop(reopened);
518
519
    config.delete().await.unwrap();
520
}
521
522
fn message_body(item: &TimelineItem) -> Option<&str> {
523
    match item {
524
        TimelineItem::Event(event) => match &event.content {
525
            TimelineContent::Message(message) => Some(&message.body),
526
            _ => None,
527
        },
528
        _ => None,
529
    }
530
}
531
532
fn media_registration() -> MediaRegistration {
533
    MediaRegistration {
534
        id: MediaId::new("test-scope#image".to_owned()),
535
        kind: MediaKind::Image,
536
        full: MediaSourceInfo {
537
            source: MediaSource::Plain(ruma::mxc_uri!("mxc://mentions.example/image").to_owned()),
538
            mimetype: Some("image/png".to_owned()),
539
            declared_size: Some(1234),
540
        },
541
        thumbnail: None,
542
    }
543
}
544
545
#[tokio::test]
546
async fn live_context_survives_capture_then_edit_and_redaction() {
547
    let server = MatrixMockServer::new().await;
548
    let room_id = room_id!("!mentions:localhost");
549
    let (client, own_user) = joined_client(&server, room_id).await;
550
    let mention = EventFactory::new()
551
        .room(room_id)
552
        .text_msg("direct mention")
553
        .mentions(Mentions::with_user_ids([own_user]))
554
        .event_id(event_id!("$live-mention:localhost"))
555
        .sender(user_id!("@bob:localhost"))
556
        .into_raw_sync();
557
    let mut mention: serde_json::Value = serde_json::from_str(mention.json().get()).unwrap();
558
    mention["origin_server_ts"] = json!(10);
559
    let following = EventFactory::new()
560
        .room(room_id)
561
        .text_msg("following")
562
        .event_id(event_id!("$live-following:localhost"))
563
        .sender(user_id!("@bob:localhost"))
564
        .into_raw_sync();
565
    let mut following: serde_json::Value = serde_json::from_str(following.json().get()).unwrap();
566
    following["origin_server_ts"] = json!(11);
567
568
    let media = Arc::new(MediaService::new(
569
        client.clone(),
570
        EngineConfig::default().max_media_bytes,
571
        EngineConfig::default().max_context_media,
572
    ));
573
    let (_handle, mut service, _updates) =
574
        Service::new(client, media, None, EngineConfig::default())
575
            .await
576
            .unwrap();
577
    let room_id = room_id.to_owned();
578
    let mention_key = MentionKey {
579
        room_id: room_id.clone(),
580
        event_id: event_id!("$live-mention:localhost").to_owned(),
581
    };
582
    let following_id = format!("{room_id}\n$live-following:localhost");
583
    let mut data = Data::default();
584
585
    service.observe(&mut data, &room_id, mention).await;
586
    service.observe(&mut data, &room_id, following).await;
587
588
    let mention_identity = id(&mention_key);
589
    let entry = data.entries.get(&mention_identity).unwrap();
590
    let revision = entry.revision;
591
    let target = entry.event.clone();
592
    service.captured(
593
        &mut data,
594
        mention_identity.clone(),
595
        revision,
596
        Ok((
597
            crate::EventContext {
598
                events_before: Vec::new(),
599
                event: target,
600
                events_after: Vec::new(),
601
                start_token: None,
602
                end_token: None,
603
            },
604
            Vec::new(),
605
            Vec::new(),
606
        )),
607
    );
608
609
    assert_eq!(
610
        data.entries[&mention_identity].capture,
611
        MentionCaptureState::Partial
612
    );
613
    assert!(data.events.contains_key(&following_id));
614
    let (saved, _) = data.context(&mention_key).unwrap();
615
    let context = saved.context.unwrap();
616
    assert_eq!(context.events_after.len(), 1);
617
    assert_eq!(message_body(&context.events_after[0]), Some("following"));
618
619
    let edit = json!({
620
        "content": {
621
            "body": "* edited following",
622
            "m.new_content": {"body": "edited following", "msgtype": "m.text"},
623
            "m.relates_to": {"event_id": "$live-following:localhost", "rel_type": "m.replace"},
624
            "msgtype": "m.text"
625
        },
626
        "event_id": "$live-following-edit:localhost",
627
        "origin_server_ts": 12,
628
        "room_id": room_id.as_str(),
629
        "sender": "@bob:localhost",
630
        "type": "m.room.message"
631
    });
632
    service.observe(&mut data, &room_id, edit).await;
633
    let (saved, _) = data.context(&mention_key).unwrap();
634
    let context = saved.context.unwrap();
635
    assert_eq!(
636
        message_body(&context.events_after[0]),
637
        Some("edited following")
638
    );
639
640
    let saved_following = data.events.get_mut(&following_id).unwrap();
641
    assert!(saved_following.raw.is_some());
642
    saved_following.media.push(media_registration());
643
    assert_eq!(saved_following.media.len(), 1);
644
    let redaction = json!({
645
        "content": {},
646
        "event_id": "$live-following-redaction:localhost",
647
        "origin_server_ts": 13,
648
        "redacts": "$live-following:localhost",
649
        "room_id": room_id.as_str(),
650
        "sender": "@bob:localhost",
651
        "type": "m.room.redaction"
652
    });
653
    service.observe(&mut data, &room_id, redaction).await;
654
655
    let saved_following = data.events.get(&following_id).unwrap();
656
    assert!(matches!(
657
        &saved_following.item,
658
        TimelineItem::Event(event) if event.content == TimelineContent::Redacted
659
    ));
660
    assert!(saved_following.raw.is_none());
661
    assert!(saved_following.media.is_empty());
662
    let (saved, _) = data.context(&mention_key).unwrap();
663
    let context = saved.context.unwrap();
664
    assert!(matches!(
665
        &context.events_after[0],
666
        TimelineItem::Event(event) if event.content == TimelineContent::Redacted
667
    ));
668
    data.windows.remove(&mention_identity);
669
    service.data = data;
670
    let CommandOutput::MentionContext(context) = service
671
        .command(MentionsCommand::RetryContext(mention_key))
672
        .await
673
        .unwrap()
674
    else {
675
        panic!("retry should return saved context");
676
    };
677
    assert_eq!(context.state, MentionCaptureState::Pending);
678
    assert!(context.context.is_some());
679
}
680
681
#[tokio::test]
682
async fn explicit_recovery_rechecks_server_support() {
683
    let server = MatrixMockServer::new().await;
684
    let (client, _) = joined_client(&server, room_id!("!mentions:localhost")).await;
685
    let config = EngineConfig::default();
686
    let media = Arc::new(MediaService::new(
687
        client.clone(),
688
        config.max_media_bytes,
689
        config.max_context_media,
690
    ));
691
    let (_, mut service, _) = Service::new(client, media, None, config).await.unwrap();
692
    service.data.coverage.historical_supported = Some(false);
693
    service.recovery_budget = 0;
694
    service.command(MentionsCommand::Recover).await.unwrap();
695
    assert_eq!(service.data.coverage.historical_supported, None);
696
    assert!(service.data.coverage.recovering);
697
    assert_eq!(service.recovery_budget, 20);
698
}
699
700
#[tokio::test]
701
async fn observing_again_keeps_the_profiles_a_capture_fetched() {
702
    let server = MatrixMockServer::new().await;
703
    let room_id = room_id!("!mentions:localhost");
704
    let (client, own_user) = joined_client(&server, room_id).await;
705
    let raw = |event: matrix_sdk::ruma::serde::Raw<_>, timestamp: u64| {
706
        let mut value: serde_json::Value = serde_json::from_str(event.json().get()).unwrap();
707
        value["origin_server_ts"] = json!(timestamp);
708
        value
709
    };
710
    let factory = EventFactory::new()
711
        .room(room_id)
712
        .sender(user_id!("@bob:localhost"));
713
    let mention = raw(
714
        factory
715
            .text_msg("ping")
716
            .mentions(Mentions::with_user_ids([own_user]))
717
            .event_id(event_id!("$ping:localhost"))
718
            .into_raw_sync(),
719
        10,
720
    );
721
    let following = raw(
722
        factory
723
            .text_msg("following")
724
            .event_id(event_id!("$following:localhost"))
725
            .into_raw_sync(),
726
        11,
727
    );
728
    let service = service(&client, None).await;
729
    let room_id = room_id.to_owned();
730
    let mut data = Data::default();
731
    service.observe(&mut data, &room_id, mention.clone()).await;
732
    let ping = format!("{room_id}\n$ping:localhost");
733
    let following_id = format!("{room_id}\n$following:localhost");
734
735
    let profiled = |item: &TimelineItem| {
736
        let mut item = item.clone();
737
        if let TimelineItem::Event(event) = &mut item {
738
            event.sender.display_name = Some("Bob".to_owned());
739
        }
740
        item
741
    };
742
    let target = profiled(&data.entries[&ping].event);
743
    let mut neighbour = normalise_observed(&following, None, &following_id)
744
        .unwrap()
745
        .0;
746
    neighbour = profiled(&neighbour);
747
    let revision = data.entries[&ping].revision;
748
    service.captured(
749
        &mut data,
750
        ping.clone(),
751
        revision,
752
        Ok((
753
            crate::EventContext {
754
                events_before: Vec::new(),
755
                event: target,
756
                events_after: vec![neighbour],
757
                start_token: None,
758
                end_token: None,
759
            },
760
            Vec::new(),
761
            Vec::new(),
762
        )),
763
    );
764
    let capture = data.entries[&ping].capture.clone();
765
766
    service.observe(&mut data, &room_id, following).await;
767
    service.observe(&mut data, &room_id, mention).await;
768
769
    let name = |item: &TimelineItem| match item {
770
        TimelineItem::Event(event) => event.sender.display_name.clone(),
771
        _ => None,
772
    };
773
    assert_eq!(name(&data.entries[&ping].event).as_deref(), Some("Bob"));
774
    assert_eq!(name(&data.events[&ping].item).as_deref(), Some("Bob"));
775
    assert_eq!(
776
        name(&data.events[&following_id].item).as_deref(),
777
        Some("Bob")
778
    );
779
    assert_eq!(data.entries[&ping].capture, capture);
780
}