fix: change incremental search indexing operation set

apps/labrinth/src/background_task.rs

@@ -100,7 +100,7 @@ impl BackgroundTask {
             }
             IncrementalIndexSearch => {
                 crate::search::incremental::consume::run(
-                    pool,
+                    ro_pool,
                     redis_pool,
                     search_backend,
                     kafka_client,

apps/labrinth/src/models/v2/search.rs

@@ -152,7 +152,9 @@ impl LegacyResultSearchProject {
             server_side,
             environment: environments,
             versions,
-            latest_version: result_search_project.version_id,
+            latest_version: result_search_project
+                .version_id
+                .unwrap_or_default(),
             categories,
 
             project_id: result_search_project.project_id,

apps/labrinth/src/routes/v3/projects.rs

@@ -1207,6 +1207,14 @@ pub async fn project_edit_internal(
 
     let mut reindex_versions = new_project.categories.is_some()
         || new_project.additional_categories.is_some();
+    let became_searchable = !project_item.inner.status.is_searchable()
+        && new_project
+            .status
+            .is_some_and(|status| status.is_searchable());
+    let became_unsearchable = project_item.inner.status.is_searchable()
+        && new_project
+            .status
+            .is_some_and(|status| !status.is_searchable());
 
     reindex_versions |= update(
         &mut transaction,
@@ -1284,7 +1292,7 @@ pub async fn project_edit_internal(
         .await
         .wrap_internal_err("committing database transaction")?;
 
-    if reindex_versions {
+    if became_unsearchable {
         db_models::DBProject::clear_cache(
             project_item.inner.id,
             project_item.inner.slug,
@@ -1295,10 +1303,20 @@ pub async fn project_edit_internal(
         .wrap_internal_err("clearing cached data from Redis")?;
         search_state
             .queue
-            .push_version_changes(
-                project_item.inner.id.into(),
-                project_item.versions.iter().copied().map(VersionId::from),
-            )
+            .push_project_removal(project_item.inner.id.into())
+            .await;
+    } else if reindex_versions || became_searchable {
+        db_models::DBProject::clear_cache(
+            project_item.inner.id,
+            project_item.inner.slug,
+            None,
+            &redis,
+        )
+        .await
+        .wrap_internal_err("clearing cached data from Redis")?;
+        search_state
+            .queue
+            .push_project_with_all_versions_change(project_item.inner.id.into())
             .await;
     } else {
         clear_project_cache_and_queue_search(
@@ -1312,17 +1330,6 @@ pub async fn project_edit_internal(
         .wrap_api_err("executing `clear_project_cache_and_queue_search`")?;
     }
 
-    // Remove no longer searchable projects from search index
-    if let (true, Some(false)) = (
-        project_item.inner.status.is_searchable(),
-        new_project.status.map(|status| status.is_searchable()),
-    ) {
-        search_state
-            .queue
-            .push_project_removal(project_item.inner.id.into())
-            .await;
-    }
-
     Ok(HttpResponse::NoContent().body(""))
 }
 
@@ -1890,7 +1897,6 @@ pub async fn projects_edit(
         changed_projects.push((
             project.inner.id,
             project.inner.slug,
-            project.versions,
             reindex_versions,
         ));
     }
@@ -1900,17 +1906,14 @@ pub async fn projects_edit(
         .await
         .wrap_internal_err("committing database transaction")?;
 
-    for (project_id, slug, versions, reindex_versions) in changed_projects {
+    for (project_id, slug, reindex_versions) in changed_projects {
         if reindex_versions {
             db_models::DBProject::clear_cache(project_id, slug, None, &redis)
                 .await
                 .wrap_internal_err("clearing cached data from Redis")?;
             search_state
                 .queue
-                .push_version_changes(
-                    project_id.into(),
-                    versions.into_iter().map(VersionId::from),
-                )
+                .push_project_with_all_versions_change(project_id.into())
                 .await;
         } else {
             clear_project_cache_and_queue_search(

apps/labrinth/src/search/backend/elasticsearch/mod.rs

@@ -910,6 +910,7 @@ impl Elasticsearch {
                 let mut document = hit["_source"].clone();
                 let object = document.as_object_mut()?;
                 object.remove("document_type");
+                object.remove("_search_tokens");
                 if filter
                     .as_ref()
                     .is_some_and(|filter| filter.has_version_filter)

apps/labrinth/src/search/incremental.rs

@@ -17,7 +17,7 @@ use crate::{
 };
 
 pub const SEARCH_PROJECT_INDEX_QUEUE_TOPIC: &str =
-    "public.labrinth.search-project-index-queue.v1";
+    "public.labrinth.search-project-index-queue.v2";
 const QUEUE_FLUSH_INTERVAL: Duration = Duration::from_secs(10);
 
 #[derive(Clone)]
@@ -40,6 +40,16 @@ impl IncrementalSearchQueue {
         self.operations.lock().await.push_project_change(project_id);
     }
 
+    pub async fn push_project_with_all_versions_change(
+        &self,
+        project_id: ProjectId,
+    ) {
+        self.operations
+            .lock()
+            .await
+            .push_project_with_all_versions_change(project_id);
+    }
+
     pub async fn push_version_changes(
         &self,
         project_id: ProjectId,
@@ -116,32 +126,55 @@ impl IncrementalSearchQueue {
 #[derive(Default)]
 struct PendingSearchIndexOperations {
     changed_project_ids: HashSet<ProjectId>,
+    changed_project_ids_with_all_versions: HashSet<ProjectId>,
     changed_project_versions: HashMap<ProjectId, HashSet<VersionId>>,
     removed_project_ids: HashSet<ProjectId>,
 }
 
 impl PendingSearchIndexOperations {
     fn is_empty(&self) -> bool {
         self.changed_project_ids.is_empty()
+            && self.changed_project_ids_with_all_versions.is_empty()
             && self.changed_project_versions.is_empty()
             && self.removed_project_ids.is_empty()
     }
 
     fn push_project_change(&mut self, project_id: ProjectId) {
-        if !self.removed_project_ids.contains(&project_id) {
+        if !self.removed_project_ids.contains(&project_id)
+            && !self
+                .changed_project_ids_with_all_versions
+                .contains(&project_id)
+            && !self.changed_project_versions.contains_key(&project_id)
+        {
             self.changed_project_ids.insert(project_id);
         }
     }
 
+    fn push_project_with_all_versions_change(&mut self, project_id: ProjectId) {
+        if self.removed_project_ids.contains(&project_id) {
+            return;
+        }
+
+        self.changed_project_ids.remove(&project_id);
+        self.changed_project_versions.remove(&project_id);
+        self.changed_project_ids_with_all_versions
+            .insert(project_id);
+    }
+
     fn push_version_change(
         &mut self,
         project_id: ProjectId,
         version_ids: impl IntoIterator<Item = VersionId>,
     ) {
-        if self.removed_project_ids.contains(&project_id) {
+        if self.removed_project_ids.contains(&project_id)
+            || self
+                .changed_project_ids_with_all_versions
+                .contains(&project_id)
+        {
             return;
         }
 
+        self.changed_project_ids.remove(&project_id);
         let version_ids = version_ids.into_iter().collect::<HashSet<_>>();
         if !version_ids.is_empty() {
             self.changed_project_versions
@@ -153,6 +186,8 @@ impl PendingSearchIndexOperations {
 
     fn push_project_removal(&mut self, project_id: ProjectId) {
         self.changed_project_ids.remove(&project_id);
+        self.changed_project_ids_with_all_versions
+            .remove(&project_id);
         self.changed_project_versions.remove(&project_id);
         self.removed_project_ids.insert(project_id);
     }
@@ -162,6 +197,9 @@ impl PendingSearchIndexOperations {
             SearchProjectIndexQueueEventData::Change { project_id } => {
                 self.push_project_change(project_id)
             }
+            SearchProjectIndexQueueEventData::ChangeWithAllVersions {
+                project_id,
+            } => self.push_project_with_all_versions_change(project_id),
             SearchProjectIndexQueueEventData::VersionChange {
                 project_id,
                 version_ids,
@@ -175,6 +213,7 @@ impl PendingSearchIndexOperations {
     fn into_events(self) -> Vec<SearchProjectIndexQueueEventData> {
         let mut events = Vec::with_capacity(
             self.changed_project_ids.len()
+                + self.changed_project_ids_with_all_versions.len()
                 + self.changed_project_versions.len()
                 + self.removed_project_ids.len(),
         );
@@ -185,6 +224,15 @@ impl PendingSearchIndexOperations {
         events.extend(self.changed_project_ids.into_iter().map(|project_id| {
             SearchProjectIndexQueueEventData::Change { project_id }
         }));
+        events.extend(
+            self.changed_project_ids_with_all_versions.into_iter().map(
+                |project_id| {
+                    SearchProjectIndexQueueEventData::ChangeWithAllVersions {
+                        project_id,
+                    }
+                },
+            ),
+        );
         events.extend(self.changed_project_versions.into_iter().map(
             |(project_id, version_ids)| {
                 SearchProjectIndexQueueEventData::VersionChange {
@@ -202,6 +250,8 @@ impl PendingSearchIndexOperations {
 pub enum SearchProjectIndexQueueEventData {
     #[serde(rename = "project_change")]
     Change { project_id: ProjectId },
+    #[serde(rename = "project_change_with_all_versions")]
+    ChangeWithAllVersions { project_id: ProjectId },
     #[serde(rename = "project_version_change")]
     VersionChange {
         project_id: ProjectId,

apps/labrinth/src/search/incremental/consume.rs

@@ -21,7 +21,10 @@ use crate::{
         SearchBackend, SearchDocumentBatch, SearchIndexUpdate,
         UploadSearchProject,
         incremental::SEARCH_PROJECT_INDEX_QUEUE_TOPIC,
-        indexing::{build_project_documents, build_version_change_documents},
+        indexing::{
+            build_project_and_all_version_documents, build_project_documents,
+            build_version_change_documents,
+        },
     },
     util::kafka::{
         INCREMENTAL_INDEX_SEARCH_TASK, KAFKA_OPERATION_INTERVAL,
@@ -132,6 +135,7 @@ async fn consume_batch(
     let start = Instant::now();
 
     let mut project_ids_to_change = HashSet::new();
+    let mut project_ids_with_all_versions_to_change = HashSet::new();
     let mut project_ids_with_version_changes = HashSet::new();
     let mut project_ids_to_remove = HashSet::new();
     let mut version_ids_to_change = HashSet::new();
@@ -180,6 +184,11 @@ async fn consume_batch(
             SearchProjectIndexQueueEventData::Change { project_id } => {
                 project_ids_to_change.insert(project_id);
             }
+            SearchProjectIndexQueueEventData::ChangeWithAllVersions {
+                project_id,
+            } => {
+                project_ids_with_all_versions_to_change.insert(project_id);
+            }
             SearchProjectIndexQueueEventData::VersionChange {
                 project_id,
                 version_ids,
@@ -198,33 +207,73 @@ async fn consume_batch(
 
     project_ids_to_change
         .retain(|project_id| !project_ids_to_remove.contains(project_id));
+    project_ids_with_all_versions_to_change
+        .retain(|project_id| !project_ids_to_remove.contains(project_id));
     project_ids_with_version_changes
         .retain(|project_id| !project_ids_to_remove.contains(project_id));
+    project_ids_with_version_changes.retain(|project_id| {
+        !project_ids_with_all_versions_to_change.contains(project_id)
+    });
     project_ids_to_change.retain(|project_id| {
         !project_ids_with_version_changes.contains(project_id)
+            && !project_ids_with_all_versions_to_change.contains(project_id)
     });
     let project_ids_to_change =
         project_ids_to_change.into_iter().collect::<Vec<_>>();
     let project_ids_with_version_changes = project_ids_with_version_changes
         .into_iter()
         .collect::<Vec<_>>();
+    let project_ids_with_all_versions_to_change =
+        project_ids_with_all_versions_to_change
+            .into_iter()
+            .collect::<Vec<_>>();
     let mut project_ids_to_remove =
         project_ids_to_remove.into_iter().collect::<Vec<_>>();
     let version_ids_to_change =
         version_ids_to_change.into_iter().collect::<Vec<_>>();
 
     info!(
         kafka.message_count = messages_to_commit.len(),
-        "Read all Kafka messages in {:.2?}, found {} projects to change, {} projects with {} version changes, and {} projects to remove",
+        "Read all Kafka messages in {:.2?}, found {} projects to change, {} projects with all versions to change, {} projects with {} version changes, and {} projects to remove",
         start.elapsed(),
         project_ids_to_change.len(),
+        project_ids_with_all_versions_to_change.len(),
         project_ids_with_version_changes.len(),
         version_ids_to_change.len(),
         project_ids_to_remove.len(),
     );
     let start = Instant::now();
     let mut documents = SearchDocumentBatch::default();
 
+    if !project_ids_with_all_versions_to_change.is_empty() {
+        let operation_start = Instant::now();
+        let changed_documents = build_project_and_all_version_documents(
+            ro_pool,
+            redis_pool,
+            &project_ids_with_all_versions_to_change,
+        )
+        .instrument(info_span!(
+            "index",
+            batch_size = project_ids_with_all_versions_to_change.len()
+        ))
+        .await
+        .wrap_err_with(|| {
+            format!(
+                "failed to build search documents for {} projects and all their versions",
+                project_ids_with_all_versions_to_change.len()
+            )
+        })?;
+        project_ids_to_remove
+            .extend(project_ids_with_all_versions_to_change.iter().copied());
+        documents.projects.extend(changed_documents.projects);
+        documents.versions.extend(changed_documents.versions);
+        info!(
+            project_count = project_ids_with_all_versions_to_change.len(),
+            "Built changed projects and all their versions in {:.2?}",
+            operation_start.elapsed()
+        );
+    }
+
     if !project_ids_with_version_changes.is_empty() {
         let operation_start = Instant::now();
         let changed_documents = build_version_change_documents(
@@ -402,6 +451,8 @@ enum SearchProjectIndexQueueEvent {
 enum SearchProjectIndexQueueEventData {
     #[serde(rename = "project_change")]
     Change { project_id: ProjectId },
+    #[serde(rename = "project_change_with_all_versions")]
+    ChangeWithAllVersions { project_id: ProjectId },
     #[serde(rename = "project_version_change")]
     VersionChange {
         project_id: ProjectId,