Skip to main content

max / makenotwork

multithreaded: chat routes The room page, the SSE stream, sending, deleting, and the two moderation actions. Chat is reachable end to end from a browser now; the client island and the owner settings screen are what remain. Route order is load-bearing. /p/{slug}/chat is registered before the /p/{slug}/{category} catch-all, or a community would only need a category slugged "chat" to shadow the room. There is a test with exactly that community in it. off 404s rather than 403s, and so does a room the viewer cannot read. A 403 would confirm the room exists and is merely shut, which is a weaker statement than "off must be total": a closed room and an absent community give the same reply. Refusals carry the status a client can act on rather than a generic rejection, and none of them are logged as faults, because a refused sender is expected traffic. Rate limiting returns 429 with Retry-After from the crate's own budget; a full hub returns 503 with Retry-After, since the caller already passed the read gate and it is a resource refusal rather than an authz one. Sending is a POST for the reason the design gives: it inherits CSRF, the per-IP governor and session auth unchanged, with no CSP change needed. Messages render through docengine's chat preset in the store closure, so sanitization stays in one place and the crate never sees markdown. The author is attached before the message is published, so every listener gets a renderable frame instead of an id to resolve. The chat moderation routes apply the same target protections the forum's ban handler does. Chat is a second door onto community_bans and must not be a way around them. The backlog is server-rendered for the first paint, so the page is readable before any JS runs and stays readable if the island fails. 21 route tests; 357 integration and 192 unit tests pass, clippy and fmt clean. The stream's happy path is not driven through the test client, which collects the whole body and would hang on a stream that never ends; that is noted where the test would have gone.
Co-Authored-By
Claude Opus 5 (1M context) <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-08-08 19:44 UTC
Signed with PGP, not checked
Commit: 64f9fad9e3c2f5970521e9838595c1ce6fd8487d
Parent: ecd6151
8 files changed, +1125 insertions, -8 deletions
M multithreaded/Cargo.lock +8 -8
@@ -4862,14 +4862,6 @@
4862 4862 source = "registry+https://github.com/rust-lang/crates.io-index"
4863 4863 checksum = "b8848ee67ecc8aedbaf3e4122217aff892639231befc6a1b58d29fff4c2cabaa"
4864 4864
4865 - [[patch.unused]]
4866 - name = "kberg"
4867 - version = "0.1.0"
4868 -
4869 - [[patch.unused]]
4870 - name = "painhours"
4871 - version = "0.1.0"
4872 -
4873 4865 [[patch.unused]]
4874 4866 name = "synckit-client"
4875 4867 version = "0.8.0"
@@ -4877,3 +4869,11 @@
4877 4869 [[patch.unused]]
4878 4870 name = "synckit-config"
4879 4871 version = "0.2.0"
4872 +
4873 + [[patch.unused]]
4874 + name = "kberg"
4875 + version = "0.1.0"
4876 +
4877 + [[patch.unused]]
4878 + name = "painhours"
4879 + version = "0.1.0"
M multithreaded/src/routes/mod.rs +13
@@ -2,6 +2,7 @@
2 2
3 3 mod account;
4 4 mod admin;
5 + mod chat;
5 6 mod flagging;
6 7 mod forum;
7 8 pub(crate) mod helpers;
@@ -214,6 +215,16 @@
214 215 "/p/{slug}/uploads/{id}/remove",
215 216 post(uploads::remove_image_handler),
216 217 )
218 + .route("/p/{slug}/chat/send", post(chat::chat_send))
219 + .route(
220 + "/p/{slug}/chat/messages/{message_id}/delete",
221 + post(chat::chat_delete_message),
222 + )
223 + .route(
224 + "/p/{slug}/chat/moderation/timeout",
225 + post(chat::chat_timeout_user),
226 + )
227 + .route("/p/{slug}/chat/moderation/ban", post(chat::chat_ban_user))
217 228 .route_layer(GovernorLayer::new(write_rate_limit.clone()));
218 229
219 230 // Search, rate limited per IP (expensive full-text queries)
@@ -320,6 +331,8 @@
320 331 "/p/{slug}/moderation/threads/{thread_id}/restore",
321 332 post(moderation::restore_thread_handler),
322 333 )
334 + .route("/p/{slug}/chat", get(chat::chat_page))
335 + .route("/p/{slug}/chat/stream", get(chat::chat_stream))
323 336 .route("/p/{slug}/{category}", get(forum::category))
324 337 .route("/p/{slug}/{category}/new", get(forum::new_thread))
325 338 .route("/p/{slug}/{category}/{thread_id}", get(forum::thread))
M multithreaded/src/templates/mod.rs +1
@@ -54,6 +54,7 @@
54 54 UserProfileTemplate,
55 55 ModerationTemplate,
56 56 ModLogTemplate,
57 + ChatTemplate,
57 58 DeletedThreadsTemplate,
58 59 AdminDashboardTemplate,
59 60 TrackedThreadsTemplate,
M multithreaded/src/templates/public.rs +37
@@ -633,6 +633,43 @@
633 633 pub pagination: Pagination,
634 634 }
635 635
636 + /// One chat message, server-rendered for the first paint.
637 + ///
638 + /// The room is hydrated by the client island after load, but the initial
639 + /// window is rendered here so the page is readable before any JS runs and
640 + /// stays readable if the island fails.
641 + pub struct ChatMessageRow {
642 + pub id: i64,
643 + pub author_id: String,
644 + pub author_name: String,
645 + pub avatar_url: Option<String>,
646 + /// Already sanitized by docengine's chat preset at insert.
647 + pub body_html: String,
648 + /// Unix seconds; the client renders it in the viewer's locale.
649 + pub created_at: i64,
650 + }
651 +
652 + /// The chat room page.
653 + #[derive(Template)]
654 + #[template(path = "pages/chat.html")]
655 + pub struct ChatTemplate {
656 + pub csrf_token: CsrfTokenOption,
657 + pub session_user: Option<TemplateSessionUser>,
658 + pub mnw_base_url: std::sync::Arc<str>,
659 + pub community_name: String,
660 + pub community_slug: String,
661 + /// Frozen or archived: the backlog shows, the composer does not.
662 + pub read_only: bool,
663 + /// Whether to render the composer at all. One question answered once in the
664 + /// handler rather than three conditions here.
665 + pub can_send: bool,
666 + pub is_moderator: bool,
667 + pub max_message_len: usize,
668 + pub messages: Vec<ChatMessageRow>,
669 + /// Highest message id in the first paint; the island resumes from it.
670 + pub cursor: i64,
671 + }
672 +
636 673 // Admin templates
637 674
638 675 /// Row for communities in admin dashboard.
M multithreaded/tests/workflows/mod.rs +1
@@ -3,6 +3,7 @@
3 3 mod admin_queries;
4 4 mod auth;
5 5 mod bans;
6 + mod chat_routes;
6 7 mod chat_storage;
7 8 mod chat_traits;
8 9 mod community_state;
A multithreaded/src/routes/chat.rs +500
@@ -1,0 +1,564 @@
1 + //! Chat routes: the room page, the SSE stream, sending, and moderation.
2 + //!
3 + //! Every handler starts the same way, by resolving the slug to a `Room` through
4 + //! [`MtChatRooms`] and refusing anything that is not open. That is deliberate
5 + //! repetition of one call, not a copied predicate: the room carries the state
6 + //! and the retention, and each handler's gates read it rather than re-deriving
7 + //! community facts. See `crate::chat::authz` for why that shape is what keeps
8 + //! chat out of the C1 authz chronic.
9 + //!
10 + //! # Why `off` 404s rather than 403s
11 + //!
12 + //! A community with chat off has no route, no hub room and no affordance. A 403
13 + //! would confirm the room exists and is merely shut, which is a different
14 + //! statement than the design's "off must be total". A closed room and an absent
15 + //! community are the same reply.
16 +
17 + use axum::{
18 + Form, Json,
19 + extract::{Path, Query, State},
20 + http::StatusCode,
21 + response::{IntoResponse, Response},
22 + };
23 + use livechat::{
24 + ChatAuthz, ChatIdentity, ChatRooms, DenyReason, MessageId, Room, RoomState, SendRejection,
25 + SendRequest, UserId,
26 + };
27 + use serde::{Deserialize, Serialize};
28 +
29 + use crate::AppState;
30 + use crate::auth::{MaybeUser, RequireUser, SessionUser};
31 + use crate::chat::{
32 + authz::MtChatAuthz, identity::MtChatIdentity, moderation::MtChatModeration, rooms::MtChatRooms,
33 + };
34 + use crate::templates::ChatTemplate;
35 +
36 + use super::{db_error, get_user_by_username, template_user};
37 +
38 + /// Backlog handed to a client with no cursor.
39 + ///
40 + /// A screenful and some scroll, not the whole retention window: a room holds up
41 + /// to `chat_max_messages` and shipping that on every page load would be a large
42 + /// response for content nobody scrolls back through.
43 + const INITIAL_BACKLOG: i64 = 50;
44 +
45 + /// Ceiling on one replay reply.
46 + ///
47 + /// A client away longer than this walks forward a page at a time, because its
48 + /// cursor advances with each batch. Without the ceiling, a client that has been
49 + /// gone since before the retention window could ask the server to materialize
50 + /// every message in the room in one query.
51 + const MAX_REPLAY: i64 = 200;
52 +
53 + // Resolution and gating
54 +
55 + /// Resolve the slug to an open-or-read-only room, or 404.
56 + async fn open_room(state: &AppState, slug: &str) -> Result<Room, Response> {
57 + let room = MtChatRooms::new(state.db.clone())
58 + .resolve(slug)
59 + .await
60 + .map_err(|e| chat_error(&e))?
61 + .ok_or_else(crate::error_page::not_found)?;
62 +
63 + if room.state == RoomState::Closed {
64 + return Err(crate::error_page::not_found());
65 + }
66 + Ok(room)
67 + }
68 +
69 + /// Build this request's authz gate from the session.
70 + ///
71 + /// The policy is re-read rather than carried on `Room`, which holds only the
72 + /// state the crate needs. One extra cheap read per request, and it keeps the
73 + /// crate's `Room` free of a Multithreaded concept.
74 + async fn gate(
75 + state: &AppState,
76 + slug: &str,
77 + user: Option<&SessionUser>,
78 + ) -> Result<MtChatAuthz, Response> {
79 + let policy = mt_db::queries::get_chat_room_by_slug(&state.db, slug)
80 + .await
81 + .map_err(db_error)?
82 + .map(|row| row.policy)
83 + .ok_or_else(crate::error_page::not_found)?;
84 +
85 + Ok(MtChatAuthz::new(
86 + state.db.clone(),
87 + policy,
88 + user.map(|u| u.user_id),
89 + user.map(|u| u.perks.clone()).unwrap_or_default(),
90 + ))
91 + }
92 +
93 + /// A crate-side failure. Every one of these is a host fault by the time it
94 + /// reaches a handler: refusals come back as values, not errors.
95 + fn chat_error(err: &livechat::ChatError) -> Response {
96 + tracing::error!(error = ?err, "chat error");
97 + crate::error_page::internal_error()
98 + }
99 +
100 + /// Refuse a reader who may not see the room, as a 404 rather than a 403.
101 + async fn require_read(
102 + authz: &MtChatAuthz,
103 + room: &Room,
104 + viewer: Option<UserId>,
105 + ) -> Result<(), Response> {
106 + let allowed = authz
107 + .can_read(viewer, room)
108 + .await
109 + .map_err(|e| chat_error(&e))?;
110 + if allowed {
111 + Ok(())
112 + } else {
113 + Err(crate::error_page::not_found())
114 + }
115 + }
116 +
117 + /// Load a backlog window and fill in each author for display.
118 + async fn backlog(
119 + state: &AppState,
120 + room: &Room,
121 + after: Option<MessageId>,
122 + ) -> Result<Vec<livechat::Message>, livechat::ChatError> {
123 + let rows = match after {
124 + Some(MessageId(cursor)) => {
125 + mt_db::queries::backlog_after(&state.db, room.id.0, cursor, MAX_REPLAY).await
126 + }
127 + None => mt_db::queries::recent_backlog(&state.db, room.id.0, INITIAL_BACKLOG).await,
128 + }
129 + .map_err(livechat::ChatError::host)?;
130 +
131 + let mut messages: Vec<_> = rows
132 + .into_iter()
133 + .map(|r| livechat::Message {
134 + id: MessageId(r.id),
135 + room_id: room.id,
136 + author_id: UserId(r.author_id),
137 + body_html: r.body_html,
138 + created_at: r.created_at,
139 + nonce: None,
140 + author: None,
141 + })
142 + .collect();
143 +
144 + MtChatIdentity::new(state.db.clone(), state.chat_identities.clone())
145 + .attach(&mut messages)
146 + .await?;
147 +
148 + Ok(messages)
149 + }
150 +
151 + // The room page
152 +
153 + #[tracing::instrument(skip_all)]
154 + pub(super) async fn chat_page(
155 + State(state): State<AppState>,
156 + Path(slug): Path<String>,
157 + MaybeUser(user): MaybeUser,
158 + session: tower_sessions::Session,
159 + ) -> Result<Response, Response> {
160 + let room = open_room(&state, &slug).await?;
161 + let authz = gate(&state, &slug, user.as_ref()).await?;
162 + let viewer = user.as_ref().map(|u| UserId(u.user_id));
163 + require_read(&authz, &room, viewer).await?;
164 +
165 + let community = super::get_community(&state.db, &slug).await?;
166 +
167 + // Whether to render the composer at all. A read-only room, a logged-out
168 + // visitor, and a member without the tier all read the same room and none of
169 + // them can type in it, so the answer is one question asked once here rather
170 + // than three conditions in the template.
171 + let can_send = match (&user, room.state) {
172 + (Some(u), RoomState::Open) => authz
173 + .can_write(UserId(u.user_id), &room)
174 + .await
175 + .map_err(|e| chat_error(&e))?
176 + .is_allowed(),
177 + _ => false,
178 + };
179 +
180 + let is_moderator = match &user {
181 + Some(u) => authz
182 + .is_moderator(UserId(u.user_id), &room)
183 + .await
184 + .map_err(|e| chat_error(&e))?,
185 + None => false,
186 + };
187 +
188 + let messages = backlog(&state, &room, None)
189 + .await
190 + .map_err(|e| chat_error(&e))?;
191 +
192 + Ok(ChatTemplate {
193 + csrf_token: Some(crate::csrf::get_or_create_token(&session).await?),
194 + session_user: user
195 + .as_ref()
196 + .map(|u| template_user(u, state.config.platform_admin_id)),
197 + mnw_base_url: state.config.mnw_base_url.clone(),
198 + community_name: community.name,
199 + community_slug: slug,
200 + read_only: room.state == RoomState::ReadOnly,
201 + can_send,
202 + is_moderator,
203 + max_message_len: livechat::MAX_MESSAGE_LEN,
204 + messages: messages.iter().map(chat_message_row).collect(),
205 + cursor: messages.last().map_or(0, |m| m.id.0),
206 + }
207 + .into_response())
208 + }
209 +
210 + /// One rendered message, for the server-rendered first paint.
211 + fn chat_message_row(m: &livechat::Message) -> crate::templates::ChatMessageRow {
212 + crate::templates::ChatMessageRow {
213 + id: m.id.0,
214 + author_id: m.author_id.0.to_string(),
215 + author_name: m
216 + .author
217 + .as_ref()
218 + .map_or_else(|| "Unknown".to_owned(), |a| a.display_name.clone()),
219 + avatar_url: m.author.as_ref().and_then(|a| a.avatar_url.clone()),
220 + body_html: m.body_html.clone(),
221 + created_at: m.created_at,
222 + }
223 + }
224 +
225 + // The stream
226 +
227 + #[derive(Deserialize)]
228 + pub(super) struct StreamQuery {
229 + /// Last message id the client holds. Absent on a first connection.
230 + after: Option<i64>,
231 + }
232 +
233 + #[tracing::instrument(skip_all)]
234 + pub(super) async fn chat_stream(
235 + State(state): State<AppState>,
236 + Path(slug): Path<String>,
237 + Query(query): Query<StreamQuery>,
238 + MaybeUser(user): MaybeUser,
239 + ) -> Result<Response, Response> {
240 + let room = open_room(&state, &slug).await?;
241 + let authz = gate(&state, &slug, user.as_ref()).await?;
242 + let viewer = user.as_ref().map(|u| UserId(u.user_id));
243 + require_read(&authz, &room, viewer).await?;
244 +
245 + // A logged-out reader still occupies a connection slot, so it needs an
246 + // identity for the per-user cap. The room id stands in: anonymous readers of
247 + // one room share a budget rather than being unbounded, which is the correct
248 + // answer when the alternative is a per-IP cap that a proxy collapses anyway.
249 + let listener = viewer.unwrap_or(UserId(room.id.0));
250 +
251 + let stream = state
252 + .chat
253 + .subscribe(room.id, listener, query.after.map(MessageId), |after| {
254 + let state = state.clone();
255 + async move { backlog(&state, &room, after).await }
256 + })
257 + .await
258 + .map_err(|e| match e {
259 + // At capacity is a resource refusal, not an authz one: the caller
260 + // already passed the read gate. 503 with a Retry-After, so a
261 + // reconnecting client backs off instead of hammering.
262 + livechat::ChatError::ConnectionLimit(scope) => {
263 + tracing::warn!(%scope, "chat connection limit reached");
264 + (
265 + StatusCode::SERVICE_UNAVAILABLE,
266 + [("Retry-After", "5")],
267 + "Too many chat connections. Try again shortly.",
268 + )
269 + .into_response()
270 + }
271 + other => chat_error(&other),
272 + })?;
273 +
274 + Ok(livechat::sse_response(stream).into_response())
275 + }
276 +
277 + // Sending
278 +
279 + #[derive(Deserialize)]
280 + pub(super) struct SendForm {
281 + body: String,
282 + /// Client-generated, so the sender's optimistically rendered message
283 + /// reconciles instead of appearing twice.
284 + nonce: Option<String>,
285 + }
286 +
287 + #[derive(Serialize)]
288 + pub(super) struct SendResponse {
289 + id: i64,
290 + #[serde(skip_serializing_if = "Option::is_none")]
291 + nonce: Option<String>,
292 + }
293 +
294 + #[tracing::instrument(skip_all)]
295 + pub(super) async fn chat_send(
296 + State(state): State<AppState>,
297 + Path(slug): Path<String>,
298 + RequireUser(user): RequireUser,
299 + Form(form): Form<SendForm>,
300 + ) -> Result<Json<SendResponse>, Response> {
301 + let room = open_room(&state, &slug).await?;
302 + let authz = gate(&state, &slug, Some(&user)).await?;
303 +
304 + let nonce = form.nonce.as_deref().and_then(livechat::Nonce::parse);
305 + let author = UserId(user.user_id);
306 + let retention_hours = i32::try_from(room.retention.max_age().as_secs() / 3600).unwrap_or(168);
307 +
308 + let sent = state
309 + .chat
310 + .send(
311 + &authz,
312 + SendRequest {
313 + room: &room,
314 + author,
315 + body: &form.body,
316 + nonce: nonce.clone(),
317 + now: std::time::Instant::now(),
318 + },
319 + |body| {
320 + let state = state.clone();
321 + async move {
322 + // Rendered here, once, through docengine's chat preset.
323 + // The crate never sees markdown and never renders: keeping
324 + // sanitization in one place is the whole reason the store
325 + // closure exists.
326 + let html = docengine::render_chat(body.as_str());
327 +
328 + let stored = mt_db::mutations::insert_chat_message(
329 + &state.db,
330 + room.id.0,
331 + author.0,
332 + &html,
333 + retention_hours,
334 + )
335 + .await
336 + .map_err(livechat::ChatError::host)?;
337 +
338 + let mut message = livechat::Message {
339 + id: MessageId(stored.id),
340 + room_id: room.id,
341 + author_id: author,
342 + body_html: html,
343 + created_at: stored.created_at,
344 + nonce: None,
345 + author: None,
346 + };
347 +
348 + // Attach the author before the message is published, so
349 + // every listener gets a renderable frame rather than an id
350 + // they have to resolve themselves.
351 + MtChatIdentity::new(state.db.clone(), state.chat_identities.clone())
352 + .attach(std::slice::from_mut(&mut message))
353 + .await?;
354 +
355 + Ok(message)
356 + }
357 + },
358 + )
359 + .await
360 + .map_err(|e| send_error(&e))?;
361 +
362 + Ok(Json(SendResponse {
363 + id: sent.id.0,
364 + nonce: nonce.map(|n| n.0),
365 + }))
366 + }
367 +
368 + /// Turn a refusal into the status a client can act on.
369 + ///
370 + /// Refusals are expected traffic and must not be logged as faults; only the
371 + /// host-error arm is a real problem.
372 + fn send_error(err: &livechat::ChatError) -> Response {
373 + let livechat::ChatError::Rejected(rejection) = err else {
374 + return chat_error(err);
375 + };
376 +
377 + match rejection {
378 + SendRejection::Invalid(message) => (StatusCode::UNPROCESSABLE_ENTITY, message.clone()),
379 + SendRejection::RoomClosed => (
380 + StatusCode::NOT_FOUND,
381 + "This room is not available.".to_owned(),
382 + ),
383 + SendRejection::RoomReadOnly => (
384 + StatusCode::FORBIDDEN,
385 + "This community is read-only.".to_owned(),
386 + ),
387 + SendRejection::Denied(reason) => return deny_response(reason),
388 + }
389 + .into_response()
390 + }
391 +
392 + fn deny_response(reason: &DenyReason) -> Response {
393 + match reason {
394 + DenyReason::RateLimited { retry_after } => (
395 + StatusCode::TOO_MANY_REQUESTS,
396 + [("Retry-After", retry_after.as_secs().max(1).to_string())],
397 + "You are sending messages too quickly.",
398 + )
399 + .into_response(),
400 + DenyReason::Anonymous => (StatusCode::UNAUTHORIZED, "Sign in to chat.").into_response(),
401 + DenyReason::NotAMember => {
402 + (StatusCode::FORBIDDEN, "Join this community to chat in it.").into_response()
403 + }
404 + DenyReason::Banned => {
405 + (StatusCode::FORBIDDEN, "You are banned from this community.").into_response()
406 + }
407 + DenyReason::Muted => {
408 + (StatusCode::FORBIDDEN, "You are muted in this community.").into_response()
409 + }
410 + DenyReason::Suspended => {
411 + (StatusCode::FORBIDDEN, "Your account has been suspended.").into_response()
412 + }
413 + DenyReason::TierRequired => (
414 + StatusCode::FORBIDDEN,
415 + "Chat here is limited to Fan+ supporters.",
416 + )
417 + .into_response(),
418 + }
419 + }
420 +
421 + // Moderation
422 +
423 + #[tracing::instrument(skip_all)]
424 + pub(super) async fn chat_delete_message(
425 + State(state): State<AppState>,
426 + Path((slug, message_id)): Path<(String, i64)>,
427 + RequireUser(user): RequireUser,
428 + ) -> Result<StatusCode, Response> {
429 + let room = open_room(&state, &slug).await?;
430 + let authz = gate(&state, &slug, Some(&user)).await?;
431 +
432 + let is_moderator = authz
433 + .is_moderator(UserId(user.user_id), &room)
434 + .await
435 + .map_err(|e| chat_error(&e))?;
436 +
437 + // The gate carries the powers; a member's gate cannot express a moderator
438 + // removal, and the impl checks authorship for everyone else.
439 + state
440 + .chat
441 + .delete_message(
442 + &MtChatModeration::new(state.db.clone(), is_moderator),
443 + UserId(user.user_id),
444 + &room,
445 + MessageId(message_id),
446 + )
447 + .await
448 + .map_err(|e| {
449 + tracing::warn!(error = ?e, "chat delete refused");
450 + (StatusCode::FORBIDDEN, "Cannot delete that message.").into_response()
451 + })?;
452 +
453 + Ok(StatusCode::NO_CONTENT)
454 + }
455 +
456 + #[derive(Deserialize)]
457 + pub(super) struct ChatModerationForm {
458 + username: String,
459 + /// Timeout length in seconds. Absent means a ban.
460 + seconds: Option<u64>,
461 + reason: Option<String>,
462 + }
463 +
464 + #[tracing::instrument(skip_all)]
465 + pub(super) async fn chat_timeout_user(
466 + State(state): State<AppState>,
467 + Path(slug): Path<String>,
468 + RequireUser(user): RequireUser,
469 + Form(form): Form<ChatModerationForm>,
470 + ) -> Result<StatusCode, Response> {
471 + let (room, target) = moderation_target(&state, &slug, &user, &form).await?;
472 +
473 + let seconds = form.seconds.unwrap_or(300).clamp(60, 86_400);
474 +
475 + state
476 + .chat
477 + .timeout_user(
478 + &MtChatModeration::new(state.db.clone(), true),
479 + UserId(user.user_id),
480 + &room,
481 + UserId(target),
482 + std::time::Duration::from_secs(seconds),
483 + )
484 + .await
485 + .map_err(|e| chat_error(&e))?;
486 +
487 + Ok(StatusCode::NO_CONTENT)
488 + }
489 +
490 + #[tracing::instrument(skip_all)]
491 + pub(super) async fn chat_ban_user(
492 + State(state): State<AppState>,
493 + Path(slug): Path<String>,
494 + RequireUser(user): RequireUser,
495 + Form(form): Form<ChatModerationForm>,
496 + ) -> Result<StatusCode, Response> {
497 + let (room, target) = moderation_target(&state, &slug, &user, &form).await?;
498 +
499 + let reason = form.reason.as_deref().filter(|r| !r.trim().is_empty());
500 +
Lines truncated
A multithreaded/templates/pages/chat.html +71
@@ -1,0 +1,71 @@
1 + {% extends "base.html" %}
2 +
3 + {% block title %}Chat — {{ community_name }} — Multithreaded{% endblock %}
4 +
5 + {# Chat is ephemeral by design, so there is nothing here worth indexing and a
6 + crawler holding an SSE connection open is a connection slot spent on nobody. #}
7 + {% block head %}<meta name="robots" content="noindex">{% endblock %}
8 +
9 + {% block header %}{% include "partials/site_header.html" %}{% endblock %}
10 +
11 + {% block content %}
12 + <div class="container">
13 + <div class="breadcrumb">
14 + <a href="/">Forums</a>
15 + <span class="sep">/</span>
16 + <a href="/p/{{ community_slug }}">{{ community_name }}</a>
17 + <span class="sep">/</span>
18 + Chat
19 + </div>
20 + <div class="page-header">
21 + <h1>Chat</h1>
22 + </div>
23 +
24 + {# The island reads its configuration from data attributes rather than an
25 + inline script, so the page needs no script-src exception. #}
26 + <div id="chat-room"
27 + class="chat-room"
28 + data-slug="{{ community_slug }}"
29 + data-cursor="{{ cursor }}"
30 + data-max-length="{{ max_message_len }}"
31 + data-can-send="{{ can_send }}"
32 + data-moderator="{{ is_moderator }}">
33 +
34 + <ol class="chat-log" id="chat-log" aria-live="polite" aria-label="Chat messages">
35 + {% for message in messages %}
36 + <li class="chat-message" data-id="{{ message.id }}" data-author="{{ message.author_id }}">
37 + {% if let Some(avatar) = message.avatar_url %}
38 + <img class="chat-avatar" src="{{ avatar }}" alt="" width="24" height="24">
39 + {% endif %}
40 + <span class="chat-author">{{ message.author_name }}</span>
41 + <time class="chat-time" datetime="{{ message.created_at }}"></time>
42 + <span class="chat-body">{{ message.body_html|safe }}</span>
43 + </li>
44 + {% endfor %}
45 + </ol>
46 +
47 + {% if messages.is_empty() %}
48 + <div class="empty-state" id="chat-empty">Nothing said yet.</div>
49 + {% endif %}
50 +
51 + {% if read_only %}
52 + <div class="chat-notice">This community is read-only, so chat is too.</div>
53 + {% elif can_send %}
54 + <form class="chat-composer" id="chat-composer" method="post"
55 + action="/p/{{ community_slug }}/chat/send">
56 + {% if let Some(token) = csrf_token %}
57 + <input type="hidden" name="csrf_token" value="{{ token }}">
58 + {% endif %}
59 + <label class="sr-only" for="chat-input">Message</label>
60 + <input type="text" id="chat-input" name="body" autocomplete="off"
61 + maxlength="{{ max_message_len }}" placeholder="Say something">
62 + <button type="submit">Send</button>
63 + </form>
64 + {% elif session_user.is_some() %}
65 + <div class="chat-notice">You cannot send messages in this room.</div>
66 + {% else %}
67 + <div class="chat-notice"><a href="/auth/login">Sign in</a> to join the conversation.</div>
68 + {% endif %}
69 + </div>
70 + </div>
71 + {% endblock %}
A multithreaded/tests/workflows/chat_routes.rs +494
@@ -1,0 +1,494 @@
1 + //! Tests for the chat routes.
2 + //!
3 + //! The trait tests cover the authz matrix as a decision; these cover what a
4 + //! browser actually gets back, which is a different question: the status code
5 + //! for each refusal, that `off` is indistinguishable from absent, that the
6 + //! route does not collide with the category catch-all, and that CSRF and the
7 + //! rate limiter apply to sending the way they apply to every other write.
8 +
9 + use axum::http::StatusCode;
10 + use mt_core::types::ChatPolicy;
11 + use uuid::Uuid;
12 +
13 + use crate::harness::TestHarness;
14 +
15 + async fn set_policy(h: &TestHarness, community: Uuid, policy: ChatPolicy) {
16 + sqlx::query("UPDATE communities SET chat_policy = $1 WHERE id = $2")
17 + .bind(policy.as_str())
18 + .bind(community)
19 + .execute(&h.db)
20 + .await
21 + .expect("set policy");
22 + }
23 +
24 + /// Insert a user without logging in as them.
25 + ///
26 + /// `login_as` mints a fresh uuid on every call and replaces the session, so
27 + /// there is no way back to an earlier user. Tests therefore create everyone up
28 + /// front and log in the actor they care about last.
29 + async fn user(h: &TestHarness, username: &str) -> Uuid {
30 + let id = Uuid::new_v4();
31 + sqlx::query("INSERT INTO users (mnw_account_id, username, display_name) VALUES ($1, $2, $3)")
32 + .bind(id)
33 + .bind(username)
34 + .bind(username)
35 + .execute(&h.db)
36 + .await
37 + .expect("insert user");
38 + id
39 + }
40 +
41 + /// A community with chat on and an owner who is not signed in yet.
42 + async fn room(h: &mut TestHarness, policy: ChatPolicy) -> (Uuid, Uuid) {
43 + let id = h.create_community("Test", "test").await;
44 + h.create_category(id, "General", "general").await;
45 + let owner = user(h, "owner").await;
46 + h.add_membership(owner, id, "owner").await;
47 + set_policy(h, id, policy).await;
48 + (id, owner)
49 + }
50 +
51 + /// Log in as an already-created user.
52 + async fn sign_in(h: &mut TestHarness, id: Uuid, username: &str) {
53 + h.client.get("/").await;
54 + h.client
55 + .post_json(
56 + "/_test/login",
57 + &serde_json::json!({ "user_id": id.to_string(), "username": username }).to_string(),
58 + )
59 + .await;
60 + }
61 +
62 + // Reachability
63 +
64 + #[sqlx::test]
65 + async fn chat_off_is_indistinguishable_from_a_community_without_chat(_pool: sqlx::PgPool) {
66 + // `off` must be total: not a 403, which would confirm the room exists and
67 + // is merely shut.
68 + let mut h = TestHarness::new().await;
69 + let (_id, _owner) = room(&mut h, ChatPolicy::Off).await;
70 +
71 + assert_eq!(
72 + h.client.get("/p/test/chat").await.status,
73 + StatusCode::NOT_FOUND
74 + );
75 + assert_eq!(
76 + h.client.get("/p/nope/chat").await.status,
77 + StatusCode::NOT_FOUND,
78 + "and a community that does not exist reads the same"
79 + );
80 + }
81 +
82 + #[sqlx::test]
83 + async fn the_chat_route_is_not_swallowed_by_the_category_catch_all(_pool: sqlx::PgPool) {
84 + // `/p/{slug}/{category}` would match `/p/test/chat` if chat were registered
85 + // after it. A community with a category actually named "chat" is the case
86 + // that would hide the bug.
87 + let mut h = TestHarness::new().await;
88 + let (id, owner) = room(&mut h, ChatPolicy::Members).await;
89 + h.create_category(id, "Chat", "chat").await;
90 + sign_in(&mut h, owner, "owner").await;
91 +
92 + let resp = h.client.get("/p/test/chat").await;
93 + assert_eq!(resp.status, StatusCode::OK);
94 + assert!(
95 + resp.text.contains("chat-room"),
96 + "the chat page must win over a category of the same slug"
97 + );
98 + }
99 +
100 + #[sqlx::test]
101 + async fn an_enabled_room_renders_its_backlog_server_side(_pool: sqlx::PgPool) {
102 + // The page must be readable before any JS runs, and stay readable if the
103 + // island fails to load.
104 + let mut h = TestHarness::new().await;
105 + let (id, owner) = room(&mut h, ChatPolicy::Members).await;
106 +
107 + mt_db::mutations::insert_chat_message(&h.db, id, owner, "<p>hello room</p>", 168)
108 + .await
109 + .unwrap();
110 + sign_in(&mut h, owner, "owner").await;
111 +
112 + let resp = h.client.get("/p/test/chat").await;
113 + assert_eq!(resp.status, StatusCode::OK);
114 + assert!(
115 + resp.text.contains("hello room"),
116 + "backlog is server-rendered"
117 + );
118 + }
119 +
120 + #[sqlx::test]
121 + async fn a_logged_out_visitor_sees_public_read_and_not_members(_pool: sqlx::PgPool) {
122 + let mut h = TestHarness::new().await;
123 + let (id, _owner) = room(&mut h, ChatPolicy::PublicRead).await;
124 +
125 + assert_eq!(h.client.get("/p/test/chat").await.status, StatusCode::OK);
126 +
127 + set_policy(&h, id, ChatPolicy::Members).await;
128 + assert_eq!(
129 + h.client.get("/p/test/chat").await.status,
130 + StatusCode::NOT_FOUND,
131 + "members-only chat is not visible logged out"
132 + );
133 + }
134 +
135 + #[sqlx::test]
136 + async fn a_logged_out_visitor_gets_no_composer(_pool: sqlx::PgPool) {
137 + let mut h = TestHarness::new().await;
138 + room(&mut h, ChatPolicy::PublicRead).await;
139 +
140 + let resp = h.client.get("/p/test/chat").await;
141 + assert!(!resp.text.contains("chat-composer"));
142 + assert!(resp.text.contains("Sign in"));
143 + }
144 +
145 + #[sqlx::test]
146 + async fn a_read_only_community_shows_the_room_without_a_composer(_pool: sqlx::PgPool) {
147 + let mut h = TestHarness::new().await;
148 + let (id, owner) = room(&mut h, ChatPolicy::Members).await;
149 +
150 + sqlx::query("UPDATE communities SET state = 'frozen' WHERE id = $1")
151 + .bind(id)
152 + .execute(&h.db)
153 + .await
154 + .unwrap();
155 + sign_in(&mut h, owner, "owner").await;
156 +
157 + let resp = h.client.get("/p/test/chat").await;
158 + assert_eq!(resp.status, StatusCode::OK);
159 + assert!(!resp.text.contains("chat-composer"));
160 + assert!(resp.text.contains("read-only"));
161 + }
162 +
163 + // Sending
164 +
165 + #[sqlx::test]
166 + async fn a_member_sends_and_gets_the_id_and_nonce_back(_pool: sqlx::PgPool) {
167 + let mut h = TestHarness::new().await;
168 + let (id, owner) = room(&mut h, ChatPolicy::Members).await;
169 + sign_in(&mut h, owner, "owner").await;
170 +
171 + h.client.get("/p/test/chat").await;
172 + let resp = h
173 + .client
174 + .post_form("/p/test/chat/send", "body=hello&nonce=n1")
175 + .await;
176 +
177 + assert_eq!(resp.status, StatusCode::OK);
178 + let json: serde_json::Value = serde_json::from_str(&resp.text).expect("json reply");
179 + assert!(json["id"].as_i64().unwrap() > 0);
180 + assert_eq!(json["nonce"], "n1", "the sender reconciles by nonce");
181 +
182 + let stored = mt_db::queries::recent_backlog(&h.db, id, 10).await.unwrap();
183 + assert_eq!(stored.len(), 1);
184 + assert!(stored[0].body_html.contains("hello"));
185 + }
186 +
187 + #[sqlx::test]
188 + async fn a_sent_message_is_rendered_and_sanitized(_pool: sqlx::PgPool) {
189 + // Sanitization happens once, at insert, through docengine's chat preset.
190 + // The store layer never sees markdown and must never see raw HTML.
191 + //
192 + // Two messages rather than one: a line starting with a raw tag is parsed
193 + // as an HTML block, so the rest of that line never reaches the inline
194 + // renderer. Asserting both properties on one message would be testing the
195 + // block parser, not the two things worth pinning.
196 + let mut h = TestHarness::new().await;
197 + let (id, owner) = room(&mut h, ChatPolicy::Members).await;
198 + sign_in(&mut h, owner, "owner").await;
199 + h.client.get("/p/test/chat").await;
200 +
201 + h.client
202 + .post_form("/p/test/chat/send", "body=%2Aemphasis%2A")
203 + .await;
204 + h.client
205 + .post_form(
206 + "/p/test/chat/send",
207 + "body=%3Cscript%3Ealert(1)%3C%2Fscript%3E",
208 + )
209 + .await;
210 +
211 + let stored = mt_db::queries::recent_backlog(&h.db, id, 10).await.unwrap();
212 + assert_eq!(stored.len(), 2);
213 + assert!(
214 + stored[0].body_html.contains("<em>"),
215 + "markdown renders: {}",
216 + stored[0].body_html
217 + );
218 + assert!(
219 + !stored[1].body_html.contains("<script"),
220 + "raw HTML must not survive: {}",
221 + stored[1].body_html
222 + );
223 + }
224 +
225 + #[sqlx::test]
226 + async fn sending_without_a_csrf_token_is_refused(_pool: sqlx::PgPool) {
227 + // Chat sends are a POST precisely so they inherit this middleware unchanged.
228 + let mut h = TestHarness::new().await;
229 + let (_id, owner) = room(&mut h, ChatPolicy::Members).await;
230 + sign_in(&mut h, owner, "owner").await;
231 +
232 + let resp = h
233 + .client
234 + .post_form_no_csrf("/p/test/chat/send", "body=hello")
235 + .await;
236 + assert_eq!(resp.status, StatusCode::FORBIDDEN);
237 + }
238 +
239 + #[sqlx::test]
240 + async fn an_empty_message_is_rejected_as_unprocessable(_pool: sqlx::PgPool) {
241 + let mut h = TestHarness::new().await;
242 + let (_id, owner) = room(&mut h, ChatPolicy::Members).await;
243 + sign_in(&mut h, owner, "owner").await;
244 +
245 + h.client.get("/p/test/chat").await;
246 + let resp = h.client.post_form("/p/test/chat/send", "body=+++").await;
247 + assert_eq!(resp.status, StatusCode::UNPROCESSABLE_ENTITY);
248 + }
249 +
250 + #[sqlx::test]
251 + async fn an_overlong_message_is_rejected(_pool: sqlx::PgPool) {
252 + let mut h = TestHarness::new().await;
253 + let (_id, owner) = room(&mut h, ChatPolicy::Members).await;
254 + sign_in(&mut h, owner, "owner").await;
255 +
256 + h.client.get("/p/test/chat").await;
257 + let body = format!("body={}", "x".repeat(livechat::MAX_MESSAGE_LEN + 1));
258 + let resp = h.client.post_form("/p/test/chat/send", &body).await;
259 + assert_eq!(resp.status, StatusCode::UNPROCESSABLE_ENTITY);
260 + }
261 +
262 + #[sqlx::test]
263 + async fn a_non_member_is_told_to_join_rather_than_that_it_failed(_pool: sqlx::PgPool) {
264 + let mut h = TestHarness::new().await;
265 + room(&mut h, ChatPolicy::PublicRead).await;
266 +
267 + let stranger = user(&h, "stranger").await;
268 + sign_in(&mut h, stranger, "stranger").await;
269 + h.client.get("/p/test/chat").await;
270 + let resp = h.client.post_form("/p/test/chat/send", "body=hi").await;
271 +
272 + assert_eq!(resp.status, StatusCode::FORBIDDEN);
273 + assert!(resp.text.contains("Join this community"));
274 + }
275 +
276 + #[sqlx::test]
277 + async fn a_muted_member_is_told_they_are_muted(_pool: sqlx::PgPool) {
278 + let mut h = TestHarness::new().await;
279 + let (id, owner) = room(&mut h, ChatPolicy::Members).await;
280 +
281 + let member = user(&h, "member").await;
282 + h.add_membership(member, id, "member").await;
283 + h.ban_user(id, member, owner, "mute").await;
284 + sign_in(&mut h, member, "member").await;
285 +
286 + h.client.get("/p/test/chat").await;
287 + let resp = h.client.post_form("/p/test/chat/send", "body=hi").await;
288 +
289 + assert_eq!(resp.status, StatusCode::FORBIDDEN);
290 + assert!(resp.text.contains("muted"));
291 + }
292 +
293 + #[sqlx::test]
294 + async fn sending_to_a_room_with_chat_off_is_a_404_not_a_403(_pool: sqlx::PgPool) {
295 + let mut h = TestHarness::new().await;
296 + let (_id, owner) = room(&mut h, ChatPolicy::Off).await;
297 + sign_in(&mut h, owner, "owner").await;
298 +
299 + h.client.get("/p/test").await;
300 + let resp = h.client.post_form("/p/test/chat/send", "body=hi").await;
301 + assert_eq!(resp.status, StatusCode::NOT_FOUND);
302 + }
303 +
304 + // Deleting
305 +
306 + #[sqlx::test]
307 + async fn an_author_deletes_their_own_message_through_the_route(_pool: sqlx::PgPool) {
308 + let mut h = TestHarness::new().await;
309 + let (id, owner) = room(&mut h, ChatPolicy::Members).await;
310 +
311 + let member = user(&h, "member").await;
312 + h.add_membership(member, id, "member").await;
313 + let mine = mt_db::mutations::insert_chat_message(&h.db, id, member, "mine", 168)
314 + .await
315 + .unwrap()
316 + .id;
317 + let theirs = mt_db::mutations::insert_chat_message(&h.db, id, owner, "theirs", 168)
318 + .await
319 + .unwrap()
320 + .id;
321 +
322 + sign_in(&mut h, member, "member").await;
323 + h.client.get("/p/test/chat").await;
324 +
325 + let ok = h
326 + .client
327 + .post_form(&format!("/p/test/chat/messages/{mine}/delete"), "")
328 + .await;
329 + assert_eq!(ok.status, StatusCode::NO_CONTENT);
330 +
331 + let refused = h
332 + .client
333 + .post_form(&format!("/p/test/chat/messages/{theirs}/delete"), "")
334 + .await;
335 + assert_eq!(refused.status, StatusCode::FORBIDDEN);
336 +
337 + let left = mt_db::queries::recent_backlog(&h.db, id, 10).await.unwrap();
338 + assert_eq!(left.len(), 1);
339 + assert_eq!(left[0].body_html, "theirs");
340 + }
341 +
342 + #[sqlx::test]
343 + async fn a_moderator_deletes_anyones_message_through_the_route(_pool: sqlx::PgPool) {
344 + let mut h = TestHarness::new().await;
345 + let (id, owner) = room(&mut h, ChatPolicy::Members).await;
346 +
347 + let member = user(&h, "member").await;
348 + h.add_membership(member, id, "member").await;
349 + let theirs = mt_db::mutations::insert_chat_message(&h.db, id, member, "spam", 168)
350 + .await
351 + .unwrap()
352 + .id;
353 +
354 + sign_in(&mut h, owner, "owner").await;
355 + h.client.get("/p/test/chat").await;
356 + let resp = h
357 + .client
358 + .post_form(&format!("/p/test/chat/messages/{theirs}/delete"), "")
359 + .await;
360 +
361 + assert_eq!(resp.status, StatusCode::NO_CONTENT);
362 + assert!(
363 + mt_db::queries::recent_backlog(&h.db, id, 10)
364 + .await
365 + .unwrap()
366 + .is_empty()
367 + );
368 + }
369 +
370 + // Moderation routes
371 +
372 + #[sqlx::test]
373 + async fn a_moderator_bans_from_chat_and_the_backlog_goes(_pool: sqlx::PgPool) {
374 + let mut h = TestHarness::new().await;
375 + let (id, owner) = room(&mut h, ChatPolicy::Members).await;
376 +
377 + let member = user(&h, "member").await;
378 + h.add_membership(member, id, "member").await;
379 + for _ in 0..3 {
380 + mt_db::mutations::insert_chat_message(&h.db, id, member, "spam", 168)
381 + .await
382 + .unwrap();
383 + }
384 +
385 + sign_in(&mut h, owner, "owner").await;
386 + h.client.get("/p/test/chat").await;
387 + let resp = h
388 + .client
389 + .post_form("/p/test/chat/moderation/ban", "username=member&reason=spam")
390 + .await;
391 +
392 + assert_eq!(resp.status, StatusCode::NO_CONTENT);
393 + assert!(
394 + mt_db::queries::recent_backlog(&h.db, id, 10)
395 + .await
396 + .unwrap()
397 + .is_empty(),
398 + "the ban purges the backlog"
399 + );
400 + assert!(
401 + mt_db::queries::is_user_banned(&h.db, id, member)
402 + .await
403 + .unwrap()
404 + );
405 + }
406 +
407 + #[sqlx::test]
408 + async fn a_member_cannot_reach_the_moderation_routes(_pool: sqlx::PgPool) {
409 + let mut h = TestHarness::new().await;
410 + let (id, _owner) = room(&mut h, ChatPolicy::Members).await;
411 +
412 + let member = user(&h, "member").await;
413 + h.add_membership(member, id, "member").await;
414 + sign_in(&mut h, member, "member").await;
415 + h.client.get("/p/test/chat").await;
416 +
417 + for (path, body) in [
418 + ("/p/test/chat/moderation/ban", "username=owner"),
419 + (
420 + "/p/test/chat/moderation/timeout",
421 + "username=owner&seconds=300",
422 + ),
423 + ] {
424 + assert_eq!(
425 + h.client.post_form(path, body).await.status,
426 + StatusCode::FORBIDDEN,
427 + "{path}"
428 + );
429 + }
430 + }
431 +
432 + #[sqlx::test]
433 + async fn chat_moderation_cannot_be_used_to_ban_an_owner(_pool: sqlx::PgPool) {
434 + // Chat is a second door onto community_bans, so it must not be a way past
435 + // the protections the forum's own ban handler applies.
436 + let mut h = TestHarness::new().await;
437 + let (id, _owner) = room(&mut h, ChatPolicy::Members).await;
438 +
439 + let moderator = user(&h, "mod").await;
440 + h.add_membership(moderator, id, "moderator").await;
441 + sign_in(&mut h, moderator, "mod").await;
442 + h.client.get("/p/test/chat").await;
443 +
444 + let resp = h
445 + .client
446 + .post_form("/p/test/chat/moderation/ban", "username=owner")
447 + .await;
448 + assert_eq!(resp.status, StatusCode::FORBIDDEN);
449 + }
450 +
451 + #[sqlx::test]
452 + async fn a_chat_timeout_mutes_the_target_with_an_expiry(_pool: sqlx::PgPool) {
453 + let mut h = TestHarness::new().await;
454 + let (id, owner) = room(&mut h, ChatPolicy::Members).await;
455 +
456 + let member = user(&h, "member").await;
457 + h.add_membership(member, id, "member").await;
458 +
459 + sign_in(&mut h, owner, "owner").await;
460 + h.client.get("/p/test/chat").await;
461 + let resp = h
462 + .client
463 + .post_form(
464 + "/p/test/chat/moderation/timeout",
465 + "username=member&seconds=600",
466 + )
467 + .await;
468 +
469 + assert_eq!(resp.status, StatusCode::NO_CONTENT);
470 + assert!(
471 + mt_db::queries::is_user_muted(&h.db, id, member)
472 + .await
473 + .unwrap()
474 + );
475 + }
476 +
477 + // The stream
478 +
479 + #[sqlx::test]
480 + async fn the_stream_is_refused_for_a_room_the_viewer_cannot_read(_pool: sqlx::PgPool) {
481 + let mut h = TestHarness::new().await;
482 + room(&mut h, ChatPolicy::Members).await;
483 +
484 + assert_eq!(
485 + h.client.get("/p/test/chat/stream").await.status,
486 + StatusCode::NOT_FOUND
487 + );
488 + }
489 +
490 + // The stream's happy path is deliberately not driven here: `TestClient`
491 + // collects the full response body, and an SSE stream never ends, so the test
492 + // would hang rather than fail. Frame format and backlog-then-live ordering are
493 + // covered by the crate's own tests (`livechat::sse`, `livechat::stream`); the
494 + // end-to-end path belongs with the client island.