diff --git a/apps/labrinth/src/database/models/organization_item.rs b/apps/labrinth/src/database/models/organization_item.rs index 845ff1235d..52a1d1c4df 100644 --- a/apps/labrinth/src/database/models/organization_item.rs +++ b/apps/labrinth/src/database/models/organization_item.rs @@ -206,6 +206,31 @@ impl DBOrganization { } } + pub async fn get_projects<'a, E>( + organization_id: DBOrganizationId, + exec: E, + ) -> Result, super::DatabaseError> + where + E: crate::database::Executor<'a, Database = sqlx::Postgres>, + { + use futures::TryStreamExt; + + let db_projects = sqlx::query!( + " + SELECT m.id FROM organizations o + INNER JOIN mods m ON m.organization_id = o.id + WHERE o.id = $1 + ", + organization_id as DBOrganizationId, + ) + .fetch(exec) + .map_ok(|m| DBProjectId(m.id)) + .try_collect::>() + .await?; + + Ok(db_projects) + } + pub async fn remove( id: DBOrganizationId, transaction: &mut PgTransaction<'_>, diff --git a/apps/labrinth/src/models/v3/projects.rs b/apps/labrinth/src/models/v3/projects.rs index 03e05e51b6..e9cef0b8e6 100644 --- a/apps/labrinth/src/models/v3/projects.rs +++ b/apps/labrinth/src/models/v3/projects.rs @@ -444,7 +444,7 @@ impl From for Link { /// Scheduled - Project is scheduled to be released in the future /// Private - Project is approved, but is not viewable to the public #[derive( - Serialize, Deserialize, Copy, Clone, Eq, PartialEq, Debug, utoipa::ToSchema, + Serialize, Deserialize, Copy, Clone, Eq, PartialEq, Hash, Debug, utoipa::ToSchema, )] #[serde(rename_all = "lowercase")] pub enum ProjectStatus { diff --git a/apps/labrinth/src/routes/internal/moderation/mod.rs b/apps/labrinth/src/routes/internal/moderation/mod.rs index b6d28c760a..e171677b2c 100644 --- a/apps/labrinth/src/routes/internal/moderation/mod.rs +++ b/apps/labrinth/src/routes/internal/moderation/mod.rs @@ -2,7 +2,7 @@ use super::ApiError; use crate::auth::get_user_from_headers; use crate::database; use crate::database::PgPool; -use crate::database::models::DBModerationLock; +use crate::database::models::{DBModerationLock, DBOrganization, DBOrganizationId, DBProjectId, DBProject}; use crate::database::models::moderation_external_item; use crate::models::ids::{OrganizationId, ProjectId}; use crate::models::projects::{ProjectStatus, VersionStatus}; @@ -20,7 +20,10 @@ use chrono::{DateTime, Utc}; use eyre::eyre; use serde::{Deserialize, Serialize}; use std::collections::HashMap; +use futures_util::future::try_join_all; use xredis::RedisPool; +use crate::routes::v3::organizations::OrganizationIds; +use crate::routes::v3::users::UserIds; pub mod external_license; mod ownership; @@ -37,6 +40,10 @@ pub fn config(cfg: &mut actix_web::web::ServiceConfig) { .service(release_lock) .service(release_lock_beacon) .service(delete_all_locks) + .service(get_user_project_grouped) + .service(get_users_project_grouped) + .service(get_organization_project_grouped) + .service(get_organizations_project_grouped) .service(web::scope("/tech-review").configure(tech_review::config)) .service( web::scope("/external-license").configure(external_license::config), @@ -215,7 +222,7 @@ pub struct DeleteAllLocksResponse { pub deleted_count: u64, } -/// List projects in the moderation queue. +/// List projects in the moderation queue. #[utoipa::path( context_path = "/moderation", tag = "moderation", @@ -1063,7 +1070,7 @@ fn row_to_ownership( }) } -/// Get project moderation metadata. +/// Get project moderation metadata. #[utoipa::path( context_path = "/moderation", tag = "moderation", @@ -1226,7 +1233,7 @@ pub enum Judgement { }, } -/// Update project moderation judgements. +/// Update project moderation judgements. #[utoipa::path( context_path = "/moderation", tag = "moderation", @@ -1328,7 +1335,7 @@ pub async fn set_project_meta( Ok(()) } -/// Acquire a moderation lock. +/// Acquire a moderation lock. /// Returns success if acquired, or info about who holds the lock if blocked. #[utoipa::path( context_path = "/moderation", @@ -1393,7 +1400,7 @@ pub async fn acquire_lock( } } -/// Override a moderation lock. +/// Override a moderation lock. #[utoipa::path( context_path = "/moderation", tag = "moderation", @@ -1444,7 +1451,7 @@ pub async fn override_lock( })) } -/// Get moderation lock status. +/// Get moderation lock status. #[utoipa::path( context_path = "/moderation", tag = "moderation", @@ -1511,7 +1518,7 @@ pub async fn get_lock_status( } } -/// Release a moderation lock. +/// Release a moderation lock. #[utoipa::path( context_path = "/moderation", tag = "moderation", @@ -1557,7 +1564,7 @@ pub async fn release_lock( Ok(web::Json(LockReleaseResponse { success: released })) } -/// Release a moderation lock by beacon. +/// Release a moderation lock by beacon. /// /// For use with `navigator.sendBeacon`, which cannot set `Authorization` or send `DELETE`. /// The body must be `text/plain` containing the same token value as the `Authorization` header @@ -1633,7 +1640,7 @@ pub async fn release_lock_beacon( Ok(web::Json(LockReleaseResponse { success: released })) } -/// Delete all moderation locks. +/// Delete all moderation locks. #[utoipa::path( context_path = "/moderation", tag = "moderation", @@ -1672,3 +1679,290 @@ pub async fn delete_all_locks( Ok(web::Json(DeleteAllLocksResponse { deleted_count })) } + +/// Get project id's for a given user with them grouped by their `ProjectStatus`. +/// +/// Only statuses with at least one project are present in the map. +#[utoipa::path( + context_path = "/moderation", + tag = "moderation", + security(("bearer_auth" = [])), + responses((status = OK, body = HashMap>)) +)] +#[get("/user/{user_id}/all-projects-grouped")] +pub async fn get_user_project_grouped( + req: HttpRequest, + info: web::Path<(String,)>, + pool: web::Data, + redis: web::Data, + session_queue: web::Data, +) -> Result>>, ApiError> { + check_is_moderator_from_headers( + &req, + &**pool, + &redis, + &session_queue, + Scopes::PROJECT_READ, + ) + .await + .wrap_auth_err("authenticating API request")?; + + let target_user = + database::models::DBUser::get(&info.into_inner().0, &**pool, &redis) + .await + .wrap_internal_err("fetching user from database")? + .wrap_not_found_err("resource not found")?; + + let counts = + user_projects_status_grouped(target_user.id, &**pool, &redis).await?; + + Ok(web::Json(counts)) +} + +/// Get project id's for a list of user's with them grouped by their `ProjectStatus`. +/// +/// Users that don't exist are silently omitted from the response; users +/// that exist but have no projects are included with an empty map. +#[utoipa::path( + context_path = "/moderation", + tag = "moderation", + security(("bearer_auth" = [])), + params(("ids" = String, Query)), + responses((status = OK, body = HashMap>>)) +)] +#[get("/users/all-projects-grouped")] +pub async fn get_users_project_grouped( + req: HttpRequest, + ids: web::Query, + pool: web::Data, + redis: web::Data, + session_queue: web::Data, +) -> Result>>>, ApiError> +{ + check_is_moderator_from_headers( + &req, + &**pool, + &redis, + &session_queue, + Scopes::PROJECT_READ, + ) + .await + .wrap_auth_err("authenticating API request")?; + + let user_ids = serde_json::from_str::>(&ids.ids) + .wrap_request_err("deserializing JSON data")?; + + if user_ids.is_empty() { + return Ok(web::Json(HashMap::new())); + } + + let target_users = + database::models::DBUser::get_many(&user_ids, &**pool, &redis) + .await + .wrap_internal_err("fetching users from database")?; + + let pool_ref = &**pool; + let redis_ref = &*redis; + + let grouped_projects_by_user = try_join_all(target_users.into_iter().map( + |target_user| async move { + let counts = user_projects_status_grouped( + target_user.id, + pool_ref, + redis_ref, + ) + .await?; + + Ok::<_, ApiError>((UserId::from(target_user.id), counts)) + }, + )) + .await? + .into_iter() + .collect::>(); + + Ok(web::Json(grouped_projects_by_user)) +} + +/// Get project id's for a given organization with them grouped by their `ProjectStatus`. +/// +/// Only statuses with at least one project are present in the map. +#[utoipa::path( + context_path = "/moderation", + tag = "moderation", + security(("bearer_auth" = [])), + responses((status = OK, body = HashMap>)) +)] +#[get("/organization/{organization_id}/all-projects-grouped")] +pub async fn get_organization_project_grouped( + req: HttpRequest, + info: web::Path<(String,)>, + pool: web::Data, + redis: web::Data, + session_queue: web::Data, +) -> Result>>, ApiError> { + check_is_moderator_from_headers( + &req, + &**pool, + &redis, + &session_queue, + Scopes::PROJECT_READ, + ) + .await + .wrap_auth_err("authenticating API request")?; + + let target_org = database::models::DBOrganization::get( + &info.into_inner().0, + &**pool, + &redis, + ) + .await + .wrap_internal_err("fetching organization from database")? + .wrap_not_found_err("resource not found")?; + + let grouped_projects = organization_projects_status_grouped( + target_org.id, + &**pool, + &redis, + ) + .await?; + + Ok(web::Json(grouped_projects)) +} + +/// Get project id's for a list of organization's with them grouped by their `ProjectStatus`. +/// +/// Organizations that don't exist are silently omitted from the +/// response; organizations that exist but have no projects are included +/// with an empty map. +#[utoipa::path( + context_path = "/moderation", + tag = "moderation", + security(("bearer_auth" = [])), + params(("ids" = String, Query)), + responses((status = OK, body = HashMap>>)) +)] +#[get("/organizations/all-projects-grouped")] +pub async fn get_organizations_project_grouped( + req: HttpRequest, + ids: web::Query, + pool: web::Data, + redis: web::Data, + session_queue: web::Data, +) -> Result< + web::Json>>>, + ApiError, +> { + check_is_moderator_from_headers( + &req, + &**pool, + &redis, + &session_queue, + Scopes::PROJECT_READ, + ) + .await + .wrap_auth_err("authenticating API request")?; + + let organization_ids = serde_json::from_str::>(&ids.ids) + .wrap_request_err("deserializing JSON data")?; + + if organization_ids.is_empty() { + return Ok(web::Json(HashMap::new())); + } + + let target_orgs = database::models::DBOrganization::get_many( + &organization_ids, + &**pool, + &redis, + ) + .await + .wrap_internal_err("fetching organizations from database")?; + + let pool_ref = &**pool; + let redis_ref = &*redis; + + let grouped_projects_by_org = try_join_all(target_orgs.into_iter().map( + |target_org| async move { + let counts = organization_projects_status_grouped( + target_org.id, + pool_ref, + redis_ref, + ) + .await?; + + Ok::<_, ApiError>((OrganizationId::from(target_org.id), counts)) + }, + )) + .await? + .into_iter() + .collect::>(); + + Ok(web::Json(grouped_projects_by_org)) +} + +/// Groups the given User projects by their `ProjectStatus`. +async fn user_projects_status_grouped<'a, E>( + user_id: database::models::DBUserId, + pool: E, + redis: &RedisPool, +) -> Result>, ApiError> +where + E: database::Executor<'a, Database = sqlx::Postgres> + + database::Acquire<'a, Database = sqlx::Postgres> + + Copy, +{ + let project_ids = + database::models::DBUser::get_projects(user_id, pool, redis) + .await + .wrap_internal_err("fetching user's projects from database")?; + + grouped_projects_for(&project_ids, pool, redis).await +} + + +/// Groups the given Organization projects by their `ProjectStatus`. +async fn organization_projects_status_grouped<'a, E>( + organization_id: DBOrganizationId, + pool: E, + redis: &RedisPool, +) -> Result>, ApiError> +where + E: database::Executor<'a, Database = sqlx::Postgres> + + database::Acquire<'a, Database = sqlx::Postgres> + + Copy, +{ + let project_ids = DBOrganization::get_projects(organization_id, pool) + .await + .wrap_internal_err("fetching project IDs from database")?; + + grouped_projects_for(&project_ids, pool, redis).await +} + +/// Groups the given input Projects by their `ProjectStatus`. +async fn grouped_projects_for<'a, E>( + project_ids: &[DBProjectId], + pool: E, + redis: &RedisPool, +) -> Result>, ApiError> +where + E: database::Executor<'a, Database = sqlx::Postgres> + + database::Acquire<'a, Database = sqlx::Postgres> + + Copy, +{ + if project_ids.is_empty() { + return Ok(HashMap::new()); + } + + let projects = + DBProject::get_many_ids(project_ids, pool, redis) + .await + .wrap_internal_err("fetching projects from database")?; + + let mut grouped_projects: HashMap> = HashMap::new(); + for project in &projects { + grouped_projects.entry(project.inner.status) + .or_default() + .push(project.inner.id.into()); + } + + Ok(grouped_projects) +} diff --git a/apps/labrinth/src/routes/internal/moderation/tech_review.rs b/apps/labrinth/src/routes/internal/moderation/tech_review.rs index c55d8fdbc7..230aacd7b5 100644 --- a/apps/labrinth/src/routes/internal/moderation/tech_review.rs +++ b/apps/labrinth/src/routes/internal/moderation/tech_review.rs @@ -14,7 +14,7 @@ use crate::{ database::{ DBProject, models::{ - DBFileId, DBProjectId, DBThread, DBThreadId, DBUser, DBVersion, + DBFileId, DBProjectId, DBThread, DBThreadId, DBUser, DBUserId, DBVersion, DBVersionId, DelphiReportId, DelphiReportIssueDetailsId, DelphiReportIssueId, delphi_report_item::{ @@ -26,7 +26,7 @@ use crate::{ }, }, models::{ - ids::{FileId, ProjectId, ThreadId, VersionId}, + ids::{FileId, ProjectId, ThreadId, ThreadMessageId, VersionId}, pats::Scopes, projects::{Project, ProjectStatus}, threads::{MessageBody, Thread}, @@ -43,6 +43,12 @@ use crate::{ util::error::Context, }; use eyre::eyre; +use futures_util::future::try_join_all; +use ariadne::ids::UserId; +use crate::database::models::{DBOrganization, DBOrganizationId}; +use crate::models::ids::OrganizationId; +use crate::routes::v3::organizations::OrganizationIds; +use crate::routes::v3::users::UserIds; pub mod global; @@ -55,7 +61,11 @@ pub fn config(cfg: &mut actix_web::web::ServiceConfig) { .service(submit_report) .service(update_issue_details) .service(update_global_issue_details) - .service(add_report); + .service(add_report) + .service(get_user_flagged_projects) + .service(get_users_flagged_projects) + .service(get_organization_flagged_projects) + .service(get_organizations_flagged_projects); } /// Arguments for searching project technical reviews. @@ -204,7 +214,7 @@ pub enum FlagReason { Delphi, } -/// Get a Delphi report issue. +/// Get a Delphi report issue. #[utoipa::path( context_path = "/moderation/tech-review", tag = "moderation", @@ -268,7 +278,7 @@ pub async fn get_issue( Ok(web::Json(row.data.0)) } -/// Get a project technical report. +/// Get a project technical report. #[utoipa::path( context_path = "/moderation/tech-review", tag = "moderation", @@ -684,7 +694,7 @@ async fn fetch_project_reports( Ok(project_reports) } -/// Search projects awaiting technical review. +/// Search projects awaiting technical review. #[utoipa::path( context_path = "/moderation/tech-review", tag = "moderation", @@ -896,7 +906,7 @@ pub async fn search_projects( })) } -/// Get a project technical review report. +/// Get a project technical review report. #[utoipa::path( context_path = "/moderation/tech-review", tag = "moderation", @@ -992,7 +1002,7 @@ pub struct SubmitReport { pub message: Option, } -/// Submit a technical review verdict. +/// Submit a technical review verdict. /// /// Before this is called, all issues for this project's reports must have been /// marked as either safe or unsafe. Otherwise, this will error with @@ -1210,7 +1220,7 @@ pub struct UpdateGlobalIssue { pub verdict: DelphiStatus, } -/// Update technical review issue details. +/// Update technical review issue details. /// /// This will not automatically reject the project for malware, but just flag /// this issue with a verdict. @@ -1497,7 +1507,7 @@ pub struct AddReport { pub file_id: FileId, } -/// Add a technical review report. +/// Add a technical review report. /// does not already exist for it. #[utoipa::path( context_path = "/moderation/tech-review", @@ -1566,3 +1576,335 @@ pub async fn add_report( Ok(web::Json(report_id)) } + +/// A user's project that is stuck in `processing` or `rejected` because the +/// most recent technical review verdict posted to its thread was `unsafe`. +#[derive(Debug, Clone, Serialize, utoipa::ToSchema)] +pub struct FlaggedProject { + pub project_id: ProjectId, + pub thread_id: ThreadId, + pub status: ProjectStatus, + /// The `tech_review` message that carried the `unsafe` verdict. + pub message_id: ThreadMessageId, + /// When that verdict was posted. + pub reviewed: DateTime, +} + +/// Get all of a user's `processing`/`rejected` projects whose most recent +/// `tech_review` thread message was an `unsafe` verdict. +#[utoipa::path( + context_path = "/moderation/tech-review", + tag = "moderation", + security(("bearer_auth" = [])), + responses((status = OK, body = Vec)) +)] +#[get("/user/{user_id}/flagged-projects")] +pub async fn get_user_flagged_projects( + req: HttpRequest, + info: web::Path<(String,)>, + pool: web::Data, + redis: web::Data, + session_queue: web::Data, +) -> Result>, ApiError> { + check_is_moderator_from_headers( + &req, + &**pool, + &redis, + &session_queue, + Scopes::PROJECT_READ, + ) + .await + .wrap_auth_err("authenticating API request")?; + + let target_user = DBUser::get(&info.into_inner().0, &**pool, &redis) + .await + .wrap_internal_err("fetching user from database")? + .wrap_not_found_err("resource not found")?; + + let flagged = + user_flagged_projects(target_user.id, &**pool, &redis).await?; + + Ok(web::Json(flagged)) +} + +/// Get all of multiple users' `processing`/`rejected` projects whose most +/// recent `tech_review` thread message was an `unsafe` verdict. Users that +/// don't exist are silently omitted from the response; users that exist +/// but have no matching projects are included with an empty list. +#[utoipa::path( + context_path = "/moderation/tech-review", + tag = "moderation", + security(("bearer_auth" = [])), + params(("ids" = String, Query)), + responses((status = OK, body = HashMap>)) +)] +#[get("/users/flagged-projects")] +pub async fn get_users_flagged_projects( + req: HttpRequest, + ids: web::Query, + pool: web::Data, + redis: web::Data, + session_queue: web::Data, +) -> Result>>, ApiError> { + check_is_moderator_from_headers( + &req, + &**pool, + &redis, + &session_queue, + Scopes::PROJECT_READ, + ) + .await + .wrap_auth_err("authenticating API request")?; + + let user_ids = serde_json::from_str::>(&ids.ids) + .wrap_request_err("deserializing JSON data")?; + + if user_ids.is_empty() { + return Ok(web::Json(HashMap::new())); + } + + let target_users = DBUser::get_many(&user_ids, &**pool, &redis) + .await + .wrap_internal_err("fetching users from database")?; + + let pool_ref = &**pool; + let redis_ref = &*redis; + + let flagged_by_user = try_join_all(target_users.into_iter().map( + |target_user| async move { + let flagged = + user_flagged_projects(target_user.id, pool_ref, redis_ref) + .await?; + + Ok::<_, ApiError>((UserId::from(target_user.id), flagged)) + }, + )) + .await? + .into_iter() + .collect::>(); + + Ok(web::Json(flagged_by_user)) +} + +/// Get all of an organization's `processing`/`rejected` projects whose most +/// recent `tech_review` thread message was an `unsafe` verdict. +#[utoipa::path( + context_path = "/moderation/tech-review", + tag = "moderation", + security(("bearer_auth" = [])), + responses((status = OK, body = Vec)) +)] +#[get("/organization/{organization_id}/flagged-projects")] +pub async fn get_organization_flagged_projects( + req: HttpRequest, + info: web::Path<(String,)>, + pool: web::Data, + redis: web::Data, + session_queue: web::Data, +) -> Result>, ApiError> { + check_is_moderator_from_headers( + &req, + &**pool, + &redis, + &session_queue, + Scopes::PROJECT_READ, + ) + .await + .wrap_auth_err("authenticating API request")?; + + let target_org = + DBOrganization::get(&info.into_inner().0, &**pool, &redis) + .await + .wrap_internal_err("fetching organization from database")? + .wrap_not_found_err("resource not found")?; + + let flagged = + organization_flagged_projects(target_org.id, &**pool, &redis).await?; + + Ok(web::Json(flagged)) +} + +/// Get all of multiple organizations' `processing`/`rejected` projects +/// whose most recent `tech_review` thread message was an `unsafe` verdict. +/// Organizations that don't exist are silently omitted from the response; +/// organizations that exist but have no matching projects are included +/// with an empty list. +#[utoipa::path( + context_path = "/moderation/tech-review", + tag = "moderation", + security(("bearer_auth" = [])), + params(("ids" = String, Query)), + responses((status = OK, body = HashMap>)) +)] +#[get("/organizations/flagged-projects")] +pub async fn get_organizations_flagged_projects( + req: HttpRequest, + ids: web::Query, + pool: web::Data, + redis: web::Data, + session_queue: web::Data, +) -> Result>>, ApiError> +{ + check_is_moderator_from_headers( + &req, + &**pool, + &redis, + &session_queue, + Scopes::PROJECT_READ, + ) + .await + .wrap_auth_err("authenticating API request")?; + + let organization_ids = serde_json::from_str::>(&ids.ids) + .wrap_request_err("deserializing JSON data")?; + + if organization_ids.is_empty() { + return Ok(web::Json(HashMap::new())); + } + + let target_orgs = + DBOrganization::get_many(&organization_ids, &**pool, &redis) + .await + .wrap_internal_err("fetching organizations from database")?; + + let pool_ref = &**pool; + let redis_ref = &*redis; + + let flagged_by_org = try_join_all(target_orgs.into_iter().map( + |target_org| async move { + let flagged = organization_flagged_projects( + target_org.id, + pool_ref, + redis_ref, + ) + .await?; + + Ok::<_, ApiError>((OrganizationId::from(target_org.id), flagged)) + }, + )) + .await? + .into_iter() + .collect::>(); + + Ok(web::Json(flagged_by_org)) +} + +/// Finds a single user's `processing`/`rejected` projects whose most recent +/// `tech_review` thread message was an `unsafe` verdict. Shared by the +/// single- and multi-user flagged-projects endpoints. +async fn user_flagged_projects<'a, E>( + user_id: DBUserId, + pool: E, + redis: &RedisPool, +) -> Result, ApiError> +where + E: crate::database::Executor<'a, Database = sqlx::Postgres> + + crate::database::Acquire<'a, Database = sqlx::Postgres> + + Copy, +{ + let project_ids = DBUser::get_projects(user_id, pool, redis) + .await + .wrap_internal_err("fetching user's projects from database")?; + + flagged_projects_among(&project_ids, pool, redis).await +} + +/// Finds a single organization's `processing`/`rejected` projects whose +/// most recent `tech_review` thread message was an `unsafe` verdict. +/// Shared by the single- and multi-organization flagged-projects +/// endpoints. +async fn organization_flagged_projects<'a, E>( + organization_id: DBOrganizationId, + pool: E, + redis: &RedisPool, +) -> Result, ApiError> +where + E: crate::database::Executor<'a, Database = sqlx::Postgres> + + crate::database::Acquire<'a, Database = sqlx::Postgres> + + Copy, +{ + let project_ids = DBOrganization::get_projects(organization_id, pool) + .await + .wrap_internal_err("fetching project IDs from database")?; + + flagged_projects_among(&project_ids, pool, redis).await +} + +/// Finds `processing`/`rejected` projects (from the given ids) whose most +/// recent `tech_review` thread message was an `unsafe` verdict. Shared by +/// the user- and organization-scoped flagged-projects helpers above. +async fn flagged_projects_among<'a, E>( + project_ids: &[DBProjectId], + pool: E, + redis: &RedisPool, +) -> Result, ApiError> +where + E: crate::database::Executor<'a, Database = sqlx::Postgres> + + crate::database::Acquire<'a, Database = sqlx::Postgres> + + Copy, +{ + if project_ids.is_empty() { + return Ok(Vec::new()); + } + + // Only `processing`/`rejected` projects are relevant, so narrow down + // before pulling any thread data. + let candidate_projects = DBProject::get_many_ids(project_ids, pool, redis) + .await + .wrap_internal_err("fetching projects from database")? + .into_iter() + .filter(|project| { + matches!( + project.inner.status, + ProjectStatus::Processing | ProjectStatus::Rejected + ) + }) + .collect::>(); + + if candidate_projects.is_empty() { + return Ok(Vec::new()); + } + + let thread_ids = candidate_projects + .iter() + .map(|project| project.thread_id) + .collect::>(); + + let threads = DBThread::get_many(&thread_ids, pool) + .await + .wrap_internal_err("fetching threads from database")? + .into_iter() + .map(|thread| (thread.id, thread)) + .collect::>(); + + Ok(candidate_projects + .into_iter() + .filter_map(|project| { + let thread = threads.get(&project.thread_id)?; + + // `DBThread::get_many` returns messages sorted oldest-first, so + // walking backwards finds the most recent `tech_review` entry + // regardless of what (if anything) was posted after it. + let last_review = thread.messages.iter().rev().find(|message| { + matches!(message.body, MessageBody::TechReview { .. }) + })?; + + let verdict = match &last_review.body { + MessageBody::TechReview { verdict } => *verdict, + _ => return None, + }; + + if verdict != DelphiVerdict::Unsafe { + return None; + } + + Some(FlaggedProject { + project_id: project.inner.id.into(), + thread_id: project.thread_id.into(), + status: project.inner.status, + message_id: last_review.id.into(), + reviewed: last_review.created, + }) + }) + .collect()) +}