filip/Stalwart
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

JMAP Registry API implementation - part 6

Browse Source
This commit is contained in:
mdecimus
2026-03-01 18:30:04 +01:00
parent 07d5748f6b
commit 9ccc2ed6f0
74 changed files with 2417 additions and 785 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
crates/store/src/backend/azure/mod.rs
View File

@@ -23,7 +23,10 @@ pub struct AzureStore {
impl AzureStore {
pub async fn open(config: structs::AzureStore) -> Result<BlobStore, String> {
let credentials = match (config.access_key, config.sas_token) {
let credentials = match (
config.access_key.secret().await?.map(|v| v.into_owned()),
config.sas_token.secret().await?.map(|v| v.into_owned()),
) {
(Some(access_key), None) => {
StorageCredentials::access_key(config.storage_account.clone(), access_key)
}

15
crates/store/src/backend/elastic/main.rs
View File

@@ -23,12 +23,15 @@ impl ElasticSearchStore {
Url::parse(&config.url).map_err(|e| format!("Invalid URL: {e}",))?;
Ok(SearchStore::ElasticSearch(Arc::new(Self {
client: config.http_auth.build_http_client(
config.http_headers,
"application/json".into(),
config.timeout,
config.allow_invalid_certs,
)?,
client: config
.http_auth
.build_http_client(
config.http_headers,
"application/json".into(),
config.timeout,
config.allow_invalid_certs,
)
.await?,
url: config.url,
num_replicas: config.num_replicas as usize,
num_shards: config.num_shards as usize,

15
crates/store/src/backend/meili/main.rs
View File

@@ -19,12 +19,15 @@ use std::{sync::Arc, time::Duration};
impl MeiliSearchStore {
pub async fn open(config: structs::MeilisearchStore) -> Result<SearchStore, String> {
let client = config.http_auth.build_http_client(
config.http_headers,
"application/json".into(),
config.timeout,
config.allow_invalid_certs,
)?;
let client = config
.http_auth
.build_http_client(
config.http_headers,
"application/json".into(),
config.timeout,
config.allow_invalid_certs,
)
.await?;
Url::parse(&config.url).map_err(|e| format!("Invalid URL: {e}",))?;

4
crates/store/src/backend/mysql/main.rs
View File

@@ -23,7 +23,7 @@ impl MysqlStore {
let mut opts = OptsBuilder::default()
.ip_or_hostname(config.host)
.user(config.auth_username)
.pass(config.auth_secret)
.pass(config.auth_secret.secret().await?.map(|v| v.into_owned()))
.db_name(Some(config.database))
.max_allowed_packet(config.max_allowed_packet.map(|v| v as usize))
.wait_timeout(config.timeout.map(|t| t.as_secs() as usize))
@@ -58,7 +58,7 @@ impl MysqlStore {
opts.clone()
.ip_or_hostname(replica.host)
.user(replica.auth_username)
.pass(replica.auth_secret)
.pass(replica.auth_secret.secret().await?.map(|v| v.into_owned()))
.db_name(Some(replica.database))
.tcp_port(replica.port as u16),
),

2
crates/store/src/backend/mysql/read.rs
View File

@@ -23,7 +23,7 @@ impl MysqlStore {
.await
.map_err(into_error)?;
let key = key.serialize(0);
conn.exec_first::<Vec<u8>, _, _>(&s, (key,))
conn.exec_first::<Vec<u8>, _, _>(&s, (&key,))
.await
.map_err(into_error)
.and_then(|r| {

4
crates/store/src/backend/postgres/main.rs
View File

@@ -25,7 +25,7 @@ impl PostgresStore {
cfg.dbname = config.database.into();
cfg.host = config.host.into();
cfg.user = config.auth_username;
cfg.password = config.auth_secret;
cfg.password = config.auth_secret.secret().await?.map(|v| v.into_owned());
cfg.port = (config.port as u16).into();
cfg.connect_timeout = config.timeout.map(|t| t.into_inner());
cfg.options = config.options;
@@ -47,7 +47,7 @@ impl PostgresStore {
cfg.dbname = replica.database.into();
cfg.host = replica.host.into();
cfg.user = replica.auth_username;
cfg.password = replica.auth_secret;
cfg.password = replica.auth_secret.secret().await?.map(|v| v.into_owned());
cfg.port = (replica.port as u16).into();
cfg.options = replica.options;
replicas.push(Store::PostgreSQL(Arc::new(PostgresStore {

2
crates/store/src/backend/redis/mod.rs
View File

@@ -64,7 +64,7 @@ impl RedisStore {
if let Some(value) = config.auth_username {
builder = builder.username(value);
}
if let Some(value) = config.auth_secret {
if let Some(value) = config.auth_secret.secret().await?.map(|v| v.into_owned()) {
builder = builder.password(value);
}
if let Some(value) = config.max_retries {

6
crates/store/src/backend/s3/mod.rs
View File

@@ -69,9 +69,9 @@ impl S3Store {
};
let credentials = Credentials::new(
config.access_key.as_deref(),
config.secret_key.as_deref(),
config.security_token.as_deref(),
config.session_token.as_deref(),
config.secret_key.secret().await?.as_deref(),
config.security_token.secret().await?.as_deref(),
config.session_token.secret().await?.as_deref(),
config.profile.as_deref(),
)
.map_err(|err| format!("Failed to create credentials: {err:?}"))?;

4
crates/store/src/registry/query.rs
View File

@@ -142,6 +142,10 @@ impl RegistryStore {
Ok(results)
}
pub async fn count(&self, query: RegistryQuery) -> trc::Result<usize> {
self.query::<AHashSet<u64>>(query).await.map(|r| r.len())
}
}
pub trait RegistryQueryResults: Default + Sized + Sync + Send {

239
crates/store/src/registry/write.rs
View File

@@ -49,18 +49,10 @@ pub enum RegistryWriteResult {
ValidationError {
errors: Vec<ValidationError>,
},
InvalidTenantId,
InvalidAccountId,
NotSupported,
}
pub struct RegistryWrite<'x> {
op: RegistryWriteOp<'x>,
current_tenant_id: Option<u32>,
current_account_id: Option<u32>,
}
pub enum RegistryWriteOp<'x> {
pub enum RegistryWrite<'x> {
Insert {
object: &'x Object,
id: Option<Id>,
@@ -85,15 +77,14 @@ impl RegistryStore {
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;
match write.op {
RegistryWriteOp::Insert {
match write {
RegistryWrite::Insert {
object: insert_object,
id,
} => {
@@ -102,7 +93,6 @@ impl RegistryStore {
object_type = object.object_type();
object_id = object_type.to_id();
object.index(&mut set_index);
object_tenant_id = set_index.tenant_id();
item_id = if let Some(id) = id {
id.id()
@@ -116,7 +106,7 @@ impl RegistryStore {
self.0.id_generator.generate()
};
}
RegistryWriteOp::Update {
RegistryWrite::Update {
object: update_object,
id,
old_object,
@@ -126,7 +116,6 @@ impl RegistryStore {
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();
@@ -148,9 +137,9 @@ impl RegistryStore {
AssertValue::Hash(old_object.revision),
);
}
RegistryWriteOp::Delete { object_id, object } => {
RegistryWrite::Delete { object_id, object } => {
return if object_id.object().flags() & OBJ_SINGLETON == 0 {
self.delete(write, object_id, object).await
self.delete(object_id, object).await
} else {
Ok(RegistryWriteResult::CannotDeleteSingleton)
};
@@ -164,19 +153,6 @@ impl RegistryStore {
return Ok(RegistryWriteResult::ValidationError { errors });
}
// Validate tenant ownership
if write.current_tenant_id.is_some()
&& (object_flags & OBJ_FILTER_TENANT) != 0
&& write.current_tenant_id != object_tenant_id
{
return Ok(RegistryWriteResult::InvalidTenantId);
}
// Validate tenant and account changes
if let Some(err) = write.validate_owner(&set_index) {
return Ok(err);
}
// Write to local registry
if self.0.local_objects.contains(&object_type) {
if generate_id {
@@ -194,6 +170,8 @@ impl RegistryStore {
}
// Validate foreign keys
let tenant_id = object.inner.member_tenant_id().map(|id| id.id());
let account_id = object.inner.account_id().map(|id| id.id());
for key in &set_index.keys {
match key {
IndexKey::ForeignKey {
@@ -224,7 +202,7 @@ impl RegistryStore {
return Ok(RegistryWriteResult::InvalidForeignKey {
object_id: *foreign_id,
});
} else if let Some(tenant_id) = object_tenant_id
} else if let Some(tenant_id) = tenant_id
&& (object_flags & OBJ_FILTER_TENANT) != 0
&& self
.0
@@ -234,7 +212,7 @@ impl RegistryStore {
index_id: Property::MemberTenantId.to_id(),
object_id,
item_id,
key: IndexValue::U64(tenant_id as u64).serialize(),
key: IndexValue::U64(tenant_id).serialize(),
},
)))
.await
@@ -244,7 +222,7 @@ impl RegistryStore {
return Ok(RegistryWriteResult::InvalidForeignKey {
object_id: *foreign_id,
});
} else if let Some(account_id) = write.current_account_id
} else if let Some(account_id) = account_id
&& (object_flags & OBJ_FILTER_ACCOUNT) != 0
&& self
.0
@@ -254,7 +232,7 @@ impl RegistryStore {
index_id: Property::AccountId.to_id(),
object_id,
item_id,
key: IndexValue::U64(account_id as u64).serialize(),
key: IndexValue::U64(account_id).serialize(),
},
)))
.await
@@ -334,7 +312,6 @@ impl RegistryStore {
async fn delete(
&self,
write: RegistryWrite<'_>,
object_id: ObjectId,
object: Option<&Object>,
) -> trc::Result<RegistryWriteResult> {
@@ -368,11 +345,47 @@ impl RegistryStore {
// Validate tenant and account changes
let mut clear_index = IndexBuilder::default();
object.index(&mut clear_index);
if let Some(err) = write.validate_owner(&clear_index) {
return Ok(err);
}
// Validate relationships
let linked = self.linked_objects(object_id).await?;
if !linked.is_empty() {
return Ok(RegistryWriteResult::CannotDeleteLinked {
object_id: ObjectId::new(object_type, id),
linked_objects: linked,
});
}
// Build deletion batch
let mut batch = BatchBuilder::new();
batch
.assert_value(
ValueClass::Registry(RegistryClass::Item {
object_id: object_type_id,
item_id,
}),
AssertValue::Hash(object.revision),
)
.clear(ValueClass::Registry(RegistryClass::Item {
object_id: object_type_id,
item_id,
}))
.clear(ValueClass::Registry(RegistryClass::IndexId {
object_id: object_type_id,
item_id,
}))
.registry_index(object_type_id, item_id, clear_index.keys.iter(), false);
self.0
.store
.write(batch.build_all())
.await
.map(|_| RegistryWriteResult::Success(Id::from(item_id)))
.caused_by(trc::location!())
}
pub async fn linked_objects(&self, object_id: ObjectId) -> trc::Result<Vec<ObjectId>> {
let object_type_id = object_id.object().to_id();
let item_id = object_id.id().id();
let mut linked = Vec::new();
let from_key = ValueKey::from(ValueClass::Registry(RegistryClass::Reference {
to_object_id: object_type_id,
@@ -411,41 +424,13 @@ impl RegistryStore {
},
)
.await
.caused_by(trc::location!())?;
if !linked.is_empty() {
return Ok(RegistryWriteResult::CannotDeleteLinked {
object_id: ObjectId::new(object_type, id),
linked_objects: linked,
});
}
// Build deletion batch
let mut batch = BatchBuilder::new();
batch
.assert_value(
ValueClass::Registry(RegistryClass::Item {
object_id: object_type_id,
item_id,
}),
AssertValue::Hash(object.revision),
)
.clear(ValueClass::Registry(RegistryClass::Item {
object_id: object_type_id,
item_id,
}))
.clear(ValueClass::Registry(RegistryClass::IndexId {
object_id: object_type_id,
item_id,
}))
.registry_index(object_type_id, item_id, clear_index.keys.iter(), false);
self.0
.store
.write(batch.build_all())
.await
.map(|_| RegistryWriteResult::Success(Id::from(item_id)))
.caused_by(trc::location!())
.map(|_| linked)
}
#[inline(always)]
pub fn assign_id(&self) -> u64 {
self.0.id_generator.generate()
}
}
@@ -538,117 +523,37 @@ impl SerializeInfallible for IndexValue<'_> {
impl<'x> RegistryWrite<'x> {
pub fn insert(object: &'x Object) -> Self {
Self {
op: RegistryWriteOp::Insert { object, id: None },
current_tenant_id: None,
current_account_id: None,
}
RegistryWrite::Insert { object, id: None }
}
pub fn insert_with_id(id: Id, object: &'x Object) -> Self {
Self {
op: RegistryWriteOp::Insert {
object,
id: Some(id),
},
current_tenant_id: None,
current_account_id: None,
RegistryWrite::Insert {
object,
id: Some(id),
}
}
pub fn update(id: Id, object: &'x Object, old_object: &'x Object) -> Self {
Self {
op: RegistryWriteOp::Update {
object,
id,
old_object,
},
current_tenant_id: None,
current_account_id: None,
RegistryWrite::Update {
object,
id,
old_object,
}
}
pub fn delete(object_id: ObjectId) -> Self {
Self {
op: RegistryWriteOp::Delete {
object_id,
object: None,
},
current_tenant_id: None,
current_account_id: None,
RegistryWrite::Delete {
object_id,
object: 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,
RegistryWrite::Delete {
object_id,
object: Some(object),
}
}
pub fn with_current_tenant_id(mut self, tenant_id: u32) -> Self {
self.current_tenant_id = Some(tenant_id);
self
}
pub fn with_current_account_id(mut self, account_id: u32) -> Self {
self.current_account_id = Some(account_id);
self
}
fn validate_owner(&self, builder: &IndexBuilder<'_>) -> Option<RegistryWriteResult> {
// Validate tenant and account changes
if let Some(tenant_id) = self.current_tenant_id {
for key in &builder.keys {
if let IndexKey::Search {
property: Property::MemberTenantId,
value,
} = key
&& value != &IndexValue::U64(tenant_id as u64)
{
return Some(RegistryWriteResult::InvalidTenantId);
}
}
}
if let Some(account_id) = self.current_account_id {
for key in &builder.keys {
if let IndexKey::Search {
property: Property::AccountId,
value,
} = key
&& value != &IndexValue::U64(account_id as u64)
{
return Some(RegistryWriteResult::InvalidTenantId);
}
}
}
None
}
}
trait FindTenantId {
fn tenant_id(&self) -> Option<u32>;
}
impl FindTenantId for IndexBuilder<'_> {
fn tenant_id(&self) -> Option<u32> {
self.keys.iter().find_map(|key| {
if let IndexKey::Search {
property: Property::MemberTenantId,
value: IndexValue::U64(tenant_id),
} = key
{
Some(*tenant_id as u32)
} else {
None
}
})
}
}
impl Display for RegistryWriteResult {
@@ -689,8 +594,6 @@ impl Display for RegistryWriteResult {
}
Ok(())
}
RegistryWriteResult::InvalidTenantId => write!(f, "Invalid tenant id"),
RegistryWriteResult::InvalidAccountId => write!(f, "Invalid account id"),
RegistryWriteResult::NotSupported => write!(f, "Operation not supported"),
}
}