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

Email query and thread merge tests passing

Browse Source
This commit is contained in:
Mauro D
2023-04-21 15:40:11 +00:00
parent 51b14ed79e
commit 46b5dc0425
20 changed files with 1401 additions and 232 deletions
Download Patch File Download Diff File Expand all files Collapse all files

27
crates/jmap/src/api/http.rs
View File

@@ -38,12 +38,17 @@ impl JMAP {
"jmap" => match (path.next().unwrap_or(""), req.method()) {
("", &Method::POST) => {
return match fetch_body(req, self.config.request_max_size).await {
Ok(bytes) => match self.handle_request(&bytes).await {
Ok(response) => response.into_http_response(),
Err(err) => err.into_http_response(),
},
Ok(bytes) => {
let delete = "fd";
//println!("<- {}", String::from_utf8_lossy(&bytes));
match self.handle_request(&bytes).await {
Ok(response) => response.into_http_response(),
Err(err) => err.into_http_response(),
}
}
Err(err) => err.into_http_response(),
}
};
}
("download", &Method::GET) => {
if let (Some(account_id), Some(blob_id), Some(name)) = (
@@ -258,8 +263,8 @@ trait ToHttpResponse {
impl ToHttpResponse for Response {
fn into_http_response(self) -> hyper::Response<BoxBody<Bytes, hyper::Error>> {
let delete = "";
println!("-> {}", serde_json::to_string_pretty(&self).unwrap());
//let delete = "";
//println!("-> {}", serde_json::to_string_pretty(&self).unwrap());
hyper::Response::builder()
.status(StatusCode::OK)
.header(header::CONTENT_TYPE, "application/json; charset=utf-8")
@@ -274,8 +279,6 @@ impl ToHttpResponse for Response {
impl ToHttpResponse for Session {
fn into_http_response(self) -> hyper::Response<BoxBody<Bytes, hyper::Error>> {
let delete = "";
println!("-> {}", serde_json::to_string_pretty(&self).unwrap());
hyper::Response::builder()
.status(StatusCode::OK)
.header(header::CONTENT_TYPE, "application/json; charset=utf-8")
@@ -315,9 +318,6 @@ impl ToHttpResponse for DownloadResponse {
impl ToHttpResponse for UploadResponse {
fn into_http_response(self) -> hyper::Response<BoxBody<Bytes, hyper::Error>> {
let delete = "";
println!("-> {}", serde_json::to_string_pretty(&self).unwrap());
hyper::Response::builder()
.status(StatusCode::OK)
.header(header::CONTENT_TYPE, "application/json; charset=utf-8")
@@ -332,9 +332,6 @@ impl ToHttpResponse for UploadResponse {
impl ToHttpResponse for RequestError {
fn into_http_response(self) -> hyper::Response<BoxBody<Bytes, hyper::Error>> {
let delete = "";
println!("-> {}", serde_json::to_string_pretty(&self).unwrap());
hyper::Response::builder()
.status(self.status)
.header(header::CONTENT_TYPE, "application/json; charset=utf-8")

1
crates/jmap/src/api/request.rs
View File

@@ -9,7 +9,6 @@ use crate::JMAP;
impl JMAP {
pub async fn handle_request(&self, bytes: &[u8]) -> Result<Response, RequestError> {
println!("<- {}", String::from_utf8_lossy(bytes));
let request = Request::parse(
bytes,
self.config.request_max_calls,

77
crates/jmap/src/email/index.rs
View File

@@ -4,7 +4,6 @@ use jmap_proto::{
object::Object,
types::{
date::UTCDate,
id::Id,
keyword::Keyword,
property::{HeaderForm, Property},
value::Value,
@@ -16,7 +15,10 @@ use mail_parser::{
Addr, GetHeader, Group, HeaderName, HeaderValue, Message, MessagePart, PartType, RfcHeader,
};
use store::{
fts::{builder::FtsIndexBuilder, Language},
fts::{
builder::{FtsIndexBuilder, MAX_TOKEN_LENGTH},
Language,
},
write::{BatchBuilder, F_BITMAP, F_INDEX, F_VALUE},
};
@@ -89,28 +91,30 @@ impl IndexMessage for BatchBuilder {
language = part_language;
for header in part.headers.into_iter().rev() {
if let HeaderName::Rfc(rfc_header) = header.name {
// Index hasHeader property
let header_num = (rfc_header as u8).to_string();
fts.index_raw_token(Property::Headers, &header_num);
match rfc_header {
RfcHeader::MessageId
| RfcHeader::InReplyTo
| RfcHeader::References
| RfcHeader::ResentMessageId => {
match &header.value {
HeaderValue::Text(id) if id.len() < MAX_ID_LENGTH => {
self.value(Property::MessageId, id.as_ref(), F_INDEX);
header.value.visit_text(|id| {
// Add ids to inverted index
if id.len() < MAX_ID_LENGTH {
println!("indexing {}: {}", rfc_header.as_str(), id);
self.value(Property::MessageId, id, F_INDEX);
}
HeaderValue::TextList(ids) => {
for id in ids {
if id.len() < MAX_ID_LENGTH {
self.value(
Property::MessageId,
id.as_ref(),
F_INDEX,
);
}
}
// Index ids without stemming
if id.len() < MAX_TOKEN_LENGTH {
fts.index_raw_token(
Property::Headers,
format!("{header_num}{id}"),
);
}
_ => (),
}
});
if matches!(
rfc_header,
@@ -135,6 +139,7 @@ impl IndexMessage for BatchBuilder {
| RfcHeader::Bcc
| RfcHeader::ReplyTo
| RfcHeader::Sender => {
let property = Property::from(rfc_header);
let seen_header = seen_headers[rfc_header as usize];
if matches!(
rfc_header,
@@ -172,13 +177,13 @@ impl IndexMessage for BatchBuilder {
}
// Index an address name or email without stemming
fts.index_raw(rfc_header, value);
fts.index_raw(u8::from(&property), value);
});
if !seen_header {
// Add address to inverted index
self.value(
rfc_header,
u8::from(&property),
if !sort_text.is_empty() {
&sort_text
} else {
@@ -192,7 +197,7 @@ impl IndexMessage for BatchBuilder {
if !seen_header {
// Add address to object
object.append(
rfc_header.into(),
property,
header
.value
.trim_text(MAX_STORED_FIELD_LENGTH)
@@ -255,6 +260,20 @@ impl IndexMessage for BatchBuilder {
// Index subject for FTS
fts.index(Property::Subject, subject, language);
}
RfcHeader::Comments | RfcHeader::Keywords | RfcHeader::ListId => {
// Index headers
header.value.visit_text(|text| {
for token in text.split_ascii_whitespace() {
if token.len() < MAX_TOKEN_LENGTH {
fts.index_raw_token(
Property::Headers,
format!("{header_num}{}", token.to_lowercase()),
);
}
}
});
}
_ => (),
}
}
@@ -370,11 +389,12 @@ impl GetContentLanguage for MessagePart<'_> {
}
}
trait VisitAddresses {
trait VisitValues {
fn visit_addresses(&self, visitor: impl FnMut(&str, bool));
fn visit_text(&self, visitor: impl FnMut(&str));
}
impl VisitAddresses for HeaderValue<'_> {
impl VisitValues for HeaderValue<'_> {
fn visit_addresses(&self, mut visitor: impl FnMut(&str, bool)) {
match self {
HeaderValue::Address(addr) => {
@@ -426,6 +446,19 @@ impl VisitAddresses for HeaderValue<'_> {
_ => (),
}
}
fn visit_text(&self, mut visitor: impl FnMut(&str)) {
match &self {
HeaderValue::Text(text) => {
visitor(text.as_ref());
}
HeaderValue::TextList(texts) => {
for text in texts {
visitor(text.as_ref());
}
}
_ => (),
}
}
}
pub trait TrimTextValue {

52
crates/jmap/src/email/ingest.rs
View File

@@ -9,9 +9,10 @@ use mail_parser::{
parsers::fields::thread::thread_name, HeaderName, HeaderValue, Message, RfcHeader,
};
use store::{
ahash::AHashSet,
query::Filter,
write::{log::ChangeLogBuilder, now, BatchBuilder, F_BITMAP, F_CLEAR, F_VALUE},
ValueKey,
BitmapKey, ValueKey,
};
use utils::map::vec_map::VecMap;
@@ -210,6 +211,7 @@ impl JMAP {
) -> Result<Option<u32>, MaybeError> {
let mut try_count = 0;
println!("-----------\nthread name: {:?}", thread_name);
loop {
// Find messages with matching references
let mut filters = Vec::with_capacity(references.len() + 3);
@@ -232,6 +234,9 @@ impl JMAP {
MaybeError::Temporary
})?
.results;
println!("found messages {:?}", results);
if results.is_empty() {
return Ok(None);
}
@@ -261,6 +266,7 @@ impl JMAP {
"Failed to obtain threadIds.");
MaybeError::Temporary
})?;
println!("found thread ids {:?}", thread_ids);
if thread_ids.len() == 1 {
return Ok(thread_ids.into_iter().next().unwrap());
}
@@ -277,6 +283,7 @@ impl JMAP {
thread_id = *thread_id_;
}
}
println!("common thread id {:?}", thread_id);
if thread_id == u32::MAX {
return Ok(None); // This should never happen
} else if thread_counts.len() == 1 {
@@ -310,19 +317,38 @@ impl JMAP {
// Move messages to the new threadId
batch.with_collection(Collection::Email);
for (document_id, old_thread_id) in results.iter().zip(thread_ids.into_iter()) {
let old_thread_id = old_thread_id.unwrap_or(u32::MAX);
for old_thread_id in thread_ids.into_iter().flatten().collect::<AHashSet<_>>() {
if thread_id != old_thread_id {
batch
.update_document(document_id)
.assert_value(Property::ThreadId, old_thread_id)
.value(Property::ThreadId, old_thread_id, F_BITMAP | F_CLEAR)
.value(Property::ThreadId, thread_id, F_VALUE | F_BITMAP);
changes.log_move(
Collection::Email,
Id::from_parts(old_thread_id, document_id),
Id::from_parts(thread_id, document_id),
)
for document_id in self
.store
.get_bitmap(BitmapKey::value(
account_id,
Collection::Email,
Property::ThreadId,
old_thread_id,
))
.await
.map_err(|err| {
tracing::error!(
event = "error",
context = "find_or_merge_thread",
error = ?err,
"Failed to obtain threadId bitmap.");
MaybeError::Temporary
})?
.unwrap_or_default()
{
batch
.update_document(document_id)
.assert_value(Property::ThreadId, old_thread_id)
.value(Property::ThreadId, old_thread_id, F_BITMAP | F_CLEAR)
.value(Property::ThreadId, thread_id, F_VALUE | F_BITMAP);
changes.log_move(
Collection::Email,
Id::from_parts(old_thread_id, document_id),
Id::from_parts(thread_id, document_id),
);
}
}
}
batch.custom(changes).map_err(|err| {

91
crates/jmap/src/email/query.rs
View File

@@ -4,8 +4,9 @@ use jmap_proto::{
object::email::QueryArguments,
types::{collection::Collection, keyword::Keyword, property::Property},
};
use mail_parser::{HeaderName, RfcHeader};
use store::{
fts::Language,
fts::{builder::MAX_TOKEN_LENGTH, Language},
query::{self, sort::Pagination},
roaring::RoaringBitmap,
ValueKey,
@@ -87,20 +88,20 @@ impl JMAP {
&text,
Language::None,
));
filters.push(query::Filter::has_text(
filters.push(query::Filter::has_text_detect(
Property::Subject,
&text,
Language::Unknown,
self.config.default_language,
));
filters.push(query::Filter::has_text(
filters.push(query::Filter::has_text_detect(
Property::TextBody,
&text,
Language::Unknown,
self.config.default_language,
));
filters.push(query::Filter::has_text(
filters.push(query::Filter::has_text_detect(
Property::Attachments,
text,
Language::Unknown,
self.config.default_language,
));
filters.push(query::Filter::End);
}
@@ -118,21 +119,78 @@ impl JMAP {
Filter::Bcc(text) => {
filters.push(query::Filter::has_text(Property::Bcc, text, Language::None))
}
Filter::Subject(text) => filters.push(query::Filter::has_text(
Filter::Subject(text) => filters.push(query::Filter::has_text_detect(
Property::Subject,
text,
Language::Unknown,
self.config.default_language,
)),
Filter::Body(text) => filters.push(query::Filter::has_text(
Filter::Body(text) => filters.push(query::Filter::has_text_detect(
Property::TextBody,
text,
Language::Unknown,
self.config.default_language,
)),
Filter::Header(header) => {
return Err(MethodError::InvalidArguments(format!(
"Querying headers '{}' is not supported.",
header.join(":")
)));
let mut header = header.into_iter();
let header_name = header.next().ok_or_else(|| {
MethodError::InvalidArguments("Header name is missing.".to_string())
})?;
if let Some(HeaderName::Rfc(header_name)) = HeaderName::parse(&header_name) {
let is_id = matches!(
header_name,
RfcHeader::MessageId
| RfcHeader::InReplyTo
| RfcHeader::References
| RfcHeader::ResentMessageId
);
let tokens = if let Some(header_value) = header.next() {
let header_num = u8::from(header_name).to_string();
header_value
.split_ascii_whitespace()
.filter_map(|token| {
if token.len() < MAX_TOKEN_LENGTH {
if is_id {
format!("{header_num}{token}")
} else {
format!("{header_num}{}", token.to_lowercase())
}
.into()
} else {
None
}
})
.collect::<Vec<_>>()
} else {
vec![]
};
match tokens.len() {
0 => {
filters.push(query::Filter::has_raw_text(
Property::Headers,
u8::from(header_name).to_string(),
));
}
1 => {
filters.push(query::Filter::has_raw_text(
Property::Headers,
tokens.into_iter().next().unwrap(),
));
}
_ => {
filters.push(query::Filter::And);
for token in tokens {
filters.push(query::Filter::has_raw_text(
Property::Headers,
token,
));
}
filters.push(query::Filter::End);
}
}
} else {
return Err(MethodError::InvalidArguments(format!(
"Querying non-RFC header '{header_name}' is not allowed.",
)));
};
}
// Non-standard
@@ -149,6 +207,9 @@ impl JMAP {
Property::ThreadId,
id.document_id(),
)),
Filter::And | Filter::Or | Filter::Not | Filter::Close => {
filters.push(cond.into());
}
other => return Err(MethodError::UnsupportedFilter(other.to_string())),
}