1#![cfg_attr(not(test), allow(unused))]
16
17use std::{collections::HashMap, rc::Rc, time::Duration};
18
19use indexed_db_futures::{Build, database::Database};
20#[cfg(target_family = "wasm")]
21use matrix_sdk_base::{
22 cross_process_lock::{CrossProcessLockGeneration, FIRST_CROSS_PROCESS_LOCK_GENERATION},
23 event_cache::thread::ThreadInfo,
24};
25use matrix_sdk_base::{
26 event_cache::{Event, Gap, store::EventCacheStore},
27 linked_chunk::{
28 ChunkIdentifier, ChunkIdentifierGenerator, ChunkMetadata, LinkedChunkId, Position,
29 RawChunk, Update,
30 },
31 timer,
32};
33use ruma::{
34 EventId, MilliSecondsSinceUnixEpoch, OwnedEventId, RoomId, events::relation::RelationType,
35};
36use tracing::{error, instrument, trace};
37use web_sys::IdbTransactionMode;
38
39use crate::{
40 event_cache_store::{
41 migrations::current::keys,
42 transaction::IndexeddbEventCacheStoreTransaction,
43 types::{ChunkType, InBandEvent, Lease, OutOfBandEvent, Thread},
44 },
45 serializer::indexed_type::{IndexedTypeSerializer, traits::Indexed},
46 transaction::TransactionError,
47};
48
49mod builder;
50mod error;
51#[cfg(all(test, target_family = "wasm"))]
52mod integration_tests;
53mod migrations;
54mod serializer;
55mod transaction;
56mod types;
57
58pub use builder::IndexeddbEventCacheStoreBuilder;
59pub use error::IndexeddbEventCacheStoreError;
60
61#[derive(Debug, Clone)]
67pub struct IndexeddbEventCacheStore {
68 inner: Rc<Database>,
70 serializer: IndexedTypeSerializer,
72}
73
74impl IndexeddbEventCacheStore {
75 pub fn builder() -> IndexeddbEventCacheStoreBuilder {
78 IndexeddbEventCacheStoreBuilder::default()
79 }
80
81 pub fn transaction<'a>(
85 &'a self,
86 stores: &[&str],
87 mode: IdbTransactionMode,
88 ) -> Result<IndexeddbEventCacheStoreTransaction<'a>, IndexeddbEventCacheStoreError> {
89 Ok(IndexeddbEventCacheStoreTransaction::new(
90 self.inner
91 .transaction(stores)
92 .with_mode(mode)
93 .build()
94 .map_err(TransactionError::from)?,
95 &self.serializer,
96 ))
97 }
98}
99
100#[cfg(target_family = "wasm")]
101#[async_trait::async_trait(?Send)]
102impl EventCacheStore for IndexeddbEventCacheStore {
103 type Error = IndexeddbEventCacheStoreError;
104
105 #[instrument(skip(self))]
106 async fn try_take_leased_lock(
107 &self,
108 lease_duration_ms: u32,
109 key: &str,
110 holder: &str,
111 ) -> Result<Option<CrossProcessLockGeneration>, IndexeddbEventCacheStoreError> {
112 let transaction =
113 self.transaction(&[Lease::OBJECT_STORE], IdbTransactionMode::Readwrite)?;
114
115 let now = Duration::from_millis(MilliSecondsSinceUnixEpoch::now().get().into());
116 let expiration = now + Duration::from_millis(lease_duration_ms.into());
117
118 let lease = match transaction.get_lease_by_id(key).await? {
119 Some(mut lease) => {
120 if lease.holder == holder {
121 lease.expiration = expiration;
123
124 Some(lease)
125 } else {
126 if lease.expiration < now {
128 lease.holder = holder.to_owned();
130 lease.expiration = expiration;
131 lease.generation += 1;
132
133 Some(lease)
134 } else {
135 None
137 }
138 }
139 }
140 None => {
141 let lease = Lease {
142 key: key.to_owned(),
143 holder: holder.to_owned(),
144 expiration,
145 generation: FIRST_CROSS_PROCESS_LOCK_GENERATION,
146 };
147
148 Some(lease)
149 }
150 };
151
152 Ok(if let Some(lease) = lease {
153 transaction.put_lease(&lease).await?;
154 transaction.commit().await?;
155
156 Some(lease.generation)
157 } else {
158 None
159 })
160 }
161
162 #[instrument(skip(self, updates))]
163 async fn handle_linked_chunk_updates(
164 &self,
165 linked_chunk_id: LinkedChunkId<'_>,
166 updates: Vec<Update<Event, Gap>>,
167 ) -> Result<(), IndexeddbEventCacheStoreError> {
168 let _timer = timer!("method");
169
170 let transaction = self.transaction(
171 &[keys::LINKED_CHUNKS, keys::GAPS, keys::EVENTS],
172 IdbTransactionMode::Readwrite,
173 )?;
174
175 for update in updates {
176 match update {
177 Update::NewItemsChunk { previous, new, next } => {
178 trace!(%linked_chunk_id, "Inserting new chunk (prev={previous:?}, new={new:?}, next={next:?})");
179 transaction
180 .add_chunk(&types::Chunk {
181 linked_chunk_id: linked_chunk_id.to_owned(),
182 identifier: new.index(),
183 previous: previous.map(|i| i.index()),
184 next: next.map(|i| i.index()),
185 chunk_type: ChunkType::Event,
186 })
187 .await?;
188 }
189 Update::NewGapChunk { previous, new, next, gap } => {
190 trace!(%linked_chunk_id, "Inserting new gap (prev={previous:?}, new={new:?}, next={next:?})");
191 transaction
192 .add_item(&types::Gap {
193 linked_chunk_id: linked_chunk_id.to_owned(),
194 chunk_identifier: new.index(),
195 token: gap.token,
196 })
197 .await?;
198 transaction
199 .add_chunk(&types::Chunk {
200 linked_chunk_id: linked_chunk_id.to_owned(),
201 identifier: new.index(),
202 previous: previous.map(|i| i.index()),
203 next: next.map(|i| i.index()),
204 chunk_type: ChunkType::Gap,
205 })
206 .await?;
207 }
208 Update::RemoveChunk(chunk_id) => {
209 trace!(%linked_chunk_id, "Removing chunk {chunk_id:?}");
210 transaction.delete_chunk_by_id(linked_chunk_id, chunk_id).await?;
211 }
212 Update::PushItems { at, items } => {
213 let chunk_identifier = at.chunk_identifier().index();
214
215 trace!(%linked_chunk_id, "pushing {} items @ {chunk_identifier}", items.len());
216
217 for (i, item) in items.into_iter().enumerate() {
218 transaction
219 .add_event(&types::Event::InBand(InBandEvent {
220 linked_chunk_id: linked_chunk_id.to_owned(),
221 content: item,
222 position: types::Position {
223 chunk_identifier,
224 index: at.index() + i,
225 },
226 }))
227 .await?;
228 }
229 }
230 Update::ReplaceItem { at, item } => {
231 let chunk_id = at.chunk_identifier().index();
232 let index = at.index();
233
234 trace!(%linked_chunk_id, "replacing item @ {chunk_id}:{index}");
235
236 transaction
237 .put_event(&types::Event::InBand(InBandEvent {
238 linked_chunk_id: linked_chunk_id.to_owned(),
239 content: item,
240 position: at.into(),
241 }))
242 .await?;
243 }
244 Update::RemoveItem { at } => {
245 let chunk_id = at.chunk_identifier().index();
246 let index = at.index();
247
248 trace!(%linked_chunk_id, "removing item @ {chunk_id}:{index}");
249
250 transaction.delete_event_by_position(linked_chunk_id, at.into()).await?;
251 }
252 Update::DetachLastItems { at } => {
253 let chunk_id = at.chunk_identifier().index();
254 let index = at.index();
255
256 trace!(%linked_chunk_id, "detaching last items @ {chunk_id}:{index}");
257
258 transaction
259 .delete_events_by_chunk_from_index(linked_chunk_id, at.into())
260 .await?;
261 }
262 Update::StartReattachItems | Update::EndReattachItems => {
263 }
265 Update::Clear => {
266 trace!(%linked_chunk_id, "clearing room");
267 transaction.delete_chunks_by_linked_chunk_id(linked_chunk_id).await?;
268 transaction.delete_events_by_linked_chunk_id(linked_chunk_id).await?;
269 transaction.delete_gaps_by_linked_chunk_id(linked_chunk_id).await?;
270 }
271 }
272 }
273 transaction.commit().await?;
274 Ok(())
275 }
276
277 #[instrument(skip(self))]
278 async fn load_all_chunks(
279 &self,
280 linked_chunk_id: LinkedChunkId<'_>,
281 ) -> Result<Vec<RawChunk<Event, Gap>>, IndexeddbEventCacheStoreError> {
282 let _ = timer!("method");
283
284 let transaction = self.transaction(
285 &[keys::LINKED_CHUNKS, keys::GAPS, keys::EVENTS],
286 IdbTransactionMode::Readonly,
287 )?;
288
289 let mut raw_chunks = Vec::new();
290 let chunks = transaction.get_chunks_by_linked_chunk_id(linked_chunk_id).await?;
291 for chunk in chunks {
292 if let Some(raw_chunk) = transaction
293 .load_chunk_by_id(linked_chunk_id, ChunkIdentifier::new(chunk.identifier))
294 .await?
295 {
296 raw_chunks.push(raw_chunk);
297 }
298 }
299 Ok(raw_chunks)
300 }
301
302 #[instrument(skip(self))]
303 async fn load_all_chunks_metadata(
304 &self,
305 linked_chunk_id: LinkedChunkId<'_>,
306 ) -> Result<Vec<ChunkMetadata>, IndexeddbEventCacheStoreError> {
307 let _ = timer!("method");
318
319 let transaction = self.transaction(
320 &[keys::LINKED_CHUNKS, keys::EVENTS, keys::GAPS],
321 IdbTransactionMode::Readonly,
322 )?;
323
324 let mut raw_chunks = Vec::new();
325 let chunks = transaction.get_chunks_by_linked_chunk_id(linked_chunk_id).await?;
326 for chunk in chunks {
327 let chunk_id = ChunkIdentifier::new(chunk.identifier);
328 let num_items =
329 transaction.get_events_count_by_chunk(linked_chunk_id, chunk_id).await?;
330 raw_chunks.push(ChunkMetadata {
331 num_items,
332 previous: chunk.previous.map(ChunkIdentifier::new),
333 identifier: ChunkIdentifier::new(chunk.identifier),
334 next: chunk.next.map(ChunkIdentifier::new),
335 });
336 }
337 Ok(raw_chunks)
338 }
339
340 #[instrument(skip(self))]
341 async fn load_last_chunk(
342 &self,
343 linked_chunk_id: LinkedChunkId<'_>,
344 ) -> Result<
345 (Option<RawChunk<Event, Gap>>, ChunkIdentifierGenerator),
346 IndexeddbEventCacheStoreError,
347 > {
348 let _timer = timer!("method");
349
350 let transaction = self.transaction(
351 &[keys::LINKED_CHUNKS, keys::EVENTS, keys::GAPS],
352 IdbTransactionMode::Readonly,
353 )?;
354
355 if transaction.get_chunks_count_by_linked_chunk_id(linked_chunk_id).await? == 0 {
356 return Ok((None, ChunkIdentifierGenerator::new_from_scratch()));
357 }
358 match transaction.get_chunk_by_next_chunk_id(linked_chunk_id, None).await {
362 Err(TransactionError::ItemIsNotUnique) => {
363 Err(IndexeddbEventCacheStoreError::ChunksContainDisjointLists)
367 }
368 Err(e) => {
369 Err(e.into())
372 }
373 Ok(None) => {
374 Err(IndexeddbEventCacheStoreError::ChunksContainCycle)
377 }
378 Ok(Some(last_chunk)) => {
379 let last_chunk_identifier = ChunkIdentifier::new(last_chunk.identifier);
380 let last_raw_chunk = transaction
381 .load_chunk_by_id(linked_chunk_id, last_chunk_identifier)
382 .await?
383 .ok_or(IndexeddbEventCacheStoreError::UnableToLoadChunk)?;
384 let max_chunk_id = transaction
385 .get_max_chunk_by_id(linked_chunk_id)
386 .await?
387 .map(|chunk| ChunkIdentifier::new(chunk.identifier))
388 .ok_or(IndexeddbEventCacheStoreError::NoMaxChunkId)?;
389 let generator =
390 ChunkIdentifierGenerator::new_from_previous_chunk_identifier(max_chunk_id);
391 Ok((Some(last_raw_chunk), generator))
392 }
393 }
394 }
395
396 #[instrument(skip(self))]
397 async fn load_previous_chunk(
398 &self,
399 linked_chunk_id: LinkedChunkId<'_>,
400 before_chunk_identifier: ChunkIdentifier,
401 ) -> Result<Option<RawChunk<Event, Gap>>, IndexeddbEventCacheStoreError> {
402 let _timer = timer!("method");
403
404 let transaction = self.transaction(
405 &[keys::LINKED_CHUNKS, keys::EVENTS, keys::GAPS],
406 IdbTransactionMode::Readonly,
407 )?;
408 if let Some(chunk) =
409 transaction.get_chunk_by_id(linked_chunk_id, before_chunk_identifier).await?
410 && let Some(previous_identifier) = chunk.previous
411 {
412 let previous_identifier = ChunkIdentifier::new(previous_identifier);
413 Ok(transaction.load_chunk_by_id(linked_chunk_id, previous_identifier).await?)
414 } else {
415 Ok(None)
416 }
417 }
418
419 #[instrument(skip(self))]
420 async fn load_thread_info(
421 &self,
422 room_id: &RoomId,
423 thread_id: &EventId,
424 ) -> Result<ThreadInfo, Self::Error> {
425 let _timer = timer!("method");
426
427 let transaction = self.transaction(&[keys::THREADS], IdbTransactionMode::Readonly)?;
428
429 if let Some(thread) = transaction.load_thread_info(room_id, thread_id).await? {
430 return Ok(thread.info);
431 }
432
433 drop(transaction);
434
435 let transaction = self.transaction(&[keys::THREADS], IdbTransactionMode::Readwrite)?;
436
437 let thread = Thread {
438 room_id: room_id.to_owned(),
439 thread_id: thread_id.to_owned(),
440 info: ThreadInfo::new(),
441 };
442 transaction.update_thread_info(&thread).await?;
443 transaction.commit().await?;
444
445 Ok(thread.info)
446 }
447
448 #[instrument(skip(self))]
449 async fn update_thread_info(
450 &self,
451 room_id: &RoomId,
452 thread_id: &EventId,
453 thread_info: &ThreadInfo,
454 ) -> Result<(), Self::Error> {
455 let _timer = timer!("method");
456
457 let transaction = self.transaction(&[keys::THREADS], IdbTransactionMode::Readwrite)?;
458
459 let thread = Thread {
460 room_id: room_id.to_owned(),
461 thread_id: thread_id.to_owned(),
462 info: thread_info.clone(),
463 };
464 transaction.update_thread_info(&thread).await?;
465 transaction.commit().await?;
466
467 Ok(())
468 }
469
470 #[instrument(skip(self))]
471 async fn clear_all_events(
472 &self,
473 room_id: Option<&RoomId>,
474 ) -> Result<(), IndexeddbEventCacheStoreError> {
475 let _timer = timer!("method");
476
477 let transaction = self.transaction(
478 &[keys::LINKED_CHUNKS, keys::EVENTS, keys::GAPS, keys::THREADS],
479 IdbTransactionMode::Readwrite,
480 )?;
481
482 match room_id {
483 None => {
485 transaction.clear::<types::Chunk>().await?;
486 transaction.clear::<types::Event>().await?;
487 transaction.clear::<types::Gap>().await?;
488 transaction.commit().await?;
489 }
490
491 Some(room_id) => {
493 {
495 for linked_chunk_id in
496 [LinkedChunkId::Room(room_id), LinkedChunkId::PinnedEvents(room_id)]
497 {
498 transaction.delete_chunks_by_linked_chunk_id(linked_chunk_id).await?;
500 transaction.delete_gaps_by_linked_chunk_id(linked_chunk_id).await?;
501 transaction.delete_events_by_linked_chunk_id(linked_chunk_id).await?;
502 }
503 }
504
505 {
507 for thread in transaction.get_threads_by_room_id(room_id).await? {
508 let linked_chunk_id = thread.linked_chunk();
509
510 transaction.delete_chunks_by_linked_chunk_id(linked_chunk_id).await?;
512 transaction.delete_gaps_by_linked_chunk_id(linked_chunk_id).await?;
513 transaction.delete_events_by_linked_chunk_id(linked_chunk_id).await?;
514 }
515 }
516
517 transaction.commit().await?;
519 }
520 }
521 Ok(())
522 }
523
524 #[instrument(skip(self, events))]
525 async fn filter_duplicated_events(
526 &self,
527 linked_chunk_id: LinkedChunkId<'_>,
528 events: Vec<OwnedEventId>,
529 ) -> Result<Vec<(OwnedEventId, Position)>, IndexeddbEventCacheStoreError> {
530 let _timer = timer!("method");
531
532 if events.is_empty() {
533 return Ok(Vec::new());
534 }
535
536 let transaction = self.transaction(&[keys::EVENTS], IdbTransactionMode::Readonly)?;
537 let mut duplicated = Vec::new();
538 for event_id in events {
539 if let Some(types::Event::InBand(event)) =
540 transaction.get_event_by_id(linked_chunk_id, &event_id).await?
541 {
542 duplicated.push((event_id, event.position.into()));
543 }
544 }
545 Ok(duplicated)
546 }
547
548 #[instrument(skip(self, event_id))]
549 async fn find_event(
550 &self,
551 room_id: &RoomId,
552 event_id: &EventId,
553 ) -> Result<Option<Event>, IndexeddbEventCacheStoreError> {
554 let _timer = timer!("method");
555
556 let transaction = self.transaction(&[keys::EVENTS], IdbTransactionMode::Readonly)?;
557 transaction
558 .get_events_by_room(room_id, event_id)
559 .await
560 .map(|mut events| events.pop().map(Into::into))
561 .map_err(Into::into)
562 }
563
564 #[instrument(skip(self, event_id, filters))]
565 async fn find_event_relations(
566 &self,
567 room_id: &RoomId,
568 event_id: &EventId,
569 filters: Option<&[RelationType]>,
570 ) -> Result<Vec<(Event, Option<Position>)>, IndexeddbEventCacheStoreError> {
571 let _timer = timer!("method");
572
573 let transaction = self.transaction(&[keys::EVENTS], IdbTransactionMode::Readonly)?;
574
575 let mut related_events = HashMap::<OwnedEventId, types::Event>::new();
576 match filters {
577 Some(relation_types) if !relation_types.is_empty() => {
578 for relation_type in relation_types {
579 let relation = (event_id, relation_type);
580 let events = transaction.get_events_by_relation(room_id, relation).await?;
581 for event in events {
582 let Some(event_id) = event.event_id() else {
583 return Err(IndexeddbEventCacheStoreError::EventWithoutId);
584 };
585 match event.linked_chunk_id() {
586 LinkedChunkId::Room(_) => {
587 related_events.insert(event_id.to_owned(), event);
589 }
590 _ => {
591 related_events
594 .entry(event_id.to_owned())
595 .or_insert_with(|| event.into_out_of_band_event());
596 }
597 }
598 }
599 }
600 }
601 _ => {
602 for event in transaction.get_events_by_related_event(room_id, event_id).await? {
603 let Some(event_id) = event.event_id() else {
604 return Err(IndexeddbEventCacheStoreError::EventWithoutId);
605 };
606 match event.linked_chunk_id() {
607 LinkedChunkId::Room(_) => {
608 related_events.insert(event_id.to_owned(), event);
610 }
611 _ => {
612 related_events
615 .entry(event_id.to_owned())
616 .or_insert_with(|| event.into_out_of_band_event());
617 }
618 }
619 }
620 }
621 }
622 Ok(related_events
623 .into_values()
624 .map(|event| {
625 let position = event.position().map(Into::into);
626 (event.into(), position)
627 })
628 .collect())
629 }
630
631 #[instrument(skip(self))]
632 async fn get_room_events(
633 &self,
634 room_id: &RoomId,
635 event_type: Option<&str>,
636 session_id: Option<&str>,
637 ) -> Result<Vec<Event>, IndexeddbEventCacheStoreError> {
638 let _timer = timer!("method");
639
640 let transaction = self.transaction(&[keys::EVENTS], IdbTransactionMode::Readonly)?;
644 transaction
645 .get_room_events(room_id)
646 .await
647 .map(|mut vec| {
648 vec.dedup_by(|a, b| {
649 if let (Some(a), Some(b)) = (a.event_id(), b.event_id()) {
650 a == b
651 } else {
652 false
653 }
654 });
655 vec.into_iter()
656 .map(Event::from)
657 .filter(|e| {
658 event_type.is_none_or(|event_type| {
659 Some(event_type) == e.kind.event_type().as_deref()
660 })
661 })
662 .filter(|e| session_id.is_none_or(|s| Some(s) == e.kind.session_id()))
663 .collect()
664 })
665 .map_err(Into::into)
666 }
667
668 #[instrument(skip(self, event))]
669 async fn save_event(
670 &self,
671 room_id: &RoomId,
672 event: Event,
673 ) -> Result<(), IndexeddbEventCacheStoreError> {
674 let _timer = timer!("method");
675
676 let Some(event_id) = event.event_id() else {
677 error!(%room_id, "Trying to save an event with no ID");
678 return Ok(());
679 };
680 let transaction = self.transaction(&[keys::EVENTS], IdbTransactionMode::Readwrite)?;
681
682 let mut events = transaction
683 .get_events_by_room(room_id, event_id)
684 .await?
685 .into_iter()
686 .map(|e| e.with_content(event.clone()))
687 .collect::<Vec<types::Event>>();
688 if events.is_empty() {
689 events.push(types::Event::OutOfBand(OutOfBandEvent {
690 linked_chunk_id: LinkedChunkId::Room(room_id).to_owned(),
691 content: event,
692 position: (),
693 }));
694 }
695 for event in events {
696 transaction.put_event(&event).await?;
697 }
698 transaction.commit().await?;
699 Ok(())
700 }
701
702 async fn optimize(&self) -> Result<(), Self::Error> {
703 Ok(())
704 }
705
706 async fn get_size(&self) -> Result<Option<usize>, Self::Error> {
707 Ok(None)
708 }
709
710 async fn close(&self) -> Result<(), Self::Error> {
711 Ok(())
712 }
713
714 async fn reopen(&self) -> Result<(), Self::Error> {
715 Ok(())
716 }
717}
718
719#[cfg(all(test, target_family = "wasm"))]
720mod tests {
721 use matrix_sdk_base::{
722 event_cache::store::EventCacheStoreError, event_cache_store_integration_tests,
723 event_cache_store_integration_tests_time,
724 };
725 use uuid::Uuid;
726
727 use crate::{
728 event_cache_store::IndexeddbEventCacheStore, indexeddb_event_cache_store_integration_tests,
729 };
730
731 mod unencrypted {
732 use super::*;
733
734 wasm_bindgen_test::wasm_bindgen_test_configure!(run_in_browser);
735
736 async fn get_event_cache_store() -> Result<IndexeddbEventCacheStore, EventCacheStoreError> {
737 let name = format!("test-event-cache-store-{}", Uuid::new_v4().as_hyphenated());
738 Ok(IndexeddbEventCacheStore::builder().database_name(name).build().await?)
739 }
740
741 event_cache_store_integration_tests!();
742 event_cache_store_integration_tests_time!();
743
744 indexeddb_event_cache_store_integration_tests!();
745 }
746
747 mod encrypted {
748 use std::sync::Arc;
749
750 use matrix_sdk_store_encryption::StoreCipher;
751
752 use super::*;
753
754 wasm_bindgen_test::wasm_bindgen_test_configure!(run_in_browser);
755
756 async fn get_event_cache_store() -> Result<IndexeddbEventCacheStore, EventCacheStoreError> {
757 let name = format!("test-event-cache-store-{}", Uuid::new_v4().as_hyphenated());
758 Ok(IndexeddbEventCacheStore::builder()
759 .database_name(name)
760 .store_cipher(Arc::new(StoreCipher::new().expect("store cipher")))
761 .build()
762 .await?)
763 }
764
765 event_cache_store_integration_tests!();
766 event_cache_store_integration_tests_time!();
767
768 indexeddb_event_cache_store_integration_tests!();
769 }
770}