Skip to main content

max / makenotwork

server: mt architectural refactors (audit 2026-07-02 follow-up) Consistency/type-safety/readability refactors; gate-green (clippy -D warnings, 437 tests, offline .sqlx build): - Move inlined handler SQL into mt-db: 7 compile-checked fns (get_flag_removal_target, is_post_removed, count_threads_in_community, get_community_suspension_reason, get_user_signature, clear/set_user_signature); flagging/admin/actions/account handlers call them. .sqlx cache regenerated. - De-dup sorted-thread SQL: collapse _sorted + _sorted_filtered into one list_threads_in_category_sorted(..., tag_slug: Option<&str>). - Push enums to the DB boundary: the list fn takes SortColumn/SortOrder (exhaustive match); ensure_membership_with_role takes CommunityRole. - Decompose auth.rs::callback into exchange_code_for_token / fetch_userinfo_with_retry / upsert_login_user; loops return on success, so the unwrap()/expect() panic paths are gone (~240 -> ~90 lines). - Split helpers.rs (886 lines) into helpers/{markdown,validation,authz}.rs + a slim mod.rs that re-exports them; all call sites unchanged.
Co-Authored-By
Claude Opus 4.8 <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-07-02 22:02 UTC
Signed with PGP, not checked
Commit: 376071e952519c28bcfb8d0a823790e65f6d2f93
Parent: 37d5fdc
23 files changed, +1455 insertions, -1168 deletions
M multithreaded/src/auth.rs +156 -128
@@ -465,6 +465,147 @@
465 465 Redirect::to(&url)
466 466 }
467 467
468 + /// Retry backoffs for the OAuth token/userinfo round trips (two retries).
469 + const OAUTH_BACKOFFS: [std::time::Duration; 2] = [
470 + std::time::Duration::from_millis(500),
471 + std::time::Duration::from_millis(1000),
472 + ];
473 +
474 + /// Exchange the authorization code for a token, retrying on transport/5xx.
475 + ///
476 + /// Returns the parsed token on success, or the `?error=` slug to redirect with.
477 + /// Parsing happens here so the caller never holds an un-parsed response — there
478 + /// is no post-loop `unwrap()` to trip if the retry logic ever changes.
479 + async fn exchange_code_for_token(
480 + http: &reqwest::Client,
481 + config: &crate::config::Config,
482 + code: &str,
483 + verifier: &str,
484 + ) -> Result<TokenResponse, &'static str> {
485 + let token_url = format!("{}/oauth/token", config.mnw_base_url);
486 + tracing::info!(%token_url, "exchanging code for token");
487 + // `attempt` is the retry counter (also logged), and the loop runs one past
488 + // the backoff array — an iterator-with-enumerate doesn't fit the N+1 shape.
489 + #[allow(clippy::needless_range_loop)]
490 + for attempt in 0..=OAUTH_BACKOFFS.len() {
491 + let res = http
492 + .post(&token_url)
493 + .json(&serde_json::json!({
494 + "grant_type": "authorization_code",
495 + "code": code,
496 + "redirect_uri": config.oauth_redirect_uri,
497 + "code_verifier": verifier,
498 + "client_id": config.oauth_client_id,
499 + }))
500 + .send()
501 + .await;
502 +
503 + match res {
504 + Ok(r) if r.status().is_server_error() => {
505 + let status = r.status();
506 + if attempt < OAUTH_BACKOFFS.len() {
507 + tracing::warn!(%status, attempt, "token exchange got 5xx, retrying");
508 + sleep(OAUTH_BACKOFFS[attempt]).await;
509 + continue;
510 + }
511 + let body = r.text().await.unwrap_or_default();
512 + tracing::error!(%status, %body, "token exchange failed after retries");
513 + return Err("token_exchange_failed");
514 + }
515 + Ok(r) if !r.status().is_success() => {
516 + let status = r.status();
517 + let body = r.text().await.unwrap_or_default();
518 + tracing::error!(%status, %body, "token exchange failed");
519 + return Err("token_exchange_failed");
520 + }
521 + Ok(r) => {
522 + return r.json().await.map_err(|e| {
523 + tracing::error!(error = %e, "token parse failed");
524 + "token_parse_failed"
525 + });
526 + }
527 + Err(e) => {
528 + if attempt < OAUTH_BACKOFFS.len() {
529 + tracing::warn!(error = %e, attempt, "token request failed, retrying");
530 + sleep(OAUTH_BACKOFFS[attempt]).await;
531 + continue;
532 + }
533 + tracing::error!(error = %e, "token request failed after retries");
534 + return Err("token_request_failed");
535 + }
536 + }
537 + }
538 + // Unreachable: the loop returns on every terminal branch. Kept total so a
539 + // future edit to the retry logic can't reintroduce a panic path.
540 + Err("token_request_failed")
541 + }
542 +
543 + /// Fetch userinfo, retrying on transport/5xx. Returns userinfo or the `?error=`
544 + /// slug to redirect with. No post-loop `expect()` — the loop returns on success.
545 + async fn fetch_userinfo_with_retry(
546 + http: &reqwest::Client,
547 + base_url: &str,
548 + access_token: &str,
549 + ) -> Result<UserinfoResponse, &'static str> {
550 + #[allow(clippy::needless_range_loop)]
551 + for attempt in 0..=OAUTH_BACKOFFS.len() {
552 + match fetch_userinfo(http, base_url, access_token).await {
553 + Ok(i) => return Ok(i),
554 + Err(UserinfoError::Transport) if attempt < OAUTH_BACKOFFS.len() => {
555 + tracing::warn!(attempt, "userinfo transport error, retrying");
556 + sleep(OAUTH_BACKOFFS[attempt]).await;
557 + continue;
558 + }
559 + Err(UserinfoError::Transport) => {
560 + tracing::error!("userinfo transport failed after retries");
561 + return Err("userinfo_fetch_failed");
562 + }
563 + Err(UserinfoError::Unauthorized) => {
564 + tracing::error!("userinfo unauthorized — token rejected");
565 + return Err("userinfo_fetch_failed");
566 + }
567 + Err(UserinfoError::BadResponse | UserinfoError::RefreshUnavailable) => {
568 + // RefreshUnavailable is unreachable from fetch_userinfo (it's a
569 + // refresh-grant outcome), but the match must be exhaustive.
570 + tracing::error!("userinfo bad response");
571 + return Err("userinfo_parse_failed");
572 + }
573 + }
574 + }
575 + Err("userinfo_fetch_failed")
576 + }
577 +
578 + /// Upsert the local user row from userinfo on login. `is_fan_plus`/`is_creator`
579 + /// are denormalised here so post rendering can JOIN the author's perks (migration
580 + /// 026). The stale-username vacate and the upsert run on one transaction so the
581 + /// freed name is visible to the insert (S2).
582 + async fn upsert_login_user(
583 + db: &sqlx::PgPool,
584 + info: &UserinfoResponse,
585 + ) -> Result<(), sqlx::Error> {
586 + let mut tx = db.begin().await?;
587 + mt_db::mutations::vacate_username_for_login(&mut tx, info.user_id, &info.username).await?;
588 + sqlx::query(
589 + r#"
590 + INSERT INTO users (mnw_account_id, username, display_name, avatar_url, is_fan_plus, is_creator)
591 + VALUES ($1, $2, $3, $4, $5, $6)
592 + ON CONFLICT (mnw_account_id) DO UPDATE
593 + SET username = $2, display_name = $3, avatar_url = $4,
594 + is_fan_plus = $5, is_creator = $6, updated_at = now()
595 + "#,
596 + )
597 + .bind(info.user_id)
598 + .bind(&info.username)
599 + .bind(&info.display_name)
600 + .bind(&info.avatar_url)
601 + .bind(info.perks.fan_plus)
602 + .bind(info.perks.is_creator)
603 + .execute(&mut *tx)
604 + .await?;
605 + tx.commit().await?;
606 + Ok(())
607 + }
608 +
468 609 /// `GET /auth/callback` — exchange code for token, fetch userinfo, create session.
469 610 #[tracing::instrument(skip_all)]
470 611 pub async fn callback(
@@ -521,141 +662,28 @@
521 662 }
522 663 };
523 664
524 - // Exchange code for token (retry up to 2 attempts on network/5xx errors)
525 - let token_url = format!("{}/oauth/token", state.config.mnw_base_url);
526 - tracing::info!(%token_url, "exchanging code for token");
527 - let backoffs = [
528 - std::time::Duration::from_millis(500),
529 - std::time::Duration::from_millis(1000),
530 - ];
531 - let mut token_res = None;
532 - for attempt in 0..=backoffs.len() {
533 - let res = state
534 - .http
535 - .post(&token_url)
536 - .json(&serde_json::json!({
537 - "grant_type": "authorization_code",
538 - "code": code,
539 - "redirect_uri": state.config.oauth_redirect_uri,
540 - "code_verifier": verifier,
541 - "client_id": state.config.oauth_client_id,
542 - }))
543 - .send()
544 - .await;
545 -
546 - match res {
547 - Ok(r) if r.status().is_server_error() => {
548 - let status = r.status();
549 - if attempt < backoffs.len() {
550 - tracing::warn!(%status, attempt, "token exchange got 5xx, retrying");
551 - sleep(backoffs[attempt]).await;
552 - continue;
553 - }
554 - let body = r.text().await.unwrap_or_default();
555 - tracing::error!(%status, %body, "token exchange failed after retries");
556 - return Redirect::to("/?error=token_exchange_failed");
557 - }
558 - Ok(r) if !r.status().is_success() => {
559 - let status = r.status();
560 - let body = r.text().await.unwrap_or_default();
561 - tracing::error!(%status, %body, "token exchange failed");
562 - return Redirect::to("/?error=token_exchange_failed");
563 - }
564 - Ok(r) => {
565 - token_res = Some(r);
566 - break;
567 - }
568 - Err(e) => {
569 - if attempt < backoffs.len() {
570 - tracing::warn!(error = %e, attempt, "token request failed, retrying");
571 - sleep(backoffs[attempt]).await;
572 - continue;
573 - }
574 - tracing::error!(error = %e, "token request failed after retries");
575 - return Redirect::to("/?error=token_request_failed");
576 - }
577 - }
578 - }
579 - // Safety: loop always either sets token_res or returns early
580 - let token_res = token_res.unwrap();
581 -
582 - let token: TokenResponse = match token_res.json().await {
665 + // Exchange code for token, then fetch userinfo — each retries on transport/5xx
666 + // and returns the `?error=` slug to redirect with on failure.
667 + let token = match exchange_code_for_token(&state.http, &state.config, &code, &verifier).await {
583 668 Ok(t) => t,
584 - Err(e) => {
585 - tracing::error!(error = %e, "token parse failed");
586 - return Redirect::to("/?error=token_parse_failed");
587 - }
669 + Err(slug) => return Redirect::to(&format!("/?error={slug}")),
588 670 };
589 671
590 - // Fetch userinfo (retry up to 2 attempts on transport / 5xx errors).
591 672 tracing::info!(base_url = %state.config.mnw_base_url, "fetching userinfo");
592 - let mut info: Option<UserinfoResponse> = None;
593 - for attempt in 0..=backoffs.len() {
594 - match fetch_userinfo(&state.http, &state.config.mnw_base_url, &token.access_token).await {
595 - Ok(i) => {
596 - info = Some(i);
597 - break;
598 - }
599 - Err(UserinfoError::Transport) if attempt < backoffs.len() => {
600 - tracing::warn!(attempt, "userinfo transport error, retrying");
601 - sleep(backoffs[attempt]).await;
602 - continue;
603 - }
604 - Err(UserinfoError::Transport) => {
605 - tracing::error!("userinfo transport failed after retries");
606 - return Redirect::to("/?error=userinfo_fetch_failed");
607 - }
608 - Err(UserinfoError::Unauthorized) => {
609 - tracing::error!("userinfo unauthorized — token rejected");
610 - return Redirect::to("/?error=userinfo_fetch_failed");
611 - }
612 - Err(UserinfoError::BadResponse | UserinfoError::RefreshUnavailable) => {
613 - // RefreshUnavailable is unreachable from fetch_userinfo (it's a
614 - // refresh-grant outcome), but the match must be exhaustive.
615 - tracing::error!("userinfo bad response");
616 - return Redirect::to("/?error=userinfo_parse_failed");
617 - }
618 - }
619 - }
620 - let info = info.expect("userinfo loop always sets value or returns");
673 + let info = match fetch_userinfo_with_retry(
674 + &state.http,
675 + &state.config.mnw_base_url,
676 + &token.access_token,
677 + )
678 + .await
679 + {
680 + Ok(i) => i,
681 + Err(slug) => return Redirect::to(&format!("/?error={slug}")),
682 + };
621 683
622 684 tracing::info!(user_id = %info.user_id, username = %info.username, "OAuth login successful");
623 685
624 - // Upsert local user. `is_fan_plus`/`is_creator` are denormalised here so
625 - // post rendering can look up the post author's perks via JOIN — see
626 - // migration 026.
627 - //
628 - // First vacate the username from any stale mirror row that still holds it:
629 - // MNW usernames are mutable/reusable, so a login that renames into a name an
630 - // old row carries would otherwise trip the `users_username_key` unique index
631 - // and lock the user out (S2). The vacate and the upsert run on one
632 - // transaction so the freed name is visible to the insert.
633 - let upsert_result = async {
634 - let mut tx = state.db.begin().await?;
635 - mt_db::mutations::vacate_username_for_login(&mut tx, info.user_id, &info.username).await?;
636 - sqlx::query(
637 - r#"
638 - INSERT INTO users (mnw_account_id, username, display_name, avatar_url, is_fan_plus, is_creator)
639 - VALUES ($1, $2, $3, $4, $5, $6)
640 - ON CONFLICT (mnw_account_id) DO UPDATE
641 - SET username = $2, display_name = $3, avatar_url = $4,
642 - is_fan_plus = $5, is_creator = $6, updated_at = now()
643 - "#,
644 - )
645 - .bind(info.user_id)
646 - .bind(&info.username)
647 - .bind(&info.display_name)
648 - .bind(&info.avatar_url)
649 - .bind(info.perks.fan_plus)
650 - .bind(info.perks.is_creator)
651 - .execute(&mut *tx)
652 - .await?;
653 - tx.commit().await?;
654 - Ok::<(), sqlx::Error>(())
655 - }
656 - .await;
657 -
658 - if let Err(e) = upsert_result {
686 + if let Err(e) = upsert_login_user(&state.db, &info).await {
659 687 tracing::error!(error = %e, "user upsert failed");
660 688 return Redirect::to("/?error=user_upsert_failed");
661 689 }
M multithreaded/src/routes/account.rs +20 -35
@@ -33,17 +33,14 @@
33 33 let user = session_user
34 34 .ok_or_else(|| Redirect::to("/auth/login").into_response())?;
35 35
36 - let row: Option<(Option<String>, Option<String>)> = sqlx::query_as(
37 - "SELECT signature_markdown, signature_html FROM users WHERE mnw_account_id = $1",
38 - )
39 - .bind(user.user_id)
40 - .fetch_optional(&state.db)
41 - .await
42 - .map_err(|e| {
43 - tracing::error!(error = ?e, "db error loading signature");
44 - StatusCode::INTERNAL_SERVER_ERROR.into_response()
45 - })?;
46 - let (signature_markdown, signature_html) = row.unwrap_or((None, None));
36 + let (signature_markdown, signature_html) =
37 + mt_db::queries::get_user_signature(&state.db, user.user_id)
38 + .await
39 + .map_err(|e| {
40 + tracing::error!(error = ?e, "db error loading signature");
41 + StatusCode::INTERNAL_SERVER_ERROR.into_response()
42 + })?
43 + .unwrap_or((None, None));
47 44
48 45 Ok(AccountSettingsTemplate {
49 46 csrf_token,
@@ -76,17 +73,12 @@
76 73 // "Clear signature" button submits with `clear=1`; takes precedence over
77 74 // the textarea content.
78 75 if form.clear.as_deref() == Some("1") {
79 - sqlx::query(
80 - "UPDATE users SET signature_markdown = NULL, signature_html = NULL \
81 - WHERE mnw_account_id = $1",
82 - )
83 - .bind(user.user_id)
84 - .execute(&state.db)
85 - .await
86 - .map_err(|e| {
87 - tracing::error!(error = ?e, "db error clearing signature");
88 - StatusCode::INTERNAL_SERVER_ERROR.into_response()
89 - })?;
76 + mt_db::mutations::clear_user_signature(&state.db, user.user_id)
77 + .await
78 + .map_err(|e| {
79 + tracing::error!(error = ?e, "db error clearing signature");
80 + StatusCode::INTERNAL_SERVER_ERROR.into_response()
81 + })?;
90 82 return Ok(Redirect::to("/account?toast=Signature+cleared"));
91 83 }
92 84
@@ -114,19 +106,12 @@
114 106 render_markdown(trimmed)
115 107 };
116 108
117 - sqlx::query(
118 - "UPDATE users SET signature_markdown = $2, signature_html = $3 \
119 - WHERE mnw_account_id = $1",
120 - )
121 - .bind(user.user_id)
122 - .bind(trimmed)
123 - .bind(&signature_html)
124 - .execute(&state.db)
125 - .await
126 - .map_err(|e| {
127 - tracing::error!(error = ?e, "db error saving signature");
128 - StatusCode::INTERNAL_SERVER_ERROR.into_response()
129 - })?;
109 + mt_db::mutations::set_user_signature(&state.db, user.user_id, trimmed, &signature_html)
110 + .await
111 + .map_err(|e| {
112 + tracing::error!(error = ?e, "db error saving signature");
113 + StatusCode::INTERNAL_SERVER_ERROR.into_response()
114 + })?;
130 115
131 116 Ok(Redirect::to("/account?toast=Signature+saved"))
132 117 }
M multithreaded/src/routes/admin.rs +7 -15
@@ -201,18 +201,12 @@
201 201 let csrf_token = Some(csrf::get_or_create_token(&session).await);
202 202 let community = get_community(&state.db, &slug).await?;
203 203
204 - let thread_count: i64 = sqlx::query_scalar(
205 - "SELECT COUNT(*) FROM threads t
206 - JOIN categories c ON c.id = t.category_id
207 - WHERE c.community_id = $1",
208 - )
209 - .bind(community.id)
210 - .fetch_one(&state.db)
211 - .await
212 - .map_err(|e| {
213 - tracing::error!(error = ?e, "db error counting threads");
214 - StatusCode::INTERNAL_SERVER_ERROR.into_response()
215 - })?;
204 + let thread_count = mt_db::queries::count_threads_in_community(&state.db, community.id)
205 + .await
206 + .map_err(|e| {
207 + tracing::error!(error = ?e, "db error counting threads");
208 + StatusCode::INTERNAL_SERVER_ERROR.into_response()
209 + })?;
216 210
217 211 let member_count = mt_db::queries::count_community_members(&state.db, community.id)
218 212 .await
@@ -222,9 +216,7 @@
222 216 })?;
223 217
224 218 let suspension_reason: Option<String> = if community.suspended_at.is_some() {
225 - sqlx::query_scalar("SELECT suspension_reason FROM communities WHERE id = $1")
226 - .bind(community.id)
227 - .fetch_one(&state.db)
219 + mt_db::queries::get_community_suspension_reason(&state.db, community.id)
228 220 .await
229 221 .ok()
230 222 .flatten()
M multithreaded/src/routes/flagging.rs +10 -20
@@ -155,26 +155,16 @@
155 155 let (community, _role) = require_mod_or_owner(&state, &slug, &user).await?;
156 156 let flag_id = parse_uuid(&flag_id_str)?;
157 157
158 - // Get the flag to find the post_id and its thread — scoped to this community
159 - let flag_row: Option<(uuid::Uuid, uuid::Uuid, uuid::Uuid)> = sqlx::query_as(
160 - "SELECT pf.post_id, p.author_id, t.id
161 - FROM post_flags pf
162 - JOIN posts p ON p.id = pf.post_id
163 - JOIN threads t ON t.id = p.thread_id
164 - JOIN categories c ON c.id = t.category_id
165 - WHERE pf.id = $1 AND c.community_id = $2",
166 - )
167 - .bind(flag_id)
168 - .bind(community.id)
169 - .fetch_optional(&state.db)
170 - .await
171 - .map_err(|e| {
172 - tracing::error!(error = ?e, "db error fetching flag");
173 - (StatusCode::INTERNAL_SERVER_ERROR, "Internal server error").into_response()
174 - })?;
175 -
176 - let (post_id, author_id, thread_id) = flag_row
177 - .ok_or_else(|| (StatusCode::NOT_FOUND, "Not found").into_response())?;
158 + // Find the flag's post + thread, scoped to this community (scoping enforced
159 + // in the query layer).
160 + let (post_id, author_id, thread_id) =
161 + mt_db::queries::get_flag_removal_target(&state.db, flag_id, community.id)
162 + .await
163 + .map_err(|e| {
164 + tracing::error!(error = ?e, "db error fetching flag");
165 + (StatusCode::INTERNAL_SERVER_ERROR, "Internal server error").into_response()
166 + })?
167 + .ok_or_else(|| (StatusCode::NOT_FOUND, "Not found").into_response())?;
178 168
179 169 // Mod-remove the post, resolve its flags, and write the audit row(s) on one
180 170 // transaction: the removal, the flag resolution, and the log all commit
M multithreaded/src/routes/internal.rs +1 -1
@@ -140,7 +140,7 @@
140 140 &state.db,
141 141 req.owner_mnw_id,
142 142 community_id,
143 - "owner",
143 + mt_core::types::CommunityRole::Owner,
144 144 )
145 145 .await
146 146 .map_err(db_error)?;
M multithreaded/tests/workflows/mutations.rs +2 -2
@@ -306,12 +306,12 @@
306 306 .await
307 307 .unwrap();
308 308
309 - mt_db::mutations::ensure_membership_with_role(&h.db, user, comm_id, "moderator")
309 + mt_db::mutations::ensure_membership_with_role(&h.db, user, comm_id, mt_core::types::CommunityRole::Moderator)
310 310 .await
311 311 .unwrap();
312 312
313 313 // Second call with same role should succeed
314 - mt_db::mutations::ensure_membership_with_role(&h.db, user, comm_id, "moderator")
314 + mt_db::mutations::ensure_membership_with_role(&h.db, user, comm_id, mt_core::types::CommunityRole::Moderator)
315 315 .await
316 316 .unwrap();
317 317
M multithreaded/tests/workflows/tags.rs +8 -1
@@ -36,7 +36,14 @@
36 36
37 37 // Check thread_tags
38 38 let threads = mt_db::queries::list_threads_in_category_sorted(
39 - &h.db, "test", "general", "activity", "desc", 10, 0,
39 + &h.db,
40 + "test",
41 + "general",
42 + mt_core::types::SortColumn::Activity,
43 + mt_core::types::SortOrder::Desc,
44 + 10,
45 + 0,
46 + None,
40 47 )
41 48 .await
42 49 .unwrap();
M multithreaded/crates/mt-db/src/mutations.rs +36 -3
@@ -1,7 +1,7 @@
1 1 //! Database write mutations — inserts, updates, deletes.
2 2
3 3 use chrono::{DateTime, Utc};
4 - use mt_core::types::{BanType, CommunityState, ModAction, ModActor};
4 + use mt_core::types::{BanType, CommunityRole, CommunityState, ModAction, ModActor};
5 5 use sqlx::PgPool;
6 6 use uuid::Uuid;
7 7
@@ -95,7 +95,7 @@
95 95 pool: &PgPool,
96 96 user_id: Uuid,
97 97 community_id: Uuid,
98 - role: &str,
98 + role: CommunityRole,
99 99 ) -> Result<(), sqlx::Error> {
100 100 sqlx::query!(
101 101 "INSERT INTO memberships (user_id, community_id, role)
@@ -103,7 +103,7 @@
103 103 ON CONFLICT (user_id, community_id) DO NOTHING",
104 104 user_id,
105 105 community_id,
106 - role,
106 + role.as_str(),
107 107 )
108 108 .execute(pool)
109 109 .await?;
@@ -1339,3 +1339,36 @@
1339 1339 .await?;
1340 1340 Ok(())
1341 1341 }
1342 +
1343 + /// Clear a user's saved signature (markdown + rendered html).
1344 + #[tracing::instrument(skip_all)]
1345 + pub async fn clear_user_signature(pool: &PgPool, user_id: Uuid) -> Result<(), sqlx::Error> {
1346 + sqlx::query!(
1347 + "UPDATE users SET signature_markdown = NULL, signature_html = NULL \
1348 + WHERE mnw_account_id = $1",
1349 + user_id,
1350 + )
1351 + .execute(pool)
1352 + .await?;
1353 + Ok(())
1354 + }
1355 +
1356 + /// Save a user's signature markdown and its pre-rendered html.
1357 + #[tracing::instrument(skip_all)]
1358 + pub async fn set_user_signature(
1359 + pool: &PgPool,
1360 + user_id: Uuid,
1361 + markdown: &str,
1362 + html: &str,
1363 + ) -> Result<(), sqlx::Error> {
1364 + sqlx::query!(
1365 + "UPDATE users SET signature_markdown = $2, signature_html = $3 \
1366 + WHERE mnw_account_id = $1",
1367 + user_id,
1368 + markdown,
1369 + html,
1370 + )
1371 + .execute(pool)
1372 + .await?;
1373 + Ok(())
1374 + }
M multithreaded/crates/mt-db/src/queries.rs +127 -69
@@ -1,7 +1,7 @@
1 1 //! Database read queries — projection structs and SQL.
2 2
3 3 use chrono::{DateTime, Utc};
4 - use mt_core::types::{BanType, CommunityRole, CommunityState, ModAction};
4 + use mt_core::types::{BanType, CommunityRole, CommunityState, ModAction, SortColumn, SortOrder};
5 5 use sqlx::PgPool;
6 6 use uuid::Uuid;
7 7
@@ -372,23 +372,49 @@
372 372 .await
373 373 }
374 374
375 - /// List threads with sorting. `sort` must be "replies" or "activity".
376 - /// `order` must be "asc" or "desc". Pinned threads always sort first.
375 + /// List threads in a category, sorted, optionally filtered to one tag.
376 + ///
377 + /// Pinned threads always sort first. `sort`/`order` are typed enums, so the
378 + /// `ORDER BY` clause is chosen from an exhaustive match of fixed literals (no
379 + /// user string reaches the SQL); when `tag_slug` is `Some`, an extra join
380 + /// restricts the results to threads carrying that tag in the same community.
377 381 #[tracing::instrument(skip_all)]
382 + #[allow(clippy::too_many_arguments)]
378 383 pub async fn list_threads_in_category_sorted(
379 384 pool: &PgPool,
380 385 community_slug: &str,
381 386 category_slug: &str,
382 - sort: &str,
383 - order: &str,
387 + sort: SortColumn,
388 + order: SortOrder,
384 389 limit: i64,
385 390 offset: i64,
391 + tag_slug: Option<&str>,
386 392 ) -> Result<Vec<ThreadWithMeta>, sqlx::Error> {
387 393 let order_clause = match (sort, order) {
388 - ("replies", "asc") => "ORDER BY t.pinned DESC, reply_count ASC, t.last_activity_at DESC",
389 - ("replies", _) => "ORDER BY t.pinned DESC, reply_count DESC, t.last_activity_at DESC",
390 - (_, "asc") => "ORDER BY t.pinned DESC, t.last_activity_at ASC",
391 - _ => "ORDER BY t.pinned DESC, t.last_activity_at DESC",
394 + (SortColumn::Replies, SortOrder::Asc) => {
395 + "ORDER BY t.pinned DESC, reply_count ASC, t.last_activity_at DESC"
396 + }
397 + (SortColumn::Replies, SortOrder::Desc) => {
398 + "ORDER BY t.pinned DESC, reply_count DESC, t.last_activity_at DESC"
399 + }
400 + (SortColumn::Activity, SortOrder::Asc) => {
401 + "ORDER BY t.pinned DESC, t.last_activity_at ASC"
402 + }
403 + (SortColumn::Activity, SortOrder::Desc) => {
404 + "ORDER BY t.pinned DESC, t.last_activity_at DESC"
405 + }
406 + };
407 +
408 + // When a tag filter is present it binds as $3, pushing limit/offset to $4/$5.
409 + let (tag_join, limit_ph, offset_ph) = if tag_slug.is_some() {
410 + (
411 + "JOIN thread_tags tt ON tt.thread_id = t.id \
412 + JOIN tags tg ON tg.id = tt.tag_id AND tg.slug = $3 AND tg.community_id = co.id",
413 + "$4",
414 + "$5",
415 + )
416 + } else {
417 + ("", "$3", "$4")
392 418 };
393 419
394 420 let query = format!(
@@ -402,19 +428,20 @@
402 428 JOIN categories c ON c.id = t.category_id
403 429 JOIN communities co ON co.id = c.community_id
404 430 JOIN users u ON u.mnw_account_id = t.author_id
431 + {tag_join}
405 432 WHERE co.slug = $1 AND c.slug = $2 AND t.deleted_at IS NULL
406 433 {order_clause}
407 - LIMIT $3 OFFSET $4"
434 + LIMIT {limit_ph} OFFSET {offset_ph}"
408 435 );
409 436
410 437 // runtime-checked: dynamic SQL (cannot use compile-time macro)
411 - sqlx::query_as::<_, ThreadWithMeta>(&query)
438 + let mut q = sqlx::query_as::<_, ThreadWithMeta>(&query)
412 439 .bind(community_slug)
413 - .bind(category_slug)
414 - .bind(limit)
415 - .bind(offset)
416 - .fetch_all(pool)
417 - .await
440 + .bind(category_slug);
441 + if let Some(tag) = tag_slug {
442 + q = q.bind(tag);
443 + }
444 + q.bind(limit).bind(offset).fetch_all(pool).await
418 445 }
419 446
420 447 #[tracing::instrument(skip_all)]
@@ -1393,59 +1420,6 @@
1393 1420 }
1394 1421 }
1395 1422
1396 - /// List threads with sorting, optionally filtered by tag slug.
1397 - #[tracing::instrument(skip_all)]
1398 - #[allow(clippy::too_many_arguments)]
1399 - pub async fn list_threads_in_category_sorted_filtered(
1400 - pool: &PgPool,
1401 - community_slug: &str,
1402 - category_slug: &str,
1403 - sort: &str,
1404 - order: &str,
1405 - limit: i64,
1406 - offset: i64,
1407 - tag_slug: Option<&str>,
1408 - ) -> Result<Vec<ThreadWithMeta>, sqlx::Error> {
1409 - if tag_slug.is_none() {
1410 - return list_threads_in_category_sorted(pool, community_slug, category_slug, sort, order, limit, offset).await;
1411 - }
1412 -
1413 - let tag = tag_slug.unwrap();
1414 - let order_clause = match (sort, order) {
1415 - ("replies", "asc") => "ORDER BY t.pinned DESC, reply_count ASC, t.last_activity_at DESC",
1416 - ("replies", _) => "ORDER BY t.pinned DESC, reply_count DESC, t.last_activity_at DESC",
1417 - (_, "asc") => "ORDER BY t.pinned DESC, t.last_activity_at ASC",
1418 - _ => "ORDER BY t.pinned DESC, t.last_activity_at DESC",
1419 - };
1420 -
1421 - let query = format!(
1422 - "SELECT t.id, t.title,
1423 - COALESCE(u.display_name, u.username) AS author_name,
1424 - u.username AS author_username,
1425 - GREATEST(t.post_count - 1, 0)::BIGINT AS reply_count,
1426 - t.last_activity_at,
1427 - t.pinned, t.locked
1428 - FROM threads t
1429 - JOIN categories c ON c.id = t.category_id
1430 - JOIN communities co ON co.id = c.community_id
1431 - JOIN users u ON u.mnw_account_id = t.author_id
1432 - JOIN thread_tags tt ON tt.thread_id = t.id
1433 - JOIN tags tg ON tg.id = tt.tag_id AND tg.slug = $3 AND tg.community_id = co.id
1434 - WHERE co.slug = $1 AND c.slug = $2 AND t.deleted_at IS NULL
1435 - {order_clause}
1436 - LIMIT $4 OFFSET $5"
1437 - );
1438 -
1439 - // runtime-checked: dynamic SQL (cannot use compile-time macro)
1440 - sqlx::query_as::<_, ThreadWithMeta>(&query)
1441 - .bind(community_slug)
1442 - .bind(category_slug)
1443 - .bind(tag)
1444 - .bind(limit)
1445 - .bind(offset)
1446 - .fetch_all(pool)
1447 - .await
1448 - }
1449 1423
1450 1424 // ============================================================================
1451 1425 // Flag queries
@@ -1779,3 +1753,87 @@
1779 1753 .fetch_one(pool)
1780 1754 .await
1781 1755 }
1756 +
1757 + /// Post-removal target for a flag, scoped to a community: `(post_id, author_id,
1758 + /// thread_id)`. Returns `None` if the flag doesn't exist or belongs to another
1759 + /// community — the community scoping is enforced here rather than in the handler.
1760 + #[tracing::instrument(skip_all)]
1761 + pub async fn get_flag_removal_target(
1762 + pool: &PgPool,
1763 + flag_id: Uuid,
1764 + community_id: Uuid,
1765 + ) -> Result<Option<(Uuid, Uuid, Uuid)>, sqlx::Error> {
1766 + let row = sqlx::query!(
1767 + r#"SELECT pf.post_id AS "post_id!", p.author_id AS "author_id!", t.id AS "thread_id!"
1768 + FROM post_flags pf
1769 + JOIN posts p ON p.id = pf.post_id
1770 + JOIN threads t ON t.id = p.thread_id
1771 + JOIN categories c ON c.id = t.category_id
1772 + WHERE pf.id = $1 AND c.community_id = $2"#,
1773 + flag_id,
1774 + community_id,
1775 + )
1776 + .fetch_optional(pool)
1777 + .await?;
1778 + Ok(row.map(|r| (r.post_id, r.author_id, r.thread_id)))
1779 + }
1780 +
1781 + /// Whether a post is mod-removed (`removed_at IS NOT NULL`).
1782 + #[tracing::instrument(skip_all)]
1783 + pub async fn is_post_removed(pool: &PgPool, post_id: Uuid) -> Result<bool, sqlx::Error> {
1784 + sqlx::query_scalar!(
1785 + r#"SELECT (removed_at IS NOT NULL) AS "removed!" FROM posts WHERE id = $1"#,
1786 + post_id,
1787 + )
1788 + .fetch_one(pool)
1789 + .await
1790 + }
1791 +
1792 + /// Count all threads in a community (across every category, including
1793 + /// soft-deleted — matches the admin dashboard's historical total).
1794 + #[tracing::instrument(skip_all)]
1795 + pub async fn count_threads_in_community(
1796 + pool: &PgPool,
1797 + community_id: Uuid,
1798 + ) -> Result<i64, sqlx::Error> {
1799 + sqlx::query_scalar!(
1800 + r#"SELECT COUNT(*) AS "count!" FROM threads t
1801 + JOIN categories c ON c.id = t.category_id
1802 + WHERE c.community_id = $1"#,
1803 + community_id,
1804 + )
1805 + .fetch_one(pool)
1806 + .await
1807 + }
1808 +
1809 + /// The stored suspension reason for a community, if any.
1810 + #[tracing::instrument(skip_all)]
1811 + pub async fn get_community_suspension_reason(
1812 + pool: &PgPool,
1813 + community_id: Uuid,
1814 + ) -> Result<Option<String>, sqlx::Error> {
1815 + sqlx::query_scalar!(
1816 + "SELECT suspension_reason FROM communities WHERE id = $1",
1817 + community_id,
1818 + )
1819 + .fetch_one(pool)
1820 + .await
1821 + }
1822 +
1823 + /// A user's saved signature: `(markdown, html)`. Either field may be `NULL`.
1824 + pub type UserSignature = (Option<String>, Option<String>);
1825 +
1826 + /// Fetch a user's saved signature, or `None` if the user row is absent.
1827 + #[tracing::instrument(skip_all)]
1828 + pub async fn get_user_signature(
1829 + pool: &PgPool,
1830 + user_id: Uuid,
1831 + ) -> Result<Option<UserSignature>, sqlx::Error> {
1832 + let row = sqlx::query!(
1833 + "SELECT signature_markdown, signature_html FROM users WHERE mnw_account_id = $1",
1834 + user_id,
1835 + )
1836 + .fetch_optional(pool)
1837 + .await?;
1838 + Ok(row.map(|r| (r.signature_markdown, r.signature_html)))
1839 + }
M multithreaded/src/routes/forum/actions.rs +2 -6
@@ -90,9 +90,7 @@
90 90 let scope = CommunityScope::<PostForEdit>::resolve(&state.db, &slug, &post_id_str).await?;
91 91 let post_id = scope.resource.id;
92 92
93 - let removed: bool = sqlx::query_scalar("SELECT removed_at IS NOT NULL FROM posts WHERE id = $1")
94 - .bind(post_id)
95 - .fetch_one(&state.db)
93 + let removed = mt_db::queries::is_post_removed(&state.db, post_id)
96 94 .await
97 95 .map_err(|e| {
98 96 tracing::error!(error = ?e, "db error checking removal status");
@@ -170,9 +168,7 @@
170 168 let scope = CommunityScope::<PostForEdit>::resolve(&state.db, &slug, &post_id_str).await?;
171 169 let post_id = scope.resource.id;
172 170
173 - let removed: bool = sqlx::query_scalar("SELECT removed_at IS NOT NULL FROM posts WHERE id = $1")
174 - .bind(post_id)
175 - .fetch_one(&state.db)
171 + let removed = mt_db::queries::is_post_removed(&state.db, post_id)
176 172 .await
177 173 .map_err(|e| {
178 174 tracing::error!(error = ?e, "db error checking removal status");
M multithreaded/src/routes/forum/views.rs +2 -2
A multithreaded/.sqlx/query-0d9bb9d5a6919ac3280ad1c0586b6114559dc610c93629d3f9f99c2cf51de866.json +22
A multithreaded/.sqlx/query-59953c28712213fdcd1d8f2a53219fe106e75f288669565848345200a7ae2286.json +22
A multithreaded/.sqlx/query-5ed94c9e85f15c46c8dd1e701dde63071ba44711af8d7c340850e56df6388cac.json +35
A multithreaded/.sqlx/query-790719db1a55f098443b09b1b9a31d161ee4aed1277323ee43f147ef327d8cd3.json +22
A multithreaded/.sqlx/query-7d9c0e757bb77446e6cace3d83b718edc18a3a020e3913b2ae0ced477ed7b24e.json +28
A multithreaded/.sqlx/query-96b258d531a69452f8ffb93123c05e112e0c2eae2b5c352d59c47979848305c3.json +16
A multithreaded/.sqlx/query-abfd2381ad3c10a2581671173e22ae35a5936d073388c6c24dfa35451a7fb74d.json +14
D multithreaded/src/routes/helpers.rs -886
A multithreaded/src/routes/helpers/authz.rs +400
A multithreaded/src/routes/helpers/markdown.rs +135
A multithreaded/src/routes/helpers/mod.rs +148
A multithreaded/src/routes/helpers/validation.rs +242