moduforge_collaboration/
sync_service.rs

1use std::sync::Arc;
2use moduforge_model::tree::Tree;
3use moduforge_state::{State, Transaction};
4use yrs::{ReadTxn, Transact};
5use crate::{
6    Result, 
7    YrsManager, 
8    WebSocketServer, 
9    RoomSnapshot,
10    mapping::Mapper,
11    TransmissionError,
12};
13use crate::ws_server::WsMessage;
14use yrs::updates::decoder::Decode;
15
16#[derive(Clone)]
17pub struct SyncService {
18    yrs_manager: Arc<YrsManager>,
19    ws_server: Arc<WebSocketServer>,
20}
21
22impl SyncService {
23    pub fn new(yrs_manager: Arc<YrsManager>, ws_server: Arc<WebSocketServer>) -> Self {
24        Self {
25            yrs_manager,
26            ws_server,
27        }
28    }
29
30    /// 将 ModuForge 的 Tree 初始化为房间快照
31    pub fn init_room_from_tree(&self, room_id: &str, _tree: &Tree) -> Result<()> {
32        tracing::info!("初始化房间 {} 从 Tree", room_id);
33        
34        // 创建房间（空的 Yrs 文档）
35        let _doc = self.yrs_manager.get_or_create_doc(room_id);
36        
37        // 这里不存储全量数据到 Yrs，只初始化空文档
38        // 前端需要全量数据时通过 get_room_snapshot 获取
39        
40        Ok(())
41    }
42
43    /// 获取房间的完整快照（用于前端首次加载）
44    pub fn get_room_snapshot(&self, room_id: &str, tree: &Tree) -> RoomSnapshot {
45        tracing::debug!("获取房间 {} 的快照", room_id);
46        Mapper::tree_to_snapshot(tree, room_id.to_string())
47    }
48
49    /// 🚀 内部方法：处理事务变更并以二进制格式批量推送到前端
50    pub async fn handle_transaction_applied(
51        &self,
52        room_id: &str,
53        transactions: &[Transaction],
54        _new_state: &State,
55        client_id: Option<String>,
56    ) -> Result<()> {
57        if transactions.is_empty() {
58            return Ok(());
59        }
60
61        tracing::info!("检测到 {} 个事务变更，开始批量同步到房间 {}", transactions.len(), room_id);
62        
63        let client_id = client_id.unwrap_or_else(|| "server".to_string());
64        let doc = self.yrs_manager.get_or_create_doc(room_id);
65        let registry = Mapper::global_registry();
66
67        // The update is generated from the transaction, so we need to capture it.
68        let update: Vec<u8>; 
69        { // Scoped to confine the transaction
70            let mut txn = doc.transact_mut();
71            for (i, transaction) in transactions.iter().enumerate() {
72                tracing::debug!("处理事务 {}/{}: {} 个操作", i + 1, transactions.len(), transaction.steps.len());
73                for (j, step) in transaction.steps.iter().enumerate() {
74                    let type_name = std::any::type_name_of_val(step.as_ref());
75                    tracing::debug!("  - 操作 {}/{}: {}", j + 1, transaction.steps.len(), type_name);
76                    
77                    if let Some(converter) = registry.find_converter(step.as_ref()) {
78                        if let Err(e) = converter.apply_to_yrs_txn(step.as_ref(), &mut txn, &client_id) {
79                            tracing::error!("应用Step到Yrs事务失败: {}", e);
80                        }
81                    } else {
82                        tracing::warn!("未找到Step转换器: {}", type_name);
83                    }
84                }
85            }
86            // At the end of the transaction, encode the changes made *within this transaction*.
87            update = txn.encode_update_v1();
88        } // Yrs transaction commits here
89
90        // 🚀 主动推送 update 给连接的客户端
91        if !update.is_empty() {
92            self.ws_server.broadcast_binary_to_room(room_id, update, Some(client_id))?;
93        }
94
95        // 发送通知消息
96        /* let change_msg = format!("应用了 {} 个事务，总共 {} 个操作", 
97            transactions.len(), 
98            transactions.iter().map(|t| t.steps.len()).sum::<usize>()
99        );
100        
101        if let Err(e) = self.notify_room_change(room_id, change_msg) {
102            tracing::error!("发送变更通知失败: {}", e);
103        } */
104
105        Ok(())
106    }
107
108
109    /// 移除房间
110    pub fn remove_room(&self, room_id: &str) {
111        tracing::info!("移除房间: {}", room_id);
112        if self.yrs_manager.remove_doc(room_id).is_some() {
113            tracing::info!("房间 {} 的 Yrs Doc 已成功移除", room_id);
114        } else {
115            tracing::warn!("尝试移除房间 {} 的 Yrs Doc，但未找到", room_id);
116        }
117    }
118
119    /// 获取服务状态
120    pub fn get_status(&self) -> SyncServiceStatus {
121        SyncServiceStatus {
122            client_count: self.ws_server.client_count(),
123            room_count: self.ws_server.room_count(),
124            rooms: self.yrs_manager.list_rooms(),
125        }
126    }
127
128    /// 获取 YrsManager 的引用（用于高级操作）
129    pub fn yrs_manager(&self) -> &Arc<YrsManager> {
130        &self.yrs_manager
131    }
132
133    /// 获取 WebSocketServer 的引用（用于高级操作）
134    pub fn ws_server(&self) -> &Arc<WebSocketServer> {
135        &self.ws_server
136    }
137
138    /// 🚀 主动推送消息到房间的所有客户端
139    pub fn broadcast_message_to_room(&self, room_id: &str, message: String) -> Result<()> {
140        self.ws_server.broadcast_to_room(room_id, message, None)
141    }
142
143    /// 🚀 主动推送二进制数据到房间的所有客户端
144    pub fn broadcast_data_to_room(&self, room_id: &str, data: Vec<u8>) -> Result<()> {
145        self.ws_server.broadcast_binary_to_room(room_id, data, None)
146    }
147
148    /// 🚀 主动推送消息到特定客户端
149    pub fn send_message_to_client(&self, client_id: &str, message: String) -> Result<()> {
150        self.ws_server.send_to_client(client_id, message)
151    }
152
153    /// 🚀 主动推送二进制数据到特定客户端
154    pub fn send_data_to_client(&self, client_id: &str, data: Vec<u8>) -> Result<()> {
155        self.ws_server.send_binary_to_client(client_id, data)
156    }
157
158    /// 🚀 主动推送 ModuForge 变更通知到房间
159    pub fn notify_room_change(&self, room_id: &str, change_description: String) -> Result<()> {
160        let notification = WsMessage::Notification { 
161            message: change_description
162        };
163        
164        let json = serde_json::to_string(&notification)?;
165        self.broadcast_message_to_room(room_id, json)
166    }
167
168    /// 🚀 为新加入的客户端处理同步请求
169    /// 1. 从权威State生成一个临时的Doc快照
170    /// 2. 将当前房间的增量更新应用到快照上
171    /// 3. 根据客户端的状态向量计算并返回最终的diff
172    pub async fn handle_sync_request(
173        &self,
174        room_id: &str,
175        tree: &Tree,
176        client_state_vector: &[u8],
177    ) -> Result<Vec<u8>> {
178        // A: 从权威State生成一个临时的Doc快照
179        let snapshot_doc = Mapper::tree_to_yrs_doc(tree);
180
181        // B: 获取当前房间的在线增量Doc
182        if let Some(online_doc) = self.yrs_manager.get_doc(room_id) {
183            // C: 将在线Doc的更新应用到快照上
184            let online_update = online_doc.transact().encode_state_as_update_v1(&Default::default());
185            let mut snapshot_txn = snapshot_doc.transact_mut();
186            let update_decoded = yrs::Update::decode_v1(&online_update)
187                .map_err(|e| TransmissionError::YrsError(format!("Failed to decode online doc update: {}", e)))?;
188            snapshot_txn.apply_update(update_decoded)?;
189        }
190
191        // D: 根据客户端的状态向量计算并返回最终的diff
192        let client_sv = yrs::StateVector::decode_v1(client_state_vector)
193            .map_err(|e| TransmissionError::YrsError(e.to_string()))?;
194        
195        let final_diff = snapshot_doc.transact().encode_diff_v1(&client_sv);
196        
197        Ok(final_diff)
198    }
199
200    /// 🚀 主动推送 JSON 格式的状态同步消息到房间
201    pub fn broadcast_json_sync_to_room(&self, room_id: &str, operation: &str, data: serde_json::Value) -> Result<()> {
202        let sync_message = WsMessage::StateSync {
203            room_id: room_id.to_string(),
204            operation: operation.to_string(),
205            data,
206            timestamp: std::time::SystemTime::now()
207                .duration_since(std::time::UNIX_EPOCH)
208                .unwrap()
209                .as_millis() as u64,
210        };
211        
212        let json = serde_json::to_string(&sync_message)?;
213        self.broadcast_message_to_room(room_id, json)
214    }
215}
216
217// 添加 Debug trait 实现
218impl std::fmt::Debug for SyncService {
219    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
220        f.debug_struct("SyncService")
221            .field("yrs_manager", &"YrsManager")
222            .field("ws_server", &"WebSocketServer")
223            .finish()
224    }
225}
226
227impl Default for SyncService {
228    fn default() -> Self {
229        let yrs_manager = Arc::new(YrsManager::new());
230        let ws_server = Arc::new(WebSocketServer::new(yrs_manager.clone()));
231        Self::new(yrs_manager, ws_server)
232    }
233}
234
235#[derive(Debug, Clone)]
236pub struct SyncServiceStatus {
237    pub client_count: usize,
238    pub room_count: usize,
239    pub rooms: Vec<String>,
240}
241
242#[cfg(test)]
243mod tests {
244    use super::*;
245    use moduforge_model::{node::Node, attrs::Attrs};
246    use std::net::{IpAddr, Ipv4Addr};
247
248    #[tokio::test]
249    async fn test_sync_service_creation() {
250        let service = SyncService::default();
251        let status = service.get_status();
252        
253        assert_eq!(status.client_count, 0);
254        assert_eq!(status.room_count, 0);
255        assert!(status.rooms.is_empty());
256    }
257}
moduforge_collaboration/sync_service.rs

moduforge_collaboration/
sync_service.rs
