clean up SearchBackend interface

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

@@ -29,7 +29,8 @@ use crate::queue::session::AuthQueue;
 use crate::routes::ApiError;
 use crate::routes::internal::delphi;
 use crate::search::{
-    SearchBackend, SearchQuery, SearchRequest, SearchResults, SearchState,
+    SearchBackend, SearchIndexUpdate, SearchQuery, SearchRequest,
+    SearchResults, SearchState,
 };
 use crate::util::error::Context;
 use crate::util::img;
@@ -2849,16 +2850,13 @@ pub async fn project_delete_internal(
             &redis,
         )
         .await?;
+        let project_id = project.inner.id.into();
         search_state
             .backend
-            .remove_project_version_documents(&[project.inner.id.into()])
-            .await
-            .wrap_internal_err(
-                "failed to remove project versions from search index",
-            )?;
-        search_state
-            .backend
-            .remove_project_documents(&[project.inner.id.into()])
+            .apply_update(SearchIndexUpdate {
+                removed_projects: std::slice::from_ref(&project_id),
+                ..SearchIndexUpdate::default()
+            })
             .await
             .wrap_internal_err("failed to remove project from search index")?;
         search_state

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

@@ -27,7 +27,7 @@ use crate::models::teams::ProjectPermissions;
 use crate::queue::file_scan::get_files_missing_attribution;
 use crate::queue::session::AuthQueue;
 use crate::routes::internal::delphi;
-use crate::search::incremental::consume::reindex_project;
+use crate::search::incremental::consume::reindex_project_versions;
 use crate::search::{SearchBackend, SearchState};
 use crate::util::error::Context;
 use crate::util::img;
@@ -1257,18 +1257,17 @@ pub async fn version_delete(
         [VersionId::from(version.inner.id)],
     )
     .await?;
-    search_backend
-        .remove_version_documents(&[version.inner.id.into()])
-        .await
-        .wrap_internal_err("failed to remove version search document")?;
-    reindex_project(
+    let project_id = version.inner.project_id.into();
+    let version_id = version.inner.id.into();
+    reindex_project_versions(
         &pool,
         &redis,
         search_backend.as_ref(),
-        version.inner.project_id.into(),
+        std::slice::from_ref(&project_id),
+        std::slice::from_ref(&version_id),
     )
     .await
-    .wrap_internal_err("failed to reindex project")?;
+    .wrap_internal_err("failed to update search index after version removal")?;
     if result.is_some() {
         Ok(HttpResponse::NoContent().body(""))
     } else {

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

@@ -1,6 +1,5 @@
 use std::sync::LazyLock;
 
-use ariadne::ids::base62_impl::to_base62;
 use async_trait::async_trait;
 use eyre::{Result, eyre};
 use itertools::Itertools;
@@ -13,7 +12,6 @@ use tracing::{debug, info};
 use crate::database::PgPool;
 use crate::database::redis::RedisPool;
 use crate::env::ENV;
-use crate::models::ids::{ProjectId, VersionId};
 use crate::routes::ApiError;
 use crate::search::backend::{
     SearchIndex, combined_search_filters, parse_search_index,
@@ -21,8 +19,9 @@ use crate::search::backend::{
 };
 use crate::search::indexing::index_local;
 use crate::search::{
-    ResultSearchProject, SearchBackend, SearchField, SearchRequest,
-    SearchResults, TasksCancelFilter, UploadSearchProject, UploadSearchVersion,
+    ResultSearchProject, SearchBackend, SearchField, SearchIndexUpdate,
+    SearchRequest, SearchResults, TasksCancelFilter, UploadSearchProject,
+    UploadSearchVersion,
 };
 use crate::util::error::Context;
 
@@ -959,6 +958,19 @@ impl Typesense {
         Ok(())
     }
 
+    async fn delete_ids_from_write_collections(
+        &self,
+        alias: &str,
+        field: &str,
+        ids: &[String],
+    ) -> Result<()> {
+        for ids in ids.chunks(DELETE_FILTER_ID_BATCH_SIZE) {
+            let filter = format!("{field}:[{}]", ids.iter().join(", "));
+            self.delete_from_write_collections(alias, &filter).await?;
+        }
+        Ok(())
+    }
+
     async fn delete_legacy_filtered_collections(&self) -> Result<()> {
         let alias = self.config.get_alias_name("projects_filtered");
         let live = self.client.get_alias(&alias).await?;
@@ -1265,101 +1277,94 @@ impl SearchBackend for Typesense {
         Ok(())
     }
 
-    async fn index_documents(
+    async fn apply_update(
         &self,
-        documents: &[UploadSearchProject],
+        update: SearchIndexUpdate<'_>,
     ) -> eyre::Result<()> {
-        if documents.is_empty() {
-            return Ok(());
-        }
-
-        let alias = self.config.get_alias_name("projects");
-        let collections = self.existing_write_collections(&alias).await?;
-        debug!(
-            ?collections,
-            num_documents = documents.len(),
-            "Inserting into collections",
-        );
-        self.import_document_batches(&collections, documents)
-            .await?;
-
-        debug!("Done importing");
-        Ok(())
-    }
-
-    async fn index_version_documents(
-        &self,
-        documents: &[UploadSearchVersion],
-    ) -> eyre::Result<()> {
-        if documents.is_empty() {
-            return Ok(());
-        }
-
-        let alias = self.config.get_alias_name("versions");
-        let collections = self.existing_write_collections(&alias).await?;
-        debug!(
-            ?collections,
-            num_documents = documents.len(),
-            "Inserting version documents into collections",
-        );
-        self.import_version_document_batches(&collections, documents)
-            .await?;
-
-        debug!("Done importing version documents");
-        Ok(())
-    }
-
-    async fn remove_project_documents(
-        &self,
-        ids: &[ProjectId],
-    ) -> eyre::Result<()> {
-        if ids.is_empty() {
-            return Ok(());
-        }
-
-        let id_list = ids
-            .iter()
-            .map(|id| to_base62(id.0))
-            .collect::<Vec<_>>()
-            .join(", ");
-        let filter = format!("project_id:[{id_list}]");
-
         let projects_alias = self.config.get_alias_name("projects");
-        self.delete_from_write_collections(&projects_alias, &filter)
+        let versions_alias = self.config.get_alias_name("versions");
+
+        let removed_project_ids = update
+            .removed_projects
+            .iter()
+            .map(ToString::to_string)
+            .collect::<Vec<_>>();
+        if !removed_project_ids.is_empty() {
+            self.delete_ids_from_write_collections(
+                &versions_alias,
+                "project_id",
+                &removed_project_ids,
+            )
+            .await?;
+            self.delete_ids_from_write_collections(
+                &projects_alias,
+                "project_id",
+                &removed_project_ids,
+            )
             .await?;
-
-        debug!("Done");
-        Ok(())
-    }
-
-    async fn remove_project_version_documents(
-        &self,
-        ids: &[ProjectId],
-    ) -> eyre::Result<()> {
-        if ids.is_empty() {
-            return Ok(());
         }
 
-        let id_list = ids.iter().map(ToString::to_string).join(", ");
-        let filter = format!("project_id:[{id_list}]");
-        let alias = self.config.get_alias_name("versions");
-        self.delete_from_write_collections(&alias, &filter).await
-    }
-
-    async fn remove_version_documents(
-        &self,
-        ids: &[VersionId],
-    ) -> eyre::Result<()> {
-        if ids.is_empty() {
-            return Ok(());
+        let version_ids = update
+            .removed_versions
+            .iter()
+            .map(ToString::to_string)
+            .chain(
+                update
+                    .versions
+                    .iter()
+                    .map(|document| document.version_id.clone()),
+            )
+            .unique()
+            .collect::<Vec<_>>();
+        if !version_ids.is_empty() {
+            self.delete_ids_from_write_collections(
+                &versions_alias,
+                "id",
+                &version_ids,
+            )
+            .await?;
         }
 
-        let alias = self.config.get_alias_name("versions");
-        for ids in ids.chunks(DELETE_FILTER_ID_BATCH_SIZE) {
-            let id_list = ids.iter().map(ToString::to_string).join(", ");
-            let filter = format!("id:[{id_list}]");
-            self.delete_from_write_collections(&alias, &filter).await?;
+        let project_ids = update
+            .projects
+            .iter()
+            .map(|document| document.project_id.clone())
+            .unique()
+            .collect::<Vec<_>>();
+        if !project_ids.is_empty() {
+            self.delete_ids_from_write_collections(
+                &projects_alias,
+                "project_id",
+                &project_ids,
+            )
+            .await?;
         }
+
+        if !update.projects.is_empty() {
+            let collections =
+                self.existing_write_collections(&projects_alias).await?;
+            debug!(
+                ?collections,
+                num_documents = update.projects.len(),
+                "Replacing project documents in collections",
+            );
+            self.import_document_batches(&collections, update.projects)
+                .await?;
+        }
+
+        if !update.versions.is_empty() {
+            let collections =
+                self.existing_write_collections(&versions_alias).await?;
+            debug!(
+                ?collections,
+                num_documents = update.versions.len(),
+                "Replacing version documents in collections",
+            );
+            self.import_version_document_batches(&collections, update.versions)
+                .await?;
+        }
+
+        debug!("Done applying search index update");
         Ok(())
     }
 

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

@@ -17,7 +17,8 @@ use crate::{
     env::ENV,
     models::ids::{ProjectId, VersionId},
     search::{
-        SearchBackend,
+        SearchBackend, SearchDocumentBatch, SearchIndexUpdate,
+        UploadSearchProject,
         incremental::SEARCH_PROJECT_INDEX_QUEUE_TOPIC,
         indexing::{index_project_documents, index_project_version_documents},
     },
@@ -214,56 +215,24 @@ async fn consume_batch(
         project_ids_to_remove.len(),
     );
     let start = Instant::now();
-
-    if !project_ids_to_remove.is_empty() {
-        let operation_start = Instant::now();
-        info!(
-            project_count = project_ids_to_remove.len(),
-            "Removing project documents"
-        );
-        search_backend
-            .remove_project_version_documents(&project_ids_to_remove)
-            .await
-            .wrap_err("failed to remove project version documents")?;
-        search_backend
-            .remove_project_documents(&project_ids_to_remove)
-            .await
-            .wrap_err("failed to remove project documents")?;
-        info!(
-            project_count = project_ids_to_remove.len(),
-            "Removed project documents in {:.2?}",
-            operation_start.elapsed()
-        );
-    }
-
-    if !version_ids_to_change.is_empty() {
-        let operation_start = Instant::now();
-        search_backend
-            .remove_version_documents(&version_ids_to_change)
-            .await
-            .wrap_err("failed to remove changed version documents")?;
-        info!(
-            version_count = version_ids_to_change.len(),
-            "Removed changed version documents in {:.2?}",
-            operation_start.elapsed()
-        );
-    }
+    let mut documents = SearchDocumentBatch::default();
 
     if !project_ids_with_version_changes.is_empty() {
         let operation_start = Instant::now();
-        reindex_changed_project_versions(
+        let changed_documents = build_changed_project_versions(
             ro_pool,
             redis_pool,
-            search_backend,
             &project_ids_with_version_changes,
             &version_ids_to_change,
         )
         .await
-        .wrap_err("failed to reindex changed project versions")?;
+        .wrap_err("failed to build changed project versions")?;
+        documents.projects.extend(changed_documents.projects);
+        documents.versions.extend(changed_documents.versions);
         info!(
             project_count = project_ids_with_version_changes.len(),
             version_count = version_ids_to_change.len(),
-            "Reindexed changed project versions in {:.2?}",
+            "Built changed project versions in {:.2?}",
             operation_start.elapsed()
         );
     }
@@ -272,23 +241,39 @@ async fn consume_batch(
         let operation_start = Instant::now();
         info!(
             project_count = project_ids_to_change.len(),
-            "Reindexing changed projects"
+            "Building changed projects"
+        );
+        documents.projects.extend(
+            build_changed_projects(ro_pool, redis_pool, &project_ids_to_change)
+                .await
+                .wrap_err("failed to build changed projects")?,
         );
-        reindex_projects(
-            ro_pool,
-            redis_pool,
-            search_backend,
-            &project_ids_to_change,
-        )
-        .await
-        .wrap_err("failed to reindex changed project batch")?;
         info!(
             project_count = project_ids_to_change.len(),
-            "Reindexed changed projects in {:.2?}",
+            "Built changed projects in {:.2?}",
             operation_start.elapsed()
         );
     }
 
+    let operation_start = Instant::now();
+    search_backend
+        .apply_update(SearchIndexUpdate {
+            projects: &documents.projects,
+            versions: &documents.versions,
+            removed_projects: &project_ids_to_remove,
+            removed_versions: &version_ids_to_change,
+        })
+        .await
+        .wrap_err("failed to apply search index update")?;
+    info!(
+        project_count = documents.projects.len(),
+        version_count = documents.versions.len(),
+        removed_project_count = project_ids_to_remove.len(),
+        removed_version_count = version_ids_to_change.len(),
+        "Applied search index update in {:.2?}",
+        operation_start.elapsed()
+    );
+
     for message in messages_to_commit {
         consumer
             .commit_message(&message, CommitMode::Async)
@@ -320,22 +305,24 @@ pub async fn reindex_projects(
     search_backend: &dyn SearchBackend,
     project_ids: &[ProjectId],
 ) -> eyre::Result<()> {
-    info!("Removing documents for batch");
-    search_backend.remove_project_documents(project_ids).await?;
-
     info!("Creating project documents");
-    index_changed_projects(ro_pool, redis_pool, search_backend, project_ids)
+    let projects =
+        build_changed_projects(ro_pool, redis_pool, project_ids).await?;
+    search_backend
+        .apply_update(SearchIndexUpdate {
+            projects: &projects,
+            ..SearchIndexUpdate::default()
+        })
         .await?;
 
     Ok(())
 }
 
-async fn index_changed_projects(
+async fn build_changed_projects(
     ro_pool: &PgPool,
     redis_pool: &RedisPool,
-    search_backend: &dyn SearchBackend,
     project_ids: &[ProjectId],
-) -> eyre::Result<()> {
+) -> eyre::Result<Vec<UploadSearchProject>> {
     let documents = index_project_documents(ro_pool, redis_pool, project_ids)
         .instrument(info_span!("index", batch_size = project_ids.len()))
         .await
@@ -346,20 +333,16 @@ async fn index_changed_projects(
             )
         })?;
 
-    info!("Fetched all project documents, indexing into backend");
-
-    search_backend.index_documents(&documents).await?;
-
-    Ok(())
+    info!("Fetched all project documents");
+    Ok(documents)
 }
 
-async fn reindex_changed_project_versions(
+async fn build_changed_project_versions(
     ro_pool: &PgPool,
     redis_pool: &RedisPool,
-    search_backend: &dyn SearchBackend,
     project_ids: &[ProjectId],
     version_ids: &[VersionId],
-) -> eyre::Result<()> {
+) -> eyre::Result<SearchDocumentBatch> {
     let documents = index_project_version_documents(
         ro_pool,
         redis_pool,
@@ -380,13 +363,31 @@ async fn reindex_changed_project_versions(
         )
     })?;
 
-    search_backend.remove_project_documents(project_ids).await?;
-    search_backend.index_documents(&documents.projects).await?;
-    search_backend
-        .index_version_documents(&documents.versions)
-        .await?;
+    Ok(documents)
+}
 
-    Ok(())
+pub async fn reindex_project_versions(
+    ro_pool: &PgPool,
+    redis_pool: &RedisPool,
+    search_backend: &dyn SearchBackend,
+    project_ids: &[ProjectId],
+    version_ids: &[VersionId],
+) -> eyre::Result<()> {
+    let documents = build_changed_project_versions(
+        ro_pool,
+        redis_pool,
+        project_ids,
+        version_ids,
+    )
+    .await?;
+    search_backend
+        .apply_update(SearchIndexUpdate {
+            projects: &documents.projects,
+            versions: &documents.versions,
+            removed_versions: version_ids,
+            ..SearchIndexUpdate::default()
+        })
+        .await
 }
 
 #[derive(Debug, Deserialize)]

apps/labrinth/src/search/mod.rs

@@ -111,29 +111,9 @@ pub trait SearchBackend: Send + Sync {
         redis: RedisPool,
     ) -> eyre::Result<()>;
 
-    async fn index_documents(
+    async fn apply_update(
         &self,
-        documents: &[UploadSearchProject],
-    ) -> eyre::Result<()>;
-
-    async fn index_version_documents(
-        &self,
-        documents: &[UploadSearchVersion],
-    ) -> eyre::Result<()>;
-
-    async fn remove_project_documents(
-        &self,
-        ids: &[ProjectId],
-    ) -> eyre::Result<()>;
-
-    async fn remove_project_version_documents(
-        &self,
-        ids: &[ProjectId],
-    ) -> eyre::Result<()>;
-
-    async fn remove_version_documents(
-        &self,
-        ids: &[VersionId],
+        update: SearchIndexUpdate<'_>,
     ) -> eyre::Result<()>;
 
     async fn tasks(&self) -> eyre::Result<Value>;
@@ -328,6 +308,17 @@ pub struct SearchDocumentBatch {
     pub versions: Vec<UploadSearchVersion>,
 }
 
+/// A logical search index mutation. Removals are applied before replacements,
+/// so a document may be present in both a removed and replacement field.
+#[derive(Debug, Clone, Copy, Default)]
+pub struct SearchIndexUpdate<'a> {
+    pub projects: &'a [UploadSearchProject],
+    pub versions: &'a [UploadSearchVersion],
+    /// Projects and all of their version documents to remove.
+    pub removed_projects: &'a [ProjectId],
+    pub removed_versions: &'a [VersionId],
+}
+
 /// Nullable fields in Typesense-bound documents should use
 /// `skip_serializing_if = "Option::is_none"` so they are omitted instead of
 /// serialized as `null`.