tdm_server_rust/collaboration/
hub.rs

1//! 在线翻译协作 Hub 核心逻辑
2//!
3//! 移植自 Poprako 的“页面级独占编辑 + 整页快照广播”协作模型：
4//!
5//! - **在线状态**：项目（话数）内每个连接一条在线记录，存 Redis 哈希 `hub:{project}:presence`
6//! - **页面锁**：每页一把独占锁，Redis 字符串 `hub:{project}:lock:{page}`（`SET NX EX` 抢占 + 续约）
7//! - **页面快照**：整页翻校单元快照，Redis 字符串 `hub:{project}:snapshot:{page}`，供重连/跨实例恢复
8//! - **跨实例广播**：本地 DashMap 直发 + Redis pub/sub 频道 `hub:fanout` 转发到其他实例
9//! - **落库**：收到快照后 debounce 写入 `editor_unit`（整页替换）
10//!
11//! ## 键命名约定
12//!
13//! - `project_key` 形如 `episode:{episodeId}`，末段解析为分片键 `episode_id`
14//! - `page_key` 形如 `page:{pageId}`，末段解析为 `editor_page.id`
15
16use crate::app::AppState;
17use crate::collaboration::signalr::{encode_message, HubMessage};
18use crate::entity::editor::EditorUnitInput;
19use crate::repository::editor_page_repo::EditorPageRepository;
20use crate::repository::editor_unit_repo::EditorUnitRepository;
21use dashmap::DashMap;
22use serde::{Deserialize, Serialize};
23use serde_json::{json, Value};
24use std::sync::atomic::{AtomicU64, Ordering};
25use std::sync::Arc;
26use std::time::Duration;
27use tokio::sync::mpsc;
28
29/// 页面锁默认存活秒数（活跃时续约）
30const LOCK_TTL_SECS: u64 = 120;
31/// 页面快照存活秒数
32const SNAPSHOT_TTL_SECS: u64 = 6 * 3600;
33/// 在线状态哈希存活秒数（活跃时刷新）
34const PRESENCE_TTL_SECS: u64 = 3600;
35/// 快照落库 debounce 毫秒
36const PERSIST_DEBOUNCE_MS: u64 = 1500;
37/// 跨实例广播 Redis 频道（逻辑名，实际加前缀）
38const FANOUT_CHANNEL: &str = "hub:fanout";
39
40/// 在线编辑者状态（与前端 `TranslatorPageEditorState` 对齐，snake_case）
41#[derive(Clone, Debug, Default, Deserialize, Serialize)]
42#[serde(default)]
43pub struct PageEditorState {
44    /// 用户 ID（组员 ID 字符串）
45    pub user_id: String,
46    /// 展示名称
47    pub display_name: String,
48    /// 头像 URL
49    pub avatar_url: Option<String>,
50    /// 页面键
51    pub page_key: String,
52    /// 页面序号
53    pub page_index: i32,
54    /// 页面名称
55    pub page_name: String,
56    /// 编辑模式（translate / proofread）
57    pub mode: String,
58    /// 编辑状态（viewing / lock_holder）
59    pub editor_state: String,
60    /// 获取时间戳（毫秒）
61    pub acquired_at: i64,
62}
63
64/// 项目协作状态（项目内全部在线编辑者）
65#[derive(Clone, Debug, Default, Deserialize, Serialize)]
66pub struct ProjectState {
67    /// 项目键
68    pub project_key: String,
69    /// 在线编辑者列表
70    pub page_editors: Vec<PageEditorState>,
71}
72
73/// 页面整页快照
74#[derive(Clone, Debug, Default, Deserialize, Serialize)]
75#[serde(default)]
76pub struct PageSnapshot {
77    /// 项目键
78    pub project_key: String,
79    /// 页面键
80    pub page_key: String,
81    /// 页面序号
82    pub page_index: i32,
83    /// 页面名称
84    pub page_name: String,
85    /// 编辑模式
86    pub mode: String,
87    /// 更新者用户 ID
88    pub updated_by_user_id: String,
89    /// 更新者展示名称
90    pub updated_by_display_name: String,
91    /// 更新时间戳（毫秒）
92    pub updated_at: i64,
93    /// 页面翻校单元快照（透传 JSON）
94    pub units: Vec<Value>,
95}
96
97/// 尝试获取页面锁结果
98#[derive(Clone, Debug, Serialize)]
99pub struct TryAcquireResult {
100    /// 是否获取成功
101    pub acquired: bool,
102    /// 当前锁持有者
103    pub editor: Option<PageEditorState>,
104}
105
106/// 加入协作项目请求
107#[derive(Clone, Debug, Default, Deserialize)]
108#[serde(default)]
109struct JoinProjectArgs {
110    /// 项目键
111    project_key: String,
112    /// 展示名称
113    display_name: String,
114    /// 头像 URL
115    avatar_url: Option<String>,
116}
117
118/// 打开页面 / 同步页面快照请求
119#[derive(Clone, Debug, Default, Deserialize)]
120#[serde(default)]
121struct OpenPageArgs {
122    /// 项目键
123    project_key: String,
124    /// 页面键
125    page_key: String,
126    /// 页面序号
127    page_index: i32,
128    /// 页面名称
129    page_name: String,
130    /// 编辑模式
131    mode: String,
132    /// 页面翻校单元快照（透传 JSON）
133    units: Vec<Value>,
134}
135
136/// 尝试获取页面锁请求
137#[derive(Clone, Debug, Default, Deserialize)]
138#[serde(default)]
139struct TryAcquireArgs {
140    /// 项目键
141    project_key: String,
142    /// 页面键
143    page_key: String,
144    /// 页面序号
145    page_index: i32,
146    /// 页面名称
147    page_name: String,
148    /// 编辑模式
149    mode: String,
150}
151
152/// 更新页面锁模式请求
153#[derive(Clone, Debug, Default, Deserialize)]
154#[serde(default)]
155struct UpdateModeArgs {
156    /// 项目键
157    project_key: String,
158    /// 页面键
159    page_key: String,
160    /// 编辑模式
161    mode: String,
162}
163
164/// 释放页面锁请求
165#[derive(Clone, Debug, Default, Deserialize)]
166#[serde(default)]
167struct ReleaseArgs {
168    /// 项目键
169    project_key: String,
170    /// 页面键
171    page_key: String,
172}
173
174/// Redis 页面锁持有者载荷
175#[derive(Clone, Debug, Deserialize, Serialize)]
176struct LockHolder {
177    /// 持锁连接 ID
178    connection_id: String,
179    /// 持锁编辑者状态
180    editor: PageEditorState,
181}
182
183/// 单个连接状态
184struct ConnState {
185    /// 用户 ID（来自 JWT，权威）
186    user_id: String,
187    /// 展示名称（来自 JoinProject）
188    display_name: String,
189    /// 头像 URL
190    avatar_url: Option<String>,
191    /// 当前所在项目键
192    project_key: Option<String>,
193    /// 当前打开的页面键
194    page_key: Option<String>,
195    /// 当前持锁页面键
196    lock_page_key: Option<String>,
197    /// 出站消息发送端
198    outbound: mpsc::UnboundedSender<String>,
199}
200
201/// 连接身份快照（同步读取后用于异步流程，避免持锁跨 await）
202#[derive(Clone)]
203struct ConnIdentity {
204    /// 用户 ID
205    user_id: String,
206    /// 展示名称
207    display_name: String,
208    /// 头像 URL
209    avatar_url: Option<String>,
210    /// 当前项目键（保留用于诊断与后续扩展）
211    #[allow(dead_code)]
212    project_key: Option<String>,
213    /// 当前持锁页面键
214    lock_page_key: Option<String>,
215}
216
217/// 协作连接注册表（本节点）
218pub struct CollaborationRegistry {
219    /// 连接表：connection_id -> 连接状态
220    connections: DashMap<String, ConnState>,
221    /// 页面快照落库 debounce 世代：page_key -> 世代计数
222    persist_gen: DashMap<String, Arc<AtomicU64>>,
223}
224
225impl Default for CollaborationRegistry {
226    fn default() -> Self {
227        Self::new()
228    }
229}
230
231impl CollaborationRegistry {
232    /// 创建空注册表
233    pub fn new() -> Self {
234        Self {
235            connections: DashMap::new(),
236            persist_gen: DashMap::new(),
237        }
238    }
239
240    /// 注册新连接，返回出站消息接收端
241    pub fn register(
242        &self,
243        connection_id: String,
244        user_id: String,
245    ) -> mpsc::UnboundedReceiver<String> {
246        let (tx, rx) = mpsc::unbounded_channel();
247        self.connections.insert(
248            connection_id,
249            ConnState {
250                user_id,
251                display_name: String::new(),
252                avatar_url: None,
253                project_key: None,
254                page_key: None,
255                lock_page_key: None,
256                outbound: tx,
257            },
258        );
259        rx
260    }
261
262    /// 获取连接出站发送端克隆（用于握手/应答/心跳回写）
263    pub fn sender(&self, connection_id: &str) -> Option<mpsc::UnboundedSender<String>> {
264        self.connections
265            .get(connection_id)
266            .map(|c| c.outbound.clone())
267    }
268
269    /// 同步读取连接身份快照
270    fn identity(&self, connection_id: &str) -> Option<ConnIdentity> {
271        self.connections.get(connection_id).map(|c| ConnIdentity {
272            user_id: c.user_id.clone(),
273            display_name: c.display_name.clone(),
274            avatar_url: c.avatar_url.clone(),
275            project_key: c.project_key.clone(),
276            lock_page_key: c.lock_page_key.clone(),
277        })
278    }
279
280    /// 向本节点同项目连接直发文本帧（可排除指定连接）
281    fn local_broadcast(&self, project_key: &str, frame: &str, exclude: Option<&str>) {
282        for entry in self.connections.iter() {
283            if Some(entry.key().as_str()) == exclude {
284                continue;
285            }
286            if entry.value().project_key.as_deref() == Some(project_key) {
287                let _ = entry.value().outbound.send(frame.to_string());
288            }
289        }
290    }
291}
292
293/// 取毫秒时间戳
294fn now_ms() -> i64 {
295    chrono::Utc::now().timestamp_millis()
296}
297
298/// 在线状态哈希逻辑键
299fn presence_key(project: &str) -> String {
300    format!("hub:{project}:presence")
301}
302
303/// 页面锁逻辑键
304fn lock_key(project: &str, page: &str) -> String {
305    format!("hub:{project}:lock:{page}")
306}
307
308/// 页面快照逻辑键
309fn snapshot_key(project: &str, page: &str) -> String {
310    format!("hub:{project}:snapshot:{page}")
311}
312
313/// 从 `episode:{id}` 解析分片键 `episode_id`
314fn parse_episode_id(project_key: &str) -> Option<i32> {
315    project_key
316        .rsplit(':')
317        .next()
318        .and_then(|s| s.trim().parse().ok())
319}
320
321/// 从 `page:{id}` 解析 `editor_page.id`
322fn parse_page_id(page_key: &str) -> Option<i64> {
323    page_key
324        .rsplit(':')
325        .next()
326        .and_then(|s| s.trim().parse().ok())
327}
328
329/// 规范化模式取值
330fn normalize_mode(mode: &str) -> String {
331    if mode == "proofread" {
332        "proofread".to_string()
333    } else {
334        "translate".to_string()
335    }
336}
337
338/// 启动跨实例广播订阅（多实例部署时转发事件到本节点连接）
339pub fn spawn_fanout(state: AppState) {
340    let redis = state.redis.clone();
341    let registry = state.collaboration.clone();
342    let self_id = redis.sender_id().to_string();
343    tokio::spawn(async move {
344        redis
345            .subscribe_loop(FANOUT_CHANNEL, move |payload| {
346                let Ok(v) = serde_json::from_str::<Value>(&payload) else {
347                    return;
348                };
349                // 跳过本节点自身发出的消息（本地已直发，避免重复）
350                if v.get("sender").and_then(Value::as_str) == Some(self_id.as_str()) {
351                    return;
352                }
353                let project = v.get("project").and_then(Value::as_str).unwrap_or("");
354                let frame = v.get("frame").and_then(Value::as_str).unwrap_or("");
355                if !project.is_empty() && !frame.is_empty() {
356                    registry.local_broadcast(project, frame, None);
357                }
358            })
359            .await;
360    });
361}
362
363/// 本地直发 + 跨实例广播一条事件
364async fn broadcast(
365    state: &AppState,
366    project: &str,
367    target: &str,
368    payload: Value,
369    exclude: Option<&str>,
370) {
371    let Ok(frame) = encode_message(&HubMessage::event(target, payload)) else {
372        return;
373    };
374    state
375        .collaboration
376        .local_broadcast(project, &frame, exclude);
377    let envelope = json!({
378        "sender": state.redis.sender_id(),
379        "project": project,
380        "frame": frame,
381    });
382    state
383        .redis
384        .publish(FANOUT_CHANNEL, &envelope.to_string())
385        .await;
386}
387
388/// 读取项目当前在线状态
389async fn get_project_state(state: &AppState, project: &str) -> ProjectState {
390    let entries = state.redis.hgetall(&presence_key(project)).await;
391    let page_editors = entries
392        .into_iter()
393        .filter_map(|(_, v)| serde_json::from_str::<PageEditorState>(&v).ok())
394        .collect();
395    ProjectState {
396        project_key: project.to_string(),
397        page_editors,
398    }
399}
400
401/// 写入/刷新连接在线状态
402async fn upsert_presence(state: &AppState, project: &str, conn_id: &str, editor: &PageEditorState) {
403    if let Ok(json) = serde_json::to_string(editor) {
404        state
405            .redis
406            .hset_ttl(&presence_key(project), conn_id, &json, PRESENCE_TTL_SECS)
407            .await;
408    }
409}
410
411/// 读取指定连接在线状态
412async fn get_presence(state: &AppState, project: &str, conn_id: &str) -> Option<PageEditorState> {
413    state
414        .redis
415        .hgetall(&presence_key(project))
416        .await
417        .into_iter()
418        .find(|(field, _)| field == conn_id)
419        .and_then(|(_, v)| serde_json::from_str::<PageEditorState>(&v).ok())
420}
421
422/// 把同页其他连接的持锁状态降级为浏览（保证一页仅一个 lock_holder）
423async fn demote_other_holders(state: &AppState, project: &str, page: &str, keep_conn: &str) {
424    for (field, v) in state.redis.hgetall(&presence_key(project)).await {
425        if field == keep_conn {
426            continue;
427        }
428        if let Ok(mut ed) = serde_json::from_str::<PageEditorState>(&v) {
429            if ed.page_key == page && ed.editor_state == "lock_holder" {
430                ed.editor_state = "viewing".to_string();
431                upsert_presence(state, project, &field, &ed).await;
432            }
433        }
434    }
435}
436
437/// 广播项目状态更新（项目内所有连接）
438async fn broadcast_project_state(state: &AppState, project: &str) {
439    let st = get_project_state(state, project).await;
440    broadcast(state, project, "ProjectStateUpdated", json!(st), None).await;
441}
442
443/// 分发一条客户端 Hub 调用
444///
445/// ## 返回
446/// - `Ok(Some(value))`：带返回值的应答
447/// - `Ok(None)`：无返回值（仍回 Completion）
448/// - `Err(msg)`：调用失败（回 Completion error）
449pub async fn handle_invocation(
450    state: &AppState,
451    conn_id: &str,
452    target: &str,
453    args: Vec<Value>,
454) -> Result<Option<Value>, String> {
455    let arg0 = args.into_iter().next().unwrap_or(Value::Null);
456    match target {
457        "JoinProject" => join_project(state, conn_id, arg0).await.map(Some),
458        "OpenPage" => open_page(state, conn_id, arg0).await.map(Some),
459        "TryAcquirePageLock" => try_acquire_lock(state, conn_id, arg0).await.map(Some),
460        "UpdatePageLockMode" => {
461            update_lock_mode(state, conn_id, arg0).await?;
462            Ok(None)
463        }
464        "ReleaseCurrentPageLock" => {
465            release_lock(state, conn_id, arg0).await?;
466            Ok(None)
467        }
468        "SyncPageSnapshot" => {
469            sync_snapshot(state, conn_id, arg0).await?;
470            Ok(None)
471        }
472        "SyncPageLiveDelta" => {
473            sync_live_delta(state, conn_id, arg0).await?;
474            Ok(None)
475        }
476        other => Err(format!("不支持的协作方法：{other}")),
477    }
478}
479
480/// 处理连接断开：释放锁、移除在线状态、广播
481pub async fn disconnect(state: &AppState, conn_id: &str) {
482    let info = state
483        .collaboration
484        .connections
485        .get(conn_id)
486        .map(|c| (c.project_key.clone(), c.lock_page_key.clone()));
487    state.collaboration.connections.remove(conn_id);
488    if let Some((Some(project), lock_page)) = info {
489        cleanup_in_project(state, conn_id, &project, lock_page).await;
490    }
491}
492
493/// 清理某连接在某项目内的锁与在线状态并广播
494async fn cleanup_in_project(
495    state: &AppState,
496    conn_id: &str,
497    project: &str,
498    lock_page: Option<String>,
499) {
500    if let Some(page) = lock_page {
501        state.redis.del(&lock_key(project, &page)).await;
502    }
503    state.redis.hdel(&presence_key(project), conn_id).await;
504    broadcast_project_state(state, project).await;
505}
506
507/// 加入协作项目
508async fn join_project(state: &AppState, conn_id: &str, arg: Value) -> Result<Value, String> {
509    let req: JoinProjectArgs = serde_json::from_value(arg).map_err(|e| e.to_string())?;
510    let project = req.project_key.trim().to_string();
511    if project.is_empty() {
512        return Err("缺少 project_key".to_string());
513    }
514    let display = if req.display_name.trim().is_empty() {
515        "协作成员".to_string()
516    } else {
517        req.display_name.trim().to_string()
518    };
519
520    // 同步更新连接身份与项目键，捕获旧项目用于清理
521    let (old_project, old_lock) = {
522        let mut c = state
523            .collaboration
524            .connections
525            .get_mut(conn_id)
526            .ok_or("连接不存在")?;
527        let old_project = c.project_key.clone();
528        let old_lock = c.lock_page_key.clone();
529        c.display_name = display;
530        c.avatar_url = req.avatar_url.clone();
531        if old_project.as_deref() != Some(project.as_str()) {
532            c.project_key = Some(project.clone());
533            c.page_key = None;
534            c.lock_page_key = None;
535        }
536        (old_project, old_lock)
537    };
538
539    // 切换项目：清理旧项目内的锁与在线状态
540    if let Some(op) = old_project {
541        if op != project {
542            cleanup_in_project(state, conn_id, &op, old_lock).await;
543        }
544    }
545
546    let st = get_project_state(state, &project).await;
547    Ok(json!(st))
548}
549
550/// 打开页面：登记在线状态并返回最新快照（无则用入参建立）
551async fn open_page(state: &AppState, conn_id: &str, arg: Value) -> Result<Value, String> {
552    let req: OpenPageArgs = serde_json::from_value(arg).map_err(|e| e.to_string())?;
553    let ident = state.collaboration.identity(conn_id).ok_or("连接不存在")?;
554    let project = req.project_key.trim().to_string();
555    if project.is_empty() || req.page_key.is_empty() {
556        return Err("缺少 project_key 或 page_key".to_string());
557    }
558
559    // 记录连接当前页面
560    {
561        if let Some(mut c) = state.collaboration.connections.get_mut(conn_id) {
562            c.page_key = Some(req.page_key.clone());
563        }
564    }
565
566    let holds_lock = ident.lock_page_key.as_deref() == Some(req.page_key.as_str());
567    let editor = PageEditorState {
568        user_id: ident.user_id.clone(),
569        display_name: ident.display_name.clone(),
570        avatar_url: ident.avatar_url.clone(),
571        page_key: req.page_key.clone(),
572        page_index: req.page_index,
573        page_name: req.page_name.clone(),
574        mode: normalize_mode(&req.mode),
575        editor_state: if holds_lock { "lock_holder" } else { "viewing" }.to_string(),
576        acquired_at: now_ms(),
577    };
578    upsert_presence(state, &project, conn_id, &editor).await;
579    broadcast_project_state(state, &project).await;
580
581    // 返回已有快照，否则用入参建立并落地 Redis
582    if let Some(existing) = state
583        .redis
584        .get_str(&snapshot_key(&project, &req.page_key))
585        .await
586    {
587        if let Ok(v) = serde_json::from_str::<Value>(&existing) {
588            return Ok(v);
589        }
590    }
591    if req.units.is_empty() {
592        return Ok(Value::Null);
593    }
594    let snapshot = PageSnapshot {
595        project_key: project.clone(),
596        page_key: req.page_key.clone(),
597        page_index: req.page_index,
598        page_name: req.page_name.clone(),
599        mode: normalize_mode(&req.mode),
600        updated_by_user_id: ident.user_id.clone(),
601        updated_by_display_name: ident.display_name.clone(),
602        updated_at: now_ms(),
603        units: req.units.clone(),
604    };
605    if let Ok(json) = serde_json::to_string(&snapshot) {
606        state
607            .redis
608            .set_str_ttl(
609                &snapshot_key(&project, &req.page_key),
610                &json,
611                SNAPSHOT_TTL_SECS,
612            )
613            .await;
614    }
615    Ok(json!(snapshot))
616}
617
618/// 尝试获取页面锁
619async fn try_acquire_lock(state: &AppState, conn_id: &str, arg: Value) -> Result<Value, String> {
620    let req: TryAcquireArgs = serde_json::from_value(arg).map_err(|e| e.to_string())?;
621    let ident = state.collaboration.identity(conn_id).ok_or("连接不存在")?;
622    let project = req.project_key.trim().to_string();
623    if project.is_empty() || req.page_key.is_empty() {
624        return Err("缺少 project_key 或 page_key".to_string());
625    }
626
627    let editor = PageEditorState {
628        user_id: ident.user_id.clone(),
629        display_name: ident.display_name.clone(),
630        avatar_url: ident.avatar_url.clone(),
631        page_key: req.page_key.clone(),
632        page_index: req.page_index,
633        page_name: req.page_name.clone(),
634        mode: normalize_mode(&req.mode),
635        editor_state: "lock_holder".to_string(),
636        acquired_at: now_ms(),
637    };
638    let holder = LockHolder {
639        connection_id: conn_id.to_string(),
640        editor: editor.clone(),
641    };
642    let holder_json = serde_json::to_string(&holder).map_err(|e| e.to_string())?;
643    let lk = lock_key(&project, &req.page_key);
644
645    let acquired = if state
646        .redis
647        .set_nx_ex(&lk, &holder_json, LOCK_TTL_SECS)
648        .await
649    {
650        true
651    } else {
652        match state
653            .redis
654            .get_str(&lk)
655            .await
656            .and_then(|v| serde_json::from_str::<LockHolder>(&v).ok())
657        {
658            // 本连接已持锁：续约
659            Some(cur) if cur.connection_id == conn_id => {
660                state
661                    .redis
662                    .set_str_ttl(&lk, &holder_json, LOCK_TTL_SECS)
663                    .await;
664                true
665            }
666            // 被他人持有：返回当前持有者
667            Some(cur) => {
668                return Ok(json!(TryAcquireResult {
669                    acquired: false,
670                    editor: Some(cur.editor),
671                }));
672            }
673            // 锁数据损坏：覆盖抢占
674            None => {
675                state
676                    .redis
677                    .set_str_ttl(&lk, &holder_json, LOCK_TTL_SECS)
678                    .await;
679                true
680            }
681        }
682    };
683
684    if acquired {
685        // 释放本连接在其它页面上的旧锁
686        if let Some(old) = ident.lock_page_key.as_ref() {
687            if old != &req.page_key {
688                state.redis.del(&lock_key(&project, old)).await;
689            }
690        }
691        if let Some(mut c) = state.collaboration.connections.get_mut(conn_id) {
692            c.lock_page_key = Some(req.page_key.clone());
693            c.page_key = Some(req.page_key.clone());
694        }
695        demote_other_holders(state, &project, &req.page_key, conn_id).await;
696        upsert_presence(state, &project, conn_id, &editor).await;
697        broadcast_project_state(state, &project).await;
698    }
699
700    Ok(json!(TryAcquireResult {
701        acquired,
702        editor: Some(editor),
703    }))
704}
705
706/// 更新页面锁编辑模式
707async fn update_lock_mode(state: &AppState, conn_id: &str, arg: Value) -> Result<(), String> {
708    let req: UpdateModeArgs = serde_json::from_value(arg).map_err(|e| e.to_string())?;
709    let ident = state.collaboration.identity(conn_id).ok_or("连接不存在")?;
710    let project = req.project_key.trim().to_string();
711    if ident.lock_page_key.as_deref() != Some(req.page_key.as_str()) {
712        return Ok(());
713    }
714    if let Some(mut editor) = get_presence(state, &project, conn_id).await {
715        editor.mode = normalize_mode(&req.mode);
716        editor.editor_state = "lock_holder".to_string();
717        upsert_presence(state, &project, conn_id, &editor).await;
718    }
719    state
720        .redis
721        .expire(&lock_key(&project, &req.page_key), LOCK_TTL_SECS)
722        .await;
723    broadcast_project_state(state, &project).await;
724    Ok(())
725}
726
727/// 释放当前页面锁
728async fn release_lock(state: &AppState, conn_id: &str, arg: Value) -> Result<(), String> {
729    let req: ReleaseArgs = serde_json::from_value(arg).map_err(|e| e.to_string())?;
730    let ident = state.collaboration.identity(conn_id).ok_or("连接不存在")?;
731    let project = req.project_key.trim().to_string();
732    if ident.lock_page_key.as_deref() != Some(req.page_key.as_str()) {
733        return Ok(());
734    }
735    state.redis.del(&lock_key(&project, &req.page_key)).await;
736    if let Some(mut c) = state.collaboration.connections.get_mut(conn_id) {
737        c.lock_page_key = None;
738    }
739    if let Some(mut editor) = get_presence(state, &project, conn_id).await {
740        editor.editor_state = "viewing".to_string();
741        upsert_presence(state, &project, conn_id, &editor).await;
742    }
743    broadcast_project_state(state, &project).await;
744    Ok(())
745}
746
747/// 同步页面快照：存 Redis、广播、debounce 落库
748async fn sync_snapshot(state: &AppState, conn_id: &str, arg: Value) -> Result<(), String> {
749    let req: OpenPageArgs = serde_json::from_value(arg).map_err(|e| e.to_string())?;
750    let ident = state.collaboration.identity(conn_id).ok_or("连接不存在")?;
751    let project = req.project_key.trim().to_string();
752    if project.is_empty() || req.page_key.is_empty() {
753        return Err("缺少 project_key 或 page_key".to_string());
754    }
755    let snapshot = PageSnapshot {
756        project_key: project.clone(),
757        page_key: req.page_key.clone(),
758        page_index: req.page_index,
759        page_name: req.page_name.clone(),
760        mode: normalize_mode(&req.mode),
761        updated_by_user_id: ident.user_id.clone(),
762        updated_by_display_name: ident.display_name.clone(),
763        updated_at: now_ms(),
764        units: req.units.clone(),
765    };
766    if let Ok(json) = serde_json::to_string(&snapshot) {
767        state
768            .redis
769            .set_str_ttl(
770                &snapshot_key(&project, &req.page_key),
771                &json,
772                SNAPSHOT_TTL_SECS,
773            )
774            .await;
775    }
776    // 活跃续约
777    if ident.lock_page_key.as_deref() == Some(req.page_key.as_str()) {
778        state
779            .redis
780            .expire(&lock_key(&project, &req.page_key), LOCK_TTL_SECS)
781            .await;
782    }
783    broadcast(
784        state,
785        &project,
786        "PageSnapshotUpdated",
787        json!(snapshot),
788        Some(conn_id),
789    )
790    .await;
791    schedule_persist(state, &project, &req.page_key, req.units, &ident.user_id);
792    Ok(())
793}
794
795/// 同步页面实时增量：仅广播，不落库
796async fn sync_live_delta(state: &AppState, conn_id: &str, arg: Value) -> Result<(), String> {
797    let project = arg
798        .get("project_key")
799        .and_then(Value::as_str)
800        .unwrap_or("")
801        .to_string();
802    if project.is_empty() {
803        return Ok(());
804    }
805    broadcast(state, &project, "PageLiveDeltaUpdated", arg, Some(conn_id)).await;
806    Ok(())
807}
808
809/// 安排快照落库（debounce：同页连续快照只在静默后写一次）
810fn schedule_persist(
811    state: &AppState,
812    project: &str,
813    page_key: &str,
814    units: Vec<Value>,
815    user_id: &str,
816) {
817    let gen = state
818        .collaboration
819        .persist_gen
820        .entry(page_key.to_string())
821        .or_insert_with(|| Arc::new(AtomicU64::new(0)))
822        .clone();
823    let target = gen.fetch_add(1, Ordering::SeqCst) + 1;
824
825    let state = state.clone();
826    let project = project.to_string();
827    let page_key = page_key.to_string();
828    let user_id = user_id.to_string();
829    tokio::spawn(async move {
830        tokio::time::sleep(Duration::from_millis(PERSIST_DEBOUNCE_MS)).await;
831        // 静默期内有更新的快照覆盖了世代号则跳过本次
832        if gen.load(Ordering::SeqCst) != target {
833            return;
834        }
835        persist_snapshot(&state, &project, &page_key, &units, &user_id).await;
836    });
837}
838
839/// 把页面快照落库到 `editor_unit`（整页替换）
840async fn persist_snapshot(
841    state: &AppState,
842    project: &str,
843    page_key: &str,
844    units: &[Value],
845    user_id: &str,
846) {
847    let Some(page_id) = parse_page_id(page_key) else {
848        return;
849    };
850    let member_id = user_id.parse::<i32>().unwrap_or(0);
851    if member_id <= 0 {
852        return;
853    }
854    // 优先用 project_key 解析分片键，失败回退查页面
855    let episode_id = match parse_episode_id(project) {
856        Some(e) => e,
857        None => match EditorPageRepository::new(state.db.clone())
858            .episode_of_page(page_id)
859            .await
860        {
861            Ok(Some(e)) => e,
862            _ => return,
863        },
864    };
865    let inputs: Vec<EditorUnitInput> = units
866        .iter()
867        .filter_map(|v| serde_json::from_value::<EditorUnitInput>(v.clone()).ok())
868        .collect();
869    if let Err(e) = EditorUnitRepository::new(state.db.clone())
870        .replace_page_units(episode_id, page_id, member_id, &inputs)
871        .await
872    {
873        tracing::warn!("协作快照落库失败 page={page_id}: {e:?}");
874    }
875}
tdm_server_rust/collaboration/hub.rs

tdm_server_rust/collaboration/
hub.rs
