mf_collab/
sync_service.rs

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