Skip to main content

max / makenotwork

Split mt-db mutations.rs into per-entity modules Break the 1393-line mutations.rs into mutations/<entity>.rs (community, category, thread, post, footnote, member, user, moderation, tracked_thread, tag, link_preview, mention, endorsement, image) behind a mutations/mod.rs facade mirroring the queries split. Private helpers vacate_stale_username and html_escape stay private, co-located with their only callers (user, community). All mt_db::mutations::* paths unchanged; fn multiset identical, build and clippy clean. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-07-13 16:10 UTC
Commit: 400bb5afe16e4e94411b9f1eb65114dfee66d902
Parent: e5ace39
16 files changed, +1401 insertions, -500 deletions
D multithreaded/crates/mt-db/src/mutations.rs -500
@@ -1,1393 +0,0 @@
1 - //! Database write mutations — inserts, updates, deletes.
2 -
3 - use chrono::{DateTime, Utc};
4 - use mt_core::types::{BanType, CommunityRole, CommunityState, ModAction, ModActor};
5 - use sqlx::PgPool;
6 - use uuid::Uuid;
7 -
8 - /// Create a new community and return its ID.
9 - #[tracing::instrument(skip_all)]
10 - pub async fn create_community(
11 - pool: &PgPool,
12 - name: &str,
13 - slug: &str,
14 - description: Option<&str>,
15 - ) -> Result<Uuid, sqlx::Error> {
16 - sqlx::query_scalar!(
17 - "INSERT INTO communities (name, slug, description)
18 - VALUES ($1, $2, $3)
19 - RETURNING id",
20 - name,
21 - slug,
22 - description,
23 - )
24 - .fetch_one(pool)
25 - .await
26 - }
27 -
28 - /// Upsert a user from MNW account data. Creates the user if they don't exist,
29 - /// updates username/display_name if they do.
30 - #[tracing::instrument(skip_all)]
31 - pub async fn upsert_user(
32 - pool: &PgPool,
33 - mnw_account_id: Uuid,
34 - username: &str,
35 - display_name: Option<&str>,
36 - ) -> Result<(), sqlx::Error> {
37 - let mut tx = pool.begin().await?;
38 - vacate_stale_username(&mut tx, mnw_account_id, username).await?;
39 - sqlx::query!(
40 - "INSERT INTO users (mnw_account_id, username, display_name)
41 - VALUES ($1, $2, $3)
42 - ON CONFLICT (mnw_account_id) DO UPDATE
43 - SET username = $2, display_name = $3, updated_at = now()",
44 - mnw_account_id,
45 - username,
46 - display_name,
47 - )
48 - .execute(&mut *tx)
49 - .await?;
50 - tx.commit().await?;
51 - Ok(())
52 - }
53 -
54 - /// Free `username` from any *other* user row that still holds it, so the caller's
55 - /// upsert can claim it without tripping the `users_username_key` unique index.
56 - ///
57 - /// MNW is the source of truth for usernames and they are reusable there; a local
58 - /// mirror row holding a name its MNW account no longer owns is stale and must
59 - /// yield. Run this on the same transaction as the upsert (a single statement
60 - /// can't, because a non-deferrable unique index is checked per-row and the
61 - /// rename wouldn't yet be visible to the insert). The vacated row keeps its data
62 - /// under a collision-proof `mnw_<account-id>` placeholder until that user's own
63 - /// next login resets it. Closes the username-collision login lockout (S2).
64 - async fn vacate_stale_username(
65 - conn: &mut sqlx::PgConnection,
66 - mnw_account_id: Uuid,
67 - username: &str,
68 - ) -> Result<(), sqlx::Error> {
69 - sqlx::query!(
70 - "UPDATE users
71 - SET username = 'mnw_' || mnw_account_id::text, updated_at = now()
72 - WHERE username = $1 AND mnw_account_id <> $2",
73 - username,
74 - mnw_account_id,
75 - )
76 - .execute(&mut *conn)
77 - .await?;
78 - Ok(())
79 - }
80 -
81 - /// Public wrapper over [`vacate_stale_username`] for the OAuth login upsert in the
82 - /// web crate, which builds its INSERT with a runtime query.
83 - pub async fn vacate_username_for_login(
84 - conn: &mut sqlx::PgConnection,
85 - mnw_account_id: Uuid,
86 - username: &str,
87 - ) -> Result<(), sqlx::Error> {
88 - vacate_stale_username(conn, mnw_account_id, username).await
89 - }
90 -
91 - /// Ensure a user has a membership in a community with the given role.
92 - /// Creates membership if none exists, does nothing if already a member.
93 - #[tracing::instrument(skip_all)]
94 - pub async fn ensure_membership_with_role(
95 - pool: &PgPool,
96 - user_id: Uuid,
97 - community_id: Uuid,
98 - role: CommunityRole,
99 - ) -> Result<(), sqlx::Error> {
100 - sqlx::query!(
101 - "INSERT INTO memberships (user_id, community_id, role)
102 - VALUES ($1, $2, $3)
103 - ON CONFLICT (user_id, community_id) DO NOTHING",
104 - user_id,
105 - community_id,
106 - role.as_str(),
107 - )
108 - .execute(pool)
109 - .await?;
110 - Ok(())
111 - }
112 -
113 - /// Atomically create an externally-referenced thread and its opening post in
114 - /// one transaction. Returns `(thread_id, op_post_id)`.
115 - ///
116 - /// The atomicity matters most here: the MNW→MT internal path retries, and a
117 - /// committed-but-post-less thread would make the retry collide with the UNIQUE
118 - /// `external_ref` index and 500 while leaking an empty thread.
119 - #[tracing::instrument(skip_all)]
120 - pub async fn create_thread_with_op_external_ref(
121 - pool: &PgPool,
122 - category_id: Uuid,
123 - author_id: Uuid,
124 - title: &str,
125 - external_ref: &str,
126 - body_markdown: &str,
127 - body_html: &str,
128 - ) -> Result<(Uuid, Uuid), sqlx::Error> {
129 - let mut tx = pool.begin().await?;
130 -
131 - let thread_id = sqlx::query_scalar!(
132 - "INSERT INTO threads (category_id, author_id, title, external_ref)
133 - VALUES ($1, $2, $3, $4)
134 - RETURNING id",
135 - category_id,
136 - author_id,
137 - title,
138 - external_ref,
139 - )
140 - .fetch_one(&mut *tx)
141 - .await?;
142 -
143 - let post_id = sqlx::query_scalar!(
144 - "INSERT INTO posts (thread_id, author_id, body_markdown, body_html)
145 - VALUES ($1, $2, $3, $4)
146 - RETURNING id",
147 - thread_id,
148 - author_id,
149 - body_markdown,
150 - body_html,
151 - )
152 - .fetch_one(&mut *tx)
153 - .await?;
154 -
155 - tx.commit().await?;
156 - Ok((thread_id, post_id))
157 - }
158 -
159 - /// Ensure a user has a membership in a community. Creates a 'member' role if none exists.
160 - #[tracing::instrument(skip_all)]
161 - pub async fn ensure_membership(
162 - pool: &PgPool,
163 - user_id: Uuid,
164 - community_id: Uuid,
165 - ) -> Result<(), sqlx::Error> {
166 - sqlx::query!(
167 - "INSERT INTO memberships (user_id, community_id, role)
168 - VALUES ($1, $2, 'member')
169 - ON CONFLICT (user_id, community_id) DO NOTHING",
170 - user_id,
171 - community_id,
172 - )
173 - .execute(pool)
174 - .await?;
175 - Ok(())
176 - }
177 -
178 - /// Low-level: insert a bare thread row (no opening post). Returns the thread ID.
179 - ///
180 - /// Handlers must NOT pair this with a separate `create_post` for the OP — that
181 - /// non-atomic sequence can leak a post-less thread. Use [`create_thread_with_op`]
182 - /// for the real "start a thread" operation. This primitive exists for tests and
183 - /// data construction that build posts explicitly.
184 - #[tracing::instrument(skip_all)]
185 - pub async fn create_thread(
186 - pool: &PgPool,
187 - category_id: Uuid,
188 - author_id: Uuid,
189 - title: &str,
190 - ) -> Result<Uuid, sqlx::Error> {
191 - sqlx::query_scalar!(
192 - "INSERT INTO threads (category_id, author_id, title)
193 - VALUES ($1, $2, $3)
194 - RETURNING id",
195 - category_id,
196 - author_id,
197 - title,
198 - )
199 - .fetch_one(pool)
200 - .await
201 - }
202 -
203 - /// Atomically create a thread and its opening post in one transaction.
204 - ///
205 - /// Returns `(thread_id, op_post_id)`. Creating the thread and OP separately
206 - /// could leave a titled, post-less thread if the second insert failed; this
207 - /// makes that state unreachable. `threads.last_activity_at` defaults to now().
208 - #[tracing::instrument(skip_all)]
209 - pub async fn create_thread_with_op(
210 - pool: &PgPool,
211 - category_id: Uuid,
212 - author_id: Uuid,
213 - title: &str,
214 - body_markdown: &str,
215 - body_html: &str,
216 - ) -> Result<(Uuid, Uuid), sqlx::Error> {
217 - let mut tx = pool.begin().await?;
218 - let ids =
219 - create_thread_with_op_tx(&mut tx, category_id, author_id, title, body_markdown, body_html)
220 - .await?;
221 - tx.commit().await?;
222 - Ok(ids)
223 - }
224 -
225 - /// Create a thread and its opening post on the caller's transaction.
226 - ///
227 - /// Lets a handler fold the thread, its OP, its mentions, and its tags into one
228 - /// atomic unit so a mid-sequence failure can't leave a thread whose @mentions or
229 - /// tags silently vanished (ultra-fuzz M-St1). The pool-taking
230 - /// [`create_thread_with_op`] wraps this in its own transaction.
231 - #[tracing::instrument(skip_all)]
232 - pub async fn create_thread_with_op_tx(
233 - conn: &mut sqlx::PgConnection,
234 - category_id: Uuid,
235 - author_id: Uuid,
236 - title: &str,
237 - body_markdown: &str,
238 - body_html: &str,
239 - ) -> Result<(Uuid, Uuid), sqlx::Error> {
240 - let thread_id = sqlx::query_scalar!(
241 - "INSERT INTO threads (category_id, author_id, title)
242 - VALUES ($1, $2, $3)
243 - RETURNING id",
244 - category_id,
245 - author_id,
246 - title,
247 - )
248 - .fetch_one(&mut *conn)
249 - .await?;
250 -
251 - let post_id = sqlx::query_scalar!(
252 - "INSERT INTO posts (thread_id, author_id, body_markdown, body_html)
253 - VALUES ($1, $2, $3, $4)
254 - RETURNING id",
255 - thread_id,
256 - author_id,
257 - body_markdown,
258 - body_html,
259 - )
260 - .fetch_one(&mut *conn)
261 - .await?;
262 -
263 - Ok((thread_id, post_id))
264 - }
265 -
266 - /// Insert a reply and bump the thread's last_activity_at atomically.
267 - ///
268 - /// Opening posts are created with the thread via [`create_thread_with_op`];
269 - /// this is for replies only. The display reply count derives from the
270 - /// denormalized `threads.post_count`, which the m029 trigger maintains on every
271 - /// post INSERT/DELETE/removed_at change — no counter to update by hand here.
272 - #[tracing::instrument(skip_all)]
273 - pub async fn create_post(
274 - pool: &PgPool,
275 - thread_id: Uuid,
276 - author_id: Uuid,
277 - body_markdown: &str,
278 - body_html: &str,
279 - ) -> Result<Uuid, sqlx::Error> {
280 - let mut tx = pool.begin().await?;
281 - let post_id = create_post_tx(&mut tx, thread_id, author_id, body_markdown, body_html)
282 - .await?
283 - // The pool-taking wrapper is for callers that have already established the
284 - // thread is live (tests, simple internal paths); a locked/deleted thread
285 - // here is a "row not inserted" error rather than a silent no-op.
286 - .ok_or(sqlx::Error::RowNotFound)?;
287 - tx.commit().await?;
288 - Ok(post_id)
289 - }
290 -
291 - /// Insert a reply and bump `last_activity_at` on the caller's transaction.
292 - ///
293 - /// Lets a handler fold the reply *and its mentions* into one atomic unit so a
294 - /// mid-sequence failure can't leave a committed reply whose @mentions silently
295 - /// vanished (ultra-fuzz M-St1). The pool-taking [`create_post`] wraps this in
296 - /// its own transaction for the simple, mention-free callers (and tests).
297 - #[tracing::instrument(skip_all)]
298 - pub async fn create_post_tx(
299 - conn: &mut sqlx::PgConnection,
300 - thread_id: Uuid,
301 - author_id: Uuid,
302 - body_markdown: &str,
303 - body_html: &str,
304 - ) -> Result<Option<Uuid>, sqlx::Error> {
305 - // Insert only while the thread is still live and unlocked, in the same
306 - // statement that reads that state — closing the TOCTOU between a handler's
307 - // snapshot-time `locked`/`deleted_at` check and this insert. Without the
308 - // guard a concurrent lock or soft-delete in the gap would let the reply
309 - // commit (and the m029 post_count trigger increment a deleted thread). None
310 - // = the thread was locked/deleted meanwhile; the caller surfaces it. This
311 - // mirrors the conditional-write pattern `mod_remove_post_cascade` uses.
312 - let post_id = sqlx::query_scalar!(
313 - "INSERT INTO posts (thread_id, author_id, body_markdown, body_html)
314 - SELECT $1, $2, $3, $4
315 - WHERE EXISTS (
316 - SELECT 1 FROM threads
317 - WHERE id = $1 AND deleted_at IS NULL AND locked = false
318 - )
319 - RETURNING id",
320 - thread_id,
321 - author_id,
322 - body_markdown,
323 - body_html,
324 - )
325 - .fetch_optional(&mut *conn)
326 - .await?;
327 -
328 - // Only bump activity if the reply actually landed.
329 - if post_id.is_some() {
330 - sqlx::query!("UPDATE threads SET last_activity_at = now() WHERE id = $1", thread_id)
331 - .execute(&mut *conn)
332 - .await?;
333 - }
334 -
335 - Ok(post_id)
336 - }
337 -
338 - /// Idempotent reply insert keyed on `external_ref`, for the internal API.
339 - ///
340 - /// Mirrors [`create_thread_with_op_external_ref`]: a retried or replayed
341 - /// server→server reply carrying the same `external_ref` inserts at most once.
342 - /// `ON CONFLICT` on the partial unique index (m030) makes the repeat a no-op,
343 - /// so the m029 post_count trigger fires exactly once and `last_activity_at` is
344 - /// bumped only on the fresh insert. Returns `(post_id, created)` — `created` is
345 - /// false when an existing post was returned. User-facing replies use the plain
346 - /// [`create_post`] (no ref); this path is internal-only.
347 - #[tracing::instrument(skip_all)]
348 - pub async fn create_post_external_ref(
349 - pool: &PgPool,
350 - thread_id: Uuid,
351 - author_id: Uuid,
352 - external_ref: &str,
353 - body_markdown: &str,
354 - body_html: &str,
355 - ) -> Result<(Uuid, bool), sqlx::Error> {
356 - let mut tx = pool.begin().await?;
357 -
358 - let inserted = sqlx::query_scalar!(
359 - "INSERT INTO posts (thread_id, author_id, body_markdown, body_html, external_ref)
360 - VALUES ($1, $2, $3, $4, $5)
361 - ON CONFLICT (external_ref) WHERE external_ref IS NOT NULL DO NOTHING
362 - RETURNING id",
363 - thread_id,
364 - author_id,
365 - body_markdown,
366 - body_html,
367 - external_ref,
368 - )
369 - .fetch_optional(&mut *tx)
370 - .await?;
371 -
372 - let result = match inserted {
373 - Some(id) => {
374 - // Fresh reply: bump thread activity (the trigger already counted it).
375 - sqlx::query!("UPDATE threads SET last_activity_at = now() WHERE id = $1", thread_id)
376 - .execute(&mut *tx)
377 - .await?;
378 - (id, true)
379 - }
380 - None => {
381 - // Replay: the reply already exists. Return its id, don't re-bump.
382 - let existing = sqlx::query_scalar!(
383 - "SELECT id FROM posts WHERE external_ref = $1",
384 - external_ref,
385 - )
386 - .fetch_one(&mut *tx)
387 - .await?;
388 - (existing, false)
389 - }
390 - };
391 -
392 - tx.commit().await?;
393 - Ok(result)
394 - }
395 -
396 - /// Insert a footnote on a post. Returns the footnote ID.
397 - #[tracing::instrument(skip_all)]
398 - pub async fn insert_footnote(
399 - pool: &PgPool,
400 - post_id: Uuid,
401 - author_id: Uuid,
402 - body_markdown: &str,
403 - body_html: &str,
404 - ) -> Result<Uuid, sqlx::Error> {
405 - sqlx::query_scalar!(
406 - "INSERT INTO post_footnotes (post_id, author_id, body_markdown, body_html)
407 - VALUES ($1, $2, $3, $4)
408 - RETURNING id",
409 - post_id,
410 - author_id,
411 - body_markdown,
412 - body_html,
413 - )
414 - .fetch_one(pool)
415 - .await
416 - }
417 -
418 - /// Test-only: mod-remove a single post without the OP-cascade.
419 - ///
420 - /// Production code must use [`mod_remove_post_cascade`], which also soft-deletes
421 - /// the thread when the post is its opening post — removing an OP without the
422 - /// cascade leaves a headless, still-repliable thread. This non-cascade variant
423 - /// exists solely so the integration suite can stage a removed post directly; it
424 - /// is gated behind the `test-support` feature so handler code cannot reach it.
425 - #[cfg(feature = "test-support")]
426 - #[tracing::instrument(skip_all)]
427 - pub async fn mod_remove_post(
428 - pool: &PgPool,
429 - post_id: Uuid,
430 - removed_by_id: Uuid,
431 - ) -> Result<bool, sqlx::Error> {
432 - let result = sqlx::query!(
433 - "UPDATE posts SET removed_by = $2, removed_at = now()
434 - WHERE id = $1 AND removed_at IS NULL",
435 - post_id,
436 - removed_by_id,
437 - )
438 - .execute(pool)
439 - .await?;
440 - Ok(result.rows_affected() > 0)
441 - }
442 -
443 - /// Outcome of [`mod_remove_post_cascade`].
444 - #[derive(Debug, Clone, Copy, PartialEq, Eq)]
445 - pub struct PostRemoval {
446 - /// The post was removed by this call (false if it was already removed).
447 - pub post_removed: bool,
448 - /// The post was the thread's opening post, so the whole thread was
449 - /// soft-deleted as well.
450 - pub thread_removed: bool,
451 - }
452 -
453 - /// Mod-remove a post and, if it is the thread's opening post, soft-delete the
454 - /// whole thread in the same transaction.
455 - ///
456 - /// The opening post is the earliest post in the thread (by `created_at`, `id`
457 - /// as tiebreak), counted regardless of removal state so the identity is stable.
458 - /// Removing an OP without this cascade would leave a headless, still-repliable
459 - /// thread; cascading deletes it so listings, the thread view, and the reply
460 - /// path all treat it as gone. Returns false flags when the post was already
461 - /// removed or was not the OP — callers can log a thread-deletion accordingly.
462 - #[tracing::instrument(skip_all)]
463 - pub async fn mod_remove_post_cascade(
464 - conn: &mut sqlx::PgConnection,
465 - post_id: Uuid,
466 - removed_by_id: Uuid,
467 - ) -> Result<PostRemoval, sqlx::Error> {
468 - let post_removed = sqlx::query!(
469 - "UPDATE posts SET removed_by = $2, removed_at = now()
470 - WHERE id = $1 AND removed_at IS NULL",
471 - post_id,
472 - removed_by_id,
473 - )
474 - .execute(&mut *conn)
475 - .await?
476 - .rows_affected()
477 - > 0;
478 -
479 - let mut thread_removed = false;
480 - if post_removed {
481 - // Soft-delete the thread only when this post is its opening post.
482 - thread_removed = sqlx::query!(
483 - "UPDATE threads SET deleted_at = now()
484 - WHERE deleted_at IS NULL
485 - AND id = (SELECT thread_id FROM posts WHERE id = $1)
486 - AND $1 = (
487 - SELECT id FROM posts
488 - WHERE thread_id = (SELECT thread_id FROM posts WHERE id = $1)
489 - ORDER BY created_at ASC, id ASC
490 - LIMIT 1
491 - )",
492 - post_id,
493 - )
494 - .execute(&mut *conn)
495 - .await?
496 - .rows_affected()
497 - > 0;
498 - }
499 -
500 - Ok(PostRemoval { post_removed, thread_removed })
Lines truncated
A multithreaded/crates/mt-db/src/mutations/category.rs +85
@@ -0,0 +1,85 @@
1 + use super::*;
2 +
3 + /// Create a new category in a community.
4 + #[tracing::instrument(skip_all)]
5 + pub async fn create_category<'e, E: sqlx::PgExecutor<'e>>(
6 + executor: E,
7 + community_id: Uuid,
8 + name: &str,
9 + slug: &str,
10 + description: Option<&str>,
11 + sort_order: i32,
12 + ) -> Result<Uuid, sqlx::Error> {
13 + sqlx::query_scalar!(
14 + "INSERT INTO categories (community_id, name, slug, description, sort_order)
15 + VALUES ($1, $2, $3, $4, $5)
16 + RETURNING id",
17 + community_id,
18 + name,
19 + slug,
20 + description,
21 + sort_order,
22 + )
23 + .fetch_one(executor)
24 + .await
25 + }
26 +
27 + /// Update a category's name and description (scoped to community).
28 + #[tracing::instrument(skip_all)]
29 + pub async fn update_category<'e, E: sqlx::PgExecutor<'e>>(
30 + executor: E,
31 + category_id: Uuid,
32 + community_id: Uuid,
33 + name: &str,
34 + description: Option<&str>,
35 + ) -> Result<bool, sqlx::Error> {
36 + let result = sqlx::query!(
37 + "UPDATE categories SET name = $2, description = $3 WHERE id = $1 AND community_id = $4",
38 + category_id,
39 + name,
40 + description,
41 + community_id,
42 + )
43 + .execute(executor)
44 + .await?;
45 + Ok(result.rows_affected() > 0)
46 + }
47 +
48 + /// Swap the sort_order of two categories atomically.
49 + #[tracing::instrument(skip_all)]
50 + pub async fn swap_category_order(
51 + pool: &PgPool,
52 + id_a: Uuid,
53 + order_a: i32,
54 + id_b: Uuid,
55 + order_b: i32,
56 + ) -> Result<(), sqlx::Error> {
57 + let mut tx = pool.begin().await?;
58 + sqlx::query!("UPDATE categories SET sort_order = $2 WHERE id = $1", id_a, order_b)
59 + .execute(&mut *tx)
60 + .await?;
61 + sqlx::query!("UPDATE categories SET sort_order = $2 WHERE id = $1", id_b, order_a)
62 + .execute(&mut *tx)
63 + .await?;
64 + tx.commit().await?;
65 + Ok(())
66 + }
67 +
68 + /// Look up a category ID by community slug + category slug.
69 + #[tracing::instrument(skip_all)]
70 + pub async fn get_category_id_by_slugs(
71 + pool: &PgPool,
72 + community_slug: &str,
73 + category_slug: &str,
74 + ) -> Result<Option<Uuid>, sqlx::Error> {
75 + sqlx::query_scalar!(
76 + "SELECT c.id
77 + FROM categories c
78 + JOIN communities co ON co.id = c.community_id
79 + WHERE co.slug = $1 AND c.slug = $2",
80 + community_slug,
81 + category_slug,
82 + )
83 + .fetch_optional(pool)
84 + .await
85 + }
A multithreaded/crates/mt-db/src/mutations/community.rs +195
@@ -0,0 +1,195 @@
1 + use super::*;
2 +
3 + /// Create a new community and return its ID.
4 + #[tracing::instrument(skip_all)]
5 + pub async fn create_community(
6 + pool: &PgPool,
7 + name: &str,
8 + slug: &str,
9 + description: Option<&str>,
10 + ) -> Result<Uuid, sqlx::Error> {
11 + sqlx::query_scalar!(
12 + "INSERT INTO communities (name, slug, description)
13 + VALUES ($1, $2, $3)
14 + RETURNING id",
15 + name,
16 + slug,
17 + description,
18 + )
19 + .fetch_one(pool)
20 + .await
21 + }
22 +
23 + /// Update a community's name and description.
24 + #[tracing::instrument(skip_all)]
25 + pub async fn update_community<'e, E: sqlx::PgExecutor<'e>>(
26 + executor: E,
27 + community_id: Uuid,
28 + name: &str,
29 + description: Option<&str>,
30 + auto_hide_threshold: Option<i32>,
31 + ) -> Result<(), sqlx::Error> {
32 + sqlx::query!(
33 + "UPDATE communities SET name = $2, description = $3, auto_hide_threshold = $4 WHERE id = $1",
34 + community_id,
35 + name,
36 + description,
37 + auto_hide_threshold,
38 + )
39 + .execute(executor)
40 + .await?;
41 + Ok(())
42 + }
43 +
44 + /// Suspend a community.
45 + #[tracing::instrument(skip_all)]
46 + pub async fn suspend_community<'e, E: sqlx::PgExecutor<'e>>(
47 + executor: E,
48 + community_id: Uuid,
49 + reason: Option<&str>,
50 + ) -> Result<(), sqlx::Error> {
51 + sqlx::query!(
52 + "UPDATE communities SET suspended_at = now(), suspension_reason = $2 WHERE id = $1",
53 + community_id,
54 + reason,
55 + )
56 + .execute(executor)
57 + .await?;
58 + Ok(())
59 + }
60 +
61 + /// Unsuspend a community.
62 + #[tracing::instrument(skip_all)]
63 + pub async fn unsuspend_community<'e, E: sqlx::PgExecutor<'e>>(
64 + executor: E,
65 + community_id: Uuid,
66 + ) -> Result<(), sqlx::Error> {
67 + sqlx::query!(
68 + "UPDATE communities SET suspended_at = NULL, suspension_reason = NULL WHERE id = $1",
69 + community_id,
70 + )
71 + .execute(executor)
72 + .await?;
73 + Ok(())
74 + }
75 +
76 + /// Set the community moderation state. See [`CommunityState`] for semantics.
77 + #[tracing::instrument(skip_all)]
78 + pub async fn set_community_state<'e, E: sqlx::PgExecutor<'e>>(
79 + executor: E,
80 + community_id: Uuid,
81 + state: CommunityState,
82 + ) -> Result<(), sqlx::Error> {
83 + sqlx::query!("UPDATE communities SET state = $2 WHERE id = $1", community_id, state.as_str())
84 + .execute(executor)
85 + .await?;
86 + Ok(())
87 + }
88 +
89 + /// Result of a clean-slate operation.
90 + pub struct CleanSlateResult {
91 + /// Number of threads deleted (excluding the system reset thread that's
92 + /// then inserted). Useful for the success toast.
93 + pub deleted_thread_count: i64,
94 + /// ID of the system "Community reset" thread created in the first
95 + /// category. `None` if the community has no categories — clean-slate
96 + /// still deletes threads but has nowhere to post the notice.
97 + pub system_thread_id: Option<Uuid>,
98 + }
99 +
100 + /// Clean-slate a community: delete all threads (and the posts / footnotes /
101 + /// endorsements / flags / read-positions that cascade from them) while
102 + /// preserving the community row, categories, memberships, bans, mutes, and
103 + /// tags. Posts a system thread "Community reset by &lt;actor&gt; on &lt;date&gt;"
104 + /// in the first category by `sort_order`.
105 + ///
106 + /// Authorization is the caller's responsibility (see `routes/admin.rs`); this
107 + /// mutation only enforces atomicity.
108 + #[tracing::instrument(skip_all)]
109 + pub async fn clean_slate_community(
110 + conn: &mut sqlx::PgConnection,
111 + community_id: Uuid,
112 + actor_id: Uuid,
113 + actor_display: &str,
114 + ) -> Result<CleanSlateResult, sqlx::Error> {
115 + // Delete every thread whose category belongs to this community. Cascades
116 + // reap posts, footnotes, endorsements, flags, read-positions, link
117 + // previews, mentions, and tag joins.
118 + let deleted = sqlx::query_scalar!(
119 + r#"WITH d AS (
120 + DELETE FROM threads
121 + WHERE category_id IN (SELECT id FROM categories WHERE community_id = $1)
122 + RETURNING 1
123 + )
124 + SELECT COUNT(*) AS "count!" FROM d"#,
125 + community_id,
126 + )
127 + .fetch_one(&mut *conn)
128 + .await?;
129 +
130 + // Pick the first category by sort_order to host the reset notice. None
131 + // means the community has no categories — nothing to post into.
132 + let first_category = sqlx::query_scalar!(
133 + "SELECT id FROM categories
134 + WHERE community_id = $1
135 + ORDER BY sort_order
136 + LIMIT 1",
137 + community_id,
138 + )
139 + .fetch_optional(&mut *conn)
140 + .await?;
141 +
142 + let system_thread_id = if let Some(cat_id) = first_category {
143 + let now = Utc::now();
144 + let date = now.format("%Y-%m-%d").to_string();
145 + let title = format!("Community reset by {actor_display} on {date}");
146 + let body_md = format!(
147 + "This community was reset by **{actor_display}** on {date}. All previous threads have been cleared. Settings, categories, members, and bans are preserved.",
148 + );
149 + let body_html = format!(
150 + "<p>This community was reset by <strong>{}</strong> on {}. All previous threads have been cleared. Settings, categories, members, and bans are preserved.</p>",
151 + html_escape(actor_display),
152 + date,
153 + );
154 +
155 + let thread_id = sqlx::query_scalar!(
156 + "INSERT INTO threads (category_id, author_id, title, pinned, locked)
157 + VALUES ($1, $2, $3, TRUE, TRUE)
158 + RETURNING id",
159 + cat_id,
160 + actor_id,
161 + title,
162 + )
163 + .fetch_one(&mut *conn)
164 + .await?;
165 +
166 + sqlx::query!(
167 + "INSERT INTO posts (thread_id, author_id, body_markdown, body_html)
168 + VALUES ($1, $2, $3, $4)",
169 + thread_id,
170 + actor_id,
171 + body_md,
172 + body_html,
173 + )
174 + .execute(&mut *conn)
175 + .await?;
176 +
177 + Some(thread_id)
178 + } else {
179 + None
180 + };
181 +
182 + Ok(CleanSlateResult { deleted_thread_count: deleted, system_thread_id })
183 + }
184 +
185 + /// Minimal HTML-escape for actor display names embedded in the reset notice.
186 + /// We render the notice as a literal HTML string (skipping the markdown
187 + /// pipeline) so we don't have to thread renderer config into the mutation.
188 + fn html_escape(input: &str) -> String {
189 + input
190 + .replace('&', "&amp;")
191 + .replace('<', "&lt;")
192 + .replace('>', "&gt;")
193 + .replace('"', "&quot;")
194 + .replace('\'', "&#39;")
195 + }
A multithreaded/crates/mt-db/src/mutations/endorsement.rs +36
@@ -0,0 +1,36 @@
1 + use super::*;
2 +
3 + /// Toggle endorsement: insert if missing, delete if exists. Returns true if now endorsed.
4 + /// Uses a transaction to prevent race conditions between concurrent toggle requests.
5 + #[tracing::instrument(skip_all)]
6 + pub async fn toggle_endorsement(
7 + pool: &PgPool,
8 + post_id: Uuid,
9 + endorser_id: Uuid,
10 + ) -> Result<bool, sqlx::Error> {
11 + let mut tx = pool.begin().await?;
12 +
13 + let result = sqlx::query!(
14 + "INSERT INTO post_endorsements (post_id, endorser_id) VALUES ($1, $2) ON CONFLICT DO NOTHING",
15 + post_id,
16 + endorser_id,
17 + )
18 + .execute(&mut *tx)
19 + .await?;
20 +
21 + if result.rows_affected() == 0 {
22 + // Already existed — remove it
23 + sqlx::query!(
24 + "DELETE FROM post_endorsements WHERE post_id = $1 AND endorser_id = $2",
25 + post_id,
26 + endorser_id,
27 + )
28 + .execute(&mut *tx)
29 + .await?;
30 + tx.commit().await?;
31 + Ok(false)
32 + } else {
33 + tx.commit().await?;
34 + Ok(true)
35 + }
36 + }
A multithreaded/crates/mt-db/src/mutations/footnote.rs +23
@@ -0,0 +1,23 @@
1 + use super::*;
2 +
3 + /// Insert a footnote on a post. Returns the footnote ID.
4 + #[tracing::instrument(skip_all)]
5 + pub async fn insert_footnote(
6 + pool: &PgPool,
7 + post_id: Uuid,
8 + author_id: Uuid,
9 + body_markdown: &str,
10 + body_html: &str,
11 + ) -> Result<Uuid, sqlx::Error> {
12 + sqlx::query_scalar!(
13 + "INSERT INTO post_footnotes (post_id, author_id, body_markdown, body_html)
14 + VALUES ($1, $2, $3, $4)
15 + RETURNING id",
16 + post_id,
17 + author_id,
18 + body_markdown,
19 + body_html,
20 + )
21 + .fetch_one(pool)
22 + .await
23 + }
A multithreaded/crates/mt-db/src/mutations/image.rs +69
@@ -0,0 +1,69 @@
1 + use super::*;
2 +
3 + /// Insert an uploaded image record.
4 + #[tracing::instrument(skip_all)]
5 + pub async fn insert_image(
6 + pool: &PgPool,
7 + uploader_id: Uuid,
8 + community_id: Uuid,
9 + s3_key: &str,
10 + filename: &str,
11 + content_type: &str,
12 + size_bytes: i64,
13 + ) -> Result<Uuid, sqlx::Error> {
14 + sqlx::query_scalar!(
15 + "INSERT INTO images (uploader_id, community_id, s3_key, filename, content_type, size_bytes)
16 + VALUES ($1, $2, $3, $4, $5, $6)
17 + RETURNING id",
18 + uploader_id,
19 + community_id,
20 + s3_key,
21 + filename,
22 + content_type,
23 + size_bytes,
24 + )
25 + .fetch_one(pool)
26 + .await
27 + }
28 +
29 + /// Mark an image as removed by a moderator.
30 + #[tracing::instrument(skip_all)]
31 + pub async fn remove_image<'e, E: sqlx::PgExecutor<'e>>(
32 + executor: E,
33 + image_id: Uuid,
34 + removed_by: Uuid,
35 + ) -> Result<(), sqlx::Error> {
36 + sqlx::query!(
37 + "UPDATE images SET removed_at = now(), removed_by = $2 WHERE id = $1 AND removed_at IS NULL",
38 + image_id,
39 + removed_by,
40 + )
41 + .execute(executor)
42 + .await?;
43 + Ok(())
44 + }
45 +
46 + /// Hard-delete an image row. Used to roll back the row inserted just before a
47 + /// failed S3 upload — the object never landed in the bucket, so there is
48 + /// nothing to reconcile and the row should simply not exist.
49 + #[tracing::instrument(skip_all)]
50 + pub async fn delete_image_row(pool: &PgPool, image_id: Uuid) -> Result<(), sqlx::Error> {
51 + sqlx::query!("DELETE FROM images WHERE id = $1", image_id)
52 + .execute(pool)
53 + .await?;
54 + Ok(())
55 + }
56 +
57 + /// Mark images whose backing S3 object has been deleted, so the reconcile sweep
58 + /// never revisits them. Called inline after a successful best-effort delete and
59 + /// in batch by the background sweep.
60 + #[tracing::instrument(skip_all)]
61 + pub async fn mark_images_s3_purged(pool: &PgPool, image_ids: &[Uuid]) -> Result<(), sqlx::Error> {
62 + if image_ids.is_empty() {
63 + return Ok(());
64 + }
65 + sqlx::query!("UPDATE images SET s3_purged_at = now() WHERE id = ANY($1)", image_ids)
66 + .execute(pool)
67 + .await?;
68 + Ok(())
69 + }
A multithreaded/crates/mt-db/src/mutations/link_preview.rs +24
@@ -0,0 +1,24 @@
1 + use super::*;
2 +
3 + /// Insert a link preview for a post. Ignores duplicates.
4 + #[tracing::instrument(skip_all)]
5 + pub async fn insert_link_preview(
6 + pool: &PgPool,
7 + post_id: Uuid,
8 + url: &str,
9 + title: Option<&str>,
10 + description: Option<&str>,
11 + ) -> Result<(), sqlx::Error> {
12 + sqlx::query!(
13 + "INSERT INTO link_previews (post_id, url, title, description)
14 + VALUES ($1, $2, $3, $4)
15 + ON CONFLICT (post_id, url) DO NOTHING",
16 + post_id,
17 + url,
18 + title,
19 + description,
20 + )
21 + .execute(pool)
22 + .await?;
23 + Ok(())
24 + }
A multithreaded/crates/mt-db/src/mutations/member.rs +42
@@ -0,0 +1,42 @@
1 + use super::*;
2 +
3 + /// Ensure a user has a membership in a community with the given role.
4 + /// Creates membership if none exists, does nothing if already a member.
5 + #[tracing::instrument(skip_all)]
6 + pub async fn ensure_membership_with_role(
7 + pool: &PgPool,
8 + user_id: Uuid,
9 + community_id: Uuid,
10 + role: CommunityRole,
11 + ) -> Result<(), sqlx::Error> {
12 + sqlx::query!(
13 + "INSERT INTO memberships (user_id, community_id, role)
14 + VALUES ($1, $2, $3)
15 + ON CONFLICT (user_id, community_id) DO NOTHING",
16 + user_id,
17 + community_id,
18 + role.as_str(),
19 + )
20 + .execute(pool)
21 + .await?;
22 + Ok(())
23 + }
24 +
25 + /// Ensure a user has a membership in a community. Creates a 'member' role if none exists.
26 + #[tracing::instrument(skip_all)]
27 + pub async fn ensure_membership(
28 + pool: &PgPool,
29 + user_id: Uuid,
30 + community_id: Uuid,
31 + ) -> Result<(), sqlx::Error> {
32 + sqlx::query!(
33 + "INSERT INTO memberships (user_id, community_id, role)
34 + VALUES ($1, $2, 'member')
35 + ON CONFLICT (user_id, community_id) DO NOTHING",
36 + user_id,
37 + community_id,
38 + )
39 + .execute(pool)
40 + .await?;
41 + Ok(())
42 + }
A multithreaded/crates/mt-db/src/mutations/mention.rs +29
@@ -0,0 +1,29 @@
1 + use super::*;
2 +
3 + /// Insert mention rows for a post (batch insert). Ignores duplicates.
4 + #[tracing::instrument(skip_all)]
5 + pub async fn insert_mentions<'e, E: sqlx::PgExecutor<'e>>(
6 + executor: E,
7 + post_id: Uuid,
8 + user_ids: &[Uuid],
9 + ) -> Result<(), sqlx::Error> {
10 + if user_ids.is_empty() {
11 + return Ok(());
12 + }
13 + // runtime-checked: dynamic SQL (cannot use compile-time macro)
14 + // Build multi-row INSERT: VALUES ($1, $2), ($1, $3), ... ON CONFLICT DO NOTHING
15 + let mut sql = String::from("INSERT INTO post_mentions (post_id, mentioned_user_id) VALUES ");
16 + for (i, _) in user_ids.iter().enumerate() {
17 + if i > 0 {
18 + sql.push_str(", ");
19 + }
20 + sql.push_str(&format!("($1, ${})", i + 2));
21 + }
22 + sql.push_str(" ON CONFLICT DO NOTHING");
23 + let mut query = sqlx::query(&sql).bind(post_id);
24 + for user_id in user_ids {
25 + query = query.bind(user_id);
26 + }
27 + query.execute(executor).await?;
28 + Ok(())
29 + }
A multithreaded/crates/mt-db/src/mutations/mod.rs +36
@@ -0,0 +1,36 @@
1 + //! Database write mutations — inserts, updates, deletes.
2 +
3 + use chrono::{DateTime, Utc};
4 + use mt_core::types::{BanType, CommunityRole, CommunityState, ModAction, ModActor};
5 + use sqlx::PgPool;
6 + use uuid::Uuid;
7 +
8 + mod community;
9 + mod category;
10 + mod thread;
11 + mod post;
12 + mod footnote;
13 + mod member;
14 + mod user;
15 + mod moderation;
16 + mod tracked_thread;
17 + mod tag;
18 + mod link_preview;
19 + mod mention;
20 + mod endorsement;
21 + mod image;
22 +
23 + pub use community::*;
24 + pub use category::*;
25 + pub use thread::*;
26 + pub use post::*;
27 + pub use footnote::*;
28 + pub use member::*;
29 + pub use user::*;
30 + pub use moderation::*;
31 + pub use tracked_thread::*;
32 + pub use tag::*;
33 + pub use link_preview::*;
34 + pub use mention::*;
35 + pub use endorsement::*;
36 + pub use image::*;
A multithreaded/crates/mt-db/src/mutations/moderation.rs +167
@@ -0,0 +1,167 @@
1 + use super::*;
2 +
3 + /// Create or update a ban/mute. Returns the ban ID.
4 + #[tracing::instrument(skip_all)]
5 + pub async fn create_community_ban<'e, E: sqlx::PgExecutor<'e>>(
6 + executor: E,
7 + community_id: Uuid,
8 + user_id: Uuid,
9 + banned_by: Uuid,
10 + ban_type: BanType,
11 + reason: Option<&str>,
12 + expires_at: Option<DateTime<Utc>>,
13 + ) -> Result<Uuid, sqlx::Error> {
14 + // runtime-checked: the full server build unifies sqlx's `time` and `chrono`
15 + // features (the session store pulls in `time`), and with both on the
16 + // compile-time macro infers the TIMESTAMPTZ bind parameter as
17 + // `time::OffsetDateTime`, which a `chrono::DateTime<Utc>` won't satisfy.
18 + // Output-column type can be overridden in the macro; a bind parameter's
19 + // cannot. This is the one write path that binds a chrono timestamp, so it
20 + // stays a runtime-checked query (chrono's Encode handles the bind directly).
21 + sqlx::query_scalar::<_, Uuid>(
22 + "INSERT INTO community_bans (community_id, user_id, banned_by, ban_type, reason, expires_at)
23 + VALUES ($1, $2, $3, $4, $5, $6)
24 + ON CONFLICT (community_id, user_id, ban_type) DO UPDATE
25 + SET banned_by = $3, reason = $5, expires_at = $6, created_at = now()
26 + RETURNING id",
27 + )
28 + .bind(community_id)
29 + .bind(user_id)
30 + .bind(banned_by)
31 + .bind(ban_type.as_str())
32 + .bind(reason)
33 + .bind(expires_at)
34 + .fetch_one(executor)
35 + .await
36 + }
37 +
38 + /// Delete all bans/mutes whose expiration has passed.
39 + #[tracing::instrument(skip_all)]
40 + pub async fn cleanup_expired_bans(pool: &PgPool, community_id: Uuid) -> Result<u64, sqlx::Error> {
41 + let result = sqlx::query!(
42 + "DELETE FROM community_bans
43 + WHERE community_id = $1 AND expires_at IS NOT NULL AND expires_at <= now()",
44 + community_id,
45 + )
46 + .execute(pool)
47 + .await?;
48 + Ok(result.rows_affected())
49 + }
50 +
51 + /// Remove a ban or mute.
52 + #[tracing::instrument(skip_all)]
53 + pub async fn remove_community_ban<'e, E: sqlx::PgExecutor<'e>>(
54 + executor: E,
55 + community_id: Uuid,
56 + user_id: Uuid,
57 + ban_type: BanType,
58 + ) -> Result<(), sqlx::Error> {
59 + sqlx::query!(
60 + "DELETE FROM community_bans
61 + WHERE community_id = $1 AND user_id = $2 AND ban_type = $3",
62 + community_id,
63 + user_id,
64 + ban_type.as_str(),
65 + )
66 + .execute(executor)
67 + .await?;
68 + Ok(())
69 + }
70 +
71 + /// Insert a mod log entry on the given executor.
72 + ///
73 + /// Generic over the executor so the audit row can be written on the *same*
74 + /// transaction as the mutation it records — handlers open one `tx`, run the
75 + /// mutation and this insert on `&mut *tx`, then commit, so an auditable action
76 + /// can never commit without its log row. A `System` actor persists as a NULL
77 + /// `actor_id` (migration 032).
78 + #[tracing::instrument(skip_all)]
79 + pub async fn insert_mod_log<'e, E>(
80 + executor: E,
81 + community_id: Option<Uuid>,
82 + actor: ModActor,
83 + action: ModAction,
84 + target_user: Option<Uuid>,
85 + target_id: Option<Uuid>,
86 + reason: Option<&str>,
87 + ) -> Result<(), sqlx::Error>
88 + where
89 + E: sqlx::PgExecutor<'e>,
90 + {
91 + sqlx::query!(
92 + "INSERT INTO mod_log (community_id, actor_id, action, target_user, target_id, reason)
93 + VALUES ($1, $2, $3, $4, $5, $6)",
94 + community_id,
95 + actor.id(),
96 + action.as_str(),
97 + target_user,
98 + target_id,
99 + reason,
100 + )
101 + .execute(executor)
102 + .await?;
103 + Ok(())
104 + }
105 +
106 + /// Insert a flag on a post. ON CONFLICT DO NOTHING (idempotent per user+post).
107 + #[tracing::instrument(skip_all)]
108 + pub async fn insert_flag(
109 + pool: &PgPool,
110 + post_id: Uuid,
111 + flagger_id: Uuid,
112 + reason: &str,
113 + detail: Option<&str>,
114 + ) -> Result<(), sqlx::Error> {
115 + sqlx::query!(
116 + "INSERT INTO post_flags (post_id, flagger_id, reason, detail)
117 + VALUES ($1, $2, $3, $4)
118 + ON CONFLICT (post_id, flagger_id) DO NOTHING",
119 + post_id,
120 + flagger_id,
121 + reason,
122 + detail,
123 + )
124 + .execute(pool)
125 + .await?;
126 + Ok(())
127 + }
128 +
129 + /// Resolve a single flag.
130 + #[tracing::instrument(skip_all)]
131 + pub async fn resolve_flag(
132 + pool: &PgPool,
133 + flag_id: Uuid,
134 + resolved_by: Uuid,
135 + resolution: &str,
136 + ) -> Result<(), sqlx::Error> {
137 + sqlx::query!(
138 + "UPDATE post_flags SET resolved_at = now(), resolved_by = $2, resolution = $3
139 + WHERE id = $1 AND resolved_at IS NULL",
140 + flag_id,
141 + resolved_by,
142 + resolution,
143 + )
144 + .execute(pool)
145 + .await?;
146 + Ok(())
147 + }
148 +
149 + /// Resolve all unresolved flags for a given post.
150 + #[tracing::instrument(skip_all)]
151 + pub async fn resolve_all_flags_for_post<'e, E: sqlx::PgExecutor<'e>>(
152 + executor: E,
153 + post_id: Uuid,
154 + resolved_by: Uuid,
155 + resolution: &str,
156 + ) -> Result<(), sqlx::Error> {
157 + sqlx::query!(
158 + "UPDATE post_flags SET resolved_at = now(), resolved_by = $2, resolution = $3
159 + WHERE post_id = $1 AND resolved_at IS NULL",
160 + post_id,
161 + resolved_by,
162 + resolution,
163 + )
164 + .execute(executor)
165 + .await?;
166 + Ok(())
167 + }
A multithreaded/crates/mt-db/src/mutations/post.rs +216
@@ -0,0 +1,216 @@
1 + use super::*;
2 +
3 + /// Insert a reply and bump the thread's last_activity_at atomically.
4 + ///
5 + /// Opening posts are created with the thread via [`create_thread_with_op`];
6 + /// this is for replies only. The display reply count derives from the
7 + /// denormalized `threads.post_count`, which the m029 trigger maintains on every
8 + /// post INSERT/DELETE/removed_at change — no counter to update by hand here.
9 + #[tracing::instrument(skip_all)]
10 + pub async fn create_post(
11 + pool: &PgPool,
12 + thread_id: Uuid,
13 + author_id: Uuid,
14 + body_markdown: &str,
15 + body_html: &str,
16 + ) -> Result<Uuid, sqlx::Error> {
17 + let mut tx = pool.begin().await?;
18 + let post_id = create_post_tx(&mut tx, thread_id, author_id, body_markdown, body_html)
19 + .await?
20 + // The pool-taking wrapper is for callers that have already established the
21 + // thread is live (tests, simple internal paths); a locked/deleted thread
22 + // here is a "row not inserted" error rather than a silent no-op.
23 + .ok_or(sqlx::Error::RowNotFound)?;
24 + tx.commit().await?;
25 + Ok(post_id)
26 + }
27 +
28 + /// Insert a reply and bump `last_activity_at` on the caller's transaction.
29 + ///
30 + /// Lets a handler fold the reply *and its mentions* into one atomic unit so a
31 + /// mid-sequence failure can't leave a committed reply whose @mentions silently
32 + /// vanished (ultra-fuzz M-St1). The pool-taking [`create_post`] wraps this in
33 + /// its own transaction for the simple, mention-free callers (and tests).
34 + #[tracing::instrument(skip_all)]
35 + pub async fn create_post_tx(
36 + conn: &mut sqlx::PgConnection,
37 + thread_id: Uuid,
38 + author_id: Uuid,
39 + body_markdown: &str,
40 + body_html: &str,
41 + ) -> Result<Option<Uuid>, sqlx::Error> {
42 + // Insert only while the thread is still live and unlocked, in the same
43 + // statement that reads that state — closing the TOCTOU between a handler's
44 + // snapshot-time `locked`/`deleted_at` check and this insert. Without the
45 + // guard a concurrent lock or soft-delete in the gap would let the reply
46 + // commit (and the m029 post_count trigger increment a deleted thread). None
47 + // = the thread was locked/deleted meanwhile; the caller surfaces it. This
48 + // mirrors the conditional-write pattern `mod_remove_post_cascade` uses.
49 + let post_id = sqlx::query_scalar!(
50 + "INSERT INTO posts (thread_id, author_id, body_markdown, body_html)
51 + SELECT $1, $2, $3, $4
52 + WHERE EXISTS (
53 + SELECT 1 FROM threads
54 + WHERE id = $1 AND deleted_at IS NULL AND locked = false
55 + )
56 + RETURNING id",
57 + thread_id,
58 + author_id,
59 + body_markdown,
60 + body_html,
61 + )
62 + .fetch_optional(&mut *conn)
63 + .await?;
64 +
65 + // Only bump activity if the reply actually landed.
66 + if post_id.is_some() {
67 + sqlx::query!("UPDATE threads SET last_activity_at = now() WHERE id = $1", thread_id)
68 + .execute(&mut *conn)
69 + .await?;
70 + }
71 +
72 + Ok(post_id)
73 + }
74 +
75 + /// Idempotent reply insert keyed on `external_ref`, for the internal API.
76 + ///
77 + /// Mirrors [`create_thread_with_op_external_ref`]: a retried or replayed
78 + /// server→server reply carrying the same `external_ref` inserts at most once.
79 + /// `ON CONFLICT` on the partial unique index (m030) makes the repeat a no-op,
80 + /// so the m029 post_count trigger fires exactly once and `last_activity_at` is
81 + /// bumped only on the fresh insert. Returns `(post_id, created)` — `created` is
82 + /// false when an existing post was returned. User-facing replies use the plain
83 + /// [`create_post`] (no ref); this path is internal-only.
84 + #[tracing::instrument(skip_all)]
85 + pub async fn create_post_external_ref(
86 + pool: &PgPool,
87 + thread_id: Uuid,
88 + author_id: Uuid,
89 + external_ref: &str,
90 + body_markdown: &str,
91 + body_html: &str,
92 + ) -> Result<(Uuid, bool), sqlx::Error> {
93 + let mut tx = pool.begin().await?;
94 +
95 + let inserted = sqlx::query_scalar!(
96 + "INSERT INTO posts (thread_id, author_id, body_markdown, body_html, external_ref)
97 + VALUES ($1, $2, $3, $4, $5)
98 + ON CONFLICT (external_ref) WHERE external_ref IS NOT NULL DO NOTHING
99 + RETURNING id",
100 + thread_id,
101 + author_id,
102 + body_markdown,
103 + body_html,
104 + external_ref,
105 + )
106 + .fetch_optional(&mut *tx)
107 + .await?;
108 +
109 + let result = match inserted {
110 + Some(id) => {
111 + // Fresh reply: bump thread activity (the trigger already counted it).
112 + sqlx::query!("UPDATE threads SET last_activity_at = now() WHERE id = $1", thread_id)
113 + .execute(&mut *tx)
114 + .await?;
115 + (id, true)
116 + }
117 + None => {
118 + // Replay: the reply already exists. Return its id, don't re-bump.
119 + let existing = sqlx::query_scalar!(
120 + "SELECT id FROM posts WHERE external_ref = $1",
121 + external_ref,
122 + )
123 + .fetch_one(&mut *tx)
124 + .await?;
125 + (existing, false)
126 + }
127 + };
128 +
129 + tx.commit().await?;
130 + Ok(result)
131 + }
132 +
133 + /// Test-only: mod-remove a single post without the OP-cascade.
134 + ///
135 + /// Production code must use [`mod_remove_post_cascade`], which also soft-deletes
136 + /// the thread when the post is its opening post — removing an OP without the
137 + /// cascade leaves a headless, still-repliable thread. This non-cascade variant
138 + /// exists solely so the integration suite can stage a removed post directly; it
139 + /// is gated behind the `test-support` feature so handler code cannot reach it.
140 + #[cfg(feature = "test-support")]
141 + #[tracing::instrument(skip_all)]
142 + pub async fn mod_remove_post(
143 + pool: &PgPool,
144 + post_id: Uuid,
145 + removed_by_id: Uuid,
146 + ) -> Result<bool, sqlx::Error> {
147 + let result = sqlx::query!(
148 + "UPDATE posts SET removed_by = $2, removed_at = now()
149 + WHERE id = $1 AND removed_at IS NULL",
150 + post_id,
151 + removed_by_id,
152 + )
153 + .execute(pool)
154 + .await?;
155 + Ok(result.rows_affected() > 0)
156 + }
157 +
158 + /// Outcome of [`mod_remove_post_cascade`].
159 + #[derive(Debug, Clone, Copy, PartialEq, Eq)]
160 + pub struct PostRemoval {
161 + /// The post was removed by this call (false if it was already removed).
162 + pub post_removed: bool,
163 + /// The post was the thread's opening post, so the whole thread was
164 + /// soft-deleted as well.
165 + pub thread_removed: bool,
166 + }
167 +
168 + /// Mod-remove a post and, if it is the thread's opening post, soft-delete the
169 + /// whole thread in the same transaction.
170 + ///
171 + /// The opening post is the earliest post in the thread (by `created_at`, `id`
172 + /// as tiebreak), counted regardless of removal state so the identity is stable.
173 + /// Removing an OP without this cascade would leave a headless, still-repliable
174 + /// thread; cascading deletes it so listings, the thread view, and the reply
175 + /// path all treat it as gone. Returns false flags when the post was already
176 + /// removed or was not the OP — callers can log a thread-deletion accordingly.
177 + #[tracing::instrument(skip_all)]
178 + pub async fn mod_remove_post_cascade(
179 + conn: &mut sqlx::PgConnection,
180 + post_id: Uuid,
181 + removed_by_id: Uuid,
182 + ) -> Result<PostRemoval, sqlx::Error> {
183 + let post_removed = sqlx::query!(
184 + "UPDATE posts SET removed_by = $2, removed_at = now()
185 + WHERE id = $1 AND removed_at IS NULL",
186 + post_id,
187 + removed_by_id,
188 + )
189 + .execute(&mut *conn)
190 + .await?
191 + .rows_affected()
192 + > 0;
193 +
194 + let mut thread_removed = false;
195 + if post_removed {
196 + // Soft-delete the thread only when this post is its opening post.
197 + thread_removed = sqlx::query!(
198 + "UPDATE threads SET deleted_at = now()
199 + WHERE deleted_at IS NULL
200 + AND id = (SELECT thread_id FROM posts WHERE id = $1)
201 + AND $1 = (
202 + SELECT id FROM posts
203 + WHERE thread_id = (SELECT thread_id FROM posts WHERE id = $1)
204 + ORDER BY created_at ASC, id ASC
205 + LIMIT 1
206 + )",
207 + post_id,
208 + )
209 + .execute(&mut *conn)
210 + .await?
211 + .rows_affected()
212 + > 0;
213 + }
214 +
215 + Ok(PostRemoval { post_removed, thread_removed })
216 + }
A multithreaded/crates/mt-db/src/mutations/tag.rs +73
@@ -0,0 +1,73 @@
1 + use super::*;
2 +
3 + /// Create a tag in a community. Returns the tag ID.
4 + #[tracing::instrument(skip_all)]
5 + pub async fn create_tag(
6 + pool: &PgPool,
7 + community_id: Uuid,
8 + name: &str,
9 + slug: &str,
10 + ) -> Result<Uuid, sqlx::Error> {
11 + sqlx::query_scalar!(
12 + "INSERT INTO tags (community_id, name, slug) VALUES ($1, $2, $3) RETURNING id",
13 + community_id,
14 + name,
15 + slug,
16 + )
17 + .fetch_one(pool)
18 + .await
19 + }
20 +
21 + /// Delete a tag (CASCADE removes thread_tags rows), scoped to community.
22 + #[tracing::instrument(skip_all)]
23 + pub async fn delete_tag(pool: &PgPool, tag_id: Uuid, community_id: Uuid) -> Result<bool, sqlx::Error> {
24 + let result = sqlx::query!("DELETE FROM tags WHERE id = $1 AND community_id = $2", tag_id, community_id)
25 + .execute(pool)
26 + .await?;
27 + Ok(result.rows_affected() > 0)
28 + }
29 +
30 + /// Set the tags for a thread (delete existing + batch insert new, in a transaction).
31 + #[tracing::instrument(skip_all)]
32 + pub async fn set_thread_tags(
33 + pool: &PgPool,
34 + thread_id: Uuid,
35 + tag_ids: &[Uuid],
36 + ) -> Result<(), sqlx::Error> {
37 + let mut tx = pool.begin().await?;
38 + set_thread_tags_tx(&mut tx, thread_id, tag_ids).await?;
39 + tx.commit().await?;
40 + Ok(())
41 + }
42 +
43 + /// Set a thread's tags on the caller's transaction, so a handler can fold tag
44 + /// assignment into the same atomic unit as the thread/OP creation (M-St1). The
45 + /// pool-taking [`set_thread_tags`] wraps this in its own transaction.
46 + #[tracing::instrument(skip_all)]
47 + pub async fn set_thread_tags_tx(
48 + conn: &mut sqlx::PgConnection,
49 + thread_id: Uuid,
50 + tag_ids: &[Uuid],
51 + ) -> Result<(), sqlx::Error> {
52 + sqlx::query!("DELETE FROM thread_tags WHERE thread_id = $1", thread_id)
53 + .execute(&mut *conn)
54 + .await?;
55 + if !tag_ids.is_empty() {
56 + // runtime-checked: dynamic SQL (cannot use compile-time macro)
57 + // Build multi-row INSERT: VALUES ($1, $2), ($1, $3), ...
58 + let mut sql = String::from("INSERT INTO thread_tags (thread_id, tag_id) VALUES ");
59 + for (i, _) in tag_ids.iter().enumerate() {
60 + if i > 0 {
61 + sql.push_str(", ");
62 + }
63 + // $1 = thread_id, $2.. = tag_ids
64 + sql.push_str(&format!("($1, ${})", i + 2));
65 + }
66 + let mut query = sqlx::query(&sql).bind(thread_id);
67 + for tag_id in tag_ids {
68 + query = query.bind(tag_id);
69 + }
70 + query.execute(&mut *conn).await?;
71 + }
72 + Ok(())
73 + }
A multithreaded/crates/mt-db/src/mutations/thread.rs +208
@@ -0,0 +1,208 @@
1 + use super::*;
2 +
3 + /// Atomically create an externally-referenced thread and its opening post in
4 + /// one transaction. Returns `(thread_id, op_post_id)`.
5 + ///
6 + /// The atomicity matters most here: the MNW→MT internal path retries, and a
7 + /// committed-but-post-less thread would make the retry collide with the UNIQUE
8 + /// `external_ref` index and 500 while leaking an empty thread.
9 + #[tracing::instrument(skip_all)]
10 + pub async fn create_thread_with_op_external_ref(
11 + pool: &PgPool,
12 + category_id: Uuid,
13 + author_id: Uuid,
14 + title: &str,
15 + external_ref: &str,
16 + body_markdown: &str,
17 + body_html: &str,
18 + ) -> Result<(Uuid, Uuid), sqlx::Error> {
19 + let mut tx = pool.begin().await?;
20 +
21 + let thread_id = sqlx::query_scalar!(
22 + "INSERT INTO threads (category_id, author_id, title, external_ref)
23 + VALUES ($1, $2, $3, $4)
24 + RETURNING id",
25 + category_id,
26 + author_id,
27 + title,
28 + external_ref,
29 + )
30 + .fetch_one(&mut *tx)
31 + .await?;
32 +
33 + let post_id = sqlx::query_scalar!(
34 + "INSERT INTO posts (thread_id, author_id, body_markdown, body_html)
35 + VALUES ($1, $2, $3, $4)
36 + RETURNING id",
37 + thread_id,
38 + author_id,
39 + body_markdown,
40 + body_html,
41 + )
42 + .fetch_one(&mut *tx)
43 + .await?;
44 +
45 + tx.commit().await?;
46 + Ok((thread_id, post_id))
47 + }
48 +
49 + /// Low-level: insert a bare thread row (no opening post). Returns the thread ID.
50 + ///
51 + /// Handlers must NOT pair this with a separate `create_post` for the OP — that
52 + /// non-atomic sequence can leak a post-less thread. Use [`create_thread_with_op`]
53 + /// for the real "start a thread" operation. This primitive exists for tests and
54 + /// data construction that build posts explicitly.
55 + #[tracing::instrument(skip_all)]
56 + pub async fn create_thread(
57 + pool: &PgPool,
58 + category_id: Uuid,
59 + author_id: Uuid,
60 + title: &str,
61 + ) -> Result<Uuid, sqlx::Error> {
62 + sqlx::query_scalar!(
63 + "INSERT INTO threads (category_id, author_id, title)
64 + VALUES ($1, $2, $3)
65 + RETURNING id",
66 + category_id,
67 + author_id,
68 + title,
69 + )
70 + .fetch_one(pool)
71 + .await
72 + }
73 +
74 + /// Atomically create a thread and its opening post in one transaction.
75 + ///
76 + /// Returns `(thread_id, op_post_id)`. Creating the thread and OP separately
77 + /// could leave a titled, post-less thread if the second insert failed; this
78 + /// makes that state unreachable. `threads.last_activity_at` defaults to now().
79 + #[tracing::instrument(skip_all)]
80 + pub async fn create_thread_with_op(
81 + pool: &PgPool,
82 + category_id: Uuid,
83 + author_id: Uuid,
84 + title: &str,
85 + body_markdown: &str,
86 + body_html: &str,
87 + ) -> Result<(Uuid, Uuid), sqlx::Error> {
88 + let mut tx = pool.begin().await?;
89 + let ids =
90 + create_thread_with_op_tx(&mut tx, category_id, author_id, title, body_markdown, body_html)
91 + .await?;
92 + tx.commit().await?;
93 + Ok(ids)
94 + }
95 +
96 + /// Create a thread and its opening post on the caller's transaction.
97 + ///
98 + /// Lets a handler fold the thread, its OP, its mentions, and its tags into one
99 + /// atomic unit so a mid-sequence failure can't leave a thread whose @mentions or
100 + /// tags silently vanished (ultra-fuzz M-St1). The pool-taking
101 + /// [`create_thread_with_op`] wraps this in its own transaction.
102 + #[tracing::instrument(skip_all)]
103 + pub async fn create_thread_with_op_tx(
104 + conn: &mut sqlx::PgConnection,
105 + category_id: Uuid,
106 + author_id: Uuid,
107 + title: &str,
108 + body_markdown: &str,
109 + body_html: &str,
110 + ) -> Result<(Uuid, Uuid), sqlx::Error> {
111 + let thread_id = sqlx::query_scalar!(
112 + "INSERT INTO threads (category_id, author_id, title)
113 + VALUES ($1, $2, $3)
114 + RETURNING id",
115 + category_id,
116 + author_id,
117 + title,
118 + )
119 + .fetch_one(&mut *conn)
120 + .await?;
121 +
122 + let post_id = sqlx::query_scalar!(
123 + "INSERT INTO posts (thread_id, author_id, body_markdown, body_html)
124 + VALUES ($1, $2, $3, $4)
125 + RETURNING id",
126 + thread_id,
127 + author_id,
128 + body_markdown,
129 + body_html,
130 + )
131 + .fetch_one(&mut *conn)
132 + .await?;
133 +
134 + Ok((thread_id, post_id))
135 + }
136 +
137 + /// Atomically auto-hide a post if pending flag count meets the threshold.
138 + /// Combines count check and removal in a single query to avoid race conditions.
139 + /// Returns true if the post was actually removed.
140 + /// Sets removed_by to NULL (system action) — the mod log records the event.
141 + #[tracing::instrument(skip_all)]
142 + pub async fn auto_hide_if_threshold_met<'e, E: sqlx::PgExecutor<'e>>(
143 + executor: E,
144 + post_id: Uuid,
145 + threshold: i32,
146 + ) -> Result<bool, sqlx::Error> {
147 + let result = sqlx::query!(
148 + "UPDATE posts SET removed_by = NULL, removed_at = now()
149 + WHERE id = $1 AND removed_at IS NULL
150 + AND (SELECT COUNT(*) FROM post_flags WHERE post_id = $1 AND resolved_at IS NULL) >= $2",
151 + post_id,
152 + threshold as i64,
153 + )
154 + .execute(executor)
155 + .await?;
156 + Ok(result.rows_affected() > 0)
157 + }
158 +
159 + /// Update a thread's title.
160 + #[tracing::instrument(skip_all)]
161 + pub async fn update_thread_title(
162 + pool: &PgPool,
163 + thread_id: Uuid,
164 + title: &str,
165 + ) -> Result<(), sqlx::Error> {
166 + sqlx::query!("UPDATE threads SET title = $2 WHERE id = $1", thread_id, title)
167 + .execute(pool)
168 + .await?;
169 + Ok(())
170 + }
171 +
172 + /// Soft-delete a thread: set deleted_at (hides from listings).
173 + #[tracing::instrument(skip_all)]
174 + pub async fn soft_delete_thread<'e, E: sqlx::PgExecutor<'e>>(
175 + executor: E,
176 + thread_id: Uuid,
177 + ) -> Result<(), sqlx::Error> {
178 + sqlx::query!("UPDATE threads SET deleted_at = now() WHERE id = $1", thread_id)
179 + .execute(executor)
180 + .await?;
181 + Ok(())
182 + }
183 +
184 + /// Set or unset the pinned flag on a thread.
185 + #[tracing::instrument(skip_all)]
186 + pub async fn set_thread_pinned<'e, E: sqlx::PgExecutor<'e>>(
187 + executor: E,
188 + thread_id: Uuid,
189 + pinned: bool,
190 + ) -> Result<(), sqlx::Error> {
191 + sqlx::query!("UPDATE threads SET pinned = $2 WHERE id = $1", thread_id, pinned)
192 + .execute(executor)
193 + .await?;
194 + Ok(())
195 + }
196 +
197 + /// Set or unset the locked flag on a thread.
198 + #[tracing::instrument(skip_all)]
199 + pub async fn set_thread_locked<'e, E: sqlx::PgExecutor<'e>>(
200 + executor: E,
201 + thread_id: Uuid,
202 + locked: bool,
203 + ) -> Result<(), sqlx::Error> {
204 + sqlx::query!("UPDATE threads SET locked = $2 WHERE id = $1", thread_id, locked)
205 + .execute(executor)
206 + .await?;
207 + Ok(())
208 + }
A multithreaded/crates/mt-db/src/mutations/tracked_thread.rs +69
@@ -0,0 +1,69 @@
1 + use super::*;
2 +
3 + /// Track a thread (upsert).
4 + #[tracing::instrument(skip_all)]
5 + pub async fn track_thread(
6 + pool: &PgPool,
7 + user_id: Uuid,
8 + thread_id: Uuid,
9 + ) -> Result<(), sqlx::Error> {
10 + sqlx::query!(
11 + "INSERT INTO tracked_threads (user_id, thread_id)
12 + VALUES ($1, $2)
13 + ON CONFLICT (user_id, thread_id) DO NOTHING",
14 + user_id,
15 + thread_id,
16 + )
17 + .execute(pool)
18 + .await?;
19 + Ok(())
20 + }
21 +
22 + /// Untrack a thread.
23 + #[tracing::instrument(skip_all)]
24 + pub async fn untrack_thread(
25 + pool: &PgPool,
26 + user_id: Uuid,
27 + thread_id: Uuid,
28 + ) -> Result<(), sqlx::Error> {
29 + sqlx::query!(
30 + "DELETE FROM tracked_threads WHERE user_id = $1 AND thread_id = $2",
31 + user_id,
32 + thread_id,
33 + )
34 + .execute(pool)
35 + .await?;
36 + Ok(())
37 + }
38 +
39 + /// Stop tracking all threads for a user.
40 + #[tracing::instrument(skip_all)]
41 + pub async fn untrack_all(
42 + pool: &PgPool,
43 + user_id: Uuid,
44 + ) -> Result<(), sqlx::Error> {
45 + sqlx::query!("DELETE FROM tracked_threads WHERE user_id = $1", user_id)
46 + .execute(pool)
47 + .await?;
48 + Ok(())
49 + }
50 +
51 + /// Update the read position for a tracked thread (set last_read_post_id to the last post).
52 + #[tracing::instrument(skip_all)]
53 + pub async fn update_read_position(
54 + pool: &PgPool,
55 + user_id: Uuid,
56 + thread_id: Uuid,
57 + last_post_id: Uuid,
58 + ) -> Result<(), sqlx::Error> {
59 + sqlx::query!(
60 + "UPDATE tracked_threads SET last_read_post_id = $3
61 + WHERE user_id = $1 AND thread_id = $2",
62 + user_id,
63 + thread_id,
64 + last_post_id,
65 + )
66 + .execute(pool)
67 + .await?;
68 + Ok(())
69 + }
A multithreaded/crates/mt-db/src/mutations/user.rs +129
@@ -0,0 +1,129 @@
1 + use super::*;
2 +
3 + /// Upsert a user from MNW account data. Creates the user if they don't exist,
4 + /// updates username/display_name if they do.
5 + #[tracing::instrument(skip_all)]
6 + pub async fn upsert_user(
7 + pool: &PgPool,
8 + mnw_account_id: Uuid,
9 + username: &str,
10 + display_name: Option<&str>,
11 + ) -> Result<(), sqlx::Error> {
12 + let mut tx = pool.begin().await?;
13 + vacate_stale_username(&mut tx, mnw_account_id, username).await?;
14 + sqlx::query!(
15 + "INSERT INTO users (mnw_account_id, username, display_name)
16 + VALUES ($1, $2, $3)
17 + ON CONFLICT (mnw_account_id) DO UPDATE
18 + SET username = $2, display_name = $3, updated_at = now()",
19 + mnw_account_id,
20 + username,
21 + display_name,
22 + )
23 + .execute(&mut *tx)
24 + .await?;
25 + tx.commit().await?;
26 + Ok(())
27 + }
28 +
29 + /// Free `username` from any *other* user row that still holds it, so the caller's
30 + /// upsert can claim it without tripping the `users_username_key` unique index.
31 + ///
32 + /// MNW is the source of truth for usernames and they are reusable there; a local
33 + /// mirror row holding a name its MNW account no longer owns is stale and must
34 + /// yield. Run this on the same transaction as the upsert (a single statement
35 + /// can't, because a non-deferrable unique index is checked per-row and the
36 + /// rename wouldn't yet be visible to the insert). The vacated row keeps its data
37 + /// under a collision-proof `mnw_<account-id>` placeholder until that user's own
38 + /// next login resets it. Closes the username-collision login lockout (S2).
39 + async fn vacate_stale_username(
40 + conn: &mut sqlx::PgConnection,
41 + mnw_account_id: Uuid,
42 + username: &str,
43 + ) -> Result<(), sqlx::Error> {
44 + sqlx::query!(
45 + "UPDATE users
46 + SET username = 'mnw_' || mnw_account_id::text, updated_at = now()
47 + WHERE username = $1 AND mnw_account_id <> $2",
48 + username,
49 + mnw_account_id,
50 + )
51 + .execute(&mut *conn)
52 + .await?;
53 + Ok(())
54 + }
55 +
56 + /// Public wrapper over [`vacate_stale_username`] for the OAuth login upsert in the
57 + /// web crate, which builds its INSERT with a runtime query.
58 + pub async fn vacate_username_for_login(
59 + conn: &mut sqlx::PgConnection,
60 + mnw_account_id: Uuid,
61 + username: &str,
62 + ) -> Result<(), sqlx::Error> {
63 + vacate_stale_username(conn, mnw_account_id, username).await
64 + }
65 +
66 + /// Suspend a user.
67 + #[tracing::instrument(skip_all)]
68 + pub async fn suspend_user<'e, E: sqlx::PgExecutor<'e>>(
69 + executor: E,
70 + user_id: Uuid,
71 + reason: Option<&str>,
72 + ) -> Result<(), sqlx::Error> {
73 + sqlx::query!(
74 + "UPDATE users SET suspended_at = now(), suspension_reason = $2 WHERE mnw_account_id = $1",
75 + user_id,
76 + reason,
77 + )
78 + .execute(executor)
79 + .await?;
80 + Ok(())
81 + }
82 +
83 + /// Unsuspend a user.
84 + #[tracing::instrument(skip_all)]
85 + pub async fn unsuspend_user<'e, E: sqlx::PgExecutor<'e>>(
86 + executor: E,
87 + user_id: Uuid,
88 + ) -> Result<(), sqlx::Error> {
89 + sqlx::query!(
90 + "UPDATE users SET suspended_at = NULL, suspension_reason = NULL WHERE mnw_account_id = $1",
91 + user_id,
92 + )
93 + .execute(executor)
94 + .await?;
95 + Ok(())
96 + }
97 +
98 + /// Clear a user's saved signature (markdown + rendered html).
99 + #[tracing::instrument(skip_all)]
100 + pub async fn clear_user_signature(pool: &PgPool, user_id: Uuid) -> Result<(), sqlx::Error> {
101 + sqlx::query!(
102 + "UPDATE users SET signature_markdown = NULL, signature_html = NULL \
103 + WHERE mnw_account_id = $1",
104 + user_id,
105 + )
106 + .execute(pool)
107 + .await?;
108 + Ok(())
109 + }
110 +
111 + /// Save a user's signature markdown and its pre-rendered html.
112 + #[tracing::instrument(skip_all)]
113 + pub async fn set_user_signature(
114 + pool: &PgPool,
115 + user_id: Uuid,
116 + markdown: &str,
117 + html: &str,
118 + ) -> Result<(), sqlx::Error> {
119 + sqlx::query!(
120 + "UPDATE users SET signature_markdown = $2, signature_html = $3 \
121 + WHERE mnw_account_id = $1",
122 + user_id,
123 + markdown,
124 + html,
125 + )
126 + .execute(pool)
127 + .await?;
128 + Ok(())
129 + }