180 lines
5.9 KiB
Rust
Raw Normal View History

2022-01-25 19:45:41 +08:00
use crate::RevisionCache;
2021-12-15 23:01:50 +08:00
use flowy_collaboration::{
2022-01-14 15:23:21 +08:00
entities::revision::{RepeatedRevision, Revision, RevisionRange, RevisionState},
util::{pair_rev_id_from_revisions, RevIdCounter},
2021-12-15 23:01:50 +08:00
};
2022-01-14 15:23:21 +08:00
use flowy_error::{FlowyError, FlowyResult};
2021-12-13 13:55:44 +08:00
use lib_infra::future::FutureResult;
2022-01-26 23:29:18 +08:00
2022-02-19 11:34:31 +08:00
use std::sync::Arc;
2022-01-14 15:23:21 +08:00
pub trait RevisionCloudService: Send + Sync {
fn fetch_object(&self, user_id: &str, object_id: &str) -> FutureResult<Vec<Revision>, FlowyError>;
}
2022-01-14 15:23:21 +08:00
pub trait RevisionObjectBuilder: Send + Sync {
type Output;
2022-02-19 11:34:31 +08:00
fn build_object(object_id: &str, revisions: Vec<Revision>) -> FlowyResult<Self::Output>;
2022-01-14 15:23:21 +08:00
}
2022-01-25 20:37:48 +08:00
pub trait RevisionCompact: Send + Sync {
fn compact_revisions(user_id: &str, object_id: &str, revisions: Vec<Revision>) -> FlowyResult<Revision>;
}
2022-01-14 15:23:21 +08:00
pub struct RevisionManager {
pub object_id: String,
2021-12-09 22:28:11 +08:00
user_id: String,
rev_id_counter: RevIdCounter,
2022-02-19 11:34:31 +08:00
rev_cache: Arc<RevisionCache>,
2022-01-22 18:48:43 +08:00
#[cfg(feature = "flowy_unit_test")]
2022-02-18 23:04:55 +08:00
rev_ack_notifier: tokio::sync::broadcast::Sender<i64>,
}
2022-01-14 15:23:21 +08:00
impl RevisionManager {
2022-02-19 11:34:31 +08:00
pub fn new(user_id: &str, object_id: &str, rev_cache: Arc<RevisionCache>) -> Self {
2021-10-07 20:46:29 +08:00
let rev_id_counter = RevIdCounter::new(0);
2022-01-22 18:48:43 +08:00
#[cfg(feature = "flowy_unit_test")]
2022-01-24 17:35:58 +08:00
let (revision_ack_notifier, _) = tokio::sync::broadcast::channel(1);
2022-01-22 18:48:43 +08:00
Self {
2022-01-14 15:23:21 +08:00
object_id: object_id.to_string(),
2021-12-09 22:28:11 +08:00
user_id: user_id.to_owned(),
rev_id_counter,
2022-02-19 11:34:31 +08:00
rev_cache,
2022-01-22 18:48:43 +08:00
#[cfg(feature = "flowy_unit_test")]
2022-02-18 23:04:55 +08:00
rev_ack_notifier: revision_ack_notifier,
}
}
2022-01-25 20:37:48 +08:00
pub async fn load<B, C>(&mut self, cloud: Arc<dyn RevisionCloudService>) -> FlowyResult<B::Output>
2022-01-14 15:23:21 +08:00
where
2022-01-25 20:37:48 +08:00
B: RevisionObjectBuilder,
C: RevisionCompact,
2022-01-14 15:23:21 +08:00
{
2022-01-23 22:33:47 +08:00
let (revisions, rev_id) = RevisionLoader {
2022-01-14 15:23:21 +08:00
object_id: self.object_id.clone(),
2021-12-18 00:23:26 +08:00
user_id: self.user_id.clone(),
2022-01-14 15:23:21 +08:00
cloud,
2022-02-19 11:34:31 +08:00
rev_cache: self.rev_cache.clone(),
2021-12-18 00:23:26 +08:00
}
2022-02-19 11:34:31 +08:00
.load()
2021-12-18 00:23:26 +08:00
.await?;
2022-01-23 22:33:47 +08:00
self.rev_id_counter.set(rev_id);
2022-02-19 11:34:31 +08:00
B::build_object(&self.object_id, revisions)
2021-10-07 20:46:29 +08:00
}
2022-01-01 14:23:58 +08:00
#[tracing::instrument(level = "debug", skip(self, revisions), err)]
2022-01-14 15:23:21 +08:00
pub async fn reset_object(&self, revisions: RepeatedRevision) -> FlowyResult<()> {
2022-01-02 10:34:42 +08:00
let rev_id = pair_rev_id_from_revisions(&revisions).1;
2022-02-19 11:34:31 +08:00
let _ = self.rev_cache.reset(revisions.into_inner()).await?;
2022-01-02 10:34:42 +08:00
self.rev_id_counter.set(rev_id);
Ok(())
2022-01-01 14:23:58 +08:00
}
2022-01-02 10:34:42 +08:00
#[tracing::instrument(level = "debug", skip(self, revision), err)]
2021-12-14 18:04:51 +08:00
pub async fn add_remote_revision(&self, revision: &Revision) -> Result<(), FlowyError> {
2022-01-01 23:09:13 +08:00
if revision.delta_data.is_empty() {
return Err(FlowyError::internal().context("Delta data should be empty"));
}
2022-01-26 23:29:18 +08:00
2022-02-19 11:34:31 +08:00
let _ = self.rev_cache.add_ack_revision(revision).await?;
2022-01-07 17:37:11 +08:00
self.rev_id_counter.set(revision.rev_id);
2021-12-13 22:46:35 +08:00
Ok(())
}
2022-01-01 16:16:06 +08:00
#[tracing::instrument(level = "debug", skip(self, revision))]
2022-01-25 20:37:48 +08:00
pub async fn add_local_revision<C>(&self, revision: &Revision) -> Result<(), FlowyError>
where
C: RevisionCompact,
{
2022-01-01 23:09:13 +08:00
if revision.delta_data.is_empty() {
return Err(FlowyError::internal().context("Delta data should be empty"));
}
2022-02-19 11:34:31 +08:00
let rev_id = self.rev_cache.add_sync_revision::<C>(revision).await?;
2022-01-26 23:29:18 +08:00
self.rev_id_counter.set(rev_id);
2021-10-07 20:46:29 +08:00
Ok(())
}
2022-01-01 16:16:06 +08:00
#[tracing::instrument(level = "debug", skip(self), err)]
2021-12-16 21:31:36 +08:00
pub async fn ack_revision(&self, rev_id: i64) -> Result<(), FlowyError> {
2022-02-19 11:34:31 +08:00
if self.rev_cache.ack_revision(rev_id).await.is_ok() {
#[cfg(feature = "flowy_unit_test")]
2022-02-18 23:04:55 +08:00
let _ = self.rev_ack_notifier.send(rev_id);
}
Ok(())
}
2022-01-24 17:35:58 +08:00
pub fn rev_id(&self) -> i64 {
self.rev_id_counter.value()
}
2022-01-02 10:34:42 +08:00
pub fn next_rev_id_pair(&self) -> (i64, i64) {
let cur = self.rev_id_counter.value();
let next = self.rev_id_counter.next();
(cur, next)
}
2021-12-25 21:44:45 +08:00
pub async fn get_revisions_in_range(&self, range: RevisionRange) -> Result<Vec<Revision>, FlowyError> {
2022-02-19 11:34:31 +08:00
let revisions = self.rev_cache.revisions_in_range(&range).await?;
2021-12-25 21:44:45 +08:00
Ok(revisions)
}
2021-12-08 21:51:06 +08:00
2022-01-25 20:37:48 +08:00
pub async fn next_sync_revision(&self) -> FlowyResult<Option<Revision>> {
2022-02-19 11:34:31 +08:00
Ok(self.rev_cache.next_sync_revision().await?)
2022-01-01 16:16:06 +08:00
}
2021-12-18 18:35:45 +08:00
2021-12-25 21:44:45 +08:00
pub async fn get_revision(&self, rev_id: i64) -> Option<Revision> {
2022-02-19 11:34:31 +08:00
self.rev_cache.get(rev_id).await.map(|record| record.revision)
2021-12-25 21:44:45 +08:00
}
}
2021-12-08 14:17:40 +08:00
2022-01-25 20:37:48 +08:00
#[cfg(feature = "flowy_unit_test")]
impl RevisionManager {
pub async fn revision_cache(&self) -> Arc<RevisionCache> {
2022-02-19 11:34:31 +08:00
self.rev_cache.clone()
2022-01-25 20:37:48 +08:00
}
2022-02-18 23:04:55 +08:00
pub fn ack_notify(&self) -> tokio::sync::broadcast::Receiver<i64> {
self.rev_ack_notifier.subscribe()
2022-01-01 16:16:06 +08:00
}
}
2021-12-18 00:23:26 +08:00
struct RevisionLoader {
2022-01-14 15:23:21 +08:00
object_id: String,
2021-12-18 00:23:26 +08:00
user_id: String,
2022-01-14 15:23:21 +08:00
cloud: Arc<dyn RevisionCloudService>,
2022-02-19 11:34:31 +08:00
rev_cache: Arc<RevisionCache>,
2021-12-18 00:23:26 +08:00
}
impl RevisionLoader {
2022-02-19 11:34:31 +08:00
async fn load(&self) -> Result<(Vec<Revision>, i64), FlowyError> {
let records = self.rev_cache.batch_get(&self.object_id)?;
2021-12-18 00:23:26 +08:00
let revisions: Vec<Revision>;
2022-01-23 22:33:47 +08:00
let mut rev_id = 0;
2021-12-18 00:23:26 +08:00
if records.is_empty() {
2022-01-14 15:23:21 +08:00
let remote_revisions = self.cloud.fetch_object(&self.user_id, &self.object_id).await?;
for revision in &remote_revisions {
2022-01-23 22:33:47 +08:00
rev_id = revision.rev_id;
2022-02-19 11:34:31 +08:00
let _ = self.rev_cache.add_ack_revision(revision).await?;
2022-01-14 15:23:21 +08:00
}
revisions = remote_revisions;
2021-12-18 00:23:26 +08:00
} else {
2022-01-25 20:37:48 +08:00
for record in &records {
rev_id = record.revision.rev_id;
if record.state == RevisionState::Sync {
// Sync the records if their state is RevisionState::Sync.
2022-02-19 11:34:31 +08:00
let _ = self.rev_cache.sync_revision(&record.revision).await?;
2022-01-23 22:33:47 +08:00
}
}
2021-12-18 00:23:26 +08:00
revisions = records.into_iter().map(|record| record.revision).collect::<_>();
}
2022-01-23 22:33:47 +08:00
if let Some(revision) = revisions.last() {
debug_assert_eq!(rev_id, revision.rev_id);
}
Ok((revisions, rev_id))
2021-12-18 00:23:26 +08:00
}
}