use std::{
collections::BTreeMap,
fmt::Debug,
num::NonZeroUsize,
sync::{Arc, RwLock as StdRwLock},
time::Duration,
};
use matrix_sdk_common::{ring_buffer::RingBuffer, timer};
use ruma::{
api::client::sync::sync_events::v4::{
self, AccountDataConfig, E2EEConfig, ExtensionsConfig, ReceiptsConfig, ToDeviceConfig,
TypingConfig,
},
OwnedRoomId,
};
use tokio::sync::{broadcast::channel, Mutex as AsyncMutex, RwLock as AsyncRwLock};
use url::Url;
use super::{
cache::{format_storage_key_prefix, restore_sliding_sync_state},
sticky_parameters::SlidingSyncStickyManager,
Error, SlidingSync, SlidingSyncInner, SlidingSyncListBuilder, SlidingSyncPositionMarkers,
SlidingSyncRoom,
};
use crate::{sliding_sync::SlidingSyncStickyParameters, Client, Result};
#[derive(Debug, Clone)]
pub struct SlidingSyncBuilder {
id: String,
storage_key: String,
sliding_sync_proxy: Option<Url>,
client: Client,
lists: Vec<SlidingSyncListBuilder>,
extensions: Option<ExtensionsConfig>,
subscriptions: BTreeMap<OwnedRoomId, v4::RoomSubscription>,
rooms: BTreeMap<OwnedRoomId, SlidingSyncRoom>,
poll_timeout: Duration,
network_timeout: Duration,
#[cfg(feature = "e2e-encryption")]
share_pos: bool,
}
impl SlidingSyncBuilder {
pub(super) fn new(id: String, client: Client) -> Result<Self, Error> {
if id.len() > 16 {
Err(Error::InvalidSlidingSyncIdentifier)
} else {
let storage_key =
format_storage_key_prefix(&id, client.user_id().ok_or(Error::UnauthenticatedUser)?);
Ok(Self {
id,
storage_key,
sliding_sync_proxy: None,
client,
lists: Vec::new(),
extensions: None,
subscriptions: BTreeMap::new(),
rooms: BTreeMap::new(),
poll_timeout: Duration::from_secs(30),
network_timeout: Duration::from_secs(30),
#[cfg(feature = "e2e-encryption")]
share_pos: false,
})
}
}
pub fn sliding_sync_proxy(mut self, value: Url) -> Self {
self.sliding_sync_proxy = Some(value);
self
}
pub fn add_list(mut self, list_builder: SlidingSyncListBuilder) -> Self {
self.lists.push(list_builder);
self
}
pub async fn add_cached_list(mut self, mut list: SlidingSyncListBuilder) -> Result<Self> {
let _timer = timer!(format!("restoring (loading+processing) list {}", list.name));
let reloaded_rooms = list.set_cached_and_reload(&self.client, &self.storage_key).await?;
for (key, frozen) in reloaded_rooms {
self.rooms
.entry(key)
.or_insert_with(|| SlidingSyncRoom::from_frozen(frozen, self.client.clone()));
}
Ok(self.add_list(list))
}
pub fn with_all_extensions(mut self) -> Self {
{
let cfg = self.extensions.get_or_insert_with(Default::default);
if cfg.to_device.enabled.is_none() {
cfg.to_device.enabled = Some(true);
}
if cfg.e2ee.enabled.is_none() {
cfg.e2ee.enabled = Some(true);
}
if cfg.account_data.enabled.is_none() {
cfg.account_data.enabled = Some(true);
}
if cfg.receipts.enabled.is_none() {
cfg.receipts.enabled = Some(true);
}
if cfg.typing.enabled.is_none() {
cfg.typing.enabled = Some(true);
}
}
self
}
pub fn with_e2ee_extension(mut self, e2ee: E2EEConfig) -> Self {
self.extensions.get_or_insert_with(Default::default).e2ee = e2ee;
self
}
pub fn without_e2ee_extension(mut self) -> Self {
self.extensions.get_or_insert_with(Default::default).e2ee = E2EEConfig::default();
self
}
pub fn with_to_device_extension(mut self, to_device: ToDeviceConfig) -> Self {
self.extensions.get_or_insert_with(Default::default).to_device = to_device;
self
}
pub fn without_to_device_extension(mut self) -> Self {
self.extensions.get_or_insert_with(Default::default).to_device = ToDeviceConfig::default();
self
}
pub fn with_account_data_extension(mut self, account_data: AccountDataConfig) -> Self {
self.extensions.get_or_insert_with(Default::default).account_data = account_data;
self
}
pub fn without_account_data_extension(mut self) -> Self {
self.extensions.get_or_insert_with(Default::default).account_data =
AccountDataConfig::default();
self
}
pub fn with_typing_extension(mut self, typing: TypingConfig) -> Self {
self.extensions.get_or_insert_with(Default::default).typing = typing;
self
}
pub fn without_typing_extension(mut self) -> Self {
self.extensions.get_or_insert_with(Default::default).typing = TypingConfig::default();
self
}
pub fn with_receipt_extension(mut self, receipt: ReceiptsConfig) -> Self {
self.extensions.get_or_insert_with(Default::default).receipts = receipt;
self
}
pub fn without_receipt_extension(mut self) -> Self {
self.extensions.get_or_insert_with(Default::default).receipts = ReceiptsConfig::default();
self
}
pub fn poll_timeout(mut self, timeout: Duration) -> Self {
self.poll_timeout = timeout;
self
}
pub fn network_timeout(mut self, timeout: Duration) -> Self {
self.network_timeout = timeout;
self
}
#[cfg(feature = "e2e-encryption")]
pub fn share_pos(mut self) -> Self {
self.share_pos = true;
self
}
pub async fn build(self) -> Result<SlidingSync> {
let client = self.client;
let (internal_channel_sender, _internal_channel_receiver) = channel(8);
let mut lists = BTreeMap::new();
for list_builder in self.lists {
let list = list_builder.build(internal_channel_sender.clone());
lists.insert(list.name().to_owned(), list);
}
let restored_fields =
restore_sliding_sync_state(&client, &self.storage_key, &lists).await?;
let (delta_token, pos) = if let Some(fields) = restored_fields {
#[cfg(feature = "e2e-encryption")]
let pos = if self.share_pos { fields.pos } else { None };
#[cfg(not(feature = "e2e-encryption"))]
let pos = None;
(fields.delta_token, pos)
} else {
(None, None)
};
#[cfg(feature = "e2e-encryption")]
let share_pos = self.share_pos;
#[cfg(not(feature = "e2e-encryption"))]
let share_pos = false;
let rooms = AsyncRwLock::new(self.rooms);
let lists = AsyncRwLock::new(lists);
let sliding_sync_proxy = self.sliding_sync_proxy.or_else(|| client.sliding_sync_proxy());
Ok(SlidingSync::new(SlidingSyncInner {
id: self.id,
sliding_sync_proxy,
client,
storage_key: self.storage_key,
share_pos,
lists,
rooms,
position: Arc::new(AsyncMutex::new(SlidingSyncPositionMarkers { pos, delta_token })),
past_positions: StdRwLock::new(RingBuffer::new(NonZeroUsize::new(20).unwrap())),
sticky: StdRwLock::new(SlidingSyncStickyManager::new(
SlidingSyncStickyParameters::new(
self.subscriptions,
self.extensions.unwrap_or_default(),
),
)),
room_unsubscriptions: Default::default(),
internal_channel: internal_channel_sender,
poll_timeout: self.poll_timeout,
network_timeout: self.network_timeout,
}))
}
}