diff --git a/Cargo.lock b/Cargo.lock index 726ebeeed2..af2269e55b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5532,6 +5532,7 @@ dependencies = [ "totp-rs", "tracing", "tracing-actix-web", + "unicode-normalization", "url", "urlencoding", "utoipa", diff --git a/Cargo.toml b/Cargo.toml index 1529ce8910..59c4c0b554 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -230,6 +230,7 @@ tracing-ecs = "0.5.0" tracing-error = "0.2.1" tracing-subscriber = "0.3.20" typed-path = "0.12.0" +unicode-normalization = "0.1.24" url = "2.5.7" urlencoding = "2.1.3" utoipa = { version = "5.4.0", features = ["actix_extras", "chrono", "decimal"] } diff --git a/apps/labrinth/.env.docker-compose b/apps/labrinth/.env.docker-compose index 81bd6934a9..aab67afd55 100644 --- a/apps/labrinth/.env.docker-compose +++ b/apps/labrinth/.env.docker-compose @@ -17,14 +17,15 @@ DATABASE_URL=postgresql://labrinth:labrinth@labrinth-postgres/labrinth DATABASE_MIN_CONNECTIONS=0 DATABASE_MAX_CONNECTIONS=16 -SEARCH_BACKEND=typesense +SEARCH_BACKEND=elasticsearch MEILISEARCH_READ_ADDR=http://localhost:7700 MEILISEARCH_WRITE_ADDRS=http://localhost:7700 MEILISEARCH_KEY=modrinth -ELASTICSEARCH_URL=http://localhost:9200 +ELASTICSEARCH_URL=http://elasticsearch0:9200 ELASTICSEARCH_INDEX_PREFIX=labrinth -ELASTICSEARCH_USERNAME=elastic -ELASTICSEARCH_PASSWORD=elastic +ELASTICSEARCH_USERNAME= +ELASTICSEARCH_PASSWORD= +ELASTICSEARCH_TYPESENSE_PARITY=true SEARCH_INDEX_CHUNK_SIZE=5000 SEARCH_INCREMENTAL_INDEX_BATCH_DELAY_SECONDS=5 SEARCH_INCREMENTAL_INDEX_BATCH_MAX_SIZE=1000 diff --git a/apps/labrinth/.env.local b/apps/labrinth/.env.local index ddc5096de8..aabd22213e 100644 --- a/apps/labrinth/.env.local +++ b/apps/labrinth/.env.local @@ -17,7 +17,7 @@ DATABASE_URL=postgresql://labrinth:labrinth@localhost/labrinth DATABASE_MIN_CONNECTIONS=0 DATABASE_MAX_CONNECTIONS=16 -SEARCH_BACKEND=typesense +SEARCH_BACKEND=elasticsearch # Meilisearch configuration MEILISEARCH_READ_ADDR=http://localhost:7700 @@ -32,7 +32,7 @@ ELASTICSEARCH_INDEX_PREFIX=labrinth # MEILISEARCH_READ_ADDR=http://localhost:7710 # MEILISEARCH_WRITE_ADDRS=http://localhost:7700,http://localhost:7701 -SEARCH_BACKEND=typesense +SEARCH_BACKEND=elasticsearch MEILISEARCH_KEY=modrinth MEILISEARCH_META_NAMESPACE= @@ -42,6 +42,7 @@ ELASTICSEARCH_URL=http://localhost:9200 ELASTICSEARCH_INDEX_PREFIX=labrinth ELASTICSEARCH_USERNAME= ELASTICSEARCH_PASSWORD= +ELASTICSEARCH_TYPESENSE_PARITY=true SEARCH_INDEX_CHUNK_SIZE=5000 SEARCH_INCREMENTAL_INDEX_BATCH_DELAY_SECONDS=5 diff --git a/apps/labrinth/Cargo.toml b/apps/labrinth/Cargo.toml index 1081074d9a..585ae3000d 100644 --- a/apps/labrinth/Cargo.toml +++ b/apps/labrinth/Cargo.toml @@ -135,6 +135,7 @@ tokio-stream = { workspace = true } totp-rs = { workspace = true, features = ["gen_secret"] } tracing = { workspace = true } tracing-actix-web = { workspace = true } +unicode-normalization = { workspace = true } url = { workspace = true } urlencoding = { workspace = true } utoipa = { workspace = true, features = ["url"] } diff --git a/apps/labrinth/src/env.rs b/apps/labrinth/src/env.rs index 20ce98f546..a4915201c3 100644 --- a/apps/labrinth/src/env.rs +++ b/apps/labrinth/src/env.rs @@ -237,6 +237,12 @@ vars! { SEARCH_TYPESENSE_DEFAULT_BUCKETING: Json = Json(crate::search::backend::typesense::Bucketing::Buckets(5)); SEARCH_TYPESENSE_DEFAULT_MAX_CANDIDATES: usize = 24usize; + ELASTICSEARCH_URL: String = "http://localhost:9200"; + ELASTICSEARCH_INDEX_PREFIX: String = "labrinth"; + ELASTICSEARCH_USERNAME: String = ""; + ELASTICSEARCH_PASSWORD: String = ""; + ELASTICSEARCH_BULK_BATCH_SIZE: usize = 1000usize; + ELASTICSEARCH_TYPESENSE_PARITY: bool = true; // storage STORAGE_BACKEND: crate::file_hosting::FileHostKind = crate::file_hosting::FileHostKind::Local; diff --git a/apps/labrinth/src/search/backend/elasticsearch/filter.rs b/apps/labrinth/src/search/backend/elasticsearch/filter.rs new file mode 100644 index 0000000000..8849315eb4 --- /dev/null +++ b/apps/labrinth/src/search/backend/elasticsearch/filter.rs @@ -0,0 +1,394 @@ +use eyre::{Result, eyre}; +use serde_json::{Value, json}; + +use crate::search::filter::{ + FilterComparison, FilterCondition, FilterExpr, FilterLiteral, + FilterPredicate, +}; + +const MAX_DNF_CLAUSES: usize = 64; +const MAX_FILTER_DEPTH: usize = 64; +const MAX_FILTER_NODES: usize = 1024; + +#[derive(Clone, Copy, PartialEq, Eq)] +enum FilterScope { + Project, + Version, + Mixed, +} + +pub(super) struct ElasticsearchFilter { + pub query: Value, + pub has_version_filter: bool, +} + +pub(super) fn serialize_filter( + filter: &FilterExpr, +) -> Result { + let (nodes, depth) = filter_complexity(filter); + if nodes > MAX_FILTER_NODES { + return Err(eyre!("search filter has too many expressions")); + } + if depth > MAX_FILTER_DEPTH { + return Err(eyre!("search filter is nested too deeply")); + } + + let mut inner_hits_index = 0; + let query = plan(filter, &mut inner_hits_index)?; + Ok(ElasticsearchFilter { + query, + has_version_filter: inner_hits_index != 0, + }) +} + +fn plan(filter: &FilterExpr, inner_hits_index: &mut usize) -> Result { + match filter_scope(filter) { + FilterScope::Project => lower(filter), + FilterScope::Version => { + version_query(lower(filter)?, inner_hits_index) + } + FilterScope::Mixed => plan_mixed(filter, inner_hits_index), + } +} + +fn plan_mixed( + filter: &FilterExpr, + inner_hits_index: &mut usize, +) -> Result { + match filter { + FilterExpr::Or(expressions) => expressions + .iter() + .map(|expression| plan(expression, inner_hits_index)) + .collect::>>() + .map(or_query), + FilterExpr::And(expressions) + if expressions.iter().all(|expression| { + filter_scope(expression) != FilterScope::Mixed + }) => + { + plan_partitioned_and(expressions, inner_hits_index) + } + _ => { + let clauses = to_dnf(filter)?; + clauses + .into_iter() + .map(|clause| plan_clause(clause, inner_hits_index)) + .collect::>>() + .map(or_query) + } + } +} + +fn plan_partitioned_and( + expressions: &[FilterExpr], + inner_hits_index: &mut usize, +) -> Result { + let mut project = Vec::new(); + let mut version = Vec::new(); + for expression in expressions { + match filter_scope(expression) { + FilterScope::Project => project.push(lower(expression)?), + FilterScope::Version => version.push(lower(expression)?), + FilterScope::Mixed => { + return Err(eyre!("could not partition mixed search filter")); + } + } + } + if !version.is_empty() { + project.push(version_query( + and_query(version), + inner_hits_index, + )?); + } + Ok(and_query(project)) +} + +fn plan_clause( + predicates: Vec<&FilterPredicate>, + inner_hits_index: &mut usize, +) -> Result { + let mut project = Vec::new(); + let mut version = Vec::new(); + for predicate in predicates { + let query = predicate_query(predicate)?; + if is_version_filter_field(predicate.field.as_str()) { + version.push(query); + } else { + project.push(query); + } + } + if !version.is_empty() { + project.push(version_query( + and_query(version), + inner_hits_index, + )?); + } + Ok(and_query(project)) +} + +fn lower(filter: &FilterExpr) -> Result { + match filter { + FilterExpr::And(expressions) => expressions + .iter() + .map(lower) + .collect::>>() + .map(and_query), + FilterExpr::Or(expressions) => expressions + .iter() + .map(lower) + .collect::>>() + .map(or_query), + FilterExpr::Predicate(predicate) => predicate_query(predicate), + FilterExpr::Not(_) => { + Err(eyre!("search filter contains an unnormalized negation")) + } + } +} + +fn predicate_query(predicate: &FilterPredicate) -> Result { + let field = exact_field(predicate.field.as_str()); + match &predicate.condition { + FilterCondition::Compare { comparison, value } => { + let value = literal_value(value)?; + Ok(match comparison { + FilterComparison::Equal => { + json!({"term": {(field): {"value": value}}}) + } + FilterComparison::NotEqual => not_query(json!({ + "term": {(field): {"value": value}} + })), + FilterComparison::GreaterThan => { + json!({"range": {(field): {"gt": value}}}) + } + FilterComparison::GreaterThanOrEqual => { + json!({"range": {(field): {"gte": value}}}) + } + FilterComparison::LessThan => { + json!({"range": {(field): {"lt": value}}}) + } + FilterComparison::LessThanOrEqual => { + json!({"range": {(field): {"lte": value}}}) + } + }) + } + FilterCondition::In { values, negated } => { + let values = values + .iter() + .map(literal_value) + .collect::>>()?; + let query = json!({"terms": {(field): values}}); + Ok(if *negated { not_query(query) } else { query }) + } + FilterCondition::Exists { negated } => { + let query = json!({"exists": {"field": field}}); + Ok(if *negated { not_query(query) } else { query }) + } + } +} + +fn literal_value(literal: &FilterLiteral) -> Result { + match literal { + FilterLiteral::String(value) => Ok(Value::String(value.clone())), + FilterLiteral::Number(value) => serde_json::from_str(value) + .map_err(|error| eyre!("invalid numeric filter literal: {error}")), + FilterLiteral::Bool(value) => Ok(Value::Bool(*value)), + } +} + +fn exact_field(field: &str) -> &str { + match field { + "name" => "name.keyword", + "author" => "author.keyword", + "summary" => "summary.keyword", + "slug" => "slug.keyword", + _ => field, + } +} + +fn version_query( + query: Value, + inner_hits_index: &mut usize, +) -> Result { + let name = format!("matching_versions_{}", *inner_hits_index); + *inner_hits_index += 1; + Ok(json!({ + "has_child": { + "type": "version", + "score_mode": "none", + "query": query, + "inner_hits": { + "name": name, + "size": 1, + "_source": ["version_id", "version_published_timestamp"], + "sort": [ + {"version_published_timestamp": {"order": "desc"}}, + {"version_id": {"order": "desc"}} + ] + } + } + })) +} + +fn and_query(queries: Vec) -> Value { + match queries.len() { + 0 => json!({"match_all": {}}), + 1 => queries.into_iter().next().unwrap_or_default(), + _ => json!({"bool": {"filter": queries}}), + } +} + +fn or_query(queries: Vec) -> Value { + match queries.len() { + 0 => json!({"match_none": {}}), + 1 => queries.into_iter().next().unwrap_or_default(), + _ => json!({ + "bool": { + "should": queries, + "minimum_should_match": 1 + } + }), + } +} + +fn not_query(query: Value) -> Value { + json!({ + "bool": { + "must": [{"match_all": {}}], + "must_not": [query] + } + }) +} + +fn filter_scope(filter: &FilterExpr) -> FilterScope { + match filter { + FilterExpr::Predicate(predicate) => { + if is_version_filter_field(predicate.field.as_str()) { + FilterScope::Version + } else { + FilterScope::Project + } + } + FilterExpr::And(expressions) | FilterExpr::Or(expressions) => { + let mut scopes = expressions.iter().map(filter_scope); + let Some(first) = scopes.next() else { + return FilterScope::Project; + }; + if scopes.all(|scope| scope == first) { + first + } else { + FilterScope::Mixed + } + } + FilterExpr::Not(expression) => filter_scope(expression), + } +} + +fn is_version_filter_field(field: &str) -> bool { + matches!( + field, + "categories" + | "project_types" + | "environment" + | "game_versions" + | "client_side" + | "server_side" + ) +} + +fn to_dnf(filter: &FilterExpr) -> Result>> { + match filter { + FilterExpr::Predicate(predicate) => Ok(vec![vec![predicate]]), + FilterExpr::Or(expressions) => { + let mut clauses = Vec::new(); + for expression in expressions.iter() { + clauses.extend(to_dnf(expression)?); + if clauses.len() > MAX_DNF_CLAUSES { + return Err(eyre!( + "search filter has too many boolean clauses" + )); + } + } + Ok(clauses) + } + FilterExpr::And(expressions) => { + let mut clauses = vec![Vec::new()]; + for expression in expressions.iter() { + let right = to_dnf(expression)?; + if clauses.len().saturating_mul(right.len()) + > MAX_DNF_CLAUSES + { + return Err(eyre!( + "search filter has too many boolean clauses" + )); + } + clauses = clauses + .into_iter() + .flat_map(|left| { + right.iter().map(move |right| { + let mut clause = left.clone(); + clause.extend(right); + clause + }) + }) + .collect(); + } + Ok(clauses) + } + FilterExpr::Not(_) => { + Err(eyre!("search filter contains an unnormalized negation")) + } + } +} + +fn filter_complexity(filter: &FilterExpr) -> (usize, usize) { + match filter { + FilterExpr::Predicate(_) => (1, 1), + FilterExpr::And(expressions) | FilterExpr::Or(expressions) => { + expressions.iter().map(filter_complexity).fold( + (1, 1), + |(nodes, depth), (child_nodes, child_depth)| { + (nodes + child_nodes, depth.max(child_depth + 1)) + }, + ) + } + FilterExpr::Not(expression) => { + let (nodes, depth) = filter_complexity(expression); + (nodes + 1, depth + 1) + } + } +} + +#[cfg(test)] +mod tests { + use super::serialize_filter; + use crate::search::filter::{normalize, parse_expression}; + use serde_json::Value; + + fn serialize(input: &str) -> Value { + let filter = normalize(parse_expression(input).unwrap()); + serialize_filter(&filter).unwrap().query + } + + #[test] + fn correlated_version_filters_use_one_join() { + let query = serialize( + "categories = fabric AND game_versions = 1.21", + ); + assert_eq!(query.to_string().matches("has_child").count(), 1); + } + + #[test] + fn project_filters_do_not_use_a_join() { + let query = serialize("license = MIT"); + assert_eq!(query.to_string().matches("has_child").count(), 0); + assert_eq!(query["term"]["license"]["value"], "MIT"); + } + + #[test] + fn mixed_boolean_filters_preserve_version_correlation() { + let query = serialize( + "(license = MIT OR categories = fabric) AND game_versions = 1.21", + ); + assert_eq!(query.to_string().matches("has_child").count(), 2); + } +} diff --git a/apps/labrinth/src/search/backend/elasticsearch/mod.rs b/apps/labrinth/src/search/backend/elasticsearch/mod.rs new file mode 100644 index 0000000000..034eea120f --- /dev/null +++ b/apps/labrinth/src/search/backend/elasticsearch/mod.rs @@ -0,0 +1,1150 @@ +//! Search implementation backed by an Elasticsearch cluster. +//! +//! Projects and versions share an index and use an Elasticsearch join field. +//! This keeps version filters correlated without duplicating every version +//! into its project document. + +use std::collections::{HashMap, HashSet}; + +use async_trait::async_trait; +use eyre::{Result, eyre}; +use itertools::Itertools; +use reqwest::{Method, Response, StatusCode}; +use serde::Serialize; +use serde_json::{Map, Value, json}; +use tracing::{debug, info, warn}; +use xredis::RedisPool; + +use crate::database::PgPool; +use crate::env::ENV; +use crate::routes::ApiError; +use crate::search::backend::{ + SearchIndex, combined_search_filters, parse_search_index, + parse_search_request, +}; +use crate::search::filter::{ + FilterExpr, from_legacy_v2_facets_json, normalize, parse_expression, +}; +use crate::search::indexing::index_local; +use crate::search::{ + ResultSearchProject, SearchBackend, SearchIndexUpdate, SearchRequest, + SearchResults, TasksCancelFilter, UploadSearchProject, UploadSearchVersion, +}; +use crate::util::error::Context; + +use self::filter::{ElasticsearchFilter, serialize_filter}; + +mod filter; +mod typesense_parity; + +const DELETE_FILTER_ID_BATCH_SIZE: usize = 1024; +const MAX_RESULT_WINDOW: usize = 10_000; +const MAX_CACHED_HITS: usize = 250; + +#[derive(Debug, Clone)] +pub struct ElasticsearchConfig { + pub url: String, + pub username: String, + pub password: String, + pub index_prefix: String, + pub meta_namespace: String, + pub index_chunk_size: i64, + pub bulk_batch_size: usize, + pub project_import_batch_size: usize, + pub typesense_parity: bool, +} + +impl ElasticsearchConfig { + pub fn new(meta_namespace: Option) -> Self { + Self { + url: ENV.ELASTICSEARCH_URL.clone(), + username: ENV.ELASTICSEARCH_USERNAME.clone(), + password: ENV.ELASTICSEARCH_PASSWORD.clone(), + index_prefix: ENV.ELASTICSEARCH_INDEX_PREFIX.clone(), + meta_namespace: meta_namespace.unwrap_or_default(), + index_chunk_size: ENV.SEARCH_INDEX_CHUNK_SIZE, + bulk_batch_size: ENV.ELASTICSEARCH_BULK_BATCH_SIZE, + project_import_batch_size: + typesense_parity::project_import_batch_size(), + typesense_parity: ENV.ELASTICSEARCH_TYPESENSE_PARITY, + } + } + + fn alias_name(&self) -> String { + if self.meta_namespace.is_empty() { + format!("{}_projects", self.index_prefix) + } else { + format!( + "{}_{}_projects", + self.meta_namespace, self.index_prefix + ) + } + } + + fn next_index_name(&self, alias: &str, use_alt: bool) -> String { + if use_alt { + format!("{alias}__alt") + } else { + format!("{alias}__current") + } + } +} + +struct ElasticsearchClient { + client: reqwest::Client, + base_url: String, + username: String, + password: String, +} + +impl ElasticsearchClient { + fn new(config: &ElasticsearchConfig) -> Self { + Self { + client: reqwest::Client::new(), + base_url: config.url.trim_end_matches('/').to_string(), + username: config.username.clone(), + password: config.password.clone(), + } + } + + fn request(&self, method: Method, path: &str) -> reqwest::RequestBuilder { + let request = self + .client + .request(method, format!("{}{}", self.base_url, path)); + if self.username.is_empty() { + request + } else { + request.basic_auth(&self.username, Some(&self.password)) + } + } + + async fn get_alias_target(&self, alias: &str) -> Result> { + let response = self + .request(Method::GET, &format!("/_alias/{alias}")) + .send() + .await + .wrap_err("failed to get Elasticsearch alias")?; + if response.status() == StatusCode::NOT_FOUND { + return Ok(None); + } + let body = response_json(response, "get Elasticsearch alias").await?; + Ok(body + .as_object() + .and_then(|indices| indices.keys().next()) + .cloned()) + } + + async fn index_exists(&self, index: &str) -> Result { + let response = self + .request(Method::HEAD, &format!("/{index}")) + .send() + .await + .wrap_err("failed to check Elasticsearch index existence")?; + Ok(response.status().is_success()) + } + + async fn create_index(&self, index: &str, schema: &Value) -> Result<()> { + let response = self + .request(Method::PUT, &format!("/{index}")) + .json(schema) + .send() + .await + .wrap_err("failed to create Elasticsearch index")?; + response_json(response, "create Elasticsearch index").await?; + Ok(()) + } + + async fn delete_index_if_exists(&self, index: &str) -> Result<()> { + let response = self + .request(Method::DELETE, &format!("/{index}")) + .send() + .await + .wrap_err("failed to delete Elasticsearch index")?; + if response.status() == StatusCode::NOT_FOUND { + return Ok(()); + } + response_json(response, "delete Elasticsearch index").await?; + Ok(()) + } + + async fn swap_alias( + &self, + alias: &str, + old_index: Option<&str>, + new_index: &str, + ) -> Result<()> { + let mut actions = Vec::new(); + if let Some(old_index) = old_index { + actions.push(json!({ + "remove": {"index": old_index, "alias": alias} + })); + } + actions.push(json!({ + "add": { + "index": new_index, + "alias": alias, + "is_write_index": true + } + })); + + let response = self + .request(Method::POST, "/_aliases") + .json(&json!({"actions": actions})) + .send() + .await + .wrap_err("failed to swap Elasticsearch alias")?; + response_json(response, "swap Elasticsearch alias").await?; + Ok(()) + } + + async fn bulk(&self, index: &str, body: String) -> Result<()> { + let response = self + .request( + Method::POST, + &format!("/{index}/_bulk?refresh=false"), + ) + .header("Content-Type", "application/x-ndjson") + .body(body) + .send() + .await + .wrap_err("failed to execute Elasticsearch bulk request")?; + let body = + response_json(response, "execute Elasticsearch bulk request") + .await?; + if body["errors"].as_bool() == Some(true) { + let failures = body["items"] + .as_array() + .into_iter() + .flatten() + .filter_map(|item| { + item.as_object()? + .values() + .next()? + .get("error") + .cloned() + }) + .unique() + .take(10) + .map(|error| error.to_string()) + .join("; "); + return Err(eyre!( + "Elasticsearch bulk request contained failures: {failures}" + )); + } + Ok(()) + } + + async fn delete_by_query( + &self, + index: &str, + query: &Value, + ) -> Result<()> { + let response = self + .request( + Method::POST, + &format!( + "/{index}/_delete_by_query?conflicts=proceed&refresh=false" + ), + ) + .json(&json!({"query": query})) + .send() + .await + .wrap_err("failed to delete Elasticsearch documents")?; + let body = + response_json(response, "delete Elasticsearch documents").await?; + if body["failures"] + .as_array() + .is_some_and(|failures| !failures.is_empty()) + { + return Err(eyre!( + "Elasticsearch delete-by-query contained failures: {}", + body["failures"] + )); + } + Ok(()) + } + + async fn refresh(&self, index: &str) -> Result<()> { + let response = self + .request(Method::POST, &format!("/{index}/_refresh")) + .send() + .await + .wrap_err("failed to refresh Elasticsearch index")?; + response_json(response, "refresh Elasticsearch index").await?; + Ok(()) + } + + async fn force_merge(&self, index: &str) -> Result<()> { + let response = self + .request( + Method::POST, + &format!( + "/{index}/_forcemerge?max_num_segments=1&flush=true" + ), + ) + .send() + .await + .wrap_err("failed to force-merge Elasticsearch index")?; + response_json(response, "force-merge Elasticsearch index").await?; + Ok(()) + } +} + +async fn response_json( + response: Response, + operation: &str, +) -> Result { + let status = response.status(); + let body = response + .text() + .await + .wrap_err_with(|| format!("failed to read response for {operation}"))?; + let json = serde_json::from_str(&body).unwrap_or_else(|_| { + json!({ + "unparsed_response": body + }) + }); + if !status.is_success() { + return Err(eyre!("{operation} failed ({status}): {json}")); + } + Ok(json) +} + +pub struct Elasticsearch { + pub config: ElasticsearchConfig, + client: ElasticsearchClient, +} + +impl Elasticsearch { + pub fn new(config: ElasticsearchConfig) -> Self { + let client = ElasticsearchClient::new(&config); + Self { config, client } + } + + + fn sort(index: SearchIndex) -> Vec { + let descending = |field: &str| { + json!({(field): {"order": "desc", "missing": "_last"}}) + }; + let mut sort = match index { + SearchIndex::Relevance => vec![ + json!({"_score": {"order": "desc"}}), + descending("log_downloads"), + descending("version_published_timestamp"), + ], + SearchIndex::Downloads => vec![ + descending("log_downloads"), + descending("version_published_timestamp"), + ], + SearchIndex::Follows => vec![ + descending("follows"), + descending("version_published_timestamp"), + ], + SearchIndex::Updated => vec![ + descending("modified_timestamp"), + descending("version_published_timestamp"), + ], + SearchIndex::Newest => vec![ + descending("created_timestamp"), + descending("version_published_timestamp"), + ], + SearchIndex::MinecraftJavaServerVerifiedPlays2w => vec![ + json!({"_score": {"order": "desc"}}), + descending("minecraft_java_server.verified_plays_2w"), + descending("minecraft_java_server.is_online"), + ], + SearchIndex::MinecraftJavaServerPlayersOnline => vec![ + json!({"_score": {"order": "desc"}}), + descending("minecraft_java_server.is_online"), + descending( + "minecraft_java_server.ping.data.players_online", + ), + ], + }; + sort.push(json!({"project_id": {"order": "asc"}})); + sort + } + + fn build_filter( + info: &SearchRequest, + ) -> Result, ApiError> { + let facet_part = if let Some(facets_json) = info.facets.as_deref() { + from_legacy_v2_facets_json(facets_json) + .wrap_request_err("failed to parse facets")? + } else { + None + }; + + let filter_part = combined_search_filters(info) + .filter(|filter| !filter.trim().is_empty()) + .map(|filter| parse_expression(&filter)) + .transpose() + .wrap_request_err("failed to parse filters")?; + + FilterExpr::and([facet_part, filter_part].into_iter().flatten()) + .map(normalize) + .map(|filter| { + serialize_filter(&filter) + .wrap_request_err("failed to build search filter") + }) + .transpose() + } + + async fn execute_search( + &self, + alias: &str, + body: &Value, + sort_only: bool, + ) -> Result { + let request_cache = if body["size"] + .as_u64() + .is_some_and(|size| size <= MAX_CACHED_HITS as u64) + { + "&request_cache=true" + } else { + "" + }; + let path = if sort_only { + format!( + "/{alias}/_search?filter_path=hits.total,hits.hits.sort&preference=labrinth{request_cache}" + ) + } else { + format!( + "/{alias}/_search?preference=labrinth{request_cache}" + ) + }; + let response = self + .client + .request(Method::POST, &path) + .json(body) + .send() + .await + .wrap_internal_err("failed to execute Elasticsearch search")?; + let mut body = response_json(response, "execute Elasticsearch search") + .await + .map_err(ApiError::Internal)?; + if sort_only && !body["hits"]["hits"].is_array() { + body["hits"]["hits"] = Value::Array(Vec::new()); + } + Ok(body) + } + + async fn open_point_in_time( + &self, + alias: &str, + ) -> Result { + let response = self + .client + .request( + Method::POST, + &format!( + "/{alias}/_pit?keep_alive=1m&preference=labrinth" + ), + ) + .send() + .await + .wrap_internal_err( + "failed to open Elasticsearch point in time", + )?; + let body = response_json(response, "open Elasticsearch point in time") + .await + .map_err(ApiError::Internal)?; + body["id"] + .as_str() + .map(ToOwned::to_owned) + .ok_or_else(|| { + ApiError::Internal(eyre!( + "Elasticsearch point in time response did not contain an ID" + )) + }) + } + + async fn close_point_in_time(&self, id: &str) -> Result<(), ApiError> { + let response = self + .client + .request(Method::DELETE, "/_pit") + .json(&json!({"id": id})) + .send() + .await + .wrap_internal_err( + "failed to close Elasticsearch point in time", + )?; + response_json(response, "close Elasticsearch point in time") + .await + .map_err(ApiError::Internal)?; + Ok(()) + } + + async fn execute_point_in_time_search( + &self, + body: &Value, + sort_only: bool, + ) -> Result { + let path = if sort_only { + "/_search?filter_path=pit_id,hits.total,hits.hits.sort" + } else { + "/_search" + }; + let response = self + .client + .request(Method::POST, path) + .json(body) + .send() + .await + .wrap_internal_err( + "failed to execute Elasticsearch point in time search", + )?; + let mut body = response_json( + response, + "execute Elasticsearch point in time search", + ) + .await + .map_err(ApiError::Internal)?; + if sort_only && !body["hits"]["hits"].is_array() { + body["hits"]["hits"] = Value::Array(Vec::new()); + } + Ok(body) + } + + #[allow(clippy::too_many_arguments)] + async fn execute_deep_search( + &self, + alias: &str, + query: &Value, + sort: &[Value], + offset: usize, + size: usize, + ) -> Result { + let mut point_in_time_id = self.open_point_in_time(alias).await?; + let result = self + .execute_deep_search_with_point_in_time( + query, + sort, + offset, + size, + &mut point_in_time_id, + ) + .await; + if let Err(error) = self.close_point_in_time(&point_in_time_id).await { + warn!( + ?error, + "failed to close Elasticsearch deep-search point in time" + ); + } + result + } + + async fn execute_deep_search_with_point_in_time( + &self, + query: &Value, + sort: &[Value], + offset: usize, + size: usize, + point_in_time_id: &mut String, + ) -> Result { + let mut remaining_offset = offset; + let mut search_after = None; + let mut total_hits = None; + while remaining_offset > 0 { + let skipped = remaining_offset.min(MAX_RESULT_WINDOW); + let mut body = Self::search_body( + query, + sort, + 0, + skipped, + total_hits.is_none(), + search_after.as_ref(), + false, + ); + body["pit"] = json!({ + "id": point_in_time_id, + "keep_alive": "1m" + }); + let body = self + .execute_point_in_time_search(&body, true) + .await?; + if let Some(id) = body["pit_id"].as_str() { + *point_in_time_id = id.to_string(); + } + if total_hits.is_none() { + let exact_total_hits = body["hits"]["total"]["value"] + .as_u64() + .unwrap_or_default() + as usize; + if offset >= exact_total_hits { + return Ok(json!({ + "hits": { + "total": { + "value": exact_total_hits, + "relation": "eq" + }, + "hits": [] + } + })); + } + total_hits = Some(exact_total_hits); + } + let pagination_hits = body["hits"]["hits"].as_array(); + let Some(anchor) = pagination_hits + .and_then(|hits| hits.last()) + .and_then(|hit| hit.get("sort")) + .cloned() + else { + return Ok(json!({ + "hits": { + "total": { + "value": total_hits.unwrap_or_default(), + "relation": "eq" + }, + "hits": [] + } + })); + }; + debug!( + requested_offset = offset, + remaining_offset, + skipped, + hit_count = pagination_hits.map_or(0, Vec::len), + ?anchor, + "advanced Elasticsearch search-after cursor" + ); + search_after = Some(anchor); + remaining_offset -= skipped; + } + + let mut body = Self::search_body( + query, + sort, + 0, + size, + true, + search_after.as_ref(), + true, + ); + body["pit"] = json!({ + "id": point_in_time_id, + "keep_alive": "1m" + }); + let body = self + .execute_point_in_time_search(&body, false) + .await?; + if let Some(id) = body["pit_id"].as_str() { + *point_in_time_id = id.to_string(); + } + Ok(body) + } + + async fn has_matches( + &self, + alias: &str, + query: &Value, + ) -> Result { + let body = json!({ + "_source": false, + "size": 1, + "terminate_after": 1, + "track_total_hits": false, + "query": query + }); + let response = self.execute_search(alias, &body, false).await?; + Ok(response["hits"]["hits"] + .as_array() + .is_some_and(|hits| !hits.is_empty())) + } + + async fn count_matches( + &self, + alias: &str, + query: &Value, + ) -> Result { + let response = self + .client + .request(Method::POST, &format!("/{alias}/_count")) + .json(&json!({"query": query})) + .send() + .await + .wrap_internal_err("failed to count Elasticsearch matches")?; + let body = response_json(response, "count Elasticsearch matches") + .await + .map_err(ApiError::Internal)?; + Ok(body["count"].as_u64().unwrap_or_default() as usize) + } + + #[allow(clippy::too_many_arguments)] + + fn search_body( + query: &Value, + sort: &[Value], + from: usize, + size: usize, + track_total_hits: bool, + search_after: Option<&Value>, + include_source: bool, + ) -> Value { + let mut body = json!({ + "from": from, + "size": size, + "track_total_hits": track_total_hits, + "query": query, + "sort": sort + }); + if let Some(search_after) = search_after { + body["search_after"] = search_after.clone(); + } + if !include_source { + body["_source"] = Value::Bool(false); + } + body + } + + async fn existing_write_indices(&self) -> Result> { + let alias = self.config.alias_name(); + let mut indices = self + .client + .get_alias_target(&alias) + .await? + .into_iter() + .collect_vec(); + + for index in [ + self.config.next_index_name(&alias, false), + self.config.next_index_name(&alias, true), + ] { + if !indices.contains(&index) + && self.client.index_exists(&index).await? + { + indices.push(index); + } + } + Ok(indices) + } + + + async fn import_versions( + &self, + indices: &[String], + documents: &[UploadSearchVersion], + ) -> Result<()> { + let batch_size = self.config.bulk_batch_size.max(1); + for documents in documents.chunks(batch_size) { + let body = versions_to_bulk(documents)?; + for index in indices { + info!( + index, + document_count = documents.len(), + content_length_bytes = body.len(), + "sending Elasticsearch version bulk request" + ); + self.client.bulk(index, body.clone()).await?; + } + } + Ok(()) + } + + async fn delete_ids( + &self, + field: &str, + ids: &[String], + ) -> Result<()> { + let indices = self.existing_write_indices().await?; + for ids in ids.chunks(DELETE_FILTER_ID_BATCH_SIZE) { + let query = json!({"terms": {(field): ids}}); + for index in &indices { + self.client.delete_by_query(index, &query).await?; + } + } + Ok(()) + } + + async fn refresh_write_indices(&self) -> Result<()> { + for index in self.existing_write_indices().await? { + self.client.refresh(&index).await?; + } + Ok(()) + } + + fn native_text_query(query: &str) -> Value { + if query.is_empty() || query.trim() == "*" { + return json!({"match_all": {}}); + } + + json!({ + "multi_match": { + "query": query, + "fields": [ + "name^15", + "indexed_name^15", + "slug^10", + "author^3", + "indexed_author^3", + "summary" + ], + "type": "best_fields", + "operator": "and", + "fuzziness": "AUTO" + } + }) + } + + async fn search_for_project_raw_native( + &self, + info: &SearchRequest, + ) -> Result { + let parsed = parse_search_request(info)?; + let search_sort = + parse_search_index(parsed.index, info.new_filters.as_deref())?; + let filter = Self::build_filter(info)?; + let mut filters = + vec![json!({"term": {"document_type": "project"}})]; + if let Some(filter) = &filter { + filters.push(filter.query.clone()); + } + let query = json!({ + "bool": { + "must": [Self::native_text_query(parsed.query)], + "filter": filters + } + }); + let sort = Self::sort(search_sort.index); + let alias = self.config.alias_name(); + let body = if parsed + .offset + .saturating_add(parsed.hits_per_page) + > MAX_RESULT_WINDOW + { + self.execute_deep_search( + &alias, + &query, + &sort, + parsed.offset, + parsed.hits_per_page, + ) + .await? + } else { + let body = Self::search_body( + &query, + &sort, + parsed.offset, + parsed.hits_per_page, + true, + None, + true, + ); + self.execute_search(&alias, &body, false).await? + }; + let total_hits = body["hits"]["total"]["value"] + .as_u64() + .unwrap_or_default() as usize; + let hits = body["hits"]["hits"] + .as_array() + .into_iter() + .flatten() + .filter_map(|hit| { + let mut document = hit["_source"].clone(); + let object = document.as_object_mut()?; + object.remove("document_type"); + object.remove("_search_tokens"); + if filter + .as_ref() + .is_some_and(|filter| filter.has_version_filter) + && let Some(version_id) = matching_version_id(hit) + { + object.insert( + "version_id".to_string(), + Value::String(version_id), + ); + } + + let metadata = info.show_metadata.then(|| { + json!({ + "score": hit["_score"], + "sort": hit["sort"] + }) + }); + let mut result: ResultSearchProject = + serde_json::from_value::(document) + .ok()? + .into(); + result.search_metadata = metadata; + Some(result) + }) + .collect(); + + Ok(SearchResults { + hits, + page: parsed.page, + hits_per_page: parsed.hits_per_page, + total_hits, + }) + } +} + +#[async_trait] +impl SearchBackend for Elasticsearch { + async fn search_for_project_raw( + &self, + info: &SearchRequest, + ) -> Result { + if self.config.typesense_parity { + self.search_for_project_raw_typesense_parity(info).await + } else { + self.search_for_project_raw_native(info).await + } + } + + async fn rebuild_index( + &self, + ro_pool: PgPool, + redis: RedisPool, + ) -> Result<()> { + info!("starting Elasticsearch project indexing"); + let alias = self.config.alias_name(); + let current = self.client.get_alias_target(&alias).await?; + let use_alt = + !current.as_deref().is_some_and(|name| name.ends_with("__alt")); + let next = self.config.next_index_name(&alias, use_alt); + + info!(index = next, "creating Elasticsearch shadow index"); + self.client.delete_index_if_exists(&next).await?; + self.client + .create_index(&next, &Self::index_schema()) + .await?; + + let mut cursor = 0_i64; + let mut chunk_index = 0_usize; + let mut total_projects = 0_usize; + let mut total_versions = 0_usize; + + loop { + info!("fetching index chunk {chunk_index}"); + chunk_index += 1; + let (documents, next_cursor) = index_local( + &ro_pool, + &redis, + cursor, + self.config.index_chunk_size, + ) + .await + .wrap_err("failed to fetch projects from local DB")?; + if documents.projects.is_empty() { + info!( + "no more documents; indexed {total_projects} projects and {total_versions} versions in {chunk_index} chunks" + ); + break; + } + + total_projects += documents.projects.len(); + total_versions += documents.versions.len(); + cursor = next_cursor; + self.import_projects( + std::slice::from_ref(&next), + &documents.projects, + ) + .await?; + self.import_versions( + std::slice::from_ref(&next), + &documents.versions, + ) + .await?; + } + + self.client.refresh(&next).await?; + info!("force-merging Elasticsearch shadow index"); + self.client.force_merge(&next).await?; + info!("swapping Elasticsearch index alias"); + self.client + .swap_alias(&alias, current.as_deref(), &next) + .await?; + if let Some(old) = current { + self.client.delete_index_if_exists(&old).await?; + } + info!("Elasticsearch indexing complete"); + Ok(()) + } + + async fn apply_update( + &self, + update: SearchIndexUpdate<'_>, + ) -> Result<()> { + let removed_project_ids = update + .removed_projects + .iter() + .map(ToString::to_string) + .collect::>(); + if !removed_project_ids.is_empty() { + self.delete_ids("project_id", &removed_project_ids).await?; + } + + let version_ids = update + .removed_versions + .iter() + .map(ToString::to_string) + .collect::>(); + if !version_ids.is_empty() { + self.delete_ids("version_id", &version_ids).await?; + } + + let indices = self.existing_write_indices().await?; + if !update.projects.is_empty() { + debug!( + ?indices, + num_documents = update.projects.len(), + "replacing Elasticsearch project documents" + ); + self.import_projects(&indices, update.projects).await?; + } + if !update.versions.is_empty() { + debug!( + ?indices, + num_documents = update.versions.len(), + "replacing Elasticsearch version documents" + ); + self.import_versions(&indices, update.versions).await?; + } + self.refresh_write_indices().await?; + debug!("done applying Elasticsearch search index update"); + Ok(()) + } + + async fn tasks(&self) -> Result { + let response = self + .client + .request(Method::GET, "/_tasks?detailed=true") + .send() + .await + .wrap_err("failed to get Elasticsearch tasks")?; + response_json(response, "get Elasticsearch tasks").await + } + + async fn tasks_cancel(&self, filter: &TasksCancelFilter) -> Result<()> { + match filter { + TasksCancelFilter::All => { + let response = self + .client + .request(Method::POST, "/_tasks/_cancel") + .send() + .await + .wrap_err("failed to cancel Elasticsearch tasks")?; + response_json(response, "cancel Elasticsearch tasks").await?; + } + TasksCancelFilter::AllEnqueued => { + // Elasticsearch executes operations immediately and does not + // expose an enqueued-task state. + } + TasksCancelFilter::Indexes { indexes } => { + let tasks = self.tasks().await?; + for (_node_id, node) in tasks["nodes"] + .as_object() + .into_iter() + .flatten() + { + for (task_id, task) in node["tasks"] + .as_object() + .into_iter() + .flatten() + { + let description = + task["description"].as_str().unwrap_or_default(); + if indexes + .iter() + .any(|index| description.contains(index)) + { + let response = self + .client + .request( + Method::POST, + &format!( + "/_tasks/{task_id}/_cancel" + ), + ) + .send() + .await + .wrap_err( + "failed to cancel Elasticsearch task", + )?; + response_json( + response, + "cancel Elasticsearch task", + ) + .await?; + } + } + } + } + } + Ok(()) + } +} + +fn matching_version_id(hit: &Value) -> Option { + hit["inner_hits"] + .as_object()? + .values() + .filter_map(|inner_hits| { + let source = inner_hits["hits"]["hits"] + .as_array()? + .first()? + .get("_source")?; + Some(( + source["version_published_timestamp"].as_i64()?, + source["version_id"].as_str()?.to_string(), + )) + }) + .max_by_key(|(published, _)| *published) + .map(|(_, version_id)| version_id) +} + +fn versions_to_bulk(documents: &[UploadSearchVersion]) -> Result { + let mut output = String::new(); + for document in documents { + let id = format!("version:{}", document.version_id); + push_json_line( + &mut output, + &json!({ + "index": { + "_id": id, + "routing": document.project_id + } + }), + )?; + + let mut source = serde_json::to_value(document) + .wrap_err("failed to serialize `UploadSearchVersion`")?; + source + .as_object_mut() + .ok_or_else(|| eyre!("version search document is not an object"))? + .insert( + "document_type".to_string(), + json!({ + "name": "version", + "parent": format!("project:{}", document.project_id) + }), + ); + push_json_line(&mut output, &source)?; + } + Ok(output) +} + +fn add_server_online_field(object: &mut Map) { + let Some(server) = object + .get_mut("minecraft_java_server") + .and_then(Value::as_object_mut) + else { + return; + }; + let is_online = server + .get("ping") + .and_then(Value::as_object) + .and_then(|ping| ping.get("data")) + .is_some_and(|data| !data.is_null()); + server.insert("is_online".to_string(), Value::Bool(is_online)); +} + +fn push_json_line( + output: &mut String, + value: &T, +) -> Result<()> { + output.push_str(&serde_json::to_string(value)?); + output.push('\n'); + Ok(()) +} diff --git a/apps/labrinth/src/search/backend/elasticsearch/typesense_parity.rs b/apps/labrinth/src/search/backend/elasticsearch/typesense_parity.rs new file mode 100644 index 0000000000..2e5801a456 --- /dev/null +++ b/apps/labrinth/src/search/backend/elasticsearch/typesense_parity.rs @@ -0,0 +1,2223 @@ +//! Optional Typesense-compatible query and indexing behavior. + +use super::*; + +use std::{ + cmp::Ordering, + collections::BTreeMap, +}; + +use itertools::Itertools; +use serde_json::{Value, json}; +use unicode_normalization::{ + UnicodeNormalization, + char::is_combining_mark, +}; + +use crate::search::backend::{ + SearchIndex, + typesense::Bucketing, +}; + +pub(super) const BUCKETED_HITS: usize = 250; +pub(super) const INTERNAL_INDEX_BATCH_SIZE: usize = 1000; +pub(super) const MAX_CANDIDATE_QUERY_LENGTH: usize = 64; +pub(super) const CANDIDATE_AGGREGATION_MULTIPLIER: usize = 128; +pub(super) const CANDIDATE_SHARD_MULTIPLIER: usize = 3; +pub(super) const SEARCH_TEXT_FIELDS: [(&str, u8); 6] = [ + ("name", 15), + ("indexed_name", 15), + ("slug", 10), + ("author", 3), + ("indexed_author", 3), + ("summary", 1), +]; +pub(super) const SEARCH_CANDIDATE_FIELDS: [(&str, &str); 6] = [ + ("name.prefix", "_search_tokens.name"), + ("indexed_name.prefix", "indexed_name"), + ("slug.prefix", "_search_tokens.slug"), + ("author.prefix", "_search_tokens.author"), + ("indexed_author.prefix", "_search_tokens.indexed_author"), + ("summary.prefix", "_search_tokens.summary"), +]; +pub(super) const CANDIDATE_SCRIPT: &str = r#" +String query = params.query; +int cost = Integer.parseInt(params.cost); +int queryLength = query.length(); +for (def candidateValue : doc[params.field]) { + String candidate = candidateValue.toString(); + int minimumLength = candidate.length() < queryLength + ? candidate.length() + : queryLength; + int maximumLength = queryLength + cost; + if (maximumLength > candidate.length()) { + maximumLength = candidate.length(); + } + int[] previousPrevious = new int[maximumLength + 1]; + int[] previous = new int[maximumLength + 1]; + int[] current = new int[maximumLength + 1]; + for (int right = 0; right <= maximumLength; right++) { + previous[right] = right; + } + for (int left = 1; left <= queryLength; left++) { + current[0] = left; + for (int right = 1; right <= maximumLength; right++) { + int substitution = + query.charAt(left - 1) == candidate.charAt(right - 1) + ? 0 + : 1; + int distance = previous[right] + 1; + int insertion = current[right - 1] + 1; + if (insertion < distance) { + distance = insertion; + } + int substitutionDistance = + previous[right - 1] + substitution; + if (substitutionDistance < distance) { + distance = substitutionDistance; + } + if ( + left > 1 + && right > 1 + && query.charAt(left - 1) == candidate.charAt(right - 2) + && query.charAt(left - 2) == candidate.charAt(right - 1) + ) { + int transposition = + previousPrevious[right - 2] + 1; + if (transposition < distance) { + distance = transposition; + } + } + current[right] = distance; + } + int[] swap = previousPrevious; + previousPrevious = previous; + previous = current; + current = swap; + } + boolean matched = false; + for ( + int candidateLength = minimumLength; + candidateLength <= maximumLength && !matched; + candidateLength++ + ) { + matched = previous[candidateLength] == cost; + } + if (matched) { + emit(candidate); + } +} +"#; + +const TWO_TYPO_WEIGHT_SCALE: f64 = 16.0; + +pub(super) fn project_import_batch_size() -> usize { + ENV.TYPESENSE_IMPORT_BATCH_SIZE +} + +pub(super) struct CandidateSelection { + pub(super) terms: Vec, + pub(super) cost: usize, + pub(super) boundary_overshoot: bool, +} + +struct CandidateTerm { + term: String, + score: f64, + batch_score: f64, +} + +#[derive(Default)] +struct CandidateTrieNode { + children: BTreeMap, + term: Option, + batch_score: f64, + path: Vec, +} + +pub(super) fn text_query( + query: &str, + allow_typos: bool, + two_typo_max_expansions: usize, +) -> Value { + if query.is_empty() || query.trim() == "*" { + return json!({"match_all": {}}); + } + + let tokens = query.split_whitespace().collect_vec(); + if tokens.is_empty() { + return json!({"match_all": {}}); + } + let token_queries = |token: &str, prefix: bool| { + let mut queries = Vec::with_capacity(SEARCH_TEXT_FIELDS.len() * 4); + let token_length = token.chars().count(); + for ((field, weight), (_, candidate_field)) in + SEARCH_TEXT_FIELDS.into_iter().zip(SEARCH_CANDIDATE_FIELDS) + { + if field != "indexed_name" { + let normalized_token = + tokenize_candidate_text( + token, + field == "name" || field == "author", + ) + .into_iter() + .next() + .unwrap_or_else(|| token.to_lowercase()); + queries.push(json!({ + "constant_score": { + "filter": { + "bool": { + "filter": [{ + "match": { + (field): { + "query": token, + "fuzziness": 0 + } + } + }, { + "term": { + (candidate_field): normalized_token + } + }] + } + }, + "boost": weight + 1 + } + })); + } + if prefix { + let prefix_query = if field == "indexed_name" { + json!({ + "prefix": { + (field): {"value": token.to_lowercase()} + } + }) + } else { + json!({ + "match_bool_prefix": { + (field): {"query": token} + } + }) + }; + queries.push(json!({ + "constant_score": { + "filter": prefix_query, + "boost": weight as f64 + 0.75 + } + })); + } + if allow_typos { + let max_expansions = + if field == "indexed_name" && token_length >= 7 { + 1 + } else if field == "indexed_name" { + 4 + } else { + 2 + }; + queries.push(json!({ + "constant_score": { + "filter": { + "match": { + (field): { + "query": token, + "fuzziness": 1, + "prefix_length": 1, + "max_expansions": max_expansions + } + } + }, + "boost": weight as f64 + 0.5 + } + })); + if field == "name" { + let transposed_prefix_query = json!({ + "match": { + "name.prefix": { + "query": token, + "fuzziness": 1, + "prefix_length": 1, + "max_expansions": 4, + "fuzzy_transpositions": true + } + } + }); + queries.push(json!({ + "constant_score": { + "filter": &transposed_prefix_query, + "boost": weight as f64 + 0.25 + } + })); + queries.push(json!({ + "constant_score": { + "filter": { + "bool": { + "must": [ + transposed_prefix_query, + { + "match": { + "name": { + "query": token, + "fuzziness": 2, + "prefix_length": 1, + "max_expansions": 24 + } + } + } + ], + "must_not": [{ + "match": { + "name.prefix": { + "query": token, + "fuzziness": 1, + "prefix_length": 1, + "max_expansions": 4, + "fuzzy_transpositions": false + } + } + }] + } + }, + "boost": weight as f64 + 0.3 + } + })); + } + queries.push(json!({ + "constant_score": { + "filter": { + "match": { + (field): { + "query": token, + "fuzziness": "AUTO:4,7", + "prefix_length": 1, + "max_expansions": + two_typo_max_expansions + } + } + }, + "boost": weight as f64 / TWO_TYPO_WEIGHT_SCALE + } + })); + } + } + if allow_typos && tokens.len() == 1 { + let chars = token.chars().collect_vec(); + if chars.len() >= 8 { + let split_at = chars.len().div_ceil(2); + let left = chars[..split_at].iter().collect::(); + let right = chars[split_at..].iter().collect::(); + queries.push(json!({ + "constant_score": { + "filter": { + "intervals": { + "name": { + "all_of": { + "ordered": true, + "max_gaps": 0, + "intervals": [ + { + "fuzzy": { + "term": left, + "fuzziness": 1, + "prefix_length": 1, + "transpositions": true + } + }, + { + "prefix": { + "prefix": right + } + } + ] + } + } + } + }, + "boost": 15.25 + } + })); + } + } + queries + }; + let queries_by_token = tokens + .iter() + .enumerate() + .map(|(index, token)| { + let is_last = index == tokens.len() - 1; + token_queries(token, is_last) + }) + .collect_vec(); + if tokens.len() == 1 { + json!({ + "dis_max": { + "queries": queries_by_token + .into_iter() + .flatten() + .collect_vec(), + "tie_breaker": 0 + } + }) + } else { + json!({ + "bool": { + "must": queries_by_token + .into_iter() + .map(|queries| { + json!({ + "dis_max": { + "queries": queries, + "tie_breaker": 0 + } + }) + }) + .collect_vec() + } + }) + } +} + +pub(super) fn uses_text_match_bucketing(index: SearchIndex) -> bool { + matches!( + index, + SearchIndex::Relevance + | SearchIndex::MinecraftJavaServerVerifiedPlays2w + | SearchIndex::MinecraftJavaServerPlayersOnline + ) +} + +pub(super) fn bucket_size( + bucketing: &Bucketing, + hit_count: usize, +) -> Option { + if hit_count == 0 { + return None; + } + + match bucketing { + Bucketing::Buckets(count) => { + let count = usize::try_from(*count).ok()?; + if count == 0 { + return None; + } + Some(hit_count.div_ceil(count)) + } + Bucketing::BucketSize(size) => { + let size = usize::try_from(*size).ok()?; + (size > 0).then_some(size) + } + } +} + +fn compare_descending_sort_values(left: &Value, right: &Value) -> Ordering { + match (left, right) { + (Value::Null, Value::Null) => Ordering::Equal, + (Value::Null, _) => Ordering::Greater, + (_, Value::Null) => Ordering::Less, + (Value::Number(left), Value::Number(right)) => right + .as_f64() + .and_then(|right| { + left.as_f64() + .and_then(|left| right.partial_cmp(&left)) + }) + .unwrap_or(Ordering::Equal), + (Value::Bool(left), Value::Bool(right)) => right.cmp(left), + (Value::String(left), Value::String(right)) => right.cmp(left), + _ => Ordering::Equal, + } +} + +fn compare_bucket_hits(left: &Value, right: &Value) -> Ordering { + let Some(left) = left["sort"].as_array() else { + return Ordering::Equal; + }; + let Some(right) = right["sort"].as_array() else { + return Ordering::Equal; + }; + + left.iter() + .zip(right) + .skip(1) + .take(left.len().saturating_sub(2)) + .map(|(left, right)| compare_descending_sort_values(left, right)) + .find(|ordering| !ordering.is_eq()) + .unwrap_or(Ordering::Equal) +} + +pub(super) fn rerank_bucketed_hits( + hits: &mut [Value], + bucketing: &Bucketing, + candidate_count: usize, + used_fuzzy_query: bool, +) { + let use_narrow_full_bucket = candidate_count >= 500; + let candidate_count = candidate_count.min(BUCKETED_HITS); + let Some(mut bucket_size) = bucket_size(bucketing, candidate_count) else { + return; + }; + if used_fuzzy_query + && use_narrow_full_bucket + && matches!(bucketing, Bucketing::Buckets(_)) + { + bucket_size = bucket_size.saturating_sub(1).max(1); + } + let hit_count = hits.len().min(BUCKETED_HITS); + for bucket in hits[..hit_count].chunks_mut(bucket_size) { + bucket.sort_by(compare_bucket_hits); + } +} + +pub(super) fn selected_candidate_text_query(terms: &[String]) -> Value { + json!({ + "dis_max": { + "queries": SEARCH_TEXT_FIELDS + .into_iter() + .map(|(field, weight)| { + json!({ + "constant_score": { + "filter": { + "terms": {(field): terms} + }, + "boost": weight + } + }) + }) + .collect_vec(), + "tie_breaker": 0 + } + }) +} + +pub(super) fn stemmed_single_name_query(query: &str) -> Value { + json!({ + "constant_score": { + "filter": { + "bool": { + "must": [{ + "prefix": { + "_search_tokens.indexed_name": { + "value": query.to_lowercase() + } + } + }, { + "match": { + "indexed_name": { + "query": query, + "fuzziness": 0 + } + } + }], + "must_not": [{ + "wildcard": { + "_search_tokens.indexed_name": { + "value": "*-*" + } + } + }] + } + }, + "boost": 16 + } + }) +} + +pub(super) fn candidate_token_text_query( + terms: &[String], + cost: usize, + query_length: usize, +) -> Value { + let (short_terms, long_terms): (Vec<_>, Vec<_>) = terms + .iter() + .cloned() + .partition(|term| term.chars().count() <= query_length + 1); + let mut queries = Vec::new(); + for ((_, field), (_, weight)) in SEARCH_CANDIDATE_FIELDS + .into_iter() + .zip(SEARCH_TEXT_FIELDS) + { + let mut add_query = |terms: &[String], boost: f64| { + if !terms.is_empty() { + queries.push(json!({ + "constant_score": { + "filter": { + "terms": {(field): terms} + }, + "boost": boost + } + })); + } + }; + match cost { + 0 => add_query(terms, weight as f64 + 0.75), + 1 => { + add_query(&short_terms, weight as f64 + 0.5); + add_query(&long_terms, weight as f64 + 0.25); + } + _ => { + add_query(terms, weight as f64 / TWO_TYPO_WEIGHT_SCALE); + } + } + } + json!({ + "dis_max": { + "queries": queries, + "tie_breaker": 0 + } + }) +} + +pub(super) fn rank_candidate_buckets( + buckets: &[Value], + query: &str, + cost: usize, + max_candidates: usize, +) -> Vec { + let candidates = buckets + .iter() + .filter(|bucket| { + bucket["allowed"]["doc_count"] + .as_u64() + .unwrap_or_default() + > 0 + }) + .filter_map(|bucket| { + let term = bucket["key"].as_str()?.to_string(); + let score = bucket["max_score"]["value"] + .as_f64() + .unwrap_or_default() as f32 as f64; + let batch_score = bucket["max_batch_score"]["value"] + .as_f64() + .unwrap_or(score) as f32 as f64; + Some(CandidateTerm { + term, + score, + batch_score, + }) + }) + .collect_vec(); + if candidates.is_empty() { + return Vec::new(); + } + + let query_chars = query.chars().collect_vec(); + let mut groups = BTreeMap::>::new(); + for (index, candidate) in candidates.iter().enumerate() { + let candidate_chars = candidate.term.chars().collect_vec(); + let minimum_length = candidate_chars.len().min(query_chars.len()); + let maximum_length = + (query_chars.len() + cost).min(candidate_chars.len()); + let prefix = (minimum_length..=maximum_length) + .find(|length| { + damerau_levenshtein_distance( + &query_chars, + &candidate_chars[..*length], + ) == cost + }) + .map(|length| { + candidate_chars[..length].iter().collect::() + }) + .unwrap_or_else(|| candidate.term.clone()); + groups.entry(prefix).or_default().push(index); + } + + let discovery_limit = max_candidates.saturating_mul(4); + let mut discovered = Vec::with_capacity(discovery_limit); + for (prefix, candidate_indices) in groups.iter().rev() { + if discovered.len() >= discovery_limit { + break; + } + let mut nodes = vec![CandidateTrieNode::default()]; + for candidate_index in candidate_indices { + let candidate = &candidates[*candidate_index]; + let mut node_index = 0; + nodes[node_index].batch_score = nodes[node_index] + .batch_score + .max(candidate.batch_score); + for byte in candidate.term.as_bytes()[prefix.len()..] + .iter() + .copied() + { + let child_index = + if let Some(index) = nodes[node_index] + .children + .get(&byte) + .copied() + { + index + } else { + let index = nodes.len(); + let mut path = nodes[node_index].path.clone(); + path.push(byte); + nodes.push(CandidateTrieNode { + path, + ..CandidateTrieNode::default() + }); + nodes[node_index].children.insert(byte, index); + index + }; + node_index = child_index; + nodes[node_index].batch_score = nodes[node_index] + .batch_score + .max(candidate.batch_score); + } + let mut path = nodes[node_index].path.clone(); + path.push(0); + let terminal_index = nodes.len(); + nodes.push(CandidateTrieNode { + term: Some(*candidate_index), + batch_score: candidate.score, + path, + ..CandidateTrieNode::default() + }); + nodes[node_index].children.insert(0, terminal_index); + } + + let compressed_node = |mut index: usize| { + while nodes[index].term.is_none() + && nodes[index].children.len() == 1 + { + index = *nodes[index] + .children + .values() + .next() + .expect("candidate trie node has one child"); + } + index + }; + let mut frontier = vec![compressed_node(0)]; + while !frontier.is_empty() && discovered.len() < discovery_limit { + let best_position = frontier + .iter() + .enumerate() + .max_by(|(_, left), (_, right)| { + nodes[**left] + .batch_score + .total_cmp(&nodes[**right].batch_score) + .then_with(|| { + nodes[**right].path.cmp(&nodes[**left].path) + }) + }) + .map(|(position, _)| position) + .unwrap_or_default(); + let node_index = frontier.swap_remove(best_position); + if let Some(candidate_index) = nodes[node_index].term { + discovered.push(candidate_index); + } + frontier.extend( + nodes[node_index] + .children + .values() + .copied() + .map(compressed_node), + ); + } + } + + discovered.sort_by(|left, right| { + candidates[*right] + .score + .total_cmp(&candidates[*left].score) + .then_with(|| { + candidates[*left].term.cmp(&candidates[*right].term) + }) + }); + let mut ranked = discovered + .into_iter() + .take(max_candidates.saturating_add(8)) + .map(|index| candidates[index].term.clone()) + .collect_vec(); + if cost == 0 && candidates.iter().any(|candidate| candidate.term == query) + { + ranked.retain(|term| term != query); + ranked.insert(0, query.to_string()); + ranked.truncate(max_candidates.saturating_add(8)); + } + ranked +} + +pub(super) fn tokenize_search_name(name: &str) -> Vec { + tokenize_candidate_text(name, true) +} + +pub(super) fn tokenize_candidate_text( + text: &str, + separate_hyphens: bool, +) -> Vec { + let mut tokens = Vec::new(); + let mut token = String::new(); + let characters = text.chars().collect_vec(); + for (index, character) in characters.iter().copied().enumerate() { + if character == ' ' + || character == '\n' + || (separate_hyphens && character == '-') + { + if !token.is_empty() { + tokens.push(std::mem::take(&mut token)); + } + } else if character.is_ascii_alphanumeric() { + token.extend(character.to_lowercase()); + } else if !character.is_ascii() { + let joins_word = matches!( + character, + '\u{2010}'..='\u{2015}' | '\u{2212}' + ) && index > 0 + && characters[index - 1].is_alphanumeric() + && characters + .get(index + 1) + .is_some_and(|next| next.is_alphanumeric()); + if joins_word { + continue; + } + if character == '\u{202f}' { + token.push(character); + continue; + } + let mut encoded = [0; 4]; + for normalized in character + .encode_utf8(&mut encoded) + .nfkd() + .filter(|normalized| !is_combining_mark(*normalized)) + { + if drops_transliterated_punctuation(normalized) { + continue; + } + if normalized.is_ascii_alphanumeric() + || !normalized.is_ascii() + { + token.extend(normalized.to_lowercase()); + } + } + } + } + if !token.is_empty() { + tokens.push(token); + } + tokens.into_iter().unique().collect() +} + +fn drops_transliterated_punctuation(character: char) -> bool { + matches!( + character, + '\u{00ab}' + | '\u{00b7}' + | '\u{00bb}' + | '\u{2018}'..='\u{201f}' + | '\u{2026}' + | '\u{2032}'..='\u{2037}' + ) +} + +pub(super) fn damerau_levenshtein_distance( + left: &[char], + right: &[char], +) -> usize { + let mut distances = vec![vec![0; right.len() + 1]; left.len() + 1]; + for (index, row) in distances.iter_mut().enumerate() { + row[0] = index; + } + for index in 0..=right.len() { + distances[0][index] = index; + } + + for left_index in 1..=left.len() { + for right_index in 1..=right.len() { + let substitution_cost = + usize::from(left[left_index - 1] != right[right_index - 1]); + let mut distance = distances[left_index - 1][right_index] + .saturating_add(1) + .min( + distances[left_index][right_index - 1] + .saturating_add(1), + ) + .min( + distances[left_index - 1][right_index - 1] + .saturating_add(substitution_cost), + ); + if left_index > 1 + && right_index > 1 + && left[left_index - 1] == right[right_index - 2] + && left[left_index - 2] == right[right_index - 1] + { + distance = distance.min( + distances[left_index - 2][right_index - 2] + .saturating_add(1), + ); + } + distances[left_index][right_index] = distance; + } + } + distances[left.len()][right.len()] +} + +impl Elasticsearch { + pub(super) async fn search_for_project_raw_typesense_parity( + &self, + info: &SearchRequest, + ) -> Result { + let parsed = parse_search_request(info)?; + let search_sort = + parse_search_index(parsed.index, info.new_filters.as_deref())?; + let filter = Self::build_filter(info)?; + + let mut filters = + vec![json!({"term": {"document_type": "project"}})]; + if let Some(filter) = &filter { + filters.push(filter.query.clone()); + } + let alias = self.config.alias_name(); + let strict_query = json!({ + "bool": { + "must": [text_query(parsed.query, false, 2)], + "filter": &filters + } + }); + let fuzzy_query = parsed + .query + .split_whitespace() + .next() + .map(|_| { + json!({ + "bool": { + "must": [text_query(parsed.query, true, 2)], + "filter": &filters + } + }) + }); + let broader_fuzzy_query = parsed + .query + .split_whitespace() + .next() + .map(|_| { + json!({ + "bool": { + "must": [text_query(parsed.query, true, 3)], + "filter": &filters + } + }) + }); + let mut query = strict_query; + let sort = Self::sort(search_sort.index); + let bucketed_relevance = uses_text_match_bucketing( + search_sort.index, + ) && parsed.offset < BUCKETED_HITS + && bucket_size( + &info.typesense_config.bucketing, + BUCKETED_HITS, + ) + .is_some(); + let requested_end = + parsed.offset.saturating_add(parsed.hits_per_page); + let remaining_offset = + if bucketed_relevance { 0 } else { parsed.offset }; + let fetch_size = if bucketed_relevance { + BUCKETED_HITS.max(requested_end) + } else { + parsed.hits_per_page + }; + let mut used_fuzzy_query = false; + let mut used_typesense_candidates = false; + let mut typesense_candidate_count_hint = None; + let mut typesense_candidate_token_query = None; + let mut typesense_candidate_documents_query = None; + let mut typesense_candidate_cost = None; + let candidate_query_length = parsed + .query + .split_whitespace() + .next() + .filter(|_| parsed.query.split_whitespace().count() == 1) + .map(|token| token.chars().count()) + .unwrap_or_default(); + let should_select_typesense_candidates = + (3..=MAX_CANDIDATE_QUERY_LENGTH) + .contains(&candidate_query_length); + if should_select_typesense_candidates + && let Some(selection) = self + .select_typesense_candidates( + &alias, + &filters, + parsed.query, + info.typesense_config.max_candidates, + ) + .await? + { + let normalized_candidate_query = parsed.query.to_lowercase(); + let filter_trailing_prefix = selection.cost > 0 + && filters.len() == 1 + && candidate_query_length < 5; + let candidate_terms = selection + .terms + .iter() + .filter(|term| { + !filter_trailing_prefix + || term.chars().count() >= candidate_query_length + || !normalized_candidate_query.starts_with(*term) + }) + .take( + info.typesense_config + .max_candidates + .clamp(1, 128) + .saturating_add(usize::from( + selection.boundary_overshoot, + )), + ) + .cloned() + .collect_vec(); + debug!( + query = parsed.query, + cost = selection.cost, + ?candidate_terms, + "selected Elasticsearch typo candidates" + ); + let selected_query = json!({ + "bool": { + "must": [ + selected_candidate_text_query( + &candidate_terms, + ) + ], + "filter": &filters + } + }); + let candidate_token_query = + candidate_token_text_query( + &candidate_terms, + selection.cost, + candidate_query_length, + ); + let candidate_documents_query = json!({ + "bool": { + "must": [&candidate_token_query], + "filter": &filters + } + }); + let should_count_candidates = selection.cost == 0; + let selected_candidate_count = + if should_count_candidates { + Some( + self.count_matches( + &alias, + &candidate_documents_query, + ) + .await?, + ) + } else { + None + }; + typesense_candidate_count_hint = selected_candidate_count; + typesense_candidate_token_query = Some(candidate_token_query); + typesense_candidate_documents_query = + Some(candidate_documents_query.clone()); + typesense_candidate_cost = Some(selection.cost); + if selection.cost == 0 { + query = json!({ + "bool": { + "must": [{ + "dis_max": { + "queries": [ + text_query( + parsed.query, + false, + 2, + ), + stemmed_single_name_query( + parsed.query, + ) + ], + "tie_breaker": 0 + } + }], + "filter": &filters + } + }); + } + let use_candidate_ranking = selection.cost > 0 + || (selection.cost == 0 + && candidate_query_length <= 4 + && selected_candidate_count + .is_some_and(|count| { + count <= BUCKETED_HITS + })); + if use_candidate_ranking { + query = if selection.cost > 0 { + candidate_documents_query.clone() + } else { + selected_query + }; + used_fuzzy_query = selection.cost > 0; + used_typesense_candidates = true; + } + } + let deep_pagination = remaining_offset + .saturating_add(fetch_size) + > MAX_RESULT_WINDOW; + let mut adjusted_candidate_query = false; + if deep_pagination + && !used_typesense_candidates + && let ( + Some(candidate_token_query), + Some(candidate_documents_query), + Some(candidate_cost), + Some(candidate_count), + ) = ( + typesense_candidate_token_query.as_ref(), + typesense_candidate_documents_query.as_ref(), + typesense_candidate_cost, + typesense_candidate_count_hint, + ) + { + let current_count = self.count_matches(&alias, &query).await?; + if let Some(adjusted) = self + .adjust_candidate_query( + &alias, + &query, + &sort, + &filters, + candidate_token_query, + candidate_documents_query, + candidate_cost, + candidate_count, + current_count, + candidate_query_length, + ) + .await? + { + query = adjusted; + adjusted_candidate_query = true; + } + } + if deep_pagination + && let Some(fuzzy_query) = &fuzzy_query + && !self.has_matches(&alias, &query).await? + { + query = fuzzy_query.clone(); + query = self + .add_fuzzy_name_promotions( + &alias, + query, + &filters, + parsed.query, + ) + .await?; + used_fuzzy_query = true; + } + + let mut body = if deep_pagination { + self + .execute_deep_search( + &alias, + &query, + &sort, + remaining_offset, + fetch_size, + ) + .await? + } else { + let request_body = Self::search_body( + &query, + &sort, + remaining_offset, + fetch_size, + true, + None, + !bucketed_relevance, + ); + self + .execute_search( + &alias, + &request_body, + bucketed_relevance, + ) + .await? + }; + if !used_typesense_candidates + && !deep_pagination + && body["hits"]["total"]["value"].as_u64() == Some(0) + && let Some(fuzzy_query) = fuzzy_query + { + query = fuzzy_query; + query = self + .add_fuzzy_name_promotions( + &alias, + query, + &filters, + parsed.query, + ) + .await?; + used_fuzzy_query = true; + let request_body = Self::search_body( + &query, + &sort, + remaining_offset, + fetch_size, + true, + None, + !bucketed_relevance, + ); + body = self + .execute_search( + &alias, + &request_body, + bucketed_relevance, + ) + .await?; + } + + if !used_typesense_candidates + && !adjusted_candidate_query + && !deep_pagination + && let ( + Some(candidate_token_query), + Some(candidate_documents_query), + Some(candidate_cost), + Some(candidate_count), + ) = ( + typesense_candidate_token_query.as_ref(), + typesense_candidate_documents_query.as_ref(), + typesense_candidate_cost, + typesense_candidate_count_hint, + ) + { + let current_count = body["hits"]["total"]["value"] + .as_u64() + .unwrap_or_default() as usize; + if let Some(adjusted) = self + .adjust_candidate_query( + &alias, + &query, + &sort, + &filters, + candidate_token_query, + candidate_documents_query, + candidate_cost, + candidate_count, + current_count, + candidate_query_length, + ) + .await? + { + query = adjusted; + let request_body = Self::search_body( + &query, + &sort, + remaining_offset, + fetch_size, + true, + None, + !bucketed_relevance, + ); + body = self + .execute_search( + &alias, + &request_body, + bucketed_relevance, + ) + .await?; + } + } + + let total_hits = body["hits"]["total"]["value"] + .as_u64() + .unwrap_or_default() as usize; + let bucketed_candidate_count = if let Some(candidate_count) = + typesense_candidate_count_hint + { + candidate_count + } else if bucketed_relevance + && used_fuzzy_query + && !used_typesense_candidates + && total_hits < BUCKETED_HITS + { + if let Some(broader_fuzzy_query) = &broader_fuzzy_query { + let broader_count = + self.count_matches(&alias, broader_fuzzy_query).await?; + broader_count.min( + total_hits.saturating_add(total_hits.div_ceil(5)), + ) + } else { + total_hits + } + } else { + total_hits + }; + if bucketed_relevance { + let project_ids = { + let hits = body["hits"]["hits"] + .as_array_mut() + .ok_or_else(|| { + ApiError::Internal(eyre!( + "elasticsearch search hits were not an array" + )) + })?; + rerank_bucketed_hits( + hits, + &info.typesense_config.bucketing, + bucketed_candidate_count, + used_fuzzy_query && !used_typesense_candidates, + ); + hits.iter() + .skip(parsed.offset) + .take(parsed.hits_per_page) + .filter_map(|hit| { + hit["sort"] + .as_array() + .and_then(|sort| sort.last()) + .and_then(Value::as_str) + .map(ToOwned::to_owned) + }) + .collect_vec() + }; + + if project_ids.is_empty() { + body["hits"]["hits"] = Value::Array(Vec::new()); + } else { + let mut selected_filters = filters.clone(); + selected_filters.push( + json!({"terms": {"project_id": &project_ids}}), + ); + let selected_query = + json!({"bool": {"filter": selected_filters}}); + let selected_body = Self::search_body( + &selected_query, + &sort, + 0, + project_ids.len(), + false, + None, + true, + ); + let mut selected_body = self + .execute_search(&alias, &selected_body, false) + .await?; + let hits = selected_body["hits"]["hits"] + .as_array_mut() + .ok_or_else(|| { + ApiError::Internal(eyre!( + "elasticsearch search hits were not an array" + )) + })?; + let mut hits_by_project_id = std::mem::take(hits) + .into_iter() + .filter_map(|hit| { + let project_id = hit["sort"] + .as_array() + .and_then(|sort| sort.last()) + .and_then(Value::as_str)? + .to_string(); + Some((project_id, hit)) + }) + .collect::>(); + *hits = project_ids + .into_iter() + .filter_map(|project_id| { + hits_by_project_id.remove(&project_id) + }) + .collect(); + body = selected_body; + } + } + let hits = body["hits"]["hits"] + .as_array() + .into_iter() + .flatten() + .filter_map(|hit| { + let mut document = hit["_source"].clone(); + let object = document.as_object_mut()?; + object.remove("document_type"); + object.remove("_search_tokens"); + if filter + .as_ref() + .is_some_and(|filter| filter.has_version_filter) + { + if let Some(version_id) = matching_version_id(hit) { + object.insert( + "version_id".to_string(), + Value::String(version_id), + ); + } + } + + let metadata = info.show_metadata.then(|| { + json!({ + "score": hit["_score"], + "sort": hit["sort"] + }) + }); + let mut result: ResultSearchProject = + serde_json::from_value::(document) + .ok()? + .into(); + result.search_metadata = metadata; + Some(result) + }) + .collect(); + + Ok(SearchResults { + hits, + page: parsed.page, + hits_per_page: parsed.hits_per_page, + total_hits, + }) + } + +} + +impl Elasticsearch { + async fn adjust_candidate_query( + &self, + alias: &str, + query: &Value, + sort: &[Value], + filters: &[Value], + candidate_token_query: &Value, + candidate_documents_query: &Value, + candidate_cost: usize, + candidate_count: usize, + current_count: usize, + candidate_query_length: usize, + ) -> Result, ApiError> { + let candidate_counts_differ = current_count != candidate_count; + let constrain_to_candidates = candidate_counts_differ + && candidate_count <= BUCKETED_HITS; + let restrict_to_candidate_terms = (candidate_cost > 0 + && !constrain_to_candidates + && (!candidate_counts_differ || filters.len() > 1)) + || (candidate_cost == 0 && candidate_count < current_count); + let promote_candidates = candidate_cost > 0 + && (candidate_query_length >= 5 + || candidate_count <= BUCKETED_HITS); + let promote_filtered_mismatch = + candidate_counts_differ && filters.len() > 1; + if !constrain_to_candidates + && !promote_candidates + && !promote_filtered_mismatch + && !restrict_to_candidate_terms + { + return Ok(None); + } + + let candidate_body = Self::search_body( + candidate_documents_query, + sort, + 0, + candidate_count.min(MAX_RESULT_WINDOW), + false, + None, + false, + ); + let candidate_body = + self.execute_search(alias, &candidate_body, true).await?; + let candidate_project_ids = candidate_body["hits"]["hits"] + .as_array() + .into_iter() + .flatten() + .filter_map(|hit| { + hit["sort"] + .as_array() + .and_then(|sort| sort.last()) + .and_then(Value::as_str) + .map(ToOwned::to_owned) + }) + .collect_vec(); + if candidate_project_ids.is_empty() { + return Ok(None); + } + let promoted_candidate_project_ids = candidate_project_ids + .iter() + .take(BUCKETED_HITS) + .collect_vec(); + + let candidate_scoring_query = json!({ + "bool": { + "must": [candidate_token_query], + "filter": [{ + "terms": { + "project_id": promoted_candidate_project_ids + } + }] + } + }); + let promoted_query = json!({ + "dis_max": { + "queries": [ + query, + candidate_scoring_query, + { + "constant_score": { + "filter": { + "bool": { + "filter": [{ + "term": { + "document_type": "project" + } + }, { + "terms": { + "project_id": + &candidate_project_ids + } + }] + } + }, + "boost": 0 + } + } + ], + "tie_breaker": 0 + } + }); + let adjusted = if constrain_to_candidates { + json!({ + "bool": { + "must": [promoted_query], + "filter": [{ + "terms": { + "project_id": &candidate_project_ids + } + }] + } + }) + } else if restrict_to_candidate_terms { + json!({ + "bool": { + "must": [promoted_query], + "filter": [candidate_token_query] + } + }) + } else { + promoted_query + }; + Ok(Some(adjusted)) + } + + async fn select_typesense_candidates( + &self, + alias: &str, + filters: &[Value], + raw_query: &str, + max_candidates: usize, + ) -> Result, ApiError> { + let mut query_tokens = raw_query.split_whitespace(); + let Some(query) = query_tokens.next() else { + return Ok(None); + }; + if query_tokens.next().is_some() { + return Ok(None); + } + + let query = query.to_lowercase(); + let query_length = query.chars().count(); + if !(3..=MAX_CANDIDATE_QUERY_LENGTH) + .contains(&query_length) + { + return Ok(None); + } + + let maximum_cost: usize = if query_length >= 7 { + 2 + } else if query_length >= 4 { + 1 + } else { + 0 + }; + let max_candidates = max_candidates.clamp(1, 128); + let aggregation_size = max_candidates + .saturating_mul(CANDIDATE_AGGREGATION_MULTIPLIER); + let shard_size = aggregation_size + .saturating_mul(CANDIDATE_SHARD_MULTIPLIER); + let native_exact_prefix = query + .chars() + .all(|character| character.is_ascii_alphanumeric()); + + for cost in 0..=maximum_cost { + if query_length > 10 && cost > 0 && filters.len() == 1 { + break; + } + let mut runtime_mappings = Map::new(); + let mut aggregations = Map::new(); + for ( + index, + ((prefix_field, candidate_field), _), + ) in SEARCH_CANDIDATE_FIELDS + .into_iter() + .zip(SEARCH_TEXT_FIELDS) + .enumerate() + { + let name = format!("candidate_{index}"); + let candidate_prefilter = if query_length > 10 && cost == 0 { + json!({ + "wildcard": { + (candidate_field): { + "value": format!("{query}*") + } + } + }) + } else if query_length > 10 { + let prefix_query = + query.chars().take(10).collect::(); + json!({ + "match": { + (prefix_field): { + "query": prefix_query, + "fuzziness": cost.saturating_add(1).min(2), + "prefix_length": 0, + "max_expansions": 1024, + "fuzzy_transpositions": true + } + } + }) + } else { + json!({ + "match": { + (prefix_field): { + "query": &query, + "fuzziness": cost, + "prefix_length": 0, + "max_expansions": 1024, + "fuzzy_transpositions": true + } + } + }) + }; + let aggregation_field = if cost == 0 + && native_exact_prefix + { + candidate_field.to_string() + } else { + runtime_mappings.insert( + name.clone(), + json!({ + "type": "keyword", + "script": { + "source": CANDIDATE_SCRIPT, + "params": { + "query": &query, + "cost": cost.to_string(), + "field": candidate_field + } + } + }), + ); + name.clone() + }; + let mut terms_aggregation = json!({ + "field": aggregation_field, + "size": aggregation_size, + "shard_size": shard_size, + "order": [ + {"max_batch_score": "desc"}, + {"_key": "asc"} + ] + }); + if cost == 0 && native_exact_prefix { + terms_aggregation["include"] = + Value::String(format!("{query}.*")); + } + aggregations.insert( + name.clone(), + json!({ + "filter": candidate_prefilter, + "aggs": { + "values": { + "terms": terms_aggregation, + "aggs": { + "max_score": { + "max": {"field": "log_downloads"} + }, + "max_batch_score": { + "max": { + "field": "_search_tokens.batch_score" + } + }, + "allowed": { + "filter": { + "bool": {"filter": filters} + } + } + } + } + } + }), + ); + } + + let mut body = json!({ + "_source": false, + "size": 0, + "track_total_hits": false, + "query": { + "term": {"document_type": "project"} + }, + "aggs": aggregations + }); + if !runtime_mappings.is_empty() { + body["runtime_mappings"] = Value::Object(runtime_mappings); + } + let response = self.execute_search(alias, &body, false).await?; + let selection_limit = max_candidates.saturating_add(8); + let mut terms = Vec::with_capacity(selection_limit); + let mut seen_terms = HashSet::with_capacity(selection_limit); + let mut boundary_overshoot = false; + for index in 0..SEARCH_CANDIDATE_FIELDS.len() { + let terms_before_field = terms.len(); + let name = format!("candidate_{index}"); + let buckets = response["aggregations"][&name]["values"] + ["buckets"] + .as_array() + .map(Vec::as_slice) + .unwrap_or_default(); + let ranked_terms = rank_candidate_buckets( + buckets, + &query, + cost, + max_candidates, + ); + for term in ranked_terms { + if seen_terms.insert(term.clone()) { + terms.push(term); + } + if terms.len() >= selection_limit { + break; + } + } + if terms.len() >= max_candidates { + boundary_overshoot = + index == SEARCH_CANDIDATE_FIELDS.len() - 1 + && terms_before_field > 0 + && terms_before_field < max_candidates + && terms.len() > max_candidates; + break; + } + } + + if !terms.is_empty() { + if cost == 0 + && let Some(index) = + terms.iter().position(|term| term == &query) + { + let exact = terms.remove(index); + terms.insert(0, exact); + } + return Ok(Some(CandidateSelection { + terms, + cost, + boundary_overshoot, + })); + } + } + + Ok(None) + } + + async fn add_fuzzy_name_promotions( + &self, + alias: &str, + query: Value, + filters: &[Value], + raw_query: &str, + ) -> Result { + let mut tokens = raw_query.split_whitespace(); + let Some(token) = tokens.next() else { + return Ok(query); + }; + if tokens.next().is_some() { + return Ok(query); + } + + let normalized_query = token.to_lowercase(); + let query_chars = normalized_query.chars().collect_vec(); + if !(4..=10).contains(&query_chars.len()) { + return Ok(query); + } + + let candidate_query = json!({ + "bool": { + "must": [{ + "match": { + "name.prefix": { + "query": token, + "fuzziness": 1, + "prefix_length": 0, + "max_expansions": 1024, + "fuzzy_transpositions": true + } + } + }], + "filter": filters + } + }); + let candidate_body = json!({ + "_source": ["project_id", "name"], + "size": 100, + "track_total_hits": false, + "query": candidate_query, + "sort": [ + {"log_downloads": {"order": "desc", "missing": "_last"}}, + { + "version_published_timestamp": { + "order": "desc", + "missing": "_last" + } + }, + {"project_id": {"order": "asc"}} + ] + }); + let candidates = + self.execute_search(alias, &candidate_body, false).await?; + let candidate_hits = candidates["hits"]["hits"] + .as_array() + .into_iter() + .flatten() + .collect_vec(); + let has_first_character_correction = + candidate_hits.iter().any(|hit| { + let Some(name) = hit["_source"]["name"].as_str() else { + return false; + }; + tokenize_search_name(name).into_iter().any(|token| { + let token_chars = token.chars().collect_vec(); + token_chars.len() <= query_chars.len() + 1 + && token_chars.first() != query_chars.first() + && damerau_levenshtein_distance( + &query_chars, + &token_chars, + ) == 1 + }) + }); + let promoted_project_ids = candidate_hits + .into_iter() + .filter_map(|hit| { + let name = hit["_source"]["name"].as_str()?; + let should_promote = + tokenize_search_name(name).into_iter().any(|token| { + let token_chars = token.chars().collect_vec(); + let distance = damerau_levenshtein_distance( + &query_chars, + &token_chars, + ); + token_chars.len() <= query_chars.len() + 1 + && ((query_chars.len() <= 7 + && token_chars.len() + == query_chars.len() + 1 + && distance == 2) + || (has_first_character_correction + && distance == 1) + || (token_chars.len() + == query_chars.len() + 1 + && distance == 1)) + }); + should_promote + .then(|| hit["_source"]["project_id"].as_str()) + .flatten() + .map(ToOwned::to_owned) + }) + .collect_vec(); + if promoted_project_ids.is_empty() { + return Ok(query); + } + + Ok(json!({ + "dis_max": { + "queries": [ + query, + { + "constant_score": { + "filter": { + "bool": { + "filter": [ + { + "term": { + "document_type": "project" + } + }, + { + "terms": { + "project_id": + promoted_project_ids + } + } + ] + } + }, + "boost": 15.5 + } + } + ], + "tie_breaker": 0 + } + })) + } + +} + +impl Elasticsearch { + pub(super) fn index_schema() -> Value { + json!({ + "settings": { + "number_of_shards": 3, + "number_of_replicas": 1, + "refresh_interval": "30s", + "index.mapping.total_fields.limit": 5000, + "analysis": { + "char_filter": { + "hyphen_separator": { + "type": "pattern_replace", + "pattern": "-", + "replacement": " " + }, + "strip_symbols": { + "type": "pattern_replace", + "pattern": r"[^\p{L}\p{N}\s]", + "replacement": "" + } + }, + "filter": { + "typesense_stemmer": { + "type": "stemmer", + "language": "english" + }, + "typesense_prefix_ngrams": { + "type": "edge_ngram", + "min_gram": 1, + "max_gram": 10 + } + }, + "analyzer": { + "typesense_text": { + "type": "custom", + "char_filter": ["strip_symbols"], + "tokenizer": "whitespace", + "filter": ["lowercase"] + }, + "typesense_hyphen_text": { + "type": "custom", + "char_filter": [ + "hyphen_separator", + "strip_symbols" + ], + "tokenizer": "whitespace", + "filter": ["lowercase"] + }, + "typesense_stemmed_text": { + "type": "custom", + "char_filter": ["strip_symbols"], + "tokenizer": "whitespace", + "filter": ["lowercase", "typesense_stemmer"] + }, + "typesense_prefix_text": { + "type": "custom", + "char_filter": [ + "hyphen_separator", + "strip_symbols" + ], + "tokenizer": "whitespace", + "filter": [ + "lowercase", + "typesense_prefix_ngrams" + ] + }, + "typesense_plain_prefix_text": { + "type": "custom", + "char_filter": ["strip_symbols"], + "tokenizer": "whitespace", + "filter": [ + "lowercase", + "typesense_prefix_ngrams" + ] + }, + "typesense_stemmed_prefix_text": { + "type": "custom", + "char_filter": ["strip_symbols"], + "tokenizer": "whitespace", + "filter": [ + "lowercase", + "typesense_stemmer", + "typesense_prefix_ngrams" + ] + } + } + } + }, + "mappings": { + "dynamic_templates": [ + { + "strings_as_keywords": { + "match_mapping_type": "string", + "mapping": { + "type": "keyword", + "ignore_above": 8191 + } + } + } + ], + "properties": { + "document_type": { + "type": "join", + "relations": {"project": "version"}, + "eager_global_ordinals": true + }, + "version_id": {"type": "keyword"}, + "project_id": {"type": "keyword"}, + "project_types": {"type": "keyword"}, + "all_project_types": {"type": "keyword"}, + "slug": { + "type": "text", + "analyzer": "typesense_text", + "index_options": "docs", + "norms": false, + "index_prefixes": { + "min_chars": 1, + "max_chars": 10 + }, + "fields": { + "keyword": {"type": "keyword", "ignore_above": 8191}, + "prefix": { + "type": "text", + "analyzer": "typesense_plain_prefix_text", + "search_analyzer": "typesense_text", + "index_options": "docs", + "norms": false + } + } + }, + "author": { + "type": "text", + "analyzer": "typesense_hyphen_text", + "index_options": "docs", + "norms": false, + "index_prefixes": { + "min_chars": 1, + "max_chars": 10 + }, + "fields": { + "keyword": {"type": "keyword", "ignore_above": 8191}, + "prefix": { + "type": "text", + "analyzer": "typesense_prefix_text", + "search_analyzer": "typesense_hyphen_text", + "index_options": "docs", + "norms": false + } + } + }, + "indexed_author": { + "type": "text", + "analyzer": "typesense_text", + "index_options": "docs", + "norms": false, + "index_prefixes": { + "min_chars": 1, + "max_chars": 10 + }, + "fields": { + "prefix": { + "type": "text", + "analyzer": "typesense_plain_prefix_text", + "search_analyzer": "typesense_text", + "index_options": "docs", + "norms": false + } + } + }, + "name": { + "type": "text", + "analyzer": "typesense_hyphen_text", + "index_options": "positions", + "norms": false, + "index_prefixes": { + "min_chars": 1, + "max_chars": 10 + }, + "fields": { + "keyword": {"type": "keyword", "ignore_above": 8191}, + "prefix": { + "type": "text", + "analyzer": "typesense_prefix_text", + "search_analyzer": "typesense_hyphen_text", + "index_options": "docs", + "norms": false + } + } + }, + "indexed_name": { + "type": "text", + "analyzer": "typesense_stemmed_text", + "index_options": "docs", + "norms": false, + "index_prefixes": { + "min_chars": 1, + "max_chars": 10 + }, + "fielddata": true, + "fields": { + "prefix": { + "type": "text", + "analyzer": "typesense_stemmed_prefix_text", + "search_analyzer": "typesense_stemmed_text", + "index_options": "docs", + "norms": false + } + } + }, + "summary": { + "type": "text", + "analyzer": "typesense_text", + "index_options": "docs", + "norms": false, + "index_prefixes": { + "min_chars": 1, + "max_chars": 10 + }, + "fields": { + "keyword": {"type": "keyword", "ignore_above": 8191}, + "prefix": { + "type": "text", + "analyzer": "typesense_plain_prefix_text", + "search_analyzer": "typesense_text", + "index_options": "docs", + "norms": false + } + } + }, + "_search_tokens": { + "properties": { + "batch_score": {"type": "double"}, + "name": {"type": "keyword"}, + "indexed_name": {"type": "keyword"}, + "slug": {"type": "keyword"}, + "author": {"type": "keyword"}, + "indexed_author": {"type": "keyword"}, + "summary": {"type": "keyword"} + } + }, + "categories": {"type": "keyword"}, + "project_categories": {"type": "keyword"}, + "display_categories": {"type": "keyword"}, + "license": {"type": "keyword"}, + "open_source": {"type": "boolean"}, + "environment": {"type": "keyword"}, + "game_versions": {"type": "keyword"}, + "client_side": {"type": "keyword"}, + "server_side": {"type": "keyword"}, + "dependency_project_ids": {"type": "keyword"}, + "compatible_dependency_project_ids": {"type": "keyword"}, + "downloads": {"type": "integer"}, + "log_downloads": {"type": "double"}, + "follows": {"type": "integer"}, + "created_timestamp": {"type": "long"}, + "modified_timestamp": {"type": "long"}, + "version_published_timestamp": {"type": "long"}, + "date_created": {"type": "date"}, + "date_modified": {"type": "date"}, + "project_loader_fields": {"type": "object", "enabled": false}, + "minecraft_java_server": { + "properties": { + "verified_plays_2w": {"type": "long"}, + "is_online": {"type": "boolean"}, + "ping": { + "properties": { + "data": { + "properties": { + "players_online": {"type": "integer"} + } + } + } + } + } + } + } + } + }) + } + +} + +impl Elasticsearch { + pub(super) async fn import_projects( + &self, + indices: &[String], + documents: &[UploadSearchProject], + ) -> Result<()> { + let batch_size = self.config.bulk_batch_size.max(1); + let import_batch_size = + self.config.project_import_batch_size.max(1); + for import_batch in documents.chunks(import_batch_size) { + for candidate_batch in + import_batch + .chunks(INTERNAL_INDEX_BATCH_SIZE) + { + let batch_score = candidate_batch + .iter() + .map(|document| document.log_downloads) + .max_by(f64::total_cmp) + .unwrap_or_default(); + for documents in candidate_batch.chunks(batch_size) { + let body = projects_to_bulk(documents, batch_score)?; + for index in indices { + info!( + index, + document_count = documents.len(), + content_length_bytes = body.len(), + "sending Elasticsearch project bulk request" + ); + self.client.bulk(index, body.clone()).await?; + } + } + } + } + Ok(()) + } + +} + +fn projects_to_bulk( + documents: &[UploadSearchProject], + batch_score: f64, +) -> Result { + let mut output = String::new(); + for document in documents { + let id = format!("project:{}", document.project_id); + push_json_line( + &mut output, + &json!({ + "index": { + "_id": id, + "routing": document.project_id + } + }), + )?; + + let mut source = serde_json::to_value(document) + .wrap_err("failed to serialize `UploadSearchProject`")?; + let object = source + .as_object_mut() + .ok_or_else(|| eyre!("project search document is not an object"))?; + object.insert( + "document_type".to_string(), + Value::String("project".to_string()), + ); + object.insert( + "_search_tokens".to_string(), + json!({ + "name": tokenize_candidate_text(&document.name, true), + "indexed_name": &document.indexed_name, + "slug": document + .slug + .as_deref() + .map(|slug| tokenize_candidate_text(slug, false)) + .unwrap_or_default(), + "author": tokenize_candidate_text(&document.author, true), + "indexed_author": tokenize_candidate_text( + &document.indexed_author, + false, + ), + "summary": tokenize_candidate_text( + &document.summary, + false, + ), + "batch_score": batch_score, + }), + ); + add_server_online_field(object); + push_json_line(&mut output, &source)?; + } + Ok(output) +} diff --git a/apps/labrinth/src/search/backend/mod.rs b/apps/labrinth/src/search/backend/mod.rs index 544cf2a557..cb38ae6782 100644 --- a/apps/labrinth/src/search/backend/mod.rs +++ b/apps/labrinth/src/search/backend/mod.rs @@ -1,8 +1,10 @@ mod common; +pub mod elasticsearch; pub mod typesense; pub use common::{ ParsedSearchRequest, SearchIndex, SearchSort, combined_search_filters, parse_search_index, parse_search_request, }; +pub use elasticsearch::{Elasticsearch, ElasticsearchConfig}; pub use typesense::{Typesense, TypesenseConfig}; diff --git a/apps/labrinth/src/search/mod.rs b/apps/labrinth/src/search/mod.rs index f98f9f4cb1..690efe149a 100644 --- a/apps/labrinth/src/search/mod.rs +++ b/apps/labrinth/src/search/mod.rs @@ -189,6 +189,7 @@ pub enum TasksCancelFilter { #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] pub enum SearchBackendKind { Typesense, + Elasticsearch, } #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, strum::EnumIter)] @@ -224,6 +225,7 @@ impl FromStr for SearchBackendKind { fn from_str(s: &str) -> Result { Ok(match s { "typesense" => SearchBackendKind::Typesense, + "elasticsearch" => SearchBackendKind::Elasticsearch, _ => return Err(InvalidSearchBackendKind), }) } @@ -438,5 +440,9 @@ pub fn backend(meta_namespace: Option) -> Box { let config = backend::TypesenseConfig::new(meta_namespace); Box::new(backend::Typesense::new(config)) } + SearchBackendKind::Elasticsearch => { + let config = backend::ElasticsearchConfig::new(meta_namespace); + Box::new(backend::Elasticsearch::new(config)) + } } } diff --git a/docker-compose.yml b/docker-compose.yml index def52d2862..64112bdbc0 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -32,6 +32,102 @@ services: interval: 3s timeout: 5s retries: 3 + elasticsearch0: + image: docker.elastic.co/elasticsearch/elasticsearch:9.4.4 + container_name: labrinth-elasticsearch0 + restart: on-failure + networks: + - elasticsearch-mesh + ports: + - '127.0.0.1:9200:9200' + volumes: + - elasticsearch0-data:/usr/share/elasticsearch/data + environment: + node.name: elasticsearch0 + cluster.name: labrinth-elasticsearch + discovery.seed_hosts: elasticsearch0,elasticsearch1,elasticsearch2 + cluster.initial_master_nodes: elasticsearch0,elasticsearch1,elasticsearch2 + bootstrap.memory_lock: 'true' + xpack.security.enabled: 'false' + xpack.security.enrollment.enabled: 'false' + ES_JAVA_OPTS: -Xms512m -Xmx512m + ulimits: + memlock: + soft: -1 + hard: -1 + healthcheck: + test: + [ + 'CMD-SHELL', + 'curl --fail http://localhost:9200/_cluster/health?wait_for_status=yellow', + ] + interval: 5s + timeout: 5s + retries: 30 + elasticsearch1: + image: docker.elastic.co/elasticsearch/elasticsearch:9.4.4 + container_name: labrinth-elasticsearch1 + restart: on-failure + networks: + - elasticsearch-mesh + ports: + - '127.0.0.1:9201:9200' + volumes: + - elasticsearch1-data:/usr/share/elasticsearch/data + environment: + node.name: elasticsearch1 + cluster.name: labrinth-elasticsearch + discovery.seed_hosts: elasticsearch0,elasticsearch1,elasticsearch2 + cluster.initial_master_nodes: elasticsearch0,elasticsearch1,elasticsearch2 + bootstrap.memory_lock: 'true' + xpack.security.enabled: 'false' + xpack.security.enrollment.enabled: 'false' + ES_JAVA_OPTS: -Xms512m -Xmx512m + ulimits: + memlock: + soft: -1 + hard: -1 + healthcheck: + test: + [ + 'CMD-SHELL', + 'curl --fail http://localhost:9200/_cluster/health?wait_for_status=yellow', + ] + interval: 5s + timeout: 5s + retries: 30 + elasticsearch2: + image: docker.elastic.co/elasticsearch/elasticsearch:9.4.4 + container_name: labrinth-elasticsearch2 + restart: on-failure + networks: + - elasticsearch-mesh + ports: + - '127.0.0.1:9202:9200' + volumes: + - elasticsearch2-data:/usr/share/elasticsearch/data + environment: + node.name: elasticsearch2 + cluster.name: labrinth-elasticsearch + discovery.seed_hosts: elasticsearch0,elasticsearch1,elasticsearch2 + cluster.initial_master_nodes: elasticsearch0,elasticsearch1,elasticsearch2 + bootstrap.memory_lock: 'true' + xpack.security.enabled: 'false' + xpack.security.enrollment.enabled: 'false' + ES_JAVA_OPTS: -Xms512m -Xmx512m + ulimits: + memlock: + soft: -1 + hard: -1 + healthcheck: + test: + [ + 'CMD-SHELL', + 'curl --fail http://localhost:9200/_cluster/health?wait_for_status=yellow', + ] + interval: 5s + timeout: 5s + retries: 30 meilisearch0: image: getmeili/meilisearch:v1.12.0 container_name: labrinth-meilisearch0 @@ -398,7 +494,7 @@ services: depends_on: postgres_db: condition: service_healthy - meilisearch: + meilisearch0: condition: service_healthy elasticsearch0: condition: service_healthy @@ -481,6 +577,8 @@ services: volumes: - ./apps/labrinth/nginx/meili-lb.conf:/etc/nginx/conf.d/default.conf:ro networks: + elasticsearch-mesh: + driver: bridge meilisearch-mesh: driver: bridge redis-cluster-mesh: