mf_collab/
sync_service.rs1use 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#[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 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 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 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 if let Some(nodes_map) = txn.get_map("nodes") {
112 node_count = nodes_map.len(&txn);
113 }
114
115 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 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 if !self.yrs_manager.room_exists(room_id) {
144 tracing::warn!("🔄 尝试下线不存在的房间: {}", room_id);
145 return Ok(None);
146 }
147
148 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 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 final_snapshot = Some(RoomSnapshot {
168 room_id: room_id.to_string(),
169 root_id: "root".to_string(), nodes: std::collections::HashMap::new(),
171 version: 0,
172 });
173 }
174 }
175 }
176
177 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 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 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 pub fn get_active_rooms(&self) -> Vec<String> {
243 self.yrs_manager.get_active_rooms()
244 }
245
246 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 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}