diff --git a/Cargo.lock b/Cargo.lock index a4a9773e..c3d66d91 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6020,6 +6020,7 @@ dependencies = [ "trc", "types", "utils", + "xxhash-rust", ] [[package]] diff --git a/crates/common/src/auth/access_token.rs b/crates/common/src/auth/access_token.rs index 61f401a3..05476a83 100644 --- a/crates/common/src/auth/access_token.rs +++ b/crates/common/src/auth/access_token.rs @@ -19,7 +19,7 @@ use registry::{ enums::Permission, structs::{self, Account}, }, - types::EnumType, + types::EnumImpl, }; use std::{ hash::{Hash, Hasher}, diff --git a/crates/common/src/auth/mod.rs b/crates/common/src/auth/mod.rs index 8eb88c66..07b5afb6 100644 --- a/crates/common/src/auth/mod.rs +++ b/crates/common/src/auth/mod.rs @@ -14,7 +14,7 @@ use directory::Credentials; use quick_cache::Equivalent; use registry::{ schema::enums::{Locale, Permission}, - types::EnumType, + types::EnumImpl, }; use std::{ hash::{Hash, Hasher}, diff --git a/crates/common/src/auth/oauth/config.rs b/crates/common/src/auth/oauth/config.rs index a917a526..f49fef87 100644 --- a/crates/common/src/auth/oauth/config.rs +++ b/crates/common/src/auth/oauth/config.rs @@ -17,7 +17,11 @@ use biscuit::{ }, jws::Secret, }; -use registry::schema::{enums::JwtSignatureAlgorithm, prelude::Object, structs::OidcProvider}; +use registry::schema::{ + enums::JwtSignatureAlgorithm, + prelude::{Object, ObjectType}, + structs::OidcProvider, +}; use ring::signature::{self, KeyPair}; use rsa::{RsaPublicKey, pkcs1::DecodeRsaPublicKey, traits::PublicKeyParts}; use store::{ @@ -88,7 +92,7 @@ impl OAuthConfig { | SignatureAlgorithm::PS384 | SignatureAlgorithm::PS512 => parse_rsa_key(&auth) .map_err(|err| { - bp.build_error(Object::OidcProvider.singleton(), err); + bp.build_error(ObjectType::OidcProvider.singleton(), err); }) .unwrap_or_else(|_| { ( @@ -102,7 +106,7 @@ impl OAuthConfig { SignatureAlgorithm::ES256 | SignatureAlgorithm::ES384 | SignatureAlgorithm::ES512 => { parse_ecdsa_key(&auth, oidc_signature_algorithm) .map_err(|err| { - bp.build_error(Object::OidcProvider.singleton(), err); + bp.build_error(ObjectType::OidcProvider.singleton(), err); }) .unwrap_or_else(|_| { ( diff --git a/crates/common/src/cache/directory.rs b/crates/common/src/cache/directory.rs index 0c615e01..3b0aa556 100644 --- a/crates/common/src/cache/directory.rs +++ b/crates/common/src/cache/directory.rs @@ -8,13 +8,13 @@ use std::sync::Arc; use crate::{Server, auth::DomainCache}; use registry::{ - schema::structs::{Account, EmailAlias, GroupAccount, UserAccount}, - types::datetime::UTCDateTime, -}; -use store::registry::{ - HashedObject, - write::{RegistryWrite, RegistryWriteResult}, + schema::{ + prelude::{Object, ObjectType}, + structs::{Account, EmailAlias, GroupAccount, UserAccount}, + }, + types::{datetime::UTCDateTime, id::ObjectId}, }; +use store::registry::write::{RegistryWrite, RegistryWriteResult}; use trc::AddContext; use types::id::Id; @@ -37,7 +37,7 @@ impl Server { Some(account_id) => { let current_account = self .registry() - .object::>(Id::from(account_id)) + .get(ObjectId::new(ObjectType::Account, account_id.into())) .await .caused_by(trc::location!())? .ok_or_else(|| { @@ -47,8 +47,9 @@ impl Server { .ctx(trc::Key::AccountName, account.email.clone()) .ctx(trc::Key::AccountId, account_id) })?; - let mut updated_account = - current_account.object.clone().into_user().ok_or_else(|| { + let mut updated_account = Account::from(current_account.clone()) + .into_user() + .ok_or_else(|| { trc::AuthEvent::Error .into_err() .details( @@ -111,7 +112,7 @@ impl Server { } if has_changes { - let updated_account = Account::User(updated_account); + let updated_account = Object::from(Account::User(updated_account)); match self .registry() .write(RegistryWrite::update( @@ -124,7 +125,7 @@ impl Server { { RegistryWriteResult::Success(id) => Ok(AccountWithId { id: id.document_id(), - account: updated_account, + account: updated_account.into(), }), failure => Err(trc::AuthEvent::Error .into_err() @@ -135,7 +136,7 @@ impl Server { } else { Ok(AccountWithId { id: account_id, - account: current_account.object, + account: Account::User(updated_account), }) } } @@ -169,7 +170,7 @@ impl Server { .into(), ); } - let account = Account::User(UserAccount { + let account = Object::from(Account::User(UserAccount { name: local.to_string(), domain_id: Id::from(domain.id), aliases, @@ -180,7 +181,7 @@ impl Server { role_ids: self.core.network.security.default_role_ids_user.clone(), secret: account.secret.unwrap_or_default(), ..Default::default() - }); + })); match self .registry() @@ -190,7 +191,7 @@ impl Server { { RegistryWriteResult::Success(id) => Ok(AccountWithId { id: id.document_id(), - account, + account: account.into(), }), failure => Err(trc::AuthEvent::Error .into_err() @@ -213,7 +214,7 @@ impl Server { Some(account_id) => { let current_account = self .registry() - .object::>(Id::from(account_id)) + .get(ObjectId::new(ObjectType::Account, account_id.into())) .await .caused_by(trc::location!())? .ok_or_else(|| { @@ -223,8 +224,9 @@ impl Server { .ctx(trc::Key::AccountName, group.email.clone()) .ctx(trc::Key::AccountId, account_id) })?; - let mut updated_account = - current_account.object.clone().into_group().ok_or_else(|| { + let mut updated_account = Account::from(current_account.clone()) + .into_group() + .ok_or_else(|| { trc::AuthEvent::Error .into_err() .details( @@ -257,12 +259,11 @@ impl Server { } if has_changes { - let updated_account = Account::Group(updated_account); match self .registry() .write(RegistryWrite::update( Id::from(account_id), - &updated_account, + &Object::from(Account::Group(updated_account)), ¤t_account, )) .await @@ -298,7 +299,7 @@ impl Server { } } - let account = Account::Group(GroupAccount { + let account = Object::from(Account::Group(GroupAccount { name: local.to_string(), domain_id: Id::from(domain.id), aliases, @@ -307,7 +308,7 @@ impl Server { member_tenant_id: domain.id_tenant.map(Id::from), role_ids: self.core.network.security.default_role_ids_group.clone(), ..Default::default() - }); + })); match self .registry() diff --git a/crates/common/src/cache/principals.rs b/crates/common/src/cache/principals.rs index 1f8ebd3f..ed9bee0b 100644 --- a/crates/common/src/cache/principals.rs +++ b/crates/common/src/cache/principals.rs @@ -21,7 +21,7 @@ use arcstr::ArcStr; use registry::{ schema::{ enums::{Locale, StorageQuota, TenantStorageQuota}, - prelude::{Object, Property}, + prelude::{ObjectType, Property}, structs::{ Account, DkimSignature, Domain, MailingList, MaskedEmail, Permissions, PermissionsList, Role, SubAddressing, Tenant, @@ -57,7 +57,7 @@ impl Server { if let Some(domain_id) = self .registry() .query::>( - RegistryQuery::new(Object::Domain).equal(Property::Name, domain), + RegistryQuery::new(ObjectType::Domain).equal(Property::Name, domain), ) .await? .into_iter() @@ -114,7 +114,7 @@ impl Server { flags |= DOMAIN_FLAG_SUB_ADDRESSING; let mut bp = Bootstrap::new(self.registry().clone()); let custom = bp.compile_expr( - ObjectId::new(Object::Domain, domain_id.into()), + ObjectId::new(ObjectType::Domain, domain_id.into()), &custom.ctx_custom_rule(), ); if bp.errors.is_empty() { @@ -177,8 +177,8 @@ impl Server { { let item_id = object.id().document_id(); let result = match object.object() { - Object::Account => EmailCache::Account(item_id), - Object::MailingList => EmailCache::MailingList(item_id), + ObjectType::Account => EmailCache::Account(item_id), + ObjectType::MailingList => EmailCache::MailingList(item_id), _ => { return Err(trc::AuthEvent::Error .into_err() @@ -595,7 +595,7 @@ impl Server { let ids = self .registry() .query::>( - RegistryQuery::new(Object::DkimSignature) + RegistryQuery::new(ObjectType::DkimSignature) .equal(Property::DomainId, domain.id), ) .await?; diff --git a/crates/common/src/cache/reload.rs b/crates/common/src/cache/reload.rs index 27f0ad4d..7706c322 100644 --- a/crates/common/src/cache/reload.rs +++ b/crates/common/src/cache/reload.rs @@ -15,7 +15,7 @@ use crate::{ }; use ahash::AHashMap; use directory::Directories; -use registry::schema::{prelude::Object, structs::BlockedIp}; +use registry::schema::{prelude::ObjectType, structs::BlockedIp}; use std::sync::Arc; use store::{InMemoryStore, LookupStores, registry::bootstrap::Bootstrap, write::now}; @@ -27,11 +27,11 @@ pub struct ReloadResult { impl Server { pub async fn reload_registry(&self, change: RegistryChange) -> trc::Result { - // TODO: check the different events triggering this, spam filter reload, etc. + let todo = "check the different events triggering this, spam filter reload, etc. make sure all are used"; let mut bootstrap = Bootstrap::init(self.registry().clone()).await; let object = match change { RegistryChange::Insert(id) => { - if matches!(id.object(), Object::BlockedIp) { + if matches!(id.object(), ObjectType::BlockedIp) { if let Some(ip) = bootstrap.get_infallible::(id.id()).await && ip.expires_at.is_none_or(|ip| ip.timestamp() > now() as i64) { @@ -62,7 +62,7 @@ impl Server { }; match object { - Object::Certificate => { + ObjectType::Certificate => { let mut certificates = AHashMap::new(); parse_certificates( &mut result.bootstrap, @@ -75,7 +75,7 @@ impl Server { .tls_certificates .store(Arc::new(certificates)); } - Object::MemoryLookupKey | Object::MemoryLookupKeyValue => { + ObjectType::MemoryLookupKey | ObjectType::MemoryLookupKeyValue => { let mut lookup = LookupStores { stores: self.inner.data.lookup_stores.load().as_ref().clone(), }; @@ -84,7 +84,7 @@ impl Server { .retain(|_, store| !matches!(store, InMemoryStore::Static(_))); lookup.parse_static(&mut result.bootstrap).await; } - Object::HttpLookup => { + ObjectType::HttpLookup => { let mut lookup = LookupStores { stores: self.inner.data.lookup_stores.load().as_ref().clone(), }; @@ -93,7 +93,7 @@ impl Server { .retain(|_, store| !matches!(store, InMemoryStore::Http(_))); lookup.parse_http(&mut result.bootstrap).await; } - Object::LookupStore => { + ObjectType::StoreLookup => { let mut lookup = LookupStores { stores: self.inner.data.lookup_stores.load().as_ref().clone(), }; diff --git a/crates/common/src/config/inner.rs b/crates/common/src/config/inner.rs index 8fe3e350..1cafcb2c 100644 --- a/crates/common/src/config/inner.rs +++ b/crates/common/src/config/inner.rs @@ -24,7 +24,10 @@ use arc_swap::ArcSwap; use mail_auth::{MX, Parameters, Txt}; use mail_send::smtp::tls::build_tls_connector; use parking_lot::RwLock; -use registry::schema::{prelude::Object, structs}; +use registry::schema::{ + prelude::{Object, ObjectType}, + structs, +}; use std::{ net::{IpAddr, Ipv4Addr, Ipv6Addr}, sync::Arc, @@ -68,7 +71,7 @@ impl Data { ) .or_else(|err| { bp.build_error( - Object::Certificate.singleton(), + ObjectType::Certificate.singleton(), format!("Failed to build self-signed TLS certificate: {err}"), ); build_self_signed_cert(vec!["localhost".to_string()]) diff --git a/crates/common/src/config/mailstore/capabilities.rs b/crates/common/src/config/mailstore/capabilities.rs index fe4b971a..b566b322 100644 --- a/crates/common/src/config/mailstore/capabilities.rs +++ b/crates/common/src/config/mailstore/capabilities.rs @@ -20,7 +20,7 @@ use jmap_proto::{ }; use registry::{ schema::structs::{Calendar, Email, SieveUserInterpreter}, - types::EnumType, + types::EnumImpl, }; use store::registry::bootstrap::Bootstrap; use types::type_state::DataType; diff --git a/crates/common/src/config/mailstore/email.rs b/crates/common/src/config/mailstore/email.rs index 5a2ce3d9..31cf680e 100644 --- a/crates/common/src/config/mailstore/email.rs +++ b/crates/common/src/config/mailstore/email.rs @@ -12,13 +12,13 @@ use registry::{ CompressionAlgo, SearchCalendarField, SearchContactField, SearchEmailField, StorageQuota, }, - prelude::Object, + prelude::ObjectType, structs::{ AddressBook, Authentication, Calendar, DataRetention, Domain, Email, Jmap, Search, SieveUserInterpreter, }, }, - types::EnumType, + types::EnumImpl, }; use std::time::Duration; use store::{ @@ -94,7 +94,7 @@ impl EmailConfig { default_domain.name } else { bp.build_error( - Object::Authentication.singleton(), + ObjectType::Authentication.singleton(), format!( "Default domain with ID {} not found", auth.default_domain_id diff --git a/crates/common/src/config/mailstore/scripts.rs b/crates/common/src/config/mailstore/scripts.rs index 3caade72..f93301ec 100644 --- a/crates/common/src/config/mailstore/scripts.rs +++ b/crates/common/src/config/mailstore/scripts.rs @@ -15,12 +15,12 @@ use crate::{ use ahash::AHashMap; use registry::{ schema::{ - prelude::Object, + prelude::ObjectType, structs::{ SieveSystemInterpreter, SieveSystemScript, SieveUserInterpreter, SieveUserScript, }, }, - types::EnumType, + types::EnumImpl, }; use sieve::{Compiler, Runtime, Sieve, compiler::grammar::Capability}; use std::sync::Arc; @@ -169,19 +169,19 @@ impl Scripting { untrusted_scripts, trusted_scripts, from_addr: bp.compile_expr( - Object::SieveSystemScript.singleton(), + ObjectType::SieveSystemScript.singleton(), &trusted.ctx_default_from_address(), ), from_name: bp.compile_expr( - Object::SieveSystemScript.singleton(), + ObjectType::SieveSystemScript.singleton(), &trusted.ctx_default_from_name(), ), return_path: bp.compile_expr( - Object::SieveSystemScript.singleton(), + ObjectType::SieveSystemScript.singleton(), &trusted.ctx_default_return_path(), ), sign: bp.compile_expr( - Object::SieveSystemScript.singleton(), + ObjectType::SieveSystemScript.singleton(), &trusted.ctx_dkim_sign_domain(), ), } diff --git a/crates/common/src/config/mailstore/spamfilter.rs b/crates/common/src/config/mailstore/spamfilter.rs index 52d78129..dd7ab967 100644 --- a/crates/common/src/config/mailstore/spamfilter.rs +++ b/crates/common/src/config/mailstore/spamfilter.rs @@ -14,7 +14,7 @@ use mail_auth::common::resolver::ToReverseName; use nlp::classifier::model::{CcfhClassifier, FhClassifier}; use registry::schema::{ enums::{ExpressionVariable, ModelSize}, - prelude::Object, + prelude::ObjectType, structs::{ self, SpamDnsblServer, SpamDnsblSettings, SpamFileExtension, SpamPyzor, SpamRule, SpamSettings, SpamTag, @@ -421,14 +421,14 @@ impl PyzorConfig { Ok(Some(address)) => address, Ok(None) => { bp.build_error( - Object::SpamPyzor.singleton(), + ObjectType::SpamPyzor.singleton(), "Invalid address: No addresses found.", ); return None; } Err(err) => { bp.build_error( - Object::SpamPyzor.singleton(), + ObjectType::SpamPyzor.singleton(), format!("Invalid address: {}", err), ); return None; diff --git a/crates/common/src/config/mod.rs b/crates/common/src/config/mod.rs index 4b868952..08fc3a1d 100644 --- a/crates/common/src/config/mod.rs +++ b/crates/common/src/config/mod.rs @@ -37,26 +37,26 @@ impl Core { let enterprise = { let enterprise = crate::enterprise::Enterprise::parse(bp).await; if enterprise.is_none() { - use registry::schema::prelude::Object; + use registry::schema::prelude::ObjectType; use store::Store; if storage.data.is_enterprise() { bp.build_error( - Object::DataStore.singleton(), + ObjectType::DataStore.singleton(), "Disabling enterprise-only data store.", ); storage.data = storage.data.downgrade_store(); } if storage.blob.is_enterprise() { bp.build_error( - Object::BlobStore.singleton(), + ObjectType::BlobStore.singleton(), "Disabling enterprise-only blob store.", ); storage.blob = storage.blob.downgrade_store(); } if storage.memory.is_enterprise() { bp.build_error( - Object::InMemoryStore.singleton(), + ObjectType::InMemoryStore.singleton(), "Disabling enterprise-only in-memory store.", ); storage.memory = storage.memory.downgrade_store(); diff --git a/crates/common/src/config/network.rs b/crates/common/src/config/network.rs index 92713575..09a1859e 100644 --- a/crates/common/src/config/network.rs +++ b/crates/common/src/config/network.rs @@ -13,10 +13,10 @@ use ahash::AHashMap; use registry::{ schema::{ enums::NodeShardType, - prelude::Object, + prelude::{Object, ObjectType}, structs::{self, Asn, HttpForm, NodeRole, NodeShard, Rate}, }, - types::EnumType, + types::EnumImpl, }; use std::{hash::Hasher, str::FromStr, time::Duration}; use xxhash_rust::xxh3::Xxh3Builder; @@ -117,7 +117,7 @@ impl ContactForm { return None; } else if form.deliver_to.is_empty() { bp.build_error( - Object::HttpForm.singleton(), + ObjectType::HttpForm.singleton(), "Contact form is enabled but no recipient addresses are configured", ); return None; @@ -331,7 +331,7 @@ impl Http { .collect::, String>>() .map_err(|e| { bp.build_error( - Object::Http.singleton(), + ObjectType::Http.singleton(), format!("Failed to parse HTTP headers: {}", e), ) }) @@ -366,9 +366,9 @@ impl Http { } Http { - response_url: bp.compile_expr(Object::Http.singleton(), &http.ctx_base_url()), + response_url: bp.compile_expr(ObjectType::Http.singleton(), &http.ctx_base_url()), allowed_endpoint: bp - .compile_expr(Object::Http.singleton(), &http.ctx_allowed_endpoints()), + .compile_expr(ObjectType::Http.singleton(), &http.ctx_allowed_endpoints()), rate_authenticated: http.rate_limit_authenticated, rate_anonymous: http.rate_limit_anonymous, response_headers: http_headers, @@ -389,7 +389,7 @@ impl AsnGeoLookupConfig { .build_headers(asn.http_headers, None) .map_err(|err| { bp.build_error( - Object::Asn.singleton(), + ObjectType::Asn.singleton(), format!("Unable to build HTTP headers: {}", err), ) }) diff --git a/crates/common/src/config/smtp/auth.rs b/crates/common/src/config/smtp/auth.rs index 70d7c23b..fd7c6c76 100644 --- a/crates/common/src/config/smtp/auth.rs +++ b/crates/common/src/config/smtp/auth.rs @@ -16,10 +16,10 @@ use mail_parser::decoders::base64::base64_decode; use registry::{ schema::{ enums::{self, ExpressionConstant}, - prelude::Object, + prelude::ObjectType, structs::{Dkim1Signature, DkimSignature, SenderAuth}, }, - types::ObjectType, + types::ObjectImpl, }; use rustls_pki_types::{PrivateKeyDer, PrivatePkcs1KeyDer, PrivatePkcs8KeyDer, pem::PemObject}; use store::registry::bootstrap::Bootstrap; @@ -87,26 +87,35 @@ impl MailAuthConfig { MailAuthConfig { dkim: DkimAuthConfig { - verify: bp.compile_expr(Object::SenderAuth.singleton(), &auth.ctx_dkim_verify()), - sign: bp.compile_expr(Object::SenderAuth.singleton(), &auth.ctx_dkim_sign_domain()), + verify: bp + .compile_expr(ObjectType::SenderAuth.singleton(), &auth.ctx_dkim_verify()), + sign: bp.compile_expr( + ObjectType::SenderAuth.singleton(), + &auth.ctx_dkim_sign_domain(), + ), strict: auth.dkim_strict, }, arc: ArcAuthConfig { - verify: bp.compile_expr(Object::SenderAuth.singleton(), &auth.ctx_arc_verify()), - //seal: bp.compile_expr(Object::SenderAuth.singleton(), &auth.ctx_arc_seal_domain()), + verify: bp.compile_expr(ObjectType::SenderAuth.singleton(), &auth.ctx_arc_verify()), + //seal: bp.compile_expr(ObjectType::SenderAuth.singleton(), &auth.ctx_arc_seal_domain()), }, spf: SpfAuthConfig { - verify_ehlo: bp - .compile_expr(Object::SenderAuth.singleton(), &auth.ctx_spf_ehlo_verify()), - verify_mail_from: bp - .compile_expr(Object::SenderAuth.singleton(), &auth.ctx_spf_from_verify()), + verify_ehlo: bp.compile_expr( + ObjectType::SenderAuth.singleton(), + &auth.ctx_spf_ehlo_verify(), + ), + verify_mail_from: bp.compile_expr( + ObjectType::SenderAuth.singleton(), + &auth.ctx_spf_from_verify(), + ), }, dmarc: DmarcAuthConfig { - verify: bp.compile_expr(Object::SenderAuth.singleton(), &auth.ctx_dmarc_verify()), + verify: bp + .compile_expr(ObjectType::SenderAuth.singleton(), &auth.ctx_dmarc_verify()), }, iprev: IpRevAuthConfig { verify: bp.compile_expr( - Object::SenderAuth.singleton(), + ObjectType::SenderAuth.singleton(), &auth.ctx_reverse_ip_verify(), ), }, @@ -116,20 +125,20 @@ impl MailAuthConfig { impl DkimSigner { pub fn new(domain: String, signature: DkimSignature) -> trc::Result { + let mut errors = vec![]; + if !signature.validate(&mut errors) { + return Err(trc::DkimEvent::BuildError + .reason("DKIM signature validation failed") + .details( + errors + .into_iter() + .map(|v| trc::Value::from(v.to_string())) + .collect::>(), + )); + } + match signature { DkimSignature::Dkim1Ed25519Sha256(signature) => { - let mut errors = vec![]; - if !signature.validate(&mut errors) { - return Err(trc::DkimEvent::BuildError - .reason("DKIM signature validation failed") - .details( - errors - .into_iter() - .map(|v| trc::Value::from(v.to_string())) - .collect::>(), - )); - } - let private_key = simple_pem_parse(&signature.private_key).ok_or_else(|| { trc::DkimEvent::BuildError .reason("Failed to parse ED25519 private key PEM") @@ -147,18 +156,6 @@ impl DkimSigner { ))) } DkimSignature::Dkim1RsaSha256(signature) => { - let mut errors = vec![]; - if !signature.validate(&mut errors) { - return Err(trc::DkimEvent::BuildError - .reason("DKIM signature validation failed") - .details( - errors - .into_iter() - .map(|v| trc::Value::from(v.to_string())) - .collect::>(), - )); - } - let key = PrivatePkcs1KeyDer::from_pem_slice(signature.private_key.as_bytes()) .map(PrivateKeyDer::Pkcs1) .or_else(|_| { @@ -188,20 +185,20 @@ impl DkimSigner { impl ArcSealer { pub fn new(selector: String, domain: String, signature: DkimSignature) -> trc::Result { + let mut errors = vec![]; + if !signature.validate(&mut errors) { + return Err(trc::DkimEvent::BuildError + .reason("DKIM signature validation failed") + .details( + errors + .into_iter() + .map(|v| trc::Value::from(v.to_string())) + .collect::>(), + )); + } + match signature { DkimSignature::Dkim1Ed25519Sha256(signature) => { - let mut errors = vec![]; - if !signature.validate(&mut errors) { - return Err(trc::DkimEvent::BuildError - .reason("DKIM signature validation failed") - .details( - errors - .into_iter() - .map(|v| trc::Value::from(v.to_string())) - .collect::>(), - )); - } - let private_key = simple_pem_parse(&signature.private_key).ok_or_else(|| { trc::DkimEvent::BuildError .reason("Failed to parse ED25519 private key PEM") @@ -219,18 +216,6 @@ impl ArcSealer { ))) } DkimSignature::Dkim1RsaSha256(signature) => { - let mut errors = vec![]; - if !signature.validate(&mut errors) { - return Err(trc::DkimEvent::BuildError - .reason("DKIM signature validation failed") - .details( - errors - .into_iter() - .map(|v| trc::Value::from(v.to_string())) - .collect::>(), - )); - } - let key = PrivatePkcs1KeyDer::from_pem_slice(signature.private_key.as_bytes()) .map(PrivateKeyDer::Pkcs1) .or_else(|_| { diff --git a/crates/common/src/config/smtp/queue.rs b/crates/common/src/config/smtp/queue.rs index 9ff31139..f430e5ef 100644 --- a/crates/common/src/config/smtp/queue.rs +++ b/crates/common/src/config/smtp/queue.rs @@ -17,7 +17,7 @@ use mail_auth::IpLookupStrategy; use mail_send::Credentials; use registry::schema::{ enums::{self, ExpressionConstant, ExpressionVariable, MtaRequiredOrOptional}, - prelude::Object, + prelude::ObjectType, structs::{ DsnReportSettings, MtaConnectionStrategy, MtaDeliveryExpiration, MtaDeliverySchedule, MtaInboundThrottle, MtaOutboundStrategy, MtaOutboundThrottle, MtaQueueQuota, MtaRoute, @@ -201,21 +201,27 @@ impl QueueConfig { let dsn = bp.setting_infallible::().await; let mut queue = QueueConfig { - route: bp.compile_expr(Object::MtaOutboundStrategy.singleton(), &st.ctx_route()), - queue: bp.compile_expr(Object::MtaOutboundStrategy.singleton(), &st.ctx_schedule()), + route: bp.compile_expr(ObjectType::MtaOutboundStrategy.singleton(), &st.ctx_route()), + queue: bp.compile_expr( + ObjectType::MtaOutboundStrategy.singleton(), + &st.ctx_schedule(), + ), connection: bp.compile_expr( - Object::MtaOutboundStrategy.singleton(), + ObjectType::MtaOutboundStrategy.singleton(), &st.ctx_connection(), ), - tls: bp.compile_expr(Object::MtaOutboundStrategy.singleton(), &st.ctx_tls()), + tls: bp.compile_expr(ObjectType::MtaOutboundStrategy.singleton(), &st.ctx_tls()), dsn: Dsn { - name: bp.compile_expr(Object::DsnReportSettings.singleton(), &dsn.ctx_from_name()), + name: bp.compile_expr( + ObjectType::DsnReportSettings.singleton(), + &dsn.ctx_from_name(), + ), address: bp.compile_expr( - Object::DsnReportSettings.singleton(), + ObjectType::DsnReportSettings.singleton(), &dsn.ctx_from_address(), ), sign: bp.compile_expr( - Object::DsnReportSettings.singleton(), + ObjectType::DsnReportSettings.singleton(), &dsn.ctx_dkim_sign_domain(), ), }, diff --git a/crates/common/src/config/smtp/report.rs b/crates/common/src/config/smtp/report.rs index 4057d4f1..c533a3aa 100644 --- a/crates/common/src/config/smtp/report.rs +++ b/crates/common/src/config/smtp/report.rs @@ -11,7 +11,7 @@ use crate::expr::{ }; use registry::schema::{ enums::ExpressionConstant, - prelude::Object, + prelude::ObjectType, structs::{ DataRetention, DkimReportSettings, DmarcReportSettings, ReportSettings, SpfReportSettings, TlsReportSettings, @@ -85,7 +85,7 @@ impl ReportConfig { ReportConfig { submitter: bp.compile_expr( - Object::ReportSettings.singleton(), + ObjectType::ReportSettings.singleton(), &report.ctx_outbound_report_submitter(), ), analysis: ReportAnalysis { @@ -99,114 +99,114 @@ impl ReportConfig { }, dkim: Report { name: bp.compile_expr( - Object::DkimReportSettings.singleton(), + ObjectType::DkimReportSettings.singleton(), &dkim.ctx_from_name(), ), address: bp.compile_expr( - Object::DkimReportSettings.singleton(), + ObjectType::DkimReportSettings.singleton(), &dkim.ctx_from_address(), ), subject: bp - .compile_expr(Object::DkimReportSettings.singleton(), &dkim.ctx_subject()), + .compile_expr(ObjectType::DkimReportSettings.singleton(), &dkim.ctx_subject()), sign: bp.compile_expr( - Object::DkimReportSettings.singleton(), + ObjectType::DkimReportSettings.singleton(), &dkim.ctx_dkim_sign_domain(), ), send: bp.compile_expr( - Object::DkimReportSettings.singleton(), + ObjectType::DkimReportSettings.singleton(), &dkim.ctx_send_frequency(), ), }, spf: Report { - name: bp.compile_expr(Object::SpfReportSettings.singleton(), &spf.ctx_from_name()), + name: bp.compile_expr(ObjectType::SpfReportSettings.singleton(), &spf.ctx_from_name()), address: bp.compile_expr( - Object::SpfReportSettings.singleton(), + ObjectType::SpfReportSettings.singleton(), &spf.ctx_from_address(), ), - subject: bp.compile_expr(Object::SpfReportSettings.singleton(), &spf.ctx_subject()), + subject: bp.compile_expr(ObjectType::SpfReportSettings.singleton(), &spf.ctx_subject()), sign: bp.compile_expr( - Object::SpfReportSettings.singleton(), + ObjectType::SpfReportSettings.singleton(), &spf.ctx_dkim_sign_domain(), ), send: bp.compile_expr( - Object::SpfReportSettings.singleton(), + ObjectType::SpfReportSettings.singleton(), &spf.ctx_send_frequency(), ), }, dmarc: Report { name: bp.compile_expr( - Object::DmarcReportSettings.singleton(), + ObjectType::DmarcReportSettings.singleton(), &dmarc.ctx_failure_from_name(), ), address: bp.compile_expr( - Object::DmarcReportSettings.singleton(), + ObjectType::DmarcReportSettings.singleton(), &dmarc.ctx_failure_from_address(), ), subject: bp.compile_expr( - Object::DmarcReportSettings.singleton(), + ObjectType::DmarcReportSettings.singleton(), &dmarc.ctx_failure_subject(), ), sign: bp.compile_expr( - Object::DmarcReportSettings.singleton(), + ObjectType::DmarcReportSettings.singleton(), &dmarc.ctx_failure_dkim_sign_domain(), ), send: bp.compile_expr( - Object::DmarcReportSettings.singleton(), + ObjectType::DmarcReportSettings.singleton(), &dmarc.ctx_failure_send_frequency(), ), }, dmarc_aggregate: AggregateReport { name: bp.compile_expr( - Object::DmarcReportSettings.singleton(), + ObjectType::DmarcReportSettings.singleton(), &dmarc.ctx_aggregate_from_name(), ), address: bp.compile_expr( - Object::DmarcReportSettings.singleton(), + ObjectType::DmarcReportSettings.singleton(), &dmarc.ctx_aggregate_from_address(), ), org_name: bp.compile_expr( - Object::DmarcReportSettings.singleton(), + ObjectType::DmarcReportSettings.singleton(), &dmarc.ctx_aggregate_org_name(), ), contact_info: bp.compile_expr( - Object::DmarcReportSettings.singleton(), + ObjectType::DmarcReportSettings.singleton(), &dmarc.ctx_aggregate_contact_info(), ), send: bp.compile_expr( - Object::DmarcReportSettings.singleton(), + ObjectType::DmarcReportSettings.singleton(), &dmarc.ctx_aggregate_send_frequency(), ), sign: bp.compile_expr( - Object::DmarcReportSettings.singleton(), + ObjectType::DmarcReportSettings.singleton(), &dmarc.ctx_aggregate_dkim_sign_domain(), ), max_size: bp.compile_expr( - Object::DmarcReportSettings.singleton(), + ObjectType::DmarcReportSettings.singleton(), &dmarc.ctx_aggregate_max_report_size(), ), }, tls: AggregateReport { - name: bp.compile_expr(Object::TlsReportSettings.singleton(), &tls.ctx_from_name()), + name: bp.compile_expr(ObjectType::TlsReportSettings.singleton(), &tls.ctx_from_name()), address: bp.compile_expr( - Object::TlsReportSettings.singleton(), + ObjectType::TlsReportSettings.singleton(), &tls.ctx_from_address(), ), org_name: bp - .compile_expr(Object::TlsReportSettings.singleton(), &tls.ctx_org_name()), + .compile_expr(ObjectType::TlsReportSettings.singleton(), &tls.ctx_org_name()), contact_info: bp.compile_expr( - Object::TlsReportSettings.singleton(), + ObjectType::TlsReportSettings.singleton(), &tls.ctx_contact_info(), ), send: bp.compile_expr( - Object::TlsReportSettings.singleton(), + ObjectType::TlsReportSettings.singleton(), &tls.ctx_send_frequency(), ), sign: bp.compile_expr( - Object::TlsReportSettings.singleton(), + ObjectType::TlsReportSettings.singleton(), &tls.ctx_dkim_sign_domain(), ), max_size: bp.compile_expr( - Object::TlsReportSettings.singleton(), + ObjectType::TlsReportSettings.singleton(), &tls.ctx_max_report_size(), ), }, diff --git a/crates/common/src/config/smtp/resolver.rs b/crates/common/src/config/smtp/resolver.rs index 9264ac1d..b5970ce0 100644 --- a/crates/common/src/config/smtp/resolver.rs +++ b/crates/common/src/config/smtp/resolver.rs @@ -16,7 +16,7 @@ use mail_auth::{ }; use registry::schema::{ enums::{DnsResolverProtocol, PolicyEnforcement}, - prelude::Object, + prelude::ObjectType, structs::{DnsResolver, MtaSts}, }; use serde::{Deserialize, Serialize}; @@ -122,7 +122,7 @@ impl Resolvers { } Err(err) => { bp.build_error( - Object::DnsResolver.singleton(), + ObjectType::DnsResolver.singleton(), format!("Failed to read system DNS config: {err}"), ); resolver_config = ResolverConfig::cloudflare(); diff --git a/crates/common/src/config/smtp/session.rs b/crates/common/src/config/smtp/session.rs index c0806269..fb400adb 100644 --- a/crates/common/src/config/smtp/session.rs +++ b/crates/common/src/config/smtp/session.rs @@ -14,7 +14,7 @@ use ahash::AHashSet; use hyper::HeaderMap; use registry::schema::{ enums::{self, ExpressionConstant, MtaStage}, - prelude::Object, + prelude::ObjectType, structs::{ MtaExtensions, MtaHook, MtaInboundSession, MtaMilter, MtaStageAuth, MtaStageConnect, MtaStageData, MtaStageEhlo, MtaStageMail, MtaStageRcpt, @@ -186,128 +186,156 @@ impl SessionConfig { SessionConfig { timeout: bp.compile_expr( - Object::MtaInboundSession.singleton(), + ObjectType::MtaInboundSession.singleton(), &session.ctx_timeout(), ), duration: bp.compile_expr( - Object::MtaInboundSession.singleton(), + ObjectType::MtaInboundSession.singleton(), &session.ctx_max_duration(), ), transfer_limit: bp.compile_expr( - Object::MtaInboundSession.singleton(), + ObjectType::MtaInboundSession.singleton(), &session.ctx_transfer_limit(), ), connect: Connect { - hostname: bp - .compile_expr(Object::MtaStageConnect.singleton(), &connect.ctx_hostname()), - script: bp.compile_expr(Object::MtaStageConnect.singleton(), &connect.ctx_script()), + hostname: bp.compile_expr( + ObjectType::MtaStageConnect.singleton(), + &connect.ctx_hostname(), + ), + script: bp.compile_expr( + ObjectType::MtaStageConnect.singleton(), + &connect.ctx_script(), + ), greeting: bp.compile_expr( - Object::MtaStageConnect.singleton(), + ObjectType::MtaStageConnect.singleton(), &connect.ctx_smtp_greeting(), ), }, ehlo: Ehlo { - script: bp.compile_expr(Object::MtaStageEhlo.singleton(), &ehlo.ctx_script()), - require: bp.compile_expr(Object::MtaStageEhlo.singleton(), &ehlo.ctx_require()), + script: bp.compile_expr(ObjectType::MtaStageEhlo.singleton(), &ehlo.ctx_script()), + require: bp.compile_expr(ObjectType::MtaStageEhlo.singleton(), &ehlo.ctx_require()), reject_non_fqdn: bp.compile_expr( - Object::MtaStageEhlo.singleton(), + ObjectType::MtaStageEhlo.singleton(), &ehlo.ctx_reject_non_fqdn(), ), }, auth: Auth { mechanisms: bp.compile_expr( - Object::MtaStageAuth.singleton(), + ObjectType::MtaStageAuth.singleton(), &auth.ctx_sasl_mechanisms(), ), - require: bp.compile_expr(Object::MtaStageAuth.singleton(), &auth.ctx_require()), + require: bp.compile_expr(ObjectType::MtaStageAuth.singleton(), &auth.ctx_require()), must_match_sender: bp.compile_expr( - Object::MtaStageAuth.singleton(), + ObjectType::MtaStageAuth.singleton(), &auth.ctx_must_match_sender(), ), - errors_max: bp - .compile_expr(Object::MtaStageAuth.singleton(), &auth.ctx_max_failures()), - errors_wait: bp - .compile_expr(Object::MtaStageAuth.singleton(), &auth.ctx_wait_on_fail()), + errors_max: bp.compile_expr( + ObjectType::MtaStageAuth.singleton(), + &auth.ctx_max_failures(), + ), + errors_wait: bp.compile_expr( + ObjectType::MtaStageAuth.singleton(), + &auth.ctx_wait_on_fail(), + ), }, mail: Mail { - script: bp.compile_expr(Object::MtaStageMail.singleton(), &mail.ctx_script()), - rewrite: bp.compile_expr(Object::MtaStageMail.singleton(), &mail.ctx_rewrite()), + script: bp.compile_expr(ObjectType::MtaStageMail.singleton(), &mail.ctx_script()), + rewrite: bp.compile_expr(ObjectType::MtaStageMail.singleton(), &mail.ctx_rewrite()), is_allowed: bp.compile_expr( - Object::MtaStageMail.singleton(), + ObjectType::MtaStageMail.singleton(), &mail.ctx_is_sender_allowed(), ), }, rcpt: Rcpt { - script: bp.compile_expr(Object::MtaStageRcpt.singleton(), &rcpt.ctx_script()), - relay: bp - .compile_expr(Object::MtaStageRcpt.singleton(), &rcpt.ctx_allow_relaying()), - rewrite: bp.compile_expr(Object::MtaStageRcpt.singleton(), &rcpt.ctx_rewrite()), - errors_max: bp - .compile_expr(Object::MtaStageRcpt.singleton(), &rcpt.ctx_max_failures()), - errors_wait: bp - .compile_expr(Object::MtaStageRcpt.singleton(), &rcpt.ctx_wait_on_fail()), - max_recipients: bp - .compile_expr(Object::MtaStageRcpt.singleton(), &rcpt.ctx_max_recipients()), + script: bp.compile_expr(ObjectType::MtaStageRcpt.singleton(), &rcpt.ctx_script()), + relay: bp.compile_expr( + ObjectType::MtaStageRcpt.singleton(), + &rcpt.ctx_allow_relaying(), + ), + rewrite: bp.compile_expr(ObjectType::MtaStageRcpt.singleton(), &rcpt.ctx_rewrite()), + errors_max: bp.compile_expr( + ObjectType::MtaStageRcpt.singleton(), + &rcpt.ctx_max_failures(), + ), + errors_wait: bp.compile_expr( + ObjectType::MtaStageRcpt.singleton(), + &rcpt.ctx_wait_on_fail(), + ), + max_recipients: bp.compile_expr( + ObjectType::MtaStageRcpt.singleton(), + &rcpt.ctx_max_recipients(), + ), }, data: Data { - script: bp.compile_expr(Object::MtaStageData.singleton(), &data.ctx_script()), + script: bp.compile_expr(ObjectType::MtaStageData.singleton(), &data.ctx_script()), spam_filter: bp.compile_expr( - Object::MtaStageData.singleton(), + ObjectType::MtaStageData.singleton(), &data.ctx_enable_spam_filter(), ), - max_messages: bp - .compile_expr(Object::MtaStageData.singleton(), &data.ctx_max_messages()), + max_messages: bp.compile_expr( + ObjectType::MtaStageData.singleton(), + &data.ctx_max_messages(), + ), max_message_size: bp.compile_expr( - Object::MtaStageData.singleton(), + ObjectType::MtaStageData.singleton(), &data.ctx_max_message_size(), ), max_received_headers: bp.compile_expr( - Object::MtaStageData.singleton(), + ObjectType::MtaStageData.singleton(), &data.ctx_max_received_headers(), ), add_received: bp.compile_expr( - Object::MtaStageData.singleton(), + ObjectType::MtaStageData.singleton(), &data.ctx_add_received_header(), ), add_received_spf: bp.compile_expr( - Object::MtaStageData.singleton(), + ObjectType::MtaStageData.singleton(), &data.ctx_add_received_spf_header(), ), add_return_path: bp.compile_expr( - Object::MtaStageData.singleton(), + ObjectType::MtaStageData.singleton(), &data.ctx_add_return_path_header(), ), add_auth_results: bp.compile_expr( - Object::MtaStageData.singleton(), + ObjectType::MtaStageData.singleton(), &data.ctx_add_auth_results_header(), ), add_message_id: bp.compile_expr( - Object::MtaStageData.singleton(), + ObjectType::MtaStageData.singleton(), &data.ctx_add_message_id_header(), ), add_date: bp.compile_expr( - Object::MtaStageData.singleton(), + ObjectType::MtaStageData.singleton(), &data.ctx_add_date_header(), ), add_delivered_to: data.add_delivered_to_header, }, extensions: Extensions { pipelining: bp - .compile_expr(Object::MtaExtensions.singleton(), &ext.ctx_pipelining()), - chunking: bp.compile_expr(Object::MtaExtensions.singleton(), &ext.ctx_chunking()), - requiretls: bp - .compile_expr(Object::MtaExtensions.singleton(), &ext.ctx_require_tls()), - dsn: bp.compile_expr(Object::MtaExtensions.singleton(), &ext.ctx_dsn()), - vrfy: bp.compile_expr(Object::MtaExtensions.singleton(), &ext.ctx_vrfy()), - expn: bp.compile_expr(Object::MtaExtensions.singleton(), &ext.ctx_expn()), - no_soliciting: bp - .compile_expr(Object::MtaExtensions.singleton(), &ext.ctx_no_soliciting()), - future_release: bp - .compile_expr(Object::MtaExtensions.singleton(), &ext.ctx_future_release()), + .compile_expr(ObjectType::MtaExtensions.singleton(), &ext.ctx_pipelining()), + chunking: bp + .compile_expr(ObjectType::MtaExtensions.singleton(), &ext.ctx_chunking()), + requiretls: bp.compile_expr( + ObjectType::MtaExtensions.singleton(), + &ext.ctx_require_tls(), + ), + dsn: bp.compile_expr(ObjectType::MtaExtensions.singleton(), &ext.ctx_dsn()), + vrfy: bp.compile_expr(ObjectType::MtaExtensions.singleton(), &ext.ctx_vrfy()), + expn: bp.compile_expr(ObjectType::MtaExtensions.singleton(), &ext.ctx_expn()), + no_soliciting: bp.compile_expr( + ObjectType::MtaExtensions.singleton(), + &ext.ctx_no_soliciting(), + ), + future_release: bp.compile_expr( + ObjectType::MtaExtensions.singleton(), + &ext.ctx_future_release(), + ), deliver_by: bp - .compile_expr(Object::MtaExtensions.singleton(), &ext.ctx_deliver_by()), - mt_priority: bp - .compile_expr(Object::MtaExtensions.singleton(), &ext.ctx_mt_priority()), + .compile_expr(ObjectType::MtaExtensions.singleton(), &ext.ctx_deliver_by()), + mt_priority: bp.compile_expr( + ObjectType::MtaExtensions.singleton(), + &ext.ctx_mt_priority(), + ), }, mta_sts_policy: Policy::try_parse(bp).await, milters: bp diff --git a/crates/common/src/config/telemetry.rs b/crates/common/src/config/telemetry.rs index 47e00399..90223feb 100644 --- a/crates/common/src/config/telemetry.rs +++ b/crates/common/src/config/telemetry.rs @@ -20,7 +20,7 @@ use opentelemetry_sdk::{ use opentelemetry_semantic_conventions::resource::SERVICE_VERSION; use registry::schema::{ enums::{EventPolicy, LogRotateFrequency}, - prelude::Object, + prelude::{Object, ObjectType}, structs::{self, EventTracingLevel, MetricsPrometheus, Tracer, WebHook}, }; use std::{collections::HashMap, sync::Arc, time::Duration}; @@ -564,7 +564,7 @@ impl Metrics { .collect::>(), Err(err) => { bp.build_error( - Object::Metrics.singleton(), + ObjectType::Metrics.singleton(), format!("Failed to build OpenTelemetry HTTP headers: {err}"), ); Default::default() @@ -589,7 +589,7 @@ impl Metrics { })), Err(err) => { bp.build_error( - Object::Metrics.singleton(), + ObjectType::Metrics.singleton(), format!("Failed to build OpenTelemetry metrics exporter: {err}"), ); None @@ -615,7 +615,7 @@ impl Metrics { })), Err(err) => { bp.build_error( - Object::Metrics.singleton(), + ObjectType::Metrics.singleton(), format!("Failed to build OpenTelemetry metrics exporter: {err}"), ); None diff --git a/crates/common/src/enterprise/config.rs b/crates/common/src/enterprise/config.rs index c1fc5871..e15c88e1 100644 --- a/crates/common/src/enterprise/config.rs +++ b/crates/common/src/enterprise/config.rs @@ -16,16 +16,15 @@ use crate::{enterprise::llm::ApiType, expr::if_block::BootstrapExprExt}; use ahash::AHashMap; use registry::schema::{ enums::AiModelType, - prelude::{Object, Property}, + prelude::{ObjectType, Property}, structs::{self, AiModel, Alert, CalendarAlarm, CalendarScheduling, DataRetention, SpamLlm}, }; use std::sync::Arc; use store::{ - registry::{HashedObject, RegistryQuery, bootstrap::Bootstrap, write::RegistryWrite}, + registry::{RegistryQuery, bootstrap::Bootstrap, write::RegistryWrite}, roaring::RoaringBitmap, }; use trc::MetricType; -use types::id::Id; use utils::template::Template; impl Enterprise { @@ -33,10 +32,8 @@ impl Enterprise { let server_hostname = bp.hostname().to_string(); let mut update_license = None; - let mut enterprise = bp - .setting_infallible::>() - .await; - let license_result = match (&enterprise.object.license_key, &enterprise.object.api_key) { + let mut enterprise = bp.setting_infallible::().await; + let license_result = match (&enterprise.license_key, &enterprise.api_key) { (Some(license_key), maybe_api_key) => { match ( LicenseKey::new(license_key, &server_hostname), @@ -77,21 +74,18 @@ impl Enterprise { let license = match license_result { Ok(license) => license, Err(err) => { - bp.build_warning(Object::Enterprise.singleton(), err.to_string()); + bp.build_warning(ObjectType::Enterprise.singleton(), err.to_string()); return None; } }; // Update the license if a new one was obtained + let logo_url = enterprise.logo_url.clone(); if let Some(license) = update_license { - enterprise.object.license_key = Some(license); + enterprise.license_key = Some(license); if let Err(err) = bp .registry - .write(RegistryWrite::update( - Id::singleton(), - &enterprise.object, - &enterprise, - )) + .write(RegistryWrite::insert(&enterprise.into())) .await { trc::error!( @@ -103,12 +97,12 @@ impl Enterprise { match bp .registry - .query::(RegistryQuery::new(Object::Account)) + .query::(RegistryQuery::new(ObjectType::Account)) .await { Ok(total) if total.len() > license.accounts as u64 => { bp.build_warning( - Object::Enterprise.singleton(), + ObjectType::Enterprise.singleton(), format!( "License key is valid but only allows {} accounts, found {}.", license.accounts, @@ -162,7 +156,7 @@ impl Enterprise { let mut enterprise = Enterprise { license, undelete_retention: dr.hold_deleted_for.map(|retention| retention.into_inner()), - logo_url: enterprise.object.logo_url, + logo_url, metrics_alerts: Default::default(), spam_filter_llm: SpamFilterLlmConfig::parse(bp, &ai_apis_ids).await, ai_apis, @@ -214,19 +208,19 @@ impl Enterprise { ( alarm.template, &mut enterprise.template_calendar_alarm, - Object::CalendarAlarm.singleton(), + ObjectType::CalendarAlarm.singleton(), Property::Template, ), ( sched.email_template, &mut enterprise.template_scheduling_email, - Object::CalendarScheduling.singleton(), + ObjectType::CalendarScheduling.singleton(), Property::EmailTemplate, ), ( sched.http_rsvp_template, &mut enterprise.template_scheduling_web, - Object::CalendarScheduling.singleton(), + ObjectType::CalendarScheduling.singleton(), Property::HttpRsvpTemplate, ), ] { @@ -253,7 +247,7 @@ impl SpamFilterLlmConfig { SpamLlm::Enable(llm) => { let Some(model) = models.get(&llm.model_id.id()).cloned() else { bp.build_error( - Object::SpamLlm.singleton(), + ObjectType::SpamLlm.singleton(), format!("Model {:?} not found in AI API configuration", llm.model_id), ); return None; diff --git a/crates/common/src/expr/eval.rs b/crates/common/src/expr/eval.rs index dab093e3..3d2b41d9 100644 --- a/crates/common/src/expr/eval.rs +++ b/crates/common/src/expr/eval.rs @@ -15,7 +15,7 @@ use compact_str::{CompactString, ToCompactString, format_compact}; use hyper::StatusCode; use registry::{ schema::prelude::Property, - types::{EnumType, id::ObjectId}, + types::{EnumImpl, id::ObjectId}, }; use std::{cmp::Ordering, fmt::Display}; use trc::{Collector, EvalEvent}; diff --git a/crates/common/src/expr/tokenizer.rs b/crates/common/src/expr/tokenizer.rs index f13a5b58..125cb807 100644 --- a/crates/common/src/expr/tokenizer.rs +++ b/crates/common/src/expr/tokenizer.rs @@ -10,7 +10,7 @@ use super::{ }; use ahash::AHashSet; use regex::Regex; -use registry::{schema::enums::ExpressionConstant, types::EnumType}; +use registry::{schema::enums::ExpressionConstant, types::EnumImpl}; use std::{borrow::Cow, iter::Peekable, slice::Iter}; use trc::MetricType; diff --git a/crates/common/src/ipc.rs b/crates/common/src/ipc.rs index 5d6ba4fe..20b42a98 100644 --- a/crates/common/src/ipc.rs +++ b/crates/common/src/ipc.rs @@ -15,7 +15,7 @@ use mail_auth::{ mta_sts::TlsRpt, report::{Record, tlsrpt::FailureDetails}, }; -use registry::{schema::prelude::Object, types::id::ObjectId}; +use registry::{schema::prelude::ObjectType, types::id::ObjectId}; use std::{ sync::{ Arc, @@ -106,7 +106,7 @@ pub enum BroadcastEvent { pub enum RegistryChange { Insert(ObjectId), Delete(ObjectId), - Reload(Object), + Reload(ObjectType), } #[derive(Debug, Clone, Copy)] @@ -209,7 +209,7 @@ impl TrainTaskController { } impl BroadcastEvent { - pub fn reload(object: Object) -> Self { + pub fn reload(object: ObjectType) -> Self { BroadcastEvent::RegistryChange(RegistryChange::Reload(object)) } } diff --git a/crates/common/src/network/security.rs b/crates/common/src/network/security.rs index 3a12a826..e48159f6 100644 --- a/crates/common/src/network/security.rs +++ b/crates/common/src/network/security.rs @@ -13,7 +13,7 @@ use ahash::AHashSet; use registry::{ schema::{ enums::BlockReason, - prelude::Object, + prelude::{HashedObject, Object, ObjectType}, structs::{self, AllowedIp, BlockedIp, Rate}, }, types::{datetime::UTCDateTime, ipmask::IpAddrOrMask}, @@ -65,26 +65,34 @@ impl Security { let mut expired_allows = Vec::new(); let now = now() as i64; - for ip in bp.list_infallible::().await { + for ip in bp.list_infallible::>().await { let id = ip.id; - let ip = ip.object; + let revision = ip.object.revision; + let ip = ip.object.object; - if ip.expires_at.is_none_or(|ip| ip.timestamp() > now) { + if ip.expires_at.as_ref().is_none_or(|ip| ip.timestamp() > now) { if let Some(ip) = ip.address.try_to_ip() { allowed_ip_addresses.insert(ip); } else { allowed_ip_networks.push(ip.address); } } else { - expired_allows.push((id, ip.address)); + expired_allows.push(( + id, + ip.address.clone(), + Object { + inner: ip.into(), + revision, + }, + )); } } if !expired_allows.is_empty() { - for (id, _) in &expired_allows { + for (id, _, object) in &expired_allows { if let Err(err) = bp .registry - .write::(RegistryWrite::delete(id.id())) + .write(RegistryWrite::delete_object(*id, object)) .await { trc::error!( @@ -98,7 +106,7 @@ impl Security { Security(trc::SecurityEvent::IpAllowExpired), Details = expired_allows .into_iter() - .map(|(_, ip)| trc::Value::from(ip.into_inner().0)) + .map(|(_, ip, _)| trc::Value::from(ip.into_inner().0)) .collect::>() ); } @@ -257,17 +265,20 @@ impl Server { let now = now() as i64; let RegistryWriteResult::Success(id) = self .registry() - .write(RegistryWrite::insert(&BlockedIp { - address: IpAddrOrMask::from_ip(ip), - created_at: UTCDateTime::from_timestamp(now), - expires_at: self - .core - .network - .security - .blocked_ip_expiration - .map(|v| UTCDateTime::from_timestamp(now + v as i64)), - reason, - })) + .write(RegistryWrite::insert( + &BlockedIp { + address: IpAddrOrMask::from_ip(ip), + created_at: UTCDateTime::from_timestamp(now), + expires_at: self + .core + .network + .security + .blocked_ip_expiration + .map(|v| UTCDateTime::from_timestamp(now + v as i64)), + reason, + } + .into(), + )) .await .caused_by(trc::location!())? else { @@ -276,7 +287,7 @@ impl Server { // Increment version self.cluster_broadcast(BroadcastEvent::RegistryChange(RegistryChange::Insert( - Object::BlockedIp.id(id), + ObjectType::BlockedIp.id(id), ))) .await; @@ -317,26 +328,34 @@ impl BlockedIps { let mut expired_blocks = Vec::new(); let now = now() as i64; - for ip in bp.list_infallible::().await { + for ip in bp.list_infallible::>().await { let id = ip.id; - let ip = ip.object; + let revision = ip.object.revision; + let ip = ip.object.object; - if ip.expires_at.is_none_or(|ip| ip.timestamp() > now) { + if ip.expires_at.as_ref().is_none_or(|ip| ip.timestamp() > now) { if let Some(ip) = ip.address.try_to_ip() { ips.blocked_ip_addresses.insert(ip); } else { ips.blocked_ip_networks.push(ip.address); } } else { - expired_blocks.push((id, ip.address)); + expired_blocks.push(( + id, + ip.address.clone(), + Object { + inner: ip.into(), + revision, + }, + )); } } if !expired_blocks.is_empty() { - for (id, _) in &expired_blocks { + for (id, _, object) in &expired_blocks { if let Err(err) = bp .registry - .write::(RegistryWrite::delete(id.id())) + .write(RegistryWrite::delete_object(*id, object)) .await { trc::error!( @@ -349,7 +368,7 @@ impl BlockedIps { Security(trc::SecurityEvent::IpBlockExpired), Details = expired_blocks .into_iter() - .map(|(_, ip)| trc::Value::from(ip.into_inner().0)) + .map(|(_, ip, _)| trc::Value::from(ip.into_inner().0)) .collect::>() ); } diff --git a/crates/common/src/scripts/plugins/llm_prompt.rs b/crates/common/src/scripts/plugins/llm_prompt.rs index 6a6bb776..4685e0ad 100644 --- a/crates/common/src/scripts/plugins/llm_prompt.rs +++ b/crates/common/src/scripts/plugins/llm_prompt.rs @@ -39,7 +39,7 @@ pub async fn exec(ctx: PluginContext<'_>) -> trc::Result { if token.has_permission(Permission::AiModelInteract) { true } else { - use registry::types::EnumType; + use registry::types::EnumImpl; trc::event!( Security(SecurityEvent::Unauthorized), diff --git a/crates/common/src/storage/mod.rs b/crates/common/src/storage/mod.rs index fb3b98e3..5226bece 100644 --- a/crates/common/src/storage/mod.rs +++ b/crates/common/src/storage/mod.rs @@ -9,9 +9,9 @@ use directory::Directory; use registry::{ schema::{ enums::{StorageQuota, TenantStorageQuota}, - prelude::Object, + prelude::ObjectType, }, - types::EnumType, + types::EnumImpl, }; use std::sync::Arc; use store::{ @@ -91,14 +91,14 @@ impl Server { pub async fn total_accounts(&self) -> trc::Result { self.registry() - .query::(RegistryQuery::new(Object::Account)) + .query::(RegistryQuery::new(ObjectType::Account)) .await .map(|r| r.len()) } pub async fn total_domains(&self) -> trc::Result { self.registry() - .query::(RegistryQuery::new(Object::Domain)) + .query::(RegistryQuery::new(ObjectType::Domain)) .await .map(|r| r.len()) } diff --git a/crates/common/src/storage/quota.rs b/crates/common/src/storage/quota.rs index b3fcfbc8..f3a732b8 100644 --- a/crates/common/src/storage/quota.rs +++ b/crates/common/src/storage/quota.rs @@ -10,7 +10,7 @@ use crate::{ }; use registry::{ schema::enums::{StorageQuota, TenantStorageQuota}, - types::EnumType, + types::EnumImpl, }; use store::{ValueKey, write::ValueClass}; use trc::AddContext; diff --git a/crates/coordinator/src/bootstrap.rs b/crates/coordinator/src/bootstrap.rs index 62d6c999..4beb6e3e 100644 --- a/crates/coordinator/src/bootstrap.rs +++ b/crates/coordinator/src/bootstrap.rs @@ -5,7 +5,7 @@ */ use crate::Coordinator; -use registry::schema::{prelude::Object, structs}; +use registry::schema::{prelude::ObjectType, structs}; use store::{InMemoryStore, registry::bootstrap::Bootstrap}; #[allow(unreachable_patterns)] @@ -54,7 +54,7 @@ impl Coordinator { match result { Ok(store) => Some(store), Err(err) => { - bp.build_error(Object::Coordinator.singleton(), err); + bp.build_error(ObjectType::Coordinator.singleton(), err); None } } diff --git a/crates/dav/src/common/propfind.rs b/crates/dav/src/common/propfind.rs index d3c61cc8..e112de69 100644 --- a/crates/dav/src/common/propfind.rs +++ b/crates/dav/src/common/propfind.rs @@ -56,7 +56,7 @@ use groupware::{ }; use http_proto::HttpResponse; use hyper::StatusCode; -use registry::schema::{enums::Permission, prelude::Object}; +use registry::schema::{enums::Permission, prelude::ObjectType}; use std::sync::Arc; use store::{ ValueKey, @@ -260,7 +260,7 @@ impl PropFindRequestHandler for Server { // Return all principals self.registry() .query::( - RegistryQuery::new(Object::Account) + RegistryQuery::new(ObjectType::Account) .with_tenant(access_token.tenant_id()), ) .await diff --git a/crates/dav/src/principal/propsearch.rs b/crates/dav/src/principal/propsearch.rs index 8a1bab39..5cf64c21 100644 --- a/crates/dav/src/principal/propsearch.rs +++ b/crates/dav/src/principal/propsearch.rs @@ -13,7 +13,7 @@ use dav_proto::schema::{ }; use http_proto::HttpResponse; use hyper::StatusCode; -use registry::schema::prelude::Object; +use registry::schema::prelude::ObjectType; use store::{registry::RegistryQuery, roaring::RoaringBitmap}; use trc::AddContext; use types::collection::Collection; @@ -49,7 +49,7 @@ impl PrincipalPropSearch for Server { let ids = self .registry() .query::( - RegistryQuery::new(Object::Account) + RegistryQuery::new(ObjectType::Account) .with_tenant(access_token.tenant_id()) .text(search_for), ) diff --git a/crates/directory/src/core/config.rs b/crates/directory/src/core/config.rs index 1d763b6b..23db4ea8 100644 --- a/crates/directory/src/core/config.rs +++ b/crates/directory/src/core/config.rs @@ -9,7 +9,7 @@ use crate::{ backend::{ldap::LdapDirectory, oidc::OpenIdDirectory, sql::SqlDirectory}, }; use registry::schema::{ - prelude::Object, + prelude::ObjectType, structs::{self, Authentication}, }; use std::{collections::HashMap, sync::Arc}; @@ -45,7 +45,7 @@ impl Directories { Some(default_directory) => default_directory.clone().into(), None => { bp.build_error( - Object::Authentication.singleton(), + ObjectType::Authentication.singleton(), format!("Default directory with ID {} not found", directory_id), ); None diff --git a/crates/email/src/message/delete.rs b/crates/email/src/message/delete.rs index f0589412..4d061990 100644 --- a/crates/email/src/message/delete.rs +++ b/crates/email/src/message/delete.rs @@ -7,7 +7,7 @@ use super::metadata::MessageData; use common::{KV_LOCK_PURGE_ACCOUNT, Server, storage::index::ObjectIndexBuilder}; use groupware::calendar::storage::ItipAutoExpunge; -use registry::schema::prelude::Object; +use registry::schema::prelude::ObjectType; use std::future::Future; use store::ahash::AHashSet; use store::registry::RegistryQuery; @@ -113,7 +113,7 @@ impl EmailDeletion for Server { async fn purge_accounts(&self, use_roles: bool) { match self .registry() - .query::>(RegistryQuery::new(Object::Account)) + .query::>(RegistryQuery::new(ObjectType::Account)) .await { Ok(account_ids) => { diff --git a/crates/http/src/auth/oauth/registration.rs b/crates/http/src/auth/oauth/registration.rs index de7bc385..5debdbe2 100644 --- a/crates/http/src/auth/oauth/registration.rs +++ b/crates/http/src/auth/oauth/registration.rs @@ -4,8 +4,6 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ -use std::future::Future; - use super::ErrorType; use crate::auth::authenticate::Authenticator; use common::{ @@ -19,11 +17,12 @@ use http_proto::{request::fetch_body, *}; use registry::{ schema::{ enums::Permission, - prelude::{Object, Property}, + prelude::{ObjectType, Property}, structs::OAuthClient, }, types::datetime::UTCDateTime, }; +use std::future::Future; use store::{ ahash::AHashSet, rand::{Rng, distr::Alphanumeric, rng}, @@ -82,16 +81,19 @@ impl ClientRegistrationHandler for Server { .collect::(); self.registry() - .write(RegistryWrite::insert(&OAuthClient { - client_id: client_id.clone(), - created_at: UTCDateTime::now(), - description: request.client_name.clone(), - contacts: request.contacts.clone(), - member_tenant_id: tenant_id.map(|id| Id::new(id as u64)), - redirect_uris: request.redirect_uris.clone(), - logo: request.logo_uri.clone(), - ..Default::default() - })) + .write(RegistryWrite::insert( + &OAuthClient { + client_id: client_id.clone(), + created_at: UTCDateTime::now(), + description: request.client_name.clone(), + contacts: request.contacts.clone(), + member_tenant_id: tenant_id.map(|id| Id::new(id as u64)), + redirect_uris: request.redirect_uris.clone(), + logo: request.logo_uri.clone(), + ..Default::default() + } + .into(), + )) .await .caused_by(trc::location!())?; @@ -124,7 +126,7 @@ impl ClientRegistrationHandler for Server { let found_registration = if let Some(client_id) = self .registry() .query::>( - RegistryQuery::new(Object::OAuthClient).equal(Property::ClientId, client_id), + RegistryQuery::new(ObjectType::OAuthClient).equal(Property::ClientId, client_id), ) .await? .iter() diff --git a/crates/jmap-proto/src/object/registry.rs b/crates/jmap-proto/src/object/registry.rs index 29124fd6..5e9c1964 100644 --- a/crates/jmap-proto/src/object/registry.rs +++ b/crates/jmap-proto/src/object/registry.rs @@ -10,7 +10,7 @@ use crate::{ }; use registry::{ jmap::RegistryValue, - schema::prelude::{Object, Property}, + schema::prelude::{ObjectType, Property}, }; use std::borrow::Cow; use types::id::Id; @@ -20,7 +20,7 @@ pub struct Registry; #[derive(Debug, Clone, PartialEq, Eq)] pub enum RegistryFilter { - Type(Object), + Type(ObjectType), Text(String), Id(Vec), Property(Property), diff --git a/crates/jmap-proto/src/request/method.rs b/crates/jmap-proto/src/request/method.rs index fafe1f3a..ee87f0f7 100644 --- a/crates/jmap-proto/src/request/method.rs +++ b/crates/jmap-proto/src/request/method.rs @@ -5,7 +5,7 @@ */ use std::{borrow::Cow, fmt::Display}; -use registry::{schema::prelude::Object, types::EnumType}; +use registry::{schema::prelude::ObjectType, types::EnumImpl}; #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub struct MethodName { @@ -36,7 +36,7 @@ pub enum MethodObject { FileNode, ParticipantIdentity, ShareNotification, - Registry(Object), + Registry(ObjectType), } #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -300,7 +300,7 @@ impl MethodName { ).or_else(|| { let (obj, fnc) = s.strip_prefix("x:")?.split_once('/')?; - let obj = Object::parse(obj)?; + let obj = ObjectType::parse(obj)?; let fnc = hashify::tiny_map!(fnc.as_bytes(), "get" => MethodFunction::Get, "set" => MethodFunction::Set, @@ -364,6 +364,15 @@ impl MethodFunction { } } +impl MethodObject { + pub fn unwrap_registry(self) -> ObjectType { + match self { + MethodObject::Registry(obj) => obj, + _ => panic!("Not a registry method object"), + } + } +} + impl<'de> serde::Deserialize<'de> for MethodName { fn deserialize(deserializer: D) -> Result where diff --git a/crates/jmap/src/api/acl.rs b/crates/jmap/src/api/acl.rs index cbaa63e5..5ce90248 100644 --- a/crates/jmap/src/api/acl.rs +++ b/crates/jmap/src/api/acl.rs @@ -10,7 +10,7 @@ use jmap_proto::{ object::{JmapRight, JmapSharedObject}, }; use jmap_tools::{JsonPointerIter, Key, Map, Property, Value}; -use registry::schema::prelude::Object; +use registry::schema::prelude::ObjectType; use store::{registry::RegistryQuery, roaring::RoaringBitmap}; use types::{ acl::{Acl, AclGrant}, @@ -242,7 +242,7 @@ impl JmapAcl for Server { let principal_ids = self .registry() - .query::(RegistryQuery::new(Object::Account)) + .query::(RegistryQuery::new(ObjectType::Account)) .await .unwrap_or_default(); diff --git a/crates/jmap/src/api/request.rs b/crates/jmap/src/api/request.rs index a6894e09..60494353 100644 --- a/crates/jmap/src/api/request.rs +++ b/crates/jmap/src/api/request.rs @@ -351,7 +351,9 @@ impl RequestHandler for Server { set_account_id_if_missing(&mut req.account_id, access_token); access_token.assert_is_member(req.account_id)?; - self.registry_get(req, access_token).await?.into() + self.registry_get(method_name.obj.unwrap_registry(), req, access_token) + .await? + .into() } }, RequestMethod::Query(req) => match req { @@ -424,7 +426,9 @@ impl RequestHandler for Server { set_account_id_if_missing(&mut req.account_id, access_token); access_token.assert_is_member(req.account_id)?; - self.registry_query(req, access_token).await?.into() + self.registry_query(method_name.obj.unwrap_registry(), req, access_token) + .await? + .into() } }, RequestMethod::Set(req) => match req { @@ -532,7 +536,9 @@ impl RequestHandler for Server { set_account_id_if_missing(&mut req.account_id, access_token); access_token.assert_is_member(req.account_id)?; - self.registry_set(req, access_token).await?.into() + self.registry_set(method_name.obj.unwrap_registry(), req, access_token) + .await? + .into() } }, RequestMethod::Changes(mut req) => { diff --git a/crates/jmap/src/principal/get.rs b/crates/jmap/src/principal/get.rs index 99c6d099..52c7c3b3 100644 --- a/crates/jmap/src/principal/get.rs +++ b/crates/jmap/src/principal/get.rs @@ -12,7 +12,7 @@ use jmap_proto::{ types::state::State, }; use jmap_tools::{Key, Map, Value}; -use registry::schema::prelude::{Object, Permission}; +use registry::schema::prelude::{ObjectType, Permission}; use std::future::Future; use store::{registry::RegistryQuery, roaring::RoaringBitmap}; use trc::AddContext; @@ -52,7 +52,7 @@ impl PrincipalGet for Server { let principal_ids = self .registry() .query::( - RegistryQuery::new(Object::Account).with_tenant(access_token.tenant_id()), + RegistryQuery::new(ObjectType::Account).with_tenant(access_token.tenant_id()), ) .await .caused_by(trc::location!())?; diff --git a/crates/jmap/src/principal/query.rs b/crates/jmap/src/principal/query.rs index 926aff9f..744949f5 100644 --- a/crates/jmap/src/principal/query.rs +++ b/crates/jmap/src/principal/query.rs @@ -14,9 +14,9 @@ use jmap_proto::{ use registry::{ schema::{ enums::AccountType, - prelude::{Object, Permission, Property}, + prelude::{ObjectType, Permission, Property}, }, - types::EnumType, + types::EnumImpl, }; use std::future::Future; use store::{ @@ -52,7 +52,7 @@ impl PrincipalQuery for Server { let principal_ids = self .registry() .query::( - RegistryQuery::new(Object::Account).with_tenant(access_token.tenant_id()), + RegistryQuery::new(ObjectType::Account).with_tenant(access_token.tenant_id()), ) .await .caused_by(trc::location!())?; @@ -86,7 +86,7 @@ impl PrincipalQuery for Server { filters.push(SearchFilter::is_in_set( self.registry() .query::( - RegistryQuery::new(Object::Account) + RegistryQuery::new(ObjectType::Account) .with_tenant(access_token.tenant_id()) .text(text), ) @@ -107,7 +107,7 @@ impl PrincipalQuery for Server { filters.push(SearchFilter::is_in_set( self.registry() .query::( - RegistryQuery::new(Object::Account) + RegistryQuery::new(ObjectType::Account) .equal(Property::Type, typ.to_id()) .with_tenant(access_token.tenant_id()), ) diff --git a/crates/jmap/src/registry/get.rs b/crates/jmap/src/registry/get.rs index 18f3207f..44aed305 100644 --- a/crates/jmap/src/registry/get.rs +++ b/crates/jmap/src/registry/get.rs @@ -9,10 +9,12 @@ use jmap_proto::{ method::get::{GetRequest, GetResponse}, object::registry::Registry, }; +use registry::schema::prelude::ObjectType; pub trait RegistryGet: Sync + Send { fn registry_get( &self, + object_type: ObjectType, request: GetRequest, access_token: &AccessToken, ) -> impl Future>> + Send; @@ -21,6 +23,7 @@ pub trait RegistryGet: Sync + Send { impl RegistryGet for Server { async fn registry_get( &self, + object_type: ObjectType, mut request: GetRequest, access_token: &AccessToken, ) -> trc::Result> { diff --git a/crates/jmap/src/registry/query.rs b/crates/jmap/src/registry/query.rs index b7582a45..4d120019 100644 --- a/crates/jmap/src/registry/query.rs +++ b/crates/jmap/src/registry/query.rs @@ -9,10 +9,12 @@ use jmap_proto::{ method::query::{QueryRequest, QueryResponse}, object::registry::Registry, }; +use registry::schema::prelude::ObjectType; pub trait RegistryQuery: Sync + Send { fn registry_query( &self, + object_type: ObjectType, request: QueryRequest, access_token: &AccessToken, ) -> impl Future> + Send; @@ -21,6 +23,7 @@ pub trait RegistryQuery: Sync + Send { impl RegistryQuery for Server { async fn registry_query( &self, + object_type: ObjectType, mut request: QueryRequest, access_token: &AccessToken, ) -> trc::Result { diff --git a/crates/jmap/src/registry/set.rs b/crates/jmap/src/registry/set.rs index a8f0fa3b..1a787d98 100644 --- a/crates/jmap/src/registry/set.rs +++ b/crates/jmap/src/registry/set.rs @@ -9,10 +9,12 @@ use jmap_proto::{ method::set::{SetRequest, SetResponse}, object::registry::Registry, }; +use registry::schema::prelude::ObjectType; pub trait RegistrySet: Sync + Send { fn registry_set( &self, + object_type: ObjectType, request: SetRequest<'_, Registry>, access_token: &AccessToken, ) -> impl Future>> + Send; @@ -21,6 +23,7 @@ pub trait RegistrySet: Sync + Send { impl RegistrySet for Server { async fn registry_set( &self, + object_type: ObjectType, mut request: SetRequest<'_, Registry>, access_token: &AccessToken, ) -> trc::Result> { diff --git a/crates/registry/Cargo.toml b/crates/registry/Cargo.toml index 379e4ef4..c0ade08c 100644 --- a/crates/registry/Cargo.toml +++ b/crates/registry/Cargo.toml @@ -12,6 +12,7 @@ serde_json = "1.0" hashify = "0.2.7" ahash = { version = "0.8" } jmap-tools = { version = "0.1" } +xxhash-rust = { version = "0.8.5", features = ["xxh3"] } [features] test_mode = [] diff --git a/crates/registry/src/jmap.rs b/crates/registry/src/jmap.rs index ca407dd5..397052c1 100644 --- a/crates/registry/src/jmap.rs +++ b/crates/registry/src/jmap.rs @@ -7,7 +7,7 @@ use crate::{ schema::prelude::Property, types::{ - EnumType, + EnumImpl, error::PatchError, string::{StringValidator, StringValidatorResult}, }, @@ -322,7 +322,7 @@ impl RegistryJsonPatch for f64 { } } -impl RegistryJsonEnumPatch for T { +impl RegistryJsonEnumPatch for T { fn patch( &mut self, pointer: JsonPointerPatch<'_>, @@ -534,7 +534,7 @@ impl MapItem for u32 { } } -impl MapItem for T { +impl MapItem for T { fn try_from_string(value: &str) -> Option { Self::parse(value) } @@ -544,7 +544,7 @@ impl MapItem for T { } } -pub fn object_type( +pub fn object_type( pointer: &JsonPointerPatch<'_>, value: &Value<'_, Property, RegistryValue>, ) -> Result { diff --git a/crates/registry/src/pickle.rs b/crates/registry/src/pickle.rs index 5386cb54..45935b03 100644 --- a/crates/registry/src/pickle.rs +++ b/crates/registry/src/pickle.rs @@ -6,7 +6,7 @@ use utils::map::vec_map::VecMap; -use crate::types::EnumType; +use crate::types::EnumImpl; use std::collections::HashMap; pub trait Pickle: Sized { @@ -47,61 +47,61 @@ impl<'x> PickledStream<'x> { impl Pickle for u16 { fn pickle(&self, out: &mut Vec) { - out.extend_from_slice(&self.to_le_bytes()); + out.extend_from_slice(&self.to_be_bytes()); } fn unpickle(stream: &mut PickledStream<'_>) -> Option { let mut arr = [0u8; std::mem::size_of::()]; arr.copy_from_slice(stream.read_bytes(std::mem::size_of::())?); - Some(u16::from_le_bytes(arr)) + Some(u16::from_be_bytes(arr)) } } impl Pickle for u64 { fn pickle(&self, out: &mut Vec) { - out.extend_from_slice(&self.to_le_bytes()); + out.extend_from_slice(&self.to_be_bytes()); } fn unpickle(stream: &mut PickledStream<'_>) -> Option { let mut arr = [0u8; std::mem::size_of::()]; arr.copy_from_slice(stream.read_bytes(std::mem::size_of::())?); - Some(u64::from_le_bytes(arr)) + Some(u64::from_be_bytes(arr)) } } impl Pickle for u32 { fn pickle(&self, out: &mut Vec) { - out.extend_from_slice(&self.to_le_bytes()); + out.extend_from_slice(&self.to_be_bytes()); } fn unpickle(stream: &mut PickledStream<'_>) -> Option { let mut arr = [0u8; std::mem::size_of::()]; arr.copy_from_slice(stream.read_bytes(std::mem::size_of::())?); - Some(u32::from_le_bytes(arr)) + Some(u32::from_be_bytes(arr)) } } impl Pickle for i64 { fn pickle(&self, out: &mut Vec) { - out.extend_from_slice(&self.to_le_bytes()); + out.extend_from_slice(&self.to_be_bytes()); } fn unpickle(stream: &mut PickledStream<'_>) -> Option { let mut arr = [0u8; std::mem::size_of::()]; arr.copy_from_slice(stream.read_bytes(std::mem::size_of::())?); - Some(i64::from_le_bytes(arr)) + Some(i64::from_be_bytes(arr)) } } impl Pickle for f64 { fn pickle(&self, out: &mut Vec) { - out.extend_from_slice(&self.to_le_bytes()); + out.extend_from_slice(&self.to_be_bytes()); } fn unpickle(stream: &mut PickledStream<'_>) -> Option { let mut arr = [0u8; std::mem::size_of::()]; arr.copy_from_slice(stream.read_bytes(std::mem::size_of::())?); - Some(f64::from_le_bytes(arr)) + Some(f64::from_be_bytes(arr)) } } @@ -121,27 +121,27 @@ impl Pickle for bool { impl Pickle for String { fn pickle(&self, out: &mut Vec) { - out.extend_from_slice(&(self.len() as u32).to_le_bytes()); + out.extend_from_slice(&(self.len() as u32).to_be_bytes()); out.extend_from_slice(self.as_bytes()); } fn unpickle(stream: &mut PickledStream<'_>) -> Option { let mut len_arr = [0u8; std::mem::size_of::()]; len_arr.copy_from_slice(stream.read_bytes(std::mem::size_of::())?); - let bytes = stream.read_bytes(u32::from_le_bytes(len_arr) as usize)?; + let bytes = stream.read_bytes(u32::from_be_bytes(len_arr) as usize)?; String::from_utf8(bytes.to_vec()).ok() } } -impl Pickle for T { +impl Pickle for T { fn pickle(&self, out: &mut Vec) { - out.extend_from_slice(&self.to_id().to_le_bytes()); + out.extend_from_slice(&self.to_id().to_be_bytes()); } fn unpickle(stream: &mut PickledStream<'_>) -> Option { let mut id_arr = [0u8; std::mem::size_of::()]; id_arr.copy_from_slice(stream.read_bytes(std::mem::size_of::())?); - Self::from_id(u16::from_le_bytes(id_arr)) + Self::from_id(u16::from_be_bytes(id_arr)) } } @@ -175,7 +175,7 @@ where T: Pickle, { fn pickle(&self, out: &mut Vec) { - out.extend_from_slice(&(self.len() as u32).to_le_bytes()); + out.extend_from_slice(&(self.len() as u32).to_be_bytes()); for item in self { item.pickle(out); } @@ -184,7 +184,7 @@ where fn unpickle(stream: &mut PickledStream<'_>) -> Option { let mut len_arr = [0u8; 4]; len_arr.copy_from_slice(stream.read_bytes(4)?); - let len = u32::from_le_bytes(len_arr) as usize; + let len = u32::from_be_bytes(len_arr) as usize; let mut vec = Vec::with_capacity(len); for _ in 0..len { vec.push(T::unpickle(stream)?); @@ -200,7 +200,7 @@ where S: std::hash::BuildHasher + Default, { fn pickle(&self, out: &mut Vec) { - out.extend_from_slice(&(self.len() as u32).to_le_bytes()); + out.extend_from_slice(&(self.len() as u32).to_be_bytes()); for (key, value) in self { key.pickle(out); value.pickle(out); @@ -210,7 +210,7 @@ where fn unpickle(stream: &mut PickledStream<'_>) -> Option { let mut len_arr = [0u8; 4]; len_arr.copy_from_slice(stream.read_bytes(4)?); - let len = u32::from_le_bytes(len_arr) as usize; + let len = u32::from_be_bytes(len_arr) as usize; let mut map = HashMap::with_capacity_and_hasher(len, S::default()); for _ in 0..len { let key = K::unpickle(stream)?; @@ -227,7 +227,7 @@ where V: Pickle, { fn pickle(&self, out: &mut Vec) { - out.extend_from_slice(&(self.len() as u32).to_le_bytes()); + out.extend_from_slice(&(self.len() as u32).to_be_bytes()); for (key, value) in self { key.pickle(out); value.pickle(out); @@ -237,7 +237,7 @@ where fn unpickle(stream: &mut PickledStream<'_>) -> Option { let mut len_arr = [0u8; 4]; len_arr.copy_from_slice(stream.read_bytes(4)?); - let len = u32::from_le_bytes(len_arr) as usize; + let len = u32::from_be_bytes(len_arr) as usize; let mut map = VecMap::with_capacity(len); for _ in 0..len { let key = K::unpickle(stream)?; diff --git a/crates/registry/src/schema/mod.rs b/crates/registry/src/schema/mod.rs index 550bf086..e1e08dbf 100644 --- a/crates/registry/src/schema/mod.rs +++ b/crates/registry/src/schema/mod.rs @@ -5,11 +5,15 @@ */ use crate::{ + pickle::Pickle, schema::{ enums::{TracingLevel, TracingLevelOpt}, - prelude::{Account, Duration, GroupAccount, HttpAuth, NodeRange, Property, UserAccount}, + prelude::{ + Account, Duration, GroupAccount, HashedObject, HttpAuth, NodeRange, Object, + ObjectInner, ObjectType, Property, UserAccount, + }, }, - types::EnumType, + types::{EnumImpl, ObjectImpl}, }; use std::{cmp::Ordering, fmt::Display}; use trc::TOTAL_EVENT_COUNT; @@ -174,7 +178,7 @@ impl From for trc::Level { } } -impl EnumType for trc::EventType { +impl EnumImpl for trc::EventType { const COUNT: usize = TOTAL_EVENT_COUNT; fn parse(s: &str) -> Option { @@ -194,7 +198,7 @@ impl EnumType for trc::EventType { } } -impl EnumType for trc::MetricType { +impl EnumImpl for trc::MetricType { const COUNT: usize = TOTAL_EVENT_COUNT; fn parse(s: &str) -> Option { @@ -225,3 +229,62 @@ impl Ord for Property { self.to_id().cmp(&other.to_id()) } } + +impl> From for Object { + fn from(value: T) -> Self { + Object { + inner: value.into(), + revision: 0, + } + } +} + +impl> From for HashedObject { + fn from(value: Object) -> Self { + HashedObject { + revision: value.revision, + object: T::from(value), + } + } +} + +impl ObjectImpl for HashedObject { + const FLAGS: u64 = T::FLAGS; + const OBJECT: ObjectType = T::OBJECT; + + fn validate(&self, errors: &mut Vec) -> bool { + self.object.validate(errors) + } + + fn index<'x>(&'x self, builder: &mut prelude::IndexBuilder<'x>) { + self.object.index(builder) + } +} + +impl Pickle for HashedObject { + fn pickle(&self, out: &mut Vec) { + T::OBJECT.pickle(out); + self.object.pickle(out); + (xxhash_rust::xxh3::xxh3_64(out) as u32).pickle(out); + } + + fn unpickle(stream: &mut crate::pickle::PickledStream<'_>) -> Option { + let _ = u16::unpickle(stream)?; + Some(Self { + object: T::unpickle(stream)?, + revision: u32::unpickle(stream)?, + }) + } +} + +impl<'de, T: ObjectImpl> serde::Deserialize<'de> for HashedObject { + fn deserialize(deserializer: D) -> Result + where + D: serde::Deserializer<'de>, + { + T::deserialize(deserializer).map(|object| Self { + object, + revision: 0, + }) + } +} diff --git a/crates/registry/src/schema/prelude.rs b/crates/registry/src/schema/prelude.rs index 64c3a8b3..4bcf032f 100644 --- a/crates/registry/src/schema/prelude.rs +++ b/crates/registry/src/schema/prelude.rs @@ -11,8 +11,8 @@ pub use crate::pickle::Pickle; pub use crate::schema::enums::*; pub use crate::schema::properties::*; pub use crate::schema::structs::*; -pub use crate::types::EnumType; -pub use crate::types::ObjectType; +pub use crate::types::EnumImpl; +pub use crate::types::ObjectImpl; pub use crate::types::datetime::UTCDateTime; pub use crate::types::duration::Duration; pub use crate::types::error::*; @@ -26,6 +26,18 @@ pub use std::str::FromStr; pub use types::id::Id; pub use utils::map::vec_map::VecMap; +#[derive(Debug, Clone)] +pub struct Object { + pub inner: ObjectInner, + pub revision: u32, +} + +#[derive(Debug, Clone, Default, Serialize)] +pub struct HashedObject { + pub object: T, + pub revision: u32, +} + #[derive(Debug)] pub struct ExpressionContext<'x> { pub expr: &'x Expression, diff --git a/crates/registry/src/types/datetime.rs b/crates/registry/src/types/datetime.rs index a5a114b0..e8a676b8 100644 --- a/crates/registry/src/types/datetime.rs +++ b/crates/registry/src/types/datetime.rs @@ -250,13 +250,13 @@ impl<'de> serde::Deserialize<'de> for UTCDateTime { impl Pickle for UTCDateTime { fn pickle(&self, out: &mut Vec) { - out.extend_from_slice(&self.0.to_le_bytes()); + out.extend_from_slice(&self.0.to_be_bytes()); } fn unpickle(data: &mut PickledStream<'_>) -> Option { let mut arr = [0u8; 8]; arr.copy_from_slice(data.read_bytes(8)?); - Some(UTCDateTime(i64::from_le_bytes(arr))) + Some(UTCDateTime(i64::from_be_bytes(arr))) } } diff --git a/crates/registry/src/types/duration.rs b/crates/registry/src/types/duration.rs index 1a9b9baa..99e54736 100644 --- a/crates/registry/src/types/duration.rs +++ b/crates/registry/src/types/duration.rs @@ -125,14 +125,14 @@ impl FromStr for Duration { impl Pickle for Duration { fn pickle(&self, out: &mut Vec) { - out.extend_from_slice(&(self.0.as_millis() as u64).to_le_bytes()); + out.extend_from_slice(&(self.0.as_millis() as u64).to_be_bytes()); } fn unpickle(data: &mut PickledStream<'_>) -> Option { let mut arr = [0u8; 8]; arr.copy_from_slice(data.read_bytes(8)?); Some(Duration(std::time::Duration::from_millis( - u64::from_le_bytes(arr), + u64::from_be_bytes(arr), ))) } } diff --git a/crates/registry/src/types/id.rs b/crates/registry/src/types/id.rs index ef73c05a..99b2b6af 100644 --- a/crates/registry/src/types/id.rs +++ b/crates/registry/src/types/id.rs @@ -7,20 +7,20 @@ use crate::{ jmap::{JsonPointerPatch, RegistryJsonPatch, RegistryValue}, pickle::{Pickle, PickledStream}, - schema::prelude::Object, - types::{EnumType, error::PatchError}, + schema::prelude::ObjectType, + types::{EnumImpl, error::PatchError}, }; use std::{fmt::Display, str::FromStr}; use types::{blob::BlobId, id::Id}; #[derive(Debug, PartialEq, Clone, Copy, Eq, Hash)] pub struct ObjectId { - object: Object, + object: ObjectType, id: Id, } impl ObjectId { - pub fn new(object: Object, id: Id) -> Self { + pub fn new(object: ObjectType, id: Id) -> Self { Self { object, id } } @@ -30,7 +30,7 @@ impl ObjectId { } #[inline(always)] - pub fn object(&self) -> Object { + pub fn object(&self) -> ObjectType { self.object } @@ -46,7 +46,7 @@ impl Display for ObjectId { } } -impl Object { +impl ObjectType { pub fn id(&self, id: Id) -> ObjectId { ObjectId::new(*self, id) } @@ -58,19 +58,19 @@ impl Object { impl Default for ObjectId { fn default() -> Self { - ObjectId::new(Object::Account, Id::default()) + ObjectId::new(ObjectType::Account, Id::default()) } } impl Pickle for Id { fn pickle(&self, out: &mut Vec) { - out.extend_from_slice(&self.id().to_le_bytes()); + out.extend_from_slice(&self.id().to_be_bytes()); } fn unpickle(data: &mut PickledStream<'_>) -> Option { let mut arr = [0u8; std::mem::size_of::()]; arr.copy_from_slice(data.read_bytes(8)?); - let id = u64::from_le_bytes(arr); + let id = u64::from_be_bytes(arr); Some(Id::new(id)) } diff --git a/crates/registry/src/types/index.rs b/crates/registry/src/types/index.rs index 29288691..47a66f48 100644 --- a/crates/registry/src/types/index.rs +++ b/crates/registry/src/types/index.rs @@ -5,7 +5,7 @@ */ use crate::{ - schema::prelude::{Object, Property}, + schema::prelude::{ObjectType, Property}, types::{id::ObjectId, ipmask::IpAddrOrMask}, }; use ahash::AHashSet; @@ -52,12 +52,12 @@ pub enum IndexValue<'x> { #[derive(Debug, Default)] pub struct IndexBuilder<'x> { - pub object: Option, + pub object: Option, pub keys: AHashSet>, } impl<'x> IndexBuilder<'x> { - pub fn object(&mut self, object: Object) { + pub fn object(&mut self, object: ObjectType) { if self.object.is_none() { self.object = Some(object); } @@ -127,7 +127,7 @@ impl<'x> IndexBuilder<'x> { }); } - pub fn foreign_key(&mut self, object: Object, id: Option, type_filter: Option) { + pub fn foreign_key(&mut self, object: ObjectType, id: Option, type_filter: Option) { if let Some(id) = id { self.keys.insert(IndexKey::ForeignKey { object_id: ObjectId::new(object, id), diff --git a/crates/registry/src/types/ipmask.rs b/crates/registry/src/types/ipmask.rs index 4d6b355e..2387aa62 100644 --- a/crates/registry/src/types/ipmask.rs +++ b/crates/registry/src/types/ipmask.rs @@ -210,12 +210,12 @@ impl Pickle for IpAddrOrMask { IpAddrOrMask::V4 { addr, mask } => { out.push(4); out.extend_from_slice(&addr.octets()); - out.extend_from_slice(&mask.to_le_bytes()); + out.extend_from_slice(&mask.to_be_bytes()); } IpAddrOrMask::V6 { addr, mask } => { out.push(6); out.extend_from_slice(&addr.octets()); - out.extend_from_slice(&mask.to_le_bytes()); + out.extend_from_slice(&mask.to_be_bytes()); } } } @@ -229,7 +229,7 @@ impl Pickle for IpAddrOrMask { mask_arr.copy_from_slice(data.read_bytes(4)?); Some(IpAddrOrMask::V4 { addr: Ipv4Addr::from(addr_arr), - mask: u32::from_le_bytes(mask_arr), + mask: u32::from_be_bytes(mask_arr), }) } 6 => { @@ -239,7 +239,7 @@ impl Pickle for IpAddrOrMask { mask_arr.copy_from_slice(data.read_bytes(16)?); Some(IpAddrOrMask::V6 { addr: Ipv6Addr::from(addr_arr), - mask: u128::from_le_bytes(mask_arr), + mask: u128::from_be_bytes(mask_arr), }) } _ => None, diff --git a/crates/registry/src/types/mod.rs b/crates/registry/src/types/mod.rs index 0f68737d..6e61a801 100644 --- a/crates/registry/src/types/mod.rs +++ b/crates/registry/src/types/mod.rs @@ -6,7 +6,7 @@ use crate::{ pickle::Pickle, - schema::prelude::Object, + schema::prelude::ObjectType, types::{error::ValidationError, index::IndexBuilder}, }; use serde::{Serialize, de::DeserializeOwned}; @@ -22,7 +22,7 @@ pub mod ipmask; pub mod socketaddr; pub mod string; -pub trait EnumType: Sized + Debug + PartialEq + Eq { +pub trait EnumImpl: Sized + Debug + PartialEq + Eq { const COUNT: usize; fn parse(s: &str) -> Option; @@ -31,12 +31,12 @@ pub trait EnumType: Sized + Debug + PartialEq + Eq { fn to_id(&self) -> u16; } -pub trait ObjectType: +pub trait ObjectImpl: Pickle + Serialize + DeserializeOwned + Default + Clone + Send + Sync { const FLAGS: u64; + const OBJECT: ObjectType; - fn object() -> Object; fn validate(&self, errors: &mut Vec) -> bool; fn index<'x>(&'x self, builder: &mut IndexBuilder<'x>); } diff --git a/crates/registry/src/types/socketaddr.rs b/crates/registry/src/types/socketaddr.rs index 87598881..e402bb1f 100644 --- a/crates/registry/src/types/socketaddr.rs +++ b/crates/registry/src/types/socketaddr.rs @@ -75,14 +75,14 @@ impl AsRef for SocketAddr { impl Pickle for SocketAddr { fn pickle(&self, out: &mut Vec) { self.0.ip().pickle(out); - out.extend_from_slice(&self.0.port().to_le_bytes()); + out.extend_from_slice(&self.0.port().to_be_bytes()); } fn unpickle(data: &mut PickledStream<'_>) -> Option { let ip = std::net::IpAddr::unpickle(data)?; let mut port_bytes = [0u8; 2]; port_bytes.copy_from_slice(data.read_bytes(2)?); - let port = u16::from_le_bytes(port_bytes); + let port = u16::from_be_bytes(port_bytes); Some(SocketAddr(std::net::SocketAddr::new(ip, port))) } } diff --git a/crates/services/src/broadcast/mod.rs b/crates/services/src/broadcast/mod.rs index 704b826b..64175a33 100644 --- a/crates/services/src/broadcast/mod.rs +++ b/crates/services/src/broadcast/mod.rs @@ -8,8 +8,8 @@ use common::ipc::{ BroadcastEvent, CacheInvalidation, CalendarAlert, PushNotification, RegistryChange, }; use registry::{ - schema::prelude::Object, - types::{EnumType, id::ObjectId}, + schema::prelude::ObjectType, + types::{EnumImpl, id::ObjectId}, }; use std::{borrow::Borrow, io::Write}; use types::{id::Id, type_state::StateChange}; @@ -180,7 +180,7 @@ where let id = self.messages.next_leb128::().ok_or(())?; Ok(Some(BroadcastEvent::RegistryChange( RegistryChange::Insert(ObjectId::new( - Object::from_id(object_id).ok_or(())?, + ObjectType::from_id(object_id).ok_or(())?, Id::new(id), )), ))) @@ -190,7 +190,7 @@ where let id = self.messages.next_leb128::().ok_or(())?; Ok(Some(BroadcastEvent::RegistryChange( RegistryChange::Delete(ObjectId::new( - Object::from_id(object_id).ok_or(())?, + ObjectType::from_id(object_id).ok_or(())?, Id::new(id), )), ))) @@ -198,7 +198,7 @@ where 6 => { let object_id = self.messages.next_leb128().ok_or(())?; Ok(Some(BroadcastEvent::RegistryChange( - RegistryChange::Reload(Object::from_id(object_id).ok_or(())?), + RegistryChange::Reload(ObjectType::from_id(object_id).ok_or(())?), ))) } 7 => { diff --git a/crates/services/src/broadcast/subscriber.rs b/crates/services/src/broadcast/subscriber.rs index 1cbd6125..7d848250 100644 --- a/crates/services/src/broadcast/subscriber.rs +++ b/crates/services/src/broadcast/subscriber.rs @@ -9,7 +9,7 @@ use common::{ BuildServer, Inner, ipc::{BroadcastEvent, HousekeeperEvent, PushEvent, PushNotification, RegistryChange}, }; -use registry::types::EnumType; +use registry::types::EnumImpl; use std::{sync::Arc, time::Duration}; use tokio::sync::watch; use trc::{ClusterEvent, ServerEvent}; diff --git a/crates/services/src/housekeeper/mod.rs b/crates/services/src/housekeeper/mod.rs index 88aa203d..91f22cc3 100644 --- a/crates/services/src/housekeeper/mod.rs +++ b/crates/services/src/housekeeper/mod.rs @@ -667,7 +667,7 @@ pub fn spawn_housekeeper(inner: Arc, mut rx: mpsc::Receiver { use common::ipc::RegistryChange; - use registry::schema::prelude::Object; + use registry::schema::prelude::ObjectType; trc::event!( Housekeeper(trc::HousekeeperEvent::Run), @@ -675,13 +675,11 @@ pub fn spawn_housekeeper(inner: Arc, mut rx: mpsc::Receiver { if let Some(new_core) = result.new_core { - use registry::schema::prelude::Object; - if let Some(enterprise) = &new_core.enterprise { let renew_in = if enterprise.license.is_near_expiration() { @@ -709,7 +707,7 @@ pub fn spawn_housekeeper(inner: Arc, mut rx: mpsc::Receiver Some(store), Err(err) => { - bp.build_error(Object::BlobStore.singleton(), err); + bp.build_error(ObjectType::BlobStore.singleton(), err); None } } diff --git a/crates/store/src/build/data.rs b/crates/store/src/build/data.rs index d3df7ee8..b8a3b45f 100644 --- a/crates/store/src/build/data.rs +++ b/crates/store/src/build/data.rs @@ -6,7 +6,7 @@ use crate::{Store, registry::bootstrap::Bootstrap}; use registry::schema::{ - prelude::Object, + prelude::ObjectType, structs::{DataStore, MetricsStore, TracingStore}, }; @@ -56,7 +56,7 @@ impl Store { match result { Ok(store) => store, Err(err) => { - bp.build_warning(Object::TracingStore.singleton(), err); + bp.build_warning(ObjectType::TracingStore.singleton(), err); None } } @@ -86,7 +86,7 @@ impl Store { match result { Ok(store) => store, Err(err) => { - bp.build_warning(Object::MetricsStore.singleton(), err); + bp.build_warning(ObjectType::MetricsStore.singleton(), err); None } } diff --git a/crates/store/src/build/memory.rs b/crates/store/src/build/memory.rs index 249e3469..d3eb4552 100644 --- a/crates/store/src/build/memory.rs +++ b/crates/store/src/build/memory.rs @@ -5,7 +5,7 @@ */ use crate::{InMemoryStore, registry::bootstrap::Bootstrap}; -use registry::schema::{prelude::Object, structs}; +use registry::schema::{prelude::ObjectType, structs}; #[allow(unreachable_patterns)] impl InMemoryStore { @@ -36,7 +36,7 @@ impl InMemoryStore { match result { Ok(store) => Some(store), Err(err) => { - bp.build_error(Object::InMemoryStore.singleton(), err); + bp.build_error(ObjectType::InMemoryStore.singleton(), err); None } } diff --git a/crates/store/src/build/registry.rs b/crates/store/src/build/registry.rs index ea6a2ac9..72f05800 100644 --- a/crates/store/src/build/registry.rs +++ b/crates/store/src/build/registry.rs @@ -5,12 +5,13 @@ */ use crate::{RegistryStore, RegistryStoreInner, Store}; +use ahash::AHashSet; use registry::{ schema::{ - prelude::Object, + prelude::ObjectType, structs::{DataStore, LocalSettings}, }, - types::id::ObjectId, + types::{EnumImpl, id::ObjectId}, }; use std::path::PathBuf; use types::id::Id; @@ -25,30 +26,49 @@ impl RegistryStore { let Some(data_store) = inner .local_registry .read() - .get(&ObjectId::new(Object::DataStore, Id::singleton())) + .get(&ObjectId::new(ObjectType::DataStore, Id::singleton())) .cloned() + .map(DataStore::from) else { return Err(format!( "{ERROR_MSG}: Missing \"DataStore\" object definition." )); }; - let data_store = serde_json::from_value::(data_store) - .map_err(|err| format!("{ERROR_MSG}: Failed to parse \"DataStore\" object: {err}"))?; + let Some(local_settings) = inner .local_registry .read() - .get(&ObjectId::new(Object::LocalSettings, Id::singleton())) + .get(&ObjectId::new(ObjectType::LocalSettings, Id::singleton())) .cloned() + .map(LocalSettings::from) else { return Err(format!( "{ERROR_MSG}: Missing \"LocalSettings\" object definition." )); }; - let local_settings = - serde_json::from_value::(local_settings).map_err(|err| { - format!("{ERROR_MSG}: Failed to parse \"LocalSettings\" object: {err}") - })?; + // Validate local objects + let mut local_objects = + AHashSet::from_iter([ObjectType::DataStore, ObjectType::LocalSettings]); + for object in local_settings.local_registry_object_types { + if let Some(object) = ObjectType::parse(&object) { + local_objects.insert(object); + } else { + return Err(format!( + "{ERROR_MSG}: LocalSettings/localRegistryObjectImpls contains invalid object type: {object}" + )); + } + } + for object_id in inner.local_registry.read().keys() { + if !local_objects.contains(&object_id.object()) { + return Err(format!( + "{ERROR_MSG}: Found object of type {:?} in local registry, but it is not listed in LocalSettings/localRegistryObjectImpls.", + object_id.object().as_str() + )); + } + } + + inner.local_objects = local_objects; inner.store = Store::build(data_store).await?; inner.node_id = local_settings.node_id; if inner.node_id == 0 { diff --git a/crates/store/src/build/search.rs b/crates/store/src/build/search.rs index 8733f3e4..608b958a 100644 --- a/crates/store/src/build/search.rs +++ b/crates/store/src/build/search.rs @@ -9,7 +9,7 @@ use crate::{ backend::{elastic::ElasticSearchStore, meili::MeiliSearchStore}, registry::bootstrap::Bootstrap, }; -use registry::schema::{prelude::Object, structs}; +use registry::schema::{prelude::ObjectType, structs}; #[allow(unreachable_patterns)] impl SearchStore { @@ -48,7 +48,7 @@ impl SearchStore { match result { Ok(store) => Some(store), Err(err) => { - bp.build_error(Object::SearchStore.singleton(), err); + bp.build_error(ObjectType::SearchStore.singleton(), err); None } } diff --git a/crates/store/src/lib.rs b/crates/store/src/lib.rs index d9ef77fa..08acd373 100644 --- a/crates/store/src/lib.rs +++ b/crates/store/src/lib.rs @@ -13,7 +13,10 @@ pub mod search; pub mod write; use ::registry::{ - schema::{enums::CompressionAlgo, prelude::Object}, + schema::{ + enums::CompressionAlgo, + prelude::{Object, ObjectType}, + }, types::id::ObjectId, }; pub use ahash; @@ -199,8 +202,8 @@ pub struct RegistryStore(pub(crate) Arc); pub struct RegistryStoreInner { pub(crate) local_path: PathBuf, - pub(crate) local_registry: RwLock>, - pub(crate) local_objects: AHashSet, + pub(crate) local_registry: RwLock>, + pub(crate) local_objects: AHashSet, pub(crate) store: Store, pub(crate) node_id: u64, pub(crate) id_generator: SnowflakeIdGenerator, diff --git a/crates/store/src/registry/bootstrap.rs b/crates/store/src/registry/bootstrap.rs index 3f870f74..c3022042 100644 --- a/crates/store/src/registry/bootstrap.rs +++ b/crates/store/src/registry/bootstrap.rs @@ -11,11 +11,11 @@ use crate::{ use ahash::AHashSet; use registry::{ schema::{ - prelude::{Object, Property}, + prelude::{Object, ObjectType, Property}, structs::Node, }, types::{ - EnumType, ObjectType, + EnumImpl, ObjectImpl, error::{Error, ValidationError, Warning}, id::ObjectId, }, @@ -56,7 +56,7 @@ impl Bootstrap { let ids = match self .registry .query::>( - RegistryQuery::new(Object::Node).equal(Property::NodeId, self.node_id()), + RegistryQuery::new(ObjectType::Node).equal(Property::NodeId, self.node_id()), ) .await { @@ -77,7 +77,10 @@ impl Bootstrap { self.node = node; } else { self.warnings.push(Warning { - object_id: ObjectId::new(Object::Node, id.map(Id::new).unwrap_or(Id::singleton())), + object_id: ObjectId::new( + ObjectType::Node, + id.map(Id::new).unwrap_or(Id::singleton()), + ), property: Some(Property::NodeId), message: format!( "No node configuration found for nodeId {}, using defaults.", @@ -88,8 +91,8 @@ impl Bootstrap { } } - pub async fn setting(&mut self) -> trc::Result { - let object_id = T::object().singleton(); + pub async fn setting>(&mut self) -> trc::Result { + let object_id = T::OBJECT.singleton(); if let Some(setting) = self.registry.object::(object_id.id()).await? { let mut errors = Vec::new(); @@ -102,13 +105,13 @@ impl Bootstrap { Ok(T::default()) } - pub async fn setting_infallible(&mut self) -> T { + pub async fn setting_infallible>(&mut self) -> T { match self.setting::().await { Ok(setting) => setting, Err(err) => { if !self.has_fatal_errors { self.errors.push(Error::Internal { - object_id: Some(T::object().singleton()), + object_id: Some(T::OBJECT.singleton()), error: err, }); self.has_fatal_errors = true; @@ -118,7 +121,7 @@ impl Bootstrap { } } - pub async fn get_infallible(&mut self, id: Id) -> Option { + pub async fn get_infallible>(&mut self, id: Id) -> Option { match self.registry.object::(id).await { Ok(Some(setting)) => { let mut errors = Vec::new(); @@ -126,7 +129,7 @@ impl Bootstrap { Some(setting) } else { self.errors.push(Error::Validation { - object_id: ObjectId::new(T::object(), id), + object_id: ObjectId::new(T::OBJECT, id), errors, }); None @@ -134,14 +137,14 @@ impl Bootstrap { } Ok(None) => { self.errors.push(Error::NotFound { - object_id: ObjectId::new(T::object(), id), + object_id: ObjectId::new(T::OBJECT, id), }); None } Err(err) => { if !self.has_fatal_errors { self.errors.push(Error::Internal { - object_id: Some(ObjectId::new(T::object(), id)), + object_id: Some(ObjectId::new(T::OBJECT, id)), error: err, }); self.has_fatal_errors = true; @@ -151,7 +154,9 @@ impl Bootstrap { } } - pub async fn list_infallible(&mut self) -> Vec> { + pub async fn list_infallible>( + &mut self, + ) -> Vec> { match self.registry.list::().await { Ok(objects) => objects .into_iter() @@ -195,7 +200,7 @@ impl Bootstrap { }); } - pub fn validate(&mut self, id: ObjectId, object: &impl ObjectType) -> bool { + pub fn validate(&mut self, id: ObjectId, object: &impl ObjectImpl) -> bool { let mut errors = Vec::new(); if object.validate(&mut errors) { true diff --git a/crates/store/src/registry/get.rs b/crates/store/src/registry/get.rs index ac0af843..58d42620 100644 --- a/crates/store/src/registry/get.rs +++ b/crates/store/src/registry/get.rs @@ -11,81 +11,61 @@ use crate::{ }; use registry::{ pickle::PickledStream, - types::{EnumType, ObjectType, id::ObjectId}, + schema::prelude::Object, + types::{EnumImpl, ObjectImpl, id::ObjectId}, }; use trc::AddContext; use types::id::Id; use utils::codec::leb128::Leb128Reader; impl RegistryStore { - pub async fn object(&self, id: Id) -> trc::Result> { - let item_id = id.id(); - let object = T::object(); - - if self.0.local_objects.contains(&object) { - let Some(item) = self - .0 - .local_registry - .read() - .get(&ObjectId::new(object, id)) - .cloned() - else { - return Ok(None); - }; - serde_json::from_value::(item).map(Some).map_err(|err| { - trc::EventType::Registry(trc::RegistryEvent::LocalParseError) - .into_err() - .caused_by(trc::location!()) - .id(item_id) - .details(object.as_str()) - .reason(err) - }) + pub async fn get(&self, object_id: ObjectId) -> trc::Result> { + if self.0.local_objects.contains(&object_id.object()) { + Ok(self.0.local_registry.read().get(&object_id).cloned()) } else { - let Some(bytes) = self - .0 + self.0 .store - .get_value::(ValueKey::from(ValueClass::Registry( - RegistryClass::Item { - object_id: object.to_id(), - item_id, - }, - ))) - .await? - else { - return Ok(None); - }; - T::unpickle(&mut PickledStream::new(&bytes.0)) - .ok_or_else(|| { - trc::EventType::Registry(trc::RegistryEvent::DeserializationError) - .into_err() - .caused_by(trc::location!()) - .id(item_id) - .details(object.as_str()) - .ctx(trc::Key::Value, bytes.0) + .get_value::(ValueKey::from(ValueClass::Registry(RegistryClass::Item { + object_id: object_id.object().to_id(), + item_id: object_id.id().id(), + }))) + .await + .and_then(|v| { + if v.as_ref() + .is_none_or(|v| v.object_type() == object_id.object()) + { + Ok(v) + } else { + Err( + trc::EventType::Registry(trc::RegistryEvent::DeserializationError) + .into_err() + .caused_by(trc::location!()) + .id(object_id.id().id()) + .details(object_id.object().as_str()) + .reason("Object type mismatch"), + ) + } }) - .map(Some) } } - pub async fn list(&self) -> trc::Result>> { - let object = T::object(); + pub async fn object>(&self, id: Id) -> trc::Result> { + self.get(ObjectId::new(T::OBJECT, id)) + .await + .map(|v| v.map(T::from)) + } + + pub async fn list>(&self) -> trc::Result>> { + let object = T::OBJECT; if self.0.local_objects.contains(&object) { let mut results = Vec::new(); for (id, item) in self.0.local_registry.read().iter() { if id.object() == object { - let item = serde_json::from_value::(item.clone()).map_err(|err| { - trc::EventType::Registry(trc::RegistryEvent::LocalParseError) - .into_err() - .caused_by(trc::location!()) - .id(id.id().id()) - .details(object.as_str()) - .reason(err) - })?; results.push(RegistryObject { id: *id, - object: item, + object: T::from(item.clone()), }); } } @@ -125,15 +105,17 @@ impl RegistryStore { .details(object.as_str()) .ctx(trc::Key::Key, key) })?; - let item = - T::unpickle(&mut PickledStream::new(value)).ok_or_else(|| { - trc::EventType::Registry(trc::RegistryEvent::DeserializationError) - .into_err() - .caused_by(trc::location!()) - .id(id) - .details(object.as_str()) - .ctx(trc::Key::Value, value) - })?; + let item = T::unpickle(&mut PickledStream::new( + value.get(U16_LEN..).unwrap_or_default(), + )) + .ok_or_else(|| { + trc::EventType::Registry(trc::RegistryEvent::DeserializationError) + .into_err() + .caused_by(trc::location!()) + .id(id) + .details(object.as_str()) + .ctx(trc::Key::Value, value) + })?; results.push(RegistryObject { id: ObjectId::new(object, Id::new(id)), object: item, @@ -150,10 +132,13 @@ impl RegistryStore { } } -struct PickledBytes(Vec); - -impl Deserialize for PickledBytes { +impl Deserialize for Object { fn deserialize(bytes: &[u8]) -> trc::Result { - Ok(Self(bytes.to_vec())) + Object::unpickle(&mut PickledStream::new(bytes)).ok_or_else(|| { + trc::EventType::Registry(trc::RegistryEvent::DeserializationError) + .into_err() + .caused_by(trc::location!()) + .ctx(trc::Key::Value, bytes) + }) } } diff --git a/crates/store/src/registry/local.rs b/crates/store/src/registry/local.rs index 579533c3..9db6793f 100644 --- a/crates/store/src/registry/local.rs +++ b/crates/store/src/registry/local.rs @@ -8,8 +8,8 @@ use crate::{RegistryStore, RegistryStoreInner, Store}; use ahash::AHashMap; use parking_lot::RwLock; use registry::{ - schema::prelude::{OBJ_SINGLETON, Object}, - types::{EnumType, id::ObjectId}, + schema::prelude::{OBJ_SINGLETON, Object, ObjectType}, + types::{EnumImpl, id::ObjectId}, }; use serde_json::{Map, Value, map::Entry}; use std::path::PathBuf; @@ -17,7 +17,7 @@ use types::id::Id; use utils::snowflake::SnowflakeIdGenerator; impl RegistryStoreInner { - pub async fn load(local_path: PathBuf) -> Result { + pub(crate) async fn load(local_path: PathBuf) -> Result { let error_msg = format!("Failed to read local registry at {}", local_path.display()); let contents = tokio::fs::read_to_string(&local_path) .await @@ -31,7 +31,7 @@ impl RegistryStoreInner { let mut local_registry = AHashMap::new(); for (key, value) in object.into_iter() { - let object_type = Object::parse(key.as_str()) + let object_type = ObjectType::parse(key.as_str()) .ok_or_else(|| format!("{error_msg}: Unrecognized object {key:?}."))?; let is_singleton = object_type.flags() & OBJ_SINGLETON != 0; let Value::Object(object) = value else { @@ -48,7 +48,24 @@ impl RegistryStoreInner { )); } if local_registry - .insert(ObjectId::new(object_type, Id::new(id)), value) + .insert(ObjectId::new(object_type, Id::new(id)), Object::deserialize(object_type, value).map_err(|err| { + format!("{error_msg}: Failed to parse object {key:?} with id {id}: {err}") + }).and_then(|obj| { + let mut errors = Vec::new(); + obj.validate(&mut errors); + if errors.is_empty() { + Ok(obj) + } else { + Err(format!( + "{error_msg}: Validation errors for object {key:?} with id {id}: {}", + errors + .into_iter() + .map(|e| e.to_string()) + .collect::>() + .join("; ") + )) + } + })?) .is_some() { return Err(format!( @@ -59,7 +76,26 @@ impl RegistryStoreInner { } else if local_registry .insert( ObjectId::new(object_type, Id::singleton()), - Value::Object(object), + Object::deserialize(object_type, object) + .map_err(|err| { + format!("{error_msg}: Failed to parse object {key:?}: {err}") + }) + .and_then(|obj| { + let mut errors = Vec::new(); + obj.validate(&mut errors); + if errors.is_empty() { + Ok(obj) + } else { + Err(format!( + "{error_msg}: Validation errors for object {key:?}: {}", + errors + .into_iter() + .map(|e| e.to_string()) + .collect::>() + .join("; ") + )) + } + })?, ) .is_some() { @@ -89,15 +125,39 @@ impl RegistryStore { match map.entry(id.object().as_str().to_string()) { Entry::Vacant(entry) => { if is_singleton { - entry.insert(value.clone()); + entry.insert(serde_json::to_value(&value.inner).map_err(|err| { + trc::EventType::Registry(trc::RegistryEvent::LocalWriteError) + .into_err() + .caused_by(trc::location!()) + .reason(err) + })?); } else { - entry.insert(Map::from_iter([(id.id().to_string(), value.clone())]).into()); + entry.insert( + Map::from_iter([( + id.id().to_string(), + serde_json::to_value(&value.inner).map_err(|err| { + trc::EventType::Registry(trc::RegistryEvent::LocalWriteError) + .into_err() + .caused_by(trc::location!()) + .reason(err) + })?, + )]) + .into(), + ); } } Entry::Occupied(mut entry) => { if !is_singleton { if let Value::Object(map) = entry.get_mut() { - map.insert(id.id().to_string(), value.clone()); + map.insert( + id.id().to_string(), + serde_json::to_value(&value.inner).map_err(|err| { + trc::EventType::Registry(trc::RegistryEvent::LocalWriteError) + .into_err() + .caused_by(trc::location!()) + .reason(err) + })?, + ); } } else { debug_assert!(false, "Unexpected double singleton assignment"); diff --git a/crates/store/src/registry/mod.rs b/crates/store/src/registry/mod.rs index c67d6ed1..e53284d5 100644 --- a/crates/store/src/registry/mod.rs +++ b/crates/store/src/registry/mod.rs @@ -11,25 +11,17 @@ pub mod query; pub mod write; use registry::{ - pickle::{Pickle, PickledStream}, - schema::prelude::{Object, Property}, - types::{ObjectType, id::ObjectId}, + schema::prelude::{ObjectType, Property}, + types::{ObjectImpl, id::ObjectId}, }; -#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)] - -pub struct HashedObject { - pub hash: u64, - pub object: T, -} - -pub struct RegistryObject { +pub struct RegistryObject { pub id: ObjectId, pub object: T, } pub struct RegistryQuery { - pub object_type: Object, + pub object_type: ObjectType, pub filters: Vec, pub account_id: Option, pub tenant_id: Option, @@ -57,30 +49,3 @@ pub enum RegistryFilterValue { U16(u16), Boolean(bool), } - -impl Pickle for HashedObject { - fn pickle(&self, out: &mut Vec) { - self.object.pickle(out); - } - - fn unpickle(stream: &mut PickledStream<'_>) -> Option { - let hash = xxhash_rust::xxh3::xxh3_64(stream.bytes()); - T::unpickle(stream).map(|object| Self { hash, object }) - } -} - -impl ObjectType for HashedObject { - const FLAGS: u64 = T::FLAGS; - - fn object() -> Object { - T::object() - } - - fn validate(&self, errors: &mut Vec) -> bool { - self.object.validate(errors) - } - - fn index<'x>(&'x self, builder: &mut registry::schema::prelude::IndexBuilder<'x>) { - self.object.index(builder) - } -} diff --git a/crates/store/src/registry/query.rs b/crates/store/src/registry/query.rs index 835b2753..b518c7c9 100644 --- a/crates/store/src/registry/query.rs +++ b/crates/store/src/registry/query.rs @@ -14,8 +14,8 @@ use crate::{ }; use ahash::AHashSet; use registry::{ - schema::prelude::{OBJ_FILTER_ACCOUNT, OBJ_FILTER_TENANT, OBJ_SINGLETON, Object, Property}, - types::EnumType, + schema::prelude::{OBJ_FILTER_ACCOUNT, OBJ_FILTER_TENANT, OBJ_SINGLETON, ObjectType, Property}, + types::EnumImpl, }; use roaring::RoaringBitmap; use std::{borrow::Cow, ops::BitAndAssign}; @@ -197,7 +197,7 @@ impl RegistryQueryResults for RoaringBitmap { } impl RegistryQuery { - pub fn new(object_type: Object) -> Self { + pub fn new(object_type: ObjectType) -> Self { Self { object_type, filters: Vec::new(), @@ -364,7 +364,7 @@ impl From for RegistryFilterValue { } } -async fn all_ids(store: &Store, object: Object) -> trc::Result { +async fn all_ids(store: &Store, object: ObjectType) -> trc::Result { let mut bm = T::default(); let object_id = object.to_id(); store @@ -394,7 +394,7 @@ async fn all_ids(store: &Store, object: Object) -> trc: async fn range_to_set( store: &Store, - object: Object, + object: ObjectType, index_id: u16, match_value: &[u8], op: RegistryFilterOp, diff --git a/crates/store/src/registry/write.rs b/crates/store/src/registry/write.rs index c31b17e7..6c321755 100644 --- a/crates/store/src/registry/write.rs +++ b/crates/store/src/registry/write.rs @@ -7,7 +7,6 @@ use crate::{ IterateParams, RegistryStore, SUBSPACE_REGISTRY, SerializeInfallible, U16_LEN, U64_LEN, ValueKey, - registry::HashedObject, write::{ AnyClass, BatchBuilder, RegistryClass, ValueClass, assert::AssertValue, @@ -16,16 +15,17 @@ use crate::{ }; use registry::{ schema::prelude::{ - OBJ_FILTER_ACCOUNT, OBJ_FILTER_TENANT, OBJ_SEQ_ID, OBJ_SINGLETON, Object, Property, + OBJ_FILTER_ACCOUNT, OBJ_FILTER_TENANT, OBJ_SEQ_ID, OBJ_SINGLETON, Object, ObjectType, + Property, }, types::{ - EnumType, ObjectType, + EnumImpl, error::ValidationError, id::ObjectId, index::{IndexBuilder, IndexKey, IndexValue}, }, }; -use std::fmt::Display; +use std::{borrow::Cow, fmt::Display}; use trc::AddContext; use types::id::Id; use utils::codec::leb128::Leb128Reader; @@ -56,42 +56,41 @@ pub enum RegistryWriteResult { NotSupported, } -pub struct RegistryWrite<'x, T: ObjectType> { - op: RegistryWriteOp<'x, T>, +pub struct RegistryWrite<'x> { + op: RegistryWriteOp<'x>, current_tenant_id: Option, current_account_id: Option, } -pub enum RegistryWriteOp<'x, T: ObjectType> { +pub enum RegistryWriteOp<'x> { Insert { - object: &'x T, + object: &'x Object, id: Option, }, Update { - object: &'x T, + object: &'x Object, id: Id, - old_object: &'x HashedObject, + old_object: &'x Object, }, Delete { - id: Id, + object_id: ObjectId, + object: Option<&'x Object>, }, } impl RegistryStore { - pub async fn write( - &self, - write: RegistryWrite<'_, T>, - ) -> trc::Result { - let object_type = T::object(); - let object_flags = T::FLAGS; - let object_id = object_type.to_id(); + pub async fn write(&self, write: RegistryWrite<'_>) -> trc::Result { let mut set_index = IndexBuilder::default(); let mut clear_index = IndexBuilder::default(); - let mut batch = BatchBuilder::new(); - let mut item_id; let object; + let object_type; + let object_flags; + let object_id; let object_tenant_id; + let mut item_id; + + let mut batch = BatchBuilder::new(); let mut write_id = true; let mut generate_id = false; @@ -101,6 +100,9 @@ impl RegistryStore { id, } => { object = insert_object; + object_flags = object.flags(); + object_type = object.object_type(); + object_id = object_type.to_id(); object.index(&mut set_index); object_tenant_id = set_index.tenant_id(); @@ -122,12 +124,15 @@ impl RegistryStore { old_object, } => { object = update_object; + object_flags = object.flags(); + object_type = object.object_type(); + object_id = object_type.to_id(); object.index(&mut set_index); object_tenant_id = set_index.tenant_id(); // Obtain changes let mut old_index = IndexBuilder::default(); - old_object.object.index(&mut old_index); + old_object.index(&mut old_index); for key in &old_index.keys { set_index.keys.remove(key); } @@ -142,12 +147,12 @@ impl RegistryStore { item_id = id.id(); batch.assert_value( ValueClass::Registry(RegistryClass::Item { object_id, item_id }), - AssertValue::Hash(old_object.hash), + AssertValue::U32(old_object.revision), ); } - RegistryWriteOp::Delete { id } => { - return if object_flags & OBJ_SINGLETON == 0 { - self.delete(write, id).await + RegistryWriteOp::Delete { object_id, object } => { + return if object_id.object().flags() & OBJ_SINGLETON == 0 { + self.delete(write, object_id, object).await } else { Ok(RegistryWriteResult::CannotDeleteSingleton) }; @@ -180,17 +185,10 @@ impl RegistryStore { return Ok(RegistryWriteResult::NotSupported); } let id = Id::new(item_id); - self.0.local_registry.write().insert( - ObjectId::new(object_type, id), - serde_json::to_value(object.clone()).map_err(|err| { - trc::EventType::Registry(trc::RegistryEvent::LocalWriteError) - .into_err() - .caused_by(trc::location!()) - .id(item_id) - .details(object_type.as_str()) - .reason(err) - })?, - ); + self.0 + .local_registry + .write() + .insert(ObjectId::new(object_type, id), object.clone()); return self .write_local_registry() .await @@ -338,13 +336,15 @@ impl RegistryStore { Ok(RegistryWriteResult::Success(Id::new(item_id))) } - async fn delete( + async fn delete( &self, - write: RegistryWrite<'_, T>, - id: Id, + write: RegistryWrite<'_>, + object_id: ObjectId, + object: Option<&Object>, ) -> trc::Result { - let object_type = T::object(); - let object_id = object_type.to_id(); + let object_type = object_id.object(); + let object_type_id = object_type.to_id(); + let id = object_id.id(); let item_id = id.id(); if self.0.local_objects.contains(&object_type) { @@ -359,7 +359,11 @@ impl RegistryStore { } // Fetch object - let Some(object) = self.object::>(id).await? else { + let object = if let Some(object) = object { + Cow::Borrowed(object) + } else if let Some(object) = self.get(object_id).await? { + Cow::Owned(object) + } else { return Ok(RegistryWriteResult::NotFound { object_id: ObjectId::new(object_type, id), }); @@ -367,7 +371,7 @@ impl RegistryStore { // Validate tenant and account changes let mut clear_index = IndexBuilder::default(); - object.object.index(&mut clear_index); + object.index(&mut clear_index); if let Some(err) = write.validate_owner(&clear_index) { return Ok(err); } @@ -376,7 +380,7 @@ impl RegistryStore { let mut linked = Vec::new(); let key = KeySerializer::new(U64_LEN + U16_LEN + 1) .write(1u8) - .write(object_id) + .write(object_type_id) .write(item_id) .finalize(); let prefix_len = key.len(); @@ -386,7 +390,7 @@ impl RegistryStore { })); let key = KeySerializer::new((U64_LEN * 2) + U16_LEN + 1) .write(1u8) - .write(object_id) + .write(object_type_id) .write(item_id) .write(u64::MAX) .finalize(); @@ -399,8 +403,8 @@ impl RegistryStore { .iterate( IterateParams::new(from_key, to_key).no_values().ascending(), |key, _| { - let object = - Object::from_id(key.deserialize_be_u16(prefix_len)?).ok_or_else(|| { + let object = ObjectType::from_id(key.deserialize_be_u16(prefix_len)?) + .ok_or_else(|| { trc::EventType::Registry(trc::RegistryEvent::DeserializationError) .into_err() .caused_by(trc::location!()) @@ -436,18 +440,21 @@ impl RegistryStore { let mut batch = BatchBuilder::new(); batch .assert_value( - ValueClass::Registry(RegistryClass::Item { object_id, item_id }), - AssertValue::Hash(object.hash), + ValueClass::Registry(RegistryClass::Item { + object_id: object_type_id, + item_id, + }), + AssertValue::U32(object.revision), ) .clear(ValueClass::Registry(RegistryClass::Item { - object_id, + object_id: object_type_id, item_id, })) .clear(ValueClass::Registry(RegistryClass::Id { - object_id, + object_id: object_type_id, item_id, })) - .registry_index(object_id, item_id, clear_index.keys.iter(), false); + .registry_index(object_type_id, item_id, clear_index.keys.iter(), false); self.0 .store @@ -461,7 +468,7 @@ impl RegistryStore { &self, from_key: RegistryClass, to_key: RegistryClass, - object: Option, + object: Option, ) -> trc::Result> { let from_key = ValueKey::from(from_key); let to_key = ValueKey::from(to_key); @@ -480,7 +487,7 @@ impl RegistryStore { } else { let object_id = key.deserialize_be_u16(key.len() - U64_LEN - U16_LEN)?; - Object::from_id(object_id).ok_or_else(|| { + ObjectType::from_id(object_id).ok_or_else(|| { trc::EventType::Registry(trc::RegistryEvent::DeserializationError) .into_err() .caused_by(trc::location!()) @@ -587,8 +594,8 @@ impl SerializeInfallible for IndexValue<'_> { } } -impl<'x, T: ObjectType> RegistryWrite<'x, T> { - pub fn insert(object: &'x T) -> Self { +impl<'x> RegistryWrite<'x> { + pub fn insert(object: &'x Object) -> Self { Self { op: RegistryWriteOp::Insert { object, id: None }, current_tenant_id: None, @@ -596,7 +603,7 @@ impl<'x, T: ObjectType> RegistryWrite<'x, T> { } } - pub fn insert_with_id(id: Id, object: &'x T) -> Self { + pub fn insert_with_id(id: Id, object: &'x Object) -> Self { Self { op: RegistryWriteOp::Insert { object, @@ -607,7 +614,7 @@ impl<'x, T: ObjectType> RegistryWrite<'x, T> { } } - pub fn update(id: Id, object: &'x T, old_object: &'x HashedObject) -> Self { + pub fn update(id: Id, object: &'x Object, old_object: &'x Object) -> Self { Self { op: RegistryWriteOp::Update { object, @@ -619,9 +626,23 @@ impl<'x, T: ObjectType> RegistryWrite<'x, T> { } } - pub fn delete(id: Id) -> Self { + pub fn delete(object_id: ObjectId) -> Self { Self { - op: RegistryWriteOp::Delete { id }, + op: RegistryWriteOp::Delete { + object_id, + object: None, + }, + current_tenant_id: None, + current_account_id: None, + } + } + + pub fn delete_object(object_id: ObjectId, object: &'x Object) -> Self { + Self { + op: RegistryWriteOp::Delete { + object_id, + object: Some(object), + }, current_tenant_id: None, current_account_id: None, } diff --git a/crates/store/src/write/assert.rs b/crates/store/src/write/assert.rs index 0920a25d..eedeebe6 100644 --- a/crates/store/src/write/assert.rs +++ b/crates/store/src/write/assert.rs @@ -12,7 +12,6 @@ pub enum AssertValue { U32(u32), U64(u64), Archive(ArchiveVersion), - Hash(u64), Some, None, } @@ -78,7 +77,6 @@ impl AssertValue { }, AssertValue::None => false, AssertValue::Some => true, - AssertValue::Hash(v) => xxhash_rust::xxh3::xxh3_64(bytes) == *v, } }