helix_im/
increment_hydration.rs

1//! UC-4.5 陌生 channel 单频道 hydration 编排。
2//!
3//! HTTP `incrementByChannelId` 仍保留通用 read relay，但其 `data=*IncrementChannel` 还会复用
4//! WS increment 解析/应用规则。channel/member 确认落库后才启动既有 sync；terminal sync commit
5//! 后从本地 channel/message 表读回并复用 render-ready query 投影，前端不参与业务计算。
6
7use crate::error::ImError;
8use crate::http_envelope::unwrap_sync_envelope;
9use crate::module::ImModule;
10use crate::state::{ChannelId, CorrelationContext, HydrationPersistSnapshot};
11use helix_core::effect::{Effect, GetSpec, ScanOrder, ScanSpec, SqlValue, StorageOp};
12use helix_core::tick::PortOutcome;
13use helix_core::{Correlation, EffectSink};
14
15const HYDRATION_MESSAGE_ORDER: &[ScanOrder] = &[
16    ScanOrder::desc("create_at"),
17    ScanOrder::desc("temporary_id"),
18];
19
20/// Classify a missing-data reply using fixed values; never log remote payloads or messages.
21fn hydration_reply_diagnostic(body: &serde_json::Value) -> (&'static str, &'static str) {
22    match body.get("status").and_then(serde_json::Value::as_str) {
23        Some("failed") => (
24            "failed",
25            if body.get("message").and_then(serde_json::Value::as_str)
26                == Some("user not member of channel")
27            {
28                "remote_not_member"
29            } else {
30                "remote_business_failure"
31            },
32        ),
33        Some("SUCCESS") => ("SUCCESS", "missing_data"),
34        Some("FINISH") => ("FINISH", "missing_data"),
35        Some(_) => ("unknown", "invalid_business_status"),
36        None => ("missing_or_invalid", "invalid_business_status"),
37    }
38}
39
40#[cfg(test)]
41mod diagnostic_tests {
42    use super::hydration_reply_diagnostic;
43    use serde_json::json;
44
45    /// Preserve the actionable failure category without exporting arbitrary server text.
46    #[test]
47    fn hydration_diagnostic_classifies_business_failure_without_remote_text() {
48        for (body, expected) in [
49            (
50                json!({"status":"failed","message":"user not member of channel"}),
51                ("failed", "remote_not_member"),
52            ),
53            (
54                json!({"status":"failed","message":"secret=private\nraw payload"}),
55                ("failed", "remote_business_failure"),
56            ),
57            (json!({"status":"SUCCESS"}), ("SUCCESS", "missing_data")),
58            (json!({"status":"FINISH"}), ("FINISH", "missing_data")),
59            (
60                json!({"status":"private-value"}),
61                ("unknown", "invalid_business_status"),
62            ),
63            (
64                json!({"status":200}),
65                ("missing_or_invalid", "invalid_business_status"),
66            ),
67            (json!({}), ("missing_or_invalid", "invalid_business_status")),
68        ] {
69            assert_eq!(hydration_reply_diagnostic(&body), expected);
70        }
71    }
72}
73
74impl ImModule {
75    /// B-rest 测试 / driver 入口：直接喂一个解析好的 `increment_channel` 帧。
76    pub fn ingest_increment(
77        &mut self,
78        inc: &crate::sync_session::IncrementChannel,
79        out: &mut EffectSink,
80    ) {
81        let api_base_url = self.config.api_base_url.clone();
82        let auth_user_id = self.config.auth_user_id.clone();
83        self.with_state_and_corr_allocator(|state, alloc| {
84            let mut ctx =
85                crate::ws::ImWsContext::new(state, 0, &api_base_url, &auth_user_id, alloc);
86            crate::ws::handlers::increment_channel::apply_increment(&mut ctx, inc, out);
87        });
88    }
89
90    fn ingest_increment_hydration(
91        &mut self,
92        inc: &crate::sync_session::IncrementChannel,
93        persist_corr: Correlation,
94        now_ms: u64,
95        out: &mut EffectSink,
96    ) -> bool {
97        let api_base_url = self.config.api_base_url.clone();
98        let auth_user_id = self.config.auth_user_id.clone();
99        self.with_state_and_corr_allocator(|state, alloc| {
100            let mut ctx =
101                crate::ws::ImWsContext::new(state, now_ms, &api_base_url, &auth_user_id, alloc);
102            crate::ws::handlers::increment_channel::apply_increment_hydration(
103                &mut ctx,
104                inc,
105                persist_corr,
106                out,
107            )
108        })
109    }
110
111    /// 记录 increment 应用在内存中写入的临时壳，供持久化失败时恢复旧状态。
112    fn hydration_snapshot(&self, channel_id: ChannelId) -> HydrationPersistSnapshot {
113        HydrationPersistSnapshot {
114            had_channel: self.state.channels.contains_key(&channel_id),
115            previous_target: self.state.increment_target.get(&channel_id).copied(),
116            was_increment_fetched: self.state.increment_fetched.contains(&channel_id),
117            was_need_sync_skip: self.state.need_sync_skip.contains(&channel_id),
118            previous_about_me_len: self.state.about_me_post_ids.len(),
119        }
120    }
121
122    /// 回滚尚未通过 durable channel/member barrier 的本地临时事实。
123    fn restore_hydration_snapshot(
124        &mut self,
125        channel_id: ChannelId,
126        snapshot: &HydrationPersistSnapshot,
127    ) {
128        if !snapshot.had_channel {
129            self.state.channels.remove(&channel_id);
130        }
131        if let Some(target) = snapshot.previous_target {
132            self.state.increment_target.insert(channel_id, target);
133        } else {
134            self.state.increment_target.remove(&channel_id);
135        }
136        if !snapshot.was_increment_fetched {
137            self.state.increment_fetched.remove(&channel_id);
138            self.state.increment_order.retain(|id| *id != channel_id);
139        }
140        if snapshot.was_need_sync_skip {
141            self.state.need_sync_skip.insert(channel_id);
142        } else {
143            self.state.need_sync_skip.remove(&channel_id);
144        }
145        self.state
146            .about_me_post_ids
147            .truncate(snapshot.previous_about_me_len);
148    }
149
150    /// 从 driver 的 rows JSON 回包解出对象列表，畸形回包 fail closed。
151    fn hydration_rows(
152        reply: &helix_core::tick::ReplyBytes,
153    ) -> Result<Vec<serde_json::Value>, ImError> {
154        let value = serde_json::from_slice::<serde_json::Value>(reply.0.as_ref())
155            .map_err(|error| ImError::Parse(format!("hydration read-back rows: {error}")))?;
156        value
157            .as_array()
158            .cloned()
159            .ok_or_else(|| ImError::Parse("hydration read-back must be an array".to_string()))
160    }
161
162    /// 从 Go G09d raw snapshot 冻结有序全角色 roster，并校验 memberCount 与唯一 user id。
163    fn ordered_hydration_roster(raw_increment: &[u8]) -> Result<Vec<serde_json::Value>, ImError> {
164        let data: serde_json::Value = serde_json::from_slice(raw_increment)
165            .map_err(|error| ImError::Parse(format!("hydration roster parse: {error}")))?;
166        let members = data
167            .get("members")
168            .and_then(serde_json::Value::as_array)
169            .ok_or_else(|| ImError::Parse("hydration roster members missing".to_string()))?;
170        let member_count = data
171            .get("memberCount")
172            .or_else(|| data.get("member_count"))
173            .and_then(serde_json::Value::as_u64)
174            .ok_or_else(|| ImError::Parse("hydration roster memberCount missing".to_string()))?;
175        if member_count as usize != members.len() || members.is_empty() {
176            return Err(ImError::Parse(
177                "hydration roster count mismatch or empty".to_string(),
178            ));
179        }
180        let mut seen = std::collections::HashSet::with_capacity(members.len());
181        let mut ordered = Vec::with_capacity(members.len());
182        for member in members {
183            let user_id = member
184                .get("userId")
185                .or_else(|| member.get("id"))
186                .and_then(serde_json::Value::as_str)
187                .filter(|value| !value.is_empty())
188                .ok_or_else(|| ImError::Parse("hydration roster member id missing".to_string()))?;
189            if !seen.insert(user_id.to_string()) {
190                return Err(ImError::Parse(
191                    "hydration roster contains duplicate member".to_string(),
192                ));
193            }
194            ordered.push(serde_json::json!({
195                "userId": user_id,
196                "teamId": member.get("teamId").and_then(serde_json::Value::as_str).unwrap_or_default(),
197                "role": member.get("role").and_then(serde_json::Value::as_str).unwrap_or("MEMBER"),
198                "nickName": member.get("nickName").and_then(serde_json::Value::as_str).unwrap_or_default(),
199            }));
200        }
201        Ok(ordered)
202    }
203
204    /// 将 durable member rows 按冻结位序重建完整 channel roster，并拒绝租户/集合漂移。
205    fn attach_hydration_roster(
206        &self,
207        channel: &mut serde_json::Value,
208        durable_rows: &[serde_json::Value],
209        ordered: &[serde_json::Value],
210    ) -> Result<(), ImError> {
211        let mut durable = std::collections::HashMap::with_capacity(durable_rows.len());
212        for row in durable_rows {
213            let user_id = row
214                .get("user_id")
215                .and_then(serde_json::Value::as_str)
216                .filter(|value| !value.is_empty())
217                .ok_or_else(|| ImError::Parse("hydration durable member id missing".to_string()))?;
218            if row.get("team_id").and_then(serde_json::Value::as_str)
219                != Some(self.config.company_id.as_str())
220            {
221                return Err(ImError::Parse(
222                    "hydration durable member tenant mismatch".to_string(),
223                ));
224            }
225            if durable.insert(user_id, row).is_some() {
226                return Err(ImError::Parse(
227                    "hydration durable member duplicate".to_string(),
228                ));
229            }
230        }
231        if durable.len() != ordered.len() {
232            return Err(ImError::Parse(
233                "hydration durable roster count mismatch".to_string(),
234            ));
235        }
236        let object = channel
237            .as_object_mut()
238            .ok_or_else(|| ImError::Parse("hydration channel shape invalid".to_string()))?;
239        let mut admins = Vec::new();
240        let mut bosses = Vec::new();
241        let mut owner = serde_json::Value::Null;
242        let mut projected_members = Vec::with_capacity(ordered.len());
243        for member in ordered {
244            let user_id = member["userId"]
245                .as_str()
246                .ok_or_else(|| ImError::Parse("hydration ordered member invalid".to_string()))?;
247            let row = durable.get(user_id).ok_or_else(|| {
248                ImError::Parse("hydration durable roster set mismatch".to_string())
249            })?;
250            let role = row
251                .get("role")
252                .and_then(serde_json::Value::as_str)
253                .unwrap_or("MEMBER");
254            let projected = serde_json::json!({
255                "userId": user_id,
256                "teamId": self.config.company_id,
257                "role": role,
258                "nickName": row.get("nick_name").and_then(serde_json::Value::as_str).unwrap_or_default(),
259            });
260            projected_members.push(projected.clone());
261            match role {
262                "ADMIN" | "MANAGER" => admins.push(projected.clone()),
263                "BOSS" => bosses.push(projected.clone()),
264                "OWNER" | "CREATOR" => owner = projected.clone(),
265                _ => {}
266            }
267        }
268        object.insert(
269            "members".to_string(),
270            serde_json::Value::Array(projected_members),
271        );
272        object.insert("adminUsers".to_string(), serde_json::Value::Array(admins));
273        object.insert("boss".to_string(), serde_json::Value::Array(bosses));
274        object.insert("owner".to_string(), owner);
275        object.insert("memberCount".to_string(), serde_json::json!(ordered.len()));
276        Ok(())
277    }
278
279    /// 发送 channel row 读回，后续阶段均由 matching corr 串联。
280    fn start_hydration_channel_readback(
281        &mut self,
282        req_id: String,
283        channel_id: ChannelId,
284        out: &mut EffectSink,
285    ) {
286        let corr = self.alloc_corr_internal();
287        out.push(Effect::Persist {
288            corr,
289            ops: vec![StorageOp::Get(GetSpec {
290                table: "channel",
291                key_col: "id",
292                key_val: SqlValue::Text(channel_id.as_str().to_string()),
293            })],
294        });
295        self.state.corr_map.insert(
296            corr,
297            CorrelationContext::HydrationChannelReadback { req_id, channel_id },
298        );
299    }
300
301    /// channel row 读回成功后，扫描完整 durable roster 并在下一阶段验证 viewer。
302    pub(crate) fn handle_hydration_channel_readback(
303        &mut self,
304        req_id: String,
305        channel_id: ChannelId,
306        outcome: &PortOutcome,
307        out: &mut EffectSink,
308    ) -> Result<(), ImError> {
309        let rows = match outcome {
310            PortOutcome::Ok(reply) => Self::hydration_rows(reply),
311            PortOutcome::Err(error) => Err(ImError::Parse(format!(
312                "channel read-back failed: {error:?}"
313            ))),
314        };
315        let channel = match rows {
316            Ok(mut rows) => rows.pop(),
317            Err(error) => {
318                self.finish_hydration_error(&req_id, channel_id, &error.to_string(), out);
319                return Ok(());
320            }
321        };
322        let Some(channel) = channel.filter(|row| {
323            row.get("id").and_then(serde_json::Value::as_str) == Some(channel_id.as_str())
324                && row.get("team_id").and_then(serde_json::Value::as_str)
325                    == Some(self.config.company_id.as_str())
326                && row.get("user_id").and_then(serde_json::Value::as_str)
327                    == Some(self.config.auth_user_id.as_str())
328        }) else {
329            self.finish_hydration_error(
330                &req_id,
331                channel_id,
332                "channel read-back scope mismatch",
333                out,
334            );
335            return Ok(());
336        };
337        let channel = crate::query::render_ready::channel::shape_channel_row(&channel);
338        let corr = self.alloc_corr_internal();
339        out.push(Effect::Persist {
340            corr,
341            ops: vec![StorageOp::Scan(ScanSpec {
342                table: "channel_member",
343                limit: None,
344                filter: Some((
345                    "channel_id",
346                    SqlValue::Text(channel_id.as_str().to_string()),
347                )),
348                order_by: &[],
349            })],
350        });
351        self.state.corr_map.insert(
352            corr,
353            CorrelationContext::HydrationMemberReadback {
354                req_id,
355                channel_id,
356                channel: Box::new(channel),
357            },
358        );
359        Ok(())
360    }
361
362    /// member row 读回成功后，读取频道消息窗口。
363    pub(crate) fn handle_hydration_member_readback(
364        &mut self,
365        req_id: String,
366        channel_id: ChannelId,
367        channel: Box<serde_json::Value>,
368        outcome: &PortOutcome,
369        out: &mut EffectSink,
370    ) -> Result<(), ImError> {
371        let rows = match outcome {
372            PortOutcome::Ok(reply) => Self::hydration_rows(reply),
373            PortOutcome::Err(error) => Err(ImError::Parse(format!(
374                "member read-back failed: {error:?}"
375            ))),
376        };
377        let durable_rows = match rows {
378            Ok(rows) => rows,
379            Err(error) => {
380                self.finish_hydration_error(&req_id, channel_id, &error.to_string(), out);
381                return Ok(());
382            }
383        };
384        let Some(member) = durable_rows
385            .iter()
386            .find(|row| {
387                row.get("user_id").and_then(serde_json::Value::as_str)
388                    == Some(self.config.auth_user_id.as_str())
389                    && row.get("team_id").and_then(serde_json::Value::as_str)
390                        == Some(self.config.company_id.as_str())
391            })
392            .cloned()
393        else {
394            self.finish_hydration_error(&req_id, channel_id, "member read-back missing", out);
395            return Ok(());
396        };
397        let Some(ordered) = self
398            .state
399            .hydration_ordered_rosters
400            .get(&channel_id)
401            .cloned()
402        else {
403            self.finish_hydration_error(&req_id, channel_id, "ordered roster missing", out);
404            return Ok(());
405        };
406        let mut channel = *channel;
407        if let Err(error) = self.attach_hydration_roster(&mut channel, &durable_rows, &ordered) {
408            self.finish_hydration_error(&req_id, channel_id, &error.to_string(), out);
409            return Ok(());
410        }
411        let corr = self.alloc_corr_internal();
412        out.push(Effect::Persist {
413            corr,
414            ops: vec![StorageOp::Scan(ScanSpec {
415                table: "message",
416                limit: Some(50),
417                filter: Some((
418                    "channel_id",
419                    SqlValue::Text(channel_id.as_str().to_string()),
420                )),
421                order_by: HYDRATION_MESSAGE_ORDER,
422            })],
423        });
424        self.state.corr_map.insert(
425            corr,
426            CorrelationContext::HydrationMessagesReadback {
427                req_id,
428                channel_id,
429                channel: Box::new(channel),
430                member: Box::new(member),
431            },
432        );
433        Ok(())
434    }
435
436    /// 消息窗口读回成功后，读取 durable cursor。
437    pub(crate) fn handle_hydration_messages_readback(
438        &mut self,
439        req_id: String,
440        channel_id: ChannelId,
441        channel: Box<serde_json::Value>,
442        member: Box<serde_json::Value>,
443        outcome: &PortOutcome,
444        out: &mut EffectSink,
445    ) -> Result<(), ImError> {
446        let messages = match outcome {
447            PortOutcome::Ok(reply) => match Self::hydration_rows(reply) {
448                Ok(rows) => serde_json::Value::Array(rows),
449                Err(error) => {
450                    self.finish_hydration_error(&req_id, channel_id, &error.to_string(), out);
451                    return Ok(());
452                }
453            },
454            PortOutcome::Err(error) => {
455                self.finish_hydration_error(
456                    &req_id,
457                    channel_id,
458                    &format!("message read-back failed: {error:?}"),
459                    out,
460                );
461                return Ok(());
462            }
463        };
464        let corr = self.alloc_corr_internal();
465        out.push(Effect::Persist {
466            corr,
467            ops: vec![StorageOp::Get(GetSpec {
468                table: "channel_event_cursor",
469                key_col: "channel_id",
470                key_val: SqlValue::Text(channel_id.as_str().to_string()),
471            })],
472        });
473        self.state.corr_map.insert(
474            corr,
475            CorrelationContext::HydrationCursorReadback {
476                req_id,
477                channel_id,
478                channel,
479                member,
480                messages: Box::new(messages),
481            },
482        );
483        Ok(())
484    }
485
486    /// cursor 读回成功后只释放一个 ChannelHydrationResult。
487    pub(crate) fn handle_hydration_cursor_readback(
488        &mut self,
489        req_id: String,
490        channel_id: ChannelId,
491        channel: Box<serde_json::Value>,
492        member: Box<serde_json::Value>,
493        messages: Box<serde_json::Value>,
494        outcome: &PortOutcome,
495        out: &mut EffectSink,
496    ) -> Result<(), ImError> {
497        let cursor = match outcome {
498            PortOutcome::Ok(reply) => match Self::hydration_rows(reply) {
499                Ok(mut rows) => rows
500                    .pop()
501                    .and_then(|row| {
502                        row.get("last_event_seq")
503                            .and_then(serde_json::Value::as_i64)
504                    })
505                    .unwrap_or_else(|| {
506                        self.state
507                            .channels
508                            .get(&channel_id)
509                            .map(|channel| channel.cursor.value().0 as i64)
510                            .unwrap_or(0)
511                    }),
512                Err(error) => {
513                    self.finish_hydration_error(&req_id, channel_id, &error.to_string(), out);
514                    return Ok(());
515                }
516            },
517            PortOutcome::Err(error) => {
518                self.finish_hydration_error(
519                    &req_id,
520                    channel_id,
521                    &format!("cursor read-back failed: {error:?}"),
522                    out,
523                );
524                return Ok(());
525            }
526        };
527        let emit_channel_increment = self
528            .state
529            .hydration_emit_channel_increment
530            .remove(&channel_id);
531        self.state.hydration_req_ids.remove(&channel_id);
532        self.state.hydration_ordered_rosters.remove(&channel_id);
533        let unread_reconcile = self
534            .state
535            .hydration_authority_unreads
536            .remove(&channel_id)
537            .and_then(|authority| {
538                member
539                    .get("unread_count")
540                    .or_else(|| member.get("unreadCount"))
541                    .and_then(serde_json::Value::as_i64)
542                    .map(|local| {
543                        serde_json::json!({
544                            "authority": authority,
545                            "local": local,
546                            "status": if authority == local { "match" } else { "mismatch" },
547                        })
548                    })
549            });
550        if emit_channel_increment {
551            out.push(
552                crate::event::MessageV3Event::new("im:channel:increment", (*channel).clone())?
553                    .into_effect(),
554            );
555        }
556        out.push(crate::read_relay::emit_read_body(
557            &req_id,
558            serde_json::json!({
559                "channelId": channel_id.as_str(),
560                "channel": *channel,
561                "member": *member,
562                "messages": *messages,
563                "cursor": cursor,
564                "completion": "hydrated",
565                "unreadReconcile": unread_reconcile,
566            }),
567        ));
568        Ok(())
569    }
570
571    /// 失败终态统一清理 hydration bookkeeping 并只回灌一次 error Result。
572    pub(crate) fn finish_hydration_error(
573        &mut self,
574        req_id: &str,
575        channel_id: ChannelId,
576        reason: &str,
577        out: &mut EffectSink,
578    ) {
579        self.state.hydration_pending.remove(&channel_id);
580        self.state.hydration_req_ids.remove(&channel_id);
581        self.state
582            .hydration_emit_channel_increment
583            .remove(&channel_id);
584        self.state.hydration_ordered_rosters.remove(&channel_id);
585        self.state.hydration_authority_unreads.remove(&channel_id);
586        out.push(crate::read_relay::emit_read_error(req_id, reason));
587    }
588
589    /// 以 channel 取回 hydration reqId 并结算一次失败，供 sync caller 处理解析/网络错误。
590    pub(crate) fn fail_hydration_for_channel(
591        &mut self,
592        channel_id: ChannelId,
593        reason: &str,
594        out: &mut EffectSink,
595    ) {
596        let Some(req_id) = self.state.hydration_req_ids.get(&channel_id).cloned() else {
597            return;
598        };
599        self.finish_hydration_error(&req_id, channel_id, reason, out);
600    }
601
602    /// Decode hydration replies; diagnostics do not change the existing Result contract.
603    pub(crate) fn handle_increment_hydration_reply(
604        &mut self,
605        req_id: &str,
606        emit_channel_increment: bool,
607        outcome: &PortOutcome,
608        now_ms: u64,
609        out: &mut EffectSink,
610    ) -> Result<(), ImError> {
611        match outcome {
612            PortOutcome::Ok(reply) => match unwrap_sync_envelope(reply.0.as_ref()) {
613                Ok(raw_body) => {
614                    let body: serde_json::Value = match serde_json::from_slice(&raw_body) {
615                        Ok(body) => body,
616                        Err(e) => {
617                            tracing::warn!(req_id, error = ?e, "increment hydration body is not json");
618                            out.push(crate::read_relay::emit_read_error(
619                                req_id,
620                                "increment hydration body is not json",
621                            ));
622                            return Ok(());
623                        }
624                    };
625                    let Some(data) = body.get("data") else {
626                        let (business_status, reason) = hydration_reply_diagnostic(&body);
627                        tracing::warn!(
628                            req_id,
629                            operation = "channel_hydration",
630                            phase = "http_reply",
631                            business_status,
632                            reason,
633                            "hydration failed"
634                        );
635                        out.push(crate::read_relay::emit_read_error(
636                            req_id,
637                            "increment hydration reply missing data",
638                        ));
639                        return Ok(());
640                    };
641                    if data.is_null() {
642                        out.push(crate::read_relay::emit_read_body(
643                            req_id,
644                            serde_json::Value::Null,
645                        ));
646                        return Ok(());
647                    }
648                    let remote_company = data
649                        .get("teamId")
650                        .or_else(|| data.get("team_id"))
651                        .and_then(serde_json::Value::as_str);
652                    if remote_company != Some(self.config.company_id.as_str()) {
653                        out.push(crate::read_relay::emit_read_error(
654                            req_id,
655                            "increment hydration tenant mismatch",
656                        ));
657                        return Ok(());
658                    }
659                    let Some(increment) = crate::ws::parser::parse_increment_channel(data) else {
660                        tracing::warn!(
661                            req_id,
662                            "increment hydration reply has invalid IncrementChannel"
663                        );
664                        out.push(crate::read_relay::emit_read_error(
665                            req_id,
666                            "increment hydration reply has invalid channel",
667                        ));
668                        return Ok(());
669                    };
670                    if let Some(authority_unread) = data
671                        .get("unreadCount")
672                        .or_else(|| data.get("unread_count"))
673                        .and_then(serde_json::Value::as_i64)
674                    {
675                        self.state
676                            .hydration_authority_unreads
677                            .insert(increment.channel_id, authority_unread);
678                    }
679                    let snapshot = self.hydration_snapshot(increment.channel_id);
680                    let persist_corr = self.alloc_corr_internal();
681                    let has_persist =
682                        self.ingest_increment_hydration(&increment, persist_corr, now_ms, out);
683                    // needSync 是服务端按其会话水位给的提示，不能覆盖本端持久化事实。独立窗口
684                    // 首次见到频道时本地 cursor=0；只要仍落后 lastEventSeq，就必须从本地 cursor
685                    // 拉 sync，把 events/messages 与 cursor 在同一 commit 链落稳后再做终态投影。
686                    let local_cursor = self
687                        .state
688                        .channels
689                        .get(&increment.channel_id)
690                        .map(|channel| channel.cursor.value())
691                        .unwrap_or(crate::state::Seq(0));
692                    let requires_sync =
693                        increment.need_sync || local_cursor < increment.last_event_seq;
694                    if requires_sync {
695                        self.state.need_sync_skip.remove(&increment.channel_id);
696                    }
697                    if has_persist {
698                        self.state.corr_map.insert(
699                            persist_corr,
700                            CorrelationContext::IncrementHydrationPersist {
701                                channel_id: increment.channel_id,
702                                req_id: req_id.to_string(),
703                                need_sync: requires_sync,
704                                raw_increment: increment.raw.as_ref().to_vec(),
705                                snapshot,
706                                emit_channel_increment,
707                            },
708                        );
709                    } else {
710                        tracing::warn!(
711                            channel_id = increment.channel_id.as_str(),
712                            "increment hydration produced no durable writes; final projection suppressed"
713                        );
714                        out.push(crate::read_relay::emit_read_error(
715                            req_id,
716                            "increment hydration produced no durable writes",
717                        ));
718                    }
719                }
720                Err(e) => {
721                    tracing::warn!(req_id, error = ?e, "increment hydration envelope decode failed");
722                    out.push(crate::read_relay::emit_read_error(
723                        req_id,
724                        "response envelope decode failed",
725                    ));
726                }
727            },
728            PortOutcome::Err(e) => {
729                tracing::warn!(req_id, error = ?e, "increment hydration http failed");
730                out.push(crate::read_relay::emit_read_error(
731                    req_id,
732                    "http request failed",
733                ));
734            }
735        }
736        Ok(())
737    }
738
739    pub(crate) fn handle_increment_hydration_persist(
740        &mut self,
741        channel_id: ChannelId,
742        req_id: String,
743        need_sync: bool,
744        raw_increment: Vec<u8>,
745        snapshot: HydrationPersistSnapshot,
746        emit_channel_increment: bool,
747        outcome: &PortOutcome,
748        out: &mut EffectSink,
749    ) -> Result<(), ImError> {
750        match outcome {
751            PortOutcome::Ok(_) => {
752                let ordered_roster = match Self::ordered_hydration_roster(&raw_increment) {
753                    Ok(roster) => roster,
754                    Err(error) => {
755                        self.restore_hydration_snapshot(channel_id, &snapshot);
756                        self.finish_hydration_error(&req_id, channel_id, &error.to_string(), out);
757                        return Ok(());
758                    }
759                };
760                self.state
761                    .hydration_ordered_rosters
762                    .insert(channel_id, ordered_roster);
763                self.after_increment_persist(
764                    channel_id,
765                    req_id,
766                    need_sync,
767                    emit_channel_increment,
768                    out,
769                )?;
770            }
771            PortOutcome::Err(e) => {
772                self.restore_hydration_snapshot(channel_id, &snapshot);
773                tracing::warn!(
774                    channel_id = channel_id.as_str(),
775                    error = ?e,
776                    "increment hydration channel/member persist failed"
777                );
778                self.finish_hydration_error(
779                    &req_id,
780                    channel_id,
781                    "channel/member persist failed",
782                    out,
783                );
784            }
785        }
786        Ok(())
787    }
788
789    fn after_increment_persist(
790        &mut self,
791        channel_id: ChannelId,
792        req_id: String,
793        need_sync: bool,
794        emit_channel_increment: bool,
795        out: &mut EffectSink,
796    ) -> Result<(), ImError> {
797        self.state.hydration_pending.insert(channel_id);
798        if emit_channel_increment {
799            self.state
800                .hydration_emit_channel_increment
801                .insert(channel_id);
802        }
803        self.state
804            .hydration_req_ids
805            .insert(channel_id, req_id.clone());
806        if !need_sync {
807            return self.finish_increment_hydration(channel_id, out);
808        }
809        let api_base_url = self.config.api_base_url.clone();
810        self.with_state_and_corr_allocator(|state, alloc| {
811            crate::sync_scheduler::enqueue_and_drain_with_trigger(
812                state,
813                &api_base_url,
814                &[channel_id],
815                crate::state::SyncTrigger::Hydration,
816                alloc,
817                out,
818            );
819        });
820        Ok(())
821    }
822
823    /// terminal sync 后从本地表读回完整 dialogList 与该 channel messages。
824    pub(crate) fn finish_increment_hydration(
825        &mut self,
826        channel_id: ChannelId,
827        out: &mut EffectSink,
828    ) -> Result<(), ImError> {
829        if !self.state.hydration_pending.remove(&channel_id) {
830            return Ok(());
831        }
832        let Some(req_id) = self.state.hydration_req_ids.get(&channel_id).cloned() else {
833            return Ok(());
834        };
835        self.start_hydration_channel_readback(req_id, channel_id, out);
836        Ok(())
837    }
838}
helix_im/increment_hydration.rs

helix_im/
increment_hydration.rs
