Skip to main content

max / makenotwork

Add livechat backlog/live stream and axum SSE adapter ChatStream joins backlog replay to the live feed. The ordering is the whole correctness question: fetch then subscribe loses every message sent between the two, so open() subscribes first and then fetches, turning a gap into an overlap, and drops live messages the backlog already covered by id. Hosts pass a fetch closure rather than a finished backlog so the order cannot be got wrong from outside. Deletes and purges pass through the overlap unfiltered. A delete landing in that window refers to a message the client already has, so suppressing it would leave deleted content on screen until reload. SSE adapter is behind an `axum` feature; the rest of the crate stays framework-agnostic. 30s keepalive clears Cloudflare's 100s idle timeout, matching the SyncKit endpoint already in production. The stream owns the hub subscription, so axum dropping the response on disconnect releases the slot and prunes the room. Design: wiki livechat-design Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-07-19 00:27 UTC
Commit: 7fc6d99e203d76f99b37e46fbf144e59cb05946d
Parent: 7126666
5 files changed, +867 insertions, -0 deletions
M shared/livechat/Cargo.lock +378
@@ -3,6 +3,28 @@
3 3 version = 4
4 4
5 5 [[package]]
6 + name = "async-stream"
7 + version = "0.3.6"
8 + source = "registry+https://github.com/rust-lang/crates.io-index"
9 + checksum = "0b5a71a6f37880a80d1d7f19efd781e4b5de42c88f0722cc13bcb6cc2cfe8476"
10 + dependencies = [
11 + "async-stream-impl",
12 + "futures-core",
13 + "pin-project-lite",
14 + ]
15 +
16 + [[package]]
17 + name = "async-stream-impl"
18 + version = "0.3.6"
19 + source = "registry+https://github.com/rust-lang/crates.io-index"
20 + checksum = "c7c24de15d275a1ecfd47a380fb4d5ec9bfe0933f309ed5e705b775596a3574d"
21 + dependencies = [
22 + "proc-macro2",
23 + "quote",
24 + "syn 2.0.119",
25 + ]
26 +
27 + [[package]]
6 28 name = "async-trait"
7 29 version = "0.1.91"
8 30 source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -14,6 +36,64 @@ dependencies = [
14 36 ]
15 37
16 38 [[package]]
39 + name = "atomic-waker"
40 + version = "1.1.2"
41 + source = "registry+https://github.com/rust-lang/crates.io-index"
42 + checksum = "1505bd5d3d116872e7271a6d4e16d81d0c8570876c8de68093a09ac269d8aac0"
43 +
44 + [[package]]
45 + name = "axum"
46 + version = "0.8.9"
47 + source = "registry+https://github.com/rust-lang/crates.io-index"
48 + checksum = "31b698c5f9a010f6573133b09e0de5408834d0c82f8d7475a89fc1867a71cd90"
49 + dependencies = [
50 + "axum-core",
51 + "bytes",
52 + "form_urlencoded",
53 + "futures-util",
54 + "http",
55 + "http-body",
56 + "http-body-util",
57 + "hyper",
58 + "hyper-util",
59 + "itoa",
60 + "matchit",
61 + "memchr",
62 + "mime",
63 + "percent-encoding",
64 + "pin-project-lite",
65 + "serde_core",
66 + "serde_json",
67 + "serde_path_to_error",
68 + "serde_urlencoded",
69 + "sync_wrapper",
70 + "tokio",
71 + "tower",
72 + "tower-layer",
73 + "tower-service",
74 + "tracing",
75 + ]
76 +
77 + [[package]]
78 + name = "axum-core"
79 + version = "0.5.6"
80 + source = "registry+https://github.com/rust-lang/crates.io-index"
81 + checksum = "08c78f31d7b1291f7ee735c1c6780ccde7785daae9a9206026862dab7d8792d1"
82 + dependencies = [
83 + "bytes",
84 + "futures-core",
85 + "http",
86 + "http-body",
87 + "http-body-util",
88 + "mime",
89 + "pin-project-lite",
90 + "sync_wrapper",
91 + "tower-layer",
92 + "tower-service",
93 + "tracing",
94 + ]
95 +
96 + [[package]]
17 97 name = "bitflags"
18 98 version = "2.13.1"
19 99 source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -26,6 +106,12 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
26 106 checksum = "72f5acc6cb2ba439de613abc23857ec3d78374d8ed5ac84e9d11336e87da8649"
27 107
28 108 [[package]]
109 + name = "bytes"
110 + version = "1.12.1"
111 + source = "registry+https://github.com/rust-lang/crates.io-index"
112 + checksum = "fc652a48c352aef3ea3aed32080501cf3ef6ed5da78602a020c991775b0aff04"
113 +
114 + [[package]]
29 115 name = "cfg-if"
30 116 version = "1.0.4"
31 117 source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -52,12 +138,36 @@ dependencies = [
52 138 ]
53 139
54 140 [[package]]
141 + name = "form_urlencoded"
142 + version = "1.2.2"
143 + source = "registry+https://github.com/rust-lang/crates.io-index"
144 + checksum = "cb4cb245038516f5f85277875cdaa4f7d2c9a0fa0468de06ed190163b1581fcf"
145 + dependencies = [
146 + "percent-encoding",
147 + ]
148 +
149 + [[package]]
150 + name = "futures-channel"
151 + version = "0.3.33"
152 + source = "registry+https://github.com/rust-lang/crates.io-index"
153 + checksum = "262590f4fe6afeb0bc83be1daa64e52657fe185690a958af7f3ad0e92085c5ae"
154 + dependencies = [
155 + "futures-core",
156 + ]
157 +
158 + [[package]]
55 159 name = "futures-core"
56 160 version = "0.3.33"
57 161 source = "registry+https://github.com/rust-lang/crates.io-index"
58 162 checksum = "2cd50c473c80f6d7c3670a752354b8e569b1a7cbfdc0419ec88e5edad85e0dc7"
59 163
60 164 [[package]]
165 + name = "futures-sink"
166 + version = "0.3.33"
167 + source = "registry+https://github.com/rust-lang/crates.io-index"
168 + checksum = "e34418ac499d6305c2fb5ad0ed2f6ac998c5f8ca209b4510f7f94242c647e307"
169 +
170 + [[package]]
61 171 name = "futures-task"
62 172 version = "0.3.33"
63 173 source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -93,6 +203,86 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
93 203 checksum = "e5274423e17b7c9fc20b6e7e208532f9b19825d82dfd615708b70edd83df41f1"
94 204
95 205 [[package]]
206 + name = "http"
207 + version = "1.4.2"
208 + source = "registry+https://github.com/rust-lang/crates.io-index"
209 + checksum = "6970f50e31d6fc17d3fa27329444bfa74e196cf62e95052a3f6fee181dba6425"
210 + dependencies = [
211 + "bytes",
212 + "itoa",
213 + ]
214 +
215 + [[package]]
216 + name = "http-body"
217 + version = "1.1.0"
218 + source = "registry+https://github.com/rust-lang/crates.io-index"
219 + checksum = "ca2a8f2913ee65f60facd6a5905613afaa448497a0230cc41ce022d93290bc2c"
220 + dependencies = [
221 + "bytes",
222 + "http",
223 + ]
224 +
225 + [[package]]
226 + name = "http-body-util"
227 + version = "0.1.4"
228 + source = "registry+https://github.com/rust-lang/crates.io-index"
229 + checksum = "e9f41fd6a08e4d4ec69df65976da761afd5ad5e58a9d4acb46bd1c953a9e3ff2"
230 + dependencies = [
231 + "bytes",
232 + "futures-core",
233 + "http",
234 + "http-body",
235 + "pin-project-lite",
236 + ]
237 +
238 + [[package]]
239 + name = "httparse"
240 + version = "1.10.1"
241 + source = "registry+https://github.com/rust-lang/crates.io-index"
242 + checksum = "6dbf3de79e51f3d586ab4cb9d5c3e2c14aa28ed23d180cf89b4df0454a69cc87"
243 +
244 + [[package]]
245 + name = "httpdate"
246 + version = "1.0.3"
247 + source = "registry+https://github.com/rust-lang/crates.io-index"
248 + checksum = "df3b46402a9d5adb4c86a0cf463f42e19994e3ee891101b1841f30a545cb49a9"
249 +
250 + [[package]]
251 + name = "hyper"
252 + version = "1.10.1"
253 + source = "registry+https://github.com/rust-lang/crates.io-index"
254 + checksum = "55281c53a1894c864990125767da440a4e630446785086f52523b20033b74498"
255 + dependencies = [
256 + "atomic-waker",
257 + "bytes",
258 + "futures-channel",
259 + "futures-core",
260 + "http",
261 + "http-body",
262 + "httparse",
263 + "httpdate",
264 + "itoa",
265 + "pin-project-lite",
266 + "smallvec",
267 + "tokio",
268 + ]
269 +
270 + [[package]]
271 + name = "hyper-util"
272 + version = "0.1.20"
273 + source = "registry+https://github.com/rust-lang/crates.io-index"
274 + checksum = "96547c2556ec9d12fb1578c4eaf448b04993e7fb79cbaad930a656880a6bdfa0"
275 + dependencies = [
276 + "bytes",
277 + "http",
278 + "http-body",
279 + "hyper",
280 + "pin-project-lite",
281 + "tokio",
282 + "tower-service",
283 + ]
284 +
285 + [[package]]
96 286 name = "itoa"
97 287 version = "1.0.18"
98 288 source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -119,12 +309,16 @@ checksum = "68ab91017fe16c622486840e4c83c9a37afeff978bd239b5293d61ece587de66"
119 309 name = "livechat"
120 310 version = "0.1.0"
121 311 dependencies = [
312 + "async-stream",
122 313 "async-trait",
314 + "axum",
123 315 "dashmap",
124 316 "serde",
125 317 "serde_json",
126 318 "thiserror",
127 319 "tokio",
320 + "tokio-stream",
321 + "tracing",
128 322 "uuid",
129 323 ]
130 324
@@ -138,12 +332,41 @@ dependencies = [
138 332 ]
139 333
140 334 [[package]]
335 + name = "log"
336 + version = "0.4.33"
337 + source = "registry+https://github.com/rust-lang/crates.io-index"
338 + checksum = "0ceec5bc11778974d1bcb055b18002eba7f4b3518b6a0081b3af5f21666da9ad"
339 +
340 + [[package]]
341 + name = "matchit"
342 + version = "0.8.4"
343 + source = "registry+https://github.com/rust-lang/crates.io-index"
344 + checksum = "47e1ffaa40ddd1f3ed91f717a33c8c0ee23fff369e3aa8772b9605cc1d22f4c3"
345 +
346 + [[package]]
141 347 name = "memchr"
142 348 version = "2.8.3"
143 349 source = "registry+https://github.com/rust-lang/crates.io-index"
144 350 checksum = "cf8baf1c55e62ffcace7a9f06f4bd9cd3f0c4beb022d3b367256b91b87513d98"
145 351
146 352 [[package]]
353 + name = "mime"
354 + version = "0.3.17"
355 + source = "registry+https://github.com/rust-lang/crates.io-index"
356 + checksum = "6877bb514081ee2a7ff5ef9de3281f14a4dd4bceac4c09388074a6b5df8a139a"
357 +
358 + [[package]]
359 + name = "mio"
360 + version = "1.2.2"
361 + source = "registry+https://github.com/rust-lang/crates.io-index"
362 + checksum = "30d65c71f1ce40ab09135ce117d742b9f8a19ff91a41a8b57ed50bc2de59c427"
363 + dependencies = [
364 + "libc",
365 + "wasi",
366 + "windows-sys",
367 + ]
368 +
369 + [[package]]
147 370 name = "once_cell"
148 371 version = "1.21.4"
149 372 source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -163,6 +386,12 @@ dependencies = [
163 386 ]
164 387
165 388 [[package]]
389 + name = "percent-encoding"
390 + version = "2.3.2"
391 + source = "registry+https://github.com/rust-lang/crates.io-index"
392 + checksum = "9b4f627cb1b25917193a259e49bdad08f671f8d9708acfd5fe0a8c1455d87220"
393 +
394 + [[package]]
166 395 name = "pin-project-lite"
167 396 version = "0.2.17"
168 397 source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -208,6 +437,12 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
208 437 checksum = "cf54715a573b99ac80df0bc206da022bcd442c974952c7b9720069370852e21f"
209 438
210 439 [[package]]
440 + name = "ryu"
441 + version = "1.0.23"
442 + source = "registry+https://github.com/rust-lang/crates.io-index"
443 + checksum = "9774ba4a74de5f7b1c1451ed6cd5285a32eddb5cccb8cc655a4e50009e06477f"
444 +
445 + [[package]]
211 446 name = "scopeguard"
212 447 version = "1.2.0"
213 448 source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -257,6 +492,29 @@ dependencies = [
257 492 ]
258 493
259 494 [[package]]
495 + name = "serde_path_to_error"
496 + version = "0.1.20"
497 + source = "registry+https://github.com/rust-lang/crates.io-index"
498 + checksum = "10a9ff822e371bb5403e391ecd83e182e0e77ba7f6fe0160b795797109d1b457"
499 + dependencies = [
500 + "itoa",
501 + "serde",
502 + "serde_core",
503 + ]
504 +
505 + [[package]]
506 + name = "serde_urlencoded"
507 + version = "0.7.1"
508 + source = "registry+https://github.com/rust-lang/crates.io-index"
509 + checksum = "d3491c14715ca2294c4d6a88f15e84739788c1d030eed8c110436aafdaa2f3fd"
510 + dependencies = [
511 + "form_urlencoded",
512 + "itoa",
513 + "ryu",
514 + "serde",
515 + ]
516 +
517 + [[package]]
260 518 name = "slab"
261 519 version = "0.4.12"
262 520 source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -269,6 +527,16 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
269 527 checksum = "8ed6a63f02c8539c91a8685a86f4099661ba3da017932f6ebbea6de3f0fa7c90"
270 528
271 529 [[package]]
530 + name = "socket2"
531 + version = "0.6.5"
532 + source = "registry+https://github.com/rust-lang/crates.io-index"
533 + checksum = "c3d1e2c7f27f8d4cb10542a02c49005dbd6e93095799d6f3be745fae9f8fedd4"
534 + dependencies = [
535 + "libc",
536 + "windows-sys",
537 + ]
538 +
539 + [[package]]
272 540 name = "syn"
273 541 version = "2.0.119"
274 542 source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -291,6 +559,12 @@ dependencies = [
291 559 ]
292 560
293 561 [[package]]
562 + name = "sync_wrapper"
563 + version = "1.0.2"
564 + source = "registry+https://github.com/rust-lang/crates.io-index"
565 + checksum = "0bf256ce5efdfa370213c1dabab5935a12e49f2c58d15e9eac2870d3b4f27263"
566 +
567 + [[package]]
294 568 name = "thiserror"
295 569 version = "2.0.19"
296 570 source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -316,8 +590,12 @@ version = "1.53.0"
316 590 source = "registry+https://github.com/rust-lang/crates.io-index"
317 591 checksum = "d988bcd52dbe076d3d46903332f58c912b87a2c49b1428419a5845154762ffee"
318 592 dependencies = [
593 + "libc",
594 + "mio",
319 595 "pin-project-lite",
596 + "socket2",
320 597 "tokio-macros",
598 + "windows-sys",
321 599 ]
322 600
323 601 [[package]]
@@ -332,6 +610,91 @@ dependencies = [
332 610 ]
333 611
334 612 [[package]]
613 + name = "tokio-stream"
614 + version = "0.1.18"
615 + source = "registry+https://github.com/rust-lang/crates.io-index"
616 + checksum = "32da49809aab5c3bc678af03902d4ccddea2a87d028d86392a4b1560c6906c70"
617 + dependencies = [
618 + "futures-core",
619 + "pin-project-lite",
620 + "tokio",
621 + "tokio-util",
622 + ]
623 +
624 + [[package]]
625 + name = "tokio-util"
626 + version = "0.7.18"
627 + source = "registry+https://github.com/rust-lang/crates.io-index"
628 + checksum = "9ae9cec805b01e8fc3fd2fe289f89149a9b66dd16786abd8b19cfa7b48cb0098"
629 + dependencies = [
630 + "bytes",
631 + "futures-core",
632 + "futures-sink",
633 + "pin-project-lite",
634 + "tokio",
635 + ]
636 +
637 + [[package]]
638 + name = "tower"
639 + version = "0.5.3"
640 + source = "registry+https://github.com/rust-lang/crates.io-index"
641 + checksum = "ebe5ef63511595f1344e2d5cfa636d973292adc0eec1f0ad45fae9f0851ab1d4"
642 + dependencies = [
643 + "futures-core",
644 + "futures-util",
645 + "pin-project-lite",
646 + "sync_wrapper",
647 + "tokio",
648 + "tower-layer",
649 + "tower-service",
650 + "tracing",
651 + ]
652 +
653 + [[package]]
654 + name = "tower-layer"
655 + version = "0.3.3"
656 + source = "registry+https://github.com/rust-lang/crates.io-index"
657 + checksum = "121c2a6cda46980bb0fcd1647ffaf6cd3fc79a013de288782836f6df9c48780e"
658 +
659 + [[package]]
660 + name = "tower-service"
661 + version = "0.3.3"
662 + source = "registry+https://github.com/rust-lang/crates.io-index"
663 + checksum = "8df9b6e13f2d32c91b9bd719c00d1958837bc7dec474d94952798cc8e69eeec3"
664 +
665 + [[package]]
666 + name = "tracing"
667 + version = "0.1.44"
668 + source = "registry+https://github.com/rust-lang/crates.io-index"
669 + checksum = "63e71662fa4b2a2c3a26f570f037eb95bb1f85397f3cd8076caed2f026a6d100"
670 + dependencies = [
671 + "log",
672 + "pin-project-lite",
673 + "tracing-attributes",
674 + "tracing-core",
675 + ]
676 +
677 + [[package]]
678 + name = "tracing-attributes"
679 + version = "0.1.31"
680 + source = "registry+https://github.com/rust-lang/crates.io-index"
681 + checksum = "7490cfa5ec963746568740651ac6781f701c9c5ea257c58e057f3ba8cf69e8da"
682 + dependencies = [
683 + "proc-macro2",
684 + "quote",
685 + "syn 2.0.119",
686 + ]
687 +
688 + [[package]]
689 + name = "tracing-core"
690 + version = "0.1.36"
691 + source = "registry+https://github.com/rust-lang/crates.io-index"
692 + checksum = "db97caf9d906fbde555dd62fa95ddba9eecfd14cb388e4f491a66d74cd5fb79a"
693 + dependencies = [
694 + "once_cell",
695 + ]
696 +
697 + [[package]]
335 698 name = "unicode-ident"
336 699 version = "1.0.24"
337 700 source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -350,6 +713,12 @@ dependencies = [
350 713 ]
351 714
352 715 [[package]]
716 + name = "wasi"
717 + version = "0.11.1+wasi-snapshot-preview1"
718 + source = "registry+https://github.com/rust-lang/crates.io-index"
719 + checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b"
720 +
721 + [[package]]
353 722 name = "wasm-bindgen"
354 723 version = "0.2.126"
355 724 source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -401,6 +770,15 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
401 770 checksum = "f0805222e57f7521d6a62e36fa9163bc891acd422f971defe97d64e70d0a4fe5"
402 771
403 772 [[package]]
773 + name = "windows-sys"
774 + version = "0.61.2"
775 + source = "registry+https://github.com/rust-lang/crates.io-index"
776 + checksum = "ae137229bcbd6cdf0f7b80a31df61766145077ddf49416a728b02cb3921ff3fc"
777 + dependencies = [
778 + "windows-link",
779 + ]
780 +
781 + [[package]]
404 782 name = "zmij"
405 783 version = "1.0.23"
406 784 source = "registry+https://github.com/rust-lang/crates.io-index"
M shared/livechat/Cargo.toml +11
@@ -3,6 +3,11 @@ name = "livechat"
3 3 version = "0.1.0"
4 4 edition = "2024"
5 5
6 + [features]
7 + default = []
8 + # Axum SSE adapter. The rest of the crate has no framework dependency.
9 + axum = ["dep:axum", "dep:async-stream", "dep:tokio-stream", "dep:serde_json", "dep:tracing"]
10 +
6 11 [dependencies]
7 12 async-trait = "0.1"
8 13 dashmap = "6"
@@ -11,6 +16,12 @@ thiserror = "2"
11 16 tokio = { version = "1", features = ["sync"] }
12 17 uuid = { version = "1", features = ["v4", "serde"] }
13 18
19 + async-stream = { version = "0.3", optional = true }
20 + axum = { version = "0.8", optional = true }
21 + serde_json = { version = "1", optional = true }
22 + tokio-stream = { version = "0.1", features = ["sync"], optional = true }
23 + tracing = { version = "0.1", optional = true }
24 +
14 25 [dev-dependencies]
15 26 serde_json = "1"
16 27 tokio = { version = "1", features = ["macros", "rt", "rt-multi-thread", "sync", "time"] }
M shared/livechat/src/lib.rs +6
@@ -44,6 +44,9 @@ mod ids;
44 44 mod message;
45 45 mod retention;
46 46 mod room;
47 + #[cfg(feature = "axum")]
48 + mod sse;
49 + mod stream;
47 50 mod traits;
48 51
49 52 pub use error::{ChatError, LimitScope};
@@ -53,6 +56,9 @@ pub use ids::{MessageId, Nonce, RoomId, UserId};
53 56 pub use message::{MAX_MESSAGE_LEN, Message, MessageBody};
54 57 pub use retention::{MAX_AGE_CEILING, MAX_MESSAGES_CEILING, Retention};
55 58 pub use room::{Room, RoomState};
59 + #[cfg(feature = "axum")]
60 + pub use sse::{KEEPALIVE_INTERVAL, sse_response};
61 + pub use stream::ChatStream;
56 62 pub use traits::{
57 63 ChatAuthz, ChatIdentity, ChatModeration, ChatRooms, DenyReason, Identity, WriteAccess,
58 64 };
A shared/livechat/src/sse.rs +147
@@ -0,0 +1,147 @@
1 + //! Axum SSE adapter. Behind the `axum` feature; the rest of the crate has no
2 + //! framework dependency.
3 + //!
4 + //! SSE rather than a WebSocket, on purpose. Upstream traffic is one `POST` per
5 + //! message typed, so bidirectionality buys nothing, and a `POST` passes through
6 + //! the CSRF, rate-limit, and session middleware both consumers already run. It
7 + //! also avoids adding a `connect-src` CSP directive on the Multithreaded side,
8 + //! which currently has none and falls back to `default-src 'self'`.
9 +
10 + use std::convert::Infallible;
11 + use std::time::Duration;
12 +
13 + use axum::response::sse::{Event, KeepAlive, Sse};
14 + use tokio_stream::Stream;
15 +
16 + use crate::event::ChatEvent;
17 + use crate::stream::ChatStream;
18 +
19 + /// Keepalive interval.
20 + ///
21 + /// Cloudflare fronts makenot.work and idles a connection out at 100 seconds. A
22 + /// 30 second comment clears that with room for a missed tick, and matches what
23 + /// the SyncKit SSE endpoint already runs in production.
24 + pub const KEEPALIVE_INTERVAL: Duration = Duration::from_secs(30);
25 +
26 + /// Serialize one event into an SSE frame.
27 + ///
28 + /// The event name is carried in the SSE `event:` field so the client can dispatch
29 + /// without parsing the payload first, and the payload is the same tagged JSON, so
30 + /// a client that prefers to read `type` can.
31 + fn frame(event: &ChatEvent) -> Event {
32 + let data = serde_json::to_string(event).unwrap_or_else(|_| {
33 + // A ChatEvent is plain data and cannot fail to serialize. If that ever
34 + // changes, a dead connection is worse than a frame the client ignores.
35 + tracing::error!("chat event failed to serialize");
36 + r#"{"type":"gap"}"#.to_owned()
37 + });
38 + Event::default().event(event.name()).data(data)
39 + }
40 +
41 + /// Wrap a [`ChatStream`] as an axum SSE response.
42 + ///
43 + /// The stream owns the hub subscription, so dropping the response (which axum
44 + /// does when the client disconnects) releases the connection slot and prunes the
45 + /// room if it was the last listener. There is nothing for a caller to remember to
46 + /// clean up.
47 + pub fn sse_response(
48 + stream: ChatStream,
49 + ) -> Sse<impl Stream<Item = Result<Event, Infallible>> + Send> {
50 + let events = async_stream::stream! {
51 + let mut stream = stream;
52 + while let Some(event) = stream.next().await {
53 + yield Ok(frame(&event));
54 + }
55 + };
56 +
57 + Sse::new(events).keep_alive(
58 + KeepAlive::new()
59 + .interval(KEEPALIVE_INTERVAL)
60 + .text("keepalive"),
61 + )
62 + }
63 +
64 + #[cfg(test)]
65 + mod tests {
66 + use super::*;
67 + use crate::ids::{MessageId, RoomId, UserId};
68 + use crate::message::Message;
69 + use uuid::Uuid;
70 +
71 + fn message(id: i64) -> Message {
72 + Message {
73 + id: MessageId(id),
74 + room_id: RoomId(Uuid::nil()),
75 + author_id: UserId(Uuid::nil()),
76 + body_html: "hi".into(),
77 + created_at: 0,
78 + nonce: None,
79 + }
80 + }
81 +
82 + /// The SSE wire format is a contract with the client island, so assert on the
83 + /// bytes rather than on the builder.
84 + ///
85 + /// `Event` exposes no accessor for its encoded form, so this reads the `Debug`
86 + /// output, which embeds the real buffer. If an axum upgrade ever makes that
87 + /// opaque these assertions go quiet rather than failing, so
88 + /// [`debug_output_still_exposes_the_buffer`] guards the technique itself.
89 + fn wire(event: &ChatEvent) -> String {
90 + format!("{:?}", frame(event))
91 + }
92 +
93 + #[test]
94 + fn debug_output_still_exposes_the_buffer() {
95 + assert!(
96 + wire(&ChatEvent::Gap).contains("event: gap\\n"),
97 + "Event's Debug no longer embeds the encoded frame, so the wire-format \
98 + assertions in this module have stopped checking anything"
99 + );
100 + }
101 +
102 + #[test]
103 + fn every_variant_frames_with_its_own_event_name() {
104 + for (event, name) in [
105 + (ChatEvent::Message(message(1)), "message"),
106 + (ChatEvent::Delete { id: MessageId(1) }, "delete"),
107 + (
108 + ChatEvent::Purge {
109 + author_id: UserId(Uuid::nil()),
110 + },
111 + "purge",
112 + ),
113 + (ChatEvent::Gap, "gap"),
114 + ] {
115 + // Match the SSE `event:` field specifically. A bare `contains(name)`
116 + // would also be satisfied by the tagged JSON in the data field, which
117 + // means it would pass even if the event name were never set.
118 + assert!(
119 + wire(&event).contains(&format!("event: {name}\\n")),
120 + "expected an SSE event field named {name}, got: {}",
121 + wire(&event)
122 + );
123 + }
124 + }
125 +
126 + #[test]
127 + fn body_html_is_carried_verbatim() {
128 + // The host renders through docengine before publishing; the transport
129 + // must not re-encode or re-escape what it was given.
130 + let mut m = message(1);
131 + m.body_html = "<em>hi</em>".into();
132 + let json = serde_json::to_string(&ChatEvent::Message(m)).unwrap();
133 + assert!(json.contains(r#"<em>hi</em>"#), "{json}");
134 + }
135 +
136 + #[test]
137 + fn a_frame_never_contains_a_bare_newline() {
138 + // SSE is newline-delimited. A payload with a raw newline would split one
139 + // event into two malformed ones, so the JSON encoding has to be the thing
140 + // that protects the framing.
141 + let mut m = message(1);
142 + m.body_html = "line one\nline two".into();
143 + let json = serde_json::to_string(&ChatEvent::Message(m)).unwrap();
144 + assert!(!json.contains('\n'), "raw newline reached the frame: {json}");
145 + assert!(json.contains(r"\n"), "newline should survive escaped: {json}");
146 + }
147 + }
A shared/livechat/src/stream.rs +325
@@ -0,0 +1,325 @@
1 + //! Backlog replay joined to the live feed.
2 + //!
3 + //! # The ordering that matters
4 + //!
5 + //! Every deploy is a symlink swap and a restart with no connection draining, so
6 + //! every client reconnects and asks to resume from a cursor. Serving that means
7 + //! reading stored messages and then following the live feed, and the order those
8 + //! two things happen in is the whole correctness question.
9 + //!
10 + //! Fetch the backlog first and then subscribe, and every message sent between the
11 + //! two is lost: it is too new to be in the query result and too old to be in the
12 + //! channel. Nothing errors, the client just has a hole exactly where it was busy
13 + //! reconnecting, which is the worst possible moment.
14 + //!
15 + //! So [`ChatStream::open`] subscribes first, then fetches. That inverts the
16 + //! problem into an overlap instead of a gap, and an overlap is fixable: anything
17 + //! the live feed replays that the backlog already covered is dropped by id. The
18 + //! host cannot get this wrong from the outside, because it hands over a fetch
19 + //! closure rather than a finished backlog.
20 +
21 + use crate::error::ChatError;
22 + use crate::event::ChatEvent;
23 + use crate::hub::{Hub, Subscription};
24 + use crate::ids::{MessageId, RoomId, UserId};
25 + use crate::message::Message;
26 +
27 + /// Backlog replay followed by the live feed, deduplicated across the seam.
28 + pub struct ChatStream {
29 + backlog: std::vec::IntoIter<Message>,
30 + live: Subscription,
31 + /// Highest id served from the backlog. Live messages at or below it are
32 + /// duplicates from the overlap window and are dropped.
33 + high_water: Option<MessageId>,
34 + }
35 +
36 + impl ChatStream {
37 + /// Subscribe, then fetch the backlog, then serve one followed by the other.
38 + ///
39 + /// `fetch` receives the client's cursor: `Some(id)` to resume after a known
40 + /// message, `None` for a fresh connection, where the host should return
41 + /// whatever recent window it wants a new arrival to see. It is called after
42 + /// the subscription is live, which is the point.
43 + ///
44 + /// The host is expected to return messages in ascending id order. If it does
45 + /// not, the seam still cannot duplicate (the high-water mark is taken from
46 + /// the maximum, not the last), but the client will see the backlog in
47 + /// whatever order it was given.
48 + ///
49 + /// # Errors
50 + ///
51 + /// [`ChatError::ConnectionLimit`] if the hub is at capacity, or whatever
52 + /// `fetch` fails with.
53 + pub async fn open<F, Fut>(
54 + hub: &Hub,
55 + room: RoomId,
56 + user: UserId,
57 + after: Option<MessageId>,
58 + fetch: F,
59 + ) -> Result<Self, ChatError>
60 + where
61 + F: FnOnce(Option<MessageId>) -> Fut,
62 + Fut: Future<Output = Result<Vec<Message>, ChatError>>,
63 + {
64 + // Subscribe first. Everything published from here on is buffered for us
65 + // while the fetch runs, so the two sources overlap rather than gap.
66 + let live = hub.subscribe(room, user)?;
67 +
68 + let backlog = fetch(after).await?;
69 + let high_water = backlog.iter().map(|m| m.id).max();
70 +
71 + Ok(Self {
72 + backlog: backlog.into_iter(),
73 + live,
74 + high_water,
75 + })
76 + }
77 +
78 + /// Next event, or `None` when the room is gone.
79 + ///
80 + /// Serves the backlog to exhaustion first, then the live feed.
81 + pub async fn next(&mut self) -> Option<ChatEvent> {
82 + if let Some(message) = self.backlog.next() {
83 + return Some(ChatEvent::Message(message));
84 + }
85 +
86 + loop {
87 + let event = self.live.recv().await?;
88 +
89 + // Drop only what the backlog already served. Deletes and purges pass
90 + // through regardless: a delete that lands during the overlap refers
91 + // to a message the client has, so suppressing it would leave that
92 + // message on screen forever.
93 + if let ChatEvent::Message(m) = &event
94 + && let Some(high) = self.high_water
95 + && m.id <= high
96 + {
97 + continue;
98 + }
99 +
100 + return Some(event);
101 + }
102 + }
103 +
104 + /// The room being served.
105 + pub fn room(&self) -> RoomId {
106 + self.live.room()
107 + }
108 + }
109 +
110 + #[cfg(test)]
111 + mod tests {
112 + use super::*;
113 + use crate::hub::HubLimits;
114 + use crate::ids::Nonce;
115 + use std::sync::Arc;
116 + use std::sync::atomic::{AtomicUsize, Ordering};
117 + use uuid::Uuid;
118 +
119 + fn message(room: RoomId, id: i64) -> Message {
120 + Message {
121 + id: MessageId(id),
122 + room_id: room,
123 + author_id: UserId(Uuid::nil()),
124 + body_html: format!("m{id}"),
125 + created_at: 0,
126 + nonce: None,
127 + }
128 + }
129 +
130 + fn ids(events: &[ChatEvent]) -> Vec<i64> {
131 + events
132 + .iter()
133 + .filter_map(|e| match e {
134 + ChatEvent::Message(m) => Some(m.id.0),
135 + _ => None,
136 + })
137 + .collect()
138 + }
139 +
140 + async fn drain(stream: &mut ChatStream, n: usize) -> Vec<ChatEvent> {
141 + let mut out = Vec::new();
142 + for _ in 0..n {
143 + match tokio::time::timeout(std::time::Duration::from_secs(5), stream.next()).await {
144 + Ok(Some(e)) => out.push(e),
145 + _ => break,
146 + }
147 + }
148 + out
149 + }
150 +
151 + #[tokio::test]
152 + async fn backlog_is_served_before_live() {
153 + let hub = Hub::new(HubLimits::default());
154 + let room = RoomId(Uuid::new_v4());
155 +
156 + let mut stream = ChatStream::open(&hub, room, UserId(Uuid::nil()), None, |_| async {
157 + Ok(vec![message(room, 1), message(room, 2)])
158 + })
159 + .await
160 + .unwrap();
161 +
162 + hub.publish(room, ChatEvent::Message(message(room, 3)));
163 +
164 + assert_eq!(ids(&drain(&mut stream, 3).await), vec![1, 2, 3]);
165 + }
166 +
167 + #[tokio::test]
168 + async fn a_message_published_during_the_fetch_is_not_lost() {
169 + // The failure this exists for: fetch-then-subscribe drops anything sent
170 + // in the window between the two.
171 + let hub = Hub::new(HubLimits::default());
172 + let room = RoomId(Uuid::new_v4());
173 + let hub_for_fetch = hub.clone();
174 +
175 + let mut stream = ChatStream::open(&hub, room, UserId(Uuid::nil()), None, |_| async move {
176 + // Published while the fetch is in flight, and deliberately absent
177 + // from what the fetch returns, exactly as a real query would miss a
178 + // row committed after it ran.
179 + hub_for_fetch.publish(room, ChatEvent::Message(message(room, 7)));
180 + Ok(vec![message(room, 5), message(room, 6)])
181 + })
182 + .await
183 + .unwrap();
184 +
185 + assert_eq!(
186 + ids(&drain(&mut stream, 3).await),
187 + vec![5, 6, 7],
188 + "the message sent during the fetch must still arrive"
189 + );
190 + }
191 +
192 + #[tokio::test]
193 + async fn the_overlap_does_not_duplicate() {
194 + // The other side of subscribe-first: a message can appear in both the
195 + // backlog and the live feed. It must be served once.
196 + let hub = Hub::new(HubLimits::default());
197 + let room = RoomId(Uuid::new_v4());
198 + let hub_for_fetch = hub.clone();
199 +
200 + let mut stream = ChatStream::open(&hub, room, UserId(Uuid::nil()), None, |_| async move {
201 + hub_for_fetch.publish(room, ChatEvent::Message(message(room, 2)));
202 + hub_for_fetch.publish(room, ChatEvent::Message(message(room, 3)));
203 + // The fetch sees both, because they committed before it ran.
204 + Ok(vec![message(room, 1), message(room, 2), message(room, 3)])
205 + })
206 + .await
207 + .unwrap();
208 +
209 + hub.publish(room, ChatEvent::Message(message(room, 4)));
210 +
211 + assert_eq!(ids(&drain(&mut stream, 4).await), vec![1, 2, 3, 4]);
212 + }
213 +
214 + #[tokio::test]
215 + async fn deletes_in_the_overlap_are_not_suppressed() {
216 + // A delete carries no id that can be compared against the high-water
217 + // mark, and it refers to a message the client already has. Dropping it
218 + // would leave a deleted message on screen until reload.
219 + let hub = Hub::new(HubLimits::default());
220 + let room = RoomId(Uuid::new_v4());
221 + let hub_for_fetch = hub.clone();
222 +
223 + let mut stream = ChatStream::open(&hub, room, UserId(Uuid::nil()), None, |_| async move {
224 + hub_for_fetch.publish(
225 + room,
226 + ChatEvent::Delete {
227 + id: MessageId(1),
228 + },
229 + );
230 + Ok(vec![message(room, 1), message(room, 2)])
231 + })
232 + .await
233 + .unwrap();
234 +
235 + let events = drain(&mut stream, 3).await;
236 + assert!(
237 + events.iter().any(|e| matches!(
238 + e,
239 + ChatEvent::Delete {
240 + id: MessageId(1)
241 + }
242 + )),
243 + "delete during the overlap must pass through: {events:?}"
244 + );
245 + }
246 +
247 + #[tokio::test]
248 + async fn the_cursor_reaches_the_fetch() {
249 + let hub = Hub::new(HubLimits::default());
250 + let room = RoomId(Uuid::new_v4());
251 + let seen = Arc::new(AtomicUsize::new(0));
252 + let seen_clone = seen.clone();
253 +
254 + let _stream = ChatStream::open(
255 + &hub,
256 + room,
257 + UserId(Uuid::nil()),
258 + Some(MessageId(42)),
259 + move |after| async move {
260 + assert_eq!(after, Some(MessageId(42)));
261 + seen_clone.fetch_add(1, Ordering::SeqCst);
262 + Ok(vec![])
263 + },
264 + )
265 + .await
266 + .unwrap();
267 +
268 + assert_eq!(seen.load(Ordering::SeqCst), 1);
269 + }
270 +
271 + #[tokio::test]
272 + async fn an_empty_backlog_goes_straight_to_live() {
273 + let hub = Hub::new(HubLimits::default());
274 + let room = RoomId(Uuid::new_v4());
275 +
276 + let mut stream =
277 + ChatStream::open(&hub, room, UserId(Uuid::nil()), None, |_| async { Ok(vec![]) })
278 + .await
279 + .unwrap();
280 +
281 + hub.publish(room, ChatEvent::Message(message(room, 1)));
282 + assert_eq!(ids(&drain(&mut stream, 1).await), vec![1]);
283 + }
284 +
285 + #[tokio::test]
286 + async fn a_failing_fetch_releases_the_connection_slot() {
287 + let hub = Hub::new(HubLimits::default());
288 + let room = RoomId(Uuid::new_v4());
289 +
290 + let result = ChatStream::open(&hub, room, UserId(Uuid::nil()), None, |_| async {
291 + Err(ChatError::MessageEmpty)
292 + })
293 + .await;
294 +
295 + assert!(result.is_err());
296 + assert_eq!(
297 + hub.connection_count(),
298 + 0,
299 + "the subscription taken before the fetch must be released when it fails"
300 + );
301 + assert_eq!(hub.room_count(), 0);
302 + }
303 +
304 + #[tokio::test]
305 + async fn the_nonce_survives_the_seam() {
306 + // The sender's own optimistic message must arrive carrying its nonce, or
307 + // the client cannot reconcile and shows it twice.
308 + let hub = Hub::new(HubLimits::default());
309 + let room = RoomId(Uuid::new_v4());
310 +
311 + let mut stream =
312 + ChatStream::open(&hub, room, UserId(Uuid::nil()), None, |_| async { Ok(vec![]) })
313 + .await
314 + .unwrap();
315 +
316 + let mut m = message(room, 1);
317 + m.nonce = Nonce::parse("abc");
318 + hub.publish(room, ChatEvent::Message(m));
319 +
320 + match drain(&mut stream, 1).await.first() {
321 + Some(ChatEvent::Message(m)) => assert_eq!(m.nonce, Nonce::parse("abc")),
322 + other => panic!("expected a message carrying its nonce, got {other:?}"),
323 + }
324 + }
325 + }