Skip to content
Sign up

flagon-io/g1t

public
00

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

gitagentscloudflarerustforgemcp
329 linesCodeBlameRaw
1//! What a workspace's work says over the last weeks (`workspace_signals`):
2//! the reads behind Home's daily signal and the Workspace overview, one
3//! row per repository the viewer can read, as `delivery_insights` reads
4//! them (insights.rs).
5//!
6//! The definitions, which the SQL beside this file (signals/) holds:
7//!
8//! - A pull request is an **agent's** when its `agent` is `agent` (opened
9//! as an agent's), g1t's agent ([`AGENT_ID`]) opened it, or a harness
10//! session is tied to it, tied as `PULL_SESSIONS` (coordination/links.rs)
11//! ties them: a `pull` session link, the session's own `pull`, or a push
12//! to its branch when it is not on a fork.
13//! - Its **owner** is the person it was requested by, else its author.
14//! - A **person's verdict** is an approval or request for changes by
15//! someone other than the owner, not g1t's agent and not a workspace
16//! agent (`agent_id` set). The first is its **first review**, as
17//! insights.sql has it.
18//! - **Rework**: an agent's pull request merged in the week that a person
19//! asked for changes on before it merged.
20//! - **Draft time**: for an agent's pull request whose first tied session
21//! started before it opened, the seconds from that start to its opening,
22//! counted in the week it opened.
23//! - **Stalled** in a week: opened more than seven days before the week
24//! ended (now, when that is earlier) and not finished by then: not
25//! merged, and not closed (a closed one's `updated_at` stands in for when
26//! it closed). Drafts count.
27//! - A day's **reviews** are comments of any kind on a pull request by a
28//! person other than its owner (not g1t's agent, a workspace agent or g1t
29//! itself); **reworked**, the ones of them asking for changes.
30//! - **Waiting**: open (not draft) pull requests with no first review.
31//! - **Authors** are owners who are people: not g1t's agent, a workspace
32//! agent (`agt_…`), a workspace (`wsp_…`) or g1t.
33use std::collections::BTreeSet;
34
35use g1t_contracts::identity::AGENT_ID;
36use g1t_contracts::repos::{ReadableArgs, Repo};
37use g1t_contracts::time::{parse_rfc3339, rfc3339};
38use g1t_contracts::work::{
39 SignalAuthor, SignalDay, SignalItem, SignalMilestone, SignalPull, SignalWeek, State, WaitingPull, WorkspaceSignals,
40 WorkspaceSignalsArgs,
41};
42use serde::{Deserialize, Serialize};
43use worker::Result;
44use worker::wasm_bindgen::JsValue;
45
46use crate::Work;
47use crate::insights::valid_day;
48use crate::items;
49
50const DAY_MS: u64 = 86_400_000;
51/// How far `since` may reach back before `until`.
52const MAX_SINCE_DAYS: u64 = 400;
53/// `pulls` returns at most this many; pulls.sql asks for one more.
54const PULLS_LIMIT: usize = 3000;
55/// `recent` in authors: the days ending with `until`.
56const RECENT_DAYS: u64 = 90;
57
58pub(crate) const WEEKS: &str = concat!(include_str!("signals/base.sql"), include_str!("signals/weeks.sql"));
59pub(crate) const DAYS: &str = concat!(include_str!("signals/base.sql"), include_str!("signals/days.sql"));
60pub(crate) const WAITING: &str = concat!(include_str!("signals/base.sql"), include_str!("signals/waiting.sql"));
61pub(crate) const PULLS: &str = concat!(include_str!("signals/base.sql"), include_str!("signals/pulls.sql"));
62pub(crate) const AUTHORS: &str = include_str!("signals/authors.sql");
63/// ?1 the workspace.
64pub(crate) const MILESTONES: &str = "SELECT number, title, state, due_on, created_at, closed_at
65 FROM milestones WHERE workspace = ?1 ORDER BY number";
66
67/// One week's bounds, as weeks.sql reads them from JSON.
68#[derive(Debug, Serialize)]
69struct WeekBounds {
70 /// Its first UTC day, `YYYY-MM-DD`.
71 start: String,
72 from: String,
73 to: String,
74 /// When it is read as ended for stalled pull requests: `to`, or now
75 /// when that is earlier.
76 cut: String,
77}
78
79/// What the queries are bound to, worked out from the arguments.
80#[derive(Debug)]
81struct Window {
82 /// JSON array of [`WeekBounds`], oldest first.
83 weeks: String,
84 /// The first of `days`, and the day after `until`, as `YYYY-MM-DDT00:00:00`.
85 days_from: String,
86 end: String,
87 /// The start of `since`.
88 since: String,
89 /// The first of the 90 days ending with `until`.
90 recent_from: String,
91}
92
93fn day_start(day: &str) -> Option<u64> {
94 parse_rfc3339(&format!("{day}T00:00:00Z"))
95}
96
97fn day_of(ms: u64) -> String {
98 rfc3339(ms)[..10].to_owned()
99}
100
101fn bound(ms: u64) -> String {
102 format!("{}T00:00:00", day_of(ms))
103}
104
105/// The bounds `a` asks for at `now`, or `None` when it asks for too many
106/// repositories or an invalid range.
107fn window(a: &WorkspaceSignalsArgs, now: u64) -> Option<Window> {
108 if a.repo_ids.len() > 100
109 || !(1..=13).contains(&a.weeks)
110 || !(1..=28).contains(&a.days)
111 || !valid_day(&a.until)
112 || !valid_day(&a.since)
113 || a.since > a.until
114 {
115 return None;
116 }
117 let until = day_start(&a.until)?;
118 let since = day_start(&a.since)?;
119 if until - since > MAX_SINCE_DAYS * DAY_MS {
120 return None;
121 }
122 let end = until + DAY_MS;
123 let first = end.checked_sub(u64::from(a.weeks) * 7 * DAY_MS)?;
124 let weeks: Vec<WeekBounds> = (0..u64::from(a.weeks))
125 .map(|i| {
126 let from = first + i * 7 * DAY_MS;
127 let to = from + 7 * DAY_MS;
128 WeekBounds {
129 start: day_of(from),
130 from: rfc3339(from),
131 to: rfc3339(to),
132 cut: rfc3339(to.min(now)),
133 }
134 })
135 .collect();
136 Some(Window {
137 weeks: serde_json::to_string(&weeks).ok()?,
138 days_from: bound(end.checked_sub(u64::from(a.days) * DAY_MS)?),
139 end: bound(end),
140 since: bound(since),
141 recent_from: bound(end.checked_sub(RECENT_DAYS * DAY_MS)?),
142 })
143}
144
145/// Whether an owner is a person: not g1t's agent, a workspace agent, a
146/// workspace or g1t itself.
147fn is_person(id: &str) -> bool {
148 id != AGENT_ID && !id.starts_with("agt_") && !id.starts_with("wsp_") && !g1t_contracts::system::is_system_id(id)
149}
150
151/// A row of waiting.sql.
152#[derive(Deserialize)]
153struct WaitingRow {
154 repo_id: String,
155 number: u32,
156 title: String,
157 created_at: String,
158 agent: u32,
159 draft_seconds: Option<f64>,
160 item: Option<u32>,
161 /// JSON array of numbers.
162 blocks: String,
163}
164
165impl From<WaitingRow> for WaitingPull {
166 fn from(row: WaitingRow) -> Self {
167 WaitingPull {
168 repo_id: row.repo_id,
169 number: row.number,
170 title: row.title,
171 created_at: row.created_at,
172 agent: row.agent != 0,
173 draft_seconds: row.draft_seconds,
174 item: row.item,
175 blocks: serde_json::from_str(&row.blocks).unwrap_or_default(),
176 }
177 }
178}
179
180/// A row of pulls.sql.
181#[derive(Deserialize)]
182struct PullRow {
183 repo_id: String,
184 number: u32,
185 title: String,
186 status: String,
187 agent: u32,
188 owner_id: String,
189 owner_name: String,
190 created_at: String,
191 merged_at: Option<String>,
192 item: Option<u32>,
193 milestone: Option<u32>,
194}
195
196impl From<PullRow> for SignalPull {
197 fn from(row: PullRow) -> Self {
198 SignalPull {
199 repo_id: row.repo_id,
200 number: row.number,
201 title: row.title,
202 status: row.status,
203 agent: row.agent != 0,
204 owner_id: row.owner_id,
205 owner_name: row.owner_name,
206 created_at: row.created_at,
207 merged_at: row.merged_at,
208 item: row.item,
209 milestone: row.milestone,
210 }
211 }
212}
213
214/// The work item numbers `pulls` and `waiting` name, each once.
215fn named_items(pulls: &[SignalPull], waiting: &[WaitingPull]) -> Vec<u32> {
216 let mut numbers: BTreeSet<u32> = pulls.iter().filter_map(|pull| pull.item).collect();
217 for pull in waiting {
218 numbers.extend(pull.item);
219 numbers.extend(pull.blocks.iter().copied());
220 }
221 numbers.into_iter().collect()
222}
223
224impl Work {
225 pub(crate) async fn workspace_signals(&self, a: WorkspaceSignalsArgs) -> Result<WorkspaceSignals> {
226 // The web loader batches larger workspaces, as it does for insights.
227 let Some(window) = window(&a, g1t_kit::now_ms()) else {
228 return Err(worker::Error::RustError(
229 "Invalid signals range or repository batch".into(),
230 ));
231 };
232 let viewer = a.viewer.clone();
233 let readable: Vec<Repo> = g1t_kit::call(
234 &self.repos,
235 "readable",
236 &ReadableArgs {
237 ids: a.repo_ids,
238 viewer: a.viewer,
239 },
240 )
241 .await?;
242 let ids: Vec<&str> = readable.iter().map(|repo| repo.id.as_str()).collect();
243 if ids.is_empty() {
244 return Ok(WorkspaceSignals::default());
245 }
246 // The workspace's own milestones and work items, only for a viewer
247 // who can read one of its repositories.
248 let in_workspace = readable.iter().any(|repo| repo.namespace == a.workspace);
249 let ids: JsValue = serde_json::to_string(&ids)?.into();
250 let workspace: JsValue = a.workspace.as_str().into();
251 let results = self
252 .timing
253 .db(
254 6,
255 self.db.batch(vec![
256 self.db.prepare(WEEKS).bind(&[ids.clone(), window.weeks.into()])?,
257 self.db.prepare(DAYS).bind(&[
258 ids.clone(),
259 window.days_from.as_str().into(),
260 window.end.as_str().into(),
261 ])?,
262 self.db.prepare(WAITING).bind(&[ids.clone(), workspace.clone()])?,
263 self.db
264 .prepare(PULLS)
265 .bind(&[ids.clone(), workspace.clone(), window.since.into()])?,
266 self.db
267 .prepare(AUTHORS)
268 .bind(&[ids, window.recent_from.into(), window.end.into()])?,
269 self.db.prepare(MILESTONES).bind(&[workspace])?,
270 ]),
271 )
272 .await?;
273 let [weeks, days, waiting, pulls, authors, milestones] = results.as_slice() else {
274 return Err(worker::Error::RustError("workspace_signals: missing results".into()));
275 };
276 let mut waiting: Vec<WaitingPull> = waiting
277 .results::<WaitingRow>()?
278 .into_iter()
279 .map(WaitingPull::from)
280 .collect();
281 let mut pulls: Vec<SignalPull> = pulls.results::<PullRow>()?.into_iter().map(SignalPull::from).collect();
282 let pulls_capped = pulls.len() > PULLS_LIMIT;
283 pulls.truncate(PULLS_LIMIT);
284 let authors: Vec<SignalAuthor> = authors
285 .results::<SignalAuthor>()?
286 .into_iter()
287 .filter(|author| is_person(&author.owner_id))
288 .collect();
289 let (items, milestones) = if in_workspace {
290 let numbers = named_items(&pulls, &waiting);
291 let visible = items::visible_numbered(self, &a.workspace, &numbers, viewer.as_ref()).await?;
292 let mut items: Vec<SignalItem> = visible
293 .into_iter()
294 .map(|held| SignalItem {
295 number: held.item.number,
296 title: held.item.title,
297 state: match held.item.state {
298 State::Open => "open",
299 State::Closed => "closed",
300 }
301 .into(),
302 milestone: held.item.milestone.map(|milestone| milestone.number),
303 })
304 .collect();
305 items.sort_by_key(|item| item.number);
306 (items, milestones.results::<SignalMilestone>()?)
307 } else {
308 (Vec::new(), Vec::new())
309 };
310 // A blocked work item the viewer may not see is not named.
311 for pull in &mut waiting {
312 pull.blocks
313 .retain(|number| items.iter().any(|item| item.number == *number));
314 }
315 Ok(WorkspaceSignals {
316 weeks: weeks.results::<SignalWeek>()?,
317 days: days.results::<SignalDay>()?,
318 waiting,
319 pulls,
320 pulls_capped,
321 authors,
322 items,
323 milestones,
324 })
325 }
326}
327
328#[cfg(test)]
329mod tests;