diff --git a/libs/collab-stream/src/client.rs b/libs/collab-stream/src/client.rs
index c989d933..2ad9d79e 100644
--- a/libs/collab-stream/src/client.rs
+++ b/libs/collab-stream/src/client.rs
@@ -12,8 +12,6 @@ use std::sync::Arc;
 use std::time::Duration;
 use tracing::error;
 
-pub const CONTROL_STREAM_KEY: &str = "af_collab_control";
-
 #[derive(Clone)]
 pub struct CollabRedisStream {
   connection_manager: ConnectionManager,
diff --git a/services/appflowy-collaborate/src/error.rs b/services/appflowy-collaborate/src/error.rs
index b5a6ea10..0f5b6f27 100644
--- a/services/appflowy-collaborate/src/error.rs
+++ b/services/appflowy-collaborate/src/error.rs
@@ -1,3 +1,4 @@
+use app_error::AppError;
 use collab::error::CollabError;
 use collab_stream::error::StreamError;
 use std::fmt::Display;
@@ -61,6 +62,21 @@ pub enum RealtimeError {
   #[error("Collab redis stream error: {0}")]
   StreamError(#[from] StreamError),
 
+  #[error("Cannot create group: {0}")]
+  CannotCreateGroup(String),
+
+  #[error("BinCodeCollab error: {0}")]
+  BincodeEncode(String),
+
+  #[error("Failed to create snapshot: {0}")]
+  CreateSnapshotFailed(String),
+
+  #[error("Failed to get latest snapshot: {0}")]
+  GetLatestSnapshotFailed(String),
+
+  #[error("Collab Schema Error: {0}")]
+  CollabSchemaError(String),
+
   #[error("failed to obtain lease: {0}")]
   Lease(Box<dyn std::error::Error + Send + Sync>),
 }
diff --git a/services/appflowy-collaborate/src/group/group_init.rs b/services/appflowy-collaborate/src/group/group_init.rs
index a5d0f604..f46b5b2a 100644
--- a/services/appflowy-collaborate/src/group/group_init.rs
+++ b/services/appflowy-collaborate/src/group/group_init.rs
@@ -400,7 +400,7 @@ impl CollabGroup {
         .state
         .collab_type
         .validate_require_data(collab)
-        .map_err(|err| RealtimeError::Internal(err.into()))
+        .map_err(|err| RealtimeError::CollabSchemaError(err.to_string()))
     })?;
     Ok(encode_collab)
   }
@@ -1159,7 +1159,7 @@ impl CollabPersister {
         .storage
         .create_snapshot(params)
         .await
-        .map_err(|err| RealtimeError::Internal(err.into()))?;
+        .map_err(|err| RealtimeError::CreateSnapshotFailed(err.to_string()))?;
 
       // 2. Generate document state with GC turned on and save it.
       tx.force_gc();
@@ -1239,12 +1239,12 @@ impl CollabPersister {
           self.collab_type.clone(),
         )
         .await
-        .map_err(|err| RealtimeError::Internal(err.into()))?;
+        .map_err(|err| RealtimeError::GetLatestSnapshotFailed(err.to_string()))?;
       match snapshot {
         None => None,
         Some(snapshot) => {
           let encoded_collab = EncodedCollab::decode_from_bytes(&snapshot.encoded_collab_v1)
-            .map_err(|err| RealtimeError::Internal(err.into()))?;
+            .map_err(|err| RealtimeError::BincodeEncode(err.to_string()))?;
           Some(encoded_collab.doc_state)
         },
       }
diff --git a/services/appflowy-collaborate/src/group/manager.rs b/services/appflowy-collaborate/src/group/manager.rs
index 78b16bcb..9aa46840 100644
--- a/services/appflowy-collaborate/src/group/manager.rs
+++ b/services/appflowy-collaborate/src/group/manager.rs
@@ -134,7 +134,7 @@ where
       .transact()
       .state_vector(),
       Err(err) if err.is_record_not_found() => StateVector::default(),
-      Err(err) => return Err(RealtimeError::Internal(err.into())),
+      Err(err) => return Err(RealtimeError::CannotCreateGroup(err.to_string())),
     };
 
     trace!(