OXYGENHUB
# a2ef53c5db26 oxygenhub / crates / hub / src / events.rs
7.3 KiB · Rust · last changed Live pages L1: an event bus of topics, versions and a replay buffer 20 days ago source blame history raw
1 
//! In-process event bus: nudges the webhook worker and streams proposal
2 
//! updates to open pages over server-sent events (hub spec §13.1).
3
4 
use hub_api::ProposalEvent;
5 
use tokio::sync::broadcast;
6
7 
pub struct Events {
8 
    pub deliveries: broadcast::Sender<()>,
9 
    pub proposals: broadcast::Sender<(i64, ProposalEvent)>,
10 
    /// Topics for live pages (oxygenhub#203).
11 
    pub live: LiveBus,
12 
}
13
14 
impl Default for Events {
15 
    fn default() -> Self {
16 
        Events { deliveries: broadcast::channel(16).0, proposals: broadcast::channel(256).0, live: LiveBus::default() }
17 
    }
18 
}
19
20 
impl Events {
21 
    pub fn notify_deliveries(&self) {
22 
        let _ = self.deliveries.send(());
23 
    }
24
25 
    /// Announce something about a proposal of a repository.
26 
    pub fn proposal(&self, repo_id: i64, event: ProposalEvent) {
27 
        let _ = self.proposals.send((repo_id, event));
28 
    }
29 
}
30
31 
/// What changed, for live pages (docs/live.md, oxygenhub#203): a topic and
32 
/// a kind, never the data. A page that hears its topic fetches the part
33 
/// that changed from its normal route, which checks the viewer's access.
34 
#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize)]
35 
pub struct LiveEvent {
36 
    /// Increasing across the hub: what `Last-Event-ID` resumes from.
37 
    pub id: u64,
38 
    pub topic: String,
39 
    /// The topic's own version: how many events it has had.
40 
    pub version: u64,
41 
    /// What happened: `push`, `comment`, `review`, `check`, `jobs`, ...
42 
    pub kind: String,
43 
}
44
45 
/// Topics (docs/live.md §4).
46 
pub mod topic {
47 
    pub fn repo(id: i64) -> String {
48 
        format!("repo:{id}")
49 
    }
50 
    pub fn proposal(repo: i64, n: i64) -> String {
51 
        format!("repo:{repo}/proposal:{n}")
52 
    }
53 
    pub fn issue(repo: i64, n: i64) -> String {
54 
        format!("repo:{repo}/issue:{n}")
55 
    }
56 
    pub fn run(id: i64) -> String {
57 
        format!("run:{id}")
58 
    }
59 
    pub fn user(id: i64) -> String {
60 
        format!("user:{id}")
61 
    }
62 
    pub fn diffusion(repo: i64) -> String {
63 
        format!("diffusion:{repo}")
64 
    }
65 
}
66
67 
/// Events kept for replay: a reconnecting page catches up from these, or is
68 
/// told to reload when it missed more.
69 
pub const REPLAY: usize = 1024;
70
71 
/// The live event bus: one broadcast that subscribers filter by topic, a
72 
/// version per topic, and the last [`REPLAY`] events.
73 
pub struct LiveBus {
74 
    tx: broadcast::Sender<LiveEvent>,
75 
    state: std::sync::Mutex<BusState>,
76 
}
77
78 
#[derive(Default)]
79 
struct BusState {
80 
    next_id: u64,
81 
    versions: std::collections::HashMap<String, u64>,
82 
    recent: std::collections::VecDeque<LiveEvent>,
83 
}
84
85 
impl Default for LiveBus {
86 
    fn default() -> Self {
87 
        LiveBus { tx: broadcast::channel(1024).0, state: std::sync::Mutex::new(BusState { next_id: 1, ..Default::default() }) }
88 
    }
89 
}
90
91 
/// What a reconnecting subscriber gets: the events it missed, or `Reload`
92 
/// when they are no longer kept.
93 
#[derive(Debug, PartialEq, Eq)]
94 
pub enum Replay {
95 
    Events(Vec<LiveEvent>),
96 
    Reload,
97 
}
98
99 
impl LiveBus {
100 
    /// Publish an event on `topic`.
101 
    pub fn publish(&self, topic: &str, kind: &str) -> LiveEvent {
102 
        let event = {
103 
            let mut s = self.state.lock().unwrap_or_else(|p| p.into_inner());
104 
            let id = s.next_id;
105 
            s.next_id += 1;
106 
            let version = {
107 
                let v = s.versions.entry(topic.to_string()).or_insert(0);
108 
                *v += 1;
109 
                *v
110 
            };
111 
            let e = LiveEvent { id, topic: topic.into(), version, kind: kind.into() };
112 
            if s.recent.len() == REPLAY {
113 
                s.recent.pop_front();
114 
            }
115 
            s.recent.push_back(e.clone());
116 
            e
117 
        };
118 
        let _ = self.tx.send(event.clone());
119 
        event
120 
    }
121
122 
    pub fn subscribe(&self) -> broadcast::Receiver<LiveEvent> {
123 
        self.tx.subscribe()
124 
    }
125
126 
    /// The newest event id (0 before any).
127 
    pub fn last_id(&self) -> u64 {
128 
        self.state.lock().unwrap_or_else(|p| p.into_inner()).next_id - 1
129 
    }
130
131 
    /// The events on `topics` after `last_id`, or `Reload` when some after
132 
    /// it are no longer kept.
133 
    pub fn since(&self, last_id: u64, topics: &[String]) -> Replay {
134 
        let s = self.state.lock().unwrap_or_else(|p| p.into_inner());
135 
        let oldest = s.recent.front().map(|e| e.id).unwrap_or(s.next_id);
136 
        if last_id + 1 < oldest {
137 
            return Replay::Reload;
138 
        }
139 
        Replay::Events(s.recent.iter().filter(|e| e.id > last_id && topics.contains(&e.topic)).cloned().collect())
140 
    }
141
142 
    /// The topics a webhook event touches: the repository always, and the
143 
    /// proposal or issue its data names.
144 
    pub fn publish_hook(&self, repo_id: i64, event: &str, data: &serde_json::Value) {
145 
        let kind = event.split('.').next().unwrap_or(event);
146 
        self.publish(&topic::repo(repo_id), kind);
147 
        let number = data["number"].as_i64().or_else(|| data["issue"]["number"].as_i64());
148 
        let is_issue = event.starts_with("issue") || event == "comment.issue";
149 
        let is_proposal = event.starts_with("proposal") || matches!(event, "review" | "comment" | "comment.proposal");
150 
        match number {
151 
            Some(n) if is_issue => {
152 
                self.publish(&topic::issue(repo_id, n), kind);
153 
            }
154 
            Some(n) if is_proposal => {
155 
                self.publish(&topic::proposal(repo_id, n), kind);
156 
            }
157 
            _ => {}
158 
        }
159 
    }
160 
}
161
162 
#[cfg(test)]
163 
mod tests {
164 
    use super::*;
165
166 
    #[test]
167 
    fn events_have_ids_and_versions_and_replay_or_say_reload() {
168 
        let bus = LiveBus::default();
169 
        let mut rx = bus.subscribe();
170 
        let a = bus.publish("repo:1", "push");
171 
        let b = bus.publish("repo:2", "push");
172 
        let c = bus.publish("repo:1", "check");
173 
        assert_eq!((a.id, b.id, c.id), (1, 2, 3));
174 
        assert_eq!((a.version, b.version, c.version), (1, 1, 2), "versions count per topic");
175 
        assert_eq!(rx.try_recv().unwrap(), a, "subscribers hear every event");
176 
        assert_eq!(bus.since(1, &["repo:1".into()]), Replay::Events(vec![c.clone()]), "only the topics asked for");
177 
        assert_eq!(bus.since(3, &["repo:1".into()]), Replay::Events(vec![]));
178 
        for _ in 0..REPLAY {
179 
            bus.publish("repo:3", "push");
180 
        }
181 
        assert_eq!(bus.since(1, &["repo:1".into()]), Replay::Reload, "too old to replay");
182 
        assert_eq!(bus.last_id(), 3 + REPLAY as u64);
183 
    }
184
185 
    #[test]
186 
    fn a_webhook_event_names_its_repository_and_its_proposal_or_issue() {
187 
        let bus = LiveBus::default();
188 
        bus.publish_hook(7, "review", &serde_json::json!({ "number": 4, "verdict": "approve" }));
189 
        bus.publish_hook(7, "issue.opened", &serde_json::json!({ "issue": { "number": 9 } }));
190 
        bus.publish_hook(7, "comment.issue", &serde_json::json!({ "number": 9 }));
191 
        bus.publish_hook(7, "push", &serde_json::json!({ "updates": [] }));
192 
        let Replay::Events(all) = bus.since(0, &["repo:7".into(), "repo:7/proposal:4".into(), "repo:7/issue:9".into()]) else { panic!() };
193 
        let seen: Vec<(String, String)> = all.into_iter().map(|e| (e.topic, e.kind)).collect();
194 
        assert_eq!(
195 
            seen,
196 
            [
197 
                ("repo:7", "review"),
198 
                ("repo:7/proposal:4", "review"),
199 
                ("repo:7", "issue"),
200 
                ("repo:7/issue:9", "issue"),
201 
                ("repo:7", "comment"),
202 
                ("repo:7/issue:9", "comment"),
203 
                ("repo:7", "push"),
204 
            ]
205 
            .map(|(t, k)| (t.to_string(), k.to_string()))
206 
        );
207 
    }
208 
}