Mirrored_Repos/AppFlowy
1
0
Fork 0
mirror of https://github.com/AppFlowy-IO/AppFlowy.git synced 2024-08-30 18:12:39 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor: md5 of revision

Browse Source
This commit is contained in:
nathan 2022-11-02 10:21:10 +08:00
parent cd8e4ddf00
commit 608a08eb76
34 changed files with 300 additions and 235 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
frontend/rust-lib/flowy-document/src/editor/document.rs
View File

@ -28,9 +28,9 @@ impl Document {
} }
} }
pub fn md5(&self) -> String { pub fn document_md5(&self) -> String {
// format!("{:x}", md5::compute(bytes)) let bytes = self.tree.to_bytes();
"".to_owned() format!("{:x}", md5::compute(&bytes))
} }
pub fn get_tree(&self) -> &NodeTree { pub fn get_tree(&self) -> &NodeTree {

2
frontend/rust-lib/flowy-document/src/editor/queue.rs
View File

@ -63,7 +63,7 @@ impl DocumentQueue {
Command::ComposeTransaction { transaction, ret } => { Command::ComposeTransaction { transaction, ret } => {
self.document.write().await.apply_transaction(transaction.clone())?; self.document.write().await.apply_transaction(transaction.clone())?;
let _ = self let _ = self
.save_local_operations(transaction, self.document.read().await.md5()) .save_local_operations(transaction, self.document.read().await.document_md5())
.await?; .await?;
let _ = ret.send(Ok(())); let _ = ret.send(Ok(()));
} }

9
frontend/rust-lib/flowy-document/src/manager.rs
View File

@ -12,11 +12,8 @@ use flowy_revision::{
RevisionCloudService, RevisionManager, RevisionPersistence, RevisionWebSocket, SQLiteRevisionSnapshotPersistence, RevisionCloudService, RevisionManager, RevisionPersistence, RevisionWebSocket, SQLiteRevisionSnapshotPersistence,
}; };
use flowy_sync::client_document::initial_delta_document_content; use flowy_sync::client_document::initial_delta_document_content;
use flowy_sync::entities::{ use flowy_sync::entities::{document::DocumentIdPB, revision::Revision, ws_data::ServerRevisionWSData};
document::DocumentIdPB, use flowy_sync::util::md5;
revision::{md5, RepeatedRevision, Revision},
ws_data::ServerRevisionWSData,
};
use lib_infra::future::FutureResult; use lib_infra::future::FutureResult;
use lib_ws::WSConnectState; use lib_ws::WSConnectState;
use std::any::Any; use std::any::Any;
@ -139,7 +136,7 @@ impl DocumentManager {
Ok(()) Ok(())
} }
pub async fn create_document<T: AsRef<str>>(&self, doc_id: T, revisions: RepeatedRevision) -> FlowyResult<()> { pub async fn create_document<T: AsRef<str>>(&self, doc_id: T, revisions: Vec<Revision>) -> FlowyResult<()> {
let doc_id = doc_id.as_ref().to_owned(); let doc_id = doc_id.as_ref().to_owned();
let db_pool = self.persistence.database.db_pool()?; let db_pool = self.persistence.database.db_pool()?;
// Maybe we could save the document to disk without creating the RevisionManager // Maybe we could save the document to disk without creating the RevisionManager

28
frontend/rust-lib/flowy-document/src/old_editor/queue.rs
View File

@ -3,7 +3,7 @@ use crate::DocumentUser;
use async_stream::stream; use async_stream::stream;
use flowy_database::ConnectionPool; use flowy_database::ConnectionPool;
use flowy_error::FlowyError; use flowy_error::FlowyError;
use flowy_revision::{OperationsMD5, RevisionManager, TransformOperations}; use flowy_revision::{RevisionMD5, RevisionManager, TransformOperations};
use flowy_sync::{ use flowy_sync::{
client_document::{history::UndoResult, ClientDocument}, client_document::{history::UndoResult, ClientDocument},
entities::revision::{RevId, Revision}, entities::revision::{RevId, Revision},
@ -70,7 +70,7 @@ impl EditDocumentQueue {
EditorCommand::ComposeLocalOperations { operations, ret } => { EditorCommand::ComposeLocalOperations { operations, ret } => {
let mut document = self.document.write().await; let mut document = self.document.write().await;
let _ = document.compose_operations(operations.clone())?; let _ = document.compose_operations(operations.clone())?;
let md5 = document.md5(); let md5 = document.document_md5();
drop(document); drop(document);
let _ = self.save_local_operations(operations, md5).await?; let _ = self.save_local_operations(operations, md5).await?;
let _ = ret.send(Ok(())); let _ = ret.send(Ok(()));
@ -78,16 +78,16 @@ impl EditDocumentQueue {
EditorCommand::ComposeRemoteOperation { client_operations, ret } => { EditorCommand::ComposeRemoteOperation { client_operations, ret } => {
let mut document = self.document.write().await; let mut document = self.document.write().await;
let _ = document.compose_operations(client_operations.clone())?; let _ = document.compose_operations(client_operations.clone())?;
let md5 = document.md5(); let md5 = document.document_md5();
drop(document); drop(document);
let _ = ret.send(Ok(md5)); let _ = ret.send(Ok(md5.into()));
} }
EditorCommand::ResetOperations { operations, ret } => { EditorCommand::ResetOperations { operations, ret } => {
let mut document = self.document.write().await; let mut document = self.document.write().await;
let _ = document.set_operations(operations); let _ = document.set_operations(operations);
let md5 = document.md5(); let md5 = document.document_md5();
drop(document); drop(document);
let _ = ret.send(Ok(md5)); let _ = ret.send(Ok(md5.into()));
} }
EditorCommand::TransformOperations { operations, ret } => { EditorCommand::TransformOperations { operations, ret } => {
let f = || async { let f = || async {
@ -114,14 +114,14 @@ impl EditDocumentQueue {
EditorCommand::Insert { index, data, ret } => { EditorCommand::Insert { index, data, ret } => {
let mut write_guard = self.document.write().await; let mut write_guard = self.document.write().await;
let operations = write_guard.insert(index, data)?; let operations = write_guard.insert(index, data)?;
let md5 = write_guard.md5(); let md5 = write_guard.document_md5();
let _ = self.save_local_operations(operations, md5).await?; let _ = self.save_local_operations(operations, md5).await?;
let _ = ret.send(Ok(())); let _ = ret.send(Ok(()));
} }
EditorCommand::Delete { interval, ret } => { EditorCommand::Delete { interval, ret } => {
let mut write_guard = self.document.write().await; let mut write_guard = self.document.write().await;
let operations = write_guard.delete(interval)?; let operations = write_guard.delete(interval)?;
let md5 = write_guard.md5(); let md5 = write_guard.document_md5();
let _ = self.save_local_operations(operations, md5).await?; let _ = self.save_local_operations(operations, md5).await?;
let _ = ret.send(Ok(())); let _ = ret.send(Ok(()));
} }
@ -132,14 +132,14 @@ impl EditDocumentQueue {
} => { } => {
let mut write_guard = self.document.write().await; let mut write_guard = self.document.write().await;
let operations = write_guard.format(interval, attribute)?; let operations = write_guard.format(interval, attribute)?;
let md5 = write_guard.md5(); let md5 = write_guard.document_md5();
let _ = self.save_local_operations(operations, md5).await?; let _ = self.save_local_operations(operations, md5).await?;
let _ = ret.send(Ok(())); let _ = ret.send(Ok(()));
} }
EditorCommand::Replace { interval, data, ret } => { EditorCommand::Replace { interval, data, ret } => {
let mut write_guard = self.document.write().await; let mut write_guard = self.document.write().await;
let operations = write_guard.replace(interval, data)?; let operations = write_guard.replace(interval, data)?;
let md5 = write_guard.md5(); let md5 = write_guard.document_md5();
let _ = self.save_local_operations(operations, md5).await?; let _ = self.save_local_operations(operations, md5).await?;
let _ = ret.send(Ok(())); let _ = ret.send(Ok(()));
} }
@ -152,14 +152,14 @@ impl EditDocumentQueue {
EditorCommand::Undo { ret } => { EditorCommand::Undo { ret } => {
let mut write_guard = self.document.write().await; let mut write_guard = self.document.write().await;
let UndoResult { operations } = write_guard.undo()?; let UndoResult { operations } = write_guard.undo()?;
let md5 = write_guard.md5(); let md5 = write_guard.document_md5();
let _ = self.save_local_operations(operations, md5).await?; let _ = self.save_local_operations(operations, md5).await?;
let _ = ret.send(Ok(())); let _ = ret.send(Ok(()));
} }
EditorCommand::Redo { ret } => { EditorCommand::Redo { ret } => {
let mut write_guard = self.document.write().await; let mut write_guard = self.document.write().await;
let UndoResult { operations } = write_guard.redo()?; let UndoResult { operations } = write_guard.redo()?;
let md5 = write_guard.md5(); let md5 = write_guard.document_md5();
let _ = self.save_local_operations(operations, md5).await?; let _ = self.save_local_operations(operations, md5).await?;
let _ = ret.send(Ok(())); let _ = ret.send(Ok(()));
} }
@ -197,11 +197,11 @@ pub(crate) enum EditorCommand {
}, },
ComposeRemoteOperation { ComposeRemoteOperation {
client_operations: DeltaTextOperations, client_operations: DeltaTextOperations,
ret: Ret<OperationsMD5>, ret: Ret<RevisionMD5>,
}, },
ResetOperations { ResetOperations {
operations: DeltaTextOperations, operations: DeltaTextOperations,
ret: Ret<OperationsMD5>, ret: Ret<RevisionMD5>,
}, },
TransformOperations { TransformOperations {
operations: DeltaTextOperations, operations: DeltaTextOperations,

7
frontend/rust-lib/flowy-document/src/old_editor/web_socket.rs
View File

@ -136,7 +136,7 @@ impl ConflictResolver<DeltaDocumentResolveOperations> for DocumentConflictResolv
fn compose_operations( fn compose_operations(
&self, &self,
operations: DeltaDocumentResolveOperations, operations: DeltaDocumentResolveOperations,
) -> BoxResultFuture<OperationsMD5, FlowyError> { ) -> BoxResultFuture<RevisionMD5, FlowyError> {
let tx = self.edit_cmd_tx.clone(); let tx = self.edit_cmd_tx.clone();
let operations = operations.into_inner(); let operations = operations.into_inner();
Box::pin(async move { Box::pin(async move {
@ -172,10 +172,7 @@ impl ConflictResolver<DeltaDocumentResolveOperations> for DocumentConflictResolv
}) })
} }
fn reset_operations( fn reset_operations(&self, operations: DeltaDocumentResolveOperations) -> BoxResultFuture<RevisionMD5, FlowyError> {
&self,
operations: DeltaDocumentResolveOperations,
) -> BoxResultFuture<OperationsMD5, FlowyError> {
let tx = self.edit_cmd_tx.clone(); let tx = self.edit_cmd_tx.clone();
let operations = operations.into_inner(); let operations = operations.into_inner();
Box::pin(async move { Box::pin(async move {

8
frontend/rust-lib/flowy-document/src/services/migration.rs
View File

@ -4,9 +4,9 @@ use crate::DocumentDatabase;
use bytes::Bytes; use bytes::Bytes;
use flowy_database::kv::KV; use flowy_database::kv::KV;
use flowy_error::FlowyResult; use flowy_error::FlowyResult;
use flowy_revision::disk::{RevisionDiskCache, RevisionRecord}; use flowy_revision::disk::{RevisionDiskCache, SyncRecord};
use flowy_sync::entities::revision::{md5, Revision}; use flowy_sync::entities::revision::Revision;
use flowy_sync::util::make_operations_from_revisions; use flowy_sync::util::{make_operations_from_revisions, md5};
use std::sync::Arc; use std::sync::Arc;
const V1_MIGRATION: &str = "DOCUMENT_V1_MIGRATION"; const V1_MIGRATION: &str = "DOCUMENT_V1_MIGRATION";
@ -44,7 +44,7 @@ impl DocumentMigration {
let bytes = Bytes::from(transaction.to_bytes()?); let bytes = Bytes::from(transaction.to_bytes()?);
let md5 = format!("{:x}", md5::compute(&bytes)); let md5 = format!("{:x}", md5::compute(&bytes));
let revision = Revision::new(&document_id, 0, 1, bytes, &self.user_id, md5); let revision = Revision::new(&document_id, 0, 1, bytes, &self.user_id, md5);
let record = RevisionRecord::new(revision); let record = SyncRecord::new(revision);
match disk_cache.create_revision_records(vec![record]) { match disk_cache.create_revision_records(vec![record]) {
Ok(_) => {} Ok(_) => {}
Err(err) => { Err(err) => {

20
frontend/rust-lib/flowy-document/src/services/persistence/rev_sqlite/document_rev_sqlite_v0.rs
View File

@ -7,7 +7,7 @@ use flowy_database::{
ConnectionPool, ConnectionPool,
}; };
use flowy_error::{internal_error, FlowyError, FlowyResult}; use flowy_error::{internal_error, FlowyError, FlowyResult};
use flowy_revision::disk::{RevisionChangeset, RevisionDiskCache, RevisionRecord, RevisionState}; use flowy_revision::disk::{RevisionChangeset, RevisionDiskCache, RevisionState, SyncRecord};
use flowy_sync::{ use flowy_sync::{
entities::revision::{RevType, Revision, RevisionRange}, entities::revision::{RevType, Revision, RevisionRange},
util::md5, util::md5,
@ -23,7 +23,7 @@ pub struct SQLiteDeltaDocumentRevisionPersistence {
impl RevisionDiskCache<Arc<ConnectionPool>> for SQLiteDeltaDocumentRevisionPersistence { impl RevisionDiskCache<Arc<ConnectionPool>> for SQLiteDeltaDocumentRevisionPersistence {
type Error = FlowyError; type Error = FlowyError;
fn create_revision_records(&self, revision_records: Vec<RevisionRecord>) -> Result<(), Self::Error> { fn create_revision_records(&self, revision_records: Vec<SyncRecord>) -> Result<(), Self::Error> {
let conn = self.pool.get().map_err(internal_error)?; let conn = self.pool.get().map_err(internal_error)?;
let _ = DeltaRevisionSql::create(revision_records, &*conn)?; let _ = DeltaRevisionSql::create(revision_records, &*conn)?;
Ok(()) Ok(())
@ -37,7 +37,7 @@ impl RevisionDiskCache<Arc<ConnectionPool>> for SQLiteDeltaDocumentRevisionPersi
&self, &self,
object_id: &str, object_id: &str,
rev_ids: Option<Vec<i64>>, rev_ids: Option<Vec<i64>>,
) -> Result<Vec<RevisionRecord>, Self::Error> { ) -> Result<Vec<SyncRecord>, Self::Error> {
let conn = self.pool.get().map_err(internal_error)?; let conn = self.pool.get().map_err(internal_error)?;
let records = DeltaRevisionSql::read(&self.user_id, object_id, rev_ids, &*conn)?; let records = DeltaRevisionSql::read(&self.user_id, object_id, rev_ids, &*conn)?;
Ok(records) Ok(records)
@ -47,7 +47,7 @@ impl RevisionDiskCache<Arc<ConnectionPool>> for SQLiteDeltaDocumentRevisionPersi
&self, &self,
object_id: &str, object_id: &str,
range: &RevisionRange, range: &RevisionRange,
) -> Result<Vec<RevisionRecord>, Self::Error> { ) -> Result<Vec<SyncRecord>, Self::Error> {
let conn = &*self.pool.get().map_err(internal_error)?; let conn = &*self.pool.get().map_err(internal_error)?;
let revisions = DeltaRevisionSql::read_with_range(&self.user_id, object_id, range.clone(), conn)?; let revisions = DeltaRevisionSql::read_with_range(&self.user_id, object_id, range.clone(), conn)?;
Ok(revisions) Ok(revisions)
@ -74,7 +74,7 @@ impl RevisionDiskCache<Arc<ConnectionPool>> for SQLiteDeltaDocumentRevisionPersi
&self, &self,
object_id: &str, object_id: &str,
deleted_rev_ids: Option<Vec<i64>>, deleted_rev_ids: Option<Vec<i64>>,
inserted_records: Vec<RevisionRecord>, inserted_records: Vec<SyncRecord>,
) -> Result<(), Self::Error> { ) -> Result<(), Self::Error> {
let conn = self.pool.get().map_err(internal_error)?; let conn = self.pool.get().map_err(internal_error)?;
conn.immediate_transaction::<_, FlowyError, _>(|| { conn.immediate_transaction::<_, FlowyError, _>(|| {
@ -97,7 +97,7 @@ impl SQLiteDeltaDocumentRevisionPersistence {
pub struct DeltaRevisionSql {} pub struct DeltaRevisionSql {}
impl DeltaRevisionSql { impl DeltaRevisionSql {
fn create(revision_records: Vec<RevisionRecord>, conn: &SqliteConnection) -> Result<(), FlowyError> { fn create(revision_records: Vec<SyncRecord>, conn: &SqliteConnection) -> Result<(), FlowyError> {
// Batch insert: https://diesel.rs/guides/all-about-inserts.html // Batch insert: https://diesel.rs/guides/all-about-inserts.html
let records = revision_records let records = revision_records
@ -143,7 +143,7 @@ impl DeltaRevisionSql {
object_id: &str, object_id: &str,
rev_ids: Option<Vec<i64>>, rev_ids: Option<Vec<i64>>,
conn: &SqliteConnection, conn: &SqliteConnection,
) -> Result<Vec<RevisionRecord>, FlowyError> { ) -> Result<Vec<SyncRecord>, FlowyError> {
let mut sql = dsl::rev_table.filter(dsl::doc_id.eq(object_id)).into_boxed(); let mut sql = dsl::rev_table.filter(dsl::doc_id.eq(object_id)).into_boxed();
if let Some(rev_ids) = rev_ids { if let Some(rev_ids) = rev_ids {
sql = sql.filter(dsl::rev_id.eq_any(rev_ids)); sql = sql.filter(dsl::rev_id.eq_any(rev_ids));
@ -162,7 +162,7 @@ impl DeltaRevisionSql {
object_id: &str, object_id: &str,
range: RevisionRange, range: RevisionRange,
conn: &SqliteConnection, conn: &SqliteConnection,
) -> Result<Vec<RevisionRecord>, FlowyError> { ) -> Result<Vec<SyncRecord>, FlowyError> {
let rev_tables = dsl::rev_table let rev_tables = dsl::rev_table
.filter(dsl::rev_id.ge(range.start)) .filter(dsl::rev_id.ge(range.start))
.filter(dsl::rev_id.le(range.end)) .filter(dsl::rev_id.le(range.end))
@ -244,7 +244,7 @@ impl std::default::Default for TextRevisionState {
} }
} }
fn mk_revision_record_from_table(user_id: &str, table: RevisionTable) -> RevisionRecord { fn mk_revision_record_from_table(user_id: &str, table: RevisionTable) -> SyncRecord {
let md5 = md5(&table.data); let md5 = md5(&table.data);
let revision = Revision::new( let revision = Revision::new(
&table.doc_id, &table.doc_id,
@ -254,7 +254,7 @@ fn mk_revision_record_from_table(user_id: &str, table: RevisionTable) -> Revisio
user_id, user_id,
md5, md5,
); );
RevisionRecord { SyncRecord {
revision, revision,
state: table.state.into(), state: table.state.into(),
write_to_disk: false, write_to_disk: false,

20
frontend/rust-lib/flowy-document/src/services/persistence/rev_sqlite/document_rev_sqlite_v1.rs
View File

@ -7,7 +7,7 @@ use flowy_database::{
ConnectionPool, ConnectionPool,
}; };
use flowy_error::{internal_error, FlowyError, FlowyResult}; use flowy_error::{internal_error, FlowyError, FlowyResult};
use flowy_revision::disk::{RevisionChangeset, RevisionDiskCache, RevisionRecord, RevisionState}; use flowy_revision::disk::{RevisionChangeset, RevisionDiskCache, RevisionState, SyncRecord};
use flowy_sync::{ use flowy_sync::{
entities::revision::{Revision, RevisionRange}, entities::revision::{Revision, RevisionRange},
util::md5, util::md5,
@ -22,7 +22,7 @@ pub struct SQLiteDocumentRevisionPersistence {
impl RevisionDiskCache<Arc<ConnectionPool>> for SQLiteDocumentRevisionPersistence { impl RevisionDiskCache<Arc<ConnectionPool>> for SQLiteDocumentRevisionPersistence {
type Error = FlowyError; type Error = FlowyError;
fn create_revision_records(&self, revision_records: Vec<RevisionRecord>) -> Result<(), Self::Error> { fn create_revision_records(&self, revision_records: Vec<SyncRecord>) -> Result<(), Self::Error> {
let conn = self.pool.get().map_err(internal_error)?; let conn = self.pool.get().map_err(internal_error)?;
let _ = DocumentRevisionSql::create(revision_records, &*conn)?; let _ = DocumentRevisionSql::create(revision_records, &*conn)?;
Ok(()) Ok(())
@ -36,7 +36,7 @@ impl RevisionDiskCache<Arc<ConnectionPool>> for SQLiteDocumentRevisionPersistenc
&self, &self,
object_id: &str, object_id: &str,
rev_ids: Option<Vec<i64>>, rev_ids: Option<Vec<i64>>,
) -> Result<Vec<RevisionRecord>, Self::Error> { ) -> Result<Vec<SyncRecord>, Self::Error> {
let conn = self.pool.get().map_err(internal_error)?; let conn = self.pool.get().map_err(internal_error)?;
let records = DocumentRevisionSql::read(&self.user_id, object_id, rev_ids, &*conn)?; let records = DocumentRevisionSql::read(&self.user_id, object_id, rev_ids, &*conn)?;
Ok(records) Ok(records)
@ -46,7 +46,7 @@ impl RevisionDiskCache<Arc<ConnectionPool>> for SQLiteDocumentRevisionPersistenc
&self, &self,
object_id: &str, object_id: &str,
range: &RevisionRange, range: &RevisionRange,
) -> Result<Vec<RevisionRecord>, Self::Error> { ) -> Result<Vec<SyncRecord>, Self::Error> {
let conn = &*self.pool.get().map_err(internal_error)?; let conn = &*self.pool.get().map_err(internal_error)?;
let revisions = DocumentRevisionSql::read_with_range(&self.user_id, object_id, range.clone(), conn)?; let revisions = DocumentRevisionSql::read_with_range(&self.user_id, object_id, range.clone(), conn)?;
Ok(revisions) Ok(revisions)
@ -73,7 +73,7 @@ impl RevisionDiskCache<Arc<ConnectionPool>> for SQLiteDocumentRevisionPersistenc
&self, &self,
object_id: &str, object_id: &str,
deleted_rev_ids: Option<Vec<i64>>, deleted_rev_ids: Option<Vec<i64>>,
inserted_records: Vec<RevisionRecord>, inserted_records: Vec<SyncRecord>,
) -> Result<(), Self::Error> { ) -> Result<(), Self::Error> {
let conn = self.pool.get().map_err(internal_error)?; let conn = self.pool.get().map_err(internal_error)?;
conn.immediate_transaction::<_, FlowyError, _>(|| { conn.immediate_transaction::<_, FlowyError, _>(|| {
@ -96,7 +96,7 @@ impl SQLiteDocumentRevisionPersistence {
struct DocumentRevisionSql {} struct DocumentRevisionSql {}
impl DocumentRevisionSql { impl DocumentRevisionSql {
fn create(revision_records: Vec<RevisionRecord>, conn: &SqliteConnection) -> Result<(), FlowyError> { fn create(revision_records: Vec<SyncRecord>, conn: &SqliteConnection) -> Result<(), FlowyError> {
// Batch insert: https://diesel.rs/guides/all-about-inserts.html // Batch insert: https://diesel.rs/guides/all-about-inserts.html
let records = revision_records let records = revision_records
.into_iter() .into_iter()
@ -142,7 +142,7 @@ impl DocumentRevisionSql {
object_id: &str, object_id: &str,
rev_ids: Option<Vec<i64>>, rev_ids: Option<Vec<i64>>,
conn: &SqliteConnection, conn: &SqliteConnection,
) -> Result<Vec<RevisionRecord>, FlowyError> { ) -> Result<Vec<SyncRecord>, FlowyError> {
let mut sql = dsl::document_rev_table let mut sql = dsl::document_rev_table
.filter(dsl::document_id.eq(object_id)) .filter(dsl::document_id.eq(object_id))
.into_boxed(); .into_boxed();
@ -163,7 +163,7 @@ impl DocumentRevisionSql {
object_id: &str, object_id: &str,
range: RevisionRange, range: RevisionRange,
conn: &SqliteConnection, conn: &SqliteConnection,
) -> Result<Vec<RevisionRecord>, FlowyError> { ) -> Result<Vec<SyncRecord>, FlowyError> {
let rev_tables = dsl::document_rev_table let rev_tables = dsl::document_rev_table
.filter(dsl::rev_id.ge(range.start)) .filter(dsl::rev_id.ge(range.start))
.filter(dsl::rev_id.le(range.end)) .filter(dsl::rev_id.le(range.end))
@ -220,7 +220,7 @@ impl std::default::Default for DocumentRevisionState {
} }
} }
fn mk_revision_record_from_table(user_id: &str, table: DocumentRevisionTable) -> RevisionRecord { fn mk_revision_record_from_table(user_id: &str, table: DocumentRevisionTable) -> SyncRecord {
let md5 = md5(&table.data); let md5 = md5(&table.data);
let revision = Revision::new( let revision = Revision::new(
&table.document_id, &table.document_id,
@ -230,7 +230,7 @@ fn mk_revision_record_from_table(user_id: &str, table: DocumentRevisionTable) ->
user_id, user_id,
md5, md5,
); );
RevisionRecord { SyncRecord {
revision, revision,
state: table.state.into(), state: table.state.into(),
write_to_disk: false, write_to_disk: false,

3
frontend/rust-lib/flowy-folder/src/services/persistence/migration.rs
View File

@ -9,11 +9,12 @@ use flowy_error::{FlowyError, FlowyResult};
use flowy_folder_data_model::revision::{AppRevision, FolderRevision, ViewRevision, WorkspaceRevision}; use flowy_folder_data_model::revision::{AppRevision, FolderRevision, ViewRevision, WorkspaceRevision};
use flowy_revision::reset::{RevisionResettable, RevisionStructReset}; use flowy_revision::reset::{RevisionResettable, RevisionStructReset};
use flowy_sync::client_folder::make_folder_rev_json_str; use flowy_sync::client_folder::make_folder_rev_json_str;
use flowy_sync::client_folder::FolderPad;
use flowy_sync::entities::revision::Revision; use flowy_sync::entities::revision::Revision;
use flowy_sync::server_folder::FolderOperationsBuilder; use flowy_sync::server_folder::FolderOperationsBuilder;
use flowy_sync::{client_folder::FolderPad, entities::revision::md5};
use crate::services::persistence::rev_sqlite::SQLiteFolderRevisionPersistence; use crate::services::persistence::rev_sqlite::SQLiteFolderRevisionPersistence;
use flowy_sync::util::md5;
use std::sync::Arc; use std::sync::Arc;
const V1_MIGRATION: &str = "FOLDER_V1_MIGRATION"; const V1_MIGRATION: &str = "FOLDER_V1_MIGRATION";

4
frontend/rust-lib/flowy-folder/src/services/persistence/mod.rs
View File

@ -11,7 +11,7 @@ use crate::{
use flowy_database::ConnectionPool; use flowy_database::ConnectionPool;
use flowy_error::{FlowyError, FlowyResult}; use flowy_error::{FlowyError, FlowyResult};
use flowy_folder_data_model::revision::{AppRevision, TrashRevision, ViewRevision, WorkspaceRevision}; use flowy_folder_data_model::revision::{AppRevision, TrashRevision, ViewRevision, WorkspaceRevision};
use flowy_revision::disk::{RevisionDiskCache, RevisionRecord, RevisionState}; use flowy_revision::disk::{RevisionDiskCache, RevisionState, SyncRecord};
use flowy_sync::{client_folder::FolderPad, entities::revision::Revision}; use flowy_sync::{client_folder::FolderPad, entities::revision::Revision};
use crate::services::persistence::rev_sqlite::SQLiteFolderRevisionPersistence; use crate::services::persistence::rev_sqlite::SQLiteFolderRevisionPersistence;
@ -112,7 +112,7 @@ impl FolderPersistence {
let json = folder.to_json()?; let json = folder.to_json()?;
let delta_data = FolderOperationsBuilder::new().insert(&json).build().json_bytes(); let delta_data = FolderOperationsBuilder::new().insert(&json).build().json_bytes();
let revision = Revision::initial_revision(user_id, folder_id.as_ref(), delta_data); let revision = Revision::initial_revision(user_id, folder_id.as_ref(), delta_data);
let record = RevisionRecord { let record = SyncRecord {
revision, revision,
state: RevisionState::Sync, state: RevisionState::Sync,
write_to_disk: true, write_to_disk: true,

20
frontend/rust-lib/flowy-folder/src/services/persistence/rev_sqlite/folder_rev_sqlite.rs
View File

@ -7,7 +7,7 @@ use flowy_database::{
ConnectionPool, ConnectionPool,
}; };
use flowy_error::{internal_error, FlowyError, FlowyResult}; use flowy_error::{internal_error, FlowyError, FlowyResult};
use flowy_revision::disk::{RevisionChangeset, RevisionDiskCache, RevisionRecord, RevisionState}; use flowy_revision::disk::{RevisionChangeset, RevisionDiskCache, RevisionState, SyncRecord};
use flowy_sync::{ use flowy_sync::{
entities::revision::{RevType, Revision, RevisionRange}, entities::revision::{RevType, Revision, RevisionRange},
util::md5, util::md5,
@ -23,7 +23,7 @@ pub struct SQLiteFolderRevisionPersistence {
impl RevisionDiskCache<Arc<ConnectionPool>> for SQLiteFolderRevisionPersistence { impl RevisionDiskCache<Arc<ConnectionPool>> for SQLiteFolderRevisionPersistence {
type Error = FlowyError; type Error = FlowyError;
fn create_revision_records(&self, revision_records: Vec<RevisionRecord>) -> Result<(), Self::Error> { fn create_revision_records(&self, revision_records: Vec<SyncRecord>) -> Result<(), Self::Error> {
let conn = self.pool.get().map_err(internal_error)?; let conn = self.pool.get().map_err(internal_error)?;
let _ = FolderRevisionSql::create(revision_records, &*conn)?; let _ = FolderRevisionSql::create(revision_records, &*conn)?;
Ok(()) Ok(())
@ -37,7 +37,7 @@ impl RevisionDiskCache<Arc<ConnectionPool>> for SQLiteFolderRevisionPersistence
&self, &self,
object_id: &str, object_id: &str,
rev_ids: Option<Vec<i64>>, rev_ids: Option<Vec<i64>>,
) -> Result<Vec<RevisionRecord>, Self::Error> { ) -> Result<Vec<SyncRecord>, Self::Error> {
let conn = self.pool.get().map_err(internal_error)?; let conn = self.pool.get().map_err(internal_error)?;
let records = FolderRevisionSql::read(&self.user_id, object_id, rev_ids, &*conn)?; let records = FolderRevisionSql::read(&self.user_id, object_id, rev_ids, &*conn)?;
Ok(records) Ok(records)
@ -47,7 +47,7 @@ impl RevisionDiskCache<Arc<ConnectionPool>> for SQLiteFolderRevisionPersistence
&self, &self,
object_id: &str, object_id: &str,
range: &RevisionRange, range: &RevisionRange,
) -> Result<Vec<RevisionRecord>, Self::Error> { ) -> Result<Vec<SyncRecord>, Self::Error> {
let conn = &*self.pool.get().map_err(internal_error)?; let conn = &*self.pool.get().map_err(internal_error)?;
let revisions = FolderRevisionSql::read_with_range(&self.user_id, object_id, range.clone(), conn)?; let revisions = FolderRevisionSql::read_with_range(&self.user_id, object_id, range.clone(), conn)?;
Ok(revisions) Ok(revisions)
@ -74,7 +74,7 @@ impl RevisionDiskCache<Arc<ConnectionPool>> for SQLiteFolderRevisionPersistence
&self, &self,
object_id: &str, object_id: &str,
deleted_rev_ids: Option<Vec<i64>>, deleted_rev_ids: Option<Vec<i64>>,
inserted_records: Vec<RevisionRecord>, inserted_records: Vec<SyncRecord>,
) -> Result<(), Self::Error> { ) -> Result<(), Self::Error> {
let conn = self.pool.get().map_err(internal_error)?; let conn = self.pool.get().map_err(internal_error)?;
conn.immediate_transaction::<_, FlowyError, _>(|| { conn.immediate_transaction::<_, FlowyError, _>(|| {
@ -97,7 +97,7 @@ impl SQLiteFolderRevisionPersistence {
struct FolderRevisionSql {} struct FolderRevisionSql {}
impl FolderRevisionSql { impl FolderRevisionSql {
fn create(revision_records: Vec<RevisionRecord>, conn: &SqliteConnection) -> Result<(), FlowyError> { fn create(revision_records: Vec<SyncRecord>, conn: &SqliteConnection) -> Result<(), FlowyError> {
// Batch insert: https://diesel.rs/guides/all-about-inserts.html // Batch insert: https://diesel.rs/guides/all-about-inserts.html
let records = revision_records let records = revision_records
@ -143,7 +143,7 @@ impl FolderRevisionSql {
object_id: &str, object_id: &str,
rev_ids: Option<Vec<i64>>, rev_ids: Option<Vec<i64>>,
conn: &SqliteConnection, conn: &SqliteConnection,
) -> Result<Vec<RevisionRecord>, FlowyError> { ) -> Result<Vec<SyncRecord>, FlowyError> {
let mut sql = dsl::rev_table.filter(dsl::doc_id.eq(object_id)).into_boxed(); let mut sql = dsl::rev_table.filter(dsl::doc_id.eq(object_id)).into_boxed();
if let Some(rev_ids) = rev_ids { if let Some(rev_ids) = rev_ids {
sql = sql.filter(dsl::rev_id.eq_any(rev_ids)); sql = sql.filter(dsl::rev_id.eq_any(rev_ids));
@ -162,7 +162,7 @@ impl FolderRevisionSql {
object_id: &str, object_id: &str,
range: RevisionRange, range: RevisionRange,
conn: &SqliteConnection, conn: &SqliteConnection,
) -> Result<Vec<RevisionRecord>, FlowyError> { ) -> Result<Vec<SyncRecord>, FlowyError> {
let rev_tables = dsl::rev_table let rev_tables = dsl::rev_table
.filter(dsl::rev_id.ge(range.start)) .filter(dsl::rev_id.ge(range.start))
.filter(dsl::rev_id.le(range.end)) .filter(dsl::rev_id.le(range.end))
@ -220,7 +220,7 @@ impl std::default::Default for TextRevisionState {
} }
} }
fn mk_revision_record_from_table(user_id: &str, table: RevisionTable) -> RevisionRecord { fn mk_revision_record_from_table(user_id: &str, table: RevisionTable) -> SyncRecord {
let md5 = md5(&table.data); let md5 = md5(&table.data);
let revision = Revision::new( let revision = Revision::new(
&table.doc_id, &table.doc_id,
@ -230,7 +230,7 @@ fn mk_revision_record_from_table(user_id: &str, table: RevisionTable) -> Revisio
user_id, user_id,
md5, md5,
); );
RevisionRecord { SyncRecord {
revision, revision,
state: table.state.into(), state: table.state.into(),
write_to_disk: false, write_to_disk: false,

8
frontend/rust-lib/flowy-folder/src/services/web_socket.rs
View File

@ -78,12 +78,12 @@ struct FolderConflictResolver {
} }
impl ConflictResolver<FolderResolveOperations> for FolderConflictResolver { impl ConflictResolver<FolderResolveOperations> for FolderConflictResolver {
fn compose_operations(&self, operations: FolderResolveOperations) -> BoxResultFuture<OperationsMD5, FlowyError> { fn compose_operations(&self, operations: FolderResolveOperations) -> BoxResultFuture<RevisionMD5, FlowyError> {
let operations = operations.into_inner(); let operations = operations.into_inner();
let folder_pad = self.folder_pad.clone(); let folder_pad = self.folder_pad.clone();
Box::pin(async move { Box::pin(async move {
let md5 = folder_pad.write().compose_remote_operations(operations)?; let md5 = folder_pad.write().compose_remote_operations(operations)?;
Ok(md5) Ok(md5.into())
}) })
} }
@ -113,11 +113,11 @@ impl ConflictResolver<FolderResolveOperations> for FolderConflictResolver {
}) })
} }
fn reset_operations(&self, operations: FolderResolveOperations) -> BoxResultFuture<OperationsMD5, FlowyError> { fn reset_operations(&self, operations: FolderResolveOperations) -> BoxResultFuture<RevisionMD5, FlowyError> {
let folder_pad = self.folder_pad.clone(); let folder_pad = self.folder_pad.clone();
Box::pin(async move { Box::pin(async move {
let md5 = folder_pad.write().reset_folder(operations.into_inner())?; let md5 = folder_pad.write().reset_folder(operations.into_inner())?;
Ok(md5) Ok(md5.into())
}) })
} }
} }

23
frontend/rust-lib/flowy-grid/src/manager.rs
View File

@ -15,7 +15,7 @@ use flowy_error::{FlowyError, FlowyResult};
use flowy_grid_data_model::revision::{BuildGridContext, GridRevision, GridViewRevision}; use flowy_grid_data_model::revision::{BuildGridContext, GridRevision, GridViewRevision};
use flowy_revision::{RevisionManager, RevisionPersistence, RevisionWebSocket, SQLiteRevisionSnapshotPersistence}; use flowy_revision::{RevisionManager, RevisionPersistence, RevisionWebSocket, SQLiteRevisionSnapshotPersistence};
use flowy_sync::client_grid::{make_grid_block_operations, make_grid_operations, make_grid_view_operations}; use flowy_sync::client_grid::{make_grid_block_operations, make_grid_operations, make_grid_view_operations};
use flowy_sync::entities::revision::{RepeatedRevision, Revision}; use flowy_sync::entities::revision::Revision;
use std::sync::Arc; use std::sync::Arc;
use tokio::sync::RwLock; use tokio::sync::RwLock;
@ -67,7 +67,7 @@ impl GridManager {
} }
#[tracing::instrument(level = "debug", skip_all, err)] #[tracing::instrument(level = "debug", skip_all, err)]
pub async fn create_grid<T: AsRef<str>>(&self, grid_id: T, revisions: RepeatedRevision) -> FlowyResult<()> { pub async fn create_grid<T: AsRef<str>>(&self, grid_id: T, revisions: Vec<Revision>) -> FlowyResult<()> {
let grid_id = grid_id.as_ref(); let grid_id = grid_id.as_ref();
let db_pool = self.grid_user.db_pool()?; let db_pool = self.grid_user.db_pool()?;
let rev_manager = self.make_grid_rev_manager(grid_id, db_pool)?; let rev_manager = self.make_grid_rev_manager(grid_id, db_pool)?;
@ -77,7 +77,7 @@ impl GridManager {
} }
#[tracing::instrument(level = "debug", skip_all, err)] #[tracing::instrument(level = "debug", skip_all, err)]
async fn create_grid_view<T: AsRef<str>>(&self, view_id: T, revisions: RepeatedRevision) -> FlowyResult<()> { async fn create_grid_view<T: AsRef<str>>(&self, view_id: T, revisions: Vec<Revision>) -> FlowyResult<()> {
let view_id = view_id.as_ref(); let view_id = view_id.as_ref();
let rev_manager = make_grid_view_rev_manager(&self.grid_user, view_id).await?; let rev_manager = make_grid_view_rev_manager(&self.grid_user, view_id).await?;
let _ = rev_manager.reset_object(revisions).await?; let _ = rev_manager.reset_object(revisions).await?;
@ -85,7 +85,7 @@ impl GridManager {
} }
#[tracing::instrument(level = "debug", skip_all, err)] #[tracing::instrument(level = "debug", skip_all, err)]
pub async fn create_grid_block<T: AsRef<str>>(&self, block_id: T, revisions: RepeatedRevision) -> FlowyResult<()> { pub async fn create_grid_block<T: AsRef<str>>(&self, block_id: T, revisions: Vec<Revision>) -> FlowyResult<()> {
let block_id = block_id.as_ref(); let block_id = block_id.as_ref();
let db_pool = self.grid_user.db_pool()?; let db_pool = self.grid_user.db_pool()?;
let rev_manager = self.make_grid_block_rev_manager(block_id, db_pool)?; let rev_manager = self.make_grid_block_rev_manager(block_id, db_pool)?;
@ -208,9 +208,8 @@ pub async fn make_grid_view_data(
// Create grid's block // Create grid's block
let grid_block_delta = make_grid_block_operations(block_meta_data); let grid_block_delta = make_grid_block_operations(block_meta_data);
let block_delta_data = grid_block_delta.json_bytes(); let block_delta_data = grid_block_delta.json_bytes();
let repeated_revision: RepeatedRevision = let revision = Revision::initial_revision(user_id, block_id, block_delta_data);
Revision::initial_revision(user_id, block_id, block_delta_data).into(); let _ = grid_manager.create_grid_block(&block_id, vec![revision]).await?;
let _ = grid_manager.create_grid_block(&block_id, repeated_revision).await?;
} }
// Will replace the grid_id with the value returned by the gen_grid_id() // Will replace the grid_id with the value returned by the gen_grid_id()
@ -220,9 +219,8 @@ pub async fn make_grid_view_data(
// Create grid // Create grid
let grid_rev_delta = make_grid_operations(&grid_rev); let grid_rev_delta = make_grid_operations(&grid_rev);
let grid_rev_delta_bytes = grid_rev_delta.json_bytes(); let grid_rev_delta_bytes = grid_rev_delta.json_bytes();
let repeated_revision: RepeatedRevision = let revision = Revision::initial_revision(user_id, &grid_id, grid_rev_delta_bytes.clone());
Revision::initial_revision(user_id, &grid_id, grid_rev_delta_bytes.clone()).into(); let _ = grid_manager.create_grid(&grid_id, vec![revision]).await?;
let _ = grid_manager.create_grid(&grid_id, repeated_revision).await?;
// Create grid view // Create grid view
let grid_view = if grid_view_revision_data.is_empty() { let grid_view = if grid_view_revision_data.is_empty() {
@ -232,9 +230,8 @@ pub async fn make_grid_view_data(
}; };
let grid_view_delta = make_grid_view_operations(&grid_view); let grid_view_delta = make_grid_view_operations(&grid_view);
let grid_view_delta_bytes = grid_view_delta.json_bytes(); let grid_view_delta_bytes = grid_view_delta.json_bytes();
let repeated_revision: RepeatedRevision = let revision = Revision::initial_revision(user_id, view_id, grid_view_delta_bytes);
Revision::initial_revision(user_id, view_id, grid_view_delta_bytes).into(); let _ = grid_manager.create_grid_view(view_id, vec![revision]).await?;
let _ = grid_manager.create_grid_view(view_id, repeated_revision).await?;
Ok(grid_rev_delta_bytes) Ok(grid_rev_delta_bytes)
} }

20
frontend/rust-lib/flowy-grid/src/services/persistence/rev_sqlite/grid_block_impl.rs
View File

@ -7,7 +7,7 @@ use flowy_database::{
ConnectionPool, ConnectionPool,
}; };
use flowy_error::{internal_error, FlowyError, FlowyResult}; use flowy_error::{internal_error, FlowyError, FlowyResult};
use flowy_revision::disk::{RevisionChangeset, RevisionDiskCache, RevisionRecord, RevisionState}; use flowy_revision::disk::{RevisionChangeset, RevisionDiskCache, RevisionState, SyncRecord};
use flowy_sync::{ use flowy_sync::{
entities::revision::{Revision, RevisionRange}, entities::revision::{Revision, RevisionRange},
util::md5, util::md5,
@ -22,7 +22,7 @@ pub struct SQLiteGridBlockRevisionPersistence {
impl RevisionDiskCache<Arc<ConnectionPool>> for SQLiteGridBlockRevisionPersistence { impl RevisionDiskCache<Arc<ConnectionPool>> for SQLiteGridBlockRevisionPersistence {
type Error = FlowyError; type Error = FlowyError;
fn create_revision_records(&self, revision_records: Vec<RevisionRecord>) -> Result<(), Self::Error> { fn create_revision_records(&self, revision_records: Vec<SyncRecord>) -> Result<(), Self::Error> {
let conn = self.pool.get().map_err(internal_error)?; let conn = self.pool.get().map_err(internal_error)?;
let _ = GridMetaRevisionSql::create(revision_records, &*conn)?; let _ = GridMetaRevisionSql::create(revision_records, &*conn)?;
Ok(()) Ok(())
@ -36,7 +36,7 @@ impl RevisionDiskCache<Arc<ConnectionPool>> for SQLiteGridBlockRevisionPersisten
&self, &self,
object_id: &str, object_id: &str,
rev_ids: Option<Vec<i64>>, rev_ids: Option<Vec<i64>>,
) -> Result<Vec<RevisionRecord>, Self::Error> { ) -> Result<Vec<SyncRecord>, Self::Error> {
let conn = self.pool.get().map_err(internal_error)?; let conn = self.pool.get().map_err(internal_error)?;
let records = GridMetaRevisionSql::read(&self.user_id, object_id, rev_ids, &*conn)?; let records = GridMetaRevisionSql::read(&self.user_id, object_id, rev_ids, &*conn)?;
Ok(records) Ok(records)
@ -46,7 +46,7 @@ impl RevisionDiskCache<Arc<ConnectionPool>> for SQLiteGridBlockRevisionPersisten
&self, &self,
object_id: &str, object_id: &str,
range: &RevisionRange, range: &RevisionRange,
) -> Result<Vec<RevisionRecord>, Self::Error> { ) -> Result<Vec<SyncRecord>, Self::Error> {
let conn = &*self.pool.get().map_err(internal_error)?; let conn = &*self.pool.get().map_err(internal_error)?;
let revisions = GridMetaRevisionSql::read_with_range(&self.user_id, object_id, range.clone(), conn)?; let revisions = GridMetaRevisionSql::read_with_range(&self.user_id, object_id, range.clone(), conn)?;
Ok(revisions) Ok(revisions)
@ -73,7 +73,7 @@ impl RevisionDiskCache<Arc<ConnectionPool>> for SQLiteGridBlockRevisionPersisten
&self, &self,
object_id: &str, object_id: &str,
deleted_rev_ids: Option<Vec<i64>>, deleted_rev_ids: Option<Vec<i64>>,
inserted_records: Vec<RevisionRecord>, inserted_records: Vec<SyncRecord>,
) -> Result<(), Self::Error> { ) -> Result<(), Self::Error> {
let conn = self.pool.get().map_err(internal_error)?; let conn = self.pool.get().map_err(internal_error)?;
conn.immediate_transaction::<_, FlowyError, _>(|| { conn.immediate_transaction::<_, FlowyError, _>(|| {
@ -95,7 +95,7 @@ impl SQLiteGridBlockRevisionPersistence {
struct GridMetaRevisionSql(); struct GridMetaRevisionSql();
impl GridMetaRevisionSql { impl GridMetaRevisionSql {
fn create(revision_records: Vec<RevisionRecord>, conn: &SqliteConnection) -> Result<(), FlowyError> { fn create(revision_records: Vec<SyncRecord>, conn: &SqliteConnection) -> Result<(), FlowyError> {
// Batch insert: https://diesel.rs/guides/all-about-inserts.html // Batch insert: https://diesel.rs/guides/all-about-inserts.html
let records = revision_records let records = revision_records
@ -142,7 +142,7 @@ impl GridMetaRevisionSql {
object_id: &str, object_id: &str,
rev_ids: Option<Vec<i64>>, rev_ids: Option<Vec<i64>>,
conn: &SqliteConnection, conn: &SqliteConnection,
) -> Result<Vec<RevisionRecord>, FlowyError> { ) -> Result<Vec<SyncRecord>, FlowyError> {
let mut sql = dsl::grid_meta_rev_table let mut sql = dsl::grid_meta_rev_table
.filter(dsl::object_id.eq(object_id)) .filter(dsl::object_id.eq(object_id))
.into_boxed(); .into_boxed();
@ -163,7 +163,7 @@ impl GridMetaRevisionSql {
object_id: &str, object_id: &str,
range: RevisionRange, range: RevisionRange,
conn: &SqliteConnection, conn: &SqliteConnection,
) -> Result<Vec<RevisionRecord>, FlowyError> { ) -> Result<Vec<SyncRecord>, FlowyError> {
let rev_tables = dsl::grid_meta_rev_table let rev_tables = dsl::grid_meta_rev_table
.filter(dsl::rev_id.ge(range.start)) .filter(dsl::rev_id.ge(range.start))
.filter(dsl::rev_id.le(range.end)) .filter(dsl::rev_id.le(range.end))
@ -219,7 +219,7 @@ impl std::default::Default for GridBlockRevisionState {
} }
} }
fn mk_revision_record_from_table(user_id: &str, table: GridBlockRevisionTable) -> RevisionRecord { fn mk_revision_record_from_table(user_id: &str, table: GridBlockRevisionTable) -> SyncRecord {
let md5 = md5(&table.data); let md5 = md5(&table.data);
let revision = Revision::new( let revision = Revision::new(
&table.object_id, &table.object_id,
@ -229,7 +229,7 @@ fn mk_revision_record_from_table(user_id: &str, table: GridBlockRevisionTable) -
user_id, user_id,
md5, md5,
); );
RevisionRecord { SyncRecord {
revision, revision,
state: table.state.into(), state: table.state.into(),
write_to_disk: false, write_to_disk: false,

20
frontend/rust-lib/flowy-grid/src/services/persistence/rev_sqlite/grid_impl.rs
View File

@ -7,7 +7,7 @@ use flowy_database::{
ConnectionPool, ConnectionPool,
}; };
use flowy_error::{internal_error, FlowyError, FlowyResult}; use flowy_error::{internal_error, FlowyError, FlowyResult};
use flowy_revision::disk::{RevisionChangeset, RevisionDiskCache, RevisionRecord, RevisionState}; use flowy_revision::disk::{RevisionChangeset, RevisionDiskCache, RevisionState, SyncRecord};
use flowy_sync::{ use flowy_sync::{
entities::revision::{Revision, RevisionRange}, entities::revision::{Revision, RevisionRange},
util::md5, util::md5,
@ -22,7 +22,7 @@ pub struct SQLiteGridRevisionPersistence {
impl RevisionDiskCache<Arc<ConnectionPool>> for SQLiteGridRevisionPersistence { impl RevisionDiskCache<Arc<ConnectionPool>> for SQLiteGridRevisionPersistence {
type Error = FlowyError; type Error = FlowyError;
fn create_revision_records(&self, revision_records: Vec<RevisionRecord>) -> Result<(), Self::Error> { fn create_revision_records(&self, revision_records: Vec<SyncRecord>) -> Result<(), Self::Error> {
let conn = self.pool.get().map_err(internal_error)?; let conn = self.pool.get().map_err(internal_error)?;
let _ = GridRevisionSql::create(revision_records, &*conn)?; let _ = GridRevisionSql::create(revision_records, &*conn)?;
Ok(()) Ok(())
@ -36,7 +36,7 @@ impl RevisionDiskCache<Arc<ConnectionPool>> for SQLiteGridRevisionPersistence {
&self, &self,
object_id: &str, object_id: &str,
rev_ids: Option<Vec<i64>>, rev_ids: Option<Vec<i64>>,
) -> Result<Vec<RevisionRecord>, Self::Error> { ) -> Result<Vec<SyncRecord>, Self::Error> {
let conn = self.pool.get().map_err(internal_error)?; let conn = self.pool.get().map_err(internal_error)?;
let records = GridRevisionSql::read(&self.user_id, object_id, rev_ids, &*conn)?; let records = GridRevisionSql::read(&self.user_id, object_id, rev_ids, &*conn)?;
Ok(records) Ok(records)
@ -46,7 +46,7 @@ impl RevisionDiskCache<Arc<ConnectionPool>> for SQLiteGridRevisionPersistence {
&self, &self,
object_id: &str, object_id: &str,
range: &RevisionRange, range: &RevisionRange,
) -> Result<Vec<RevisionRecord>, Self::Error> { ) -> Result<Vec<SyncRecord>, Self::Error> {
let conn = &*self.pool.get().map_err(internal_error)?; let conn = &*self.pool.get().map_err(internal_error)?;
let revisions = GridRevisionSql::read_with_range(&self.user_id, object_id, range.clone(), conn)?; let revisions = GridRevisionSql::read_with_range(&self.user_id, object_id, range.clone(), conn)?;
Ok(revisions) Ok(revisions)
@ -73,7 +73,7 @@ impl RevisionDiskCache<Arc<ConnectionPool>> for SQLiteGridRevisionPersistence {
&self, &self,
object_id: &str, object_id: &str,
deleted_rev_ids: Option<Vec<i64>>, deleted_rev_ids: Option<Vec<i64>>,
inserted_records: Vec<RevisionRecord>, inserted_records: Vec<SyncRecord>,
) -> Result<(), Self::Error> { ) -> Result<(), Self::Error> {
let conn = self.pool.get().map_err(internal_error)?; let conn = self.pool.get().map_err(internal_error)?;
conn.immediate_transaction::<_, FlowyError, _>(|| { conn.immediate_transaction::<_, FlowyError, _>(|| {
@ -95,7 +95,7 @@ impl SQLiteGridRevisionPersistence {
struct GridRevisionSql(); struct GridRevisionSql();
impl GridRevisionSql { impl GridRevisionSql {
fn create(revision_records: Vec<RevisionRecord>, conn: &SqliteConnection) -> Result<(), FlowyError> { fn create(revision_records: Vec<SyncRecord>, conn: &SqliteConnection) -> Result<(), FlowyError> {
// Batch insert: https://diesel.rs/guides/all-about-inserts.html // Batch insert: https://diesel.rs/guides/all-about-inserts.html
let records = revision_records let records = revision_records
.into_iter() .into_iter()
@ -141,7 +141,7 @@ impl GridRevisionSql {
object_id: &str, object_id: &str,
rev_ids: Option<Vec<i64>>, rev_ids: Option<Vec<i64>>,
conn: &SqliteConnection, conn: &SqliteConnection,
) -> Result<Vec<RevisionRecord>, FlowyError> { ) -> Result<Vec<SyncRecord>, FlowyError> {
let mut sql = dsl::grid_rev_table.filter(dsl::object_id.eq(object_id)).into_boxed(); let mut sql = dsl::grid_rev_table.filter(dsl::object_id.eq(object_id)).into_boxed();
if let Some(rev_ids) = rev_ids { if let Some(rev_ids) = rev_ids {
sql = sql.filter(dsl::rev_id.eq_any(rev_ids)); sql = sql.filter(dsl::rev_id.eq_any(rev_ids));
@ -160,7 +160,7 @@ impl GridRevisionSql {
object_id: &str, object_id: &str,
range: RevisionRange, range: RevisionRange,
conn: &SqliteConnection, conn: &SqliteConnection,
) -> Result<Vec<RevisionRecord>, FlowyError> { ) -> Result<Vec<SyncRecord>, FlowyError> {
let rev_tables = dsl::grid_rev_table let rev_tables = dsl::grid_rev_table
.filter(dsl::rev_id.ge(range.start)) .filter(dsl::rev_id.ge(range.start))
.filter(dsl::rev_id.le(range.end)) .filter(dsl::rev_id.le(range.end))
@ -217,7 +217,7 @@ impl std::default::Default for GridRevisionState {
} }
} }
fn mk_revision_record_from_table(user_id: &str, table: GridRevisionTable) -> RevisionRecord { fn mk_revision_record_from_table(user_id: &str, table: GridRevisionTable) -> SyncRecord {
let md5 = md5(&table.data); let md5 = md5(&table.data);
let revision = Revision::new( let revision = Revision::new(
&table.object_id, &table.object_id,
@ -227,7 +227,7 @@ fn mk_revision_record_from_table(user_id: &str, table: GridRevisionTable) -> Rev
user_id, user_id,
md5, md5,
); );
RevisionRecord { SyncRecord {
revision, revision,
state: table.state.into(), state: table.state.into(),
write_to_disk: false, write_to_disk: false,

20
frontend/rust-lib/flowy-grid/src/services/persistence/rev_sqlite/grid_view_impl.rs
View File

@ -7,7 +7,7 @@ use flowy_database::{
ConnectionPool, ConnectionPool,
}; };
use flowy_error::{internal_error, FlowyError, FlowyResult}; use flowy_error::{internal_error, FlowyError, FlowyResult};
use flowy_revision::disk::{RevisionChangeset, RevisionDiskCache, RevisionRecord, RevisionState}; use flowy_revision::disk::{RevisionChangeset, RevisionDiskCache, RevisionState, SyncRecord};
use flowy_sync::{ use flowy_sync::{
entities::revision::{Revision, RevisionRange}, entities::revision::{Revision, RevisionRange},
util::md5, util::md5,
@ -31,7 +31,7 @@ impl SQLiteGridViewRevisionPersistence {
impl RevisionDiskCache<Arc<ConnectionPool>> for SQLiteGridViewRevisionPersistence { impl RevisionDiskCache<Arc<ConnectionPool>> for SQLiteGridViewRevisionPersistence {
type Error = FlowyError; type Error = FlowyError;
fn create_revision_records(&self, revision_records: Vec<RevisionRecord>) -> Result<(), Self::Error> { fn create_revision_records(&self, revision_records: Vec<SyncRecord>) -> Result<(), Self::Error> {
let conn = self.pool.get().map_err(internal_error)?; let conn = self.pool.get().map_err(internal_error)?;
let _ = GridViewRevisionSql::create(revision_records, &*conn)?; let _ = GridViewRevisionSql::create(revision_records, &*conn)?;
Ok(()) Ok(())
@ -45,7 +45,7 @@ impl RevisionDiskCache<Arc<ConnectionPool>> for SQLiteGridViewRevisionPersistenc
&self, &self,
object_id: &str, object_id: &str,
rev_ids: Option<Vec<i64>>, rev_ids: Option<Vec<i64>>,
) -> Result<Vec<RevisionRecord>, Self::Error> { ) -> Result<Vec<SyncRecord>, Self::Error> {
let conn = self.pool.get().map_err(internal_error)?; let conn = self.pool.get().map_err(internal_error)?;
let records = GridViewRevisionSql::read(&self.user_id, object_id, rev_ids, &*conn)?; let records = GridViewRevisionSql::read(&self.user_id, object_id, rev_ids, &*conn)?;
Ok(records) Ok(records)
@ -55,7 +55,7 @@ impl RevisionDiskCache<Arc<ConnectionPool>> for SQLiteGridViewRevisionPersistenc
&self, &self,
object_id: &str, object_id: &str,
range: &RevisionRange, range: &RevisionRange,
) -> Result<Vec<RevisionRecord>, Self::Error> { ) -> Result<Vec<SyncRecord>, Self::Error> {
let conn = &*self.pool.get().map_err(internal_error)?; let conn = &*self.pool.get().map_err(internal_error)?;
let revisions = GridViewRevisionSql::read_with_range(&self.user_id, object_id, range.clone(), conn)?; let revisions = GridViewRevisionSql::read_with_range(&self.user_id, object_id, range.clone(), conn)?;
Ok(revisions) Ok(revisions)
@ -82,7 +82,7 @@ impl RevisionDiskCache<Arc<ConnectionPool>> for SQLiteGridViewRevisionPersistenc
&self, &self,
object_id: &str, object_id: &str,
deleted_rev_ids: Option<Vec<i64>>, deleted_rev_ids: Option<Vec<i64>>,
inserted_records: Vec<RevisionRecord>, inserted_records: Vec<SyncRecord>,
) -> Result<(), Self::Error> { ) -> Result<(), Self::Error> {
let conn = self.pool.get().map_err(internal_error)?; let conn = self.pool.get().map_err(internal_error)?;
conn.immediate_transaction::<_, FlowyError, _>(|| { conn.immediate_transaction::<_, FlowyError, _>(|| {
@ -95,7 +95,7 @@ impl RevisionDiskCache<Arc<ConnectionPool>> for SQLiteGridViewRevisionPersistenc
struct GridViewRevisionSql(); struct GridViewRevisionSql();
impl GridViewRevisionSql { impl GridViewRevisionSql {
fn create(revision_records: Vec<RevisionRecord>, conn: &SqliteConnection) -> Result<(), FlowyError> { fn create(revision_records: Vec<SyncRecord>, conn: &SqliteConnection) -> Result<(), FlowyError> {
// Batch insert: https://diesel.rs/guides/all-about-inserts.html // Batch insert: https://diesel.rs/guides/all-about-inserts.html
let records = revision_records let records = revision_records
.into_iter() .into_iter()
@ -141,7 +141,7 @@ impl GridViewRevisionSql {
object_id: &str, object_id: &str,
rev_ids: Option<Vec<i64>>, rev_ids: Option<Vec<i64>>,
conn: &SqliteConnection, conn: &SqliteConnection,
) -> Result<Vec<RevisionRecord>, FlowyError> { ) -> Result<Vec<SyncRecord>, FlowyError> {
let mut sql = dsl::grid_view_rev_table let mut sql = dsl::grid_view_rev_table
.filter(dsl::object_id.eq(object_id)) .filter(dsl::object_id.eq(object_id))
.into_boxed(); .into_boxed();
@ -162,7 +162,7 @@ impl GridViewRevisionSql {
object_id: &str, object_id: &str,
range: RevisionRange, range: RevisionRange,
conn: &SqliteConnection, conn: &SqliteConnection,
) -> Result<Vec<RevisionRecord>, FlowyError> { ) -> Result<Vec<SyncRecord>, FlowyError> {
let rev_tables = dsl::grid_view_rev_table let rev_tables = dsl::grid_view_rev_table
.filter(dsl::rev_id.ge(range.start)) .filter(dsl::rev_id.ge(range.start))
.filter(dsl::rev_id.le(range.end)) .filter(dsl::rev_id.le(range.end))
@ -219,7 +219,7 @@ impl std::default::Default for GridViewRevisionState {
} }
} }
fn mk_revision_record_from_table(user_id: &str, table: GridViewRevisionTable) -> RevisionRecord { fn mk_revision_record_from_table(user_id: &str, table: GridViewRevisionTable) -> SyncRecord {
let md5 = md5(&table.data); let md5 = md5(&table.data);
let revision = Revision::new( let revision = Revision::new(
&table.object_id, &table.object_id,
@ -229,7 +229,7 @@ fn mk_revision_record_from_table(user_id: &str, table: GridViewRevisionTable) ->
user_id, user_id,
md5, md5,
); );
RevisionRecord { SyncRecord {
revision, revision,
state: table.state.into(), state: table.state.into(),
write_to_disk: false, write_to_disk: false,

25
frontend/rust-lib/flowy-revision/src/cache/disk.rs vendored
View File

@ -5,23 +5,20 @@ use std::sync::Arc;
pub trait RevisionDiskCache<Connection>: Sync + Send { pub trait RevisionDiskCache<Connection>: Sync + Send {
type Error: Debug; type Error: Debug;
fn create_revision_records(&self, revision_records: Vec<RevisionRecord>) -> Result<(), Self::Error>; fn create_revision_records(&self, revision_records: Vec<SyncRecord>) -> Result<(), Self::Error>;
fn get_connection(&self) -> Result<Connection, Self::Error>; fn get_connection(&self) -> Result<Connection, Self::Error>;
// Read all the records if the rev_ids is None // Read all the records if the rev_ids is None
fn read_revision_records( fn read_revision_records(&self, object_id: &str, rev_ids: Option<Vec<i64>>)
&self, -> Result<Vec<SyncRecord>, Self::Error>;
object_id: &str,
rev_ids: Option<Vec<i64>>,
) -> Result<Vec<RevisionRecord>, Self::Error>;
// Read the revision which rev_id >= range.start && rev_id <= range.end // Read the revision which rev_id >= range.start && rev_id <= range.end
fn read_revision_records_with_range( fn read_revision_records_with_range(
&self, &self,
object_id: &str, object_id: &str,
range: &RevisionRange, range: &RevisionRange,
) -> Result<Vec<RevisionRecord>, Self::Error>; ) -> Result<Vec<SyncRecord>, Self::Error>;
fn update_revision_record(&self, changesets: Vec<RevisionChangeset>) -> FlowyResult<()>; fn update_revision_record(&self, changesets: Vec<RevisionChangeset>) -> FlowyResult<()>;
@ -34,7 +31,7 @@ pub trait RevisionDiskCache<Connection>: Sync + Send {
&self, &self,
object_id: &str, object_id: &str,
deleted_rev_ids: Option<Vec<i64>>, deleted_rev_ids: Option<Vec<i64>>,
inserted_records: Vec<RevisionRecord>, inserted_records: Vec<SyncRecord>,
) -> Result<(), Self::Error>; ) -> Result<(), Self::Error>;
} }
@ -44,7 +41,7 @@ where
{ {
type Error = FlowyError; type Error = FlowyError;
fn create_revision_records(&self, revision_records: Vec<RevisionRecord>) -> Result<(), Self::Error> { fn create_revision_records(&self, revision_records: Vec<SyncRecord>) -> Result<(), Self::Error> {
(**self).create_revision_records(revision_records) (**self).create_revision_records(revision_records)
} }
@ -56,7 +53,7 @@ where
&self, &self,
object_id: &str, object_id: &str,
rev_ids: Option<Vec<i64>>, rev_ids: Option<Vec<i64>>,
) -> Result<Vec<RevisionRecord>, Self::Error> { ) -> Result<Vec<SyncRecord>, Self::Error> {
(**self).read_revision_records(object_id, rev_ids) (**self).read_revision_records(object_id, rev_ids)
} }
@ -64,7 +61,7 @@ where
&self, &self,
object_id: &str, object_id: &str,
range: &RevisionRange, range: &RevisionRange,
) -> Result<Vec<RevisionRecord>, Self::Error> { ) -> Result<Vec<SyncRecord>, Self::Error> {
(**self).read_revision_records_with_range(object_id, range) (**self).read_revision_records_with_range(object_id, range)
} }
@ -80,20 +77,20 @@ where
&self, &self,
object_id: &str, object_id: &str,
deleted_rev_ids: Option<Vec<i64>>, deleted_rev_ids: Option<Vec<i64>>,
inserted_records: Vec<RevisionRecord>, inserted_records: Vec<SyncRecord>,
) -> Result<(), Self::Error> { ) -> Result<(), Self::Error> {
(**self).delete_and_insert_records(object_id, deleted_rev_ids, inserted_records) (**self).delete_and_insert_records(object_id, deleted_rev_ids, inserted_records)
} }
} }
#[derive(Clone, Debug)] #[derive(Clone, Debug)]
pub struct RevisionRecord { pub struct SyncRecord {
pub revision: Revision, pub revision: Revision,
pub state: RevisionState, pub state: RevisionState,
pub write_to_disk: bool, pub write_to_disk: bool,
} }
impl RevisionRecord { impl SyncRecord {
pub fn new(revision: Revision) -> Self { pub fn new(revision: Revision) -> Self {
Self { Self {
revision, revision,

42
frontend/rust-lib/flowy-revision/src/cache/memory.rs vendored
View File

@ -1,4 +1,4 @@
use crate::disk::RevisionRecord; use crate::disk::SyncRecord;
use crate::REVISION_WRITE_INTERVAL_IN_MILLIS; use crate::REVISION_WRITE_INTERVAL_IN_MILLIS;
use dashmap::DashMap; use dashmap::DashMap;
use flowy_error::{FlowyError, FlowyResult}; use flowy_error::{FlowyError, FlowyResult};
@ -7,15 +7,15 @@ use std::{borrow::Cow, sync::Arc, time::Duration};
use tokio::{sync::RwLock, task::JoinHandle}; use tokio::{sync::RwLock, task::JoinHandle};
pub(crate) trait RevisionMemoryCacheDelegate: Send + Sync { pub(crate) trait RevisionMemoryCacheDelegate: Send + Sync {
fn checkpoint_tick(&self, records: Vec<RevisionRecord>) -> FlowyResult<()>; fn send_sync(&self, records: Vec<SyncRecord>) -> FlowyResult<()>;
fn receive_ack(&self, object_id: &str, rev_id: i64); fn receive_ack(&self, object_id: &str, rev_id: i64);
} }
pub(crate) struct RevisionMemoryCache { pub(crate) struct RevisionMemoryCache {
object_id: String, object_id: String,
revs_map: Arc<DashMap<i64, RevisionRecord>>, revs_map: Arc<DashMap<i64, SyncRecord>>,
delegate: Arc<dyn RevisionMemoryCacheDelegate>, delegate: Arc<dyn RevisionMemoryCacheDelegate>,
pending_write_revs: Arc<RwLock<Vec<i64>>>, defer_write_revs: Arc<RwLock<Vec<i64>>>,
defer_save: RwLock<Option<JoinHandle<()>>>, defer_save: RwLock<Option<JoinHandle<()>>>,
} }
@ -25,7 +25,7 @@ impl RevisionMemoryCache {
object_id: object_id.to_owned(), object_id: object_id.to_owned(),
revs_map: Arc::new(DashMap::new()), revs_map: Arc::new(DashMap::new()),
delegate, delegate,
pending_write_revs: Arc::new(RwLock::new(vec![])), defer_write_revs: Arc::new(RwLock::new(vec![])),
defer_save: RwLock::new(None), defer_save: RwLock::new(None),
} }
} }
@ -34,7 +34,7 @@ impl RevisionMemoryCache {
self.revs_map.contains_key(rev_id) self.revs_map.contains_key(rev_id)
} }
pub(crate) async fn add<'a>(&'a self, record: Cow<'a, RevisionRecord>) { pub(crate) async fn add<'a>(&'a self, record: Cow<'a, SyncRecord>) {
let record = match record { let record = match record {
Cow::Borrowed(record) => record.clone(), Cow::Borrowed(record) => record.clone(),
Cow::Owned(record) => record, Cow::Owned(record) => record,
@ -43,11 +43,11 @@ impl RevisionMemoryCache {
let rev_id = record.revision.rev_id; let rev_id = record.revision.rev_id;
self.revs_map.insert(rev_id, record); self.revs_map.insert(rev_id, record);
let mut write_guard = self.pending_write_revs.write().await; let mut write_guard = self.defer_write_revs.write().await;
if !write_guard.contains(&rev_id) { if !write_guard.contains(&rev_id) {
write_guard.push(rev_id); write_guard.push(rev_id);
drop(write_guard); drop(write_guard);
self.make_checkpoint().await; self.tick_checkpoint().await;
} }
} }
@ -57,8 +57,8 @@ impl RevisionMemoryCache {
Some(mut record) => record.ack(), Some(mut record) => record.ack(),
} }
if self.pending_write_revs.read().await.contains(rev_id) { if self.defer_write_revs.read().await.contains(rev_id) {
self.make_checkpoint().await; self.tick_checkpoint().await;
} else { } else {
// The revision must be saved on disk if the pending_write_revs // The revision must be saved on disk if the pending_write_revs
// doesn't contains the rev_id. // doesn't contains the rev_id.
@ -66,7 +66,7 @@ impl RevisionMemoryCache {
} }
} }
pub(crate) async fn get(&self, rev_id: &i64) -> Option<RevisionRecord> { pub(crate) async fn get(&self, rev_id: &i64) -> Option<SyncRecord> {
self.revs_map.get(rev_id).map(|r| r.value().clone()) self.revs_map.get(rev_id).map(|r| r.value().clone())
} }
@ -80,21 +80,21 @@ impl RevisionMemoryCache {
} }
} }
pub(crate) async fn get_with_range(&self, range: &RevisionRange) -> Result<Vec<RevisionRecord>, FlowyError> { pub(crate) async fn get_with_range(&self, range: &RevisionRange) -> Result<Vec<SyncRecord>, FlowyError> {
let revs = range let revs = range
.iter() .iter()
.flat_map(|rev_id| self.revs_map.get(&rev_id).map(|record| record.clone())) .flat_map(|rev_id| self.revs_map.get(&rev_id).map(|record| record.clone()))
.collect::<Vec<RevisionRecord>>(); .collect::<Vec<SyncRecord>>();
Ok(revs) Ok(revs)
} }
pub(crate) async fn reset_with_revisions(&self, revision_records: Vec<RevisionRecord>) { pub(crate) async fn reset_with_revisions(&self, revision_records: Vec<SyncRecord>) {
self.revs_map.clear(); self.revs_map.clear();
if let Some(handler) = self.defer_save.write().await.take() { if let Some(handler) = self.defer_save.write().await.take() {
handler.abort(); handler.abort();
} }
let mut write_guard = self.pending_write_revs.write().await; let mut write_guard = self.defer_write_revs.write().await;
write_guard.clear(); write_guard.clear();
for record in revision_records { for record in revision_records {
write_guard.push(record.revision.rev_id); write_guard.push(record.revision.rev_id);
@ -102,21 +102,21 @@ impl RevisionMemoryCache {
} }
drop(write_guard); drop(write_guard);
self.make_checkpoint().await; self.tick_checkpoint().await;
} }
async fn make_checkpoint(&self) { async fn tick_checkpoint(&self) {
// https://github.com/async-graphql/async-graphql/blob/ed8449beec3d9c54b94da39bab33cec809903953/src/dataloader/mod.rs#L362 // https://github.com/async-graphql/async-graphql/blob/ed8449beec3d9c54b94da39bab33cec809903953/src/dataloader/mod.rs#L362
if let Some(handler) = self.defer_save.write().await.take() { if let Some(handler) = self.defer_save.write().await.take() {
handler.abort(); handler.abort();
} }
if self.pending_write_revs.read().await.is_empty() { if self.defer_write_revs.read().await.is_empty() {
return; return;
} }
let rev_map = self.revs_map.clone(); let rev_map = self.revs_map.clone();
let pending_write_revs = self.pending_write_revs.clone(); let pending_write_revs = self.defer_write_revs.clone();
let delegate = self.delegate.clone(); let delegate = self.delegate.clone();
*self.defer_save.write().await = Some(tokio::spawn(async move { *self.defer_save.write().await = Some(tokio::spawn(async move {
@ -128,7 +128,7 @@ impl RevisionMemoryCache {
// //
// Use saturating_sub and split_off ? // Use saturating_sub and split_off ?
// https://stackoverflow.com/questions/28952411/what-is-the-idiomatic-way-to-pop-the-last-n-elements-in-a-mutable-vec // https://stackoverflow.com/questions/28952411/what-is-the-idiomatic-way-to-pop-the-last-n-elements-in-a-mutable-vec
let mut save_records: Vec<RevisionRecord> = vec![]; let mut save_records: Vec<SyncRecord> = vec![];
revs_write_guard.iter().for_each(|rev_id| match rev_map.get(rev_id) { revs_write_guard.iter().for_each(|rev_id| match rev_map.get(rev_id) {
None => {} None => {}
Some(value) => { Some(value) => {
@ -136,7 +136,7 @@ impl RevisionMemoryCache {
} }
}); });
if delegate.checkpoint_tick(save_records).is_ok() { if delegate.send_sync(save_records).is_ok() {
revs_write_guard.clear(); revs_write_guard.clear();
drop(revs_write_guard); drop(revs_write_guard);
} }

4
frontend/rust-lib/flowy-revision/src/cache/reset.rs vendored
View File

@ -1,4 +1,4 @@
use crate::disk::{RevisionDiskCache, RevisionRecord}; use crate::disk::{RevisionDiskCache, SyncRecord};
use crate::{RevisionLoader, RevisionPersistence}; use crate::{RevisionLoader, RevisionPersistence};
use bytes::Bytes; use bytes::Bytes;
use flowy_error::{FlowyError, FlowyResult}; use flowy_error::{FlowyError, FlowyResult};
@ -76,7 +76,7 @@ where
let bytes = self.target.reset_data(revisions)?; let bytes = self.target.reset_data(revisions)?;
let revision = Revision::initial_revision(&self.user_id, self.target.target_id(), bytes); let revision = Revision::initial_revision(&self.user_id, self.target.target_id(), bytes);
let record = RevisionRecord::new(revision); let record = SyncRecord::new(revision);
tracing::trace!("Reset {} revision record object", self.target.target_id()); tracing::trace!("Reset {} revision record object", self.target.target_id());
let _ = self let _ = self

15
frontend/rust-lib/flowy-revision/src/conflict_resolve.rs
View File

@ -1,4 +1,4 @@
use crate::RevisionManager; use crate::{RevisionMD5, RevisionManager};
use bytes::Bytes; use bytes::Bytes;
use flowy_error::{FlowyError, FlowyResult}; use flowy_error::{FlowyError, FlowyResult};
use flowy_sync::entities::{ use flowy_sync::entities::{
@ -8,8 +8,6 @@ use flowy_sync::entities::{
use lib_infra::future::BoxResultFuture; use lib_infra::future::BoxResultFuture;
use std::{convert::TryFrom, sync::Arc}; use std::{convert::TryFrom, sync::Arc};
pub type OperationsMD5 = String;
pub struct TransformOperations<Operations> { pub struct TransformOperations<Operations> {
pub client_operations: Operations, pub client_operations: Operations,
pub server_operations: Option<Operations>, pub server_operations: Option<Operations>,
@ -28,12 +26,12 @@ pub trait ConflictResolver<Operations>
where where
Operations: Send + Sync, Operations: Send + Sync,
{ {
fn compose_operations(&self, operations: Operations) -> BoxResultFuture<OperationsMD5, FlowyError>; fn compose_operations(&self, operations: Operations) -> BoxResultFuture<RevisionMD5, FlowyError>;
fn transform_operations( fn transform_operations(
&self, &self,
operations: Operations, operations: Operations,
) -> BoxResultFuture<TransformOperations<Operations>, FlowyError>; ) -> BoxResultFuture<TransformOperations<Operations>, FlowyError>;
fn reset_operations(&self, operations: Operations) -> BoxResultFuture<OperationsMD5, FlowyError>; fn reset_operations(&self, operations: Operations) -> BoxResultFuture<RevisionMD5, FlowyError>;
} }
pub trait ConflictRevisionSink: Send + Sync + 'static { pub trait ConflictRevisionSink: Send + Sync + 'static {
@ -129,9 +127,8 @@ where
// The server_prime is None means the client local revisions conflict with the // The server_prime is None means the client local revisions conflict with the
// // server, and it needs to override the client delta. // // server, and it needs to override the client delta.
let md5 = self.resolver.reset_operations(client_operations).await?; let md5 = self.resolver.reset_operations(client_operations).await?;
let repeated_revision = RepeatedRevision::new(revisions); debug_assert!(md5.is_equal(&revisions.last().unwrap().md5));
assert_eq!(repeated_revision.last().unwrap().md5, md5); let _ = self.rev_manager.reset_object(revisions).await?;
let _ = self.rev_manager.reset_object(repeated_revision).await?;
Ok(None) Ok(None)
} }
Some(server_operations) => { Some(server_operations) => {
@ -158,7 +155,7 @@ fn make_client_and_server_revision<Operations, Connection>(
rev_manager: &Arc<RevisionManager<Connection>>, rev_manager: &Arc<RevisionManager<Connection>>,
client_operations: Operations, client_operations: Operations,
server_operations: Option<Operations>, server_operations: Option<Operations>,
md5: String, md5: RevisionMD5,
) -> (Revision, Option<Revision>) ) -> (Revision, Option<Revision>)
where where
Operations: OperationsSerializer, Operations: OperationsSerializer,

63
frontend/rust-lib/flowy-revision/src/rev_manager.rs
View File

@ -3,8 +3,8 @@ use crate::{RevisionPersistence, RevisionSnapshotDiskCache, RevisionSnapshotMana
use bytes::Bytes; use bytes::Bytes;
use flowy_error::{FlowyError, FlowyResult}; use flowy_error::{FlowyError, FlowyResult};
use flowy_sync::{ use flowy_sync::{
entities::revision::{RepeatedRevision, Revision, RevisionRange}, entities::revision::{Revision, RevisionRange},
util::{pair_rev_id_from_revisions, RevIdCounter}, util::{md5, pair_rev_id_from_revisions, RevIdCounter},
}; };
use lib_infra::future::FutureResult; use lib_infra::future::FutureResult;
use std::sync::Arc; use std::sync::Arc;
@ -143,9 +143,9 @@ impl<Connection: 'static> RevisionManager<Connection> {
} }
#[tracing::instrument(level = "debug", skip(self, revisions), err)] #[tracing::instrument(level = "debug", skip(self, revisions), err)]
pub async fn reset_object(&self, revisions: RepeatedRevision) -> FlowyResult<()> { pub async fn reset_object(&self, revisions: Vec<Revision>) -> FlowyResult<()> {
let rev_id = pair_rev_id_from_revisions(&revisions).1; let rev_id = pair_rev_id_from_revisions(&revisions).1;
let _ = self.rev_persistence.reset(revisions.into_inner()).await?; let _ = self.rev_persistence.reset(revisions).await?;
self.rev_id_counter.set(rev_id); self.rev_id_counter.set(rev_id);
Ok(()) Ok(())
} }
@ -191,7 +191,7 @@ impl<Connection: 'static> RevisionManager<Connection> {
pub fn next_rev_id_pair(&self) -> (i64, i64) { pub fn next_rev_id_pair(&self) -> (i64, i64) {
let cur = self.rev_id_counter.value(); let cur = self.rev_id_counter.value();
let next = self.rev_id_counter.next(); let next = self.rev_id_counter.next_id();
(cur, next) (cur, next)
} }
@ -283,3 +283,56 @@ impl<Connection: 'static> RevisionLoader<Connection> {
Ok(revisions) Ok(revisions)
} }
} }
/// Represents as the md5 of the revision object after applying the
/// revision. For example, RevisionMD5 will be the md5 of the document
/// content.
#[derive(Debug, Clone)]
pub struct RevisionMD5(String);
impl RevisionMD5 {
pub fn from_bytes<T: AsRef<[u8]>>(bytes: T) -> Result<Self, FlowyError> {
Ok(RevisionMD5(md5(bytes)))
}
pub fn into_inner(self) -> String {
self.0
}
pub fn is_equal(&self, s: &str) -> bool {
self.0 == s
}
}
impl std::convert::From<RevisionMD5> for String {
fn from(md5: RevisionMD5) -> Self {
md5.0
}
}
impl std::convert::From<&str> for RevisionMD5 {
fn from(s: &str) -> Self {
Self(s.to_owned())
}
}
impl std::convert::From<String> for RevisionMD5 {
fn from(s: String) -> Self {
Self(s)
}
}
impl std::ops::Deref for RevisionMD5 {
type Target = String;
fn deref(&self) -> &Self::Target {
&self.0
}
}
impl PartialEq<Self> for RevisionMD5 {
fn eq(&self, other: &Self) -> bool {
self.0 == other.0
}
}
impl std::cmp::Eq for RevisionMD5 {}

28
frontend/rust-lib/flowy-revision/src/rev_persistence.rs
View File

@ -2,10 +2,9 @@ use crate::cache::{
disk::{RevisionChangeset, RevisionDiskCache}, disk::{RevisionChangeset, RevisionDiskCache},
memory::RevisionMemoryCacheDelegate, memory::RevisionMemoryCacheDelegate,
}; };
use crate::disk::{RevisionRecord, RevisionState}; use crate::disk::{RevisionState, SyncRecord};
use crate::memory::RevisionMemoryCache; use crate::memory::RevisionMemoryCache;
use crate::RevisionCompress; use crate::RevisionCompress;
use flowy_error::{internal_error, FlowyError, FlowyResult}; use flowy_error::{internal_error, FlowyError, FlowyResult};
use flowy_sync::entities::revision::{Revision, RevisionRange}; use flowy_sync::entities::revision::{Revision, RevisionRange};
use std::collections::VecDeque; use std::collections::VecDeque;
@ -20,10 +19,13 @@ pub struct RevisionPersistence<Connection> {
object_id: String, object_id: String,
disk_cache: Arc<dyn RevisionDiskCache<Connection, Error = FlowyError>>, disk_cache: Arc<dyn RevisionDiskCache<Connection, Error = FlowyError>>,
memory_cache: Arc<RevisionMemoryCache>, memory_cache: Arc<RevisionMemoryCache>,
sync_seq: RwLock<RevisionSyncSequence>, sync_seq: RwLock<DeferSyncSequence>,
} }
impl<Connection: 'static> RevisionPersistence<Connection> { impl<Connection> RevisionPersistence<Connection>
where
Connection: 'static,
{
pub fn new<C>(user_id: &str, object_id: &str, disk_cache: C) -> RevisionPersistence<Connection> pub fn new<C>(user_id: &str, object_id: &str, disk_cache: C) -> RevisionPersistence<Connection>
where where
C: 'static + RevisionDiskCache<Connection, Error = FlowyError>, C: 'static + RevisionDiskCache<Connection, Error = FlowyError>,
@ -39,7 +41,7 @@ impl<Connection: 'static> RevisionPersistence<Connection> {
) -> RevisionPersistence<Connection> { ) -> RevisionPersistence<Connection> {
let object_id = object_id.to_owned(); let object_id = object_id.to_owned();
let user_id = user_id.to_owned(); let user_id = user_id.to_owned();
let sync_seq = RwLock::new(RevisionSyncSequence::new()); let sync_seq = RwLock::new(DeferSyncSequence::new());
let memory_cache = Arc::new(RevisionMemoryCache::new(&object_id, Arc::new(disk_cache.clone()))); let memory_cache = Arc::new(RevisionMemoryCache::new(&object_id, Arc::new(disk_cache.clone())));
Self { Self {
user_id, user_id,
@ -131,7 +133,7 @@ impl<Connection: 'static> RevisionPersistence<Connection> {
pub(crate) async fn reset(&self, revisions: Vec<Revision>) -> FlowyResult<()> { pub(crate) async fn reset(&self, revisions: Vec<Revision>) -> FlowyResult<()> {
let records = revisions let records = revisions
.into_iter() .into_iter()
.map(|revision| RevisionRecord { .map(|revision| SyncRecord {
revision, revision,
state: RevisionState::Sync, state: RevisionState::Sync,
write_to_disk: false, write_to_disk: false,
@ -151,7 +153,7 @@ impl<Connection: 'static> RevisionPersistence<Connection> {
tracing::warn!("Duplicate revision: {}:{}-{:?}", self.object_id, revision.rev_id, state); tracing::warn!("Duplicate revision: {}:{}-{:?}", self.object_id, revision.rev_id, state);
return Ok(()); return Ok(());
} }
let record = RevisionRecord { let record = SyncRecord {
revision, revision,
state, state,
write_to_disk, write_to_disk,
@ -172,7 +174,7 @@ impl<Connection: 'static> RevisionPersistence<Connection> {
Ok(()) Ok(())
} }
pub async fn get(&self, rev_id: i64) -> Option<RevisionRecord> { pub async fn get(&self, rev_id: i64) -> Option<SyncRecord> {
match self.memory_cache.get(&rev_id).await { match self.memory_cache.get(&rev_id).await {
None => match self None => match self
.disk_cache .disk_cache
@ -192,7 +194,7 @@ impl<Connection: 'static> RevisionPersistence<Connection> {
} }
} }
pub fn batch_get(&self, doc_id: &str) -> FlowyResult<Vec<RevisionRecord>> { pub fn batch_get(&self, doc_id: &str) -> FlowyResult<Vec<SyncRecord>> {
self.disk_cache.read_revision_records(doc_id, None) self.disk_cache.read_revision_records(doc_id, None)
} }
@ -225,7 +227,7 @@ impl<Connection: 'static> RevisionPersistence<Connection> {
} }
impl<C> RevisionMemoryCacheDelegate for Arc<dyn RevisionDiskCache<C, Error = FlowyError>> { impl<C> RevisionMemoryCacheDelegate for Arc<dyn RevisionDiskCache<C, Error = FlowyError>> {
fn checkpoint_tick(&self, mut records: Vec<RevisionRecord>) -> FlowyResult<()> { fn send_sync(&self, mut records: Vec<SyncRecord>) -> FlowyResult<()> {
records.retain(|record| record.write_to_disk); records.retain(|record| record.write_to_disk);
if !records.is_empty() { if !records.is_empty() {
tracing::Span::current().record( tracing::Span::current().record(
@ -251,10 +253,10 @@ impl<C> RevisionMemoryCacheDelegate for Arc<dyn RevisionDiskCache<C, Error = Flo
} }
#[derive(Default)] #[derive(Default)]
struct RevisionSyncSequence(VecDeque<i64>); struct DeferSyncSequence(VecDeque<i64>);
impl RevisionSyncSequence { impl DeferSyncSequence {
fn new() -> Self { fn new() -> Self {
RevisionSyncSequence::default() DeferSyncSequence::default()
} }
fn add(&mut self, new_rev_id: i64) -> FlowyResult<()> { fn add(&mut self, new_rev_id: i64) -> FlowyResult<()> {

2
frontend/rust-lib/flowy-revision/src/ws_manager.rs
View File

@ -28,7 +28,7 @@ pub trait RevisionWSDataStream: Send + Sync {
} }
// The sink provides the data that will be sent through the web socket to the // The sink provides the data that will be sent through the web socket to the
// backend. // server.
pub trait RevisionWebSocketSink: Send + Sync { pub trait RevisionWebSocketSink: Send + Sync {
fn next(&self) -> FutureResult<Option<ClientRevisionWSData>, FlowyError>; fn next(&self) -> FutureResult<Option<ClientRevisionWSData>, FlowyError>;
} }

1
frontend/rust-lib/flowy-revision/tests/main.rs Normal file
View File

@ -0,0 +1 @@

15
frontend/rust-lib/flowy-sdk/src/deps_resolve/folder_deps.rs
View File

@ -18,7 +18,7 @@ use flowy_net::{
http_server::folder::FolderHttpCloudService, local_server::LocalServer, ws::connection::FlowyWebSocketConnect, http_server::folder::FolderHttpCloudService, local_server::LocalServer, ws::connection::FlowyWebSocketConnect,
}; };
use flowy_revision::{RevisionWebSocket, WSStateReceiver}; use flowy_revision::{RevisionWebSocket, WSStateReceiver};
use flowy_sync::entities::revision::{RepeatedRevision, Revision}; use flowy_sync::entities::revision::Revision;
use flowy_sync::entities::ws_data::ClientRevisionWSData; use flowy_sync::entities::ws_data::ClientRevisionWSData;
use flowy_user::services::UserSession; use flowy_user::services::UserSession;
use futures_core::future::BoxFuture; use futures_core::future::BoxFuture;
@ -151,12 +151,12 @@ impl ViewDataProcessor for DocumentViewDataProcessor {
) -> FutureResult<(), FlowyError> { ) -> FutureResult<(), FlowyError> {
// Only accept Document type // Only accept Document type
debug_assert_eq!(layout, ViewLayoutTypePB::Document); debug_assert_eq!(layout, ViewLayoutTypePB::Document);
let repeated_revision: RepeatedRevision = Revision::initial_revision(user_id, view_id, view_data).into(); let revision = Revision::initial_revision(user_id, view_id, view_data);
let view_id = view_id.to_string(); let view_id = view_id.to_string();
let manager = self.0.clone(); let manager = self.0.clone();
FutureResult::new(async move { FutureResult::new(async move {
let _ = manager.create_document(view_id, repeated_revision).await?; let _ = manager.create_document(view_id, vec![revision]).await?;
Ok(()) Ok(())
}) })
} }
@ -194,9 +194,8 @@ impl ViewDataProcessor for DocumentViewDataProcessor {
let document_content = self.0.initial_document_content(); let document_content = self.0.initial_document_content();
FutureResult::new(async move { FutureResult::new(async move {
let delta_data = Bytes::from(document_content); let delta_data = Bytes::from(document_content);
let repeated_revision: RepeatedRevision = let revision = Revision::initial_revision(&user_id, &view_id, delta_data.clone());
Revision::initial_revision(&user_id, &view_id, delta_data.clone()).into(); let _ = manager.create_document(view_id, vec![revision]).await?;
let _ = manager.create_document(view_id, repeated_revision).await?;
Ok(delta_data) Ok(delta_data)
}) })
} }
@ -226,11 +225,11 @@ impl ViewDataProcessor for GridViewDataProcessor {
_layout: ViewLayoutTypePB, _layout: ViewLayoutTypePB,
delta_data: Bytes, delta_data: Bytes,
) -> FutureResult<(), FlowyError> { ) -> FutureResult<(), FlowyError> {
let repeated_revision: RepeatedRevision = Revision::initial_revision(user_id, view_id, delta_data).into(); let revision = Revision::initial_revision(user_id, view_id, delta_data);
let view_id = view_id.to_string(); let view_id = view_id.to_string();
let grid_manager = self.0.clone(); let grid_manager = self.0.clone();
FutureResult::new(async move { FutureResult::new(async move {
let _ = grid_manager.create_grid(view_id, repeated_revision).await?; let _ = grid_manager.create_grid(view_id, vec![revision]).await?;
Ok(()) Ok(())
}) })
} }

5
shared-lib/flowy-sync/src/client_document/document_pad.rs
View File

@ -1,3 +1,4 @@
use crate::util::md5;
use crate::{ use crate::{
client_document::{ client_document::{
history::{History, UndoResult}, history::{History, UndoResult},
@ -77,9 +78,9 @@ impl ClientDocument {
&self.operations &self.operations
} }
pub fn md5(&self) -> String { pub fn document_md5(&self) -> String {
let bytes = self.to_bytes(); let bytes = self.to_bytes();
format!("{:x}", md5::compute(bytes)) md5(&bytes)
} }
pub fn set_notify(&mut self, notify: mpsc::UnboundedSender<()>) { pub fn set_notify(&mut self, notify: mpsc::UnboundedSender<()>) {

12
shared-lib/flowy-sync/src/client_folder/folder_pad.rs
View File

@ -1,9 +1,9 @@
use crate::errors::internal_error; use crate::errors::internal_error;
use crate::server_folder::{FolderOperations, FolderOperationsBuilder}; use crate::server_folder::{FolderOperations, FolderOperationsBuilder};
use crate::util::cal_diff; use crate::util::{cal_diff, md5};
use crate::{ use crate::{
client_folder::builder::FolderPadBuilder, client_folder::builder::FolderPadBuilder,
entities::revision::{md5, Revision}, entities::revision::Revision,
errors::{CollaborateError, CollaborateResult}, errors::{CollaborateError, CollaborateResult},
}; };
use flowy_folder_data_model::revision::{AppRevision, FolderRevision, TrashRevision, ViewRevision, WorkspaceRevision}; use flowy_folder_data_model::revision::{AppRevision, FolderRevision, TrashRevision, ViewRevision, WorkspaceRevision};
@ -61,7 +61,7 @@ impl FolderPad {
self.folder_rev = folder.folder_rev; self.folder_rev = folder.folder_rev;
self.operations = folder.operations; self.operations = folder.operations;
Ok(self.md5()) Ok(self.folder_md5())
} }
pub fn compose_remote_operations(&mut self, operations: FolderOperations) -> CollaborateResult<String> { pub fn compose_remote_operations(&mut self, operations: FolderOperations) -> CollaborateResult<String> {
@ -313,7 +313,7 @@ impl FolderPad {
} }
} }
pub fn md5(&self) -> String { pub fn folder_md5(&self) -> String {
md5(&self.operations.json_bytes()) md5(&self.operations.json_bytes())
} }
@ -345,7 +345,7 @@ impl FolderPad {
self.operations = self.operations.compose(&operations)?; self.operations = self.operations.compose(&operations)?;
Ok(Some(FolderChangeset { Ok(Some(FolderChangeset {
operations, operations,
md5: self.md5(), md5: self.folder_md5(),
})) }))
} }
} }
@ -383,7 +383,7 @@ impl FolderPad {
self.operations = self.operations.compose(&operations)?; self.operations = self.operations.compose(&operations)?;
Ok(Some(FolderChangeset { Ok(Some(FolderChangeset {
operations, operations,
md5: self.md5(), md5: self.folder_md5(),
})) }))
} }
} }

4
shared-lib/flowy-sync/src/client_grid/block_revision_pad.rs
View File

@ -1,6 +1,6 @@
use crate::entities::revision::{md5, RepeatedRevision, Revision}; use crate::entities::revision::{RepeatedRevision, Revision};
use crate::errors::{CollaborateError, CollaborateResult}; use crate::errors::{CollaborateError, CollaborateResult};
use crate::util::{cal_diff, make_operations_from_revisions}; use crate::util::{cal_diff, make_operations_from_revisions, md5};
use flowy_grid_data_model::revision::{ use flowy_grid_data_model::revision::{
gen_block_id, gen_row_id, CellRevision, GridBlockRevision, RowChangeset, RowRevision, gen_block_id, gen_row_id, CellRevision, GridBlockRevision, RowChangeset, RowRevision,
}; };

8
shared-lib/flowy-sync/src/client_grid/grid_revision_pad.rs
View File

@ -1,6 +1,6 @@
use crate::entities::revision::{md5, RepeatedRevision, Revision}; use crate::entities::revision::{RepeatedRevision, Revision};
use crate::errors::{internal_error, CollaborateError, CollaborateResult}; use crate::errors::{internal_error, CollaborateError, CollaborateResult};
use crate::util::{cal_diff, make_operations_from_revisions}; use crate::util::{cal_diff, make_operations_from_revisions, md5};
use flowy_grid_data_model::revision::{ use flowy_grid_data_model::revision::{
gen_block_id, gen_grid_id, FieldRevision, FieldTypeRevision, GridBlockMetaRevision, GridBlockMetaRevisionChangeset, gen_block_id, gen_grid_id, FieldRevision, FieldTypeRevision, GridBlockMetaRevision, GridBlockMetaRevisionChangeset,
@ -315,7 +315,7 @@ impl GridRevisionPad {
}) })
} }
pub fn md5(&self) -> String { pub fn grid_md5(&self) -> String {
md5(&self.operations.json_bytes()) md5(&self.operations.json_bytes())
} }
@ -343,7 +343,7 @@ impl GridRevisionPad {
self.operations = self.operations.compose(&operations)?; self.operations = self.operations.compose(&operations)?;
Ok(Some(GridRevisionChangeset { Ok(Some(GridRevisionChangeset {
operations, operations,
md5: self.md5(), md5: self.grid_md5(),
})) }))
} }
} }

4
shared-lib/flowy-sync/src/client_grid/view_revision_pad.rs
View File

@ -1,6 +1,6 @@
use crate::entities::revision::{md5, Revision}; use crate::entities::revision::Revision;
use crate::errors::{internal_error, CollaborateError, CollaborateResult}; use crate::errors::{internal_error, CollaborateError, CollaborateResult};
use crate::util::{cal_diff, make_operations_from_revisions}; use crate::util::{cal_diff, make_operations_from_revisions, md5};
use flowy_grid_data_model::revision::{ use flowy_grid_data_model::revision::{
FieldRevision, FieldTypeRevision, FilterConfigurationRevision, FilterConfigurationsByFieldId, GridViewRevision, FieldRevision, FieldTypeRevision, FilterConfigurationRevision, FilterConfigurationsByFieldId, GridViewRevision,
GroupConfigurationRevision, GroupConfigurationsByFieldId, LayoutRevision, GroupConfigurationRevision, GroupConfigurationsByFieldId, LayoutRevision,

57
shared-lib/flowy-sync/src/entities/revision.rs
View File

@ -1,3 +1,4 @@
use crate::util::md5;
use bytes::Bytes; use bytes::Bytes;
use flowy_derive::{ProtoBuf, ProtoBuf_Enum}; use flowy_derive::{ProtoBuf, ProtoBuf_Enum};
use std::{convert::TryFrom, fmt::Formatter, ops::RangeInclusive}; use std::{convert::TryFrom, fmt::Formatter, ops::RangeInclusive};
@ -36,6 +37,34 @@ impl std::convert::From<Vec<u8>> for Revision {
} }
impl Revision { impl Revision {
pub fn new<T: Into<String>>(
object_id: &str,
base_rev_id: i64,
rev_id: i64,
bytes: Bytes,
user_id: &str,
md5: T,
) -> Revision {
let user_id = user_id.to_owned();
let object_id = object_id.to_owned();
let bytes = bytes.to_vec();
let base_rev_id = base_rev_id;
let rev_id = rev_id;
if base_rev_id != 0 {
debug_assert!(base_rev_id != rev_id);
}
Self {
base_rev_id,
rev_id,
bytes,
md5: md5.into(),
object_id,
ty: RevType::DeprecatedLocal,
user_id,
}
}
pub fn is_empty(&self) -> bool { pub fn is_empty(&self) -> bool {
self.base_rev_id == self.rev_id self.base_rev_id == self.rev_id
} }
@ -52,28 +81,6 @@ impl Revision {
let md5 = md5(&bytes); let md5 = md5(&bytes);
Self::new(object_id, 0, 0, bytes, user_id, md5) Self::new(object_id, 0, 0, bytes, user_id, md5)
} }
pub fn new(object_id: &str, base_rev_id: i64, rev_id: i64, bytes: Bytes, user_id: &str, md5: String) -> Revision {
let user_id = user_id.to_owned();
let object_id = object_id.to_owned();
let bytes = bytes.to_vec();
let base_rev_id = base_rev_id;
let rev_id = rev_id;
if base_rev_id != 0 {
debug_assert!(base_rev_id != rev_id);
}
Self {
base_rev_id,
rev_id,
bytes,
md5,
object_id,
ty: RevType::DeprecatedLocal,
user_id,
}
}
} }
impl std::fmt::Debug for Revision { impl std::fmt::Debug for Revision {
@ -209,12 +216,6 @@ impl RevisionRange {
} }
} }
#[inline]
pub fn md5<T: AsRef<[u8]>>(data: T) -> String {
let md5 = format!("{:x}", md5::compute(data));
md5
}
#[derive(Debug, ProtoBuf_Enum, Clone, Eq, PartialEq)] #[derive(Debug, ProtoBuf_Enum, Clone, Eq, PartialEq)]
pub enum RevType { pub enum RevType {
DeprecatedLocal = 0, DeprecatedLocal = 0,

3
shared-lib/flowy-sync/src/util.rs
View File

@ -49,7 +49,8 @@ impl RevIdCounter {
pub fn new(n: i64) -> Self { pub fn new(n: i64) -> Self {
Self(AtomicI64::new(n)) Self(AtomicI64::new(n))
} }
pub fn next(&self) -> i64 {
pub fn next_id(&self) -> i64 {
let _ = self.0.fetch_add(1, SeqCst); let _ = self.0.fetch_add(1, SeqCst);
self.value() self.value()
} }

16
shared-lib/lib-ot/src/core/node_tree/tree.rs
View File

@ -35,9 +35,19 @@ impl NodeTree {
Ok(tree) Ok(tree)
} }
pub fn from_bytes(bytes: Vec<u8>, context: NodeTreeContext) -> Result<Self, OTError> { pub fn from_bytes(bytes: &[u8]) -> Result<Self, OTError> {
let operations = NodeOperations::from_bytes(bytes)?; let tree: NodeTree = serde_json::from_slice(bytes).map_err(|e| OTError::serde().context(e))?;
Self::from_operations(operations, context) Ok(tree)
}
pub fn to_bytes(&self) -> Vec<u8> {
match serde_json::to_vec(self) {
Ok(bytes) => bytes,
Err(e) => {
tracing::error!("{}", e);
vec![]
}
}
} }
pub fn from_operations<T: Into<NodeOperations>>(operations: T, context: NodeTreeContext) -> Result<Self, OTError> { pub fn from_operations<T: Into<NodeOperations>>(operations: T, context: NodeTreeContext) -> Result<Self, OTError> {

13
shared-lib/lib-ot/tests/node/serde_test.rs
View File

@ -1,4 +1,6 @@
use lib_ot::core::{AttributeBuilder, Changeset, NodeData, NodeDataBuilder, NodeOperation, NodeTree, Path}; use lib_ot::core::{
AttributeBuilder, Changeset, NodeData, NodeDataBuilder, NodeOperation, NodeTree, NodeTreeContext, Path,
};
use lib_ot::text_delta::DeltaTextOperationBuilder; use lib_ot::text_delta::DeltaTextOperationBuilder;
#[test] #[test]
@ -26,6 +28,7 @@ fn operation_insert_node_with_children_serde_test() {
r#"{"op":"insert","path":[0,1],"nodes":[{"type":"text","children":[{"type":"sub_text"}]}]}"# r#"{"op":"insert","path":[0,1],"nodes":[{"type":"text","children":[{"type":"sub_text"}]}]}"#
); );
} }
#[test] #[test]
fn operation_update_node_attributes_serde_test() { fn operation_update_node_attributes_serde_test() {
let operation = NodeOperation::Update { let operation = NodeOperation::Update {
@ -102,6 +105,14 @@ fn node_tree_serialize_test() {
assert_eq!(json, TREE_JSON); assert_eq!(json, TREE_JSON);
} }
#[test]
fn node_tree_serde_test() {
let tree: NodeTree = serde_json::from_str(TREE_JSON).unwrap();
let bytes = tree.to_bytes();
let tree = NodeTree::from_bytes(&bytes).unwrap();
assert_eq!(bytes, tree.to_bytes());
}
#[allow(dead_code)] #[allow(dead_code)]
const TREE_JSON: &str = r#"{ const TREE_JSON: &str = r#"{
"type": "editor", "type": "editor",