diff --git a/crates/common/src/auth/authentication.rs b/crates/common/src/auth/authentication.rs index f028477b..24cd4c30 100644 --- a/crates/common/src/auth/authentication.rs +++ b/crates/common/src/auth/authentication.rs @@ -207,7 +207,7 @@ impl Server { }?; // Enforce alias login restrictions - if is_alias_login && !token.has_permission(Permission::AuthenticateAlias) { + if is_alias_login && !token.has_permission(Permission::AuthenticateWithAlias) { return Err(trc::AuthEvent::Failed .into_err() .ctx(trc::Key::AccountName, auth_as_address.to_string()) diff --git a/crates/common/src/config/mailstore/jmap.rs b/crates/common/src/config/mailstore/jmap.rs index 2592df54..098656e2 100644 --- a/crates/common/src/config/mailstore/jmap.rs +++ b/crates/common/src/config/mailstore/jmap.rs @@ -40,6 +40,7 @@ pub struct JmapConfig { pub push_timeout: Duration, pub push_verify_timeout: Duration, pub push_throttle: Duration, + pub push_total_shards: u32, pub web_socket_throttle: Duration, pub web_socket_timeout: Duration, @@ -79,6 +80,7 @@ impl JmapConfig { push_timeout: jmap.push_request_timeout.into_inner(), push_verify_timeout: jmap.push_verify_timeout.into_inner(), push_throttle: jmap.push_throttle.into_inner(), + push_total_shards: jmap.push_shards_total as u32, capabilities: BaseCapabilities::default(), }; diff --git a/crates/common/src/config/network.rs b/crates/common/src/config/network.rs index 900bdff9..14efc137 100644 --- a/crates/common/src/config/network.rs +++ b/crates/common/src/config/network.rs @@ -10,12 +10,11 @@ use crate::{ network::security::Security, }; use registry::schema::{ - enums::{ClusterShardedTaskType, ClusterTaskType}, + enums::ClusterTaskType, prelude::ObjectType, structs::{self, Asn, ClusterTaskGroup, HttpForm, Rate, SystemSettings, TaskManager}, }; -use std::{hash::Hasher, str::FromStr, time::Duration}; -use xxhash_rust::xxh3::Xxh3Builder; +use std::{str::FromStr, time::Duration}; #[derive(Clone)] pub struct Network { @@ -51,33 +50,18 @@ pub struct ContactForm { pub field_honey_pot: Option, } -#[derive(Clone, Default)] +#[derive(Clone)] pub struct ClusterRoles { - pub store_maintenance: ClusterRole, - pub account_maintenance: ClusterRole, - pub push_notifications: ClusterRole, - pub search_indexing: ClusterRole, - pub spam_training: ClusterRole, - pub imip_processing: ClusterRole, - pub merge_threads: ClusterRole, - pub calendar_alerts: ClusterRole, - pub dns_acme: ClusterRole, - pub calculate_metrics: ClusterRole, - pub push_metrics: ClusterRole, - pub outbound_mta: ClusterRole, - pub task_scheduler: ClusterRole, - pub task_manager: ClusterRole, -} - -#[derive(Clone, Copy, Default)] -pub enum ClusterRole { - #[default] - Enabled, - Disabled, - Sharded { - shard_id: u32, - total_shards: u32, - }, + pub store_maintenance: bool, + pub account_maintenance: bool, + pub push_notifications: bool, + pub search_indexing: bool, + pub spam_training: bool, + pub metrics_calculate: bool, + pub metrics_push: bool, + pub outbound_mta: bool, + pub task_scheduler: bool, + pub task_manager: bool, } #[derive(Clone, Default)] @@ -164,140 +148,20 @@ impl Network { ClusterTaskGroup::EnableAll => {} ClusterTaskGroup::DisableAll => { for network_role in network.roles.all_mut() { - network_role.set_role(false); + *network_role = false; } } ClusterTaskGroup::EnableSome(group) => { for network_role in network.roles.all_mut() { - network_role.set_role(false); + *network_role = false; } for task_type in group.task_types.iter() { - match task_type { - ClusterTaskType::StoreMaintenance => { - network.roles.store_maintenance.set_role(true); - } - ClusterTaskType::AccountMaintenance => { - network.roles.account_maintenance.set_role(true); - } - ClusterTaskType::DnsAndAcme => { - network.roles.dns_acme.set_role(true); - } - ClusterTaskType::CalculateMetrics => { - network.roles.calculate_metrics.set_role(true); - } - ClusterTaskType::PushMetrics => { - network.roles.push_metrics.set_role(true); - } - ClusterTaskType::PushNotifications => { - network.roles.push_notifications.set_role(true); - } - ClusterTaskType::SearchIndexing => { - network.roles.search_indexing.set_role(true); - } - ClusterTaskType::SpamClassifierTraining => { - network.roles.spam_training.set_role(true); - } - ClusterTaskType::ImipProcessing => { - network.roles.imip_processing.set_role(true); - } - ClusterTaskType::CalendarAlerts => { - network.roles.calendar_alerts.set_role(true); - } - ClusterTaskType::MergeThreads => { - network.roles.merge_threads.set_role(true); - } - ClusterTaskType::OutboundMta => { - network.roles.outbound_mta.set_role(true); - } - ClusterTaskType::TaskQueueProcessing => { - network.roles.task_manager.set_role(true); - } - ClusterTaskType::TaskScheduler => { - network.roles.task_scheduler.set_role(true); - } - } + network.roles.set_role(*task_type, true); } } ClusterTaskGroup::DisableSome(group) => { for task_type in group.task_types.iter() { - match task_type { - ClusterTaskType::StoreMaintenance => { - network.roles.store_maintenance.set_role(true); - } - ClusterTaskType::AccountMaintenance => { - network.roles.account_maintenance.set_role(false); - } - ClusterTaskType::DnsAndAcme => { - network.roles.dns_acme.set_role(false); - } - ClusterTaskType::CalculateMetrics => { - network.roles.calculate_metrics.set_role(false); - } - ClusterTaskType::PushMetrics => { - network.roles.push_metrics.set_role(false); - } - ClusterTaskType::PushNotifications => { - network.roles.push_notifications.set_role(false); - } - ClusterTaskType::SearchIndexing => { - network.roles.search_indexing.set_role(false); - } - ClusterTaskType::SpamClassifierTraining => { - network.roles.spam_training.set_role(false); - } - ClusterTaskType::ImipProcessing => { - network.roles.imip_processing.set_role(false); - } - ClusterTaskType::CalendarAlerts => { - network.roles.calendar_alerts.set_role(false); - } - ClusterTaskType::MergeThreads => { - network.roles.merge_threads.set_role(false); - } - ClusterTaskType::OutboundMta => { - network.roles.outbound_mta.set_role(false); - } - ClusterTaskType::TaskQueueProcessing => { - network.roles.task_manager.set_role(false); - } - ClusterTaskType::TaskScheduler => { - network.roles.task_scheduler.set_role(false); - } - } - } - } - } - - if role.shard_size > 1 { - for task_type in role.shard_task_types.iter() { - let network_role = match task_type { - ClusterShardedTaskType::StoreMaintenance => { - &mut network.roles.store_maintenance - } - ClusterShardedTaskType::AccountMaintenance => { - &mut network.roles.account_maintenance - } - ClusterShardedTaskType::DnsAndAcme => &mut network.roles.dns_acme, - ClusterShardedTaskType::PushNotifications => { - &mut network.roles.push_notifications - } - ClusterShardedTaskType::SearchIndexing => { - &mut network.roles.search_indexing - } - ClusterShardedTaskType::ImipProcessing => { - &mut network.roles.imip_processing - } - ClusterShardedTaskType::CalendarAlerts => { - &mut network.roles.calendar_alerts - } - ClusterShardedTaskType::MergeThreads => &mut network.roles.merge_threads, - }; - - if network_role.is_enabled_or_sharded() { - *network_role = ClusterRole::Sharded { - shard_id: bp.registry.cluster_role_shard() as u32, - total_shards: role.shard_size as u32, - }; + network.roles.set_role(*task_type, false); } } } @@ -408,104 +272,52 @@ impl AsnGeoLookupConfig { } } -impl ClusterRole { - pub fn is_enabled_or_sharded(&self) -> bool { - debug_assert!(!self.is_uninit() && !self.is_seen_role()); - matches!(self, ClusterRole::Enabled | ClusterRole::Sharded { .. }) - } - - pub fn is_enabled_for_integer(&self, value: u64) -> bool { - debug_assert!(!self.is_uninit() && !self.is_seen_role()); - match self { - ClusterRole::Enabled => true, - ClusterRole::Disabled => false, - ClusterRole::Sharded { - shard_id, - total_shards, - } => (value as u32 % total_shards) == *shard_id, - } - } - - pub fn is_enabled_for_hash(&self, item: &impl std::hash::Hash) -> bool { - debug_assert!(!self.is_uninit() && !self.is_seen_role()); - match self { - ClusterRole::Enabled => true, - ClusterRole::Disabled => false, - ClusterRole::Sharded { - shard_id, - total_shards, - } => { - let mut hasher = Xxh3Builder::new().with_seed(201179).build(); - item.hash(&mut hasher); - hasher.finish() % (*total_shards as u64) == *shard_id as u64 - } - } - } - - fn set_uninit(&mut self) { - *self = ClusterRole::Sharded { - shard_id: u32::MAX, - total_shards: u32::MAX, - }; - } - - fn set_role(&mut self, is_member: bool) -> bool { - if self.is_uninit() { - if is_member { - *self = ClusterRole::Enabled; - } else { - *self = ClusterRole::Sharded { - shard_id: u32::MAX, - total_shards: 0, - }; - } - true - } else { - false - } - } - - fn is_seen_role(&self) -> bool { - matches!(self, ClusterRole::Sharded { - shard_id, - total_shards, - } if *shard_id == u32::MAX && *total_shards == 0) - } - - fn is_uninit(&self) -> bool { - matches!(self, ClusterRole::Sharded { - shard_id, - total_shards, - } if *shard_id == u32::MAX && *total_shards == u32::MAX) - } - - fn finalize(&mut self) { - if self.is_uninit() { - *self = ClusterRole::Enabled; - } else if self.is_seen_role() { - *self = ClusterRole::Disabled; - } - } -} - impl ClusterRoles { - fn all_mut(&mut self) -> impl Iterator { + fn all_mut(&mut self) -> impl Iterator { [ &mut self.store_maintenance, &mut self.account_maintenance, &mut self.push_notifications, &mut self.search_indexing, &mut self.spam_training, - &mut self.imip_processing, - &mut self.merge_threads, - &mut self.calendar_alerts, - &mut self.dns_acme, &mut self.outbound_mta, - &mut self.calculate_metrics, - &mut self.push_metrics, &mut self.task_manager, &mut self.task_scheduler, + &mut self.metrics_calculate, + &mut self.metrics_push, ] .into_iter() } + + fn set_role(&mut self, role: ClusterTaskType, enabled: bool) { + match role { + ClusterTaskType::StoreMaintenance => self.store_maintenance = enabled, + ClusterTaskType::AccountMaintenance => self.account_maintenance = enabled, + ClusterTaskType::PushNotifications => self.push_notifications = enabled, + ClusterTaskType::SearchIndexing => self.search_indexing = enabled, + ClusterTaskType::SpamClassifierTraining => self.spam_training = enabled, + ClusterTaskType::MetricsCalculate => self.metrics_calculate = enabled, + ClusterTaskType::MetricsPush => self.metrics_push = enabled, + ClusterTaskType::OutboundMta => self.outbound_mta = enabled, + ClusterTaskType::TaskQueueProcessing => self.task_manager = enabled, + ClusterTaskType::TaskScheduler => self.task_scheduler = enabled, + } + } +} + +impl Default for ClusterRoles { + fn default() -> Self { + ClusterRoles { + store_maintenance: true, + account_maintenance: true, + push_notifications: true, + search_indexing: true, + spam_training: true, + metrics_calculate: true, + metrics_push: true, + outbound_mta: true, + task_manager: true, + task_scheduler: true, + } + } } diff --git a/crates/common/src/config/server/tls.rs b/crates/common/src/config/server/tls.rs index 9bee9bdd..a01d2137 100644 --- a/crates/common/src/config/server/tls.rs +++ b/crates/common/src/config/server/tls.rs @@ -14,7 +14,7 @@ use hickory_proto::rr::dnssec::KeyPair; use rcgen::generate_simple_self_signed; use registry::schema::{ enums, - structs::{self, Certificate, DnsServer}, + structs::{self, Certificate, DnsServer, SystemSettings}, }; use ring::signature::{EcdsaKeyPair, Ed25519KeyPair}; use rustls::{ @@ -268,6 +268,8 @@ pub(crate) async fn parse_certificates( certificates: &mut AHashMap, Arc>, subject_names: &mut AHashSet>, ) { + let system = bp.setting_infallible::().await; + // Parse certificates for cert_obj in bp.list_infallible::().await { let secret = match cert_obj.object.private_key.secret().await { @@ -347,7 +349,11 @@ pub(crate) async fn parse_certificates( } // Add default certificate - if cert_obj.object.default { + if system + .default_certificate_id + .as_ref() + .is_some_and(|id| *id == cert_obj.id.id()) + { certificates.insert("*".into(), cert.clone()); } } diff --git a/crates/common/src/expr/if_block.rs b/crates/common/src/expr/if_block.rs index 8e70cd12..c5218b46 100644 --- a/crates/common/src/expr/if_block.rs +++ b/crates/common/src/expr/if_block.rs @@ -81,7 +81,7 @@ impl Expression { } } -pub(crate) trait BootstrapExprExt { +pub trait BootstrapExprExt { fn compile_expr(&mut self, id: ObjectId, expr_ctx: &ExpressionContext<'_>) -> IfBlock; fn compile_default_expr(&mut self, id: ObjectId, expr_ctx: &ExpressionContext<'_>) -> IfBlock; fn try_compile_expr( diff --git a/crates/common/src/manager/boot.rs b/crates/common/src/manager/boot.rs index a9fbd687..380715e0 100644 --- a/crates/common/src/manager/boot.rs +++ b/crates/common/src/manager/boot.rs @@ -284,7 +284,7 @@ impl BootManager { .cluster_role() .unwrap_or("[default]") .to_string(), - Details = bootstrap.registry.cluster_role_shard() + Details = bootstrap.registry.cluster_push_shard() ); } diff --git a/crates/common/src/scripts/plugins/llm_prompt.rs b/crates/common/src/scripts/plugins/llm_prompt.rs index 4685e0ad..3e0b0fe6 100644 --- a/crates/common/src/scripts/plugins/llm_prompt.rs +++ b/crates/common/src/scripts/plugins/llm_prompt.rs @@ -36,7 +36,7 @@ pub async fn exec(ctx: PluginContext<'_>) -> trc::Result { if ctx.access_token.is_none_or(|token| { use registry::schema::enums::Permission; - if token.has_permission(Permission::AiModelInteract) { + if token.has_permission(Permission::InteractAi) { true } else { use registry::types::EnumImpl; @@ -44,7 +44,7 @@ pub async fn exec(ctx: PluginContext<'_>) -> trc::Result { trc::event!( Security(SecurityEvent::Unauthorized), AccountId = token.account_id(), - Details = Permission::AiModelInteract.as_str(), + Details = Permission::InteractAi.as_str(), SpanId = ctx.session_id, ); false diff --git a/crates/dav/src/common/propfind.rs b/crates/dav/src/common/propfind.rs index 4df62ac6..a1be5c97 100644 --- a/crates/dav/src/common/propfind.rs +++ b/crates/dav/src/common/propfind.rs @@ -200,7 +200,7 @@ impl PropFindRequestHandler for Server { } else if access_token.has_account_access(account_id) || (self.core.groupware.allow_directory_query && access_token.has_permission(Permission::DavPrincipalList)) - || access_token.has_permission(Permission::AccountQuery) + || access_token.has_permission(Permission::SysAccountQuery) { self.prepare_principal_propfind_response( access_token, @@ -255,7 +255,7 @@ impl PropFindRequestHandler for Server { ) } else if (self.core.groupware.allow_directory_query && access_token.has_permission(Permission::DavPrincipalList)) - || access_token.has_permission(Permission::AccountQuery) + || access_token.has_permission(Permission::SysAccountQuery) { // Return all principals self.registry() diff --git a/crates/dav/src/principal/propsearch.rs b/crates/dav/src/principal/propsearch.rs index 5cf64c21..def9d590 100644 --- a/crates/dav/src/principal/propsearch.rs +++ b/crates/dav/src/principal/propsearch.rs @@ -13,7 +13,7 @@ use dav_proto::schema::{ }; use http_proto::HttpResponse; use hyper::StatusCode; -use registry::schema::prelude::ObjectType; +use registry::schema::prelude::{ObjectType, Property}; use store::{registry::RegistryQuery, roaring::RoaringBitmap}; use trc::AddContext; use types::collection::Collection; @@ -51,7 +51,7 @@ impl PrincipalPropSearch for Server { .query::( RegistryQuery::new(ObjectType::Account) .with_tenant(access_token.tenant_id()) - .text(search_for), + .text(Property::Text, search_for), ) .await .caused_by(trc::location!())?; diff --git a/crates/dav/src/request.rs b/crates/dav/src/request.rs index b4b7ad4f..56ec939e 100644 --- a/crates/dav/src/request.rs +++ b/crates/dav/src/request.rs @@ -179,7 +179,7 @@ impl DavRequestDispatcher for Server { Report::AclPrincipalPropSet(report) => { // Validate permissions if !self.core.groupware.allow_directory_query - && !access_token.has_permission(Permission::AccountQuery) + && !access_token.has_permission(Permission::SysAccountQuery) { return Err(DavError::Condition( DavErrorCondition::new( @@ -198,7 +198,7 @@ impl DavRequestDispatcher for Server { Report::PrincipalMatch(report) => { // Validate permissions if !self.core.groupware.allow_directory_query - && !access_token.has_permission(Permission::AccountQuery) + && !access_token.has_permission(Permission::SysAccountQuery) { return Err(DavError::Condition( DavErrorCondition::new( @@ -218,7 +218,7 @@ impl DavRequestDispatcher for Server { if resource == DavResourceName::Principal { // Validate permissions if !self.core.groupware.allow_directory_query - && !access_token.has_permission(Permission::AccountQuery) + && !access_token.has_permission(Permission::SysAccountQuery) { return Err(DavError::Condition( DavErrorCondition::new( diff --git a/crates/jmap-proto/src/object/registry.rs b/crates/jmap-proto/src/object/registry.rs index 5e9c1964..e4dc73e8 100644 --- a/crates/jmap-proto/src/object/registry.rs +++ b/crates/jmap-proto/src/object/registry.rs @@ -8,10 +8,7 @@ use crate::{ object::{AnyId, JmapObject, JmapObjectId}, request::deserialize::DeserializeArguments, }; -use registry::{ - jmap::RegistryValue, - schema::prelude::{ObjectType, Property}, -}; +use registry::{jmap::RegistryValue, schema::prelude::Property, types::EnumImpl}; use std::borrow::Cow; use types::id::Id; @@ -20,16 +17,25 @@ pub struct Registry; #[derive(Debug, Clone, PartialEq, Eq)] pub enum RegistryFilter { - Type(ObjectType), - Text(String), - Id(Vec), - Property(Property), + Property { + property: Property, + operator: RegistryFilterOperator, + value: serde_json::Value, + }, _T(String), } +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum RegistryFilterOperator { + Equal, + GreaterThan, + GreaterThanOrEqual, + LessThan, + LessThanOrEqual, +} + #[derive(Debug, Clone, PartialEq, Eq)] pub enum RegistryComparator { - Id, Property(Property), _T(String), } @@ -119,18 +125,31 @@ impl<'de> DeserializeArguments<'de> for RegistryFilter { where A: serde::de::MapAccess<'de>, { - hashify::fnc_map!(key.as_bytes(), - b"text" => { - *self = RegistryFilter::Text(map.next_value()?); - }, - b"id" => { - *self = RegistryFilter::Id(map.next_value()?); - }, - _ => { - *self = RegistryFilter::_T(key.to_string()); - let _ = map.next_value::()?; - } - ); + if let Some(property) = Property::parse(key) { + let value = map.next_value()?; + *self = RegistryFilter::Property { + property, + operator: RegistryFilterOperator::Equal, + value, + }; + return Ok(()); + } else if let Some((property, operator)) = key.rsplit_once("Is") + && let (Some(property), Some(operator)) = ( + Property::parse(property), + RegistryFilterOperator::parse(operator), + ) + { + let value = map.next_value()?; + *self = RegistryFilter::Property { + property, + operator, + value, + }; + return Ok(()); + } + + *self = RegistryFilter::_T(key.to_string()); + let _ = map.next_value::()?; Ok(()) } @@ -143,14 +162,12 @@ impl<'de> DeserializeArguments<'de> for RegistryComparator { { if key == "property" { let value = map.next_value::>()?; - hashify::fnc_map!(value.as_bytes(), - b"id" => { - *self = RegistryComparator::Id; - }, - _ => { - *self = RegistryComparator::_T(key.to_string()); - } - ); + + if let Some(property) = Property::parse(value.as_ref()) { + *self = RegistryComparator::Property(property); + } else { + *self = RegistryComparator::_T(value.into_owned()); + } } else { let _ = map.next_value::()?; } @@ -159,6 +176,17 @@ impl<'de> DeserializeArguments<'de> for RegistryComparator { } } +impl RegistryFilterOperator { + pub fn parse(value: &str) -> Option { + hashify::tiny_map!(value.as_bytes(), + b"GreaterThan" => RegistryFilterOperator::GreaterThan, + b"GreaterThanOrEqual" => RegistryFilterOperator::GreaterThanOrEqual, + b"LessThan" => RegistryFilterOperator::LessThan, + b"LessThanOrEqual" => RegistryFilterOperator::LessThanOrEqual, + ) + } +} + impl Default for RegistryFilter { fn default() -> Self { RegistryFilter::_T("".to_string()) diff --git a/crates/jmap/src/api/query.rs b/crates/jmap/src/api/query.rs index 196abaa6..4254041f 100644 --- a/crates/jmap/src/api/query.rs +++ b/crates/jmap/src/api/query.rs @@ -15,7 +15,7 @@ pub struct QueryResponseBuilder { requested_position: i32, position: i32, pub limit: usize, - anchor: u32, + anchor: u64, anchor_offset: i32, has_anchor: bool, anchor_found: bool, @@ -41,18 +41,13 @@ impl QueryResponseBuilder { (std::cmp::min(max_results, total_results), max_results) }; - let (has_anchor, anchor) = request - .anchor - .map(|anchor| (true, anchor.document_id())) - .unwrap_or((false, 0)); - QueryResponseBuilder { requested_position: request.position.unwrap_or(0), position: request.position.unwrap_or(0), limit: limit_total, - anchor, + has_anchor: request.anchor.is_some(), + anchor: request.anchor.map(|anchor| anchor.id()).unwrap_or(0), anchor_offset: request.anchor_offset.unwrap_or(0), - has_anchor, anchor_found: false, response: QueryResponse { account_id: request.account_id, @@ -80,7 +75,7 @@ impl QueryResponseBuilder { } pub fn add_id(&mut self, id: Id) -> bool { - let document_id = id.document_id(); + let id_u64 = id.id(); // Pagination if !self.has_anchor { @@ -98,7 +93,7 @@ impl QueryResponseBuilder { } } else if self.anchor_offset >= 0 { if !self.anchor_found { - if document_id != self.anchor { + if id_u64 != self.anchor { return true; } self.anchor_found = true; @@ -113,7 +108,7 @@ impl QueryResponseBuilder { } } } else { - self.anchor_found = document_id == self.anchor; + self.anchor_found = id_u64 == self.anchor; self.response.ids.push(id); if self.anchor_found { diff --git a/crates/jmap/src/changes/query.rs b/crates/jmap/src/changes/query.rs index 899314c1..87f13d4f 100644 --- a/crates/jmap/src/changes/query.rs +++ b/crates/jmap/src/changes/query.rs @@ -10,7 +10,7 @@ use crate::{ calendar_event_notification::query::CalendarEventNotificationQuery, contact::query::ContactCardQuery, email::query::EmailQuery, file::query::FileNodeQuery, mailbox::query::MailboxQuery, share_notification::query::ShareNotificationQuery, - sieve::query::SieveScriptQuery, submission::query::EmailSubmissionQuery, + submission::query::EmailSubmissionQuery, }; use common::{Server, auth::AccessToken}; use jmap_proto::{ diff --git a/crates/jmap/src/principal/availability.rs b/crates/jmap/src/principal/availability.rs index 1ebb6a84..32618fbf 100644 --- a/crates/jmap/src/principal/availability.rs +++ b/crates/jmap/src/principal/availability.rs @@ -62,7 +62,7 @@ impl PrincipalGetAvailability for Server { access_token: &AccessToken, ) -> trc::Result { if !self.core.groupware.allow_directory_query - && !access_token.has_permission(Permission::AccountQuery) + && !access_token.has_permission(Permission::SysAccountQuery) { return Err(trc::JmapEvent::Forbidden .into_err() diff --git a/crates/jmap/src/principal/get.rs b/crates/jmap/src/principal/get.rs index 091fed9d..ec9240ea 100644 --- a/crates/jmap/src/principal/get.rs +++ b/crates/jmap/src/principal/get.rs @@ -32,7 +32,7 @@ impl PrincipalGet for Server { access_token: &AccessToken, ) -> trc::Result> { if !self.core.groupware.allow_directory_query - && !access_token.has_permission(Permission::AccountQuery) + && !access_token.has_permission(Permission::SysAccountQuery) { return Err(trc::JmapEvent::Forbidden .into_err() diff --git a/crates/jmap/src/principal/query.rs b/crates/jmap/src/principal/query.rs index acd1aa96..7d1afc2e 100644 --- a/crates/jmap/src/principal/query.rs +++ b/crates/jmap/src/principal/query.rs @@ -42,7 +42,7 @@ impl PrincipalQuery for Server { access_token: &AccessToken, ) -> trc::Result { if !self.core.groupware.allow_directory_query - && !access_token.has_permission(Permission::AccountQuery) + && !access_token.has_permission(Permission::SysAccountQuery) { return Err(trc::JmapEvent::Forbidden .into_err() @@ -88,7 +88,7 @@ impl PrincipalQuery for Server { .query::( RegistryQuery::new(ObjectType::Account) .with_tenant(access_token.tenant_id()) - .text(text), + .text(Property::Text, text), ) .await .caused_by(trc::location!())?, diff --git a/crates/jmap/src/registry/mapping/account.rs b/crates/jmap/src/registry/mapping/account.rs index 773a77d7..240a1717 100644 --- a/crates/jmap/src/registry/mapping/account.rs +++ b/crates/jmap/src/registry/mapping/account.rs @@ -4,9 +4,15 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ -use crate::registry::{ - mapping::{RegistryGetResponse, RegistrySetResponse, principal::build_set_error}, - set::map_write_error, +use crate::{ + api::query::QueryResponseBuilder, + registry::{ + mapping::{ + RegistryGetResponse, RegistryQueryResponse, RegistrySetResponse, + principal::build_set_error, + }, + set::map_write_error, + }, }; use common::{ auth::{ @@ -648,6 +654,12 @@ pub(crate) async fn account_get( Ok(get) } +pub(crate) async fn credential_query( + mut query: RegistryQueryResponse<'_>, +) -> trc::Result { + todo!() +} + fn validate_credential_permissions( access_token: &AccessToken, credential: &SecondaryCredential, diff --git a/crates/jmap/src/registry/mapping/action.rs b/crates/jmap/src/registry/mapping/action.rs index 90b1d4e9..19657613 100644 --- a/crates/jmap/src/registry/mapping/action.rs +++ b/crates/jmap/src/registry/mapping/action.rs @@ -4,8 +4,7 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ -use std::time::Instant; - +use crate::registry::mapping::{RegistrySetResponse, map_bootstrap_error}; use common::{ Server, config::mailstore::spamfilter::SpamFilterAction, @@ -26,17 +25,16 @@ use registry::{ prelude::{ObjectType, Property}, structs::{Action, DmarcTroubleshoot, SpamClassify, SpamClassifyTag}, }, - types::{ObjectImpl, error::Error}, + types::{EnumImpl, ObjectImpl}, }; use smtp_proto::{MAIL_BODY_7BIT, MAIL_BODY_8BITMIME, MAIL_BODY_BINARYMIME, MAIL_SMTPUTF8}; use spam_filter::{ SpamFilterInput, analysis::{init::SpamFilterInit, score::SpamFilterAnalyzeScore}, }; +use std::time::Instant; use store::write::now; -use crate::registry::mapping::RegistrySetResponse; - pub(crate) async fn action_set( mut set: RegistrySetResponse<'_>, ) -> trc::Result> { @@ -72,6 +70,17 @@ pub(crate) async fn action_set( continue 'outer; } + if !set.access_token.has_permission(action.permission()) { + set.response.not_created.append( + id, + SetError::forbidden().with_description(format!( + "Insufficient permissions to perform action of type {}", + action.object_type().as_str() + )), + ); + continue 'outer; + } + match action { Action::ReloadSettings | Action::ReloadTlsCertificates @@ -524,20 +533,3 @@ async fn dmarc_troubleshoot( Some(request) } - -fn map_bootstrap_error(error: Vec) -> SetError { - match error.into_iter().next().unwrap() { - Error::Validation { object_id, errors } => SetError::new(SetErrorType::ValidationFailed) - .with_validation_errors(errors) - .with_object_id(object_id), - Error::Build { object_id, message } => SetError::new(SetErrorType::ValidationFailed) - .with_description(message) - .with_object_id(object_id), - Error::Internal { object_id, error } => SetError::new(SetErrorType::Forbidden) - .with_description(error.to_string()) - .with_object_id_opt(object_id), - Error::NotFound { object_id } => { - SetError::new(SetErrorType::NotFound).with_object_id(object_id) - } - } -} diff --git a/crates/jmap/src/registry/mapping/archived_item.rs b/crates/jmap/src/registry/mapping/archived_item.rs index 991d6db9..91b11ecf 100644 --- a/crates/jmap/src/registry/mapping/archived_item.rs +++ b/crates/jmap/src/registry/mapping/archived_item.rs @@ -4,7 +4,10 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ -use crate::registry::mapping::{RegistryGetResponse, RegistrySetResponse}; +use crate::{ + api::query::QueryResponseBuilder, + registry::mapping::{RegistryGetResponse, RegistryQueryResponse, RegistrySetResponse}, +}; use jmap_proto::error::set::SetError; use jmap_tools::{Key, Value}; use registry::{ @@ -230,11 +233,15 @@ pub(crate) async fn archived_item_get( let ids = if let Some(ids) = get.ids.take() { ids } else { + let query = if !get.is_account_filtered { + RegistryQuery::new(get.object_type).greater_than_or_equal(Property::AccountId, 0u64) + } else { + RegistryQuery::new(get.object_type).with_account(get.account_id) + }; + get.server .registry() - .query::>( - RegistryQuery::new(get.object_type).with_account(get.account_id), - ) + .query::>(query) .await? .into_iter() .take(get.server.core.jmap.get_max_objects) @@ -271,3 +278,9 @@ pub(crate) async fn archived_item_get( Ok(get) } + +pub(crate) async fn archived_item_query( + mut query: RegistryQueryResponse<'_>, +) -> trc::Result { + todo!() +} diff --git a/crates/jmap/src/registry/mapping/log.rs b/crates/jmap/src/registry/mapping/log.rs index 8f9dfd4c..e5685b17 100644 --- a/crates/jmap/src/registry/mapping/log.rs +++ b/crates/jmap/src/registry/mapping/log.rs @@ -4,7 +4,10 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ -use crate::registry::mapping::RegistryGetResponse; +use crate::{ + api::query::QueryResponseBuilder, + registry::mapping::{RegistryGetResponse, RegistryQueryResponse}, +}; use chrono::DateTime; use registry::{ jmap::IntoValue, @@ -65,6 +68,12 @@ pub(crate) async fn log_get( Ok(get) } +pub(crate) async fn log_query( + mut query: RegistryQueryResponse<'_>, +) -> trc::Result { + todo!() +} + fn line_numbers( path: impl AsRef, filter: &str, diff --git a/crates/jmap/src/registry/mapping/mod.rs b/crates/jmap/src/registry/mapping/mod.rs index 9708812f..38522da5 100644 --- a/crates/jmap/src/registry/mapping/mod.rs +++ b/crates/jmap/src/registry/mapping/mod.rs @@ -6,14 +6,15 @@ use common::{Server, auth::AccessToken}; use jmap_proto::{ - error::set::SetError, - method::{get::GetResponse, set::SetResponse}, + error::set::{SetError, SetErrorType}, + method::{get::GetResponse, query::QueryRequest, set::SetResponse}, object::registry::Registry, }; use jmap_tools::Map; use registry::{ jmap::{JmapValue, RegistryValue}, schema::prelude::{ObjectType, Property}, + types::error::Error, }; use std::net::IpAddr; use store::ahash::AHashSet; @@ -61,6 +62,13 @@ pub(crate) struct RegistrySetResponse<'x> { pub is_account_filtered: bool, } +pub(crate) struct RegistryQueryResponse<'x> { + pub server: &'x Server, + pub access_token: &'x AccessToken, + pub object_type: ObjectType, + pub request: QueryRequest, +} + pub type ValidationResult = trc::Result>>; pub struct ObjectResponse { @@ -85,3 +93,20 @@ impl Default for ObjectResponse { } } } + +pub(crate) fn map_bootstrap_error(error: Vec) -> SetError { + match error.into_iter().next().unwrap() { + Error::Validation { object_id, errors } => SetError::new(SetErrorType::ValidationFailed) + .with_validation_errors(errors) + .with_object_id(object_id), + Error::Build { object_id, message } => SetError::new(SetErrorType::ValidationFailed) + .with_description(message) + .with_object_id(object_id), + Error::Internal { object_id, error } => SetError::new(SetErrorType::Forbidden) + .with_description(error.to_string()) + .with_object_id_opt(object_id), + Error::NotFound { object_id } => { + SetError::new(SetErrorType::NotFound).with_object_id(object_id) + } + } +} diff --git a/crates/jmap/src/registry/mapping/queued_message.rs b/crates/jmap/src/registry/mapping/queued_message.rs index 77191242..9b2d78e9 100644 --- a/crates/jmap/src/registry/mapping/queued_message.rs +++ b/crates/jmap/src/registry/mapping/queued_message.rs @@ -4,7 +4,10 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ -use crate::registry::mapping::{RegistryGetResponse, RegistrySetResponse}; +use crate::{ + api::query::QueryResponseBuilder, + registry::mapping::{RegistryGetResponse, RegistryQueryResponse, RegistrySetResponse}, +}; use common::{ Server, config::smtp::queue::{ArchivedQueueExpiry, QueueName}, @@ -259,6 +262,12 @@ pub(crate) async fn queued_message_get( Ok(get) } +pub(crate) async fn queued_message_query( + mut query: RegistryQueryResponse<'_>, +) -> trc::Result { + todo!() +} + async fn tenant_domains(server: &Server, tenant_id: u32) -> trc::Result> { let domain_ids = server .registry() diff --git a/crates/jmap/src/registry/mapping/report.rs b/crates/jmap/src/registry/mapping/report.rs index 21aca2fc..43a95af7 100644 --- a/crates/jmap/src/registry/mapping/report.rs +++ b/crates/jmap/src/registry/mapping/report.rs @@ -4,9 +4,14 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ -use crate::registry::mapping::{RegistryGetResponse, RegistrySetResponse}; -use common::Server; -use jmap_proto::error::set::SetError; +use crate::{ + api::query::QueryResponseBuilder, + registry::{ + mapping::{RegistryGetResponse, RegistryQueryResponse, RegistrySetResponse}, + query::RegistryQueryFilters, + }, +}; +use jmap_proto::{error::set::SetError, types::state::State}; use jmap_tools::{Key, Value}; use registry::{ jmap::IntoValue, @@ -16,10 +21,10 @@ use registry::{ use smtp::reporting::index::{ExternalReportIndex, InternalReportIndex}; use std::str::FromStr; use store::{ - IterateParams, U16_LEN, ValueKey, + U64_LEN, ValueKey, ahash::AHashSet, - registry::RegistryQuery, - write::{BatchBuilder, RegistryClass, ValueClass, key::DeserializeBigEndian}, + registry::{RegistryFilter, RegistryFilterValue, RegistryQuery}, + write::{BatchBuilder, RegistryClass, ValueClass, key::KeySerializer}, }; use trc::AddContext; use types::id::Id; @@ -184,7 +189,20 @@ pub(crate) async fn report_get( .map(Id::from) .collect() } else { - internal_report_ids(get.server, object_id, get.server.core.jmap.get_max_objects).await? + get.server + .registry() + .query::>(RegistryQuery::new(get.object_type).filter( + RegistryFilter::greater_than( + Property::Domain, + RegistryFilterValue::Bytes(vec![]), + true, + ), + )) + .await? + .into_iter() + .take(get.server.core.jmap.get_max_objects) + .map(Id::from) + .collect() }; let tenant_id = get.access_token.tenant_id().map(Id::from); @@ -210,46 +228,189 @@ pub(crate) async fn report_get( Ok(get) } -async fn internal_report_ids( - server: &Server, - object_id: u16, - max_results: usize, -) -> trc::Result> { - let mut events = Vec::with_capacity(8); +pub(crate) async fn report_query( + mut req: RegistryQueryResponse<'_>, +) -> trc::Result { + let mut query = store::registry::RegistryQuery::new(req.object_type) + .with_tenant(req.access_token.tenant_id()); + let is_internal = matches!( + req.object_type, + ObjectType::DmarcInternalReport | ObjectType::TlsInternalReport + ); - let from_key = ValueKey::from(ValueClass::Registry(RegistryClass::PrimaryKey { - object_id: object_id.into(), - index_id: Property::Domain.to_id(), - key: vec![], - })); - let to_key = ValueKey::from(ValueClass::Registry(RegistryClass::PrimaryKey { - object_id: object_id.into(), - index_id: Property::Domain.to_id(), - key: vec![ - u8::MAX, - u8::MAX, - u8::MAX, - u8::MAX, - u8::MAX, - u8::MAX, - u8::MAX, - u8::MAX, - ], - })); + req.request + .extract_filters(|property, op, value| match property { + Property::Domain => { + if let serde_json::Value::String(value) = value { + match req.object_type { + ObjectType::DmarcInternalReport => { + query.filters.push(RegistryFilter::greater_than_or_equal( + property, + RegistryFilterValue::Bytes( + KeySerializer::new(value.len() + U64_LEN) + .write(value.as_str()) + .write(0u64) + .finalize(), + ), + true, + )); + query.filters.push(RegistryFilter::less_than_or_equal( + property, + RegistryFilterValue::Bytes( + KeySerializer::new(value.len() + U64_LEN) + .write(value.as_str()) + .write(u64::MAX) + .finalize(), + ), + true, + )); - server - .store() - .iterate( - IterateParams::new(from_key, to_key).ascending(), - |key, value| { - if !value.is_empty() { - events.push(key.deserialize_be_u64(U16_LEN)?.into()); + true + } + ObjectType::TlsInternalReport => { + query + .filters + .push(RegistryFilter::equal(property, value, true)); + true + } + _ => false, + } + } else { + false } + } + Property::Text if !is_internal => { + if let serde_json::Value::String(value) = value { + query.filters.push(RegistryFilter::text(property, value)); + true + } else { + false + } + } + Property::MemberTenantId if !is_internal => { + if req.access_token.tenant_id().is_none() + && let Some(id) = value.as_str().and_then(|s| Id::from_str(s).ok()) + { + query + .filters + .push(RegistryFilter::equal(property, id.id(), false)); + true + } else { + false + } + } + Property::TotalFailedSessions | Property::TotalSuccessfulSessions if !is_internal => { + if let Some(value) = value.as_u64() { + query.filters.push(store::registry::RegistryFilter { + property, + op, + value: value.into(), + is_pk: false, + }); + true + } else { + false + } + } + Property::ExpiresAt if !is_internal => { + if let Some(value) = value + .as_str() + .and_then(|value| UTCDateTime::from_str(value).ok()) + { + query.filters.push(store::registry::RegistryFilter { + property, + op, + value: (value.timestamp() as u64).into(), + is_pk: false, + }); + true + } else { + false + } + } + _ => false, + })?; - Ok(events.len() < max_results) - }, - ) - .await - .caused_by(trc::location!()) - .map(|_| events) + let (comparator, is_ascending) = req.request.extract_comparator()?; + + if !query.has_filters() { + if is_internal { + query.filters.push(RegistryFilter::greater_than( + Property::Domain, + RegistryFilterValue::Bytes(vec![]), + true, + )); + } else { + query.filters.push(RegistryFilter::greater_than( + Property::ExpiresAt, + 0u64, + false, + )); + } + } + + let matches = req.server.registry().query::>(query).await?; + let results = match comparator { + Property::Id => { + let mut results = matches.into_iter().collect::>(); + if is_ascending { + results.sort_unstable(); + } else { + results.sort_unstable_by(|a, b| b.cmp(a)); + } + results + } + Property::Domain if is_internal => { + if !matches.is_empty() { + req.server + .registry() + .sort_by_pk( + req.object_type, + Property::Domain, + Some(matches), + is_ascending, + ) + .await? + } else { + vec![] + } + } + Property::ExpiresAt if !is_internal => { + if !matches.is_empty() { + req.server + .registry() + .sort_by_index( + req.object_type, + Property::ExpiresAt, + Some(matches), + is_ascending, + ) + .await? + } else { + vec![] + } + } + property => { + return Err(trc::JmapEvent::UnsupportedSort.into_err().details(format!( + "Property {} is not supported for sorting", + property + ))); + } + }; + + // Build response + let mut response = QueryResponseBuilder::new( + results.len(), + req.server.core.jmap.query_max_results, + State::Initial, + &req.request, + ); + + for id in results { + if !response.add_id(id.into()) { + break; + } + } + + Ok(response) } diff --git a/crates/jmap/src/registry/mapping/spam_sample.rs b/crates/jmap/src/registry/mapping/spam_sample.rs index f9087c1c..7caec320 100644 --- a/crates/jmap/src/registry/mapping/spam_sample.rs +++ b/crates/jmap/src/registry/mapping/spam_sample.rs @@ -5,8 +5,9 @@ */ use crate::{ + api::query::QueryResponseBuilder, blob::download::BlobDownload, - registry::mapping::{RegistryGetResponse, RegistrySetResponse}, + registry::mapping::{RegistryGetResponse, RegistryQueryResponse, RegistrySetResponse}, }; use jmap_proto::error::set::SetError; use jmap_tools::{JsonPointer, JsonPointerItem, Key}; @@ -298,3 +299,9 @@ pub(crate) async fn spam_sample_get( Ok(get) } + +pub(crate) async fn spam_sample_query( + mut query: RegistryQueryResponse<'_>, +) -> trc::Result { + todo!() +} diff --git a/crates/jmap/src/registry/mapping/task.rs b/crates/jmap/src/registry/mapping/task.rs index 526a672d..5f709ad8 100644 --- a/crates/jmap/src/registry/mapping/task.rs +++ b/crates/jmap/src/registry/mapping/task.rs @@ -4,7 +4,10 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ -use crate::registry::mapping::{RegistryGetResponse, RegistrySetResponse}; +use crate::{ + api::query::QueryResponseBuilder, + registry::mapping::{RegistryGetResponse, RegistryQueryResponse, RegistrySetResponse}, +}; use common::Server; use jmap_proto::error::set::{SetError, SetErrorType}; use jmap_tools::{JsonPointer, JsonPointerItem, Key}; @@ -59,6 +62,17 @@ pub(crate) async fn task_set( continue 'outer; } + if !set.access_token.has_permission(task.permission()) { + set.response.not_created.append( + id, + SetError::forbidden().with_description(format!( + "Insufficient permissions to create task of type {}", + task.object_type().as_str() + )), + ); + continue 'outer; + } + let task_type = task.object_type(); match task_type { TaskType::IndexDocument @@ -136,6 +150,17 @@ pub(crate) async fn task_set( continue; }; + if !set.access_token.has_permission(task.permission()) { + set.response.not_updated.append( + id, + SetError::forbidden().with_description(format!( + "Insufficient permissions to update task of type {}", + task.object_type().as_str() + )), + ); + continue 'outer; + } + if !set.server.try_lock_task(task_id).await { set.response.not_updated.append( id, @@ -205,6 +230,17 @@ pub(crate) async fn task_set( continue; }; + if !set.access_token.has_permission(task.permission()) { + set.response.not_destroyed.append( + id, + SetError::forbidden().with_description(format!( + "Insufficient permissions to destroy task of type {}", + task.object_type().as_str() + )), + ); + continue; + } + if !set.server.try_lock_task(task_id).await { set.response.not_destroyed.append( id, @@ -335,6 +371,12 @@ pub(crate) async fn task_get( Ok(get) } +pub(crate) async fn task_query( + mut query: RegistryQueryResponse<'_>, +) -> trc::Result { + todo!() +} + async fn task_ids(server: &Server, max_results: usize) -> trc::Result> { let mut events = Vec::with_capacity(8); diff --git a/crates/jmap/src/registry/mapping/telemetry.rs b/crates/jmap/src/registry/mapping/telemetry.rs index 46109316..33e40ab3 100644 --- a/crates/jmap/src/registry/mapping/telemetry.rs +++ b/crates/jmap/src/registry/mapping/telemetry.rs @@ -4,7 +4,10 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ -use crate::registry::mapping::RegistryGetResponse; +use crate::{ + api::query::QueryResponseBuilder, + registry::mapping::{RegistryGetResponse, RegistryQueryResponse}, +}; use common::Server; use registry::{ jmap::IntoValue, @@ -100,6 +103,18 @@ pub(crate) async fn metric_get( Ok(get) } +pub(crate) async fn trace_query( + mut query: RegistryQueryResponse<'_>, +) -> trc::Result { + todo!() +} + +pub(crate) async fn metric_query( + mut query: RegistryQueryResponse<'_>, +) -> trc::Result { + todo!() +} + async fn metric_ids(server: &Server, max_results: usize) -> trc::Result> { let mut events = Vec::with_capacity(8); diff --git a/crates/jmap/src/registry/query.rs b/crates/jmap/src/registry/query.rs index 4d120019..8da9de22 100644 --- a/crates/jmap/src/registry/query.rs +++ b/crates/jmap/src/registry/query.rs @@ -4,12 +4,42 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ +use crate::{ + api::query::QueryResponseBuilder, + registry::mapping::{ + RegistryQueryResponse, + account::credential_query, + archived_item::archived_item_query, + log::log_query, + queued_message::queued_message_query, + report::report_query, + spam_sample::spam_sample_query, + task::task_query, + telemetry::{metric_query, trace_query}, + }, +}; use common::{Server, auth::AccessToken}; use jmap_proto::{ - method::query::{QueryRequest, QueryResponse}, - object::registry::Registry, + method::query::{Comparator, Filter, QueryRequest, QueryResponse}, + object::registry::{Registry, RegistryComparator, RegistryFilter, RegistryFilterOperator}, + types::state::State, }; -use registry::schema::prelude::ObjectType; +use registry::{ + schema::{ + enums::{AccountType, Permission}, + prelude::{ObjectType, Property}, + }, + types::{ + EnumImpl, + index::{IndexSchemaType, IndexSchemaValueType}, + }, +}; +use std::str::FromStr; +use store::{ + ahash::AHashSet, + registry::{RegistryFilterOp, RegistryFilterValue}, +}; +use types::id::Id; pub trait RegistryQuery: Sync + Send { fn registry_query( @@ -27,6 +57,302 @@ impl RegistryQuery for Server { mut request: QueryRequest, access_token: &AccessToken, ) -> trc::Result { - todo!() + match object_type { + ObjectType::ArfExternalReport + | ObjectType::DmarcExternalReport + | ObjectType::TlsExternalReport + | ObjectType::DmarcInternalReport + | ObjectType::TlsInternalReport => report_query(RegistryQueryResponse { + server: self, + access_token, + object_type, + request, + }) + .await + .and_then(|response| response.build()), + + ObjectType::ArchivedItem => archived_item_query(RegistryQueryResponse { + server: self, + access_token, + object_type, + request, + }) + .await + .and_then(|response| response.build()), + + ObjectType::SpamTrainingSample => spam_sample_query(RegistryQueryResponse { + server: self, + access_token, + object_type, + request, + }) + .await + .and_then(|response| response.build()), + + ObjectType::QueuedMessage => queued_message_query(RegistryQueryResponse { + server: self, + access_token, + object_type, + request, + }) + .await + .and_then(|response| response.build()), + + ObjectType::Credential => credential_query(RegistryQueryResponse { + server: self, + access_token, + object_type, + request, + }) + .await + .and_then(|response| response.build()), + + ObjectType::Task => task_query(RegistryQueryResponse { + server: self, + access_token, + object_type, + request, + }) + .await + .and_then(|response| response.build()), + + ObjectType::Log => log_query(RegistryQueryResponse { + server: self, + access_token, + object_type, + request, + }) + .await + .and_then(|response| response.build()), + + ObjectType::Metric => metric_query(RegistryQueryResponse { + server: self, + access_token, + object_type, + request, + }) + .await + .and_then(|response| response.build()), + + ObjectType::Trace => trace_query(RegistryQueryResponse { + server: self, + access_token, + object_type, + request, + }) + .await + .and_then(|response| response.build()), + + ObjectType::Action => Err(trc::JmapEvent::InvalidArguments + .into_err() + .details("Actions cannot be queried")), + + _ => { + let mut query = store::registry::RegistryQuery::new(object_type) + .with_tenant(access_token.tenant_id()); + let can_impersonate = access_token.has_permission(Permission::Impersonate); + if !can_impersonate { + query = query.with_account(request.account_id.document_id()); + } + let indexes = object_type.indexes(); + request.extract_filters(|property, op, value| match property { + Property::MemberTenantId if access_token.tenant_id().is_some() => true, + Property::AccountId if !can_impersonate => true, + property => { + let Some(index) = indexes.iter().find(|i| i.prop == property) else { + return false; + }; + let is_pk = index.typ == IndexSchemaType::Unique; + + let value = match (index.value, value) { + (IndexSchemaValueType::Keyword, serde_json::Value::String(value)) => { + Some(RegistryFilterValue::from(value)) + } + (IndexSchemaValueType::Text, serde_json::Value::String(value)) => { + query.push_text(property, value); + return true; + } + (IndexSchemaValueType::Number, serde_json::Value::Number(value)) => { + value + .as_i64() + .map(|value| RegistryFilterValue::from(value as u64)) + } + (IndexSchemaValueType::Enum, serde_json::Value::String(value)) + if (property == Property::Type + && object_type == ObjectType::Account) => + { + AccountType::parse(&value) + .map(|id| RegistryFilterValue::from(id.to_id())) + } + (IndexSchemaValueType::Boolean, serde_json::Value::Bool(value)) => { + Some(RegistryFilterValue::from(value)) + } + (IndexSchemaValueType::Id, serde_json::Value::String(value)) => { + Id::from_str(&value) + .ok() + .map(|id| RegistryFilterValue::from(id.id())) + } + _ => None, + }; + + if let Some(value) = value { + query.filters.push(store::registry::RegistryFilter { + property, + op, + value, + is_pk, + }); + + true + } else { + false + } + } + })?; + + let (comparator, is_ascending) = request.extract_comparator()?; + let matches = if query.has_filters() || matches!(comparator, Property::Id) { + let matches = self.registry().query::>(query).await?; + if matches.is_empty() { + return QueryResponseBuilder::new( + 0, + self.core.jmap.query_max_results, + State::Initial, + &request, + ) + .build(); + } + matches.into() + } else { + None + }; + + let results = match comparator { + Property::Id => { + let mut results = matches.unwrap().into_iter().collect::>(); + if is_ascending { + results.sort_unstable(); + } else { + results.sort_unstable_by(|a, b| b.cmp(a)); + } + results + } + property => { + let Some(index) = indexes + .iter() + .find(|i| i.prop == property && i.value != IndexSchemaValueType::Text) + else { + return Err(trc::JmapEvent::UnsupportedSort.into_err().details( + format!("Property {} is not supported for sorting", property), + )); + }; + + if index.typ == IndexSchemaType::Search { + self.registry() + .sort_by_index(object_type, index.prop, matches, is_ascending) + .await? + } else { + self.registry() + .sort_by_pk(object_type, index.prop, matches, is_ascending) + .await? + } + } + }; + + // Build response + let mut response = QueryResponseBuilder::new( + results.len(), + self.core.jmap.query_max_results, + State::Initial, + &request, + ); + + for id in results { + if !response.add_id(id.into()) { + break; + } + } + + response.build() + } + } + } +} + +pub(crate) trait RegistryQueryFilters { + fn extract_filters( + &mut self, + cb: impl FnMut(Property, RegistryFilterOp, serde_json::Value) -> bool, + ) -> trc::Result<()>; + + fn extract_comparator(&mut self) -> trc::Result<(Property, bool)>; +} + +impl RegistryQueryFilters for QueryRequest { + fn extract_filters( + &mut self, + mut cb: impl FnMut(Property, RegistryFilterOp, serde_json::Value) -> bool, + ) -> trc::Result<()> { + for cond in std::mem::take(&mut self.filter) { + match cond { + Filter::Property(cond) => match cond { + RegistryFilter::Property { + property, + operator, + value, + } => { + let operator = match operator { + RegistryFilterOperator::Equal => RegistryFilterOp::Equal, + RegistryFilterOperator::GreaterThan => RegistryFilterOp::GreaterThan, + RegistryFilterOperator::GreaterThanOrEqual => { + RegistryFilterOp::GreaterEqualThan + } + RegistryFilterOperator::LessThan => RegistryFilterOp::LowerThan, + RegistryFilterOperator::LessThanOrEqual => { + RegistryFilterOp::LowerEqualThan + } + }; + if !cb(property, operator, value) { + return Err(trc::JmapEvent::UnsupportedFilter.into_err().details( + format!( + "Filter on property {} is not supported or invalid", + property + ), + )); + } + } + RegistryFilter::_T(other) => { + return Err(trc::JmapEvent::UnsupportedFilter + .into_err() + .details(other.to_string())); + } + }, + Filter::And | Filter::Close => {} + Filter::Or | Filter::Not => { + return Err(trc::JmapEvent::UnsupportedFilter + .into_err() + .details("Only AND is supported in filters".to_string())); + } + } + } + + Ok(()) + } + + fn extract_comparator(&mut self) -> trc::Result<(Property, bool)> { + let comparator = self + .sort + .take() + .unwrap_or_default() + .into_iter() + .next() + .unwrap_or_else(|| Comparator::ascending(RegistryComparator::Property(Property::Id))); + + match comparator.property { + RegistryComparator::Property(property) => Ok((property, comparator.is_ascending)), + RegistryComparator::_T(other) => Err(trc::JmapEvent::UnsupportedSort + .into_err() + .details(format!("Property {} is not supported for sorting", other))), + } } } diff --git a/crates/jmap/src/registry/set.rs b/crates/jmap/src/registry/set.rs index 1082a644..1d493f8e 100644 --- a/crates/jmap/src/registry/set.rs +++ b/crates/jmap/src/registry/set.rs @@ -12,6 +12,7 @@ use crate::registry::mapping::{ action::action_set, archived_item::archived_item_set, dkim::validate_dkim_signature, + map_bootstrap_error, masked_email::validate_masked_email, principal::{ schedule_account_destruction, validate_account, validate_role, validate_tenant_quota, @@ -22,7 +23,10 @@ use crate::registry::mapping::{ spam_sample::spam_sample_set, task::task_set, }; -use common::{Server, auth::AccessToken, cache::invalidate::CacheInvalidationBuilder}; +use common::{ + Server, auth::AccessToken, cache::invalidate::CacheInvalidationBuilder, + expr::if_block::BootstrapExprExt, +}; use http_proto::HttpSessionData; use jmap_proto::{ error::set::{SetError, SetErrorType}, @@ -43,7 +47,10 @@ use registry::{ }, types::id::ObjectId, }; -use store::registry::write::{RegistryWrite, RegistryWriteResult}; +use store::registry::{ + bootstrap::Bootstrap, + write::{RegistryWrite, RegistryWriteResult}, +}; use trc::AddContext; use types::id::Id; use utils::map::vec_map::VecMap; @@ -72,9 +79,6 @@ impl RegistrySet for Server { access_token: &AccessToken, session: &HttpSessionData, ) -> trc::Result> { - let todo = "list"; - // locks for expensive tasks should be longer or renewed - // Validate expressions let object_flags = object_type.flags(); let is_singleton = (object_flags & OBJ_SINGLETON) != 0; let has_account_id = (object_flags & OBJ_FILTER_ACCOUNT) != 0; @@ -418,6 +422,24 @@ impl RegistrySet for Server { } }; + // Validate expressions + if let Some(expressions) = new_object.inner.expression_ctxs() { + let mut bp = Bootstrap::new_uninitialized(self.registry().clone()); + + for expression in expressions { + bp.compile_expr(ObjectId::new(object_type, 0u64.into()), &expression); + if !bp.errors.is_empty() { + set.failed( + modification, + map_bootstrap_error(bp.errors) + .with_object_id_opt(None) + .with_property(expression.property), + ); + continue 'outer; + } + } + } + // Save object let result = match &modification { Modification::Create(_) => { diff --git a/crates/registry/src/schema/prelude.rs b/crates/registry/src/schema/prelude.rs index 67dc58f9..611ed495 100644 --- a/crates/registry/src/schema/prelude.rs +++ b/crates/registry/src/schema/prelude.rs @@ -21,7 +21,7 @@ pub use crate::types::datetime::UTCDateTime; pub use crate::types::duration::Duration; pub use crate::types::error::*; pub use crate::types::float::Float; -pub use crate::types::index::IndexBuilder; +pub use crate::types::index::{IndexBuilder, IndexSchema, IndexSchemaType, IndexSchemaValueType}; pub use crate::types::ipaddr::IpAddr; pub use crate::types::ipmask::IpAddrOrMask; pub use crate::types::list::List; diff --git a/crates/registry/src/types/index.rs b/crates/registry/src/types/index.rs index 3858ff9d..ae6dc142 100644 --- a/crates/registry/src/types/index.rs +++ b/crates/registry/src/types/index.rs @@ -46,6 +46,31 @@ pub enum IndexValue<'x> { None, } +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub struct IndexSchema { + pub prop: Property, + pub typ: IndexSchemaType, + pub value: IndexSchemaValueType, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +#[repr(u8)] +pub enum IndexSchemaType { + Unique, + Search, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +#[repr(u8)] +pub enum IndexSchemaValueType { + Keyword, + Text, + Number, + Enum, + Boolean, + Id, +} + #[derive(Debug, Default)] pub struct IndexBuilder<'x> { @@ -135,6 +160,12 @@ impl<'x> IndexBuilder<'x> { } } +impl IndexSchema { + pub const fn new(prop: Property, typ: IndexSchemaType, value: IndexSchemaValueType) -> Self { + Self { prop, typ, value } + } +} + impl From for IndexValue<'_> { fn from(value: u64) -> Self { IndexValue::U64(value) diff --git a/crates/registry/src/utils/task.rs b/crates/registry/src/utils/task.rs index 1957d8f3..57740c20 100644 --- a/crates/registry/src/utils/task.rs +++ b/crates/registry/src/utils/task.rs @@ -4,7 +4,10 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ -use crate::schema::prelude::{Task, TaskStatus, TaskStatusPending, UTCDateTime}; +use crate::schema::{ + enums::Permission, + prelude::{Action, Task, TaskStatus, TaskStatusPending, UTCDateTime}, +}; impl Task { pub fn set_status(&mut self, status: TaskStatus) { @@ -60,6 +63,41 @@ impl Task { TaskStatus::Failed(_) => u64::MAX, } } + + pub fn permission(&self) -> Permission { + match self { + Task::IndexDocument(_) => Permission::TaskIndexDocument, + Task::UnindexDocument(_) => Permission::TaskUnindexDocument, + Task::IndexTrace(_) => Permission::TaskIndexTrace, + Task::CalendarAlarmEmail(_) => Permission::TaskCalendarAlarmEmail, + Task::CalendarAlarmNotification(_) => Permission::TaskCalendarAlarmNotification, + Task::CalendarItipMessage(_) => Permission::TaskCalendarItipMessage, + Task::MergeThreads(_) => Permission::TaskMergeThreads, + Task::DmarcReport(_) => Permission::TaskDmarcReport, + Task::TlsReport(_) => Permission::TaskTlsReport, + Task::RestoreArchivedItem(_) => Permission::TaskRestoreArchivedItem, + Task::DestroyAccount(_) => Permission::TaskDestroyAccount, + Task::AccountMaintenance(_) => Permission::TaskAccountMaintenance, + Task::StoreMaintenance(_) => Permission::TaskStoreMaintenance, + Task::SpamFilterMaintenance(_) => Permission::TaskSpamFilterMaintenance, + } + } +} + +impl Action { + pub fn permission(&self) -> Permission { + match self { + Action::ReloadSettings => Permission::ActionReloadSettings, + Action::ReloadTlsCertificates => Permission::ActionReloadTlsCertificates, + Action::ReloadLookupStores => Permission::ActionReloadLookupStores, + Action::ReloadBlockedIps => Permission::ActionReloadBlockedIps, + Action::TroubleshootDmarc(_) => Permission::ActionTroubleshootDmarc, + Action::ClassifySpam(_) => Permission::ActionClassifySpam, + Action::InvalidateCaches => Permission::ActionInvalidateCaches, + Action::PauseMtaQueue => Permission::ActionPauseMtaQueue, + Action::ResumeMtaQueue => Permission::ActionResumeMtaQueue, + } + } } impl TaskStatus { diff --git a/crates/services/src/state_manager/push.rs b/crates/services/src/state_manager/push.rs index c3021f7f..7bcd1942 100644 --- a/crates/services/src/state_manager/push.rs +++ b/crates/services/src/state_manager/push.rs @@ -30,8 +30,6 @@ pub fn spawn_push_manager(inner: Arc) -> mpsc::Sender { let (push_tx_, mut push_rx) = mpsc::channel::(IPC_CHANNEL_BUFFER); let push_tx = push_tx_.clone(); - tokio::spawn(async move {}); - tokio::spawn(async move { let mut push_servers: AHashMap = AHashMap::default(); let mut account_push_ids: AHashMap> = AHashMap::default(); @@ -44,85 +42,84 @@ pub fn spawn_push_manager(inner: Arc) -> mpsc::Sender { { let server = inner.build_server(); - match server - .document_ids( - u32::MAX, - Collection::Principal, - PrincipalField::PushSubscriptions, - ) - .await - { - Ok(account_ids) => { - for account_id in account_ids { - if server - .core - .network - .roles - .push_notifications - .is_enabled_for_integer(account_id as u64) - { - // Load push subscriptions for account - let (subscriptions, member_account_ids) = - match load_push_subscriptions(&server, account_id).await { - Ok(subscriptions) => subscriptions, - Err(err) => { - trc::error!(err.caused_by(trc::location!())); - continue; - } - }; - let current_time = now(); - for subscription in subscriptions - .subscriptions - .into_iter() - .filter(|s| s.verified && s.expires > current_time) + if server.core.network.roles.push_notifications { + match server + .document_ids( + u32::MAX, + Collection::Principal, + PrincipalField::PushSubscriptions, + ) + .await + { + Ok(account_ids) => { + for account_id in account_ids { + if server.core.jmap.push_total_shards <= 1 + || account_id % server.core.jmap.push_total_shards + == server.registry().cluster_push_shard() { - let id = Id::from_parts(subscription.id, account_id); - let subscription = Arc::new(subscription); + // Load push subscriptions for account + let (subscriptions, member_account_ids) = + match load_push_subscriptions(&server, account_id).await { + Ok(subscriptions) => subscriptions, + Err(err) => { + trc::error!(err.caused_by(trc::location!())); + continue; + } + }; + let current_time = now(); + for subscription in subscriptions + .subscriptions + .into_iter() + .filter(|s| s.verified && s.expires > current_time) + { + let id = Id::from_parts(subscription.id, account_id); + let subscription = Arc::new(subscription); - for account_id in &member_account_ids { - account_push_ids.entry(*account_id).or_default().insert(id); + for account_id in &member_account_ids { + account_push_ids.entry(*account_id).or_default().insert(id); + } + push_servers.insert( + id, + PushRegistration { + member_account_ids: member_account_ids.clone(), + num_attempts: 0, + last_request: Instant::now() + - (server.core.jmap.push_throttle + + Duration::from_millis(1)), + notifications: Vec::new(), + server: subscription.clone(), + in_flight: false, + }, + ); } - push_servers.insert( - id, - PushRegistration { - member_account_ids: member_account_ids.clone(), - num_attempts: 0, - last_request: Instant::now() - - (server.core.jmap.push_throttle - + Duration::from_millis(1)), - notifications: Vec::new(), - server: subscription.clone(), - in_flight: false, - }, - ); } } } + Err(err) => { + trc::error!(err.caused_by(trc::location!())); + } } - Err(err) => { - trc::error!(err.caused_by(trc::location!())); - } - } - // Subscribe to push events - if !account_push_ids.is_empty() - && server - .inner - .ipc - .push_tx - .clone() - .send(PushEvent::PushServerRegister { - activate: account_push_ids.keys().copied().collect(), - expired: vec![], - }) - .await - .is_err() - { - trc::event!( - Server(ServerEvent::ThreadError), - Details = "Error sending state change.", - CausedBy = trc::location!() - ); + // Subscribe to push events + if !account_push_ids.is_empty() + && server + .inner + .ipc + .push_tx + .clone() + .send(PushEvent::PushServerRegister { + activate: account_push_ids.keys().copied().collect(), + expired: vec![], + }) + .await + .is_err() + { + trc::event!( + Server(ServerEvent::ThreadError), + Details = "Error sending state change.", + CausedBy = trc::location!() + ); + } } } @@ -142,12 +139,9 @@ pub fn spawn_push_manager(inner: Arc) -> mpsc::Sender { match event_or_timeout { Ok(Some(event)) => match event { Event::Update { account_id } => { - if !server - .core - .network - .roles - .push_notifications - .is_enabled_for_integer(account_id as u64) + if server.core.jmap.push_total_shards > 1 + && account_id % server.core.jmap.push_total_shards + != server.registry().cluster_push_shard() { continue; } diff --git a/crates/services/src/task_manager/maintenance.rs b/crates/services/src/task_manager/maintenance.rs index f8c4b393..330eb226 100644 --- a/crates/services/src/task_manager/maintenance.rs +++ b/crates/services/src/task_manager/maintenance.rs @@ -185,11 +185,34 @@ async fn store_maintenance( ); } TaskStoreMaintenanceType::PurgeBlob => { - server - .store() - .purge_blobs(server.blob_store().clone()) - .await - .caused_by(trc::location!())?; + if let Some(shard_index) = task.shard_index { + server + .store() + .purge_blobs(server.blob_store().clone(), shard_index as u8) + .await + .caused_by(trc::location!())?; + } else { + let mut batch = BatchBuilder::new(); + let now = now() as i64; + for shard_index in 0..=u8::MAX { + batch.schedule_task(Task::StoreMaintenance(TaskStoreMaintenance { + maintenance_type: TaskStoreMaintenanceType::PurgeBlob, + shard_index: Some(shard_index as u64), + status: TaskStatus::at(now), + })); + + if batch.is_large_batch() { + server.core.storage.data.write(batch.build_all()).await?; + server.notify_task_queue(); + batch = BatchBuilder::new(); + } + } + + if !batch.is_empty() { + server.core.storage.data.write(batch.build_all()).await?; + server.notify_task_queue(); + } + } } TaskStoreMaintenanceType::RemoveGreylist | TaskStoreMaintenanceType::RemoveLockQueueMessage diff --git a/crates/services/src/task_manager/manager.rs b/crates/services/src/task_manager/manager.rs index 99fbedea..66552c85 100644 --- a/crates/services/src/task_manager/manager.rs +++ b/crates/services/src/task_manager/manager.rs @@ -46,15 +46,18 @@ use trc::TaskManagerEvent; use utils::snowflake::SnowflakeIdGenerator; pub fn spawn_task_manager(inner: Arc) { - if !inner - .build_server() - .core - .network - .roles - .task_manager - .is_enabled_or_sharded() { - return; + let server = inner.build_server(); + let roles = &server.core.network.roles; + + if !roles.account_maintenance + && !roles.store_maintenance + && !roles.search_indexing + && !roles.spam_training + && !roles.task_manager + { + return; + } } trc::event!(TaskManager(TaskManagerEvent::ManagerStarted)); @@ -364,28 +367,21 @@ impl TaskQueueManager for Server { let roles = &self.core.network.roles; for (task_job, task_type_idx) in tasks { let enabled = match task_job.typ { - TaskType::IndexDocument | TaskType::UnindexDocument | TaskType::IndexTrace => roles - .search_indexing - .is_enabled_for_integer(task_job.id_hash()), - TaskType::CalendarAlarmEmail | TaskType::CalendarAlarmNotification => roles - .calendar_alerts - .is_enabled_for_integer(task_job.id_hash()), - TaskType::CalendarItipMessage => roles - .imip_processing - .is_enabled_for_integer(task_job.id_hash()), - TaskType::MergeThreads => roles - .merge_threads - .is_enabled_for_integer(task_job.id_hash()), - TaskType::AccountMaintenance | TaskType::DestroyAccount => roles - .account_maintenance - .is_enabled_for_integer(task_job.id_hash()), - TaskType::StoreMaintenance => roles - .store_maintenance - .is_enabled_for_integer(task_job.id_hash()), - TaskType::SpamFilterMaintenance => roles - .spam_training - .is_enabled_for_integer(task_job.id_hash()), - TaskType::DmarcReport | TaskType::TlsReport | TaskType::RestoreArchivedItem => true, + TaskType::IndexDocument | TaskType::UnindexDocument | TaskType::IndexTrace => { + roles.search_indexing + } + TaskType::AccountMaintenance | TaskType::DestroyAccount => { + roles.account_maintenance + } + TaskType::StoreMaintenance => roles.store_maintenance, + TaskType::SpamFilterMaintenance => roles.spam_training, + TaskType::CalendarAlarmEmail + | TaskType::CalendarAlarmNotification + | TaskType::CalendarItipMessage + | TaskType::MergeThreads + | TaskType::DmarcReport + | TaskType::TlsReport + | TaskType::RestoreArchivedItem => true, }; if enabled { diff --git a/crates/services/src/task_manager/mod.rs b/crates/services/src/task_manager/mod.rs index db5e3f48..a67a3039 100644 --- a/crates/services/src/task_manager/mod.rs +++ b/crates/services/src/task_manager/mod.rs @@ -11,7 +11,6 @@ use registry::types::EnumImpl; use std::future::Future; use std::time::Instant; use store::write::Operation; -use store::xxhash_rust::xxh3::xxh3_64; use store::{ahash::AHashMap, write::now}; use tokio::sync::mpsc; use trc::TaskManagerEvent; @@ -113,10 +112,3 @@ impl TaskResult { } } } - -impl TaskJob { - #[inline(always)] - pub fn id_hash(&self) -> u64 { - xxh3_64(&self.id.to_le_bytes()) - } -} diff --git a/crates/services/src/task_manager/scheduler.rs b/crates/services/src/task_manager/scheduler.rs index c260e8de..632adb5e 100644 --- a/crates/services/src/task_manager/scheduler.rs +++ b/crates/services/src/task_manager/scheduler.rs @@ -168,7 +168,7 @@ pub fn spawn_task_scheduler(inner: Arc) { let server = inner.build_server(); let roles = &server.core.network.roles; - let mut batch = (roles.task_scheduler.is_enabled_or_sharded()).then(BatchBuilder::new); + let mut batch = (roles.task_scheduler).then(BatchBuilder::new); while let Some(event) = queue.pop() { match event.event { @@ -188,6 +188,7 @@ pub fn spawn_task_scheduler(inner: Arc) { batch.schedule_task(Task::StoreMaintenance(TaskStoreMaintenance { maintenance_type: TaskStoreMaintenanceType::PurgeAccounts, status: TaskStatus::now(), + shard_index: None, })); } } @@ -206,6 +207,7 @@ pub fn spawn_task_scheduler(inner: Arc) { batch.schedule_task(Task::StoreMaintenance(TaskStoreMaintenance { maintenance_type: TaskStoreMaintenanceType::PurgeData, status: TaskStatus::now(), + shard_index: None, })); } } @@ -224,6 +226,7 @@ pub fn spawn_task_scheduler(inner: Arc) { batch.schedule_task(Task::StoreMaintenance(TaskStoreMaintenance { maintenance_type: TaskStoreMaintenanceType::PurgeBlob, status: TaskStatus::now(), + shard_index: None, })); } } @@ -249,7 +252,7 @@ pub fn spawn_task_scheduler(inner: Arc) { if let Some(otel) = &server.core.metrics.otel { queue.schedule(Instant::now() + otel.interval, Event::OtelMetrics); - if roles.push_metrics.is_enabled_or_sharded() { + if roles.metrics_push { let otel = otel.clone(); // SPDX-SnippetBegin @@ -291,13 +294,7 @@ pub fn spawn_task_scheduler(inner: Arc) { let server = server.clone(); tokio::spawn(async move { let elapsed = Instant::now(); - if server - .core - .network - .roles - .calculate_metrics - .is_enabled_or_sharded() - { + if server.core.network.roles.metrics_calculate { // SPDX-SnippetBegin // SPDX-FileCopyrightText: 2020 Stalwart Labs LLC // SPDX-License-Identifier: LicenseRef-SEL diff --git a/crates/smtp/src/lib.rs b/crates/smtp/src/lib.rs index f0bd3d1b..957d4dce 100644 --- a/crates/smtp/src/lib.rs +++ b/crates/smtp/src/lib.rs @@ -37,14 +37,7 @@ impl StartQueueManager for BootManager { impl SpawnQueueManager for IpcReceivers { fn spawn_queue_manager(&mut self, inner: Arc) { - if inner - .build_server() - .core - .network - .roles - .outbound_mta - .is_enabled_or_sharded() - { + if inner.build_server().core.network.roles.outbound_mta { // Spawn queue manager self.queue_rx.take().unwrap().spawn(inner.clone()); diff --git a/crates/smtp/src/reporting/index.rs b/crates/smtp/src/reporting/index.rs index f35513b6..0dc5fbad 100644 --- a/crates/smtp/src/reporting/index.rs +++ b/crates/smtp/src/reporting/index.rs @@ -6,6 +6,7 @@ use registry::{ schema::{ + enums::DmarcActionDisposition, prelude::{ObjectType, Property}, structs::{ ArfExternalReport, DmarcExternalReport, DmarcInternalReport, Task, TaskDmarcReport, @@ -108,17 +109,23 @@ pub trait ExternalReportIndex: ObjectImpl { fn domains(&self) -> impl Iterator; + fn success_fail_count(&self) -> (u64, u64); + fn write_ops(&self, batch: &mut BatchBuilder, item_id: u64, is_set: bool) { let object_id = Self::OBJECT.to_id(); let mut index_builder = IndexBuilder::default(); for text in self.text() { - index_builder.text(Property::Domain, text); + index_builder.text(Property::Text, text); } if let Some(tenant_id) = self.tenant_id() { index_builder.search(Property::MemberTenantId, tenant_id.id()); } + let (success_count, fail_count) = self.success_fail_count(); + index_builder.search(Property::TotalSuccessfulSessions, success_count); + index_builder.search(Property::TotalFailedSessions, fail_count); + index_builder.search(Property::ExpiresAt, self.expires_at()); batch.registry_index(object_id, item_id, index_builder.keys.iter(), is_set); @@ -228,6 +235,10 @@ impl ExternalReportIndex for ArfExternalReport { fn expires_at(&self) -> u64 { self.expires_at.timestamp() as u64 } + + fn success_fail_count(&self) -> (u64, u64) { + (self.report.incidents, 0) + } } impl ExternalReportIndex for DmarcExternalReport { @@ -266,6 +277,21 @@ impl ExternalReportIndex for DmarcExternalReport { fn expires_at(&self) -> u64 { self.expires_at.timestamp() as u64 } + + fn success_fail_count(&self) -> (u64, u64) { + let mut success_count = 0; + let mut fail_count = 0; + + for record in self.report.records.iter() { + if record.evaluated_disposition == DmarcActionDisposition::Pass { + success_count += std::cmp::min(record.count, 1); + } else { + fail_count += std::cmp::min(record.count, 1); + } + } + + (success_count, fail_count) + } } impl ExternalReportIndex for TlsExternalReport { @@ -304,6 +330,18 @@ impl ExternalReportIndex for TlsExternalReport { fn expires_at(&self) -> u64 { self.expires_at.timestamp() as u64 } + + fn success_fail_count(&self) -> (u64, u64) { + let mut success_count = 0; + let mut fail_count = 0; + + for policy in self.report.policies.iter() { + success_count += std::cmp::min(policy.total_successful_sessions, 1); + fail_count += std::cmp::min(policy.total_failed_sessions, 1); + } + + (success_count, fail_count) + } } #[inline(always)] diff --git a/crates/store/src/build/registry.rs b/crates/store/src/build/registry.rs index f6285b7a..cfbf069c 100644 --- a/crates/store/src/build/registry.rs +++ b/crates/store/src/build/registry.rs @@ -165,8 +165,8 @@ impl RegistryStore { self.0.env_cluster_role.as_deref() } - pub fn cluster_role_shard(&self) -> u64 { - self.0.env_cluster_role_shard_id + pub fn cluster_push_shard(&self) -> u32 { + self.0.env_push_shard_id } pub fn local_hostname(&self) -> &str { diff --git a/crates/store/src/lib.rs b/crates/store/src/lib.rs index 4061e20d..e06329ea 100644 --- a/crates/store/src/lib.rs +++ b/crates/store/src/lib.rs @@ -211,7 +211,7 @@ pub struct RegistryStoreInner { pub(crate) env_recovery_mode: bool, pub(crate) env_recovery_admin: Option<(String, String)>, pub(crate) env_cluster_role: Option, - pub(crate) env_cluster_role_shard_id: u64, + pub(crate) env_push_shard_id: u32, pub(crate) env_hostname: String, pub(crate) id_generator: SnowflakeIdGenerator, } diff --git a/crates/store/src/registry/bootstrap.rs b/crates/store/src/registry/bootstrap.rs index fe1e3758..4f7dbf8e 100644 --- a/crates/store/src/registry/bootstrap.rs +++ b/crates/store/src/registry/bootstrap.rs @@ -37,16 +37,6 @@ impl Bootstrap { for role in bp.list_infallible::().await { if role.object.name == role_name { - if bp.registry.cluster_role_shard() >= role.object.shard_size { - bp.build_error( - ObjectType::ClusterRole.singleton(), - format!( - "Cluster role \"{role_name}\" has shard size of {}, which is smaller than the configured shard id {}.", - role.object.shard_size, - bp.registry.cluster_role_shard() - ), - ); - } bp.role = Some(role.object); return bp; } diff --git a/crates/store/src/registry/local.rs b/crates/store/src/registry/local.rs index 6b9e49fc..35c0b5bb 100644 --- a/crates/store/src/registry/local.rs +++ b/crates/store/src/registry/local.rs @@ -32,9 +32,9 @@ impl RegistryStoreInner { env_cluster_role: std::env::var("STALWART_ROLE") .ok() .filter(|r| !r.is_empty()), - env_cluster_role_shard_id: std::env::var("STALWART_ROLE_SHARD") + env_push_shard_id: std::env::var("STALWART_PUSH_SHARD") .ok() - .and_then(|id| id.parse::().ok().and_then(|v| v.checked_sub(1))) + .and_then(|id| id.parse::().ok().and_then(|v| v.checked_sub(1))) .unwrap_or(0), env_hostname: std::env::var("STALWART_HOSTNAME") .ok() diff --git a/crates/store/src/registry/mod.rs b/crates/store/src/registry/mod.rs index 132a2cf4..10009f8f 100644 --- a/crates/store/src/registry/mod.rs +++ b/crates/store/src/registry/mod.rs @@ -41,6 +41,7 @@ pub struct RegistryFilter { pub property: Property, pub op: RegistryFilterOp, pub value: RegistryFilterValue, + pub is_pk: bool, } #[derive(Debug, PartialEq, Clone, Copy, Eq, Hash)] @@ -61,6 +62,7 @@ pub enum RegistryFilterOp { pub enum RegistryFilterValue { String(String), + Bytes(Vec), U64(u64), U16(u16), Boolean(bool), diff --git a/crates/store/src/registry/query.rs b/crates/store/src/registry/query.rs index 0668db11..502ba688 100644 --- a/crates/store/src/registry/query.rs +++ b/crates/store/src/registry/query.rs @@ -5,7 +5,8 @@ */ use crate::{ - IterateParams, RegistryStore, SUBSPACE_REGISTRY_IDX, Store, U16_LEN, U64_LEN, ValueKey, + IterateParams, RegistryStore, SUBSPACE_REGISTRY_IDX, SUBSPACE_REGISTRY_PK, Store, U16_LEN, + U64_LEN, ValueKey, registry::{RegistryFilter, RegistryFilterOp, RegistryFilterValue, RegistryQuery}, write::{ AnyClass, RegistryClass, ValueClass, @@ -61,7 +62,7 @@ impl RegistryStore { Cow::Owned(word.to_lowercase()) }; - let result = range_to_set( + let result = index_range( &self.0.store, query.object_type, filter.property.to_id(), @@ -91,28 +92,42 @@ impl RegistryStore { .details("TextMatch operator only supports string values")); } } else { - let result = range_to_set( - &self.0.store, - query.object_type, - filter.property.to_id(), - match &filter.value { - RegistryFilterValue::String(v) => v.as_bytes(), - RegistryFilterValue::U64(v) => { - u64_buffer = v.to_be_bytes(); - &u64_buffer - } - RegistryFilterValue::U16(v) => { - u16_buffer = v.to_be_bytes(); - &u16_buffer - } - RegistryFilterValue::Boolean(v) => { - bool_buffer[0] = *v as u8; - &bool_buffer - } - }, - filter.op, - ) - .await?; + let value = match &filter.value { + RegistryFilterValue::String(v) => v.as_bytes(), + RegistryFilterValue::Bytes(v) => v.as_slice(), + RegistryFilterValue::U64(v) => { + u64_buffer = v.to_be_bytes(); + &u64_buffer + } + RegistryFilterValue::U16(v) => { + u16_buffer = v.to_be_bytes(); + &u16_buffer + } + RegistryFilterValue::Boolean(v) => { + bool_buffer[0] = *v as u8; + &bool_buffer + } + }; + + let result = if !filter.is_pk { + index_range( + &self.0.store, + query.object_type, + filter.property.to_id(), + value, + filter.op, + ) + .await? + } else { + pk_range( + &self.0.store, + query.object_type, + filter.property.to_id(), + value, + filter.op, + ) + .await? + }; if !results.has_items() { results = result; @@ -132,229 +147,122 @@ impl RegistryStore { pub async fn count(&self, query: RegistryQuery) -> trc::Result { self.query::>(query).await.map(|r| r.len()) } -} -pub trait RegistryQueryResults: Default + Sized + Sync + Send { - fn push(&mut self, id: u64); - fn has_items(&self) -> bool; - fn intersect(&mut self, other: &Self); -} - -impl RegistryQueryResults for AHashSet { - fn push(&mut self, id: u64) { - self.insert(id); - } - - fn has_items(&self) -> bool { - !self.is_empty() - } - - fn intersect(&mut self, other: &Self) { - self.retain(|id| other.contains(id)); - } -} - -impl RegistryQueryResults for RoaringBitmap { - fn push(&mut self, id: u64) { - self.insert(id as u32); - } - - fn has_items(&self) -> bool { - !self.is_empty() - } - - fn intersect(&mut self, other: &Self) { - self.bitand_assign(other); - } -} - -impl RegistryQuery { - pub fn new(object_type: ObjectType) -> Self { - Self { - object_type, - filters: Vec::new(), - } - } - - pub fn with_account(mut self, account_id: u32) -> Self { - if self.object_type.flags() & OBJ_FILTER_ACCOUNT != 0 { - let filter = RegistryFilter::equal(Property::AccountId, account_id); - if self.filters.is_empty() { - self.filters.push(filter); - } else { - self.filters.insert(0, filter); - } - } - self - } - - pub fn with_account_opt(self, account_id: Option) -> Self { - if let Some(account_id) = account_id { - self.with_account(account_id) - } else { - self - } - } - - pub fn with_tenant(mut self, tenant_id: Option) -> Self { - if let Some(tenant_id) = tenant_id - && self.object_type.flags() & OBJ_FILTER_TENANT != 0 - { - let filter = RegistryFilter::equal(Property::MemberTenantId, tenant_id); - if self.filters.is_empty() { - self.filters.push(filter); - } else { - self.filters.insert(0, filter); - } - } - self - } - - pub fn equal(mut self, property: Property, value: impl Into) -> Self { - self.filters.push(RegistryFilter::equal(property, value)); - self - } - - pub fn equal_opt( - mut self, + pub async fn sort_by_index( + &self, + object: ObjectType, property: Property, - value: Option>, - ) -> Self { - if let Some(value) = value { - self.filters.push(RegistryFilter::equal(property, value)); - } - self + mut ids: Option>, + ascending: bool, + ) -> trc::Result> { + let mut ids_sorted = Vec::with_capacity(ids.as_ref().map_or(0, |ids| ids.len())); + + let object_id = object.to_id(); + let index_id = property.to_id(); + let begin = ValueKey::from(ValueClass::Any(AnyClass { + subspace: SUBSPACE_REGISTRY_IDX, + key: KeySerializer::new(U16_LEN * 2) + .write(object_id) + .write(index_id) + .finalize(), + })); + let end = ValueKey::from(ValueClass::Any(AnyClass { + subspace: SUBSPACE_REGISTRY_IDX, + key: KeySerializer::new((U16_LEN * 2) + U64_LEN) + .write(object_id) + .write(index_id) + .write(u64::MAX) + .finalize(), + })); + + self.0 + .store + .iterate( + IterateParams::new(begin, end) + .no_values() + .set_ascending(ascending), + |key, _| { + let id = key.deserialize_be_u64(key.len() - U64_LEN)?; + if let Some(ids) = ids.as_mut() { + if ids.remove(&id) { + ids_sorted.push(id); + } + Ok(!ids.is_empty()) + } else { + ids_sorted.push(id); + Ok(true) + } + }, + ) + .await + .caused_by(trc::location!()) + .map(|_| { + if let Some(mut ids) = ids + && !ids.is_empty() + { + ids_sorted.extend(ids.drain()); + } + + ids_sorted + }) } - pub fn greater_than( - mut self, + pub async fn sort_by_pk( + &self, + object: ObjectType, property: Property, - value: impl Into, - ) -> Self { - self.filters - .push(RegistryFilter::greater_than(property, value)); - self - } + mut ids: Option>, + ascending: bool, + ) -> trc::Result> { + let mut ids_sorted = Vec::with_capacity(ids.as_ref().map_or(0, |ids| ids.len())); - pub fn less_than(mut self, property: Property, value: impl Into) -> Self { - self.filters - .push(RegistryFilter::less_than(property, value)); - self - } + let object_id = object.to_id(); + let index_id = property.to_id(); + let begin = ValueKey::from(ValueClass::Any(AnyClass { + subspace: SUBSPACE_REGISTRY_PK, + key: KeySerializer::new(U16_LEN * 2) + .write(object_id) + .write(index_id) + .finalize(), + })); + let end = ValueKey::from(ValueClass::Any(AnyClass { + subspace: SUBSPACE_REGISTRY_PK, + key: KeySerializer::new((U16_LEN * 2) + U64_LEN) + .write(object_id) + .write(index_id) + .write(u64::MAX) + .finalize(), + })); - pub fn greater_than_or_equal( - mut self, - property: Property, - value: impl Into, - ) -> Self { - self.filters - .push(RegistryFilter::greater_than_or_equal(property, value)); - self - } + self.0 + .store + .iterate( + IterateParams::new(begin, end).set_ascending(ascending), + |_, value| { + let id = value.deserialize_be_u64(U16_LEN)?; - pub fn less_than_or_equal( - mut self, - property: Property, - value: impl Into, - ) -> Self { - self.filters - .push(RegistryFilter::less_than_or_equal(property, value)); - self - } + if let Some(ids) = ids.as_mut() { + if ids.remove(&id) { + ids_sorted.push(id); + } + Ok(!ids.is_empty()) + } else { + ids_sorted.push(id); + Ok(true) + } + }, + ) + .await + .caused_by(trc::location!()) + .map(|_| { + if let Some(mut ids) = ids + && !ids.is_empty() + { + ids_sorted.extend(ids.drain()); + } - pub fn text(mut self, value: impl Into) -> Self { - self.filters.push(RegistryFilter::text(value)); - self - } - - pub fn text_opt(mut self, value: Option>) -> Self { - if let Some(value) = value { - self.filters.push(RegistryFilter::text(value)); - } - self - } -} - -impl RegistryFilter { - pub fn text(value: impl Into) -> Self { - Self { - property: Property::Contents, - op: RegistryFilterOp::TextMatch, - value: RegistryFilterValue::String(value.into()), - } - } - - pub fn equal(property: Property, value: impl Into) -> Self { - Self { - property, - op: RegistryFilterOp::Equal, - value: value.into(), - } - } - - pub fn greater_than(property: Property, value: impl Into) -> Self { - Self { - property, - op: RegistryFilterOp::GreaterThan, - value: value.into(), - } - } - - pub fn less_than(property: Property, value: impl Into) -> Self { - Self { - property, - op: RegistryFilterOp::LowerThan, - value: value.into(), - } - } - - pub fn greater_than_or_equal( - property: Property, - value: impl Into, - ) -> Self { - Self { - property, - op: RegistryFilterOp::GreaterEqualThan, - value: value.into(), - } - } - - pub fn less_than_or_equal(property: Property, value: impl Into) -> Self { - Self { - property, - op: RegistryFilterOp::LowerEqualThan, - value: value.into(), - } - } -} - -impl From for RegistryFilterValue { - fn from(value: String) -> Self { - RegistryFilterValue::String(value) - } -} - -impl From<&str> for RegistryFilterValue { - fn from(value: &str) -> Self { - RegistryFilterValue::String(value.to_string()) - } -} - -impl From for RegistryFilterValue { - fn from(value: u64) -> Self { - RegistryFilterValue::U64(value) - } -} - -impl From for RegistryFilterValue { - fn from(value: u32) -> Self { - RegistryFilterValue::U64(value as u64) - } -} - -impl From for RegistryFilterValue { - fn from(value: u16) -> Self { - RegistryFilterValue::U16(value) + ids_sorted + }) } } @@ -384,11 +292,11 @@ async fn all_ids(store: &Store, object: ObjectType) -> }, ) .await - .caused_by(trc::location!())?; - Ok(bm) + .caused_by(trc::location!()) + .map(|_| bm) } -async fn range_to_set( +async fn index_range( store: &Store, object: ObjectType, index_id: u16, @@ -474,3 +382,361 @@ async fn range_to_set( .caused_by(trc::location!()) .map(|_| bm) } + +async fn pk_range( + store: &Store, + object: ObjectType, + index_id: u16, + match_value: &[u8], + op: RegistryFilterOp, +) -> trc::Result { + let object_id = object.to_id(); + let ((from_value, from_index_id), (end_value, end_index_id)) = match op { + RegistryFilterOp::LowerThan => ((&[][..], object_id), (match_value, object_id)), + RegistryFilterOp::LowerEqualThan => ((&[][..], object_id), (match_value, object_id)), + RegistryFilterOp::GreaterThan => ((match_value, object_id), (&[][..], object_id + 1)), + RegistryFilterOp::GreaterEqualThan => ((match_value, object_id), (&[][..], object_id + 1)), + RegistryFilterOp::Equal | RegistryFilterOp::TextMatch => { + ((match_value, object_id), (match_value, object_id)) + } + }; + + let begin = ValueKey::from(ValueClass::Any(AnyClass { + subspace: SUBSPACE_REGISTRY_PK, + key: KeySerializer::new((U16_LEN * 2) + from_value.len()) + .write(object_id) + .write(from_index_id) + .write(from_value) + .finalize(), + })); + let end = ValueKey::from(ValueClass::Any(AnyClass { + subspace: SUBSPACE_REGISTRY_PK, + key: KeySerializer::new((U16_LEN * 2) + end_value.len()) + .write(object_id) + .write(end_index_id) + .write(end_value) + .finalize(), + })); + + let mut bm = T::default(); + let prefix = KeySerializer::new(U16_LEN * 2) + .write(object_id) + .write(index_id) + .finalize(); + let prefix_len = prefix.len(); + + store + .iterate(IterateParams::new(begin, end).ascending(), |key, _| { + if !key.starts_with(&prefix) { + return Ok(false); + } + + let id_pos = key.len() - U64_LEN; + let value = key + .get(prefix_len..id_pos) + .ok_or_else(|| trc::Error::corrupted_key(key, None, trc::location!()))?; + + let matches = match op { + RegistryFilterOp::LowerThan => value < match_value, + RegistryFilterOp::LowerEqualThan => value <= match_value, + RegistryFilterOp::GreaterThan => value > match_value, + RegistryFilterOp::GreaterEqualThan => value >= match_value, + RegistryFilterOp::Equal | RegistryFilterOp::TextMatch => value == match_value, + }; + + if matches { + bm.push(value.deserialize_be_u64(U16_LEN)?); + } + + Ok(true) + }) + .await + .caused_by(trc::location!()) + .map(|_| bm) +} + +pub trait RegistryQueryResults: Default + Sized + Sync + Send { + fn push(&mut self, id: u64); + fn has_items(&self) -> bool; + fn intersect(&mut self, other: &Self); +} + +impl RegistryQueryResults for AHashSet { + fn push(&mut self, id: u64) { + self.insert(id); + } + + fn has_items(&self) -> bool { + !self.is_empty() + } + + fn intersect(&mut self, other: &Self) { + self.retain(|id| other.contains(id)); + } +} + +impl RegistryQueryResults for RoaringBitmap { + fn push(&mut self, id: u64) { + self.insert(id as u32); + } + + fn has_items(&self) -> bool { + !self.is_empty() + } + + fn intersect(&mut self, other: &Self) { + self.bitand_assign(other); + } +} + +impl RegistryQuery { + pub fn new(object_type: ObjectType) -> Self { + Self { + object_type, + filters: Vec::new(), + } + } + + pub fn with_account(mut self, account_id: u32) -> Self { + if self.object_type.flags() & OBJ_FILTER_ACCOUNT != 0 { + let filter = RegistryFilter::equal(Property::AccountId, account_id, false); + if self.filters.is_empty() { + self.filters.push(filter); + } else { + self.filters.insert(0, filter); + } + } + self + } + + pub fn with_account_opt(self, account_id: Option) -> Self { + if let Some(account_id) = account_id { + self.with_account(account_id) + } else { + self + } + } + + pub fn with_tenant(mut self, tenant_id: Option) -> Self { + if let Some(tenant_id) = tenant_id + && self.object_type.flags() & OBJ_FILTER_TENANT != 0 + { + let filter = RegistryFilter::equal(Property::MemberTenantId, tenant_id, false); + if self.filters.is_empty() { + self.filters.push(filter); + } else { + self.filters.insert(0, filter); + } + } + self + } + + pub fn filter(mut self, filter: RegistryFilter) -> Self { + self.filters.push(filter); + self + } + + pub fn equal(mut self, property: Property, value: impl Into) -> Self { + self.filters + .push(RegistryFilter::equal(property, value, false)); + self + } + + pub fn equal_pk( + mut self, + property: Property, + value: impl Into, + is_pk: bool, + ) -> Self { + self.filters + .push(RegistryFilter::equal(property, value, is_pk)); + self + } + + pub fn push_equal_pk( + &mut self, + property: Property, + value: impl Into, + is_pk: bool, + ) { + self.filters + .push(RegistryFilter::equal(property, value, is_pk)); + } + + pub fn equal_opt( + mut self, + property: Property, + value: Option>, + ) -> Self { + if let Some(value) = value { + self.filters + .push(RegistryFilter::equal(property, value, false)); + } + self + } + + pub fn greater_than( + mut self, + property: Property, + value: impl Into, + ) -> Self { + self.filters + .push(RegistryFilter::greater_than(property, value, false)); + self + } + + pub fn less_than(mut self, property: Property, value: impl Into) -> Self { + self.filters + .push(RegistryFilter::less_than(property, value, false)); + self + } + + pub fn greater_than_or_equal( + mut self, + property: Property, + value: impl Into, + ) -> Self { + self.filters.push(RegistryFilter::greater_than_or_equal( + property, value, false, + )); + self + } + + pub fn less_than_or_equal( + mut self, + property: Property, + value: impl Into, + ) -> Self { + self.filters + .push(RegistryFilter::less_than_or_equal(property, value, false)); + self + } + + pub fn text(mut self, property: Property, value: impl Into) -> Self { + self.filters.push(RegistryFilter::text(property, value)); + self + } + + pub fn text_opt(mut self, property: Property, value: Option>) -> Self { + if let Some(value) = value { + self.filters.push(RegistryFilter::text(property, value)); + } + self + } + + pub fn push_text(&mut self, property: Property, value: impl Into) { + self.filters.push(RegistryFilter::text(property, value)); + } + + pub fn has_filters(&self) -> bool { + !self.filters.is_empty() + } +} + +impl RegistryFilter { + pub fn text(property: Property, value: impl Into) -> Self { + Self { + property, + op: RegistryFilterOp::TextMatch, + value: RegistryFilterValue::String(value.into()), + is_pk: false, + } + } + + pub fn equal(property: Property, value: impl Into, is_pk: bool) -> Self { + Self { + property, + op: RegistryFilterOp::Equal, + value: value.into(), + is_pk, + } + } + + pub fn greater_than( + property: Property, + value: impl Into, + is_pk: bool, + ) -> Self { + Self { + property, + op: RegistryFilterOp::GreaterThan, + value: value.into(), + is_pk, + } + } + + pub fn less_than( + property: Property, + value: impl Into, + is_pk: bool, + ) -> Self { + Self { + property, + op: RegistryFilterOp::LowerThan, + value: value.into(), + is_pk, + } + } + + pub fn greater_than_or_equal( + property: Property, + value: impl Into, + is_pk: bool, + ) -> Self { + Self { + property, + op: RegistryFilterOp::GreaterEqualThan, + value: value.into(), + is_pk, + } + } + + pub fn less_than_or_equal( + property: Property, + value: impl Into, + is_pk: bool, + ) -> Self { + Self { + property, + op: RegistryFilterOp::LowerEqualThan, + value: value.into(), + is_pk, + } + } +} + +impl From for RegistryFilterValue { + fn from(value: String) -> Self { + RegistryFilterValue::String(value) + } +} + +impl From<&str> for RegistryFilterValue { + fn from(value: &str) -> Self { + RegistryFilterValue::String(value.to_string()) + } +} + +impl From for RegistryFilterValue { + fn from(value: u64) -> Self { + RegistryFilterValue::U64(value) + } +} + +impl From for RegistryFilterValue { + fn from(value: u32) -> Self { + RegistryFilterValue::U64(value as u64) + } +} + +impl From for RegistryFilterValue { + fn from(value: u16) -> Self { + RegistryFilterValue::U16(value) + } +} + +impl From for RegistryFilterValue { + fn from(value: bool) -> Self { + RegistryFilterValue::Boolean(value) + } +} diff --git a/crates/store/src/write/blob.rs b/crates/store/src/write/blob.rs index 05af93b8..01e1b603 100644 --- a/crates/store/src/write/blob.rs +++ b/crates/store/src/write/blob.rs @@ -79,118 +79,117 @@ impl Store { self.get_value::<()>(key).await.map(|v| v.is_some()) } - pub async fn purge_blobs(&self, blob_store: BlobStore) -> trc::Result<()> { + pub async fn purge_blobs(&self, blob_store: BlobStore, shard_index: u8) -> trc::Result<()> { let mut total_active = 0; let mut total_deleted = 0; let started = Instant::now(); - for byte in 0..=u8::MAX { - // Validate linked blobs - let mut from_hash = BlobHash::default(); - let mut to_hash = BlobHash::new_max(); - from_hash.0[0] = byte; - to_hash.0[0] = byte; - let from_key = ValueKey { - account_id: 0, - collection: 0, - document_id: 0, - class: ValueClass::Blob(BlobOp::Commit { hash: from_hash }), - }; - let to_key = ValueKey { - account_id: u32::MAX, - collection: u8::MAX, - document_id: u32::MAX, - class: ValueClass::Blob(BlobOp::Link { - hash: to_hash, - to: BlobLink::Document, - }), - }; + // Validate linked blobs + let mut from_hash = BlobHash::default(); + let mut to_hash = BlobHash::new_max(); + from_hash.0[0] = shard_index; + to_hash.0[0] = shard_index; + let from_key = ValueKey { + account_id: 0, + collection: 0, + document_id: 0, + class: ValueClass::Blob(BlobOp::Commit { hash: from_hash }), + }; + let to_key = ValueKey { + account_id: u32::MAX, + collection: u8::MAX, + document_id: u32::MAX, + class: ValueClass::Blob(BlobOp::Link { + hash: to_hash, + to: BlobLink::Document, + }), + }; - let mut state = BlobPurgeState::new(); - self.iterate( - IterateParams::new(from_key, to_key).ascending(), - |key, value| { - let hash = - BlobHash::try_from_hash_slice(key.get(0..BLOB_HASH_LEN).ok_or_else( - || trc::Error::corrupted_key(key, value.into(), trc::location!()), - )?) - .unwrap(); + let mut state = BlobPurgeState::new(); + self.iterate( + IterateParams::new(from_key, to_key).ascending(), + |key, value| { + let hash = + BlobHash::try_from_hash_slice(key.get(0..BLOB_HASH_LEN).ok_or_else(|| { + trc::Error::corrupted_key(key, value.into(), trc::location!()) + })?) + .unwrap(); - state.update_hash(hash); - state.process_key(key, value)?; + state.update_hash(hash); + state.process_key(key, value)?; - Ok(true) - }, - ) - .await - .caused_by(trc::location!())?; + Ok(true) + }, + ) + .await + .caused_by(trc::location!())?; - state.finalize(BlobHash::default()); + state.finalize(BlobHash::default()); - // Delete expired or unlinked blobs - for (_, op) in &state.delete_keys { - if let BlobOp::Commit { hash } = op { - blob_store - .delete_blob(hash.as_ref()) - .await - .caused_by(trc::location!())?; - } - } - - // Delete hashes - let mut batch = BatchBuilder::new(); - for (account_id, op) in state.delete_keys { - if batch.is_large_batch() { - self.write(batch.build_all()) - .await - .caused_by(trc::location!())?; - batch = BatchBuilder::new(); - } - - if let Some(account_id) = account_id { - batch.with_account_id(account_id); - } - - batch.any_op(Operation::Value { - class: ValueClass::Blob(op), - op: ValueOp::Clear, - }); - } - for (account_id, object_id) in state.delete_registry { - if batch.is_large_batch() { - self.write(batch.build_all()) - .await - .caused_by(trc::location!())?; - batch = BatchBuilder::new(); - } - - let item_id = object_id.id().id(); - let object_id = object_id.object().to_id(); - - batch - .clear(ValueClass::Registry(RegistryClass::Index { - index_id: Property::AccountId.to_id(), - object_id, - item_id, - key: (account_id as u64).serialize(), - })) - .clear(ValueClass::Registry(RegistryClass::Item { - object_id, - item_id, - })); - } - if !batch.is_empty() { - self.write(batch.build_all()) + // Delete expired or unlinked blobs + for (_, op) in &state.delete_keys { + if let BlobOp::Commit { hash } = op { + blob_store + .delete_blob(hash.as_ref()) .await .caused_by(trc::location!())?; } - - total_active += state.total_active - 1; // Exclude default hash - total_deleted += state.total_deleted; } + // Delete hashes + let mut batch = BatchBuilder::new(); + for (account_id, op) in state.delete_keys { + if batch.is_large_batch() { + self.write(batch.build_all()) + .await + .caused_by(trc::location!())?; + batch = BatchBuilder::new(); + } + + if let Some(account_id) = account_id { + batch.with_account_id(account_id); + } + + batch.any_op(Operation::Value { + class: ValueClass::Blob(op), + op: ValueOp::Clear, + }); + } + for (account_id, object_id) in state.delete_registry { + if batch.is_large_batch() { + self.write(batch.build_all()) + .await + .caused_by(trc::location!())?; + batch = BatchBuilder::new(); + } + + let item_id = object_id.id().id(); + let object_id = object_id.object().to_id(); + + batch + .clear(ValueClass::Registry(RegistryClass::Index { + index_id: Property::AccountId.to_id(), + object_id, + item_id, + key: (account_id as u64).serialize(), + })) + .clear(ValueClass::Registry(RegistryClass::Item { + object_id, + item_id, + })); + } + if !batch.is_empty() { + self.write(batch.build_all()) + .await + .caused_by(trc::location!())?; + } + + total_active += state.total_active - 1; // Exclude default hash + total_deleted += state.total_deleted; + trc::event!( Store(StoreEvent::BlobStorePurged), + Id = shard_index as u16, Expires = total_deleted, Total = total_active, Elapsed = started.elapsed()