281 lines
9.4 KiB
Rust
Raw Normal View History

2022-01-01 16:16:06 +08:00
use crate::{
errors::FlowyError,
services::doc::{revision::RevisionCache, RevisionRecord},
};
2021-12-18 00:23:26 +08:00
use bytes::Bytes;
2022-01-01 16:16:06 +08:00
use dashmap::DashMap;
2021-12-15 23:01:50 +08:00
use flowy_collaboration::{
2021-12-22 21:13:52 +08:00
entities::{
2021-12-23 23:17:57 +08:00
doc::DocumentInfo,
revision::{RepeatedRevision, Revision, RevisionRange, RevisionState},
2021-12-22 21:13:52 +08:00
},
2021-12-15 23:01:50 +08:00
util::{md5, RevIdCounter},
};
2021-12-14 18:04:51 +08:00
use flowy_error::FlowyResult;
2021-12-13 13:55:44 +08:00
use lib_infra::future::FutureResult;
2021-12-07 22:32:34 +08:00
use lib_ot::{
2021-12-18 00:23:26 +08:00
core::{Operation, OperationTransformable},
2022-01-01 16:16:06 +08:00
errors::OTError,
2021-12-07 22:32:34 +08:00
rich_text::RichTextDelta,
};
2022-01-01 16:16:06 +08:00
use std::{collections::VecDeque, sync::Arc};
use tokio::sync::RwLock;
pub trait RevisionServer: Send + Sync {
2021-12-23 23:17:57 +08:00
fn fetch_document(&self, doc_id: &str) -> FutureResult<DocumentInfo, FlowyError>;
}
pub struct RevisionManager {
doc_id: String,
2021-12-09 22:28:11 +08:00
user_id: String,
rev_id_counter: RevIdCounter,
2021-12-08 14:17:40 +08:00
cache: Arc<RevisionCache>,
2022-01-01 16:16:06 +08:00
sync_seq: Arc<RevisionSyncSeq>,
}
impl RevisionManager {
2021-12-16 21:31:36 +08:00
pub fn new(user_id: &str, doc_id: &str, cache: Arc<RevisionCache>) -> Self {
2021-10-07 20:46:29 +08:00
let rev_id_counter = RevIdCounter::new(0);
2022-01-01 16:16:06 +08:00
let sync_seq = Arc::new(RevisionSyncSeq::new());
Self {
doc_id: doc_id.to_string(),
2021-12-09 22:28:11 +08:00
user_id: user_id.to_owned(),
rev_id_counter,
2021-12-08 14:17:40 +08:00
cache,
2022-01-01 16:16:06 +08:00
sync_seq,
}
}
2021-12-18 00:23:26 +08:00
pub async fn load_document(&mut self, server: Arc<dyn RevisionServer>) -> FlowyResult<RichTextDelta> {
let revisions = RevisionLoader {
doc_id: self.doc_id.clone(),
user_id: self.user_id.clone(),
server,
cache: self.cache.clone(),
}
.load()
.await?;
let doc = mk_doc_from_revisions(&self.doc_id, revisions)?;
2021-12-25 21:44:45 +08:00
self.rev_id_counter.set(doc.rev_id);
2021-10-07 20:46:29 +08:00
Ok(doc.delta()?)
}
2022-01-01 14:23:58 +08:00
#[tracing::instrument(level = "debug", skip(self, revisions), err)]
pub async fn reset_document(&self, revisions: RepeatedRevision) -> FlowyResult<()> {
self.cache.reset_document(&self.doc_id, revisions.into_inner())
}
2022-01-01 16:16:06 +08:00
#[tracing::instrument(level = "debug", skip(self, revision))]
2021-12-14 18:04:51 +08:00
pub async fn add_remote_revision(&self, revision: &Revision) -> Result<(), FlowyError> {
2021-12-25 21:44:45 +08:00
self.rev_id_counter.set(revision.rev_id);
let _ = self.cache.add(revision.clone(), RevisionState::Ack, true).await?;
2021-12-13 22:46:35 +08:00
Ok(())
}
2022-01-01 16:16:06 +08:00
#[tracing::instrument(level = "debug", skip(self, revision))]
2021-12-14 18:04:51 +08:00
pub async fn add_local_revision(&self, revision: &Revision) -> Result<(), FlowyError> {
let record = self.cache.add(revision.clone(), RevisionState::Local, true).await?;
2022-01-01 16:16:06 +08:00
self.sync_seq.add_revision(record).await?;
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-01-01 16:16:06 +08:00
if self.sync_seq.ack(&rev_id).await.is_ok() {
self.cache.ack(rev_id).await;
}
Ok(())
}
pub fn rev_id(&self) -> i64 { self.rev_id_counter.value() }
2022-01-01 16:16:06 +08:00
pub fn set_rev_id(&self, rev_id: i64) { self.rev_id_counter.set(rev_id); }
pub fn next_rev_id(&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> {
2021-11-27 19:19:41 +08:00
debug_assert!(range.doc_id == self.doc_id);
2021-12-08 14:17:40 +08:00
let revisions = self.cache.revisions_in_range(range.clone()).await?;
2021-12-25 21:44:45 +08:00
Ok(revisions)
}
2021-12-08 21:51:06 +08:00
2022-01-01 16:16:06 +08:00
pub fn next_sync_revision(&self) -> FutureResult<Option<Revision>, FlowyError> {
let sync_seq = self.sync_seq.clone();
let cache = self.cache.clone();
FutureResult::new(async move {
match sync_seq.next_sync_revision().await {
None => match sync_seq.next_sync_rev_id().await {
None => Ok(None),
Some(rev_id) => Ok(cache.get(rev_id).await.map(|record| record.revision)),
},
Some((_, record)) => Ok(Some(record.revision)),
}
})
}
2021-12-18 18:35:45 +08:00
2021-12-20 20:59:33 +08:00
pub async fn latest_revision(&self) -> Revision { self.cache.latest_revision().await }
2021-12-25 21:44:45 +08:00
pub async fn get_revision(&self, rev_id: i64) -> Option<Revision> {
2022-01-01 16:16:06 +08:00
self.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-01 16:16:06 +08:00
struct RevisionSyncSeq {
revs_map: Arc<DashMap<i64, RevisionRecord>>,
local_revs: Arc<RwLock<VecDeque<i64>>>,
}
impl std::default::Default for RevisionSyncSeq {
fn default() -> Self {
let local_revs = Arc::new(RwLock::new(VecDeque::new()));
RevisionSyncSeq {
revs_map: Arc::new(DashMap::new()),
local_revs,
}
}
}
impl RevisionSyncSeq {
fn new() -> Self { RevisionSyncSeq::default() }
async fn add_revision(&self, record: RevisionRecord) -> Result<(), OTError> {
// The last revision's rev_id must be greater than the new one.
if let Some(rev_id) = self.local_revs.read().await.back() {
if *rev_id >= record.revision.rev_id {
return Err(OTError::revision_id_conflict()
.context(format!("The new revision's id must be greater than {}", rev_id)));
}
}
self.local_revs.write().await.push_back(record.revision.rev_id);
self.revs_map.insert(record.revision.rev_id, record);
Ok(())
}
async fn ack(&self, rev_id: &i64) -> FlowyResult<()> {
if let Some(pop_rev_id) = self.next_sync_rev_id().await {
if &pop_rev_id != rev_id {
let desc = format!(
"The ack rev_id:{} is not equal to the current rev_id:{}",
rev_id, pop_rev_id
);
// tracing::error!("{}", desc);
return Err(FlowyError::internal().context(desc));
}
tracing::debug!("pop revision {}", pop_rev_id);
self.revs_map.remove(&pop_rev_id);
let _ = self.local_revs.write().await.pop_front();
}
Ok(())
}
async fn next_sync_revision(&self) -> Option<(i64, RevisionRecord)> {
match self.local_revs.read().await.front() {
None => None,
Some(rev_id) => self.revs_map.get(rev_id).map(|r| (*r.key(), r.value().clone())),
}
}
async fn next_sync_rev_id(&self) -> Option<i64> { self.local_revs.read().await.front().copied() }
2021-12-08 14:17:40 +08:00
}
2021-12-18 00:23:26 +08:00
struct RevisionLoader {
doc_id: String,
user_id: String,
server: Arc<dyn RevisionServer>,
cache: Arc<RevisionCache>,
}
impl RevisionLoader {
async fn load(&self) -> Result<Vec<Revision>, FlowyError> {
2022-01-01 16:16:06 +08:00
let records = self.cache.batch_get(&self.doc_id)?;
2021-12-18 00:23:26 +08:00
let revisions: Vec<Revision>;
if records.is_empty() {
let doc = self.server.fetch_document(&self.doc_id).await?;
let delta_data = Bytes::from(doc.text.clone());
2021-12-18 00:23:26 +08:00
let doc_md5 = md5(&delta_data);
let revision = Revision::new(
2021-12-26 23:59:45 +08:00
&doc.doc_id,
2021-12-18 00:23:26 +08:00
doc.base_rev_id,
doc.rev_id,
delta_data,
&self.user_id,
doc_md5,
);
let _ = self.cache.add(revision.clone(), RevisionState::Ack, true).await?;
2021-12-18 00:23:26 +08:00
revisions = vec![revision];
} else {
for record in &records {
match record.state {
2022-01-01 16:16:06 +08:00
RevisionState::Local => {
//
match self
.cache
.add(record.revision.clone(), RevisionState::Local, false)
.await
{
2022-01-01 16:16:06 +08:00
Ok(_) => {},
Err(e) => tracing::error!("{}", e),
}
2021-12-18 00:23:26 +08:00
},
2022-01-01 14:23:58 +08:00
RevisionState::Ack => {},
2021-12-18 00:23:26 +08:00
}
}
revisions = records.into_iter().map(|record| record.revision).collect::<_>();
}
Ok(revisions)
}
}
2021-12-23 23:17:57 +08:00
fn mk_doc_from_revisions(doc_id: &str, revisions: Vec<Revision>) -> FlowyResult<DocumentInfo> {
2021-12-18 00:23:26 +08:00
let (base_rev_id, rev_id) = revisions.last().unwrap().pair_rev_id();
let mut delta = RichTextDelta::new();
for (_, revision) in revisions.into_iter().enumerate() {
match RichTextDelta::from_bytes(revision.delta_data) {
Ok(local_delta) => {
delta = delta.compose(&local_delta)?;
},
Err(e) => {
tracing::error!("Deserialize delta from revision failed: {}", e);
},
}
}
correct_delta_if_need(&mut delta);
2021-12-23 23:17:57 +08:00
Result::<DocumentInfo, FlowyError>::Ok(DocumentInfo {
2021-12-26 23:59:45 +08:00
doc_id: doc_id.to_owned(),
text: delta.to_json(),
2021-12-18 00:23:26 +08:00
rev_id,
base_rev_id,
})
}
fn correct_delta_if_need(delta: &mut RichTextDelta) {
if delta.ops.last().is_none() {
return;
}
let data = delta.ops.last().as_ref().unwrap().get_data();
if !data.ends_with('\n') {
log::error!("❌The op must end with newline. Correcting it by inserting newline op");
delta.ops.push(Operation::Insert("\n".into()));
}
}
2022-01-01 16:16:06 +08:00
#[cfg(feature = "flowy_unit_test")]
impl RevisionSyncSeq {
#[allow(dead_code)]
pub fn revs_map(&self) -> Arc<DashMap<i64, RevisionRecord>> { self.revs_map.clone() }
#[allow(dead_code)]
pub fn pending_revs(&self) -> Arc<RwLock<VecDeque<i64>>> { self.local_revs.clone() }
}
#[cfg(feature = "flowy_unit_test")]
impl RevisionManager {
pub fn revision_cache(&self) -> Arc<RevisionCache> { self.cache.clone() }
}