From 45af5efa4c1f16e3ad0d9be48fde77ab873bd5be Mon Sep 17 00:00:00 2001 From: mdecimus <11444311+mdecimus@users.noreply.github.com> Date: Wed, 11 Mar 2026 16:33:49 +0100 Subject: [PATCH] JMAP Registry API implementation - part 13 --- Cargo.lock | 18 +- crates/common/src/auth/permissions.rs | 84 +++- crates/common/src/cache/principals.rs | 15 +- crates/common/src/config/smtp/queue.rs | 2 + crates/common/src/manager/boot.rs | 105 +--- crates/common/src/manager/defaults.rs | 459 ++++++++++++++++++ crates/common/src/manager/mod.rs | 1 + crates/common/src/network/dkim.rs | 49 ++ crates/common/src/network/mod.rs | 1 + crates/common/src/sharing/acl.rs | 12 +- crates/common/src/storage/mod.rs | 10 +- crates/coordinator/src/bootstrap.rs | 4 +- crates/dav/src/common/acl.rs | 4 +- crates/http/Cargo.toml | 1 - crates/http/src/auth/oauth/registration.rs | 10 +- crates/jmap/Cargo.toml | 1 - crates/jmap/src/addressbook/set.rs | 7 +- crates/jmap/src/calendar/set.rs | 7 +- crates/jmap/src/file/set.rs | 7 +- crates/jmap/src/mailbox/set.rs | 3 +- crates/jmap/src/registry/get.rs | 14 +- crates/jmap/src/registry/mapping/account.rs | 122 ++++- .../src/registry/mapping/archived_item.rs | 94 +++- crates/jmap/src/registry/mapping/dkim.rs | 47 +- crates/jmap/src/registry/mapping/log.rs | 400 ++++++++++++--- .../jmap/src/registry/mapping/masked_email.rs | 12 +- crates/jmap/src/registry/mapping/principal.rs | 25 +- .../jmap/src/registry/mapping/public_key.rs | 9 +- .../src/registry/mapping/queued_message.rs | 259 +++++++++- crates/jmap/src/registry/mapping/report.rs | 67 +-- .../jmap/src/registry/mapping/spam_sample.rs | 95 +++- crates/jmap/src/registry/mapping/task.rs | 138 +++++- crates/jmap/src/registry/mapping/telemetry.rs | 261 +++++++++- crates/jmap/src/registry/query.rs | 89 +++- crates/jmap/src/registry/set.rs | 6 +- crates/migration/src/lib.rs | 2 +- .../src/task_manager/destroy_account.rs | 95 +++- .../services/src/task_manager/maintenance.rs | 69 ++- crates/smtp/src/queue/spool.rs | 7 +- crates/store/Cargo.toml | 1 + crates/store/src/registry/local.rs | 3 +- crates/store/src/registry/mod.rs | 14 +- crates/store/src/registry/query.rs | 300 +++++++++--- crates/store/src/registry/write.rs | 20 +- 44 files changed, 2476 insertions(+), 473 deletions(-) create mode 100644 crates/common/src/manager/defaults.rs create mode 100644 crates/common/src/network/dkim.rs diff --git a/Cargo.lock b/Cargo.lock index 29da8ef2..0308b7c1 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2920,7 +2920,6 @@ dependencies = [ "pkcs8", "quick-xml 0.38.4", "registry", - "rev_lines", "rkyv", "rsa", "serde", @@ -3589,7 +3588,6 @@ dependencies = [ "rand 0.9.2", "registry", "reqwest", - "rev_lines", "rkyv", "rsa", "serde", @@ -5675,6 +5673,12 @@ dependencies = [ "nibble_vec", ] +[[package]] +name = "radsort" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "019b4b213425016d7d84a153c4c73afb0946fbb4840e4eece7ba8848b9d6da22" + [[package]] name = "rancor" version = "0.1.1" @@ -6092,15 +6096,6 @@ version = "0.7.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1e061d1b48cb8d38042de4ae0a7a6401009d6143dc80d2e2d6f31f0bdd6470c7" -[[package]] -name = "rev_lines" -version = "0.3.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ed62916ac7a5ccbf13fa5e1d303029ff015600fee841756dfc134a1ac62bf05f" -dependencies = [ - "thiserror 1.0.69", -] - [[package]] name = "rfc6979" version = "0.4.0" @@ -7431,6 +7426,7 @@ dependencies = [ "num_cpus", "parking_lot", "r2d2", + "radsort", "rand 0.9.2", "rayon", "redis", diff --git a/crates/common/src/auth/permissions.rs b/crates/common/src/auth/permissions.rs index 8341fcd4..a65e6c13 100644 --- a/crates/common/src/auth/permissions.rs +++ b/crates/common/src/auth/permissions.rs @@ -169,6 +169,13 @@ pub(crate) fn build_permissions_list(permissions_in: &Permissions) -> Vec, + pub group: Vec, + pub tenant: Vec, + pub superuser: Vec, +} + impl PermissionsGroup { pub fn with_merge(mut self, merge: bool) -> Self { self.merge = merge; @@ -197,13 +204,8 @@ impl PermissionsGroup { } pub fn user() -> Self { - let todo = "fix"; let mut permissions = PermissionsGroup::default(); - for permission in [ - Permission::Authenticate, - Permission::JmapParticipantIdentityGet, - Permission::JmapParticipantIdentityChanges, - ] { + for permission in DefaultPermissions::default().user { permissions.enabled.set(permission as usize); } @@ -211,6 +213,76 @@ impl PermissionsGroup { } } +impl Default for DefaultPermissions { + fn default() -> Self { + let mut default = Self { + user: Default::default(), + group: Default::default(), + tenant: Default::default(), + superuser: Default::default(), + }; + + for permission_id in 0..Permission::COUNT { + let permission = Permission::from_id(permission_id as u16).unwrap(); + match permission { + Permission::Authenticate + | Permission::AuthenticateWithAlias + | Permission::InteractAi => { + default.user.push(permission); + } + Permission::Impersonate + | Permission::UnlimitedRequests + | Permission::UnlimitedUploads + | Permission::LiveMetrics + | Permission::LiveTracing => { + default.superuser.push(permission); + } + Permission::FetchAnyBlob | Permission::LiveDeliveryTest => { + default.superuser.push(permission); + default.tenant.push(permission); + } + permission => { + let name = permission.as_str(); + if name.starts_with("jmap") + || name.starts_with("imap") + || name.starts_with("pop3") + || name.starts_with("calendar") + || name.starts_with("email") + || name.starts_with("dav") + || name.starts_with("sieve") + || name.starts_with("sysMaskedEmail") + || name.starts_with("sysArchivedItem") + || name.starts_with("sysAccountSettings") + || name.starts_with("sysPublicKey") + || name.starts_with("sysSpamTrainingSample") + { + default.user.push(permission); + default.group.push(permission); + } else if name.starts_with("sysCredential") { + default.user.push(permission); + } else if name.starts_with("sysDomain") + || name.starts_with("sysDkimSignature") + || name.starts_with("sysAccount") + || name.starts_with("sysRole") + || name.starts_with("sysOAuthClient") + || name.starts_with("sysMailingList") + || name.starts_with("sysExternalReport") + || name.starts_with("sysDnsServer") + || name.starts_with("sysQueuedMessage") + { + default.tenant.push(permission); + default.superuser.push(permission); + } else { + default.superuser.push(permission); + } + } + } + } + + default + } +} + impl From for PermissionsGroup { fn from(value: PermissionsList) -> Self { Self::from(&value) diff --git a/crates/common/src/cache/principals.rs b/crates/common/src/cache/principals.rs index 11575b67..80bc8189 100644 --- a/crates/common/src/cache/principals.rs +++ b/crates/common/src/cache/principals.rs @@ -22,7 +22,6 @@ use crate::{ encryption::{EncryptionMethod, parse_public_key}, }, }; -use ahash::AHashSet; use arcstr::ArcStr; use registry::{ schema::{ @@ -60,7 +59,7 @@ impl Server { if domain_names_negative.get(domain).is_none() { if let Some(domain_id) = self .registry() - .query::>( + .query::>( RegistryQuery::new(ObjectType::Domain).equal(Property::Name, domain), ) .await? @@ -68,7 +67,7 @@ impl Server { .next() { // Cache positive result - let domain_id = domain_id as u32; + let domain_id = domain_id.document_id(); let domain = self.domain_by_id(domain_id).await?; if let Some(domain) = &domain { for name in domain.names.iter() { @@ -664,20 +663,20 @@ impl Server { Err(guard) => { let ids = self .registry() - .query::>( + .query::>( RegistryQuery::new(ObjectType::DkimSignature) .equal(Property::DomainId, domain.id), ) .await?; let mut signatures = Vec::with_capacity(ids.len()); for id in ids { - if let Some(signature) = - self.registry().object::(id.into()).await? - { + if let Some(signature) = self.registry().object::(id).await? { match DkimSigner::new(domain.names[0].to_string(), signature).await { Ok(signer) => signatures.push(signer), Err(err) => { - trc::error!(err.ctx(trc::Key::Id, id).caused_by(trc::location!())); + trc::error!( + err.ctx(trc::Key::Id, id.id()).caused_by(trc::location!()) + ); } } } diff --git a/crates/common/src/config/smtp/queue.rs b/crates/common/src/config/smtp/queue.rs index ca3361ba..99925f23 100644 --- a/crates/common/src/config/smtp/queue.rs +++ b/crates/common/src/config/smtp/queue.rs @@ -273,6 +273,8 @@ impl QueueConfig { 30 * 60, 60 * 60, 2 * 60 * 60, + 24 * 60 * 60, + 3 * 24 * 60 * 60, ], MtaDeliveryScheduleIntervalsOrDefault::Custom(intervals) => intervals .intervals diff --git a/crates/common/src/manager/boot.rs b/crates/common/src/manager/boot.rs index 380715e0..07858c89 100644 --- a/crates/common/src/manager/boot.rs +++ b/crates/common/src/manager/boot.rs @@ -11,6 +11,7 @@ use crate::{ network::AsnGeoLookupConfig, server::Listeners, storage::Storage, telemetry::Telemetry, }, ipc::{BroadcastEvent, PushEvent, QueueEvent, ReportingEvent, TrainTaskController}, + manager::defaults::BootstrapDefaults, }; use arc_swap::ArcSwap; use pwhash::sha512_crypt; @@ -67,107 +68,6 @@ enum StoreOp { None, } -pub const DEFAULT_SETTINGS: &[(&str, &str)] = &[ - ("oauth.key", "abc"), - ("queue.quota.size.messages", "100000"), - ("queue.quota.size.size", "10737418240"), - ("queue.quota.size.enable", "true"), - ("queue.limiter.inbound.ip.key", "remote_ip"), - ("queue.limiter.inbound.ip.rate", "5/1s"), - ("queue.limiter.inbound.ip.enable", "true"), - ("queue.limiter.inbound.sender.key.0", "sender_domain"), - ("queue.limiter.inbound.sender.key.1", "rcpt"), - ("queue.limiter.inbound.sender.rate", "25/1h"), - ("queue.limiter.inbound.sender.enable", "true"), - ("report.analysis.addresses", "postmaster@*"), - ("queue.virtual.local.threads-per-node", "25"), - ("queue.virtual.local.description", "Local delivery queue"), - ("queue.virtual.remote.threads-per-node", "50"), - ("queue.virtual.remote.description", "Remote delivery queue"), - ("queue.virtual.dsn.threads-per-node", "5"), - ( - "queue.virtual.dsn.description", - "Delivery Status Notification delivery queue", - ), - ("queue.virtual.report.threads-per-node", "5"), - ( - "queue.virtual.report.description", - "DMARC and TLS report delivery queue", - ), - ("queue.schedule.local.queue-name", "local"), - ("queue.schedule.local.retry.0", "2m"), - ("queue.schedule.local.retry.1", "5m"), - ("queue.schedule.local.retry.2", "10m"), - ("queue.schedule.local.retry.3", "15m"), - ("queue.schedule.local.retry.4", "30m"), - ("queue.schedule.local.retry.5", "1h"), - ("queue.schedule.local.retry.6", "2h"), - ("queue.schedule.local.notify.0", "1d"), - ("queue.schedule.local.notify.1", "3d"), - ("queue.schedule.local.expire-type", "ttl"), - ("queue.schedule.local.expire", "3d"), - ( - "queue.schedule.local.description", - "Local delivery schedule", - ), - ("queue.schedule.remote.queue-name", "remote"), - ("queue.schedule.remote.retry.0", "2m"), - ("queue.schedule.remote.retry.1", "5m"), - ("queue.schedule.remote.retry.2", "10m"), - ("queue.schedule.remote.retry.3", "15m"), - ("queue.schedule.remote.retry.4", "30m"), - ("queue.schedule.remote.retry.5", "1h"), - ("queue.schedule.remote.retry.6", "2h"), - ("queue.schedule.remote.notify.0", "1d"), - ("queue.schedule.remote.notify.1", "3d"), - ("queue.schedule.remote.expire-type", "ttl"), - ("queue.schedule.remote.expire", "3d"), - ( - "queue.schedule.remote.description", - "Remote delivery schedule", - ), - ("queue.schedule.dsn.queue-name", "dsn"), - ("queue.schedule.dsn.retry.0", "15m"), - ("queue.schedule.dsn.retry.1", "30m"), - ("queue.schedule.dsn.retry.2", "1h"), - ("queue.schedule.dsn.retry.3", "2h"), - ("queue.schedule.dsn.expire-type", "attempts"), - ("queue.schedule.dsn.max-attempts", "10"), - ( - "queue.schedule.dsn.description", - "Delivery Status Notification delivery schedule", - ), - ("queue.schedule.report.queue-name", "report"), - ("queue.schedule.report.retry.0", "30m"), - ("queue.schedule.report.retry.1", "1h"), - ("queue.schedule.report.retry.2", "2h"), - ("queue.schedule.report.expire-type", "attempts"), - ("queue.schedule.report.max-attempts", "8"), - ( - "queue.schedule.report.description", - "DMARC and TLS report delivery schedule", - ), - ("queue.tls.invalid-tls.allow-invalid-certs", "true"), - ( - "queue.tls.invalid-tls.description", - "Allow invalid TLS certificates", - ), - ("queue.tls.default.allow-invalid-certs", "false"), - ("queue.tls.default.description", "Default TLS settings"), - ("queue.route.local.type", "local"), - ("queue.route.local.description", "Local delivery route"), - ("queue.route.mx.type", "mx"), - ("queue.route.mx.limits.multihomed", "2"), - ("queue.route.mx.limits.mx", "5"), - ("queue.route.mx.ip-lookup", "ipv4_then_ipv6"), - ("queue.route.mx.description", "MX delivery route"), - ("queue.connection.default.timeout.connect", "5m"), - ( - "queue.connection.default.description", - "Default connection settings", - ), -]; - impl BootManager { pub async fn init() -> Self { let mut config_path = std::env::var("CONFIG_PATH").ok(); @@ -250,7 +150,8 @@ impl BootManager { match import_export { StoreOp::None => { - let todo = "add default settings, hostname, download filter rules, webadmin"; + // Add safe defaults if missing + bootstrap.insert_safe_defaults().await; // Parse components let core = Box::pin(Core::parse(&mut bootstrap, storage)).await; diff --git a/crates/common/src/manager/defaults.rs b/crates/common/src/manager/defaults.rs new file mode 100644 index 00000000..d031797d --- /dev/null +++ b/crates/common/src/manager/defaults.rs @@ -0,0 +1,459 @@ +/* + * SPDX-FileCopyrightText: 2020 Stalwart Labs LLC + * + * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL + */ + +use crate::{auth::permissions::DefaultPermissions, network::dkim::generate_dkim_private_key}; +use registry::{ + schema::{ + enums::{DkimSignatureType, MtaInboundThrottleKey, MtaIpStrategy}, + prelude::ObjectType, + structs::{ + Authentication, Dkim1Signature, DkimPrivateKey, DkimSignature, Domain, + MtaConnectionStrategy, MtaDeliveryExpiration, MtaDeliveryExpirationAttempts, + MtaDeliveryExpirationTtl, MtaDeliverySchedule, MtaDeliveryScheduleInterval, + MtaDeliveryScheduleIntervals, MtaDeliveryScheduleIntervalsOrDefault, + MtaInboundThrottle, MtaQueueQuota, MtaRoute, MtaRouteCommon, MtaRouteMx, + MtaTlsStrategy, MtaVirtualQueue, OidcProvider, Rate, Role, SecretKey, SecretKeyValue, + SecretText, SecretTextValue, SystemSettings, + }, + }, + types::{duration::Duration, error::Error, list::List, map::Map}, +}; +use store::{ + rand::{Rng, distr::Alphanumeric, rng}, + registry::{ + bootstrap::Bootstrap, + write::{RegistryWrite, RegistryWriteResult}, + }, + write::now, +}; +use types::id::Id; + +pub trait BootstrapDefaults { + fn insert_safe_defaults(&mut self) -> impl Future + Send; +} + +impl BootstrapDefaults for Bootstrap { + async fn insert_safe_defaults(&mut self) { + if let Err(error) = insert_safe_defaults(self).await { + self.errors.push(Error::Internal { + object_id: None, + error, + }); + } + } +} + +async fn insert_safe_defaults(bp: &mut Bootstrap) -> trc::Result<()> { + if bp.registry.count_object(ObjectType::MtaQueueQuota).await? == 0 { + bp.registry + .write(RegistryWrite::insert( + &MtaQueueQuota { + description: "Global queue quota".to_string().into(), + enable: true, + messages: 100000.into(), + size: 10737418240.into(), + ..Default::default() + } + .into(), + )) + .await?; + } + + if bp + .registry + .count_object(ObjectType::MtaInboundThrottle) + .await? + == 0 + { + for object in [ + MtaInboundThrottle { + description: "Sender IP throttle".to_string().into(), + enable: true, + key: Map::new(vec![MtaInboundThrottleKey::RemoteIp]), + rate: Rate { + count: 5, + period: Duration::from_millis(1000), + }, + ..Default::default() + }, + MtaInboundThrottle { + description: "Sender address to recipient throttle".to_string().into(), + enable: true, + key: Map::new(vec![ + MtaInboundThrottleKey::SenderDomain, + MtaInboundThrottleKey::Rcpt, + ]), + rate: Rate { + count: 25, + period: Duration::from_millis(60 * 60 * 1000), + }, + ..Default::default() + }, + ] { + bp.registry + .write(RegistryWrite::insert(&object.into())) + .await?; + } + } + + if bp + .registry + .count_object(ObjectType::MtaVirtualQueue) + .await? + == 0 + && bp + .registry + .count_object(ObjectType::MtaDeliverySchedule) + .await? + == 0 + { + for (id, object) in [ + MtaVirtualQueue { + description: "Local delivery queue".to_string().into(), + name: "local".into(), + threads_per_node: 25, + }, + MtaVirtualQueue { + description: "Remote delivery queue".to_string().into(), + name: "remote".into(), + threads_per_node: 50, + }, + MtaVirtualQueue { + description: "Delivery Status Notification delivery queue" + .to_string() + .into(), + name: "dsn".into(), + threads_per_node: 5, + }, + MtaVirtualQueue { + description: "DMARC and TLS report delivery queue".to_string().into(), + name: "report".into(), + threads_per_node: 5, + }, + ] + .into_iter() + .enumerate() + { + bp.registry + .write(RegistryWrite::insert_with_id( + (id as u64).into(), + &object.into(), + )) + .await?; + } + + for (id, object) in [ + MtaDeliverySchedule { + name: "local".into(), + description: "Local delivery schedule".to_string().into(), + expiry: MtaDeliveryExpiration::Ttl(MtaDeliveryExpirationTtl { + expire: Duration::from_millis(3 * 24 * 60 * 60 * 1000), + }), + notify: MtaDeliveryScheduleIntervalsOrDefault::Default, + retry: MtaDeliveryScheduleIntervalsOrDefault::Default, + queue_id: 0u64.into(), + }, + MtaDeliverySchedule { + name: "remote".into(), + description: "Remote delivery schedule".to_string().into(), + expiry: MtaDeliveryExpiration::Ttl(MtaDeliveryExpirationTtl { + expire: Duration::from_millis(3 * 24 * 60 * 60 * 1000), + }), + notify: MtaDeliveryScheduleIntervalsOrDefault::Default, + retry: MtaDeliveryScheduleIntervalsOrDefault::Default, + queue_id: 1u64.into(), + }, + MtaDeliverySchedule { + name: "dsn".into(), + description: "Delivery Status Notification delivery schedule" + .to_string() + .into(), + expiry: MtaDeliveryExpiration::Attempts(MtaDeliveryExpirationAttempts { + max_attempts: 10, + }), + notify: MtaDeliveryScheduleIntervalsOrDefault::Default, + retry: MtaDeliveryScheduleIntervalsOrDefault::Custom( + MtaDeliveryScheduleIntervals { + intervals: List::from_iter([ + MtaDeliveryScheduleInterval { + duration: Duration::from_millis(15 * 60 * 1000), + }, + MtaDeliveryScheduleInterval { + duration: Duration::from_millis(30 * 60 * 1000), + }, + MtaDeliveryScheduleInterval { + duration: Duration::from_millis(60 * 60 * 1000), + }, + MtaDeliveryScheduleInterval { + duration: Duration::from_millis(2 * 60 * 60 * 1000), + }, + ]), + }, + ), + queue_id: 2u64.into(), + }, + MtaDeliverySchedule { + name: "report".into(), + description: "DMARC and TLS report delivery schedule".to_string().into(), + expiry: MtaDeliveryExpiration::Attempts(MtaDeliveryExpirationAttempts { + max_attempts: 8, + }), + notify: MtaDeliveryScheduleIntervalsOrDefault::Custom(Default::default()), + retry: MtaDeliveryScheduleIntervalsOrDefault::Custom( + MtaDeliveryScheduleIntervals { + intervals: List::from_iter([ + MtaDeliveryScheduleInterval { + duration: Duration::from_millis(30 * 60 * 1000), + }, + MtaDeliveryScheduleInterval { + duration: Duration::from_millis(60 * 60 * 1000), + }, + MtaDeliveryScheduleInterval { + duration: Duration::from_millis(2 * 60 * 60 * 1000), + }, + ]), + }, + ), + queue_id: 3u64.into(), + }, + ] + .into_iter() + .enumerate() + { + bp.registry + .write(RegistryWrite::insert_with_id( + (id as u64).into(), + &object.into(), + )) + .await?; + } + } + + if bp.registry.count_object(ObjectType::MtaTlsStrategy).await? == 0 { + for object in [ + MtaTlsStrategy { + name: "invalid-tls".into(), + description: "Allow invalid TLS certificates".to_string().into(), + allow_invalid_certs: true, + ..Default::default() + }, + MtaTlsStrategy { + name: "default".into(), + description: "Default TLS settings".to_string().into(), + allow_invalid_certs: false, + ..Default::default() + }, + ] { + bp.registry + .write(RegistryWrite::insert(&object.into())) + .await?; + } + } + + if bp.registry.count_object(ObjectType::MtaRoute).await? == 0 { + for object in [ + MtaRoute::Mx(MtaRouteMx { + description: "MX delivery route".to_string().into(), + ip_lookup_strategy: MtaIpStrategy::V4ThenV6, + max_multihomed: 2, + max_mx_hosts: 2, + name: "default".into(), + }), + MtaRoute::Local(MtaRouteCommon { + description: "Local delivery route".to_string().into(), + name: "local".into(), + }), + ] { + bp.registry + .write(RegistryWrite::insert(&object.into())) + .await?; + } + } + + if bp + .registry + .count_object(ObjectType::MtaConnectionStrategy) + .await? + == 0 + { + bp.registry + .write(RegistryWrite::insert( + &MtaConnectionStrategy { + name: "default".into(), + description: "Default connection strategy".to_string().into(), + ..Default::default() + } + .into(), + )) + .await?; + } + + if bp.registry.count_object(ObjectType::OidcProvider).await? == 0 { + bp.registry + .write(RegistryWrite::insert( + &OidcProvider { + encryption_key: SecretKey::Value(SecretKeyValue { + secret: rng() + .sample_iter(Alphanumeric) + .take(64) + .map(char::from) + .collect::(), + }), + signature_key: SecretText::Text(SecretTextValue { + secret: rng() + .sample_iter(Alphanumeric) + .take(64) + .map(char::from) + .collect::(), + }), + ..Default::default() + } + .into(), + )) + .await?; + } + + if bp.registry.count_object(ObjectType::Role).await? == 0 { + let permissions = DefaultPermissions::default(); + let mut role_ids = Vec::with_capacity(4); + + for role in [ + Role { + description: "User".into(), + enabled_permissions: Map::new(permissions.user), + ..Default::default() + }, + Role { + description: "Group".into(), + enabled_permissions: Map::new(permissions.group), + ..Default::default() + }, + Role { + description: "Tenant Administrator".into(), + enabled_permissions: Map::new(permissions.tenant), + ..Default::default() + }, + Role { + description: "Superuser".into(), + enabled_permissions: Map::new(permissions.superuser), + ..Default::default() + }, + ] { + match bp + .registry + .write(RegistryWrite::insert(&role.into())) + .await? + { + RegistryWriteResult::Success(id) => role_ids.push(id), + err => { + bp.build_error( + ObjectType::Role.singleton(), + format!("Failed to insert default role: {err}"), + ); + } + } + } + + if bp.registry.count_object(ObjectType::Authentication).await? == 0 && role_ids.len() == 4 { + bp.registry + .write(RegistryWrite::insert( + &Authentication { + default_user_role_ids: Map::new(vec![role_ids[0]]), + default_group_role_ids: Map::new(vec![role_ids[1]]), + default_tenant_role_ids: Map::new(vec![role_ids[2], role_ids[0]]), + ..Default::default() + } + .into(), + )) + .await?; + } + } + + let mut default_domain_id = None; + if bp.registry.count_object(ObjectType::Domain).await? == 0 { + match bp + .registry + .write(RegistryWrite::insert( + &Domain { + name: psl::domain_str(bp.registry.local_hostname()) + .unwrap_or("localhost.localdomain") + .to_string(), + is_enabled: true, + ..Default::default() + } + .into(), + )) + .await? + { + RegistryWriteResult::Success(id) => { + default_domain_id = Some(id); + } + err => { + bp.build_error( + ObjectType::Domain.singleton(), + format!("Failed to insert default domain: {err}"), + ); + } + } + + if let Some(domain_id) = default_domain_id { + let now = now(); + let signature_rsa = DkimSignature::Dkim1RsaSha256(Dkim1Signature { + domain_id, + enabled: true, + selector: format!("rsa-{now}"), + private_key: DkimPrivateKey::Value(SecretTextValue { + secret: generate_dkim_private_key(DkimSignatureType::Dkim1RsaSha256) + .await? + .map_err(|err| { + trc::EventType::Dkim(trc::DkimEvent::BuildError) + .into_err() + .reason(err) + .caused_by(trc::location!()) + })?, + }), + ..Default::default() + }); + let signature_ed = DkimSignature::Dkim1Ed25519Sha256(Dkim1Signature { + domain_id, + enabled: true, + selector: format!("ed-{now}"), + private_key: DkimPrivateKey::Value(SecretTextValue { + secret: generate_dkim_private_key(DkimSignatureType::Dkim1Ed25519Sha256) + .await? + .map_err(|err| { + trc::EventType::Dkim(trc::DkimEvent::BuildError) + .into_err() + .reason(err) + .caused_by(trc::location!()) + })?, + }), + ..Default::default() + }); + + for signature in [signature_rsa, signature_ed] { + bp.registry + .write(RegistryWrite::insert(&signature.into())) + .await?; + } + } + } + + if bp.registry.count_object(ObjectType::SystemSettings).await? == 0 { + bp.registry + .write(RegistryWrite::insert( + &SystemSettings { + default_hostname: bp.registry.local_hostname().to_string(), + default_domain_id: default_domain_id.unwrap_or(Id::new(0)), + ..Default::default() + } + .into(), + )) + .await?; + } + + Ok(()) +} diff --git a/crates/common/src/manager/mod.rs b/crates/common/src/manager/mod.rs index 8115e103..9bfbdcde 100644 --- a/crates/common/src/manager/mod.rs +++ b/crates/common/src/manager/mod.rs @@ -13,6 +13,7 @@ pub mod application; pub mod backup; pub mod boot; pub mod console; +pub mod defaults; pub mod restore; pub const WEBADMIN_KEY: &[u8] = "STALWART_WEBADMIN".as_bytes(); diff --git a/crates/common/src/network/dkim.rs b/crates/common/src/network/dkim.rs new file mode 100644 index 00000000..9c470958 --- /dev/null +++ b/crates/common/src/network/dkim.rs @@ -0,0 +1,49 @@ +/* + * SPDX-FileCopyrightText: 2020 Stalwart Labs LLC + * + * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL + */ + +use mail_auth::dkim::generate::DkimKeyPair; +use mail_builder::encoders::base64::base64_encode; +use registry::schema::enums::DkimSignatureType; + +pub async fn generate_dkim_private_key( + key_type: DkimSignatureType, +) -> trc::Result> { + let private_key = tokio::task::spawn_blocking(move || match key_type { + DkimSignatureType::Dkim1RsaSha256 => { + DkimKeyPair::generate_rsa(2048).map(|key| (key, "RSA PRIVATE KEY")) + } + DkimSignatureType::Dkim1Ed25519Sha256 => { + DkimKeyPair::generate_ed25519().map(|key| (key, "PRIVATE KEY")) + } + }) + .await + .map_err(|err| { + trc::EventType::Server(trc::ServerEvent::ThreadError) + .reason(err) + .caused_by(trc::location!()) + })?; + + Ok(private_key + .map(|(private_key, pk_type)| { + let mut pem = format!("-----BEGIN {pk_type}-----\n").into_bytes(); + let mut lf_count = 65; + for ch in base64_encode(private_key.private_key()).unwrap_or_default() { + pem.push(ch); + lf_count -= 1; + if lf_count == 0 { + pem.push(b'\n'); + lf_count = 65; + } + } + if lf_count != 65 { + pem.push(b'\n'); + } + pem.extend_from_slice(format!("-----END {pk_type}-----\n").as_bytes()); + + String::from_utf8(pem).unwrap_or_default() + }) + .map_err(|err| err.to_string())) +} diff --git a/crates/common/src/network/mod.rs b/crates/common/src/network/mod.rs index bd609c9d..26ef2d21 100644 --- a/crates/common/src/network/mod.rs +++ b/crates/common/src/network/mod.rs @@ -26,6 +26,7 @@ use utils::snowflake::SnowflakeIdGenerator; pub mod acme; pub mod asn; +pub mod dkim; pub mod dns; pub mod limiter; pub mod listen; diff --git a/crates/common/src/sharing/acl.rs b/crates/common/src/sharing/acl.rs index 0126e27e..f07c4f18 100644 --- a/crates/common/src/sharing/acl.rs +++ b/crates/common/src/sharing/acl.rs @@ -8,7 +8,11 @@ use crate::{Server, cache::invalidate::CacheInvalidationBuilder, ipc::CacheInval use types::acl::{AclGrant, ArchivedAclGrant}; impl Server { - pub async fn refresh_acls(&self, acl_changes: &[AclGrant], current: Option<&[AclGrant]>) { + pub async fn refresh_acls( + &self, + acl_changes: &[AclGrant], + current: Option<&[AclGrant]>, + ) -> trc::Result<()> { let mut changed_principals = CacheInvalidationBuilder::default(); if let Some(acl_current) = current { for current_item in acl_current { @@ -44,14 +48,14 @@ impl Server { } } - self.invalidate_caches(changed_principals).await; + self.invalidate_caches(changed_principals).await } pub async fn refresh_archived_acls( &self, acl_changes: &[AclGrant], acl_current: &[ArchivedAclGrant], - ) { + ) -> trc::Result<()> { let mut changed_principals = CacheInvalidationBuilder::default(); for current_item in acl_current.iter() { @@ -83,6 +87,6 @@ impl Server { } } - self.invalidate_caches(changed_principals).await; + self.invalidate_caches(changed_principals).await } } diff --git a/crates/common/src/storage/mod.rs b/crates/common/src/storage/mod.rs index 9810ad73..f65943b8 100644 --- a/crates/common/src/storage/mod.rs +++ b/crates/common/src/storage/mod.rs @@ -14,7 +14,7 @@ use registry::{ types::EnumImpl, }; use std::sync::Arc; -use store::{BlobStore, InMemoryStore, RegistryStore, SearchStore, Store, registry::RegistryQuery}; +use store::{BlobStore, InMemoryStore, RegistryStore, SearchStore, Store}; pub mod archive; pub mod blob; @@ -88,15 +88,11 @@ impl Server { } pub async fn total_accounts(&self) -> trc::Result { - self.registry() - .count(RegistryQuery::new(ObjectType::Account)) - .await + self.registry().count_object(ObjectType::Account).await } pub async fn total_domains(&self) -> trc::Result { - self.registry() - .count(RegistryQuery::new(ObjectType::Domain)) - .await + self.registry().count_object(ObjectType::Domain).await } #[cfg(not(feature = "enterprise"))] diff --git a/crates/coordinator/src/bootstrap.rs b/crates/coordinator/src/bootstrap.rs index 832715e6..56247f95 100644 --- a/crates/coordinator/src/bootstrap.rs +++ b/crates/coordinator/src/bootstrap.rs @@ -10,12 +10,12 @@ use store::{InMemoryStore, registry::bootstrap::Bootstrap}; #[allow(unreachable_patterns)] impl Coordinator { - pub async fn build(bp: &mut Bootstrap, in_memory: &InMemoryStore) -> Option { + pub async fn build(bp: &mut Bootstrap, _in_memory: &InMemoryStore) -> Option { let result = match bp.setting_infallible::().await { structs::Coordinator::Disabled => Ok(Coordinator::None), #[cfg(feature = "redis")] structs::Coordinator::Default => { - if let InMemoryStore::Redis(redis) = &in_memory { + if let InMemoryStore::Redis(redis) = &_in_memory { Ok(Coordinator::Redis(redis.clone())) } else { Err( diff --git a/crates/dav/src/common/acl.rs b/crates/dav/src/common/acl.rs index 79a4d5b2..52de50be 100644 --- a/crates/dav/src/common/acl.rs +++ b/crates/dav/src/common/acl.rs @@ -139,7 +139,9 @@ impl DavAclHandler for Server { if grants.len() != acls.len() || acls.iter().zip(grants.iter()).any(|(a, b)| a != b) { // Refresh ACLs - self.refresh_archived_acls(&grants, acls).await; + self.refresh_archived_acls(&grants, acls) + .await + .caused_by(trc::location!())?; let mut batch = BatchBuilder::new(); match container { diff --git a/crates/http/Cargo.toml b/crates/http/Cargo.toml index c59337af..cf6b53a7 100644 --- a/crates/http/Cargo.toml +++ b/crates/http/Cargo.toml @@ -40,7 +40,6 @@ pkcs8 = { version = "0.10.2", features = ["alloc", "std"] } rsa = "0.9.2" sha1 = "0.10" sha2 = "0.10" -rev_lines = "0.3.0" rkyv = { version = "0.8.10", features = ["little_endian"] } form-data = { version = "0.6.0", features = ["sync"], default-features = false } mime = "0.3.17" diff --git a/crates/http/src/auth/oauth/registration.rs b/crates/http/src/auth/oauth/registration.rs index 23780d18..45eec1ce 100644 --- a/crates/http/src/auth/oauth/registration.rs +++ b/crates/http/src/auth/oauth/registration.rs @@ -24,7 +24,6 @@ use registry::{ }; use std::future::Future; use store::{ - ahash::AHashSet, rand::{Rng, distr::Alphanumeric, rng}, registry::{RegistryQuery, write::RegistryWrite}, }; @@ -125,24 +124,23 @@ impl ClientRegistrationHandler for Server { // Fetch client registration let found_registration = if let Some(client_id) = self .registry() - .query::>( + .query::>( RegistryQuery::new(ObjectType::OAuthClient).equal(Property::ClientId, client_id), ) .await? - .iter() - .next() + .first() { if let Some(redirect_uri) = redirect_uri { let client = self .registry() - .object::(Id::new(*client_id)) + .object::(*client_id) .await? .ok_or_else(|| { trc::StoreEvent::UnexpectedError .into_err() .details("OAuth client not found.") .caused_by(trc::location!()) - .ctx(trc::Key::Id, *client_id) + .ctx(trc::Key::Id, client_id.id()) })?; if client.redirect_uris.iter().any(|uri| uri == redirect_uri) { return Ok(None); diff --git a/crates/jmap/Cargo.toml b/crates/jmap/Cargo.toml index 5cc28454..62747706 100644 --- a/crates/jmap/Cargo.toml +++ b/crates/jmap/Cargo.toml @@ -53,7 +53,6 @@ rsa = "0.9.2" rkyv = { version = "0.8.10", features = ["little_endian"] } compact_str = "0.9.0" hashify = "0.2" -rev_lines = "0.3.0" [features] test_mode = [] diff --git a/crates/jmap/src/addressbook/set.rs b/crates/jmap/src/addressbook/set.rs index 0693bb82..e3a344c1 100644 --- a/crates/jmap/src/addressbook/set.rs +++ b/crates/jmap/src/addressbook/set.rs @@ -101,7 +101,9 @@ impl AddressBookSet for Server { continue 'create; } - self.refresh_acls(&address_book.acls, None).await; + self.refresh_acls(&address_book.acls, None) + .await + .caused_by(trc::location!())?; } // Insert record @@ -191,7 +193,8 @@ impl AddressBookSet for Server { &new_address_book.acls, address_book.inner.acls.as_slice(), ) - .await; + .await + .caused_by(trc::location!())?; } // Update record diff --git a/crates/jmap/src/calendar/set.rs b/crates/jmap/src/calendar/set.rs index ee537fcc..6fd9a36e 100644 --- a/crates/jmap/src/calendar/set.rs +++ b/crates/jmap/src/calendar/set.rs @@ -106,7 +106,9 @@ impl CalendarSet for Server { continue 'create; } - self.refresh_acls(&calendar.acls, None).await; + self.refresh_acls(&calendar.acls, None) + .await + .caused_by(trc::location!())?; } // Insert record @@ -192,7 +194,8 @@ impl CalendarSet for Server { continue 'update; } self.refresh_archived_acls(&new_calendar.acls, calendar.inner.acls.as_slice()) - .await; + .await + .caused_by(trc::location!())?; } // Update record diff --git a/crates/jmap/src/file/set.rs b/crates/jmap/src/file/set.rs index d89cea45..7517a005 100644 --- a/crates/jmap/src/file/set.rs +++ b/crates/jmap/src/file/set.rs @@ -163,7 +163,9 @@ impl FileNodeSet for Server { continue 'create; } - self.refresh_acls(&file_node.acls, None).await; + self.refresh_acls(&file_node.acls, None) + .await + .caused_by(trc::location!())?; } // Insert record @@ -299,7 +301,8 @@ impl FileNodeSet for Server { .as_slice(), ), ) - .await; + .await + .caused_by(trc::location!())?; } // Update record diff --git a/crates/jmap/src/mailbox/set.rs b/crates/jmap/src/mailbox/set.rs index 59597e09..2aa67152 100644 --- a/crates/jmap/src/mailbox/set.rs +++ b/crates/jmap/src/mailbox/set.rs @@ -581,7 +581,8 @@ impl MailboxSet for Server { &changes.acls, current.as_ref().map(|m| m.inner.acls.as_slice()), ) - .await; + .await + .caused_by(trc::location!())?; } // Validate diff --git a/crates/jmap/src/registry/get.rs b/crates/jmap/src/registry/get.rs index 877a5798..d522d1c9 100644 --- a/crates/jmap/src/registry/get.rs +++ b/crates/jmap/src/registry/get.rs @@ -192,21 +192,15 @@ impl RegistryGet for Server { let ids = if let Some(ids) = get.ids.take() { ids } else { - let mut ids = self - .registry() - .query::>( + self.registry() + .query::>( RegistryQuery::new(object_type) .with_tenant(access_token.tenant_id()) - .with_account_opt(is_account_filtered.then_some(get.account_id)), + .with_account_opt(is_account_filtered.then_some(get.account_id)) + .with_limit(self.core.jmap.get_max_objects), ) .await .caused_by(trc::location!())? - .into_iter() - .take(self.core.jmap.get_max_objects) - .map(Id::new) - .collect::>(); - ids.sort_unstable(); - ids }; get.response.list.reserve(ids.len()); diff --git a/crates/jmap/src/registry/mapping/account.rs b/crates/jmap/src/registry/mapping/account.rs index 240a1717..63c08fb8 100644 --- a/crates/jmap/src/registry/mapping/account.rs +++ b/crates/jmap/src/registry/mapping/account.rs @@ -11,6 +11,7 @@ use crate::{ RegistryGetResponse, RegistryQueryResponse, RegistrySetResponse, principal::build_set_error, }, + query::RegistryQueryFilters, set::map_write_error, }, }; @@ -24,20 +25,24 @@ use common::{ ipc::CacheInvalidation, }; use directory::core::secret::{hash_secret, verify_otp_auth, verify_secret_hash}; -use jmap_proto::error::set::SetError; +use jmap_proto::{error::set::SetError, types::state::State}; use jmap_tools::{JsonPointer, JsonPointerItem, Key, Map, Value}; use registry::{ jmap::{IntoValue, JsonPointerPatch, MaybeUnpatched, RegistryJsonPatch, RegistryValue}, schema::{ - enums::StorageQuota, + enums::{CredentialType, StorageQuota}, prelude::{MASKED_PASSWORD, Object, ObjectInner, ObjectType, Property}, structs::{ Account, AccountSettings, Credential, CredentialPermissions, SecondaryCredential, }, }, - types::id::ObjectId, + types::{EnumImpl, datetime::UTCDateTime, id::ObjectId}, +}; +use std::str::FromStr; +use store::registry::{ + RegistryFilterOp, + write::{RegistryWrite, RegistryWriteResult}, }; -use store::registry::write::{RegistryWrite, RegistryWriteResult}; use trc::AddContext; use types::id::Id; use utils::map::vec_map::VecMap; @@ -657,7 +662,114 @@ pub(crate) async fn account_get( pub(crate) async fn credential_query( mut query: RegistryQueryResponse<'_>, ) -> trc::Result { - todo!() + let Some(Account::User(account)) = query + .server + .registry() + .object::(query.request.account_id) + .await? + else { + return Err(trc::JmapEvent::Forbidden + .into_err() + .details("Account not found.")); + }; + + let mut credential_type = None; + let mut expires_at_filter = None; + + query + .request + .extract_filters(|property, op, value| match property { + Property::Type => { + if let Some(typ) = value.as_str().and_then(CredentialType::parse) { + credential_type = Some(typ); + true + } else { + false + } + } + Property::ExpiresAt => { + if let Some(value) = value + .as_str() + .and_then(|value| UTCDateTime::from_str(value).ok()) + { + expires_at_filter = Some((op, value)); + true + } else { + false + } + } + _ => false, + })?; + + let mut matches = Vec::new(); + for credential in account.credentials.iter() { + if credential_type.is_none_or(|typ| credential.object_type() == typ) { + let (credential_id, expires_at) = match credential { + Credential::Password(credential) => { + (credential.credential_id, credential.expires_at) + } + Credential::AppPassword(credential) => { + (credential.credential_id, credential.expires_at) + } + Credential::ApiKey(credential) => (credential.credential_id, credential.expires_at), + }; + if expires_at_filter.is_none_or(|(op, filter_value)| { + expires_at.is_some_and(|expires_at| match op { + RegistryFilterOp::Equal => expires_at == filter_value, + RegistryFilterOp::GreaterThan => expires_at > filter_value, + RegistryFilterOp::GreaterEqualThan => expires_at >= filter_value, + RegistryFilterOp::LowerThan => expires_at < filter_value, + RegistryFilterOp::LowerEqualThan => expires_at <= filter_value, + RegistryFilterOp::TextMatch => false, + }) + }) { + matches.push((credential_id, expires_at)); + } + } + } + + let params = query + .request + .extract_parameters(query.server.core.jmap.query_max_results, None)?; + + match params.sort_by { + Property::ExpiresAt => { + if params.sort_ascending { + matches.sort_by(|a, b| a.1.cmp(&b.1).then_with(|| a.0.cmp(&b.0))); + } else { + matches.sort_by(|a, b| b.1.cmp(&a.1).then_with(|| a.0.cmp(&b.0))); + } + } + Property::Id => { + if params.sort_ascending { + matches.sort_by(|a, b| a.0.cmp(&b.0)); + } else { + matches.sort_by(|a, b| b.0.cmp(&a.0)); + } + } + property => { + return Err(trc::JmapEvent::UnsupportedSort.into_err().details(format!( + "Property {} is not supported for sorting", + property + ))); + } + } + + // Build response + let mut response = QueryResponseBuilder::new( + matches.len(), + query.server.core.jmap.query_max_results, + State::Initial, + &query.request, + ); + + for (id, _) in matches { + if !response.add_id(id) { + break; + } + } + + Ok(response) } fn validate_credential_permissions( diff --git a/crates/jmap/src/registry/mapping/archived_item.rs b/crates/jmap/src/registry/mapping/archived_item.rs index 91b11ecf..e8aa7deb 100644 --- a/crates/jmap/src/registry/mapping/archived_item.rs +++ b/crates/jmap/src/registry/mapping/archived_item.rs @@ -6,15 +6,18 @@ use crate::{ api::query::QueryResponseBuilder, - registry::mapping::{RegistryGetResponse, RegistryQueryResponse, RegistrySetResponse}, + registry::{ + mapping::{RegistryGetResponse, RegistryQueryResponse, RegistrySetResponse}, + query::RegistryQueryFilters, + }, }; -use jmap_proto::error::set::SetError; +use jmap_proto::{error::set::SetError, types::state::State}; use jmap_tools::{Key, Value}; use registry::{ jmap::IntoValue, pickle::Pickle, schema::{ - enums::ArchivedItemStatus, + enums::{ArchivedItemStatus, Permission}, prelude::{Object, ObjectType, Property}, structs::{ArchivedItem, Task, TaskRestoreArchivedItem, TaskStatus}, }, @@ -23,7 +26,6 @@ use registry::{ use std::str::FromStr; use store::{ SerializeInfallible, ValueKey, - ahash::AHashSet, registry::RegistryQuery, write::{BatchBuilder, BlobLink, BlobOp, RegistryClass, ValueClass, assert::AssertValue}, }; @@ -237,16 +239,10 @@ pub(crate) async fn archived_item_get( RegistryQuery::new(get.object_type).greater_than_or_equal(Property::AccountId, 0u64) } else { RegistryQuery::new(get.object_type).with_account(get.account_id) - }; + } + .with_limit(get.server.core.jmap.get_max_objects); - get.server - .registry() - .query::>(query) - .await? - .into_iter() - .take(get.server.core.jmap.get_max_objects) - .map(Id::from) - .collect() + get.server.registry().query::>(query).await? }; for id in ids { @@ -280,7 +276,75 @@ pub(crate) async fn archived_item_get( } pub(crate) async fn archived_item_query( - mut query: RegistryQueryResponse<'_>, + mut req: RegistryQueryResponse<'_>, ) -> trc::Result { - todo!() + let can_impersonate = req.access_token.has_permission(Permission::Impersonate); + let mut account_id = None; + + req.request + .extract_filters(|property, _, value| match property { + Property::AccountId if can_impersonate => { + if let Some(id) = value.as_str().and_then(|s| Id::from_str(s).ok()) { + account_id = Some(id); + true + } else { + false + } + } + + _ => false, + })?; + + let mut query = if let Some(account_id) = account_id { + RegistryQuery::new(req.object_type).with_account(account_id.document_id()) + } else if !can_impersonate { + RegistryQuery::new(req.object_type).with_account(req.request.account_id.document_id()) + } else { + RegistryQuery::new(req.object_type).greater_than_or_equal(Property::AccountId, 0u64) + }; + + let params = req + .request + .extract_parameters(req.server.core.jmap.query_max_results, Some(Property::Id))?; + + if let Some(limit) = params.limit { + query = query.with_limit(limit); + if let Some(anchor) = params.anchor { + query = query.with_anchor(anchor); + } else if let Some(position) = params.position { + query = query.with_index_start(position); + } + } + + let mut results = req.server.registry().query::>(query).await?; + + match params.sort_by { + Property::Id => { + if !params.sort_ascending { + results.sort_unstable_by(|a, b| b.cmp(a)); + } + } + 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) { + break; + } + } + + Ok(response) } diff --git a/crates/jmap/src/registry/mapping/dkim.rs b/crates/jmap/src/registry/mapping/dkim.rs index fb0af70b..80ed71db 100644 --- a/crates/jmap/src/registry/mapping/dkim.rs +++ b/crates/jmap/src/registry/mapping/dkim.rs @@ -7,15 +7,18 @@ use crate::registry::mapping::{ ObjectResponse, RegistrySetResponse, ValidationResult, principal::validate_tenant_quota, }; -use common::config::smtp::auth::{DkimSigner, rsa_key_parse, simple_pem_parse}; +use common::{ + config::smtp::auth::{DkimSigner, rsa_key_parse, simple_pem_parse}, + network::dkim::generate_dkim_private_key, +}; use jmap_proto::error::set::SetError; -use mail_auth::{common::crypto::Ed25519Key, dkim::generate::DkimKeyPair}; +use mail_auth::common::crypto::Ed25519Key; use mail_builder::encoders::base64::base64_encode; use pkcs8::Document; use registry::{ jmap::IntoValue, schema::{ - enums::{DkimSignatureType, TenantStorageQuota}, + enums::TenantStorageQuota, prelude::{MASKED_PASSWORD, Property}, structs::{DkimPrivateKey, DkimSignature, SecretTextValue}, }, @@ -47,41 +50,9 @@ pub(crate) async fn validate_dkim_signature( *pk = old_key.private_key().clone(); } if pk == &DkimPrivateKey::Generate { - let private_key = tokio::task::spawn_blocking(move || match key_type { - DkimSignatureType::Dkim1RsaSha256 => { - DkimKeyPair::generate_rsa(2048).map(|key| (key, "RSA PRIVATE KEY")) - } - DkimSignatureType::Dkim1Ed25519Sha256 => { - DkimKeyPair::generate_ed25519().map(|key| (key, "PRIVATE KEY")) - } - }) - .await - .map_err(|err| { - trc::EventType::Server(trc::ServerEvent::ThreadError) - .reason(err) - .caused_by(trc::location!()) - })?; - - match private_key { - Ok((private_key, pk_type)) => { - let mut pem = format!("-----BEGIN {pk_type}-----\n").into_bytes(); - let mut lf_count = 65; - for ch in base64_encode(private_key.private_key()).unwrap_or_default() { - pem.push(ch); - lf_count -= 1; - if lf_count == 0 { - pem.push(b'\n'); - lf_count = 65; - } - } - if lf_count != 65 { - pem.push(b'\n'); - } - pem.extend_from_slice(format!("-----END {pk_type}-----\n").as_bytes()); - - let pk_value = DkimPrivateKey::Value(SecretTextValue { - secret: String::from_utf8(pem).unwrap_or_default(), - }); + match generate_dkim_private_key(key_type).await? { + Ok(secret) => { + let pk_value = DkimPrivateKey::Value(SecretTextValue { secret }); response .object diff --git a/crates/jmap/src/registry/mapping/log.rs b/crates/jmap/src/registry/mapping/log.rs index e5685b17..ea3f2fff 100644 --- a/crates/jmap/src/registry/mapping/log.rs +++ b/crates/jmap/src/registry/mapping/log.rs @@ -6,21 +6,24 @@ use crate::{ api::query::QueryResponseBuilder, - registry::mapping::{RegistryGetResponse, RegistryQueryResponse}, + registry::{ + mapping::{RegistryGetResponse, RegistryQueryResponse}, + query::RegistryQueryFilters, + }, }; use chrono::DateTime; +use jmap_proto::types::state::State; use registry::{ jmap::IntoValue, - schema::{enums::TracingLevel, structs::Log}, + schema::{enums::TracingLevel, prelude::Property, structs::Log}, types::{EnumImpl, datetime::UTCDateTime}, }; -use rev_lines::RevLines; use std::{ fs::{self, File}, - io, + io::{self, BufRead, BufReader, Read, Seek, SeekFrom}, path::Path, }; -use store::ahash::AHashSet; +use store::ahash::AHashMap; use tokio::sync::oneshot; use trc::EventType; use types::id::Id; @@ -34,17 +37,14 @@ pub(crate) async fn log_get( .details("No log tracers configured on the server")); }; - let ids = if let Some(ids) = get.ids.take() { - ids.into_iter().map(|id| id.id()).collect::>() - } else { - (0u64..get.server.core.jmap.get_max_objects as u64).collect() - }; + let ids = get.ids.take(); - if !ids.is_empty() { + if ids.as_ref().is_none_or(|ids| !ids.is_empty()) { // TODO: Use worker pool + let limit = get.server.core.jmap.get_max_objects; let (tx, rx) = oneshot::channel(); tokio::task::spawn_blocking(move || { - let _ = tx.send(read_log_entries(path, ids)); + let _ = tx.send(read_log_entries(path, ids, limit)); }); rx.await @@ -69,84 +69,198 @@ pub(crate) async fn log_get( } pub(crate) async fn log_query( - mut query: RegistryQueryResponse<'_>, + mut req: RegistryQueryResponse<'_>, ) -> trc::Result { - todo!() + let Some(path) = req.server.core.metrics.log_path.clone() else { + return Err(trc::JmapEvent::InvalidArguments + .into_err() + .details("No log tracers configured on the server")); + }; + + let mut filter = None; + + req.request + .extract_filters(|property, _, value| match property { + Property::Text => { + if let serde_json::Value::String(due) = value { + filter = Some(due); + true + } else { + false + } + } + _ => false, + })?; + + let params = req + .request + .extract_parameters(req.server.core.jmap.query_max_results, Property::Id.into())?; + + if params.sort_by != Property::Id { + return Err(trc::JmapEvent::UnsupportedSort + .into_err() + .details("Only sorting by 'id' is supported for logs")); + } + + if req.request.calculate_total.unwrap_or(false) { + return Err(trc::JmapEvent::CannotCalculateChanges + .into_err() + .details("Calculating total is not supported for logs")); + } + + if req.request.anchor_offset.unwrap_or(0) != 0 || req.request.position.unwrap_or(0) != 0 { + return Err(trc::JmapEvent::InvalidArguments + .into_err() + .details("Pagination is only possible using anchors for logs")); + } + + let (tx, rx) = oneshot::channel(); + let anchor = params.anchor.unwrap_or(0); + let limit = params + .limit + .unwrap_or(req.server.core.jmap.query_max_results); + tokio::task::spawn_blocking(move || { + let _ = tx.send(read_log_offsets(path, filter.as_deref(), anchor, limit)); + }); + + // Build response + let mut response = QueryResponseBuilder::new( + req.server.core.jmap.query_max_results, + req.server.core.jmap.query_max_results, + State::Initial, + &req.request, + ); + + response.response.ids = rx + .await + .map_err(|err| { + trc::EventType::Server(trc::ServerEvent::ThreadError) + .reason(err) + .caused_by(trc::location!()) + })? + .map_err(|err| { + trc::EventType::Telemetry(trc::TelemetryEvent::LogError) + .reason(err) + .details("Failed to read log files") + .caused_by(trc::location!()) + })?; + + Ok(response) } -fn line_numbers( +fn read_log_offsets( path: impl AsRef, - filter: &str, - mut offset: usize, + filter: Option<&str>, + anchor: u64, limit: usize, -) -> io::Result<(usize, Vec)> { +) -> io::Result> { let mut logs = fs::read_dir(path)?.collect::, _>>()?; - let mut total = 0; - - // Sort the entries by file name in reverse order. logs.sort_by_key(|b| std::cmp::Reverse(b.file_name())); let mut entries = Vec::with_capacity(limit); - let mut logs = logs.into_iter(); - let mut current_line = 0u64; - while let Some(log) = logs.next() { - if log.file_type()?.is_file() { - let mut rev_lines = RevLines::new(File::open(log.path())?); + let mut file_number = 0u64; + let mut found_anchor = false; + let file_anchor = anchor >> 48; - while let Some(line) = rev_lines.next() { - let line = line.map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?; + 'outer: for log in logs.into_iter() { + if !log.file_type()?.is_file() { + continue; + } - if filter.is_empty() || line.contains(filter) { - total += 1; - if offset == 0 { - entries.push(Id::from(current_line)); - if entries.len() == limit { - if rev_lines.next().is_some() || logs.next().is_some() { - total += limit; - } + if !found_anchor && file_anchor != file_number { + file_number += 1; + continue; + } - return Ok((total, entries)); - } - } else { - offset -= 1; - } + let mut rev_lines = RevLines::new(File::open(log.path())?); + rev_lines.0.init_reader()?; + + let mut offset = rev_lines.0.reader_cursor; + + for line in rev_lines { + let line = line?; + offset = offset.saturating_sub(line.len() as u64 + 1); // +1 for the newline character + let id = (file_number << 48) | offset; + + if !found_anchor { + found_anchor = id == anchor; + continue; + } + + if filter.is_none_or(|filter| line.contains(filter)) { + entries.push(Id::from(id)); + if entries.len() == limit { + break 'outer; } - - current_line += 1; } } + + file_number += 1; } - Ok((total, entries)) + Ok(entries) } -fn read_log_entries(path: impl AsRef, lines: AHashSet) -> io::Result> { +fn read_log_entries( + path: impl AsRef, + ids: Option>, + limit: usize, +) -> io::Result> { + let path = path.as_ref(); + let ids = if let Some(mut ids) = ids { + ids.truncate(limit); + ids + } else { + read_log_offsets(path, None, 0, limit)? + }; + let mut logs = fs::read_dir(path)?.collect::, _>>()?; // Sort the entries by file name in reverse order. logs.sort_by_key(|b| std::cmp::Reverse(b.file_name())); - let mut entries = Vec::with_capacity(lines.len()); - let mut current_line = 0; + let mut entries = Vec::with_capacity(ids.len()); + + // Group files and offsets + let mut offset_map = AHashMap::new(); + let total_ids = ids.len(); + for id in ids { + let file_number = id.id() >> 48; + let offset = id.id() & 0xFFFFFFFFFFFF; + offset_map + .entry(file_number) + .or_insert_with(Vec::new) + .push(offset); + } + + let mut file_number = 0u64; + let mut line = String::with_capacity(256); 'outer: for log in logs.into_iter() { - if log.file_type()?.is_file() { - for line in RevLines::new(File::open(log.path())?) { - let line = line.map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?; + if !log.file_type()?.is_file() { + continue; + } - if lines.contains(¤t_line) - && let Some(log) = log_from_line(&line) - { - entries.push((Id::from(current_line), log)); + if let Some(offsets) = offset_map.get(&file_number) { + let mut reader = BufReader::new(File::open(log.path())?); - if entries.len() == lines.len() { + for offset in offsets { + // seek to the offset and read the line + reader.seek(SeekFrom::Start(*offset))?; + line.clear(); + reader.read_line(&mut line)?; + + if let Some(log) = log_from_line(&line) { + entries.push((Id::from((file_number << 48) | *offset), log)); + + if entries.len() == total_ids { break 'outer; } } - - current_line += 1; } } + + file_number += 1; } Ok(entries) @@ -166,3 +280,175 @@ fn log_from_line(line: &str) -> Option { details: details.trim().to_string(), }) } + +/* + * SPDX-FileCopyrightText: 2017 Michael Coyne + * + * SPDX-License-Identifier: MIT + */ + +// Adapted from https://github.com/mjc-gh/rev_lines/blob/main/src/lib.rs + +static DEFAULT_SIZE: usize = 4096; +static LF_BYTE: u8 = b'\n'; + +/// `RevLines` struct +pub struct RawRevLines { + reader: BufReader, + reader_cursor: u64, + buffer: Vec, + buffer_end: usize, + read_len: usize, +} + +impl RawRevLines { + /// Create a new `RawRevLines` struct from a Reader. + /// Internal buffering for iteration will default to 4096 bytes at a time. + pub fn new(reader: R) -> RawRevLines { + RawRevLines::with_capacity(DEFAULT_SIZE, reader) + } + + /// Create a new `RawRevLines` struct from a Reader`. + /// Internal buffering for iteration will use `cap` bytes at a time. + pub fn with_capacity(cap: usize, reader: R) -> RawRevLines { + RawRevLines { + reader: BufReader::new(reader), + reader_cursor: u64::MAX, + buffer: vec![0; cap], + buffer_end: 0, + read_len: 0, + } + } + + pub fn init_reader(&mut self) -> io::Result<()> { + // Move cursor to the end of the file and store the cursor position + self.reader_cursor = self.reader.seek(SeekFrom::End(0))?; + // Next read will be the full buffer size or the remaining bytes in the file + self.read_len = std::cmp::min(self.buffer.len(), self.reader_cursor as usize); + // Move cursor just before the next bytes to read + self.reader.seek_relative(-(self.read_len as i64))?; + // Update the cursor position + self.reader_cursor -= self.read_len as u64; + + self.read_to_buffer()?; + + // Handle any trailing new line characters for the reader + // so the first next call does not return Some("") + if self.buffer_end > 0 + && let Some(last_byte) = self.buffer.get(self.buffer_end - 1) + && *last_byte == LF_BYTE + { + self.buffer_end -= 1; + } + + Ok(()) + } + + fn read_to_buffer(&mut self) -> io::Result<()> { + // Read the next bytes into the buffer, self.read_len was already prepared for that + self.reader.read_exact(&mut self.buffer[0..self.read_len])?; + // Specify which part of the buffer is valid + self.buffer_end = self.read_len; + + // Determine what the next read length will be + let next_read_len = std::cmp::min(self.buffer.len(), self.reader_cursor as usize); + // Move the cursor just in front of the next read + self.reader + .seek_relative(-((self.read_len + next_read_len) as i64))?; + // Update cursor position + self.reader_cursor -= next_read_len as u64; + + // Store the next read length, it'll be used in the next call + self.read_len = next_read_len; + + Ok(()) + } + + fn next_line(&mut self) -> io::Result>> { + // Reader cursor will only ever be u64::MAX if the reader has not been initialized + // If by some chance the reader is initialized with a file of length u64::MAX this will still work, + // as some read length value is subtracted from the cursor position right away + if self.reader_cursor == u64::MAX { + self.init_reader()?; + } + + // For most sane scenarios, where size of the buffer is greater than the length of the line, + // the result will only contain one and at most two elements, making the flattening trivial. + // At the same time, instead of pushing one element at a time, it allows us to copy a subslice of the buffer, + // which is very performant on modern architectures. + let mut result: Vec> = Vec::new(); + + 'outer: loop { + // Current buffer was read to completion, read new contents + if self.buffer_end == 0 { + // Read the of minimum between the desired + // buffer size or remaining length of the reader + self.read_to_buffer()?; + } + + // If buffer_end is still 0, it means the reader is empty + if self.buffer_end == 0 { + if result.is_empty() { + return Ok(None); + } else { + break; + } + } + + let buffer_length = self.buffer_end; + + for ch in self.buffer[..self.buffer_end].iter().rev() { + self.buffer_end -= 1; + // Found a new line character to break on + if *ch == LF_BYTE { + result.push(self.buffer[self.buffer_end + 1..buffer_length].to_vec()); + break 'outer; + } + } + + result.push(self.buffer[..buffer_length].to_vec()); + } + + Ok(Some(result.into_iter().rev().flatten().collect())) + } +} + +impl Iterator for RawRevLines { + type Item = io::Result>; + + fn next(&mut self) -> Option>> { + self.next_line().transpose() + } +} + +pub struct RevLines(RawRevLines); + +impl RevLines { + /// Create a new `RawRevLines` struct from a Reader. + /// Internal buffering for iteration will default to 4096 bytes at a time. + pub fn new(reader: R) -> RevLines { + RevLines(RawRevLines::new(reader)) + } + + /// Create a new `RawRevLines` struct from a Reader`. + /// Internal buffering for iteration will use `cap` bytes at a time. + pub fn with_capacity(cap: usize, reader: R) -> RevLines { + RevLines(RawRevLines::with_capacity(cap, reader)) + } +} + +impl Iterator for RevLines { + type Item = Result; + + fn next(&mut self) -> Option> { + let line = match self.0.next_line().transpose()? { + Ok(line) => line, + Err(error) => return Some(Err(error)), + }; + + Some( + String::from_utf8(line) + .map_err(|_| std::io::Error::new(std::io::ErrorKind::InvalidData, "Invalid UTF-8")), + ) + } +} diff --git a/crates/jmap/src/registry/mapping/masked_email.rs b/crates/jmap/src/registry/mapping/masked_email.rs index a414676c..afa924a5 100644 --- a/crates/jmap/src/registry/mapping/masked_email.rs +++ b/crates/jmap/src/registry/mapping/masked_email.rs @@ -16,7 +16,10 @@ use registry::{ structs::MaskedEmail, }, }; -use store::{registry::RegistryQuery, write::now}; +use store::{ + registry::{RegistryObjectCounter, RegistryQuery}, + write::now, +}; use utils::{DomainPart, map::vec_map::VecMap}; pub(crate) async fn validate_masked_email( @@ -32,8 +35,11 @@ pub(crate) async fn validate_masked_email( let num_masked = set .server .registry() - .count(RegistryQuery::new(ObjectType::MaskedEmail).with_account(set.account_id)) - .await? as u32; + .query::( + RegistryQuery::new(ObjectType::MaskedEmail).with_account(set.account_id), + ) + .await? + .0 as u32; let account = set.server.account(set.account_id).await?; let masked_quota = set .server diff --git a/crates/jmap/src/registry/mapping/principal.rs b/crates/jmap/src/registry/mapping/principal.rs index 72023cf6..e6a02065 100644 --- a/crates/jmap/src/registry/mapping/principal.rs +++ b/crates/jmap/src/registry/mapping/principal.rs @@ -20,7 +20,10 @@ use registry::{ }, types::EnumImpl, }; -use store::{registry::RegistryQuery, write::BatchBuilder}; +use store::{ + registry::{RegistryObjectCounter, RegistryQuery}, + write::BatchBuilder, +}; use trc::AddContext; use types::id::Id; @@ -325,11 +328,21 @@ pub(crate) async fn validate_tenant_quota( TenantStorageQuota::MaxDkimKeys => (ObjectType::DkimSignature, None, "DKIM keys"), TenantStorageQuota::MaxDiskQuota => unreachable!(), }; - let mut query = RegistryQuery::new(object_type).with_tenant(tenant_id.into()); - if let Some(type_filter) = type_filter { - query = query.equal(Property::Type, type_filter.to_id()); - } - let count = set.server.registry().count(query).await? as u32; + let query = RegistryQuery::new(object_type).with_tenant(tenant_id.into()); + let count = if let Some(type_filter) = type_filter { + set.server + .registry() + .query::>(query.equal(Property::Type, type_filter.to_id())) + .await? + .len() as u32 + } else { + set.server + .registry() + .query::(query) + .await? + .0 as u32 + }; + if count >= quotas { return Ok(Err(SetError::over_quota().with_description(format!( "You have exceeded your quota of {} {}.", diff --git a/crates/jmap/src/registry/mapping/public_key.rs b/crates/jmap/src/registry/mapping/public_key.rs index 60b55339..036af889 100644 --- a/crates/jmap/src/registry/mapping/public_key.rs +++ b/crates/jmap/src/registry/mapping/public_key.rs @@ -12,7 +12,7 @@ use registry::schema::{ prelude::{ObjectType, Property}, structs::PublicKey, }; -use store::registry::RegistryQuery; +use store::registry::{RegistryObjectCounter, RegistryQuery}; pub(crate) async fn validate_public_key( set: &RegistrySetResponse<'_>, @@ -30,8 +30,11 @@ pub(crate) async fn validate_public_key( let num_masked = set .server .registry() - .count(RegistryQuery::new(ObjectType::PublicKey).with_account(set.account_id)) - .await? as u32; + .query::( + RegistryQuery::new(ObjectType::PublicKey).with_account(set.account_id), + ) + .await? + .0 as u32; let account = set.server.account(set.account_id).await?; let masked_quota = set .server diff --git a/crates/jmap/src/registry/mapping/queued_message.rs b/crates/jmap/src/registry/mapping/queued_message.rs index 9b2d78e9..ba61758e 100644 --- a/crates/jmap/src/registry/mapping/queued_message.rs +++ b/crates/jmap/src/registry/mapping/queued_message.rs @@ -4,22 +4,27 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ +use std::str::FromStr; + use crate::{ api::query::QueryResponseBuilder, - registry::mapping::{RegistryGetResponse, RegistryQueryResponse, RegistrySetResponse}, + registry::{ + mapping::{RegistryGetResponse, RegistryQueryResponse, RegistrySetResponse}, + query::RegistryQueryFilters, + }, }; use common::{ Server, config::smtp::queue::{ArchivedQueueExpiry, QueueName}, ipc::QueueEvent, }; -use jmap_proto::error::set::SetError; +use jmap_proto::{error::set::SetError, object::registry::RegistryComparator, types::state::State}; use jmap_tools::{JsonPointer, JsonPointerItem, Key}; use registry::{ jmap::{IntoValue, JsonPointerPatch, RegistryJsonPatch}, schema::{ enums::{DeliveryErrorType, MessageFlag, RecipientFlag}, - prelude::ObjectType, + prelude::{ObjectType, Property}, structs::{ DeliveryError, QueueExpiry, QueueExpiryAttempts, QueueExpiryTtl, QueuedMessage, QueuedRecipient, RecipientStatus, ServerResponse, @@ -34,10 +39,10 @@ use smtp::queue::{ Schedule, Status, spool::SmtpSpool, }; use store::{ - IterateParams, U64_LEN, ValueKey, + Deserialize, IterateParams, U64_LEN, ValueKey, ahash::AHashSet, - registry::RegistryQuery, - write::{QueueClass, ValueClass, key::DeserializeBigEndian}, + registry::{RegistryFilterOp, RegistryQuery}, + write::{AlignedBytes, Archive, QueueClass, ValueClass, key::DeserializeBigEndian}, }; use trc::AddContext; use types::{blob::BlobId, blob_hash::BlobHash, id::Id}; @@ -263,23 +268,253 @@ pub(crate) async fn queued_message_get( } pub(crate) async fn queued_message_query( - mut query: RegistryQueryResponse<'_>, + mut req: RegistryQueryResponse<'_>, ) -> trc::Result { - todo!() + let mut due_from = 0u64; + let mut due_to = u64::MAX; + let mut queue_name = None; + let mut filter_text = None; + let mut filter_from = None; + let mut filter_to = None; + + // Obtain tenant domains + let tenant_domains = if let Some(tenant_id) = req.access_token.tenant_id() { + Some(tenant_domains(req.server, tenant_id).await?) + } else { + None + }; + + req.request + .extract_filters(|property, op, value| match property { + Property::Due => { + if let Some(due) = value.as_str().and_then(|s| UTCDateTime::from_str(s).ok()) { + let due = due.timestamp() as u64; + let (from, to) = match op { + RegistryFilterOp::Equal => (due, due), + RegistryFilterOp::GreaterThan => (due + 1, u64::MAX), + RegistryFilterOp::GreaterEqualThan => (due, u64::MAX), + RegistryFilterOp::LowerThan => (0, due - 1), + RegistryFilterOp::LowerEqualThan => (0, due), + _ => return false, + }; + + // Intersect with existing range + due_from = due_from.max(from); + due_to = due_to.min(to); + + due_from <= due_to + } else { + false + } + } + Property::QueueName => { + if let Some(value) = value.as_str().and_then(QueueName::new) { + queue_name = Some(value); + true + } else { + false + } + } + Property::ReturnPath => { + if let serde_json::Value::String(name) = value { + filter_from = Some(name); + true + } else { + false + } + } + Property::To => { + if let serde_json::Value::String(name) = value { + filter_to = Some(name); + true + } else { + false + } + } + Property::Text => { + if let serde_json::Value::String(name) = value { + filter_text = Some(name); + true + } else { + false + } + } + _ => false, + })?; + + if req + .request + .sort + .as_ref() + .and_then(|sort| sort.first()) + .is_some_and(|comp| !matches!(comp.property, RegistryComparator::Property(Property::Due))) + { + return Err(trc::JmapEvent::UnsupportedSort + .into_err() + .details("Only sorting by 'due' is supported for queued messages".to_string())); + } + + let params = req + .request + .extract_parameters(req.server.core.jmap.query_max_results, None)?; + + let has_filters = filter_text.is_some() || filter_from.is_some() || filter_to.is_some(); + if has_filters || tenant_domains.is_some() { + let from_key = ValueKey::from(ValueClass::Queue(QueueClass::Message(0))); + let to_key = ValueKey::from(ValueClass::Queue(QueueClass::Message(u64::MAX))); + + let mut results = Vec::with_capacity(8); + req.server + .core + .storage + .data + .iterate( + IterateParams::new(from_key, to_key).ascending(), + |key, value| { + let message_ = as Deserialize>::deserialize(value) + .add_context(|ctx| ctx.ctx(trc::Key::Key, key))?; + let message = message_ + .unarchive::() + .add_context(|ctx| ctx.ctx(trc::Key::Key, key))?; + + if let Some(due) = message.next_delivery_event(queue_name) + && tenant_domains + .as_ref() + .is_none_or(|domains| message.has_domain(domains)) + && (due_from..=due_to).contains(&due) + && queue_name + .as_ref() + .is_none_or(|q| message.recipients.iter().any(|r| &r.queue == q)) + && (!has_filters + || (filter_text + .as_ref() + .map(|text| { + message.return_path.contains(text) + || message + .recipients + .iter() + .any(|r| r.address().contains(text)) + }) + .unwrap_or_else(|| { + filter_from + .as_ref() + .is_none_or(|from| message.return_path.contains(from)) + && filter_to.as_ref().is_none_or(|to| { + message + .recipients + .iter() + .any(|r| r.address().contains(to)) + }) + }))) + { + results.push((key.deserialize_be_u64(0)?, due)); + } + + Ok(true) + }, + ) + .await + .caused_by(trc::location!())?; + + // Build response + let mut response = QueryResponseBuilder::new( + results.len(), + req.server.core.jmap.query_max_results, + State::Initial, + &req.request, + ); + + if params.sort_ascending { + results.sort_by_key(|(_, due)| *due); + } else { + results.sort_by_key(|(_, due)| u64::MAX - *due); + } + + for (id, _) in results { + if !response.add_id(id.into()) { + break; + } + } + + Ok(response) + } else { + // Build response + let mut response = QueryResponseBuilder::new( + req.server.core.jmap.query_max_results, + req.server.core.jmap.query_max_results, + State::Initial, + &req.request, + ); + + if response.response.total.is_some() { + response.response.total = Some(0); + } + + let from_key = ValueKey::from(ValueClass::Queue(QueueClass::MessageEvent( + store::write::QueueEvent { + due: due_from, + queue_id: 0, + queue_name: [0; 8], + }, + ))); + let to_key = ValueKey::from(ValueClass::Queue(QueueClass::MessageEvent( + store::write::QueueEvent { + due: due_to, + queue_id: u64::MAX, + queue_name: [u8::MAX; 8], + }, + ))); + + let mut seen_ids = AHashSet::with_capacity(8); + req.server + .store() + .iterate( + IterateParams::new(from_key, to_key) + .set_ascending(params.sort_ascending) + .no_values(), + |key, _| { + let id = key.deserialize_be_u64(U64_LEN)?; + if queue_name.is_none_or(|queue_name| { + queue_name.as_slice() == key.get(U64_LEN * 2..).unwrap_or_default() + }) && seen_ids.insert(id) + { + if let Some(total) = response.response.total.as_mut() { + *total += 1; + if !response.is_full() { + response.add_id(id.into()); + } + Ok(true) + } else { + Ok(response.add_id(id.into())) + } + } else { + Ok(true) + } + }, + ) + .await + .caused_by(trc::location!())?; + + if let (Some(total), Some(limit)) = (response.response.total, response.response.limit) + && total < limit + { + response.response.limit = None; + } + + Ok(response) + } } async fn tenant_domains(server: &Server, tenant_id: u32) -> trc::Result> { let domain_ids = server .registry() - .query::>( - RegistryQuery::new(ObjectType::Domain).with_tenant(tenant_id.into()), - ) + .query::>(RegistryQuery::new(ObjectType::Domain).with_tenant(tenant_id.into())) .await?; let mut domains = AHashSet::with_capacity(domain_ids.len()); for domain_id in domain_ids { - if let Some(domain) = server.domain_by_id(domain_id as u32).await? { + if let Some(domain) = server.domain_by_id(domain_id.document_id()).await? { domains.extend(domain.names.iter().map(|name| name.to_string())); } } diff --git a/crates/jmap/src/registry/mapping/report.rs b/crates/jmap/src/registry/mapping/report.rs index 43a95af7..da01a22a 100644 --- a/crates/jmap/src/registry/mapping/report.rs +++ b/crates/jmap/src/registry/mapping/report.rs @@ -22,7 +22,6 @@ use smtp::reporting::index::{ExternalReportIndex, InternalReportIndex}; use std::str::FromStr; use store::{ U64_LEN, ValueKey, - ahash::AHashSet, registry::{RegistryFilter, RegistryFilterValue, RegistryQuery}, write::{BatchBuilder, RegistryClass, ValueClass, key::KeySerializer}, }; @@ -119,7 +118,7 @@ pub(crate) async fn report_set( if let Some(report) = set .server .store() - .get_value::(ValueKey::from(key.clone())) + .get_value::(ValueKey::from(key)) .await? .filter(|report| { !set.is_tenant_filtered || report.inner.member_tenant_id() == tenant_id @@ -175,34 +174,32 @@ pub(crate) async fn report_get( | ObjectType::ArfExternalReport ) { if get.is_tenant_filtered { - get.server.registry().query::>( - RegistryQuery::new(get.object_type).with_tenant(get.access_token.tenant_id()), + get.server.registry().query::>( + RegistryQuery::new(get.object_type) + .with_tenant(get.access_token.tenant_id()) + .with_limit(get.server.core.jmap.get_max_objects), ) } else { - get.server.registry().query::>( - RegistryQuery::new(get.object_type).greater_than(Property::ExpiresAt, 0u64), + get.server.registry().query::>( + RegistryQuery::new(get.object_type) + .greater_than(Property::ExpiresAt, 0u64) + .with_limit(get.server.core.jmap.get_max_objects), ) } .await? - .into_iter() - .take(get.server.core.jmap.get_max_objects) - .map(Id::from) - .collect() } else { get.server .registry() - .query::>(RegistryQuery::new(get.object_type).filter( - RegistryFilter::greater_than( - Property::Domain, - RegistryFilterValue::Bytes(vec![]), - true, - ), - )) + .query::>( + RegistryQuery::new(get.object_type) + .filter(RegistryFilter::greater_than( + Property::Domain, + RegistryFilterValue::Bytes(vec![]), + true, + )) + .with_limit(get.server.core.jmap.get_max_objects), + ) .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); @@ -331,7 +328,9 @@ pub(crate) async fn report_query( _ => false, })?; - let (comparator, is_ascending) = req.request.extract_comparator()?; + let params = req + .request + .extract_parameters(req.server.core.jmap.query_max_results, Some(Property::Id))?; if !query.has_filters() { if is_internal { @@ -348,14 +347,20 @@ pub(crate) async fn report_query( )); } } + if let Some(limit) = params.limit { + query = query.with_limit(limit); + if let Some(anchor) = params.anchor { + query = query.with_anchor(anchor); + } else if let Some(position) = params.position { + query = query.with_index_start(position); + } + } - let matches = req.server.registry().query::>(query).await?; - let results = match comparator { + let matches = req.server.registry().query::>(query).await?; + let results = match params.sort_by { Property::Id => { - let mut results = matches.into_iter().collect::>(); - if is_ascending { - results.sort_unstable(); - } else { + let mut results = matches; + if !params.sort_ascending { results.sort_unstable_by(|a, b| b.cmp(a)); } results @@ -368,7 +373,7 @@ pub(crate) async fn report_query( req.object_type, Property::Domain, Some(matches), - is_ascending, + params.sort_ascending, ) .await? } else { @@ -383,7 +388,7 @@ pub(crate) async fn report_query( req.object_type, Property::ExpiresAt, Some(matches), - is_ascending, + params.sort_ascending, ) .await? } else { @@ -407,7 +412,7 @@ pub(crate) async fn report_query( ); for id in results { - if !response.add_id(id.into()) { + if !response.add_id(id) { break; } } diff --git a/crates/jmap/src/registry/mapping/spam_sample.rs b/crates/jmap/src/registry/mapping/spam_sample.rs index 7caec320..5ebf83d8 100644 --- a/crates/jmap/src/registry/mapping/spam_sample.rs +++ b/crates/jmap/src/registry/mapping/spam_sample.rs @@ -4,18 +4,24 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ +use std::str::FromStr; + use crate::{ api::query::QueryResponseBuilder, blob::download::BlobDownload, - registry::mapping::{RegistryGetResponse, RegistryQueryResponse, RegistrySetResponse}, + registry::{ + mapping::{RegistryGetResponse, RegistryQueryResponse, RegistrySetResponse}, + query::RegistryQueryFilters, + }, }; -use jmap_proto::error::set::SetError; +use jmap_proto::{error::set::SetError, types::state::State}; use jmap_tools::{JsonPointer, JsonPointerItem, Key}; use mail_parser::{MessageParser, parsers::fields::thread::thread_name}; use registry::{ jmap::{IntoValue, JsonPointerPatch, RegistryJsonPatch}, pickle::Pickle, schema::{ + enums::Permission, prelude::{ObjectType, Property}, structs::SpamTrainingSample, }, @@ -23,7 +29,6 @@ use registry::{ }; use store::{ SerializeInfallible, ValueKey, - ahash::AHashSet, registry::RegistryQuery, write::{BatchBuilder, BlobLink, BlobOp, RegistryClass, ValueClass, now}, }; @@ -254,16 +259,10 @@ pub(crate) async fn spam_sample_get( RegistryQuery::new(get.object_type).greater_than_or_equal(Property::AccountId, 0u64) } else { RegistryQuery::new(get.object_type).with_account(get.account_id) - }; + } + .with_limit(get.server.core.jmap.get_max_objects); - get.server - .registry() - .query::>(query) - .await? - .into_iter() - .take(get.server.core.jmap.get_max_objects) - .map(Id::from) - .collect() + get.server.registry().query::>(query).await? }; for id in ids { @@ -301,7 +300,75 @@ pub(crate) async fn spam_sample_get( } pub(crate) async fn spam_sample_query( - mut query: RegistryQueryResponse<'_>, + mut req: RegistryQueryResponse<'_>, ) -> trc::Result { - todo!() + let can_impersonate = req.access_token.has_permission(Permission::Impersonate); + let mut account_id = None; + + req.request + .extract_filters(|property, _, value| match property { + Property::AccountId if can_impersonate => { + if let Some(id) = value.as_str().and_then(|s| Id::from_str(s).ok()) { + account_id = Some(id); + true + } else { + false + } + } + + _ => false, + })?; + + let mut query = if let Some(account_id) = account_id { + RegistryQuery::new(req.object_type).with_account(account_id.document_id()) + } else if !can_impersonate { + RegistryQuery::new(req.object_type).with_account(req.request.account_id.document_id()) + } else { + RegistryQuery::new(req.object_type).greater_than_or_equal(Property::AccountId, 0u64) + }; + + let params = req + .request + .extract_parameters(req.server.core.jmap.query_max_results, Some(Property::Id))?; + + if let Some(limit) = params.limit { + query = query.with_limit(limit); + if let Some(anchor) = params.anchor { + query = query.with_anchor(anchor); + } else if let Some(position) = params.position { + query = query.with_index_start(position); + } + } + + let mut results = req.server.registry().query::>(query).await?; + + match params.sort_by { + Property::Id => { + if !params.sort_ascending { + results.sort_unstable_by(|a, b| b.cmp(a)); + } + } + 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) { + break; + } + } + + Ok(response) } diff --git a/crates/jmap/src/registry/mapping/task.rs b/crates/jmap/src/registry/mapping/task.rs index 5f709ad8..64b1a8c4 100644 --- a/crates/jmap/src/registry/mapping/task.rs +++ b/crates/jmap/src/registry/mapping/task.rs @@ -6,23 +6,37 @@ use crate::{ api::query::QueryResponseBuilder, - registry::mapping::{RegistryGetResponse, RegistryQueryResponse, RegistrySetResponse}, + registry::{ + mapping::{RegistryGetResponse, RegistryQueryResponse, RegistrySetResponse}, + query::RegistryQueryFilters, + }, }; use common::Server; -use jmap_proto::error::set::{SetError, SetErrorType}; +use jmap_proto::{ + error::set::{SetError, SetErrorType}, + object::registry::RegistryComparator, + types::state::State, +}; use jmap_tools::{JsonPointer, JsonPointerItem, Key}; use registry::{ jmap::{IntoValue, JsonPointerPatch, RegistryJsonPatch}, pickle::Pickle, - schema::{enums::TaskType, prelude::Object, structs::Task}, + schema::{ + enums::{TaskStatusType, TaskType}, + prelude::{Object, Property}, + structs::Task, + }, types::{ EnumImpl, ObjectImpl, + datetime::UTCDateTime, index::{IndexBuilder, IndexKey}, }, }; use services::task_manager::lock::TaskLockManager; +use std::str::FromStr; use store::{ IterateParams, SerializeInfallible, U64_LEN, ValueKey, + registry::RegistryFilterOp, write::{BatchBuilder, RegistryClass, TaskQueueClass, ValueClass, key::DeserializeBigEndian}, }; use trc::AddContext; @@ -372,14 +386,118 @@ pub(crate) async fn task_get( } pub(crate) async fn task_query( - mut query: RegistryQueryResponse<'_>, + mut req: RegistryQueryResponse<'_>, ) -> trc::Result { - todo!() + let mut due_from = 0u64; + let mut due_to = u64::MAX; + + req.request + .extract_filters(|property, op, value| match property { + Property::Due => { + if let Some(due) = value.as_str().and_then(|s| UTCDateTime::from_str(s).ok()) { + let due = due.timestamp() as u64; + let (from, to) = match op { + RegistryFilterOp::Equal => (due, due), + RegistryFilterOp::GreaterThan => (due + 1, u64::MAX), + RegistryFilterOp::GreaterEqualThan => (due, u64::MAX), + RegistryFilterOp::LowerThan => (0, due - 1), + RegistryFilterOp::LowerEqualThan => (0, due), + _ => return false, + }; + + // Intersect with existing range + due_from = due_from.max(from); + due_to = due_to.min(to); + + due_from <= due_to + } else { + false + } + } + Property::Status => { + if let Some(typ) = value.as_str().and_then(TaskStatusType::parse) { + if typ == TaskStatusType::Failed { + due_from = u64::MAX; + due_to = u64::MAX; + } + true + } else { + false + } + } + _ => false, + })?; + + if req + .request + .sort + .as_ref() + .and_then(|sort| sort.first()) + .is_some_and(|comp| !matches!(comp.property, RegistryComparator::Property(Property::Due))) + { + return Err(trc::JmapEvent::UnsupportedSort + .into_err() + .details("Only sorting by 'due' is supported for tasks".to_string())); + } + + let params = req + .request + .extract_parameters(req.server.core.jmap.query_max_results, None)?; + + // Build response + let mut response = QueryResponseBuilder::new( + req.server.core.jmap.query_max_results, + req.server.core.jmap.query_max_results, + State::Initial, + &req.request, + ); + + if response.response.total.is_some() { + response.response.total = Some(0); + } + + let from_key = ValueKey::from(ValueClass::TaskQueue(TaskQueueClass::Due { + id: 0, + due: due_from, + })); + let to_key = ValueKey::from(ValueClass::TaskQueue(TaskQueueClass::Due { + id: u64::MAX, + due: due_to, + })); + + req.server + .store() + .iterate( + IterateParams::new(from_key, to_key) + .set_ascending(params.sort_ascending) + .no_values(), + |key, _| { + let id = key.deserialize_be_u64(U64_LEN)?; + if let Some(total) = response.response.total.as_mut() { + *total += 1; + if !response.is_full() { + response.add_id(id.into()); + } + Ok(true) + } else { + Ok(response.add_id(id.into())) + } + }, + ) + .await + .caused_by(trc::location!())?; + + if let (Some(total), Some(limit)) = (response.response.total, response.response.limit) + && total < limit + { + response.response.limit = None; + } + + Ok(response) } async fn task_ids(server: &Server, max_results: usize) -> trc::Result> { - let mut events = Vec::with_capacity(8); - + let mut tasks = Vec::with_capacity(8); let from_key = ValueKey::from(ValueClass::TaskQueue(TaskQueueClass::Due { id: 0, due: 0 })); let to_key = ValueKey::from(ValueClass::TaskQueue(TaskQueueClass::Due { id: u64::MAX, @@ -391,12 +509,12 @@ async fn task_ids(server: &Server, max_results: usize) -> trc::Result> { .iterate( IterateParams::new(from_key, to_key).ascending().no_values(), |key, _| { - events.push(key.deserialize_be_u64(U64_LEN)?.into()); + tasks.push(key.deserialize_be_u64(U64_LEN)?.into()); - Ok(events.len() < max_results) + Ok(tasks.len() < max_results) }, ) .await .caused_by(trc::location!()) - .map(|_| events) + .map(|_| tasks) } diff --git a/crates/jmap/src/registry/mapping/telemetry.rs b/crates/jmap/src/registry/mapping/telemetry.rs index 33e40ab3..a7e6e99b 100644 --- a/crates/jmap/src/registry/mapping/telemetry.rs +++ b/crates/jmap/src/registry/mapping/telemetry.rs @@ -4,11 +4,17 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ +use std::str::FromStr; + use crate::{ api::query::QueryResponseBuilder, - registry::mapping::{RegistryGetResponse, RegistryQueryResponse}, + registry::{ + mapping::{RegistryGetResponse, RegistryQueryResponse}, + query::RegistryQueryFilters, + }, }; use common::Server; +use jmap_proto::types::state::State; use registry::{ jmap::IntoValue, schema::prelude::{Object, Property}, @@ -16,10 +22,14 @@ use registry::{ }; use store::{ IterateParams, ValueKey, - search::{SearchComparator, SearchField, SearchFilter, SearchQuery}, + registry::RegistryFilterOp, + search::{ + SearchComparator, SearchField, SearchFilter, SearchOperator, SearchQuery, + TracingSearchField, + }, write::{SearchIndex, TelemetryClass, ValueClass, key::DeserializeBigEndian, now}, }; -use trc::AddContext; +use trc::{AddContext, EventType}; use types::id::Id; use utils::snowflake::SnowflakeIdGenerator; @@ -104,15 +114,252 @@ pub(crate) async fn metric_get( } pub(crate) async fn trace_query( - mut query: RegistryQueryResponse<'_>, + mut req: RegistryQueryResponse<'_>, ) -> trc::Result { - todo!() + let mut tracing_query = Vec::new(); + tracing_query.push(SearchFilter::And); + + req.request + .extract_filters(|property, op, value| match property { + Property::Timestamp => { + if let Some(id) = value + .as_str() + .and_then(|s| UTCDateTime::from_str(s).ok()) + .and_then(|dt| SnowflakeIdGenerator::from_timestamp(dt.timestamp() as u64)) + { + let op = match op { + RegistryFilterOp::Equal => SearchOperator::Equal, + RegistryFilterOp::GreaterThan => SearchOperator::GreaterThan, + RegistryFilterOp::GreaterEqualThan => SearchOperator::GreaterEqualThan, + RegistryFilterOp::LowerThan => SearchOperator::LowerThan, + RegistryFilterOp::LowerEqualThan => SearchOperator::LowerEqualThan, + _ => return false, + }; + + tracing_query.push(SearchFilter::Operator { + field: SearchField::Id, + op, + value: id.into(), + }); + + true + } else { + false + } + } + Property::Event => { + if let Some(typ) = value.as_str().and_then(EventType::parse) { + tracing_query.push(SearchFilter::eq( + TracingSearchField::EventType, + typ.to_id() as u64, + )); + true + } else { + false + } + } + Property::QueueId => { + if let Some(queue_id) = value.as_str().and_then(|s| Id::from_str(s).ok()) { + tracing_query + .push(SearchFilter::eq(TracingSearchField::QueueId, queue_id.id())); + true + } else { + false + } + } + Property::Text => { + if let Some(query) = value.as_str() { + let mut buf = String::with_capacity(query.len()); + let mut in_quote = false; + for ch in query.chars() { + if ch.is_ascii_whitespace() { + if in_quote { + buf.push(' '); + } else if !buf.is_empty() { + tracing_query.push(SearchFilter::has_keyword( + TracingSearchField::Keywords, + buf, + )); + buf = String::new(); + } + } else if ch == '"' { + buf.push(ch); + if in_quote { + if !buf.is_empty() { + tracing_query.push(SearchFilter::has_keyword( + TracingSearchField::Keywords, + buf, + )); + buf = String::new(); + } + in_quote = false; + } else { + in_quote = true; + } + } else { + buf.push(ch); + } + } + if !buf.is_empty() { + tracing_query + .push(SearchFilter::has_keyword(TracingSearchField::Keywords, buf)); + } + true + } else { + false + } + } + + _ => false, + })?; + + if !tracing_query.iter().any(|f| { + matches!( + f, + SearchFilter::Operator { + field: SearchField::Tracing( + TracingSearchField::Keywords | TracingSearchField::QueueId + ) | SearchField::Id, + .. + } + ) + }) { + tracing_query.push(SearchFilter::gt( + SearchField::Id, + SnowflakeIdGenerator::from_timestamp(now() - 86400).unwrap_or_default(), + )); + } + tracing_query.push(SearchFilter::End); + + let params = req + .request + .extract_parameters(req.server.core.jmap.query_max_results, None)?; + + if !matches!(params.sort_by, Property::Id | Property::Timestamp) { + return Err(trc::JmapEvent::UnsupportedSort.into_err().details(format!( + "Property {} is not supported for sorting", + params.sort_by + ))); + } + + let results = req + .server + .search_store() + .query_global( + SearchQuery::new(SearchIndex::Tracing) + .with_filters(tracing_query) + .with_comparator(SearchComparator::Field { + field: SearchField::Id, + ascending: params.sort_ascending, + }), + ) + .await?; + + // 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) } pub(crate) async fn metric_query( - mut query: RegistryQueryResponse<'_>, + mut req: RegistryQueryResponse<'_>, ) -> trc::Result { - todo!() + let mut ts_from = 0u64; + let mut ts_to = u64::MAX; + + req.request + .extract_filters(|property, op, value| match property { + Property::Timestamp => { + if let Some(ts) = value.as_str().and_then(|s| UTCDateTime::from_str(s).ok()) { + let ts = ts.timestamp() as u64; + let (from, to) = match op { + RegistryFilterOp::Equal => (ts, ts), + RegistryFilterOp::GreaterThan => (ts + 1, u64::MAX), + RegistryFilterOp::GreaterEqualThan => (ts, u64::MAX), + RegistryFilterOp::LowerThan => (0, ts - 1), + RegistryFilterOp::LowerEqualThan => (0, ts), + _ => return false, + }; + + // Intersect with existing range + ts_from = ts_from.max(from); + ts_to = ts_to.min(to); + + true + } else { + false + } + } + _ => false, + })?; + + let params = req + .request + .extract_parameters(req.server.core.jmap.query_max_results, None)?; + + if ts_from != 0 { + ts_from = SnowflakeIdGenerator::from_timestamp(ts_from).unwrap_or(0); + } + + if ts_to != u64::MAX { + ts_to = SnowflakeIdGenerator::from_timestamp(ts_to).unwrap_or(u64::MAX); + } + + let from_key = ValueKey::from(ValueClass::Telemetry(TelemetryClass::Metric(ts_from))); + let to_key = ValueKey::from(ValueClass::Telemetry(TelemetryClass::Metric(ts_to))); + + // Build response + let mut response = QueryResponseBuilder::new( + req.server.core.jmap.query_max_results, + req.server.core.jmap.query_max_results, + State::Initial, + &req.request, + ); + + if response.response.total.is_some() { + response.response.total = Some(0); + } + + req.server + .metrics_store() + .iterate( + IterateParams::new(from_key, to_key) + .set_ascending(params.sort_ascending) + .no_values(), + |key, _| { + let id = key.deserialize_be_u64(0)?; + if let Some(total) = response.response.total.as_mut() { + *total += 1; + if !response.is_full() { + response.add_id(id.into()); + } + Ok(true) + } else { + Ok(response.add_id(id.into())) + } + }, + ) + .await + .caused_by(trc::location!())?; + + if let (Some(total), Some(limit)) = (response.response.total, response.response.limit) + && total < limit + { + response.response.limit = None; + } + + Ok(response) } async fn metric_ids(server: &Server, max_results: usize) -> trc::Result> { diff --git a/crates/jmap/src/registry/query.rs b/crates/jmap/src/registry/query.rs index 8da9de22..206312fc 100644 --- a/crates/jmap/src/registry/query.rs +++ b/crates/jmap/src/registry/query.rs @@ -35,10 +35,7 @@ use registry::{ }, }; use std::str::FromStr; -use store::{ - ahash::AHashSet, - registry::{RegistryFilterOp, RegistryFilterValue}, -}; +use store::registry::{RegistryFilterOp, RegistryFilterValue}; use types::id::Id; pub trait RegistryQuery: Sync + Send { @@ -210,9 +207,19 @@ impl RegistryQuery for Server { } })?; - let (comparator, is_ascending) = request.extract_comparator()?; - let matches = if query.has_filters() || matches!(comparator, Property::Id) { - let matches = self.registry().query::>(query).await?; + let params = request + .extract_parameters(self.core.jmap.query_max_results, Some(Property::Id))?; + if let Some(limit) = params.limit { + query = query.with_limit(limit); + if let Some(anchor) = params.anchor { + query = query.with_anchor(anchor); + } else if let Some(position) = params.position { + query = query.with_index_start(position); + } + } + + let matches = if query.has_filters() || params.sort_by == Property::Id { + let matches = self.registry().query::>(query).await?; if matches.is_empty() { return QueryResponseBuilder::new( 0, @@ -227,12 +234,10 @@ impl RegistryQuery for Server { None }; - let results = match comparator { + let results = match params.sort_by { Property::Id => { - let mut results = matches.unwrap().into_iter().collect::>(); - if is_ascending { - results.sort_unstable(); - } else { + let mut results = matches.unwrap(); + if !params.sort_ascending { results.sort_unstable_by(|a, b| b.cmp(a)); } results @@ -249,11 +254,16 @@ impl RegistryQuery for Server { if index.typ == IndexSchemaType::Search { self.registry() - .sort_by_index(object_type, index.prop, matches, is_ascending) + .sort_by_index( + object_type, + index.prop, + matches, + params.sort_ascending, + ) .await? } else { self.registry() - .sort_by_pk(object_type, index.prop, matches, is_ascending) + .sort_by_pk(object_type, index.prop, matches, params.sort_ascending) .await? } } @@ -268,7 +278,7 @@ impl RegistryQuery for Server { ); for id in results { - if !response.add_id(id.into()) { + if !response.add_id(id) { break; } } @@ -285,7 +295,19 @@ pub(crate) trait RegistryQueryFilters { cb: impl FnMut(Property, RegistryFilterOp, serde_json::Value) -> bool, ) -> trc::Result<()>; - fn extract_comparator(&mut self) -> trc::Result<(Property, bool)>; + fn extract_parameters( + &mut self, + max_results: usize, + external_filter: Option, + ) -> trc::Result; +} + +pub(crate) struct RegistryQueryParameters { + pub sort_by: Property, + pub sort_ascending: bool, + pub anchor: Option, + pub position: Option, + pub limit: Option, } impl RegistryQueryFilters for QueryRequest { @@ -339,7 +361,11 @@ impl RegistryQueryFilters for QueryRequest { Ok(()) } - fn extract_comparator(&mut self) -> trc::Result<(Property, bool)> { + fn extract_parameters( + &mut self, + max_results: usize, + external_filter: Option, + ) -> trc::Result { let comparator = self .sort .take() @@ -349,7 +375,34 @@ impl RegistryQueryFilters for QueryRequest { .unwrap_or_else(|| Comparator::ascending(RegistryComparator::Property(Property::Id))); match comparator.property { - RegistryComparator::Property(property) => Ok((property, comparator.is_ascending)), + RegistryComparator::Property(property) => { + if external_filter.is_some_and(|f| f == property) + && !self.calculate_total.unwrap_or(false) + && self.anchor_offset.is_none_or(|offset| offset == 0) + && self.position.is_none_or(|pos| pos > 0) + { + Ok(RegistryQueryParameters { + sort_by: property, + sort_ascending: comparator.is_ascending, + anchor: self.anchor.take().map(|anchor| anchor.id()), + position: self.position.take().map(|pos| pos as u64), + limit: self + .limit + .take() + .map(|limit| std::cmp::min(limit, max_results)) + .unwrap_or(max_results) + .into(), + }) + } else { + Ok(RegistryQueryParameters { + sort_by: property, + sort_ascending: comparator.is_ascending, + anchor: None, + position: None, + limit: None, + }) + } + } 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 1d493f8e..0d827b14 100644 --- a/crates/jmap/src/registry/set.rs +++ b/crates/jmap/src/registry/set.rs @@ -512,7 +512,11 @@ impl RegistrySet for Server { .write(RegistryWrite::Delete { object_id, object: Some(&object), - force: object_type == ObjectType::Account, // Force delete accounts to allow recovery, but not other objects + allowed_orphan_types: if object_type == ObjectType::Account { + &[ObjectType::PublicKey, ObjectType::MaskedEmail] + } else { + &[] + }, }) .await? { diff --git a/crates/migration/src/lib.rs b/crates/migration/src/lib.rs index 63f7ceee..7fefa868 100644 --- a/crates/migration/src/lib.rs +++ b/crates/migration/src/lib.rs @@ -13,7 +13,7 @@ v013::migrate_v0_13, v014::{SUBSPACE_BITMAP_ID, migrate_principal_v0_14, migrate_v0_14}, };*/ -use common::{DATABASE_SCHEMA_VERSION, Server, manager::boot::DEFAULT_SETTINGS}; +use common::{DATABASE_SCHEMA_VERSION, Server}; use std::time::Duration; use store::{ Deserialize, IterateParams, SUBSPACE_PROPERTY, SUBSPACE_QUEUE_MESSAGE, SUBSPACE_REGISTRY, diff --git a/crates/services/src/task_manager/destroy_account.rs b/crates/services/src/task_manager/destroy_account.rs index 4f3a2b15..2113dfed 100644 --- a/crates/services/src/task_manager/destroy_account.rs +++ b/crates/services/src/task_manager/destroy_account.rs @@ -8,16 +8,25 @@ use crate::task_manager::TaskResult; use common::Server; use email::{message::metadata::MessageMetadata, sieve::SieveScript}; use groupware::file::FileNode; -use registry::schema::structs::TaskDestroyAccount; +use registry::{ + schema::{ + prelude::{ObjectType, Property}, + structs::{ArchivedItem, TaskDestroyAccount}, + }, + types::EnumImpl, +}; use store::{ + SerializeInfallible, ValueKey, + registry::RegistryQuery, search::SearchQuery, - write::{BatchBuilder, BlobLink, BlobOp, SearchIndex, ValueClass}, + write::{BatchBuilder, BlobLink, BlobOp, RegistryClass, SearchIndex, ValueClass}, }; use trc::AddContext; use types::{ blob_hash::BlobHash, collection::Collection, field::{EmailField, Field}, + id::Id, }; pub(crate) trait DestroyAccountTask: Sync + Send { @@ -43,7 +52,87 @@ impl DestroyAccountTask for Server { async fn destroy_account(server: &Server, task: &TaskDestroyAccount) -> trc::Result { let account_id = task.account_id.document_id(); - let todo = "destroy spam samples, undelete, registry objects, etc."; + + // Destroy public keys and masked emails + for object in [ObjectType::PublicKey, ObjectType::MaskedEmail] { + let mut batch = BatchBuilder::new(); + let ids = server + .registry() + .query::>(RegistryQuery::new(object).with_account(account_id)) + .await?; + let object_id = object.to_id(); + + for id in ids { + batch + .clear(ValueClass::Registry(RegistryClass::Item { + object_id, + item_id: id.id(), + })) + .clear(ValueClass::Registry(RegistryClass::IndexId { + object_id, + item_id: id.id(), + })) + .clear(ValueClass::Registry(RegistryClass::Index { + index_id: Property::AccountId as u16, + object_id, + item_id: id.id(), + key: (account_id as u64).serialize(), + })) + .clear(ValueClass::Registry(RegistryClass::Reference { + to_object_id: ObjectType::Account as u16, + to_item_id: account_id as u64, + from_object_id: object_id, + from_item_id: id.id(), + })); + } + + if !batch.is_empty() { + server.store().write(batch.build_all()).await?; + } + } + + // Remove archived items + let mut batch = BatchBuilder::new(); + let ids = server + .registry() + .query::>(RegistryQuery::new(ObjectType::ArchivedItem).with_account(account_id)) + .await?; + for id in ids { + let object_id = ObjectType::ArchivedItem.to_id(); + let item_id = id.id(); + + if let Some(item) = server + .store() + .get_value::(ValueKey::from(ValueClass::Registry(RegistryClass::Item { + object_id, + item_id, + }))) + .await? + { + let until = item.archived_until().timestamp() as u64; + let blob_hash = item.into_blob_id().hash; + + batch + .with_account_id(account_id) + .clear(BlobOp::Link { + hash: blob_hash, + to: BlobLink::Temporary { until }, + }) + .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() { + server.store().write(batch.build_all()).await?; + } // Remove search index for index in [ diff --git a/crates/services/src/task_manager/maintenance.rs b/crates/services/src/task_manager/maintenance.rs index 330eb226..9f8120b9 100644 --- a/crates/services/src/task_manager/maintenance.rs +++ b/crates/services/src/task_manager/maintenance.rs @@ -26,22 +26,27 @@ use groupware::{ contact::{AddressBook, ContactCard}, file::FileNode, }; -use registry::schema::{ - enums::{TaskAccountMaintenanceType, TaskStoreMaintenanceType}, - prelude::ObjectType, - structs::{Task, TaskAccountMaintenance, TaskStatus, TaskStoreMaintenance}, +use registry::{ + schema::{ + enums::{TaskAccountMaintenanceType, TaskStoreMaintenanceType}, + prelude::{Object, ObjectInner, ObjectType, Property}, + structs::{Task, TaskAccountMaintenance, TaskStatus, TaskStoreMaintenance}, + }, + types::EnumImpl, }; +use smtp::reporting::index::ExternalReportIndex; use store::{ Serialize, ValueKey, rand::{self, Rng}, - registry::RegistryQuery, + registry::{RegistryFilter, RegistryQuery}, roaring::RoaringBitmap, - write::{AlignedBytes, Archive, Archiver, BatchBuilder, ValueClass, now}, + write::{AlignedBytes, Archive, Archiver, BatchBuilder, RegistryClass, ValueClass, now}, }; use trc::{AddContext, StoreEvent}; use types::{ collection::Collection, field::{EmailField, MailboxField}, + id::Id, }; pub(crate) trait MaintenanceTask: Sync + Send { @@ -123,9 +128,55 @@ async fn store_maintenance( reindex_telemetry(server).await?; } TaskStoreMaintenanceType::PurgeData => { - let todo = "make sure all store types are purged, in memory, metrics, tracing, etc"; - let todo = - "make sure spam samples with their indexes and undelete items are purged as well"; + // Delete expired external reports + let now = now(); + let mut batch = BatchBuilder::new(); + for object in [ + ObjectType::DmarcExternalReport, + ObjectType::TlsExternalReport, + ObjectType::ArfExternalReport, + ] { + let ids = server + .registry() + .query::>(RegistryQuery::new(object).filter(RegistryFilter::less_than( + Property::ExpiresAt, + now, + false, + ))) + .await?; + let object_id = object.to_id(); + for id in ids { + let item_id = id.id(); + if let Some(report) = server + .store() + .get_value::(ValueKey::from(ValueClass::Registry( + RegistryClass::Item { object_id, item_id }, + ))) + .await? + { + match &report.inner { + ObjectInner::DmarcExternalReport(report) => { + report.write_ops(&mut batch, item_id, false); + } + ObjectInner::TlsExternalReport(report) => { + report.write_ops(&mut batch, item_id, false); + } + ObjectInner::ArfExternalReport(report) => { + report.write_ops(&mut batch, item_id, false); + } + _ => {} + } + + if batch.is_large_batch() { + server.store().write(batch.build_all()).await?; + batch = BatchBuilder::new(); + } + } + } + } + if !batch.is_empty() { + server.store().write(batch.build_all()).await?; + } let started = Instant::now(); diff --git a/crates/smtp/src/queue/spool.rs b/crates/smtp/src/queue/spool.rs index 1ea033b9..ee569893 100644 --- a/crates/smtp/src/queue/spool.rs +++ b/crates/smtp/src/queue/spool.rs @@ -13,6 +13,7 @@ use crate::queue::{ FROM_AUTHENTICATED, FROM_AUTOGENERATED, FROM_DSN, FROM_REPORT, FROM_UNAUTHENTICATED, FROM_UNAUTHENTICATED_DMARC, MessageWrapper, }; +use ahash::AHashSet; use common::config::smtp::queue::QueueName; use common::ipc::QueueEvent; use common::{KV_LOCK_QUEUE_MESSAGE, Server}; @@ -777,14 +778,14 @@ impl MessageWrapper { } impl ArchivedMessage { - pub fn has_domain(&self, domains: &[String]) -> bool { + pub fn has_domain(&self, domains: &AHashSet) -> bool { self.recipients.iter().any(|r| { let domain = r.address.domain_part(); - domains.iter().any(|dd| dd == domain) + domains.contains(domain) }) || self .return_path .rsplit_once('@') - .is_some_and(|(_, domain)| domains.iter().any(|dd| dd == domain)) + .is_some_and(|(_, domain)| domains.contains(domain)) } pub fn next_delivery_event(&self, queue: Option) -> Option { diff --git a/crates/store/Cargo.toml b/crates/store/Cargo.toml index 0062f099..7ada5e86 100644 --- a/crates/store/Cargo.toml +++ b/crates/store/Cargo.toml @@ -53,6 +53,7 @@ rkyv = { version = "0.8.10", features = ["little_endian"] } compact_str = "0.9.0" rustls_021 = { package = "rustls", version = "0.21", default-features = false, features = ["dangerous_configuration"], optional = true } gethostname = "1.1.0" +radsort = "0.1.1" [dev-dependencies] tokio = { version = "1.47", features = ["full"] } diff --git a/crates/store/src/registry/local.rs b/crates/store/src/registry/local.rs index 35c0b5bb..1ff2cd98 100644 --- a/crates/store/src/registry/local.rs +++ b/crates/store/src/registry/local.rs @@ -39,7 +39,8 @@ impl RegistryStoreInner { env_hostname: std::env::var("STALWART_HOSTNAME") .ok() .filter(|h| !h.is_empty()) - .unwrap_or_else(|| gethostname::gethostname().to_string_lossy().into_owned()), + .unwrap_or_else(|| gethostname::gethostname().to_string_lossy().into_owned()) + .to_lowercase(), } } diff --git a/crates/store/src/registry/mod.rs b/crates/store/src/registry/mod.rs index 10009f8f..b48c8a25 100644 --- a/crates/store/src/registry/mod.rs +++ b/crates/store/src/registry/mod.rs @@ -33,8 +33,20 @@ pub struct RegistryObject { } pub struct RegistryQuery { - pub object_type: ObjectType, + pub(crate) object_type: ObjectType, pub filters: Vec, + pub(crate) start: RegistryQueryStart, + pub(crate) limit: Option, +} + +#[derive(Debug, Clone, Copy, Default)] +pub struct RegistryObjectCounter(pub usize); + +#[derive(Debug, Clone, Copy)] +pub(crate) enum RegistryQueryStart { + Index(u64), + Anchor(u64), + None, } pub struct RegistryFilter { diff --git a/crates/store/src/registry/query.rs b/crates/store/src/registry/query.rs index 502ba688..8e95e1d7 100644 --- a/crates/store/src/registry/query.rs +++ b/crates/store/src/registry/query.rs @@ -7,7 +7,10 @@ use crate::{ IterateParams, RegistryStore, SUBSPACE_REGISTRY_IDX, SUBSPACE_REGISTRY_PK, Store, U16_LEN, U64_LEN, ValueKey, - registry::{RegistryFilter, RegistryFilterOp, RegistryFilterValue, RegistryQuery}, + registry::{ + RegistryFilter, RegistryFilterOp, RegistryFilterValue, RegistryObjectCounter, + RegistryQuery, RegistryQueryStart, + }, write::{ AnyClass, RegistryClass, ValueClass, key::{DeserializeBigEndian, KeySerializer}, @@ -15,7 +18,7 @@ use crate::{ }; use ahash::AHashSet; use registry::{ - schema::prelude::{OBJ_FILTER_ACCOUNT, OBJ_FILTER_TENANT, OBJ_SINGLETON, ObjectType, Property}, + schema::prelude::{OBJ_FILTER_ACCOUNT, OBJ_FILTER_TENANT, ObjectType, Property}, types::EnumImpl, }; use roaring::RoaringBitmap; @@ -25,30 +28,20 @@ use types::id::Id; impl RegistryStore { pub async fn query(&self, query: RegistryQuery) -> trc::Result { - let flags = query.object_type.flags(); - if flags & OBJ_SINGLETON != 0 { - if query.filters.is_empty() { - let mut results = T::default(); - results.push(Id::singleton().id()); - return Ok(results); - } else { - return Err(trc::EventType::Registry(trc::RegistryEvent::NotSupported) - .into_err() - .details("Singletons do not support searching")); - } - } else if query.filters.is_empty() { - return all_ids::(&self.0.store, query.object_type).await; + if query.filters.is_empty() { + return all_ids::(&self.0.store, query).await; } let mut u64_buffer; let mut u16_buffer; let mut bool_buffer = [0u8; 1]; - let mut results = T::default(); - for filter in query.filters { + let mut results = ResultsPagination::::new(&query); + for filter in &query.filters { if filter.op == RegistryFilterOp::TextMatch { - if let RegistryFilterValue::String(text) = filter.value { - let mut matches = T::default(); + if let RegistryFilterValue::String(text) = &filter.value { + let mut matches = ResultsPagination::::new(&query); + for word in text .split(|c: char| !c.is_alphanumeric()) .filter(|s| s.len() > 1) @@ -62,29 +55,32 @@ impl RegistryStore { Cow::Owned(word.to_lowercase()) }; - let result = index_range( + let mut result = ResultsPagination::::new(&query); + + index_range( &self.0.store, query.object_type, filter.property.to_id(), word.as_bytes(), RegistryFilterOp::Equal, + &mut result, ) .await?; - if !matches.has_items() { + if !matches.list.has_items() { matches = result; } else { - matches.intersect(&result); - if !matches.has_items() { + matches.list.intersect(&result.list); + if !matches.list.has_items() { break; } } } - if !results.has_items() { + if !results.list.has_items() { results = matches; } else { - results.intersect(&matches); + results.list.intersect(&matches.list); } } else { return Err(trc::EventType::Registry(trc::RegistryEvent::NotSupported) @@ -109,13 +105,15 @@ impl RegistryStore { } }; - let result = if !filter.is_pk { + let mut result = ResultsPagination::::new(&query); + if !filter.is_pk { index_range( &self.0.store, query.object_type, filter.property.to_id(), value, filter.op, + &mut result, ) .await? } else { @@ -125,36 +123,40 @@ impl RegistryStore { filter.property.to_id(), value, filter.op, + &mut result, ) .await? }; - if !results.has_items() { + if !results.list.has_items() { results = result; } else { - results.intersect(&result); + results.list.intersect(&result.list); } } - if !results.has_items() { - return Ok(results); + if !results.list.has_items() { + break; } } - Ok(results) + Ok(results.finalize()) } - pub async fn count(&self, query: RegistryQuery) -> trc::Result { - self.query::>(query).await.map(|r| r.len()) + pub async fn count_object(&self, object_type: ObjectType) -> trc::Result { + self.query::(RegistryQuery::new(object_type)) + .await + .map(|r| r.0) } pub async fn sort_by_index( &self, object: ObjectType, property: Property, - mut ids: Option>, + ids: Option>, ascending: bool, - ) -> trc::Result> { + ) -> trc::Result> { + let mut ids = ids.map(|ids| ids.into_iter().collect::>()); let mut ids_sorted = Vec::with_capacity(ids.as_ref().map_or(0, |ids| ids.len())); let object_id = object.to_id(); @@ -182,7 +184,7 @@ impl RegistryStore { .no_values() .set_ascending(ascending), |key, _| { - let id = key.deserialize_be_u64(key.len() - U64_LEN)?; + let id = Id::from(key.deserialize_be_u64(key.len() - U64_LEN)?); if let Some(ids) = ids.as_mut() { if ids.remove(&id) { ids_sorted.push(id); @@ -211,9 +213,10 @@ impl RegistryStore { &self, object: ObjectType, property: Property, - mut ids: Option>, + ids: Option>, ascending: bool, - ) -> trc::Result> { + ) -> trc::Result> { + let mut ids = ids.map(|ids| ids.into_iter().collect::>()); let mut ids_sorted = Vec::with_capacity(ids.as_ref().map_or(0, |ids| ids.len())); let object_id = object.to_id(); @@ -239,7 +242,7 @@ impl RegistryStore { .iterate( IterateParams::new(begin, end).set_ascending(ascending), |_, value| { - let id = value.deserialize_be_u64(U16_LEN)?; + let id = Id::from(value.deserialize_be_u64(U16_LEN)?); if let Some(ids) = ids.as_mut() { if ids.remove(&id) { @@ -266,15 +269,22 @@ impl RegistryStore { } } -async fn all_ids(store: &Store, object: ObjectType) -> trc::Result { +async fn all_ids(store: &Store, query: RegistryQuery) -> trc::Result { let mut bm = T::default(); - let object_id = object.to_id(); + let object_id = query.object_type.to_id(); + + let (item_id, mut offset) = match query.start { + RegistryQueryStart::Index(index) => (0, index), + RegistryQueryStart::Anchor(anchor) => (anchor + 1, 0), + RegistryQueryStart::None => (0, 0), + }; + store .iterate( IterateParams::new( ValueKey::from(ValueClass::Registry(RegistryClass::IndexId { object_id, - item_id: 0u64, + item_id, })), ValueKey::from(ValueClass::Registry(RegistryClass::IndexId { object_id, @@ -284,11 +294,15 @@ async fn all_ids(store: &Store, object: ObjectType) -> .no_values() .ascending(), |key, _| { - if key.len() == U64_LEN + U16_LEN { - bm.push(key.deserialize_be_u64(key.len() - U64_LEN)?); + if offset == 0 { + if key.len() == U64_LEN + U16_LEN { + bm.push(key.deserialize_be_u64(key.len() - U64_LEN)?); + } + Ok(query.limit.is_none_or(|limit| bm.count() < limit)) + } else { + offset -= 1; + Ok(true) } - - Ok(true) }, ) .await @@ -302,7 +316,8 @@ async fn index_range( index_id: u16, match_value: &[u8], op: RegistryFilterOp, -) -> trc::Result { + results: &mut ResultsPagination, +) -> trc::Result<()> { let object_id = object.to_id(); let ((from_value, from_doc_id, from_index_id), (end_value, end_doc_id, end_index_id)) = match op { @@ -343,7 +358,6 @@ async fn index_range( .finalize(), })); - let mut bm = T::default(); let prefix = KeySerializer::new(U16_LEN * 2) .write(object_id) .write(index_id) @@ -372,15 +386,15 @@ async fn index_range( }; if matches { - bm.push(key.deserialize_be_u64(id_pos)?); + Ok(results.push(key.deserialize_be_u64(id_pos)?)) + } else { + Ok(true) } - - Ok(true) }, ) .await .caused_by(trc::location!()) - .map(|_| bm) + .inspect(|_| results.list.sort()) } async fn pk_range( @@ -389,7 +403,8 @@ async fn pk_range( index_id: u16, match_value: &[u8], op: RegistryFilterOp, -) -> trc::Result { + results: &mut ResultsPagination, +) -> 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)), @@ -418,7 +433,6 @@ async fn pk_range( .finalize(), })); - let mut bm = T::default(); let prefix = KeySerializer::new(U16_LEN * 2) .write(object_id) .write(index_id) @@ -445,25 +459,28 @@ async fn pk_range( }; if matches { - bm.push(value.deserialize_be_u64(U16_LEN)?); + Ok(results.push(value.deserialize_be_u64(U16_LEN)?)) + } else { + Ok(true) } - - Ok(true) }) .await .caused_by(trc::location!()) - .map(|_| bm) + .inspect(|_| results.list.sort()) } pub trait RegistryQueryResults: Default + Sized + Sync + Send { fn push(&mut self, id: u64); fn has_items(&self) -> bool; fn intersect(&mut self, other: &Self); + fn count(&self) -> usize; + fn sort(&mut self); + fn into_list(self) -> impl Iterator; } -impl RegistryQueryResults for AHashSet { +impl RegistryQueryResults for Vec { fn push(&mut self, id: u64) { - self.insert(id); + self.push(Id::new(id)); } fn has_items(&self) -> bool { @@ -471,7 +488,45 @@ impl RegistryQueryResults for AHashSet { } fn intersect(&mut self, other: &Self) { - self.retain(|id| other.contains(id)); + let a = self; + let b = other; + let mut i = 0; + let mut j = 0; + let mut write = 0; + + while i < a.len() && j < b.len() { + if a[i] < b[j] { + let target = b[j]; + let remain = &a[i..]; + i += remain.partition_point(|&x| x < target); + } else if a[i] > b[j] { + let target = a[i]; + let remain = &b[j..]; + j += remain.partition_point(|&x| x < target); + } else { + a[write] = a[i]; + write += 1; + i += 1; + j += 1; + } + } + a.truncate(write); + } + + fn count(&self) -> usize { + self.len() + } + + fn sort(&mut self) { + match self.len() { + 0 | 1 => {} + ..3000 => self.sort_unstable(), + _ => radsort::sort_by_key(self, |id| id.id()), + } + } + + fn into_list(self) -> impl Iterator { + self.into_iter().map(|id| id.id()) } } @@ -487,6 +542,112 @@ impl RegistryQueryResults for RoaringBitmap { fn intersect(&mut self, other: &Self) { self.bitand_assign(other); } + + fn count(&self) -> usize { + self.len() as usize + } + + fn sort(&mut self) {} + + fn into_list(self) -> impl Iterator { + self.into_iter().map(|id| id as u64) + } +} + +impl RegistryQueryResults for RegistryObjectCounter { + fn push(&mut self, _: u64) { + self.0 += 1; + } + + fn has_items(&self) -> bool { + self.0 > 0 + } + + fn intersect(&mut self, _: &Self) { + unimplemented!() + } + + fn count(&self) -> usize { + self.0 + } + + fn sort(&mut self) {} + + fn into_list(self) -> impl Iterator { + Vec::new().into_iter() + } +} + +struct ResultsPagination { + list: T, + offset: usize, + anchor: Option, + limit: Option, + deferred_pagination: bool, +} + +impl ResultsPagination { + fn new(query: &RegistryQuery) -> Self { + let (anchor, offset) = match query.start { + RegistryQueryStart::Index(index) => (None, index), + RegistryQueryStart::Anchor(anchor) => (Some(anchor), 0), + RegistryQueryStart::None => (None, 0), + }; + + Self { + list: T::default(), + offset: offset as usize, + anchor, + limit: query.limit, + deferred_pagination: query.filters.len() > 1 + || query.filters.first().is_some_and(|f| { + if let (RegistryFilterOp::TextMatch, RegistryFilterValue::String(value)) = + (&f.op, &f.value) + { + value.chars().any(|c| !c.is_alphanumeric()) && value.len() > 1 + } else { + false + } + }), + } + } + + fn push(&mut self, id: u64) -> bool { + if !self.deferred_pagination { + if self.offset > 0 { + self.offset -= 1; + true + } else if let Some(anchor) = self.anchor { + if id == anchor { + self.anchor = None; + } + true + } else { + self.list.push(id); + self.limit.is_none_or(|limit| self.list.count() < limit) + } + } else { + self.list.push(id); + true + } + } + + fn finalize(mut self) -> T { + if self.deferred_pagination + && self.list.has_items() + && (self.limit.is_some() || self.anchor.is_some() || self.offset > 0) + { + let list = std::mem::take(&mut self.list); + self.deferred_pagination = false; + + for item in list.into_list() { + if !self.push(item) { + break; + } + } + } + self.list + } } impl RegistryQuery { @@ -494,9 +655,26 @@ impl RegistryQuery { Self { object_type, filters: Vec::new(), + start: RegistryQueryStart::None, + limit: None, } } + pub fn with_anchor(mut self, anchor: u64) -> Self { + self.start = RegistryQueryStart::Anchor(anchor); + self + } + + pub fn with_index_start(mut self, index: u64) -> Self { + self.start = RegistryQueryStart::Index(index); + self + } + + pub fn with_limit(mut self, limit: usize) -> Self { + self.limit = Some(limit); + self + } + 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); diff --git a/crates/store/src/registry/write.rs b/crates/store/src/registry/write.rs index f1a66f12..7ac73e3b 100644 --- a/crates/store/src/registry/write.rs +++ b/crates/store/src/registry/write.rs @@ -65,7 +65,7 @@ pub enum RegistryWrite<'x> { Delete { object_id: ObjectId, object: Option<&'x Object>, - force: bool, + allowed_orphan_types: &'x [ObjectType], }, } @@ -141,10 +141,10 @@ impl RegistryStore { RegistryWrite::Delete { object_id, object, - force, + allowed_orphan_types, } => { return if object_id.object().flags() & OBJ_SINGLETON == 0 { - self.delete(object_id, object, force).await + self.delete(object_id, object, allowed_orphan_types).await } else { Ok(RegistryWriteResult::CannotDeleteSingleton) }; @@ -319,7 +319,7 @@ impl RegistryStore { &self, object_id: ObjectId, object: Option<&Object>, - force: bool, + allowed_orphan_types: &[ObjectType], ) -> trc::Result { let object_type = object_id.object(); let object_type_id = object_type.to_id(); @@ -342,8 +342,12 @@ impl RegistryStore { object.index(&mut clear_index); // Validate relationships - if !force { - let linked = self.linked_objects(object_id).await?; + let mut linked = self.linked_objects(object_id).await?; + if !linked.is_empty() { + if !allowed_orphan_types.is_empty() { + linked.retain(|object_id| !allowed_orphan_types.contains(&object_id.object())); + } + if !linked.is_empty() { return Ok(RegistryWriteResult::CannotDeleteLinked { object_id: ObjectId::new(object_type, id), @@ -542,7 +546,7 @@ impl<'x> RegistryWrite<'x> { RegistryWrite::Delete { object_id, object: None, - force: false, + allowed_orphan_types: &[], } } @@ -550,7 +554,7 @@ impl<'x> RegistryWrite<'x> { RegistryWrite::Delete { object_id, object: Some(object), - force: false, + allowed_orphan_types: &[], } } }