helix_im/
module.rs

1//! ImModule：实现 helix-core Module trait 的 IM 业务模块
2//!
3//! ## 职责
4//!
5//! - 实现 `Module` trait（4 方法：name / accepts / handle / on_start+on_stop）
6//! - 持有 ImState（纯数据，无 I/O）
7//! - 通过 ACL-1 翻译 IM 概念 ↔ core Effect/Tick
8//!
9//! ## 不变量（关键）
10//!
11//! - `handle` 严格同步，零 await，零 I/O，零 spawn
12//! - 所有副作用通过 EffectSink 输出，由 driver 异步兑现
13
14use crate::acl;
15use crate::channel::Channel;
16use crate::state::{ChannelId, ConnState, CorrelationContext, ImState, SendStatus, TemporaryId};
17use helix_core::effect::TimerId;
18use helix_core::{CoreError, EffectSink, Module, Tick};
19
20mod chain;
21mod trait_impl;
22
23/// Driver 向 sans-I/O IM 模块同步当前 host 身份的内部命令。
24///
25/// 该命令不属于 public client API；driver 必须直接构造 `AppCommand`，避免移动端绕过
26/// header/profile 边界向业务 payload 注入身份。
27pub const RUNTIME_IDENTITY_COMMAND: &str = "__helix_runtime_identity";
28
29/// 产生本地权威投影 ACK 的宿主平台，固定枚举避免指标标签基数失控。
30#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
31pub enum ClientPlatform {
32    #[default]
33    Native,
34    Ffi,
35    Web,
36}
37
38impl ClientPlatform {
39    /// 返回 Go ACK 合同和指标共用的稳定低基数字符串。
40    pub const fn as_str(self) -> &'static str {
41        match self {
42            Self::Native => "native",
43            Self::Ffi => "ffi",
44            Self::Web => "web",
45        }
46    }
47}
48
49/// IM 业务模块配置
50pub struct ImConfig {
51    /// WS endpoint（如 "wss://api.example.com/ws"）
52    pub ws_url: String,
53    /// Go IM HTTP API base URL（通常含 `/api/cses`）。
54    pub api_base_url: String,
55    /// Java base（host/前端显式注入 `env.restHost`，无 `/api/cses`）。
56    ///
57    /// 媒体 prepare/verify 与 `Gateway::Default` 命令使用；不得从 Go `api_base_url` 推导。
58    pub default_api_base_url: String,
59    /// 当前登录用户 Id26；host 注入，供未读 sender 豁免、可见性门控与 channel 归属列使用。
60    pub auth_user_id: String,
61    /// 当前用户所属公司/团队 Id；host 注入，供 `posts/create` 的 teamId/userSnapshot 使用。
62    pub company_id: String,
63    pub user_name: String,
64    pub org_name: String,
65    pub dept_name: String,
66    /// 当前宿主平台；只用于三端 ACK 归属，不接受业务 payload 覆盖。
67    pub client_platform: ClientPlatform,
68    pub ping_interval_ms: u64,
69    pub send_timeout_ms: u64,
70    /// 由 host 注入的纯离线 sync 诊断配置，默认关闭。
71    pub offline_sync_diagnostics: crate::sync::SyncDiagnosticsConfig,
72}
73
74impl ImConfig {
75    /// 当前登录用户身份视图（userSnapshot 注入源，posts/create body §1）。
76    pub fn user_identity(&self) -> crate::outbound::send_build::UserIdentity<'_> {
77        crate::outbound::send_build::UserIdentity {
78            user_id: &self.auth_user_id,
79            team_id: &self.company_id,
80            user_name: &self.user_name,
81            org_name: &self.org_name,
82            dept_name: &self.dept_name,
83        }
84    }
85}
86
87impl Default for ImConfig {
88    fn default() -> Self {
89        Self {
90            ws_url: String::new(),
91            api_base_url: String::new(),
92            default_api_base_url: String::new(),
93            auth_user_id: String::new(),
94            company_id: String::new(),
95            user_name: String::new(),
96            org_name: String::new(),
97            dept_name: String::new(),
98            client_platform: ClientPlatform::Native,
99            ping_interval_ms: 8_000,
100            send_timeout_ms: 15_000,
101            offline_sync_diagnostics: crate::sync::SyncDiagnosticsConfig::disabled(),
102        }
103    }
104}
105
106/// IM 业务模块（实现 helix_core::Module trait）
107pub struct ImModule {
108    pub(crate) config: ImConfig,
109    pub(crate) sync_timing: crate::sync_observation::Timing,
110    pub(crate) diagnostics: crate::diagnostics::Session,
111    pub(crate) state: ImState,
112    /// 平台注入的本地存储能力；coverage 仍由 query 状态机逐次证明。
113    pub(crate) local_store_mode: crate::query::LocalStoreMode,
114    /// Correlation 分配器（自增）
115    ///
116    /// BLK-1b：模块从 1 开始，避免与 engine 的 Correlation 冲突。
117    /// 多模块场景由各模块从不同起点分配；更健壮的方案是 core 统一 IdSource。
118    next_corr: u64,
119    /// TimerId 分配器（自增）
120    ///
121    /// BLK-1b：从 100 开始，避免与 engine 内部潜在分配冲突。
122    next_timer: u64,
123    /// 发送临时 ID 的 module 内单调序号；与 step 的 now_ms 共同形成确定性 ID。
124    next_temporary_id: u64,
125}
126
127impl ImModule {
128    /// 排队一个原子业务写，并把 MessageV3 terminal events 绑定到 matching PersistOk。
129    #[doc(hidden)]
130    pub fn queue_message_v3_commit(
131        &mut self,
132        ops: Vec<helix_core::effect::StorageOp>,
133        terminal_events: Vec<crate::event::MessageV3Event>,
134        out: &mut EffectSink,
135    ) -> helix_core::Correlation {
136        let corr = self.alloc_corr_internal();
137        let terminal_events = terminal_events
138            .into_iter()
139            .map(crate::event::MessageV3Event::into_bytes)
140            .collect();
141        self.state.corr_map.insert(
142            corr,
143            CorrelationContext::MessageV3Commit { terminal_events },
144        );
145        out.push(helix_core::Effect::PersistAtomic { corr, ops });
146        corr
147    }
148
149    /// PC/Flutter 默认使用 durable host store。
150    pub fn new(config: ImConfig) -> Self {
151        Self::new_with_local_store(config, crate::query::LocalStoreMode::Durable)
152    }
153
154    /// Web 可注入 Session/Disabled；三端业务 query API 保持一致。
155    pub fn new_with_local_store(
156        config: ImConfig,
157        local_store_mode: crate::query::LocalStoreMode,
158    ) -> Self {
159        Self {
160            config,
161            diagnostics: crate::diagnostics::Session::default(),
162            sync_timing: Default::default(),
163            state: ImState::new(),
164            local_store_mode,
165            next_corr: 1,
166            next_timer: 100,
167            next_temporary_id: 1,
168        }
169    }
170
171    pub fn local_store_mode(&self) -> crate::query::LocalStoreMode {
172        self.local_store_mode
173    }
174
175    pub fn alloc_corr(&mut self) -> helix_core::Correlation {
176        let c = helix_core::Correlation::from_raw(self.next_corr);
177        self.next_corr += 1;
178        c
179    }
180
181    pub fn alloc_timer(&mut self) -> TimerId {
182        let t = TimerId::from_raw(self.next_timer);
183        self.next_timer += 1;
184        t
185    }
186
187    pub(crate) fn alloc_temporary_id(
188        &mut self,
189        now_ms: u64,
190    ) -> Result<TemporaryId, crate::error::ImError> {
191        let sequence = self.next_temporary_id;
192        self.next_temporary_id = sequence.checked_add(1).ok_or_else(|| {
193            crate::error::ImError::Parse("temporary_id sequence exhausted".to_string())
194        })?;
195        Ok(TemporaryId::mint(now_ms, sequence))
196    }
197
198    /// 测试 / driver 初始化用：预注册一个 channel（初始 cursor 值由调用方提供）
199    pub fn register_channel(&mut self, id: ChannelId, initial_cursor: u64) {
200        self.state
201            .channels
202            .insert(id, Channel::new(id, initial_cursor));
203    }
204
205    /// 查询 per-channel cursor（测试 / 诊断用）
206    pub fn cursor_for(&self, channel_id: ChannelId) -> Option<u64> {
207        self.state
208            .channels
209            .get(&channel_id)
210            .map(|ch| ch.cursor.value().0)
211    }
212
213    /// 查询已确认的 type7 terminal tombstone 水位（测试 / 诊断用）。
214    pub fn terminal_event_seq_for(&self, channel_id: ChannelId) -> Option<u64> {
215        self.state
216            .channels
217            .get(&channel_id)
218            .and_then(|channel| channel.terminal_event_seq())
219            .map(|seq| seq.0)
220    }
221
222    /// 查询 pending_sends 数量（测试用）
223    pub fn pending_send_count(&self) -> usize {
224        self.state.pending_sends.len()
225    }
226
227    /// 查询文字接龙 mutation 的本地状态，供 bridge 映射稳定状态而非重算业务事实。
228    pub fn chain_mutation_state(
229        &self,
230        client_mutation_id: &str,
231    ) -> Option<crate::chain::ChainMutationState> {
232        self.state
233            .chain_mutations
234            .get(client_mutation_id)
235            .map(|mutation| mutation.state)
236    }
237
238    /// 查询由 Helix 生成的稳定 operationId，供 reconcile 继续使用同一业务操作。
239    pub fn chain_operation_id(&self, client_mutation_id: &str) -> Option<&str> {
240        self.state
241            .chain_mutations
242            .get(client_mutation_id)
243            .map(|mutation| mutation.operation_id.as_str())
244    }
245
246    /// B4：全局在途 sync 数 + 待 dispatch 队列长（HX-C011 证伪性质测试结构化断言用）。
247    pub fn sync_inflight(&self) -> usize {
248        self.state.sync_scheduler.inflight()
249    }
250    pub fn sync_pending_len(&self) -> usize {
251        self.state.sync_scheduler.pending_len()
252    }
253
254    /// 查询某 temporary_id 的 PendingSend 状态（测试 / 诊断用）。
255    /// 用于 C1 echo→reconcile 断言「对账后 PendingSend 推进到 Sent」。
256    pub fn pending_send_status(&self, temporary_id: &str) -> Option<SendStatus> {
257        self.state
258            .pending_sends
259            .get(&TemporaryId(temporary_id.to_string()))
260            .map(|ps| ps.status)
261    }
262
263    /// 查询 increment_fetched 是否含某 channel（测试 / 诊断用）。
264    pub fn increment_fetched_contains(&self, channel_id: ChannelId) -> bool {
265        self.state.increment_fetched.contains(&channel_id)
266    }
267
268    /// 查询某 channel 的 increment_target（服务端水位，测试 / 诊断用）。
269    /// cursor（本地确认）与 target（服务端水位）严格解耦——HX-C008 回归断言锚点。
270    pub fn increment_target_for(&self, channel_id: ChannelId) -> Option<u64> {
271        self.state
272            .increment_target
273            .get(&channel_id)
274            .map(|seq| seq.0)
275    }
276
277    /// B-rest 测试 / driver 入口：直接喂 `increment_channel_end` 信号。
278    /// `Some(ch)`=子 topic 结束；`None`=global end（触发增量群 sync）。
279    pub fn ingest_increment_end(&mut self, ch: Option<ChannelId>, out: &mut EffectSink) {
280        let api_base_url = self.config.api_base_url.as_str();
281        let auth_user_id = self.config.auth_user_id.as_str();
282        let next_corr = &mut self.next_corr;
283        let mut alloc_corr = || {
284            let corr = helix_core::Correlation::from_raw(*next_corr);
285            *next_corr += 1;
286            corr
287        };
288        let mut ctx = crate::ws::ImWsContext::new(
289            &mut self.state,
290            0,
291            api_base_url,
292            auth_user_id,
293            &mut alloc_corr,
294        );
295        crate::ws::handlers::increment_channel_end::apply_increment_end(&mut ctx, ch, out);
296    }
297
298    /// 内部 alloc_corr（不借用 self.state，仅借用分配器）
299    pub(crate) fn alloc_corr_internal(&mut self) -> helix_core::Correlation {
300        let c = helix_core::Correlation::from_raw(self.next_corr);
301        self.next_corr += 1;
302        c
303    }
304
305    /// 拆借 state 与 correlation 分配器，供 sibling module 复用私有 `next_corr` 的顺序分配。
306    pub(crate) fn with_state_and_corr_allocator<R>(
307        &mut self,
308        f: impl FnOnce(&mut ImState, &mut dyn FnMut() -> helix_core::Correlation) -> R,
309    ) -> R {
310        let next_corr = &mut self.next_corr;
311        let mut alloc_corr = || {
312            let corr = helix_core::Correlation::from_raw(*next_corr);
313            *next_corr += 1;
314            corr
315        };
316        f(&mut self.state, &mut alloc_corr)
317    }
318
319    /// `Tick::Inbound` 解析后的统一 dispatch：`WsFrame::parse` → WS registry by `action`（hello /
320    /// increment_channel(_end) / post / 19 事件全集 / dead-action no-op / 无 action 的 legacy event）。
321    /// 未知 action → `UnsupportedWsAction`（可观测）。各 handler 在 `ImWsContext` 上操作 state，副作用
322    /// 经 `out.push(Effect::…)`，保持 handle 同步纯函数。
323    fn dispatch_ws_frame(
324        &mut self,
325        frame: &crate::ws::WsFrame,
326        now_ms: u64,
327        out: &mut EffectSink,
328    ) -> Result<(), CoreError> {
329        self.diagnose_inbound(frame);
330        let module_name = self.name();
331        let api_base_url = self.config.api_base_url.as_str();
332        let auth_user_id = self.config.auth_user_id.as_str();
333        let next_corr = &mut self.next_corr;
334        let mut alloc_corr = || {
335            let corr = helix_core::Correlation::from_raw(*next_corr);
336            *next_corr += 1;
337            corr
338        };
339        let (result, timeline_refreshes) = {
340            let mut ctx = crate::ws::ImWsContext::new(
341                &mut self.state,
342                now_ms,
343                api_base_url,
344                auth_user_id,
345                &mut alloc_corr,
346            );
347            let result = crate::ws::dispatch_ws(&mut ctx, frame, out);
348            let timeline_refreshes = ctx.take_attached_timeline_refreshes();
349            (result, timeline_refreshes)
350        };
351        result.map_err(|e| CoreError::ModuleError {
352            module: module_name,
353            source: Box::new(e),
354        })?;
355
356        // The post handler has already appended its PersistFire write. The host
357        // driver serializes Persist/PersistFire by the same `message` table key,
358        // so this local-first query observes that accepted durable write before
359        // producing the next MessageV3 event.
360        for (channel_id, causation_id) in timeline_refreshes {
361            self.refresh_attached_latest_timeline(channel_id, causation_id, out)
362                .map_err(|e| CoreError::ModuleError {
363                    module: module_name,
364                    source: Box::new(e),
365                })?;
366        }
367        Ok(())
368    }
369}
helix_im/module.rs

helix_im/
module.rs
