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) render_scope: crate::render_scope::RenderScope,
110    pub(crate) scope_hydration: crate::scope_hydration::ScopeHydration,
111    pub(crate) sync_timing: crate::sync_observation::Timing,
112    pub(crate) diagnostics: crate::diagnostics::Session,
113    pub(crate) state: ImState,
114    /// 平台注入的本地存储能力；coverage 仍由 query 状态机逐次证明。
115    pub(crate) local_store_mode: crate::query::LocalStoreMode,
116    /// Correlation 分配器（自增）
117    ///
118    /// BLK-1b：模块从 1 开始，避免与 engine 的 Correlation 冲突。
119    /// 多模块场景由各模块从不同起点分配；更健壮的方案是 core 统一 IdSource。
120    next_corr: u64,
121    /// TimerId 分配器（自增）
122    ///
123    /// BLK-1b：从 100 开始，避免与 engine 内部潜在分配冲突。
124    next_timer: u64,
125    /// 发送临时 ID 的 module 内单调序号；与 step 的 now_ms 共同形成确定性 ID。
126    next_temporary_id: u64,
127}
128
129impl ImModule {
130    /// 排队一个原子业务写，并把 MessageV3 terminal events 绑定到 matching PersistOk。
131    #[doc(hidden)]
132    pub fn queue_message_v3_commit(
133        &mut self,
134        ops: Vec<helix_core::effect::StorageOp>,
135        terminal_events: Vec<crate::event::MessageV3Event>,
136        out: &mut EffectSink,
137    ) -> helix_core::Correlation {
138        let corr = self.alloc_corr_internal();
139        self.render_scope.stage_ops(corr, &ops);
140        let terminal_events = terminal_events
141            .into_iter()
142            .map(crate::event::MessageV3Event::into_bytes)
143            .collect();
144        self.state.corr_map.insert(
145            corr,
146            CorrelationContext::MessageV3Commit { terminal_events },
147        );
148        out.push(helix_core::Effect::PersistAtomic { corr, ops });
149        corr
150    }
151
152    /// PC/Flutter 默认使用 durable host store。
153    pub fn new(config: ImConfig) -> Self {
154        Self::new_with_local_store(config, crate::query::LocalStoreMode::Durable)
155    }
156
157    /// Web 可注入 Session/Disabled；三端业务 query API 保持一致。
158    pub fn new_with_local_store(
159        config: ImConfig,
160        local_store_mode: crate::query::LocalStoreMode,
161    ) -> Self {
162        Self {
163            config,
164            render_scope: Default::default(),
165            scope_hydration: Default::default(),
166            diagnostics: crate::diagnostics::Session::default(),
167            sync_timing: Default::default(),
168            state: ImState::new(),
169            local_store_mode,
170            next_corr: 1,
171            next_timer: 100,
172            next_temporary_id: 1,
173        }
174    }
175
176    pub fn local_store_mode(&self) -> crate::query::LocalStoreMode {
177        self.local_store_mode
178    }
179
180    pub fn alloc_corr(&mut self) -> helix_core::Correlation {
181        let c = helix_core::Correlation::from_raw(self.next_corr);
182        self.next_corr += 1;
183        c
184    }
185
186    pub fn alloc_timer(&mut self) -> TimerId {
187        let t = TimerId::from_raw(self.next_timer);
188        self.next_timer += 1;
189        t
190    }
191
192    pub(crate) fn alloc_temporary_id(
193        &mut self,
194        now_ms: u64,
195    ) -> Result<TemporaryId, crate::error::ImError> {
196        let sequence = self.next_temporary_id;
197        self.next_temporary_id = sequence.checked_add(1).ok_or_else(|| {
198            crate::error::ImError::Parse("temporary_id sequence exhausted".to_string())
199        })?;
200        Ok(TemporaryId::mint(now_ms, sequence))
201    }
202
203    /// 测试 / driver 初始化用：预注册一个 channel（初始 cursor 值由调用方提供）
204    pub fn register_channel(&mut self, id: ChannelId, initial_cursor: u64) {
205        self.state
206            .channels
207            .insert(id, Channel::new(id, initial_cursor));
208    }
209
210    /// 查询 per-channel cursor（测试 / 诊断用）
211    pub fn cursor_for(&self, channel_id: ChannelId) -> Option<u64> {
212        self.state
213            .channels
214            .get(&channel_id)
215            .map(|ch| ch.cursor.value().0)
216    }
217
218    /// 查询已确认的 type7 terminal tombstone 水位（测试 / 诊断用）。
219    pub fn terminal_event_seq_for(&self, channel_id: ChannelId) -> Option<u64> {
220        self.state
221            .channels
222            .get(&channel_id)
223            .and_then(|channel| channel.terminal_event_seq())
224            .map(|seq| seq.0)
225    }
226
227    /// 查询 pending_sends 数量（测试用）
228    pub fn pending_send_count(&self) -> usize {
229        self.state.pending_sends.len()
230    }
231
232    /// 查询文字接龙 mutation 的本地状态，供 bridge 映射稳定状态而非重算业务事实。
233    pub fn chain_mutation_state(
234        &self,
235        client_mutation_id: &str,
236    ) -> Option<crate::chain::ChainMutationState> {
237        self.state
238            .chain_mutations
239            .get(client_mutation_id)
240            .map(|mutation| mutation.state)
241    }
242
243    /// 查询由 Helix 生成的稳定 operationId，供 reconcile 继续使用同一业务操作。
244    pub fn chain_operation_id(&self, client_mutation_id: &str) -> Option<&str> {
245        self.state
246            .chain_mutations
247            .get(client_mutation_id)
248            .map(|mutation| mutation.operation_id.as_str())
249    }
250
251    /// B4：全局在途 sync 数 + 待 dispatch 队列长（HX-C011 证伪性质测试结构化断言用）。
252    pub fn sync_inflight(&self) -> usize {
253        self.state.sync_scheduler.inflight()
254    }
255    pub fn sync_pending_len(&self) -> usize {
256        self.state.sync_scheduler.pending_len()
257    }
258
259    /// 查询某 temporary_id 的 PendingSend 状态（测试 / 诊断用）。
260    /// 用于 C1 echo→reconcile 断言「对账后 PendingSend 推进到 Sent」。
261    pub fn pending_send_status(&self, temporary_id: &str) -> Option<SendStatus> {
262        self.state
263            .pending_sends
264            .get(&TemporaryId(temporary_id.to_string()))
265            .map(|ps| ps.status)
266    }
267
268    /// 查询 increment_fetched 是否含某 channel（测试 / 诊断用）。
269    pub fn increment_fetched_contains(&self, channel_id: ChannelId) -> bool {
270        self.state.increment_fetched.contains(&channel_id)
271    }
272
273    /// 查询某 channel 的 increment_target（服务端水位，测试 / 诊断用）。
274    /// cursor（本地确认）与 target（服务端水位）严格解耦——HX-C008 回归断言锚点。
275    pub fn increment_target_for(&self, channel_id: ChannelId) -> Option<u64> {
276        self.state
277            .increment_target
278            .get(&channel_id)
279            .map(|seq| seq.0)
280    }
281
282    /// B-rest 测试 / driver 入口：直接喂 `increment_channel_end` 信号。
283    /// `Some(ch)`=子 topic 结束；`None`=global end（触发增量群 sync）。
284    pub fn ingest_increment_end(&mut self, ch: Option<ChannelId>, out: &mut EffectSink) {
285        let start = out.as_slice().len();
286        let api_base_url = self.config.api_base_url.as_str();
287        let auth_user_id = self.config.auth_user_id.as_str();
288        let next_corr = &mut self.next_corr;
289        let mut alloc_corr = || {
290            let corr = helix_core::Correlation::from_raw(*next_corr);
291            *next_corr += 1;
292            corr
293        };
294        let mut ctx = crate::ws::ImWsContext::new(
295            &mut self.state,
296            0,
297            api_base_url,
298            auth_user_id,
299            &mut alloc_corr,
300        );
301        crate::ws::handlers::increment_channel_end::apply_increment_end(&mut ctx, ch, out);
302        self.render_scope.guard_effects(&self.config, start, out);
303    }
304
305    /// 内部 alloc_corr（不借用 self.state，仅借用分配器）
306    pub(crate) fn alloc_corr_internal(&mut self) -> helix_core::Correlation {
307        let c = helix_core::Correlation::from_raw(self.next_corr);
308        self.next_corr += 1;
309        c
310    }
311
312    /// 拆借 state 与 correlation 分配器，供 sibling module 复用私有 `next_corr` 的顺序分配。
313    pub(crate) fn with_state_and_corr_allocator<R>(
314        &mut self,
315        f: impl FnOnce(&mut ImState, &mut dyn FnMut() -> helix_core::Correlation) -> R,
316    ) -> R {
317        let next_corr = &mut self.next_corr;
318        let mut alloc_corr = || {
319            let corr = helix_core::Correlation::from_raw(*next_corr);
320            *next_corr += 1;
321            corr
322        };
323        f(&mut self.state, &mut alloc_corr)
324    }
325
326    /// `Tick::Inbound` 解析后的统一 dispatch：`WsFrame::parse` → WS registry by `action`（hello /
327    /// increment_channel(_end) / post / 19 事件全集 / dead-action no-op / 无 action 的 legacy event）。
328    /// 未知 action → `UnsupportedWsAction`（可观测）。各 handler 在 `ImWsContext` 上操作 state，副作用
329    /// 经 `out.push(Effect::…)`，保持 handle 同步纯函数。
330    fn dispatch_ws_frame(
331        &mut self,
332        frame: &crate::ws::WsFrame,
333        now_ms: u64,
334        out: &mut EffectSink,
335    ) -> Result<(), CoreError> {
336        self.diagnose_inbound(frame);
337        let module_name = self.name();
338        let api_base_url = self.config.api_base_url.as_str();
339        let auth_user_id = self.config.auth_user_id.as_str();
340        // Only hello needs the local watermark scope.  Do not rescan the channel
341        // ownership map for every ordinary WS frame.
342        let hello_scope = (frame.action().ok() == Some("hello"))
343            .then(|| self.render_scope.scoped_channel_ids(&self.config));
344        let next_corr = &mut self.next_corr;
345        let mut alloc_corr = || {
346            let corr = helix_core::Correlation::from_raw(*next_corr);
347            *next_corr += 1;
348            corr
349        };
350        let (result, timeline_refreshes) = {
351            let mut ctx = crate::ws::ImWsContext::new(
352                &mut self.state,
353                now_ms,
354                api_base_url,
355                auth_user_id,
356                &mut alloc_corr,
357            );
358            if let Some(channel_ids) = hello_scope.as_deref() {
359                ctx.set_scoped_channel_ids(channel_ids);
360            }
361            let result = crate::ws::dispatch_ws(&mut ctx, frame, out);
362            let timeline_refreshes = ctx.take_attached_timeline_refreshes();
363            (result, timeline_refreshes)
364        };
365        result.map_err(|e| CoreError::ModuleError {
366            module: module_name,
367            source: Box::new(e),
368        })?;
369
370        // The post handler has already appended its PersistFire write. The host
371        // driver serializes Persist/PersistFire by the same `message` table key,
372        // so this local-first query observes that accepted durable write before
373        // producing the next MessageV3 event.
374        for (channel_id, causation_id) in timeline_refreshes {
375            self.refresh_attached_latest_timeline(channel_id, causation_id, out)
376                .map_err(|e| CoreError::ModuleError {
377                    module: module_name,
378                    source: Box::new(e),
379                })?;
380        }
381        Ok(())
382    }
383}
helix_im/module.rs

helix_im/
module.rs
