Search backend refactor with typesense impl (#5528)

* initial elasticsearch impl

* working elastic cluster

* replace SearchError with ApiError for preparation of search backend

* start factoring meili out to trait

* move meili to backend

* update routes to use search backend trait

* wip

* Update projects.rs

* search backend is only init'd once in config

* wip

* wip: backend agnostic

* change search internal routes to delegate to backend

* initial elasticsearch impl

* fix filtering

* elastic impl

* refactor indexing into its own module

* clean up elastic code

* fix ci

* fix tests

* fix elastic health check

* fix up env rebase

* fix compile

* dummy commit to update github pr

* Fix rebase

* Elastic basic https auth

* Fix duplicate projects showing up

* Fix up tests

* Replace search `ApiErrors` with `eyre::Reports`, propagate background task errors

* clean up agents files

* make index chunk size configurable

* make `match_phrase` in elastic case-insensitive

* use current/next indices and swap between them

* test case for error body

* Fix failing case

* da merge

* factor out common stuff from search backends

* allow fetching hit metadata from search results

* allow customising elasticsearch search config

* bit of docs

* add mappings to indices for elastic

* Implement Typesense

* wip

* fix up some sort fields stuff

* use different approach to filterable field sets

* remove a bunch of search fields which weren't used for filtering

* bucket text matches

* Bucketing by text_match for typesense

* fix tombi lint

* fix some sentry errors and dont prioritise 2+ term matches

* tweak ts query settings

* expose some more search settings

* query sort changes

* small fixes

* should fix pagination stuff

* fix healthcheck maybe

* ragebait ci

* tests

* tests

* revert environment
This commit is contained in:
aecsocket
2026-03-12 18:58:55 +01:00
committed by GitHub
parent 1c1683adb6
commit f0224dfff7
36 changed files with 3848 additions and 762 deletions
+123
View File
@@ -0,0 +1,123 @@
use crate::routes::ApiError;
use crate::search::SearchRequest;
use crate::util::error::Context;
use eyre::eyre;
use std::borrow::Cow;
pub struct ParsedSearchRequest<'a> {
pub offset: usize,
pub hits_per_page: usize,
pub page: usize,
pub index: &'a str,
pub query: &'a str,
}
pub fn parse_search_request(
info: &SearchRequest,
) -> Result<ParsedSearchRequest<'_>, ApiError> {
let offset = info
.offset
.as_deref()
.unwrap_or("0")
.parse::<usize>()
.wrap_request_err("invalid offset")?;
let limit = info
.limit
.as_deref()
.unwrap_or("10")
.parse::<usize>()
.wrap_request_err("invalid limit")?
.min(100);
let hits_per_page = if limit == 0 { 1 } else { limit };
Ok(ParsedSearchRequest {
offset,
hits_per_page,
page: offset / hits_per_page + 1,
index: info.index.as_deref().unwrap_or("relevance"),
query: info.query.as_deref().unwrap_or_default(),
})
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum SearchIndex {
Relevance,
Downloads,
Follows,
Updated,
Newest,
MinecraftJavaServerVerifiedPlays2w,
MinecraftJavaServerPlayersOnline,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum SearchIndexName {
Projects,
ProjectsFiltered,
}
pub struct SearchSort {
pub index_name: SearchIndexName,
pub index: SearchIndex,
}
pub fn parse_search_index(
index: &str,
new_filters: Option<&str>,
) -> Result<SearchSort, ApiError> {
let projects_name = SearchIndexName::Projects;
let projects_filtered_name = SearchIndexName::ProjectsFiltered;
// TODO: this is a dumb hack, the frontend should pass the project type it's filtering directly
let is_server = new_filters
.is_some_and(|f| f.contains("project_types = minecraft_java_server"));
Ok(match index {
"relevance" => SearchSort {
index_name: projects_name,
index: if is_server {
SearchIndex::MinecraftJavaServerVerifiedPlays2w
} else {
SearchIndex::Relevance
},
},
"downloads" => SearchSort {
index_name: projects_filtered_name,
index: SearchIndex::Downloads,
},
"follows" => SearchSort {
index_name: projects_name,
index: SearchIndex::Follows,
},
"updated" | "date_modified" => SearchSort {
index_name: projects_name,
index: SearchIndex::Updated,
},
"newest" | "date_created" => SearchSort {
index_name: projects_name,
index: SearchIndex::Newest,
},
"minecraft_java_server.verified_plays_2w" => SearchSort {
index_name: projects_name,
index: SearchIndex::MinecraftJavaServerVerifiedPlays2w,
},
"minecraft_java_server.ping.data.players_online" => SearchSort {
index_name: projects_name,
index: SearchIndex::MinecraftJavaServerPlayersOnline,
},
i => return Err(ApiError::Request(eyre!("invalid index '{i}'"))),
})
}
pub fn combined_search_filters(info: &SearchRequest) -> Option<Cow<'_, str>> {
if let Some(filters) = info.new_filters.as_deref() {
return Some(filters.into());
}
match (info.filters.as_deref(), info.version.as_deref()) {
(Some(f), Some(v)) => Some(format!("({f}) AND ({v})").into()),
(Some(f), None) => Some(f.into()),
(None, Some(v)) => Some(v.into()),
(None, None) => None,
}
}
File diff suppressed because it is too large Load Diff
@@ -0,0 +1,689 @@
use std::sync::LazyLock;
use std::time::Duration;
use crate::database::PgPool;
use crate::database::redis::RedisPool;
use crate::env::ENV;
use crate::search::backend::meilisearch::MeilisearchConfig;
use crate::search::indexing::index_local;
use crate::search::{SearchField, UploadSearchProject};
use crate::util::error::Context;
use ariadne::ids::base62_impl::to_base62;
use eyre::{Result, eyre};
use futures::StreamExt;
use futures::stream::FuturesOrdered;
use meilisearch_sdk::client::{Client, SwapIndexes};
use meilisearch_sdk::indexes::Index;
use meilisearch_sdk::settings::{PaginationSetting, Settings};
use meilisearch_sdk::task_info::TaskInfo;
use tracing::{Instrument, error, info, info_span, instrument};
// // The chunk size for adding projects to the indexing database. If the request size
// // is too large (>10MiB) then the request fails with an error. This chunk size
// // assumes a max average size of 4KiB per project to avoid this cap.
//
// Set this to 50k for better observability
const MEILISEARCH_CHUNK_SIZE: usize = 50000; // 10_000_000
fn search_operation_timeout() -> std::time::Duration {
std::time::Duration::from_millis(ENV.SEARCH_OPERATION_TIMEOUT)
}
pub async fn remove_documents(
ids: &[crate::models::ids::VersionId],
config: &MeilisearchConfig,
) -> Result<()> {
let mut indexes = get_indexes_for_indexing(config, false, false)
.await
.wrap_err("failed to get current indexes")?;
let indexes_next = get_indexes_for_indexing(config, true, false)
.await
.wrap_err("failed to get next indexes")?;
for list in &mut indexes {
for alt_list in &indexes_next {
list.extend(alt_list.iter().cloned());
}
}
let client = config
.make_batch_client()
.wrap_err("failed to create batch client")?;
let client = &client;
let ids_base62 = ids.iter().map(|x| to_base62(x.0)).collect::<Vec<_>>();
let mut deletion_tasks = FuturesOrdered::new();
client.across_all(indexes, |index_list, client| {
for index in index_list {
let owned_client = client.clone();
let ids_base62_ref = &ids_base62;
deletion_tasks.push_back(async move {
index
.delete_documents(ids_base62_ref)
.await
.wrap_err_with(|| {
eyre!("failed to request to delete documents {ids_base62_ref:?}")
})?
.wait_for_completion(
&owned_client,
None,
Some(Duration::from_secs(15)),
)
.await
.wrap_err_with(|| {
eyre!("failed to delete documents {ids_base62_ref:?}")
})
});
}
});
while let Some(result) = deletion_tasks.next().await {
result?;
}
Ok(())
}
pub async fn index_projects(
ro_pool: PgPool,
redis: RedisPool,
config: &MeilisearchConfig,
) -> Result<()> {
info!("Indexing projects.");
info!("Ensuring current indexes exists");
// First, ensure current index exists (so no error happens- current index should be worst-case empty, not missing)
get_indexes_for_indexing(config, false, false)
.await
.wrap_err("failed to get indexes for indexing")?;
info!("Deleting surplus indexes");
// Then, delete the next index if it still exists
let indices = get_indexes_for_indexing(config, true, false)
.await
.wrap_err("failed to get next indexes to delete")?;
for client_indices in indices {
for index in client_indices {
index.delete().await.wrap_err("failed to delete an index")?;
}
}
info!("Recreating next index");
// Recreate the next index for indexing
let indices = get_indexes_for_indexing(config, true, true)
.await
.wrap_internal_err("failed to recreate next index")?;
let all_loader_fields =
crate::database::models::loader_fields::LoaderField::get_fields_all(
&ro_pool, &redis,
)
.await
.wrap_internal_err("failed to get all loader fields")?
.into_iter()
.map(|x| x.field)
.collect::<Vec<_>>();
info!("Gathering local projects");
let mut cursor = 0;
let mut idx = 0;
let mut total = 0;
loop {
info!("Gathering index data chunk {idx}");
idx += 1;
let (uploads, next_cursor) =
index_local(&ro_pool, &redis, cursor, 10000).await?;
total += uploads.len();
if uploads.is_empty() {
info!(
"No more projects to index, indexed {total} projects after {idx} chunks"
);
break;
}
cursor = next_cursor;
add_projects_batch_client(
&indices,
uploads,
all_loader_fields.clone(),
config,
)
.await?;
}
info!("Swapping indexes");
// Swap the index
swap_index(config, "projects").await?;
swap_index(config, "projects_filtered").await?;
info!("Deleting old indexes");
// Delete the now-old index
for index_list in indices {
for index in index_list {
index.delete().await?;
}
}
info!("Done adding projects.");
Ok(())
}
pub async fn swap_index(
config: &MeilisearchConfig,
index_name: &str,
) -> Result<()> {
let client = config.make_batch_client()?;
let index_name_next = config.get_index_name(index_name, true);
let index_name = config.get_index_name(index_name, false);
let swap_indices = SwapIndexes {
indexes: (index_name_next, index_name),
rename: None,
};
let swap_indices_ref = &swap_indices;
// is it "indexes" or "indices"? who knows! roll a die!
client
.with_all_clients("swap_indexes", |client| async move {
let task = client
.swap_indexes([swap_indices_ref])
.await
.wrap_err("failed to swap indices")?;
monitor_task(
client,
task,
Duration::from_secs(60 * 10), // 10 minutes
Some(Duration::from_secs(1)),
)
.await?;
Ok(())
})
.await?;
Ok(())
}
#[instrument(skip(config))]
pub async fn get_indexes_for_indexing(
config: &MeilisearchConfig,
next: bool, // Get the 'next' one
update_settings: bool,
) -> Result<Vec<Vec<Index>>> {
let client = config.make_batch_client()?;
let project_name = config.get_index_name("projects", next);
let project_filtered_name =
config.get_index_name("projects_filtered", next);
let project_name_ref = &project_name;
let project_filtered_name_ref = &project_filtered_name;
let results = client
.with_all_clients("get_indexes_for_indexing", |client| async move {
let projects_index = create_or_update_index(
client,
project_name_ref,
Some(&[
"words",
"typo",
"proximity",
"attribute",
"exactness",
"sort",
]),
update_settings,
)
.await?;
let projects_filtered_index = create_or_update_index(
client,
project_filtered_name_ref,
Some(&[
"sort",
"words",
"typo",
"proximity",
"attribute",
"exactness",
]),
update_settings,
)
.await?;
Ok(vec![projects_index, projects_filtered_index])
})
.await?;
Ok(results)
}
#[instrument(skip_all, fields(name))]
async fn create_or_update_index(
client: &Client,
name: &str,
custom_rules: Option<&'static [&'static str]>,
update_settings: bool,
) -> Result<Index, meilisearch_sdk::errors::Error> {
info!("Updating/creating index");
match client.get_index(name).await {
Ok(index) => {
info!("Updating index settings.");
let mut settings = default_settings();
if let Some(custom_rules) = custom_rules {
settings = settings.with_ranking_rules(custom_rules);
}
if update_settings {
info!("Updating index settings");
index
.set_settings(&settings)
.await
.inspect_err(|e| {
error!("Error setting index settings: {e:?}")
})?
.wait_for_completion(
client,
None,
Some(search_operation_timeout()),
)
.await
.inspect_err(|e| {
error!(
"Error setting index settings while waiting: {e:?}"
)
})?;
}
info!("Done performing index settings set.");
Ok(index)
}
_ => {
info!("Creating index.");
// Only create index and set settings if the index doesn't already exist
let task = client.create_index(name, Some("version_id")).await?;
let task = task
.wait_for_completion(
client,
None,
Some(search_operation_timeout()),
)
.await
.inspect_err(|e| {
error!("Error creating index while waiting: {e:?}")
})?;
let index = task
.try_make_index(client)
.map_err(|x| x.unwrap_failure())?;
let mut settings = default_settings();
if let Some(custom_rules) = custom_rules {
settings = settings.with_ranking_rules(custom_rules);
}
if update_settings {
index
.set_settings(&settings)
.await
.inspect_err(|e| {
error!("Error setting index settings: {e:?}")
})?
.wait_for_completion(
client,
None,
Some(search_operation_timeout()),
)
.await
.inspect_err(|e| {
error!(
"Error setting index settings while waiting: {e:?}"
)
})?;
}
Ok(index)
}
}
}
#[instrument(skip_all, fields(%index.uid, mods.len = mods.len()))]
async fn add_to_index(
client: &Client,
index: &Index,
mods: &[UploadSearchProject],
) -> Result<()> {
for chunk in mods.chunks(MEILISEARCH_CHUNK_SIZE) {
info!(
"Adding chunk of {} versions starting with version id {}",
chunk.len(),
chunk[0].version_id
);
let now = std::time::Instant::now();
let task = index
.add_or_replace(chunk, Some("version_id"))
.await
.inspect_err(|e| error!("Error adding chunk to index: {e:?}"))?;
monitor_task(
client,
task,
Duration::from_secs(60 * 5), // Timeout after 10 minutes
Some(Duration::from_secs(1)), // Poll once every second
)
.await?;
info!(
"Added chunk of {} projects to index in {:.2} seconds",
chunk.len(),
now.elapsed().as_secs_f64()
);
}
Ok(())
}
async fn monitor_task(
client: &Client,
task: TaskInfo,
timeout: Duration,
poll: Option<Duration>,
) -> Result<()> {
let now = std::time::Instant::now();
let id = task.get_task_uid();
let mut interval = tokio::time::interval(Duration::from_secs(30));
interval.reset();
let wait = task.wait_for_completion(client, poll, Some(timeout));
tokio::select! {
biased;
result = wait => {
info!("Task {id} completed in {:.2} seconds: {result:?}", now.elapsed().as_secs_f64());
result?;
}
_ = interval.tick() => {
struct Id(u32);
impl AsRef<u32> for Id {
fn as_ref(&self) -> &u32 {
&self.0
}
}
// it takes an AsRef<u32> but u32 itself doesn't impl it lol
if let Ok(task) = client.get_task(Id(id)).await {
if task.is_pending() {
info!("Task {id} is still pending after {:.2} seconds", now.elapsed().as_secs_f64());
}
} else {
error!("Error getting task {id}");
}
}
};
Ok(())
}
#[instrument(skip_all, fields(index.uid = %index.uid))]
async fn update_and_add_to_index(
client: &Client,
index: &Index,
projects: &[UploadSearchProject],
_additional_fields: &[String],
) -> Result<()> {
// TODO: Uncomment this- hardcoding loader_fields is a band-aid fix, and will be fixed soon
// let mut new_filterable_attributes: Vec<String> = index.get_filterable_attributes().await?;
// let mut new_displayed_attributes = index.get_displayed_attributes().await?;
// // Check if any 'additional_fields' are not already in the index
// // Only add if they are not already in the index
// let new_fields = additional_fields
// .iter()
// .filter(|x| !new_filterable_attributes.contains(x))
// .collect::<Vec<_>>();
// if !new_fields.is_empty() {
// info!("Adding new fields to index: {:?}", new_fields);
// new_filterable_attributes.extend(new_fields.iter().map(|s: &&String| s.to_string()));
// new_displayed_attributes.extend(new_fields.iter().map(|s| s.to_string()));
// // Adds new fields to the index
// let filterable_task = index
// .set_filterable_attributes(new_filterable_attributes)
// .await?;
// let displayable_task = index
// .set_displayed_attributes(new_displayed_attributes)
// .await?;
// // Allow a long timeout for adding new attributes- it only needs to happen the once
// filterable_task
// .wait_for_completion(client, None, Some(search_operation_timeout() * 100))
// .await?;
// displayable_task
// .wait_for_completion(client, None, Some(search_operation_timeout() * 100))
// .await?;
// }
info!("Adding to index.");
add_to_index(client, index, projects).await?;
Ok(())
}
pub async fn add_projects_batch_client(
indices: &[Vec<Index>],
projects: Vec<UploadSearchProject>,
additional_fields: Vec<String>,
config: &MeilisearchConfig,
) -> Result<()> {
let client = config.make_batch_client()?;
let index_references = indices
.iter()
.map(|x| x.iter().collect())
.collect::<Vec<Vec<&Index>>>();
let mut tasks = FuturesOrdered::new();
let mut id = 0;
client.across_all(index_references, |index_list, client| {
let span = info_span!("add_projects_batch", client.idx = id);
id += 1;
for index in index_list {
let owned_client = client.clone();
let projects_ref = &projects;
let additional_fields_ref = &additional_fields;
tasks.push_back(
async move {
update_and_add_to_index(
&owned_client,
index,
projects_ref,
additional_fields_ref,
)
.await
}
.instrument(span.clone()),
);
}
});
while let Some(result) = tasks.next().await {
result?;
}
Ok(())
}
fn default_settings() -> Settings {
Settings::new()
.with_distinct_attribute(Some("project_id"))
.with_displayed_attributes(DEFAULT_DISPLAYED_ATTRIBUTES)
.with_searchable_attributes(DEFAULT_SEARCHABLE_ATTRIBUTES)
.with_sortable_attributes(DEFAULT_SORTABLE_ATTRIBUTES)
.with_filterable_attributes(&*MEILI_FILTERABLE_ATTRIBUTES)
.with_pagination(PaginationSetting {
max_total_hits: 2147483647,
})
}
pub struct MeilisearchFieldSpec {
pub path: &'static str,
pub filterable: bool,
}
impl SearchField {
pub const fn meilisearch_spec(self) -> MeilisearchFieldSpec {
match self {
SearchField::Categories => MeilisearchFieldSpec {
path: "categories",
filterable: true,
},
SearchField::ProjectTypes => MeilisearchFieldSpec {
path: "project_types",
filterable: true,
},
SearchField::ProjectId => MeilisearchFieldSpec {
path: "project_id",
filterable: true,
},
SearchField::OpenSource => MeilisearchFieldSpec {
path: "open_source",
filterable: true,
},
SearchField::Environment => MeilisearchFieldSpec {
path: "environment",
filterable: true,
},
SearchField::GameVersions => MeilisearchFieldSpec {
path: "game_versions",
filterable: true,
},
SearchField::ClientSide => MeilisearchFieldSpec {
path: "client_side",
filterable: true,
},
SearchField::ServerSide => MeilisearchFieldSpec {
path: "server_side",
filterable: true,
},
SearchField::MinecraftServerRegion => MeilisearchFieldSpec {
path: "minecraft_server.region",
filterable: true,
},
SearchField::MinecraftServerLanguages => MeilisearchFieldSpec {
path: "minecraft_server.languages",
filterable: true,
},
SearchField::MinecraftJavaServerContentKind => {
MeilisearchFieldSpec {
path: "minecraft_java_server.content.kind",
filterable: true,
}
}
SearchField::MinecraftJavaServerContentSupportedGameVersions => {
MeilisearchFieldSpec {
path: "minecraft_java_server.content.supported_game_versions",
filterable: true,
}
}
SearchField::MinecraftJavaServerPingData => MeilisearchFieldSpec {
path: "minecraft_java_server.ping.data",
filterable: true,
},
}
}
}
static MEILI_FILTERABLE_ATTRIBUTES: LazyLock<Vec<&'static str>> =
LazyLock::new(|| {
use strum::IntoEnumIterator;
SearchField::iter()
.filter_map(|field| {
let spec = field.meilisearch_spec();
spec.filterable.then_some(spec.path)
})
.collect()
});
const DEFAULT_DISPLAYED_ATTRIBUTES: &[&str] = &[
"project_id",
"version_id",
"project_types",
"slug",
"author",
"name",
"summary",
"categories",
"display_categories",
"downloads",
"follows",
"icon_url",
"date_created",
"date_modified",
"latest_version",
"license",
"gallery",
"featured_gallery",
"color",
// Note: loader fields are not here, but are added on as they are needed (so they can be dynamically added depending on which exist).
// TODO: remove these- as they should be automatically populated. This is a band-aid fix.
"environment",
"game_versions",
"mrpack_loaders",
// V2 legacy fields for logical consistency
"client_side",
"server_side",
// Non-searchable fields for filling out the Project model.
"license_url",
"monetization_status",
"team_id",
"thread_id",
"versions",
"date_published",
"date_queued",
"status",
"requested_status",
"games",
"organization_id",
"links",
"gallery_items",
"loaders", // search uses loaders as categories- this is purely for the Project model.
"project_loader_fields",
"minecraft_mod",
"minecraft_server",
"minecraft_java_server",
"minecraft_bedrock_server",
];
const DEFAULT_SEARCHABLE_ATTRIBUTES: &[&str] =
&["name", "summary", "author", "slug"];
const DEFAULT_SORTABLE_ATTRIBUTES: &[&str] = &[
"downloads",
"follows",
"date_created",
"date_modified",
"version_published_timestamp",
"minecraft_java_server.verified_plays_2w",
"minecraft_java_server.ping.data.players_online",
];
@@ -0,0 +1,489 @@
use crate::database::PgPool;
use crate::database::redis::RedisPool;
use crate::env::ENV;
use crate::models::ids::VersionId;
use crate::routes::ApiError;
use crate::search::backend::{
SearchIndex, SearchIndexName, combined_search_filters, parse_search_index,
parse_search_request,
};
use crate::search::{
ResultSearchProject, SearchBackend, SearchRequest, SearchResults,
TasksCancelFilter,
};
use crate::util::error::Context;
use async_trait::async_trait;
use eyre::Result;
use futures::TryStreamExt;
use futures::stream::FuturesOrdered;
use itertools::Itertools;
use meilisearch_sdk::client::Client;
use meilisearch_sdk::tasks::{Task, TasksCancelQuery};
use serde::Serialize;
use serde_json::Value;
use std::collections::HashMap;
use std::fmt::Write;
use std::time::Duration;
use tracing::{Instrument, info_span};
pub mod indexing;
#[derive(Debug, Clone)]
pub struct MeilisearchReadClient {
pub client: Client,
}
impl std::ops::Deref for MeilisearchReadClient {
type Target = Client;
fn deref(&self) -> &Self::Target {
&self.client
}
}
pub struct BatchClient {
pub clients: Vec<Client>,
}
impl BatchClient {
pub fn new(clients: Vec<Client>) -> Self {
Self { clients }
}
pub async fn with_all_clients<'a, T, G, Fut>(
&'a self,
task_name: &str,
generator: G,
) -> Result<Vec<T>>
where
G: Fn(&'a Client) -> Fut,
Fut: Future<Output = Result<T>> + 'a,
{
let mut tasks = FuturesOrdered::new();
for (idx, client) in self.clients.iter().enumerate() {
tasks.push_back(generator(client).instrument(info_span!(
"client_task",
task.name = task_name,
client.idx = idx,
)));
}
let results = tasks.try_collect::<Vec<T>>().await?;
Ok(results)
}
pub fn across_all<T, F, R>(&self, data: Vec<T>, mut predicate: F) -> Vec<R>
where
F: FnMut(T, &Client) -> R,
{
assert_eq!(
data.len(),
self.clients.len(),
"mismatch between data len and meilisearch client count"
);
self.clients
.iter()
.zip(data)
.map(|(client, item)| predicate(item, client))
.collect()
}
}
#[derive(Debug, Clone)]
pub struct MeilisearchConfig {
pub addresses: Vec<String>,
pub read_lb_address: String,
pub key: String,
pub meta_namespace: String,
}
impl MeilisearchConfig {
pub fn new(meta_namespace: Option<String>) -> Self {
Self {
addresses: ENV.MEILISEARCH_WRITE_ADDRS.0.clone(),
key: ENV.MEILISEARCH_KEY.clone(),
meta_namespace: meta_namespace.unwrap_or_default(),
read_lb_address: ENV.MEILISEARCH_READ_ADDR.clone(),
}
}
pub fn make_loadbalanced_read_client(
&self,
) -> Result<MeilisearchReadClient, meilisearch_sdk::errors::Error> {
Ok(MeilisearchReadClient {
client: Client::new(&self.read_lb_address, Some(&self.key))?,
})
}
pub fn make_batch_client(
&self,
) -> Result<BatchClient, meilisearch_sdk::errors::Error> {
Ok(BatchClient::new(
self.addresses
.iter()
.map(|address| {
Client::new(address.as_str(), Some(self.key.as_str()))
})
.collect::<Result<Vec<_>, _>>()?,
))
}
pub fn get_index_name(&self, index: &str, next: bool) -> String {
let alt = if next { "_alt" } else { "" };
format!("{}_{}_{}", self.meta_namespace, index, alt)
}
}
pub struct Meilisearch {
pub config: MeilisearchConfig,
}
impl Meilisearch {
pub fn new(config: MeilisearchConfig) -> Self {
Self { config }
}
fn get_sort_index(
&self,
index: &str,
new_filters: Option<&str>,
) -> Result<(String, &'static [&'static str]), ApiError> {
let sort = parse_search_index(index, new_filters)?;
let index_name = match sort.index_name {
SearchIndexName::Projects => {
self.config.get_index_name("projects", false)
}
SearchIndexName::ProjectsFiltered => {
self.config.get_index_name("projects_filtered", false)
}
};
Ok(match sort.index {
SearchIndex::Relevance => (
index_name,
&["downloads:desc", "version_published_timestamp:desc"],
),
SearchIndex::Downloads => (
index_name,
&["downloads:desc", "version_published_timestamp:desc"],
),
SearchIndex::Follows => (
index_name,
&["follows:desc", "version_published_timestamp:desc"],
),
SearchIndex::Updated => (
index_name,
&["date_modified:desc", "version_published_timestamp:desc"],
),
SearchIndex::Newest => (
index_name,
&["date_created:desc", "version_published_timestamp:desc"],
),
SearchIndex::MinecraftJavaServerVerifiedPlays2w => (
index_name,
&[
"minecraft_java_server.verified_plays_2w:desc",
"minecraft_java_server.ping.data.players_online:desc",
"version_published_timestamp:desc",
],
),
SearchIndex::MinecraftJavaServerPlayersOnline => (
index_name,
&[
"minecraft_java_server.ping.data.players_online:desc",
"version_published_timestamp:desc",
],
),
})
}
}
#[async_trait]
impl SearchBackend for Meilisearch {
async fn search_for_project_raw(
&self,
info: &SearchRequest,
) -> Result<SearchResults, ApiError> {
let parsed = parse_search_request(info)?;
let (index_name, sort_name) =
self.get_sort_index(parsed.index, info.new_filters.as_deref())?;
let client = self
.config
.make_loadbalanced_read_client()
.wrap_internal_err("failed to make load-balanced read client")?;
let meilisearch_index = client
.get_index(index_name)
.await
.wrap_internal_err("failed to get index")?;
let mut filter_string = String::new();
let results = {
let mut query = meilisearch_index.search();
query
.with_page(parsed.page)
.with_hits_per_page(parsed.hits_per_page)
.with_query(parsed.query)
.with_sort(sort_name);
if let Some(new_filters) = info.new_filters.as_deref() {
query.with_filter(new_filters);
} else {
let facets = if let Some(facets) = &info.facets {
let facets =
serde_json::from_str::<Vec<Vec<Value>>>(facets)
.wrap_request_err("failed to parse facets")?;
Some(facets)
} else {
None
};
let filters =
combined_search_filters(info).unwrap_or_else(|| "".into());
if let Some(facets) = facets {
let facets: Vec<Vec<Vec<String>>> =
facets
.into_iter()
.map(|facets| {
facets
.into_iter()
.map(|facet| {
if facet.is_array() {
serde_json::from_value::<Vec<String>>(facet)
.unwrap_or_default()
} else {
vec![
serde_json::from_value::<String>(facet)
.unwrap_or_default(),
]
}
})
.collect_vec()
})
.collect_vec();
filter_string.push('(');
for (index, facet_outer_list) in facets.iter().enumerate() {
filter_string.push('(');
for (facet_outer_index, facet_inner_list) in
facet_outer_list.iter().enumerate()
{
filter_string.push('(');
for (facet_inner_index, facet) in
facet_inner_list.iter().enumerate()
{
filter_string
.push_str(&facet.replace(':', " = "));
if facet_inner_index
!= (facet_inner_list.len() - 1)
{
filter_string.push_str(" AND ")
}
}
filter_string.push(')');
if facet_outer_index != (facet_outer_list.len() - 1)
{
filter_string.push_str(" OR ")
}
}
filter_string.push(')');
if index != (facets.len() - 1) {
filter_string.push_str(" AND ")
}
}
filter_string.push(')');
if !filters.is_empty() {
write!(filter_string, " AND ({filters})")
.expect("write should not fail");
}
} else {
filter_string.push_str(&filters);
}
if !filter_string.is_empty() {
query.with_filter(&filter_string);
}
}
if info.show_metadata {
query.with_show_ranking_score(true);
query.with_show_ranking_score_details(true);
query.execute().await?
} else {
query.execute::<ResultSearchProject>().await?
}
};
if info.show_metadata {
let hits = results
.hits
.into_iter()
.map(|hit| {
let metadata = serde_json::to_value(&hit)
.ok()
.and_then(|value| value.as_object().cloned())
.map(|mut value| {
value.remove("_formatted");
value.remove("_matchesPosition");
value.remove("_federation");
let result = value.remove("result");
let metadata = Value::Object(value);
(result, metadata)
});
let (result, metadata) =
metadata.unwrap_or((None, Value::Null));
let mut result = result
.and_then(|value| {
serde_json::from_value::<ResultSearchProject>(value)
.ok()
})
.unwrap_or(hit.result);
if !metadata.is_null() {
result.search_metadata = Some(metadata);
}
result
})
.collect();
Ok(SearchResults {
hits,
page: results.page.unwrap_or_default(),
hits_per_page: results.hits_per_page.unwrap_or_default(),
total_hits: results.total_hits.unwrap_or_default(),
})
} else {
Ok(SearchResults {
hits: results.hits.into_iter().map(|r| r.result).collect(),
page: results.page.unwrap_or_default(),
hits_per_page: results.hits_per_page.unwrap_or_default(),
total_hits: results.total_hits.unwrap_or_default(),
})
}
}
async fn index_projects(
&self,
ro_pool: PgPool,
redis: RedisPool,
) -> eyre::Result<()> {
indexing::index_projects(ro_pool, redis, &self.config).await?;
Ok(())
}
async fn remove_documents(&self, ids: &[VersionId]) -> eyre::Result<()> {
indexing::remove_documents(ids, &self.config).await?;
Ok(())
}
async fn tasks(&self) -> eyre::Result<Value> {
let client = self
.config
.make_batch_client()
.wrap_internal_err("failed to make batch client")?;
let tasks = client
.with_all_clients("get_tasks", async |client| {
let tasks = client.get_tasks().await?;
Ok(tasks.results)
})
.await
.wrap_internal_err("failed to get tasks")?;
#[derive(Serialize)]
struct MeiliTask<Time> {
uid: u32,
status: &'static str,
duration: Option<Duration>,
enqueued_at: Option<Time>,
}
#[derive(Serialize)]
struct TaskList<Time> {
by_instance: HashMap<String, Vec<MeiliTask<Time>>>,
}
let response = tasks
.into_iter()
.enumerate()
.map(|(idx, instance_tasks)| {
let tasks = instance_tasks
.into_iter()
.filter_map(|task| {
Some(match task {
Task::Enqueued { content } => MeiliTask {
uid: content.uid,
status: "enqueued",
duration: None,
enqueued_at: Some(content.enqueued_at),
},
Task::Processing { content } => MeiliTask {
uid: content.uid,
status: "processing",
duration: None,
enqueued_at: Some(content.enqueued_at),
},
Task::Failed { content } => MeiliTask {
uid: content.task.uid,
status: "failed",
duration: Some(content.task.duration),
enqueued_at: Some(content.task.enqueued_at),
},
Task::Succeeded { .. } => return None,
})
})
.collect();
(idx.to_string(), tasks)
})
.collect::<HashMap<String, Vec<MeiliTask<_>>>>();
let response = serde_json::to_value(TaskList {
by_instance: response,
})
.wrap_internal_err("failed to serialize tasks response")?;
Ok(response)
}
async fn tasks_cancel(
&self,
filter: &TasksCancelFilter,
) -> eyre::Result<()> {
let client = self
.config
.make_batch_client()
.wrap_internal_err("failed to make batch client")?;
let all_results = client
.with_all_clients("cancel_tasks", async |client| {
let mut q = TasksCancelQuery::new(client);
match filter {
TasksCancelFilter::All => {}
TasksCancelFilter::Indexes { indexes } => {
q.with_index_uids(indexes.iter().map(|s| s.as_str()));
}
TasksCancelFilter::AllEnqueued => {
q.with_statuses(["enqueued"]);
}
};
let result = client.cancel_tasks_with(&q).await;
Ok(result)
})
.await
.wrap_internal_err("failed to cancel tasks")?;
for r in all_results {
r.wrap_internal_err("failed to cancel tasks")?;
}
Ok(())
}
}
+12
View File
@@ -0,0 +1,12 @@
mod common;
pub mod elasticsearch;
pub mod meilisearch;
pub mod typesense;
pub use common::{
ParsedSearchRequest, SearchIndex, SearchIndexName, SearchSort,
combined_search_filters, parse_search_index, parse_search_request,
};
pub use elasticsearch::Elasticsearch;
pub use meilisearch::{Meilisearch, MeilisearchConfig};
pub use typesense::{Typesense, TypesenseConfig};
File diff suppressed because it is too large Load Diff