Skip to main content

mf_collab/
sync_service.rs

1use std::sync::Arc;
2use yrs::{Map, ReadTxn as _, Transact};
3use serde::{Deserialize, Serialize};
4
5use crate::error::Result;
6use crate::yrs_manager::YrsManager;
7use crate::RoomSnapshot;
8
9/// 房间状态枚举
10#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
11pub enum RoomStatus {
12    /// 房间不存在
13    NotExists,
14    /// 房间已创建但未初始化数据
15    Created,
16    /// 房间已初始化并有数据
17    Initialized,
18    /// 房间正在下线中
19    Shutting,
20    /// 房间已下线
21    Offline,
22}
23
24/// 房间信息结构
25#[derive(Debug, Clone, Serialize, Deserialize)]
26pub struct RoomInfo {
27    pub room_id: String,
28    pub status: RoomStatus,
29    pub node_count: usize,
30    pub client_count: usize,
31    pub last_activity: std::time::SystemTime,
32}
33
34#[derive(Clone)]
35pub struct SyncService {
36    yrs_manager: Arc<YrsManager>,
37    client_id: String,
38}
39
40impl SyncService {
41    pub fn new(yrs_manager: Arc<YrsManager>) -> Self {
42        Self { yrs_manager, client_id: "server".to_string() }
43    }
44
45    /// 初始化房间,确保 Yrs 文档存在
46    pub fn init_room(
47        &self,
48        room_id: &str,
49    ) {
50        tracing::info!("🔄 初始化房间: {}", room_id);
51        self.yrs_manager.get_or_create_awareness(room_id);
52    }
53
54    /// 检查房间是否已初始化(有数据)
55    pub async fn is_room_initialized(
56        &self,
57        room_id: &str,
58    ) -> bool {
59        if let Some(awareness_ref) = self.yrs_manager.get_awareness_ref(room_id)
60        {
61            let awareness = awareness_ref.read().await;
62            let doc = awareness.doc();
63            let txn = doc.transact();
64
65            if let Some(nodes_map) = txn.get_map("nodes") {
66                nodes_map.len(&txn) > 0
67            } else {
68                false
69            }
70        } else {
71            false
72        }
73    }
74
75    /// 获取房间状态信息
76    pub async fn get_room_status(
77        &self,
78        room_id: &str,
79    ) -> RoomStatus {
80        if !self.yrs_manager.room_exists(room_id) {
81            return RoomStatus::NotExists;
82        }
83
84        if self.is_room_initialized(room_id).await {
85            RoomStatus::Initialized
86        } else {
87            RoomStatus::Created
88        }
89    }
90
91    /// 获取房间详细信息
92    pub async fn get_room_info(
93        &self,
94        room_id: &str,
95    ) -> Option<RoomInfo> {
96        if !self.yrs_manager.room_exists(room_id) {
97            return None;
98        }
99
100        let status = self.get_room_status(room_id).await;
101        let mut node_count = 0;
102        let mut client_count = 0;
103
104        if let Some(awareness_ref) = self.yrs_manager.get_awareness_ref(room_id)
105        {
106            if let Ok(awareness) = awareness_ref.try_read() {
107                let doc = awareness.doc();
108                let txn = doc.transact();
109
110                // 获取节点数量
111                if let Some(nodes_map) = txn.get_map("nodes") {
112                    node_count = nodes_map.len(&txn);
113                }
114
115                // 获取客户端数量
116                client_count = awareness.clients().len();
117            }
118        }
119
120        Some(RoomInfo {
121            room_id: room_id.to_string(),
122            status,
123            node_count: node_count as usize,
124            client_count,
125            last_activity: std::time::SystemTime::now(),
126        })
127    }
128
129    /// 房间下线 - 核心下线方法
130    /// 1. 断开所有客户端
131    /// 2. 可选保存数据
132    /// 3. 清理资源
133    pub async fn offline_room(
134        &self,
135        room_id: &str,
136        save_data: bool,
137    ) -> Result<Option<RoomSnapshot>> {
138        tracing::info!("🔄 开始下线房间: {}", room_id);
139
140        let mut final_snapshot = None;
141
142        // 1. 检查房间是否存在
143        if !self.yrs_manager.room_exists(room_id) {
144            tracing::warn!("🔄 尝试下线不存在的房间: {}", room_id);
145            return Ok(None);
146        }
147
148        // 2. 如果需要保存数据,先创建快照
149        if save_data {
150            if let Some(awareness_ref) =
151                self.yrs_manager.get_awareness_ref(room_id)
152            {
153                let awareness = awareness_ref.read().await;
154                let doc = awareness.doc();
155                let txn = doc.transact();
156
157                // 从 Yrs 文档重建 Tree 快照
158                if let Some(nodes_map) = txn.get_map("nodes") {
159                    let node_count = nodes_map.len(&txn);
160                    tracing::info!(
161                        "🔄 保存 {} 个节点 from room: {}",
162                        node_count,
163                        room_id
164                    );
165
166                    // 创建简化的快照(实际项目中可能需要完整的 Tree 重建)
167                    final_snapshot = Some(RoomSnapshot {
168                        room_id: room_id.to_string(),
169                        root_id: "root".to_string(), // 简化处理
170                        nodes: std::collections::HashMap::new(),
171                        version: 0,
172                    });
173                }
174            }
175        }
176
177        // 3. 从 YrsManager 中移除房间(这会自动断开客户端)
178        if let Some(_awareness_ref) =
179            self.yrs_manager.remove_room(room_id).await
180        {
181            tracing::info!("🔄 房间 '{}' 成功下线", room_id);
182        } else {
183            tracing::error!("🔄 从 YrsManager 中移除房间 '{}' 失败", room_id);
184            return Err(crate::error::TransmissionError::SyncError(format!(
185                "Failed to offline room: {room_id}"
186            )));
187        }
188
189        Ok(final_snapshot)
190    }
191
192    /// 强制房间下线(用于紧急情况)
193    pub async fn force_offline_room(
194        &self,
195        room_id: &str,
196    ) -> Result<bool> {
197        tracing::warn!("Force offlining room: {}", room_id);
198
199        let success = self.yrs_manager.force_cleanup_room(room_id).await;
200
201        if success {
202            tracing::info!("Room '{}' force offlined successfully", room_id);
203        } else {
204            tracing::error!("Failed to force offline room: {}", room_id);
205        }
206
207        Ok(success)
208    }
209
210    /// 批量下线房间
211    pub async fn offline_rooms(
212        &self,
213        room_ids: &[String],
214        save_data: bool,
215    ) -> Result<Vec<(String, Option<RoomSnapshot>)>> {
216        tracing::info!("🔄 批量下线 {} 个房间", room_ids.len());
217
218        let mut results = Vec::new();
219
220        for room_id in room_ids {
221            match self.offline_room(room_id, save_data).await {
222                Ok(snapshot) => {
223                    results.push((room_id.clone(), snapshot));
224                },
225                Err(e) => {
226                    tracing::error!("🔄 下线房间 '{}' 失败: {}", room_id, e);
227                    results.push((room_id.clone(), None));
228                },
229            }
230        }
231
232        tracing::info!(
233            "🔄 批量下线完成: {}/{} 个房间成功下线",
234            results.iter().filter(|(_, snapshot)| snapshot.is_some()).count(),
235            room_ids.len()
236        );
237
238        Ok(results)
239    }
240
241    /// 获取所有活跃房间列表
242    pub fn get_active_rooms(&self) -> Vec<String> {
243        self.yrs_manager.get_active_rooms()
244    }
245
246    /// 获取房间统计信息
247    pub async fn get_rooms_stats(&self) -> Vec<RoomInfo> {
248        let room_ids = self.get_active_rooms();
249        let mut stats = Vec::new();
250
251        for room_id in room_ids {
252            if let Some(info) = self.get_room_info(&room_id).await {
253                stats.push(info);
254            }
255        }
256
257        stats
258    }
259
260    /// 获取 YrsManager 的引用(用于高级操作)
261    pub fn yrs_manager(&self) -> &Arc<YrsManager> {
262        &self.yrs_manager
263    }
264}
265
266impl std::fmt::Debug for SyncService {
267    fn fmt(
268        &self,
269        f: &mut std::fmt::Formatter<'_>,
270    ) -> std::fmt::Result {
271        f.debug_struct("SyncService")
272            .field("client_id", &self.client_id)
273            .finish()
274    }
275}