From f7c2fe10c1d34c8c0f2ce38db7aa34c5b162bb3b Mon Sep 17 00:00:00 2001 From: mdecimus <11444311+mdecimus@users.noreply.github.com> Date: Wed, 11 Feb 2026 21:30:17 +0000 Subject: [PATCH] Update all modules to use registry - part 3 --- Cargo.lock | 2 + crates/common/src/auth/access_token.rs | 21 +- crates/common/src/auth/authentication.rs | 6 +- crates/common/src/auth/mod.rs | 3 - crates/common/src/auth/oauth/config.rs | 2 +- crates/common/src/cache/directory.rs | 8 +- crates/common/src/cache/invalidate.rs | 5 +- crates/common/src/cache/reload.rs | 237 ++++++++-------- crates/common/src/config/inner.rs | 20 +- crates/common/src/config/mailstore/email.rs | 7 +- crates/common/src/config/mailstore/jmap.rs | 2 +- .../common/src/config/mailstore/spamfilter.rs | 14 +- crates/common/src/config/mod.rs | 186 +++---------- crates/common/src/config/network.rs | 14 +- crates/common/src/config/server/listener.rs | 31 +-- crates/common/src/config/server/mod.rs | 3 +- crates/common/src/config/smtp/mod.rs | 1 - crates/common/src/config/smtp/session.rs | 13 +- crates/common/src/config/storage.rs | 32 ++- crates/common/src/config/telemetry.rs | 45 ++- crates/common/src/enterprise/config.rs | 39 +-- crates/common/src/enterprise/mod.rs | 28 +- crates/common/src/expr/if_block.rs | 17 +- crates/common/src/ipc.rs | 11 +- crates/common/src/lib.rs | 4 +- .../manager/{webadmin.rs => application.rs} | 6 +- crates/common/src/manager/boot.rs | 256 +++--------------- crates/common/src/manager/mod.rs | 18 +- crates/common/src/network/acme/cache.rs | 1 - crates/common/src/network/listen.rs | 12 - crates/common/src/storage/blob.rs | 4 +- crates/common/src/storage/index.rs | 5 +- crates/common/src/storage/mod.rs | 14 +- crates/common/src/storage/quota.rs | 10 +- crates/common/src/storage/state.rs | 1 - crates/common/src/telemetry/metrics/store.rs | 6 +- crates/common/src/telemetry/tracers/store.rs | 2 +- crates/coordinator/src/bootstrap.rs | 1 + crates/directory/Cargo.toml | 1 + crates/directory/src/core/config.rs | 7 +- crates/directory/src/lib.rs | 8 +- crates/http-proto/src/response.rs | 2 +- crates/http/src/autoconfig/mod.rs | 2 +- crates/http/src/management/telemetry.rs | 65 ++--- crates/http/src/request.rs | 6 +- crates/jmap/src/api/request.rs | 7 +- crates/jmap/src/blob/upload.rs | 2 - crates/jmap/src/principal/query.rs | 138 +++++----- crates/main/src/main.rs | 6 +- crates/registry/Cargo.toml | 1 + crates/registry/src/pickle.rs | 57 +++- crates/registry/src/schema/mod.rs | 7 +- crates/registry/src/schema/prelude.rs | 4 +- crates/registry/src/types/index.rs | 168 ++++++++++++ crates/registry/src/types/ipmask.rs | 3 +- crates/registry/src/types/mod.rs | 11 +- crates/services/src/broadcast/mod.rs | 4 +- crates/services/src/housekeeper/mod.rs | 110 +++----- crates/services/src/lib.rs | 4 +- crates/services/src/task_manager/alarm.rs | 2 +- crates/services/src/task_manager/imip.rs | 5 +- crates/services/src/task_manager/index.rs | 8 +- crates/smtp/src/inbound/hooks/message.rs | 28 +- crates/smtp/src/inbound/milter/client.rs | 2 +- crates/smtp/src/inbound/milter/mod.rs | 14 +- crates/smtp/src/queue/spool.rs | 2 +- crates/spam-filter/src/modules/classifier.rs | 4 +- .../src/backend/composite/read_replica.rs | 4 - crates/store/src/backend/http/config.rs | 2 +- crates/store/src/backend/memory/mod.rs | 6 +- crates/store/src/bootstrap/data.rs | 31 --- crates/store/src/{bootstrap => build}/blob.rs | 17 ++ crates/store/src/build/data.rs | 94 +++++++ .../store/src/{bootstrap => build}/lookup.rs | 14 +- .../store/src/{bootstrap => build}/memory.rs | 17 ++ crates/store/src/{bootstrap => build}/mod.rs | 1 + crates/store/src/build/registry.rs | 29 ++ .../store/src/{bootstrap => build}/search.rs | 0 crates/store/src/dispatch/store.rs | 2 +- crates/store/src/lib.rs | 31 ++- crates/store/src/registry/bootstrap.rs | 8 + crates/store/src/registry/mod.rs | 3 +- crates/store/src/registry/query.rs | 15 +- tests/Cargo.toml | 4 +- tests/src/jmap/server/webhooks.rs | 2 +- tests/src/smtp/inbound/milter.rs | 2 +- 86 files changed, 1010 insertions(+), 1037 deletions(-) rename crates/common/src/manager/{webadmin.rs => application.rs} (98%) create mode 100644 crates/registry/src/types/index.rs delete mode 100644 crates/store/src/bootstrap/data.rs rename crates/store/src/{bootstrap => build}/blob.rs (83%) create mode 100644 crates/store/src/build/data.rs rename crates/store/src/{bootstrap => build}/lookup.rs (88%) rename crates/store/src/{bootstrap => build}/memory.rs (76%) rename crates/store/src/{bootstrap => build}/mod.rs (92%) create mode 100644 crates/store/src/build/registry.rs rename crates/store/src/{bootstrap => build}/search.rs (100%) diff --git a/Cargo.lock b/Cargo.lock index 192527fb..992420e0 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1768,6 +1768,7 @@ dependencies = [ "mail-parser", "md5 0.8.0", "nlp", + "nohash-hasher", "password-hash", "pbkdf2", "proc_macros", @@ -6029,6 +6030,7 @@ checksum = "7a2d987857b319362043e95f5353c0535c1f58eec5336fdfcf626430af7def58" name = "registry" version = "0.15.4" dependencies = [ + "ahash", "hashify", "serde", "serde_json", diff --git a/crates/common/src/auth/access_token.rs b/crates/common/src/auth/access_token.rs index cad0e6c5..87902b29 100644 --- a/crates/common/src/auth/access_token.rs +++ b/crates/common/src/auth/access_token.rs @@ -14,7 +14,6 @@ use crate::{ network::limiter::{ConcurrencyLimiter, LimiterResult}, }; use ahash::AHasher; -use chrono::format::Item; use registry::{ schema::{ enums::Permission, @@ -165,13 +164,13 @@ impl Server { let credential_scopes = account .credentials .into_iter() - .filter_map(|pass| { - let expires_at = pass + .filter_map(|(credential_id, credential)| { + let expires_at = credential .expires_at .map(|v| v.timestamp() as u64) .unwrap_or(u64::MAX); if expires_at > now { - let permissions = match pass.permissions { + let permissions = match credential.permissions { structs::Permissions::Inherit => permissions.clone().finalize(), structs::Permissions::Merge(merge) => { let mut permissions = permissions.clone(); @@ -183,7 +182,7 @@ impl Server { } }; Some(AccessScope { - credential_id: pass.credential_id as u32, + credential_id, permissions, expires_at, }) @@ -502,7 +501,6 @@ impl AccessToken { } pub fn assert_is_valid(self) -> trc::Result { - let todo = "use this function"; if self .inner .scopes @@ -514,7 +512,7 @@ impl AccessToken { Err(trc::SecurityEvent::Unauthorized .into_err() .ctx(trc::Key::AccountId, self.inner.account_id) - .reason("Access token expired.")) + .reason("Credential expired.")) } } @@ -741,11 +739,6 @@ impl AccessScope { expires_at: u64::MAX, } } - - pub fn expires_at(mut self, expires_at: u64) -> Self { - self.expires_at = expires_at; - self - } } fn hash_account(account: &Account) -> u64 { @@ -756,8 +749,8 @@ fn hash_account(account: &Account) -> u64 { account.member_tenant_id.hash(&mut s); account.role_ids.hash(&mut s); hash_permissions(&mut s, &account.permissions); - for credential in &account.credentials { - credential.credential_id.hash(&mut s); + for (credential_id, credential) in &account.credentials { + credential_id.hash(&mut s); credential.expires_at.hash(&mut s); hash_permissions(&mut s, &credential.permissions); } diff --git a/crates/common/src/auth/authentication.rs b/crates/common/src/auth/authentication.rs index 4c4ef7a0..4e287335 100644 --- a/crates/common/src/auth/authentication.rs +++ b/crates/common/src/auth/authentication.rs @@ -7,7 +7,7 @@ use crate::{ Server, auth::{ - AccessToken, AuthRequest, EmailCache, + AccessToken, AuthRequest, credential::{ApiKey, AppPassword}, oauth::GrantType, }, @@ -288,8 +288,8 @@ impl Server { .and_then(|account| account.into_user()) { // Find credential by credential_id - for credential in account.credentials.iter() { - if credential.credential_id as u32 == credential_id { + for (id, credential) in &account.credentials { + if *id == credential_id { if !verify_secret_hash(&credential.secret, secret).await? { return Err(trc::AuthEvent::Failed .into_err() diff --git a/crates/common/src/auth/mod.rs b/crates/common/src/auth/mod.rs index 35858e71..674c5d39 100644 --- a/crates/common/src/auth/mod.rs +++ b/crates/common/src/auth/mod.rs @@ -32,9 +32,6 @@ pub const FALLBACK_ADMIN_ID: u32 = u32::MAX; const PERMISSIONS_BITSET_SIZE: usize = Permission::COUNT.div_ceil(std::mem::size_of::()); pub type Permissions = Bitset; -//pub type IdMap = HashMap, nohash_hasher::BuildNoHashHasher>; -//pub type NameMap = AHashMap>; - #[derive(Debug, Clone, Copy)] pub enum EmailCache { Account(u32), diff --git a/crates/common/src/auth/oauth/config.rs b/crates/common/src/auth/oauth/config.rs index 35573d3b..a917a526 100644 --- a/crates/common/src/auth/oauth/config.rs +++ b/crates/common/src/auth/oauth/config.rs @@ -6,7 +6,7 @@ use crate::{ config::{build_ecdsa_pem, build_rsa_keypair}, - manager::webadmin::Resource, + manager::application::Resource, }; use biscuit::{ jwa::{Algorithm, SignatureAlgorithm}, diff --git a/crates/common/src/cache/directory.rs b/crates/common/src/cache/directory.rs index ced89656..76e85e5f 100644 --- a/crates/common/src/cache/directory.rs +++ b/crates/common/src/cache/directory.rs @@ -4,15 +4,11 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ -use registry::schema::{enums::Locale, prelude::Object}; -use store::write::now; +use registry::schema::enums::Locale; use crate::{ Server, - auth::{ - AccountCache, AccountInfo, AccountTenantIds, DomainCache, EmailCache, RoleCache, - TenantCache, - }, + auth::{AccountCache, AccountInfo, AccountTenantIds, DomainCache, RoleCache, TenantCache}, config::smtp::auth::DkimSigner, storage::ObjectQuota, }; diff --git a/crates/common/src/cache/invalidate.rs b/crates/common/src/cache/invalidate.rs index 5dc78510..65253312 100644 --- a/crates/common/src/cache/invalidate.rs +++ b/crates/common/src/cache/invalidate.rs @@ -26,12 +26,13 @@ impl Server { } CacheInvalidation::Domain(id) => { cache.domains.remove(id); + cache.dkim_signers.remove(id); } CacheInvalidation::Account(id) => { cache.accounts.remove(id); } - CacheInvalidation::Group(id) => { - cache.accounts.remove(id); + CacheInvalidation::DkimSignature(id) => { + cache.dkim_signers.remove(id); } CacheInvalidation::Tenant(id) => { cache.tenants.remove(id); diff --git a/crates/common/src/cache/reload.rs b/crates/common/src/cache/reload.rs index 9d3fc00b..90deb62e 100644 --- a/crates/common/src/cache/reload.rs +++ b/crates/common/src/cache/reload.rs @@ -6,133 +6,140 @@ use crate::{ Core, Server, - config::{server::Listeners, telemetry::Telemetry}, + config::{ + server::{Listeners, tls::parse_certificates}, + storage::Storage, + telemetry::Telemetry, + }, ipc::RegistryChange, }; use ahash::AHashMap; -use arc_swap::ArcSwap; -use store::registry::bootstrap::Bootstrap; +use directory::Directories; +use registry::schema::{prelude::Object, structs::BlockedIp}; +use std::sync::Arc; +use store::{InMemoryStore, LookupStores, registry::bootstrap::Bootstrap, write::now}; pub struct ReloadResult { - pub bp: Bootstrap, + pub bootstrap: Bootstrap, pub new_core: Option, pub tracers: Option, } impl Server { - async fn reload_blocked_ips(&self) -> trc::Result { - todo!() - /*let mut config = self - .core - .storage - .config - .build_config(BLOCKED_IP_KEY) - .await?; - *self.inner.data.blocked_ips.write() = BlockedIps::parse(&mut config).blocked_ip_addresses; - - Ok(config.into())*/ - } - - async fn reload_certificates(&self) -> trc::Result { - todo!() - /*let mut config = self.core.storage.config.build_config("certificate").await?; - let mut certificates = self.inner.data.tls_certificates.load().as_ref().clone(); - - parse_certificates(&mut config, &mut certificates, &mut Default::default()); - - self.inner.data.tls_certificates.store(certificates.into()); - - Ok(config.into())*/ - } - - async fn reload_lookups(&self) -> trc::Result { - todo!() - /*let mut config = self.core.storage.config.build_config("lookup").await?; - let mut stores = Stores::default(); - stores.parse_static_stores(&mut config, true); - - let mut core = self.core.as_ref().clone(); - for (id, store) in stores.in_memory_stores { - core.storage.lookups.insert(id, store); - } - - Ok(ReloadResult { - config, - new_core: core.into(), - tracers: None, - })*/ - } - pub async fn reload_registry(&self, change: RegistryChange) -> trc::Result { // TODO: check the different events triggering this, spam filter reload, etc. - todo!() - /*let mut config = self.core.storage.config.build_config("").await?; - - // Load stores - let mut stores = Stores { - stores: self.core.storage.stores.clone(), - blob_stores: self.core.storage.blobs.clone(), - search_stores: self.core.storage.ftss.clone(), - in_memory_stores: self.core.storage.lookups.clone(), - purge_schedules: Default::default(), - }; - stores.parse_stores(&mut config).await; - stores.parse_in_memory(&mut config, true).await; - - // Parse tracers - let tracers = Telemetry::parse(&mut config, &stores); - - if !config.errors.is_empty() { - return Ok(config.into()); - } - - // Build manager - let manager = ConfigManager { - cfg_local: ArcSwap::from_pointee( - self.core.storage.config.cfg_local.load().as_ref().clone(), - ), - cfg_local_path: self.core.storage.config.cfg_local_path.clone(), - cfg_local_patterns: Patterns::parse(&mut config).into(), - cfg_store: config - .value("storage.data") - .and_then(|id| stores.stores.get(id)) - .cloned() - .unwrap_or_default(), - }; - - // Parse settings and build shared core - let core = Box::pin(Core::parse(&mut config, stores, manager)).await; - if !config.errors.is_empty() { - return Ok(config.into()); - } - - // Update TLS certificates - let mut new_certificates = AHashMap::new(); - parse_certificates(&mut config, &mut new_certificates, &mut Default::default()); - let mut current_certificates = self.inner.data.tls_certificates.load().as_ref().clone(); - for (cert_id, cert) in new_certificates { - current_certificates.insert(cert_id, cert); - } - self.inner - .data - .tls_certificates - .store(current_certificates.into()); - - // Update blocked IPs - *self.inner.data.blocked_ips.write() = BlockedIps::parse(&mut config).blocked_ip_addresses; - - // Parser servers - let mut servers = Listeners::parse(&mut config); - servers.parse_tcp_acceptors(&mut config, self.inner.clone()); - - Ok(if config.errors.is_empty() { - ReloadResult { - config, - new_core: core.into(), - tracers: tracers.into(), + let mut bootstrap = Bootstrap::new(self.registry().clone()); + let object = match change { + RegistryChange::Insert(id) => { + if matches!(id.object(), Object::BlockedIp) { + if let Some(ip) = bootstrap.get_infallible::(id).await + && ip.expires_at.is_none_or(|ip| ip.timestamp() > now() as i64) + { + let mut ips = self.inner.data.blocked_ips.write(); + if let Some(ip) = ip.address.try_to_ip() { + ips.blocked_ip_addresses.insert(ip); + } else { + ips.blocked_ip_networks.push(ip.address); + } + } + return Ok(ReloadResult { + bootstrap, + new_core: None, + tracers: None, + }); + } else { + id.object() + } } - } else { - config.into() - })*/ + RegistryChange::Delete(id) => id.object(), + RegistryChange::Reload(object) => object, + }; + + let mut result = ReloadResult { + bootstrap, + new_core: None, + tracers: None, + }; + + match object { + Object::Certificate => { + let mut certificates = AHashMap::new(); + parse_certificates( + &mut result.bootstrap, + &mut certificates, + &mut Default::default(), + ) + .await; + self.inner + .data + .tls_certificates + .store(Arc::new(certificates)); + } + Object::MemoryLookupKey | Object::MemoryLookupKeyValue => { + let mut lookup = LookupStores { + stores: self.inner.data.lookup_stores.load().as_ref().clone(), + }; + lookup + .stores + .retain(|_, store| !matches!(store, InMemoryStore::Static(_))); + lookup.parse_static(&mut result.bootstrap).await; + } + Object::HttpLookup => { + let mut lookup = LookupStores { + stores: self.inner.data.lookup_stores.load().as_ref().clone(), + }; + lookup + .stores + .retain(|_, store| !matches!(store, InMemoryStore::Http(_))); + lookup.parse_http(&mut result.bootstrap).await; + } + Object::LookupStore => { + let mut lookup = LookupStores { + stores: self.inner.data.lookup_stores.load().as_ref().clone(), + }; + lookup.stores.retain(|_, store| { + matches!(store, InMemoryStore::Static(_) | InMemoryStore::Http(_)) + }); + lookup.parse_stores(&mut result.bootstrap).await; + } + _ => { + // Load stores + let directory = Directories::build(&mut result.bootstrap).await; + let storage = &self.core.storage; + let storage = Storage { + registry: storage.registry.clone(), + data: storage.data.clone(), + blob: storage.blob.clone(), + search: storage.search.clone(), + metrics: storage.metrics.clone(), + tracing: storage.tracing.clone(), + memory: storage.memory.clone(), + coordinator: storage.coordinator.clone(), + directory: directory.default_directory, + directories: directory.directories, + }; + + // Parse tracers + let tracers = Telemetry::parse(&mut result.bootstrap, &storage).await; + + if result.bootstrap.errors.is_empty() { + let core = Box::pin(Core::parse(&mut result.bootstrap, storage)).await; + + if result.bootstrap.errors.is_empty() { + let mut servers = Listeners::parse(&mut result.bootstrap).await; + servers + .parse_tcp_acceptors(&mut result.bootstrap, self.inner.clone()) + .await; + + if result.bootstrap.errors.is_empty() { + result.new_core = Some(core); + result.tracers = Some(tracers); + } + } + } + } + } + + Ok(result) } } diff --git a/crates/common/src/config/inner.rs b/crates/common/src/config/inner.rs index e5384376..8fe3e350 100644 --- a/crates/common/src/config/inner.rs +++ b/crates/common/src/config/inner.rs @@ -29,7 +29,7 @@ use std::{ net::{IpAddr, Ipv4Addr, Ipv6Addr}, sync::Arc, }; -use store::registry::bootstrap::Bootstrap; +use store::{LookupStores, registry::bootstrap::Bootstrap}; use utils::{ cache::{Cache, CacheWithTtl}, snowflake::SnowflakeIdGenerator, @@ -40,7 +40,7 @@ impl Data { // Parse certificates let mut certificates = AHashMap::new(); let mut subject_names = AHashSet::new(); - parse_certificates(bp, &mut certificates, &mut subject_names); + parse_certificates(bp, &mut certificates, &mut subject_names).await; if subject_names.is_empty() { subject_names.insert("localhost".into()); } @@ -52,7 +52,10 @@ impl Data { panic!("Invalid system time, panicking to avoid data corruption"); } - let todo = "TODO: WebAdminManager initialization"; + let todo = "TODO: WebApplicationManager initialization"; + + let blocked_ips = BlockedIps::parse(bp).await; + let lookup_stores = LookupStores::build(bp).await; Data { spam_classifier: ArcSwap::from_pointee(SpamClassifier::default()), @@ -72,19 +75,16 @@ impl Data { }) .ok() .map(Arc::new), - blocked_ips: RwLock::new(BlockedIps::parse(bp).await), + lookup_stores: ArcSwap::from_pointee(lookup_stores.stores), + blocked_ips: RwLock::new(blocked_ips), jmap_id_gen: id_generator.clone(), queue_id_gen: id_generator.clone(), span_id_gen: id_generator, queue_status: true.into(), - webadmin: Default::default(), /*config - .value("webadmin.path") - .map(|path| WebAdminManager::new(path.into())) - .unwrap_or_default(),*/ + applications: Default::default(), logos: Default::default(), smtp_connectors: TlsConnectors::default(), asn_geo_data: Default::default(), - lookup_stores: Default::default(), } } } @@ -220,7 +220,7 @@ impl Default for Data { queue_id_gen: Default::default(), span_id_gen: Default::default(), queue_status: true.into(), - webadmin: Default::default(), + applications: Default::default(), logos: Default::default(), smtp_connectors: Default::default(), asn_geo_data: Default::default(), diff --git a/crates/common/src/config/mailstore/email.rs b/crates/common/src/config/mailstore/email.rs index ac1edf37..08ab2ff7 100644 --- a/crates/common/src/config/mailstore/email.rs +++ b/crates/common/src/config/mailstore/email.rs @@ -8,7 +8,10 @@ use ahash::{AHashMap, AHashSet}; use nlp::language::Language; use registry::{ schema::{ - enums::{SearchCalendarField, SearchContactField, SearchEmailField, StorageQuota}, + enums::{ + CompressionAlgo, SearchCalendarField, SearchContactField, SearchEmailField, + StorageQuota, + }, structs::{ AddressBook, Calendar, DataRetention, Email, Jmap, OidcProvider, Search, SieveUserInterpreter, @@ -54,6 +57,7 @@ pub struct EmailConfig { pub index_fields: AHashMap>, pub max_objects: ObjectQuota, + pub compression: CompressionAlgo, pub account_purge_frequency: SimpleCron, pub data_purge_frequency: SimpleCron, @@ -250,6 +254,7 @@ impl EmailConfig { account_purge_frequency: dr.expunge_schedule.into(), data_purge_frequency: dr.data_cleanup_schedule.into(), blob_purge_frequency: dr.blob_cleanup_schedule.into(), + compression: email.compression_algorithm, } } } diff --git a/crates/common/src/config/mailstore/jmap.rs b/crates/common/src/config/mailstore/jmap.rs index 060ca147..2592df54 100644 --- a/crates/common/src/config/mailstore/jmap.rs +++ b/crates/common/src/config/mailstore/jmap.rs @@ -83,7 +83,7 @@ impl JmapConfig { }; // Add capabilities - jmap.add_capabilities(bp); + jmap.add_capabilities(bp).await; jmap } } diff --git a/crates/common/src/config/mailstore/spamfilter.rs b/crates/common/src/config/mailstore/spamfilter.rs index d025b00e..52d78129 100644 --- a/crates/common/src/config/mailstore/spamfilter.rs +++ b/crates/common/src/config/mailstore/spamfilter.rs @@ -316,49 +316,49 @@ impl DnsBlServer { zone: bp.compile_expr(obj.id, &server.ctx_zone()), tags: bp.compile_expr(obj.id, &server.ctx_tag()), scope: Element::Any, - id: server.name, + id: server.description, } .into(), SpamDnsblServer::Url(server) if server.enable => DnsBlServer { zone: bp.compile_expr(obj.id, &server.ctx_zone()), tags: bp.compile_expr(obj.id, &server.ctx_tag()), scope: Element::Url, - id: server.name, + id: server.description, } .into(), SpamDnsblServer::Domain(server) if server.enable => DnsBlServer { zone: bp.compile_expr(obj.id, &server.ctx_zone()), tags: bp.compile_expr(obj.id, &server.ctx_tag()), scope: Element::Domain, - id: server.name, + id: server.description, } .into(), SpamDnsblServer::Email(server) if server.enable => DnsBlServer { zone: bp.compile_expr(obj.id, &server.ctx_zone()), tags: bp.compile_expr(obj.id, &server.ctx_tag()), scope: Element::Email, - id: server.name, + id: server.description, } .into(), SpamDnsblServer::Ip(server) if server.enable => DnsBlServer { zone: bp.compile_expr(obj.id, &server.ctx_zone()), tags: bp.compile_expr(obj.id, &server.ctx_tag()), scope: Element::Ip, - id: server.name, + id: server.description, } .into(), SpamDnsblServer::Header(server) if server.enable => DnsBlServer { zone: bp.compile_expr(obj.id, &server.ctx_zone()), tags: bp.compile_expr(obj.id, &server.ctx_tag()), scope: Element::Header, - id: server.name, + id: server.description, } .into(), SpamDnsblServer::Body(server) if server.enable => DnsBlServer { zone: bp.compile_expr(obj.id, &server.ctx_zone()), tags: bp.compile_expr(obj.id, &server.ctx_tag()), scope: Element::Body, - id: server.name, + id: server.description, } .into(), _ => None, diff --git a/crates/common/src/config/mod.rs b/crates/common/src/config/mod.rs index efd05edb..75f5b8b7 100644 --- a/crates/common/src/config/mod.rs +++ b/crates/common/src/config/mod.rs @@ -8,17 +8,15 @@ use self::{mailstore::jmap::JmapConfig, smtp::SmtpConfig, storage::Storage}; use crate::{ Core, Network, auth::oauth::config::OAuthConfig, - config::mailstore::{imap::ImapConfig, scripts::Scripting, spamfilter::SpamFilterConfig}, - expr::*, + config::mailstore::{ + email::EmailConfig, imap::ImapConfig, scripts::Scripting, spamfilter::SpamFilterConfig, + }, }; use arc_swap::ArcSwap; -use coordinator::Coordinator; -use directory::{Directories, Directory}; use groupware::GroupwareConfig; use hyper::HeaderMap; use ring::signature::{EcdsaKeyPair, RsaKeyPair}; -use std::sync::Arc; -use store::{BlobStore, InMemoryStore, SearchStore, Store, registry::bootstrap::Bootstrap}; +use store::registry::bootstrap::Bootstrap; use telemetry::Metrics; pub mod groupware; @@ -31,139 +29,45 @@ pub mod storage; pub mod telemetry; impl Core { - pub async fn parse(bp: &mut Bootstrap) -> Self { - todo!() - /*let mut data = config - .value_require("storage.data") - .map(|id| id.to_string()) - .and_then(|id| { - if let Some(store) = stores.stores.get(&id) { - store.clone().into() - } else { - config.new_parse_error("storage.data", format!("Data store {id:?} not found")); - None - } - }) - .unwrap_or_default(); - - #[cfg(not(feature = "enterprise"))] - let is_enterprise = false; - + pub async fn parse(bp: &mut Bootstrap, mut storage: Storage) -> Self { // SPDX-SnippetBegin // SPDX-FileCopyrightText: 2020 Stalwart Labs LLC // SPDX-License-Identifier: LicenseRef-SEL #[cfg(feature = "enterprise")] - let enterprise = - crate::enterprise::Enterprise::parse(config, &config_manager, &stores, &data).await; + let enterprise = { + let enterprise = crate::enterprise::Enterprise::parse(bp).await; + if enterprise.is_none() { + use registry::schema::prelude::Object; + use store::Store; - #[cfg(feature = "enterprise")] - let is_enterprise = enterprise.is_some(); - - #[cfg(feature = "enterprise")] - if !is_enterprise { - if data.is_enterprise_store() { - config - .new_build_error("storage.data", "SQL read replicas is an Enterprise feature"); - data = Store::None; + if storage.data.is_enterprise() { + bp.build_error( + Object::DataStore.singleton(), + "Disabling enterprise-only data store.", + ); + storage.data = storage.data.downgrade_store(); + } + if storage.blob.is_enterprise() { + bp.build_error( + Object::BlobStore.singleton(), + "Disabling enterprise-only blob store.", + ); + storage.blob = storage.blob.downgrade_store(); + } + if storage.memory.is_enterprise() { + bp.build_error( + Object::InMemoryStore.singleton(), + "Disabling enterprise-only in-memory store.", + ); + storage.memory = storage.memory.downgrade_store(); + } + storage.metrics = Store::None; + storage.metrics = Store::None; } - stores.disable_enterprise_only(); - } + enterprise + }; // SPDX-SnippetEnd - let mut blob = config - .value_require("storage.blob") - .map(|id| id.to_string()) - .and_then(|id| { - if let Some(store) = stores.blob_stores.get(&id) { - store.clone().into() - } else { - config.new_parse_error("storage.blob", format!("Blob store {id:?} not found")); - None - } - }) - .unwrap_or_default(); - let mut lookup = config - .value_require("storage.lookup") - .map(|id| id.to_string()) - .and_then(|id| { - if let Some(store) = stores.in_memory_stores.get(&id) { - store.clone().into() - } else { - config.new_parse_error( - "storage.lookup", - format!("In-memory store {id:?} not found"), - ); - None - } - }) - .unwrap_or_default(); - let mut fts = config - .value_require("storage.fts") - .map(|id| id.to_string()) - .and_then(|id| { - if let Some(store) = stores.search_stores.get(&id) { - store.clone().into() - } else { - config.new_parse_error( - "storage.fts", - format!("Full-text store {id:?} not found"), - ); - None - } - }) - .unwrap_or_default(); - let pubsub = Coordinator::None; /*config - .value("cluster.coordinator") - .map(|id| id.to_string()) - .and_then(|id| { - if let Some(store) = stores.pubsub_stores.get(&id) { - store.clone().into() - } else { - config.new_parse_error( - "cluster.coordinator", - format!("Coordinator backend {id:?} not found"), - ); - None - } - }) - .unwrap_or_default();*/ - let mut directories = - Directories::parse(config, &stores, data.clone(), is_enterprise).await; - let directory = config - .value_require("storage.directory") - .map(|id| id.to_string()) - .and_then(|id| { - if let Some(directory) = directories.directories.get(&id) { - directory.clone().into() - } else { - config.new_parse_error( - "storage.directory", - format!("Directory {id:?} not found"), - ); - None - } - }) - .unwrap_or_else(|| Arc::new(Directory::default())); - directories - .directories - .insert("*".to_string(), directory.clone()); - - // If any of the stores are missing, disable all stores to avoid data loss - if matches!(data, Store::None) - || matches!(&blob.backend, BlobBackend::Store(Store::None)) - || matches!(lookup, InMemoryStore::Store(Store::None)) - || matches!(fts, SearchStore::Store(Store::None)) - { - data = Store::default(); - blob = BlobStore::default(); - lookup = InMemoryStore::default(); - fts = SearchStore::default(); - config.new_build_error( - "storage.*", - "One or more stores are missing, disabling all stores", - ) - } - Self { // SPDX-SnippetBegin // SPDX-FileCopyrightText: 2020 Stalwart Labs LLC @@ -177,24 +81,12 @@ impl Core { jmap: JmapConfig::parse(bp).await, imap: ImapConfig::parse(bp).await, oauth: OAuthConfig::parse(bp).await, - metrics: Metrics::parse(bp.await), + metrics: Metrics::parse(bp).await, spam: SpamFilterConfig::parse(bp).await, + email: EmailConfig::parse(bp).await, groupware: GroupwareConfig::parse(bp).await, - storage: Storage { - data, - blob, - fts, - lookup, - pubsub, - directory, - directories: directories.directories, - purge_schedules: stores.purge_schedules, - stores: stores.stores, - lookups: stores.in_memory_stores, - blobs: stores.blob_stores, - ftss: stores.search_stores, - }, - }*/ + storage, + } } pub fn into_shared(self) -> ArcSwap { diff --git a/crates/common/src/config/network.rs b/crates/common/src/config/network.rs index 864d2a92..92713575 100644 --- a/crates/common/src/config/network.rs +++ b/crates/common/src/config/network.rs @@ -468,23 +468,17 @@ impl ClusterRole { } fn is_seen_role(&self) -> bool { - match self { - ClusterRole::Sharded { + matches!(self, ClusterRole::Sharded { shard_id, total_shards, - } if *shard_id == u32::MAX && *total_shards == 0 => true, - _ => false, - } + } if *shard_id == u32::MAX && *total_shards == 0) } fn is_uninit(&self) -> bool { - match self { - ClusterRole::Sharded { + matches!(self, ClusterRole::Sharded { shard_id, total_shards, - } if *shard_id == u32::MAX && *total_shards == u32::MAX => true, - _ => false, - } + } if *shard_id == u32::MAX && *total_shards == u32::MAX) } fn finalize(&mut self) { diff --git a/crates/common/src/config/server/listener.rs b/crates/common/src/config/server/listener.rs index f9ff89e6..9c11efa9 100644 --- a/crates/common/src/config/server/listener.rs +++ b/crates/common/src/config/server/listener.rs @@ -92,25 +92,25 @@ impl Listeners { return; } - if let Some(send_size) = listener.socket_send_buffer_size { - if let Err(err) = socket.set_send_buffer_size(send_size as u32) { - bp.build_error(id, format!("Failed to set SO_SNDBUF: {err}")); - return; - } + if let Some(send_size) = listener.socket_send_buffer_size + && let Err(err) = socket.set_send_buffer_size(send_size as u32) + { + bp.build_error(id, format!("Failed to set SO_SNDBUF: {err}")); + return; } - if let Some(recv_size) = listener.socket_receive_buffer_size { - if let Err(err) = socket.set_recv_buffer_size(recv_size as u32) { - bp.build_error(id, format!("Failed to set SO_RCVBUF: {err}")); - return; - } + if let Some(recv_size) = listener.socket_receive_buffer_size + && let Err(err) = socket.set_recv_buffer_size(recv_size as u32) + { + bp.build_error(id, format!("Failed to set SO_RCVBUF: {err}")); + return; } - if let Some(tos) = listener.socket_tos_v4 { - if let Err(err) = socket.set_tos_v4(tos as u32) { - bp.build_error(id, format!("Failed to set IP_TOS: {err}")); - return; - } + if let Some(tos) = listener.socket_tos_v4 + && let Err(err) = socket.set_tos_v4(tos as u32) + { + bp.build_error(id, format!("Failed to set IP_TOS: {err}")); + return; } listeners.push(TcpListener { @@ -118,7 +118,6 @@ impl Listeners { addr, ttl: listener.socket_ttl.map(|v| v as u32), backlog: listener.socket_backlog.map(|v| v as u32), - linger: listener.socket_linger.map(|d| d.into_inner()), nodelay: listener.socket_no_delay, }); } diff --git a/crates/common/src/config/server/mod.rs b/crates/common/src/config/server/mod.rs index 90a994d4..ee6ef496 100644 --- a/crates/common/src/config/server/mod.rs +++ b/crates/common/src/config/server/mod.rs @@ -11,7 +11,7 @@ use registry::{ types::{id::Id, ipmask::IpAddrOrMask}, }; use serde::{Deserialize, Serialize}; -use std::{fmt::Display, net::SocketAddr, sync::Arc, time::Duration}; +use std::{fmt::Display, net::SocketAddr, sync::Arc}; use store::registry::RegistryObject; use tokio::net::TcpSocket; use utils::snowflake::SnowflakeIdGenerator; @@ -46,7 +46,6 @@ pub struct TcpListener { // TCP options pub ttl: Option, - pub linger: Option, pub nodelay: bool, } diff --git a/crates/common/src/config/smtp/mod.rs b/crates/common/src/config/smtp/mod.rs index c75b720d..6d1fd512 100644 --- a/crates/common/src/config/smtp/mod.rs +++ b/crates/common/src/config/smtp/mod.rs @@ -14,7 +14,6 @@ use self::{ auth::MailAuthConfig, queue::QueueConfig, report::ReportConfig, resolver::Resolvers, session::SessionConfig, }; -use super::*; use crate::expr::Expression; use registry::{schema::structs::Rate, types::id::Id}; use store::registry::bootstrap::Bootstrap; diff --git a/crates/common/src/config/smtp/session.rs b/crates/common/src/config/smtp/session.rs index d73dc54f..301dee18 100644 --- a/crates/common/src/config/smtp/session.rs +++ b/crates/common/src/config/smtp/session.rs @@ -6,7 +6,10 @@ use self::resolver::Policy; use super::*; -use crate::expr::if_block::{BootstrapExprExt, IfBlock}; +use crate::expr::{ + Variable, + if_block::{BootstrapExprExt, IfBlock}, +}; use ahash::AHashSet; use hyper::HeaderMap; use registry::schema::{ @@ -124,7 +127,7 @@ pub struct Data { #[derive(Clone)] pub struct Milter { pub enable: IfBlock, - pub id: Arc, + pub id: Id, pub addrs: Vec, pub hostname: String, pub port: u16, @@ -150,7 +153,7 @@ pub enum MilterVersion { #[derive(Clone)] pub struct MTAHook { pub enable: IfBlock, - pub id: String, + pub id: Id, pub url: String, pub timeout: Duration, pub headers: HeaderMap, @@ -317,7 +320,7 @@ impl SessionConfig { Some(Milter { enable: bp.compile_expr(id, &milter.ctx_enable()), - id: Arc::new(milter.name.into()), + id, addrs: format!("{}:{}", milter.hostname, milter.port) .to_socket_addrs() .map_err(|err| { @@ -360,7 +363,7 @@ impl SessionConfig { Some(MTAHook { enable: bp.compile_expr(id, &hook.ctx_enable()), - id: hook.name, + id, url: hook.url, timeout: hook.timeout.into_inner(), headers: hook diff --git a/crates/common/src/config/storage.rs b/crates/common/src/config/storage.rs index 9354060a..8fa99a9c 100644 --- a/crates/common/src/config/storage.rs +++ b/crates/common/src/config/storage.rs @@ -5,10 +5,11 @@ */ use coordinator::Coordinator; -use directory::Directory; -use registry::schema::enums::CompressionAlgo; +use directory::{Directories, Directory}; use std::{collections::HashMap, sync::Arc}; -use store::{BlobStore, InMemoryStore, RegistryStore, SearchStore, Store}; +use store::{ + BlobStore, InMemoryStore, RegistryStore, SearchStore, Store, registry::bootstrap::Bootstrap, +}; pub type IdMap = HashMap, nohash_hasher::BuildNoHashHasher>; @@ -17,10 +18,31 @@ pub struct Storage { pub registry: RegistryStore, pub data: Store, pub blob: BlobStore, - pub fts: SearchStore, + pub search: SearchStore, pub memory: InMemoryStore, + pub metrics: Store, + pub tracing: Store, pub coordinator: Coordinator, pub directory: Option>, pub directories: IdMap, - pub compression: CompressionAlgo, +} + +impl Storage { + pub async fn parse(bp: &mut Bootstrap) -> Self { + let memory = InMemoryStore::build(bp).await.unwrap_or_default(); + let directory = Directories::build(bp).await; + + Storage { + registry: bp.registry.clone(), + data: bp.data_store.clone(), + blob: BlobStore::build(bp).await.unwrap_or_default(), + search: SearchStore::build(bp).await.unwrap_or_default(), + coordinator: Coordinator::build(bp, &memory).await.unwrap_or_default(), + memory, + tracing: Store::build_tracing(bp).await.unwrap_or_default(), + metrics: Store::build_metrics(bp).await.unwrap_or_default(), + directory: directory.default_directory, + directories: directory.directories, + } + } } diff --git a/crates/common/src/config/telemetry.rs b/crates/common/src/config/telemetry.rs index 4d6022ad..872fd18b 100644 --- a/crates/common/src/config/telemetry.rs +++ b/crates/common/src/config/telemetry.rs @@ -4,6 +4,7 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ +use crate::config::storage::Storage; use ahash::{AHashMap, AHashSet}; use base64::{Engine, engine::general_purpose::STANDARD}; use hyper::HeaderMap; @@ -139,9 +140,9 @@ pub struct PrometheusMetrics { } impl Telemetry { - pub async fn parse(bp: &mut Bootstrap) -> Self { + pub async fn parse(bp: &mut Bootstrap, storage: &Storage) -> Self { let mut telemetry = Telemetry { - tracers: Tracers::parse(bp).await, + tracers: Tracers::parse(bp, storage).await, metrics: Interests::default(), }; @@ -159,7 +160,7 @@ impl Telemetry { } impl Tracers { - pub async fn parse(bp: &mut Bootstrap) -> Self { + pub async fn parse(bp: &mut Bootstrap, storage: &Storage) -> Self { // Parse custom logging levels let mut custom_levels = AHashMap::new(); for level in bp.list_infallible::().await { @@ -428,32 +429,22 @@ impl Tracers { // Parse tracing history #[cfg(feature = "enterprise")] - { - use registry::schema::structs::TelemetryHistory; + if storage.tracing.is_active() { + let mut tracer = TelemetrySubscriber { + id: "history".to_string(), + interests: Default::default(), + lossy: false, + typ: TelemetrySubscriberType::StoreTracer(StoreTracer { + store: storage.tracing.clone(), + }), + }; - let todo = "update store"; - - if bp - .setting_infallible::() - .await - .enable_tracing_history - { - let mut tracer = TelemetrySubscriber { - id: "history".to_string(), - interests: Default::default(), - lossy: false, - typ: TelemetrySubscriberType::StoreTracer(StoreTracer { - store: store::Store::None, - }), - }; - - for event_type in StoreTracer::default_events() { - tracer.interests.set(event_type); - global_interests.set(event_type); - } - - tracers.push(tracer); + for event_type in StoreTracer::default_events() { + tracer.interests.set(event_type); + global_interests.set(event_type); } + + tracers.push(tracer); } // SPDX-SnippetEnd diff --git a/crates/common/src/enterprise/config.rs b/crates/common/src/enterprise/config.rs index b5c49859..b0218684 100644 --- a/crates/common/src/enterprise/config.rs +++ b/crates/common/src/enterprise/config.rs @@ -9,8 +9,8 @@ */ use super::{ - AlertContent, AlertContentToken, AlertMethod, Enterprise, MetricAlert, MetricStore, - SpamFilterLlmConfig, TraceStore, Undelete, license::LicenseKey, llm::AiApiConfig, + AlertContent, AlertContentToken, AlertMethod, Enterprise, MetricAlert, SpamFilterLlmConfig, + license::LicenseKey, llm::AiApiConfig, }; use crate::{enterprise::llm::ApiType, expr::if_block::BootstrapExprExt}; use ahash::AHashMap; @@ -20,13 +20,12 @@ use registry::{ prelude::{Object, Property}, structs::{ self, AiModel, Alert, CalendarAlarm, CalendarScheduling, DataRetention, SpamLlm, - TelemetryHistory, }, }, types::id::Id, }; use std::sync::Arc; -use store::{Store, registry::bootstrap::Bootstrap}; +use store::registry::bootstrap::Bootstrap; use trc::MetricType; use utils::template::Template; @@ -118,34 +117,13 @@ impl Enterprise { _ => (), } - let telemetry = bp.setting_infallible::().await; let dr = bp.setting_infallible::().await; - let todo = "map stores"; - let trace_store = if telemetry.enable_tracing_history { - TraceStore { - retention: dr.hold_traces_for.map(|d| d.into_inner()), - store: Store::None, - } - .into() - } else { - None - }; - let metrics_store = if telemetry.enable_metric_history { - MetricStore { - retention: dr.hold_metrics_for.map(|d| d.into_inner()), - store: Store::None, - interval: telemetry.metrics_collection_interval.into(), - } - .into() - } else { - None - }; // Parse AI APIs let mut ai_apis = AHashMap::new(); let mut ai_apis_ids = AHashMap::new(); for api in bp.list_infallible::().await { - let id = api.id.clone(); + let id = api.id; let api = api.object; let api = Arc::new(AiApiConfig { id: api.name, @@ -173,18 +151,17 @@ impl Enterprise { // Build the enterprise configuration let mut enterprise = Enterprise { license, - undelete: dr.hold_deleted_for.map(|retention| Undelete { - retention: retention.into_inner(), - }), + undelete_retention: dr.hold_deleted_for.map(|retention| retention.into_inner()), logo_url: enterprise.logo_url, - trace_store, - metrics_store, metrics_alerts: Default::default(), spam_filter_llm: SpamFilterLlmConfig::parse(bp, &ai_apis_ids).await, ai_apis, template_calendar_alarm: None, template_scheduling_email: None, template_scheduling_web: None, + trace_retention: dr.hold_traces_for.map(|d| d.into_inner()), + metrics_retention: dr.hold_metrics_for.map(|d| d.into_inner()), + metrics_interval: dr.metrics_collection_interval.into(), }; // Parse metric alerts diff --git a/crates/common/src/enterprise/mod.rs b/crates/common/src/enterprise/mod.rs index e27105c0..6bcc1399 100644 --- a/crates/common/src/enterprise/mod.rs +++ b/crates/common/src/enterprise/mod.rs @@ -16,7 +16,7 @@ pub mod undelete; use crate::{ Core, Server, config::groupware::CalendarTemplateVariable, expr::Expression, - manager::webadmin::Resource, + manager::application::Resource, }; use ahash::{AHashMap, AHashSet}; use license::LicenseKey; @@ -27,7 +27,6 @@ use registry::{ types::id::Id, }; use std::{sync::Arc, time::Duration}; -use store::Store; use trc::{AddContext, MetricType}; use utils::{HttpLimitResponse, cron::SimpleCron, template::Template}; @@ -35,9 +34,10 @@ use utils::{HttpLimitResponse, cron::SimpleCron, template::Template}; pub struct Enterprise { pub license: LicenseKey, pub logo_url: Option, - pub undelete: Option, - pub trace_store: Option, - pub metrics_store: Option, + pub undelete_retention: Option, + pub trace_retention: Option, + pub metrics_retention: Option, + pub metrics_interval: SimpleCron, pub metrics_alerts: Vec, pub ai_apis: AHashMap>, pub spam_filter_llm: Option, @@ -59,24 +59,6 @@ pub struct SpamFilterLlmConfig { pub confidence: AHashSet, } -#[derive(Clone)] -pub struct Undelete { - pub retention: Duration, -} - -#[derive(Clone)] -pub struct TraceStore { - pub retention: Option, - pub store: Store, -} - -#[derive(Clone)] -pub struct MetricStore { - pub retention: Option, - pub store: Store, - pub interval: SimpleCron, -} - #[derive(Clone, Debug)] pub struct MetricAlert { pub id: Id, diff --git a/crates/common/src/expr/if_block.rs b/crates/common/src/expr/if_block.rs index bb0a3bbb..a88c7dd3 100644 --- a/crates/common/src/expr/if_block.rs +++ b/crates/common/src/expr/if_block.rs @@ -98,7 +98,7 @@ impl BootstrapExprExt for Bootstrap { return IfBlock::empty(id, expr_ctx.property); } - if let Some(if_block) = self.try_compile_expr(id, expr_ctx, &expr_ctx.expr) { + if let Some(if_block) = self.try_compile_expr(id, expr_ctx, expr_ctx.expr) { if_block } else { self.compile_default_expr(id, expr_ctx) @@ -122,7 +122,6 @@ impl BootstrapExprExt for Bootstrap { ) -> Option { // Parse conditions let mut if_then = Vec::with_capacity(expr.match_.len()); - let default; if expr.else_.is_empty() { if !expr.match_.is_empty() { @@ -152,19 +151,17 @@ impl BootstrapExprExt for Bootstrap { .with_variables(expr_ctx.allowed_variables) .with_constants(expr_ctx.allowed_constants); - match ExpressionParser::new(Tokenizer::new(&expr.else_, &token_map)).parse() { - Ok(expr) => { - default = expr; - } + let default = match ExpressionParser::new(Tokenizer::new(&expr.else_, &token_map)).parse() { + Ok(expr) => expr, Err(err) => { self.invalid_property( id, expr_ctx.property, - &format!("Error parsing 'else' expression: {}", err), + format!("Error parsing 'else' expression: {}", err), ); return None; } - } + }; for (num, match_) in expr.match_.iter().enumerate() { match ExpressionParser::new(Tokenizer::new(&match_.if_, &token_map)).parse() { @@ -180,7 +177,7 @@ impl BootstrapExprExt for Bootstrap { self.invalid_property( id, expr_ctx.property, - &format!( + format!( "Error parsing 'then' expression in condition #{}: {}", num + 1, err @@ -194,7 +191,7 @@ impl BootstrapExprExt for Bootstrap { self.invalid_property( id, expr_ctx.property, - &format!( + format!( "Error parsing 'if' expression in condition #{}: {}", num + 1, err diff --git a/crates/common/src/ipc.rs b/crates/common/src/ipc.rs index e41517cf..6c6d2129 100644 --- a/crates/common/src/ipc.rs +++ b/crates/common/src/ipc.rs @@ -23,7 +23,6 @@ use std::{ }, time::Instant, }; -use store::{BlobStore, InMemoryStore, Store}; use tokio::sync::{Semaphore, SemaphorePermit, mpsc}; use types::type_state::{DataType, StateChange}; use utils::map::bitmap::Bitmap; @@ -39,13 +38,9 @@ pub enum HousekeeperEvent { } pub enum PurgeType { - Data(Store), - Blobs { - store: Store, - blob_store: BlobStore, - }, + Data, + Blob, Lookup { - store: InMemoryStore, prefix: Option>, }, Account { @@ -120,7 +115,7 @@ pub enum CacheInvalidation { DavResources(u32), Domain(u32), Account(u32), - Group(u32), + DkimSignature(u32), Tenant(u32), Role(u32), List(u32), diff --git a/crates/common/src/lib.rs b/crates/common/src/lib.rs index 810ce8d1..b1469a36 100644 --- a/crates/common/src/lib.rs +++ b/crates/common/src/lib.rs @@ -40,7 +40,7 @@ use config::{ }; use ipc::{BroadcastEvent, HousekeeperEvent, PushEvent, QueueEvent, ReportingEvent}; use mail_auth::{MX, Txt}; -use manager::webadmin::{Resource, WebAdminManager}; +use manager::application::{Resource, WebApplicationManager}; use parking_lot::{Mutex, RwLock}; use rustls::sign::CertifiedKey; use std::sync::atomic::AtomicU64; @@ -159,7 +159,7 @@ pub struct Data { pub span_id_gen: SnowflakeIdGenerator, pub queue_status: AtomicBool, - pub webadmin: WebAdminManager, + pub applications: WebApplicationManager, pub logos: Mutex, Option>>>>, pub smtp_connectors: TlsConnectors, diff --git a/crates/common/src/manager/webadmin.rs b/crates/common/src/manager/application.rs similarity index 98% rename from crates/common/src/manager/webadmin.rs rename to crates/common/src/manager/application.rs index 5bf5d328..c518ada9 100644 --- a/crates/common/src/manager/webadmin.rs +++ b/crates/common/src/manager/application.rs @@ -15,7 +15,7 @@ use std::{ }; use store::BlobStore; -pub struct WebAdminManager { +pub struct WebApplicationManager { bundle_path: TempDir, routes: ArcSwap>>, } @@ -35,7 +35,7 @@ impl Resource { } } -impl WebAdminManager { +impl WebApplicationManager { pub fn new(base_path: PathBuf) -> Self { Self { bundle_path: TempDir::new(base_path), @@ -193,7 +193,7 @@ fn unpack_error(err: std::io::Error) -> trc::Error { .details("Failed to unpack webadmin bundle") } -impl Default for WebAdminManager { +impl Default for WebApplicationManager { fn default() -> Self { Self::new(std::env::temp_dir()) } diff --git a/crates/common/src/manager/boot.rs b/crates/common/src/manager/boot.rs index 06a32e05..fff22f0c 100644 --- a/crates/common/src/manager/boot.rs +++ b/crates/common/src/manager/boot.rs @@ -4,10 +4,12 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ -use super::{WEBADMIN_KEY, backup::BackupParams, console::store_console}; +use super::{backup::BackupParams, console::store_console}; use crate::{ - Caches, Core, Data, IPC_CHANNEL_BUFFER, Inner, Ipc, - config::{network::AsnGeoLookupConfig, server::Listeners, telemetry::Telemetry}, + BuildServer, Caches, Core, Data, IPC_CHANNEL_BUFFER, Inner, Ipc, + config::{ + network::AsnGeoLookupConfig, server::Listeners, storage::Storage, telemetry::Telemetry, + }, ipc::{ BroadcastEvent, HousekeeperEvent, PushEvent, QueueEvent, ReportingEvent, TrainTaskController, @@ -21,6 +23,7 @@ use std::{ sync::Arc, }; use store::{ + RegistryStore, rand::{Rng, distr::Alphanumeric, rng}, registry::bootstrap::Bootstrap, }; @@ -28,7 +31,7 @@ use tokio::sync::{Notify, mpsc}; use utils::{UnwrapFailure, failed}; pub struct BootManager { - pub bp: Bootstrap, + pub bootstrap: Bootstrap, pub inner: Arc, pub servers: Listeners, pub ipc_rxs: IpcReceivers, @@ -69,6 +72,7 @@ enum StoreOp { } pub const DEFAULT_SETTINGS: &[(&str, &str)] = &[ + ("oauth.key", "abc"), ("queue.quota.size.messages", "100000"), ("queue.quota.size.size", "10737418240"), ("queue.quota.size.enable", "true"), @@ -170,8 +174,7 @@ pub const DEFAULT_SETTINGS: &[(&str, &str)] = &[ impl BootManager { pub async fn init() -> Self { - todo!() - /*let mut config_path = std::env::var("CONFIG_PATH").ok(); + let mut config_path = std::env::var("CONFIG_PATH").ok(); let mut import_export = StoreOp::None; if config_path.is_none() { @@ -232,193 +235,28 @@ impl BootManager { } } - // Read main configuration file - let cfg_local_path = PathBuf::from(config_path.unwrap()); - let mut config = Config::default(); - match std::fs::read_to_string(&cfg_local_path) { - Ok(value) => { - config.parse(&value).failed("Invalid configuration file"); - } - Err(err) => { - config.new_build_error("*", format!("Could not read configuration file: {err}")); - } - } - let cfg_local = config.keys.clone(); + // Initialize registry + let registry = RegistryStore::init(PathBuf::from(config_path.unwrap())); + let mut bootstrap = Bootstrap::new(registry); - // Resolve environment macros - config.resolve_macros(&["env"]).await; + // Start listeners + let mut servers = Listeners::parse(&mut bootstrap).await; + servers.bind_and_drop_priv(&mut bootstrap); - // Parser servers - let mut servers = Listeners::parse(&mut config).await; - - // Bind ports and drop privileges - servers.bind_and_drop_priv(&mut config); - - // Resolve file and configuration macros - config.resolve_macros(&["file", "cfg"]).await; - - // Load stores - let mut stores = Stores::parse(&mut config).await; - let local_patterns = Patterns::parse(&mut config); - - // Build local keys and warn about database keys defined in the local configuration - let mut warn_keys = Vec::new(); - for key in config.keys.keys() { - if !local_patterns.is_local_key(key) { - warn_keys.push(key.clone()); - } - } - for warn_key in warn_keys { - config.new_build_warning( - warn_key, - concat!( - "Database key defined in local configuration, this might cause issues. ", - "See https://stalw.art/docs/configuration/overview/#loc", - "al-and-database-settings" - ), - ); - } - - // Build manager - let manager = ConfigManager { - cfg_local: ArcSwap::from_pointee(cfg_local), - cfg_local_path, - cfg_local_patterns: local_patterns.into(), - cfg_store: config - .value("storage.data") - .and_then(|id| stores.stores.get(id)) - .cloned() - .unwrap_or_default(), - }; - - // Extend configuration with settings stored in the db - if !manager.cfg_store.is_none() { - for (key, value) in manager - .db_list("", false) - .await - .failed("Failed to read database configuration") - { - if manager.cfg_local_patterns.is_local_key(&key) { - config.new_build_warning( - &key, - concat!( - "Local key defined in database, this might cause issues. ", - "See https://stalw.art/docs/configuration/overview/#loc", - "al-and-database-settings" - ), - ); - } - - config.keys.entry(key).or_insert(value); - } - } + // Parse storage + let storage = Storage::parse(&mut bootstrap).await; // Parse telemetry - let telemetry = Telemetry::parse(&mut config); + let telemetry = Telemetry::parse(&mut bootstrap, &storage).await; match import_export { StoreOp::None => { - // Add hostname lookup if missing - let mut insert_keys = Vec::new(); + let todo = "add default settings, hostname, download filter rules, webadmin"; - // Generate an OAuth key if missing - if config - .value("oauth.key") - .filter(|v| !v.is_empty()) - .is_none() - { - insert_keys.push(ConfigKey::from(( - "oauth.key", - rng() - .sample_iter(Alphanumeric) - .take(64) - .map(char::from) - .collect::(), - ))); - } - - // Download Spam filter rules if missing - if config.value("version.spam-filter").is_none() { - match manager.fetch_spam_rules().await { - Ok(external_config) => { - trc::event!( - Config(trc::ConfigEvent::ImportExternal), - Version = external_config.version.to_string(), - Id = "spam-filter" - ); - insert_keys.extend(external_config.keys); - } - Err(err) => { - config.new_build_error( - "*", - format!("Failed to fetch spam filter: {err}"), - ); - } - } - - // Add default settings - for key in DEFAULT_SETTINGS { - insert_keys.push(ConfigKey::from(*key)); - } - } - - // Download webadmin if missing - if let Some(blob_store) = config - .value("storage.blob") - .and_then(|id| stores.blob_stores.get(id)) - { - match blob_store.get_blob(WEBADMIN_KEY, 0..usize::MAX).await { - Ok(Some(_)) => (), - Ok(None) => match manager.fetch_resource("webadmin").await { - Ok(bytes) => match blob_store.put_blob(WEBADMIN_KEY, &bytes).await { - Ok(_) => { - trc::event!( - Resource(trc::ResourceEvent::DownloadExternal), - Id = "webadmin" - ); - } - Err(err) => { - config.new_build_error( - "*", - format!("Failed to store webadmin blob: {err}"), - ); - } - }, - Err(err) => { - config.new_build_error( - "*", - format!("Failed to download webadmin: {err}"), - ); - } - }, - Err(err) => config - .new_build_error("*", format!("Failed to access webadmin blob: {err}")), - } - } - - // Add missing settings - if !insert_keys.is_empty() { - for item in &insert_keys { - config.keys.insert(item.key.clone(), item.value.clone()); - } - - if let Err(err) = manager.set(insert_keys, true).await { - config - .new_build_error("*", format!("Failed to update configuration: {err}")); - } - } - - // Parse in-memory stores - stores.parse_in_memory(&mut config, false).await; - - // Parse settings - let core = Box::pin(Core::parse(&mut config, stores, manager)).await; - - // Parse data - let data = Data::parse(&mut config); - - // Parse caches - let cache = Caches::parse(&mut config); + // Parse components + let core = Box::pin(Core::parse(&mut bootstrap, storage)).await; + let data = Data::parse(&mut bootstrap).await; + let cache = Caches::parse(&mut bootstrap).await; // Enable telemetry @@ -437,40 +275,12 @@ impl BootManager { Version = env!("CARGO_PKG_VERSION"), ); - // Webadmin auto-update - // Disabled temporarily until selective updates are implemented - /*if config - .property_or_default::("webadmin.auto-update", "false") - .unwrap_or_default() - { - if let Err(err) = data.webadmin.update(&core).await { - trc::event!( - Resource(trc::ResourceEvent::Error), - Details = "Failed to update webadmin", - CausedBy = err - ); - } - }*/ - - // Spam filter auto-update - if config - .property_or_default::("spam-filter.auto-update", "false") - .unwrap_or_default() - && let Err(err) = core.storage.config.update_spam_rules(false, false).await - { - trc::event!( - Resource(trc::ResourceEvent::Error), - Details = "Failed to update spam-filter", - CausedBy = err - ); - } - // Build shared inner let has_remote_asn = matches!( core.network.asn_geo_lookup, AsnGeoLookupConfig::Resource { .. } ); - let (ipc, ipc_rxs) = build_ipc(!core.storage.pubsub.is_none()); + let (ipc, ipc_rxs) = build_ipc(!core.storage.coordinator.is_none()); let inner = Arc::new(Inner { shared_core: ArcSwap::from_pointee(core), data, @@ -495,11 +305,13 @@ impl BootManager { } // Parse TCP acceptors - servers.parse_tcp_acceptors(&mut config, inner.clone()); + servers + .parse_tcp_acceptors(&mut bootstrap, inner.clone()) + .await; BootManager { inner, - config, + bootstrap, servers, ipc_rxs, } @@ -509,7 +321,7 @@ impl BootManager { telemetry.enable(false); // Parse settings and backup - Box::pin(Core::parse(&mut config, stores, manager)) + Box::pin(Core::parse(&mut bootstrap, storage)) .await .backup(path) .await; @@ -520,7 +332,7 @@ impl BootManager { telemetry.enable(false); // Parse settings and restore - Box::pin(Core::parse(&mut config, stores, manager)) + Box::pin(Core::parse(&mut bootstrap, storage)) .await .restore(path) .await; @@ -529,7 +341,7 @@ impl BootManager { StoreOp::Console => { // Store console store_console( - Box::pin(Core::parse(&mut config, stores, manager)) + Box::pin(Core::parse(&mut bootstrap, storage)) .await .storage .data, @@ -537,7 +349,7 @@ impl BootManager { .await; std::process::exit(0); } - }*/ + } } } @@ -591,7 +403,7 @@ fn quickstart(path: impl Into) { }); std::fs::write( - path.join("etc").join("config.toml"), + path.join("etc").join("registry.json"), QUICKSTART_CONFIG .replace("_P_", &path.to_string_lossy()) .replace("_S_", &sha512_crypt::hash(&admin_pass).unwrap()), @@ -599,7 +411,7 @@ fn quickstart(path: impl Into) { .failed("Failed to write configuration file"); eprintln!( - "✅ Configuration file written to {}/etc/config.toml", + "✅ Local registry initialized at {}/etc/registry.json", path.to_string_lossy() ); eprintln!("🔑 Your administrator account is 'admin' with password '{admin_pass}'."); diff --git a/crates/common/src/manager/mod.rs b/crates/common/src/manager/mod.rs index c8eb503a..8115e103 100644 --- a/crates/common/src/manager/mod.rs +++ b/crates/common/src/manager/mod.rs @@ -9,32 +9,16 @@ use hyper::HeaderMap; use std::time::Duration; use utils::HttpLimitResponse; +pub mod application; pub mod backup; pub mod boot; pub mod console; pub mod restore; -pub mod webadmin; -const DEFAULT_SPAMFILTER_URL: &str = - "https://github.com/stalwartlabs/spam-filter/releases/latest/download/spam-filter.toml"; pub const WEBADMIN_KEY: &[u8] = "STALWART_WEBADMIN".as_bytes(); pub const SPAM_TRAINER_KEY: &[u8] = "STALWART_SPAM_TRAIN_DATA.lz4".as_bytes(); pub const SPAM_CLASSIFIER_KEY: &[u8] = "STALWART_SPAM_CLASSIFIER_MODEL.lz4".as_bytes(); -// SPDX-SnippetBegin -// SPDX-FileCopyrightText: 2020 Stalwart Labs LLC -// SPDX-License-Identifier: LicenseRef-SEL -#[cfg(feature = "enterprise")] -const DEFAULT_WEBADMIN_URL: &str = - "https://github.com/stalwartlabs/webadmin/releases/latest/download/webadmin.zip"; -// SPDX-SnippetEnd - -#[cfg(not(feature = "enterprise"))] -const DEFAULT_WEBADMIN_URL: &str = - "https://github.com/stalwartlabs/webadmin/releases/latest/download/webadmin-oss.zip"; - -const MAX_SIZE: usize = 100 * 1024 * 1024; - pub async fn fetch_resource( url: &str, headers: Option, diff --git a/crates/common/src/network/acme/cache.rs b/crates/common/src/network/acme/cache.rs index f7ee4cce..e7cb8b7a 100644 --- a/crates/common/src/network/acme/cache.rs +++ b/crates/common/src/network/acme/cache.rs @@ -6,7 +6,6 @@ use super::AcmeProvider; use crate::Server; -use base64::{Engine, engine::general_purpose::URL_SAFE_NO_PAD}; use trc::AddContext; impl Server { diff --git a/crates/common/src/network/listen.rs b/crates/common/src/network/listen.rs index 43f1b553..b086af05 100644 --- a/crates/common/src/network/listen.rs +++ b/crates/common/src/network/listen.rs @@ -17,7 +17,6 @@ use rustls::crypto::ring::cipher_suite::TLS13_AES_128_GCM_SHA256; use std::{ net::{IpAddr, SocketAddr}, sync::Arc, - time::Duration, }; use store::registry::bootstrap::Bootstrap; use tokio::{net::TcpStream, sync::watch}; @@ -55,7 +54,6 @@ impl Listener { let opts = SocketOpts { nodelay: listener.nodelay, ttl: listener.ttl, - linger: listener.linger, }; // Bind socket @@ -263,7 +261,6 @@ impl BuildSession for Arc { pub struct SocketOpts { pub nodelay: bool, pub ttl: Option, - pub linger: Option, } impl SocketOpts { @@ -285,15 +282,6 @@ impl SocketOpts { Details = "Failed to set TTL", ); } - if self.linger.is_some() - && let Err(err) = stream.set_linger(self.linger) - { - trc::event!( - Network(trc::NetworkEvent::SetOptError), - Reason = err.to_string(), - Details = "Failed to set LINGER", - ); - } } } diff --git a/crates/common/src/storage/blob.rs b/crates/common/src/storage/blob.rs index 01e0ae8f..bde449ab 100644 --- a/crates/common/src/storage/blob.rs +++ b/crates/common/src/storage/blob.rs @@ -63,7 +63,7 @@ impl Server { self.core .storage .blob - .put_blob(hash.as_ref(), data, self.core.storage.compression) + .put_blob(hash.as_ref(), data, self.core.email.compression) .await .caused_by(trc::location!())?; @@ -126,7 +126,7 @@ impl Server { self.core .storage .blob - .put_blob(hash.as_ref(), data, self.core.storage.compression) + .put_blob(hash.as_ref(), data, self.core.email.compression) .await .caused_by(trc::location!())?; diff --git a/crates/common/src/storage/index.rs b/crates/common/src/storage/index.rs index d70d6647..17512375 100644 --- a/crates/common/src/storage/index.rs +++ b/crates/common/src/storage/index.rs @@ -4,10 +4,7 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ -use crate::{ - auth::{AccessToken, AccountInfo, AccountTenantIds}, - sharing::notification::ShareNotification, -}; +use crate::{auth::AccountTenantIds, sharing::notification::ShareNotification}; use rkyv::{ option::ArchivedOption, primitive::{ArchivedU32, ArchivedU64}, diff --git a/crates/common/src/storage/mod.rs b/crates/common/src/storage/mod.rs index 8f933ca1..a8717fad 100644 --- a/crates/common/src/storage/mod.rs +++ b/crates/common/src/storage/mod.rs @@ -49,7 +49,7 @@ impl Server { #[inline(always)] pub fn search_store(&self) -> &SearchStore { - &self.core.storage.fts + &self.core.storage.search } #[inline(always)] @@ -57,6 +57,16 @@ impl Server { &self.core.storage.memory } + #[inline(always)] + pub fn tracing_store(&self) -> &Store { + &self.core.storage.tracing + } + + #[inline(always)] + pub fn metrics_store(&self) -> &Store { + &self.core.storage.metrics + } + #[inline(always)] pub fn get_directory(&self, id: &u32) -> Option<&Arc> { self.core.storage.directories.get(id) @@ -88,7 +98,7 @@ impl Server { pub async fn logo_resource( &self, _: &str, - ) -> trc::Result>>> { + ) -> trc::Result>>> { Ok(None) } } diff --git a/crates/common/src/storage/quota.rs b/crates/common/src/storage/quota.rs index 7e116fa7..0b9b40a8 100644 --- a/crates/common/src/storage/quota.rs +++ b/crates/common/src/storage/quota.rs @@ -4,6 +4,10 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ +use crate::{ + Server, + storage::{ObjectQuota, TenantQuota}, +}; use registry::{ schema::enums::{StorageQuota, TenantStorageQuota}, types::EnumType, @@ -11,12 +15,6 @@ use registry::{ use store::write::DirectoryClass; use trc::AddContext; -use crate::{ - Server, - auth::AccountCache, - storage::{ObjectQuota, TenantQuota}, -}; - impl Server { pub async fn get_used_quota_account(&self, account_id: u32) -> trc::Result { self.core diff --git a/crates/common/src/storage/state.rs b/crates/common/src/storage/state.rs index 2e9b563f..979fb4a9 100644 --- a/crates/common/src/storage/state.rs +++ b/crates/common/src/storage/state.rs @@ -69,7 +69,6 @@ impl Server { } pub async fn cluster_broadcast(&self, event: BroadcastEvent) { - let todo = "refactor event names"; if let Some(broadcast_tx) = &self.inner.ipc.broadcast_tx.clone() && broadcast_tx.send(event).await.is_err() { diff --git a/crates/common/src/telemetry/metrics/store.rs b/crates/common/src/telemetry/metrics/store.rs index fdda3007..b172fc67 100644 --- a/crates/common/src/telemetry/metrics/store.rs +++ b/crates/common/src/telemetry/metrics/store.rs @@ -8,11 +8,11 @@ * */ -use std::{future::Future, sync::Arc, time::Duration}; - +use crate::Core; use ahash::AHashMap; use parking_lot::Mutex; use serde::{Deserialize, Serialize}; +use std::{future::Future, sync::Arc, time::Duration}; use store::{ IterateParams, Store, U32_LEN, U64_LEN, ValueKey, write::{ @@ -24,8 +24,6 @@ use store::{ use trc::*; use utils::codec::leb128::Leb128Reader; -use crate::Core; - pub trait MetricsStore: Sync + Send { fn write_metrics( &self, diff --git a/crates/common/src/telemetry/tracers/store.rs b/crates/common/src/telemetry/tracers/store.rs index 3c7e38c7..b75a5f28 100644 --- a/crates/common/src/telemetry/tracers/store.rs +++ b/crates/common/src/telemetry/tracers/store.rs @@ -157,7 +157,7 @@ impl TracingStore for Store { impl StoreTracer { pub fn default_events() -> impl IntoIterator { EventType::variants() - .into_iter() + .iter() .filter(|event| { !event.is_raw_io() && matches!( diff --git a/crates/coordinator/src/bootstrap.rs b/crates/coordinator/src/bootstrap.rs index 3ec480e7..62d6c999 100644 --- a/crates/coordinator/src/bootstrap.rs +++ b/crates/coordinator/src/bootstrap.rs @@ -61,6 +61,7 @@ impl Coordinator { } } +#[cfg(feature = "redis")] fn unwrap_redis(store: InMemoryStore) -> Coordinator { if let InMemoryStore::Redis(redis) = store { Coordinator::Redis(redis) diff --git a/crates/directory/Cargo.toml b/crates/directory/Cargo.toml index 54a8a518..364e7fd3 100644 --- a/crates/directory/Cargo.toml +++ b/crates/directory/Cargo.toml @@ -38,6 +38,7 @@ serde_json = "1.0" base64 = "0.22" rkyv = { version = "0.8.10", features = ["little_endian"] } compact_str = { version = "0.9.0", features = ["rkyv", "serde"] } +nohash-hasher = "0.2.0" [dev-dependencies] tokio = { version = "1.47", features = ["full"] } diff --git a/crates/directory/src/core/config.rs b/crates/directory/src/core/config.rs index 68786700..8b819a34 100644 --- a/crates/directory/src/core/config.rs +++ b/crates/directory/src/core/config.rs @@ -8,17 +8,16 @@ use crate::{ Directories, backend::{ldap::LdapDirectory, oidc::OpenIdDirectory, sql::SqlDirectory}, }; -use ahash::AHashMap; use registry::schema::{ prelude::Object, structs::{self, Authentication}, }; -use std::sync::Arc; +use std::{collections::HashMap, sync::Arc}; use store::registry::bootstrap::Bootstrap; impl Directories { pub async fn build(bp: &mut Bootstrap) -> Self { - let mut directories = AHashMap::new(); + let mut directories = HashMap::default(); for directory in bp.list_infallible::().await { let id = directory.id; @@ -32,7 +31,7 @@ impl Directories { match result { Ok(directory) => { - directories.insert(id, Arc::new(directory)); + directories.insert(id.id() as u32, Arc::new(directory)); } Err(err) => { bp.build_error(id, err); diff --git a/crates/directory/src/lib.rs b/crates/directory/src/lib.rs index 96b09cea..801ca052 100644 --- a/crates/directory/src/lib.rs +++ b/crates/directory/src/lib.rs @@ -7,12 +7,10 @@ #![warn(clippy::large_futures)] use crate::backend::oidc::OpenIdDirectory; -use ahash::AHashMap; use backend::{ldap::LdapDirectory, sql::SqlDirectory}; use deadpool::managed::PoolError; use ldap3::LdapError; -use registry::types::id::Id; -use std::{fmt::Debug, sync::Arc}; +use std::{collections::HashMap, fmt::Debug, sync::Arc}; pub mod backend; pub mod core; @@ -57,10 +55,10 @@ pub struct Group { pub description: Option, } -#[derive(Default, Clone, Debug)] +#[derive(Clone, Debug)] pub struct Directories { pub default_directory: Option>, - pub directories: AHashMap>, + pub directories: HashMap, nohash_hasher::BuildNoHashHasher>, } impl Debug for Directory { diff --git a/crates/http-proto/src/response.rs b/crates/http-proto/src/response.rs index 44322f03..e6221fb5 100644 --- a/crates/http-proto/src/response.rs +++ b/crates/http-proto/src/response.rs @@ -4,7 +4,7 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ -use common::manager::webadmin::Resource; +use common::manager::application::Resource; use http_body_util::{BodyExt, Full}; use hyper::{ StatusCode, diff --git a/crates/http/src/autoconfig/mod.rs b/crates/http/src/autoconfig/mod.rs index 6a9793f2..7969b488 100644 --- a/crates/http/src/autoconfig/mod.rs +++ b/crates/http/src/autoconfig/mod.rs @@ -4,7 +4,7 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ -use common::{Server, manager::webadmin::Resource}; +use common::{Server, manager::application::Resource}; use http_proto::*; use quick_xml::Reader; use quick_xml::events::Event; diff --git a/crates/http/src/management/telemetry.rs b/crates/http/src/management/telemetry.rs index d7322649..d30a1c1e 100644 --- a/crates/http/src/management/telemetry.rs +++ b/crates/http/src/management/telemetry.rs @@ -154,16 +154,12 @@ impl TelemetryApi for Server { tracing_query.push(SearchFilter::End); - let store = &self - .core - .enterprise - .as_ref() - .and_then(|e| e.trace_store.as_ref()) - .ok_or_else(|| { - trc::ManageEvent::NotSupported - .ctx(trc::Key::Details, "No tracing store has been configured") - })? - .store; + let store = self.tracing_store(); + + if !store.is_active() { + return Err(trc::ManageEvent::NotSupported + .ctx(trc::Key::Details, "No tracing store has been configured")); + } let span_ids = self .search_store() @@ -354,16 +350,11 @@ impl TelemetryApi for Server { // Validate the access token access_token.enforce_permission(Permission::TracingGet)?; - let store = &self - .core - .enterprise - .as_ref() - .and_then(|e| e.trace_store.as_ref()) - .ok_or_else(|| { - trc::ManageEvent::NotSupported - .ctx(trc::Key::Details, "No tracing store has been configured") - })? - .store; + let store = self.tracing_store(); + if !store.is_active() { + return Err(trc::ManageEvent::NotSupported + .ctx(trc::Key::Details, "No tracing store has been configured")); + } let mut events = Vec::new(); for span_id in id @@ -415,6 +406,7 @@ impl TelemetryApi for Server { .into_http_response()) } ("metrics", None, &Method::GET) => { + let todo = "move to registry"; // Validate the access token access_token.enforce_permission(Permission::MetricsList)?; @@ -426,25 +418,20 @@ impl TelemetryApi for Server { .parse::("after") .map(|t| t.into_inner()) .unwrap_or(0); - let results = self - .core - .enterprise - .as_ref() - .and_then(|e| e.metrics_store.as_ref()) - .ok_or_else(|| { - trc::ManageEvent::Error - .ctx(trc::Key::Details, "No metrics store has been defined") - .ctx( - trc::Key::Reason, - concat!( - "You need to configure a metrics ", - "store in order to use this feature." - ), - ) - })? - .store - .query_metrics(after, before) - .await?; + + if !self.metrics_store().is_active() { + return Err(trc::ManageEvent::Error + .ctx(trc::Key::Details, "No metrics store has been defined") + .ctx( + trc::Key::Reason, + concat!( + "You need to configure a metrics ", + "store in order to use this feature." + ), + )); + } + + let results = self.metrics_store().query_metrics(after, before).await?; let mut metrics = Vec::with_capacity(results.len()); for metric in results { diff --git a/crates/http/src/request.rs b/crates/http/src/request.rs index 00840586..3e2d3dca 100644 --- a/crates/http/src/request.rs +++ b/crates/http/src/request.rs @@ -21,7 +21,7 @@ use common::{ BuildServer, Inner, KV_ACME, Server, auth::{AccessToken, oauth::GrantType}, ipc::PushEvent, - manager::webadmin::Resource, + manager::application::Resource, network::{SessionData, SessionManager, SessionStream}, }; use dav::{DavMethod, request::DavRequestHandler}; @@ -565,7 +565,7 @@ impl ParseHttp for Server { } } - let resource = self.inner.data.webadmin.get("logo.svg").await?; + let resource = self.inner.data.applications.get("logo.svg").await?; if !resource.is_empty() { return Ok(resource.into_http_response()); @@ -599,7 +599,7 @@ impl ParseHttp for Server { let resource = self .inner .data - .webadmin + .applications .get(path.strip_prefix('/').unwrap_or(path)) .await?; diff --git a/crates/jmap/src/api/request.rs b/crates/jmap/src/api/request.rs index 9b411ee0..8b5f93ea 100644 --- a/crates/jmap/src/api/request.rs +++ b/crates/jmap/src/api/request.rs @@ -369,10 +369,9 @@ impl RequestHandler for Server { self.sieve_script_query(req).await?.into() } - QueryRequestMethod::Principal(req) => self - .principal_query(req, access_token, session) - .await? - .into(), + QueryRequestMethod::Principal(req) => { + self.principal_query(req, access_token).await?.into() + } QueryRequestMethod::Quota(mut req) => { set_account_id_if_missing(&mut req.account_id, access_token); access_token.assert_is_member(req.account_id)?; diff --git a/crates/jmap/src/blob/upload.rs b/crates/jmap/src/blob/upload.rs index 0d64e0fa..b569f1ad 100644 --- a/crates/jmap/src/blob/upload.rs +++ b/crates/jmap/src/blob/upload.rs @@ -4,8 +4,6 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ -use std::sync::Arc; - use super::{UploadResponse, download::BlobDownload}; use common::{Server, auth::AccessToken}; use jmap_proto::{ diff --git a/crates/jmap/src/principal/query.rs b/crates/jmap/src/principal/query.rs index df946e21..f4fd1e26 100644 --- a/crates/jmap/src/principal/query.rs +++ b/crates/jmap/src/principal/query.rs @@ -6,13 +6,18 @@ use crate::api::query::QueryResponseBuilder; use common::{Server, auth::AccessToken}; -use http_proto::HttpSessionData; use jmap_proto::{ method::query::{Filter, QueryRequest, QueryResponse}, object::principal::{Principal, PrincipalFilter, PrincipalType}, types::state::State, }; -use registry::schema::prelude::{Object, Permission, Property}; +use registry::{ + schema::{ + enums::AccountType, + prelude::{Object, Permission, Property}, + }, + types::EnumType, +}; use std::future::Future; use store::{ registry::RegistryQuery, @@ -27,7 +32,6 @@ pub trait PrincipalQuery: Sync + Send { &self, request: QueryRequest, access_token: &AccessToken, - session: &HttpSessionData, ) -> impl Future> + Send; } @@ -36,7 +40,6 @@ impl PrincipalQuery for Server { &self, mut request: QueryRequest, access_token: &AccessToken, - session: &HttpSessionData, ) -> trc::Result { if !self.core.groupware.allow_directory_query && !access_token.has_permission(Permission::IndividualList) @@ -58,74 +61,73 @@ impl PrincipalQuery for Server { let mut filters = Vec::with_capacity(request.filter.len()); for cond in std::mem::take(&mut request.filter) { match cond { - Filter::Property(cond) => { - match cond { - PrincipalFilter::Name(name) | PrincipalFilter::Email(name) => { - if let Some(account_id) = self.account_id(&name).await? { - filters.push(SearchFilter::is_in_set( - RoaringBitmap::from_sorted_iter([account_id]).unwrap(), - )); - } - } - PrincipalFilter::AccountIds(ids) => { + Filter::Property(cond) => match cond { + PrincipalFilter::Name(name) | PrincipalFilter::Email(name) => { + if let Some(account_id) = self.account_id(&name).await? { filters.push(SearchFilter::is_in_set( - ids.into_iter() - .filter_map(|id| { - let id = id.document_id(); - if principal_ids.contains(id) { - Some(id) - } else { - None - } - }) - .collect::(), + RoaringBitmap::from_sorted_iter([account_id]).unwrap(), )); } - PrincipalFilter::Text(text) => { - filters.push(SearchFilter::is_in_set( - self.registry() - .query::( - RegistryQuery::new(Object::Account) - .equal_opt( - Property::MemberTenantId, - access_token.tenant_id(), - ) - .text(text), - ) - .await - .caused_by(trc::location!())?, - )); - } - PrincipalFilter::Type(principal_type) => { - let todo = "make sure this works"; - let typ = match principal_type { - PrincipalType::Individual => Object::UserAccount, - PrincipalType::Group => Object::GroupAccount, - PrincipalType::Resource - | PrincipalType::Location - | PrincipalType::Other => { - filters.push(SearchFilter::is_in_set(Default::default())); - continue; - } - }; - - filters.push(SearchFilter::is_in_set( - self.registry() - .query::(RegistryQuery::new(typ).equal_opt( - Property::MemberTenantId, - access_token.tenant_id(), - )) - .await - .caused_by(trc::location!())?, - )); - } - other => { - return Err(trc::JmapEvent::UnsupportedFilter - .into_err() - .details(other.to_string())); - } } - } + PrincipalFilter::AccountIds(ids) => { + filters.push(SearchFilter::is_in_set( + ids.into_iter() + .filter_map(|id| { + let id = id.document_id(); + if principal_ids.contains(id) { + Some(id) + } else { + None + } + }) + .collect::(), + )); + } + PrincipalFilter::Text(text) => { + filters.push(SearchFilter::is_in_set( + self.registry() + .query::( + RegistryQuery::new(Object::Account) + .equal_opt( + Property::MemberTenantId, + access_token.tenant_id(), + ) + .text(text), + ) + .await + .caused_by(trc::location!())?, + )); + } + PrincipalFilter::Type(principal_type) => { + let typ = match principal_type { + PrincipalType::Individual => AccountType::User, + PrincipalType::Group => AccountType::Group, + _ => { + filters.push(SearchFilter::is_in_set(Default::default())); + continue; + } + }; + + filters.push(SearchFilter::is_in_set( + self.registry() + .query::( + RegistryQuery::new(Object::Account) + .equal(Property::Type, typ.to_id()) + .equal_opt( + Property::MemberTenantId, + access_token.tenant_id(), + ), + ) + .await + .caused_by(trc::location!())?, + )); + } + other => { + return Err(trc::JmapEvent::UnsupportedFilter + .into_err() + .details(other.to_string())); + } + }, Filter::And => { filters.push(SearchFilter::And); } diff --git a/crates/main/src/main.rs b/crates/main/src/main.rs index 7b8ec878..972e6b9a 100644 --- a/crates/main/src/main.rs +++ b/crates/main/src/main.rs @@ -9,7 +9,7 @@ #![warn(clippy::cast_possible_wrap)] #![warn(clippy::cast_sign_loss)] -use common::{config::server::ServerProtocol, core::BuildServer, manager::boot::BootManager}; +use common::{BuildServer, config::server::ServerProtocol, manager::boot::BootManager}; use http::HttpSessionManager; use imap::core::ImapSessionManager; use managesieve::core::ManageSieveSessionManager; @@ -47,8 +47,8 @@ async fn main() -> std::io::Result<()> { init.start_queue_manager(); // Log configuration errors - init.config.log_errors(); - init.config.log_warnings(); + init.bootstrap.log_errors(); + init.bootstrap.log_warnings(); // SPDX-SnippetBegin // SPDX-FileCopyrightText: 2020 Stalwart Labs LLC diff --git a/crates/registry/Cargo.toml b/crates/registry/Cargo.toml index fa515bf2..4ab41838 100644 --- a/crates/registry/Cargo.toml +++ b/crates/registry/Cargo.toml @@ -10,6 +10,7 @@ types = { path = "../types" } serde = { version = "1.0", features = ["derive"]} serde_json = "1.0" hashify = "0.2.7" +ahash = { version = "0.8" } [features] test_mode = [] diff --git a/crates/registry/src/pickle.rs b/crates/registry/src/pickle.rs index 7efad4ff..7b0993f6 100644 --- a/crates/registry/src/pickle.rs +++ b/crates/registry/src/pickle.rs @@ -4,6 +4,8 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ +use utils::map::vec_map::VecMap; + use crate::types::EnumType; use std::collections::HashMap; @@ -45,8 +47,8 @@ impl Pickle for u16 { } fn unpickle(stream: &mut PickledStream<'_>) -> Option { - let mut arr = [0u8; 2]; - arr.copy_from_slice(stream.read_bytes(2)?); + let mut arr = [0u8; std::mem::size_of::()]; + arr.copy_from_slice(stream.read_bytes(std::mem::size_of::())?); Some(u16::from_le_bytes(arr)) } } @@ -57,20 +59,32 @@ impl Pickle for u64 { } fn unpickle(stream: &mut PickledStream<'_>) -> Option { - let mut arr = [0u8; 8]; - arr.copy_from_slice(stream.read_bytes(8)?); + let mut arr = [0u8; std::mem::size_of::()]; + arr.copy_from_slice(stream.read_bytes(std::mem::size_of::())?); Some(u64::from_le_bytes(arr)) } } +impl Pickle for u32 { + fn pickle(&self, out: &mut Vec) { + out.extend_from_slice(&self.to_le_bytes()); + } + + fn unpickle(stream: &mut PickledStream<'_>) -> Option { + let mut arr = [0u8; std::mem::size_of::()]; + arr.copy_from_slice(stream.read_bytes(std::mem::size_of::())?); + Some(u32::from_le_bytes(arr)) + } +} + impl Pickle for i64 { fn pickle(&self, out: &mut Vec) { out.extend_from_slice(&self.to_le_bytes()); } fn unpickle(stream: &mut PickledStream<'_>) -> Option { - let mut arr = [0u8; 8]; - arr.copy_from_slice(stream.read_bytes(8)?); + let mut arr = [0u8; std::mem::size_of::()]; + arr.copy_from_slice(stream.read_bytes(std::mem::size_of::())?); Some(i64::from_le_bytes(arr)) } } @@ -81,8 +95,8 @@ impl Pickle for f64 { } fn unpickle(stream: &mut PickledStream<'_>) -> Option { - let mut arr = [0u8; 8]; - arr.copy_from_slice(stream.read_bytes(8)?); + let mut arr = [0u8; std::mem::size_of::()]; + arr.copy_from_slice(stream.read_bytes(std::mem::size_of::())?); Some(f64::from_le_bytes(arr)) } } @@ -203,6 +217,33 @@ where } } +impl Pickle for VecMap +where + K: Pickle + std::hash::Hash + Eq, + V: Pickle, +{ + fn pickle(&self, out: &mut Vec) { + out.extend_from_slice(&(self.len() as u32).to_le_bytes()); + for (key, value) in self { + key.pickle(out); + value.pickle(out); + } + } + + fn unpickle(stream: &mut PickledStream<'_>) -> Option { + let mut len_arr = [0u8; 4]; + len_arr.copy_from_slice(stream.read_bytes(4)?); + let len = u32::from_le_bytes(len_arr) as usize; + let mut map = VecMap::with_capacity(len); + for _ in 0..len { + let key = K::unpickle(stream)?; + let value = V::unpickle(stream)?; + map.append(key, value); + } + Some(map) + } +} + impl Pickle for trc::EventType { fn pickle(&self, out: &mut Vec) { out.extend_from_slice(&self.to_id().to_le_bytes()); diff --git a/crates/registry/src/schema/mod.rs b/crates/registry/src/schema/mod.rs index 2b278451..faa8256a 100644 --- a/crates/registry/src/schema/mod.rs +++ b/crates/registry/src/schema/mod.rs @@ -11,11 +11,12 @@ use crate::{ }, types::EnumType, }; -use std::{collections::HashMap, fmt::Display}; +use std::fmt::Display; use utils::{ Client, HeaderMap, cron::SimpleCron, http::{build_http_client, build_http_headers}, + map::vec_map::VecMap, }; #[allow(clippy::derivable_impls)] @@ -67,7 +68,7 @@ impl Account { impl HttpAuth { pub fn build_headers( &self, - extra_headers: HashMap, + extra_headers: VecMap, content_type: Option<&str>, ) -> Result { match self { @@ -93,7 +94,7 @@ impl HttpAuth { pub fn build_http_client( &self, - extra_headers: HashMap, + extra_headers: VecMap, content_type: Option<&str>, timeout: Duration, allow_invalid_certs: bool, diff --git a/crates/registry/src/schema/prelude.rs b/crates/registry/src/schema/prelude.rs index 6fbe25f6..8a5ad16a 100644 --- a/crates/registry/src/schema/prelude.rs +++ b/crates/registry/src/schema/prelude.rs @@ -9,17 +9,19 @@ pub use crate::schema::enums::*; pub use crate::schema::properties::*; pub use crate::schema::structs::*; pub use crate::types::EnumType; +pub use crate::types::ObjectIndex; pub use crate::types::ObjectType; pub use crate::types::datetime::UTCDateTime; pub use crate::types::duration::Duration; pub use crate::types::error::*; pub use crate::types::id::Id; +pub use crate::types::index::IndexBuilder; pub use crate::types::ipaddr::IpAddr; pub use crate::types::ipmask::IpAddrOrMask; pub use crate::types::socketaddr::SocketAddr; pub use serde::{Deserialize, Serialize}; -pub use std::collections::HashMap; pub use std::str::FromStr; +pub use utils::map::vec_map::VecMap; #[derive(Debug)] pub struct ExpressionContext<'x> { diff --git a/crates/registry/src/types/index.rs b/crates/registry/src/types/index.rs new file mode 100644 index 00000000..7e8e9fff --- /dev/null +++ b/crates/registry/src/types/index.rs @@ -0,0 +1,168 @@ +/* + * SPDX-FileCopyrightText: 2020 Stalwart Labs LLC + * + * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL + */ + +use crate::{ + schema::prelude::{Object, Property}, + types::ipmask::IpAddrOrMask, +}; +use ahash::AHashSet; +use std::borrow::Cow; + +#[derive(Debug, Clone, PartialEq, Eq, Copy, Hash)] +pub enum IndexType { + Unique, + Search, + TextSearch, + GlobalUnique, +} + +#[derive(Debug, Clone, PartialEq, Eq, Hash)] +pub enum IndexValue<'x> { + Text(Cow<'x, str>), + Bytes(Vec), + U64(u64), + I64(i64), + U16(u16), +} + +#[derive(Debug, Clone, PartialEq, Eq, Hash)] +pub struct IndexKey<'x> { + pub property: Property, + pub typ: IndexType, + pub value: IndexValue<'x>, +} + +#[derive(Debug, Default)] + +pub struct IndexBuilder<'x> { + pub object: Option, + pub keys: AHashSet>, +} + +impl<'x> IndexBuilder<'x> { + pub fn object(&mut self, object: Object) { + if self.object.is_none() { + self.object = Some(object); + } + } + + pub fn typ(&mut self, typ: u16) { + self.keys.insert(IndexKey { + property: Property::Type, + typ: IndexType::Search, + value: IndexValue::U16(typ), + }); + } + + pub fn unique(&mut self, property: Property, value: impl Into>) { + self.keys.insert(IndexKey { + property, + typ: IndexType::Unique, + value: value.into(), + }); + } + + pub fn search(&mut self, property: Property, value: impl Into>) { + self.keys.insert(IndexKey { + property, + typ: IndexType::Search, + value: value.into(), + }); + } + + pub fn text(&mut self, property: Property, value: &'x str) { + for word in value + .split(|c: char| !c.is_alphanumeric()) + .filter(|s| s.len() > 1) + { + if word + .chars() + .all(|ch| ch.is_lowercase() || !ch.is_alphabetic()) + { + self.keys.insert(IndexKey { + property, + typ: IndexType::TextSearch, + value: IndexValue::Text(Cow::Borrowed(word)), + }); + } else { + self.keys.insert(IndexKey { + property, + typ: IndexType::TextSearch, + value: IndexValue::Text(Cow::Owned(word.to_lowercase())), + }); + } + } + } + + pub fn global_unique(&mut self, property: Property, value: impl Into>) { + self.keys.insert(IndexKey { + property, + typ: IndexType::GlobalUnique, + value: value.into(), + }); + } +} + +impl From for IndexValue<'_> { + fn from(value: u64) -> Self { + IndexValue::U64(value) + } +} + +impl From<&u64> for IndexValue<'_> { + fn from(value: &u64) -> Self { + IndexValue::U64(*value) + } +} + +impl From for IndexValue<'_> { + fn from(value: i64) -> Self { + IndexValue::I64(value) + } +} + +impl From<&i64> for IndexValue<'_> { + fn from(value: &i64) -> Self { + IndexValue::I64(*value) + } +} + +impl<'x> From<&'x IpAddrOrMask> for IndexValue<'x> { + fn from(value: &'x IpAddrOrMask) -> Self { + match value { + IpAddrOrMask::V4 { addr, mask } => { + let mut bytes = Vec::with_capacity(8); + bytes.extend_from_slice(&addr.octets()); + bytes.extend_from_slice(&mask.to_be_bytes()); + IndexValue::Bytes(bytes) + } + IpAddrOrMask::V6 { addr, mask } => { + let mut bytes = Vec::with_capacity(24); + bytes.extend_from_slice(&addr.octets()); + bytes.extend_from_slice(&mask.to_be_bytes()); + IndexValue::Bytes(bytes) + } + } + } +} + +impl<'x> From<&'x trc::EventType> for IndexValue<'x> { + fn from(value: &'x trc::EventType) -> Self { + IndexValue::U16(value.to_id()) + } +} + +impl<'x> From<&'x str> for IndexValue<'x> { + fn from(value: &'x str) -> Self { + IndexValue::Text(value.into()) + } +} + +impl<'x> From<&'x String> for IndexValue<'x> { + fn from(value: &'x String) -> Self { + IndexValue::Text(Cow::Borrowed(value.as_str())) + } +} diff --git a/crates/registry/src/types/ipmask.rs b/crates/registry/src/types/ipmask.rs index 4279cd59..19d0b03c 100644 --- a/crates/registry/src/types/ipmask.rs +++ b/crates/registry/src/types/ipmask.rs @@ -4,14 +4,13 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ +use crate::pickle::{Pickle, PickledStream}; use std::{ fmt::{Display, Formatter}, net::{IpAddr, Ipv4Addr, Ipv6Addr}, str::FromStr, }; -use crate::pickle::{Pickle, PickledStream}; - #[derive(Debug, Clone, PartialEq, Eq)] pub enum IpAddrOrMask { V4 { addr: Ipv4Addr, mask: u32 }, diff --git a/crates/registry/src/types/mod.rs b/crates/registry/src/types/mod.rs index b8508871..7cc9f5af 100644 --- a/crates/registry/src/types/mod.rs +++ b/crates/registry/src/types/mod.rs @@ -4,12 +4,17 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ -use crate::{pickle::Pickle, schema::prelude::Object, types::error::ValidationError}; +use crate::{ + pickle::Pickle, + schema::prelude::Object, + types::{error::ValidationError, index::IndexBuilder}, +}; pub mod datetime; pub mod duration; pub mod error; pub mod id; +pub mod index; pub mod ipaddr; pub mod ipmask; pub mod socketaddr; @@ -27,3 +32,7 @@ pub trait ObjectType: Pickle + Default + Clone + Send + Sync { fn object() -> Object; fn validate(&self, errors: &mut Vec) -> bool; } + +pub trait ObjectIndex<'x>: Send + Sync { + fn index(&'x self, builder: &mut IndexBuilder<'x>); +} diff --git a/crates/services/src/broadcast/mod.rs b/crates/services/src/broadcast/mod.rs index 633566cd..5aceac0a 100644 --- a/crates/services/src/broadcast/mod.rs +++ b/crates/services/src/broadcast/mod.rs @@ -101,7 +101,7 @@ impl BroadcastBatch> { CacheInvalidation::DavResources(id) => (1u8, *id), CacheInvalidation::Domain(id) => (2u8, *id), CacheInvalidation::Account(id) => (3u8, *id), - CacheInvalidation::Group(id) => (4u8, *id), + CacheInvalidation::DkimSignature(id) => (4u8, *id), CacheInvalidation::Tenant(id) => (5u8, *id), CacheInvalidation::Role(id) => (6u8, *id), CacheInvalidation::List(id) => (7u8, *id), @@ -206,7 +206,7 @@ where 1 => CacheInvalidation::DavResources(id), 2 => CacheInvalidation::Domain(id), 3 => CacheInvalidation::Account(id), - 4 => CacheInvalidation::Group(id), + 4 => CacheInvalidation::DkimSignature(id), 5 => CacheInvalidation::Tenant(id), 6 => CacheInvalidation::Role(id), 7 => CacheInvalidation::List(id), diff --git a/crates/services/src/housekeeper/mod.rs b/crates/services/src/housekeeper/mod.rs index c9934088..88aa203d 100644 --- a/crates/services/src/housekeeper/mod.rs +++ b/crates/services/src/housekeeper/mod.rs @@ -170,9 +170,9 @@ pub fn spawn_housekeeper(inner: Arc, mut rx: mpsc::Receiver, mut rx: mpsc::Receiver, mut rx: mpsc::Receiver { @@ -430,10 +423,8 @@ pub fn spawn_housekeeper(inner: Arc, mut rx: mpsc::Receiver { @@ -615,23 +606,25 @@ pub fn spawn_housekeeper(inner: Arc, mut rx: mpsc::Receiver { - if let Some(metrics_store) = &server - .core - .enterprise - .as_ref() - .and_then(|e| e.metrics_store.as_ref()) - { + if server.core.storage.metrics.is_active() { trc::event!( Housekeeper(trc::HousekeeperEvent::Run), Type = "metrics_internal" ); queue.schedule( - Instant::now() + metrics_store.interval.time_to_next(), + Instant::now() + + server + .core + .enterprise + .as_ref() + .unwrap() + .metrics_interval + .time_to_next(), ActionClass::InternalMetrics, ); - let metrics_store = metrics_store.store.clone(); + let metrics_store = server.core.storage.metrics.clone(); let metrics_history = metrics_history.clone(); let core = server.core.clone(); tokio::spawn(async move { @@ -742,9 +735,9 @@ impl Purge for Server { async fn purge(&self, purge: PurgeType) { // Lock task let (lock_type, lock_name) = match &purge { - PurgeType::Data(_) => ("data", [0u8].into()), - PurgeType::Blobs { .. } => ("blob", [1u8].into()), - PurgeType::Lookup { prefix: None, .. } => ("in-memory", [2u8].into()), + PurgeType::Data => ("data", [0u8].into()), + PurgeType::Blob => ("blob", [1u8].into()), + PurgeType::Lookup { prefix: None } => ("in-memory", [2u8].into()), PurgeType::Lookup { .. } => ("in-memory-prefix", None), PurgeType::Account { .. } => ("account", None), }; @@ -770,27 +763,8 @@ impl Purge for Server { let time = Instant::now(); match purge { - PurgeType::Data(store) => { - // SPDX-SnippetBegin - // SPDX-FileCopyrightText: 2020 Stalwart Labs LLC - // SPDX-License-Identifier: LicenseRef-SEL - #[cfg(feature = "enterprise")] - let trace_retention = self - .core - .enterprise - .as_ref() - .and_then(|e| e.trace_store.as_ref()) - .and_then(|t| t.retention); - #[cfg(feature = "enterprise")] - let metrics_retention = self - .core - .enterprise - .as_ref() - .and_then(|e| e.metrics_store.as_ref()) - .and_then(|m| m.retention); - // SPDX-SnippetEnd - - if let Err(err) = store.purge_store().await { + PurgeType::Data => { + if let Err(err) = self.store().purge_store().await { trc::error!(err.details("Failed to purge data store")); } @@ -798,14 +772,14 @@ impl Purge for Server { // SPDX-FileCopyrightText: 2020 Stalwart Labs LLC // SPDX-License-Identifier: LicenseRef-SEL #[cfg(feature = "enterprise")] - if let Some(trace_retention) = trace_retention - && let Some(trace_store) = self - .core - .enterprise - .as_ref() - .and_then(|e| e.trace_store.as_ref()) - && let Err(err) = trace_store - .store + if let Some(trace_retention) = self + .core + .enterprise + .as_ref() + .and_then(|e| e.trace_retention) + && self.tracing_store().is_active() + && let Err(err) = self + .tracing_store() .purge_spans(trace_retention, self.search_store().into()) .await { @@ -813,32 +787,32 @@ impl Purge for Server { } #[cfg(feature = "enterprise")] - if let Some(metrics_retention) = metrics_retention - && let Some(metrics_store) = self - .core - .enterprise - .as_ref() - .and_then(|e| e.metrics_store.as_ref()) - && let Err(err) = metrics_store.store.purge_metrics(metrics_retention).await + if let Some(metrics_retention) = self + .core + .enterprise + .as_ref() + .and_then(|e| e.metrics_retention) + && self.metrics_store().is_active() + && let Err(err) = self.metrics_store().purge_metrics(metrics_retention).await { trc::error!(err.details("Failed to purge metrics")); } // SPDX-SnippetEnd } - PurgeType::Blobs { store, blob_store } => { - if let Err(err) = store.purge_blobs(blob_store).await { + PurgeType::Blob => { + if let Err(err) = self.store().purge_blobs(self.blob_store().clone()).await { trc::error!(err.details("Failed to purge blob store")); } } - PurgeType::Lookup { store, prefix } => { + PurgeType::Lookup { prefix } => { if let Some(prefix) = prefix { - if let Err(err) = store.key_delete_prefix(&prefix).await { + if let Err(err) = self.in_memory_store().key_delete_prefix(&prefix).await { trc::error!( err.details("Failed to delete key prefix") .ctx(trc::Key::Key, prefix) ); } - } else if let Err(err) = store.purge_in_memory_store().await { + } else if let Err(err) = self.in_memory_store().purge_in_memory_store().await { trc::error!(err.details("Failed to purge in-memory store")); } } diff --git a/crates/services/src/lib.rs b/crates/services/src/lib.rs index 6fe30e98..7b37dc40 100644 --- a/crates/services/src/lib.rs +++ b/crates/services/src/lib.rs @@ -33,14 +33,14 @@ impl StartServices for BootManager { if let Err(err) = self .inner .data - .webadmin + .applications .unpack(&self.inner.shared_core.load().storage.blob) .await { trc::event!( Resource(trc::ResourceEvent::Error), Reason = err, - Details = "Failed to unpack webadmin bundle" + Details = "Failed to unpack application bundle" ); } diff --git a/crates/services/src/task_manager/alarm.rs b/crates/services/src/task_manager/alarm.rs index 963808d4..fe87c565 100644 --- a/crates/services/src/task_manager/alarm.rs +++ b/crates/services/src/task_manager/alarm.rs @@ -543,7 +543,7 @@ async fn build_template( // Validate recipient let rcpt_to = if let Some(rcpt_to) = rcpt_to { if server.core.groupware.alarms_allow_external_recipients - || account_info.addresses().any(|email| email == &rcpt_to) + || account_info.addresses().any(|email| email == rcpt_to) { rcpt_to } else { diff --git a/crates/services/src/task_manager/imip.rs b/crates/services/src/task_manager/imip.rs index ff2c93af..80476775 100644 --- a/crates/services/src/task_manager/imip.rs +++ b/crates/services/src/task_manager/imip.rs @@ -168,10 +168,7 @@ async fn send_imip( // Build message let message = MessageBuilder::new() .from(( - account_info - .description() - .as_deref() - .unwrap_or(account_info.name()), + account_info.description().unwrap_or(account_info.name()), itip_message.from.as_str(), )) .to(recipient.as_str()) diff --git a/crates/services/src/task_manager/index.rs b/crates/services/src/task_manager/index.rs index 709ff4e2..1e01c728 100644 --- a/crates/services/src/task_manager/index.rs +++ b/crates/services/src/task_manager/index.rs @@ -574,7 +574,7 @@ async fn build_tracing_span_document( }; let span_id = ((account_id as u64) << 32) | document_id as u64; - let span = store.store.get_span(span_id).await?; + let span = server.tracing_store().get_span(span_id).await?; if !span.is_empty() { Ok(Some(build_span_document(span_id, span, index_fields))) @@ -630,11 +630,11 @@ async fn delete_email_metadata( use common::enterprise::undelete::DeletedItemType; use email::message::metadata::ArchivedMetadataHeaderName; - if let Some(undelete) = server + if let Some(undelete_retention) = server .core .enterprise .as_ref() - .and_then(|e| e.undelete.as_ref()) + .and_then(|e| e.undelete_retention.as_ref()) { use common::enterprise::undelete::DeletedItem; use email::message::metadata::MESSAGE_RECEIVED_MASK; @@ -668,7 +668,7 @@ async fn delete_email_metadata( } }); let now = now(); - let until = now + undelete.retention.as_secs(); + let until = now + undelete_retention.as_secs(); let blob_hash = BlobHash::from(&metadata.blob_hash); batch .set( diff --git a/crates/smtp/src/inbound/hooks/message.rs b/crates/smtp/src/inbound/hooks/message.rs index 695c39f7..3df9b89c 100644 --- a/crates/smtp/src/inbound/hooks/message.rs +++ b/crates/smtp/src/inbound/hooks/message.rs @@ -4,18 +4,7 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ -use std::time::Instant; - -use ahash::AHashMap; -use common::{ - DAEMON_NAME, - config::smtp::session::{MTAHook, Stage}, - network::SessionStream, -}; - -use mail_auth::AuthenticatedMessage; -use trc::MtaHookEvent; - +use super::{Action, Queue, Response, client::send_mta_hook_request}; use crate::{ core::Session, inbound::{ @@ -27,8 +16,15 @@ use crate::{ }, queue::QueueId, }; - -use super::{Action, Queue, Response, client::send_mta_hook_request}; +use ahash::AHashMap; +use common::{ + DAEMON_NAME, + config::smtp::session::{MTAHook, Stage}, + network::SessionStream, +}; +use mail_auth::AuthenticatedMessage; +use std::time::Instant; +use trc::MtaHookEvent; impl Session { pub async fn run_mta_hooks( @@ -65,7 +61,7 @@ impl Session { Action::Quarantine => MtaHookEvent::ActionQuarantine, }), SpanId = self.data.session_id, - Id = mta_hook.id.clone(), + Id = mta_hook.id.to_string(), Elapsed = time.elapsed(), ); @@ -156,7 +152,7 @@ impl Session { trc::event!( MtaHook(MtaHookEvent::Error), SpanId = self.data.session_id, - Id = mta_hook.id.clone(), + Id = mta_hook.id.to_string(), Reason = err, Elapsed = time.elapsed(), ); diff --git a/crates/smtp/src/inbound/milter/client.rs b/crates/smtp/src/inbound/milter/client.rs index 83d70a4a..fb86cda5 100644 --- a/crates/smtp/src/inbound/milter/client.rs +++ b/crates/smtp/src/inbound/milter/client.rs @@ -49,7 +49,7 @@ impl MilterClient { | SMFIF_ADDRCPT_PAR, ), flags_protocol: config.flags_protocol.unwrap_or(0x42), - id: config.id.clone(), + id: config.id, }); } Err(err) => { diff --git a/crates/smtp/src/inbound/milter/mod.rs b/crates/smtp/src/inbound/milter/mod.rs index e687b217..7bfe9eae 100644 --- a/crates/smtp/src/inbound/milter/mod.rs +++ b/crates/smtp/src/inbound/milter/mod.rs @@ -4,14 +4,12 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ -use std::{borrow::Cow, fmt::Display, net::IpAddr, sync::Arc, time::Duration}; - -use common::config::smtp::session::MilterVersion; - -use serde::{Deserialize, Serialize}; -use tokio::io::{AsyncRead, AsyncWrite}; - use self::receiver::Receiver; +use common::config::smtp::session::MilterVersion; +use registry::types::id::Id; +use serde::{Deserialize, Serialize}; +use std::{borrow::Cow, fmt::Display, net::IpAddr, time::Duration}; +use tokio::io::{AsyncRead, AsyncWrite}; pub mod client; pub mod macros; @@ -30,7 +28,7 @@ pub struct MilterClient { options: u32, flags_actions: u32, flags_protocol: u32, - id: Arc, + id: Id, session_id: u64, } diff --git a/crates/smtp/src/queue/spool.rs b/crates/smtp/src/queue/spool.rs index 2423e379..fa42a96e 100644 --- a/crates/smtp/src/queue/spool.rs +++ b/crates/smtp/src/queue/spool.rs @@ -377,7 +377,7 @@ impl MessageWrapper { .put_blob( self.message.blob_hash.as_slice(), message.as_ref(), - server.core.storage.compression, + server.core.email.compression, ) .await { diff --git a/crates/spam-filter/src/modules/classifier.rs b/crates/spam-filter/src/modules/classifier.rs index c6d94e76..e0e09934 100644 --- a/crates/spam-filter/src/modules/classifier.rs +++ b/crates/spam-filter/src/modules/classifier.rs @@ -506,7 +506,7 @@ impl SpamClassifier for Server { &Archiver::new(trainer) .serialize() .caused_by(trc::location!())?, - self.core.storage.compression, + self.core.email.compression, ) .await .caused_by(trc::location!())?; @@ -514,7 +514,7 @@ impl SpamClassifier for Server { .put_blob( SPAM_CLASSIFIER_KEY, &classifier.serialize().caused_by(trc::location!())?, - self.core.storage.compression, + self.core.email.compression, ) .await .caused_by(trc::location!())?; diff --git a/crates/store/src/backend/composite/read_replica.rs b/crates/store/src/backend/composite/read_replica.rs index 0f14bd9d..12c72d93 100644 --- a/crates/store/src/backend/composite/read_replica.rs +++ b/crates/store/src/backend/composite/read_replica.rs @@ -238,8 +238,4 @@ impl SQLReadReplica { pub fn primary_store(&self) -> &Store { &self.primary } - - pub fn into_primary(self) -> Store { - self.primary - } } diff --git a/crates/store/src/backend/http/config.rs b/crates/store/src/backend/http/config.rs index 6182c75a..a93e634f 100644 --- a/crates/store/src/backend/http/config.rs +++ b/crates/store/src/backend/http/config.rs @@ -45,7 +45,7 @@ impl LookupStores { id: http.namespace, }; - match self.stores.entry(http_config.id.clone()) { + match self.stores.entry(http_config.id.as_str().into()) { Entry::Vacant(entry) => { let store = HttpStore { entries: ArcSwap::from_pointee(AHashMap::new()), diff --git a/crates/store/src/backend/memory/mod.rs b/crates/store/src/backend/memory/mod.rs index 05e1f9c0..5b831e75 100644 --- a/crates/store/src/backend/memory/mod.rs +++ b/crates/store/src/backend/memory/mod.rs @@ -56,8 +56,10 @@ impl LookupStores { } for (namespace, store) in lookups { - self.stores - .insert(namespace, InMemoryStore::Static(store.into())); + self.stores.insert( + namespace.into_boxed_str(), + InMemoryStore::Static(store.into()), + ); } } } diff --git a/crates/store/src/bootstrap/data.rs b/crates/store/src/bootstrap/data.rs deleted file mode 100644 index a13a023f..00000000 --- a/crates/store/src/bootstrap/data.rs +++ /dev/null @@ -1,31 +0,0 @@ -/* - * SPDX-FileCopyrightText: 2020 Stalwart Labs LLC - * - * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL - */ - -use crate::Store; -use registry::schema::structs::DataStore; - -impl Store { - pub async fn build(config: DataStore) -> Result { - #[allow(unreachable_patterns)] - match config { - #[cfg(feature = "rocks")] - DataStore::RocksDb(store) => crate::backend::rocksdb::RocksDbStore::open(store).await, - #[cfg(feature = "foundation")] - DataStore::FoundationDb(store) => { - crate::backend::foundationdb::FdbStore::open(store).await - } - #[cfg(feature = "postgres")] - DataStore::PostgreSql(store) => { - crate::backend::postgres::PostgresStore::open(store).await - } - #[cfg(feature = "mysql")] - DataStore::MySql(store) => crate::backend::mysql::MysqlStore::open(store).await, - #[cfg(feature = "sqlite")] - DataStore::Sqlite(store) => crate::backend::sqlite::SqliteStore::open(store), - _ => Err("Binary was not compiled with the selected data store backend".to_string()), - } - } -} diff --git a/crates/store/src/bootstrap/blob.rs b/crates/store/src/build/blob.rs similarity index 83% rename from crates/store/src/bootstrap/blob.rs rename to crates/store/src/build/blob.rs index 9a68c788..83ec83ef 100644 --- a/crates/store/src/bootstrap/blob.rs +++ b/crates/store/src/build/blob.rs @@ -57,4 +57,21 @@ impl BlobStore { } } } + + // SPDX-SnippetBegin + // SPDX-FileCopyrightText: 2020 Stalwart Labs LLC + // SPDX-License-Identifier: LicenseRef-SEL + #[cfg(feature = "enterprise")] + pub fn downgrade_store(self) -> BlobStore { + match self { + BlobStore::Sharded(_) => BlobStore::default(), + other => other, + } + } + + #[cfg(feature = "enterprise")] + pub fn is_enterprise(&self) -> bool { + matches!(self, BlobStore::Sharded(_)) + } + // SPDX-SnippetEnd } diff --git a/crates/store/src/build/data.rs b/crates/store/src/build/data.rs new file mode 100644 index 00000000..d3df7ee8 --- /dev/null +++ b/crates/store/src/build/data.rs @@ -0,0 +1,94 @@ +/* + * SPDX-FileCopyrightText: 2020 Stalwart Labs LLC + * + * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL + */ + +use crate::{Store, registry::bootstrap::Bootstrap}; +use registry::schema::{ + prelude::Object, + structs::{DataStore, MetricsStore, TracingStore}, +}; + +impl Store { + pub async fn build(config: DataStore) -> Result { + #[allow(unreachable_patterns)] + match config { + #[cfg(feature = "rocks")] + DataStore::RocksDb(store) => crate::backend::rocksdb::RocksDbStore::open(store).await, + #[cfg(feature = "foundation")] + DataStore::FoundationDb(store) => { + crate::backend::foundationdb::FdbStore::open(store).await + } + #[cfg(feature = "postgres")] + DataStore::PostgreSql(store) => { + crate::backend::postgres::PostgresStore::open(store).await + } + #[cfg(feature = "mysql")] + DataStore::MySql(store) => crate::backend::mysql::MysqlStore::open(store).await, + #[cfg(feature = "sqlite")] + DataStore::Sqlite(store) => crate::backend::sqlite::SqliteStore::open(store), + _ => Err("Binary was not compiled with the selected data store backend".to_string()), + } + } + + pub async fn build_tracing(bp: &mut Bootstrap) -> Option { + let result = match bp.setting_infallible::().await { + TracingStore::Disabled => Ok(None), + TracingStore::Default => Ok(Some(bp.data_store.clone())), + #[cfg(feature = "foundation")] + TracingStore::FoundationDb(store) => { + crate::backend::foundationdb::FdbStore::open(store) + .await + .map(Some) + } + #[cfg(feature = "postgres")] + TracingStore::PostgreSql(store) => crate::backend::postgres::PostgresStore::open(store) + .await + .map(Some), + #[cfg(feature = "mysql")] + TracingStore::MySql(store) => crate::backend::mysql::MysqlStore::open(store) + .await + .map(Some), + _ => Err("Binary was not compiled with the selected tracing store backend".to_string()), + }; + + match result { + Ok(store) => store, + Err(err) => { + bp.build_warning(Object::TracingStore.singleton(), err); + None + } + } + } + + pub async fn build_metrics(bp: &mut Bootstrap) -> Option { + let result = match bp.setting_infallible::().await { + MetricsStore::Disabled => Ok(None), + MetricsStore::Default => Ok(Some(bp.data_store.clone())), + #[cfg(feature = "foundation")] + MetricsStore::FoundationDb(store) => { + crate::backend::foundationdb::FdbStore::open(store) + .await + .map(Some) + } + #[cfg(feature = "postgres")] + MetricsStore::PostgreSql(store) => crate::backend::postgres::PostgresStore::open(store) + .await + .map(Some), + #[cfg(feature = "mysql")] + MetricsStore::MySql(store) => crate::backend::mysql::MysqlStore::open(store) + .await + .map(Some), + _ => Err("Binary was not compiled with the selected metrics store backend".to_string()), + }; + + match result { + Ok(store) => store, + Err(err) => { + bp.build_warning(Object::MetricsStore.singleton(), err); + None + } + } + } +} diff --git a/crates/store/src/bootstrap/lookup.rs b/crates/store/src/build/lookup.rs similarity index 88% rename from crates/store/src/bootstrap/lookup.rs rename to crates/store/src/build/lookup.rs index 72ebef6a..8a8da41f 100644 --- a/crates/store/src/bootstrap/lookup.rs +++ b/crates/store/src/build/lookup.rs @@ -8,8 +8,16 @@ use crate::{InMemoryStore, LookupStores, registry::bootstrap::Bootstrap}; use registry::schema::structs::{LookupStore, StoreLookup}; use std::collections::hash_map::Entry; -#[allow(unreachable_patterns)] impl LookupStores { + pub async fn build(bp: &mut Bootstrap) -> Self { + let mut stores = LookupStores::default(); + stores.parse_stores(bp).await; + stores.parse_static(bp).await; + stores.parse_http(bp).await; + stores + } + + #[allow(unreachable_patterns)] pub async fn parse_stores(&mut self, bp: &mut Bootstrap) { for store in bp.list_infallible::().await { let id = store.id; @@ -53,7 +61,7 @@ impl LookupStores { }; match result { - Ok(lookup) => match self.stores.entry(store.namespace.clone()) { + Ok(lookup) => match self.stores.entry(store.namespace.as_str().into()) { Entry::Vacant(entry) => { entry.insert(lookup); } @@ -61,7 +69,7 @@ impl LookupStores { bp.build_error( id, format!( - "An lookup store with the {} namespace already exists", + "A lookup store with the {} namespace already exists", store.namespace ), ); diff --git a/crates/store/src/bootstrap/memory.rs b/crates/store/src/build/memory.rs similarity index 76% rename from crates/store/src/bootstrap/memory.rs rename to crates/store/src/build/memory.rs index e7c6c4d4..249e3469 100644 --- a/crates/store/src/bootstrap/memory.rs +++ b/crates/store/src/build/memory.rs @@ -41,4 +41,21 @@ impl InMemoryStore { } } } + + // SPDX-SnippetBegin + // SPDX-FileCopyrightText: 2020 Stalwart Labs LLC + // SPDX-License-Identifier: LicenseRef-SEL + #[cfg(feature = "enterprise")] + pub fn downgrade_store(self) -> InMemoryStore { + match self { + InMemoryStore::Sharded(_) => InMemoryStore::default(), + other => other, + } + } + + #[cfg(feature = "enterprise")] + pub fn is_enterprise(&self) -> bool { + matches!(self, InMemoryStore::Sharded(_)) + } + // SPDX-SnippetEnd } diff --git a/crates/store/src/bootstrap/mod.rs b/crates/store/src/build/mod.rs similarity index 92% rename from crates/store/src/bootstrap/mod.rs rename to crates/store/src/build/mod.rs index 80cda2a0..e150c7c5 100644 --- a/crates/store/src/bootstrap/mod.rs +++ b/crates/store/src/build/mod.rs @@ -8,4 +8,5 @@ pub mod blob; pub mod data; pub mod lookup; pub mod memory; +pub mod registry; pub mod search; diff --git a/crates/store/src/build/registry.rs b/crates/store/src/build/registry.rs new file mode 100644 index 00000000..20198f13 --- /dev/null +++ b/crates/store/src/build/registry.rs @@ -0,0 +1,29 @@ +/* + * SPDX-FileCopyrightText: 2020 Stalwart Labs LLC + * + * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL + */ + +use crate::RegistryStore; +use std::path::PathBuf; + +impl RegistryStore { + pub fn init(local: PathBuf) -> Self { + let todo = "environment variables and reading from files"; + + /* + + match std::fs::read_to_string(&cfg_local_path) { + Ok(value) => { + config.parse(&value).failed("Invalid local registry file"); + } + Err(err) => { + config.new_build_error("*", format!("Could not read registry file: {err}")); + } + } + + */ + + todo!() + } +} diff --git a/crates/store/src/bootstrap/search.rs b/crates/store/src/build/search.rs similarity index 100% rename from crates/store/src/bootstrap/search.rs rename to crates/store/src/build/search.rs diff --git a/crates/store/src/dispatch/store.rs b/crates/store/src/dispatch/store.rs index 34aa8401..0310219d 100644 --- a/crates/store/src/dispatch/store.rs +++ b/crates/store/src/dispatch/store.rs @@ -469,7 +469,7 @@ impl Store { // SPDX-SnippetBegin // SPDX-FileCopyrightText: 2020 Stalwart Labs LLC // SPDX-License-Identifier: LicenseRef-SEL - #[cfg(feature = "enterprise")] + #[cfg(all(feature = "enterprise", any(feature = "postgres", feature = "mysql")))] Store::SQLReadReplica(store) => Box::pin(store.primary_store().create_tables()).await, // SPDX-SnippetEnd _ => Ok(()), diff --git a/crates/store/src/lib.rs b/crates/store/src/lib.rs index 8dabed42..ed1c0d2f 100644 --- a/crates/store/src/lib.rs +++ b/crates/store/src/lib.rs @@ -5,7 +5,7 @@ */ pub mod backend; -pub mod bootstrap; +pub mod build; pub mod dispatch; pub mod query; pub mod registry; @@ -126,7 +126,7 @@ pub struct IterateParams { #[derive(Clone, Default)] pub struct LookupStores { - pub stores: AHashMap, + pub stores: AHashMap, InMemoryStore>, } #[derive(Clone, Default)] @@ -261,6 +261,12 @@ impl From for InMemoryStore { } } +impl Default for BlobStore { + fn default() -> Self { + Self::Store(Store::None) + } +} + impl Default for InMemoryStore { fn default() -> Self { Self::Store(Store::None) @@ -603,6 +609,11 @@ impl Store { matches!(self, Self::None) } + #[inline(always)] + pub fn is_active(&self) -> bool { + !matches!(self, Self::None) + } + #[inline(always)] pub fn is_sql(&self) -> bool { match self { @@ -646,7 +657,16 @@ impl Store { // SPDX-FileCopyrightText: 2020 Stalwart Labs LLC // SPDX-License-Identifier: LicenseRef-SEL #[cfg(feature = "enterprise")] - pub fn is_enterprise_store(&self) -> bool { + pub fn downgrade_store(self) -> Self { + match self { + #[cfg(any(feature = "postgres", feature = "mysql"))] + Store::SQLReadReplica(store) => store.primary_store().clone(), + other => other, + } + } + + #[cfg(feature = "enterprise")] + pub fn is_enterprise(&self) -> bool { match self { #[cfg(any(feature = "postgres", feature = "mysql"))] Store::SQLReadReplica(_) => true, @@ -654,11 +674,6 @@ impl Store { } } // SPDX-SnippetEnd - - #[cfg(not(feature = "enterprise"))] - pub fn is_enterprise_store(&self) -> bool { - false - } } impl std::fmt::Debug for Store { diff --git a/crates/store/src/registry/bootstrap.rs b/crates/store/src/registry/bootstrap.rs index 8ab9d701..958be310 100644 --- a/crates/store/src/registry/bootstrap.rs +++ b/crates/store/src/registry/bootstrap.rs @@ -174,4 +174,12 @@ impl Bootstrap { pub fn hostname(&self) -> &str { &self.node.hostname } + + pub fn log_errors(&self) { + let todo = "implement"; + } + + pub fn log_warnings(&self) { + let todo = "implement"; + } } diff --git a/crates/store/src/registry/mod.rs b/crates/store/src/registry/mod.rs index e3993dfb..75a83e4c 100644 --- a/crates/store/src/registry/mod.rs +++ b/crates/store/src/registry/mod.rs @@ -40,6 +40,7 @@ pub enum RegistryFilterOp { pub enum RegistryFilterValue { String(String), - Integer(u64), + U64(u64), + U16(u16), Boolean(bool), } diff --git a/crates/store/src/registry/query.rs b/crates/store/src/registry/query.rs index 9b2c35e9..4ff3940d 100644 --- a/crates/store/src/registry/query.rs +++ b/crates/store/src/registry/query.rs @@ -8,7 +8,10 @@ use crate::{ RegistryStore, registry::{RegistryFilter, RegistryFilterOp, RegistryFilterValue, RegistryQuery}, }; -use registry::schema::prelude::{Object, Property}; +use registry::{ + schema::prelude::{Object, Property}, + types::EnumType, +}; use roaring::RoaringBitmap; impl RegistryStore { @@ -193,12 +196,18 @@ impl From<&str> for RegistryFilterValue { impl From for RegistryFilterValue { fn from(value: u64) -> Self { - RegistryFilterValue::Integer(value) + RegistryFilterValue::U64(value) } } impl From for RegistryFilterValue { fn from(value: u32) -> Self { - RegistryFilterValue::Integer(value as u64) + RegistryFilterValue::U64(value as u64) + } +} + +impl From for RegistryFilterValue { + fn from(value: u16) -> Self { + RegistryFilterValue::U16(value) } } diff --git a/tests/Cargo.toml b/tests/Cargo.toml index 6e824d9b..a40da9d7 100644 --- a/tests/Cargo.toml +++ b/tests/Cargo.toml @@ -4,9 +4,9 @@ version = "0.15.5" edition = "2024" [features] -default = ["sqlite", "postgres", "mysql", "rocks", "s3", "redis", "nats", "azure", "foundationdb"] +#default = ["sqlite", "postgres", "mysql", "rocks", "s3", "redis", "nats", "azure", "foundationdb"] #default = ["sqlite", "postgres", "mysql", "rocks", "s3", "redis", "foundationdb"] -#default = ["rocks"] +default = ["postgres"] sqlite = ["store/sqlite", "directory/sqlite"] foundationdb = ["store/foundation", "common/foundation"] postgres = ["store/postgres", "directory/postgres"] diff --git a/tests/src/jmap/server/webhooks.rs b/tests/src/jmap/server/webhooks.rs index c842fa41..e08d8e09 100644 --- a/tests/src/jmap/server/webhooks.rs +++ b/tests/src/jmap/server/webhooks.rs @@ -14,7 +14,7 @@ use std::{ use crate::jmap::JMAPTest; use base64::{Engine, engine::general_purpose::STANDARD}; -use common::manager::webadmin::Resource; +use common::manager::application::Resource; use http_proto::{ToHttpResponse, request::fetch_body}; use hyper::{body, server::conn::http1, service::service_fn}; use hyper_util::rt::TokioIo; diff --git a/tests/src/smtp/inbound/milter.rs b/tests/src/smtp/inbound/milter.rs index be3ab0c0..f364a4c7 100644 --- a/tests/src/smtp/inbound/milter.rs +++ b/tests/src/smtp/inbound/milter.rs @@ -11,7 +11,7 @@ use common::{ Core, config::smtp::session::{Milter, MilterVersion, Stage}, expr::if_block::IfBlock, - manager::webadmin::Resource, + manager::application::Resource, }; use http_proto::{ToHttpResponse, request::fetch_body};