Skip to content
Sign up

flagon-io/g1t

public
0

Where people and agents ship software together. The open-source git platform for the whole job: issues, agents, checks and deploys to the edge.

g1t.sh
gitagentscloudflarerustforgemcp
231 linesCodeBlameRaw

Pick any line to see why it is the way it is: the commit, the pull request and issue it came from, and what the agent was thinking.

Merge branch 'worktree-agent-ad531323a21f4c975'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
17use g1t_contracts::events::{Event, ListArgs};
18use serde_json::Value;
19use worker::{Fetcher, Result};
20
21/// The longest a call holds on, in seconds.
22pub const MAX_WAIT_SECONDS: u64 = 25;
23/// How often a held call asks again.
24const POLL_MS: u64 = 1_000;
25/// The most events one call returns; the events service's own page.
26const MAX_LIMIT: u32 = 200;
27
28/// What a `list_events` call asks for, read from its input.
29#[derive(Debug, Default, PartialEq)]
30pub 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.
42fn 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
62fn 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
70fn 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.
79pub 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
122impl 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.
140pub 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.
150pub 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)]
166mod 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}

This file's history is long; its oldest lines are credited to the oldest commit read.