filip/Stalwart
1
0
Fork 0
Code Issues Pull Requests Actions 2 Packages Projects Releases Wiki Activity

ES code reorganization

Browse Source
This commit is contained in:
mdecimus
2025-12-13 15:51:57 +01:00
parent afdb764097
commit 8e85a616ff
3 changed files with 145 additions and 207 deletions
Download Patch File Download Diff File Expand all files Collapse all files

160
crates/store/src/backend/elastic/main.rs
View File

@@ -7,8 +7,8 @@
use crate::{
backend::elastic::ElasticSearchStore,
search::{
CalendarSearchField, ContactSearchField, EmailSearchField, FileSearchField, SearchField,
SearchableField, TracingSearchField,
CalendarSearchField, ContactSearchField, EmailSearchField, SearchableField,
TracingSearchField,
},
write::SearchIndex,
};
@@ -80,7 +80,7 @@ impl ElasticSearchStore {
T::primary_keys()
.iter()
.chain(T::all_fields())
.map(|field| (field.es_field().to_string(), field.es_schema()))
.map(|field| (field.field_name().to_string(), field.es_schema()))
.collect::<serde_json::Map<String, Value>>(),
),
);
@@ -107,7 +107,7 @@ impl ElasticSearchStore {
assert_success(
self.client
.put(format!("{}/{}", self.url, T::index().es_index_name()))
.put(format!("{}/{}", self.url, T::index().index_name()))
.body(body)
.send()
.await,
@@ -126,7 +126,7 @@ impl ElasticSearchStore {
] {
assert_success(
self.client
.delete(format!("{}/{}", self.url, index.es_index_name()))
.delete(format!("{}/{}", self.url, index.index_name()))
.send()
.await,
)
@@ -153,153 +153,3 @@ pub(crate) async fn assert_success(response: Result<Response, Error>) -> trc::Re
Err(err) => Err(trc::StoreEvent::ElasticsearchError.reason(err)),
}
}
impl SearchIndex {
pub fn es_index_name(&self) -> &'static str {
match self {
SearchIndex::Email => "st_email",
SearchIndex::Calendar => "st_calendar",
SearchIndex::Contacts => "st_contact",
SearchIndex::File => "st_file",
SearchIndex::Tracing => "st_tracing",
SearchIndex::InMemory => unreachable!(),
}
}
}
impl SearchField {
pub fn es_field(&self) -> &'static str {
match self {
SearchField::AccountId => "acc_id",
SearchField::DocumentId => "doc_id",
SearchField::Id => "id",
SearchField::Email(field) => match field {
EmailSearchField::From => "from",
EmailSearchField::To => "to",
EmailSearchField::Cc => "cc",
EmailSearchField::Bcc => "bcc",
EmailSearchField::Subject => "subj",
EmailSearchField::Body => "body",
EmailSearchField::Attachment => "attach",
EmailSearchField::ReceivedAt => "rcvd",
EmailSearchField::SentAt => "sent",
EmailSearchField::Size => "size",
EmailSearchField::HasAttachment => "has_att",
EmailSearchField::Headers => "headers",
},
SearchField::Calendar(field) => match field {
CalendarSearchField::Title => "title",
CalendarSearchField::Description => "desc",
CalendarSearchField::Location => "loc",
CalendarSearchField::Owner => "owner",
CalendarSearchField::Attendee => "attendee",
CalendarSearchField::Start => "start",
CalendarSearchField::Uid => "uid",
},
SearchField::Contact(field) => match field {
ContactSearchField::Member => "member",
ContactSearchField::Kind => "kind",
ContactSearchField::Name => "name",
ContactSearchField::Nickname => "nick",
ContactSearchField::Organization => "org",
ContactSearchField::Email => "email",
ContactSearchField::Phone => "phone",
ContactSearchField::OnlineService => "online",
ContactSearchField::Address => "addr",
ContactSearchField::Note => "note",
ContactSearchField::Uid => "uid",
},
SearchField::File(field) => match field {
FileSearchField::Name => "name",
FileSearchField::Content => "content",
},
SearchField::Tracing(field) => match field {
TracingSearchField::EventType => "ev_type",
TracingSearchField::QueueId => "queue_id",
TracingSearchField::Keywords => "keywords",
},
}
}
pub fn es_schema(&self) -> Value {
match self {
SearchField::AccountId
| SearchField::DocumentId
| SearchField::Email(EmailSearchField::Size) => json!({
"type": "integer"
}),
SearchField::Id
| SearchField::Email(EmailSearchField::SentAt | EmailSearchField::ReceivedAt)
| SearchField::Calendar(CalendarSearchField::Start)
| SearchField::Tracing(TracingSearchField::QueueId | TracingSearchField::EventType) => {
json!({
"type": "long"
})
}
SearchField::Email(EmailSearchField::HasAttachment) => json!({
"type": "boolean"
}),
SearchField::Calendar(CalendarSearchField::Uid)
| SearchField::Contact(ContactSearchField::Uid) => json!({
"type": "keyword",
}),
SearchField::Email(
EmailSearchField::From | EmailSearchField::To | EmailSearchField::Subject,
) => json!({
"type": "text",
"fields": {
"keyword": {
"type": "keyword"
}
}
}),
SearchField::Email(EmailSearchField::Headers) => {
json!({
"type": "object",
"enabled": true
})
}
#[cfg(feature = "test_mode")]
SearchField::Email(EmailSearchField::Bcc | EmailSearchField::Cc) => {
json!({
"type": "text",
"fields": {
"keyword": {
"type": "keyword"
}
}
})
}
#[cfg(not(feature = "test_mode"))]
SearchField::Email(EmailSearchField::Bcc | EmailSearchField::Cc) => {
json!({
"type": "text"
})
}
SearchField::Email(EmailSearchField::Body | EmailSearchField::Attachment)
| SearchField::Calendar(
CalendarSearchField::Title
| CalendarSearchField::Description
| CalendarSearchField::Location
| CalendarSearchField::Owner
| CalendarSearchField::Attendee,
)
| SearchField::Contact(
ContactSearchField::Member
| ContactSearchField::Kind
| ContactSearchField::Name
| ContactSearchField::Nickname
| ContactSearchField::Organization
| ContactSearchField::Email
| ContactSearchField::Phone
| ContactSearchField::OnlineService
| ContactSearchField::Address
| ContactSearchField::Note,
)
| SearchField::File(FileSearchField::Name | FileSearchField::Content)
| SearchField::Tracing(TracingSearchField::Keywords) => json!({
"type": "text"
}),
}
}
}

123
crates/store/src/backend/elastic/mod.rs
View File

@@ -4,7 +4,10 @@
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
*/
use crate::search::*;
use reqwest::Client;
use serde::{Deserialize, Deserializer};
use serde_json::{Value, json};
pub mod main;
pub mod search;
@@ -13,3 +16,123 @@ pub struct ElasticSearchStore {
client: Client,
url: String,
}
#[derive(Debug, Deserialize)]
pub struct SearchResponse {
pub hits: Hits,
}
#[derive(Debug, Deserialize)]
pub struct Hits {
pub total: Total,
pub hits: Vec<Hit>,
}
#[derive(Debug, Deserialize)]
pub struct Total {
pub value: u64,
}
#[derive(Debug, Deserialize)]
pub struct Hit {
#[serde(rename = "_id", deserialize_with = "deserialize_string_to_u64")]
pub id: u64,
}
#[derive(Debug, Deserialize)]
pub struct DeleteByQueryResponse {
pub deleted: u64,
}
impl SearchField {
pub fn es_schema(&self) -> Value {
match self {
SearchField::AccountId
| SearchField::DocumentId
| SearchField::Email(EmailSearchField::Size) => json!({
"type": "integer"
}),
SearchField::Id
| SearchField::Email(EmailSearchField::SentAt | EmailSearchField::ReceivedAt)
| SearchField::Calendar(CalendarSearchField::Start)
| SearchField::Tracing(TracingSearchField::QueueId | TracingSearchField::EventType) => {
json!({
"type": "long"
})
}
SearchField::Email(EmailSearchField::HasAttachment) => json!({
"type": "boolean"
}),
SearchField::Calendar(CalendarSearchField::Uid)
| SearchField::Contact(ContactSearchField::Uid) => json!({
"type": "keyword",
}),
SearchField::Email(
EmailSearchField::From | EmailSearchField::To | EmailSearchField::Subject,
) => json!({
"type": "text",
"fields": {
"keyword": {
"type": "keyword"
}
}
}),
SearchField::Email(EmailSearchField::Headers) => {
json!({
"type": "object",
"enabled": true
})
}
#[cfg(feature = "test_mode")]
SearchField::Email(EmailSearchField::Bcc | EmailSearchField::Cc) => {
json!({
"type": "text",
"fields": {
"keyword": {
"type": "keyword"
}
}
})
}
#[cfg(not(feature = "test_mode"))]
SearchField::Email(EmailSearchField::Bcc | EmailSearchField::Cc) => {
json!({
"type": "text"
})
}
SearchField::Email(EmailSearchField::Body | EmailSearchField::Attachment)
| SearchField::Calendar(
CalendarSearchField::Title
| CalendarSearchField::Description
| CalendarSearchField::Location
| CalendarSearchField::Owner
| CalendarSearchField::Attendee,
)
| SearchField::Contact(
ContactSearchField::Member
| ContactSearchField::Kind
| ContactSearchField::Name
| ContactSearchField::Nickname
| ContactSearchField::Organization
| ContactSearchField::Email
| ContactSearchField::Phone
| ContactSearchField::OnlineService
| ContactSearchField::Address
| ContactSearchField::Note,
)
| SearchField::File(FileSearchField::Name | FileSearchField::Content)
| SearchField::Tracing(TracingSearchField::Keywords) => json!({
"type": "text"
}),
}
}
}
fn deserialize_string_to_u64<'de, D>(deserializer: D) -> Result<u64, D::Error>
where
D: Deserializer<'de>,
{
<&str>::deserialize(deserializer)?
.parse::<u64>()
.map_err(serde::de::Error::custom)
}

69
crates/store/src/backend/elastic/search.rs
View File

@@ -5,44 +5,18 @@
*/
use crate::{
backend::elastic::{ElasticSearchStore, main::assert_success},
backend::elastic::{
DeleteByQueryResponse, ElasticSearchStore, SearchResponse, main::assert_success,
},
search::{
IndexDocument, SearchComparator, SearchDocumentId, SearchField, SearchFilter,
SearchOperator, SearchQuery, SearchValue,
},
write::SearchIndex,
};
use serde::{Deserialize, Deserializer};
use serde_json::{Map, Value, json};
use std::fmt::Write;
#[derive(Debug, Deserialize)]
pub struct SearchResponse {
pub hits: Hits,
}
#[derive(Debug, Deserialize)]
pub struct Hits {
pub total: Total,
pub hits: Vec<Hit>,
}
#[derive(Debug, Deserialize)]
pub struct Total {
pub value: u64,
}
#[derive(Debug, Deserialize)]
pub struct Hit {
#[serde(rename = "_id", deserialize_with = "deserialize_string_to_u64")]
pub id: u64,
}
#[derive(Debug, Deserialize)]
pub struct DeleteByQueryResponse {
pub deleted: u64,
}
impl ElasticSearchStore {
pub async fn index(&self, documents: Vec<IndexDocument>) -> trc::Result<()> {
let mut request = String::with_capacity(512);
@@ -63,7 +37,7 @@ impl ElasticSearchStore {
let _ = writeln!(
&mut request,
"{{\"index\":{{\"_index\":\"{}\",\"_id\":{id}}}}}",
document.index.es_index_name()
document.index.index_name()
);
json_serialize(&mut request, &document);
request.push('\n');
@@ -99,7 +73,7 @@ impl ElasticSearchStore {
let response = assert_success(
self.client
.post(format!("{}/{}/_search", self.url, index.es_index_name()))
.post(format!("{}/{}/_search", self.url, index.index_name()))
.body(serde_json::to_string(&query).unwrap_or_default())
.send()
.await,
@@ -142,7 +116,7 @@ impl ElasticSearchStore {
.post(format!(
"{}/{}/_delete_by_query",
self.url,
filter.index.es_index_name()
filter.index.index_name()
))
.body(serde_json::to_string(&query).unwrap_or_default())
.send()
@@ -161,7 +135,7 @@ impl ElasticSearchStore {
}
pub async fn refresh_index(&self, index: SearchIndex) -> trc::Result<()> {
let url = format!("{}/{}/_refresh", self.url, index.es_index_name());
let url = format!("{}/{}/_refresh", self.url, index.index_name());
assert_success(self.client.post(url).send().await)
.await
@@ -190,14 +164,14 @@ fn build_query(filters: &[SearchFilter]) -> Value {
if op != &SearchOperator::Equal {
conditions.push(json!({
"match": { field.es_field(): {
"match": { field.field_name(): {
"query": value,
"operator": "and"
} }
}));
} else {
conditions.push(json!({
"match_phrase": { field.es_field(): value }
"match_phrase": { field.field_name(): value }
}));
}
} else {
@@ -213,19 +187,19 @@ fn build_query(filters: &[SearchFilter]) -> Value {
if op == &SearchOperator::Equal {
json!({
"term": {
format!("{}.{}.keyword", field.es_field(), key): value
format!("{}.{}.keyword", field.field_name(), key): value
}
})
} else {
json!({
"match": {
format!("{}.{}", field.es_field(), key): value
format!("{}.{}", field.field_name(), key): value
}
})
}
} else {
json!({
"exists": { "field": format!("{}.{}", field.es_field(), key) }
"exists": { "field": format!("{}.{}", field.field_name(), key) }
})
};
@@ -236,7 +210,7 @@ fn build_query(filters: &[SearchFilter]) -> Value {
let cond = match op {
SearchOperator::Equal | SearchOperator::Contains => json!({
"term": { field.es_field(): value }
"term": { field.field_name(): value }
}),
op => {
let op = match op {
@@ -248,7 +222,7 @@ fn build_query(filters: &[SearchFilter]) -> Value {
};
json!({
"range": { field.es_field(): { op: value } }
"range": { field.field_name(): { op: value } }
})
}
};
@@ -310,9 +284,9 @@ fn build_sort(sort: &[SearchComparator]) -> Value {
.filter_map(|comp| match comp {
SearchComparator::Field { field, ascending } => {
let field = if field.is_text() {
format!("{}.keyword", field.es_field())
format!("{}.keyword", field.field_name())
} else {
field.es_field().to_string()
field.field_name().to_string()
};
Some(json!({
@@ -332,7 +306,7 @@ fn json_serialize(request: &mut String, document: &IndexDocument) {
request.push(',');
}
let _ = write!(request, "{:?}:", k.es_field());
let _ = write!(request, "{:?}:", k.field_name());
match v {
SearchValue::Text { value, .. } => {
json_serialize_str(request, value);
@@ -385,12 +359,3 @@ fn json_serialize_str(request: &mut String, value: &str) {
}
request.push('"');
}
fn deserialize_string_to_u64<'de, D>(deserializer: D) -> Result<u64, D::Error>
where
D: Deserializer<'de>,
{
<&str>::deserialize(deserializer)?
.parse::<u64>()
.map_err(serde::de::Error::custom)
}