helix-im 0.1.5

基于 Helix Core 的确定性 MessageV3 IM 业务模块
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104

//! `post_schedule_canceled` action handler（排程消息撤：channel has_schedule_post=0，spec §S1）。
//!
//! 行为真源：现网 `handlers/schedule.rs:35 handle_canceled` → `Channel::update_fields(
//! has_schedule_post=false)` → emit `im:channel:schedule-canceled{channelId, hasSchedulePost:false}`。
//!
//! wire（`SchedulePostData` types.rs:367）：`channelId`。helix：定点列 patch
//! （`channel_set_cols`：UPDATE channel SET has_schedule_post=0 WHERE id=?）。缺 channelId → noop。

use helix_core::{Effect, EffectSink};

use crate::error::ImError;

use super::super::{ImWsContext, WsFrame, WsHandlerRegistration, WsMessageHandler};

const SCHEDULE_CANCELED_ACTION: &str = "post_schedule_canceled";

struct ScheduleCanceledHandler;

impl WsMessageHandler for ScheduleCanceledHandler {
    /// 返回该 handler 消费的唯一 G09 authority action。
    fn action(&self) -> &'static str {
        SCHEDULE_CANCELED_ACTION
    }

    /// 将 G09 cancel authority 放入原子持久屏障，成功前不发布终态事件。
    fn handle(
        &self,
        ctx: &mut ImWsContext<'_>,
        frame: &WsFrame,
        out: &mut EffectSink,
    ) -> Result<(), ImError> {
        let Ok(data) = frame.data_required() else {
            return Ok(());
        };
        let Some(fact) = crate::channel::write::ScheduleFact::from_canceled_ws(
            data,
            frame.event_seq().map(|seq| seq.0),
        ) else {
            return Ok(());
        };
        let requested_by_viewer = ctx
            .state
            .pending_schedule_cancel_requests
            .contains_key(&fact.channel_id);
        if !requested_by_viewer
            && (fact.owner_user_id.is_empty() || fact.owner_user_id != ctx.auth_user_id)
        {
            return Ok(());
        }
        let committed = ctx
            .state
            .committed_schedule_revisions
            .get(&fact.channel_id)
            .copied()
            .unwrap_or(0);
        let inflight = ctx
            .state
            .inflight_schedule_revisions
            .get(&fact.channel_id)
            .copied()
            .unwrap_or(0);
        if fact.revision <= committed.max(inflight) {
            return Ok(());
        }
        let channel_id = fact.channel_id;
        let revision = fact.revision;
        let causation_id = ctx
            .state
            .pending_schedule_cancel_requests
            .get(&channel_id)
            .cloned();
        let corr = ctx.alloc_corr();
        ctx.state
            .inflight_schedule_revisions
            .insert(channel_id, revision);
        ctx.state.corr_map.insert(
            corr,
            crate::state::CorrelationContext::ScheduleCanceledPersist {
                channel_id,
                revision,
                causation_id,
            },
        );
        out.push(Effect::PersistAtomic {
            corr,
            ops: fact.storage_ops(),
        });
        Ok(())
    }
}

static SCHEDULE_CANCELED_HANDLER: ScheduleCanceledHandler = ScheduleCanceledHandler;
#[cfg(target_arch = "wasm32")]
/// 在 wasm inventory 不可自动发现时保留静态 handler。
pub(super) fn inventory_link_anchor() {
    std::hint::black_box(&SCHEDULE_CANCELED_HANDLER);
}

inventory::submit! {
    WsHandlerRegistration {
        action: SCHEDULE_CANCELED_ACTION,
        handler: &SCHEDULE_CANCELED_HANDLER,
    }
}