chore: code cleanup

libs/collab-stream/src/client.rs

@@ -3,13 +3,12 @@ use crate::error::StreamError;
 use crate::lease::{Lease, LeaseAcquisition};
 use crate::model::{
   AwarenessStreamUpdate, AwarenessStreamUpdateBatch, CollabStreamUpdate, CollabStreamUpdateBatch,
-  CollabUpdateEvent, MessageId,
+  MessageId,
 };
-use crate::pubsub::{CollabStreamPub, CollabStreamSub};
 use crate::stream_group::{StreamConfig, StreamGroup};
 use futures::Stream;
 use redis::aio::ConnectionManager;
-use redis::streams::{StreamRangeReply, StreamReadOptions, StreamReadReply};
+use redis::streams::{StreamRangeReply, StreamReadOptions};
 use redis::{AsyncCommands, FromRedisValue};
 use std::time::Duration;
 use tracing::error;

libs/collab-stream/src/lease.rs

@@ -1,11 +1,8 @@
-use crate::client::CollabRedisStream;
 use crate::error::StreamError;
 use async_trait::async_trait;
-use chrono::{DateTime, NaiveDateTime};
 use redis::aio::ConnectionManager;
-use redis::{RedisResult, Value};
+use redis::Value;
-use std::sync::Arc;
+use std::time::{Duration, SystemTime, UNIX_EPOCH};
-use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
 
 const RELEASE_SCRIPT: &str = r#"
 if redis.call("GET", KEYS[1]) == ARGV[1] then

services/appflowy-collaborate/src/application.rs

@@ -77,8 +77,6 @@ pub async fn run_actix_server(
     rt_cmd_recv,
     state.redis_connection_manager.clone(),
     Duration::from_secs(config.collab.group_persistence_interval_secs),
-    config.collab.edit_state_max_count,
-    config.collab.edit_state_max_secs,
     state.indexer_provider.clone(),
   )
   .await

services/appflowy-collaborate/src/group/cmd.rs

@@ -2,7 +2,6 @@ use async_stream::stream;
 use collab::core::origin::CollabOrigin;
 use collab::entity::EncodedCollab;
 use dashmap::DashMap;
-use futures::channel::mpsc::{SendError, Sender};
 use futures::Sink;
 use futures_util::StreamExt;
 use std::collections::HashMap;
@@ -333,22 +332,22 @@ impl<T> Sink<T> for NullSender<T> {
   type Error = RealtimeError;
 
   #[inline]
-  fn poll_ready(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+  fn poll_ready(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
     Poll::Ready(Ok(()))
   }
 
   #[inline]
-  fn start_send(self: Pin<&mut Self>, item: T) -> Result<(), Self::Error> {
+  fn start_send(self: Pin<&mut Self>, _: T) -> Result<(), Self::Error> {
     Ok(())
   }
 
   #[inline]
-  fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+  fn poll_flush(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
     Poll::Ready(Ok(()))
   }
 
   #[inline]
-  fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+  fn poll_close(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
     Poll::Ready(Ok(()))
   }
 }

services/appflowy-collaborate/src/group/group_init.rs

@@ -1,11 +1,9 @@
 use crate::error::RealtimeError;
-use crate::indexer::{Indexer, IndexerProvider};
+use crate::indexer::Indexer;
 use crate::metrics::CollabRealtimeMetrics;
-use crate::state::RedisConnectionManager;
 use anyhow::anyhow;
 use app_error::AppError;
-use arc_swap::{ArcSwap, ArcSwapAny, ArcSwapOption};
+use arc_swap::ArcSwap;
-use bytes::Bytes;
 use collab::core::collab::DataSource;
 use collab::core::origin::CollabOrigin;
 use collab::entity::EncodedCollab;
@@ -17,7 +15,7 @@ use collab_rt_entity::{
   AckCode, AwarenessSync, BroadcastSync, CollabAck, MessageByObjectId, MsgId,
 };
 use collab_rt_entity::{ClientCollabMessage, CollabMessage};
-use collab_rt_protocol::{decode_update, Message, MessageReader, RTProtocolError, SyncMessage};
+use collab_rt_protocol::{Message, MessageReader, RTProtocolError, SyncMessage};
 use collab_stream::client::CollabRedisStream;
 use collab_stream::collab_update_sink::{AwarenessUpdateSink, CollabUpdateSink};
 use collab_stream::error::StreamError;
@@ -29,24 +27,19 @@ use collab_stream::stream_group::StreamGroup;
 use dashmap::DashMap;
 use database::collab::{CollabStorage, GetCollabOrigin};
 use database_entity::dto::{
-  AFCollabEmbeddings, CollabParams, InsertSnapshotParams, QueryCollabParams, SnapshotData,
+  AFCollabEmbeddings, CollabParams, InsertSnapshotParams, QueryCollabParams,
 };
 use futures::{pin_mut, Sink, Stream};
 use futures_util::{SinkExt, StreamExt};
 use std::collections::VecDeque;
-use std::fmt::{Display, Formatter};
+use std::sync::atomic::{AtomicU32, Ordering};
-use std::os::linux::raw::stat;
-use std::pin::Pin;
-use std::sync::atomic::{AtomicBool, AtomicI64, AtomicU32, Ordering};
 use std::sync::Arc;
 use std::time::{Duration, Instant};
-use tokio::sync::{mpsc, Mutex};
+use tokio::sync::mpsc;
-use tokio::time::{interval, MissedTickBehavior};
+use tokio::time::MissedTickBehavior;
-use tokio_util::sync::{CancellationToken, DropGuard};
+use tokio_util::sync::CancellationToken;
-use tracing::{error, event, info, trace};
+use tracing::{error, info, trace};
-use yrs::encoding::read::Error;
+use yrs::updates::decoder::{Decode, DecoderV1};
-use yrs::sync::AwarenessUpdate;
-use yrs::updates::decoder::{Decode, DecoderV1, DecoderV2};
 use yrs::updates::encoder::{Encode, Encoder, EncoderV1};
 use yrs::{ReadTxn, StateVector, Update};
 
@@ -100,6 +93,7 @@ impl CollabGroup {
     S: CollabStorage,
   {
     let persister = CollabPersister::new(
+      uid,
       workspace_id.clone(),
       object_id.clone(),
       collab_type.clone(),
@@ -174,7 +168,7 @@ impl CollabGroup {
 
   /// Task used to receive collab updates from Redis.
   async fn inbound_task(state: Arc<CollabGroupState>) -> Result<(), RealtimeError> {
-    let mut updates = state.persister.collab_redis_stream.collab_updates(
+    let updates = state.persister.collab_redis_stream.collab_updates(
       &state.workspace_id,
       &state.object_id,
       None,
@@ -233,7 +227,7 @@ impl CollabGroup {
 
   /// Task used to receive awareness updates from Redis.
   async fn inbound_awareness_task(state: Arc<CollabGroupState>) -> Result<(), RealtimeError> {
-    let mut updates = state.persister.collab_redis_stream.awareness_updates(
+    let updates = state.persister.collab_redis_stream.awareness_updates(
       &state.workspace_id,
       &state.object_id,
       None,
@@ -852,6 +846,7 @@ impl Drop for Subscription {
 }
 
 struct CollabPersister {
+  uid: i64,
   workspace_id: String,
   object_id: String,
   collab_type: CollabType,
@@ -868,6 +863,7 @@ impl CollabPersister {
   pub const GRACE_PERIOD_MS: u64 = 1000 * 60; // 5min
 
   pub fn new(
+    uid: i64,
     workspace_id: String,
     object_id: String,
     collab_type: CollabType,
@@ -878,6 +874,7 @@ impl CollabPersister {
     let update_sink = collab_redis_stream.collab_update_sink(&workspace_id, &object_id);
     let awareness_sink = collab_redis_stream.awareness_update_sink(&workspace_id, &object_id);
     Self {
+      uid,
       workspace_id,
       object_id,
       collab_type,
@@ -936,7 +933,7 @@ impl CollabPersister {
     // 2. consume all Redis updates on top of it (keep redis msg id)
     let mut last_message_id = None;
     let mut tx = collab.transact_mut();
-    let mut stream = self.collab_redis_stream.collab_updates(
+    let stream = self.collab_redis_stream.collab_updates(
       &self.workspace_id,
       &self.object_id,
       None, //TODO: store Redis last msg id somewhere in doc state snapshot and replay from there
@@ -1038,10 +1035,9 @@ impl CollabPersister {
         Err(err) => tracing::warn!("failed to fetch embeddings `{}`: {}", self.object_id, err),
       }
 
-      let uid = 0; //FIXME: what UID should go there?
       self
         .storage
-        .insert_or_update_collab(&self.workspace_id, &uid, params, true)
+        .queue_insert_or_update_collab(&self.workspace_id, &self.uid, params, true)
         .await
         .map_err(|err| RealtimeError::Internal(err.into()))?;
 

services/appflowy-collaborate/src/group/manager.rs

@@ -4,7 +4,7 @@ use std::time::Duration;
 use collab::core::collab::DataSource;
 use collab::core::origin::CollabOrigin;
 use collab::entity::EncodedCollab;
-use collab::lock::{Mutex, RwLock};
+use collab::lock::Mutex;
 use collab::preclude::Collab;
 use collab_entity::CollabType;
 use tracing::{error, instrument, trace};
@@ -34,8 +34,6 @@ pub struct GroupManager<S> {
   collab_redis_stream: Arc<CollabRedisStream>,
   control_event_stream: Arc<Mutex<StreamGroup>>,
   persistence_interval: Duration,
-  edit_state_max_count: u32,
-  edit_state_max_secs: i64,
   indexer_provider: Arc<IndexerProvider>,
 }
 
@@ -50,8 +48,6 @@ where
     metrics_calculate: Arc<CollabRealtimeMetrics>,
     collab_stream: CollabRedisStream,
     persistence_interval: Duration,
-    edit_state_max_count: u32,
-    edit_state_max_secs: i64,
     indexer_provider: Arc<IndexerProvider>,
   ) -> Result<Self, RealtimeError> {
     let collab_stream = Arc::new(collab_stream);
@@ -68,8 +64,6 @@ where
       collab_redis_stream: collab_stream,
       control_event_stream,
       persistence_interval,
-      edit_state_max_count,
-      edit_state_max_secs,
       indexer_provider,
     })
   }
@@ -107,17 +101,18 @@ where
     client_msg_router: &mut ClientMessageRouter,
   ) -> Result<(), RealtimeError> {
     // Lock the group and subscribe the user to the group.
-    if let Some(group) = self.state.get_mut_group(object_id).await {
+    if let Some(mut e) = self.state.get_mut_group(object_id).await {
+      let group = e.value_mut();
       trace!("[realtime]: {} subscribe group:{}", user, object_id,);
       let (sink, stream) = client_msg_router.init_client_communication::<CollabMessage>(
-        &group.workspace_id,
+        group.workspace_id(),
         user,
         object_id,
         self.access_control.clone(),
       );
       group.subscribe(user, message_origin.clone(), sink, stream);
       // explicitly drop the group to release the lock.
-      drop(group);
+      drop(e);
 
       self.state.insert_user(user, object_id).await?;
     } else {
@@ -160,7 +155,7 @@ where
     }
 
     let result = load_collab(user.uid, object_id, params, self.storage.clone()).await;
-    let (collab, encode_collab) = {
+    let encode_collab = {
       let (mut collab, encode_collab) = match result {
         Ok(value) => value,
         Err(err) => {
@@ -178,8 +173,7 @@ where
       };
 
       collab.initialize();
-      let collab = Arc::new(RwLock::from(collab));
+      encode_collab
-      (collab, encode_collab)
     };
 
     let cloned_control_event_stream = self.control_event_stream.clone();

services/appflowy-collaborate/src/rt_server.rs

@@ -53,8 +53,6 @@ where
     command_recv: CLCommandReceiver,
     redis_connection_manager: RedisConnectionManager,
     group_persistence_interval: Duration,
-    edit_state_max_count: u32,
-    edit_state_max_secs: i64,
     indexer_provider: Arc<IndexerProvider>,
   ) -> Result<Self, RealtimeError> {
     let enable_custom_runtime = get_env_var("APPFLOWY_COLLABORATE_MULTI_THREAD", "false")
@@ -76,8 +74,6 @@ where
         metrics.clone(),
         collab_stream,
         group_persistence_interval,
-        edit_state_max_count,
-        edit_state_max_secs,
         indexer_provider.clone(),
       )
       .await?,

src/application.rs

@@ -135,8 +135,6 @@ pub async fn run_actix_server(
     rt_cmd_recv,
     state.redis_connection_manager.clone(),
     Duration::from_secs(config.collab.group_persistence_interval_secs),
-    config.collab.edit_state_max_count,
-    config.collab.edit_state_max_secs,
     state.indexer_provider.clone(),
   )
   .await