initial commit

This commit is contained in:
mpostma
2020-12-12 13:32:06 +01:00
commit 8c0ab106c7
54 changed files with 13428 additions and 0 deletions

137
src/analytics.rs Normal file
View File

@ -0,0 +1,137 @@
use std::hash::{Hash, Hasher};
use std::{error, thread};
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
use log::error;
use serde::Serialize;
use serde_qs as qs;
use siphasher::sip::SipHasher;
use walkdir::WalkDir;
use crate::Data;
use crate::Opt;
const AMPLITUDE_API_KEY: &str = "f7fba398780e06d8fe6666a9be7e3d47";
#[derive(Debug, Serialize)]
struct EventProperties {
database_size: u64,
last_update_timestamp: Option<i64>, //timestamp
number_of_documents: Vec<u64>,
}
impl EventProperties {
fn from(data: Data) -> Result<EventProperties, Box<dyn error::Error>> {
let mut index_list = Vec::new();
let reader = data.db.main_read_txn()?;
for index_uid in data.db.indexes_uids() {
if let Some(index) = data.db.open_index(&index_uid) {
let number_of_documents = index.main.number_of_documents(&reader)?;
index_list.push(number_of_documents);
}
}
let database_size = WalkDir::new(&data.db_path)
.into_iter()
.filter_map(|entry| entry.ok())
.filter_map(|entry| entry.metadata().ok())
.filter(|metadata| metadata.is_file())
.fold(0, |acc, m| acc + m.len());
let last_update_timestamp = data.db.last_update(&reader)?.map(|u| u.timestamp());
Ok(EventProperties {
database_size,
last_update_timestamp,
number_of_documents: index_list,
})
}
}
#[derive(Debug, Serialize)]
struct UserProperties<'a> {
env: &'a str,
start_since_days: u64,
user_email: Option<String>,
server_provider: Option<String>,
}
#[derive(Debug, Serialize)]
struct Event<'a> {
user_id: &'a str,
event_type: &'a str,
device_id: &'a str,
time: u64,
app_version: &'a str,
user_properties: UserProperties<'a>,
event_properties: Option<EventProperties>,
}
#[derive(Debug, Serialize)]
struct AmplitudeRequest<'a> {
api_key: &'a str,
event: &'a str,
}
pub fn analytics_sender(data: Data, opt: Opt) {
let username = whoami::username();
let hostname = whoami::hostname();
let platform = whoami::platform();
let uid = username + &hostname + &platform.to_string();
let mut hasher = SipHasher::new();
uid.hash(&mut hasher);
let hash = hasher.finish();
let uid = format!("{:X}", hash);
let platform = platform.to_string();
let first_start = Instant::now();
loop {
let n = SystemTime::now().duration_since(UNIX_EPOCH).unwrap();
let user_id = &uid;
let device_id = &platform;
let time = n.as_secs();
let event_type = "runtime_tick";
let elapsed_since_start = first_start.elapsed().as_secs() / 86_400; // One day
let event_properties = EventProperties::from(data.clone()).ok();
let app_version = env!("CARGO_PKG_VERSION").to_string();
let app_version = app_version.as_str();
let user_email = std::env::var("MEILI_USER_EMAIL").ok();
let server_provider = std::env::var("MEILI_SERVER_PROVIDER").ok();
let user_properties = UserProperties {
env: &opt.env,
start_since_days: elapsed_since_start,
user_email,
server_provider,
};
let event = Event {
user_id,
event_type,
device_id,
time,
app_version,
user_properties,
event_properties
};
let event = serde_json::to_string(&event).unwrap();
let request = AmplitudeRequest {
api_key: AMPLITUDE_API_KEY,
event: &event,
};
let body = qs::to_string(&request).unwrap();
let response = ureq::post("https://api.amplitude.com/httpapi").send_string(&body);
if !response.ok() {
let body = response.into_string().unwrap();
error!("Unsuccessful call to Amplitude: {}", body);
}
thread::sleep(Duration::from_secs(3600)) // one hour
}
}

162
src/data.rs Normal file
View File

@ -0,0 +1,162 @@
use std::error::Error;
use std::ops::Deref;
use std::path::PathBuf;
use std::sync::Arc;
use meilisearch_core::{Database, DatabaseOptions, Index};
use sha2::Digest;
use crate::error::{Error as MSError, ResponseError};
use crate::index_update_callback;
use crate::option::Opt;
#[derive(Clone)]
pub struct Data {
inner: Arc<DataInner>,
}
impl Deref for Data {
type Target = DataInner;
fn deref(&self) -> &Self::Target {
&self.inner
}
}
#[derive(Clone)]
pub struct DataInner {
pub db: Arc<Database>,
pub db_path: String,
pub dumps_dir: PathBuf,
pub dump_batch_size: usize,
pub api_keys: ApiKeys,
pub server_pid: u32,
pub http_payload_size_limit: usize,
}
#[derive(Clone)]
pub struct ApiKeys {
pub public: Option<String>,
pub private: Option<String>,
pub master: Option<String>,
}
impl ApiKeys {
pub fn generate_missing_api_keys(&mut self) {
if let Some(master_key) = &self.master {
if self.private.is_none() {
let key = format!("{}-private", master_key);
let sha = sha2::Sha256::digest(key.as_bytes());
self.private = Some(format!("{:x}", sha));
}
if self.public.is_none() {
let key = format!("{}-public", master_key);
let sha = sha2::Sha256::digest(key.as_bytes());
self.public = Some(format!("{:x}", sha));
}
}
}
}
impl Data {
pub fn new(opt: Opt) -> Result<Data, Box<dyn Error>> {
let db_path = opt.db_path.clone();
let dumps_dir = opt.dumps_dir.clone();
let dump_batch_size = opt.dump_batch_size;
let server_pid = std::process::id();
let db_opt = DatabaseOptions {
main_map_size: opt.max_mdb_size,
update_map_size: opt.max_udb_size,
};
let http_payload_size_limit = opt.http_payload_size_limit;
let db = Arc::new(Database::open_or_create(opt.db_path, db_opt)?);
let mut api_keys = ApiKeys {
master: opt.master_key,
private: None,
public: None,
};
api_keys.generate_missing_api_keys();
let inner_data = DataInner {
db: db.clone(),
db_path,
dumps_dir,
dump_batch_size,
api_keys,
server_pid,
http_payload_size_limit,
};
let data = Data {
inner: Arc::new(inner_data),
};
let callback_context = data.clone();
db.set_update_callback(Box::new(move |index_uid, status| {
index_update_callback(&index_uid, &callback_context, status);
}));
Ok(data)
}
fn create_index(&self, uid: &str) -> Result<Index, ResponseError> {
if !uid
.chars()
.all(|x| x.is_ascii_alphanumeric() || x == '-' || x == '_')
{
return Err(MSError::InvalidIndexUid.into());
}
let created_index = self.db.create_index(&uid).map_err(|e| match e {
meilisearch_core::Error::IndexAlreadyExists => e.into(),
_ => ResponseError::from(MSError::create_index(e)),
})?;
self.db.main_write::<_, _, ResponseError>(|mut writer| {
created_index.main.put_name(&mut writer, uid)?;
created_index
.main
.created_at(&writer)?
.ok_or(MSError::internal("Impossible to read created at"))?;
created_index
.main
.updated_at(&writer)?
.ok_or(MSError::internal("Impossible to read updated at"))?;
Ok(())
})?;
Ok(created_index)
}
pub fn get_or_create_index<F, R>(&self, uid: &str, f: F) -> Result<R, ResponseError>
where
F: FnOnce(&Index) -> Result<R, ResponseError>,
{
let mut index_has_been_created = false;
let index = match self.db.open_index(&uid) {
Some(index) => index,
None => {
index_has_been_created = true;
self.create_index(&uid)?
}
};
match f(&index) {
Ok(r) => Ok(r),
Err(err) => {
if index_has_been_created {
let _ = self.db.delete_index(&uid);
}
Err(err)
}
}
}
}

425
src/dump.rs Normal file
View File

@ -0,0 +1,425 @@
use std::fs::{create_dir_all, File};
use std::io::prelude::*;
use std::path::{Path, PathBuf};
use std::sync::Mutex;
use std::thread;
use actix_web::web;
use chrono::offset::Utc;
use indexmap::IndexMap;
use log::{error, info};
use meilisearch_core::{MainWriter, MainReader, UpdateReader};
use meilisearch_core::settings::Settings;
use meilisearch_core::update::{apply_settings_update, apply_documents_addition};
use once_cell::sync::Lazy;
use serde::{Deserialize, Serialize};
use serde_json::json;
use tempfile::TempDir;
use crate::Data;
use crate::error::{Error, ResponseError};
use crate::helpers::compression;
use crate::routes::index;
use crate::routes::index::IndexResponse;
// Mutex to share dump progress.
static DUMP_INFO: Lazy<Mutex<Option<DumpInfo>>> = Lazy::new(Mutex::default);
#[derive(Debug, Serialize, Deserialize, Copy, Clone)]
enum DumpVersion {
V1,
}
impl DumpVersion {
const CURRENT: Self = Self::V1;
}
#[derive(Debug, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct DumpMetadata {
indexes: Vec<crate::routes::index::IndexResponse>,
db_version: String,
dump_version: DumpVersion,
}
impl DumpMetadata {
/// Create a DumpMetadata with the current dump version of meilisearch.
pub fn new(indexes: Vec<crate::routes::index::IndexResponse>, db_version: String) -> Self {
DumpMetadata {
indexes,
db_version,
dump_version: DumpVersion::CURRENT,
}
}
/// Extract DumpMetadata from `metadata.json` file present at provided `dir_path`
fn from_path(dir_path: &Path) -> Result<Self, Error> {
let path = dir_path.join("metadata.json");
let file = File::open(path)?;
let reader = std::io::BufReader::new(file);
let metadata = serde_json::from_reader(reader)?;
Ok(metadata)
}
/// Write DumpMetadata in `metadata.json` file at provided `dir_path`
fn to_path(&self, dir_path: &Path) -> Result<(), Error> {
let path = dir_path.join("metadata.json");
let file = File::create(path)?;
serde_json::to_writer(file, &self)?;
Ok(())
}
}
/// Extract Settings from `settings.json` file present at provided `dir_path`
fn settings_from_path(dir_path: &Path) -> Result<Settings, Error> {
let path = dir_path.join("settings.json");
let file = File::open(path)?;
let reader = std::io::BufReader::new(file);
let metadata = serde_json::from_reader(reader)?;
Ok(metadata)
}
/// Write Settings in `settings.json` file at provided `dir_path`
fn settings_to_path(settings: &Settings, dir_path: &Path) -> Result<(), Error> {
let path = dir_path.join("settings.json");
let file = File::create(path)?;
serde_json::to_writer(file, settings)?;
Ok(())
}
/// Import settings and documents of a dump with version `DumpVersion::V1` in specified index.
fn import_index_v1(
data: &Data,
dumps_dir: &Path,
index_uid: &str,
document_batch_size: usize,
write_txn: &mut MainWriter,
) -> Result<(), Error> {
// open index
let index = data
.db
.open_index(index_uid)
.ok_or(Error::index_not_found(index_uid))?;
// index dir path in dump dir
let index_path = &dumps_dir.join(index_uid);
// extract `settings.json` file and import content
let settings = settings_from_path(&index_path)?;
let settings = settings.to_update().map_err(|e| Error::dump_failed(format!("importing settings for index {}; {}", index_uid, e)))?;
apply_settings_update(write_txn, &index, settings)?;
// create iterator over documents in `documents.jsonl` to make batch importation
// create iterator over documents in `documents.jsonl` to make batch importation
let documents = {
let file = File::open(&index_path.join("documents.jsonl"))?;
let reader = std::io::BufReader::new(file);
let deserializer = serde_json::Deserializer::from_reader(reader);
deserializer.into_iter::<IndexMap<String, serde_json::Value>>()
};
// batch import document every `document_batch_size`:
// create a Vec to bufferize documents
let mut values = Vec::with_capacity(document_batch_size);
// iterate over documents
for document in documents {
// push document in buffer
values.push(document?);
// if buffer is full, create and apply a batch, and clean buffer
if values.len() == document_batch_size {
let batch = std::mem::replace(&mut values, Vec::with_capacity(document_batch_size));
apply_documents_addition(write_txn, &index, batch)?;
}
}
// apply documents remaining in the buffer
if !values.is_empty() {
apply_documents_addition(write_txn, &index, values)?;
}
// sync index information: stats, updated_at, last_update
if let Err(e) = crate::index_update_callback_txn(index, index_uid, data, write_txn) {
return Err(Error::Internal(e));
}
Ok(())
}
/// Import dump from `dump_path` in database.
pub fn import_dump(
data: &Data,
dump_path: &Path,
document_batch_size: usize,
) -> Result<(), Error> {
info!("Importing dump from {:?}...", dump_path);
// create a temporary directory
let tmp_dir = TempDir::new()?;
let tmp_dir_path = tmp_dir.path();
// extract dump in temporary directory
compression::from_tar_gz(dump_path, tmp_dir_path)?;
// read dump metadata
let metadata = DumpMetadata::from_path(&tmp_dir_path)?;
// choose importation function from DumpVersion of metadata
let import_index = match metadata.dump_version {
DumpVersion::V1 => import_index_v1,
};
// remove indexes which have same `uid` than indexes to import and create empty indexes
let existing_index_uids = data.db.indexes_uids();
for index in metadata.indexes.iter() {
if existing_index_uids.contains(&index.uid) {
data.db.delete_index(index.uid.clone())?;
}
index::create_index_sync(&data.db, index.uid.clone(), index.name.clone(), index.primary_key.clone())?;
}
// import each indexes content
data.db.main_write::<_, _, Error>(|mut writer| {
for index in metadata.indexes {
import_index(&data, tmp_dir_path, &index.uid, document_batch_size, &mut writer)?;
}
Ok(())
})?;
info!("Dump importation from {:?} succeed", dump_path);
Ok(())
}
#[derive(Debug, Serialize, Deserialize, PartialEq, Clone)]
#[serde(rename_all = "snake_case")]
pub enum DumpStatus {
Done,
InProgress,
Failed,
}
#[derive(Debug, Serialize, Clone)]
#[serde(rename_all = "camelCase")]
pub struct DumpInfo {
pub uid: String,
pub status: DumpStatus,
#[serde(skip_serializing_if = "Option::is_none", flatten)]
pub error: Option<serde_json::Value>,
}
impl DumpInfo {
pub fn new(uid: String, status: DumpStatus) -> Self {
Self { uid, status, error: None }
}
pub fn with_error(mut self, error: ResponseError) -> Self {
self.status = DumpStatus::Failed;
self.error = Some(json!(error));
self
}
pub fn dump_already_in_progress(&self) -> bool {
self.status == DumpStatus::InProgress
}
pub fn get_current() -> Option<Self> {
DUMP_INFO.lock().unwrap().clone()
}
pub fn set_current(&self) {
*DUMP_INFO.lock().unwrap() = Some(self.clone());
}
}
/// Generate uid from creation date
fn generate_uid() -> String {
Utc::now().format("%Y%m%d-%H%M%S%3f").to_string()
}
/// Infer dumps_dir from dump_uid
pub fn compressed_dumps_dir(dumps_dir: &Path, dump_uid: &str) -> PathBuf {
dumps_dir.join(format!("{}.dump", dump_uid))
}
/// Write metadata in dump
fn dump_metadata(data: &web::Data<Data>, dir_path: &Path, indexes: Vec<IndexResponse>) -> Result<(), Error> {
let (db_major, db_minor, db_patch) = data.db.version();
let metadata = DumpMetadata::new(indexes, format!("{}.{}.{}", db_major, db_minor, db_patch));
metadata.to_path(dir_path)
}
/// Export settings of provided index in dump
fn dump_index_settings(data: &web::Data<Data>, reader: &MainReader, dir_path: &Path, index_uid: &str) -> Result<(), Error> {
let settings = crate::routes::setting::get_all_sync(data, reader, index_uid)?;
settings_to_path(&settings, dir_path)
}
/// Export updates of provided index in dump
fn dump_index_updates(data: &web::Data<Data>, reader: &UpdateReader, dir_path: &Path, index_uid: &str) -> Result<(), Error> {
let updates_path = dir_path.join("updates.jsonl");
let updates = crate::routes::index::get_all_updates_status_sync(data, reader, index_uid)?;
let file = File::create(updates_path)?;
for update in updates {
serde_json::to_writer(&file, &update)?;
writeln!(&file)?;
}
Ok(())
}
/// Export documents of provided index in dump
fn dump_index_documents(data: &web::Data<Data>, reader: &MainReader, dir_path: &Path, index_uid: &str) -> Result<(), Error> {
let documents_path = dir_path.join("documents.jsonl");
let file = File::create(documents_path)?;
let dump_batch_size = data.dump_batch_size;
let mut offset = 0;
loop {
let documents = crate::routes::document::get_all_documents_sync(data, reader, index_uid, offset, dump_batch_size, None)?;
if documents.is_empty() { break; } else { offset += dump_batch_size; }
for document in documents {
serde_json::to_writer(&file, &document)?;
writeln!(&file)?;
}
}
Ok(())
}
/// Write error with a context.
fn fail_dump_process<E: std::error::Error>(dump_info: DumpInfo, context: &str, error: E) {
let error_message = format!("{}; {}", context, error);
error!("Something went wrong during dump process: {}", &error_message);
dump_info.with_error(Error::dump_failed(error_message).into()).set_current();
}
/// Main function of dump.
fn dump_process(data: web::Data<Data>, dumps_dir: PathBuf, dump_info: DumpInfo) {
// open read transaction on Update
let update_reader = match data.db.update_read_txn() {
Ok(r) => r,
Err(e) => {
fail_dump_process(dump_info, "creating RO transaction on updates", e);
return ;
}
};
// open read transaction on Main
let main_reader = match data.db.main_read_txn() {
Ok(r) => r,
Err(e) => {
fail_dump_process(dump_info, "creating RO transaction on main", e);
return ;
}
};
// create a temporary directory
let tmp_dir = match TempDir::new() {
Ok(tmp_dir) => tmp_dir,
Err(e) => {
fail_dump_process(dump_info, "creating temporary directory", e);
return ;
}
};
let tmp_dir_path = tmp_dir.path();
// fetch indexes
let indexes = match crate::routes::index::list_indexes_sync(&data, &main_reader) {
Ok(indexes) => indexes,
Err(e) => {
fail_dump_process(dump_info, "listing indexes", e);
return ;
}
};
// create metadata
if let Err(e) = dump_metadata(&data, &tmp_dir_path, indexes.clone()) {
fail_dump_process(dump_info, "generating metadata", e);
return ;
}
// export settings, updates and documents for each indexes
for index in indexes {
let index_path = tmp_dir_path.join(&index.uid);
// create index sub-dircetory
if let Err(e) = create_dir_all(&index_path) {
fail_dump_process(dump_info, &format!("creating directory for index {}", &index.uid), e);
return ;
}
// export settings
if let Err(e) = dump_index_settings(&data, &main_reader, &index_path, &index.uid) {
fail_dump_process(dump_info, &format!("generating settings for index {}", &index.uid), e);
return ;
}
// export documents
if let Err(e) = dump_index_documents(&data, &main_reader, &index_path, &index.uid) {
fail_dump_process(dump_info, &format!("generating documents for index {}", &index.uid), e);
return ;
}
// export updates
if let Err(e) = dump_index_updates(&data, &update_reader, &index_path, &index.uid) {
fail_dump_process(dump_info, &format!("generating updates for index {}", &index.uid), e);
return ;
}
}
// compress dump in a file named `{dump_uid}.dump` in `dumps_dir`
if let Err(e) = crate::helpers::compression::to_tar_gz(&tmp_dir_path, &compressed_dumps_dir(&dumps_dir, &dump_info.uid)) {
fail_dump_process(dump_info, "compressing dump", e);
return ;
}
// update dump info to `done`
let resume = DumpInfo::new(
dump_info.uid,
DumpStatus::Done
);
resume.set_current();
}
pub fn init_dump_process(data: &web::Data<Data>, dumps_dir: &Path) -> Result<DumpInfo, Error> {
create_dir_all(dumps_dir).map_err(|e| Error::dump_failed(format!("creating temporary directory {}", e)))?;
// check if a dump is already in progress
if let Some(resume) = DumpInfo::get_current() {
if resume.dump_already_in_progress() {
return Err(Error::dump_conflict())
}
}
// generate a new dump info
let info = DumpInfo::new(
generate_uid(),
DumpStatus::InProgress
);
info.set_current();
let data = data.clone();
let dumps_dir = dumps_dir.to_path_buf();
let info_cloned = info.clone();
// run dump process in a new thread
thread::spawn(move ||
dump_process(data, dumps_dir, info_cloned)
);
Ok(info)
}

307
src/error.rs Normal file
View File

@ -0,0 +1,307 @@
use std::error;
use std::fmt;
use actix_http::ResponseBuilder;
use actix_web as aweb;
use actix_web::error::{JsonPayloadError, QueryPayloadError};
use actix_web::http::StatusCode;
use serde::ser::{Serialize, Serializer, SerializeStruct};
use meilisearch_error::{ErrorCode, Code};
#[derive(Debug)]
pub struct ResponseError {
inner: Box<dyn ErrorCode>,
}
impl error::Error for ResponseError {}
impl ErrorCode for ResponseError {
fn error_code(&self) -> Code {
self.inner.error_code()
}
}
impl fmt::Display for ResponseError {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
self.inner.fmt(f)
}
}
impl From<Error> for ResponseError {
fn from(error: Error) -> ResponseError {
ResponseError { inner: Box::new(error) }
}
}
impl From<meilisearch_core::Error> for ResponseError {
fn from(err: meilisearch_core::Error) -> ResponseError {
ResponseError { inner: Box::new(err) }
}
}
impl From<meilisearch_schema::Error> for ResponseError {
fn from(err: meilisearch_schema::Error) -> ResponseError {
ResponseError { inner: Box::new(err) }
}
}
impl From<FacetCountError> for ResponseError {
fn from(err: FacetCountError) -> ResponseError {
ResponseError { inner: Box::new(err) }
}
}
impl Serialize for ResponseError {
fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
where
S: Serializer,
{
let struct_name = "ResponseError";
let field_count = 4;
let mut state = serializer.serialize_struct(struct_name, field_count)?;
state.serialize_field("message", &self.to_string())?;
state.serialize_field("errorCode", &self.error_name())?;
state.serialize_field("errorType", &self.error_type())?;
state.serialize_field("errorLink", &self.error_url())?;
state.end()
}
}
impl aweb::error::ResponseError for ResponseError {
fn error_response(&self) -> aweb::HttpResponse {
ResponseBuilder::new(self.status_code()).json(&self)
}
fn status_code(&self) -> StatusCode {
self.http_status()
}
}
#[derive(Debug)]
pub enum Error {
BadParameter(String, String),
BadRequest(String),
CreateIndex(String),
DocumentNotFound(String),
IndexNotFound(String),
IndexAlreadyExists(String),
Internal(String),
InvalidIndexUid,
InvalidToken(String),
MissingAuthorizationHeader,
NotFound(String),
OpenIndex(String),
RetrieveDocument(u32, String),
SearchDocuments(String),
PayloadTooLarge,
UnsupportedMediaType,
DumpAlreadyInProgress,
DumpProcessFailed(String),
}
impl error::Error for Error {}
impl ErrorCode for Error {
fn error_code(&self) -> Code {
use Error::*;
match self {
BadParameter(_, _) => Code::BadParameter,
BadRequest(_) => Code::BadRequest,
CreateIndex(_) => Code::CreateIndex,
DocumentNotFound(_) => Code::DocumentNotFound,
IndexNotFound(_) => Code::IndexNotFound,
IndexAlreadyExists(_) => Code::IndexAlreadyExists,
Internal(_) => Code::Internal,
InvalidIndexUid => Code::InvalidIndexUid,
InvalidToken(_) => Code::InvalidToken,
MissingAuthorizationHeader => Code::MissingAuthorizationHeader,
NotFound(_) => Code::NotFound,
OpenIndex(_) => Code::OpenIndex,
RetrieveDocument(_, _) => Code::RetrieveDocument,
SearchDocuments(_) => Code::SearchDocuments,
PayloadTooLarge => Code::PayloadTooLarge,
UnsupportedMediaType => Code::UnsupportedMediaType,
DumpAlreadyInProgress => Code::DumpAlreadyInProgress,
DumpProcessFailed(_) => Code::DumpProcessFailed,
}
}
}
#[derive(Debug)]
pub enum FacetCountError {
AttributeNotSet(String),
SyntaxError(String),
UnexpectedToken { found: String, expected: &'static [&'static str] },
NoFacetSet,
}
impl error::Error for FacetCountError {}
impl ErrorCode for FacetCountError {
fn error_code(&self) -> Code {
Code::BadRequest
}
}
impl FacetCountError {
pub fn unexpected_token(found: impl ToString, expected: &'static [&'static str]) -> FacetCountError {
let found = found.to_string();
FacetCountError::UnexpectedToken { expected, found }
}
}
impl From<serde_json::error::Error> for FacetCountError {
fn from(other: serde_json::error::Error) -> FacetCountError {
FacetCountError::SyntaxError(other.to_string())
}
}
impl fmt::Display for FacetCountError {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
use FacetCountError::*;
match self {
AttributeNotSet(attr) => write!(f, "Attribute {} is not set as facet", attr),
SyntaxError(msg) => write!(f, "Syntax error: {}", msg),
UnexpectedToken { expected, found } => write!(f, "Unexpected {} found, expected {:?}", found, expected),
NoFacetSet => write!(f, "Can't perform facet count, as no facet is set"),
}
}
}
impl Error {
pub fn internal(err: impl fmt::Display) -> Error {
Error::Internal(err.to_string())
}
pub fn bad_request(err: impl fmt::Display) -> Error {
Error::BadRequest(err.to_string())
}
pub fn missing_authorization_header() -> Error {
Error::MissingAuthorizationHeader
}
pub fn invalid_token(err: impl fmt::Display) -> Error {
Error::InvalidToken(err.to_string())
}
pub fn not_found(err: impl fmt::Display) -> Error {
Error::NotFound(err.to_string())
}
pub fn index_not_found(err: impl fmt::Display) -> Error {
Error::IndexNotFound(err.to_string())
}
pub fn document_not_found(err: impl fmt::Display) -> Error {
Error::DocumentNotFound(err.to_string())
}
pub fn bad_parameter(param: impl fmt::Display, err: impl fmt::Display) -> Error {
Error::BadParameter(param.to_string(), err.to_string())
}
pub fn open_index(err: impl fmt::Display) -> Error {
Error::OpenIndex(err.to_string())
}
pub fn create_index(err: impl fmt::Display) -> Error {
Error::CreateIndex(err.to_string())
}
pub fn invalid_index_uid() -> Error {
Error::InvalidIndexUid
}
pub fn retrieve_document(doc_id: u32, err: impl fmt::Display) -> Error {
Error::RetrieveDocument(doc_id, err.to_string())
}
pub fn search_documents(err: impl fmt::Display) -> Error {
Error::SearchDocuments(err.to_string())
}
pub fn dump_conflict() -> Error {
Error::DumpAlreadyInProgress
}
pub fn dump_failed(message: String) -> Error {
Error::DumpProcessFailed(message)
}
}
impl fmt::Display for Error {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::BadParameter(param, err) => write!(f, "Url parameter {} error: {}", param, err),
Self::BadRequest(err) => f.write_str(err),
Self::CreateIndex(err) => write!(f, "Impossible to create index; {}", err),
Self::DocumentNotFound(document_id) => write!(f, "Document with id {} not found", document_id),
Self::IndexNotFound(index_uid) => write!(f, "Index {} not found", index_uid),
Self::IndexAlreadyExists(index_uid) => write!(f, "Index {} already exists", index_uid),
Self::Internal(err) => f.write_str(err),
Self::InvalidIndexUid => f.write_str("Index must have a valid uid; Index uid can be of type integer or string only composed of alphanumeric characters, hyphens (-) and underscores (_)."),
Self::InvalidToken(err) => write!(f, "Invalid API key: {}", err),
Self::MissingAuthorizationHeader => f.write_str("You must have an authorization token"),
Self::NotFound(err) => write!(f, "{} not found", err),
Self::OpenIndex(err) => write!(f, "Impossible to open index; {}", err),
Self::RetrieveDocument(id, err) => write!(f, "Impossible to retrieve the document with id: {}; {}", id, err),
Self::SearchDocuments(err) => write!(f, "Impossible to search documents; {}", err),
Self::PayloadTooLarge => f.write_str("Payload too large"),
Self::UnsupportedMediaType => f.write_str("Unsupported media type"),
Self::DumpAlreadyInProgress => f.write_str("Another dump is already in progress"),
Self::DumpProcessFailed(message) => write!(f, "Dump process failed: {}", message),
}
}
}
impl From<std::io::Error> for Error {
fn from(err: std::io::Error) -> Error {
Error::Internal(err.to_string())
}
}
impl From<actix_http::Error> for Error {
fn from(err: actix_http::Error) -> Error {
Error::Internal(err.to_string())
}
}
impl From<meilisearch_core::Error> for Error {
fn from(err: meilisearch_core::Error) -> Error {
Error::Internal(err.to_string())
}
}
impl From<serde_json::error::Error> for Error {
fn from(err: serde_json::error::Error) -> Error {
Error::Internal(err.to_string())
}
}
impl From<JsonPayloadError> for Error {
fn from(err: JsonPayloadError) -> Error {
match err {
JsonPayloadError::Deserialize(err) => Error::BadRequest(format!("Invalid JSON: {}", err)),
JsonPayloadError::Overflow => Error::PayloadTooLarge,
JsonPayloadError::ContentType => Error::UnsupportedMediaType,
JsonPayloadError::Payload(err) => Error::BadRequest(format!("Problem while decoding the request: {}", err)),
}
}
}
impl From<QueryPayloadError> for Error {
fn from(err: QueryPayloadError) -> Error {
match err {
QueryPayloadError::Deserialize(err) => Error::BadRequest(format!("Invalid query parameters: {}", err)),
}
}
}
pub fn payload_error_handler<E: Into<Error>>(err: E) -> ResponseError {
let error: Error = err.into();
error.into()
}

View File

@ -0,0 +1,103 @@
use std::cell::RefCell;
use std::pin::Pin;
use std::rc::Rc;
use std::task::{Context, Poll};
use actix_service::{Service, Transform};
use actix_web::{dev::ServiceRequest, dev::ServiceResponse, web};
use futures::future::{err, ok, Future, Ready};
use crate::error::{Error, ResponseError};
use crate::Data;
#[derive(Clone)]
pub enum Authentication {
Public,
Private,
Admin,
}
impl<S: 'static, B> Transform<S> for Authentication
where
S: Service<Request = ServiceRequest, Response = ServiceResponse<B>, Error = actix_web::Error>,
S::Future: 'static,
B: 'static,
{
type Request = ServiceRequest;
type Response = ServiceResponse<B>;
type Error = actix_web::Error;
type InitError = ();
type Transform = LoggingMiddleware<S>;
type Future = Ready<Result<Self::Transform, Self::InitError>>;
fn new_transform(&self, service: S) -> Self::Future {
ok(LoggingMiddleware {
acl: self.clone(),
service: Rc::new(RefCell::new(service)),
})
}
}
pub struct LoggingMiddleware<S> {
acl: Authentication,
service: Rc<RefCell<S>>,
}
#[allow(clippy::type_complexity)]
impl<S, B> Service for LoggingMiddleware<S>
where
S: Service<Request = ServiceRequest, Response = ServiceResponse<B>, Error = actix_web::Error> + 'static,
S::Future: 'static,
B: 'static,
{
type Request = ServiceRequest;
type Response = ServiceResponse<B>;
type Error = actix_web::Error;
type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>>>>;
fn poll_ready(&mut self, cx: &mut Context) -> Poll<Result<(), Self::Error>> {
self.service.poll_ready(cx)
}
fn call(&mut self, req: ServiceRequest) -> Self::Future {
let mut svc = self.service.clone();
// This unwrap is left because this error should never appear. If that's the case, then
// it means that actix-web has an issue or someone changes the type `Data`.
let data = req.app_data::<web::Data<Data>>().unwrap();
if data.api_keys.master.is_none() {
return Box::pin(svc.call(req));
}
let auth_header = match req.headers().get("X-Meili-API-Key") {
Some(auth) => match auth.to_str() {
Ok(auth) => auth,
Err(_) => return Box::pin(err(ResponseError::from(Error::MissingAuthorizationHeader).into())),
},
None => {
return Box::pin(err(ResponseError::from(Error::MissingAuthorizationHeader).into()));
}
};
let authenticated = match self.acl {
Authentication::Admin => data.api_keys.master.as_deref() == Some(auth_header),
Authentication::Private => {
data.api_keys.master.as_deref() == Some(auth_header)
|| data.api_keys.private.as_deref() == Some(auth_header)
}
Authentication::Public => {
data.api_keys.master.as_deref() == Some(auth_header)
|| data.api_keys.private.as_deref() == Some(auth_header)
|| data.api_keys.public.as_deref() == Some(auth_header)
}
};
if authenticated {
Box::pin(svc.call(req))
} else {
Box::pin(err(
ResponseError::from(Error::InvalidToken(auth_header.to_string())).into()
))
}
}
}

View File

@ -0,0 +1,27 @@
use flate2::Compression;
use flate2::read::GzDecoder;
use flate2::write::GzEncoder;
use std::fs::{create_dir_all, File};
use std::path::Path;
use tar::{Builder, Archive};
use crate::error::Error;
pub fn to_tar_gz(src: &Path, dest: &Path) -> Result<(), Error> {
let f = File::create(dest)?;
let gz_encoder = GzEncoder::new(f, Compression::default());
let mut tar_encoder = Builder::new(gz_encoder);
tar_encoder.append_dir_all(".", src)?;
let gz_encoder = tar_encoder.into_inner()?;
gz_encoder.finish()?;
Ok(())
}
pub fn from_tar_gz(src: &Path, dest: &Path) -> Result<(), Error> {
let f = File::open(src)?;
let gz = GzDecoder::new(f);
let mut ar = Archive::new(gz);
create_dir_all(dest)?;
ar.unpack(dest)?;
Ok(())
}

652
src/helpers/meilisearch.rs Normal file
View File

@ -0,0 +1,652 @@
use std::cmp::Ordering;
use std::collections::{HashMap, HashSet};
use std::hash::{Hash, Hasher};
use std::time::Instant;
use indexmap::IndexMap;
use log::error;
use meilisearch_core::{Filter, MainReader};
use meilisearch_core::facets::FacetFilter;
use meilisearch_core::criterion::*;
use meilisearch_core::settings::RankingRule;
use meilisearch_core::{Highlight, Index, RankedMap};
use meilisearch_schema::{FieldId, Schema};
use meilisearch_tokenizer::is_cjk;
use serde::{Deserialize, Serialize};
use serde_json::Value;
use siphasher::sip::SipHasher;
use slice_group_by::GroupBy;
use crate::error::{Error, ResponseError};
pub trait IndexSearchExt {
fn new_search(&self, query: Option<String>) -> SearchBuilder;
}
impl IndexSearchExt for Index {
fn new_search(&self, query: Option<String>) -> SearchBuilder {
SearchBuilder {
index: self,
query,
offset: 0,
limit: 20,
attributes_to_crop: None,
attributes_to_retrieve: None,
attributes_to_highlight: None,
filters: None,
matches: false,
facet_filters: None,
facets: None,
}
}
}
pub struct SearchBuilder<'a> {
index: &'a Index,
query: Option<String>,
offset: usize,
limit: usize,
attributes_to_crop: Option<HashMap<String, usize>>,
attributes_to_retrieve: Option<HashSet<String>>,
attributes_to_highlight: Option<HashSet<String>>,
filters: Option<String>,
matches: bool,
facet_filters: Option<FacetFilter>,
facets: Option<Vec<(FieldId, String)>>
}
impl<'a> SearchBuilder<'a> {
pub fn offset(&mut self, value: usize) -> &SearchBuilder {
self.offset = value;
self
}
pub fn limit(&mut self, value: usize) -> &SearchBuilder {
self.limit = value;
self
}
pub fn attributes_to_crop(&mut self, value: HashMap<String, usize>) -> &SearchBuilder {
self.attributes_to_crop = Some(value);
self
}
pub fn attributes_to_retrieve(&mut self, value: HashSet<String>) -> &SearchBuilder {
self.attributes_to_retrieve = Some(value);
self
}
pub fn add_retrievable_field(&mut self, value: String) -> &SearchBuilder {
let attributes_to_retrieve = self.attributes_to_retrieve.get_or_insert(HashSet::new());
attributes_to_retrieve.insert(value);
self
}
pub fn attributes_to_highlight(&mut self, value: HashSet<String>) -> &SearchBuilder {
self.attributes_to_highlight = Some(value);
self
}
pub fn add_facet_filters(&mut self, filters: FacetFilter) -> &SearchBuilder {
self.facet_filters = Some(filters);
self
}
pub fn filters(&mut self, value: String) -> &SearchBuilder {
self.filters = Some(value);
self
}
pub fn get_matches(&mut self) -> &SearchBuilder {
self.matches = true;
self
}
pub fn add_facets(&mut self, facets: Vec<(FieldId, String)>) -> &SearchBuilder {
self.facets = Some(facets);
self
}
pub fn search(self, reader: &MainReader) -> Result<SearchResult, ResponseError> {
let schema = self
.index
.main
.schema(reader)?
.ok_or(Error::internal("missing schema"))?;
let ranked_map = self.index.main.ranked_map(reader)?.unwrap_or_default();
// Change criteria
let mut query_builder = match self.get_criteria(reader, &ranked_map, &schema)? {
Some(criteria) => self.index.query_builder_with_criteria(criteria),
None => self.index.query_builder(),
};
if let Some(filter_expression) = &self.filters {
let filter = Filter::parse(filter_expression, &schema)?;
let index = &self.index;
query_builder.with_filter(move |id| {
let reader = &reader;
let filter = &filter;
match filter.test(reader, index, id) {
Ok(res) => res,
Err(e) => {
log::warn!("unexpected error during filtering: {}", e);
false
}
}
});
}
if let Some(field) = self.index.main.distinct_attribute(reader)? {
let index = &self.index;
query_builder.with_distinct(1, move |id| {
match index.document_attribute_bytes(reader, id, field) {
Ok(Some(bytes)) => {
let mut s = SipHasher::new();
bytes.hash(&mut s);
Some(s.finish())
}
_ => None,
}
});
}
query_builder.set_facet_filter(self.facet_filters);
query_builder.set_facets(self.facets);
let start = Instant::now();
let result = query_builder.query(reader, self.query.as_deref(), self.offset..(self.offset + self.limit));
let search_result = result.map_err(Error::search_documents)?;
let time_ms = start.elapsed().as_millis() as usize;
let mut all_attributes: HashSet<&str> = HashSet::new();
let mut all_formatted: HashSet<&str> = HashSet::new();
match &self.attributes_to_retrieve {
Some(to_retrieve) => {
all_attributes.extend(to_retrieve.iter().map(String::as_str));
if let Some(to_highlight) = &self.attributes_to_highlight {
all_formatted.extend(to_highlight.iter().map(String::as_str));
}
if let Some(to_crop) = &self.attributes_to_crop {
all_formatted.extend(to_crop.keys().map(String::as_str));
}
all_attributes.extend(&all_formatted);
},
None => {
all_attributes.extend(schema.displayed_name());
// If we specified at least one attribute to highlight or crop then
// all available attributes will be returned in the _formatted field.
if self.attributes_to_highlight.is_some() || self.attributes_to_crop.is_some() {
all_formatted.extend(all_attributes.iter().cloned());
}
},
}
let mut hits = Vec::with_capacity(self.limit);
for doc in search_result.documents {
let mut document: IndexMap<String, Value> = self
.index
.document(reader, Some(&all_attributes), doc.id)
.map_err(|e| Error::retrieve_document(doc.id.0, e))?
.ok_or(Error::internal(
"Impossible to retrieve the document; Corrupted data",
))?;
let mut formatted = document.iter()
.filter(|(key, _)| all_formatted.contains(key.as_str()))
.map(|(k, v)| (k.clone(), v.clone()))
.collect();
let mut matches = doc.highlights.clone();
// Crops fields if needed
if let Some(fields) = &self.attributes_to_crop {
crop_document(&mut formatted, &mut matches, &schema, fields);
}
// Transform to readable matches
if let Some(attributes_to_highlight) = &self.attributes_to_highlight {
let matches = calculate_matches(
&matches,
self.attributes_to_highlight.clone(),
&schema,
);
formatted = calculate_highlights(&formatted, &matches, attributes_to_highlight);
}
let matches_info = if self.matches {
Some(calculate_matches(&matches, self.attributes_to_retrieve.clone(), &schema))
} else {
None
};
if let Some(attributes_to_retrieve) = &self.attributes_to_retrieve {
document.retain(|key, _| attributes_to_retrieve.contains(&key.to_string()))
}
let hit = SearchHit {
document,
formatted,
matches_info,
};
hits.push(hit);
}
let results = SearchResult {
hits,
offset: self.offset,
limit: self.limit,
nb_hits: search_result.nb_hits,
exhaustive_nb_hits: search_result.exhaustive_nb_hit,
processing_time_ms: time_ms,
query: self.query.unwrap_or_default(),
facets_distribution: search_result.facets,
exhaustive_facets_count: search_result.exhaustive_facets_count,
};
Ok(results)
}
pub fn get_criteria(
&self,
reader: &MainReader,
ranked_map: &'a RankedMap,
schema: &Schema,
) -> Result<Option<Criteria<'a>>, ResponseError> {
let ranking_rules = self.index.main.ranking_rules(reader)?;
if let Some(ranking_rules) = ranking_rules {
let mut builder = CriteriaBuilder::with_capacity(7 + ranking_rules.len());
for rule in ranking_rules {
match rule {
RankingRule::Typo => builder.push(Typo),
RankingRule::Words => builder.push(Words),
RankingRule::Proximity => builder.push(Proximity),
RankingRule::Attribute => builder.push(Attribute),
RankingRule::WordsPosition => builder.push(WordsPosition),
RankingRule::Exactness => builder.push(Exactness),
RankingRule::Asc(field) => {
match SortByAttr::lower_is_better(&ranked_map, &schema, &field) {
Ok(rule) => builder.push(rule),
Err(err) => error!("Error during criteria builder; {:?}", err),
}
}
RankingRule::Desc(field) => {
match SortByAttr::higher_is_better(&ranked_map, &schema, &field) {
Ok(rule) => builder.push(rule),
Err(err) => error!("Error during criteria builder; {:?}", err),
}
}
}
}
builder.push(DocumentId);
return Ok(Some(builder.build()));
}
Ok(None)
}
}
#[derive(Debug, Clone, Eq, PartialEq, Serialize, Deserialize)]
pub struct MatchPosition {
pub start: usize,
pub length: usize,
}
impl PartialOrd for MatchPosition {
fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
Some(self.cmp(other))
}
}
impl Ord for MatchPosition {
fn cmp(&self, other: &Self) -> Ordering {
match self.start.cmp(&other.start) {
Ordering::Equal => self.length.cmp(&other.length),
_ => self.start.cmp(&other.start),
}
}
}
pub type HighlightInfos = HashMap<String, Value>;
pub type MatchesInfos = HashMap<String, Vec<MatchPosition>>;
// pub type RankingInfos = HashMap<String, u64>;
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SearchHit {
#[serde(flatten)]
pub document: IndexMap<String, Value>,
#[serde(rename = "_formatted", skip_serializing_if = "IndexMap::is_empty")]
pub formatted: IndexMap<String, Value>,
#[serde(rename = "_matchesInfo", skip_serializing_if = "Option::is_none")]
pub matches_info: Option<MatchesInfos>,
}
#[derive(Debug, Clone, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct SearchResult {
pub hits: Vec<SearchHit>,
pub offset: usize,
pub limit: usize,
pub nb_hits: usize,
pub exhaustive_nb_hits: bool,
pub processing_time_ms: usize,
pub query: String,
#[serde(skip_serializing_if = "Option::is_none")]
pub facets_distribution: Option<HashMap<String, HashMap<String, usize>>>,
#[serde(skip_serializing_if = "Option::is_none")]
pub exhaustive_facets_count: Option<bool>,
}
/// returns the start index and the length on the crop.
fn aligned_crop(text: &str, match_index: usize, context: usize) -> (usize, usize) {
let is_word_component = |c: &char| c.is_alphanumeric() && !is_cjk(*c);
let word_end_index = |mut index| {
if text.chars().nth(index - 1).map_or(false, |c| is_word_component(&c)) {
index += text.chars().skip(index).take_while(is_word_component).count();
}
index
};
if context == 0 {
// count need to be at least 1 for cjk queries to return something
return (match_index, 1 + text.chars().skip(match_index).take_while(is_word_component).count());
}
let start = match match_index.saturating_sub(context) {
0 => 0,
n => {
let word_end_index = word_end_index(n);
// skip whitespaces if any
word_end_index + text.chars().skip(word_end_index).take_while(char::is_ascii_whitespace).count()
}
};
let end = word_end_index(match_index + context);
(start, end - start)
}
fn crop_text(
text: &str,
matches: impl IntoIterator<Item = Highlight>,
context: usize,
) -> (String, Vec<Highlight>) {
let mut matches = matches.into_iter().peekable();
let char_index = matches.peek().map(|m| m.char_index as usize).unwrap_or(0);
let (start, count) = aligned_crop(text, char_index, context);
// TODO do something about double allocation
let text = text
.chars()
.skip(start)
.take(count)
.collect::<String>()
.trim()
.to_string();
// update matches index to match the new cropped text
let matches = matches
.take_while(|m| (m.char_index as usize) + (m.char_length as usize) <= start + count)
.map(|m| Highlight {
char_index: m.char_index - start as u16,
..m
})
.collect();
(text, matches)
}
fn crop_document(
document: &mut IndexMap<String, Value>,
matches: &mut Vec<Highlight>,
schema: &Schema,
fields: &HashMap<String, usize>,
) {
matches.sort_unstable_by_key(|m| (m.char_index, m.char_length));
for (field, length) in fields {
let attribute = match schema.id(field) {
Some(attribute) => attribute,
None => continue,
};
let selected_matches = matches
.iter()
.filter(|m| FieldId::new(m.attribute) == attribute)
.cloned();
if let Some(Value::String(ref mut original_text)) = document.get_mut(field) {
let (cropped_text, cropped_matches) =
crop_text(original_text, selected_matches, *length);
*original_text = cropped_text;
matches.retain(|m| FieldId::new(m.attribute) != attribute);
matches.extend_from_slice(&cropped_matches);
}
}
}
fn calculate_matches(
matches: &[Highlight],
attributes_to_retrieve: Option<HashSet<String>>,
schema: &Schema,
) -> MatchesInfos {
let mut matches_result: HashMap<String, Vec<MatchPosition>> = HashMap::new();
for m in matches.iter() {
if let Some(attribute) = schema.name(FieldId::new(m.attribute)) {
if let Some(ref attributes_to_retrieve) = attributes_to_retrieve {
if !attributes_to_retrieve.contains(attribute) {
continue;
}
}
if !schema.displayed_name().contains(attribute) {
continue;
}
if let Some(pos) = matches_result.get_mut(attribute) {
pos.push(MatchPosition {
start: m.char_index as usize,
length: m.char_length as usize,
});
} else {
let mut positions = Vec::new();
positions.push(MatchPosition {
start: m.char_index as usize,
length: m.char_length as usize,
});
matches_result.insert(attribute.to_string(), positions);
}
}
}
for (_, val) in matches_result.iter_mut() {
val.sort_unstable();
val.dedup();
}
matches_result
}
fn calculate_highlights(
document: &IndexMap<String, Value>,
matches: &MatchesInfos,
attributes_to_highlight: &HashSet<String>,
) -> IndexMap<String, Value> {
let mut highlight_result = document.clone();
for (attribute, matches) in matches.iter() {
if attributes_to_highlight.contains(attribute) {
if let Some(Value::String(value)) = document.get(attribute) {
let value: Vec<_> = value.chars().collect();
let mut highlighted_value = String::new();
let mut index = 0;
let longest_matches = matches
.linear_group_by_key(|m| m.start)
.map(|group| group.last().unwrap())
.filter(move |m| m.start >= index);
for m in longest_matches {
let before = value.get(index..m.start);
let highlighted = value.get(m.start..(m.start + m.length));
if let (Some(before), Some(highlighted)) = (before, highlighted) {
highlighted_value.extend(before);
highlighted_value.push_str("<em>");
highlighted_value.extend(highlighted);
highlighted_value.push_str("</em>");
index = m.start + m.length;
} else {
error!("value: {:?}; index: {:?}, match: {:?}", value, index, m);
}
}
highlighted_value.extend(value[index..].iter());
highlight_result.insert(attribute.to_string(), Value::String(highlighted_value));
};
}
}
highlight_result
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn aligned_crops() {
let text = r#"En ce début de trentième millénaire, l'Empire n'a jamais été aussi puissant, aussi étendu à travers toute la galaxie. C'est dans sa capitale, Trantor, que l'éminent savant Hari Seldon invente la psychohistoire, une science toute nouvelle, à base de psychologie et de mathématiques, qui lui permet de prédire l'avenir... C'est-à-dire l'effondrement de l'Empire d'ici cinq siècles et au-delà, trente mille années de chaos et de ténèbres. Pour empêcher cette catastrophe et sauver la civilisation, Seldon crée la Fondation."#;
// simple test
let (start, length) = aligned_crop(&text, 6, 2);
let cropped = text.chars().skip(start).take(length).collect::<String>().trim().to_string();
assert_eq!("début", cropped);
// first word test
let (start, length) = aligned_crop(&text, 0, 1);
let cropped = text.chars().skip(start).take(length).collect::<String>().trim().to_string();
assert_eq!("En", cropped);
// last word test
let (start, length) = aligned_crop(&text, 510, 2);
let cropped = text.chars().skip(start).take(length).collect::<String>().trim().to_string();
assert_eq!("Fondation", cropped);
// CJK tests
let text = "this isのス foo myタイリ test";
// mixed charset
let (start, length) = aligned_crop(&text, 5, 3);
let cropped = text.chars().skip(start).take(length).collect::<String>().trim().to_string();
assert_eq!("isの", cropped);
// split regular word / CJK word, no space
let (start, length) = aligned_crop(&text, 7, 1);
let cropped = text.chars().skip(start).take(length).collect::<String>().trim().to_string();
assert_eq!("", cropped);
}
#[test]
fn calculate_matches() {
let mut matches = Vec::new();
matches.push(Highlight { attribute: 0, char_index: 0, char_length: 3});
matches.push(Highlight { attribute: 0, char_index: 0, char_length: 2});
let mut attributes_to_retrieve: HashSet<String> = HashSet::new();
attributes_to_retrieve.insert("title".to_string());
let schema = Schema::with_primary_key("title");
let matches_result = super::calculate_matches(&matches, Some(attributes_to_retrieve), &schema);
let mut matches_result_expected: HashMap<String, Vec<MatchPosition>> = HashMap::new();
let mut positions = Vec::new();
positions.push(MatchPosition {
start: 0,
length: 2,
});
positions.push(MatchPosition {
start: 0,
length: 3,
});
matches_result_expected.insert("title".to_string(), positions);
assert_eq!(matches_result, matches_result_expected);
}
#[test]
fn calculate_highlights() {
let data = r#"{
"title": "Fondation (Isaac ASIMOV)",
"description": "En ce début de trentième millénaire, l'Empire n'a jamais été aussi puissant, aussi étendu à travers toute la galaxie. C'est dans sa capitale, Trantor, que l'éminent savant Hari Seldon invente la psychohistoire, une science toute nouvelle, à base de psychologie et de mathématiques, qui lui permet de prédire l'avenir... C'est-à-dire l'effondrement de l'Empire d'ici cinq siècles et au-delà, trente mille années de chaos et de ténèbres. Pour empêcher cette catastrophe et sauver la civilisation, Seldon crée la Fondation."
}"#;
let document: IndexMap<String, Value> = serde_json::from_str(data).unwrap();
let mut attributes_to_highlight = HashSet::new();
attributes_to_highlight.insert("title".to_string());
attributes_to_highlight.insert("description".to_string());
let mut matches = HashMap::new();
let mut m = Vec::new();
m.push(MatchPosition {
start: 0,
length: 9,
});
matches.insert("title".to_string(), m);
let mut m = Vec::new();
m.push(MatchPosition {
start: 510,
length: 9,
});
matches.insert("description".to_string(), m);
let result = super::calculate_highlights(&document, &matches, &attributes_to_highlight);
let mut result_expected = IndexMap::new();
result_expected.insert(
"title".to_string(),
Value::String("<em>Fondation</em> (Isaac ASIMOV)".to_string()),
);
result_expected.insert("description".to_string(), Value::String("En ce début de trentième millénaire, l'Empire n'a jamais été aussi puissant, aussi étendu à travers toute la galaxie. C'est dans sa capitale, Trantor, que l'éminent savant Hari Seldon invente la psychohistoire, une science toute nouvelle, à base de psychologie et de mathématiques, qui lui permet de prédire l'avenir... C'est-à-dire l'effondrement de l'Empire d'ici cinq siècles et au-delà, trente mille années de chaos et de ténèbres. Pour empêcher cette catastrophe et sauver la civilisation, Seldon crée la <em>Fondation</em>.".to_string()));
assert_eq!(result, result_expected);
}
#[test]
fn highlight_longest_match() {
let data = r#"{
"title": "Ice"
}"#;
let document: IndexMap<String, Value> = serde_json::from_str(data).unwrap();
let mut attributes_to_highlight = HashSet::new();
attributes_to_highlight.insert("title".to_string());
let mut matches = HashMap::new();
let mut m = Vec::new();
m.push(MatchPosition {
start: 0,
length: 2,
});
m.push(MatchPosition {
start: 0,
length: 3,
});
matches.insert("title".to_string(), m);
let result = super::calculate_highlights(&document, &matches, &attributes_to_highlight);
let mut result_expected = IndexMap::new();
result_expected.insert(
"title".to_string(),
Value::String("<em>Ice</em>".to_string()),
);
assert_eq!(result, result_expected);
}
}

7
src/helpers/mod.rs Normal file
View File

@ -0,0 +1,7 @@
pub mod authentication;
pub mod meilisearch;
pub mod normalize_path;
pub mod compression;
pub use authentication::Authentication;
pub use normalize_path::NormalizePath;

View File

@ -0,0 +1,86 @@
/// From https://docs.rs/actix-web/3.0.0-alpha.2/src/actix_web/middleware/normalize.rs.html#34
use actix_http::Error;
use actix_service::{Service, Transform};
use actix_web::{
dev::ServiceRequest,
dev::ServiceResponse,
http::uri::{PathAndQuery, Uri},
};
use futures::future::{ok, Ready};
use regex::Regex;
use std::task::{Context, Poll};
pub struct NormalizePath;
impl<S, B> Transform<S> for NormalizePath
where
S: Service<Request = ServiceRequest, Response = ServiceResponse<B>, Error = Error>,
S::Future: 'static,
{
type Request = ServiceRequest;
type Response = ServiceResponse<B>;
type Error = Error;
type InitError = ();
type Transform = NormalizePathNormalization<S>;
type Future = Ready<Result<Self::Transform, Self::InitError>>;
fn new_transform(&self, service: S) -> Self::Future {
ok(NormalizePathNormalization {
service,
merge_slash: Regex::new("//+").unwrap(),
})
}
}
pub struct NormalizePathNormalization<S> {
service: S,
merge_slash: Regex,
}
impl<S, B> Service for NormalizePathNormalization<S>
where
S: Service<Request = ServiceRequest, Response = ServiceResponse<B>, Error = Error>,
S::Future: 'static,
{
type Request = ServiceRequest;
type Response = ServiceResponse<B>;
type Error = Error;
type Future = S::Future;
fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
self.service.poll_ready(cx)
}
fn call(&mut self, mut req: ServiceRequest) -> Self::Future {
let head = req.head_mut();
// always add trailing slash, might be an extra one
let path = head.uri.path().to_string() + "/";
if self.merge_slash.find(&path).is_some() {
// normalize multiple /'s to one /
let path = self.merge_slash.replace_all(&path, "/");
let path = if path.len() > 1 {
path.trim_end_matches('/')
} else {
&path
};
let mut parts = head.uri.clone().into_parts();
let pq = parts.path_and_query.as_ref().unwrap();
let path = if let Some(q) = pq.query() {
bytes::Bytes::from(format!("{}?{}", path, q))
} else {
bytes::Bytes::copy_from_slice(path.as_bytes())
};
parts.path_and_query = Some(PathAndQuery::from_maybe_shared(path).unwrap());
let uri = Uri::from_parts(parts).unwrap();
req.match_info_mut().get_mut().update(&uri);
req.head_mut().uri = uri;
}
self.service.call(req)
}
}

104
src/lib.rs Normal file
View File

@ -0,0 +1,104 @@
#![allow(clippy::or_fun_call)]
pub mod data;
pub mod error;
pub mod helpers;
pub mod models;
pub mod option;
pub mod routes;
pub mod analytics;
pub mod snapshot;
pub mod dump;
use actix_http::Error;
use actix_service::ServiceFactory;
use actix_web::{dev, web, App};
use chrono::Utc;
use log::error;
use meilisearch_core::{Index, MainWriter, ProcessedUpdateResult};
pub use option::Opt;
pub use self::data::Data;
use self::error::{payload_error_handler, ResponseError};
pub fn create_app(
data: &Data,
enable_frontend: bool,
) -> App<
impl ServiceFactory<
Config = (),
Request = dev::ServiceRequest,
Response = dev::ServiceResponse<actix_http::body::Body>,
Error = Error,
InitError = (),
>,
actix_http::body::Body,
> {
let app = App::new()
.data(data.clone())
.app_data(
web::JsonConfig::default()
.limit(data.http_payload_size_limit)
.content_type(|_mime| true) // Accept all mime types
.error_handler(|err, _req| payload_error_handler(err).into()),
)
.app_data(
web::QueryConfig::default()
.error_handler(|err, _req| payload_error_handler(err).into())
)
.configure(routes::document::services)
.configure(routes::index::services)
.configure(routes::search::services)
.configure(routes::setting::services)
.configure(routes::stop_words::services)
.configure(routes::synonym::services)
.configure(routes::health::services)
.configure(routes::stats::services)
.configure(routes::key::services)
.configure(routes::dump::services);
if enable_frontend {
app
.service(routes::load_html)
.service(routes::load_css)
} else {
app
}
}
pub fn index_update_callback_txn(index: Index, index_uid: &str, data: &Data, mut writer: &mut MainWriter) -> Result<(), String> {
if let Err(e) = data.db.compute_stats(&mut writer, index_uid) {
return Err(format!("Impossible to compute stats; {}", e));
}
if let Err(e) = data.db.set_last_update(&mut writer, &Utc::now()) {
return Err(format!("Impossible to update last_update; {}", e));
}
if let Err(e) = index.main.put_updated_at(&mut writer) {
return Err(format!("Impossible to update updated_at; {}", e));
}
Ok(())
}
pub fn index_update_callback(index_uid: &str, data: &Data, status: ProcessedUpdateResult) {
if status.error.is_some() {
return;
}
if let Some(index) = data.db.open_index(index_uid) {
let db = &data.db;
let res = db.main_write::<_, _, ResponseError>(|mut writer| {
if let Err(e) = index_update_callback_txn(index, index_uid, data, &mut writer) {
error!("{}", e);
}
Ok(())
});
match res {
Ok(_) => (),
Err(e) => error!("{}", e),
}
}
}

169
src/main.rs Normal file
View File

@ -0,0 +1,169 @@
use std::{env, thread};
use actix_cors::Cors;
use actix_web::{middleware, HttpServer};
use main_error::MainError;
use meilisearch_http::helpers::NormalizePath;
use meilisearch_http::{create_app, index_update_callback, Data, Opt};
use structopt::StructOpt;
use meilisearch_http::{snapshot, dump};
mod analytics;
#[cfg(target_os = "linux")]
#[global_allocator]
static ALLOC: jemallocator::Jemalloc = jemallocator::Jemalloc;
#[actix_web::main]
async fn main() -> Result<(), MainError> {
let opt = Opt::from_args();
#[cfg(all(not(debug_assertions), feature = "sentry"))]
let _sentry = sentry::init((
if !opt.no_sentry {
Some(opt.sentry_dsn.clone())
} else {
None
},
sentry::ClientOptions {
release: sentry::release_name!(),
..Default::default()
},
));
match opt.env.as_ref() {
"production" => {
if opt.master_key.is_none() {
return Err(
"In production mode, the environment variable MEILI_MASTER_KEY is mandatory"
.into(),
);
}
#[cfg(all(not(debug_assertions), feature = "sentry"))]
if !opt.no_sentry && _sentry.is_enabled() {
sentry::integrations::panic::register_panic_handler(); // TODO: This shouldn't be needed when upgrading to sentry 0.19.0. These integrations are turned on by default when using `sentry::init`.
sentry::integrations::env_logger::init(None, Default::default());
}
}
"development" => {
env_logger::Builder::from_env(env_logger::Env::default().default_filter_or("info")).init();
}
_ => unreachable!(),
}
if let Some(path) = &opt.import_snapshot {
snapshot::load_snapshot(&opt.db_path, path, opt.ignore_snapshot_if_db_exists, opt.ignore_missing_snapshot)?;
}
let data = Data::new(opt.clone())?;
if !opt.no_analytics {
let analytics_data = data.clone();
let analytics_opt = opt.clone();
thread::spawn(move || analytics::analytics_sender(analytics_data, analytics_opt));
}
let data_cloned = data.clone();
data.db.set_update_callback(Box::new(move |name, status| {
index_update_callback(name, &data_cloned, status);
}));
if let Some(path) = &opt.import_dump {
dump::import_dump(&data, path, opt.dump_batch_size)?;
}
if opt.schedule_snapshot {
snapshot::schedule_snapshot(data.clone(), &opt.snapshot_dir, opt.snapshot_interval_sec.unwrap_or(86400))?;
}
print_launch_resume(&opt, &data);
let enable_frontend = opt.env != "production";
let http_server = HttpServer::new(move || {
create_app(&data, enable_frontend)
.wrap(
Cors::default()
.send_wildcard()
.allowed_headers(vec!["content-type", "x-meili-api-key"])
.max_age(86_400) // 24h
)
.wrap(middleware::Logger::default())
.wrap(middleware::Compress::default())
.wrap(NormalizePath)
});
if let Some(config) = opt.get_ssl_config()? {
http_server
.bind_rustls(opt.http_addr, config)?
.run()
.await?;
} else {
http_server.bind(opt.http_addr)?.run().await?;
}
Ok(())
}
pub fn print_launch_resume(opt: &Opt, data: &Data) {
let ascii_name = r#"
888b d888 d8b 888 d8b .d8888b. 888
8888b d8888 Y8P 888 Y8P d88P Y88b 888
88888b.d88888 888 Y88b. 888
888Y88888P888 .d88b. 888 888 888 "Y888b. .d88b. 8888b. 888d888 .d8888b 88888b.
888 Y888P 888 d8P Y8b 888 888 888 "Y88b. d8P Y8b "88b 888P" d88P" 888 "88b
888 Y8P 888 88888888 888 888 888 "888 88888888 .d888888 888 888 888 888
888 " 888 Y8b. 888 888 888 Y88b d88P Y8b. 888 888 888 Y88b. 888 888
888 888 "Y8888 888 888 888 "Y8888P" "Y8888 "Y888888 888 "Y8888P 888 888
"#;
eprintln!("{}", ascii_name);
eprintln!("Database path:\t\t{:?}", opt.db_path);
eprintln!("Server listening on:\t{:?}", opt.http_addr);
eprintln!("Environment:\t\t{:?}", opt.env);
eprintln!("Commit SHA:\t\t{:?}", env!("VERGEN_SHA").to_string());
eprintln!(
"Build date:\t\t{:?}",
env!("VERGEN_BUILD_TIMESTAMP").to_string()
);
eprintln!(
"Package version:\t{:?}",
env!("CARGO_PKG_VERSION").to_string()
);
#[cfg(all(not(debug_assertions), feature = "sentry"))]
eprintln!(
"Sentry DSN:\t\t{:?}",
if !opt.no_sentry {
&opt.sentry_dsn
} else {
"Disabled"
}
);
eprintln!(
"Amplitude Analytics:\t{:?}",
if !opt.no_analytics {
"Enabled"
} else {
"Disabled"
}
);
eprintln!();
if data.api_keys.master.is_some() {
eprintln!("A Master Key has been set. Requests to MeiliSearch won't be authorized unless you provide an authentication key.");
} else {
eprintln!("No master key found; The server will accept unidentified requests. \
If you need some protection in development mode, please export a key: export MEILI_MASTER_KEY=xxx");
}
eprintln!();
eprintln!("Documentation:\t\thttps://docs.meilisearch.com");
eprintln!("Source code:\t\thttps://github.com/meilisearch/meilisearch");
eprintln!("Contact:\t\thttps://docs.meilisearch.com/resources/contact.html or bonjour@meilisearch.com");
eprintln!();
}

1
src/models/mod.rs Normal file
View File

@ -0,0 +1 @@
pub mod update_operation;

View File

@ -0,0 +1,33 @@
use std::fmt;
#[allow(dead_code)]
#[derive(Debug)]
pub enum UpdateOperation {
ClearAllDocuments,
DocumentsAddition,
DocumentsDeletion,
SynonymsUpdate,
SynonymsDeletion,
StopWordsAddition,
StopWordsDeletion,
Schema,
Config,
}
impl fmt::Display for UpdateOperation {
fn fmt(&self, f: &mut fmt::Formatter) -> std::fmt::Result {
use UpdateOperation::*;
match self {
ClearAllDocuments => write!(f, "ClearAllDocuments"),
DocumentsAddition => write!(f, "DocumentsAddition"),
DocumentsDeletion => write!(f, "DocumentsDeletion"),
SynonymsUpdate => write!(f, "SynonymsUpdate"),
SynonymsDeletion => write!(f, "SynonymsDelettion"),
StopWordsAddition => write!(f, "StopWordsAddition"),
StopWordsDeletion => write!(f, "StopWordsDeletion"),
Schema => write!(f, "Schema"),
Config => write!(f, "Config"),
}
}
}

221
src/option.rs Normal file
View File

@ -0,0 +1,221 @@
use std::{error, fs};
use std::io::{BufReader, Read};
use std::path::PathBuf;
use std::sync::Arc;
use rustls::internal::pemfile::{certs, pkcs8_private_keys, rsa_private_keys};
use rustls::{
AllowAnyAnonymousOrAuthenticatedClient, AllowAnyAuthenticatedClient, NoClientAuth,
RootCertStore,
};
use structopt::StructOpt;
const POSSIBLE_ENV: [&str; 2] = ["development", "production"];
#[derive(Debug, Default, Clone, StructOpt)]
pub struct Opt {
/// The destination where the database must be created.
#[structopt(long, env = "MEILI_DB_PATH", default_value = "./data.ms")]
pub db_path: String,
/// The address on which the http server will listen.
#[structopt(long, env = "MEILI_HTTP_ADDR", default_value = "127.0.0.1:7700")]
pub http_addr: String,
/// The master key allowing you to do everything on the server.
#[structopt(long, env = "MEILI_MASTER_KEY")]
pub master_key: Option<String>,
/// The Sentry DSN to use for error reporting. This defaults to the MeiliSearch Sentry project.
/// You can disable sentry all together using the `--no-sentry` flag or `MEILI_NO_SENTRY` environment variable.
#[cfg(all(not(debug_assertions), feature = "sentry"))]
#[structopt(long, env = "SENTRY_DSN", default_value = "https://5ddfa22b95f241198be2271aaf028653@sentry.io/3060337")]
pub sentry_dsn: String,
/// Disable Sentry error reporting.
#[cfg(all(not(debug_assertions), feature = "sentry"))]
#[structopt(long, env = "MEILI_NO_SENTRY")]
pub no_sentry: bool,
/// This environment variable must be set to `production` if you are running in production.
/// If the server is running in development mode more logs will be displayed,
/// and the master key can be avoided which implies that there is no security on the updates routes.
/// This is useful to debug when integrating the engine with another service.
#[structopt(long, env = "MEILI_ENV", default_value = "development", possible_values = &POSSIBLE_ENV)]
pub env: String,
/// Do not send analytics to Meili.
#[structopt(long, env = "MEILI_NO_ANALYTICS")]
pub no_analytics: bool,
/// The maximum size, in bytes, of the main lmdb database directory
#[structopt(long, env = "MEILI_MAX_MDB_SIZE", default_value = "107374182400")] // 100GB
pub max_mdb_size: usize,
/// The maximum size, in bytes, of the update lmdb database directory
#[structopt(long, env = "MEILI_MAX_UDB_SIZE", default_value = "107374182400")] // 100GB
pub max_udb_size: usize,
/// The maximum size, in bytes, of accepted JSON payloads
#[structopt(long, env = "MEILI_HTTP_PAYLOAD_SIZE_LIMIT", default_value = "10485760")] // 10MB
pub http_payload_size_limit: usize,
/// Read server certificates from CERTFILE.
/// This should contain PEM-format certificates
/// in the right order (the first certificate should
/// certify KEYFILE, the last should be a root CA).
#[structopt(long, env = "MEILI_SSL_CERT_PATH", parse(from_os_str))]
pub ssl_cert_path: Option<PathBuf>,
/// Read private key from KEYFILE. This should be a RSA
/// private key or PKCS8-encoded private key, in PEM format.
#[structopt(long, env = "MEILI_SSL_KEY_PATH", parse(from_os_str))]
pub ssl_key_path: Option<PathBuf>,
/// Enable client authentication, and accept certificates
/// signed by those roots provided in CERTFILE.
#[structopt(long, env = "MEILI_SSL_AUTH_PATH", parse(from_os_str))]
pub ssl_auth_path: Option<PathBuf>,
/// Read DER-encoded OCSP response from OCSPFILE and staple to certificate.
/// Optional
#[structopt(long, env = "MEILI_SSL_OCSP_PATH", parse(from_os_str))]
pub ssl_ocsp_path: Option<PathBuf>,
/// Send a fatal alert if the client does not complete client authentication.
#[structopt(long, env = "MEILI_SSL_REQUIRE_AUTH")]
pub ssl_require_auth: bool,
/// SSL support session resumption
#[structopt(long, env = "MEILI_SSL_RESUMPTION")]
pub ssl_resumption: bool,
/// SSL support tickets.
#[structopt(long, env = "MEILI_SSL_TICKETS")]
pub ssl_tickets: bool,
/// Defines the path of the snapshot file to import.
/// This option will, by default, stop the process if a database already exist or if no snapshot exists at
/// the given path. If this option is not specified no snapshot is imported.
#[structopt(long)]
pub import_snapshot: Option<PathBuf>,
/// The engine will ignore a missing snapshot and not return an error in such case.
#[structopt(long, requires = "import-snapshot")]
pub ignore_missing_snapshot: bool,
/// The engine will skip snapshot importation and not return an error in such case.
#[structopt(long, requires = "import-snapshot")]
pub ignore_snapshot_if_db_exists: bool,
/// Defines the directory path where meilisearch will create snapshot each snapshot_time_gap.
#[structopt(long, env = "MEILI_SNAPSHOT_DIR", default_value = "snapshots/")]
pub snapshot_dir: PathBuf,
/// Activate snapshot scheduling.
#[structopt(long, env = "MEILI_SCHEDULE_SNAPSHOT")]
pub schedule_snapshot: bool,
/// Defines time interval, in seconds, between each snapshot creation.
#[structopt(long, env = "MEILI_SNAPSHOT_INTERVAL_SEC")]
pub snapshot_interval_sec: Option<u64>,
/// Folder where dumps are created when the dump route is called.
#[structopt(long, env = "MEILI_DUMPS_DIR", default_value = "dumps/")]
pub dumps_dir: PathBuf,
/// Import a dump from the specified path, must be a `.tar.gz` file.
#[structopt(long, conflicts_with = "import-snapshot")]
pub import_dump: Option<PathBuf>,
/// The batch size used in the importation process, the bigger it is the faster the dump is created.
#[structopt(long, env = "MEILI_DUMP_BATCH_SIZE", default_value = "1024")]
pub dump_batch_size: usize,
}
impl Opt {
pub fn get_ssl_config(&self) -> Result<Option<rustls::ServerConfig>, Box<dyn error::Error>> {
if let (Some(cert_path), Some(key_path)) = (&self.ssl_cert_path, &self.ssl_key_path) {
let client_auth = match &self.ssl_auth_path {
Some(auth_path) => {
let roots = load_certs(auth_path.to_path_buf())?;
let mut client_auth_roots = RootCertStore::empty();
for root in roots {
client_auth_roots.add(&root).unwrap();
}
if self.ssl_require_auth {
AllowAnyAuthenticatedClient::new(client_auth_roots)
} else {
AllowAnyAnonymousOrAuthenticatedClient::new(client_auth_roots)
}
}
None => NoClientAuth::new(),
};
let mut config = rustls::ServerConfig::new(client_auth);
config.key_log = Arc::new(rustls::KeyLogFile::new());
let certs = load_certs(cert_path.to_path_buf())?;
let privkey = load_private_key(key_path.to_path_buf())?;
let ocsp = load_ocsp(&self.ssl_ocsp_path)?;
config
.set_single_cert_with_ocsp_and_sct(certs, privkey, ocsp, vec![])
.map_err(|_| "bad certificates/private key")?;
if self.ssl_resumption {
config.set_persistence(rustls::ServerSessionMemoryCache::new(256));
}
if self.ssl_tickets {
config.ticketer = rustls::Ticketer::new();
}
Ok(Some(config))
} else {
Ok(None)
}
}
}
fn load_certs(filename: PathBuf) -> Result<Vec<rustls::Certificate>, Box<dyn error::Error>> {
let certfile = fs::File::open(filename).map_err(|_| "cannot open certificate file")?;
let mut reader = BufReader::new(certfile);
Ok(certs(&mut reader).map_err(|_| "cannot read certificate file")?)
}
fn load_private_key(filename: PathBuf) -> Result<rustls::PrivateKey, Box<dyn error::Error>> {
let rsa_keys = {
let keyfile =
fs::File::open(filename.clone()).map_err(|_| "cannot open private key file")?;
let mut reader = BufReader::new(keyfile);
rsa_private_keys(&mut reader).map_err(|_| "file contains invalid rsa private key")?
};
let pkcs8_keys = {
let keyfile = fs::File::open(filename).map_err(|_| "cannot open private key file")?;
let mut reader = BufReader::new(keyfile);
pkcs8_private_keys(&mut reader)
.map_err(|_| "file contains invalid pkcs8 private key (encrypted keys not supported)")?
};
// prefer to load pkcs8 keys
if !pkcs8_keys.is_empty() {
Ok(pkcs8_keys[0].clone())
} else {
assert!(!rsa_keys.is_empty());
Ok(rsa_keys[0].clone())
}
}
fn load_ocsp(filename: &Option<PathBuf>) -> Result<Vec<u8>, Box<dyn error::Error>> {
let mut ret = Vec::new();
if let Some(ref name) = filename {
fs::File::open(name)
.map_err(|_| "cannot open ocsp file")?
.read_to_end(&mut ret)
.map_err(|_| "cannot read oscp file")?;
}
Ok(ret)
}

266
src/routes/document.rs Normal file
View File

@ -0,0 +1,266 @@
use std::collections::{BTreeSet, HashSet};
use actix_web::{delete, get, post, put};
use actix_web::{web, HttpResponse};
use indexmap::IndexMap;
use meilisearch_core::{update, MainReader};
use serde_json::Value;
use serde::Deserialize;
use crate::Data;
use crate::error::{Error, ResponseError};
use crate::helpers::Authentication;
use crate::routes::{IndexParam, IndexUpdateResponse};
type Document = IndexMap<String, Value>;
#[derive(Deserialize)]
struct DocumentParam {
index_uid: String,
document_id: String,
}
pub fn services(cfg: &mut web::ServiceConfig) {
cfg.service(get_document)
.service(delete_document)
.service(get_all_documents)
.service(add_documents)
.service(update_documents)
.service(delete_documents)
.service(clear_all_documents);
}
#[get(
"/indexes/{index_uid}/documents/{document_id}",
wrap = "Authentication::Public"
)]
async fn get_document(
data: web::Data<Data>,
path: web::Path<DocumentParam>,
) -> Result<HttpResponse, ResponseError> {
let index = data
.db
.open_index(&path.index_uid)
.ok_or(Error::index_not_found(&path.index_uid))?;
let reader = data.db.main_read_txn()?;
let internal_id = index
.main
.external_to_internal_docid(&reader, &path.document_id)?
.ok_or(Error::document_not_found(&path.document_id))?;
let document: Document = index
.document(&reader, None, internal_id)?
.ok_or(Error::document_not_found(&path.document_id))?;
Ok(HttpResponse::Ok().json(document))
}
#[delete(
"/indexes/{index_uid}/documents/{document_id}",
wrap = "Authentication::Private"
)]
async fn delete_document(
data: web::Data<Data>,
path: web::Path<DocumentParam>,
) -> Result<HttpResponse, ResponseError> {
let index = data
.db
.open_index(&path.index_uid)
.ok_or(Error::index_not_found(&path.index_uid))?;
let mut documents_deletion = index.documents_deletion();
documents_deletion.delete_document_by_external_docid(path.document_id.clone());
let update_id = data.db.update_write(|w| documents_deletion.finalize(w))?;
Ok(HttpResponse::Accepted().json(IndexUpdateResponse::with_id(update_id)))
}
#[derive(Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
struct BrowseQuery {
offset: Option<usize>,
limit: Option<usize>,
attributes_to_retrieve: Option<String>,
}
pub fn get_all_documents_sync(
data: &web::Data<Data>,
reader: &MainReader,
index_uid: &str,
offset: usize,
limit: usize,
attributes_to_retrieve: Option<&String>
) -> Result<Vec<Document>, Error> {
let index = data
.db
.open_index(index_uid)
.ok_or(Error::index_not_found(index_uid))?;
let documents_ids: Result<BTreeSet<_>, _> = index
.documents_fields_counts
.documents_ids(reader)?
.skip(offset)
.take(limit)
.collect();
let attributes: Option<HashSet<&str>> = attributes_to_retrieve
.map(|a| a.split(',').collect());
let mut documents = Vec::new();
for document_id in documents_ids? {
if let Ok(Some(document)) =
index.document::<Document>(reader, attributes.as_ref(), document_id)
{
documents.push(document);
}
}
Ok(documents)
}
#[get("/indexes/{index_uid}/documents", wrap = "Authentication::Public")]
async fn get_all_documents(
data: web::Data<Data>,
path: web::Path<IndexParam>,
params: web::Query<BrowseQuery>,
) -> Result<HttpResponse, ResponseError> {
let offset = params.offset.unwrap_or(0);
let limit = params.limit.unwrap_or(20);
let index_uid = &path.index_uid;
let reader = data.db.main_read_txn()?;
let documents = get_all_documents_sync(
&data,
&reader,
index_uid,
offset,
limit,
params.attributes_to_retrieve.as_ref()
)?;
Ok(HttpResponse::Ok().json(documents))
}
fn find_primary_key(document: &IndexMap<String, Value>) -> Option<String> {
for key in document.keys() {
if key.to_lowercase().contains("id") {
return Some(key.to_string());
}
}
None
}
#[derive(Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
struct UpdateDocumentsQuery {
primary_key: Option<String>,
}
async fn update_multiple_documents(
data: web::Data<Data>,
path: web::Path<IndexParam>,
params: web::Query<UpdateDocumentsQuery>,
body: web::Json<Vec<Document>>,
is_partial: bool,
) -> Result<HttpResponse, ResponseError> {
let update_id = data.get_or_create_index(&path.index_uid, |index| {
let reader = data.db.main_read_txn()?;
let mut schema = index
.main
.schema(&reader)?
.ok_or(meilisearch_core::Error::SchemaMissing)?;
if schema.primary_key().is_none() {
let id = match &params.primary_key {
Some(id) => id.to_string(),
None => body
.first()
.and_then(find_primary_key)
.ok_or(meilisearch_core::Error::MissingPrimaryKey)?,
};
schema.set_primary_key(&id).map_err(Error::bad_request)?;
data.db.main_write(|w| index.main.put_schema(w, &schema))?;
}
let mut document_addition = if is_partial {
index.documents_partial_addition()
} else {
index.documents_addition()
};
for document in body.into_inner() {
document_addition.update_document(document);
}
Ok(data.db.update_write(|w| document_addition.finalize(w))?)
})?;
return Ok(HttpResponse::Accepted().json(IndexUpdateResponse::with_id(update_id)));
}
#[post("/indexes/{index_uid}/documents", wrap = "Authentication::Private")]
async fn add_documents(
data: web::Data<Data>,
path: web::Path<IndexParam>,
params: web::Query<UpdateDocumentsQuery>,
body: web::Json<Vec<Document>>,
) -> Result<HttpResponse, ResponseError> {
update_multiple_documents(data, path, params, body, false).await
}
#[put("/indexes/{index_uid}/documents", wrap = "Authentication::Private")]
async fn update_documents(
data: web::Data<Data>,
path: web::Path<IndexParam>,
params: web::Query<UpdateDocumentsQuery>,
body: web::Json<Vec<Document>>,
) -> Result<HttpResponse, ResponseError> {
update_multiple_documents(data, path, params, body, true).await
}
#[post(
"/indexes/{index_uid}/documents/delete-batch",
wrap = "Authentication::Private"
)]
async fn delete_documents(
data: web::Data<Data>,
path: web::Path<IndexParam>,
body: web::Json<Vec<Value>>,
) -> Result<HttpResponse, ResponseError> {
let index = data
.db
.open_index(&path.index_uid)
.ok_or(Error::index_not_found(&path.index_uid))?;
let mut documents_deletion = index.documents_deletion();
for document_id in body.into_inner() {
let document_id = update::value_to_string(&document_id);
documents_deletion.delete_document_by_external_docid(document_id);
}
let update_id = data.db.update_write(|w| documents_deletion.finalize(w))?;
Ok(HttpResponse::Accepted().json(IndexUpdateResponse::with_id(update_id)))
}
#[delete("/indexes/{index_uid}/documents", wrap = "Authentication::Private")]
async fn clear_all_documents(
data: web::Data<Data>,
path: web::Path<IndexParam>,
) -> Result<HttpResponse, ResponseError> {
let index = data
.db
.open_index(&path.index_uid)
.ok_or(Error::index_not_found(&path.index_uid))?;
let update_id = data.db.update_write(|w| index.clear_all(w))?;
Ok(HttpResponse::Accepted().json(IndexUpdateResponse::with_id(update_id)))
}

64
src/routes/dump.rs Normal file
View File

@ -0,0 +1,64 @@
use std::fs::File;
use std::path::Path;
use actix_web::{get, post};
use actix_web::{HttpResponse, web};
use serde::{Deserialize, Serialize};
use crate::dump::{DumpInfo, DumpStatus, compressed_dumps_dir, init_dump_process};
use crate::Data;
use crate::error::{Error, ResponseError};
use crate::helpers::Authentication;
pub fn services(cfg: &mut web::ServiceConfig) {
cfg.service(trigger_dump)
.service(get_dump_status);
}
#[post("/dumps", wrap = "Authentication::Private")]
async fn trigger_dump(
data: web::Data<Data>,
) -> Result<HttpResponse, ResponseError> {
let dumps_dir = Path::new(&data.dumps_dir);
match init_dump_process(&data, &dumps_dir) {
Ok(resume) => Ok(HttpResponse::Accepted().json(resume)),
Err(e) => Err(e.into())
}
}
#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
struct DumpStatusResponse {
status: String,
}
#[derive(Deserialize)]
struct DumpParam {
dump_uid: String,
}
#[get("/dumps/{dump_uid}/status", wrap = "Authentication::Private")]
async fn get_dump_status(
data: web::Data<Data>,
path: web::Path<DumpParam>,
) -> Result<HttpResponse, ResponseError> {
let dumps_dir = Path::new(&data.dumps_dir);
let dump_uid = &path.dump_uid;
if let Some(resume) = DumpInfo::get_current() {
if &resume.uid == dump_uid {
return Ok(HttpResponse::Ok().json(resume));
}
}
if File::open(compressed_dumps_dir(Path::new(dumps_dir), dump_uid)).is_ok() {
let resume = DumpInfo::new(
dump_uid.into(),
DumpStatus::Done
);
Ok(HttpResponse::Ok().json(resume))
} else {
Err(Error::not_found("dump does not exist").into())
}
}

13
src/routes/health.rs Normal file
View File

@ -0,0 +1,13 @@
use actix_web::get;
use actix_web::{web, HttpResponse};
use crate::error::ResponseError;
pub fn services(cfg: &mut web::ServiceConfig) {
cfg.service(get_health);
}
#[get("/health")]
async fn get_health() -> Result<HttpResponse, ResponseError> {
Ok(HttpResponse::NoContent().finish())
}

388
src/routes/index.rs Normal file
View File

@ -0,0 +1,388 @@
use actix_web::{delete, get, post, put};
use actix_web::{web, HttpResponse};
use chrono::{DateTime, Utc};
use log::error;
use meilisearch_core::{Database, MainReader, UpdateReader};
use meilisearch_core::update::UpdateStatus;
use rand::seq::SliceRandom;
use serde::{Deserialize, Serialize};
use crate::Data;
use crate::error::{Error, ResponseError};
use crate::helpers::Authentication;
use crate::routes::IndexParam;
pub fn services(cfg: &mut web::ServiceConfig) {
cfg.service(list_indexes)
.service(get_index)
.service(create_index)
.service(update_index)
.service(delete_index)
.service(get_update_status)
.service(get_all_updates_status);
}
fn generate_uid() -> String {
let mut rng = rand::thread_rng();
let sample = b"abcdefghijklmnopqrstuvwxyz0123456789";
sample
.choose_multiple(&mut rng, 8)
.map(|c| *c as char)
.collect()
}
#[derive(Debug, Serialize, Deserialize, Clone)]
#[serde(rename_all = "camelCase")]
pub struct IndexResponse {
pub name: String,
pub uid: String,
created_at: DateTime<Utc>,
updated_at: DateTime<Utc>,
pub primary_key: Option<String>,
}
pub fn list_indexes_sync(data: &web::Data<Data>, reader: &MainReader) -> Result<Vec<IndexResponse>, ResponseError> {
let mut indexes = Vec::new();
for index_uid in data.db.indexes_uids() {
let index = data.db.open_index(&index_uid);
match index {
Some(index) => {
let name = index.main.name(reader)?.ok_or(Error::internal(
"Impossible to get the name of an index",
))?;
let created_at = index
.main
.created_at(reader)?
.ok_or(Error::internal(
"Impossible to get the create date of an index",
))?;
let updated_at = index
.main
.updated_at(reader)?
.ok_or(Error::internal(
"Impossible to get the last update date of an index",
))?;
let primary_key = match index.main.schema(reader) {
Ok(Some(schema)) => match schema.primary_key() {
Some(primary_key) => Some(primary_key.to_owned()),
None => None,
},
_ => None,
};
let index_response = IndexResponse {
name,
uid: index_uid,
created_at,
updated_at,
primary_key,
};
indexes.push(index_response);
}
None => error!(
"Index {} is referenced in the indexes list but cannot be found",
index_uid
),
}
}
Ok(indexes)
}
#[get("/indexes", wrap = "Authentication::Private")]
async fn list_indexes(data: web::Data<Data>) -> Result<HttpResponse, ResponseError> {
let reader = data.db.main_read_txn()?;
let indexes = list_indexes_sync(&data, &reader)?;
Ok(HttpResponse::Ok().json(indexes))
}
#[get("/indexes/{index_uid}", wrap = "Authentication::Private")]
async fn get_index(
data: web::Data<Data>,
path: web::Path<IndexParam>,
) -> Result<HttpResponse, ResponseError> {
let index = data
.db
.open_index(&path.index_uid)
.ok_or(Error::index_not_found(&path.index_uid))?;
let reader = data.db.main_read_txn()?;
let name = index.main.name(&reader)?.ok_or(Error::internal(
"Impossible to get the name of an index",
))?;
let created_at = index
.main
.created_at(&reader)?
.ok_or(Error::internal(
"Impossible to get the create date of an index",
))?;
let updated_at = index
.main
.updated_at(&reader)?
.ok_or(Error::internal(
"Impossible to get the last update date of an index",
))?;
let primary_key = match index.main.schema(&reader) {
Ok(Some(schema)) => match schema.primary_key() {
Some(primary_key) => Some(primary_key.to_owned()),
None => None,
},
_ => None,
};
let index_response = IndexResponse {
name,
uid: path.index_uid.clone(),
created_at,
updated_at,
primary_key,
};
Ok(HttpResponse::Ok().json(index_response))
}
#[derive(Debug, Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
struct IndexCreateRequest {
name: Option<String>,
uid: Option<String>,
primary_key: Option<String>,
}
pub fn create_index_sync(
database: &std::sync::Arc<Database>,
uid: String,
name: String,
primary_key: Option<String>,
) -> Result<IndexResponse, Error> {
let created_index = database
.create_index(&uid)
.map_err(|e| match e {
meilisearch_core::Error::IndexAlreadyExists => Error::IndexAlreadyExists(uid.clone()),
_ => Error::create_index(e)
})?;
let index_response = database.main_write::<_, _, Error>(|mut write_txn| {
created_index.main.put_name(&mut write_txn, &name)?;
let created_at = created_index
.main
.created_at(&write_txn)?
.ok_or(Error::internal("Impossible to read created at"))?;
let updated_at = created_index
.main
.updated_at(&write_txn)?
.ok_or(Error::internal("Impossible to read updated at"))?;
if let Some(id) = primary_key.clone() {
if let Some(mut schema) = created_index.main.schema(&write_txn)? {
schema
.set_primary_key(&id)
.map_err(Error::bad_request)?;
created_index.main.put_schema(&mut write_txn, &schema)?;
}
}
let index_response = IndexResponse {
name,
uid,
created_at,
updated_at,
primary_key,
};
Ok(index_response)
})?;
Ok(index_response)
}
#[post("/indexes", wrap = "Authentication::Private")]
async fn create_index(
data: web::Data<Data>,
body: web::Json<IndexCreateRequest>,
) -> Result<HttpResponse, ResponseError> {
if let (None, None) = (body.name.clone(), body.uid.clone()) {
return Err(Error::bad_request(
"Index creation must have an uid",
).into());
}
let uid = match &body.uid {
Some(uid) => {
if uid
.chars()
.all(|x| x.is_ascii_alphanumeric() || x == '-' || x == '_')
{
uid.to_owned()
} else {
return Err(Error::InvalidIndexUid.into());
}
}
None => loop {
let uid = generate_uid();
if data.db.open_index(&uid).is_none() {
break uid;
}
},
};
let name = body.name.as_ref().unwrap_or(&uid).to_string();
let index_response = create_index_sync(&data.db, uid, name, body.primary_key.clone())?;
Ok(HttpResponse::Created().json(index_response))
}
#[derive(Debug, Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
struct UpdateIndexRequest {
name: Option<String>,
primary_key: Option<String>,
}
#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
struct UpdateIndexResponse {
name: String,
uid: String,
created_at: DateTime<Utc>,
updated_at: DateTime<Utc>,
primary_key: Option<String>,
}
#[put("/indexes/{index_uid}", wrap = "Authentication::Private")]
async fn update_index(
data: web::Data<Data>,
path: web::Path<IndexParam>,
body: web::Json<IndexCreateRequest>,
) -> Result<HttpResponse, ResponseError> {
let index = data
.db
.open_index(&path.index_uid)
.ok_or(Error::index_not_found(&path.index_uid))?;
data.db.main_write::<_, _, ResponseError>(|writer| {
if let Some(name) = &body.name {
index.main.put_name(writer, name)?;
}
if let Some(id) = body.primary_key.clone() {
if let Some(mut schema) = index.main.schema(writer)? {
schema.set_primary_key(&id)?;
index.main.put_schema(writer, &schema)?;
}
}
index.main.put_updated_at(writer)?;
Ok(())
})?;
let reader = data.db.main_read_txn()?;
let name = index.main.name(&reader)?.ok_or(Error::internal(
"Impossible to get the name of an index",
))?;
let created_at = index
.main
.created_at(&reader)?
.ok_or(Error::internal(
"Impossible to get the create date of an index",
))?;
let updated_at = index
.main
.updated_at(&reader)?
.ok_or(Error::internal(
"Impossible to get the last update date of an index",
))?;
let primary_key = match index.main.schema(&reader) {
Ok(Some(schema)) => match schema.primary_key() {
Some(primary_key) => Some(primary_key.to_owned()),
None => None,
},
_ => None,
};
let index_response = IndexResponse {
name,
uid: path.index_uid.clone(),
created_at,
updated_at,
primary_key,
};
Ok(HttpResponse::Ok().json(index_response))
}
#[delete("/indexes/{index_uid}", wrap = "Authentication::Private")]
async fn delete_index(
data: web::Data<Data>,
path: web::Path<IndexParam>,
) -> Result<HttpResponse, ResponseError> {
if data.db.delete_index(&path.index_uid)? {
Ok(HttpResponse::NoContent().finish())
} else {
Err(Error::index_not_found(&path.index_uid).into())
}
}
#[derive(Deserialize)]
struct UpdateParam {
index_uid: String,
update_id: u64,
}
#[get(
"/indexes/{index_uid}/updates/{update_id}",
wrap = "Authentication::Private"
)]
async fn get_update_status(
data: web::Data<Data>,
path: web::Path<UpdateParam>,
) -> Result<HttpResponse, ResponseError> {
let index = data
.db
.open_index(&path.index_uid)
.ok_or(Error::index_not_found(&path.index_uid))?;
let reader = data.db.update_read_txn()?;
let status = index.update_status(&reader, path.update_id)?;
match status {
Some(status) => Ok(HttpResponse::Ok().json(status)),
None => Err(Error::NotFound(format!(
"Update {}",
path.update_id
)).into()),
}
}
pub fn get_all_updates_status_sync(
data: &web::Data<Data>,
reader: &UpdateReader,
index_uid: &str,
) -> Result<Vec<UpdateStatus>, Error> {
let index = data
.db
.open_index(index_uid)
.ok_or(Error::index_not_found(index_uid))?;
Ok(index.all_updates_status(reader)?)
}
#[get("/indexes/{index_uid}/updates", wrap = "Authentication::Private")]
async fn get_all_updates_status(
data: web::Data<Data>,
path: web::Path<IndexParam>,
) -> Result<HttpResponse, ResponseError> {
let reader = data.db.update_read_txn()?;
let response = get_all_updates_status_sync(&data, &reader, &path.index_uid)?;
Ok(HttpResponse::Ok().json(response))
}

26
src/routes/key.rs Normal file
View File

@ -0,0 +1,26 @@
use actix_web::web;
use actix_web::HttpResponse;
use actix_web::get;
use serde::Serialize;
use crate::helpers::Authentication;
use crate::Data;
pub fn services(cfg: &mut web::ServiceConfig) {
cfg.service(list);
}
#[derive(Serialize)]
struct KeysResponse {
private: Option<String>,
public: Option<String>,
}
#[get("/keys", wrap = "Authentication::Admin")]
async fn list(data: web::Data<Data>) -> HttpResponse {
let api_keys = data.api_keys.clone();
HttpResponse::Ok().json(KeysResponse {
private: api_keys.private,
public: api_keys.public,
})
}

44
src/routes/mod.rs Normal file
View File

@ -0,0 +1,44 @@
use actix_web::{get, HttpResponse};
use serde::{Deserialize, Serialize};
pub mod document;
pub mod health;
pub mod index;
pub mod key;
pub mod search;
pub mod setting;
pub mod stats;
pub mod stop_words;
pub mod synonym;
pub mod dump;
#[derive(Deserialize)]
pub struct IndexParam {
index_uid: String,
}
#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
pub struct IndexUpdateResponse {
pub update_id: u64,
}
impl IndexUpdateResponse {
pub fn with_id(update_id: u64) -> Self {
Self { update_id }
}
}
#[get("/")]
pub async fn load_html() -> HttpResponse {
HttpResponse::Ok()
.content_type("text/html; charset=utf-8")
.body(include_str!("../../public/interface.html").to_string())
}
#[get("/bulma.min.css")]
pub async fn load_css() -> HttpResponse {
HttpResponse::Ok()
.content_type("text/css; charset=utf-8")
.body(include_str!("../../public/bulma.min.css").to_string())
}

269
src/routes/search.rs Normal file
View File

@ -0,0 +1,269 @@
use std::collections::{HashMap, HashSet};
use actix_web::{get, post, web, HttpResponse};
use log::warn;
use serde::{Deserialize, Serialize};
use serde_json::Value;
use crate::error::{Error, FacetCountError, ResponseError};
use crate::helpers::meilisearch::{IndexSearchExt, SearchResult};
use crate::helpers::Authentication;
use crate::routes::IndexParam;
use crate::Data;
use meilisearch_core::facets::FacetFilter;
use meilisearch_schema::{FieldId, Schema};
pub fn services(cfg: &mut web::ServiceConfig) {
cfg.service(search_with_post).service(search_with_url_query);
}
#[derive(Serialize, Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
pub struct SearchQuery {
q: Option<String>,
offset: Option<usize>,
limit: Option<usize>,
attributes_to_retrieve: Option<String>,
attributes_to_crop: Option<String>,
crop_length: Option<usize>,
attributes_to_highlight: Option<String>,
filters: Option<String>,
matches: Option<bool>,
facet_filters: Option<String>,
facets_distribution: Option<String>,
}
#[get("/indexes/{index_uid}/search", wrap = "Authentication::Public")]
async fn search_with_url_query(
data: web::Data<Data>,
path: web::Path<IndexParam>,
params: web::Query<SearchQuery>,
) -> Result<HttpResponse, ResponseError> {
let search_result = params.search(&path.index_uid, data)?;
Ok(HttpResponse::Ok().json(search_result))
}
#[derive(Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
pub struct SearchQueryPost {
q: Option<String>,
offset: Option<usize>,
limit: Option<usize>,
attributes_to_retrieve: Option<Vec<String>>,
attributes_to_crop: Option<Vec<String>>,
crop_length: Option<usize>,
attributes_to_highlight: Option<Vec<String>>,
filters: Option<String>,
matches: Option<bool>,
facet_filters: Option<Value>,
facets_distribution: Option<Vec<String>>,
}
impl From<SearchQueryPost> for SearchQuery {
fn from(other: SearchQueryPost) -> SearchQuery {
SearchQuery {
q: other.q,
offset: other.offset,
limit: other.limit,
attributes_to_retrieve: other.attributes_to_retrieve.map(|attrs| attrs.join(",")),
attributes_to_crop: other.attributes_to_crop.map(|attrs| attrs.join(",")),
crop_length: other.crop_length,
attributes_to_highlight: other.attributes_to_highlight.map(|attrs| attrs.join(",")),
filters: other.filters,
matches: other.matches,
facet_filters: other.facet_filters.map(|f| f.to_string()),
facets_distribution: other.facets_distribution.map(|f| format!("{:?}", f)),
}
}
}
#[post("/indexes/{index_uid}/search", wrap = "Authentication::Public")]
async fn search_with_post(
data: web::Data<Data>,
path: web::Path<IndexParam>,
params: web::Json<SearchQueryPost>,
) -> Result<HttpResponse, ResponseError> {
let query: SearchQuery = params.0.into();
let search_result = query.search(&path.index_uid, data)?;
Ok(HttpResponse::Ok().json(search_result))
}
impl SearchQuery {
fn search(
&self,
index_uid: &str,
data: web::Data<Data>,
) -> Result<SearchResult, ResponseError> {
let index = data
.db
.open_index(index_uid)
.ok_or(Error::index_not_found(index_uid))?;
let reader = data.db.main_read_txn()?;
let schema = index
.main
.schema(&reader)?
.ok_or(Error::internal("Impossible to retrieve the schema"))?;
let query = self
.q
.clone()
.and_then(|q| if q.is_empty() { None } else { Some(q) });
let mut search_builder = index.new_search(query);
if let Some(offset) = self.offset {
search_builder.offset(offset);
}
if let Some(limit) = self.limit {
search_builder.limit(limit);
}
let available_attributes = schema.displayed_name();
let mut restricted_attributes: HashSet<&str>;
match &self.attributes_to_retrieve {
Some(attributes_to_retrieve) => {
let attributes_to_retrieve: HashSet<&str> =
attributes_to_retrieve.split(',').collect();
if attributes_to_retrieve.contains("*") {
restricted_attributes = available_attributes.clone();
} else {
restricted_attributes = HashSet::new();
for attr in attributes_to_retrieve {
if available_attributes.contains(attr) {
restricted_attributes.insert(attr);
search_builder.add_retrievable_field(attr.to_string());
} else {
warn!("The attributes {:?} present in attributesToCrop parameter doesn't exist", attr);
}
}
}
}
None => {
restricted_attributes = available_attributes.clone();
}
}
if let Some(ref facet_filters) = self.facet_filters {
let attrs = index
.main
.attributes_for_faceting(&reader)?
.unwrap_or_default();
search_builder.add_facet_filters(FacetFilter::from_str(
facet_filters,
&schema,
&attrs,
)?);
}
if let Some(facets) = &self.facets_distribution {
match index.main.attributes_for_faceting(&reader)? {
Some(ref attrs) => {
let field_ids = prepare_facet_list(&facets, &schema, attrs)?;
search_builder.add_facets(field_ids);
}
None => return Err(FacetCountError::NoFacetSet.into()),
}
}
if let Some(attributes_to_crop) = &self.attributes_to_crop {
let default_length = self.crop_length.unwrap_or(200);
let mut final_attributes: HashMap<String, usize> = HashMap::new();
for attribute in attributes_to_crop.split(',') {
let mut attribute = attribute.split(':');
let attr = attribute.next();
let length = attribute
.next()
.and_then(|s| s.parse().ok())
.unwrap_or(default_length);
match attr {
Some("*") => {
for attr in &restricted_attributes {
final_attributes.insert(attr.to_string(), length);
}
}
Some(attr) => {
if available_attributes.contains(attr) {
final_attributes.insert(attr.to_string(), length);
} else {
warn!("The attributes {:?} present in attributesToCrop parameter doesn't exist", attr);
}
}
None => (),
}
}
search_builder.attributes_to_crop(final_attributes);
}
if let Some(attributes_to_highlight) = &self.attributes_to_highlight {
let mut final_attributes: HashSet<String> = HashSet::new();
for attribute in attributes_to_highlight.split(',') {
if attribute == "*" {
for attr in &restricted_attributes {
final_attributes.insert(attr.to_string());
}
} else if available_attributes.contains(attribute) {
final_attributes.insert(attribute.to_string());
} else {
warn!("The attributes {:?} present in attributesToHighlight parameter doesn't exist", attribute);
}
}
search_builder.attributes_to_highlight(final_attributes);
}
if let Some(filters) = &self.filters {
search_builder.filters(filters.to_string());
}
if let Some(matches) = self.matches {
if matches {
search_builder.get_matches();
}
}
search_builder.search(&reader)
}
}
/// Parses the incoming string into an array of attributes for which to return a count. It returns
/// a Vec of attribute names ascociated with their id.
///
/// An error is returned if the array is malformed, or if it contains attributes that are
/// unexisting, or not set as facets.
fn prepare_facet_list(
facets: &str,
schema: &Schema,
facet_attrs: &[FieldId],
) -> Result<Vec<(FieldId, String)>, FacetCountError> {
let json_array = serde_json::from_str(facets)?;
match json_array {
Value::Array(vals) => {
let wildcard = Value::String("*".to_string());
if vals.iter().any(|f| f == &wildcard) {
let attrs = facet_attrs
.iter()
.filter_map(|&id| schema.name(id).map(|n| (id, n.to_string())))
.collect();
return Ok(attrs);
}
let mut field_ids = Vec::with_capacity(facet_attrs.len());
for facet in vals {
match facet {
Value::String(facet) => {
if let Some(id) = schema.id(&facet) {
if !facet_attrs.contains(&id) {
return Err(FacetCountError::AttributeNotSet(facet));
}
field_ids.push((id, facet));
}
}
bad_val => return Err(FacetCountError::unexpected_token(bad_val, &["String"])),
}
}
Ok(field_ids)
}
bad_val => Err(FacetCountError::unexpected_token(bad_val, &["[String]"])),
}
}

547
src/routes/setting.rs Normal file
View File

@ -0,0 +1,547 @@
use std::collections::{BTreeMap, BTreeSet};
use actix_web::{delete, get, post};
use actix_web::{web, HttpResponse};
use meilisearch_core::{MainReader, UpdateWriter};
use meilisearch_core::settings::{Settings, SettingsUpdate, UpdateState, DEFAULT_RANKING_RULES};
use meilisearch_schema::Schema;
use crate::Data;
use crate::error::{Error, ResponseError};
use crate::helpers::Authentication;
use crate::routes::{IndexParam, IndexUpdateResponse};
pub fn services(cfg: &mut web::ServiceConfig) {
cfg.service(update_all)
.service(get_all)
.service(delete_all)
.service(get_rules)
.service(update_rules)
.service(delete_rules)
.service(get_distinct)
.service(update_distinct)
.service(delete_distinct)
.service(get_searchable)
.service(update_searchable)
.service(delete_searchable)
.service(get_displayed)
.service(update_displayed)
.service(delete_displayed)
.service(get_attributes_for_faceting)
.service(delete_attributes_for_faceting)
.service(update_attributes_for_faceting);
}
pub fn update_all_settings_txn(
data: &web::Data<Data>,
settings: SettingsUpdate,
index_uid: &str,
write_txn: &mut UpdateWriter,
) -> Result<u64, Error> {
let index = data
.db
.open_index(index_uid)
.ok_or(Error::index_not_found(index_uid))?;
let update_id = index.settings_update(write_txn, settings)?;
Ok(update_id)
}
#[post("/indexes/{index_uid}/settings", wrap = "Authentication::Private")]
async fn update_all(
data: web::Data<Data>,
path: web::Path<IndexParam>,
body: web::Json<Settings>,
) -> Result<HttpResponse, ResponseError> {
let update_id = data.get_or_create_index(&path.index_uid, |index| {
Ok(data.db.update_write::<_, _, ResponseError>(|writer| {
let settings = body.into_inner().to_update().map_err(Error::bad_request)?;
let update_id = index.settings_update(writer, settings)?;
Ok(update_id)
})?)
})?;
Ok(HttpResponse::Accepted().json(IndexUpdateResponse::with_id(update_id)))
}
pub fn get_all_sync(data: &web::Data<Data>, reader: &MainReader, index_uid: &str) -> Result<Settings, Error> {
let index = data
.db
.open_index(index_uid)
.ok_or(Error::index_not_found(index_uid))?;
let stop_words: BTreeSet<String> = index.main.stop_words(&reader)?.into_iter().collect();
let synonyms_list = index.main.synonyms(reader)?;
let mut synonyms = BTreeMap::new();
let index_synonyms = &index.synonyms;
for synonym in synonyms_list {
let list = index_synonyms.synonyms(reader, synonym.as_bytes())?;
synonyms.insert(synonym, list);
}
let ranking_rules = index
.main
.ranking_rules(reader)?
.unwrap_or(DEFAULT_RANKING_RULES.to_vec())
.into_iter()
.map(|r| r.to_string())
.collect();
let schema = index.main.schema(&reader)?;
let distinct_attribute = match (index.main.distinct_attribute(reader)?, &schema) {
(Some(id), Some(schema)) => schema.name(id).map(str::to_string),
_ => None,
};
let attributes_for_faceting = match (&schema, &index.main.attributes_for_faceting(&reader)?) {
(Some(schema), Some(attrs)) => attrs
.iter()
.filter_map(|&id| schema.name(id))
.map(str::to_string)
.collect(),
_ => vec![],
};
let searchable_attributes = schema.as_ref().map(get_indexed_attributes);
let displayed_attributes = schema.as_ref().map(get_displayed_attributes);
Ok(Settings {
ranking_rules: Some(Some(ranking_rules)),
distinct_attribute: Some(distinct_attribute),
searchable_attributes: Some(searchable_attributes),
displayed_attributes: Some(displayed_attributes),
stop_words: Some(Some(stop_words)),
synonyms: Some(Some(synonyms)),
attributes_for_faceting: Some(Some(attributes_for_faceting)),
})
}
#[get("/indexes/{index_uid}/settings", wrap = "Authentication::Private")]
async fn get_all(
data: web::Data<Data>,
path: web::Path<IndexParam>,
) -> Result<HttpResponse, ResponseError> {
let reader = data.db.main_read_txn()?;
let settings = get_all_sync(&data, &reader, &path.index_uid)?;
Ok(HttpResponse::Ok().json(settings))
}
#[delete("/indexes/{index_uid}/settings", wrap = "Authentication::Private")]
async fn delete_all(
data: web::Data<Data>,
path: web::Path<IndexParam>,
) -> Result<HttpResponse, ResponseError> {
let index = data
.db
.open_index(&path.index_uid)
.ok_or(Error::index_not_found(&path.index_uid))?;
let settings = SettingsUpdate {
ranking_rules: UpdateState::Clear,
distinct_attribute: UpdateState::Clear,
primary_key: UpdateState::Clear,
searchable_attributes: UpdateState::Clear,
displayed_attributes: UpdateState::Clear,
stop_words: UpdateState::Clear,
synonyms: UpdateState::Clear,
attributes_for_faceting: UpdateState::Clear,
};
let update_id = data
.db
.update_write(|w| index.settings_update(w, settings))?;
Ok(HttpResponse::Accepted().json(IndexUpdateResponse::with_id(update_id)))
}
#[get(
"/indexes/{index_uid}/settings/ranking-rules",
wrap = "Authentication::Private"
)]
async fn get_rules(
data: web::Data<Data>,
path: web::Path<IndexParam>,
) -> Result<HttpResponse, ResponseError> {
let index = data
.db
.open_index(&path.index_uid)
.ok_or(Error::index_not_found(&path.index_uid))?;
let reader = data.db.main_read_txn()?;
let ranking_rules = index
.main
.ranking_rules(&reader)?
.unwrap_or(DEFAULT_RANKING_RULES.to_vec())
.into_iter()
.map(|r| r.to_string())
.collect::<Vec<String>>();
Ok(HttpResponse::Ok().json(ranking_rules))
}
#[post(
"/indexes/{index_uid}/settings/ranking-rules",
wrap = "Authentication::Private"
)]
async fn update_rules(
data: web::Data<Data>,
path: web::Path<IndexParam>,
body: web::Json<Option<Vec<String>>>,
) -> Result<HttpResponse, ResponseError> {
let update_id = data.get_or_create_index(&path.index_uid, |index| {
let settings = Settings {
ranking_rules: Some(body.into_inner()),
..Settings::default()
};
let settings = settings.to_update().map_err(Error::bad_request)?;
Ok(data
.db
.update_write(|w| index.settings_update(w, settings))?)
})?;
Ok(HttpResponse::Accepted().json(IndexUpdateResponse::with_id(update_id)))
}
#[delete(
"/indexes/{index_uid}/settings/ranking-rules",
wrap = "Authentication::Private"
)]
async fn delete_rules(
data: web::Data<Data>,
path: web::Path<IndexParam>,
) -> Result<HttpResponse, ResponseError> {
let index = data
.db
.open_index(&path.index_uid)
.ok_or(Error::index_not_found(&path.index_uid))?;
let settings = SettingsUpdate {
ranking_rules: UpdateState::Clear,
..SettingsUpdate::default()
};
let update_id = data
.db
.update_write(|w| index.settings_update(w, settings))?;
Ok(HttpResponse::Accepted().json(IndexUpdateResponse::with_id(update_id)))
}
#[get(
"/indexes/{index_uid}/settings/distinct-attribute",
wrap = "Authentication::Private"
)]
async fn get_distinct(
data: web::Data<Data>,
path: web::Path<IndexParam>,
) -> Result<HttpResponse, ResponseError> {
let index = data
.db
.open_index(&path.index_uid)
.ok_or(Error::index_not_found(&path.index_uid))?;
let reader = data.db.main_read_txn()?;
let distinct_attribute_id = index.main.distinct_attribute(&reader)?;
let schema = index.main.schema(&reader)?;
let distinct_attribute = match (schema, distinct_attribute_id) {
(Some(schema), Some(id)) => schema.name(id).map(str::to_string),
_ => None,
};
Ok(HttpResponse::Ok().json(distinct_attribute))
}
#[post(
"/indexes/{index_uid}/settings/distinct-attribute",
wrap = "Authentication::Private"
)]
async fn update_distinct(
data: web::Data<Data>,
path: web::Path<IndexParam>,
body: web::Json<Option<String>>,
) -> Result<HttpResponse, ResponseError> {
let update_id = data.get_or_create_index(&path.index_uid, |index| {
let settings = Settings {
distinct_attribute: Some(body.into_inner()),
..Settings::default()
};
let settings = settings.to_update().map_err(Error::bad_request)?;
Ok(data
.db
.update_write(|w| index.settings_update(w, settings))?)
})?;
Ok(HttpResponse::Accepted().json(IndexUpdateResponse::with_id(update_id)))
}
#[delete(
"/indexes/{index_uid}/settings/distinct-attribute",
wrap = "Authentication::Private"
)]
async fn delete_distinct(
data: web::Data<Data>,
path: web::Path<IndexParam>,
) -> Result<HttpResponse, ResponseError> {
let index = data
.db
.open_index(&path.index_uid)
.ok_or(Error::index_not_found(&path.index_uid))?;
let settings = SettingsUpdate {
distinct_attribute: UpdateState::Clear,
..SettingsUpdate::default()
};
let update_id = data
.db
.update_write(|w| index.settings_update(w, settings))?;
Ok(HttpResponse::Accepted().json(IndexUpdateResponse::with_id(update_id)))
}
#[get(
"/indexes/{index_uid}/settings/searchable-attributes",
wrap = "Authentication::Private"
)]
async fn get_searchable(
data: web::Data<Data>,
path: web::Path<IndexParam>,
) -> Result<HttpResponse, ResponseError> {
let index = data
.db
.open_index(&path.index_uid)
.ok_or(Error::index_not_found(&path.index_uid))?;
let reader = data.db.main_read_txn()?;
let schema = index.main.schema(&reader)?;
let searchable_attributes: Option<Vec<String>> = schema.as_ref().map(get_indexed_attributes);
Ok(HttpResponse::Ok().json(searchable_attributes))
}
#[post(
"/indexes/{index_uid}/settings/searchable-attributes",
wrap = "Authentication::Private"
)]
async fn update_searchable(
data: web::Data<Data>,
path: web::Path<IndexParam>,
body: web::Json<Option<Vec<String>>>,
) -> Result<HttpResponse, ResponseError> {
let update_id = data.get_or_create_index(&path.index_uid, |index| {
let settings = Settings {
searchable_attributes: Some(body.into_inner()),
..Settings::default()
};
let settings = settings.to_update().map_err(Error::bad_request)?;
Ok(data
.db
.update_write(|w| index.settings_update(w, settings))?)
})?;
Ok(HttpResponse::Accepted().json(IndexUpdateResponse::with_id(update_id)))
}
#[delete(
"/indexes/{index_uid}/settings/searchable-attributes",
wrap = "Authentication::Private"
)]
async fn delete_searchable(
data: web::Data<Data>,
path: web::Path<IndexParam>,
) -> Result<HttpResponse, ResponseError> {
let index = data
.db
.open_index(&path.index_uid)
.ok_or(Error::index_not_found(&path.index_uid))?;
let settings = SettingsUpdate {
searchable_attributes: UpdateState::Clear,
..SettingsUpdate::default()
};
let update_id = data
.db
.update_write(|w| index.settings_update(w, settings))?;
Ok(HttpResponse::Accepted().json(IndexUpdateResponse::with_id(update_id)))
}
#[get(
"/indexes/{index_uid}/settings/displayed-attributes",
wrap = "Authentication::Private"
)]
async fn get_displayed(
data: web::Data<Data>,
path: web::Path<IndexParam>,
) -> Result<HttpResponse, ResponseError> {
let index = data
.db
.open_index(&path.index_uid)
.ok_or(Error::index_not_found(&path.index_uid))?;
let reader = data.db.main_read_txn()?;
let schema = index.main.schema(&reader)?;
let displayed_attributes = schema.as_ref().map(get_displayed_attributes);
Ok(HttpResponse::Ok().json(displayed_attributes))
}
#[post(
"/indexes/{index_uid}/settings/displayed-attributes",
wrap = "Authentication::Private"
)]
async fn update_displayed(
data: web::Data<Data>,
path: web::Path<IndexParam>,
body: web::Json<Option<BTreeSet<String>>>,
) -> Result<HttpResponse, ResponseError> {
let update_id = data.get_or_create_index(&path.index_uid, |index| {
let settings = Settings {
displayed_attributes: Some(body.into_inner()),
..Settings::default()
};
let settings = settings.to_update().map_err(Error::bad_request)?;
Ok(data
.db
.update_write(|w| index.settings_update(w, settings))?)
})?;
Ok(HttpResponse::Accepted().json(IndexUpdateResponse::with_id(update_id)))
}
#[delete(
"/indexes/{index_uid}/settings/displayed-attributes",
wrap = "Authentication::Private"
)]
async fn delete_displayed(
data: web::Data<Data>,
path: web::Path<IndexParam>,
) -> Result<HttpResponse, ResponseError> {
let index = data
.db
.open_index(&path.index_uid)
.ok_or(Error::index_not_found(&path.index_uid))?;
let settings = SettingsUpdate {
displayed_attributes: UpdateState::Clear,
..SettingsUpdate::default()
};
let update_id = data
.db
.update_write(|w| index.settings_update(w, settings))?;
Ok(HttpResponse::Accepted().json(IndexUpdateResponse::with_id(update_id)))
}
#[get(
"/indexes/{index_uid}/settings/attributes-for-faceting",
wrap = "Authentication::Private"
)]
async fn get_attributes_for_faceting(
data: web::Data<Data>,
path: web::Path<IndexParam>,
) -> Result<HttpResponse, ResponseError> {
let index = data
.db
.open_index(&path.index_uid)
.ok_or(Error::index_not_found(&path.index_uid))?;
let attributes_for_faceting = data.db.main_read::<_, _, ResponseError>(|reader| {
let schema = index.main.schema(reader)?;
let attrs = index.main.attributes_for_faceting(reader)?;
let attr_names = match (&schema, &attrs) {
(Some(schema), Some(attrs)) => attrs
.iter()
.filter_map(|&id| schema.name(id))
.map(str::to_string)
.collect(),
_ => vec![],
};
Ok(attr_names)
})?;
Ok(HttpResponse::Ok().json(attributes_for_faceting))
}
#[post(
"/indexes/{index_uid}/settings/attributes-for-faceting",
wrap = "Authentication::Private"
)]
async fn update_attributes_for_faceting(
data: web::Data<Data>,
path: web::Path<IndexParam>,
body: web::Json<Option<Vec<String>>>,
) -> Result<HttpResponse, ResponseError> {
let update_id = data.get_or_create_index(&path.index_uid, |index| {
let settings = Settings {
attributes_for_faceting: Some(body.into_inner()),
..Settings::default()
};
let settings = settings.to_update().map_err(Error::bad_request)?;
Ok(data
.db
.update_write(|w| index.settings_update(w, settings))?)
})?;
Ok(HttpResponse::Accepted().json(IndexUpdateResponse::with_id(update_id)))
}
#[delete(
"/indexes/{index_uid}/settings/attributes-for-faceting",
wrap = "Authentication::Private"
)]
async fn delete_attributes_for_faceting(
data: web::Data<Data>,
path: web::Path<IndexParam>,
) -> Result<HttpResponse, ResponseError> {
let index = data
.db
.open_index(&path.index_uid)
.ok_or(Error::index_not_found(&path.index_uid))?;
let settings = SettingsUpdate {
attributes_for_faceting: UpdateState::Clear,
..SettingsUpdate::default()
};
let update_id = data
.db
.update_write(|w| index.settings_update(w, settings))?;
Ok(HttpResponse::Accepted().json(IndexUpdateResponse::with_id(update_id)))
}
fn get_indexed_attributes(schema: &Schema) -> Vec<String> {
if schema.is_indexed_all() {
["*"].iter().map(|s| s.to_string()).collect()
} else {
schema
.indexed_name()
.iter()
.map(|s| s.to_string())
.collect()
}
}
fn get_displayed_attributes(schema: &Schema) -> BTreeSet<String> {
if schema.is_displayed_all() {
["*"].iter().map(|s| s.to_string()).collect()
} else {
schema
.displayed_name()
.iter()
.map(|s| s.to_string())
.collect()
}
}

134
src/routes/stats.rs Normal file
View File

@ -0,0 +1,134 @@
use std::collections::{HashMap, BTreeMap};
use actix_web::web;
use actix_web::HttpResponse;
use actix_web::get;
use chrono::{DateTime, Utc};
use log::error;
use serde::Serialize;
use walkdir::WalkDir;
use crate::error::{Error, ResponseError};
use crate::helpers::Authentication;
use crate::routes::IndexParam;
use crate::Data;
pub fn services(cfg: &mut web::ServiceConfig) {
cfg.service(index_stats)
.service(get_stats)
.service(get_version);
}
#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
struct IndexStatsResponse {
number_of_documents: u64,
is_indexing: bool,
fields_distribution: BTreeMap<String, usize>,
}
#[get("/indexes/{index_uid}/stats", wrap = "Authentication::Private")]
async fn index_stats(
data: web::Data<Data>,
path: web::Path<IndexParam>,
) -> Result<HttpResponse, ResponseError> {
let index = data
.db
.open_index(&path.index_uid)
.ok_or(Error::index_not_found(&path.index_uid))?;
let reader = data.db.main_read_txn()?;
let number_of_documents = index.main.number_of_documents(&reader)?;
let fields_distribution = index.main.fields_distribution(&reader)?.unwrap_or_default();
let update_reader = data.db.update_read_txn()?;
let is_indexing =
data.db.is_indexing(&update_reader, &path.index_uid)?
.ok_or(Error::internal(
"Impossible to know if the database is indexing",
))?;
Ok(HttpResponse::Ok().json(IndexStatsResponse {
number_of_documents,
is_indexing,
fields_distribution,
}))
}
#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
struct StatsResult {
database_size: u64,
last_update: Option<DateTime<Utc>>,
indexes: HashMap<String, IndexStatsResponse>,
}
#[get("/stats", wrap = "Authentication::Private")]
async fn get_stats(data: web::Data<Data>) -> Result<HttpResponse, ResponseError> {
let mut index_list = HashMap::new();
let reader = data.db.main_read_txn()?;
let update_reader = data.db.update_read_txn()?;
let indexes_set = data.db.indexes_uids();
for index_uid in indexes_set {
let index = data.db.open_index(&index_uid);
match index {
Some(index) => {
let number_of_documents = index.main.number_of_documents(&reader)?;
let fields_distribution = index.main.fields_distribution(&reader)?.unwrap_or_default();
let is_indexing = data.db.is_indexing(&update_reader, &index_uid)?.ok_or(
Error::internal("Impossible to know if the database is indexing"),
)?;
let response = IndexStatsResponse {
number_of_documents,
is_indexing,
fields_distribution,
};
index_list.insert(index_uid, response);
}
None => error!(
"Index {:?} is referenced in the indexes list but cannot be found",
index_uid
),
}
}
let database_size = WalkDir::new(&data.db_path)
.into_iter()
.filter_map(|entry| entry.ok())
.filter_map(|entry| entry.metadata().ok())
.filter(|metadata| metadata.is_file())
.fold(0, |acc, m| acc + m.len());
let last_update = data.db.last_update(&reader)?;
Ok(HttpResponse::Ok().json(StatsResult {
database_size,
last_update,
indexes: index_list,
}))
}
#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
struct VersionResponse {
commit_sha: String,
build_date: String,
pkg_version: String,
}
#[get("/version", wrap = "Authentication::Private")]
async fn get_version() -> HttpResponse {
HttpResponse::Ok().json(VersionResponse {
commit_sha: env!("VERGEN_SHA").to_string(),
build_date: env!("VERGEN_BUILD_TIMESTAMP").to_string(),
pkg_version: env!("CARGO_PKG_VERSION").to_string(),
})
}

79
src/routes/stop_words.rs Normal file
View File

@ -0,0 +1,79 @@
use actix_web::{web, HttpResponse};
use actix_web::{delete, get, post};
use meilisearch_core::settings::{SettingsUpdate, UpdateState};
use std::collections::BTreeSet;
use crate::error::{Error, ResponseError};
use crate::helpers::Authentication;
use crate::routes::{IndexParam, IndexUpdateResponse};
use crate::Data;
pub fn services(cfg: &mut web::ServiceConfig) {
cfg.service(get).service(update).service(delete);
}
#[get(
"/indexes/{index_uid}/settings/stop-words",
wrap = "Authentication::Private"
)]
async fn get(
data: web::Data<Data>,
path: web::Path<IndexParam>,
) -> Result<HttpResponse, ResponseError> {
let index = data
.db
.open_index(&path.index_uid)
.ok_or(Error::index_not_found(&path.index_uid))?;
let reader = data.db.main_read_txn()?;
let stop_words = index.main.stop_words(&reader)?;
Ok(HttpResponse::Ok().json(stop_words))
}
#[post(
"/indexes/{index_uid}/settings/stop-words",
wrap = "Authentication::Private"
)]
async fn update(
data: web::Data<Data>,
path: web::Path<IndexParam>,
body: web::Json<BTreeSet<String>>,
) -> Result<HttpResponse, ResponseError> {
let update_id = data.get_or_create_index(&path.index_uid, |index| {
let settings = SettingsUpdate {
stop_words: UpdateState::Update(body.into_inner()),
..SettingsUpdate::default()
};
Ok(data
.db
.update_write(|w| index.settings_update(w, settings))?)
})?;
Ok(HttpResponse::Accepted().json(IndexUpdateResponse::with_id(update_id)))
}
#[delete(
"/indexes/{index_uid}/settings/stop-words",
wrap = "Authentication::Private"
)]
async fn delete(
data: web::Data<Data>,
path: web::Path<IndexParam>,
) -> Result<HttpResponse, ResponseError> {
let index = data
.db
.open_index(&path.index_uid)
.ok_or(Error::index_not_found(&path.index_uid))?;
let settings = SettingsUpdate {
stop_words: UpdateState::Clear,
..SettingsUpdate::default()
};
let update_id = data
.db
.update_write(|w| index.settings_update(w, settings))?;
Ok(HttpResponse::Accepted().json(IndexUpdateResponse::with_id(update_id)))
}

90
src/routes/synonym.rs Normal file
View File

@ -0,0 +1,90 @@
use std::collections::BTreeMap;
use actix_web::{web, HttpResponse};
use actix_web::{delete, get, post};
use indexmap::IndexMap;
use meilisearch_core::settings::{SettingsUpdate, UpdateState};
use crate::error::{Error, ResponseError};
use crate::helpers::Authentication;
use crate::routes::{IndexParam, IndexUpdateResponse};
use crate::Data;
pub fn services(cfg: &mut web::ServiceConfig) {
cfg.service(get).service(update).service(delete);
}
#[get(
"/indexes/{index_uid}/settings/synonyms",
wrap = "Authentication::Private"
)]
async fn get(
data: web::Data<Data>,
path: web::Path<IndexParam>,
) -> Result<HttpResponse, ResponseError> {
let index = data
.db
.open_index(&path.index_uid)
.ok_or(Error::index_not_found(&path.index_uid))?;
let reader = data.db.main_read_txn()?;
let synonyms_list = index.main.synonyms(&reader)?;
let mut synonyms = IndexMap::new();
let index_synonyms = &index.synonyms;
for synonym in synonyms_list {
let list = index_synonyms.synonyms(&reader, synonym.as_bytes())?;
synonyms.insert(synonym, list);
}
Ok(HttpResponse::Ok().json(synonyms))
}
#[post(
"/indexes/{index_uid}/settings/synonyms",
wrap = "Authentication::Private"
)]
async fn update(
data: web::Data<Data>,
path: web::Path<IndexParam>,
body: web::Json<BTreeMap<String, Vec<String>>>,
) -> Result<HttpResponse, ResponseError> {
let update_id = data.get_or_create_index(&path.index_uid, |index| {
let settings = SettingsUpdate {
synonyms: UpdateState::Update(body.into_inner()),
..SettingsUpdate::default()
};
Ok(data
.db
.update_write(|w| index.settings_update(w, settings))?)
})?;
Ok(HttpResponse::Accepted().json(IndexUpdateResponse::with_id(update_id)))
}
#[delete(
"/indexes/{index_uid}/settings/synonyms",
wrap = "Authentication::Private"
)]
async fn delete(
data: web::Data<Data>,
path: web::Path<IndexParam>,
) -> Result<HttpResponse, ResponseError> {
let index = data
.db
.open_index(&path.index_uid)
.ok_or(Error::index_not_found(&path.index_uid))?;
let settings = SettingsUpdate {
synonyms: UpdateState::Clear,
..SettingsUpdate::default()
};
let update_id = data
.db
.update_write(|w| index.settings_update(w, settings))?;
Ok(HttpResponse::Accepted().json(IndexUpdateResponse::with_id(update_id)))
}

96
src/snapshot.rs Normal file
View File

@ -0,0 +1,96 @@
use crate::Data;
use crate::error::Error;
use crate::helpers::compression;
use log::error;
use std::fs::create_dir_all;
use std::path::Path;
use std::thread;
use std::time::{Duration};
use tempfile::TempDir;
pub fn load_snapshot(
db_path: &str,
snapshot_path: &Path,
ignore_snapshot_if_db_exists: bool,
ignore_missing_snapshot: bool
) -> Result<(), Error> {
let db_path = Path::new(db_path);
if !db_path.exists() && snapshot_path.exists() {
compression::from_tar_gz(snapshot_path, db_path)
} else if db_path.exists() && !ignore_snapshot_if_db_exists {
Err(Error::Internal(format!("database already exists at {:?}, try to delete it or rename it", db_path.canonicalize().unwrap_or(db_path.into()))))
} else if !snapshot_path.exists() && !ignore_missing_snapshot {
Err(Error::Internal(format!("snapshot doesn't exist at {:?}", snapshot_path.canonicalize().unwrap_or(snapshot_path.into()))))
} else {
Ok(())
}
}
pub fn create_snapshot(data: &Data, snapshot_path: &Path) -> Result<(), Error> {
let tmp_dir = TempDir::new()?;
data.db.copy_and_compact_to_path(tmp_dir.path())?;
compression::to_tar_gz(tmp_dir.path(), snapshot_path).map_err(|e| Error::Internal(format!("something went wrong during snapshot compression: {}", e)))
}
pub fn schedule_snapshot(data: Data, snapshot_dir: &Path, time_gap_s: u64) -> Result<(), Error> {
if snapshot_dir.file_name().is_none() {
return Err(Error::Internal("invalid snapshot file path".to_string()));
}
let db_name = Path::new(&data.db_path).file_name().ok_or_else(|| Error::Internal("invalid database name".to_string()))?;
create_dir_all(snapshot_dir)?;
let snapshot_path = snapshot_dir.join(format!("{}.snapshot", db_name.to_str().unwrap_or("data.ms")));
thread::spawn(move || loop {
if let Err(e) = create_snapshot(&data, &snapshot_path) {
error!("Unsuccessful snapshot creation: {}", e);
}
thread::sleep(Duration::from_secs(time_gap_s));
});
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
use std::io::prelude::*;
use std::fs;
#[test]
fn test_pack_unpack() {
let tempdir = TempDir::new().unwrap();
let test_dir = tempdir.path();
let src_dir = test_dir.join("src");
let dest_dir = test_dir.join("complex/destination/path/");
let archive_path = test_dir.join("archive.snapshot");
let file_1_relative = Path::new("file1.txt");
let subdir_relative = Path::new("subdir/");
let file_2_relative = Path::new("subdir/file2.txt");
create_dir_all(src_dir.join(subdir_relative)).unwrap();
fs::File::create(src_dir.join(file_1_relative)).unwrap().write_all(b"Hello_file_1").unwrap();
fs::File::create(src_dir.join(file_2_relative)).unwrap().write_all(b"Hello_file_2").unwrap();
assert!(compression::to_tar_gz(&src_dir, &archive_path).is_ok());
assert!(archive_path.exists());
assert!(load_snapshot(&dest_dir.to_str().unwrap(), &archive_path, false, false).is_ok());
assert!(dest_dir.exists());
assert!(dest_dir.join(file_1_relative).exists());
assert!(dest_dir.join(subdir_relative).exists());
assert!(dest_dir.join(file_2_relative).exists());
let contents = fs::read_to_string(dest_dir.join(file_1_relative)).unwrap();
assert_eq!(contents, "Hello_file_1");
let contents = fs::read_to_string(dest_dir.join(file_2_relative)).unwrap();
assert_eq!(contents, "Hello_file_2");
}
}