Skip to main content

max / makenotwork

18.7 KB · 566 lines History Blame Raw
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 viewer_id: user.as_ref().map(|u| u.user_id.to_string()),
204 max_message_len: livechat::MAX_MESSAGE_LEN,
205 messages: messages.iter().map(chat_message_row).collect(),
206 cursor: messages.last().map_or(0, |m| m.id.0),
207 }
208 .into_response())
209 }
210
211 /// One rendered message, for the server-rendered first paint.
212 fn chat_message_row(m: &livechat::Message) -> crate::templates::ChatMessageRow {
213 crate::templates::ChatMessageRow {
214 id: m.id.0,
215 author_id: m.author_id.0.to_string(),
216 author_name: m
217 .author
218 .as_ref()
219 .map_or_else(|| "Unknown".to_owned(), |a| a.display_name.clone()),
220 avatar_url: m.author.as_ref().and_then(|a| a.avatar_url.clone()),
221 body_html: m.body_html.clone(),
222 created_at: m.created_at,
223 }
224 }
225
226 // The stream
227
228 #[derive(Deserialize)]
229 pub(super) struct StreamQuery {
230 /// Last message id the client holds. Absent on a first connection.
231 after: Option<i64>,
232 }
233
234 #[tracing::instrument(skip_all)]
235 pub(super) async fn chat_stream(
236 State(state): State<AppState>,
237 Path(slug): Path<String>,
238 Query(query): Query<StreamQuery>,
239 MaybeUser(user): MaybeUser,
240 ) -> Result<Response, Response> {
241 let room = open_room(&state, &slug).await?;
242 let authz = gate(&state, &slug, user.as_ref()).await?;
243 let viewer = user.as_ref().map(|u| UserId(u.user_id));
244 require_read(&authz, &room, viewer).await?;
245
246 // A logged-out reader still occupies a connection slot, so it needs an
247 // identity for the per-user cap. The room id stands in: anonymous readers of
248 // one room share a budget rather than being unbounded, which is the correct
249 // answer when the alternative is a per-IP cap that a proxy collapses anyway.
250 let listener = viewer.unwrap_or(UserId(room.id.0));
251
252 let stream = state
253 .chat
254 .subscribe(room.id, listener, query.after.map(MessageId), |after| {
255 let state = state.clone();
256 async move { backlog(&state, &room, after).await }
257 })
258 .await
259 .map_err(|e| match e {
260 // At capacity is a resource refusal, not an authz one: the caller
261 // already passed the read gate. 503 with a Retry-After, so a
262 // reconnecting client backs off instead of hammering.
263 livechat::ChatError::ConnectionLimit(scope) => {
264 tracing::warn!(%scope, "chat connection limit reached");
265 (
266 StatusCode::SERVICE_UNAVAILABLE,
267 [("Retry-After", "5")],
268 "Too many chat connections. Try again shortly.",
269 )
270 .into_response()
271 }
272 other => chat_error(&other),
273 })?;
274
275 Ok(livechat::sse_response(stream).into_response())
276 }
277
278 // Sending
279
280 #[derive(Deserialize)]
281 pub(super) struct SendForm {
282 body: String,
283 /// Client-generated, so the sender's optimistically rendered message
284 /// reconciles instead of appearing twice.
285 nonce: Option<String>,
286 }
287
288 #[derive(Serialize)]
289 pub(super) struct SendResponse {
290 id: i64,
291 #[serde(skip_serializing_if = "Option::is_none")]
292 nonce: Option<String>,
293 }
294
295 #[tracing::instrument(skip_all)]
296 pub(super) async fn chat_send(
297 State(state): State<AppState>,
298 Path(slug): Path<String>,
299 RequireUser(user): RequireUser,
300 Form(form): Form<SendForm>,
301 ) -> Result<Json<SendResponse>, Response> {
302 let room = open_room(&state, &slug).await?;
303 let authz = gate(&state, &slug, Some(&user)).await?;
304
305 let nonce = form.nonce.as_deref().and_then(livechat::Nonce::parse);
306 let author = UserId(user.user_id);
307 let retention_hours = i32::try_from(room.retention.max_age().as_secs() / 3600).unwrap_or(168);
308
309 let sent = state
310 .chat
311 .send(
312 &authz,
313 SendRequest {
314 room: &room,
315 author,
316 body: &form.body,
317 nonce: nonce.clone(),
318 now: std::time::Instant::now(),
319 },
320 |body| {
321 let state = state.clone();
322 async move {
323 // Rendered here, once, through docengine's chat preset.
324 // The crate never sees markdown and never renders: keeping
325 // sanitization in one place is the whole reason the store
326 // closure exists.
327 let html = docengine::render_chat(body.as_str());
328
329 let stored = mt_db::mutations::insert_chat_message(
330 &state.db,
331 room.id.0,
332 author.0,
333 &html,
334 retention_hours,
335 )
336 .await
337 .map_err(livechat::ChatError::host)?;
338
339 let mut message = livechat::Message {
340 id: MessageId(stored.id),
341 room_id: room.id,
342 author_id: author,
343 body_html: html,
344 created_at: stored.created_at,
345 nonce: None,
346 author: None,
347 };
348
349 // Attach the author before the message is published, so
350 // every listener gets a renderable frame rather than an id
351 // they have to resolve themselves.
352 MtChatIdentity::new(state.db.clone(), state.chat_identities.clone())
353 .attach(std::slice::from_mut(&mut message))
354 .await?;
355
356 Ok(message)
357 }
358 },
359 )
360 .await
361 .map_err(|e| send_error(&e))?;
362
363 Ok(Json(SendResponse {
364 id: sent.id.0,
365 nonce: nonce.map(|n| n.0),
366 }))
367 }
368
369 /// Turn a refusal into the status a client can act on.
370 ///
371 /// Refusals are expected traffic and must not be logged as faults; only the
372 /// host-error arm is a real problem.
373 fn send_error(err: &livechat::ChatError) -> Response {
374 let livechat::ChatError::Rejected(rejection) = err else {
375 return chat_error(err);
376 };
377
378 match rejection {
379 SendRejection::Invalid(message) => (StatusCode::UNPROCESSABLE_ENTITY, message.clone()),
380 SendRejection::RoomClosed => (
381 StatusCode::NOT_FOUND,
382 "This room is not available.".to_owned(),
383 ),
384 SendRejection::RoomReadOnly => (
385 StatusCode::FORBIDDEN,
386 "This community is read-only.".to_owned(),
387 ),
388 SendRejection::Denied(reason) => return deny_response(reason),
389 }
390 .into_response()
391 }
392
393 fn deny_response(reason: &DenyReason) -> Response {
394 match reason {
395 DenyReason::RateLimited { retry_after } => (
396 StatusCode::TOO_MANY_REQUESTS,
397 [("Retry-After", retry_after.as_secs().max(1).to_string())],
398 "You are sending messages too quickly.",
399 )
400 .into_response(),
401 DenyReason::Anonymous => (StatusCode::UNAUTHORIZED, "Sign in to chat.").into_response(),
402 DenyReason::NotAMember => {
403 (StatusCode::FORBIDDEN, "Join this community to chat in it.").into_response()
404 }
405 DenyReason::Banned => {
406 (StatusCode::FORBIDDEN, "You are banned from this community.").into_response()
407 }
408 DenyReason::Muted => {
409 (StatusCode::FORBIDDEN, "You are muted in this community.").into_response()
410 }
411 DenyReason::Suspended => {
412 (StatusCode::FORBIDDEN, "Your account has been suspended.").into_response()
413 }
414 DenyReason::TierRequired => (
415 StatusCode::FORBIDDEN,
416 "Chat here is limited to Fan+ supporters.",
417 )
418 .into_response(),
419 }
420 }
421
422 // Moderation
423
424 #[tracing::instrument(skip_all)]
425 pub(super) async fn chat_delete_message(
426 State(state): State<AppState>,
427 Path((slug, message_id)): Path<(String, i64)>,
428 RequireUser(user): RequireUser,
429 ) -> Result<StatusCode, Response> {
430 let room = open_room(&state, &slug).await?;
431 let authz = gate(&state, &slug, Some(&user)).await?;
432
433 let is_moderator = authz
434 .is_moderator(UserId(user.user_id), &room)
435 .await
436 .map_err(|e| chat_error(&e))?;
437
438 // The gate carries the powers; a member's gate cannot express a moderator
439 // removal, and the impl checks authorship for everyone else.
440 state
441 .chat
442 .delete_message(
443 &MtChatModeration::new(state.db.clone(), is_moderator),
444 UserId(user.user_id),
445 &room,
446 MessageId(message_id),
447 )
448 .await
449 .map_err(|e| {
450 tracing::warn!(error = ?e, "chat delete refused");
451 (StatusCode::FORBIDDEN, "Cannot delete that message.").into_response()
452 })?;
453
454 Ok(StatusCode::NO_CONTENT)
455 }
456
457 #[derive(Deserialize)]
458 pub(super) struct ChatModerationForm {
459 username: String,
460 /// Timeout length in seconds. Absent means a ban.
461 seconds: Option<u64>,
462 reason: Option<String>,
463 }
464
465 #[tracing::instrument(skip_all)]
466 pub(super) async fn chat_timeout_user(
467 State(state): State<AppState>,
468 Path(slug): Path<String>,
469 RequireUser(user): RequireUser,
470 Form(form): Form<ChatModerationForm>,
471 ) -> Result<StatusCode, Response> {
472 let (room, target) = moderation_target(&state, &slug, &user, &form).await?;
473
474 let seconds = form.seconds.unwrap_or(300).clamp(60, 86_400);
475
476 state
477 .chat
478 .timeout_user(
479 &MtChatModeration::new(state.db.clone(), true),
480 UserId(user.user_id),
481 &room,
482 UserId(target),
483 std::time::Duration::from_secs(seconds),
484 )
485 .await
486 .map_err(|e| chat_error(&e))?;
487
488 Ok(StatusCode::NO_CONTENT)
489 }
490
491 #[tracing::instrument(skip_all)]
492 pub(super) async fn chat_ban_user(
493 State(state): State<AppState>,
494 Path(slug): Path<String>,
495 RequireUser(user): RequireUser,
496 Form(form): Form<ChatModerationForm>,
497 ) -> Result<StatusCode, Response> {
498 let (room, target) = moderation_target(&state, &slug, &user, &form).await?;
499
500 let reason = form.reason.as_deref().filter(|r| !r.trim().is_empty());
501
502 // Bans the user and removes their backlog in one action, announced to the
503 // room as a single purge event.
504 state
505 .chat
506 .ban_user(
507 &MtChatModeration::new(state.db.clone(), true),
508 UserId(user.user_id),
509 &room,
510 UserId(target),
511 reason,
512 )
513 .await
514 .map_err(|e| chat_error(&e))?;
515
516 Ok(StatusCode::NO_CONTENT)
517 }
518
519 /// Shared front half of the two moderation handlers: resolve the room, prove
520 /// the actor moderates it, resolve the target, and refuse the targets that are
521 /// off limits.
522 async fn moderation_target(
523 state: &AppState,
524 slug: &str,
525 user: &SessionUser,
526 form: &ChatModerationForm,
527 ) -> Result<(Room, uuid::Uuid), Response> {
528 let room = open_room(state, slug).await?;
529 let authz = gate(state, slug, Some(user)).await?;
530
531 if !authz
532 .is_moderator(UserId(user.user_id), &room)
533 .await
534 .map_err(|e| chat_error(&e))?
535 {
536 return Err((StatusCode::FORBIDDEN, "Forbidden").into_response());
537 }
538
539 let target = get_user_by_username(&state.db, form.username.trim()).await?;
540
541 // The same protections the forum's ban handler applies. Chat is a second
542 // door onto `community_bans`, so it must not be a way around them.
543 if state.config.platform_admin_id == Some(target) {
544 return Err((
545 StatusCode::FORBIDDEN,
546 "Cannot act on the platform administrator.",
547 )
548 .into_response());
549 }
550
551 let target_role = super::get_role(&state.db, target, room.id.0).await?;
552 if super::is_mod_or_owner(target_role) {
553 let actor_role = super::get_role(&state.db, user.user_id, room.id.0).await?;
554 if !super::is_owner(actor_role) {
555 return Err(
556 (StatusCode::FORBIDDEN, "Only owners can act on moderators.").into_response(),
557 );
558 }
559 }
560 if super::is_owner(target_role) {
561 return Err((StatusCode::FORBIDDEN, "Cannot act on an owner.").into_response());
562 }
563
564 Ok((room, target))
565 }
566