events.rs 97.3%

107 of 110 lines run, 28 of 37 functions · run #488 on proposal/340 · run missed

//! Live pages' events (docs/live.md, oxygenhub#203): what a topic is, and
//! the replay buffer that numbers each process's events for `Last-Event-ID`.
//! They travel on the event bus ([`crate::bus`]), which carries them between
//! processes too.

use tokio::sync::broadcast;

/// What changed, for live pages (docs/live.md, oxygenhub#203): a topic and
/// a kind, never the data. A page that hears its topic fetches the part
/// that changed from its normal route, which checks the viewer's access.
#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize)]
pub struct LiveEvent {
    /// Increasing in this process: what `Last-Event-ID` resumes from. Each
    /// process numbers the events it hears, its own and other processes'.
    pub id: u64,
    pub topic: String,
    /// The topic's own version: how many events it has had.
    pub version: u64,
    /// What happened: `push`, `comment`, `review`, `check`, `jobs`, ...
    pub kind: String,
}

/// Topics (docs/live.md §4).
pub mod topic {
    pub fn repo(id: i64) -> String {
        format!("repo:{id}")
    }
    pub fn proposal(repo: i64, n: i64) -> String {
        format!("repo:{repo}/proposal:{n}")
    }
    pub fn issue(repo: i64, n: i64) -> String {
        format!("repo:{repo}/issue:{n}")
    }
    pub fn run(id: i64) -> String {
        format!("run:{id}")
    }
    pub fn user(id: i64) -> String {
        format!("user:{id}")
    }
    pub fn diffusion(repo: i64) -> String {
        format!("diffusion:{repo}")
    }
}

/// Events kept for replay: a reconnecting page catches up from these, or is
/// told to reload when it missed more.
pub const REPLAY: usize = 1024;

/// The live pages' buffer: one broadcast that subscribers filter by topic, a
/// version per topic, and the last [`REPLAY`] events.
pub struct LiveBus {
    tx: broadcast::Sender<LiveEvent>,
    state: std::sync::Mutex<BusState>,
}

#[derive(Default)]
struct BusState {
    next_id: u64,
    versions: std::collections::HashMap<String, u64>,
    recent: std::collections::VecDeque<LiveEvent>,
}

impl Default for LiveBus {
    fn default() -> Self {
        LiveBus { tx: broadcast::channel(1024).0, state: std::sync::Mutex::new(BusState { next_id: 1, ..Default::default() }) }
    }
}

/// What a reconnecting subscriber gets: the events it missed, or `Reload`
/// when they are no longer kept.
#[derive(Debug, PartialEq, Eq)]
pub enum Replay {
    Events(Vec<LiveEvent>),
    Reload,
}

impl LiveBus {
    /// Record an event on `topic` and tell this process's subscribers.
    /// Producers publish through the bus ([`crate::bus::Bus::live`]), which
    /// records it here and in every other process.
    pub fn publish(&self, topic: &str, kind: &str) -> LiveEvent {
        let event = {
            let mut s = self.state.lock().unwrap_or_else(|p| p.into_inner());
            let id = s.next_id;
            s.next_id += 1;
            let version = {
                let v = s.versions.entry(topic.to_string()).or_insert(0);
                *v += 1;
                *v
            };
            let e = LiveEvent { id, topic: topic.into(), version, kind: kind.into() };
            if s.recent.len() == REPLAY {
                s.recent.pop_front();
            }
            s.recent.push_back(e.clone());
            e
        };
        let _ = self.tx.send(event.clone());
        event
    }

    pub fn subscribe(&self) -> broadcast::Receiver<LiveEvent> {
        self.tx.subscribe()
    }

    /// The newest event id (0 before any).
    pub fn last_id(&self) -> u64 {
        self.state.lock().unwrap_or_else(|p| p.into_inner()).next_id - 1
    }

    /// The events on `topics` after `last_id`, or `Reload` when some after
    /// it are no longer kept.
    pub fn since(&self, last_id: u64, topics: &[String]) -> Replay {
        let s = self.state.lock().unwrap_or_else(|p| p.into_inner());
        // An id this process never gave: from before it restarted, or from
        // another process's numbering (a page that reconnected through
        // another web node, docs/roles.md). What it missed is unknown.
        if last_id >= s.next_id {
            return Replay::Reload;
        }
        let oldest = s.recent.front().map(|e| e.id).unwrap_or(s.next_id);
        if last_id + 1 < oldest {
            return Replay::Reload;
        }
        Replay::Events(s.recent.iter().filter(|e| e.id > last_id && topics.contains(&e.topic)).cloned().collect())
    }
}

/// The live topics a webhook event touches, with the kind for each: the
/// repository always, and the proposal or issue its data names.
pub fn hook_topics(repo_id: i64, event: &str, data: &serde_json::Value) -> Vec<(String, String)> {
    let kind = event.split('.').next().unwrap_or(event).to_string();
    let mut topics = vec![(topic::repo(repo_id), kind.clone())];
    let number = data["number"].as_i64().or_else(|| data["issue"]["number"].as_i64());
    let is_issue = event.starts_with("issue") || event == "comment.issue";
    let is_proposal = event.starts_with("proposal") || matches!(event, "review" | "comment" | "comment.proposal");
    match number {
        Some(n) if is_issue => topics.push((topic::issue(repo_id, n), kind)),
        Some(n) if is_proposal => topics.push((topic::proposal(repo_id, n), kind)),
        _ => {}
    }
    topics
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn events_have_ids_and_versions_and_replay_or_say_reload() {
        let bus = LiveBus::default();
        let mut rx = bus.subscribe();
        let a = bus.publish("repo:1", "push");
        let b = bus.publish("repo:2", "push");
        let c = bus.publish("repo:1", "check");
        assert_eq!((a.id, b.id, c.id), (1, 2, 3));
        assert_eq!((a.version, b.version, c.version), (1, 1, 2), "versions count per topic");
        assert_eq!(rx.try_recv().unwrap(), a, "subscribers hear every event");
        assert_eq!(bus.since(1, &["repo:1".into()]), Replay::Events(vec![c.clone()]), "only the topics asked for");
        assert_eq!(bus.since(3, &["repo:1".into()]), Replay::Events(vec![]));
        assert_eq!(bus.since(4, &["repo:1".into()]), Replay::Reload, "an id never given here: another process's, or before a restart");
        for _ in 0..REPLAY {
            bus.publish("repo:3", "push");
        }
        assert_eq!(bus.since(1, &["repo:1".into()]), Replay::Reload, "too old to replay");
        assert_eq!(bus.last_id(), 3 + REPLAY as u64);
    }

    #[test]
    fn a_webhook_event_names_its_repository_and_its_proposal_or_issue() {
        let mut seen = vec![];
        seen.extend(hook_topics(7, "review", &serde_json::json!({ "number": 4, "verdict": "approve" })));
        seen.extend(hook_topics(7, "issue.opened", &serde_json::json!({ "issue": { "number": 9 } })));
        seen.extend(hook_topics(7, "comment.issue", &serde_json::json!({ "number": 9 })));
        seen.extend(hook_topics(7, "push", &serde_json::json!({ "updates": [] })));
        assert_eq!(
            seen,
            [
                ("repo:7", "review"),
                ("repo:7/proposal:4", "review"),
                ("repo:7", "issue"),
                ("repo:7/issue:9", "issue"),
                ("repo:7", "comment"),
                ("repo:7/issue:9", "comment"),
                ("repo:7", "push"),
            ]
            .map(|(t, k)| (t.to_string(), k.to_string()))
        );
    }
}

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

`/`	Search
`g` `c`	Code
`g` `h`	History
`g` `b`	Bookmarks
`g` `t`	Trust
`y`	Permalink to this state (files and trees)
`?`	This sheet

Coverage / crates/hub/src/events.rs 97.3%

Keyboard