575 lines
19 KiB
Rust
575 lines
19 KiB
Rust
/*
|
|
* SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <hello@stalw.art>
|
|
*
|
|
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
|
|
*/
|
|
|
|
use crate::{
|
|
cache::calcard::{build_scheduling_resources, path_from_scheduling, resource_from_scheduling},
|
|
calendar::{Calendar, CalendarEvent, CalendarPreferences},
|
|
contact::{AddressBook, ContactCard},
|
|
file::FileNode,
|
|
};
|
|
use ahash::AHashSet;
|
|
use calcard::{
|
|
build_calcard_resources, build_simple_hierarchy, resource_from_addressbook,
|
|
resource_from_calendar, resource_from_card, resource_from_event,
|
|
};
|
|
use common::{CacheSwap, DavResource, DavResources, Server, auth::AccessToken};
|
|
use file::{build_file_resources, build_nested_hierarchy, resource_from_file};
|
|
use jmap_proto::types::collection::{Collection, SyncCollection};
|
|
use std::{sync::Arc, time::Instant};
|
|
use store::{
|
|
ahash::AHashMap,
|
|
query::log::{Change, Query},
|
|
write::{AlignedBytes, Archive, BatchBuilder},
|
|
};
|
|
use tokio::sync::Semaphore;
|
|
use trc::{AddContext, StoreEvent};
|
|
|
|
pub mod calcard;
|
|
pub mod file;
|
|
|
|
pub trait GroupwareCache: Sync + Send {
|
|
fn fetch_dav_resources(
|
|
&self,
|
|
access_token: &AccessToken,
|
|
account_id: u32,
|
|
collection: SyncCollection,
|
|
) -> impl Future<Output = trc::Result<Arc<DavResources>>> + Send;
|
|
|
|
fn create_default_addressbook(
|
|
&self,
|
|
access_token: &AccessToken,
|
|
account_id: u32,
|
|
account_name: &str,
|
|
) -> impl Future<Output = trc::Result<Option<u32>>> + Send;
|
|
|
|
fn create_default_calendar(
|
|
&self,
|
|
access_token: &AccessToken,
|
|
account_id: u32,
|
|
account_name: &str,
|
|
) -> impl Future<Output = trc::Result<Option<u32>>> + Send;
|
|
|
|
fn get_or_create_default_calendar(
|
|
&self,
|
|
access_token: &AccessToken,
|
|
account_id: u32,
|
|
account_name: &str,
|
|
) -> impl Future<Output = trc::Result<Option<u32>>> + Send;
|
|
|
|
fn cached_dav_resources(
|
|
&self,
|
|
account_id: u32,
|
|
collection: SyncCollection,
|
|
) -> Option<Arc<DavResources>>;
|
|
}
|
|
|
|
impl GroupwareCache for Server {
|
|
async fn fetch_dav_resources(
|
|
&self,
|
|
access_token: &AccessToken,
|
|
account_id: u32,
|
|
collection: SyncCollection,
|
|
) -> trc::Result<Arc<DavResources>> {
|
|
let cache_store = match collection {
|
|
SyncCollection::Calendar => &self.inner.cache.events,
|
|
SyncCollection::AddressBook => &self.inner.cache.contacts,
|
|
SyncCollection::FileNode => &self.inner.cache.files,
|
|
SyncCollection::CalendarScheduling => &self.inner.cache.scheduling,
|
|
_ => unreachable!(),
|
|
};
|
|
let cache_ = match cache_store.get_value_or_guard_async(&account_id).await {
|
|
Ok(cache) => cache,
|
|
Err(guard) => {
|
|
let start_time = Instant::now();
|
|
let cache = full_cache_build(
|
|
self,
|
|
account_id,
|
|
collection,
|
|
Arc::new(Semaphore::new(1)),
|
|
access_token,
|
|
)
|
|
.await?;
|
|
|
|
if guard.insert(CacheSwap::new(cache.clone())).is_err() {
|
|
cache_store.insert(account_id, CacheSwap::new(cache.clone()));
|
|
}
|
|
|
|
trc::event!(
|
|
Store(StoreEvent::CacheMiss),
|
|
AccountId = account_id,
|
|
Collection = collection.as_str(),
|
|
Total = cache.resources.len(),
|
|
ChangeId = cache.highest_change_id,
|
|
Elapsed = start_time.elapsed(),
|
|
);
|
|
|
|
return Ok(cache);
|
|
}
|
|
};
|
|
|
|
// Obtain current state
|
|
let cache = cache_.load_full();
|
|
let start_time = Instant::now();
|
|
let changes = self
|
|
.core
|
|
.storage
|
|
.data
|
|
.changes(
|
|
account_id,
|
|
collection,
|
|
Query::Since(cache.highest_change_id),
|
|
)
|
|
.await
|
|
.caused_by(trc::location!())?;
|
|
|
|
// Regenerate cache if the change log has been truncated
|
|
if changes.is_truncated {
|
|
let cache = full_cache_build(
|
|
self,
|
|
account_id,
|
|
collection,
|
|
cache.update_lock.clone(),
|
|
access_token,
|
|
)
|
|
.await?;
|
|
cache_.update(cache.clone());
|
|
|
|
trc::event!(
|
|
Store(StoreEvent::CacheStale),
|
|
AccountId = account_id,
|
|
Collection = collection.as_str(),
|
|
ChangeId = cache.highest_change_id,
|
|
Total = cache.resources.len(),
|
|
Elapsed = start_time.elapsed(),
|
|
);
|
|
|
|
return Ok(cache);
|
|
}
|
|
|
|
// Verify changes
|
|
if changes.changes.is_empty() {
|
|
trc::event!(
|
|
Store(StoreEvent::CacheHit),
|
|
AccountId = account_id,
|
|
Collection = collection.as_str(),
|
|
ChangeId = cache.highest_change_id,
|
|
Elapsed = start_time.elapsed(),
|
|
);
|
|
|
|
return Ok(cache);
|
|
}
|
|
|
|
// Lock for updates
|
|
let _permit = cache.update_lock.acquire().await;
|
|
let cache = cache_.load_full();
|
|
if cache.highest_change_id >= changes.to_change_id {
|
|
trc::event!(
|
|
Store(StoreEvent::CacheHit),
|
|
AccountId = account_id,
|
|
Collection = collection.as_str(),
|
|
ChangeId = cache.highest_change_id,
|
|
Elapsed = start_time.elapsed(),
|
|
);
|
|
|
|
return Ok(cache);
|
|
}
|
|
|
|
let num_changes = changes.changes.len();
|
|
let cache = if !matches!(collection, SyncCollection::CalendarScheduling) {
|
|
let mut updated_resources = AHashMap::with_capacity(8);
|
|
let has_no_children = collection == SyncCollection::FileNode;
|
|
|
|
process_changes(
|
|
self,
|
|
account_id,
|
|
collection,
|
|
has_no_children,
|
|
&mut updated_resources,
|
|
changes.changes,
|
|
)
|
|
.await?;
|
|
|
|
let mut rebuild_hierarchy = false;
|
|
let mut resources = Vec::with_capacity(cache.resources.len());
|
|
|
|
for resource in &cache.resources {
|
|
let is_container = has_no_children || resource.is_container();
|
|
if let Some(updated_resource) =
|
|
updated_resources.remove(&(is_container, resource.document_id))
|
|
{
|
|
if let Some(updated_resource) = updated_resource {
|
|
rebuild_hierarchy =
|
|
rebuild_hierarchy || updated_resource.has_hierarchy_changes(resource);
|
|
resources.push(updated_resource);
|
|
} else {
|
|
// Deleted resource
|
|
rebuild_hierarchy = true;
|
|
}
|
|
} else {
|
|
resources.push(resource.clone());
|
|
}
|
|
}
|
|
|
|
// Add new resources
|
|
for resource in updated_resources.into_values().flatten() {
|
|
resources.push(resource);
|
|
rebuild_hierarchy = true;
|
|
}
|
|
|
|
if rebuild_hierarchy {
|
|
let mut cache = DavResources {
|
|
base_path: cache.base_path.clone(),
|
|
paths: Default::default(),
|
|
resources,
|
|
item_change_id: changes.item_change_id.unwrap_or(cache.item_change_id),
|
|
container_change_id: changes
|
|
.container_change_id
|
|
.unwrap_or(cache.container_change_id),
|
|
highest_change_id: changes.to_change_id,
|
|
size: std::mem::size_of::<DavResources>() as u64,
|
|
update_lock: cache.update_lock.clone(),
|
|
};
|
|
|
|
if matches!(collection, SyncCollection::FileNode) {
|
|
build_nested_hierarchy(&mut cache);
|
|
} else {
|
|
build_simple_hierarchy(&mut cache);
|
|
}
|
|
cache
|
|
} else {
|
|
DavResources {
|
|
base_path: cache.base_path.clone(),
|
|
paths: cache.paths.clone(),
|
|
resources,
|
|
item_change_id: changes.item_change_id.unwrap_or(cache.item_change_id),
|
|
container_change_id: changes
|
|
.container_change_id
|
|
.unwrap_or(cache.container_change_id),
|
|
highest_change_id: changes.to_change_id,
|
|
size: cache.size,
|
|
update_lock: cache.update_lock.clone(),
|
|
}
|
|
}
|
|
} else {
|
|
let mut delete_ids = AHashSet::with_capacity(changes.changes.len());
|
|
let mut resources = Vec::with_capacity(cache.resources.len());
|
|
let mut paths = AHashSet::with_capacity(cache.paths.len());
|
|
|
|
for change in changes.changes {
|
|
match change {
|
|
Change::InsertItem(document_id) => {
|
|
let document_id = document_id as u32;
|
|
paths.insert(path_from_scheduling(document_id, resources.len(), false));
|
|
resources.push(resource_from_scheduling(document_id, false));
|
|
}
|
|
Change::DeleteItem(document_id) => {
|
|
delete_ids.insert(document_id as u32);
|
|
}
|
|
_ => {}
|
|
}
|
|
}
|
|
|
|
for resource in &cache.resources {
|
|
if !delete_ids.contains(&resource.document_id) {
|
|
paths.insert(path_from_scheduling(
|
|
resource.document_id,
|
|
resources.len(),
|
|
resource.is_container(),
|
|
));
|
|
resources.push(resource.clone());
|
|
}
|
|
}
|
|
|
|
DavResources {
|
|
base_path: cache.base_path.clone(),
|
|
paths,
|
|
resources,
|
|
item_change_id: changes.item_change_id.unwrap_or(cache.item_change_id),
|
|
container_change_id: changes
|
|
.container_change_id
|
|
.unwrap_or(cache.container_change_id),
|
|
highest_change_id: changes.to_change_id,
|
|
size: cache.size,
|
|
update_lock: cache.update_lock.clone(),
|
|
}
|
|
};
|
|
|
|
let cache = Arc::new(cache);
|
|
cache_.update(cache.clone());
|
|
|
|
trc::event!(
|
|
Store(StoreEvent::CacheUpdate),
|
|
AccountId = account_id,
|
|
Collection = collection.as_str(),
|
|
ChangeId = cache.highest_change_id,
|
|
Details = num_changes,
|
|
Total = cache.resources.len(),
|
|
Elapsed = start_time.elapsed(),
|
|
);
|
|
|
|
Ok(cache)
|
|
}
|
|
|
|
async fn create_default_addressbook(
|
|
&self,
|
|
access_token: &AccessToken,
|
|
account_id: u32,
|
|
account_name: &str,
|
|
) -> trc::Result<Option<u32>> {
|
|
if let Some(name) = &self.core.groupware.default_addressbook_name {
|
|
let mut batch = BatchBuilder::new();
|
|
let document_id = self
|
|
.store()
|
|
.assign_document_ids(account_id, Collection::AddressBook, 1)
|
|
.await?;
|
|
AddressBook {
|
|
name: name.clone(),
|
|
display_name: format!(
|
|
"{} ({})",
|
|
self.core
|
|
.groupware
|
|
.default_addressbook_display_name
|
|
.as_ref()
|
|
.unwrap_or(name),
|
|
account_name
|
|
)
|
|
.into(),
|
|
is_default: true,
|
|
..Default::default()
|
|
}
|
|
.insert(access_token, account_id, document_id, &mut batch)?;
|
|
self.commit_batch(batch).await?;
|
|
Ok(Some(document_id))
|
|
} else {
|
|
Ok(None)
|
|
}
|
|
}
|
|
|
|
async fn create_default_calendar(
|
|
&self,
|
|
access_token: &AccessToken,
|
|
account_id: u32,
|
|
account_name: &str,
|
|
) -> trc::Result<Option<u32>> {
|
|
if let Some(name) = &self.core.groupware.default_calendar_name {
|
|
let mut batch = BatchBuilder::new();
|
|
let document_id = self
|
|
.store()
|
|
.assign_document_ids(account_id, Collection::Calendar, 1)
|
|
.await?;
|
|
Calendar {
|
|
name: name.clone(),
|
|
preferences: vec![CalendarPreferences {
|
|
account_id,
|
|
name: format!(
|
|
"{} ({})",
|
|
self.core
|
|
.groupware
|
|
.default_calendar_display_name
|
|
.as_ref()
|
|
.unwrap_or(name),
|
|
account_name
|
|
),
|
|
..Default::default()
|
|
}],
|
|
..Default::default()
|
|
}
|
|
.insert(access_token, account_id, document_id, &mut batch)?;
|
|
self.commit_batch(batch).await?;
|
|
Ok(Some(document_id))
|
|
} else {
|
|
Ok(None)
|
|
}
|
|
}
|
|
|
|
async fn get_or_create_default_calendar(
|
|
&self,
|
|
access_token: &AccessToken,
|
|
account_id: u32,
|
|
account_name: &str,
|
|
) -> trc::Result<Option<u32>> {
|
|
match self
|
|
.get_document_ids(account_id, Collection::Calendar)
|
|
.await
|
|
{
|
|
Ok(Some(ids)) if !ids.is_empty() => Ok(ids.iter().next()),
|
|
_ => {
|
|
self.create_default_calendar(access_token, account_id, account_name)
|
|
.await
|
|
}
|
|
}
|
|
}
|
|
|
|
fn cached_dav_resources(
|
|
&self,
|
|
account_id: u32,
|
|
collection: SyncCollection,
|
|
) -> Option<Arc<DavResources>> {
|
|
(match collection {
|
|
SyncCollection::Calendar => &self.inner.cache.events,
|
|
SyncCollection::AddressBook => &self.inner.cache.contacts,
|
|
SyncCollection::FileNode => &self.inner.cache.files,
|
|
_ => unreachable!(),
|
|
})
|
|
.get(&account_id)
|
|
.map(|cache| cache.load_full())
|
|
}
|
|
}
|
|
|
|
async fn process_changes(
|
|
server: &Server,
|
|
account_id: u32,
|
|
collection: SyncCollection,
|
|
has_no_children: bool,
|
|
updated_resources: &mut AHashMap<(bool, u32), Option<DavResource>>,
|
|
changes: Vec<Change>,
|
|
) -> trc::Result<()> {
|
|
for change in changes {
|
|
match change {
|
|
Change::InsertItem(id) | Change::UpdateItem(id) => {
|
|
let document_id = id as u32;
|
|
if let Some(archive) = server
|
|
.get_archive(account_id, collection.collection(false), document_id)
|
|
.await
|
|
.caused_by(trc::location!())?
|
|
{
|
|
updated_resources.insert(
|
|
(has_no_children, document_id),
|
|
Some(resource_from_archive(
|
|
archive,
|
|
document_id,
|
|
collection,
|
|
false,
|
|
)?),
|
|
);
|
|
} else {
|
|
updated_resources.insert((has_no_children, document_id), None);
|
|
}
|
|
}
|
|
Change::DeleteItem(id) => {
|
|
updated_resources.insert((has_no_children, id as u32), None);
|
|
}
|
|
Change::InsertContainer(id) | Change::UpdateContainer(id) => {
|
|
let document_id = id as u32;
|
|
if let Some(archive) = server
|
|
.get_archive(account_id, collection.collection(true), document_id)
|
|
.await
|
|
.caused_by(trc::location!())?
|
|
{
|
|
updated_resources.insert(
|
|
(true, document_id),
|
|
Some(resource_from_archive(
|
|
archive,
|
|
document_id,
|
|
collection,
|
|
true,
|
|
)?),
|
|
);
|
|
} else {
|
|
updated_resources.insert((true, document_id), None);
|
|
}
|
|
}
|
|
Change::DeleteContainer(id) => {
|
|
updated_resources.insert((true, id as u32), None);
|
|
}
|
|
Change::UpdateContainerProperty(_) => (),
|
|
}
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
async fn full_cache_build(
|
|
server: &Server,
|
|
account_id: u32,
|
|
collection: SyncCollection,
|
|
update_lock: Arc<Semaphore>,
|
|
access_token: &AccessToken,
|
|
) -> trc::Result<Arc<DavResources>> {
|
|
match collection {
|
|
SyncCollection::Calendar => {
|
|
build_calcard_resources(
|
|
server,
|
|
access_token,
|
|
account_id,
|
|
SyncCollection::Calendar,
|
|
Collection::Calendar,
|
|
Collection::CalendarEvent,
|
|
update_lock,
|
|
)
|
|
.await
|
|
}
|
|
SyncCollection::AddressBook => {
|
|
build_calcard_resources(
|
|
server,
|
|
access_token,
|
|
account_id,
|
|
SyncCollection::AddressBook,
|
|
Collection::AddressBook,
|
|
Collection::ContactCard,
|
|
update_lock,
|
|
)
|
|
.await
|
|
}
|
|
SyncCollection::FileNode => build_file_resources(server, account_id, update_lock).await,
|
|
SyncCollection::CalendarScheduling => {
|
|
build_scheduling_resources(server, account_id, update_lock).await
|
|
}
|
|
_ => unreachable!(),
|
|
}
|
|
.map(Arc::new)
|
|
}
|
|
|
|
fn resource_from_archive(
|
|
archive: Archive<AlignedBytes>,
|
|
document_id: u32,
|
|
collection: SyncCollection,
|
|
is_container: bool,
|
|
) -> trc::Result<DavResource> {
|
|
Ok(match collection {
|
|
SyncCollection::Calendar => {
|
|
if is_container {
|
|
resource_from_calendar(
|
|
archive
|
|
.unarchive::<Calendar>()
|
|
.caused_by(trc::location!())?,
|
|
document_id,
|
|
)
|
|
} else {
|
|
resource_from_event(
|
|
archive
|
|
.unarchive::<CalendarEvent>()
|
|
.caused_by(trc::location!())?,
|
|
document_id,
|
|
)
|
|
}
|
|
}
|
|
SyncCollection::AddressBook => {
|
|
if is_container {
|
|
resource_from_addressbook(
|
|
archive
|
|
.unarchive::<AddressBook>()
|
|
.caused_by(trc::location!())?,
|
|
document_id,
|
|
)
|
|
} else {
|
|
resource_from_card(
|
|
archive
|
|
.unarchive::<ContactCard>()
|
|
.caused_by(trc::location!())?,
|
|
document_id,
|
|
)
|
|
}
|
|
}
|
|
SyncCollection::FileNode => resource_from_file(
|
|
archive
|
|
.unarchive::<FileNode>()
|
|
.caused_by(trc::location!())?,
|
|
document_id,
|
|
),
|
|
_ => unreachable!(),
|
|
})
|
|
}
|