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

JMAP for Calendars implementation (part 4)

Browse Source
This commit is contained in:
mdecimus
2025-10-09 19:02:33 +02:00
parent ec4963f46b
commit b84e3e85dd
56 changed files with 2008 additions and 355 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
crates/common/src/auth/access_token.rs
View File

@@ -409,6 +409,11 @@ impl AccessToken {
self.primary_id
}
#[inline(always)]
pub fn tenant_id(&self) -> Option<u32> {
self.tenant.as_ref().map(|t| t.id)
}
pub fn secondary_ids(&self) -> impl Iterator<Item = &u32> {
self.member_of
.iter()

24
crates/common/src/auth/mod.rs
View File

@@ -158,7 +158,7 @@ impl Server {
AccountName = username.to_string(),
SpanId = req.session_id,
AccountId = principal.id(),
Type = principal.typ().as_str(),
Type = principal.typ().description(),
);
return Ok(principal);
@@ -316,25 +316,3 @@ impl CredentialsUsername for Credentials<String> {
}
}
}
pub trait AsTenantId {
fn tenant_id(&self) -> Option<u32>;
}
impl AsTenantId for Option<u32> {
fn tenant_id(&self) -> Option<u32> {
*self
}
}
impl AsTenantId for AccessToken {
fn tenant_id(&self) -> Option<u32> {
self.tenant.map(|t| t.id)
}
}
impl AsTenantId for ResourceToken {
fn tenant_id(&self) -> Option<u32> {
self.tenant.map(|t| t.id)
}
}

4
crates/common/src/config/jmap/settings.rs
View File

@@ -18,6 +18,7 @@ pub struct JmapConfig {
pub changes_max_results: Option<usize>,
pub changes_max_history: Option<usize>,
pub share_notification_max_history: Option<Duration>,
pub request_max_size: usize,
pub request_max_calls: usize,
@@ -223,6 +224,9 @@ impl JmapConfig {
changes_max_history: config
.property_or_default::<Option<usize>>("changes.max-history", "10000")
.unwrap_or_default(),
share_notification_max_history: config
.property_or_default::<Option<Duration>>("sharing.max-history", "30d")
.unwrap_or_default(),
snippet_max_results: config
.property("jmap.protocol.search-snippet.max-results")
.unwrap_or(100),

176
crates/common/src/core.rs
View File

@@ -41,6 +41,7 @@ use types::{
field::{EmailField, Field},
type_state::{DataType, StateChange},
};
use utils::snowflake::SnowflakeIdGenerator;
impl Server {
#[inline(always)]
@@ -670,106 +671,131 @@ impl Server {
Ok(assigned_ids)
}
pub async fn delete_changes(&self, account_id: u32, max_entries: usize) -> trc::Result<()> {
for sync_collection in [
SyncCollection::Email,
SyncCollection::Thread,
SyncCollection::Identity,
SyncCollection::EmailSubmission,
SyncCollection::SieveScript,
SyncCollection::FileNode,
SyncCollection::AddressBook,
SyncCollection::Calendar,
] {
let collection = sync_collection.into();
let from_key = LogKey {
account_id,
collection,
change_id: 0,
};
let to_key = LogKey {
account_id,
collection,
change_id: u64::MAX,
};
pub async fn delete_changes(
&self,
account_id: u32,
max_entries: Option<usize>,
max_duration: Option<Duration>,
) -> trc::Result<()> {
if let Some(max_entries) = max_entries {
for sync_collection in [
SyncCollection::Email,
SyncCollection::Thread,
SyncCollection::Identity,
SyncCollection::EmailSubmission,
SyncCollection::SieveScript,
SyncCollection::FileNode,
SyncCollection::AddressBook,
SyncCollection::Calendar,
] {
let collection = sync_collection.into();
let from_key = LogKey {
account_id,
collection,
change_id: 0,
};
let to_key = LogKey {
account_id,
collection,
change_id: u64::MAX,
};
let mut first_change_id = 0;
let mut num_changes = 0;
let mut first_change_id = 0;
let mut num_changes = 0;
self.store()
.iterate(
IterateParams::new(from_key, to_key)
.descending()
.no_values(),
|key, _| {
first_change_id = key.deserialize_be_u64(key.len() - U64_LEN)?;
num_changes += 1;
Ok(num_changes <= max_entries)
},
)
.await
.caused_by(trc::location!())?;
if num_changes > max_entries {
self.store()
.delete_range(
LogKey {
account_id,
collection,
change_id: 0,
},
LogKey {
account_id,
collection,
change_id: first_change_id,
.iterate(
IterateParams::new(from_key, to_key)
.descending()
.no_values(),
|key, _| {
first_change_id = key.deserialize_be_u64(key.len() - U64_LEN)?;
num_changes += 1;
Ok(num_changes <= max_entries)
},
)
.await
.caused_by(trc::location!())?;
// Delete vanished items
if let Some(vanished_collection) =
sync_collection.vanished_collection().map(u8::from)
{
if num_changes > max_entries {
self.store()
.delete_range(
LogKey {
account_id,
collection: vanished_collection,
collection,
change_id: 0,
},
LogKey {
account_id,
collection: vanished_collection,
collection,
change_id: first_change_id,
},
)
.await
.caused_by(trc::location!())?;
}
// Write truncation entry for cache
let mut batch = BatchBuilder::new();
batch.with_account_id(account_id).set(
ValueClass::Any(AnyClass {
subspace: SUBSPACE_LOGS,
key: LogKey {
account_id,
collection,
change_id: first_change_id,
}
.serialize(0),
}),
Vec::new(),
);
self.store()
.write(batch.build_all())
.await
.caused_by(trc::location!())?;
// Delete vanished items
if let Some(vanished_collection) =
sync_collection.vanished_collection().map(u8::from)
{
self.store()
.delete_range(
LogKey {
account_id,
collection: vanished_collection,
change_id: 0,
},
LogKey {
account_id,
collection: vanished_collection,
change_id: first_change_id,
},
)
.await
.caused_by(trc::location!())?;
}
// Write truncation entry for cache
let mut batch = BatchBuilder::new();
batch.with_account_id(account_id).set(
ValueClass::Any(AnyClass {
subspace: SUBSPACE_LOGS,
key: LogKey {
account_id,
collection,
change_id: first_change_id,
}
.serialize(0),
}),
Vec::new(),
);
self.store()
.write(batch.build_all())
.await
.caused_by(trc::location!())?;
}
}
}
if let Some(max_duration) = max_duration {
self.store()
.delete_range(
LogKey {
account_id,
collection: SyncCollection::ShareNotification.into(),
change_id: 0,
},
LogKey {
account_id,
collection: SyncCollection::ShareNotification.into(),
change_id: SnowflakeIdGenerator::from_duration(max_duration)
.unwrap_or_default(),
},
)
.await
.caused_by(trc::location!())?;
}
Ok(())
}

1
crates/common/src/sharing/mod.rs
View File

@@ -11,6 +11,7 @@ use utils::map::bitmap::Bitmap;
pub mod acl;
pub mod document;
pub mod notification;
pub mod resources;
pub trait EffectiveAcl {

75
crates/common/src/sharing/notification.rs Normal file
View File

@@ -0,0 +1,75 @@
/*
* SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <hello@stalw.art>
*
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
*/
use store::{Deserialize, SerializeInfallible, U32_LEN, U64_LEN, write::key::KeySerializer};
use types::{acl::Acl, collection::Collection};
use utils::map::bitmap::Bitmap;
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct ShareNotification {
pub object_account_id: u32,
pub object_id: u32,
pub object_type: Collection,
pub changed_by: u32,
pub old_rights: Bitmap<Acl>,
pub new_rights: Bitmap<Acl>,
pub name: String,
}
impl SerializeInfallible for ShareNotification {
fn serialize(&self) -> Vec<u8> {
KeySerializer::new(U64_LEN * 2 + U32_LEN * 3 + 1 + self.name.len())
.write(self.object_account_id)
.write(self.object_id)
.write(self.object_type as u8)
.write(self.changed_by)
.write(self.old_rights.bitmap)
.write(self.new_rights.bitmap)
.write(self.name.as_bytes())
.finalize()
}
}
impl Deserialize for ShareNotification {
fn deserialize(bytes: &[u8]) -> trc::Result<Self> {
Self::deserialize_from_slice(bytes)
.ok_or(trc::StoreEvent::DataCorruption.caused_by(trc::location!()))
}
}
impl ShareNotification {
fn deserialize_from_slice(bytes: &[u8]) -> Option<Self> {
Some(Self {
object_account_id: bytes
.get(..U32_LEN)
.and_then(|b| b.try_into().ok())
.map(u32::from_be_bytes)?,
object_id: bytes
.get(U32_LEN..U32_LEN * 2)
.and_then(|b| b.try_into().ok())
.map(u32::from_be_bytes)?,
object_type: bytes.get(U32_LEN * 2).copied().map(Collection::from)?,
changed_by: bytes
.get(U32_LEN * 2 + 1..U32_LEN * 3 + 1)
.and_then(|b| b.try_into().ok())
.map(u32::from_be_bytes)?,
old_rights: bytes
.get(U32_LEN * 3 + 1..U32_LEN * 3 + U64_LEN + 1)
.and_then(|b| b.try_into().ok())
.map(u64::from_be_bytes)
.map(Bitmap::from)?,
new_rights: bytes
.get(U32_LEN * 3 + U64_LEN + 1..U32_LEN * 3 + U64_LEN * 2 + 1)
.and_then(|b| b.try_into().ok())
.map(u64::from_be_bytes)
.map(Bitmap::from)?,
name: bytes
.get(U32_LEN * 3 + U64_LEN * 2 + 1..)
.and_then(|b| String::from_utf8(b.to_vec()).ok())
.unwrap_or_default(),
})
}
}

163
crates/common/src/storage/index.rs
View File

@@ -4,7 +4,7 @@
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
*/
use crate::auth::AsTenantId;
use crate::{auth::AccessToken, sharing::notification::ShareNotification};
use ahash::AHashSet;
use rkyv::{
option::ArchivedOption,
@@ -14,9 +14,18 @@ use rkyv::{
use std::{borrow::Cow, fmt::Debug};
use store::{
Serialize, SerializeInfallible,
write::{Archive, Archiver, BatchBuilder, BlobOp, DirectoryClass, IntoOperations, TagValue},
write::{
Archive, Archiver, BatchBuilder, BlobOp, DirectoryClass, IntoOperations, TagValue,
ValueClass,
},
};
use types::{acl::AclGrant, blob_hash::BlobHash, collection::SyncCollection, field::Field};
use types::{
acl::AclGrant,
blob_hash::BlobHash,
collection::{Collection, SyncCollection},
field::Field,
};
use utils::{map::bitmap::Bitmap, snowflake::SnowflakeIdGenerator};
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum IndexValue<'x> {
@@ -230,6 +239,7 @@ pub trait IndexableAndSerializableObject:
#[derive(Debug)]
pub struct ObjectIndexBuilder<C: IndexableObject, N: IndexableAndSerializableObject> {
changed_by: u32,
tenant_id: Option<u32>,
current: Option<Archive<C>>,
changes: Option<N>,
@@ -247,6 +257,7 @@ impl<C: IndexableObject, N: IndexableAndSerializableObject> ObjectIndexBuilder<C
current: None,
changes: None,
tenant_id: None,
changed_by: u32::MAX,
}
}
@@ -277,8 +288,9 @@ impl<C: IndexableObject, N: IndexableAndSerializableObject> ObjectIndexBuilder<C
self.current.as_ref()
}
pub fn with_tenant_id(mut self, tenant: &impl AsTenantId) -> Self {
self.tenant_id = tenant.tenant_id();
pub fn with_access_token(mut self, access_token: &AccessToken) -> Self {
self.tenant_id = access_token.tenant.as_ref().map(|t| t.id);
self.changed_by = access_token.primary_id();
self
}
}
@@ -291,7 +303,7 @@ impl<C: IndexableObject, N: IndexableAndSerializableObject> IntoOperations
(None, Some(changes)) => {
// Insertion
for item in changes.index_values() {
build_index(batch, item, self.tenant_id, true);
build_index(batch, item, self.changed_by, self.tenant_id, true);
}
if N::is_versioned() {
let (offset, bytes) = Archiver::new(changes).serialize_versioned()?;
@@ -305,7 +317,7 @@ impl<C: IndexableObject, N: IndexableAndSerializableObject> IntoOperations
batch.assert_value(Field::ARCHIVE, &current);
for (current, change) in current.inner.index_values().zip(changes.index_values()) {
if current != change {
merge_index(batch, current, change, self.tenant_id)?;
merge_index(batch, current, change, self.changed_by, self.tenant_id)?;
} else {
match current {
IndexValue::LogContainer { sync_collection } => {
@@ -332,7 +344,7 @@ impl<C: IndexableObject, N: IndexableAndSerializableObject> IntoOperations
// Deletion
batch.assert_value(Field::ARCHIVE, &current);
for item in current.inner.index_values() {
build_index(batch, item, self.tenant_id, false);
build_index(batch, item, self.changed_by, self.tenant_id, false);
}
batch.clear(Field::ARCHIVE);
@@ -344,7 +356,13 @@ impl<C: IndexableObject, N: IndexableAndSerializableObject> IntoOperations
}
}
fn build_index(batch: &mut BatchBuilder, item: IndexValue<'_>, tenant_id: Option<u32>, set: bool) {
fn build_index(
batch: &mut BatchBuilder,
item: IndexValue<'_>,
changed_by: u32,
tenant_id: Option<u32>,
set: bool,
) {
match item {
IndexValue::Index { field, value } => {
if !value.is_empty() {
@@ -381,11 +399,52 @@ fn build_index(batch: &mut BatchBuilder, item: IndexValue<'_>, tenant_id: Option
}
}
IndexValue::Acl { value } => {
let object_account_id = batch.last_account_id().unwrap_or_default();
let object_type = batch.last_collection().unwrap_or(Collection::None);
let object_id = batch.last_document_id().unwrap_or_default();
let notification_id = SnowflakeIdGenerator::from_sequence_and_node_id(
object_type as u64 ^ object_account_id as u64,
None,
)
.unwrap_or_default();
for item in value.as_ref() {
if set {
batch.acl_grant(item.account_id, item.grants.bitmap.serialize());
batch.set(
ValueClass::ShareNotification {
notification_id,
notify_account_id: item.account_id,
},
ShareNotification {
object_account_id,
object_id,
object_type,
changed_by,
old_rights: Default::default(),
new_rights: item.grants,
name: Default::default(),
}
.serialize(),
);
} else {
batch.acl_revoke(item.account_id);
batch.set(
ValueClass::ShareNotification {
notification_id,
notify_account_id: item.account_id,
},
ShareNotification {
object_account_id,
object_id,
object_type,
changed_by,
old_rights: item.grants,
new_rights: Default::default(),
name: Default::default(),
}
.serialize(),
);
}
}
}
@@ -432,6 +491,7 @@ fn merge_index(
batch: &mut BatchBuilder,
current: IndexValue<'_>,
change: IndexValue<'_>,
changed_by: u32,
tenant_id: Option<u32>,
) -> trc::Result<()> {
match (current, change) {
@@ -499,7 +559,23 @@ fn merge_index(
batch.set(BlobOp::Link { hash: new_hash }, vec![]);
}
(IndexValue::Acl { value: old_acl }, IndexValue::Acl { value: new_acl }) => {
match (!old_acl.is_empty(), !new_acl.is_empty()) {
let has_old_acl = !old_acl.is_empty();
let has_new_acl = !new_acl.is_empty();
if !has_old_acl && !has_new_acl {
return Ok(());
}
let object_account_id = batch.last_account_id().unwrap_or_default();
let object_type = batch.last_collection().unwrap_or(Collection::None);
let object_id = batch.last_document_id().unwrap_or_default();
let notification_id = SnowflakeIdGenerator::from_sequence_and_node_id(
object_type as u64 ^ object_account_id as u64,
None,
)
.unwrap_or_default();
match (has_old_acl, has_new_acl) {
(true, true) => {
// Remove deleted ACLs
for current_item in old_acl.as_ref() {
@@ -508,22 +584,57 @@ fn merge_index(
.any(|item| item.account_id == current_item.account_id)
{
batch.acl_revoke(current_item.account_id);
batch.set(
ValueClass::ShareNotification {
notification_id,
notify_account_id: current_item.account_id,
},
ShareNotification {
object_account_id,
object_id,
object_type,
changed_by,
old_rights: current_item.grants,
new_rights: Default::default(),
name: Default::default(),
}
.serialize(),
);
}
}
// Update ACLs
for item in new_acl.as_ref() {
let mut add_item = true;
let mut old_rights = Bitmap::default();
for current_item in old_acl.as_ref() {
if item.account_id == current_item.account_id {
if item.grants == current_item.grants {
add_item = false;
} else {
old_rights = current_item.grants;
}
break;
}
}
if add_item {
batch.acl_grant(item.account_id, item.grants.bitmap.serialize());
batch.set(
ValueClass::ShareNotification {
notification_id,
notify_account_id: item.account_id,
},
ShareNotification {
object_account_id,
object_id,
object_type,
changed_by,
old_rights,
new_rights: item.grants,
name: Default::default(),
}
.serialize(),
);
}
}
}
@@ -531,12 +642,44 @@ fn merge_index(
// Add all ACLs
for item in new_acl.as_ref() {
batch.acl_grant(item.account_id, item.grants.bitmap.serialize());
batch.set(
ValueClass::ShareNotification {
notification_id,
notify_account_id: item.account_id,
},
ShareNotification {
object_account_id,
object_id,
object_type,
changed_by,
old_rights: Default::default(),
new_rights: item.grants,
name: Default::default(),
}
.serialize(),
);
}
}
(true, false) => {
// Remove all ACLs
for item in old_acl.as_ref() {
batch.acl_revoke(item.account_id);
batch.set(
ValueClass::ShareNotification {
notification_id,
notify_account_id: item.account_id,
},
ShareNotification {
object_account_id,
object_id,
object_type,
changed_by,
old_rights: item.grants,
new_rights: Default::default(),
name: Default::default(),
}
.serialize(),
);
}
}
_ => {}