starred/stalwart
1
0
Fork 0
mirror of https://github.com/stalwartlabs/stalwart.git synced 2026-10-29 00:02:43 +00:00
Code Issues Packages Projects Releases Wiki Activity

Migration implementation (part 1)

Browse Source
This commit is contained in:
mdecimus
2025-12-10 19:13:04 +01:00
parent efd1d86ff3
commit f1fe0345ef
86 changed files with 2553 additions and 356 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+30 -29
View File
@@ -1147,7 +1147,7 @@ dependencies = [
[[package]]
name = "common"
version = "0.14.1"
version = "0.15.0"
dependencies = [
"aes-gcm-siv",
"ahash",
@@ -1637,7 +1637,7 @@ checksum = "2a2330da5de22e8a3cb63252ce2abb30116bf5265e89c0e01bc17015ce30a476"
[[package]]
name = "dav"
version = "0.14.1"
version = "0.15.0"
dependencies = [
"calcard 0.3.1",
"chrono",
@@ -1659,7 +1659,7 @@ dependencies = [
[[package]]
name = "dav-proto"
version = "0.14.1"
version = "0.15.0"
dependencies = [
"calcard 0.3.1",
"chrono",
@@ -1825,7 +1825,7 @@ dependencies = [
[[package]]
name = "directory"
version = "0.14.1"
version = "0.15.0"
dependencies = [
"ahash",
"argon2",
@@ -2078,7 +2078,7 @@ dependencies = [
[[package]]
name = "email"
version = "0.14.1"
version = "0.15.0"
dependencies = [
"aes",
"aes-gcm",
@@ -2198,7 +2198,7 @@ dependencies = [
[[package]]
name = "event_macro"
version = "0.14.1"
version = "0.15.0"
dependencies = [
"proc-macro2",
"quote",
@@ -2686,7 +2686,7 @@ dependencies = [
[[package]]
name = "groupware"
version = "0.14.1"
version = "0.15.0"
dependencies = [
"ahash",
"calcard 0.3.1",
@@ -2991,7 +2991,7 @@ dependencies = [
[[package]]
name = "http"
version = "0.14.1"
version = "0.15.0"
dependencies = [
"async-stream",
"base64 0.22.1",
@@ -3101,7 +3101,7 @@ dependencies = [
[[package]]
name = "http_proto"
version = "0.14.1"
version = "0.15.0"
dependencies = [
"common",
"compact_str",
@@ -3402,7 +3402,7 @@ checksum = "09e54e57b4c48b40f7aec75635392b12b3421fa26fe8b4332e63138ed278459c"
[[package]]
name = "imap"
version = "0.14.1"
version = "0.15.0"
dependencies = [
"ahash",
"common",
@@ -3429,7 +3429,7 @@ dependencies = [
[[package]]
name = "imap_proto"
version = "0.14.1"
version = "0.15.0"
dependencies = [
"ahash",
"chrono",
@@ -3690,7 +3690,7 @@ dependencies = [
[[package]]
name = "jmap"
version = "0.14.1"
version = "0.15.0"
dependencies = [
"aes-gcm",
"aes-gcm-siv",
@@ -3800,7 +3800,7 @@ dependencies = [
[[package]]
name = "jmap_proto"
version = "0.14.1"
version = "0.15.0"
dependencies = [
"ahash",
"calcard 0.3.1",
@@ -4260,7 +4260,7 @@ dependencies = [
[[package]]
name = "managesieve"
version = "0.14.1"
version = "0.15.0"
dependencies = [
"ahash",
"common",
@@ -4369,7 +4369,7 @@ checksum = "c797b9d6bb23aab2fc369c65f871be49214f5c759af65bde26ffaaa2b646b492"
[[package]]
name = "migration"
version = "0.14.1"
version = "0.15.0"
dependencies = [
"base64 0.22.1",
"bincode 1.3.3",
@@ -4391,6 +4391,7 @@ dependencies = [
"serde_json",
"sieve-rs",
"smtp",
"smtp-proto",
"store",
"tokio",
"trc",
@@ -4609,7 +4610,7 @@ dependencies = [
[[package]]
name = "nlp"
version = "0.14.1"
version = "0.15.0"
dependencies = [
"ahash",
"bincode 1.3.3",
@@ -5352,7 +5353,7 @@ dependencies = [
[[package]]
name = "pop3"
version = "0.14.1"
version = "0.15.0"
dependencies = [
"common",
"directory",
@@ -5551,7 +5552,7 @@ dependencies = [
[[package]]
name = "proc_macros"
version = "0.14.1"
version = "0.15.0"
dependencies = [
"proc-macro2",
"quote",
@@ -7100,7 +7101,7 @@ dependencies = [
[[package]]
name = "services"
version = "0.14.1"
version = "0.15.0"
dependencies = [
"aes-gcm",
"aes-gcm-siv",
@@ -7317,7 +7318,7 @@ checksum = "67b1b7a3b5fe4f1376887184045fcf45c69e92af734b7aaddc05fb777b6fbd03"
[[package]]
name = "smtp"
version = "0.14.1"
version = "0.15.0"
dependencies = [
"ahash",
"blake3",
@@ -7419,7 +7420,7 @@ dependencies = [
[[package]]
name = "spam-filter"
version = "0.14.1"
version = "0.15.0"
dependencies = [
"common",
"compact_str",
@@ -7517,14 +7518,14 @@ checksum = "6ce2be8dc25455e1f91df71bfa12ad37d7af1092ae736f3a6cd0e37bc7810596"
[[package]]
name = "stalwart"
version = "0.14.1"
version = "0.15.0"
dependencies = [
"common",
"dav",
"directory",
"email",
"groupware",
"http 0.14.1",
"http 0.15.0",
"imap",
"jemallocator",
"jmap",
@@ -7543,7 +7544,7 @@ dependencies = [
[[package]]
name = "stalwart-cli"
version = "0.14.1"
version = "0.15.0"
dependencies = [
"clap",
"console",
@@ -7574,7 +7575,7 @@ checksum = "a2eb9349b6444b326872e140eb1cf5e7c522154d69e7a0ffb0fb81c06b37543f"
[[package]]
name = "store"
version = "0.14.1"
version = "0.15.0"
dependencies = [
"ahash",
"arc-swap",
@@ -7759,7 +7760,7 @@ dependencies = [
"form_urlencoded",
"futures",
"groupware",
"http 0.14.1",
"http 0.15.0",
"http-body-util",
"http_proto",
"hyper 1.8.1",
@@ -8294,7 +8295,7 @@ dependencies = [
[[package]]
name = "trc"
version = "0.14.1"
version = "0.15.0"
dependencies = [
"ahash",
"base64 0.22.1",
@@ -8375,7 +8376,7 @@ checksum = "562d481066bde0658276a35467c4af00bdc6ee726305698a55b86e61d7ad82bb"
[[package]]
name = "types"
version = "0.14.1"
version = "0.15.0"
dependencies = [
"blake3",
"compact_str",
@@ -8572,7 +8573,7 @@ checksum = "06abde3611657adf66d383f00b093d7faecc7fa57071cce2578660c9f1010821"
[[package]]
name = "utils"
version = "0.14.1"
version = "0.15.0"
dependencies = [
"ahash",
"base64 0.22.1",
crates/cli/Cargo.toml
+1 -1
View File
@@ -5,7 +5,7 @@ authors = ["Stalwart Labs LLC <hello@stalw.art>"]
license = "AGPL-3.0-only OR LicenseRef-SEL"
repository = "https://github.com/stalwartlabs/cli"
homepage = "https://github.com/stalwartlabs/cli"
version = "0.14.1"
version = "0.15.0"
edition = "2024"
readme = "README.md"
crates/common/Cargo.toml
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "common"
version = "0.14.1"
version = "0.15.0"
edition = "2024"
build = "build.rs"
crates/common/src/config/spamfilter.rs
-4
View File
@@ -82,7 +82,6 @@ pub enum SpamFilterAction<T> {
pub struct ClassifierConfig {
pub w_params: FtrlParameters,
pub i_params: Option<FtrlParameters>,
pub num_epochs: usize,
pub reservoir_capacity: usize,
pub min_ham_samples: u64,
pub min_spam_samples: u64,
@@ -474,9 +473,6 @@ impl ClassifierConfig {
ClassifierConfig {
w_params,
i_params,
num_epochs: config
.property_or_default("spam-filter.classifier.training.epochs", "3")
.unwrap_or(3),
reservoir_capacity: config
.property_or_default("spam-filter.classifier.reservoir-capacity", "1024")
.unwrap_or(1024),
crates/common/src/ipc.rs
+44 -2
View File
@@ -15,9 +15,15 @@ use mail_auth::{
mta_sts::TlsRpt,
report::{Record, tlsrpt::FailureDetails},
};
use std::{sync::Arc, time::Instant};
use std::{
sync::{
Arc,
atomic::{AtomicBool, Ordering},
},
time::Instant,
};
use store::{BlobStore, InMemoryStore, Store};
use tokio::sync::mpsc;
use tokio::sync::{Semaphore, SemaphorePermit, mpsc};
use types::type_state::{DataType, StateChange};
use utils::map::bitmap::Bitmap;
@@ -154,6 +160,42 @@ pub enum PolicyType {
None,
}
pub struct TrainTaskController {
semaphore: Semaphore,
stop_flag: AtomicBool,
}
impl Default for TrainTaskController {
fn default() -> Self {
Self {
semaphore: Semaphore::new(1),
stop_flag: AtomicBool::new(false),
}
}
}
impl TrainTaskController {
pub fn try_run(&self) -> Option<SemaphorePermit<'_>> {
let permit = self.semaphore.try_acquire().ok()?;
self.stop_flag.store(false, Ordering::SeqCst);
Some(permit)
}
pub fn is_running(&self) -> bool {
self.semaphore.available_permits() == 0
}
pub fn stop(&self) {
self.stop_flag.store(true, Ordering::SeqCst);
}
pub fn should_stop(&self) -> bool {
self.stop_flag.load(Ordering::SeqCst)
}
}
pub trait ToHash {
fn to_hash(&self) -> u64;
}
crates/common/src/lib.rs
+5 -2
View File
@@ -72,7 +72,7 @@ pub mod enterprise;
pub use psl;
use crate::config::spamfilter::SpamClassifier;
use crate::{config::spamfilter::SpamClassifier, ipc::TrainTaskController};
pub static VERSION_PRIVATE: &str = env!("CARGO_PKG_VERSION");
pub static VERSION_PUBLIC: &str = "1.0.0";
@@ -89,10 +89,11 @@ Schema history:
2 - v0.12.4
3 - v0.13.0
4 - v0.14.0
5 - v0.15.0
*/
pub const DATABASE_SCHEMA_VERSION: u32 = 4;
pub const DATABASE_SCHEMA_VERSION: u32 = 5;
pub const LONG_1D_SLUMBER: Duration = Duration::from_secs(60 * 60 * 24);
pub const LONG_1Y_SLUMBER: Duration = Duration::from_secs(60 * 60 * 24 * 365);
@@ -246,6 +247,7 @@ pub struct Ipc {
pub queue_tx: mpsc::Sender<QueueEvent>,
pub report_tx: mpsc::Sender<ReportingEvent>,
pub broadcast_tx: Option<mpsc::Sender<BroadcastEvent>>,
pub train_task_controller: Arc<TrainTaskController>,
}
pub struct TlsConnectors {
@@ -504,6 +506,7 @@ impl Default for Ipc {
queue_tx: mpsc::channel(IPC_CHANNEL_BUFFER).0,
report_tx: mpsc::channel(IPC_CHANNEL_BUFFER).0,
broadcast_tx: None,
train_task_controller: Arc::new(TrainTaskController::default()),
}
}
}
crates/common/src/manager/boot.rs
+5 -1
View File
@@ -14,7 +14,10 @@ use crate::{
Caches, Core, Data, IPC_CHANNEL_BUFFER, Inner, Ipc,
config::{network::AsnGeoLookupConfig, server::Listeners, telemetry::Telemetry},
core::BuildServer,
ipc::{BroadcastEvent, HousekeeperEvent, PushEvent, QueueEvent, ReportingEvent},
ipc::{
BroadcastEvent, HousekeeperEvent, PushEvent, QueueEvent, ReportingEvent,
TrainTaskController,
},
};
use arc_swap::ArcSwap;
use pwhash::sha512_crypt;
@@ -562,6 +565,7 @@ pub fn build_ipc(has_pubsub: bool) -> (Ipc, IpcReceivers) {
report_tx,
broadcast_tx: has_pubsub.then_some(broadcast_tx),
task_tx: Arc::new(Notify::new()),
train_task_controller: Arc::new(TrainTaskController::default()),
},
IpcReceivers {
push_rx: Some(push_rx),
crates/dav-proto/Cargo.toml
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "dav-proto"
version = "0.14.1"
version = "0.15.0"
edition = "2024"
[dependencies]
crates/dav/Cargo.toml
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "dav"
version = "0.14.1"
version = "0.15.0"
edition = "2024"
[dependencies]
crates/directory/Cargo.toml
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "directory"
version = "0.14.1"
version = "0.15.0"
edition = "2024"
[dependencies]
crates/email/Cargo.toml
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "email"
version = "0.14.1"
version = "0.15.0"
edition = "2024"
[dependencies]
crates/email/src/mailbox/manage.rs
+5 -1
View File
@@ -47,7 +47,11 @@ impl MailboxFnc for Server {
SpecialUse::Drafts => DRAFTS_ID,
SpecialUse::Sent => SENT_ID,
SpecialUse::Archive => ARCHIVE_ID,
SpecialUse::None | SpecialUse::Important => {
SpecialUse::None
| SpecialUse::Important
| SpecialUse::Memos
| SpecialUse::Scheduled
| SpecialUse::Snoozed => {
last_document_id += 1;
last_document_id
}
crates/email/src/message/crypto.rs
+2
View File
@@ -104,10 +104,12 @@ pub enum EncryptionType {
PGP {
algo: Algorithm,
certs: String,
allow_spam_training: bool,
},
SMIME {
algo: Algorithm,
certs: String,
allow_spam_training: bool,
},
#[default]
Disabled,
crates/email/src/message/ingest.rs
+1 -1
View File
@@ -993,7 +993,7 @@ impl std::hash::Hash for MergeThreadIds<AHashSet<u32>> {
}
}
pub(crate) struct ThreadInfo;
pub struct ThreadInfo;
impl ThreadInfo {
pub fn serialize(thread_id: u32, ref_ids: &[CheekyHash]) -> Vec<u8> {
crates/email/src/message/metadata.rs
+128 -1
View File
@@ -127,6 +127,53 @@ pub enum MetadataHeaderName {
ArcAuthenticationResults,
ArcMessageSignature,
ArcSeal,
// Delivery/Routing
DeliveredTo,
XOriginalTo,
ReturnReceiptTo,
DispositionNotificationTo,
ErrorsTo,
// Authentication
AuthenticationResults,
ReceivedSpf,
// Spam/Virus
XSpamStatus,
XSpamScore,
XSpamFlag,
XSpamResult,
// Priority
Importance,
Priority,
XPriority,
XMSMailPriority,
// Client/Agent
XMailer,
UserAgent,
XMimeOLE,
// Network/Origin
XOriginatingIp,
XForwardedTo,
XForwardedFor,
// Auto-response
AutoSubmitted,
XAutoResponseSuppress,
Precedence,
// Organization/Threading
Organization,
ThreadIndex,
ThreadTopic,
// List (additional)
ListUnsubscribePost,
FeedbackId,
}
#[derive(Debug, PartialEq, Eq, Clone, rkyv::Serialize, rkyv::Deserialize, rkyv::Archive)]
@@ -691,7 +738,57 @@ impl From<HeaderName<'_>> for MetadataHeaderName {
HeaderName::ArcMessageSignature => MetadataHeaderName::ArcMessageSignature,
HeaderName::ArcSeal => MetadataHeaderName::ArcSeal,
HeaderName::Other(value) => {
MetadataHeaderName::Other(value.into_owned().into_boxed_str())
let name = hashify::tiny_map_ignore_case!(value.as_bytes(),
// Delivery/Routing
"Delivered-To" => MetadataHeaderName::DeliveredTo,
"X-Original-To" => MetadataHeaderName::XOriginalTo,
"Return-Receipt-To" => MetadataHeaderName::ReturnReceiptTo,
"Disposition-Notification-To" => MetadataHeaderName::DispositionNotificationTo,
"Errors-To" => MetadataHeaderName::ErrorsTo,
// Authentication
"Authentication-Results" => MetadataHeaderName::AuthenticationResults,
"Received-SPF" => MetadataHeaderName::ReceivedSpf,
// Spam/Virus
"X-Spam-Status" => MetadataHeaderName::XSpamStatus,
"X-Spam-Score" => MetadataHeaderName::XSpamScore,
"X-Spam-Flag" => MetadataHeaderName::XSpamFlag,
"X-Spam-Result" => MetadataHeaderName::XSpamResult,
// Priority
"Importance" => MetadataHeaderName::Importance,
"Priority" => MetadataHeaderName::Priority,
"X-Priority" => MetadataHeaderName::XPriority,
"X-MSMail-Priority" => MetadataHeaderName::XMSMailPriority,
// Client/Agent
"X-Mailer" => MetadataHeaderName::XMailer,
"User-Agent" => MetadataHeaderName::UserAgent,
"X-MimeOLE" => MetadataHeaderName::XMimeOLE,
// Network/Origin
"X-Originating-IP" => MetadataHeaderName::XOriginatingIp,
"X-Forwarded-To" => MetadataHeaderName::XForwardedTo,
"X-Forwarded-For" => MetadataHeaderName::XForwardedFor,
// Auto-response
"Auto-Submitted" => MetadataHeaderName::AutoSubmitted,
"X-Auto-Response-Suppress" => MetadataHeaderName::XAutoResponseSuppress,
"Precedence" => MetadataHeaderName::Precedence,
// Organization/Threading
"Organization" => MetadataHeaderName::Organization,
"Thread-Index" => MetadataHeaderName::ThreadIndex,
"Thread-Topic" => MetadataHeaderName::ThreadTopic,
// List (additional)
"List-Unsubscribe-Post" => MetadataHeaderName::ListUnsubscribePost,
"Feedback-ID" => MetadataHeaderName::FeedbackId,
);
name.unwrap_or_else(|| {
MetadataHeaderName::Other(value.into_owned().into_boxed_str())
})
}
other => MetadataHeaderName::Other(other.as_str().to_string().into_boxed_str()),
}
@@ -787,6 +884,7 @@ impl ArchivedMessageMetadataContents {
}
}
#[derive(Default)]
pub struct MessageDataBuilder {
pub mailboxes: Vec<UidMailbox>,
pub keywords: Vec<Keyword>,
@@ -1086,6 +1184,35 @@ impl ArchivedMetadataHeaderName {
ArchivedMetadataHeaderName::ArcMessageSignature => "ARC-Message-Signature",
ArchivedMetadataHeaderName::ArcSeal => "ARC-Seal",
ArchivedMetadataHeaderName::DkimSignature => "DKIM-Signature",
ArchivedMetadataHeaderName::DeliveredTo => "Delivered-To",
ArchivedMetadataHeaderName::XOriginalTo => "X-Original-To",
ArchivedMetadataHeaderName::ReturnReceiptTo => "Return-Receipt-To",
ArchivedMetadataHeaderName::DispositionNotificationTo => "Disposition-Notification-To",
ArchivedMetadataHeaderName::ErrorsTo => "Errors-To",
ArchivedMetadataHeaderName::AuthenticationResults => "Authentication-Results",
ArchivedMetadataHeaderName::ReceivedSpf => "Received-SPF",
ArchivedMetadataHeaderName::XSpamStatus => "X-Spam-Status",
ArchivedMetadataHeaderName::XSpamScore => "X-Spam-Score",
ArchivedMetadataHeaderName::XSpamFlag => "X-Spam-Flag",
ArchivedMetadataHeaderName::XSpamResult => "X-Spam-Result",
ArchivedMetadataHeaderName::Importance => "Importance",
ArchivedMetadataHeaderName::Priority => "Priority",
ArchivedMetadataHeaderName::XPriority => "X-Priority",
ArchivedMetadataHeaderName::XMSMailPriority => "X-MSMail-Priority",
ArchivedMetadataHeaderName::XMailer => "X-Mailer",
ArchivedMetadataHeaderName::UserAgent => "User-Agent",
ArchivedMetadataHeaderName::XMimeOLE => "X-MimeOLE",
ArchivedMetadataHeaderName::XOriginatingIp => "X-Originating-IP",
ArchivedMetadataHeaderName::XForwardedTo => "X-Forwarded-To",
ArchivedMetadataHeaderName::XForwardedFor => "X-Forwarded-For",
ArchivedMetadataHeaderName::AutoSubmitted => "Auto-Submitted",
ArchivedMetadataHeaderName::XAutoResponseSuppress => "X-Auto-Response-Suppress",
ArchivedMetadataHeaderName::Precedence => "Precedence",
ArchivedMetadataHeaderName::Organization => "Organization",
ArchivedMetadataHeaderName::ThreadIndex => "Thread-Index",
ArchivedMetadataHeaderName::ThreadTopic => "Thread-Topic",
ArchivedMetadataHeaderName::ListUnsubscribePost => "List-Unsubscribe-Post",
ArchivedMetadataHeaderName::FeedbackId => "Feedback-ID",
ArchivedMetadataHeaderName::Other(name) => name.as_ref(),
}
}
crates/groupware/Cargo.toml
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "groupware"
version = "0.14.1"
version = "0.15.0"
edition = "2024"
[dependencies]
crates/http-proto/Cargo.toml
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "http_proto"
version = "0.14.1"
version = "0.15.0"
edition = "2024"
[dependencies]
crates/http/Cargo.toml
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "http"
version = "0.14.1"
version = "0.15.0"
edition = "2024"
[dependencies]
crates/http/src/management/crypto.rs
+31 -9
View File
@@ -7,8 +7,8 @@
use common::{Server, auth::AccessToken};
use directory::backend::internal::manage;
use email::message::crypto::{
EncryptMessage, EncryptMessageError, EncryptionMethod, EncryptionParams, EncryptionType,
try_parse_certs,
ENCRYPT_TRAIN_SPAM_FILTER, EncryptMessage, EncryptMessageError, EncryptionMethod,
EncryptionParams, EncryptionType, try_parse_certs,
};
use http_proto::*;
use mail_builder::encoders::base64::base64_encode_mime;
@@ -52,6 +52,7 @@ impl CryptoHandler for Server {
.caused_by(trc::location!())?;
let algo = params.algo();
let method = params.method();
let allow_spam_training = params.can_train_spam_filter();
let mut certs = Vec::new();
certs.extend_from_slice(b"-----STALWART CERTIFICATE-----\r\n");
let _ = base64_encode_mime(&params_.into_inner(), &mut certs, false);
@@ -59,8 +60,16 @@ impl CryptoHandler for Server {
let certs = String::from_utf8(certs).unwrap_or_default();
match method {
EncryptionMethod::PGP => EncryptionType::PGP { algo, certs },
EncryptionMethod::SMIME => EncryptionType::SMIME { algo, certs },
EncryptionMethod::PGP => EncryptionType::PGP {
algo,
certs,
allow_spam_training,
},
EncryptionMethod::SMIME => EncryptionType::SMIME {
algo,
certs,
allow_spam_training,
},
}
} else {
EncryptionType::Disabled
@@ -80,9 +89,17 @@ impl CryptoHandler for Server {
let request = serde_json::from_slice::<EncryptionType>(body.as_deref().unwrap_or_default())
.map_err(|err| trc::ResourceEvent::BadParameters.into_err().reason(err))?;
let (method, algo, mut certs) = match request {
EncryptionType::PGP { algo, certs } => (EncryptionMethod::PGP, algo, certs),
EncryptionType::SMIME { algo, certs } => (EncryptionMethod::SMIME, algo, certs),
let (method, algo, mut certs, allow_spam_training) = match request {
EncryptionType::PGP {
algo,
certs,
allow_spam_training,
} => (EncryptionMethod::PGP, algo, certs, allow_spam_training),
EncryptionType::SMIME {
algo,
certs,
allow_spam_training,
} => (EncryptionMethod::SMIME, algo, certs, allow_spam_training),
EncryptionType::Disabled => {
// Disable encryption at rest
let mut batch = BatchBuilder::new();
@@ -110,12 +127,17 @@ impl CryptoHandler for Server {
}
// Parse certificates
let todo = "fetch privacy spam train";
let certs = try_parse_certs(method, certs.into_bytes())
.map_err(|err| manage::error(err, None::<u32>))?;
let num_certs = certs.len();
let params = Archiver::new(EncryptionParams {
flags: method.flags() | algo.flags(),
flags: method.flags()
| algo.flags()
| if allow_spam_training {
ENCRYPT_TRAIN_SPAM_FILTER
} else {
0
},
certs,
})
.serialize()
crates/http/src/management/spam.rs
+32 -3
View File
@@ -21,6 +21,7 @@ use serde_json::json;
use spam_filter::{
SpamFilterInput,
analysis::{init::SpamFilterInit, score::SpamFilterAnalyzeScore},
modules::classifier::SpamClassifier,
};
use std::future::Future;
use std::net::IpAddr;
@@ -137,14 +138,42 @@ impl ManageSpamHandler for Server {
}))
.into_http_response())
}
(Some("train"), _, &Method::GET) => {
(Some("train"), request, &Method::GET) => {
// Validate the access token
access_token.assert_has_permission(Permission::SpamFilterTrain)?;
let todo = "implement";
let result = match request {
Some("start") | Some("reset") => {
if !self.inner.ipc.train_task_controller.is_running() {
let reset = matches!(request, Some("reset"));
let server = self.clone();
tokio::spawn(async move {
if let Err(err) = server.spam_train(reset).await {
trc::error!(err.caused_by(trc::location!()));
}
});
true
} else {
false
}
}
Some("stop") => {
if self.inner.ipc.train_task_controller.is_running() {
self.inner.ipc.train_task_controller.stop();
true
} else {
false
}
}
Some("status") => self.inner.ipc.train_task_controller.is_running(),
_ => {
return Err(trc::ResourceEvent::NotFound.into_err());
}
};
Ok(JsonResponse::new(json!({
"data": (),
"data": result,
}))
.into_http_response())
}
crates/imap-proto/Cargo.toml
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "imap_proto"
version = "0.14.1"
version = "0.15.0"
edition = "2024"
[dependencies]
crates/imap-proto/src/parser/create.rs
+3
View File
@@ -53,6 +53,9 @@ impl Request<Command> {
"\\Sent" => Some(Attribute::Sent),
"\\Trash" => Some(Attribute::Trash),
"\\Important" => Some(Attribute::Important),
"\\Memos" => Some(Attribute::Memos),
"\\Scheduled" => Some(Attribute::Scheduled),
"\\Snoozed" => Some(Attribute::Snoozed),
"\\All" => None,
);
crates/imap-proto/src/parser/mod.rs
+34
View File
@@ -106,6 +106,23 @@ impl Flag {
"$NotJunk" => Flag::NotJunk,
"$Phishing" => Flag::Phishing,
"$Important" => Flag::Important,
"$autosent" => Flag::Autosent,
"$canunsubscribe" => Flag::CanUnsubscribe,
"$followed" => Flag::Followed,
"$hasattachment" => Flag::HasAttachment,
"$hasmemo" => Flag::HasMemo,
"$hasnoattachment" => Flag::HasNoAttachment,
"$imported" => Flag::Imported,
"$istrusted" => Flag::IsTrusted,
"$MailFlagBit0" => Flag::MailFlagBit0,
"$MailFlagBit1" => Flag::MailFlagBit1,
"$MailFlagBit2" => Flag::MailFlagBit2,
"$maskedemail" => Flag::MaskedEmail,
"$memo" => Flag::Memo,
"$muted" => Flag::Muted,
"$new" => Flag::New,
"$notify" => Flag::Notify,
"$unsubscribed" => Flag::Unsubscribed,
);
if let Some(flag) = flag {
@@ -135,6 +152,23 @@ impl Flag {
"$deleted" => Flag::Deleted,
"$forwarded" => Flag::Forwarded,
"$mdnsent" => Flag::MDNSent,
"$autosent" => Flag::Autosent,
"$canunsubscribe" => Flag::CanUnsubscribe,
"$followed" => Flag::Followed,
"$hasattachment" => Flag::HasAttachment,
"$hasmemo" => Flag::HasMemo,
"$hasnoattachment" => Flag::HasNoAttachment,
"$imported" => Flag::Imported,
"$istrusted" => Flag::IsTrusted,
"$MailFlagBit0" => Flag::MailFlagBit0,
"$MailFlagBit1" => Flag::MailFlagBit1,
"$MailFlagBit2" => Flag::MailFlagBit2,
"$maskedemail" => Flag::MaskedEmail,
"$memo" => Flag::Memo,
"$muted" => Flag::Muted,
"$new" => Flag::New,
"$notify" => Flag::Notify,
"$unsubscribed" => Flag::Unsubscribed,
)
.unwrap_or_else(|| Flag::Keyword(value.into_boxed_str()))
} else {
crates/imap-proto/src/protocol/list.rs
+9
View File
@@ -71,6 +71,9 @@ pub enum Attribute {
Sent,
Trash,
Important,
Memos,
Scheduled,
Snoozed,
}
#[derive(Debug, Clone, PartialEq, Eq)]
@@ -142,6 +145,9 @@ impl Attribute {
Attribute::Sent => b"\\Sent",
Attribute::Trash => b"\\Trash",
Attribute::Important => b"\\Important",
Attribute::Memos => b"\\Memos",
Attribute::Scheduled => b"\\Scheduled",
Attribute::Snoozed => b"\\Snoozed",
});
}
}
@@ -157,6 +163,9 @@ impl TryFrom<&str> for Attribute {
"sent" => Attribute::Sent,
"trash" => Attribute::Trash,
"important" => Attribute::Important,
"memos" => Attribute::Memos,
"scheduled" => Attribute::Scheduled,
"snoozed" => Attribute::Snoozed,
)
.ok_or(())
}
crates/imap-proto/src/protocol/mod.rs
+85
View File
@@ -246,6 +246,23 @@ pub enum Flag {
Deleted,
Forwarded,
MDNSent,
Autosent,
CanUnsubscribe,
Followed,
HasAttachment,
HasMemo,
HasNoAttachment,
Imported,
IsTrusted,
MailFlagBit0,
MailFlagBit1,
MailFlagBit2,
MaskedEmail,
Memo,
Muted,
New,
Notify,
Unsubscribed,
Keyword(Box<str>),
}
@@ -264,6 +281,23 @@ impl Flag {
Flag::Deleted => b"\\Deleted",
Flag::Forwarded => b"$Forwarded",
Flag::MDNSent => b"$MDNSent",
Flag::Autosent => b"$autosent",
Flag::CanUnsubscribe => b"$canunsubscribe",
Flag::Followed => b"$followed",
Flag::HasAttachment => b"$hasattachment",
Flag::HasMemo => b"$hasmemo",
Flag::HasNoAttachment => b"$hasnoattachment",
Flag::Imported => b"$imported",
Flag::IsTrusted => b"$istrusted",
Flag::MailFlagBit0 => b"$MailFlagBit0",
Flag::MailFlagBit1 => b"$MailFlagBit1",
Flag::MailFlagBit2 => b"$MailFlagBit2",
Flag::MaskedEmail => b"$maskedemail",
Flag::Memo => b"$memo",
Flag::Muted => b"$muted",
Flag::New => b"$new",
Flag::Notify => b"$notify",
Flag::Unsubscribed => b"$unsubscribed",
Flag::Keyword(keyword) => keyword.as_bytes(),
});
}
@@ -284,6 +318,23 @@ impl From<Keyword> for Flag {
Keyword::Deleted => Flag::Deleted,
Keyword::Forwarded => Flag::Forwarded,
Keyword::MdnSent => Flag::MDNSent,
Keyword::Autosent => Flag::Autosent,
Keyword::CanUnsubscribe => Flag::CanUnsubscribe,
Keyword::Followed => Flag::Followed,
Keyword::HasAttachment => Flag::HasAttachment,
Keyword::HasMemo => Flag::HasMemo,
Keyword::HasNoAttachment => Flag::HasNoAttachment,
Keyword::Imported => Flag::Imported,
Keyword::IsTrusted => Flag::IsTrusted,
Keyword::MailFlagBit0 => Flag::MailFlagBit0,
Keyword::MailFlagBit1 => Flag::MailFlagBit1,
Keyword::MailFlagBit2 => Flag::MailFlagBit2,
Keyword::MaskedEmail => Flag::MaskedEmail,
Keyword::Memo => Flag::Memo,
Keyword::Muted => Flag::Muted,
Keyword::New => Flag::New,
Keyword::Notify => Flag::Notify,
Keyword::Unsubscribed => Flag::Unsubscribed,
Keyword::Other(value) => Flag::Keyword(value),
}
}
@@ -304,6 +355,23 @@ impl From<&ArchivedKeyword> for Flag {
ArchivedKeyword::Deleted => Flag::Deleted,
ArchivedKeyword::Forwarded => Flag::Forwarded,
ArchivedKeyword::MdnSent => Flag::MDNSent,
ArchivedKeyword::Autosent => Flag::Autosent,
ArchivedKeyword::CanUnsubscribe => Flag::CanUnsubscribe,
ArchivedKeyword::Followed => Flag::Followed,
ArchivedKeyword::HasAttachment => Flag::HasAttachment,
ArchivedKeyword::HasMemo => Flag::HasMemo,
ArchivedKeyword::HasNoAttachment => Flag::HasNoAttachment,
ArchivedKeyword::Imported => Flag::Imported,
ArchivedKeyword::IsTrusted => Flag::IsTrusted,
ArchivedKeyword::MailFlagBit0 => Flag::MailFlagBit0,
ArchivedKeyword::MailFlagBit1 => Flag::MailFlagBit1,
ArchivedKeyword::MailFlagBit2 => Flag::MailFlagBit2,
ArchivedKeyword::MaskedEmail => Flag::MaskedEmail,
ArchivedKeyword::Memo => Flag::Memo,
ArchivedKeyword::Muted => Flag::Muted,
ArchivedKeyword::New => Flag::New,
ArchivedKeyword::Notify => Flag::Notify,
ArchivedKeyword::Unsubscribed => Flag::Unsubscribed,
ArchivedKeyword::Other(value) => Flag::Keyword(value.as_ref().into()),
}
}
@@ -324,6 +392,23 @@ impl From<Flag> for Keyword {
Flag::Deleted => Keyword::Deleted,
Flag::Forwarded => Keyword::Forwarded,
Flag::MDNSent => Keyword::MdnSent,
Flag::Autosent => Keyword::Autosent,
Flag::CanUnsubscribe => Keyword::CanUnsubscribe,
Flag::Followed => Keyword::Followed,
Flag::HasAttachment => Keyword::HasAttachment,
Flag::HasMemo => Keyword::HasMemo,
Flag::HasNoAttachment => Keyword::HasNoAttachment,
Flag::Imported => Keyword::Imported,
Flag::IsTrusted => Keyword::IsTrusted,
Flag::MailFlagBit0 => Keyword::MailFlagBit0,
Flag::MailFlagBit1 => Keyword::MailFlagBit1,
Flag::MailFlagBit2 => Keyword::MailFlagBit2,
Flag::MaskedEmail => Keyword::MaskedEmail,
Flag::Memo => Keyword::Memo,
Flag::Muted => Keyword::Muted,
Flag::New => Keyword::New,
Flag::Notify => Keyword::Notify,
Flag::Unsubscribed => Keyword::Unsubscribed,
Flag::Keyword(value) => Keyword::from_boxed_other(value),
}
}
crates/imap/Cargo.toml
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "imap"
version = "0.14.1"
version = "0.15.0"
edition = "2024"
[dependencies]
crates/imap/src/core/mailbox.rs
+3
View File
@@ -176,6 +176,9 @@ impl<T: SessionStream> SessionData<T> {
SpecialUse::Archive => Some(Attribute::Archive),
SpecialUse::Sent => Some(Attribute::Sent),
SpecialUse::Important => Some(Attribute::Important),
SpecialUse::Memos => Some(Attribute::Memos),
SpecialUse::Scheduled => Some(Attribute::Scheduled),
SpecialUse::Snoozed => Some(Attribute::Snoozed),
_ => None,
},
total_messages: cache.in_mailbox(mailbox.document_id).count() as u64,
crates/imap/src/op/create.rs
+3
View File
@@ -329,6 +329,9 @@ fn attr_to_role(attr: Attribute) -> SpecialUse {
Attribute::Sent => SpecialUse::Sent,
Attribute::Trash => SpecialUse::Trash,
Attribute::Important => SpecialUse::Important,
Attribute::Memos => SpecialUse::Memos,
Attribute::Scheduled => SpecialUse::Scheduled,
Attribute::Snoozed => SpecialUse::Snoozed,
_ => SpecialUse::None,
}
}
crates/jmap-proto/Cargo.toml
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "jmap_proto"
version = "0.14.1"
version = "0.15.0"
edition = "2024"
[dependencies]
crates/jmap/Cargo.toml
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "jmap"
version = "0.14.1"
version = "0.15.0"
edition = "2024"
[dependencies]
crates/main/Cargo.toml
+1 -1
View File
@@ -7,7 +7,7 @@ homepage = "https://stalw.art"
keywords = ["imap", "jmap", "smtp", "email", "mail", "webdav", "server"]
categories = ["email"]
license = "AGPL-3.0-only OR LicenseRef-SEL"
version = "0.14.1"
version = "0.15.0"
edition = "2024"
[[bin]]
crates/managesieve/Cargo.toml
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "managesieve"
version = "0.14.1"
version = "0.15.0"
edition = "2024"
[dependencies]
crates/migration/Cargo.toml
+2 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "migration"
version = "0.14.1"
version = "0.15.0"
edition = "2024"
[dependencies]
@@ -18,6 +18,7 @@ dav-proto = { path = "../dav-proto" }
proc_macros = { path = "../utils/proc-macros" }
mail-parser = { version = "0.11", features = ["full_encoding"] }
mail-auth = { version = "0.7.1", features = ["rkyv"] }
smtp-proto = { version = "0.2", features = ["rkyv", "serde"] }
sieve-rs = { version = "0.7", features = ["rkyv"] }
calcard_latest = { package = "calcard", version = "0.3", features = ["rkyv"] }
calcard_v01 = { package = "calcard", version = "0.1", features = ["rkyv"] }
crates/migration/src/addressbook_v2.rs
+5 -4
View File
@@ -7,12 +7,14 @@
use common::Server;
use groupware::contact::{AddressBook, AddressBookPreferences};
use store::{
Serialize,
write::{Archiver, BatchBuilder, serialize::rkyv_deserialize},
Serialize, ValueKey,
write::{AlignedBytes, Archive, Archiver, BatchBuilder, serialize::rkyv_deserialize},
};
use trc::AddContext;
use types::{acl::AclGrant, collection::Collection, dead_property::DeadProperty, field::Field};
use crate::get_document_ids;
#[derive(
rkyv::Archive, rkyv::Deserialize, rkyv::Serialize, Debug, Default, Clone, PartialEq, Eq,
)]
@@ -31,8 +33,7 @@ pub struct AddressBookV2 {
}
pub(crate) async fn migrate_addressbook_v013(server: &Server, account_id: u32) -> trc::Result<u64> {
let document_ids = server
.get_document_ids(account_id, Collection::AddressBook)
let document_ids = get_document_ids(server, account_id, Collection::AddressBook)
.await
.caused_by(trc::location!())?
.unwrap_or_default();
crates/migration/src/blob.rs
+271
View File
@@ -0,0 +1,271 @@
/*
* SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <hello@stalw.art>
*
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
*/
use common::Server;
use store::{
IterateParams, SUBSPACE_BLOB_LINK, Serialize, SerializeInfallible, U32_LEN, U64_LEN, ValueKey,
write::{
AnyClass, Archiver, BatchBuilder, BlobLink, BlobOp, ValueClass, key::DeserializeBigEndian,
now,
},
};
use trc::AddContext;
use types::blob_hash::{BLOB_HASH_LEN, BlobHash};
const SUBSPACE_BLOB_RESERVE: u8 = b'j';
pub(crate) async fn migrate_blobs_v014(server: &Server) -> trc::Result<()> {
for byte in 0..=u8::MAX {
// Validate linked blobs
let mut from_hash = BlobHash::default();
let mut to_hash = BlobHash::new_max();
from_hash.0[0] = byte;
to_hash.0[0] = byte;
let from_key = ValueKey {
account_id: 0,
collection: 0,
document_id: 0,
class: ValueClass::Blob(BlobOp::Commit { hash: from_hash }),
};
let to_key = ValueKey {
account_id: u32::MAX,
collection: u8::MAX,
document_id: u32::MAX,
class: ValueClass::Blob(BlobOp::Link {
hash: to_hash,
to: BlobLink::Document,
}),
};
let mut keys = Vec::new();
server
.store()
.iterate(
IterateParams::new(from_key, to_key).ascending().no_values(),
|key, value| {
if key.len() == BLOB_HASH_LEN + U64_LEN + 1 {
let hash =
BlobHash::try_from_hash_slice(key.get(0..BLOB_HASH_LEN).ok_or_else(
|| trc::Error::corrupted_key(key, value.into(), trc::location!()),
)?)
.unwrap();
let account_id = key.deserialize_be_u32(BLOB_HASH_LEN)?;
let document_id = key.deserialize_be_u32(BLOB_HASH_LEN + U32_LEN + 1)?;
let collection = key[BLOB_HASH_LEN + U32_LEN];
if account_id == u32::MAX && document_id == u32::MAX && collection == 0 {
keys.push((key.to_vec(), BlobOp::Commit { hash }));
} else if collection == u8::MAX {
keys.push((
key.to_vec(),
BlobOp::Link {
hash,
to: BlobLink::Id {
id: ((account_id as u64) << 32) | document_id as u64,
},
},
));
}
}
Ok(true)
},
)
.await
.caused_by(trc::location!())?;
let mut batch = BatchBuilder::new();
for (key, op) in keys {
batch
.clear(ValueClass::Any(AnyClass {
subspace: SUBSPACE_BLOB_LINK,
key,
}))
.set(op, vec![]);
if batch.is_large_batch() {
server
.store()
.write(batch.build_all())
.await
.caused_by(trc::location!())?;
batch = BatchBuilder::new();
}
}
if !batch.is_empty() {
server
.store()
.write(batch.build_all())
.await
.caused_by(trc::location!())?;
}
}
enum OldType {
Quota { size: u32 },
Undelete { deleted_at: u64, size: u32 },
Temp,
None,
}
struct OldBlobEntry {
account_id: u32,
until: u64,
hash: BlobHash,
blob_type: OldType,
old_key: Vec<u8>,
}
let mut entries = Vec::new();
let now = now();
server
.store()
.iterate(
IterateParams::new(
ValueKey::from(ValueClass::Any(AnyClass {
subspace: SUBSPACE_BLOB_RESERVE,
key: vec![0u8],
})),
ValueKey::from(ValueClass::Any(AnyClass {
subspace: SUBSPACE_BLOB_RESERVE,
key: vec![u8::MAX; 32],
})),
)
.ascending(),
|key, value| {
if key.len() == BLOB_HASH_LEN + U64_LEN + U32_LEN {
let account_id = key.deserialize_be_u32(0)?;
let hash = BlobHash::try_from_hash_slice(
key.get(U32_LEN..BLOB_HASH_LEN + U32_LEN).ok_or_else(|| {
trc::Error::corrupted_key(key, value.into(), trc::location!())
})?,
)
.unwrap();
let until = key.deserialize_be_u64(BLOB_HASH_LEN + U32_LEN)?;
let blob_type = if until > now {
if value.len() == U32_LEN {
let size = value.deserialize_be_u32(0)?;
if size != 0 {
OldType::Quota { size }
} else {
OldType::Temp
}
} else if value.len() == U64_LEN + U32_LEN {
let size = value.deserialize_be_u32(0)?;
let deleted_at = value.deserialize_be_u64(U32_LEN)?;
OldType::Undelete { deleted_at, size }
} else {
OldType::Temp
}
} else {
OldType::None
};
entries.push(OldBlobEntry {
account_id,
until,
hash,
blob_type,
old_key: key.to_vec(),
});
}
Ok(true)
},
)
.await
.caused_by(trc::location!())?;
let mut batch = BatchBuilder::new();
for entry in entries {
batch
.clear(ValueClass::Any(AnyClass {
subspace: SUBSPACE_BLOB_RESERVE,
key: entry.old_key,
}))
.with_account_id(entry.account_id);
match entry.blob_type {
OldType::Quota { size } => {
batch
.set(
BlobOp::Link {
hash: entry.hash.clone(),
to: BlobLink::Temporary { until: entry.until },
},
vec![BlobLink::QUOTA_LINK],
)
.set(
BlobOp::Quota {
hash: entry.hash,
until: entry.until,
},
size.serialize(),
);
}
OldType::Undelete { deleted_at, size } => {
#[cfg(feature = "enterprise")]
{
batch
.set(
BlobOp::Link {
hash: entry.hash.clone(),
to: BlobLink::Temporary { until: entry.until },
},
vec![BlobLink::UNDELETE_LINK],
)
.set(
BlobOp::Undelete {
hash: entry.hash,
until: entry.until,
},
Archiver::new(common::enterprise::undelete::DeletedItem {
typ: common::enterprise::undelete::DeletedItemType::Email {
from: "unknown".into(),
subject: "unknown".into(),
received_at: deleted_at,
},
size,
deleted_at,
})
.serialize()
.caused_by(trc::location!())?,
);
}
}
OldType::Temp => {
batch.set(
BlobOp::Link {
hash: entry.hash,
to: BlobLink::Temporary { until: entry.until },
},
vec![],
);
}
OldType::None => (),
}
if batch.is_large_batch() {
server
.store()
.write(batch.build_all())
.await
.caused_by(trc::location!())?;
batch = BatchBuilder::new();
}
}
if !batch.is_empty() {
server
.store()
.write(batch.build_all())
.await
.caused_by(trc::location!())?;
}
Ok(())
}
crates/migration/src/calendar_v2.rs
+4 -5
View File
@@ -7,13 +7,13 @@
use common::Server;
use groupware::calendar::{Calendar, CalendarPreferences, Timezone};
use store::{
Serialize,
write::{Archiver, BatchBuilder, serialize::rkyv_deserialize},
Serialize, ValueKey,
write::{AlignedBytes, Archive, Archiver, BatchBuilder, serialize::rkyv_deserialize},
};
use trc::AddContext;
use types::{acl::AclGrant, collection::Collection, dead_property::DeadProperty, field::Field};
use crate::event_v2::migrate_icalendar_v02;
use crate::{event_v2::migrate_icalendar_v02, get_document_ids};
#[derive(
rkyv::Archive, rkyv::Deserialize, rkyv::Serialize, Debug, Default, Clone, PartialEq, Eq,
@@ -62,8 +62,7 @@ pub struct DefaultAlertV2 {
}
pub(crate) async fn migrate_calendar_v013(server: &Server, account_id: u32) -> trc::Result<u64> {
let document_ids = server
.get_document_ids(account_id, Collection::Calendar)
let document_ids = get_document_ids(server, account_id, Collection::Calendar)
.await
.caused_by(trc::location!())?
.unwrap_or_default();
crates/migration/src/contact_v2.rs
+5 -4
View File
@@ -7,12 +7,14 @@
use common::{DavName, Server};
use groupware::contact::ContactCard;
use store::{
Serialize,
write::{Archiver, BatchBuilder, serialize::rkyv_deserialize},
Serialize, ValueKey,
write::{AlignedBytes, Archive, Archiver, BatchBuilder, serialize::rkyv_deserialize},
};
use trc::AddContext;
use types::{collection::Collection, dead_property::DeadProperty, field::Field};
use crate::get_document_ids;
#[derive(
rkyv::Archive, rkyv::Deserialize, rkyv::Serialize, Debug, Default, Clone, PartialEq, Eq,
)]
@@ -27,8 +29,7 @@ pub struct ContactCardV2 {
}
pub(crate) async fn migrate_contacts_v013(server: &Server, account_id: u32) -> trc::Result<u64> {
let document_ids = server
.get_document_ids(account_id, Collection::ContactCard)
let document_ids = get_document_ids(server, account_id, Collection::ContactCard)
.await
.caused_by(trc::location!())?
.unwrap_or_default();
crates/migration/src/email.rs → crates/migration/src/email_v1.rs
+160 -108
View File
@@ -5,28 +5,25 @@
*/
use super::{LegacyBincode, get_properties};
use crate::{email_v2::LegacyKeyword, get_bitmap, get_document_ids, v014::SUBSPACE_BITMAP_TAG};
use common::Server;
use email::{
mailbox::UidMailbox,
message::{
index::{MAX_ID_LENGTH, VisitText},
metadata::{
MessageData, MessageMetadata, MessageMetadataContents, MessageMetadataPart,
MetadataPartType,
},
message::metadata::{
MESSAGE_HAS_ATTACHMENT, MESSAGE_RECEIVED_MASK, MessageDataBuilder, MessageMetadata,
MessageMetadataContents, MessageMetadataPart, MetadataHeader, MetadataPartType,
PART_ENCODING_BASE64, PART_ENCODING_PROBLEM, PART_ENCODING_QP, PART_SIZE_MASK,
},
};
use mail_parser::{
Address, Attribute, ContentType, DateTime, Encoding, Header, HeaderName, HeaderValue, Received,
Address, Attribute, ContentType, DateTime, Encoding, HeaderName, HeaderValue, Received,
};
use std::{borrow::Cow, collections::VecDeque};
use store::{
BitmapKey, Deserialize, SUBSPACE_BITMAP_TAG, SUBSPACE_INDEXES, SUBSPACE_PROPERTY, Serialize,
U64_LEN, ValueKey,
Deserialize, SUBSPACE_INDEXES, SUBSPACE_PROPERTY, Serialize, U32_LEN, U64_LEN, ValueKey,
ahash::AHashMap,
write::{
AlignedBytes, AnyKey, Archive, Archiver, BatchBuilder, BitmapClass, TagValue, ValueClass,
key::KeySerializer,
AlignedBytes, AnyKey, Archive, Archiver, BatchBuilder, ValueClass, key::KeySerializer,
},
};
use trc::AddContext;
@@ -41,14 +38,13 @@ use utils::codec::leb128::Leb128Iterator;
const FIELD_KEYWORDS: u8 = 4;
const FIELD_THREAD_ID: u8 = 33;
const FIELD_CID: u8 = 76;
const FIELD_MAILBOX_IDS: u8 = 7;
pub(crate) const FIELD_MAILBOX_IDS: u8 = 7;
const BM_MARKER: u8 = 1 << 7;
pub(crate) async fn migrate_emails(server: &Server, account_id: u32) -> trc::Result<u64> {
pub(crate) async fn migrate_emails_v011(server: &Server, account_id: u32) -> trc::Result<u64> {
// Obtain email ids
let mut message_ids = server
.get_document_ids(account_id, Collection::Email)
let mut message_ids = get_document_ids(server, account_id, Collection::Email)
.await
.caused_by(trc::location!())?
.unwrap_or_default();
@@ -56,22 +52,33 @@ pub(crate) async fn migrate_emails(server: &Server, account_id: u32) -> trc::Res
if num_emails == 0 {
return Ok(0);
}
let tombstoned_ids = server
.store()
.get_bitmap(BitmapKey {
account_id,
collection: Collection::Email.into(),
class: BitmapClass::Tag {
field: FIELD_MAILBOX_IDS,
value: TagValue::Id(u32::MAX - 1),
},
document_id: 0,
})
.await
.caused_by(trc::location!())?
.unwrap_or_default();
let tombstoned_ids = get_bitmap(
server,
AnyKey {
subspace: SUBSPACE_BITMAP_TAG,
key: KeySerializer::new(U64_LEN + U32_LEN + 1)
.write(account_id)
.write(u8::from(Collection::Email))
.write(FIELD_MAILBOX_IDS)
.write_leb128(u32::MAX - 1)
.finalize(),
},
AnyKey {
subspace: SUBSPACE_BITMAP_TAG,
key: KeySerializer::new(U64_LEN + U32_LEN + 1)
.write(account_id)
.write(u8::from(Collection::Email))
.write(FIELD_MAILBOX_IDS)
.write_leb128(u32::MAX - 1)
.finalize(),
},
)
.await
.caused_by(trc::location!())?
.unwrap_or_default();
let mut message_data: AHashMap<u32, MessageData> = AHashMap::with_capacity(num_emails as usize);
let mut message_data: AHashMap<u32, MessageDataBuilder> =
AHashMap::with_capacity(num_emails as usize);
let mut did_migrate = false;
// Obtain mailboxes
@@ -94,7 +101,8 @@ pub(crate) async fn migrate_emails(server: &Server, account_id: u32) -> trc::Res
.await
.caused_by(trc::location!())?
{
message_data.entry(message_id).or_default().keywords = keywords.0;
message_data.entry(message_id).or_default().keywords =
keywords.0.into_iter().map(Into::into).collect();
}
// Obtain threadIds
@@ -106,36 +114,6 @@ pub(crate) async fn migrate_emails(server: &Server, account_id: u32) -> trc::Res
message_data.entry(message_id).or_default().thread_id = thread_id;
}
// Write message data
for (message_id, data) in message_data {
if !tombstoned_ids.contains(message_id) {
message_ids.insert(message_id);
let mut batch = BatchBuilder::new();
batch
.with_account_id(account_id)
.with_collection(Collection::Email)
.with_document(message_id);
for mailbox in &data.mailboxes {
batch.untag(EmailField::MailboxIds, TagValue::Id(mailbox.mailbox_id));
}
did_migrate = true;
batch.set(
Field::ARCHIVE,
Archiver::new(data)
.serialize()
.caused_by(trc::location!())?,
);
server
.store()
.write(batch.build_all())
.await
.caused_by(trc::location!())?;
}
}
// Migrate message metadata
for message_id in &message_ids {
match server
@@ -149,6 +127,9 @@ pub(crate) async fn migrate_emails(server: &Server, account_id: u32) -> trc::Res
.await
{
Ok(Some(legacy_metadata)) => {
message_data.entry(message_id).or_default().size =
legacy_metadata.inner.size as u32;
let metadata = MessageMetadata::from_legacy(legacy_metadata.inner);
let mut batch = BatchBuilder::new();
@@ -157,7 +138,8 @@ pub(crate) async fn migrate_emails(server: &Server, account_id: u32) -> trc::Res
.with_collection(Collection::Email)
.with_document(message_id);
for header in metadata.root_part().headers.iter().rev() {
let todo = "reindex";
/*for header in metadata.root_part().headers.iter().rev() {
if matches!(header.name, HeaderName::MessageId) {
header.value.visit_text(|id| {
if id.len() < MAX_ID_LENGTH {
@@ -165,7 +147,7 @@ pub(crate) async fn migrate_emails(server: &Server, account_id: u32) -> trc::Res
}
});
}
}
}*/
batch.set(
EmailField::Metadata,
@@ -204,6 +186,36 @@ pub(crate) async fn migrate_emails(server: &Server, account_id: u32) -> trc::Res
}
}
// Write message data
for (message_id, data) in message_data {
if !tombstoned_ids.contains(message_id) {
message_ids.insert(message_id);
let mut batch = BatchBuilder::new();
batch
.with_account_id(account_id)
.with_collection(Collection::Email)
.with_document(message_id);
/*for mailbox in &data.mailboxes {
batch.untag(EmailField::MailboxIds, TagValue::Id(mailbox.mailbox_id));
}*/
did_migrate = true;
batch.set(
Field::ARCHIVE,
Archiver::new(data.seal())
.serialize()
.caused_by(trc::location!())?,
);
server
.store()
.write(batch.build_all())
.await
.caused_by(trc::location!())?;
}
}
// Delete keyword bitmaps
for field in [FIELD_KEYWORDS, FIELD_KEYWORDS | BM_MARKER] {
server
@@ -312,26 +324,12 @@ pub trait FromLegacy {
impl FromLegacy for MessageMetadata {
fn from_legacy(legacy: LegacyMessageMetadata<'_>) -> Self {
let mut metadata = MessageMetadata {
contents: vec![],
blob_hash: legacy.blob_hash,
size: legacy.size as u32,
received_at: legacy.received_at,
preview: legacy.preview,
has_attachments: legacy.has_attachments,
raw_headers: legacy.raw_headers,
};
let mut contents = Vec::new();
let mut messages = VecDeque::from([legacy.contents]);
let mut message_id = 0;
while let Some(message) = messages.pop_front() {
let mut contents = MessageMetadataContents {
html_body: message.html_body.into_iter().map(|c| c as u16).collect(),
text_body: message.text_body.into_iter().map(|c| c as u16).collect(),
attachments: message.attachments.into_iter().map(|c| c as u16).collect(),
parts: Vec::with_capacity(message.parts.len()),
};
let mut parts = Vec::new();
for part in message.parts {
let body = match part.body {
@@ -349,36 +347,90 @@ impl FromLegacy for MessageMetadata {
}
};
contents.parts.push(MessageMetadataPart {
let flags = match part.encoding {
Encoding::None => 0,
Encoding::QuotedPrintable => PART_ENCODING_QP,
Encoding::Base64 => PART_ENCODING_BASE64,
} | (if part.is_encoding_problem {
PART_ENCODING_PROBLEM
} else {
0
}) | (part.size as u32 & PART_SIZE_MASK);
parts.push(MessageMetadataPart {
headers: part
.headers
.into_iter()
.map(|hdr| Header {
name: hdr.name.into_owned(),
value: hdr.value.into(),
offset_field: hdr.offset_field as u32,
offset_start: hdr.offset_start as u32,
offset_end: hdr.offset_end as u32,
.map(|hdr| MetadataHeader {
value: if matches!(
&hdr.name,
HeaderName::Subject
| HeaderName::From
| HeaderName::To
| HeaderName::Cc
| HeaderName::Date
| HeaderName::Bcc
| HeaderName::ReplyTo
| HeaderName::Sender
| HeaderName::Comments
| HeaderName::InReplyTo
| HeaderName::Keywords
| HeaderName::MessageId
| HeaderName::References
| HeaderName::ResentMessageId
| HeaderName::ContentDescription
| HeaderName::ContentId
| HeaderName::ContentLanguage
| HeaderName::ContentLocation
| HeaderName::ContentTransferEncoding
| HeaderName::ContentType
| HeaderName::ContentDisposition
| HeaderName::ListId
) {
HeaderValue::from(hdr.value)
} else {
HeaderValue::Empty
}
.into(),
name: hdr.name.into(),
base_offset: hdr.offset_field as u32,
start: (hdr.offset_start - hdr.offset_field) as u16,
end: (hdr.offset_end - hdr.offset_field) as u16,
})
.collect(),
is_encoding_problem: part.is_encoding_problem,
encoding: part.encoding,
flags,
body,
size: part.size as u32,
offset_header: part.offset_header as u32,
offset_body: part.offset_body as u32,
offset_end: part.offset_end as u32,
});
}
metadata.contents.push(contents);
contents.push(MessageMetadataContents {
html_body: message.html_body.into_iter().map(|c| c as u16).collect(),
text_body: message.text_body.into_iter().map(|c| c as u16).collect(),
attachments: message.attachments.into_iter().map(|c| c as u16).collect(),
parts: parts.into_boxed_slice(),
});
}
metadata
MessageMetadata {
blob_body_offset: contents.first().unwrap().root_part().offset_body,
contents: contents.into_boxed_slice(),
blob_hash: legacy.blob_hash,
preview: legacy.preview.into_boxed_str(),
raw_headers: legacy.raw_headers.into_boxed_slice(),
rcvd_attach: (if legacy.has_attachments {
MESSAGE_HAS_ATTACHMENT
} else {
0
}) | (legacy.received_at & MESSAGE_RECEIVED_MASK),
}
}
}
pub struct Mailboxes(Vec<UidMailbox>);
pub struct Keywords(Vec<Keyword>);
pub struct Keywords(Vec<LegacyKeyword>);
impl Deserialize for Mailboxes {
fn deserialize(bytes: &[u8]) -> trc::Result<Self> {
@@ -418,27 +470,27 @@ impl Deserialize for Keywords {
}
}
fn deserialize_keyword(bytes: &mut std::slice::Iter<'_, u8>) -> Option<Keyword> {
fn deserialize_keyword(bytes: &mut std::slice::Iter<'_, u8>) -> Option<LegacyKeyword> {
match bytes.next_leb128::<usize>()? {
SEEN => Some(Keyword::Seen),
DRAFT => Some(Keyword::Draft),
FLAGGED => Some(Keyword::Flagged),
ANSWERED => Some(Keyword::Answered),
RECENT => Some(Keyword::Recent),
IMPORTANT => Some(Keyword::Important),
PHISHING => Some(Keyword::Phishing),
JUNK => Some(Keyword::Junk),
NOTJUNK => Some(Keyword::NotJunk),
DELETED => Some(Keyword::Deleted),
FORWARDED => Some(Keyword::Forwarded),
MDN_SENT => Some(Keyword::MdnSent),
SEEN => Some(LegacyKeyword::Seen),
DRAFT => Some(LegacyKeyword::Draft),
FLAGGED => Some(LegacyKeyword::Flagged),
ANSWERED => Some(LegacyKeyword::Answered),
RECENT => Some(LegacyKeyword::Recent),
IMPORTANT => Some(LegacyKeyword::Important),
PHISHING => Some(LegacyKeyword::Phishing),
JUNK => Some(LegacyKeyword::Junk),
NOTJUNK => Some(LegacyKeyword::NotJunk),
DELETED => Some(LegacyKeyword::Deleted),
FORWARDED => Some(LegacyKeyword::Forwarded),
MDN_SENT => Some(LegacyKeyword::MdnSent),
other => {
let len = other - OTHER;
let mut keyword = Vec::with_capacity(len);
for _ in 0..len {
keyword.push(*bytes.next()?);
}
Some(Keyword::Other(String::from_utf8(keyword).ok()?))
Some(LegacyKeyword::Other(String::from_utf8(keyword).ok()?))
}
}
}
@@ -559,9 +611,9 @@ impl From<LegacyHeaderValue<'_>> for HeaderValue<'static> {
}
}
pub(crate) fn encode_message_id(message_id: &str) -> Vec<u8> {
/*pub(crate) fn encode_message_id(message_id: &str) -> Vec<u8> {
let mut msg_id = Vec::with_capacity(message_id.len() + 1);
msg_id.extend_from_slice(message_id.as_bytes());
msg_id.push(0);
msg_id
}
}*/
crates/migration/src/email_v2.rs
+435
View File
@@ -0,0 +1,435 @@
/*
* SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <hello@stalw.art>
*
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
*/
use crate::{email_v1::FIELD_MAILBOX_IDS, get_bitmap, v014::SUBSPACE_BITMAP_TAG};
use common::Server;
use email::{
mailbox::UidMailbox,
message::{
index::extractors::VisitTextArchived,
ingest::ThreadInfo,
metadata::{
MESSAGE_HAS_ATTACHMENT, MESSAGE_RECEIVED_MASK, MessageData, MessageMetadata,
MessageMetadataContents, MessageMetadataPart, MetadataHeader, MetadataHeaderName,
MetadataHeaderValue, MetadataPartType, PART_ENCODING_BASE64, PART_ENCODING_PROBLEM,
PART_ENCODING_QP, PART_SIZE_MASK,
},
},
};
use mail_parser::{Encoding, Header, parsers::fields::thread::thread_name};
use store::{
SUBSPACE_INDEXES, Serialize, U32_LEN, U64_LEN, ValueKey,
write::{
AlignedBytes, AnyKey, Archive, Archiver, BatchBuilder, IndexPropertyClass, ValueClass,
key::KeySerializer,
},
};
use trc::AddContext;
use types::{blob_hash::BlobHash, collection::Collection, field::EmailField, keyword::*};
use utils::cheeky_hash::CheekyHash;
pub(crate) async fn migrate_emails_v014(server: &Server, account_id: u32) -> trc::Result<u64> {
let tombstoned_ids = get_bitmap(
server,
AnyKey {
subspace: SUBSPACE_BITMAP_TAG,
key: KeySerializer::new(U64_LEN + U32_LEN + 1)
.write(account_id)
.write(u8::from(Collection::Email))
.write(FIELD_MAILBOX_IDS)
.write_leb128(u32::MAX - 1)
.finalize(),
},
AnyKey {
subspace: SUBSPACE_BITMAP_TAG,
key: KeySerializer::new(U64_LEN + U32_LEN + 1)
.write(account_id)
.write(u8::from(Collection::Email))
.write(FIELD_MAILBOX_IDS)
.write_leb128(u32::MAX - 1)
.finalize(),
},
)
.await
.caused_by(trc::location!())?
.unwrap_or_default();
let todo = "delete indexes";
let mut migrate = Vec::new();
server
.archives(
account_id,
Collection::Email,
&(),
|document_id, archive| {
match archive.deserialize_untrusted::<LegacyMessageData>() {
Ok(metadata) => {
migrate.push((document_id, metadata));
}
Err(err) => {
if archive.deserialize_untrusted::<MessageData>().is_err() {
return Err(err
.account_id(account_id)
.document_id(document_id)
.caused_by(trc::location!()));
}
}
}
Ok(true)
},
)
.await
.caused_by(trc::location!())?;
let num_emails = migrate.len();
for (document_id, legacy_data) in migrate {
let mut batch = BatchBuilder::new();
batch
.with_account_id(account_id)
.with_collection(Collection::Email)
.with_document(document_id);
if !tombstoned_ids.contains(document_id) {
let (size, metadata) = match server
.store()
.get_value::<Archive<AlignedBytes>>(ValueKey::property(
account_id,
Collection::Email,
document_id,
EmailField::Metadata,
))
.await?
{
Some(metadata) => match metadata.deserialize_untrusted::<LegacyMessageMetadata>() {
Ok(legacy) => (legacy.size, MessageMetadata::from(legacy)),
Err(err) => match metadata.deserialize_untrusted::<MessageMetadata>() {
Ok(metadata) => (metadata.root_part().offset_end, metadata),
Err(_) => {
return Err(err
.account_id(account_id)
.document_id(document_id)
.caused_by(trc::location!()));
}
},
},
None => {
batch.clear(EmailField::Archive).clear(EmailField::Metadata);
continue;
}
};
let data = MessageData {
mailboxes: legacy_data.mailboxes.into_boxed_slice(),
keywords: legacy_data.keywords.into_iter().map(Into::into).collect(),
thread_id: legacy_data.thread_id,
size,
};
let mut message_ids = Vec::new();
let mut subject = "";
for header in &metadata.contents[0].parts[0].headers {
match &header.name {
MetadataHeaderName::MessageId => {
header.value.visit_text(|id| {
if !id.is_empty() {
message_ids.push(CheekyHash::new(id.as_bytes()));
}
});
}
MetadataHeaderName::InReplyTo
| MetadataHeaderName::References
| MetadataHeaderName::ResentMessageId => {
header.value.visit_text(|id| {
if !id.is_empty() {
message_ids.push(CheekyHash::new(id.as_bytes()));
}
});
}
MetadataHeaderName::Subject if subject.is_empty() => {
subject = thread_name(match &header.value {
MetadataHeaderValue::Text(text) => text.as_ref(),
MetadataHeaderValue::TextList(list) if !list.is_empty() => {
list.first().unwrap().as_ref()
}
_ => "",
});
}
_ => (),
}
}
batch
.set(
ValueClass::IndexProperty(IndexPropertyClass::Hash {
property: EmailField::Threading.into(),
hash: CheekyHash::new(if !subject.is_empty() { subject } else { "!" }),
}),
ThreadInfo::serialize(legacy_data.thread_id, &message_ids),
)
.set(
EmailField::Archive,
Archiver::new(data)
.serialize()
.caused_by(trc::location!())?,
)
.set(
EmailField::Metadata,
Archiver::new(metadata)
.serialize()
.caused_by(trc::location!())?,
);
} else {
batch.clear(EmailField::Archive).clear(EmailField::Metadata);
}
server
.store()
.write(batch.build_all())
.await
.caused_by(trc::location!())?;
}
/*
EmailField::From => 87,
EmailField::To => 35,
EmailField::Cc => 74,
EmailField::Bcc => 69,
EmailField::Subject => 29,
EmailField::Size => 27,
EmailField::References => 20,
EmailField::MailboxIds => 7,
EmailField::ReceivedAt => 19,
EmailField::SentAt => 26,
EmailField::HasAttachment => 89,
*/
for index in [87u8, 35, 74, 69, 29, 27, 20, 7, 19, 26, 89] {
server
.store()
.delete_range(
AnyKey {
subspace: SUBSPACE_INDEXES,
key: KeySerializer::new(U64_LEN * 3)
.write(account_id)
.write(u8::from(Collection::Email))
.write(index)
.finalize(),
},
AnyKey {
subspace: SUBSPACE_INDEXES,
key: KeySerializer::new(U64_LEN * 4)
.write(account_id)
.write(u8::from(Collection::Email))
.write(index)
.write(&[u8::MAX; 8][..])
.finalize(),
},
)
.await
.caused_by(trc::location!())?;
}
Ok(num_emails as u64)
}
#[derive(rkyv::Serialize, rkyv::Deserialize, rkyv::Archive, Debug, Default)]
pub struct LegacyMessageData {
pub mailboxes: Vec<UidMailbox>,
pub keywords: Vec<LegacyKeyword>,
pub thread_id: u32,
}
#[derive(rkyv::Serialize, rkyv::Deserialize, rkyv::Archive, Debug)]
pub struct LegacyMessageMetadata {
pub contents: Vec<LegacyMessageMetadataContents>,
pub blob_hash: BlobHash,
pub size: u32,
pub received_at: u64,
pub preview: String,
pub has_attachments: bool,
pub raw_headers: Vec<u8>,
}
impl From<LegacyMessageMetadata> for MessageMetadata {
fn from(legacy: LegacyMessageMetadata) -> Self {
MessageMetadata {
blob_body_offset: legacy
.contents
.first()
.unwrap()
.parts
.first()
.unwrap()
.offset_body,
contents: legacy.contents.into_iter().map(Into::into).collect(),
blob_hash: legacy.blob_hash,
preview: legacy.preview.into_boxed_str(),
raw_headers: legacy.raw_headers.into_boxed_slice(),
rcvd_attach: (if legacy.has_attachments {
MESSAGE_HAS_ATTACHMENT
} else {
0
}) | (legacy.received_at & MESSAGE_RECEIVED_MASK),
}
}
}
#[derive(rkyv::Serialize, rkyv::Deserialize, rkyv::Archive, Debug)]
pub struct LegacyMessageMetadataContents {
pub html_body: Vec<u16>,
pub text_body: Vec<u16>,
pub attachments: Vec<u16>,
pub parts: Vec<LegacyMessageMetadataPart>,
}
impl From<LegacyMessageMetadataContents> for MessageMetadataContents {
fn from(contents: LegacyMessageMetadataContents) -> Self {
MessageMetadataContents {
html_body: contents.html_body.into_boxed_slice(),
text_body: contents.text_body.into_boxed_slice(),
attachments: contents.attachments.into_boxed_slice(),
parts: contents.parts.into_iter().map(Into::into).collect(),
}
}
}
#[derive(rkyv::Serialize, rkyv::Deserialize, rkyv::Archive, Debug)]
pub struct LegacyMessageMetadataPart {
pub headers: Vec<Header<'static>>,
pub is_encoding_problem: bool,
pub body: LegacyMetadataPartType,
pub encoding: Encoding,
pub size: u32,
pub offset_header: u32,
pub offset_body: u32,
pub offset_end: u32,
}
impl From<LegacyMessageMetadataPart> for MessageMetadataPart {
fn from(part: LegacyMessageMetadataPart) -> Self {
let flags = match part.encoding {
Encoding::None => 0,
Encoding::QuotedPrintable => PART_ENCODING_QP,
Encoding::Base64 => PART_ENCODING_BASE64,
} | (if part.is_encoding_problem {
PART_ENCODING_PROBLEM
} else {
0
}) | (part.size & PART_SIZE_MASK);
MessageMetadataPart {
headers: part
.headers
.into_iter()
.map(|hdr| MetadataHeader {
value: hdr.value.into(),
name: hdr.name.into(),
base_offset: hdr.offset_field,
start: (hdr.offset_start - hdr.offset_field) as u16,
end: (hdr.offset_end - hdr.offset_field) as u16,
})
.collect(),
flags,
body: part.body.into(),
offset_header: part.offset_header,
offset_body: part.offset_body,
offset_end: part.offset_end,
}
}
}
#[derive(rkyv::Serialize, rkyv::Deserialize, rkyv::Archive, Debug)]
pub enum LegacyMetadataPartType {
Text,
Html,
Binary,
InlineBinary,
Message(u16),
Multipart(Vec<u16>),
}
impl From<LegacyMetadataPartType> for MetadataPartType {
fn from(value: LegacyMetadataPartType) -> Self {
match value {
LegacyMetadataPartType::Text => MetadataPartType::Text,
LegacyMetadataPartType::Html => MetadataPartType::Html,
LegacyMetadataPartType::Binary => MetadataPartType::Binary,
LegacyMetadataPartType::InlineBinary => MetadataPartType::InlineBinary,
LegacyMetadataPartType::Message(id) => MetadataPartType::Message(id),
LegacyMetadataPartType::Multipart(children) => {
MetadataPartType::Multipart(children.into_boxed_slice())
}
}
}
}
#[derive(
rkyv::Serialize,
rkyv::Deserialize,
rkyv::Archive,
Debug,
Clone,
PartialEq,
Eq,
Hash,
Default,
PartialOrd,
Ord,
serde::Serialize,
)]
#[serde(untagged)]
#[rkyv(derive(PartialEq), compare(PartialEq))]
pub enum LegacyKeyword {
#[serde(rename(serialize = "$seen"))]
Seen,
#[serde(rename(serialize = "$draft"))]
Draft,
#[serde(rename(serialize = "$flagged"))]
Flagged,
#[serde(rename(serialize = "$answered"))]
Answered,
#[default]
#[serde(rename(serialize = "$recent"))]
Recent,
#[serde(rename(serialize = "$important"))]
Important,
#[serde(rename(serialize = "$phishing"))]
Phishing,
#[serde(rename(serialize = "$junk"))]
Junk,
#[serde(rename(serialize = "$notjunk"))]
NotJunk,
#[serde(rename(serialize = "$deleted"))]
Deleted,
#[serde(rename(serialize = "$forwarded"))]
Forwarded,
#[serde(rename(serialize = "$mdnsent"))]
MdnSent,
Other(String),
}
impl From<LegacyKeyword> for Keyword {
fn from(kw: LegacyKeyword) -> Self {
match kw {
LegacyKeyword::Seen => Keyword::Seen,
LegacyKeyword::Draft => Keyword::Draft,
LegacyKeyword::Flagged => Keyword::Flagged,
LegacyKeyword::Answered => Keyword::Answered,
LegacyKeyword::Recent => Keyword::Recent,
LegacyKeyword::Important => Keyword::Important,
LegacyKeyword::Phishing => Keyword::Phishing,
LegacyKeyword::Junk => Keyword::Junk,
LegacyKeyword::NotJunk => Keyword::NotJunk,
LegacyKeyword::Deleted => Keyword::Deleted,
LegacyKeyword::Forwarded => Keyword::Forwarded,
LegacyKeyword::MdnSent => Keyword::MdnSent,
LegacyKeyword::Other(s) => Keyword::Other(s.into_boxed_str()),
}
}
}
crates/migration/src/encryption.rs → crates/migration/src/encryption_v1.rs
+8 -6
View File
@@ -13,13 +13,15 @@ use store::{
use trc::AddContext;
use types::{collection::Collection, field::PrincipalField};
pub(crate) async fn migrate_encryption_params(
use crate::encryption_v2::LegacyEncryptionParams;
pub(crate) async fn migrate_encryption_params_v011(
server: &Server,
account_id: u32,
) -> trc::Result<u64> {
match server
.store()
.get_value::<LegacyEncryptionParams>(ValueKey {
.get_value::<VeryOldLegacyEncryptionParams>(ValueKey {
account_id,
collection: Collection::Principal.into(),
document_id: 0,
@@ -35,7 +37,7 @@ pub(crate) async fn migrate_encryption_params(
.with_document(0)
.set(
PrincipalField::EncryptionKeys,
Archiver::new(legacy.0)
Archiver::new(EncryptionParams::from(legacy.0))
.serialize()
.caused_by(trc::location!())?,
);
@@ -67,16 +69,16 @@ pub(crate) async fn migrate_encryption_params(
Ok(0)
}
struct LegacyEncryptionParams(EncryptionParams);
struct VeryOldLegacyEncryptionParams(LegacyEncryptionParams);
impl Deserialize for LegacyEncryptionParams {
impl Deserialize for VeryOldLegacyEncryptionParams {
fn deserialize(bytes: &[u8]) -> trc::Result<Self> {
let version = *bytes
.first()
.ok_or_else(|| trc::StoreEvent::DataCorruption.caused_by(trc::location!()))?;
match version {
1 if bytes.len() > 1 => bincode::deserialize(&bytes[1..])
.map(LegacyEncryptionParams)
.map(VeryOldLegacyEncryptionParams)
.map_err(|err| {
trc::EventType::Store(trc::StoreEvent::DeserializeError)
.reason(err)
crates/migration/src/encryption_v2.rs
+132
View File
@@ -0,0 +1,132 @@
/*
* SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <hello@stalw.art>
*
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
*/
use common::Server;
use email::message::crypto::{
ENCRYPT_ALGO_AES128, ENCRYPT_ALGO_AES256, ENCRYPT_METHOD_PGP, ENCRYPT_METHOD_SMIME,
EncryptionParams,
};
use store::{
Deserialize, Serialize, ValueKey,
write::{AlignedBytes, Archive, Archiver, BatchBuilder, ValueClass},
};
use trc::AddContext;
use types::{collection::Collection, field::PrincipalField};
#[derive(
rkyv::Serialize,
rkyv::Deserialize,
rkyv::Archive,
Debug,
Clone,
Copy,
PartialEq,
Eq,
serde::Serialize,
serde::Deserialize,
)]
pub enum EncryptionMethod {
PGP,
SMIME,
}
#[derive(
rkyv::Serialize,
rkyv::Deserialize,
rkyv::Archive,
Debug,
Clone,
Copy,
serde::Serialize,
serde::Deserialize,
)]
#[rkyv(derive(Clone, Copy))]
pub enum Algorithm {
Aes128,
Aes256,
}
#[derive(
Clone,
rkyv::Serialize,
rkyv::Deserialize,
rkyv::Archive,
Debug,
serde::Serialize,
serde::Deserialize,
)]
pub struct LegacyEncryptionParams {
pub method: EncryptionMethod,
pub algo: Algorithm,
pub certs: Vec<Vec<u8>>,
}
pub(crate) async fn migrate_encryption_params_v014(
server: &Server,
account_id: u32,
) -> trc::Result<u64> {
let Some(params) = server
.store()
.get_value::<Archive<AlignedBytes>>(ValueKey {
account_id,
collection: Collection::Principal.into(),
document_id: 0,
class: ValueClass::from(PrincipalField::EncryptionKeys),
})
.await
.caused_by(trc::location!())?
else {
return Ok(0);
};
match params.deserialize_untrusted::<LegacyEncryptionParams>() {
Ok(legacy) => {
let mut batch = BatchBuilder::new();
batch
.with_account_id(account_id)
.with_collection(Collection::Principal)
.with_document(0)
.set(
PrincipalField::EncryptionKeys,
Archiver::new(EncryptionParams::from(legacy))
.serialize()
.caused_by(trc::location!())?,
);
server
.store()
.write(batch.build_all())
.await
.caused_by(trc::location!())?;
Ok(1)
}
Err(err) => {
if params.deserialize_untrusted::<EncryptionParams>().is_err() {
return Err(err.account_id(account_id).caused_by(trc::location!()));
}
Ok(0)
}
}
}
impl From<LegacyEncryptionParams> for EncryptionParams {
fn from(legacy: LegacyEncryptionParams) -> Self {
EncryptionParams {
flags: match legacy.method {
EncryptionMethod::PGP => ENCRYPT_METHOD_PGP,
EncryptionMethod::SMIME => ENCRYPT_METHOD_SMIME,
} | match legacy.algo {
Algorithm::Aes128 => ENCRYPT_ALGO_AES128,
Algorithm::Aes256 => ENCRYPT_ALGO_AES256,
},
certs: legacy
.certs
.into_iter()
.map(|c| c.into_boxed_slice())
.collect(),
}
}
}
crates/migration/src/event_v1.rs
+5 -7
View File
@@ -7,14 +7,14 @@
use common::{DavName, Server};
use groupware::calendar::{AlarmDelta, CalendarEvent, CalendarEventData, ComponentTimeRange};
use store::{
Serialize,
Serialize, ValueKey,
rand::{self, seq::SliceRandom},
write::{Archiver, BatchBuilder, serialize::rkyv_deserialize},
write::{AlignedBytes, Archive, Archiver, BatchBuilder, serialize::rkyv_deserialize},
};
use trc::AddContext;
use types::{collection::Collection, dead_property::DeadProperty, field::Field};
use crate::event_v2::migrate_icalendar_v02;
use crate::{event_v2::migrate_icalendar_v02, get_document_ids};
#[derive(
rkyv::Archive, rkyv::Deserialize, rkyv::Serialize, Debug, Default, Clone, PartialEq, Eq,
@@ -62,8 +62,7 @@ pub struct AlarmV1 {
pub(crate) async fn migrate_calendar_events_v012(server: &Server) -> trc::Result<()> {
// Obtain email ids
let account_ids = server
.get_document_ids(u32::MAX, Collection::Principal)
let account_ids = get_document_ids(server, u32::MAX, Collection::Principal)
.await
.caused_by(trc::location!())?
.unwrap_or_default();
@@ -77,8 +76,7 @@ pub(crate) async fn migrate_calendar_events_v012(server: &Server) -> trc::Result
account_ids.shuffle(&mut rand::rng());
for account_id in account_ids {
let document_ids = server
.get_document_ids(account_id, Collection::CalendarEvent)
let document_ids = get_document_ids(server, account_id, Collection::CalendarEvent)
.await
.caused_by(trc::location!())?
.unwrap_or_default();
crates/migration/src/event_v2.rs
+6 -6
View File
@@ -9,12 +9,14 @@ use groupware::calendar::{
Alarm, CalendarEvent, CalendarEventData, CalendarEventNotification, ComponentTimeRange,
};
use store::{
Serialize,
write::{Archiver, BatchBuilder, serialize::rkyv_deserialize},
Serialize, ValueKey,
write::{AlignedBytes, Archive, Archiver, BatchBuilder, serialize::rkyv_deserialize},
};
use trc::AddContext;
use types::{collection::Collection, dead_property::DeadProperty, field::Field};
use crate::get_document_ids;
#[derive(
rkyv::Archive, rkyv::Deserialize, rkyv::Serialize, Debug, Default, Clone, PartialEq, Eq,
)]
@@ -67,8 +69,7 @@ pub(crate) async fn migrate_calendar_events_v013(
server: &Server,
account_id: u32,
) -> trc::Result<u64> {
let document_ids = server
.get_document_ids(account_id, Collection::CalendarEvent)
let document_ids = get_document_ids(server, account_id, Collection::CalendarEvent)
.await
.caused_by(trc::location!())?
.unwrap_or_default();
@@ -146,8 +147,7 @@ pub(crate) async fn migrate_calendar_scheduling_v013(
server: &Server,
account_id: u32,
) -> trc::Result<u64> {
let document_ids = server
.get_document_ids(account_id, Collection::CalendarEventNotification)
let document_ids = get_document_ids(server, account_id, Collection::CalendarEventNotification)
.await
.caused_by(trc::location!())?
.unwrap_or_default();
crates/migration/src/identity.rs → crates/migration/src/identity_v1.rs
+7 -5
View File
@@ -5,7 +5,10 @@
*/
use super::object::Object;
use crate::object::{FromLegacy, Property, Value};
use crate::{
get_document_ids,
object::{FromLegacy, Property, Value},
};
use common::Server;
use email::identity::{EmailAddress, Identity};
use store::{
@@ -15,10 +18,9 @@ use store::{
use trc::AddContext;
use types::{collection::Collection, field::Field};
pub(crate) async fn migrate_identities(server: &Server, account_id: u32) -> trc::Result<u64> {
// Obtain email ids
let identity_ids = server
.get_document_ids(account_id, Collection::Identity)
pub(crate) async fn migrate_identities_v011(server: &Server, account_id: u32) -> trc::Result<u64> {
// Obtain identity ids
let identity_ids = get_document_ids(server, account_id, Collection::Identity)
.await
.caused_by(trc::location!())?
.unwrap_or_default();
crates/migration/src/identity_v2.rs
+39
View File
@@ -0,0 +1,39 @@
/*
* SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <hello@stalw.art>
*
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
*/
use crate::get_document_ids;
use common::Server;
use store::write::BatchBuilder;
use trc::AddContext;
use types::{collection::Collection, field::IdentityField};
pub(crate) async fn migrate_identities_v014(server: &Server, account_id: u32) -> trc::Result<u64> {
let identity_ids = get_document_ids(server, account_id, Collection::Identity)
.await
.caused_by(trc::location!())?
.unwrap_or_default();
let num_identities = identity_ids.len();
if num_identities == 0 {
return Ok(0);
}
let mut batch = BatchBuilder::new();
batch.with_account_id(account_id);
for document_id in identity_ids {
batch
.with_document(document_id)
.tag(IdentityField::DocumentId);
}
server
.store()
.write(batch.build_all())
.await
.caused_by(trc::location!())?;
Ok(num_identities as u64)
}
crates/migration/src/lib.rs
+81 -13
View File
@@ -4,55 +4,67 @@
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
*/
/*use crate::{
queue::{migrate_queue_v011, migrate_queue_v012},
use crate::{
queue_v1::{migrate_queue_v011, migrate_queue_v012},
v011::migrate_v0_11,
v012::migrate_v0_12,
v013::migrate_v0_13,
};*/
v014::SUBSPACE_BITMAP_ID,
};
use common::{DATABASE_SCHEMA_VERSION, Server, manager::boot::DEFAULT_SETTINGS};
use std::time::Duration;
use store::{
Deserialize, IterateParams, SUBSPACE_PROPERTY, SUBSPACE_QUEUE_MESSAGE, SUBSPACE_REPORT_IN,
SUBSPACE_REPORT_OUT, SUBSPACE_SETTINGS, SerializeInfallible, U32_LEN, Value, ValueKey,
dispatch::DocumentSet,
write::{AnyClass, AnyKey, BatchBuilder, ValueClass, key::DeserializeBigEndian},
roaring::RoaringBitmap,
write::{
AnyClass, AnyKey, BatchBuilder, ValueClass,
key::{DeserializeBigEndian, KeySerializer},
},
};
use trc::AddContext;
use types::collection::Collection;
/*pub mod addressbook_v2;
pub mod addressbook_v2;
pub mod blob;
pub mod calendar_v2;
pub mod changelog;
pub mod contact_v2;
pub mod email;
pub mod encryption;
pub mod email_v1;
pub mod email_v2;
pub mod encryption_v1;
pub mod encryption_v2;
pub mod event_v1;
pub mod event_v2;
pub mod identity;
pub mod identity_v1;
pub mod identity_v2;
pub mod mailbox;
pub mod object;
pub mod principal_v1;
pub mod principal_v2;
pub mod push_v1;
pub mod push_v2;
pub mod queue;
pub mod queue_v1;
pub mod queue_v2;
pub mod report;
pub mod sieve_v1;
pub mod sieve_v2;
pub mod submission;
pub mod tasks;
pub mod tasks_v1;
pub mod tasks_v2;
pub mod threads;
pub mod v011;
pub mod v012;
pub mod v013;*/
pub mod v013;
pub mod v014;
const LOCK_WAIT_TIME_ACCOUNT: u64 = 3 * 60;
const LOCK_WAIT_TIME_CORE: u64 = 5 * 60;
const LOCK_RETRY_TIME: Duration = Duration::from_secs(30);
pub async fn try_migrate(server: &Server) -> trc::Result<()> {
/*if let Some(version) = std::env::var("FORCE_MIGRATE_QUEUE")
if let Some(version) = std::env::var("FORCE_MIGRATE_QUEUE")
.ok()
.and_then(|s| s.parse::<u32>().ok())
{
@@ -170,7 +182,7 @@ pub async fn try_migrate(server: &Server) -> trc::Result<()> {
.store()
.write(batch.build_all())
.await
.caused_by(trc::location!())?;*/
.caused_by(trc::location!())?;
Ok(())
}
@@ -269,6 +281,62 @@ where
.map(|_| results)
}
pub async fn get_document_ids(
server: &Server,
account_id: u32,
collection: Collection,
) -> trc::Result<Option<RoaringBitmap>> {
let collection: u8 = collection.into();
get_bitmap(
server,
AnyKey {
subspace: SUBSPACE_BITMAP_ID,
key: KeySerializer::new(U32_LEN + 1)
.write(account_id)
.write(collection)
.write(0u32)
.finalize(),
},
AnyKey {
subspace: SUBSPACE_BITMAP_ID,
key: KeySerializer::new(U32_LEN + 1)
.write(account_id)
.write(collection)
.write(u32::MAX)
.finalize(),
},
)
.await
}
pub async fn get_bitmap(
server: &Server,
from_key: AnyKey<Vec<u8>>,
to_key: AnyKey<Vec<u8>>,
) -> trc::Result<Option<RoaringBitmap>> {
let mut results = RoaringBitmap::new();
server
.core
.storage
.data
.iterate(
IterateParams::new(from_key, to_key).no_values(),
|key, _| {
results.insert(key.deserialize_be_u32(key.len() - U32_LEN)?);
Ok(true)
},
)
.await
.caused_by(trc::location!())
.map(|_| {
if !results.is_empty() {
Some(results)
} else {
None
}
})
}
pub struct LegacyBincode<T: serde::de::DeserializeOwned> {
pub inner: T,
}
crates/migration/src/mailbox.rs
+7 -5
View File
@@ -5,12 +5,15 @@
*/
use super::object::Object;
use crate::object::{FromLegacy, Property, Value};
use crate::{
get_document_ids,
object::{FromLegacy, Property, Value},
v014::{SUBSPACE_BITMAP_TAG, SUBSPACE_BITMAP_TEXT},
};
use common::Server;
use email::mailbox::Mailbox;
use store::{
SUBSPACE_BITMAP_TAG, SUBSPACE_BITMAP_TEXT, SUBSPACE_INDEXES, Serialize, U64_LEN, ValueKey,
rand,
SUBSPACE_INDEXES, Serialize, U64_LEN, ValueKey, rand,
write::{
AlignedBytes, AnyKey, Archive, Archiver, BatchBuilder, ValueClass, key::KeySerializer,
},
@@ -21,8 +24,7 @@ use utils::config::utils::ParseValue;
pub(crate) async fn migrate_mailboxes(server: &Server, account_id: u32) -> trc::Result<u64> {
// Obtain email ids
let mailbox_ids = server
.get_document_ids(account_id, Collection::Mailbox)
let mailbox_ids = get_document_ids(server, account_id, Collection::Mailbox)
.await
.caused_by(trc::location!())?
.unwrap_or_default();
crates/migration/src/object.rs
+3 -1
View File
@@ -332,7 +332,9 @@ impl DeserializeFrom for Keyword {
for _ in 0..len {
keyword.push(*bytes.next()?);
}
Some(Keyword::Other(String::from_utf8(keyword).ok()?))
Some(Keyword::Other(
String::from_utf8(keyword).ok()?.into_boxed_str(),
))
}
}
}
crates/migration/src/principal_v1.rs
+8 -8
View File
@@ -5,9 +5,10 @@
*/
use crate::{
email::migrate_emails, encryption::migrate_encryption_params, identity::migrate_identities,
mailbox::migrate_mailboxes, push_v1::migrate_push_subscriptions_v011,
sieve_v1::migrate_sieve_v011, submission::migrate_email_submissions, threads::migrate_threads,
email_v1::migrate_emails_v011, encryption_v1::migrate_encryption_params_v011, get_document_ids,
identity_v1::migrate_identities_v011, mailbox::migrate_mailboxes,
push_v1::migrate_push_subscriptions_v011, sieve_v1::migrate_sieve_v011,
submission::migrate_email_submissions, threads::migrate_threads,
};
use common::Server;
use directory::{
@@ -29,8 +30,7 @@ use utils::codec::leb128::Leb128Iterator;
pub(crate) async fn migrate_principals_v0_11(server: &Server) -> trc::Result<RoaringBitmap> {
// Obtain email ids
let principal_ids = server
.get_document_ids(u32::MAX, Collection::Principal)
let principal_ids = get_document_ids(server, u32::MAX, Collection::Principal)
.await
.caused_by(trc::location!())?
.unwrap_or_default();
@@ -122,13 +122,13 @@ pub(crate) async fn migrate_principals_v0_11(server: &Server) -> trc::Result<Roa
pub(crate) async fn migrate_principal_v0_11(server: &Server, account_id: u32) -> trc::Result<()> {
let start_time = Instant::now();
let num_emails = migrate_emails(server, account_id)
let num_emails = migrate_emails_v011(server, account_id)
.await
.caused_by(trc::location!())?;
let num_mailboxes = migrate_mailboxes(server, account_id)
.await
.caused_by(trc::location!())?;
let num_params = migrate_encryption_params(server, account_id)
let num_params = migrate_encryption_params_v011(server, account_id)
.await
.caused_by(trc::location!())?;
let num_subscriptions = migrate_push_subscriptions_v011(server, account_id)
@@ -143,7 +143,7 @@ pub(crate) async fn migrate_principal_v0_11(server: &Server, account_id: u32) ->
let num_threads = migrate_threads(server, account_id)
.await
.caused_by(trc::location!())?;
let num_identities = migrate_identities(server, account_id)
let num_identities = migrate_identities_v011(server, account_id)
.await
.caused_by(trc::location!())?;
crates/migration/src/principal_v2.rs
+2 -2
View File
@@ -22,14 +22,14 @@ use crate::{
calendar_v2::migrate_calendar_v013,
contact_v2::migrate_contacts_v013,
event_v2::{migrate_calendar_events_v013, migrate_calendar_scheduling_v013},
get_document_ids,
push_v2::migrate_push_subscriptions_v013,
sieve_v2::migrate_sieve_v013,
};
pub(crate) async fn migrate_principals_v0_13(server: &Server) -> trc::Result<RoaringBitmap> {
// Obtain email ids
let principal_ids = server
.get_document_ids(u32::MAX, Collection::Principal)
let principal_ids = get_document_ids(server, u32::MAX, Collection::Principal)
.await
.caused_by(trc::location!())?
.unwrap_or_default();
crates/migration/src/push_v1.rs
+11 -7
View File
@@ -5,7 +5,10 @@
*/
use super::object::Object;
use crate::object::{FromLegacy, Property, Value};
use crate::{
get_document_ids,
object::{FromLegacy, Property, Value},
};
use base64::{Engine, engine::general_purpose};
use common::Server;
use email::push::{Keys, PushSubscription, PushSubscriptions};
@@ -25,8 +28,7 @@ pub(crate) async fn migrate_push_subscriptions_v011(
account_id: u32,
) -> trc::Result<u64> {
// Obtain email ids
let push_subscription_ids = server
.get_document_ids(account_id, Collection::PushSubscription)
let push_subscription_ids = get_document_ids(server, account_id, Collection::PushSubscription)
.await
.caused_by(trc::location!())?
.unwrap_or_default();
@@ -69,12 +71,14 @@ pub(crate) async fn migrate_push_subscriptions_v011(
batch
.with_account_id(u32::MAX)
.with_collection(Collection::PushSubscription)
.create_document(account_id)
.with_account_id(account_id);
.with_collection(Collection::Principal)
.with_document(account_id)
.tag(PrincipalField::PushSubscriptions)
.with_account_id(account_id)
.with_collection(Collection::PushSubscription);
for subscription in &subscriptions {
batch.delete_document(subscription.id).clear(Field::ARCHIVE);
batch.with_document(subscription.id).clear(Field::ARCHIVE);
}
batch
crates/migration/src/push_v2.rs
+10 -9
View File
@@ -7,8 +7,8 @@
use common::Server;
use email::push::{Keys, PushSubscription, PushSubscriptions};
use store::{
Serialize,
write::{Archiver, BatchBuilder, now},
Serialize, ValueKey,
write::{AlignedBytes, Archive, Archiver, BatchBuilder, now},
};
use trc::AddContext;
use types::{
@@ -18,13 +18,14 @@ use types::{
};
use utils::map::bitmap::Bitmap;
use crate::get_document_ids;
pub(crate) async fn migrate_push_subscriptions_v013(
server: &Server,
account_id: u32,
) -> trc::Result<u64> {
// Obtain email ids
let push_ids = server
.get_document_ids(account_id, Collection::PushSubscription)
let push_ids = get_document_ids(server, account_id, Collection::PushSubscription)
.await
.caused_by(trc::location!())?
.unwrap_or_default();
@@ -78,10 +79,9 @@ pub(crate) async fn migrate_push_subscriptions_v013(
// Delete archived and document ids
batch
.with_account_id(account_id)
.with_collection(Collection::PushSubscription)
.create_document(account_id);
.with_collection(Collection::PushSubscription);
for subscription in &subscriptions {
batch.delete_document(subscription.id).clear(Field::ARCHIVE);
batch.with_document(subscription.id).clear(Field::ARCHIVE);
}
subscriptions.retain(|s| s.verified && s.expires > now);
@@ -89,8 +89,9 @@ pub(crate) async fn migrate_push_subscriptions_v013(
if !subscriptions.is_empty() {
batch
.with_account_id(u32::MAX)
.with_collection(Collection::PushSubscription)
.create_document(account_id)
.with_collection(Collection::Principal)
.with_document(account_id)
.tag(PrincipalField::PushSubscriptions)
.with_account_id(account_id)
.with_collection(Collection::Principal)
.with_document(0)
crates/migration/src/queue.rs → crates/migration/src/queue_v1.rs
+50 -31
View File
@@ -4,15 +4,19 @@
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
*/
use crate::LegacyBincode;
use crate::{
LegacyBincode,
queue_v2::{LegacyHostResponse, LegacyQuotaKey},
};
use common::{
Server,
config::smtp::queue::{DEFAULT_QUEUE_NAME, QueueExpiry, QueueName},
};
use smtp::queue::{
Error, ErrorDetails, HostResponse, Message, QueueId, QuotaKey, Recipient, Schedule, Status,
Error, ErrorDetails, HostResponse, Message, QueueId, Recipient, Schedule, Status,
UnexpectedResponse,
};
use smtp_proto::Response;
use std::net::{IpAddr, Ipv4Addr};
use store::{
IterateParams, SUBSPACE_QUEUE_EVENT, Serialize, U64_LEN, ValueKey,
@@ -272,7 +276,7 @@ where
Message {
created: message.created,
blob_hash: message.blob_hash,
return_path: message.return_path_lcase,
return_path: message.return_path_lcase.into_boxed_str(),
recipients: message
.recipients
.into_iter()
@@ -289,7 +293,14 @@ where
Status::PermanentFailure(migrate_legacy_error(&domain.domain, err))
}
},
Status::Completed(details) => Status::Completed(details),
Status::Completed(details) => Status::Completed(HostResponse {
hostname: details.hostname.into_boxed_str(),
response: Response {
code: details.response.code,
esc: details.response.esc,
message: details.response.message.into_boxed_str(),
},
}),
Status::TemporaryFailure(err) => {
Status::TemporaryFailure(migrate_host_response(err))
}
@@ -298,7 +309,7 @@ where
}
};
rcpt.flags = r.flags;
rcpt.orcpt = r.orcpt;
rcpt.orcpt = r.orcpt.map(|o| o.into_boxed_str());
rcpt.retry = domain.retry.clone();
rcpt.notify = domain.notify.clone();
rcpt.queue = QueueName::default();
@@ -307,10 +318,10 @@ where
})
.collect(),
flags: message.flags,
env_id: message.env_id,
env_id: message.env_id.map(|e| e.into_boxed_str()),
priority: message.priority,
size: message.size.as_u64(),
quota_keys: message.quota_keys,
quota_keys: message.quota_keys.into_iter().map(Into::into).collect(),
received_from_ip: IpAddr::V4(Ipv4Addr::LOCALHOST),
received_via_port: 0,
}
@@ -339,53 +350,61 @@ impl AsU64 for u64 {
fn migrate_legacy_error(domain: &str, err: &LegacyError) -> ErrorDetails {
match err {
LegacyError::DnsError(err) => ErrorDetails {
entity: domain.to_string(),
details: Error::DnsError(err.clone()),
entity: domain.into(),
details: Error::DnsError(err.as_str().into()),
},
LegacyError::UnexpectedResponse(err) => ErrorDetails {
entity: err.hostname.entity.to_string(),
entity: err.hostname.entity.as_str().into(),
details: Error::UnexpectedResponse(UnexpectedResponse {
command: err.hostname.details.clone(),
response: err.response.clone(),
command: err.hostname.details.as_str().into(),
response: Response {
code: err.response.code,
esc: err.response.esc,
message: err.response.message.as_str().into(),
},
}),
},
LegacyError::ConnectionError(err) => ErrorDetails {
entity: err.entity.to_string(),
details: Error::ConnectionError(err.details.clone()),
entity: err.entity.as_str().into(),
details: Error::ConnectionError(err.details.as_str().into()),
},
LegacyError::TlsError(err) => ErrorDetails {
entity: err.entity.to_string(),
details: Error::TlsError(err.details.clone()),
entity: err.entity.as_str().into(),
details: Error::TlsError(err.details.as_str().into()),
},
LegacyError::DaneError(err) => ErrorDetails {
entity: err.entity.to_string(),
details: Error::DaneError(err.details.clone()),
entity: err.entity.as_str().into(),
details: Error::DaneError(err.details.as_str().into()),
},
LegacyError::MtaStsError(err) => ErrorDetails {
entity: domain.to_string(),
details: Error::MtaStsError(err.clone()),
entity: domain.into(),
details: Error::MtaStsError(err.as_str().into()),
},
LegacyError::RateLimited => ErrorDetails {
entity: domain.to_string(),
entity: domain.into(),
details: Error::RateLimited,
},
LegacyError::ConcurrencyLimited => ErrorDetails {
entity: domain.to_string(),
entity: domain.into(),
details: Error::ConcurrencyLimited,
},
LegacyError::Io(err) => ErrorDetails {
entity: domain.to_string(),
details: Error::Io(err.clone()),
entity: domain.into(),
details: Error::Io(err.as_str().into()),
},
}
}
fn migrate_host_response(response: HostResponse<LegacyErrorDetails>) -> ErrorDetails {
fn migrate_host_response(response: LegacyHostResponse<LegacyErrorDetails>) -> ErrorDetails {
ErrorDetails {
entity: response.hostname.entity,
entity: response.hostname.entity.into_boxed_str(),
details: Error::UnexpectedResponse(UnexpectedResponse {
command: response.hostname.details,
response: response.response,
command: response.hostname.details.into_boxed_str(),
response: Response {
code: response.response.code,
esc: response.response.esc,
message: response.response.message.into_boxed_str(),
},
}),
}
}
@@ -419,7 +438,7 @@ pub struct LegacyMessage<SIZE, IDX> {
pub priority: i16,
pub size: SIZE,
pub quota_keys: Vec<QuotaKey>,
pub quota_keys: Vec<LegacyQuotaKey>,
#[serde(skip)]
#[rkyv(with = rkyv::with::Skip)]
@@ -440,7 +459,7 @@ pub struct LegacyRecipient<IDX> {
pub domain_idx: IDX,
pub address: String,
pub address_lcase: String,
pub status: Status<HostResponse<String>, HostResponse<LegacyErrorDetails>>,
pub status: Status<LegacyHostResponse<String>, LegacyHostResponse<LegacyErrorDetails>>,
pub flags: u64,
pub orcpt: Option<String>,
}
@@ -475,7 +494,7 @@ pub struct LegacyDomain {
)]
pub enum LegacyError {
DnsError(String),
UnexpectedResponse(HostResponse<LegacyErrorDetails>),
UnexpectedResponse(LegacyHostResponse<LegacyErrorDetails>),
ConnectionError(LegacyErrorDetails),
TlsError(LegacyErrorDetails),
DaneError(LegacyErrorDetails),
crates/migration/src/queue_v2.rs
+332
View File
@@ -0,0 +1,332 @@
/*
* SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <hello@stalw.art>
*
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
*/
use common::{
Server,
config::smtp::queue::{QueueExpiry, QueueName},
};
use smtp::queue::{
Error, ErrorDetails, HostResponse, Message, QuotaKey, Recipient, Schedule, Status,
UnexpectedResponse,
};
use smtp_proto::Response;
use std::net::IpAddr;
use store::{
Deserialize, IterateParams, Serialize, ValueKey,
write::{
AlignedBytes, Archive, Archiver, BatchBuilder, QueueClass, ValueClass,
key::DeserializeBigEndian, now,
},
};
use trc::AddContext;
use types::blob_hash::BlobHash;
#[derive(rkyv::Serialize, rkyv::Deserialize, rkyv::Archive, Debug, Clone, PartialEq, Eq)]
pub struct LegacyMessage {
pub created: u64,
pub blob_hash: BlobHash,
pub return_path: String,
pub recipients: Vec<LegacyRecipient>,
pub received_from_ip: IpAddr,
pub received_via_port: u16,
pub flags: u64,
pub env_id: Option<String>,
pub priority: i16,
pub size: u64,
pub quota_keys: Vec<LegacyQuotaKey>,
}
#[derive(
rkyv::Serialize,
rkyv::Deserialize,
rkyv::Archive,
Debug,
Clone,
PartialEq,
Eq,
serde::Deserialize,
)]
pub struct LegacyRecipient {
pub address: String,
pub retry: Schedule<u32>,
pub notify: Schedule<u32>,
pub expires: QueueExpiry,
pub queue: QueueName,
pub status: Status<LegacyHostResponse<String>, LegacyErrorDetails>,
pub flags: u64,
pub orcpt: Option<String>,
}
#[derive(
Debug,
Clone,
PartialEq,
Eq,
rkyv::Serialize,
rkyv::Deserialize,
rkyv::Archive,
serde::Deserialize,
)]
pub struct LegacyHostResponse<T> {
pub hostname: T,
pub response: Response<String>,
}
#[derive(
Debug,
Clone,
PartialEq,
Eq,
rkyv::Serialize,
rkyv::Deserialize,
rkyv::Archive,
serde::Deserialize,
)]
pub struct LegacyUnexpectedResponse {
pub command: String,
pub response: Response<String>,
}
#[derive(
Debug,
Clone,
PartialEq,
Eq,
rkyv::Serialize,
rkyv::Deserialize,
rkyv::Archive,
Default,
serde::Deserialize,
)]
pub struct LegacyErrorDetails {
pub entity: String,
pub details: LegacyError,
}
#[derive(
Debug,
Clone,
PartialEq,
Eq,
rkyv::Serialize,
rkyv::Deserialize,
rkyv::Archive,
serde::Deserialize,
Default,
)]
pub enum LegacyError {
DnsError(String),
UnexpectedResponse(LegacyUnexpectedResponse),
ConnectionError(String),
TlsError(String),
DaneError(String),
MtaStsError(String),
RateLimited,
#[default]
ConcurrencyLimited,
Io(String),
}
#[derive(
rkyv::Serialize,
rkyv::Deserialize,
rkyv::Archive,
Debug,
Clone,
PartialEq,
Eq,
serde::Deserialize,
)]
pub enum LegacyQuotaKey {
Size { key: Vec<u8>, id: u64 },
Count { key: Vec<u8>, id: u64 },
}
pub(crate) async fn migrate_queue_v014(server: &Server) -> trc::Result<()> {
let mut count = 0;
let now = now();
let mut messages = Vec::new();
server
.store()
.iterate(
IterateParams::new(
ValueKey::from(ValueClass::Queue(QueueClass::Message(0))),
ValueKey::from(ValueClass::Queue(QueueClass::Message(u64::MAX))),
),
|key, value| {
let archive = <Archive<AlignedBytes> as Deserialize>::deserialize(value)
.caused_by(trc::location!())?;
match archive.deserialize_untrusted::<LegacyMessage>() {
Ok(message) => {
messages.push((key.deserialize_be_u64(0)?, Message::from(message)));
}
Err(err) => {
if archive.deserialize_untrusted::<Message>().is_err() {
return Err(err.caused_by(trc::location!()));
}
}
}
count += 1;
Ok(true)
},
)
.await
.caused_by(trc::location!());
let mut batch = BatchBuilder::new();
for (queue_id, message) in messages {
batch.set(
ValueClass::Queue(QueueClass::Message(queue_id)),
Archiver::new(message)
.serialize()
.caused_by(trc::location!())?,
);
if batch.is_large_batch() {
server
.store()
.write(batch.build_all())
.await
.caused_by(trc::location!())?;
batch = BatchBuilder::new();
}
}
if !batch.is_empty() {
server
.store()
.write(batch.build_all())
.await
.caused_by(trc::location!())?;
}
if count > 0 {
trc::event!(
Server(trc::ServerEvent::Startup),
Details = format!("Migrated {count} queued messages",)
);
}
Ok(())
}
impl From<LegacyMessage> for Message {
fn from(legacy: LegacyMessage) -> Self {
Message {
created: legacy.created,
blob_hash: legacy.blob_hash,
return_path: legacy.return_path.into_boxed_str(),
recipients: legacy.recipients.into_iter().map(|r| r.into()).collect(),
received_from_ip: legacy.received_from_ip,
received_via_port: legacy.received_via_port,
flags: legacy.flags,
env_id: legacy.env_id.map(|s| s.into_boxed_str()),
priority: legacy.priority,
size: legacy.size,
quota_keys: legacy.quota_keys.into_iter().map(|qk| qk.into()).collect(),
}
}
}
impl From<LegacyRecipient> for Recipient {
fn from(legacy: LegacyRecipient) -> Self {
Recipient {
address: legacy.address.into_boxed_str(),
retry: legacy.retry,
notify: legacy.notify,
expires: legacy.expires,
queue: legacy.queue,
status: match legacy.status {
Status::Scheduled => Status::Scheduled,
Status::Completed(status) => Status::Completed(status.into()),
Status::TemporaryFailure(status) => Status::TemporaryFailure(status.into()),
Status::PermanentFailure(status) => Status::PermanentFailure(status.into()),
},
flags: legacy.flags,
orcpt: legacy.orcpt.map(|s| s.into_boxed_str()),
}
}
}
impl From<LegacyErrorDetails> for ErrorDetails {
fn from(legacy: LegacyErrorDetails) -> Self {
ErrorDetails {
entity: legacy.entity.into_boxed_str(),
details: legacy.details.into(),
}
}
}
impl From<LegacyQuotaKey> for QuotaKey {
fn from(legacy: LegacyQuotaKey) -> Self {
match legacy {
LegacyQuotaKey::Size { key, id } => QuotaKey::Size {
key: key.into(),
id,
},
LegacyQuotaKey::Count { key, id } => QuotaKey::Count {
key: key.into(),
id,
},
}
}
}
impl From<LegacyError> for Error {
fn from(legacy: LegacyError) -> Self {
match legacy {
LegacyError::DnsError(s) => Error::DnsError(s.into_boxed_str()),
LegacyError::UnexpectedResponse(ur) => Error::UnexpectedResponse(ur.into()),
LegacyError::ConnectionError(s) => Error::ConnectionError(s.into_boxed_str()),
LegacyError::TlsError(s) => Error::TlsError(s.into_boxed_str()),
LegacyError::DaneError(s) => Error::DaneError(s.into_boxed_str()),
LegacyError::MtaStsError(s) => Error::MtaStsError(s.into_boxed_str()),
LegacyError::RateLimited => Error::RateLimited,
LegacyError::ConcurrencyLimited => Error::ConcurrencyLimited,
LegacyError::Io(s) => Error::Io(s.into_boxed_str()),
}
}
}
impl From<LegacyUnexpectedResponse> for UnexpectedResponse {
fn from(legacy: LegacyUnexpectedResponse) -> Self {
UnexpectedResponse {
command: legacy.command.into_boxed_str(),
response: Response {
code: legacy.response.code,
esc: legacy.response.esc,
message: legacy.response.message.into_boxed_str(),
},
}
}
}
impl From<LegacyHostResponse<String>> for HostResponse<Box<str>> {
fn from(legacy: LegacyHostResponse<String>) -> Self {
HostResponse {
hostname: legacy.hostname.into_boxed_str(),
response: Response {
code: legacy.response.code,
esc: legacy.response.esc,
message: legacy.response.message.into_boxed_str(),
},
}
}
}
crates/migration/src/sieve_v1.rs
+7 -5
View File
@@ -5,12 +5,15 @@
*/
use super::object::Object;
use crate::object::{Property, TryFromLegacy, Value};
use crate::{
get_document_ids,
object::{Property, TryFromLegacy, Value},
v014::SUBSPACE_BITMAP_TEXT,
};
use common::Server;
use email::sieve::{SieveScript, VacationResponse};
use store::{
SUBSPACE_BITMAP_TEXT, SUBSPACE_INDEXES, SUBSPACE_PROPERTY, Serialize, SerializeInfallible,
U64_LEN, ValueKey,
SUBSPACE_INDEXES, SUBSPACE_PROPERTY, Serialize, SerializeInfallible, U64_LEN, ValueKey,
write::{
AlignedBytes, AnyKey, Archive, Archiver, BatchBuilder, ValueClass, key::KeySerializer,
},
@@ -23,8 +26,7 @@ use types::{
pub(crate) async fn migrate_sieve_v011(server: &Server, account_id: u32) -> trc::Result<u64> {
// Obtain email ids
let script_ids = server
.get_document_ids(account_id, Collection::SieveScript)
let script_ids = get_document_ids(server, account_id, Collection::SieveScript)
.await
.caused_by(trc::location!())?
.unwrap_or_default();
crates/migration/src/sieve_v2.rs
+5 -4
View File
@@ -7,8 +7,8 @@
use common::Server;
use email::sieve::{SieveScript, VacationResponse};
use store::{
Serialize, SerializeInfallible,
write::{Archiver, BatchBuilder},
Serialize, SerializeInfallible, ValueKey,
write::{AlignedBytes, Archive, Archiver, BatchBuilder},
};
use trc::AddContext;
use types::{
@@ -17,10 +17,11 @@ use types::{
field::{Field, PrincipalField},
};
use crate::get_document_ids;
pub(crate) async fn migrate_sieve_v013(server: &Server, account_id: u32) -> trc::Result<u64> {
// Obtain email ids
let script_ids = server
.get_document_ids(account_id, Collection::SieveScript)
let script_ids = get_document_ids(server, account_id, Collection::SieveScript)
.await
.caused_by(trc::location!())?
.unwrap_or_default();
crates/migration/src/submission.rs
+21 -11
View File
@@ -5,16 +5,20 @@
*/
use super::object::Object;
use crate::object::{FromLegacy, Property, Value};
use crate::{
get_document_ids,
object::{FromLegacy, Property, Value},
v014::{SUBSPACE_BITMAP_TAG, SUBSPACE_BITMAP_TEXT},
};
use common::Server;
use email::submission::{
Address, Delivered, DeliveryStatus, EmailSubmission, Envelope, UndoStatus,
};
use store::{
SUBSPACE_BITMAP_TAG, SUBSPACE_BITMAP_TEXT, SUBSPACE_INDEXES, Serialize, SerializeInfallible,
U64_LEN, ValueKey,
SUBSPACE_INDEXES, Serialize, U32_LEN, U64_LEN, ValueKey,
write::{
AlignedBytes, AnyKey, Archive, Archiver, BatchBuilder, ValueClass, key::KeySerializer,
AlignedBytes, AnyKey, Archive, Archiver, BatchBuilder, IndexPropertyClass, ValueClass,
key::KeySerializer,
},
};
use trc::AddContext;
@@ -29,8 +33,7 @@ pub(crate) async fn migrate_email_submissions(
account_id: u32,
) -> trc::Result<u64> {
// Obtain email ids
let email_submission_ids = server
.get_document_ids(account_id, Collection::EmailSubmission)
let email_submission_ids = get_document_ids(server, account_id, Collection::EmailSubmission)
.await
.caused_by(trc::location!())?
.unwrap_or_default();
@@ -83,11 +86,18 @@ pub(crate) async fn migrate_email_submissions(
.with_account_id(account_id)
.with_collection(Collection::EmailSubmission)
.with_document(email_submission_id)
.index(EmailSubmissionField::UndoStatus, es.undo_status.as_index())
.index(EmailSubmissionField::EmailId, es.email_id.serialize())
.index(EmailSubmissionField::ThreadId, es.thread_id.serialize())
.index(EmailSubmissionField::IdentityId, es.identity_id.serialize())
.index(EmailSubmissionField::SendAt, es.send_at.serialize())
.set(
ValueClass::IndexProperty(IndexPropertyClass::Integer {
property: EmailSubmissionField::Metadata.into(),
value: es.send_at,
}),
KeySerializer::new(U32_LEN * 3 + 1)
.write(es.email_id)
.write(es.thread_id)
.write(es.identity_id)
.write(es.undo_status.as_index())
.finalize(),
)
.set(
Field::ARCHIVE,
Archiver::new(es).serialize().caused_by(trc::location!())?,
crates/migration/src/tasks.rs → crates/migration/src/tasks_v1.rs
+14 -4
View File
@@ -6,8 +6,12 @@
use common::Server;
use store::{
IterateParams, SUBSPACE_TASK_QUEUE, ValueKey,
write::{AnyClass, BatchBuilder, TaskQueueClass, ValueClass, key::DeserializeBigEndian, now},
IterateParams, SUBSPACE_TASK_QUEUE, U64_LEN, ValueKey,
write::{
AnyClass, BatchBuilder, ValueClass,
key::{DeserializeBigEndian, KeySerializer},
now,
},
};
use trc::AddContext;
@@ -16,13 +20,19 @@ pub(crate) async fn migrate_tasks_v011(server: &Server) -> trc::Result<()> {
account_id: 0,
collection: 0,
document_id: 0,
class: ValueClass::TaskQueue(TaskQueueClass::IndexEmail { due: 0 }),
class: ValueClass::Any(AnyClass {
subspace: SUBSPACE_TASK_QUEUE,
key: KeySerializer::new(U64_LEN).write(0u64).finalize(),
}),
};
let to_key = ValueKey::<ValueClass> {
account_id: u32::MAX,
collection: u8::MAX,
document_id: u32::MAX,
class: ValueClass::TaskQueue(TaskQueueClass::IndexEmail { due: u64::MAX }),
class: ValueClass::Any(AnyClass {
subspace: SUBSPACE_TASK_QUEUE,
key: KeySerializer::new(U64_LEN).write(u64::MAX).finalize(),
}),
};
let now = now();
crates/migration/src/tasks_v2.rs
+79
View File
@@ -0,0 +1,79 @@
/*
* SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <hello@stalw.art>
*
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
*/
use common::Server;
use store::{
IterateParams, SUBSPACE_TASK_QUEUE, U32_LEN, U64_LEN, ValueKey,
write::{AnyClass, BatchBuilder, ValueClass, key::KeySerializer},
};
use trc::AddContext;
pub(crate) async fn migrate_tasks_v014(server: &Server) -> trc::Result<()> {
let from_key = ValueKey::<ValueClass> {
account_id: 0,
collection: 0,
document_id: 0,
class: ValueClass::Any(AnyClass {
subspace: SUBSPACE_TASK_QUEUE,
key: KeySerializer::new(U64_LEN).write(0u64).finalize(),
}),
};
let to_key = ValueKey::<ValueClass> {
account_id: u32::MAX,
collection: u8::MAX,
document_id: u32::MAX,
class: ValueClass::Any(AnyClass {
subspace: SUBSPACE_TASK_QUEUE,
key: KeySerializer::new(U64_LEN).write(u64::MAX).finalize(),
}),
};
let mut delete_tasks = Vec::new();
server
.core
.storage
.data
.iterate(
IterateParams::new(from_key, to_key).ascending().no_values(),
|key, _| {
match key.get(U64_LEN + U32_LEN) {
Some(0..=2) => {
delete_tasks.push(key.to_vec());
}
None => {
return Err(trc::Error::corrupted_key(key, None, trc::location!()));
}
_ => {}
};
Ok(true)
},
)
.await
.caused_by(trc::location!())?;
if !delete_tasks.is_empty() {
let num_migrated = delete_tasks.len();
let mut batch = BatchBuilder::new();
for key in delete_tasks {
batch.clear(ValueClass::Any(AnyClass {
subspace: SUBSPACE_TASK_QUEUE,
key,
}));
}
server
.store()
.write(batch.build_all())
.await
.caused_by(trc::location!())?;
trc::event!(
Server(trc::ServerEvent::Startup),
Details = format!("Migrated {num_migrated} tasks")
);
}
Ok(())
}
crates/migration/src/threads.rs
+4 -3
View File
@@ -6,16 +6,17 @@
use common::Server;
use store::{
SUBSPACE_BITMAP_ID, U64_LEN,
U64_LEN,
write::{AnyKey, key::KeySerializer},
};
use trc::AddContext;
use types::collection::Collection;
use crate::{get_document_ids, v014::SUBSPACE_BITMAP_ID};
pub(crate) async fn migrate_threads(server: &Server, account_id: u32) -> trc::Result<u64> {
// Obtain email ids
let thread_ids = server
.get_document_ids(account_id, Collection::Thread)
let thread_ids = get_document_ids(server, account_id, Collection::Thread)
.await
.caused_by(trc::location!())?
.unwrap_or_default();
crates/migration/src/v011.rs
+3 -3
View File
@@ -7,8 +7,9 @@
use crate::{
LOCK_RETRY_TIME, LOCK_WAIT_TIME_ACCOUNT, LOCK_WAIT_TIME_CORE,
changelog::reset_changelog,
get_document_ids,
principal_v1::{migrate_principal_v0_11, migrate_principals_v0_11},
queue::migrate_queue_v011,
queue_v1::migrate_queue_v011,
report::migrate_reports,
};
use common::{KV_LOCK_HOUSEKEEPER, Server};
@@ -64,8 +65,7 @@ pub(crate) async fn migrate_v0_11(server: &Server) -> trc::Result<()> {
.await
.caused_by(trc::location!())?;
} else {
principal_ids = server
.get_document_ids(u32::MAX, Collection::Principal)
principal_ids = get_document_ids(server, u32::MAX, Collection::Principal)
.await
.caused_by(trc::location!())?
.unwrap_or_default();
crates/migration/src/v012.rs
+1 -1
View File
@@ -6,7 +6,7 @@
use crate::{
LOCK_RETRY_TIME, LOCK_WAIT_TIME_CORE, event_v1::migrate_calendar_events_v012,
queue::migrate_queue_v012, tasks::migrate_tasks_v011,
queue_v1::migrate_queue_v012, tasks_v1::migrate_tasks_v011,
};
use common::{KV_LOCK_HOUSEKEEPER, Server};
use trc::AddContext;
crates/migration/src/v013.rs
+2 -3
View File
@@ -5,7 +5,7 @@
*/
use crate::{
LOCK_RETRY_TIME, LOCK_WAIT_TIME_ACCOUNT, LOCK_WAIT_TIME_CORE,
LOCK_RETRY_TIME, LOCK_WAIT_TIME_ACCOUNT, LOCK_WAIT_TIME_CORE, get_document_ids,
principal_v2::{migrate_principal_v0_13, migrate_principals_v0_13},
};
use common::{KV_LOCK_HOUSEKEEPER, Server};
@@ -56,8 +56,7 @@ pub(crate) async fn migrate_v0_13(server: &Server) -> trc::Result<()> {
.await
.caused_by(trc::location!())?;
} else {
principal_ids = server
.get_document_ids(u32::MAX, Collection::Principal)
principal_ids = get_document_ids(server, u32::MAX, Collection::Principal)
.await
.caused_by(trc::location!())?
.unwrap_or_default();
crates/migration/src/v014.rs
+95
View File
@@ -0,0 +1,95 @@
/*
* SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <hello@stalw.art>
*
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
*/
use common::Server;
use store::{
SUBSPACE_INDEXES, U64_LEN,
write::{AnyKey, key::KeySerializer},
};
use trc::AddContext;
use types::collection::Collection;
pub const SUBSPACE_BITMAP_ID: u8 = b'b';
pub const SUBSPACE_BITMAP_TAG: u8 = b'c';
pub const SUBSPACE_BITMAP_TEXT: u8 = b'v';
pub const SUBSPACE_FTS_INDEX: u8 = b'g';
pub const SUBSPACE_TELEMETRY_INDEX: u8 = b'w';
pub(crate) async fn migrate_v0_14(server: &Server) -> trc::Result<()> {
/*
- ContactField
- CalendarField
- EmailSubmissionField
- CalendarNotificationField
*/
todo!()
}
pub(crate) async fn migrate_indexes(server: &Server, account_id: u32) -> trc::Result<()> {
/*
EmailSubmissionField::UndoStatus => 41,
EmailSubmissionField::EmailId => 83,
EmailSubmissionField::ThreadId => 33,
EmailSubmissionField::IdentityId => 95,
EmailSubmissionField::SendAt => 24,
*/
/*
ContactField::Created => 2,
ContactField::Updated => 3,
ContactField::Text => 4,
*/
/*
CalendarField::Text => 1,
CalendarField::Created => 2,
CalendarField::Updated => 3,
CalendarField::Start => 4,
CalendarField::EventId => 5,
*/
for (collection, fields) in [
(Collection::EmailSubmission, &[41u8, 83, 33, 95, 24][..]),
(Collection::ContactCard, &[2, 3, 4][..]),
(Collection::CalendarEvent, &[1, 2, 3, 4][..]),
(Collection::CalendarEventNotification, &[2, 5][..]),
] {
for index in fields {
server
.store()
.delete_range(
AnyKey {
subspace: SUBSPACE_INDEXES,
key: KeySerializer::new(U64_LEN * 3)
.write(account_id)
.write(u8::from(collection))
.write(*index)
.finalize(),
},
AnyKey {
subspace: SUBSPACE_INDEXES,
key: KeySerializer::new(U64_LEN * 4)
.write(account_id)
.write(u8::from(collection))
.write(*index)
.write(&[u8::MAX; 8][..])
.finalize(),
},
)
.await
.caused_by(trc::location!())?;
}
}
Ok(())
}
crates/nlp/Cargo.toml
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "nlp"
version = "0.14.1"
version = "0.15.0"
edition = "2024"
[dependencies]
crates/pop3/Cargo.toml
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "pop3"
version = "0.14.1"
version = "0.15.0"
edition = "2024"
[dependencies]
crates/services/Cargo.toml
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "services"
version = "0.14.1"
version = "0.15.0"
edition = "2024"
[dependencies]
crates/smtp/Cargo.toml
+1 -1
View File
@@ -7,7 +7,7 @@ homepage = "https://stalw.art/smtp"
keywords = ["smtp", "email", "mail", "server"]
categories = ["email"]
license = "AGPL-3.0-only OR LicenseRef-SEL"
version = "0.14.1"
version = "0.15.0"
edition = "2024"
[dependencies]
crates/smtp/src/inbound/data.rs
+1 -1
View File
@@ -742,7 +742,7 @@ impl<T: SessionStream> Session<T> {
size: 0,
env_id: mail_from.dsn_info.map(|i| i.into_boxed_str()),
blob_hash: Default::default(),
quota_keys: Vec::new(),
quota_keys: Default::default(),
received_from_ip: self.data.remote_ip,
received_via_port: self.data.local_port,
};
crates/smtp/src/queue/mod.rs
+3 -3
View File
@@ -68,7 +68,7 @@ pub struct Message {
pub priority: i16,
pub size: u64,
pub quota_keys: Vec<QuotaKey>,
pub quota_keys: Box<[QuotaKey]>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
@@ -91,8 +91,8 @@ pub struct MessageWrapper {
serde::Deserialize,
)]
pub enum QuotaKey {
Size { key: Vec<u8>, id: u64 },
Count { key: Vec<u8>, id: u64 },
Size { key: Box<[u8]>, id: u64 },
Count { key: Box<[u8]>, id: u64 },
}
#[derive(
crates/smtp/src/queue/quota.rs
+9 -6
View File
@@ -112,7 +112,7 @@ impl HasQueueQuota for Server {
}
}
message.message.quota_keys = quota_keys;
message.message.quota_keys = quota_keys.into_boxed_slice();
true
}
@@ -147,7 +147,7 @@ impl HasQueueQuota for Server {
return false;
} else {
refs.push(QuotaKey::Size {
key: key.as_ref().to_vec(),
key: key.as_ref().into(),
id,
});
}
@@ -167,7 +167,7 @@ impl HasQueueQuota for Server {
return false;
} else {
refs.push(QuotaKey::Count {
key: key.as_ref().to_vec(),
key: key.as_ref().into(),
id,
});
}
@@ -202,11 +202,14 @@ impl MessageWrapper {
for quota_key in std::mem::take(&mut self.message.quota_keys) {
match quota_key {
QuotaKey::Count { id, key } if quota_ids.contains(&id) => {
batch.add(ValueClass::Queue(QueueClass::QuotaCount(key)), -1);
batch.add(
ValueClass::Queue(QueueClass::QuotaCount(key.into_vec())),
-1,
);
}
QuotaKey::Size { id, key } if quota_ids.contains(&id) => {
batch.add(
ValueClass::Queue(QueueClass::QuotaSize(key)),
ValueClass::Queue(QueueClass::QuotaSize(key.into_vec())),
-(self.message.size as i64),
);
}
@@ -215,7 +218,7 @@ impl MessageWrapper {
}
}
}
self.message.quota_keys = quota_keys;
self.message.quota_keys = quota_keys.into_boxed_slice();
}
}
}
crates/smtp/src/queue/spool.rs
+5 -5
View File
@@ -90,7 +90,7 @@ impl SmtpSpool for Server {
priority: 0,
size: 0,
blob_hash: Default::default(),
quota_keys: Vec::new(),
quota_keys: Default::default(),
received_from_ip: IpAddr::V4(Ipv4Addr::LOCALHOST),
received_via_port: 0,
},
@@ -417,11 +417,11 @@ impl MessageWrapper {
for quota_key in &self.message.quota_keys {
match quota_key {
QuotaKey::Count { key, .. } => {
batch.add(ValueClass::Queue(QueueClass::QuotaCount(key.clone())), 1);
batch.add(ValueClass::Queue(QueueClass::QuotaCount(key.to_vec())), 1);
}
QuotaKey::Size { key, .. } => {
batch.add(
ValueClass::Queue(QueueClass::QuotaSize(key.clone())),
ValueClass::Queue(QueueClass::QuotaSize(key.to_vec())),
self.message.size as i64,
);
}
@@ -691,11 +691,11 @@ impl MessageWrapper {
for quota_key in self.message.quota_keys {
match quota_key {
QuotaKey::Count { key, .. } => {
batch.add(ValueClass::Queue(QueueClass::QuotaCount(key)), -1);
batch.add(ValueClass::Queue(QueueClass::QuotaCount(key.to_vec())), -1);
}
QuotaKey::Size { key, .. } => {
batch.add(
ValueClass::Queue(QueueClass::QuotaSize(key)),
ValueClass::Queue(QueueClass::QuotaSize(key.to_vec())),
-(self.message.size as i64),
);
}
crates/spam-filter/Cargo.toml
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "spam-filter"
version = "0.14.1"
version = "0.15.0"
edition = "2024"
[dependencies]
crates/spam-filter/src/modules/classifier.rs
+32 -1
View File
@@ -96,6 +96,17 @@ impl SpamClassifier for Server {
return Ok(());
};
let _permit = self
.inner
.ipc
.train_task_controller
.try_run()
.ok_or_else(|| {
trc::EventType::Spam(SpamEvent::TrainCompleted)
.reason("Spam training task is already running")
.caused_by(trc::location!())
})?;
let started = Instant::now();
trc::event!(Spam(SpamEvent::TrainStarted));
@@ -279,7 +290,16 @@ impl SpamClassifier for Server {
samples.shuffle(&mut StdRng::seed_from_u64(42));
// Spawn training task
let task = trainer.trainer.spawn(config.num_epochs)?;
let epochs = match trainer.reservoir.ham.total_seen
+ trainer.reservoir.spam.total_seen
+ spam_count as u64
+ ham_count as u64
{
0..=2500 => 3, // Bootstrap
2_501..=10_000 => 2, // Refinement
_ => 1, // Full online training
};
let task = trainer.trainer.spawn(epochs)?;
let is_fh = matches!(task, TrainTask::Fh { .. });
// Train
@@ -346,6 +366,17 @@ impl SpamClassifier for Server {
));
}
}
// Look for stop requests
if self.inner.ipc.train_task_controller.should_stop() {
trc::event!(
Spam(SpamEvent::TrainCompleted),
Reason = "Training task was stopped",
Total = fh_samples.len() + ccfh_samples.len(),
Elapsed = started.elapsed()
);
return Ok(());
}
}
// Send batch for training
crates/store/Cargo.toml
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "store"
version = "0.14.1"
version = "0.15.0"
edition = "2024"
[dependencies]
crates/trc/Cargo.toml
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "trc"
version = "0.14.1"
version = "0.15.0"
edition = "2024"
[dependencies]
crates/trc/event-macro/Cargo.toml
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "event_macro"
version = "0.14.1"
version = "0.15.0"
edition = "2024"
[lib]
crates/types/Cargo.toml
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "types"
version = "0.14.1"
version = "0.15.0"
edition = "2024"
[dependencies]
crates/types/src/keyword.rs
+223 -2
View File
@@ -19,7 +19,24 @@ pub const NOTJUNK: usize = 8;
pub const DELETED: usize = 9;
pub const FORWARDED: usize = 10;
pub const MDN_SENT: usize = 11;
pub const OTHER: usize = 12;
pub const AUTOSENT: usize = 12;
pub const CANUNSUBSCRIBE: usize = 13;
pub const FOLLOWED: usize = 14;
pub const HASATTACHMENT: usize = 15;
pub const HASMEMO: usize = 16;
pub const HASNOATTACHMENT: usize = 17;
pub const IMPORTED: usize = 18;
pub const ISTRUSTED: usize = 19;
pub const MAILFLAGBIT0: usize = 20;
pub const MAILFLAGBIT1: usize = 21;
pub const MAILFLAGBIT2: usize = 22;
pub const MASKEDEMAIL: usize = 23;
pub const MEMO: usize = 24;
pub const MUTED: usize = 25;
pub const NEW: usize = 26;
pub const NOTIFY: usize = 27;
pub const UNSUBSCRIBED: usize = 28;
pub const OTHER: usize = 29;
#[derive(
rkyv::Serialize,
@@ -64,6 +81,40 @@ pub enum Keyword {
Forwarded,
#[serde(rename(serialize = "$mdnsent"))]
MdnSent,
#[serde(rename(serialize = "$autosent"))]
Autosent,
#[serde(rename(serialize = "$canunsubscribe"))]
CanUnsubscribe,
#[serde(rename(serialize = "$followed"))]
Followed,
#[serde(rename(serialize = "$hasattachment"))]
HasAttachment,
#[serde(rename(serialize = "$hasmemo"))]
HasMemo,
#[serde(rename(serialize = "$hasnoattachment"))]
HasNoAttachment,
#[serde(rename(serialize = "$imported"))]
Imported,
#[serde(rename(serialize = "$istrusted"))]
IsTrusted,
#[serde(rename(serialize = "$MailFlagBit0"))]
MailFlagBit0,
#[serde(rename(serialize = "$MailFlagBit1"))]
MailFlagBit1,
#[serde(rename(serialize = "$MailFlagBit2"))]
MailFlagBit2,
#[serde(rename(serialize = "$maskedemail"))]
MaskedEmail,
#[serde(rename(serialize = "$memo"))]
Memo,
#[serde(rename(serialize = "$muted"))]
Muted,
#[serde(rename(serialize = "$new"))]
New,
#[serde(rename(serialize = "$notify"))]
Notify,
#[serde(rename(serialize = "$unsubscribed"))]
Unsubscribed,
}
impl Keyword {
@@ -107,7 +158,24 @@ impl Keyword {
"notjunk" => Keyword::NotJunk,
"deleted" => Keyword::Deleted,
"forwarded" => Keyword::Forwarded,
"mdnsent" => Keyword::MdnSent
"mdnsent" => Keyword::MdnSent,
"autosent" => Keyword::Autosent,
"canunsubscribe" => Keyword::CanUnsubscribe,
"followed" => Keyword::Followed,
"hasattachment" => Keyword::HasAttachment,
"hasmemo" => Keyword::HasMemo,
"hasnoattachment" => Keyword::HasNoAttachment,
"imported" => Keyword::Imported,
"istrusted" => Keyword::IsTrusted,
"mailflagbit0" => Keyword::MailFlagBit0,
"mailflagbit1" => Keyword::MailFlagBit1,
"mailflagbit2" => Keyword::MailFlagBit2,
"maskedemail" => Keyword::MaskedEmail,
"memo" => Keyword::Memo,
"muted" => Keyword::Muted,
"new" => Keyword::New,
"notify" => Keyword::Notify,
"unsubscribed" => Keyword::Unsubscribed,
)
})
}
@@ -126,6 +194,23 @@ impl Keyword {
Keyword::Deleted => Ok(DELETED as u32),
Keyword::Forwarded => Ok(FORWARDED as u32),
Keyword::MdnSent => Ok(MDN_SENT as u32),
Keyword::Autosent => Ok(AUTOSENT as u32),
Keyword::CanUnsubscribe => Ok(CANUNSUBSCRIBE as u32),
Keyword::Followed => Ok(FOLLOWED as u32),
Keyword::HasAttachment => Ok(HASATTACHMENT as u32),
Keyword::HasMemo => Ok(HASMEMO as u32),
Keyword::HasNoAttachment => Ok(HASNOATTACHMENT as u32),
Keyword::Imported => Ok(IMPORTED as u32),
Keyword::IsTrusted => Ok(ISTRUSTED as u32),
Keyword::MailFlagBit0 => Ok(MAILFLAGBIT0 as u32),
Keyword::MailFlagBit1 => Ok(MAILFLAGBIT1 as u32),
Keyword::MailFlagBit2 => Ok(MAILFLAGBIT2 as u32),
Keyword::MaskedEmail => Ok(MASKEDEMAIL as u32),
Keyword::Memo => Ok(MEMO as u32),
Keyword::Muted => Ok(MUTED as u32),
Keyword::New => Ok(NEW as u32),
Keyword::Notify => Ok(NOTIFY as u32),
Keyword::Unsubscribed => Ok(UNSUBSCRIBED as u32),
Keyword::Other(string) => Err(string.as_ref()),
}
}
@@ -144,6 +229,23 @@ impl Keyword {
Keyword::Deleted => Ok(DELETED as u32),
Keyword::Forwarded => Ok(FORWARDED as u32),
Keyword::MdnSent => Ok(MDN_SENT as u32),
Keyword::Autosent => Ok(AUTOSENT as u32),
Keyword::CanUnsubscribe => Ok(CANUNSUBSCRIBE as u32),
Keyword::Followed => Ok(FOLLOWED as u32),
Keyword::HasAttachment => Ok(HASATTACHMENT as u32),
Keyword::HasMemo => Ok(HASMEMO as u32),
Keyword::HasNoAttachment => Ok(HASNOATTACHMENT as u32),
Keyword::Imported => Ok(IMPORTED as u32),
Keyword::IsTrusted => Ok(ISTRUSTED as u32),
Keyword::MailFlagBit0 => Ok(MAILFLAGBIT0 as u32),
Keyword::MailFlagBit1 => Ok(MAILFLAGBIT1 as u32),
Keyword::MailFlagBit2 => Ok(MAILFLAGBIT2 as u32),
Keyword::MaskedEmail => Ok(MASKEDEMAIL as u32),
Keyword::Memo => Ok(MEMO as u32),
Keyword::Muted => Ok(MUTED as u32),
Keyword::New => Ok(NEW as u32),
Keyword::Notify => Ok(NOTIFY as u32),
Keyword::Unsubscribed => Ok(UNSUBSCRIBED as u32),
Keyword::Other(string) => Err(string),
}
}
@@ -162,6 +264,23 @@ impl Keyword {
DELETED => Ok(Keyword::Deleted),
FORWARDED => Ok(Keyword::Forwarded),
MDN_SENT => Ok(Keyword::MdnSent),
AUTOSENT => Ok(Keyword::Autosent),
CANUNSUBSCRIBE => Ok(Keyword::CanUnsubscribe),
FOLLOWED => Ok(Keyword::Followed),
HASATTACHMENT => Ok(Keyword::HasAttachment),
HASMEMO => Ok(Keyword::HasMemo),
HASNOATTACHMENT => Ok(Keyword::HasNoAttachment),
IMPORTED => Ok(Keyword::Imported),
ISTRUSTED => Ok(Keyword::IsTrusted),
MAILFLAGBIT0 => Ok(Keyword::MailFlagBit0),
MAILFLAGBIT1 => Ok(Keyword::MailFlagBit1),
MAILFLAGBIT2 => Ok(Keyword::MailFlagBit2),
MASKEDEMAIL => Ok(Keyword::MaskedEmail),
MEMO => Ok(Keyword::Memo),
MUTED => Ok(Keyword::Muted),
NEW => Ok(Keyword::New),
NOTIFY => Ok(Keyword::Notify),
UNSUBSCRIBED => Ok(Keyword::Unsubscribed),
_ => Err(id),
}
}
@@ -188,6 +307,23 @@ impl Display for Keyword {
Keyword::Deleted => write!(f, "$deleted"),
Keyword::Forwarded => write!(f, "$forwarded"),
Keyword::MdnSent => write!(f, "$mdnsent"),
Keyword::Autosent => write!(f, "$autosent"),
Keyword::CanUnsubscribe => write!(f, "$canunsubscribe"),
Keyword::Followed => write!(f, "$followed"),
Keyword::HasAttachment => write!(f, "$hasattachment"),
Keyword::HasMemo => write!(f, "$hasmemo"),
Keyword::HasNoAttachment => write!(f, "$hasnoattachment"),
Keyword::Imported => write!(f, "$imported"),
Keyword::IsTrusted => write!(f, "$istrusted"),
Keyword::MailFlagBit0 => write!(f, "$MailFlagBit0"),
Keyword::MailFlagBit1 => write!(f, "$MailFlagBit1"),
Keyword::MailFlagBit2 => write!(f, "$MailFlagBit2"),
Keyword::MaskedEmail => write!(f, "$maskedemail"),
Keyword::Memo => write!(f, "$memo"),
Keyword::Muted => write!(f, "$muted"),
Keyword::New => write!(f, "$new"),
Keyword::Notify => write!(f, "$notify"),
Keyword::Unsubscribed => write!(f, "$unsubscribed"),
Keyword::Other(s) => write!(f, "{}", s),
}
}
@@ -208,6 +344,23 @@ impl Display for ArchivedKeyword {
ArchivedKeyword::Deleted => write!(f, "$deleted"),
ArchivedKeyword::Forwarded => write!(f, "$forwarded"),
ArchivedKeyword::MdnSent => write!(f, "$mdnsent"),
ArchivedKeyword::Autosent => write!(f, "$autosent"),
ArchivedKeyword::CanUnsubscribe => write!(f, "$canunsubscribe"),
ArchivedKeyword::Followed => write!(f, "$followed"),
ArchivedKeyword::HasAttachment => write!(f, "$hasattachment"),
ArchivedKeyword::HasMemo => write!(f, "$hasmemo"),
ArchivedKeyword::HasNoAttachment => write!(f, "$hasnoattachment"),
ArchivedKeyword::Imported => write!(f, "$imported"),
ArchivedKeyword::IsTrusted => write!(f, "$istrusted"),
ArchivedKeyword::MailFlagBit0 => write!(f, "$MailFlagBit0"),
ArchivedKeyword::MailFlagBit1 => write!(f, "$MailFlagBit1"),
ArchivedKeyword::MailFlagBit2 => write!(f, "$MailFlagBit2"),
ArchivedKeyword::MaskedEmail => write!(f, "$maskedemail"),
ArchivedKeyword::Memo => write!(f, "$memo"),
ArchivedKeyword::Muted => write!(f, "$muted"),
ArchivedKeyword::New => write!(f, "$new"),
ArchivedKeyword::Notify => write!(f, "$notify"),
ArchivedKeyword::Unsubscribed => write!(f, "$unsubscribed"),
ArchivedKeyword::Other(s) => write!(f, "{}", s),
}
}
@@ -228,6 +381,23 @@ impl From<Keyword> for Vec<u8> {
Keyword::Deleted => vec![DELETED as u8],
Keyword::Forwarded => vec![FORWARDED as u8],
Keyword::MdnSent => vec![MDN_SENT as u8],
Keyword::Autosent => vec![AUTOSENT as u8],
Keyword::CanUnsubscribe => vec![CANUNSUBSCRIBE as u8],
Keyword::Followed => vec![FOLLOWED as u8],
Keyword::HasAttachment => vec![HASATTACHMENT as u8],
Keyword::HasMemo => vec![HASMEMO as u8],
Keyword::HasNoAttachment => vec![HASNOATTACHMENT as u8],
Keyword::Imported => vec![IMPORTED as u8],
Keyword::IsTrusted => vec![ISTRUSTED as u8],
Keyword::MailFlagBit0 => vec![MAILFLAGBIT0 as u8],
Keyword::MailFlagBit1 => vec![MAILFLAGBIT1 as u8],
Keyword::MailFlagBit2 => vec![MAILFLAGBIT2 as u8],
Keyword::MaskedEmail => vec![MASKEDEMAIL as u8],
Keyword::Memo => vec![MEMO as u8],
Keyword::Muted => vec![MUTED as u8],
Keyword::New => vec![NEW as u8],
Keyword::Notify => vec![NOTIFY as u8],
Keyword::Unsubscribed => vec![UNSUBSCRIBED as u8],
Keyword::Other(string) => string.as_bytes().to_vec(),
}
}
@@ -256,6 +426,23 @@ impl ArchivedKeyword {
ArchivedKeyword::Deleted => Ok(DELETED as u32),
ArchivedKeyword::Forwarded => Ok(FORWARDED as u32),
ArchivedKeyword::MdnSent => Ok(MDN_SENT as u32),
ArchivedKeyword::Autosent => Ok(AUTOSENT as u32),
ArchivedKeyword::CanUnsubscribe => Ok(CANUNSUBSCRIBE as u32),
ArchivedKeyword::Followed => Ok(FOLLOWED as u32),
ArchivedKeyword::HasAttachment => Ok(HASATTACHMENT as u32),
ArchivedKeyword::HasMemo => Ok(HASMEMO as u32),
ArchivedKeyword::HasNoAttachment => Ok(HASNOATTACHMENT as u32),
ArchivedKeyword::Imported => Ok(IMPORTED as u32),
ArchivedKeyword::IsTrusted => Ok(ISTRUSTED as u32),
ArchivedKeyword::MailFlagBit0 => Ok(MAILFLAGBIT0 as u32),
ArchivedKeyword::MailFlagBit1 => Ok(MAILFLAGBIT1 as u32),
ArchivedKeyword::MailFlagBit2 => Ok(MAILFLAGBIT2 as u32),
ArchivedKeyword::MaskedEmail => Ok(MASKEDEMAIL as u32),
ArchivedKeyword::Memo => Ok(MEMO as u32),
ArchivedKeyword::Muted => Ok(MUTED as u32),
ArchivedKeyword::New => Ok(NEW as u32),
ArchivedKeyword::Notify => Ok(NOTIFY as u32),
ArchivedKeyword::Unsubscribed => Ok(UNSUBSCRIBED as u32),
ArchivedKeyword::Other(string) => Err(string.as_ref()),
}
}
@@ -274,6 +461,23 @@ impl ArchivedKeyword {
ArchivedKeyword::Deleted => Keyword::Deleted,
ArchivedKeyword::Forwarded => Keyword::Forwarded,
ArchivedKeyword::MdnSent => Keyword::MdnSent,
ArchivedKeyword::Autosent => Keyword::Autosent,
ArchivedKeyword::CanUnsubscribe => Keyword::CanUnsubscribe,
ArchivedKeyword::Followed => Keyword::Followed,
ArchivedKeyword::HasAttachment => Keyword::HasAttachment,
ArchivedKeyword::HasMemo => Keyword::HasMemo,
ArchivedKeyword::HasNoAttachment => Keyword::HasNoAttachment,
ArchivedKeyword::Imported => Keyword::Imported,
ArchivedKeyword::IsTrusted => Keyword::IsTrusted,
ArchivedKeyword::MailFlagBit0 => Keyword::MailFlagBit0,
ArchivedKeyword::MailFlagBit1 => Keyword::MailFlagBit1,
ArchivedKeyword::MailFlagBit2 => Keyword::MailFlagBit2,
ArchivedKeyword::MaskedEmail => Keyword::MaskedEmail,
ArchivedKeyword::Memo => Keyword::Memo,
ArchivedKeyword::Muted => Keyword::Muted,
ArchivedKeyword::New => Keyword::New,
ArchivedKeyword::Notify => Keyword::Notify,
ArchivedKeyword::Unsubscribed => Keyword::Unsubscribed,
ArchivedKeyword::Other(other) => Keyword::Other(other.as_ref().into()),
}
}
@@ -294,6 +498,23 @@ impl From<&ArchivedKeyword> for Keyword {
ArchivedKeyword::Deleted => Keyword::Deleted,
ArchivedKeyword::Forwarded => Keyword::Forwarded,
ArchivedKeyword::MdnSent => Keyword::MdnSent,
ArchivedKeyword::Autosent => Keyword::Autosent,
ArchivedKeyword::CanUnsubscribe => Keyword::CanUnsubscribe,
ArchivedKeyword::Followed => Keyword::Followed,
ArchivedKeyword::HasAttachment => Keyword::HasAttachment,
ArchivedKeyword::HasMemo => Keyword::HasMemo,
ArchivedKeyword::HasNoAttachment => Keyword::HasNoAttachment,
ArchivedKeyword::Imported => Keyword::Imported,
ArchivedKeyword::IsTrusted => Keyword::IsTrusted,
ArchivedKeyword::MailFlagBit0 => Keyword::MailFlagBit0,
ArchivedKeyword::MailFlagBit1 => Keyword::MailFlagBit1,
ArchivedKeyword::MailFlagBit2 => Keyword::MailFlagBit2,
ArchivedKeyword::MaskedEmail => Keyword::MaskedEmail,
ArchivedKeyword::Memo => Keyword::Memo,
ArchivedKeyword::Muted => Keyword::Muted,
ArchivedKeyword::New => Keyword::New,
ArchivedKeyword::Notify => Keyword::Notify,
ArchivedKeyword::Unsubscribed => Keyword::Unsubscribed,
ArchivedKeyword::Other(string) => Keyword::Other(string.as_ref().into()),
}
}
crates/types/src/special_use.rs
+15
View File
@@ -31,6 +31,9 @@ pub enum SpecialUse {
Shared,
Important,
None,
Memos,
Scheduled,
Snoozed,
}
impl SpecialUse {
@@ -44,6 +47,9 @@ impl SpecialUse {
b"sent" => SpecialUse::Sent,
b"shared" => SpecialUse::Shared,
b"important" => SpecialUse::Important,
b"memos" => SpecialUse::Memos,
b"scheduled" => SpecialUse::Scheduled,
b"snoozed" => SpecialUse::Snoozed,
)
}
@@ -57,6 +63,9 @@ impl SpecialUse {
SpecialUse::Sent => Some("sent"),
SpecialUse::Shared => Some("shared"),
SpecialUse::Important => Some("important"),
SpecialUse::Memos => Some("memos"),
SpecialUse::Scheduled => Some("scheduled"),
SpecialUse::Snoozed => Some("snoozed"),
SpecialUse::None => None,
}
}
@@ -73,6 +82,9 @@ impl ArchivedSpecialUse {
ArchivedSpecialUse::Sent => Some("sent"),
ArchivedSpecialUse::Shared => Some("shared"),
ArchivedSpecialUse::Important => Some("important"),
ArchivedSpecialUse::Memos => Some("memos"),
ArchivedSpecialUse::Scheduled => Some("scheduled"),
ArchivedSpecialUse::Snoozed => Some("snoozed"),
ArchivedSpecialUse::None => None,
}
}
@@ -89,6 +101,9 @@ impl From<&ArchivedSpecialUse> for SpecialUse {
ArchivedSpecialUse::Sent => SpecialUse::Sent,
ArchivedSpecialUse::Shared => SpecialUse::Shared,
ArchivedSpecialUse::Important => SpecialUse::Important,
ArchivedSpecialUse::Memos => SpecialUse::Memos,
ArchivedSpecialUse::Scheduled => SpecialUse::Scheduled,
ArchivedSpecialUse::Snoozed => SpecialUse::Snoozed,
ArchivedSpecialUse::None => SpecialUse::None,
}
}
crates/utils/Cargo.toml
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "utils"
version = "0.14.1"
version = "0.15.0"
edition = "2024"
[dependencies]
crates/utils/proc-macros/Cargo.toml
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "proc_macros"
version = "0.14.1"
version = "0.15.0"
edition = "2024"
[lib]
tests/src/jmap/mail/crypto.rs
+2
View File
@@ -43,10 +43,12 @@ pub async fn test(params: &mut JMAPTest) {
EncryptionMethod::PGP => EncryptionType::PGP {
algo,
certs: certs.clone(),
allow_spam_training: true,
},
EncryptionMethod::SMIME => EncryptionType::SMIME {
algo,
certs: certs.clone(),
allow_spam_training: true,
},
};
tests/src/smtp/lookup/utils.rs
+1 -1
View File
@@ -164,7 +164,7 @@ async fn strategies() {
env_id: None,
priority: 0,
size: 978,
quota_keys: vec![],
quota_keys: Default::default(),
};
assert_eq!(
tests/src/smtp/queue/dsn.rs
+1 -1
View File
@@ -86,7 +86,7 @@ async fn generate_dsn() {
env_id: None,
priority: 0,
blob_hash: BlobHash::generate(dsn_original.as_bytes()),
quota_keys: vec![],
quota_keys: Default::default(),
received_from_ip: IpAddr::V4(Ipv4Addr::LOCALHOST),
received_via_port: 0,
},
tests/src/smtp/queue/manager.rs
+1 -1
View File
@@ -175,7 +175,7 @@ pub fn new_message(queue_id: u64) -> MessageWrapper {
flags: 0,
env_id: None,
priority: 0,
quota_keys: vec![],
quota_keys: Default::default(),
blob_hash: Default::default(),
received_from_ip: IpAddr::V4(Ipv4Addr::LOCALHOST),
received_via_port: 0,