Skip to main content

max / makenotwork

mt: bind every mod-log write to its mutation's transaction (ultra-fuzz Run #5 S3 + M-DI1) Auditable moderation actions now write their mod_log row on the same transaction as the mutation, so an action can never commit without its correctly-attributed audit row. Adds ModActor{System,User} + migration 032 (nullable mod_log.actor_id); makes insert_mod_log and the auditable mutations executor-generic; routes all 23 log sites through begin_tx/audit/commit_tx and deletes the fire-and-forget log_mod_action. Flag-threshold auto-hide is logged as System (NULL actor_id), not the member who tripped it. Also: gate the dead non-cascade mod_remove_post behind mt-db's test-support feature (unreachable from production handlers), and filter mod-removed posts / soft-deleted threads out of quote verification. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-06-23 05:52 UTC
Commit: f624d4a36379d8f317915300f08686f0e10f323f
Parent: 5a51b7a
16 files changed, +453 insertions, -254 deletions
A multithreaded/.sqlx/query-947028eba046c3be1515a1a0adc8102bd343d5ec539a18aa296f53c508d5e3e2.json +28
@@ -0,0 +1,28 @@
1 + {
2 + "db_name": "PostgreSQL",
3 + "query": "SELECT p.author_id, p.body_markdown\n FROM posts p\n JOIN threads t ON t.id = p.thread_id\n WHERE p.id = $1 AND p.removed_at IS NULL AND t.deleted_at IS NULL",
4 + "describe": {
5 + "columns": [
6 + {
7 + "ordinal": 0,
8 + "name": "author_id",
9 + "type_info": "Uuid"
10 + },
11 + {
12 + "ordinal": 1,
13 + "name": "body_markdown",
14 + "type_info": "Text"
15 + }
16 + ],
17 + "parameters": {
18 + "Left": [
19 + "Uuid"
20 + ]
21 + },
22 + "nullable": [
23 + false,
24 + false
25 + ]
26 + },
27 + "hash": "947028eba046c3be1515a1a0adc8102bd343d5ec539a18aa296f53c508d5e3e2"
28 + }
D multithreaded/.sqlx/query-c36c7dec98b8dd65d16ebb39e878b4da92252ea6cf08f57b3557afae25bf5950.json -28
@@ -1,28 +0,0 @@
1 - {
2 - "db_name": "PostgreSQL",
3 - "query": "SELECT author_id, body_markdown FROM posts WHERE id = $1",
4 - "describe": {
5 - "columns": [
6 - {
7 - "ordinal": 0,
8 - "name": "author_id",
9 - "type_info": "Uuid"
10 - },
11 - {
12 - "ordinal": 1,
13 - "name": "body_markdown",
14 - "type_info": "Text"
15 - }
16 - ],
17 - "parameters": {
18 - "Left": [
19 - "Uuid"
20 - ]
21 - },
22 - "nullable": [
23 - false,
24 - false
25 - ]
26 - },
27 - "hash": "c36c7dec98b8dd65d16ebb39e878b4da92252ea6cf08f57b3557afae25bf5950"
28 - }
M multithreaded/Cargo.toml +3
@@ -98,3 +98,6 @@ time = "0.3"
98 98 http-body-util = "0.1"
99 99 wiremock = "0.6"
100 100 pom-contract = { path = "../shared/pom-contract" }
101 + # Enable mt-db's setup-only mutations for the integration suite. Resolver 2 keeps
102 + # this feature out of the normal/production build.
103 + mt-db = { workspace = true, features = ["test-support"] }
M multithreaded/crates/mt-core/src/types.rs +23
@@ -1,6 +1,7 @@
1 1 //! Typed enums replacing raw string values across the codebase.
2 2
3 3 use serde::Serialize;
4 + use sqlx::types::Uuid;
4 5 use std::fmt;
5 6
6 7 // ============================================================================
@@ -217,6 +218,28 @@ impl ModAction {
217 218 }
218 219 }
219 220
221 + /// Who performed a moderation action, for attribution in the mod log.
222 + ///
223 + /// A `System` action (e.g. flag-threshold auto-hide) is recorded with a NULL
224 + /// `actor_id`, mirroring the `posts.removed_by IS NULL` convention — never the
225 + /// user who happened to trip the threshold. `User` carries the moderator's
226 + /// `mnw_account_id`.
227 + #[derive(Debug, Clone, Copy, PartialEq, Eq)]
228 + pub enum ModActor {
229 + System,
230 + User(Uuid),
231 + }
232 +
233 + impl ModActor {
234 + /// The `actor_id` to persist: `None` for system actions, `Some(id)` for a user.
235 + pub fn id(self) -> Option<Uuid> {
236 + match self {
237 + Self::System => None,
238 + Self::User(id) => Some(id),
239 + }
240 + }
241 + }
242 +
220 243 impl fmt::Display for ModAction {
221 244 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
222 245 f.write_str(self.as_str())
M multithreaded/crates/mt-db/Cargo.toml +5
@@ -3,6 +3,11 @@ name = "mt-db"
3 3 version.workspace = true
4 4 edition.workspace = true
5 5
6 + [features]
7 + # Exposes setup-only mutations (e.g. `mod_remove_post`) used by the integration
8 + # test suite. Off in normal/production builds so handler code cannot call them.
9 + test-support = []
10 +
6 11 [dependencies]
7 12 mt-core = { workspace = true }
8 13 sqlx = { workspace = true }
M multithreaded/crates/mt-db/src/mutations.rs +83 -71
@@ -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};
4 + use mt_core::types::{BanType, CommunityState, ModAction, ModActor};
5 5 use sqlx::PgPool;
6 6 use uuid::Uuid;
7 7
@@ -319,8 +319,14 @@ pub async fn insert_footnote(
319 319 .await
320 320 }
321 321
322 - /// Mod-remove a post: set removed_by/removed_at. Content stays intact for audit.
323 - /// Returns true if the post was actually removed (false if already removed).
322 + /// Test-only: mod-remove a single post without the OP-cascade.
323 + ///
324 + /// Production code must use [`mod_remove_post_cascade`], which also soft-deletes
325 + /// the thread when the post is its opening post — removing an OP without the
326 + /// cascade leaves a headless, still-repliable thread. This non-cascade variant
327 + /// exists solely so the integration suite can stage a removed post directly; it
328 + /// is gated behind the `test-support` feature so handler code cannot reach it.
329 + #[cfg(feature = "test-support")]
324 330 #[tracing::instrument(skip_all)]
325 331 pub async fn mod_remove_post(
326 332 pool: &PgPool,
@@ -359,19 +365,17 @@ pub struct PostRemoval {
359 365 /// removed or was not the OP — callers can log a thread-deletion accordingly.
360 366 #[tracing::instrument(skip_all)]
361 367 pub async fn mod_remove_post_cascade(
362 - pool: &PgPool,
368 + conn: &mut sqlx::PgConnection,
363 369 post_id: Uuid,
364 370 removed_by_id: Uuid,
365 371 ) -> Result<PostRemoval, sqlx::Error> {
366 - let mut tx = pool.begin().await?;
367 -
368 372 let post_removed = sqlx::query!(
369 373 "UPDATE posts SET removed_by = $2, removed_at = now()
370 374 WHERE id = $1 AND removed_at IS NULL",
371 375 post_id,
372 376 removed_by_id,
373 377 )
374 - .execute(&mut *tx)
378 + .execute(&mut *conn)
375 379 .await?
376 380 .rows_affected()
377 381 > 0;
@@ -391,13 +395,12 @@ pub async fn mod_remove_post_cascade(
391 395 )",
392 396 post_id,
393 397 )
394 - .execute(&mut *tx)
398 + .execute(&mut *conn)
395 399 .await?
396 400 .rows_affected()
397 401 > 0;
398 402 }
399 403
400 - tx.commit().await?;
401 404 Ok(PostRemoval { post_removed, thread_removed })
402 405 }
403 406
@@ -406,8 +409,8 @@ pub async fn mod_remove_post_cascade(
406 409 /// Returns true if the post was actually removed.
407 410 /// Sets removed_by to NULL (system action) — the mod log records the event.
408 411 #[tracing::instrument(skip_all)]
409 - pub async fn auto_hide_if_threshold_met(
410 - pool: &PgPool,
412 + pub async fn auto_hide_if_threshold_met<'e, E: sqlx::PgExecutor<'e>>(
413 + executor: E,
411 414 post_id: Uuid,
412 415 threshold: i32,
413 416 ) -> Result<bool, sqlx::Error> {
@@ -418,7 +421,7 @@ pub async fn auto_hide_if_threshold_met(
418 421 post_id,
419 422 threshold as i64,
420 423 )
421 - .execute(pool)
424 + .execute(executor)
422 425 .await?;
423 426 Ok(result.rows_affected() > 0)
424 427 }
@@ -438,43 +441,46 @@ pub async fn update_thread_title(
438 441
439 442 /// Soft-delete a thread: set deleted_at (hides from listings).
440 443 #[tracing::instrument(skip_all)]
441 - pub async fn soft_delete_thread(pool: &PgPool, thread_id: Uuid) -> Result<(), sqlx::Error> {
444 + pub async fn soft_delete_thread<'e, E: sqlx::PgExecutor<'e>>(
445 + executor: E,
446 + thread_id: Uuid,
447 + ) -> Result<(), sqlx::Error> {
442 448 sqlx::query!("UPDATE threads SET deleted_at = now() WHERE id = $1", thread_id)
443 - .execute(pool)
449 + .execute(executor)
444 450 .await?;
445 451 Ok(())
446 452 }
447 453
448 454 /// Set or unset the pinned flag on a thread.
449 455 #[tracing::instrument(skip_all)]
450 - pub async fn set_thread_pinned(
451 - pool: &PgPool,
456 + pub async fn set_thread_pinned<'e, E: sqlx::PgExecutor<'e>>(
457 + executor: E,
452 458 thread_id: Uuid,
453 459 pinned: bool,
454 460 ) -> Result<(), sqlx::Error> {
455 461 sqlx::query!("UPDATE threads SET pinned = $2 WHERE id = $1", thread_id, pinned)
456 - .execute(pool)
462 + .execute(executor)
457 463 .await?;
458 464 Ok(())
459 465 }
460 466
461 467 /// Set or unset the locked flag on a thread.
462 468 #[tracing::instrument(skip_all)]
463 - pub async fn set_thread_locked(
464 - pool: &PgPool,
469 + pub async fn set_thread_locked<'e, E: sqlx::PgExecutor<'e>>(
470 + executor: E,
465 471 thread_id: Uuid,
466 472 locked: bool,
467 473 ) -> Result<(), sqlx::Error> {
468 474 sqlx::query!("UPDATE threads SET locked = $2 WHERE id = $1", thread_id, locked)
469 - .execute(pool)
475 + .execute(executor)
470 476 .await?;
471 477 Ok(())
472 478 }
473 479
474 480 /// Update a community's name and description.
475 481 #[tracing::instrument(skip_all)]
476 - pub async fn update_community(
477 - pool: &PgPool,
482 + pub async fn update_community<'e, E: sqlx::PgExecutor<'e>>(
483 + executor: E,
478 484 community_id: Uuid,
479 485 name: &str,
480 486 description: Option<&str>,
@@ -487,15 +493,15 @@ pub async fn update_community(
487 493 description,
488 494 auto_hide_threshold,
489 495 )
490 - .execute(pool)
496 + .execute(executor)
491 497 .await?;
492 498 Ok(())
493 499 }
494 500
495 501 /// Create a new category in a community.
496 502 #[tracing::instrument(skip_all)]
497 - pub async fn create_category(
498 - pool: &PgPool,
503 + pub async fn create_category<'e, E: sqlx::PgExecutor<'e>>(
504 + executor: E,
499 505 community_id: Uuid,
500 506 name: &str,
501 507 slug: &str,
@@ -512,14 +518,14 @@ pub async fn create_category(
512 518 description,
513 519 sort_order,
514 520 )
515 - .fetch_one(pool)
521 + .fetch_one(executor)
516 522 .await
517 523 }
518 524
519 525 /// Update a category's name and description (scoped to community).
520 526 #[tracing::instrument(skip_all)]
521 - pub async fn update_category(
522 - pool: &PgPool,
527 + pub async fn update_category<'e, E: sqlx::PgExecutor<'e>>(
528 + executor: E,
523 529 category_id: Uuid,
524 530 community_id: Uuid,
525 531 name: &str,
@@ -532,7 +538,7 @@ pub async fn update_category(
532 538 description,
533 539 community_id,
534 540 )
535 - .execute(pool)
541 + .execute(executor)
536 542 .await?;
537 543 Ok(result.rows_affected() > 0)
538 544 }
@@ -582,8 +588,8 @@ pub async fn get_category_id_by_slugs(
582 588
583 589 /// Create or update a ban/mute. Returns the ban ID.
584 590 #[tracing::instrument(skip_all)]
585 - pub async fn create_community_ban(
586 - pool: &PgPool,
591 + pub async fn create_community_ban<'e, E: sqlx::PgExecutor<'e>>(
592 + executor: E,
587 593 community_id: Uuid,
588 594 user_id: Uuid,
589 595 banned_by: Uuid,
@@ -611,7 +617,7 @@ pub async fn create_community_ban(
611 617 .bind(ban_type.as_str())
612 618 .bind(reason)
613 619 .bind(expires_at)
614 - .fetch_one(pool)
620 + .fetch_one(executor)
615 621 .await
616 622 }
617 623
@@ -630,8 +636,8 @@ pub async fn cleanup_expired_bans(pool: &PgPool, community_id: Uuid) -> Result<u
630 636
631 637 /// Remove a ban or mute.
632 638 #[tracing::instrument(skip_all)]
633 - pub async fn remove_community_ban(
634 - pool: &PgPool,
639 + pub async fn remove_community_ban<'e, E: sqlx::PgExecutor<'e>>(
640 + executor: E,
635 641 community_id: Uuid,
636 642 user_id: Uuid,
637 643 ban_type: BanType,
@@ -643,33 +649,42 @@ pub async fn remove_community_ban(
643 649 user_id,
644 650 ban_type.as_str(),
645 651 )
646 - .execute(pool)
652 + .execute(executor)
647 653 .await?;
648 654 Ok(())
649 655 }
650 656
651 - /// Insert a mod log entry.
657 + /// Insert a mod log entry on the given executor.
658 + ///
659 + /// Generic over the executor so the audit row can be written on the *same*
660 + /// transaction as the mutation it records — handlers open one `tx`, run the
661 + /// mutation and this insert on `&mut *tx`, then commit, so an auditable action
662 + /// can never commit without its log row. A `System` actor persists as a NULL
663 + /// `actor_id` (migration 032).
652 664 #[tracing::instrument(skip_all)]
653 - pub async fn insert_mod_log(
654 - pool: &PgPool,
665 + pub async fn insert_mod_log<'e, E>(
666 + executor: E,
655 667 community_id: Option<Uuid>,
656 - actor_id: Uuid,
668 + actor: ModActor,
657 669 action: ModAction,
658 670 target_user: Option<Uuid>,
659 671 target_id: Option<Uuid>,
660 672 reason: Option<&str>,
661 - ) -> Result<(), sqlx::Error> {
673 + ) -> Result<(), sqlx::Error>
674 + where
675 + E: sqlx::PgExecutor<'e>,
676 + {
662 677 sqlx::query!(
663 678 "INSERT INTO mod_log (community_id, actor_id, action, target_user, target_id, reason)
664 679 VALUES ($1, $2, $3, $4, $5, $6)",
665 680 community_id,
666 - actor_id,
681 + actor.id(),
667 682 action.as_str(),
668 683 target_user,
669 684 target_id,
670 685 reason,
671 686 )
672 - .execute(pool)
687 + .execute(executor)
673 688 .await?;
674 689 Ok(())
675 690 }
@@ -680,8 +695,8 @@ pub async fn insert_mod_log(
680 695
681 696 /// Suspend a community.
682 697 #[tracing::instrument(skip_all)]
683 - pub async fn suspend_community(
684 - pool: &PgPool,
698 + pub async fn suspend_community<'e, E: sqlx::PgExecutor<'e>>(
699 + executor: E,
685 700 community_id: Uuid,
686 701 reason: Option<&str>,
687 702 ) -> Result<(), sqlx::Error> {
@@ -690,35 +705,35 @@ pub async fn suspend_community(
690 705 community_id,
691 706 reason,
692 707 )
693 - .execute(pool)
708 + .execute(executor)
694 709 .await?;
695 710 Ok(())
696 711 }
697 712
698 713 /// Unsuspend a community.
699 714 #[tracing::instrument(skip_all)]
700 - pub async fn unsuspend_community(
701 - pool: &PgPool,
715 + pub async fn unsuspend_community<'e, E: sqlx::PgExecutor<'e>>(
716 + executor: E,
702 717 community_id: Uuid,
703 718 ) -> Result<(), sqlx::Error> {
704 719 sqlx::query!(
705 720 "UPDATE communities SET suspended_at = NULL, suspension_reason = NULL WHERE id = $1",
706 721 community_id,
707 722 )
708 - .execute(pool)
723 + .execute(executor)
709 724 .await?;
710 725 Ok(())
711 726 }
712 727
713 728 /// Set the community moderation state. See [`CommunityState`] for semantics.
714 729 #[tracing::instrument(skip_all)]
715 - pub async fn set_community_state(
716 - pool: &PgPool,
730 + pub async fn set_community_state<'e, E: sqlx::PgExecutor<'e>>(
731 + executor: E,
717 732 community_id: Uuid,
718 733 state: CommunityState,
719 734 ) -> Result<(), sqlx::Error> {
720 735 sqlx::query!("UPDATE communities SET state = $2 WHERE id = $1", community_id, state.as_str())
721 - .execute(pool)
736 + .execute(executor)
722 737 .await?;
723 738 Ok(())
724 739 }
@@ -744,13 +759,11 @@ pub struct CleanSlateResult {
744 759 /// mutation only enforces atomicity.
745 760 #[tracing::instrument(skip_all)]
746 761 pub async fn clean_slate_community(
747 - pool: &PgPool,
762 + conn: &mut sqlx::PgConnection,
748 763 community_id: Uuid,
749 764 actor_id: Uuid,
750 765 actor_display: &str,
751 766 ) -> Result<CleanSlateResult, sqlx::Error> {
752 - let mut tx = pool.begin().await?;
753 -
754 767 // Delete every thread whose category belongs to this community. Cascades
755 768 // reap posts, footnotes, endorsements, flags, read-positions, link
756 769 // previews, mentions, and tag joins.
@@ -763,7 +776,7 @@ pub async fn clean_slate_community(
763 776 SELECT COUNT(*) AS "count!" FROM d"#,
764 777 community_id,
765 778 )
766 - .fetch_one(&mut *tx)
779 + .fetch_one(&mut *conn)
767 780 .await?;
768 781
769 782 // Pick the first category by sort_order to host the reset notice. None
@@ -775,7 +788,7 @@ pub async fn clean_slate_community(
775 788 LIMIT 1",
776 789 community_id,
777 790 )
778 - .fetch_optional(&mut *tx)
791 + .fetch_optional(&mut *conn)
779 792 .await?;
780 793
781 794 let system_thread_id = if let Some(cat_id) = first_category {
@@ -799,7 +812,7 @@ pub async fn clean_slate_community(
799 812 actor_id,
800 813 title,
801 814 )
802 - .fetch_one(&mut *tx)
815 + .fetch_one(&mut *conn)
803 816 .await?;
804 817
805 818 sqlx::query!(
@@ -810,7 +823,7 @@ pub async fn clean_slate_community(
810 823 body_md,
811 824 body_html,
812 825 )
813 - .execute(&mut *tx)
826 + .execute(&mut *conn)
814 827 .await?;
815 828
816 829 Some(thread_id)
@@ -818,7 +831,6 @@ pub async fn clean_slate_community(
818 831 None
819 832 };
820 833
821 - tx.commit().await?;
822 834 Ok(CleanSlateResult { deleted_thread_count: deleted, system_thread_id })
823 835 }
824 836
@@ -836,8 +848,8 @@ fn html_escape(input: &str) -> String {
836 848
837 849 /// Suspend a user.
838 850 #[tracing::instrument(skip_all)]
839 - pub async fn suspend_user(
840 - pool: &PgPool,
851 + pub async fn suspend_user<'e, E: sqlx::PgExecutor<'e>>(
852 + executor: E,
841 853 user_id: Uuid,
842 854 reason: Option<&str>,
843 855 ) -> Result<(), sqlx::Error> {
@@ -846,22 +858,22 @@ pub async fn suspend_user(
846 858 user_id,
847 859 reason,
848 860 )
849 - .execute(pool)
861 + .execute(executor)
850 862 .await?;
851 863 Ok(())
852 864 }
853 865
854 866 /// Unsuspend a user.
855 867 #[tracing::instrument(skip_all)]
856 - pub async fn unsuspend_user(
857 - pool: &PgPool,
868 + pub async fn unsuspend_user<'e, E: sqlx::PgExecutor<'e>>(
869 + executor: E,
858 870 user_id: Uuid,
859 871 ) -> Result<(), sqlx::Error> {
860 872 sqlx::query!(
861 873 "UPDATE users SET suspended_at = NULL, suspension_reason = NULL WHERE mnw_account_id = $1",
862 874 user_id,
863 875 )
864 - .execute(pool)
876 + .execute(executor)
865 877 .await?;
866 878 Ok(())
867 879 }
@@ -1050,8 +1062,8 @@ pub async fn resolve_flag(
1050 1062
1051 1063 /// Resolve all unresolved flags for a given post.
1052 1064 #[tracing::instrument(skip_all)]
1053 - pub async fn resolve_all_flags_for_post(
1054 - pool: &PgPool,
1065 + pub async fn resolve_all_flags_for_post<'e, E: sqlx::PgExecutor<'e>>(
1066 + executor: E,
1055 1067 post_id: Uuid,
1056 1068 resolved_by: Uuid,
1057 1069 resolution: &str,
@@ -1063,7 +1075,7 @@ pub async fn resolve_all_flags_for_post(
1063 1075 resolved_by,
1064 1076 resolution,
1065 1077 )
1066 - .execute(pool)
1078 + .execute(executor)
1067 1079 .await?;
1068 1080 Ok(())
1069 1081 }
@@ -1198,8 +1210,8 @@ pub async fn insert_image(
1198 1210
1199 1211 /// Mark an image as removed by a moderator.
1200 1212 #[tracing::instrument(skip_all)]
1201 - pub async fn remove_image(
1202 - pool: &PgPool,
1213 + pub async fn remove_image<'e, E: sqlx::PgExecutor<'e>>(
1214 + executor: E,
1203 1215 image_id: Uuid,
1204 1216 removed_by: Uuid,
1205 1217 ) -> Result<(), sqlx::Error> {
@@ -1208,7 +1220,7 @@ pub async fn remove_image(
1208 1220 image_id,
1209 1221 removed_by,
1210 1222 )
1211 - .execute(pool)
1223 + .execute(executor)
1212 1224 .await?;
1213 1225 Ok(())
1214 1226 }
M multithreaded/crates/mt-db/src/queries.rs +7 -1
@@ -735,8 +735,14 @@ pub async fn get_post_body_markdown(
735 735 pool: &PgPool,
736 736 post_id: Uuid,
737 737 ) -> Result<Option<(Uuid, String)>, sqlx::Error> {
738 + // Only live posts can be quoted: a mod-removed post (or one in a soft-deleted
739 + // thread) must fail quote verification so its text cannot be re-surfaced into
740 + // a live post by quoting it back.
738 741 sqlx::query!(
739 - "SELECT author_id, body_markdown FROM posts WHERE id = $1",
742 + "SELECT p.author_id, p.body_markdown
743 + FROM posts p
744 + JOIN threads t ON t.id = p.thread_id
745 + WHERE p.id = $1 AND p.removed_at IS NULL AND t.deleted_at IS NULL",
740 746 post_id,
741 747 )
742 748 .fetch_optional(pool)
A multithreaded/migrations/032_mod_log_system_actor.sql +6
@@ -0,0 +1,6 @@
1 + -- Allow a NULL actor_id in the mod log to denote a system action (e.g. a
2 + -- flag-threshold auto-hide), mirroring the posts.removed_by IS NULL convention.
3 + -- Previously every entry required a user actor, which forced auto-hide to record
4 + -- the flagger who tripped the threshold as the "moderator" — a false attribution
5 + -- on an auditable, exportable ledger. Additive: existing rows are unaffected.
6 + ALTER TABLE mod_log ALTER COLUMN actor_id DROP NOT NULL;
M multithreaded/src/routes/admin.rs +33 -28
@@ -13,10 +13,10 @@ use crate::csrf;
13 13 use crate::templates::*;
14 14 use crate::AppState;
15 15
16 - use mt_core::types::ModAction;
16 + use mt_core::types::{ModAction, ModActor};
17 17
18 18 use super::{
19 - get_community, log_mod_action, parse_uuid, template_user, AdminSearchQuery,
19 + audit, begin_tx, commit_tx, get_community, parse_uuid, template_user, AdminSearchQuery,
20 20 CleanSlateForm, SuspendForm,
21 21 };
22 22
@@ -89,17 +89,18 @@ pub(super) async fn suspend_community_handler(
89 89 let community_id = parse_uuid(&id)?;
90 90 let reason = form.reason.as_deref().filter(|r| !r.trim().is_empty());
91 91
92 - mt_db::mutations::suspend_community(&state.db, community_id, reason)
92 + let mut tx = begin_tx(&state.db).await?;
93 + mt_db::mutations::suspend_community(&mut *tx, community_id, reason)
93 94 .await
94 95 .map_err(|e| {
95 96 tracing::error!(error = ?e, "db error suspending community");
96 97 (StatusCode::INTERNAL_SERVER_ERROR, "Internal server error").into_response()
97 98 })?;
98 -
99 - log_mod_action(
100 - &state.db, None, admin.user_id,
99 + audit(
100 + &mut tx, None, ModActor::User(admin.user_id),
101 101 ModAction::SuspendCommunity, None, Some(community_id), reason,
102 - ).await;
102 + ).await?;
103 + commit_tx(tx).await?;
103 104
104 105 Ok(Redirect::to("/_admin?toast=Community+suspended"))
105 106 }
@@ -112,17 +113,18 @@ pub(super) async fn unsuspend_community_handler(
112 113 ) -> Result<Redirect, Response> {
113 114 let community_id = parse_uuid(&id)?;
114 115
115 - mt_db::mutations::unsuspend_community(&state.db, community_id)
116 + let mut tx = begin_tx(&state.db).await?;
117 + mt_db::mutations::unsuspend_community(&mut *tx, community_id)
116 118 .await
117 119 .map_err(|e| {
118 120 tracing::error!(error = ?e, "db error unsuspending community");
119 121 (StatusCode::INTERNAL_SERVER_ERROR, "Internal server error").into_response()
120 122 })?;
121 -
122 - log_mod_action(
123 - &state.db, None, admin.user_id,
123 + audit(
124 + &mut tx, None, ModActor::User(admin.user_id),
124 125 ModAction::UnsuspendCommunity, None, Some(community_id), None,
125 - ).await;
126 + ).await?;
127 + commit_tx(tx).await?;
126 128
127 129 Ok(Redirect::to("/_admin?toast=Community+unsuspended"))
128 130 }
@@ -137,17 +139,18 @@ pub(super) async fn suspend_user_handler(
137 139 let user_id = parse_uuid(&id)?;
138 140 let reason = form.reason.as_deref().filter(|r| !r.trim().is_empty());
139 141
140 - mt_db::mutations::suspend_user(&state.db, user_id, reason)
142 + let mut tx = begin_tx(&state.db).await?;
143 + mt_db::mutations::suspend_user(&mut *tx, user_id, reason)
141 144 .await
142 145 .map_err(|e| {
143 146 tracing::error!(error = ?e, "db error suspending user");
144 147 (StatusCode::INTERNAL_SERVER_ERROR, "Internal server error").into_response()
145 148 })?;
146 -
147 - log_mod_action(
148 - &state.db, None, admin.user_id,
149 + audit(
150 + &mut tx, None, ModActor::User(admin.user_id),
149 151 ModAction::SuspendUser, Some(user_id), None, reason,
150 - ).await;
152 + ).await?;
153 + commit_tx(tx).await?;
151 154
152 155 Ok(Redirect::to("/_admin?toast=User+suspended"))
153 156 }
@@ -160,17 +163,18 @@ pub(super) async fn unsuspend_user_handler(
160 163 ) -> Result<Redirect, Response> {
161 164 let user_id = parse_uuid(&id)?;
162 165
163 - mt_db::mutations::unsuspend_user(&state.db, user_id)
166 + let mut tx = begin_tx(&state.db).await?;
167 + mt_db::mutations::unsuspend_user(&mut *tx, user_id)
164 168 .await
165 169 .map_err(|e| {
166 170 tracing::error!(error = ?e, "db error unsuspending user");
167 171 (StatusCode::INTERNAL_SERVER_ERROR, "Internal server error").into_response()
168 172 })?;
169 -
170 - log_mod_action(
171 - &state.db, None, admin.user_id,
173 + audit(
174 + &mut tx, None, ModActor::User(admin.user_id),
172 175 ModAction::UnsuspendUser, Some(user_id), None, None,
173 - ).await;
176 + ).await?;
177 + commit_tx(tx).await?;
174 178
175 179 Ok(Redirect::to("/_admin?toast=User+unsuspended"))
176 180 }
@@ -253,8 +257,9 @@ pub(super) async fn admin_community_clean_slate_handler(
253 257 .into_response());
254 258 }
255 259
260 + let mut tx = begin_tx(&state.db).await?;
256 261 let result = mt_db::mutations::clean_slate_community(
257 - &state.db,
262 + &mut tx,
258 263 community.id,
259 264 admin.user_id,
260 265 &admin.username,
@@ -264,17 +269,17 @@ pub(super) async fn admin_community_clean_slate_handler(
264 269 tracing::error!(error = ?e, "clean-slate failed");
265 270 (StatusCode::INTERNAL_SERVER_ERROR, "Internal server error").into_response()
266 271 })?;
267 -
268 - log_mod_action(
269 - &state.db,
272 + audit(
273 + &mut tx,
270 274 Some(community.id),
271 - admin.user_id,
275 + ModActor::User(admin.user_id),
272 276 ModAction::CleanSlateCommunity,
273 277 None,
274 278 result.system_thread_id,
275 279 Some(&format!("deleted {} threads", result.deleted_thread_count)),
276 280 )
277 - .await;
281 + .await?;
282 + commit_tx(tx).await?;
278 283
279 284 Ok(Redirect::to(&format!(
280 285 "/_admin/communities/{slug}?toast=Community+reset"
M multithreaded/src/routes/flagging.rs +40 -23
@@ -11,10 +11,10 @@ use serde::Deserialize;
11 11 use crate::auth::MaybeUser;
12 12 use crate::AppState;
13 13
14 - use mt_core::types::ModAction;
14 + use mt_core::types::{ModAction, ModActor};
15 15
16 16 use super::{
17 - check_community_access, field_error, get_community, log_mod_action, parse_uuid,
17 + audit, begin_tx, check_community_access, commit_tx, field_error, get_community, parse_uuid,
18 18 require_mod_or_owner,
19 19 };
20 20
@@ -81,21 +81,32 @@ pub(super) async fn flag_post_handler(
81 81 (StatusCode::INTERNAL_SERVER_ERROR, "Internal server error").into_response()
82 82 })?;
83 83
84 - // Auto-hide: atomically check flag count and remove post if threshold met
84 + // Auto-hide: atomically check flag count and remove post if threshold met.
85 + // The removal and its audit row commit on one transaction, so the auto-hide
86 + // can never land without its log entry — and the log records `System` as the
87 + // actor (NULL actor_id), not the member who happened to trip the threshold.
88 + // Best-effort: a failure here is traced but does not fail the flag submission
89 + // (the flag itself already committed; the post will re-trip on the next flag).
85 90 if let Some(threshold) = community.auto_hide_threshold
86 91 && threshold > 0
87 92 {
88 - match mt_db::mutations::auto_hide_if_threshold_met(
89 - &state.db, post_id, threshold,
90 - ).await {
91 - Ok(true) => {
92 - log_mod_action(
93 - &state.db, Some(community.id), user.user_id,
93 + let hide = async {
94 + let mut tx = state.db.begin().await?;
95 + let hidden =
96 + mt_db::mutations::auto_hide_if_threshold_met(&mut *tx, post_id, threshold).await?;
97 + if hidden {
98 + mt_db::mutations::insert_mod_log(
99 + &mut *tx, Some(community.id), ModActor::System,
94 100 ModAction::AutoHidePost, Some(post_data.author_id), Some(post_id), None,
95 - ).await;
101 + )
102 + .await?;
96 103 }
97 - Ok(false) => {} // threshold not met or already removed
98 - Err(e) => tracing::error!(error = ?e, "auto-hide: failed to check/remove post"),
104 + tx.commit().await?;
105 + Ok::<(), sqlx::Error>(())
106 + }
107 + .await;
108 + if let Err(e) = hide {
109 + tracing::error!(error = ?e, "auto-hide: failed to hide/log post");
99 110 }
100 111 }
101 112
@@ -175,33 +186,39 @@ pub(super) async fn remove_flagged_post_handler(
175 186 let (post_id, author_id, thread_id) = flag_row
176 187 .ok_or_else(|| (StatusCode::NOT_FOUND, "Not found").into_response())?;
177 188
178 - // Mod-remove the post (idempotent); if it is the OP, the whole thread is
179 - // soft-deleted in the same transaction.
180 - let removal = mt_db::mutations::mod_remove_post_cascade(&state.db, post_id, user.user_id)
189 + // Mod-remove the post, resolve its flags, and write the audit row(s) on one
190 + // transaction: the removal, the flag resolution, and the log all commit
191 + // together or not at all. If the post is the OP, the whole thread is
192 + // soft-deleted and that deletion is logged on the same tx.
193 + let mut tx = begin_tx(&state.db).await?;
194 + let removal = mt_db::mutations::mod_remove_post_cascade(&mut tx, post_id, user.user_id)
181 195 .await
182 196 .map_err(|e| {
183 197 tracing::error!(error = ?e, "db error removing flagged post");
184 198 (StatusCode::INTERNAL_SERVER_ERROR, "Internal server error").into_response()
185 199 })?;
186 200
187 - // Resolve all flags on this post
188 - mt_db::mutations::resolve_all_flags_for_post(&state.db, post_id, user.user_id, "removed")
201 + mt_db::mutations::resolve_all_flags_for_post(&mut *tx, post_id, user.user_id, "removed")
189 202 .await
190 203 .map_err(|e| {
191 204 tracing::error!(error = ?e, "db error resolving flags");
192 205 (StatusCode::INTERNAL_SERVER_ERROR, "Internal server error").into_response()
193 206 })?;
194 207
195 - log_mod_action(
196 - &state.db, Some(community.id), user.user_id,
208 + audit(
209 + &mut tx, Some(community.id), ModActor::User(user.user_id),
197 210 ModAction::RemovePostViaFlag, Some(author_id), Some(post_id), None,
198 - ).await;
211 + ).await?;
199 212
200 213 if removal.thread_removed {
201 - log_mod_action(
202 - &state.db, Some(community.id), user.user_id,
214 + audit(
215 + &mut tx, Some(community.id), ModActor::User(user.user_id),
203 216 ModAction::DeleteThread, Some(author_id), Some(thread_id), None,
204 - ).await;
217 + ).await?;
218 + }
219 + commit_tx(tx).await?;
220 +
221 + if removal.thread_removed {
205 222 return Ok(Redirect::to(&format!(
206 223 "/p/{slug}/moderation?toast=Thread+removed"
207 224 )));
M multithreaded/src/routes/forum/posts.rs +11 -9
@@ -17,12 +17,13 @@ use crate::AppState;
17 17 use mt_core::types::ModAction;
18 18
19 19 use super::super::{
20 - check_user_post_rate, check_write_access, check_write_state, get_community, get_thread,
21 - is_mod_or_owner, log_mod_action, parse_uuid, reject_embeds_for_free_user, render_markdown,
22 - render_markdown_plus, render_markdown_with_mentions, template_user, validate_body,
23 - validate_title, get_role, WriteScope,
20 + audit, begin_tx, check_user_post_rate, check_write_access, check_write_state, commit_tx,
21 + get_community, get_thread, is_mod_or_owner, parse_uuid, reject_embeds_for_free_user,
22 + render_markdown, render_markdown_plus, render_markdown_with_mentions, template_user,
23 + validate_body, validate_title, get_role, WriteScope,
24 24 CreateReplyForm, CreateThreadForm,
25 25 };
26 + use mt_core::types::ModActor;
26 27
27 28 // ============================================================================
28 29 // Quote verification
@@ -479,17 +480,18 @@ pub(in crate::routes) async fn delete_thread_handler(
479 480 }
480 481
481 482 let thread_id = parse_uuid(&thread_id_str)?;
482 - mt_db::mutations::soft_delete_thread(&state.db, thread_id)
483 + let mut tx = begin_tx(&state.db).await?;
484 + mt_db::mutations::soft_delete_thread(&mut *tx, thread_id)
483 485 .await
484 486 .map_err(|e| {
485 487 tracing::error!(error = ?e, "db error deleting thread");
486 488 StatusCode::INTERNAL_SERVER_ERROR.into_response()
487 489 })?;
488 -
489 - log_mod_action(
490 - &state.db, Some(thread_data.community_id), user.user_id,
490 + audit(
491 + &mut tx, Some(thread_data.community_id), ModActor::User(user.user_id),
491 492 ModAction::DeleteThread, Some(thread_data.author_id), Some(thread_id), None,
492 - ).await;
493 + ).await?;
494 + commit_tx(tx).await?;
493 495
494 496 Ok(Redirect::to(&format!(
495 497 "/p/{slug}/{category_slug}?toast=Thread+deleted"
M multithreaded/src/routes/helpers.rs +39 -9
@@ -7,7 +7,7 @@ use axum::{
7 7 use chrono::{DateTime, Duration, Utc};
8 8 use uuid::Uuid;
9 9
10 - use mt_core::types::{CommunityRole, CommunityState, ModAction};
10 + use mt_core::types::{CommunityRole, CommunityState, ModAction, ModActor};
11 11
12 12 use crate::auth;
13 13 use crate::templates::*;
@@ -155,23 +155,53 @@ pub(crate) async fn get_user_by_username(
155 155 .ok_or_else(|| (StatusCode::UNPROCESSABLE_ENTITY, "User not found.").into_response())
156 156 }
157 157
158 - /// Fire-and-forget mod log entry. Logs errors but never fails the request.
159 - pub(crate) async fn log_mod_action(
158 + /// Begin a transaction, mapping a DB error to a branded 500.
159 + #[allow(clippy::result_large_err)]
160 + pub(crate) async fn begin_tx(
160 161 db: &sqlx::PgPool,
162 + ) -> Result<sqlx::Transaction<'static, sqlx::Postgres>, Response> {
163 + db.begin().await.map_err(|e| {
164 + tracing::error!(error = ?e, "db error opening transaction");
165 + crate::error_page::internal_error()
166 + })
167 + }
168 +
169 + /// Commit a transaction, mapping a DB error to a branded 500.
170 + #[allow(clippy::result_large_err)]
171 + pub(crate) async fn commit_tx(
172 + tx: sqlx::Transaction<'_, sqlx::Postgres>,
173 + ) -> Result<(), Response> {
174 + tx.commit().await.map_err(|e| {
175 + tracing::error!(error = ?e, "db error committing transaction");
176 + crate::error_page::internal_error()
177 + })
178 + }
179 +
180 + /// Write a mod-log entry on the caller's transaction.
181 + ///
182 + /// The audit row is written on the *same* `tx` as the mutation it records, so
183 + /// the moderation action and its log either both commit or both roll back — an
184 + /// auditable action can never land without its correctly-attributed audit row
185 + /// (closes the fire-and-forget gap). A `System` actor (e.g. flag-threshold
186 + /// auto-hide) is recorded as a NULL `actor_id`, never the user who tripped it.
187 + #[allow(clippy::result_large_err)]
188 + pub(crate) async fn audit(
189 + tx: &mut sqlx::PgConnection,
161 190 community_id: Option<Uuid>,
162 - actor_id: Uuid,
191 + actor: ModActor,
163 192 action: ModAction,
164 193 target_user: Option<Uuid>,
165 194 target_id: Option<Uuid>,
166 195 reason: Option<&str>,
167 - ) {
168 - if let Err(e) = mt_db::mutations::insert_mod_log(
169 - db, community_id, actor_id, action, target_user, target_id, reason,
196 + ) -> Result<(), Response> {
197 + mt_db::mutations::insert_mod_log(
198 + &mut *tx, community_id, actor, action, target_user, target_id, reason,
170 199 )
171 200 .await
172 - {
201 + .map_err(|e| {
173 202 tracing::error!(error = %e, "failed to insert mod log");
174 - }
203 + crate::error_page::internal_error()
204 + })
175 205 }
176 206
177 207 /// Convert a session user to a template session user.
M multithreaded/src/routes/moderation.rs +63 -45
@@ -16,10 +16,11 @@ use crate::AppState;
16 16 use mt_core::types::{BanType, ModAction};
17 17
18 18 use super::{
19 - field_error, get_role, get_thread, get_user_by_username, is_mod_or_owner, is_owner,
20 - log_mod_action, parse_duration, parse_uuid, require_mod_or_owner, template_user, BanForm,
21 - PageQuery, UnbanForm,
19 + audit, begin_tx, commit_tx, field_error, get_role, get_thread, get_user_by_username,
20 + is_mod_or_owner, is_owner, parse_duration, parse_uuid, require_mod_or_owner, template_user,
21 + BanForm, PageQuery, UnbanForm,
22 22 };
23 + use mt_core::types::ModActor;
23 24
24 25 #[tracing::instrument(skip_all)]
25 26 pub(super) async fn pin_thread_handler(
@@ -38,18 +39,20 @@ pub(super) async fn pin_thread_handler(
38 39 }
39 40
40 41 let new_pinned = !thread_data.pinned;
41 - mt_db::mutations::set_thread_pinned(&state.db, thread_data.id, new_pinned)
42 + let action = if new_pinned { ModAction::PinThread } else { ModAction::UnpinThread };
43 +
44 + let mut tx = begin_tx(&state.db).await?;
45 + mt_db::mutations::set_thread_pinned(&mut *tx, thread_data.id, new_pinned)
42 46 .await
43 47 .map_err(|e| {
44 48 tracing::error!(error = ?e, "db error toggling pin");
45 49 StatusCode::INTERNAL_SERVER_ERROR.into_response()
46 50 })?;
47 -
48 - let action = if new_pinned { ModAction::PinThread } else { ModAction::UnpinThread };
49 - log_mod_action(
50 - &state.db, Some(thread_data.community_id), user.user_id,
51 + audit(
52 + &mut tx, Some(thread_data.community_id), ModActor::User(user.user_id),
51 53 action, None, Some(thread_data.id), None,
52 - ).await;
54 + ).await?;
55 + commit_tx(tx).await?;
53 56
54 57 let toast = if new_pinned { "Thread+pinned" } else { "Thread+unpinned" };
55 58 Ok(Redirect::to(&format!(
@@ -74,18 +77,20 @@ pub(super) async fn lock_thread_handler(
74 77 }
75 78
76 79 let new_locked = !thread_data.locked;
77 - mt_db::mutations::set_thread_locked(&state.db, thread_data.id, new_locked)
80 + let action = if new_locked { ModAction::LockThread } else { ModAction::UnlockThread };
81 +
82 + let mut tx = begin_tx(&state.db).await?;
83 + mt_db::mutations::set_thread_locked(&mut *tx, thread_data.id, new_locked)
78 84 .await
79 85 .map_err(|e| {
80 86 tracing::error!(error = ?e, "db error toggling lock");
81 87 StatusCode::INTERNAL_SERVER_ERROR.into_response()
82 88 })?;
83 -
84 - let action = if new_locked { ModAction::LockThread } else { ModAction::UnlockThread };
85 - log_mod_action(
86 - &state.db, Some(thread_data.community_id), user.user_id,
89 + audit(
90 + &mut tx, Some(thread_data.community_id), ModActor::User(user.user_id),
87 91 action, None, Some(thread_data.id), None,
88 - ).await;
92 + ).await?;
93 + commit_tx(tx).await?;
89 94
90 95 let toast = if new_locked { "Thread+locked" } else { "Thread+unlocked" };
91 96 Ok(Redirect::to(&format!(
@@ -122,26 +127,35 @@ pub(super) async fn mod_remove_post_handler(
122 127 return Err(StatusCode::FORBIDDEN.into_response());
123 128 }
124 129
125 - let removal = mt_db::mutations::mod_remove_post_cascade(&state.db, post_id, user.user_id)
130 + // Resolve the thread id up front so the cascade-delete log can be written on
131 + // the same transaction as the removal.
132 + let thread_id = parse_uuid(&thread_id_str)?;
133 +
134 + let mut tx = begin_tx(&state.db).await?;
135 + let removal = mt_db::mutations::mod_remove_post_cascade(&mut tx, post_id, user.user_id)
126 136 .await
127 137 .map_err(|e| {
128 138 tracing::error!(error = ?e, "db error removing post");
129 139 StatusCode::INTERNAL_SERVER_ERROR.into_response()
130 140 })?;
131 141
132 - log_mod_action(
133 - &state.db, Some(post_data.community_id), user.user_id,
142 + audit(
143 + &mut tx, Some(post_data.community_id), ModActor::User(user.user_id),
134 144 ModAction::RemovePost, Some(post_data.author_id), Some(post_id), None,
135 - ).await;
145 + ).await?;
136 146
137 - // Removing the opening post cascades to soft-deleting the whole thread; the
138 - // thread page now 404s, so send the mod back to the category listing.
147 + // Removing the opening post cascades to soft-deleting the whole thread; log
148 + // the thread deletion on the same tx so the two records commit together.
139 149 if removal.thread_removed {
140 - let thread_id = parse_uuid(&thread_id_str)?;
141 - log_mod_action(
142 - &state.db, Some(post_data.community_id), user.user_id,
150 + audit(
151 + &mut tx, Some(post_data.community_id), ModActor::User(user.user_id),
143 152 ModAction::DeleteThread, Some(post_data.author_id), Some(thread_id), None,
144 - ).await;
153 + ).await?;
154 + }
155 + commit_tx(tx).await?;
156 +
157 + // The thread page now 404s, so send the mod back to the category listing.
158 + if removal.thread_removed {
145 159 return Ok(Redirect::to(&format!(
146 160 "/p/{slug}/{category_slug}?toast=Thread+removed"
147 161 )));
@@ -278,8 +292,9 @@ pub(super) async fn ban_user_handler(
278 292 return Err(field_error("reason", "Reason too long (max 1024 bytes)."));
279 293 }
280 294
295 + let mut tx = begin_tx(&state.db).await?;
281 296 mt_db::mutations::create_community_ban(
282 - &state.db, community.id, target_id, user.user_id,
297 + &mut *tx, community.id, target_id, user.user_id,
283 298 BanType::Ban, reason, expires_at,
284 299 )
285 300 .await
@@ -287,11 +302,11 @@ pub(super) async fn ban_user_handler(
287 302 tracing::error!(error = ?e, "db error creating ban");
288 303 StatusCode::INTERNAL_SERVER_ERROR.into_response()
289 304 })?;
290 -
291 - log_mod_action(
292 - &state.db, Some(community.id), user.user_id,
305 + audit(
306 + &mut tx, Some(community.id), ModActor::User(user.user_id),
293 307 ModAction::Ban, Some(target_id), None, reason,
294 - ).await;
308 + ).await?;
309 + commit_tx(tx).await?;
295 310
296 311 Ok(Redirect::to(&format!(
297 312 "/p/{slug}/moderation?toast=User+banned"
@@ -312,17 +327,18 @@ pub(super) async fn unban_user_handler(
312 327
313 328 let target_id = get_user_by_username(&state.db, form.username.trim()).await?;
314 329
315 - mt_db::mutations::remove_community_ban(&state.db, community.id, target_id, BanType::Ban)
330 + let mut tx = begin_tx(&state.db).await?;
331 + mt_db::mutations::remove_community_ban(&mut *tx, community.id, target_id, BanType::Ban)
316 332 .await
317 333 .map_err(|e| {
318 334 tracing::error!(error = ?e, "db error removing ban");
319 335 StatusCode::INTERNAL_SERVER_ERROR.into_response()
320 336 })?;
321 -
322 - log_mod_action(
323 - &state.db, Some(community.id), user.user_id,
337 + audit(
338 + &mut tx, Some(community.id), ModActor::User(user.user_id),
324 339 ModAction::Unban, Some(target_id), None, None,
325 - ).await;
340 + ).await?;
341 + commit_tx(tx).await?;
326 342
327 343 Ok(Redirect::to(&format!(
328 344 "/p/{slug}/moderation?toast=User+unbanned"
@@ -363,8 +379,9 @@ pub(super) async fn mute_user_handler(
363 379 return Err(field_error("reason", "Reason too long (max 1024 bytes)."));
364 380 }
365 381
382 + let mut tx = begin_tx(&state.db).await?;
366 383 mt_db::mutations::create_community_ban(
367 - &state.db, community.id, target_id, user.user_id,
384 + &mut *tx, community.id, target_id, user.user_id,
368 385 BanType::Mute, reason, expires_at,
369 386 )
370 387 .await
@@ -372,11 +389,11 @@ pub(super) async fn mute_user_handler(
372 389 tracing::error!(error = ?e, "db error creating mute");
373 390 StatusCode::INTERNAL_SERVER_ERROR.into_response()
374 391 })?;
375 -
376 - log_mod_action(
377 - &state.db, Some(community.id), user.user_id,
392 + audit(
393 + &mut tx, Some(community.id), ModActor::User(user.user_id),
378 394 ModAction::Mute, Some(target_id), None, reason,
379 - ).await;
395 + ).await?;
396 + commit_tx(tx).await?;
380 397
381 398 Ok(Redirect::to(&format!(
382 399 "/p/{slug}/moderation?toast=User+muted"
@@ -397,17 +414,18 @@ pub(super) async fn unmute_user_handler(
397 414
398 415 let target_id = get_user_by_username(&state.db, form.username.trim()).await?;
399 416
400 - mt_db::mutations::remove_community_ban(&state.db, community.id, target_id, BanType::Mute)
417 + let mut tx = begin_tx(&state.db).await?;
418 + mt_db::mutations::remove_community_ban(&mut *tx, community.id, target_id, BanType::Mute)
401 419 .await
402 420 .map_err(|e| {
403 421 tracing::error!(error = ?e, "db error removing mute");
404 422 StatusCode::INTERNAL_SERVER_ERROR.into_response()
405 423 })?;
406 -
407 - log_mod_action(
408 - &state.db, Some(community.id), user.user_id,
424 + audit(
425 + &mut tx, Some(community.id), ModActor::User(user.user_id),
409 426 ModAction::Unmute, Some(target_id), None, None,
410 - ).await;
427 + ).await?;
428 + commit_tx(tx).await?;
411 429
412 430 Ok(Redirect::to(&format!(
413 431 "/p/{slug}/moderation?toast=User+unmuted"
M multithreaded/src/routes/settings.rs +29 -23
@@ -13,10 +13,10 @@ use crate::csrf;
13 13 use crate::templates::*;
14 14 use crate::AppState;
15 15
16 - use mt_core::types::{CommunityState, ModAction};
16 + use mt_core::types::{CommunityState, ModAction, ModActor};
17 17
18 18 use super::{
19 - log_mod_action, parse_uuid, require_mod_or_superadmin, require_owner, template_user,
19 + audit, begin_tx, commit_tx, parse_uuid, require_mod_or_superadmin, require_owner, template_user,
20 20 validate_title, CreateCategoryForm, CreateTagForm, DeleteTagForm, EditCategoryFormData,
21 21 MoveCategoryForm, SetCommunityStateForm, UpdateCommunityForm,
22 22 };
@@ -116,17 +116,18 @@ pub(super) async fn update_community_handler(
116 116 .and_then(|s| s.trim().parse::<i32>().ok())
117 117 .filter(|&n| n > 0);
118 118
119 - mt_db::mutations::update_community(&state.db, community.id, name, desc_opt, threshold)
119 + let mut tx = begin_tx(&state.db).await?;
120 + mt_db::mutations::update_community(&mut *tx, community.id, name, desc_opt, threshold)
120 121 .await
121 122 .map_err(|e| {
122 123 tracing::error!(error = ?e, "db error updating community");
123 124 StatusCode::INTERNAL_SERVER_ERROR.into_response()
124 125 })?;
125 -
126 - log_mod_action(
127 - &state.db, Some(community.id), user.user_id,
126 + audit(
127 + &mut tx, Some(community.id), ModActor::User(user.user_id),
128 128 ModAction::EditSettings, None, None, None,
129 - ).await;
129 + ).await?;
130 + commit_tx(tx).await?;
130 131
131 132 Ok(Redirect::to(&format!(
132 133 "/p/{slug}/settings?toast=Settings+saved"
@@ -175,17 +176,18 @@ pub(super) async fn create_category_handler(
175 176 })?;
176 177 let next_order = existing.iter().map(|c| c.sort_order).max().unwrap_or(0) + 1;
177 178
178 - mt_db::mutations::create_category(&state.db, community.id, name, &cat_slug, desc_opt, next_order)
179 + let mut tx = begin_tx(&state.db).await?;
180 + mt_db::mutations::create_category(&mut *tx, community.id, name, &cat_slug, desc_opt, next_order)
179 181 .await
180 182 .map_err(|e| {
181 183 tracing::error!(error = ?e, "db error creating category");
182 184 StatusCode::INTERNAL_SERVER_ERROR.into_response()
183 185 })?;
184 -
185 - log_mod_action(
186 - &state.db, Some(community.id), user.user_id,
186 + audit(
187 + &mut tx, Some(community.id), ModActor::User(user.user_id),
187 188 ModAction::CreateCategory, None, None, Some(name),
188 - ).await;
189 + ).await?;
190 + commit_tx(tx).await?;
189 191
190 192 Ok(Redirect::to(&format!(
191 193 "/p/{slug}/settings?toast=Category+created"
@@ -253,20 +255,23 @@ pub(super) async fn edit_category_handler(
253 255 }
254 256 let desc_opt = if description.is_empty() { None } else { Some(description) };
255 257
256 - let updated = mt_db::mutations::update_category(&state.db, cat_id, community.id, name, desc_opt)
258 + let mut tx = begin_tx(&state.db).await?;
259 + let updated = mt_db::mutations::update_category(&mut *tx, cat_id, community.id, name, desc_opt)
257 260 .await
258 261 .map_err(|e| {
259 262 tracing::error!(error = ?e, "db error updating category");
260 263 StatusCode::INTERNAL_SERVER_ERROR.into_response()
261 264 })?;
262 265 if !updated {
266 + // Nothing changed (no such category in this community); the tx rolls back
267 + // on drop, so no empty audit row is written.
263 268 return Err(StatusCode::NOT_FOUND.into_response());
264 269 }
265 -
266 - log_mod_action(
267 - &state.db, Some(community.id), user.user_id,
270 + audit(
271 + &mut tx, Some(community.id), ModActor::User(user.user_id),
268 272 ModAction::EditCategory, None, Some(cat_id), None,
269 - ).await;
273 + ).await?;
274 + commit_tx(tx).await?;
270 275
271 276 Ok(Redirect::to(&format!(
272 277 "/p/{slug}/settings?toast=Category+updated"
@@ -424,23 +429,24 @@ pub(super) async fn set_community_state_handler(
424 429 )));
425 430 }
426 431
427 - mt_db::mutations::set_community_state(&state.db, community.id, new_state)
432 + let mut tx = begin_tx(&state.db).await?;
433 + mt_db::mutations::set_community_state(&mut *tx, community.id, new_state)
428 434 .await
429 435 .map_err(|e| {
430 436 tracing::error!(error = ?e, "db error setting community state");
431 437 StatusCode::INTERNAL_SERVER_ERROR.into_response()
432 438 })?;
433 -
434 - log_mod_action(
435 - &state.db,
439 + audit(
440 + &mut tx,
436 441 Some(community.id),
437 - user.user_id,
442 + ModActor::User(user.user_id),
438 443 ModAction::ChangeCommunityState,
439 444 None,
440 445 None,
441 446 Some(new_state.as_str()),
442 447 )
443 - .await;
448 + .await?;
449 + commit_tx(tx).await?;
444 450
445 451 Ok(Redirect::to(&format!(
446 452 "/p/{slug}/settings?toast=Community+state+updated"
M multithreaded/src/routes/uploads.rs +20 -17
@@ -6,7 +6,7 @@ use axum::{
6 6 http::{StatusCode, header},
7 7 response::{IntoResponse, Response},
8 8 };
9 - use mt_core::types::ModAction;
9 + use mt_core::types::{ModAction, ModActor};
10 10
11 11 use crate::auth::MaybeUser;
12 12 use crate::storage;
@@ -226,17 +226,32 @@ pub(super) async fn remove_image_handler(
226 226 return Err(StatusCode::NOT_FOUND.into_response());
227 227 }
228 228
229 - mt_db::mutations::remove_image(&state.db, image_id, user.user_id)
229 + // Mark the image removed and write the audit row on one transaction, so the
230 + // removal can never land without its log entry.
231 + let mut tx = super::begin_tx(&state.db).await?;
232 + mt_db::mutations::remove_image(&mut *tx, image_id, user.user_id)
230 233 .await
231 234 .map_err(|e| {
232 235 tracing::error!(error = ?e, "db error removing image");
233 236 StatusCode::INTERNAL_SERVER_ERROR.into_response()
234 237 })?;
238 + super::audit(
239 + &mut tx,
240 + Some(community.id),
241 + ModActor::User(user.user_id),
242 + ModAction::RemoveImage,
243 + None,
244 + Some(image_id),
245 + None,
246 + )
247 + .await?;
248 + super::commit_tx(tx).await?;
235 249
236 250 // Delete the backing S3 object so removed images don't accumulate in the
237 - // bucket forever. Best-effort: the DB row is already marked removed (serve
238 - // returns 410). On success, record s3_purged_at so the reconcile sweep skips
239 - // it; on failure, leave it unmarked so the sweep retries it later.
251 + // bucket forever. Best-effort, after the removal+log have committed: the DB
252 + // row is already marked removed (serve returns 410). On success, record
253 + // s3_purged_at so the reconcile sweep skips it; on failure, leave it unmarked
254 + // so the sweep retries it later.
240 255 if let Some(s3) = state.s3.as_ref() {
241 256 match s3.delete(&image.s3_key).await {
242 257 Ok(()) => {
@@ -250,17 +265,5 @@ pub(super) async fn remove_image_handler(
250 265 }
251 266 }
252 267
253 - // Log mod action
254 - super::log_mod_action(
255 - &state.db,
256 - Some(community.id),
257 - user.user_id,
258 - ModAction::RemoveImage,
259 - None,
260 - Some(image_id),
261 - None,
262 - )
263 - .await;
264 -
265 268 Ok(StatusCode::OK)
266 269 }
M multithreaded/tests/workflows/flagging.rs +63
@@ -62,6 +62,52 @@ async fn auto_hide_threshold_removes_post_at_threshold() {
62 62 assert!(removed, "Post should be auto-hidden after 2 flags (threshold=2)");
63 63 }
64 64
65 + /// Regression (ultra-fuzz Run #5 S3): a flag-threshold auto-hide is a *system*
66 + /// action — its mod-log row must carry a NULL actor_id, never the member who
67 + /// happened to trip the threshold. And it must actually exist (the log is bound
68 + /// to the auto-hide's transaction, not fire-and-forget).
69 + #[tokio::test]
70 + async fn auto_hide_logs_system_actor_not_flagger() {
71 + let mut h = TestHarness::new().await;
72 + let author_id = h.login_as("sysauthor").await;
73 + let comm_id = h.create_community("Sys", "sys").await;
74 + let cat_id = h.create_category(comm_id, "General", "general").await;
75 + h.add_membership(author_id, comm_id, "member").await;
76 +
77 + sqlx::query("UPDATE communities SET auto_hide_threshold = 1 WHERE id = $1")
78 + .bind(comm_id)
79 + .execute(&h.db)
80 + .await
81 + .unwrap();
82 +
83 + let thread_id = h
84 + .create_thread_with_post(cat_id, author_id, "Sys Hide", "Content")
85 + .await;
86 + let post_id = mt_db::queries::list_posts_in_thread(&h.db, thread_id)
87 + .await
88 + .unwrap()[0]
89 + .id;
90 +
91 + // One flag from a member trips the threshold of 1.
92 + let flagger = h.login_as("tripper").await;
93 + h.add_membership(flagger, comm_id, "member").await;
94 + let thread_url = format!("/p/sys/general/{}", thread_id);
95 + h.client.get(&thread_url).await;
96 + let flag_url = format!("/p/sys/general/{}/posts/{}/flag", thread_id, post_id);
97 + h.client.post_form(&flag_url, "reason=spam").await;
98 +
99 + // The audit row exists (atomic with the removal) and is attributed to System.
100 + let actor: Option<uuid::Uuid> = sqlx::query_scalar(
101 + "SELECT actor_id FROM mod_log WHERE target_id = $1 AND action = 'auto_hide_post'",
102 + )
103 + .bind(post_id)
104 + .fetch_one(&h.db)
105 + .await
106 + .expect("auto-hide must write exactly one mod_log row");
107 + assert_eq!(actor, None, "auto-hide must be logged as System (NULL actor_id), not the flagger");
108 + assert_ne!(actor, Some(flagger), "the flagger must never be recorded as the auto-hide actor");
109 + }
110 +
65 111 #[tokio::test]
66 112 async fn auto_hide_disabled_when_threshold_null() {
67 113 let mut h = TestHarness::new().await;
@@ -442,4 +488,21 @@ async fn remove_flagged_op_cascades_to_thread_delete() {
442 488
443 489 let resp = h.client.get(&format!("/p/test/general/{}", thread_id)).await;
444 490 assert_eq!(resp.status.as_u16(), 404, "Cascaded thread should 404");
491 +
492 + // Regression (ultra-fuzz Run #5 M-DI1): the removal, the flag resolution, and
493 + // BOTH audit rows commit on one transaction. Assert both log rows landed and
494 + // are attributed to the acting moderator (User), not dropped fire-and-forget.
495 + let log_rows: Vec<(String, Option<uuid::Uuid>)> = sqlx::query_as(
496 + "SELECT action, actor_id FROM mod_log
497 + WHERE community_id = $1 AND action IN ('remove_post_via_flag', 'delete_thread')
498 + ORDER BY action",
499 + )
500 + .bind(comm_id)
501 + .fetch_all(&h.db)
502 + .await
503 + .unwrap();
504 + assert_eq!(log_rows.len(), 2, "OP removal via flag must log both remove_post_via_flag and delete_thread");
505 + for (action, actor) in &log_rows {
506 + assert_eq!(*actor, Some(mod_id), "{action} must be attributed to the acting moderator");
507 + }
445 508 }