| 1 | //! `list_events`: a repository's timeline, and following it. |
| 2 | //! |
| 3 | //! Without `after` it is the timeline, newest first, paged back with |
| 4 | //! `before`. With `after` (an event id) it is what the log wrote after that |
| 5 | //! event, oldest first, and the last event's id is the cursor for the next |
| 6 | //! call. With `after` and `wait`, a call that finds nothing holds on until |
| 7 | //! something matching is written or `wait` seconds pass, asking the events |
| 8 | //! service about once a second, so an agent with no public address can |
| 9 | //! follow a repository with one request at a time instead of a webhook. |
| 10 | //! |
| 11 | //! Holding a request costs a Worker wall time, not CPU time: between polls |
| 12 | //! it awaits a timer, which is not billed or limited as CPU. Each poll is |
| 13 | //! one subrequest, so a call makes at most `MAX_WAIT_SECONDS` + 1 of them, |
| 14 | //! well under the per-request limit, and the hold stays far below the time |
| 15 | //! clients and proxies give an idle response. |
| 16 | |
| 17 | use g1t_contracts::events::{Event, ListArgs}; |
| 18 | use serde_json::Value; |
| 19 | use worker::{Fetcher, Result}; |
| 20 | |
| 21 | /// The longest a call holds on, in seconds. |
| 22 | pub const MAX_WAIT_SECONDS: u64 = 25; |
| 23 | /// How often a held call asks again. |
| 24 | const POLL_MS: u64 = 1_000; |
| 25 | /// The most events one call returns; the events service's own page. |
| 26 | const MAX_LIMIT: u32 = 200; |
| 27 | |
| 28 | /// What a `list_events` call asks for, read from its input. |
| 29 | #[derive(Debug, Default, PartialEq)] |
| 30 | pub struct Asked { |
| 31 | pub after: Option<String>, |
| 32 | pub before: Option<String>, |
| 33 | pub types: Vec<String>, |
| 34 | pub numbers: Vec<u32>, |
| 35 | pub since: Option<String>, |
| 36 | pub limit: Option<u32>, |
| 37 | /// Milliseconds to hold on; zero without `after`. |
| 38 | pub wait_ms: u64, |
| 39 | } |
| 40 | |
| 41 | /// A list given as an array or as one comma-separated string. |
| 42 | fn list(value: &Value) -> Vec<String> { |
| 43 | let items: Vec<String> = match value { |
| 44 | Value::String(text) => text.split(',').map(str::to_owned).collect(), |
| 45 | Value::Array(items) => items |
| 46 | .iter() |
| 47 | .map(|item| match item { |
| 48 | Value::String(text) => text.clone(), |
| 49 | other => other.to_string(), |
| 50 | }) |
| 51 | .collect(), |
| 52 | Value::Number(number) => vec![number.to_string()], |
| 53 | _ => Vec::new(), |
| 54 | }; |
| 55 | items |
| 56 | .into_iter() |
| 57 | .map(|item| item.trim().trim_start_matches('#').to_owned()) |
| 58 | .filter(|item| !item.is_empty()) |
| 59 | .collect() |
| 60 | } |
| 61 | |
| 62 | fn text(input: &Value, key: &str) -> Option<String> { |
| 63 | input[key] |
| 64 | .as_str() |
| 65 | .map(str::trim) |
| 66 | .filter(|value| !value.is_empty()) |
| 67 | .map(str::to_owned) |
| 68 | } |
| 69 | |
| 70 | fn whole(value: &Value) -> Option<u64> { |
| 71 | match value { |
| 72 | Value::Number(number) => number.as_u64().or_else(|| number.as_f64().filter(|n| *n >= 0.0).map(|n| n as u64)), |
| 73 | Value::String(digits) => digits.trim().parse().ok(), |
| 74 | _ => None, |
| 75 | } |
| 76 | } |
| 77 | |
| 78 | /// Reads a call's input, or says what is wrong with it. |
| 79 | pub fn asked(input: &Value) -> std::result::Result<Asked, String> { |
| 80 | let mut numbers = Vec::new(); |
| 81 | for item in list(&input["numbers"]) { |
| 82 | match item.parse::<u32>() { |
| 83 | Ok(number) if number > 0 => numbers.push(number), |
| 84 | _ => return Err(format!("numbers are issue and pull request numbers, and {item} is not one.")), |
| 85 | } |
| 86 | } |
| 87 | numbers.sort_unstable(); |
| 88 | numbers.dedup(); |
| 89 | let mut types = list(&input["types"]); |
| 90 | types.sort(); |
| 91 | types.dedup(); |
| 92 | let limit = match &input["limit"] { |
| 93 | Value::Null => None, |
| 94 | value => match whole(value) { |
| 95 | Some(limit) if limit > 0 => Some(u32::try_from(limit).unwrap_or(MAX_LIMIT).min(MAX_LIMIT)), |
| 96 | _ => return Err("limit is a whole number from 1 to 200.".to_owned()), |
| 97 | }, |
| 98 | }; |
| 99 | let after = text(input, "after"); |
| 100 | let wait = match &input["wait"] { |
| 101 | Value::Null => 0, |
| 102 | value => match whole(value) { |
| 103 | Some(seconds) => seconds.min(MAX_WAIT_SECONDS), |
| 104 | None => return Err(format!("wait is a number of seconds, at most {MAX_WAIT_SECONDS}.")), |
| 105 | }, |
| 106 | }; |
| 107 | let before = text(input, "before"); |
| 108 | if after.is_some() && before.is_some() { |
| 109 | return Err("Give after to follow the timeline or before to page back through it, not both.".to_owned()); |
| 110 | } |
| 111 | Ok(Asked { |
| 112 | wait_ms: if after.is_some() { wait * 1_000 } else { 0 }, |
| 113 | after, |
| 114 | before, |
| 115 | types, |
| 116 | numbers, |
| 117 | since: text(input, "since"), |
| 118 | limit, |
| 119 | }) |
| 120 | } |
| 121 | |
| 122 | impl Asked { |
| 123 | fn args(&self, repo_id: &str) -> ListArgs { |
| 124 | ListArgs { |
| 125 | repo_id: Some(repo_id.to_owned()), |
| 126 | types: self.types.clone(), |
| 127 | before: self.before.clone(), |
| 128 | after: self.after.clone(), |
| 129 | numbers: self.numbers.clone(), |
| 130 | since: self.since.clone(), |
| 131 | limit: self.limit, |
| 132 | ..ListArgs::default() |
| 133 | } |
| 134 | } |
| 135 | } |
| 136 | |
| 137 | /// How long to sleep before asking again, at `now`, for a call that holds |
| 138 | /// on until `deadline`: `None` when it is time to answer with nothing. |
| 139 | /// The last poll falls on the deadline itself. |
| 140 | pub fn next_poll(now: u64, deadline: u64) -> Option<u64> { |
| 141 | if now >= deadline { |
| 142 | None |
| 143 | } else { |
| 144 | Some(POLL_MS.min(deadline - now)) |
| 145 | } |
| 146 | } |
| 147 | |
| 148 | /// The events of the repository `repo_id` as asked, holding on for them |
| 149 | /// when asked to. |
| 150 | pub async fn list_events(events: &Fetcher, repo_id: &str, asked: &Asked) -> Result<Vec<Event>> { |
| 151 | let args = asked.args(repo_id); |
| 152 | let deadline = g1t_kit::now_ms() + asked.wait_ms; |
| 153 | loop { |
| 154 | let found: Vec<Event> = g1t_kit::call(events, "list", &args).await?; |
| 155 | if !found.is_empty() { |
| 156 | return Ok(found); |
| 157 | } |
| 158 | match next_poll(g1t_kit::now_ms(), deadline) { |
| 159 | Some(ms) => worker::Delay::from(std::time::Duration::from_millis(ms)).await, |
| 160 | None => return Ok(found), |
| 161 | } |
| 162 | } |
| 163 | } |
| 164 | |
| 165 | #[cfg(test)] |
| 166 | mod tests { |
| 167 | use super::*; |
| 168 | use serde_json::json; |
| 169 | |
| 170 | #[test] |
| 171 | fn lists_come_as_commas_or_arrays() { |
| 172 | let a = asked(&json!({ |
| 173 | "types": "work.claimed, integration.failed,,work.claimed", |
| 174 | "numbers": "12,#7, 12", |
| 175 | })) |
| 176 | .unwrap(); |
| 177 | assert_eq!(a.types, vec!["integration.failed", "work.claimed"]); |
| 178 | assert_eq!(a.numbers, vec![7, 12]); |
| 179 | let b = asked(&json!({ "types": ["push"], "numbers": [3, "4"] })).unwrap(); |
| 180 | assert_eq!(b.types, vec!["push"]); |
| 181 | assert_eq!(b.numbers, vec![3, 4]); |
| 182 | assert!(asked(&json!({ "numbers": "seven" })).is_err()); |
| 183 | assert!(asked(&json!({ "numbers": [0] })).is_err()); |
| 184 | } |
| 185 | |
| 186 | #[test] |
| 187 | fn wait_holds_only_a_follow_and_never_past_25_seconds() { |
| 188 | assert_eq!(asked(&json!({ "wait": 10 })).unwrap().wait_ms, 0, "nothing to wait for without after"); |
| 189 | assert_eq!(asked(&json!({ "after": "evt_1", "wait": "10" })).unwrap().wait_ms, 10_000); |
| 190 | assert_eq!(asked(&json!({ "after": "evt_1", "wait": 600 })).unwrap().wait_ms, 25_000); |
| 191 | assert_eq!(asked(&json!({ "after": "evt_1" })).unwrap().wait_ms, 0); |
| 192 | assert!(asked(&json!({ "after": "evt_1", "wait": "soon" })).is_err()); |
| 193 | assert!(asked(&json!({ "after": "evt_1", "wait": -1 })).is_err()); |
| 194 | } |
| 195 | |
| 196 | #[test] |
| 197 | fn limits_are_kept_to_a_page() { |
| 198 | assert_eq!(asked(&json!({ "limit": "500" })).unwrap().limit, Some(200)); |
| 199 | assert_eq!(asked(&json!({ "limit": 20 })).unwrap().limit, Some(20)); |
| 200 | assert_eq!(asked(&json!({})).unwrap().limit, None); |
| 201 | assert!(asked(&json!({ "limit": 0 })).is_err()); |
| 202 | } |
| 203 | |
| 204 | #[test] |
| 205 | fn after_and_before_do_not_mix() { |
| 206 | assert!(asked(&json!({ "after": "evt_2", "before": "evt_1" })).is_err()); |
| 207 | let a = asked(&json!({ "after": "evt_2", "since": "2026-10-12T00:00:00Z" })).unwrap(); |
| 208 | let args = a.args("rep_1"); |
| 209 | assert_eq!(args.after.as_deref(), Some("evt_2")); |
| 210 | assert_eq!(args.since.as_deref(), Some("2026-10-12T00:00:00Z")); |
| 211 | assert_eq!(args.repo_id.as_deref(), Some("rep_1")); |
| 212 | } |
| 213 | |
| 214 | #[test] |
| 215 | fn polls_come_about_once_a_second_and_the_last_on_the_deadline() { |
| 216 | let start = 1_000_000; |
| 217 | let deadline = start + 25_000; |
| 218 | let mut now = start; |
| 219 | let mut polls = 1; // the first, before any sleep |
| 220 | while let Some(ms) = next_poll(now, deadline) { |
| 221 | assert!(ms > 0 && ms <= 1_000); |
| 222 | now += ms; |
| 223 | polls += 1; |
| 224 | } |
| 225 | assert_eq!(now, deadline); |
| 226 | assert_eq!(polls, 26, "one call makes at most 26 polls"); |
| 227 | assert_eq!(next_poll(5, 5), None); |
| 228 | assert_eq!(next_poll(5, 4), None, "a slow poll past the deadline answers"); |
| 229 | assert_eq!(next_poll(0, 300), Some(300)); |
| 230 | } |
| 231 | } |