Skip to main content

max / makenotwork

15.7 KB · 507 lines History Blame Raw
1 //! Tests for the chat message store: backlog replay, the ban purge, and both
2 //! halves of retention.
3 //!
4 //! These drive `mt_db` directly rather than going through routes, because the
5 //! routes do not exist yet and because what is worth proving here is the SQL:
6 //! the cursor semantics the reconnect path depends on, and the two sweep
7 //! statements, which are the ones a unit test over a mock cannot check.
8
9 use mt_db::{mutations, queries};
10 use uuid::Uuid;
11
12 use crate::harness::TestHarness;
13
14 /// Default retention, matching `livechat::Retention::forum_default()` and the
15 /// column default in migration 039.
16 const FORUM_HOURS: i32 = 168;
17
18 struct Room {
19 id: Uuid,
20 alice: Uuid,
21 bob: Uuid,
22 }
23
24 async fn room(h: &mut TestHarness) -> Room {
25 let alice = h.login_as("alice").await;
26 let id = h.create_community("Test", "test").await;
27 h.add_membership(alice, id, "owner").await;
28 let bob = h.login_as("bob").await;
29 h.add_membership(bob, id, "member").await;
30 Room { id, alice, bob }
31 }
32
33 async fn say(h: &TestHarness, room: &Room, author: Uuid, body: &str) -> i64 {
34 mutations::insert_chat_message(&h.db, room.id, author, body, FORUM_HOURS)
35 .await
36 .expect("insert")
37 .id
38 }
39
40 /// Everything currently in the room, oldest first.
41 async fn all(h: &TestHarness, room: &Room) -> Vec<String> {
42 queries::recent_backlog(&h.db, room.id, 1_000)
43 .await
44 .expect("backlog")
45 .into_iter()
46 .map(|m| m.body_html)
47 .collect()
48 }
49
50 #[sqlx::test]
51 async fn ids_are_monotonic_and_the_backlog_reads_oldest_first(_pool: sqlx::PgPool) {
52 let mut h = TestHarness::new().await;
53 let r = room(&mut h).await;
54
55 let first = say(&h, &r, r.alice, "one").await;
56 let second = say(&h, &r, r.bob, "two").await;
57 let third = say(&h, &r, r.alice, "three").await;
58
59 assert!(first < second && second < third, "ids order the room");
60 assert_eq!(all(&h, &r).await, vec!["one", "two", "three"]);
61 }
62
63 #[sqlx::test]
64 async fn a_cursor_replays_only_what_came_after_it(_pool: sqlx::PgPool) {
65 // The reconnect path: every deploy drops every connection, and the client
66 // resumes from the last id it saw.
67 let mut h = TestHarness::new().await;
68 let r = room(&mut h).await;
69
70 say(&h, &r, r.alice, "before").await;
71 let cursor = say(&h, &r, r.alice, "last seen").await;
72 say(&h, &r, r.bob, "missed").await;
73 say(&h, &r, r.bob, "also missed").await;
74
75 let replay: Vec<_> = queries::backlog_after(&h.db, r.id, cursor, 100)
76 .await
77 .unwrap()
78 .into_iter()
79 .map(|m| m.body_html)
80 .collect();
81
82 assert_eq!(replay, vec!["missed", "also missed"]);
83 }
84
85 #[sqlx::test]
86 async fn a_long_absence_is_capped_and_resumable(_pool: sqlx::PgPool) {
87 // A client away longer than the window must not be able to make the server
88 // materialize the whole room in one reply. It gets the oldest slice past
89 // its cursor and walks forward.
90 let mut h = TestHarness::new().await;
91 let r = room(&mut h).await;
92
93 for i in 0..10 {
94 say(&h, &r, r.alice, &format!("m{i}")).await;
95 }
96
97 let first = queries::backlog_after(&h.db, r.id, 0, 4).await.unwrap();
98 assert_eq!(first.len(), 4);
99 assert_eq!(first[0].body_html, "m0", "oldest first, not newest");
100
101 let next = queries::backlog_after(&h.db, r.id, first[3].id, 4)
102 .await
103 .unwrap();
104 assert_eq!(next[0].body_html, "m4", "the cursor advances without a gap");
105 }
106
107 #[sqlx::test]
108 async fn a_fresh_connection_gets_the_newest_window_in_reading_order(_pool: sqlx::PgPool) {
109 let mut h = TestHarness::new().await;
110 let r = room(&mut h).await;
111
112 for i in 0..10 {
113 say(&h, &r, r.alice, &format!("m{i}")).await;
114 }
115
116 let window: Vec<_> = queries::recent_backlog(&h.db, r.id, 3)
117 .await
118 .unwrap()
119 .into_iter()
120 .map(|m| m.body_html)
121 .collect();
122
123 assert_eq!(window, vec!["m7", "m8", "m9"], "newest three, ascending");
124 }
125
126 #[sqlx::test]
127 async fn rooms_do_not_leak_into_each_other(_pool: sqlx::PgPool) {
128 let mut h = TestHarness::new().await;
129 let r = room(&mut h).await;
130
131 let other = h.create_community("Other", "other").await;
132 h.add_membership(r.alice, other, "owner").await;
133
134 say(&h, &r, r.alice, "in test").await;
135 mutations::insert_chat_message(&h.db, other, r.alice, "in other", FORUM_HOURS)
136 .await
137 .unwrap();
138
139 assert_eq!(all(&h, &r).await, vec!["in test"]);
140 }
141
142 #[sqlx::test]
143 async fn the_author_lookup_is_scoped_to_the_room(_pool: sqlx::PgPool) {
144 // A message id is a guessable integer. Asking who wrote one without saying
145 // which room it is in would confirm authorship across a community the
146 // caller cannot read.
147 let mut h = TestHarness::new().await;
148 let r = room(&mut h).await;
149
150 let other = h.create_community("Other", "other").await;
151 h.add_membership(r.alice, other, "owner").await;
152 let elsewhere = mutations::insert_chat_message(&h.db, other, r.alice, "secret", FORUM_HOURS)
153 .await
154 .unwrap()
155 .id;
156
157 let mine = say(&h, &r, r.bob, "mine").await;
158
159 assert_eq!(
160 queries::chat_message_author(&h.db, r.id, mine)
161 .await
162 .unwrap(),
163 Some(r.bob)
164 );
165 assert_eq!(
166 queries::chat_message_author(&h.db, r.id, elsewhere)
167 .await
168 .unwrap(),
169 None,
170 "a message in another room is indistinguishable from absent"
171 );
172 }
173
174 #[sqlx::test]
175 async fn a_purge_removes_one_author_and_leaves_the_room(_pool: sqlx::PgPool) {
176 let mut h = TestHarness::new().await;
177 let r = room(&mut h).await;
178
179 say(&h, &r, r.alice, "alice one").await;
180 say(&h, &r, r.bob, "bob spam").await;
181 say(&h, &r, r.alice, "alice two").await;
182 say(&h, &r, r.bob, "bob spam again").await;
183
184 let removed = mutations::purge_author_messages(&h.db, r.id, r.bob)
185 .await
186 .unwrap();
187
188 assert_eq!(removed, 2);
189 assert_eq!(all(&h, &r).await, vec!["alice one", "alice two"]);
190 }
191
192 #[sqlx::test]
193 async fn a_purge_does_not_reach_the_same_author_in_another_room(_pool: sqlx::PgPool) {
194 // Banning someone from one community must not erase them from another.
195 let mut h = TestHarness::new().await;
196 let r = room(&mut h).await;
197
198 let other = h.create_community("Other", "other").await;
199 h.add_membership(r.bob, other, "member").await;
200 mutations::insert_chat_message(&h.db, other, r.bob, "innocent", FORUM_HOURS)
201 .await
202 .unwrap();
203
204 say(&h, &r, r.bob, "spam").await;
205 mutations::purge_author_messages(&h.db, r.id, r.bob)
206 .await
207 .unwrap();
208
209 let survivors = queries::recent_backlog(&h.db, other, 100).await.unwrap();
210 assert_eq!(survivors.len(), 1, "the other room is untouched");
211 }
212
213 #[sqlx::test]
214 async fn deleting_one_message_reports_whether_it_was_there(_pool: sqlx::PgPool) {
215 let mut h = TestHarness::new().await;
216 let r = room(&mut h).await;
217
218 let id = say(&h, &r, r.alice, "oops").await;
219
220 assert_eq!(
221 mutations::delete_chat_message(&h.db, r.id, id)
222 .await
223 .unwrap(),
224 1
225 );
226 assert_eq!(
227 mutations::delete_chat_message(&h.db, r.id, id)
228 .await
229 .unwrap(),
230 0,
231 "a second delete is a no-op, not an error"
232 );
233 assert!(all(&h, &r).await.is_empty());
234 }
235
236 #[sqlx::test]
237 async fn the_sweep_takes_expired_messages_and_spares_live_ones(_pool: sqlx::PgPool) {
238 let mut h = TestHarness::new().await;
239 let r = room(&mut h).await;
240
241 say(&h, &r, r.alice, "fresh").await;
242 let stale = say(&h, &r, r.alice, "stale").await;
243
244 // Expire one row directly. Reaching past the insert path is the point:
245 // waiting out a real retention window is not a test.
246 sqlx::query("UPDATE chat_messages SET expires_at = now() - interval '1 hour' WHERE id = $1")
247 .bind(stale)
248 .execute(&h.db)
249 .await
250 .unwrap();
251
252 let swept = mutations::sweep_expired_chat_messages(&h.db).await.unwrap();
253
254 assert_eq!(swept, 1);
255 assert_eq!(all(&h, &r).await, vec!["fresh"]);
256 }
257
258 #[sqlx::test]
259 async fn shortening_retention_applies_to_messages_already_sent(_pool: sqlx::PgPool) {
260 // The cost of stamping expiry at insert: a shortened window has to reach
261 // back, or an owner who just cut retention from 30 days to 1 would still be
262 // holding 30 days of chat.
263 let mut h = TestHarness::new().await;
264 let r = room(&mut h).await;
265
266 // Sent under a long policy, and old enough that a short one expires it.
267 let old = say(&h, &r, r.alice, "old").await;
268 sqlx::query(
269 "UPDATE chat_messages
270 SET created_at = now() - interval '48 hours',
271 expires_at = now() + interval '600 hours'
272 WHERE id = $1",
273 )
274 .bind(old)
275 .execute(&h.db)
276 .await
277 .unwrap();
278
279 say(&h, &r, r.alice, "recent").await;
280
281 // Owner cuts the window to 24 hours.
282 let restamped = mutations::recompute_chat_expiry(&h.db, r.id, 24)
283 .await
284 .unwrap();
285 assert_eq!(restamped, 2, "every message in the room is restamped");
286
287 let swept = mutations::sweep_expired_chat_messages(&h.db).await.unwrap();
288 assert_eq!(swept, 1, "the 48-hour-old message is now past a 24h window");
289 assert_eq!(all(&h, &r).await, vec!["recent"]);
290 }
291
292 #[sqlx::test]
293 async fn recompute_is_measured_from_send_time_not_from_now(_pool: sqlx::PgPool) {
294 // Restamping from now() would silently extend every old message's life by
295 // the full window each time an owner touched the setting.
296 let mut h = TestHarness::new().await;
297 let r = room(&mut h).await;
298
299 let old = say(&h, &r, r.alice, "old").await;
300 sqlx::query("UPDATE chat_messages SET created_at = now() - interval '10 hours' WHERE id = $1")
301 .bind(old)
302 .execute(&h.db)
303 .await
304 .unwrap();
305
306 mutations::recompute_chat_expiry(&h.db, r.id, 6)
307 .await
308 .unwrap();
309
310 assert_eq!(
311 mutations::sweep_expired_chat_messages(&h.db).await.unwrap(),
312 1,
313 "a 10-hour-old message under a 6-hour window is already expired"
314 );
315 }
316
317 #[sqlx::test]
318 async fn the_count_cap_trims_the_oldest_and_keeps_the_newest(_pool: sqlx::PgPool) {
319 // The other half of retention: age expires a quiet room, the cap bounds a
320 // busy one that would reach its age limit holding far more.
321 let mut h = TestHarness::new().await;
322 let r = room(&mut h).await;
323
324 sqlx::query("UPDATE communities SET chat_max_messages = 3 WHERE id = $1")
325 .bind(r.id)
326 .execute(&h.db)
327 .await
328 .unwrap();
329
330 for i in 0..10 {
331 say(&h, &r, r.alice, &format!("m{i}")).await;
332 }
333
334 let trimmed = mutations::trim_chat_rooms_to_cap(&h.db).await.unwrap();
335
336 assert_eq!(trimmed, 7);
337 assert_eq!(all(&h, &r).await, vec!["m7", "m8", "m9"]);
338 }
339
340 #[sqlx::test]
341 async fn the_cap_is_per_room_not_global(_pool: sqlx::PgPool) {
342 // One statement trims the whole table, so the partition boundary is the
343 // thing that can be wrong: a room under its own cap must not lose messages
344 // because another room is over.
345 let mut h = TestHarness::new().await;
346 let r = room(&mut h).await;
347
348 let quiet = h.create_community("Quiet", "quiet").await;
349 h.add_membership(r.alice, quiet, "owner").await;
350
351 sqlx::query("UPDATE communities SET chat_max_messages = 2 WHERE id = $1")
352 .bind(r.id)
353 .execute(&h.db)
354 .await
355 .unwrap();
356
357 for i in 0..5 {
358 say(&h, &r, r.alice, &format!("loud{i}")).await;
359 }
360 for i in 0..2 {
361 mutations::insert_chat_message(&h.db, quiet, r.alice, &format!("q{i}"), FORUM_HOURS)
362 .await
363 .unwrap();
364 }
365
366 assert_eq!(mutations::trim_chat_rooms_to_cap(&h.db).await.unwrap(), 3);
367 assert_eq!(all(&h, &r).await, vec!["loud3", "loud4"]);
368 assert_eq!(
369 queries::recent_backlog(&h.db, quiet, 100)
370 .await
371 .unwrap()
372 .len(),
373 2,
374 "the room under its cap kept everything"
375 );
376 }
377
378 #[sqlx::test]
379 async fn a_room_exactly_at_its_cap_loses_nothing(_pool: sqlx::PgPool) {
380 let mut h = TestHarness::new().await;
381 let r = room(&mut h).await;
382
383 sqlx::query("UPDATE communities SET chat_max_messages = 3 WHERE id = $1")
384 .bind(r.id)
385 .execute(&h.db)
386 .await
387 .unwrap();
388
389 for i in 0..3 {
390 say(&h, &r, r.alice, &format!("m{i}")).await;
391 }
392
393 assert_eq!(
394 mutations::trim_chat_rooms_to_cap(&h.db).await.unwrap(),
395 0,
396 "the cap is inclusive"
397 );
398 assert_eq!(all(&h, &r).await.len(), 3);
399 }
400
401 #[sqlx::test]
402 async fn deleting_a_community_takes_its_chat_with_it(_pool: sqlx::PgPool) {
403 let mut h = TestHarness::new().await;
404 let r = room(&mut h).await;
405 say(&h, &r, r.alice, "hello").await;
406
407 sqlx::query("DELETE FROM communities WHERE id = $1")
408 .bind(r.id)
409 .execute(&h.db)
410 .await
411 .expect("the cascade must not be blocked by a chat row");
412
413 let left: i64 =
414 sqlx::query_scalar("SELECT COUNT(*) FROM chat_messages WHERE community_id = $1")
415 .bind(r.id)
416 .fetch_one(&h.db)
417 .await
418 .unwrap();
419 assert_eq!(left, 0);
420 }
421
422 #[sqlx::test]
423 async fn retention_columns_refuse_values_past_the_crate_ceilings(_pool: sqlx::PgPool) {
424 // The schema restates `livechat`'s ceilings so a bad UPDATE fails at the
425 // database rather than at whichever call site forgot to validate.
426 let mut h = TestHarness::new().await;
427 let r = room(&mut h).await;
428
429 for (column, value) in [
430 ("chat_retention_hours", 721),
431 ("chat_retention_hours", 0),
432 ("chat_max_messages", 20_001),
433 ("chat_max_messages", 0),
434 ] {
435 let result = sqlx::query(&format!(
436 "UPDATE communities SET {column} = $1 WHERE id = $2"
437 ))
438 .bind(value)
439 .bind(r.id)
440 .execute(&h.db)
441 .await;
442 assert!(result.is_err(), "{column} = {value} must be refused");
443 }
444 }
445
446 // The scheduled sweep
447
448 #[sqlx::test]
449 async fn one_sweep_round_enforces_both_halves_of_retention(_pool: sqlx::PgPool) {
450 // The two statements answer different questions and both must run every
451 // round. A room busy enough to be over its cap is usually one whose
452 // messages are all too new to have expired, so gating the trim on the
453 // expiry sweep finding something would skip it exactly when it is needed.
454 let mut h = TestHarness::new().await;
455 let r = room(&mut h).await;
456
457 sqlx::query("UPDATE communities SET chat_max_messages = 2 WHERE id = $1")
458 .bind(r.id)
459 .execute(&h.db)
460 .await
461 .unwrap();
462
463 // Five fresh messages: nothing is expired, and three are over the cap.
464 for i in 0..5 {
465 say(&h, &r, r.alice, &format!("m{i}")).await;
466 }
467
468 let (expired, trimmed) = multithreaded::maintenance::sweep_chat_once(&h.db).await;
469
470 assert_eq!(expired, 0, "nothing was old enough to expire");
471 assert_eq!(trimmed, 3, "the cap still bit");
472 assert_eq!(all(&h, &r).await, vec!["m3", "m4"]);
473 }
474
475 #[sqlx::test]
476 async fn a_sweep_round_over_an_empty_table_is_a_no_op(_pool: sqlx::PgPool) {
477 let h = TestHarness::new().await;
478 assert_eq!(
479 multithreaded::maintenance::sweep_chat_once(&h.db).await,
480 (0, 0)
481 );
482 }
483
484 #[sqlx::test]
485 async fn the_sweep_is_convergent(_pool: sqlx::PgPool) {
486 // Steady-state work is zero: a second round immediately after the first
487 // must find nothing, or the sweep would churn the same rows every interval.
488 let mut h = TestHarness::new().await;
489 let r = room(&mut h).await;
490
491 sqlx::query("UPDATE communities SET chat_max_messages = 2 WHERE id = $1")
492 .bind(r.id)
493 .execute(&h.db)
494 .await
495 .unwrap();
496 for i in 0..5 {
497 say(&h, &r, r.alice, &format!("m{i}")).await;
498 }
499
500 multithreaded::maintenance::sweep_chat_once(&h.db).await;
501 assert_eq!(
502 multithreaded::maintenance::sweep_chat_once(&h.db).await,
503 (0, 0),
504 "the second round must find nothing left to do"
505 );
506 }
507