fix up search pagination

apps/labrinth/.env.docker-compose

@@ -17,7 +17,7 @@ DATABASE_URL=postgresql://labrinth:labrinth@labrinth-postgres/labrinth
 DATABASE_MIN_CONNECTIONS=0
 DATABASE_MAX_CONNECTIONS=16
 
-SEARCH_BACKEND=typesense
+SEARCH_BACKEND=elasticsearch
 MEILISEARCH_READ_ADDR=http://localhost:7700
 MEILISEARCH_WRITE_ADDRS=http://localhost:7700
 MEILISEARCH_KEY=modrinth

apps/labrinth/.env.local

@@ -17,7 +17,7 @@ DATABASE_URL=postgresql://labrinth:labrinth@localhost/labrinth
 DATABASE_MIN_CONNECTIONS=0
 DATABASE_MAX_CONNECTIONS=16
 
-SEARCH_BACKEND=typesense
+SEARCH_BACKEND=elasticsearch
 
 # Meilisearch configuration
 MEILISEARCH_READ_ADDR=http://localhost:7700
@@ -32,7 +32,7 @@ ELASTICSEARCH_INDEX_PREFIX=labrinth
 # MEILISEARCH_READ_ADDR=http://localhost:7710
 # MEILISEARCH_WRITE_ADDRS=http://localhost:7700,http://localhost:7701
 
-SEARCH_BACKEND=typesense
+SEARCH_BACKEND=elasticsearch
 
 MEILISEARCH_KEY=modrinth
 MEILISEARCH_META_NAMESPACE=

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

@@ -35,6 +35,7 @@ use self::filter::{ElasticsearchFilter, serialize_filter};
 mod filter;
 
 const DELETE_FILTER_ID_BATCH_SIZE: usize = 1024;
+const MAX_RESULT_WINDOW: usize = 10_000;
 
 #[derive(Debug, Clone)]
 pub struct ElasticsearchConfig {
@@ -674,6 +675,56 @@ impl Elasticsearch {
 			.transpose()
 	}
 
+	async fn execute_search(
+		&self,
+		alias: &str,
+		body: &Value,
+		sort_only: bool,
+	) -> Result<Value, ApiError> {
+		let path = if sort_only {
+			format!(
+				"/{alias}/_search?filter_path=hits.total,hits.hits.sort"
+			)
+		} else {
+			format!("/{alias}/_search")
+		};
+		let response = self
+			.client
+			.request(Method::POST, &path)
+			.json(body)
+			.send()
+			.await
+			.wrap_internal_err("failed to execute Elasticsearch search")?;
+		response_json(response, "execute Elasticsearch search")
+			.await
+			.map_err(ApiError::Internal)
+	}
+
+	fn search_body(
+		query: &Value,
+		sort: &[Value],
+		from: usize,
+		size: usize,
+		track_total_hits: bool,
+		search_after: Option<&Value>,
+		include_source: bool,
+	) -> Value {
+		let mut body = json!({
+			"from": from,
+			"size": size,
+			"track_total_hits": track_total_hits,
+			"query": query,
+			"sort": sort
+		});
+		if let Some(search_after) = search_after {
+			body["search_after"] = search_after.clone();
+		}
+		if !include_source {
+			body["_source"] = Value::Bool(false);
+		}
+		body
+	}
+
 	async fn existing_write_indices(&self) -> Result<Vec<String>> {
 		let alias = self.config.alias_name();
 		let mut indices = self
@@ -783,25 +834,69 @@ impl SearchBackend for Elasticsearch {
 				"filter": filters
 			}
 		});
-		let body = json!({
-			"from": parsed.offset,
-			"size": parsed.hits_per_page,
-			"track_total_hits": true,
-			"query": query,
-			"sort": Self::sort(search_sort.index)
-		});
-
 		let alias = self.config.alias_name();
-		let response = self
+		let sort = Self::sort(search_sort.index);
-			.client
+		let mut remaining_offset = parsed.offset;
-			.request(Method::POST, &format!("/{alias}/_search"))
+		let mut search_after = None;
-			.json(&body)
+		let mut total_hits = None;
-			.send()
+		let deep_pagination = remaining_offset
-			.await
+			.saturating_add(parsed.hits_per_page)
-			.wrap_internal_err("failed to execute Elasticsearch search")?;
+			> MAX_RESULT_WINDOW;
-		let body = response_json(response, "execute Elasticsearch search")
+
-			.await
+		while deep_pagination && remaining_offset > 0 {
-			.map_err(ApiError::Internal)?;
+			let skipped = remaining_offset.min(MAX_RESULT_WINDOW);
+			let body = Self::search_body(
+				&query,
+				&sort,
+				0,
+				skipped,
+				total_hits.is_none(),
+				search_after.as_ref(),
+				false,
+			);
+			let body = self.execute_search(&alias, &body, true).await?;
+			if total_hits.is_none() {
+				let exact_total_hits = body["hits"]["total"]["value"]
+					.as_u64()
+					.unwrap_or_default()
+					as usize;
+				if parsed.offset >= exact_total_hits {
+					return Ok(SearchResults {
+						hits: Vec::new(),
+						page: parsed.page,
+						hits_per_page: parsed.hits_per_page,
+						total_hits: exact_total_hits,
+					});
+				}
+				total_hits = Some(exact_total_hits);
+			}
+			let Some(anchor) = body["hits"]["hits"]
+				.as_array()
+				.and_then(|hits| hits.last())
+				.and_then(|hit| hit.get("sort"))
+				.cloned()
+			else {
+				return Ok(SearchResults {
+					hits: Vec::new(),
+					page: parsed.page,
+					hits_per_page: parsed.hits_per_page,
+					total_hits: total_hits.unwrap_or_default(),
+				});
+			};
+			search_after = Some(anchor);
+			remaining_offset -= skipped;
+		}
+
+		let body = Self::search_body(
+			&query,
+			&sort,
+			remaining_offset,
+			parsed.hits_per_page,
+			true,
+			search_after.as_ref(),
+			true,
+		);
+		let body = self.execute_search(&alias, &body, false).await?;
 
 		let total_hits = body["hits"]["total"]["value"]
 			.as_u64()