mf_collab/
sync_service.rs1use 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#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
11pub enum RoomStatus {
12 NotExists,
14 Created,
16 Initialized,
18 Shutting,
20 Offline,
22}
23
24#[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 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 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 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 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 if let Some(nodes_map) = txn.get_map("nodes") {
115 node_count = nodes_map.len(&txn);
116 }
117
118 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 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 if !self.yrs_manager.room_exists(room_id) {
147 tracing::warn!("🔄 尝试下线不存在的房间: {}", room_id);
148 return Ok(None);
149 }
150
151 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 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 final_snapshot = Some(RoomSnapshot {
171 room_id: room_id.to_string(),
172 root_id: "root".to_string(), nodes: std::collections::HashMap::new(),
174 version: 0,
175 });
176 }
177 }
178 }
179
180 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 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 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 pub fn get_active_rooms(&self) -> Vec<String> {
247 self.yrs_manager.get_active_rooms()
248 }
249
250 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 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}