Batched search indexing queries

apps/labrinth/src/search/indexing/local_import.rs

@@ -20,10 +20,12 @@ use crate::routes::v2_reroute;
 use crate::search::UploadSearchProject;
 use sqlx::postgres::PgPool;
 
-pub async fn index_local(
+pub async fn index_local_chunk(
     pool: &PgPool,
-) -> Result<Vec<UploadSearchProject>, IndexingError> {
-    info!("Indexing local projects!");
+    cursor: i64,
+    limit: i64,
+) -> Result<(Vec<UploadSearchProject>, i64), IndexingError> {
+    info!(cursor = cursor, limit = limit, "Indexing local projects");
 
     // todo: loaders, project type, game versions
     struct PartialProject {
@@ -45,13 +47,16 @@ pub async fn index_local(
         SELECT m.id id, m.name name, m.summary summary, m.downloads downloads, m.follows follows,
         m.icon_url icon_url, m.updated updated, m.approved approved, m.published, m.license license, m.slug slug, m.color
         FROM mods m
-        WHERE m.status = ANY($1)
-        GROUP BY m.id;
+        WHERE m.status = ANY($1) AND m.id > $2
+        GROUP BY m.id
+        LIMIT $3
         ",
         &*crate::models::projects::ProjectStatus::iterator()
         .filter(|x| x.is_searchable())
         .map(|x| x.to_string())
         .collect::<Vec<String>>(),
+        cursor,
+        limit,
     )
         .fetch(pool)
         .map_ok(|m| {
@@ -74,6 +79,10 @@ pub async fn index_local(
 
     let project_ids = db_projects.iter().map(|x| x.id.0).collect::<Vec<i64>>();
 
+    let Some(largest) = project_ids.iter().max() else {
+        return Ok((vec![], i64::MAX));
+    };
+
     struct PartialGallery {
         url: String,
         featured: bool,
@@ -412,7 +421,7 @@ pub async fn index_local(
         }
     }
 
-    Ok(uploads)
+    Ok((uploads, *largest))
 }
 
 struct PartialVersion {

apps/labrinth/src/search/indexing/mod.rs

@@ -8,7 +8,7 @@ use crate::search::{SearchConfig, UploadSearchProject};
 use ariadne::ids::base62_impl::to_base62;
 use futures::StreamExt;
 use futures::stream::FuturesOrdered;
-use local_import::index_local;
+use local_import::index_local_chunk;
 use meilisearch_sdk::client::{Client, SwapIndexes};
 use meilisearch_sdk::indexes::Index;
 use meilisearch_sdk::settings::{PaginationSetting, Settings};
@@ -34,8 +34,8 @@ pub enum IndexingError {
 
 // The chunk size for adding projects to the indexing database. If the request size
 // is too large (>10MiB) then the request fails with an error.  This chunk size
-// assumes a max average size of 4KiB per project to avoid this cap.
-const MEILISEARCH_CHUNK_SIZE: usize = 10000000;
+// assumes a max average size of 8KiB per project to avoid this cap.
+const MEILISEARCH_CHUNK_SIZE: usize = 5000000;
 const TIMEOUT: std::time::Duration = std::time::Duration::from_secs(60);
 
 pub async fn remove_documents(
@@ -115,15 +115,35 @@ pub async fn index_projects(
         .map(|x| x.field)
         .collect::<Vec<_>>();
 
-    let uploads = index_local(&pool).await?;
+    let mut cursor = 0;
+    let mut idx = 0;
+    let mut total = 0;
 
-    add_projects_batch_client(
-        &indices,
-        uploads,
-        all_loader_fields.clone(),
-        config,
-    )
-    .await?;
+    loop {
+        info!("Gathering index data chunk {idx}");
+        idx += 1;
+
+        let (uploads, next_cursor) =
+            index_local_chunk(&pool, cursor, 10000).await?;
+        total += uploads.len();
+
+        if uploads.is_empty() {
+            info!(
+                "No more projects to index, indexed {total} projects after {idx} chunks"
+            );
+            break;
+        }
+
+        cursor = next_cursor;
+
+        add_projects_batch_client(
+            &indices,
+            uploads,
+            all_loader_fields.clone(),
+            config,
+        )
+        .await?;
+    }
 
     // Swap the index
     swap_index(config, "projects").await?;