fix: moderation issues bugs

apps/labrinth/src/models/v3/thread_issues.rs

@@ -98,6 +98,7 @@ pub enum ThreadIssueTarget {
         links: HashMap<String, TextTarget>,
     },
     AddGalleryImages {
+        #[serde(default)]
         original_image_ids: Vec<GalleryImageId>,
         auto_resolve: bool,
     },

apps/labrinth/src/routes/internal/delphi/mod.rs

@@ -221,6 +221,12 @@ async fn ingest_report_deserialized(
         return Ok(());
     }
 
+    tech_review_queue::lock_projects(
+        &[DBProjectId::from(report.project_id)],
+        &mut transaction,
+    )
+    .await?;
+
     AdvisoryLock::DelphiFile(file_id)
         .acquire(&mut transaction)
         .await
@@ -305,18 +311,22 @@ async fn ingest_report_deserialized(
     .await
     .wrap_internal_err("failed to apply delphi rules to new issue details")?;
 
-    tech_review_queue::sync_projects(
+    let exited_projects = tech_review_queue::sync_projects(
         &[DBProjectId::from(report.project_id)],
         tech_review_queue::TechReviewRemovalReason::ScanCompleted,
         &mut transaction,
     )
     .await
     .wrap_api_err("synchronizing project technical-review state")?;
 
-    transaction
-        .commit()
-        .await
-        .wrap_internal_err("failed to commit Delphi ingest transaction")?;
+    let prepared =
+        crate::routes::v3::projects::mutation::prepare_technical_review_exits(
+            &exited_projects,
+            &mut transaction,
+            &redis,
+        )
+        .await?;
+    prepared.commit(transaction, &redis).await?;
 
     if !clean {
         report.send_to_slack(&pool, &redis).await.ok();

apps/labrinth/src/routes/internal/delphi/tech_review_queue.rs

@@ -23,6 +23,26 @@ pub enum TechReviewRemovalReason {
     FileDeleted,
 }
 
+pub(crate) async fn lock_projects(
+    project_ids: &[DBProjectId],
+    txn: &mut PgTransaction<'_>,
+) -> Result<(), ApiError> {
+    sqlx::query!(
+        r#"
+		SELECT id
+		FROM mods
+		WHERE id = ANY($1::bigint[])
+		ORDER BY id
+		FOR NO KEY UPDATE
+		"#,
+        &project_ids.iter().map(|id| id.0).collect::<Vec<_>>(),
+    )
+    .fetch_all(&mut *txn)
+    .await
+    .wrap_internal_err("locking projects for technical review transitions")?;
+    Ok(())
+}
+
 pub async fn add_projects(
     project_ids: &[DBProjectId],
     txn: &mut PgTransaction<'_>,
@@ -32,6 +52,8 @@ pub async fn add_projects(
         return Ok(());
     }
 
+    lock_projects(&project_ids, txn).await?;
+
     let rows = sqlx::query!(
         r#"
         WITH inserted AS (
@@ -76,12 +98,14 @@ pub async fn remove_projects(
     project_ids: &[DBProjectId],
     reason: TechReviewRemovalReason,
     txn: &mut PgTransaction<'_>,
-) -> Result<(), ApiError> {
+) -> Result<Vec<DBProjectId>, ApiError> {
     let project_ids = project_ids.iter().copied().unique().collect::<Vec<_>>();
     if project_ids.is_empty() {
-        return Ok(());
+        return Ok(Vec::new());
     }
 
+    lock_projects(&project_ids, txn).await?;
+
     let rows = sqlx::query!(
         r#"
         WITH removed AS (
@@ -108,6 +132,8 @@ pub async fn remove_projects(
         "failed to remove projects from technical review queue",
     )?;
 
+    let removed_project_ids = rows.iter().map(|row| row.project_id).collect();
+
     let body = match reason {
         TechReviewRemovalReason::RulesChanged
         | TechReviewRemovalReason::ScanCompleted => {
@@ -130,19 +156,21 @@ pub async fn remove_projects(
         .wrap_internal_err("failed to add exiting technical review message")?;
     }
 
-    Ok(())
+    Ok(removed_project_ids)
 }
 
 pub async fn sync_projects(
     project_ids: &[DBProjectId],
     removal_reason: TechReviewRemovalReason,
     txn: &mut PgTransaction<'_>,
-) -> Result<(), ApiError> {
+) -> Result<Vec<DBProjectId>, ApiError> {
     let project_ids = project_ids.iter().copied().unique().collect::<Vec<_>>();
     if project_ids.is_empty() {
-        return Ok(());
+        return Ok(Vec::new());
     }
 
+    lock_projects(&project_ids, txn).await?;
+
     let rows = sqlx::query!(
         r#"
         SELECT

apps/labrinth/src/routes/internal/moderation/tech_review.rs

@@ -1262,7 +1262,7 @@ pub async fn submit_report(
         .wrap_internal_err("failed to add tech review message")?;
     }
 
-    let prepared = if verdict == DelphiVerdict::Unsafe {
+    let mut prepared = if verdict == DelphiVerdict::Unsafe {
         Some(
             crate::routes::v3::projects::mutation::prepare_mutations(
                 &[project_id],
@@ -1276,7 +1276,7 @@ pub async fn submit_report(
         None
     };
 
-    sqlx::query!(
+    let removed = sqlx::query!(
         r#"
         DELETE FROM delphi_tech_review_queue
         WHERE project_id = $1
@@ -1289,12 +1289,20 @@ pub async fn submit_report(
         "failed to remove project from technical review queue",
     )?;
 
+    if prepared.is_none() && removed.rows_affected() > 0 {
+        prepared = Some(crate::routes::v3::projects::mutation::prepare_technical_review_exits(
+			&[project_id], &mut txn, &redis,
+		).await?);
+    }
+
     if let Some(prepared) = prepared {
         prepared.commit(txn, &redis).await?;
-        search_state
-            .queue
-            .push_project_change(project_id.into())
-            .await;
+        if verdict == DelphiVerdict::Unsafe {
+            search_state
+                .queue
+                .push_project_change(project_id.into())
+                .await;
+        }
     } else {
         txn.commit()
             .await
@@ -1463,17 +1471,22 @@ pub async fn update_issue_details(
         .map(|row| row.project_id)
         .collect::<Vec<_>>();
 
-    tech_review_queue::sync_projects(
+    let exited_projects = tech_review_queue::sync_projects(
         &affected_project_ids,
         tech_review_queue::TechReviewRemovalReason::RulesChanged,
         &mut txn,
     )
     .await
     .wrap_api_err("executing `tech_review_queue::sync_projects`")?;
 
-    txn.commit()
-        .await
-        .wrap_internal_err("failed to commit transaction")?;
+    let prepared =
+        crate::routes::v3::projects::mutation::prepare_technical_review_exits(
+            &exited_projects,
+            &mut txn,
+            &redis,
+        )
+        .await?;
+    prepared.commit(txn, &redis).await?;
 
     Ok(())
 }
@@ -1595,7 +1608,7 @@ pub async fn update_global_issue_details(
         "failed to fetch projects affected by global detail updates",
     )?;
 
-    tech_review_queue::sync_projects(
+    let exited_projects = tech_review_queue::sync_projects(
         &affected_projects
             .into_iter()
             .map(|row| row.project_id)
@@ -1606,9 +1619,14 @@ pub async fn update_global_issue_details(
     .await
     .wrap_api_err("executing `tech_review_queue::sync_projects`")?;
 
-    txn.commit()
-        .await
-        .wrap_internal_err("failed to commit transaction")?;
+    let prepared =
+        crate::routes::v3::projects::mutation::prepare_technical_review_exits(
+            &exited_projects,
+            &mut txn,
+            &redis,
+        )
+        .await?;
+    prepared.commit(txn, &redis).await?;
 
     Ok(())
 }

apps/labrinth/src/routes/internal/moderation/tech_review/rules_scan.rs

@@ -401,7 +401,7 @@ pub async fn scan_rules(
     let _ = sender.send(Bytes::from_static(b": connected\n\n"));
 
     actix_web::rt::spawn(async move {
-        match run_scan(transaction, &sender).await {
+        match run_scan(transaction, &sender, &redis).await {
             Ok(summary) => {
                 let event = RuleScanEvent {
                     phase: "complete",
@@ -461,7 +461,22 @@ pub async fn scan_rules(
 async fn run_scan(
     mut transaction: PgTransaction<'static>,
     sender: &mpsc::UnboundedSender<Bytes>,
+    redis: &RedisPool,
 ) -> Result<ScanSummary> {
+    let project_ids = sqlx::query!(
+        r#"
+		SELECT DISTINCT project_id AS "project_id!: DBProjectId"
+		FROM delphi_issue_details_with_statuses
+		"#,
+    )
+    .fetch_all(&mut transaction)
+    .await
+    .wrap_err("fetching projects affected by delphi rule scan")?
+    .into_iter()
+    .map(|row| row.project_id)
+    .collect::<Vec<_>>();
+    tech_review_queue::lock_projects(&project_ids, &mut transaction).await?;
+
     sqlx::query!("LOCK TABLE delphi_rules IN SHARE MODE")
         .execute(&mut transaction)
         .await
@@ -755,7 +770,7 @@ async fn run_scan(
         &mut transaction,
     )
     .await?;
-    tech_review_queue::remove_projects(
+    let exited_projects = tech_review_queue::remove_projects(
         &affected_projects
             .iter()
             .filter(|project| !project.new_needs_review)
@@ -778,10 +793,14 @@ async fn run_scan(
         .await
         .wrap_err("failed to delete retired delphi rules")?;
 
-    transaction
-        .commit()
-        .await
-        .wrap_err("failed to commit the delphi rule scan")?;
+    let prepared =
+        crate::routes::v3::projects::mutation::prepare_technical_review_exits(
+            &exited_projects,
+            &mut transaction,
+            redis,
+        )
+        .await?;
+    prepared.commit(transaction, redis).await?;
 
     Ok(ScanSummary {
         revision,