diff --git a/apps/labrinth/.env.docker-compose b/apps/labrinth/.env.docker-compose index 55bc2f232b..871cbd8133 100644 --- a/apps/labrinth/.env.docker-compose +++ b/apps/labrinth/.env.docker-compose @@ -25,9 +25,12 @@ ELASTICSEARCH_INDEX_PREFIX=labrinth ELASTICSEARCH_USERNAME=elastic ELASTICSEARCH_PASSWORD=elastic SEARCH_INDEX_CHUNK_SIZE=5000 +SEARCH_INCREMENTAL_INDEX_BATCH_DELAY_SECONDS=5 +SEARCH_INCREMENTAL_INDEX_BATCH_MAX_SIZE=1000 TYPESENSE_URL=http://localhost:8108 TYPESENSE_API_KEY=modrinth TYPESENSE_INDEX_PREFIX=labrinth +TYPESENSE_IMPORT_BATCH_SIZE=5000 REDIS_URL=redis://labrinth-redis REDIS_MIN_CONNECTIONS=0 diff --git a/apps/labrinth/.env.local b/apps/labrinth/.env.local index 364f90f1e5..25ef3e5ab5 100644 --- a/apps/labrinth/.env.local +++ b/apps/labrinth/.env.local @@ -43,9 +43,12 @@ ELASTICSEARCH_USERNAME= ELASTICSEARCH_PASSWORD= SEARCH_INDEX_CHUNK_SIZE=5000 +SEARCH_INCREMENTAL_INDEX_BATCH_DELAY_SECONDS=5 +SEARCH_INCREMENTAL_INDEX_BATCH_MAX_SIZE=1000 TYPESENSE_URL=http://localhost:8108 TYPESENSE_API_KEY=modrinth TYPESENSE_INDEX_PREFIX=labrinth +TYPESENSE_IMPORT_BATCH_SIZE=5000 REDIS_URL=redis://localhost REDIS_MIN_CONNECTIONS=0 diff --git a/apps/labrinth/src/background_task.rs b/apps/labrinth/src/background_task.rs index 31c51e2367..b49692fa81 100644 --- a/apps/labrinth/src/background_task.rs +++ b/apps/labrinth/src/background_task.rs @@ -19,7 +19,7 @@ use crate::util::anrok; use actix_web::web; use clap::ValueEnum; use eyre::WrapErr; -use tracing::info; +use tracing::{info, instrument}; #[derive(ValueEnum, Debug, Copy, Clone, PartialEq, Eq)] #[clap(rename_all = "kebab_case")] @@ -50,6 +50,7 @@ pub enum BackgroundTask { impl BackgroundTask { #[allow(clippy::too_many_arguments)] + #[instrument(skip_all, fields(background_task = ?self))] pub async fn run( self, pool: PgPool, @@ -176,7 +177,7 @@ pub async fn index_search( search_backend: web::Data, ) -> eyre::Result<()> { info!("Indexing local database"); - search_backend.index_projects(ro_pool, redis_pool).await + search_backend.rebuild_index(ro_pool, redis_pool).await } pub async fn release_scheduled(pool: PgPool) -> eyre::Result<()> { diff --git a/apps/labrinth/src/env.rs b/apps/labrinth/src/env.rs index 8c8e32b0b7..f6a4705cd6 100644 --- a/apps/labrinth/src/env.rs +++ b/apps/labrinth/src/env.rs @@ -38,7 +38,7 @@ macro_rules! vars { )] let $field: Option<$ty> = { let mut default = None::<$ty>; - $( default = Some({ $default }.into()); )? + $( default = Some(<$ty>::from({ $default })); )? match parse_value::<$ty>(stringify!($field), default) { Ok(value) => Some(value), @@ -160,9 +160,13 @@ vars! { // search SEARCH_BACKEND: crate::search::SearchBackendKind = crate::search::SearchBackendKind::Typesense; SEARCH_INDEX_CHUNK_SIZE: i64 = 5000i64; + SEARCH_INCREMENTAL_INDEX_BATCH_DELAY_SECONDS: u64 = 5u64; + SEARCH_INCREMENTAL_INDEX_BATCH_MAX_SIZE: usize = 1000usize; TYPESENSE_URL: String = "http://localhost:8108"; TYPESENSE_API_KEY: String = "modrinth"; TYPESENSE_INDEX_PREFIX: String = "labrinth"; + TYPESENSE_IMPORT_BATCH_SIZE: usize = 5000usize; + TYPESENSE_DELETE_BATCH_SIZE: usize = 10_000usize; // storage STORAGE_BACKEND: crate::file_hosting::FileHostKind = crate::file_hosting::FileHostKind::Local; diff --git a/apps/labrinth/src/lib.rs b/apps/labrinth/src/lib.rs index e05e536fd2..76d2003479 100644 --- a/apps/labrinth/src/lib.rs +++ b/apps/labrinth/src/lib.rs @@ -131,30 +131,6 @@ pub fn app_setup( )); if enable_background_tasks { - // The interval in seconds at which the local database is indexed - // for searching. Defaults to 1 hour if unset. - let local_index_interval = - Duration::from_secs(ENV.LOCAL_INDEX_INTERVAL); - let pool_ref = pool.clone(); - let redis_pool_ref = redis_pool.clone(); - let search_backend_ref = search_backend.clone(); - scheduler.run(local_index_interval, move || { - let pool_ref = pool_ref.clone(); - let redis_pool_ref = redis_pool_ref.clone(); - let search_backend = search_backend_ref.clone(); - async move { - if let Err(err) = background_task::index_search( - pool_ref, - redis_pool_ref, - search_backend, - ) - .await - { - warn!("Failed to index search: {err:?}"); - } - } - }); - // Changes statuses of scheduled projects/versions let pool_ref = pool.clone(); // TODO: Clear cache when these are run diff --git a/apps/labrinth/src/queue/server_ping.rs b/apps/labrinth/src/queue/server_ping.rs index 5b6305961b..1941da8f40 100644 --- a/apps/labrinth/src/queue/server_ping.rs +++ b/apps/labrinth/src/queue/server_ping.rs @@ -181,8 +181,9 @@ impl ServerPingQueue { None, &self.redis, ); - let queue_search = - self.incremental_search_queue.push(*project_id); + let queue_search = self + .incremental_search_queue + .push_project_change(*project_id); let (clear_cache_result, _) = join(clear_cache, queue_search).await; diff --git a/apps/labrinth/src/routes/internal/admin.rs b/apps/labrinth/src/routes/internal/admin.rs index acd899e618..30905800f3 100644 --- a/apps/labrinth/src/routes/internal/admin.rs +++ b/apps/labrinth/src/routes/internal/admin.rs @@ -8,7 +8,7 @@ use crate::queue::analytics::AnalyticsQueue; use crate::queue::session::AuthQueue; use crate::routes::ApiError; use crate::search::SearchBackend; -use crate::search::incremental::consume::reindex_project; +use crate::search::incremental::consume::reindex_project_document; use crate::util::date::get_current_tenths_of_ms; use crate::util::error::Context; use crate::util::guards::admin_key_guard; @@ -330,7 +330,7 @@ pub async fn force_reindex( ) -> Result { let redis = redis.get_ref(); search_backend - .index_projects(pool.as_ref().clone(), redis.clone()) + .rebuild_index(pool.as_ref().clone(), redis.clone()) .await .wrap_internal_err("failed to index projects")?; Ok(HttpResponse::NoContent().finish()) @@ -355,7 +355,7 @@ pub async fn force_reindex_project( search_backend: web::Data, ) -> Result { let (project_id,) = path.into_inner(); - reindex_project( + reindex_project_document( pool.as_ref(), redis.as_ref(), search_backend.as_ref(), diff --git a/apps/labrinth/src/routes/v2/versions.rs b/apps/labrinth/src/routes/v2/versions.rs index 30ae89ac3c..353bb450f2 100644 --- a/apps/labrinth/src/routes/v2/versions.rs +++ b/apps/labrinth/src/routes/v2/versions.rs @@ -11,7 +11,7 @@ use crate::models::projects::{ use crate::models::v2::projects::LegacyVersion; use crate::queue::session::AuthQueue; use crate::routes::{v2_reroute, v3}; -use crate::search::{SearchBackend, SearchState}; +use crate::search::SearchState; use actix_web::{HttpRequest, HttpResponse, delete, get, patch, web}; use serde::{Deserialize, Serialize}; use validator::Validate; @@ -488,7 +488,6 @@ pub async fn version_delete( pool: web::Data, redis: web::Data, session_queue: web::Data, - search_backend: web::Data, search_state: web::Data, ) -> Result { // Returns NoContent, so we don't need to convert the response @@ -498,7 +497,6 @@ pub async fn version_delete( pool, redis, session_queue, - search_backend, search_state, ) .await diff --git a/apps/labrinth/src/routes/v3/projects.rs b/apps/labrinth/src/routes/v3/projects.rs index 578cd74e3f..18afc646e6 100644 --- a/apps/labrinth/src/routes/v3/projects.rs +++ b/apps/labrinth/src/routes/v3/projects.rs @@ -85,7 +85,11 @@ pub async fn clear_project_cache_and_queue_search( redis, ) .await?; - search_state.queue.push(project_id.into()).await; + + search_state + .queue + .push_project_change(project_id.into()) + .await; Ok(()) } @@ -1053,10 +1057,10 @@ pub async fn project_edit_internal( edit: Option>, mut component: &mut Option, perms: ProjectPermissions, - ) -> Result<(), ApiError> { + ) -> Result { let Some(edit) = edit else { // component is not specified in the input JSON - leave alone - return Ok(()); + return Ok(false); }; if !perms.contains(ProjectPermissions::EDIT_DETAILS) { @@ -1095,10 +1099,12 @@ pub async fn project_edit_internal( } } - Ok(()) + Ok(true) } - update( + let mut reindex_versions = false; + + reindex_versions |= update( &mut transaction, id, new_project.minecraft_server, @@ -1106,7 +1112,7 @@ pub async fn project_edit_internal( perms, ) .await?; - update( + reindex_versions |= update( &mut transaction, id, new_project.minecraft_java_server, @@ -1114,7 +1120,7 @@ pub async fn project_edit_internal( perms, ) .await?; - update( + reindex_versions |= update( &mut transaction, id, new_project.minecraft_bedrock_server, @@ -1167,14 +1173,31 @@ pub async fn project_edit_internal( transaction.commit().await?; - clear_project_cache_and_queue_search( - &redis, - &search_state, - project_item.inner.id, - project_item.inner.slug, - None, - ) - .await?; + if reindex_versions { + db_models::DBProject::clear_cache( + project_item.inner.id, + project_item.inner.slug, + None, + &redis, + ) + .await?; + search_state + .queue + .push_version_changes( + project_item.inner.id.into(), + project_item.versions.iter().copied().map(VersionId::from), + ) + .await; + } else { + clear_project_cache_and_queue_search( + &redis, + &search_state, + project_item.inner.id, + project_item.inner.slug, + None, + ) + .await?; + } // Remove no longer searchable projects from search index if let (true, Some(false)) = ( @@ -1182,16 +1205,9 @@ pub async fn project_edit_internal( new_project.status.map(|status| status.is_searchable()), ) { search_state - .backend - .remove_documents( - &project_item - .versions - .into_iter() - .map(|x| x.into()) - .collect::>(), - ) - .await - .wrap_internal_err("failed to remove documents")?; + .queue + .push_project_removal(project_item.inner.id.into()) + .await; } Ok(HttpResponse::NoContent().body("")) @@ -2791,27 +2807,18 @@ pub async fn project_delete_internal( .await .wrap_internal_err("failed to commit transaction")?; - search_state - .backend - .remove_documents( - &project - .versions - .into_iter() - .map(|x| x.into()) - .collect::>(), - ) - .await - .wrap_internal_err("failed to remove project version documents")?; - if result.is_some() { - clear_project_cache_and_queue_search( - &redis, - &search_state, + db_models::DBProject::clear_cache( project.inner.id, project.inner.slug, None, + &redis, ) .await?; + search_state + .queue + .push_project_removal(project.inner.id.into()) + .await; Ok(()) } else { Err(ApiError::NotFound) diff --git a/apps/labrinth/src/routes/v3/version_creation.rs b/apps/labrinth/src/routes/v3/version_creation.rs index 2d22d62d13..dea4235d6f 100644 --- a/apps/labrinth/src/routes/v3/version_creation.rs +++ b/apps/labrinth/src/routes/v3/version_creation.rs @@ -184,19 +184,20 @@ pub async fn version_create( if let Err(e) = rollback_result { return Err(e.into()); } - } else if let Ok((_, project_id)) = &result { + } else if let Ok((_, project_id, version_id)) = &result { transaction.commit().await?; - super::projects::clear_project_cache_and_queue_search( - &redis, - &search_state, - *project_id, - None, - Some(true), - ) - .await?; + models::DBProject::clear_cache(*project_id, None, Some(true), &redis) + .await?; + search_state + .queue + .push_version_changes( + (*project_id).into(), + [VersionId::from(*version_id)], + ) + .await; } - result.map(|(response, _)| response) + result.map(|(response, _, _)| response) } #[allow(clippy::too_many_arguments)] @@ -210,7 +211,8 @@ async fn version_create_inner( pool: &PgPool, session_queue: &AuthQueue, http: &reqwest::Client, -) -> Result<(HttpResponse, models::DBProjectId), CreateError> { +) -> Result<(HttpResponse, models::DBProjectId, models::DBVersionId), CreateError> +{ let mut initial_version_data = None; let mut version_builder = None; let mut selected_loaders = None; @@ -568,7 +570,11 @@ async fn version_create_inner( } } - Ok((HttpResponse::Ok().json(response), project_id)) + Ok(( + HttpResponse::Ok().json(response), + project_id, + models::DBVersionId::from(version_id), + )) } /// Add files to an existing version. @@ -635,17 +641,18 @@ pub async fn upload_file_to_version( let mut transaction = client.begin().await?; let mut uploaded_files = Vec::new(); - let version_id = models::DBVersionId::from(url_data.into_inner().0); + let version_id = url_data.into_inner().0; + let db_version_id = models::DBVersionId::from(version_id); let result = upload_file_to_version_inner( req, &mut payload, - client, + client.clone(), &mut transaction, redis.clone(), &**file_host, &mut uploaded_files, - version_id, + db_version_id, &session_queue, &http, ) @@ -665,14 +672,12 @@ pub async fn upload_file_to_version( } } else if let Ok((_, project_id)) = &result { transaction.commit().await?; - super::projects::clear_project_cache_and_queue_search( - &redis, - &search_state, - *project_id, - None, - Some(true), - ) - .await?; + models::DBProject::clear_cache(*project_id, None, Some(true), &redis) + .await?; + search_state + .queue + .push_version_changes((*project_id).into(), [version_id]) + .await; } result.map(|(response, _)| response) diff --git a/apps/labrinth/src/routes/v3/versions.rs b/apps/labrinth/src/routes/v3/versions.rs index 9d4f33e602..55b922d7d9 100644 --- a/apps/labrinth/src/routes/v3/versions.rs +++ b/apps/labrinth/src/routes/v3/versions.rs @@ -27,8 +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::{SearchBackend, SearchState}; -use crate::util::error::Context; +use crate::search::SearchState; use crate::util::img; use crate::util::validate::validation_errors_to_string; use actix_web::{HttpRequest, HttpResponse, delete, get, patch, web}; @@ -853,14 +852,20 @@ pub async fn version_edit_helper( transaction.commit().await?; database::models::DBVersion::clear_cache(&version_item, &redis) .await?; - super::projects::clear_project_cache_and_queue_search( - &redis, - &search_state, + database::models::DBProject::clear_cache( version_item.inner.project_id, None, Some(true), + &redis, ) .await?; + search_state + .queue + .push_version_changes( + version_item.inner.project_id.into(), + [VersionId::from(version_item.inner.id)], + ) + .await; Ok(HttpResponse::NoContent().body("")) } else { Err(ApiError::CustomAuthentication( @@ -1129,19 +1134,9 @@ pub async fn version_delete_route( pool: web::Data, redis: web::Data, session_queue: web::Data, - search_backend: web::Data, search_state: web::Data, ) -> Result { - version_delete( - req, - info, - pool, - redis, - session_queue, - search_backend, - search_state, - ) - .await + version_delete(req, info, pool, redis, session_queue, search_state).await } pub async fn version_delete( @@ -1150,7 +1145,6 @@ pub async fn version_delete( pool: web::Data, redis: web::Data, session_queue: web::Data, - search_backend: web::Data, search_state: web::Data, ) -> Result { let user = get_user_from_headers( @@ -1246,18 +1240,20 @@ pub async fn version_delete( transaction.commit().await?; - super::projects::clear_project_cache_and_queue_search( - &redis, - &search_state, + database::models::DBProject::clear_cache( version.inner.project_id, None, Some(true), + &redis, ) .await?; - search_backend - .remove_documents(&[version.inner.id.into()]) - .await - .wrap_internal_err("failed to remove documents")?; + search_state + .queue + .push_version_changes( + version.inner.project_id.into(), + [VersionId::from(version.inner.id)], + ) + .await; if result.is_some() { Ok(HttpResponse::NoContent().body("")) } else { diff --git a/apps/labrinth/src/search/backend/common.rs b/apps/labrinth/src/search/backend/common.rs index 5192a83dc3..9589275186 100644 --- a/apps/labrinth/src/search/backend/common.rs +++ b/apps/labrinth/src/search/backend/common.rs @@ -50,14 +50,7 @@ pub enum SearchIndex { MinecraftJavaServerPlayersOnline, } -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum SearchIndexName { - Projects, - ProjectsFiltered, -} - pub struct SearchSort { - pub index_name: SearchIndexName, pub index: SearchIndex, } @@ -65,16 +58,12 @@ pub fn parse_search_index( index: &str, new_filters: Option<&str>, ) -> Result { - let projects_name = SearchIndexName::Projects; - let projects_filtered_name = SearchIndexName::ProjectsFiltered; - // TODO: this is a dumb hack, the frontend should pass the project type it's filtering directly let is_server = new_filters .is_some_and(|f| f.contains("project_types = minecraft_java_server")); Ok(match index { "relevance" => SearchSort { - index_name: projects_name, index: if is_server { SearchIndex::MinecraftJavaServerVerifiedPlays2w } else { @@ -82,27 +71,21 @@ pub fn parse_search_index( }, }, "downloads" => SearchSort { - index_name: projects_filtered_name, index: SearchIndex::Downloads, }, "follows" => SearchSort { - index_name: projects_name, index: SearchIndex::Follows, }, "updated" | "date_modified" => SearchSort { - index_name: projects_name, index: SearchIndex::Updated, }, "newest" | "date_created" => SearchSort { - index_name: projects_name, index: SearchIndex::Newest, }, "minecraft_java_server.verified_plays_2w" => SearchSort { - index_name: projects_name, index: SearchIndex::MinecraftJavaServerVerifiedPlays2w, }, "minecraft_java_server.ping.data.players_online" => SearchSort { - index_name: projects_name, index: SearchIndex::MinecraftJavaServerPlayersOnline, }, i => return Err(ApiError::Request(eyre!("invalid index '{i}'"))), diff --git a/apps/labrinth/src/search/backend/mod.rs b/apps/labrinth/src/search/backend/mod.rs index 307cde1040..544cf2a557 100644 --- a/apps/labrinth/src/search/backend/mod.rs +++ b/apps/labrinth/src/search/backend/mod.rs @@ -2,7 +2,7 @@ mod common; pub mod typesense; pub use common::{ - ParsedSearchRequest, SearchIndex, SearchIndexName, SearchSort, - combined_search_filters, parse_search_index, parse_search_request, + ParsedSearchRequest, SearchIndex, SearchSort, combined_search_filters, + parse_search_index, parse_search_request, }; pub use typesense::{Typesense, TypesenseConfig}; diff --git a/apps/labrinth/src/search/backend/typesense/filter_rewrite.rs b/apps/labrinth/src/search/backend/typesense/filter_rewrite.rs new file mode 100644 index 0000000000..ca8f221ddc --- /dev/null +++ b/apps/labrinth/src/search/backend/typesense/filter_rewrite.rs @@ -0,0 +1,442 @@ +use std::sync::LazyLock; + +use eyre::{Result, WrapErr, eyre}; +use itertools::Itertools; +use regex::Regex; +use serde_json::Value; + +use crate::search::SearchField; + +#[derive(Clone, Default)] +struct JoinedFilterClause { + project: Vec, + version: Vec, +} + +pub(super) fn rewrite_filter_for_join( + filter: &str, + versions_collection: &str, +) -> Result { + const MAX_CLAUSES: usize = 256; + + fn parse(expression: &str) -> Result> { + let expression = trim_outer_parentheses(expression.trim()); + + let or_parts = split_top_level(expression, "||"); + if or_parts.len() > 1 { + let mut clauses = Vec::new(); + for part in or_parts { + clauses.extend(parse(part)?); + if clauses.len() > MAX_CLAUSES { + return Err(eyre!( + "search filter has too many boolean clauses" + )); + } + } + return Ok(clauses); + } + + let and_parts = split_top_level(expression, "&&"); + if and_parts.len() > 1 { + let mut clauses = vec![JoinedFilterClause::default()]; + for part in and_parts { + let right = parse(part)?; + if clauses.len().saturating_mul(right.len()) > MAX_CLAUSES { + return Err(eyre!( + "search filter has too many boolean clauses" + )); + } + clauses = clauses + .into_iter() + .cartesian_product(right) + .map(|(mut left, right)| { + left.project.extend(right.project); + left.version.extend(right.version); + left + }) + .collect(); + } + return Ok(clauses); + } + + let field = filter_field(expression).ok_or_else(|| { + eyre!("could not determine filter field in `{expression}`") + })?; + let mut clause = JoinedFilterClause::default(); + if field == "categories" { + let project_expression = + expression.replacen("categories", "project_categories", 1); + if is_negative_filter(expression) { + clause.project.push(project_expression); + clause.version.push(expression.to_string()); + Ok(vec![clause]) + } else { + Ok(vec![ + JoinedFilterClause { + project: vec![project_expression], + version: Vec::new(), + }, + JoinedFilterClause { + project: Vec::new(), + version: vec![expression.to_string()], + }, + ]) + } + } else { + if is_version_filter_field(field) { + clause.version.push(expression.to_string()); + } else { + clause.project.push(expression.to_string()); + } + Ok(vec![clause]) + } + } + + let clauses = parse(filter)?; + Ok(clauses + .into_iter() + .map(|clause| { + let mut parts = clause.project; + if !clause.version.is_empty() { + parts.push(format!( + "${versions_collection}({})", + clause.version.join(" && ") + )); + } + if parts.len() == 1 { + parts.pop().unwrap_or_default() + } else { + format!("({})", parts.join(" && ")) + } + }) + .join(" || ")) +} + +fn is_version_filter_field(field: &str) -> bool { + ::iter().any(|search_field| { + search_field.is_version_field() + && search_field.typesense_spec().path == field + }) +} + +fn is_negative_filter(expression: &str) -> bool { + expression + .split_once(':') + .is_some_and(|(_, value)| value.trim_start().starts_with("!=")) +} + +fn filter_field(expression: &str) -> Option<&str> { + let operator = expression.find(':')?; + let field = expression[..operator].trim(); + (!field.is_empty() + && field.chars().all(|character| { + character.is_ascii_alphanumeric() || "_.".contains(character) + })) + .then_some(field) +} + +fn trim_outer_parentheses(mut expression: &str) -> &str { + while expression.starts_with('(') + && expression.ends_with(')') + && matching_outer_parentheses(expression) + { + expression = expression[1..expression.len() - 1].trim(); + } + expression +} + +fn matching_outer_parentheses(expression: &str) -> bool { + let mut depth = 0; + let mut quote = None; + let mut escaped = false; + + for (index, character) in expression.char_indices() { + if escaped { + escaped = false; + continue; + } + if character == '\\' { + escaped = true; + continue; + } + if let Some(active_quote) = quote { + if character == active_quote { + quote = None; + } + continue; + } + if matches!(character, '\'' | '"' | '`') { + quote = Some(character); + continue; + } + match character { + '(' => depth += 1, + ')' => { + depth -= 1; + if depth == 0 && index + character.len_utf8() < expression.len() + { + return false; + } + } + _ => {} + } + } + + depth == 0 +} + +fn split_top_level<'a>(expression: &'a str, operator: &str) -> Vec<&'a str> { + let mut parts = Vec::new(); + let mut start = 0; + let mut parentheses = 0; + let mut brackets = 0; + let mut quote = None; + let mut escaped = false; + let bytes = expression.as_bytes(); + let mut index = 0; + + while index < bytes.len() { + let character = expression[index..].chars().next().unwrap_or_default(); + let width = character.len_utf8(); + if escaped { + escaped = false; + index += width; + continue; + } + if character == '\\' { + escaped = true; + index += width; + continue; + } + if let Some(active_quote) = quote { + if character == active_quote { + quote = None; + } + index += width; + continue; + } + if matches!(character, '\'' | '"' | '`') { + quote = Some(character); + index += width; + continue; + } + match character { + '(' => parentheses += 1, + ')' => parentheses -= 1, + '[' => brackets += 1, + ']' => brackets -= 1, + _ => {} + } + + if parentheses == 0 + && brackets == 0 + && expression[index..].starts_with(operator) + { + parts.push(expression[start..index].trim()); + index += operator.len(); + start = index; + continue; + } + index += width; + } + + if parts.is_empty() { + vec![expression] + } else { + parts.push(expression[start..].trim()); + parts + } +} + +/// Translates a Meilisearch filter expression into Typesense `filter_by` +/// syntax. +/// +/// Transformations (applied in order): +/// 1. `field (NOT )IN [v1, v2]` → `field:[v1, v2]` / `field:!=[v1, v2]` +/// 2. `field op value` for op ∈ {`!=`, `>=`, `<=`, `>`, `<`, `=`} +/// → `field:op value` +/// 3. `AND` / `OR` (case-insensitive) → `&&` / `||` +pub(super) fn meili_to_typesense(filter: &str) -> String { + static IN_RE: LazyLock = LazyLock::new(|| { + Regex::new( + r"(?i)\b([a-zA-Z_.][a-zA-Z0-9_.]*)\s+(NOT\s+)?IN\s*\[([^\]]*)\]", + ) + .expect("valid regex") + }); + static EXISTS_RE: LazyLock = LazyLock::new(|| { + Regex::new(r"(?i)\b([a-zA-Z_.][a-zA-Z0-9_.]*)\s+(NOT\s+)?EXISTS\b") + .expect("valid regex") + }); + static CMP_RE: LazyLock = LazyLock::new(|| { + Regex::new(r"([a-zA-Z_.][a-zA-Z0-9_.]*)\s*(!=|>=|<=|>|<|=)\s*") + .expect("valid regex") + }); + static AND_RE: LazyLock = + LazyLock::new(|| Regex::new(r"(?i)\bAND\b").expect("valid regex")); + static OR_RE: LazyLock = + LazyLock::new(|| Regex::new(r"(?i)\bOR\b").expect("valid regex")); + + // Step 1 – IN / NOT IN + let s = IN_RE.replace_all(filter, |caps: ®ex::Captures<'_>| { + let field = caps.get(1).map(|m| m.as_str()).unwrap_or_default(); + let is_not = caps.get(2).is_some(); + let values = caps.get(3).map(|m| m.as_str()).unwrap_or_default(); + if is_not { + format!("{field}:!=[{values}]") + } else { + format!("{field}:[{values}]") + } + }); + + let s = EXISTS_RE.replace_all(&s, |caps: ®ex::Captures<'_>| { + let field = caps.get(1).map(|m| m.as_str()).unwrap_or_default(); + let is_not = caps.get(2).is_some(); + + match field { + "minecraft_java_server.ping.data" => format!( + "minecraft_java_server.is_online:= {}", + if is_not { "false" } else { "true" } + ), + _ => caps + .get(0) + .map(|m| m.as_str()) + .unwrap_or_default() + .to_string(), + } + }); + + // Step 2 – comparison operators (field op value → field:op value). + let s = CMP_RE.replace_all(&s, |caps: ®ex::Captures<'_>| { + let field = caps.get(1).map(|m| m.as_str()).unwrap_or_default(); + let op = caps.get(2).map(|m| m.as_str()).unwrap_or_default(); + format!("{field}:{op} ") + }); + + // Step 3 – logical operators + let s = AND_RE.replace_all(&s, " && "); + let s = OR_RE.replace_all(&s, " || "); + s.into_owned() +} + +/// Converts the legacy Meilisearch `facets` JSON array into a Typesense +/// `filter_by` string. The outer array items are AND-ed together; the inner +/// array items are OR-ed together. +pub(super) fn facets_to_typesense(facets_json: &str) -> Result { + let facets = serde_json::from_str::>>(facets_json) + .wrap_err("failed to parse facets JSON")?; + + let and_parts: Vec = facets + .into_iter() + .map(|or_group| { + let or_parts: Vec = or_group + .into_iter() + .map(|facet| { + let conditions: Vec = if facet.is_array() { + serde_json::from_value::>(facet) + .unwrap_or_default() + } else { + vec![ + serde_json::from_value::(facet) + .unwrap_or_default(), + ] + }; + let and_conds: Vec = conditions + .into_iter() + .map(|condition| { + condition_to_typesense_filter(&condition) + }) + .collect(); + if and_conds.len() == 1 { + and_conds.into_iter().next().unwrap_or_default() + } else { + format!("({})", and_conds.join(" && ")) + } + }) + .collect(); + if or_parts.len() == 1 { + or_parts.into_iter().next().unwrap_or_default() + } else { + format!("({})", or_parts.join(" || ")) + } + }) + .collect(); + + Ok(and_parts.join(" && ")) +} + +/// Converts a single facet condition such as `"categories:mods"`, +/// `"categories=mods"`, or `"downloads!=100"` into a Typesense filter clause. +fn condition_to_typesense_filter(condition: &str) -> String { + // Match multi-character operators before their single-character prefixes, + // and range/inequality operators before the plain `=` equality arm. + for operator in ["!=", ">=", "<=", ">", "<"] { + if let Some((field, value)) = condition.split_once(operator) { + return format!("{}:{} {}", field.trim(), operator, value.trim()); + } + } + if let Some((field, value)) = condition.split_once(':') { + return format!("{}:= {}", field.trim(), value.trim()); + } + if let Some((field, value)) = condition.split_once('=') { + return format!("{}:= {}", field.trim(), value.trim()); + } + condition.to_string() +} + +#[cfg(test)] +mod tests { + use super::rewrite_filter_for_join; + + #[test] + fn project_filters_do_not_join_versions() { + assert_eq!( + rewrite_filter_for_join("license:= MIT", "versions").unwrap(), + "license:= MIT" + ); + } + + #[test] + fn correlated_version_filters_share_one_join() { + assert_eq!( + rewrite_filter_for_join( + "categories:= fabric && game_versions:= 1.21", + "versions", + ) + .unwrap(), + "(project_categories:= fabric && $versions(game_versions:= 1.21)) || $versions(categories:= fabric && game_versions:= 1.21)" + ); + } + + #[test] + fn project_and_version_filters_are_partitioned() { + assert_eq!( + rewrite_filter_for_join( + "license:= MIT && categories:= fabric", + "versions", + ) + .unwrap(), + "(license:= MIT && project_categories:= fabric) || (license:= MIT && $versions(categories:= fabric))" + ); + } + + #[test] + fn mixed_boolean_filters_preserve_version_correlation() { + assert_eq!( + rewrite_filter_for_join( + "(license:= MIT || categories:= fabric) && game_versions:= 1.21", + "versions", + ) + .unwrap(), + "(license:= MIT && $versions(game_versions:= 1.21)) || (project_categories:= fabric && $versions(game_versions:= 1.21)) || $versions(categories:= fabric && game_versions:= 1.21)" + ); + } + + #[test] + fn negative_categories_require_project_and_version_exclusion() { + assert_eq!( + rewrite_filter_for_join("categories:!= fabric", "versions") + .unwrap(), + "(project_categories:!= fabric && $versions(categories:!= fabric))" + ); + } +} diff --git a/apps/labrinth/src/search/backend/typesense/mod.rs b/apps/labrinth/src/search/backend/typesense/mod.rs index 6a9d1b91f9..28d628644c 100644 --- a/apps/labrinth/src/search/backend/typesense/mod.rs +++ b/apps/labrinth/src/search/backend/typesense/mod.rs @@ -1,30 +1,37 @@ use std::sync::LazyLock; -use ariadne::ids::base62_impl::to_base62; use async_trait::async_trait; use eyre::{Result, eyre}; -use regex::Regex; +use itertools::Itertools; use reqwest::Method; use serde::{Deserialize, Serialize}; use serde_json::{Value, json}; -use tracing::{info, warn}; +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, SearchIndexName, combined_search_filters, parse_search_index, + SearchIndex, combined_search_filters, parse_search_index, parse_search_request, }; use crate::search::indexing::index_local; use crate::search::{ - ResultSearchProject, SearchBackend, SearchField, SearchRequest, - SearchResults, TasksCancelFilter, UploadSearchProject, + ResultSearchProject, SearchBackend, SearchField, SearchIndexUpdate, + SearchRequest, SearchResults, TasksCancelFilter, UploadSearchProject, + UploadSearchVersion, }; use crate::util::error::Context; +use self::filter_rewrite::{ + facets_to_typesense, meili_to_typesense, rewrite_filter_for_join, +}; + +mod filter_rewrite; + +const DELETE_FILTER_ID_BATCH_SIZE: usize = 256; + #[derive(Debug, Clone)] pub struct TypesenseConfig { pub url: String, @@ -32,6 +39,8 @@ pub struct TypesenseConfig { pub index_prefix: String, pub meta_namespace: String, pub index_chunk_size: i64, + pub import_batch_size: usize, + pub delete_batch_size: usize, } #[derive(Serialize, Deserialize, Debug, Clone)] @@ -160,6 +169,8 @@ impl TypesenseConfig { index_prefix: ENV.TYPESENSE_INDEX_PREFIX.clone(), meta_namespace: meta_namespace.unwrap_or_default(), index_chunk_size: ENV.SEARCH_INDEX_CHUNK_SIZE, + import_batch_size: ENV.TYPESENSE_IMPORT_BATCH_SIZE, + delete_batch_size: ENV.TYPESENSE_DELETE_BATCH_SIZE, } } @@ -235,6 +246,22 @@ impl TypesenseClient { Ok(()) } + async fn delete_alias_if_exists(&self, alias: &str) -> Result<()> { + let resp = self + .request(Method::DELETE, &format!("/aliases/{alias}")) + .send() + .await + .wrap_err("failed to DELETE Typesense alias")?; + if resp.status() == reqwest::StatusCode::NOT_FOUND { + return Ok(()); + } + if !resp.status().is_success() { + let body = resp.json::().await.unwrap_or_default(); + return Err(eyre!("failed to delete alias `{alias}`: {body}")); + } + Ok(()) + } + async fn collection_exists(&self, name: &str) -> Result { let resp = self .request(Method::GET, &format!("/collections/{name}")) @@ -300,21 +327,35 @@ impl TypesenseClient { )); } // Typesense always returns HTTP 200; individual lines signal per-doc success. - let error_count = body + let failures = body .lines() - .filter(|l| !l.trim().is_empty()) - .filter(|l| { - serde_json::from_str::(l) - .ok() - .and_then(|v| v["success"].as_bool()) - .map(|ok| !ok) - .unwrap_or(false) + .filter(|line| !line.trim().is_empty()) + .filter_map(|line| match serde_json::from_str::(line) { + Ok(result) if result["success"].as_bool() == Some(true) => None, + Ok(result) => Some( + result["error"] + .as_str() + .unwrap_or( + "Typesense returned an unsuccessful import result", + ) + .to_string(), + ), + Err(err) => Some(format!( + "failed to parse Typesense import result: {err}" + )), }) - .count(); - if error_count > 0 { - warn!( - "{error_count} document(s) failed to import into `{collection}`" - ); + .collect::>(); + if !failures.is_empty() { + let failure_count = failures.len(); + let errors = failures + .into_iter() + .unique() + .take(10) + .collect::>() + .join("; "); + return Err(eyre!( + "{failure_count} document(s) failed to import into `{collection}`: {errors}" + )); } Ok(()) } @@ -323,12 +364,13 @@ impl TypesenseClient { &self, collection: &str, filter_by: &str, + batch_size: usize, ) -> Result<()> { let resp = self .request( Method::DELETE, &format!( - "/collections/{collection}/documents?filter_by={}&batch_size=1000", + "/collections/{collection}/documents?filter_by={}&batch_size={batch_size}", urlencoding::encode(filter_by) ), ) @@ -363,6 +405,18 @@ pub struct TypesenseFieldSpec { } impl SearchField { + const fn is_version_field(self) -> bool { + matches!( + self, + Self::Categories + | Self::ProjectTypes + | Self::Environment + | Self::GameVersions + | Self::ClientSide + | Self::ServerSide + ) + } + pub const fn typesense_spec(self) -> TypesenseFieldSpec { match self { SearchField::Categories => TypesenseFieldSpec { @@ -564,7 +618,7 @@ impl Typesense { Self { config, client } } - fn collection_schema(name: &str) -> Value { + fn project_collection_schema(name: &str) -> Value { let mut fields = vec![ json!({"name": "summary", "type": "string", "facet": false}), json!({"name": "slug", "type": "string", "facet": false}), @@ -580,6 +634,7 @@ impl Typesense { json!({"name": "minecraft_java_server.is_online", "type": "bool", "sort": true, "optional": true}), json!({"name": "minecraft_java_server.ping.data.players_online", "type": "int32", "sort": true, "optional": true}), json!({"name": "dependencies", "type": "object[]", "optional": true}), + json!({"name": "project_categories", "type": "string[]", "facet": true, "optional": true}), ]; fields.extend(TYPESENSE_SEARCH_FIELDS.iter().cloned()); @@ -591,6 +646,50 @@ impl Typesense { }) } + fn version_collection_schema( + name: &str, + projects_collection: &str, + ) -> Value { + use strum::IntoEnumIterator; + + let mut fields = SearchField::iter() + .filter(|field| field.is_version_field()) + .map(|field| { + let spec = field.typesense_spec(); + json!({ + "name": spec.path, + "type": spec.ty, + "facet": spec.facet, + "optional": spec.optional, + "token_separators": spec.token_separators, + }) + }) + .collect::>(); + fields.extend([ + json!({ + "name": "version_id", + "type": "string", + }), + json!({ + "name": "project_id", + "type": "string", + "reference": format!("{projects_collection}.id"), + "async_reference": true, + "cascade_delete": false, + }), + json!({ + "name": "version_published_timestamp", + "type": "int64", + "sort": true, + }), + ]); + + json!({ + "name": name, + "fields": fields, + }) + } + fn text_match_sort_field(request_config: &RequestConfig) -> String { match request_config.bucketing { Bucketing::Buckets(count) => { @@ -670,12 +769,7 @@ impl Typesense { request_config: &RequestConfig, ) -> Result<(String, String), ApiError> { let sort = parse_search_index(index, new_filters)?; - let alias = match sort.index_name { - SearchIndexName::Projects => self.config.get_alias_name("projects"), - SearchIndexName::ProjectsFiltered => { - self.config.get_alias_name("projects_filtered") - } - }; + let alias = self.config.get_alias_name("projects"); Ok((alias, self.get_sort_fields(sort.index, request_config))) } @@ -684,7 +778,10 @@ impl Typesense { /// Handles the new-style filter string, legacy facets JSON, and the legacy /// `filters`/`version` fields, translating each from Meilisearch filter /// syntax to Typesense filter syntax. - fn build_filter(info: &SearchRequest) -> Result, ApiError> { + fn build_filter( + info: &SearchRequest, + versions_collection: &str, + ) -> Result, ApiError> { let facet_part = if let Some(facets_json) = info.facets.as_deref() { Some( facets_to_typesense(facets_json) @@ -705,29 +802,113 @@ impl Typesense { let filter_part = new_filters_part.or(legacy_part); - Ok(match (facet_part, filter_part) { + let filter = match (facet_part, filter_part) { (Some(f), Some(l)) if !f.is_empty() && !l.is_empty() => { Some(format!("({f}) && ({l})")) } (Some(f), _) if !f.is_empty() => Some(f), (_, Some(l)) if !l.is_empty() => Some(l), _ => None, - }) + }; + + filter + .map(|filter| { + rewrite_filter_for_join(&filter, versions_collection) + .wrap_request_err("failed to rewrite search filter") + }) + .transpose() } - /// Ensures the alias and its backing collection both exist, creating them - /// when necessary so reads succeed before the first full index run. - async fn ensure_collection(&self, alias: &str) -> Result<()> { - if self.client.get_alias(alias).await?.is_some() { - return Ok(()); + async fn import_document_batches( + &self, + collections: &[String], + documents: &[T], + serialize: fn(&[T]) -> Result, + ) -> Result<()> { + let batch_size = self.config.import_batch_size.max(1); + + for batch in documents.chunks(batch_size) { + let jsonl = serialize(batch)?; + + for collection in collections { + info!( + collection, + document_count = batch.len(), + content_length_bytes = jsonl.len(), + "sending Typesense document import" + ); + self.client + .import_documents(collection, jsonl.clone()) + .await?; + } + } + + Ok(()) + } + + async fn existing_write_collections( + &self, + alias: &str, + ) -> Result> { + let mut collections = self + .client + .get_alias(alias) + .await? + .into_iter() + .collect_vec(); + + for collection in [ + self.config.get_next_collection_name(alias, true), + self.config.get_next_collection_name(alias, false), + ] { + if !collections.contains(&collection) + && self.client.collection_exists(&collection).await? + { + collections.push(collection); + } } - let name = self.config.get_next_collection_name(alias, false); - if !self.client.collection_exists(&name).await? { - self.client - .create_collection(&Self::collection_schema(&name)) - .await?; + + Ok(collections) + } + + async fn delete_ids_from_write_collections( + &self, + alias: &str, + field: &str, + ids: &[String], + ) -> Result<()> { + let collections = self.existing_write_collections(alias).await?; + for ids in ids.chunks(DELETE_FILTER_ID_BATCH_SIZE) { + let filter = format!("{field}:[{}]", ids.iter().join(", ")); + for collection in &collections { + self.client + .delete_documents_by_filter( + collection, + &filter, + self.config.delete_batch_size, + ) + .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?; + let shadow_alt = self.config.get_next_collection_name(&alias, true); + let shadow_current = + self.config.get_next_collection_name(&alias, false); + + self.client.delete_alias_if_exists(&alias).await?; + for collection in live + .into_iter() + .chain([shadow_alt, shadow_current]) + .unique() + { + self.client.delete_collection_if_exists(&collection).await?; } - self.client.upsert_alias(alias, &name).await?; + Ok(()) } } @@ -744,7 +925,8 @@ impl SearchBackend for Typesense { info.new_filters.as_deref(), &info.typesense_config, )?; - let filter_by = Self::build_filter(info)?; + let versions_alias = self.config.get_alias_name("versions"); + let filter_by = Self::build_filter(info, &versions_alias)?; let q = if parsed.query.is_empty() { "*" @@ -782,10 +964,6 @@ impl SearchBackend for Typesense { ("sort_by", sort_by.to_string()), ("page", parsed.page.to_string()), ("per_page", parsed.hits_per_page.to_string()), - ("group_by", "project_id".to_string()), - ("group_limit", "1".to_string()), - ("facet_by", "project_id".to_string()), - ("max_facet_values", "0".to_string()), ( "max_candidates", info.typesense_config.max_candidates.to_string(), @@ -801,6 +979,14 @@ impl SearchBackend for Typesense { } if let Some(filter) = &filter_by { params.push(("filter_by", filter.clone())); + if filter.contains(&format!("${versions_alias}(")) { + params.push(( + "include_fields", + format!( + "${versions_alias}(version_id, sort_by: version_published_timestamp:desc, limit:1, strategy: nest_array) as matching_versions" + ), + )); + } } let resp = self @@ -841,27 +1027,35 @@ impl SearchBackend for Typesense { .cloned() .unwrap_or(body); - let total_hits = body["facet_counts"] - .as_array() - .and_then(|facets| { - facets.iter().find(|facet| { - facet["field_name"].as_str() == Some("project_id") - }) - }) - .and_then(|facet| facet["stats"]["total_values"].as_u64()) - .unwrap_or_else(|| body["found"].as_u64().unwrap_or(0)) - as usize; + let total_hits = body["found"].as_u64().unwrap_or(0) as usize; - let hits = body["grouped_hits"] + let hits = body["hits"] .as_array() .cloned() .unwrap_or_default() .into_iter() - .filter_map(|group| { - let hit = group["hits"].as_array()?.first()?.clone(); + .filter_map(|hit| { let mut doc = hit.get("document")?.clone(); if let Some(obj) = doc.as_object_mut() { obj.remove("id"); + let matching_version_id = + obj.remove("matching_versions").and_then(|versions| { + versions + .as_array() + .and_then(|versions| versions.first()) + .or_else(|| { + versions.as_object().map(|_| &versions) + }) + .and_then(|version| version.get("version_id")) + .and_then(Value::as_str) + .map(ToString::to_string) + }); + if let Some(version_id) = matching_version_id { + obj.insert( + "version_id".to_string(), + Value::String(version_id), + ); + } } let metadata = info.show_metadata.then(|| { @@ -895,7 +1089,7 @@ impl SearchBackend for Typesense { }) } - async fn index_projects( + async fn rebuild_index( &self, ro_pool: PgPool, redis: RedisPool, @@ -903,54 +1097,53 @@ impl SearchBackend for Typesense { info!("starting project indexing"); let projects_alias = self.config.get_alias_name("projects"); - let filtered_alias = self.config.get_alias_name("projects_filtered"); - - // Guarantee current aliases exist so reads keep working during re-index. - self.ensure_collection(&projects_alias).await?; - self.ensure_collection(&filtered_alias).await?; + let versions_alias = self.config.get_alias_name("versions"); - // Toggle the shadow collection name between __current and __alt. let projects_current = self.client.get_alias(&projects_alias).await?; - let filtered_current = self.client.get_alias(&filtered_alias).await?; + let versions_current = self.client.get_alias(&versions_alias).await?; let projects_use_alt = !projects_current .as_deref() .is_some_and(|n| n.ends_with("__alt")); - let filtered_use_alt = !filtered_current + let versions_use_alt = !versions_current .as_deref() .is_some_and(|n| n.ends_with("__alt")); let projects_next = self .config .get_next_collection_name(&projects_alias, projects_use_alt); - let filtered_next = self + let versions_next = self .config - .get_next_collection_name(&filtered_alias, filtered_use_alt); + .get_next_collection_name(&versions_alias, versions_use_alt); - info!("shadow collections `{projects_next}` and `{filtered_next}`"); + info!("shadow collections `{projects_next}` and `{versions_next}`"); self.client - .delete_collection_if_exists(&projects_next) + .delete_collection_if_exists(&versions_next) .await?; self.client - .delete_collection_if_exists(&filtered_next) + .delete_collection_if_exists(&projects_next) .await?; self.client - .create_collection(&Self::collection_schema(&projects_next)) + .create_collection(&Self::project_collection_schema(&projects_next)) .await?; self.client - .create_collection(&Self::collection_schema(&filtered_next)) + .create_collection(&Self::version_collection_schema( + &versions_next, + &projects_next, + )) .await?; let mut cursor = 0_i64; let mut chunk_idx = 0_usize; - let mut total = 0_usize; + let mut total_projects = 0_usize; + let mut total_versions = 0_usize; loop { info!("fetching index chunk {chunk_idx}"); chunk_idx += 1; - let (uploads, next_cursor) = index_local( + let (documents, next_cursor) = index_local( &ro_pool, &redis, cursor, @@ -959,21 +1152,29 @@ impl SearchBackend for Typesense { .await .wrap_err("failed to fetch projects from local DB")?; - if uploads.is_empty() { + if documents.projects.is_empty() { info!( - "no more documents; indexed {total} in {chunk_idx} chunks" + "no more documents; indexed {total_projects} projects and {total_versions} versions in {chunk_idx} chunks" ); break; } - total += uploads.len(); + total_projects += documents.projects.len(); + total_versions += documents.versions.len(); cursor = next_cursor; - let jsonl = documents_to_jsonl(&uploads)?; - self.client - .import_documents(&projects_next, jsonl.clone()) - .await?; - self.client.import_documents(&filtered_next, jsonl).await?; + self.import_document_batches( + std::slice::from_ref(&projects_next), + &documents.projects, + documents_to_jsonl, + ) + .await?; + self.import_document_batches( + std::slice::from_ref(&versions_next), + &documents.versions, + version_documents_to_jsonl, + ) + .await?; } info!("swapping aliases"); @@ -981,122 +1182,119 @@ impl SearchBackend for Typesense { .upsert_alias(&projects_alias, &projects_next) .await?; self.client - .upsert_alias(&filtered_alias, &filtered_next) + .upsert_alias(&versions_alias, &versions_next) .await?; info!("cleaning up old collections"); - if let Some(old) = projects_current { + if let Some(old) = versions_current { self.client.delete_collection_if_exists(&old).await?; } - if let Some(old) = filtered_current { + if let Some(old) = projects_current { self.client.delete_collection_if_exists(&old).await?; } + self.delete_legacy_filtered_collections().await?; + info!("indexing complete"); Ok(()) } - async fn index_documents( + async fn apply_update( &self, - documents: &[UploadSearchProject], + update: SearchIndexUpdate<'_>, ) -> eyre::Result<()> { - if documents.is_empty() { - return Ok(()); - } - - let jsonl = documents_to_jsonl(documents)?; - for alias in [ - self.config.get_alias_name("projects"), - self.config.get_alias_name("projects_filtered"), - ] { - let live = self.client.get_alias(&alias).await?; - let shadow_alt = self.config.get_next_collection_name(&alias, true); - let shadow_current = - self.config.get_next_collection_name(&alias, false); + let projects_alias = self.config.get_alias_name("projects"); + let versions_alias = self.config.get_alias_name("versions"); - for collection in - live.into_iter().chain([shadow_alt, shadow_current]) - { - if self.client.collection_exists(&collection).await? { - self.client - .import_documents(&collection, jsonl.clone()) - .await?; - } - } + let removed_project_ids = update + .removed_projects + .iter() + .map(ToString::to_string) + .collect::>(); + 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?; } - Ok(()) - } - - async fn remove_project_documents( - &self, - ids: &[ProjectId], - ) -> 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::>(); + if !version_ids.is_empty() { + self.delete_ids_from_write_collections( + &versions_alias, + "id", + &version_ids, + ) + .await?; } - let id_list = ids + let project_ids = update + .projects .iter() - .map(|id| to_base62(id.0)) - .collect::>() - .join(", "); - let filter = format!("project_id:[{id_list}]"); - - for alias in [ - self.config.get_alias_name("projects"), - self.config.get_alias_name("projects_filtered"), - ] { - let live = self.client.get_alias(&alias).await?; - let shadow_alt = self.config.get_next_collection_name(&alias, true); - let shadow_current = - self.config.get_next_collection_name(&alias, false); - - for collection in - live.into_iter().chain([shadow_alt, shadow_current]) - { - if self.client.collection_exists(&collection).await? { - self.client - .delete_documents_by_filter(&collection, &filter) - .await?; - } - } + .map(|document| document.project_id.clone()) + .unique() + .collect::>(); + if !project_ids.is_empty() { + self.delete_ids_from_write_collections( + &projects_alias, + "project_id", + &project_ids, + ) + .await?; } - Ok(()) - } - async fn remove_documents(&self, ids: &[VersionId]) -> eyre::Result<()> { - if ids.is_empty() { - return Ok(()); + 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, + documents_to_jsonl, + ) + .await?; } - let id_list = ids - .iter() - .map(|id| to_base62(id.0)) - .collect::>() - .join(", "); - let filter = format!("id:[{id_list}]"); - - for alias in [ - self.config.get_alias_name("projects"), - self.config.get_alias_name("projects_filtered"), - ] { - // Delete from both the live collection and any shadow collections. - let live = self.client.get_alias(&alias).await?; - let shadow_alt = self.config.get_next_collection_name(&alias, true); - let shadow_current = - self.config.get_next_collection_name(&alias, false); - - for collection in - live.into_iter().chain([shadow_alt, shadow_current]) - { - if self.client.collection_exists(&collection).await? { - self.client - .delete_documents_by_filter(&collection, &filter) - .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_document_batches( + &collections, + update.versions, + version_documents_to_jsonl, + ) + .await?; } + + debug!("Done applying search index update"); Ok(()) } @@ -1115,7 +1313,7 @@ impl SearchBackend for Typesense { /// Serialises a batch of [`UploadSearchProject`]s to a JSONL string suitable /// for the Typesense bulk-import endpoint. Each document gets an `id` field -/// equal to `version_id` so Typesense can use it as the primary key. +/// equal to `project_id` so Typesense can use it as the primary key. fn documents_to_jsonl(uploads: &[UploadSearchProject]) -> Result { let mut out = String::new(); for upload in uploads { @@ -1123,7 +1321,7 @@ fn documents_to_jsonl(uploads: &[UploadSearchProject]) -> Result { .wrap_err("failed to serialise UploadSearchProject")?; if let Some(obj) = doc.as_object_mut() { let id = obj - .get("version_id") + .get("project_id") .and_then(Value::as_str) .unwrap_or_default() .to_string(); @@ -1147,135 +1345,21 @@ fn documents_to_jsonl(uploads: &[UploadSearchProject]) -> Result { Ok(out) } -/// Translates a Meilisearch filter expression into Typesense `filter_by` -/// syntax. -/// -/// Transformations (applied in order): -/// 1. `field (NOT )IN [v1, v2]` → `field:[v1, v2]` / `field:!=[v1, v2]` -/// 2. `field op value` for op ∈ {`!=`, `>=`, `<=`, `>`, `<`, `=`} -/// → `field:op value` -/// 3. `AND` / `OR` (case-insensitive) → `&&` / `||` -fn meili_to_typesense(filter: &str) -> String { - static IN_RE: LazyLock = LazyLock::new(|| { - Regex::new( - r"(?i)\b([a-zA-Z_.][a-zA-Z0-9_.]*)\s+(NOT\s+)?IN\s*\[([^\]]*)\]", - ) - .expect("valid regex") - }); - static EXISTS_RE: LazyLock = LazyLock::new(|| { - Regex::new(r"(?i)\b([a-zA-Z_.][a-zA-Z0-9_.]*)\s+(NOT\s+)?EXISTS\b") - .expect("valid regex") - }); - static CMP_RE: LazyLock = LazyLock::new(|| { - Regex::new(r"([a-zA-Z_.][a-zA-Z0-9_.]*)\s*(!=|>=|<=|>|<|=)\s*") - .expect("valid regex") - }); - static AND_RE: LazyLock = - LazyLock::new(|| Regex::new(r"(?i)\bAND\b").expect("valid regex")); - static OR_RE: LazyLock = - LazyLock::new(|| Regex::new(r"(?i)\bOR\b").expect("valid regex")); - - // Step 1 – IN / NOT IN - let s = IN_RE.replace_all(filter, |caps: ®ex::Captures<'_>| { - let field = caps.get(1).map(|m| m.as_str()).unwrap_or_default(); - let is_not = caps.get(2).is_some(); - let values = caps.get(3).map(|m| m.as_str()).unwrap_or_default(); - if is_not { - format!("{field}:!=[{values}]") - } else { - format!("{field}:[{values}]") - } - }); - - let s = EXISTS_RE.replace_all(&s, |caps: ®ex::Captures<'_>| { - let field = caps.get(1).map(|m| m.as_str()).unwrap_or_default(); - let is_not = caps.get(2).is_some(); - - match field { - "minecraft_java_server.ping.data" => format!( - "minecraft_java_server.is_online:= {}", - if is_not { "false" } else { "true" } - ), - _ => caps - .get(0) - .map(|m| m.as_str()) - .unwrap_or_default() - .to_string(), - } - }); - - // Step 2 – comparison operators (field op value → field:op value). - let s = CMP_RE.replace_all(&s, |caps: ®ex::Captures<'_>| { - let field = caps.get(1).map(|m| m.as_str()).unwrap_or_default(); - let op = caps.get(2).map(|m| m.as_str()).unwrap_or_default(); - format!("{field}:{op} ") - }); - - // Step 3 – logical operators - let s = AND_RE.replace_all(&s, " && "); - let s = OR_RE.replace_all(&s, " || "); - s.into_owned() -} - -/// Converts the legacy Meilisearch `facets` JSON array into a Typesense -/// `filter_by` string. The outer array items are AND-ed together; the inner -/// array items are OR-ed together. -fn facets_to_typesense(facets_json: &str) -> Result { - let facets = serde_json::from_str::>>(facets_json) - .wrap_err("failed to parse facets JSON")?; - - let and_parts: Vec = facets - .into_iter() - .map(|or_group| { - let or_parts: Vec = or_group - .into_iter() - .map(|facet| { - let conditions: Vec = if facet.is_array() { - serde_json::from_value::>(facet) - .unwrap_or_default() - } else { - vec![ - serde_json::from_value::(facet) - .unwrap_or_default(), - ] - }; - let and_conds: Vec = conditions - .into_iter() - .map(|c| condition_to_typesense_filter(&c)) - .collect(); - if and_conds.len() == 1 { - and_conds.into_iter().next().unwrap_or_default() - } else { - format!("({})", and_conds.join(" && ")) - } - }) - .collect(); - if or_parts.len() == 1 { - or_parts.into_iter().next().unwrap_or_default() - } else { - format!("({})", or_parts.join(" || ")) - } - }) - .collect(); - - Ok(and_parts.join(" && ")) -} - -/// Converts a single facet condition such as `"categories:mods"`, -/// `"categories=mods"`, or `"downloads!=100"` into a Typesense filter clause. -fn condition_to_typesense_filter(cond: &str) -> String { - // Match multi-character operators before their single-character prefixes, - // and range/inequality operators before the plain `=` equality arm. - for op in ["!=", ">=", "<=", ">", "<"] { - if let Some((field, value)) = cond.split_once(op) { - return format!("{}:{} {}", field.trim(), op, value.trim()); +fn version_documents_to_jsonl( + uploads: &[UploadSearchVersion], +) -> Result { + let mut out = String::new(); + for upload in uploads { + let mut document = serde_json::to_value(upload) + .wrap_err("failed to serialise UploadSearchVersion")?; + if let Some(object) = document.as_object_mut() { + object.insert( + "id".to_string(), + Value::String(upload.version_id.clone()), + ); } + out.push_str(&serde_json::to_string(&document)?); + out.push('\n'); } - if let Some((field, value)) = cond.split_once(':') { - return format!("{}:= {}", field.trim(), value.trim()); - } - if let Some((field, value)) = cond.split_once('=') { - return format!("{}:= {}", field.trim(), value.trim()); - } - cond.to_string() + Ok(out) } diff --git a/apps/labrinth/src/search/incremental.rs b/apps/labrinth/src/search/incremental.rs index 07c5a05b55..f96c63f508 100644 --- a/apps/labrinth/src/search/incremental.rs +++ b/apps/labrinth/src/search/incremental.rs @@ -1,43 +1,66 @@ pub mod consume; -use std::{mem, sync::Arc}; +use std::{ + collections::{HashMap, HashSet}, + mem, + sync::Arc, + time::Duration, +}; use rdkafka::{producer::FutureRecord, util::Timeout}; use serde::Serialize; use tokio::sync::Mutex; use crate::{ - models::ids::ProjectId, + models::ids::{ProjectId, VersionId}, util::kafka::{KAFKA_OPERATION_INTERVAL, KafkaClientState, KafkaEvent}, }; pub const SEARCH_PROJECT_INDEX_QUEUE_TOPIC: &str = "public.labrinth.search-project-index-queue.v1"; +const QUEUE_FLUSH_INTERVAL: Duration = Duration::from_secs(10); #[derive(Clone)] pub struct IncrementalSearchQueue { - operations: Arc>>, + operations: Arc>, kafka_client: actix_web::web::Data, } impl IncrementalSearchQueue { pub fn new(kafka_client: actix_web::web::Data) -> Self { Self { - operations: Arc::new(Mutex::new(Vec::new())), + operations: Arc::new(Mutex::new( + PendingSearchIndexOperations::default(), + )), kafka_client, } } - pub async fn push(&self, project_id: ProjectId) { + pub async fn push_project_change(&self, project_id: ProjectId) { + self.operations.lock().await.push_project_change(project_id); + } + + pub async fn push_version_changes( + &self, + project_id: ProjectId, + version_ids: impl IntoIterator, + ) { + self.operations + .lock() + .await + .push_version_change(project_id, version_ids); + } + + pub async fn push_project_removal(&self, project_id: ProjectId) { self.operations .lock() .await - .push(SearchIndexOperation { project_id }); + .push_project_removal(project_id); } pub async fn run(self) { loop { - tokio::time::sleep(KAFKA_OPERATION_INTERVAL).await; + tokio::time::sleep(QUEUE_FLUSH_INTERVAL).await; if let Err(err) = self.drain().await { tracing::error!( @@ -57,13 +80,11 @@ impl IncrementalSearchQueue { return Ok(()); } - let mut operations = operations.into_iter(); + let mut operations = operations.into_events().into_iter(); while let Some(operation) = operations.next() { let event = KafkaEvent::new( SEARCH_PROJECT_INDEX_QUEUE_TOPIC, - SearchProjectIndexQueueEventData { - project_id: operation.project_id, - }, + operation.clone(), ); let event_id = event.event_metadata.event_id; let key = event_id.to_string(); @@ -79,8 +100,10 @@ impl IncrementalSearchQueue { .await { let mut queued_operations = self.operations.lock().await; - queued_operations.push(operation); - queued_operations.extend(operations); + queued_operations.push_event(operation); + for operation in operations { + queued_operations.push_event(operation); + } return Err(err.into()); } @@ -90,12 +113,100 @@ impl IncrementalSearchQueue { } } -#[derive(Debug, Clone)] -pub struct SearchIndexOperation { - pub project_id: ProjectId, +#[derive(Default)] +struct PendingSearchIndexOperations { + changed_project_ids: HashSet, + changed_project_versions: HashMap>, + removed_project_ids: HashSet, +} + +impl PendingSearchIndexOperations { + fn is_empty(&self) -> bool { + self.changed_project_ids.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) { + self.changed_project_ids.insert(project_id); + } + } + + fn push_version_change( + &mut self, + project_id: ProjectId, + version_ids: impl IntoIterator, + ) { + if self.removed_project_ids.contains(&project_id) { + return; + } + + let version_ids = version_ids.into_iter().collect::>(); + if !version_ids.is_empty() { + self.changed_project_versions + .entry(project_id) + .or_default() + .extend(version_ids); + } + } + + fn push_project_removal(&mut self, project_id: ProjectId) { + self.changed_project_ids.remove(&project_id); + self.changed_project_versions.remove(&project_id); + self.removed_project_ids.insert(project_id); + } + + fn push_event(&mut self, event: SearchProjectIndexQueueEventData) { + match event { + SearchProjectIndexQueueEventData::Change { project_id } => { + self.push_project_change(project_id) + } + SearchProjectIndexQueueEventData::VersionChange { + project_id, + version_ids, + } => self.push_version_change(project_id, version_ids), + SearchProjectIndexQueueEventData::Removal { project_id } => { + self.push_project_removal(project_id) + } + } + } + + fn into_events(self) -> Vec { + let mut events = Vec::with_capacity( + self.changed_project_ids.len() + + self.changed_project_versions.len() + + self.removed_project_ids.len(), + ); + + events.extend(self.removed_project_ids.into_iter().map(|project_id| { + SearchProjectIndexQueueEventData::Removal { project_id } + })); + events.extend(self.changed_project_ids.into_iter().map(|project_id| { + SearchProjectIndexQueueEventData::Change { project_id } + })); + events.extend(self.changed_project_versions.into_iter().map( + |(project_id, version_ids)| { + SearchProjectIndexQueueEventData::VersionChange { + project_id, + version_ids: version_ids.into_iter().collect(), + } + }, + )); + events + } } -#[derive(Debug, Serialize)] -pub struct SearchProjectIndexQueueEventData { - pub project_id: ProjectId, +#[derive(Debug, Clone, Serialize)] +#[serde(tag = "type", rename_all = "snake_case")] +pub enum SearchProjectIndexQueueEventData { + #[serde(rename = "project_change")] + Change { project_id: ProjectId }, + #[serde(rename = "project_version_change")] + VersionChange { + project_id: ProjectId, + version_ids: Vec, + }, + #[serde(rename = "project_removal")] + Removal { project_id: ProjectId }, } diff --git a/apps/labrinth/src/search/incremental/consume.rs b/apps/labrinth/src/search/incremental/consume.rs index 69e3b30f54..1af65b5a32 100644 --- a/apps/labrinth/src/search/incremental/consume.rs +++ b/apps/labrinth/src/search/incremental/consume.rs @@ -1,20 +1,26 @@ use actix_web::web; use eyre::WrapErr; -use futures::FutureExt; use rdkafka::{ Message, consumer::{CommitMode, Consumer, StreamConsumer}, message::BorrowedMessage, }; use serde::Deserialize; -use std::collections::HashSet; +use std::{ + collections::HashSet, + time::{Duration, Instant}, +}; +use tracing::{Instrument, info, info_span}; use crate::{ database::{PgPool, redis::RedisPool}, - models::ids::ProjectId, + env::ENV, + models::ids::{ProjectId, VersionId}, search::{ - SearchBackend, incremental::SEARCH_PROJECT_INDEX_QUEUE_TOPIC, - indexing::index_project_documents, + SearchBackend, SearchDocumentBatch, SearchIndexUpdate, + UploadSearchProject, + incremental::SEARCH_PROJECT_INDEX_QUEUE_TOPIC, + indexing::{build_project_documents, build_version_change_documents}, }, util::kafka::{ INCREMENTAL_INDEX_SEARCH_TASK, KAFKA_OPERATION_INTERVAL, @@ -22,8 +28,6 @@ use crate::{ }, }; -const BATCH_SIZE: usize = 100; - pub async fn run( ro_pool: PgPool, redis_pool: RedisPool, @@ -60,25 +64,60 @@ async fn consume( search_backend: &dyn SearchBackend, consumer: &StreamConsumer, ) -> eyre::Result<()> { + // keep buffer capacity (pre-)allocated + let mut messages = Vec::with_capacity(1024); loop { - let mut messages = Vec::with_capacity(BATCH_SIZE); - messages.push( - consumer - .recv() - .await - .wrap_err("failed to receive Kafka message")?, - ); + messages.clear(); + + // wait for a first message to come in... + let first_message = consumer + .recv() + .await + .wrap_err("failed to receive Kafka message")?; + messages.push(first_message); - while messages.len() < BATCH_SIZE { - let Some(message) = consumer.recv().now_or_never() else { - break; - }; + let delay = Duration::from_secs( + ENV.SEARCH_INCREMENTAL_INDEX_BATCH_DELAY_SECONDS, + ); + info!( + "Received initial Kafka message; waiting {delay:.2?} for more to batch", + ); - messages.push(message.wrap_err("failed to receive Kafka message")?); + // ..then wait a while for more messages to batch up + // so that we can process a big batch to reindex. + // we stop until either we've reached the max batch size, + // or we've waited enough time - whichever is first. + // + // do a little trick with an `AsyncFnMut` closure + // so that we can explicitly specify the return type + let mut collect_more_messages = async || -> eyre::Result<()> { + while messages.len() < ENV.SEARCH_INCREMENTAL_INDEX_BATCH_MAX_SIZE { + let message = consumer + .recv() + .await + .wrap_err("failed to receive Kafka message")?; + messages.push(message); + } + eyre::Ok(()) + }; + match tokio::time::timeout(delay, collect_more_messages()).await { + Ok(Ok(())) | Err(_) => {} + Ok(Err(err)) => { + return Err( + err.wrap_err("failed to receive more Kafka messages") + ); + } } - consume_batch(ro_pool, redis_pool, search_backend, consumer, messages) - .await?; + info!("Consuming batch of {} messages", messages.len()); + consume_batch( + ro_pool, + redis_pool, + search_backend, + consumer, + messages.drain(..), + ) + .await?; } } @@ -87,10 +126,14 @@ async fn consume_batch( redis_pool: &RedisPool, search_backend: &dyn SearchBackend, consumer: &StreamConsumer, - messages: Vec>, + messages: impl IntoIterator>, ) -> eyre::Result<()> { - let mut project_ids = Vec::new(); - let mut seen_project_ids = HashSet::new(); + let start = Instant::now(); + + let mut project_ids_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(); let mut messages_to_commit = Vec::new(); for message in messages { @@ -125,25 +168,149 @@ async fn consume_batch( } }; - if seen_project_ids.insert(event.project_id) { - project_ids.push(event.project_id); + let event = match event { + SearchProjectIndexQueueEvent::Current(event) => event, + SearchProjectIndexQueueEvent::Legacy { project_id } => { + SearchProjectIndexQueueEventData::Change { project_id } + } + }; + + match event { + SearchProjectIndexQueueEventData::Change { project_id } => { + project_ids_to_change.insert(project_id); + } + SearchProjectIndexQueueEventData::VersionChange { + project_id, + version_ids, + } => { + if !version_ids.is_empty() { + project_ids_with_version_changes.insert(project_id); + version_ids_to_change.extend(version_ids); + } + } + SearchProjectIndexQueueEventData::Removal { project_id } => { + project_ids_to_remove.insert(project_id); + } } messages_to_commit.push(message); } - if project_ids.is_empty() { - return Ok(()); - } + project_ids_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_to_change.retain(|project_id| { + !project_ids_with_version_changes.contains(project_id) + }); + let project_ids_to_change = + project_ids_to_change.into_iter().collect::>(); + let project_ids_with_version_changes = project_ids_with_version_changes + .into_iter() + .collect::>(); + let mut project_ids_to_remove = + project_ids_to_remove.into_iter().collect::>(); + let version_ids_to_change = + version_ids_to_change.into_iter().collect::>(); - tracing::info!( + info!( kafka.message_count = messages_to_commit.len(), - project_count = project_ids.len(), - "Consumed incremental search index event batch" + "Read all Kafka messages in {:.2?}, found {} projects to change, {} projects with {} version changes, and {} projects to remove", + start.elapsed(), + project_ids_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_version_changes.is_empty() { + let operation_start = Instant::now(); + let changed_documents = build_version_change_documents( + ro_pool, + redis_pool, + &project_ids_with_version_changes, + &version_ids_to_change, + ) + .instrument(info_span!( + "index", + batch_size = project_ids_with_version_changes.len(), + version_count = version_ids_to_change.len() + )) + .await + .wrap_err_with(|| { + format!( + "failed to build search documents for {} projects and {} versions", + project_ids_with_version_changes.len(), + version_ids_to_change.len() + ) + })?; + project_ids_to_remove.extend(missing_project_document_ids( + &project_ids_with_version_changes, + &changed_documents.projects, + )); + 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(), + "Built changed project versions in {:.2?}", + operation_start.elapsed() + ); + } + + if !project_ids_to_change.is_empty() { + let operation_start = Instant::now(); + info!( + project_count = project_ids_to_change.len(), + "Building changed projects" + ); + let changed_project_documents = build_project_documents( + ro_pool, + redis_pool, + &project_ids_to_change, + ) + .instrument(info_span!( + "index", + batch_size = project_ids_to_change.len() + )) + .await + .wrap_err_with(|| { + format!( + "failed to build search documents for {} projects", + project_ids_to_change.len() + ) + })?; + project_ids_to_remove.extend(missing_project_document_ids( + &project_ids_to_change, + &changed_project_documents, + )); + documents.projects.extend(changed_project_documents); + info!( + project_count = project_ids_to_change.len(), + "Built changed projects in {:.2?}", + operation_start.elapsed() + ); + } - reindex_projects(ro_pool, redis_pool, search_backend, &project_ids) + 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 reindex project batch")?; + .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 @@ -151,45 +318,94 @@ async fn consume_batch( .wrap_err("failed to commit Kafka message")?; } + info!( + "Changed {} projects and removed {} projects in {:.2?}", + project_ids_to_change.len(), + project_ids_to_remove.len(), + start.elapsed() + ); + Ok(()) } -pub async fn reindex_project( +pub async fn reindex_project_document( ro_pool: &PgPool, redis_pool: &RedisPool, search_backend: &dyn SearchBackend, project_id: ProjectId, ) -> eyre::Result<()> { - reindex_projects(ro_pool, redis_pool, search_backend, &[project_id]).await + reindex_project_documents( + ro_pool, + redis_pool, + search_backend, + &[project_id], + ) + .await } -pub async fn reindex_projects( +pub async fn reindex_project_documents( ro_pool: &PgPool, redis_pool: &RedisPool, search_backend: &dyn SearchBackend, project_ids: &[ProjectId], ) -> eyre::Result<()> { - search_backend.remove_project_documents(project_ids).await?; + info!("Creating project documents"); + let projects = build_project_documents(ro_pool, redis_pool, project_ids) + .instrument(info_span!("index", batch_size = project_ids.len())) + .await + .wrap_err_with(|| { + format!( + "failed to build search documents for {} projects", + project_ids.len() + ) + })?; + let removed_projects = missing_project_document_ids(project_ids, &projects); + search_backend + .apply_update(SearchIndexUpdate { + projects: &projects, + removed_projects: &removed_projects, + ..SearchIndexUpdate::default() + }) + .await?; - let mut documents = Vec::new(); - for project_id in project_ids { - documents.extend( - index_project_documents(ro_pool, redis_pool, *project_id) - .await - .wrap_err_with(|| { - format!( - "failed to build project {project_id} search documents" - ) - })?, - ); - } + Ok(()) +} - search_backend.index_documents(&documents).await?; +fn missing_project_document_ids( + project_ids: &[ProjectId], + documents: &[UploadSearchProject], +) -> Vec { + let built_project_ids = documents + .iter() + .map(|project| project.project_id.as_str()) + .collect::>(); - Ok(()) + project_ids + .iter() + .copied() + .filter(|project_id| { + !built_project_ids.contains(project_id.to_string().as_str()) + }) + .collect() } #[derive(Debug, Deserialize)] -struct SearchProjectIndexQueueEvent { - project_id: ProjectId, +#[serde(untagged)] +enum SearchProjectIndexQueueEvent { + Current(SearchProjectIndexQueueEventData), + Legacy { project_id: ProjectId }, +} + +#[derive(Debug, Deserialize)] +#[serde(tag = "type", rename_all = "snake_case")] +enum SearchProjectIndexQueueEventData { + #[serde(rename = "project_change")] + Change { project_id: ProjectId }, + #[serde(rename = "project_version_change")] + VersionChange { + project_id: ProjectId, + version_ids: Vec, + }, + #[serde(rename = "project_removal")] + Removal { project_id: ProjectId }, } diff --git a/apps/labrinth/src/search/indexing.rs b/apps/labrinth/src/search/indexing.rs index 50fc706206..93ec7f16d0 100644 --- a/apps/labrinth/src/search/indexing.rs +++ b/apps/labrinth/src/search/indexing.rs @@ -5,7 +5,7 @@ use futures::TryStreamExt; use heck::ToKebabCase; use itertools::Itertools; use regex::Regex; -use std::collections::HashMap; +use std::collections::{HashMap, HashSet}; use std::sync::LazyLock; use tracing::{info, warn}; @@ -20,11 +20,14 @@ use crate::database::models::{ }; use crate::database::redis::RedisPool; use crate::models::exp; -use crate::models::ids::ProjectId; +use crate::models::ids::{ProjectId, VersionId}; use crate::models::projects::{DependencyType, from_duplicate_version_fields}; use crate::models::v2::projects::LegacyProject; use crate::routes::v2_reroute; -use crate::search::{SearchProjectDependency, UploadSearchProject}; +use crate::search::{ + SearchDocumentBatch, SearchProjectDependency, UploadSearchProject, + UploadSearchVersion, +}; use crate::util::error::Context; struct PartialProject { @@ -68,7 +71,7 @@ pub async fn index_local( redis: &RedisPool, cursor: i64, limit: i64, -) -> eyre::Result<(Vec, i64)> { +) -> eyre::Result<(SearchDocumentBatch, i64)> { info!("Indexing local projects!"); let searchable_statuses = searchable_statuses(); @@ -111,20 +114,52 @@ pub async fn index_local( let project_ids = db_projects.iter().map(|x| x.id.0).collect::>(); let Some(largest) = project_ids.iter().max() else { - return Ok((vec![], i64::MAX)); + return Ok((SearchDocumentBatch::default(), i64::MAX)); }; - let uploads = build_search_documents(pool, redis, db_projects).await?; - Ok((uploads, *largest)) + let documents = + build_search_documents(pool, redis, db_projects, None).await?; + Ok((documents, *largest)) } -pub async fn index_project_documents( +pub async fn build_project_documents( pool: &PgPool, redis: &RedisPool, - project_id: ProjectId, + project_ids: &[ProjectId], ) -> eyre::Result> { + let version_ids = HashSet::new(); + Ok( + build_search_document_batch(pool, redis, project_ids, &version_ids) + .await? + .projects, + ) +} + +pub async fn build_version_change_documents( + pool: &PgPool, + redis: &RedisPool, + project_ids: &[ProjectId], + version_ids: &[VersionId], +) -> eyre::Result { + let version_ids = version_ids + .iter() + .copied() + .map(DBVersionId::from) + .collect::>(); + build_search_document_batch(pool, redis, project_ids, &version_ids).await +} + +async fn build_search_document_batch( + pool: &PgPool, + redis: &RedisPool, + project_ids: &[ProjectId], + version_ids: &HashSet, +) -> eyre::Result { let searchable_statuses = searchable_statuses(); - let project_ids = vec![DBProjectId::from(project_id).0]; + let project_ids = project_ids + .iter() + .map(|project_id| DBProjectId::from(*project_id).0) + .collect::>(); let db_projects = sqlx::query!( r#" @@ -158,14 +193,15 @@ pub async fn index_project_documents( .await .wrap_err("failed to fetch project")?; - build_search_documents(pool, redis, db_projects).await + build_search_documents(pool, redis, db_projects, Some(version_ids)).await } async fn build_search_documents( pool: &PgPool, redis: &RedisPool, db_projects: Vec, -) -> eyre::Result> { + version_ids: Option<&HashSet>, +) -> eyre::Result { let searchable_statuses = searchable_statuses(); let project_ids = db_projects.iter().map(|x| x.id.0).collect::>(); let project_components = db_projects @@ -281,7 +317,7 @@ async fn build_search_documents( .await?; info!("Indexing local versions!"); - let mut versions = index_versions(pool, project_ids.clone()).await?; + let mut versions = load_project_versions(pool, project_ids.clone()).await?; info!("Indexing local org owners!"); @@ -333,7 +369,7 @@ async fn build_search_documents( .await?; info!("Getting all loader fields!"); - let loader_fields: Vec = sqlx::query!( + let loader_field_definitions: Vec = sqlx::query!( " SELECT DISTINCT id, field, field_type, enum_type, min_val, max_val, optional FROM loader_fields lf @@ -351,7 +387,8 @@ async fn build_search_documents( }) .try_collect() .await?; - let loader_fields: Vec<&QueryLoaderField> = loader_fields.iter().collect(); + let loader_field_definitions: Vec<&QueryLoaderField> = + loader_field_definitions.iter().collect(); info!("Getting all loader field enum values!"); @@ -376,7 +413,8 @@ async fn build_search_documents( .await?; info!("Indexing loaders, project types!"); - let mut uploads = Vec::new(); + let mut project_uploads = Vec::new(); + let mut version_uploads = Vec::new(); let total_len = db_projects.len(); let mut count = 0; @@ -475,176 +513,233 @@ async fn build_search_documents( .collect::>(); if let Some(versions) = versions.remove(&project.id) { - // Aggregated project loader fields + let Some(latest_version) = versions.iter().max_by(|a, b| { + a.date_published + .cmp(&b.date_published) + .then_with(|| a.id.0.cmp(&b.id.0)) + }) else { + continue; + }; + let project_version_fields = versions .iter() .flat_map(|x| x.version_fields.clone()) .collect::>(); let aggregated_version_fields = VersionField::from_query_json( project_version_fields, - &loader_fields, + &loader_field_definitions, &loader_field_enum_values, true, ); - let project_loader_fields = + let unvectorized_loader_fields = aggregated_version_fields + .iter() + .map(|field| { + (field.field_name.clone(), field.value.serialize_internal()) + }) + .collect(); + let mut loader_fields = from_duplicate_version_fields(aggregated_version_fields); + let project_loader_fields = loader_fields.clone(); - // aggregated project loaders - let project_loaders = versions + let mut project_loaders = versions .iter() .flat_map(|x| x.loaders.clone()) .collect::>(); + project_loaders.sort(); + project_loaders.dedup(); - // all valid project types across every version of the project, so that - // filters can exclude projects that have *any* version of a given - // project type (unlike the version-specific `project_types` field). - let mut all_project_types = versions + let mut project_types = versions .iter() .flat_map(|x| x.project_types.clone()) .collect::>(); - all_project_types.sort(); - all_project_types.dedup(); + project_types.sort(); + project_types.dedup(); exp::compat::correct_project_types( &project.components, - &mut all_project_types, + &mut project_types, ); - for version in versions { + let project_id = ProjectId::from(project.id).to_string(); + version_uploads.extend(versions.iter().filter_map(|version| { + if version_ids.is_some_and(|version_ids| { + !version_ids.contains(&version.id) + }) { + return None; + } + let version_fields = VersionField::from_query_json( - version.version_fields, - &loader_fields, + version.version_fields.clone(), + &loader_field_definitions, &loader_field_enum_values, false, ); let unvectorized_loader_fields = version_fields .iter() - .map(|vf| { - (vf.field_name.clone(), vf.value.serialize_internal()) + .map(|field| { + ( + field.field_name.clone(), + field.value.serialize_internal(), + ) }) .collect(); - let mut loader_fields = - from_duplicate_version_fields(version_fields); - let mut project_types = version.project_types; - + let mut fields = from_duplicate_version_fields(version_fields); + let mut version_project_types = version.project_types.clone(); exp::compat::correct_project_types( &project.components, - &mut project_types, + &mut version_project_types, ); - let mut version_loaders = version.loaders; - - // Uses version loaders, not project loaders. - let mut categories = categories.clone(); - categories.append(&mut version_loaders.clone()); - - let display_categories = display_categories.clone(); - categories.append(&mut version_loaders); - // SPECIAL BEHAVIOUR // Todo: revisit. // For consistency with v2 searching, we consider the loader field 'mrpack_loaders' to be a category. // These were previously considered the loader, and in v2, the loader is a category for searching. // So to avoid breakage or awkward conversions, we just consider those loader_fields to be categories. - // The loaders are kept in loader_fields as well, so that no information is lost on retrieval. - let mrpack_loaders = loader_fields + // The loaders are kept in the project document's aggregated loader fields as well, so that no information is lost on retrieval. + let mut version_categories = version.loaders.clone(); + let mrpack_loaders = fields .get("mrpack_loaders") - .cloned() - .map(|x| { - x.into_iter() - .filter_map(|x| x.as_str().map(String::from)) - .collect::>() - }) - .unwrap_or_default(); - categories.extend(mrpack_loaders); - if loader_fields.contains_key("mrpack_loaders") { - categories.retain(|x| *x != "mrpack"); + .into_iter() + .flatten() + .filter_map(|value| value.as_str().map(String::from)) + .collect::>(); + version_categories.extend(mrpack_loaders); + if fields.contains_key("mrpack_loaders") { + version_categories.retain(|category| category != "mrpack"); } + version_categories.sort(); + version_categories.dedup(); - // SPECIAL BEHAVIOUR: - // For consistency with v2 searching, we manually input the - // client_side and server_side fields from the loader fields into - // separate loader fields. - // 'client_side' and 'server_side' remain supported by meilisearch even though they are no longer v3 fields. let (_, v2_og_project_type) = - LegacyProject::get_project_type(&project_types); + LegacyProject::get_project_type(&version_project_types); let (client_side, server_side) = v2_reroute::convert_v3_side_types_to_v2_side_types( &unvectorized_loader_fields, Some(&v2_og_project_type), ); - if let Ok(client_side) = serde_json::to_value(client_side) { - loader_fields - .insert("client_side".to_string(), vec![client_side]); + fields.insert("client_side".to_string(), vec![client_side]); } if let Ok(server_side) = serde_json::to_value(server_side) { - loader_fields - .insert("server_side".to_string(), vec![server_side]); + fields.insert("server_side".to_string(), vec![server_side]); } - - let components = project - .components - .clone() - .into_query( - ProjectId::from(project.id), - &project_query_context, + fields.retain(|field, _| { + matches!( + field.as_str(), + "environment" + | "game_versions" + | "client_side" + | "server_side" ) - .wrap_err("failed to populate query components")?; - - let usp = UploadSearchProject { - version_id: crate::models::ids::VersionId::from(version.id) - .to_string(), - project_id: crate::models::ids::ProjectId::from(project.id) - .to_string(), - name: project.name.clone(), - indexed_name: normalize_for_search(&project.name), - summary: project.summary.clone(), - categories: categories.clone(), - display_categories: display_categories.clone(), - follows: project.follows, - downloads: project.downloads, - log_downloads: (project.downloads.max(1) as f64).ln(), - icon_url: project.icon_url.clone(), - author: username.clone(), - author_id: ariadne::ids::UserId::from(user_id).to_string(), - organization: org_name.clone(), - organization_id: org_id.map(|e| { - crate::models::ids::OrganizationId::from(e).to_string() - }), - indexed_author: normalize_for_search(&username), - date_created: project.approved, - created_timestamp: project.approved.timestamp(), - date_modified: project.updated, - modified_timestamp: project.updated.timestamp(), + }); + + Some(UploadSearchVersion { + version_id: VersionId::from(version.id).to_string(), + project_id: project_id.clone(), + categories: version_categories, + project_types: version_project_types, version_published_timestamp: version .date_published .timestamp(), - license: license.clone(), - slug: project.slug.clone(), - // TODO - project_types, - all_project_types: all_project_types.clone(), - gallery: gallery.clone(), - featured_gallery: featured_gallery.clone(), - open_source, - color: project.color.map(|x| x as u32), - dependency_project_ids: dependency_project_ids.clone(), - compatible_dependency_project_ids: - compatible_dependency_project_ids.clone(), - dependencies: dependencies.clone(), - loader_fields, - project_loader_fields: project_loader_fields.clone(), - // 'loaders' is aggregate of all versions' loaders - loaders: project_loaders.clone(), - components, - }; + loader_fields: fields, + }) + })); + + let mut project_categories = categories; + project_categories.sort(); + project_categories.dedup(); + let mut categories = project_categories.clone(); + categories.extend(project_loaders.iter().cloned()); + + let mrpack_loaders = loader_fields + .get("mrpack_loaders") + .into_iter() + .flatten() + .filter_map(|value| value.as_str().map(String::from)) + .collect::>(); + categories.extend(mrpack_loaders); + if loader_fields.contains_key("mrpack_loaders") { + categories.retain(|category| category != "mrpack"); + } + categories.sort(); + categories.dedup(); + + let (_, v2_og_project_type) = + LegacyProject::get_project_type(&project_types); + let (client_side, server_side) = + v2_reroute::convert_v3_side_types_to_v2_side_types( + &unvectorized_loader_fields, + Some(&v2_og_project_type), + ); - uploads.push(usp); + if let Ok(client_side) = serde_json::to_value(client_side) { + loader_fields + .insert("client_side".to_string(), vec![client_side]); + } + if let Ok(server_side) = serde_json::to_value(server_side) { + loader_fields + .insert("server_side".to_string(), vec![server_side]); } + + let components = project + .components + .clone() + .into_query(ProjectId::from(project.id), &project_query_context) + .wrap_err("failed to populate query components")?; + let indexed_name = normalize_for_search(&project.name); + + project_uploads.push(UploadSearchProject { + version_id: crate::models::ids::VersionId::from( + latest_version.id, + ) + .to_string(), + project_id, + name: project.name, + indexed_name, + summary: project.summary, + categories, + project_categories, + display_categories, + follows: project.follows, + downloads: project.downloads, + log_downloads: (project.downloads.max(1) as f64).ln(), + icon_url: project.icon_url, + author: username.clone(), + author_id: ariadne::ids::UserId::from(user_id).to_string(), + organization: org_name, + organization_id: org_id.map(|id| { + crate::models::ids::OrganizationId::from(id).to_string() + }), + indexed_author: normalize_for_search(&username), + date_created: project.approved, + created_timestamp: project.approved.timestamp(), + date_modified: project.updated, + modified_timestamp: project.updated.timestamp(), + version_published_timestamp: latest_version + .date_published + .timestamp(), + license, + slug: project.slug, + project_types: project_types.clone(), + all_project_types: project_types, + gallery, + featured_gallery, + open_source, + color: project.color.map(|x| x as u32), + dependency_project_ids, + compatible_dependency_project_ids, + dependencies, + project_loader_fields, + loader_fields, + loaders: project_loaders, + components, + }); } } - Ok(uploads) + Ok(SearchDocumentBatch { + projects: project_uploads, + versions: version_uploads, + }) } struct PartialVersion { @@ -655,7 +750,7 @@ struct PartialVersion { date_published: DateTime, } -async fn index_versions( +async fn load_project_versions( pool: &PgPool, project_ids: Vec, ) -> Result>> { diff --git a/apps/labrinth/src/search/mod.rs b/apps/labrinth/src/search/mod.rs index e7b55dd9f3..33bfe26f51 100644 --- a/apps/labrinth/src/search/mod.rs +++ b/apps/labrinth/src/search/mod.rs @@ -105,24 +105,17 @@ pub trait SearchBackend: Send + Sync { info: &SearchRequest, ) -> Result; - async fn index_projects( + async fn rebuild_index( &self, ro_pool: PgPool, redis: RedisPool, ) -> eyre::Result<()>; - async fn index_documents( + async fn apply_update( &self, - documents: &[UploadSearchProject], + update: SearchIndexUpdate<'_>, ) -> eyre::Result<()>; - async fn remove_project_documents( - &self, - ids: &[ProjectId], - ) -> eyre::Result<()>; - - async fn remove_documents(&self, ids: &[VersionId]) -> eyre::Result<()>; - async fn tasks(&self) -> eyre::Result; async fn tasks_cancel( @@ -233,8 +226,12 @@ impl FromStr for SearchBackendKind { } } +/// Nullable fields in Typesense-bound documents should use +/// `skip_serializing_if = "Option::is_none"` so they are omitted instead of +/// serialized as `null`. #[derive(Serialize, Deserialize, Debug, Clone)] pub struct UploadSearchProject { + /// ID of the most recently published version. pub version_id: String, pub project_id: String, // @@ -244,20 +241,25 @@ pub struct UploadSearchProject { pub slug: Option, pub author: String, pub author_id: String, + #[serde(skip_serializing_if = "Option::is_none")] pub organization: Option, + #[serde(skip_serializing_if = "Option::is_none")] pub organization_id: Option, pub indexed_author: String, pub name: String, pub indexed_name: String, pub summary: String, pub categories: Vec, + pub project_categories: Vec, pub display_categories: Vec, pub follows: i32, pub downloads: i32, pub log_downloads: f64, + #[serde(skip_serializing_if = "Option::is_none")] pub icon_url: Option, pub license: String, pub gallery: Vec, + #[serde(skip_serializing_if = "Option::is_none")] pub featured_gallery: Option, /// RFC 3339 formatted creation date of the project pub date_created: DateTime, @@ -267,9 +269,10 @@ pub struct UploadSearchProject { pub date_modified: DateTime, /// Unix timestamp of the last major modification pub modified_timestamp: i64, - /// Unix timestamp of the publication date of the version + /// Unix timestamp of the most recently published version. pub version_published_timestamp: i64, pub open_source: bool, + #[serde(skip_serializing_if = "Option::is_none")] pub color: Option, #[serde(default)] pub dependency_project_ids: Vec, @@ -288,12 +291,45 @@ pub struct UploadSearchProject { pub loader_fields: HashMap>, } +#[derive(Serialize, Deserialize, Debug, Clone)] +pub struct UploadSearchVersion { + pub version_id: String, + pub project_id: String, + pub categories: Vec, + pub project_types: Vec, + pub version_published_timestamp: i64, + #[serde(flatten)] + pub loader_fields: HashMap>, +} + +#[derive(Debug, Default)] +pub struct SearchDocumentBatch { + pub projects: Vec, + pub versions: Vec, +} + +/// 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`. #[derive(Serialize, Deserialize, Debug, Clone, ToSchema)] pub struct SearchProjectDependency { pub project_id: String, pub dependency_type: DependencyType, pub name: String, + #[serde(skip_serializing_if = "Option::is_none")] pub slug: Option, + #[serde(skip_serializing_if = "Option::is_none")] pub icon_url: Option, } @@ -307,6 +343,7 @@ pub struct SearchResults { #[derive(Serialize, Deserialize, Debug, Clone, ToSchema)] pub struct ResultSearchProject { + /// ID of the most recently published version. pub version_id: String, pub project_id: String, pub project_types: Vec, diff --git a/apps/labrinth/tests/search.rs b/apps/labrinth/tests/search.rs index c009aa929c..f583d4a387 100644 --- a/apps/labrinth/tests/search.rs +++ b/apps/labrinth/tests/search.rs @@ -213,18 +213,7 @@ async fn index_swaps() { test_env.api.remove_project("alpha", USER_USER_PAT).await; assert_status!(&resp, StatusCode::NO_CONTENT); - // We should wait for deletions to be indexed - let projects = test_env - .api - .search_deserialized( - None, - Some(json!([["categories:fabric"]])), - USER_USER_PAT, - ) - .await; - assert_eq!(projects.total_hits, 0); - - // When we reindex, it should be still gone + // When we reindex, the deleted project should be gone let resp = test_env.api.reset_search_index().await; assert_status!(&resp, StatusCode::NO_CONTENT);