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.

g1t.sh
gitagentscloudflarerustforgemcp
936 linesCodeBlameRaw
1//! `get_workspace_activity`: what happened across a workspace's
2//! repositories over a span, newest first, as one list a person (or Ask
3//! g1t, or an agent) can read: pull requests merged and opened, issues
4//! opened and closed, releases, deployments and pushes, each with its
5//! repository, who did it, its title where it has one and its address on
6//! the site. "What shipped today?" is one call.
7//!
8//! The repositories are those of the workspace the caller can read, as the
9//! repos service lists them for them; the events service reads their events
10//! as one list (`repo_ids`). Pull requests' titles, which their events do
11//! not carry, come from the work service's newest pull requests of the
12//! repositories that had any; people's names from identity. Everything
13//! after the reads is [`shape`], a plain function, tested on its own.
14//!
15//! Shipping on g1t is reaching the default branch, then production. For the
16//! repositories whose default branch moved, `landed` lists the commits, each
17//! with its author's g1t username where their address is one's
18//! (commits.rs), and whether it is in production ([`production`]): carried
19//! by a successful deploy (a run of a deploy workflow on the default branch,
20//! for a repository that deploys itself, or a project's production
21//! deployment on g1t.page), still deploying, or failed to.
22
23use std::collections::{BTreeSet, HashMap};
24
25use g1t_contracts::actions::{RunsArgs, WorkflowRun};
26use g1t_contracts::events::{Event, ListArgs as ListEventsArgs, MAX_LIST_REPOS};
27use g1t_contracts::identity::UsernamesArgs;
28use g1t_contracts::repos::{Commit, ListArgs as ListReposArgs, LogArgs, Repo, RepoPath};
29use g1t_contracts::time::{parse_rfc3339, rfc3339};
30use g1t_contracts::work::{MAX_PULLS_FOR_REPOS, PullsForReposArgs, RepoPulls};
31use g1t_contracts::{FailureCode, Outcome, Viewer};
32use serde_json::{Value, json};
33use worker::Result;
34
35use crate::commits::{self, Authors};
36use crate::operations::Services;
37
38/// The span when none is given: the last day.
39const DEFAULT_SPAN_MS: u64 = 24 * 60 * 60 * 1000;
40/// The longest span one call reads.
41const MAX_SPAN_MS: u64 = 31 * 24 * 60 * 60 * 1000;
42pub const DEFAULT_LIMIT: u32 = 100;
43pub const MAX_LIMIT: u32 = 200;
44
45/// The kinds a caller may ask for, and the events each is.
46pub const KINDS: &[(&str, &[&str])] = &[
47 ("merged", &["pull.merged"]),
48 ("pulls", &["pull.opened", "pull.merged", "pull.closed"]),
49 ("issues", &["issue.opened", "issue.closed"]),
50 ("releases", &["release.published"]),
51 ("deployments", &["deployment.succeeded", "deployment.failed"]),
52 ("pushes", &["git.push"]),
53 ("reviews", &["review.completed"]),
54 ("comments", &["comment.created"]),
55 ("packages", &["package.published"]),
56 ("runs", &["workflow.completed"]),
57];
58
59/// The most repositories whose default-branch commits are read, and the
60/// most commits that `landed` lists across them.
61const LANDED_REPOS: usize = 8;
62const LANDED_MAX: usize = 40;
63/// How far back each of those repositories' history is read.
64const LANDED_LOG: u32 = 60;
65/// How many of each of those repositories' default-branch runs in the span
66/// are read for its deploys.
67const LANDED_RUNS: u32 = 100;
68
69/// What is read when no kinds are given: everything but comments, which
70/// would crowd out the rest on a busy day.
71const DEFAULT_KINDS: &[&str] = &["pulls", "issues", "releases", "deployments", "pushes", "reviews", "packages", "runs"];
72
73pub fn description() -> &'static str {
74 "What happened across a workspace's repositories that you can read, newest first: pull requests merged, opened and closed, issues opened and closed, releases published, deployments that succeeded or failed, pushes (with `default_branch` true for what landed on the default branch without a pull request), reviews, packages, and workflow runs that finished (`workflow`, `run_number`, `conclusion`, `run_id` for workflow get_run, and `production` true for a run of a deploy workflow on the default branch: a repository that deploys itself). Each event has its `kind`, `time`, `repo` (owner/name), `actor` (a username), and where it has them its `number`, `title`, `url` (its page on g1t), `branch`, `commit`, `commits`, `tag` and `error`. `summary` counts merged pull requests, releases, production deployments (projects' production deployments and successful deploy runs on default branches), failed ones, default-branch pushes and failed runs, and in `shipped` the landed commits by where they are. `landed` lists the commits made on default branches in the span, one line each (`repo`, `commit`, `subject`, `author` as git wrote it, `username` when the author's address is a g1t account's, `time`, `url`), and, where the repository deploys, `production`: `live` (a successful deploy carried it, named in `deploy` with its `url`), `deploying`, `failed` (its deploy failed and none since succeeded) or `not_deployed`. That is what shipped: what reached the default branch, and which of it is live. The span is `since` to `until` (RFC 3339; the last 24 hours when `since` is left out, at most 31 days); narrow it with `kinds` (merged, pulls, issues, releases, deployments, pushes, reviews, comments, packages, runs; all but comments when left out) and `limit` (100 unless you say, at most 200). `complete` is false when more happened than was read: narrow the span or the kinds. For one repository's full timeline use list_events."
75}
76
77pub fn input() -> Value {
78 json!({
79 "type": "object",
80 "properties": {
81 "workspace": { "type": "string", "description": "The workspace's slug, e.g. \"flagon-io\"." },
82 "since": { "type": "string", "description": "RFC 3339: the start of the span, inclusive. The last 24 hours when left out; for \"today\", midnight UTC." },
83 "until": { "type": "string", "description": "RFC 3339: the end of the span, exclusive. Now when left out." },
84 "kinds": {
85 "type": "array",
86 "items": { "type": "string", "enum": KINDS.iter().map(|(name, _)| *name).collect::<Vec<_>>() },
87 "description": "What to read: merged (merged pull requests only), pulls, issues, releases, deployments, pushes, reviews, comments, packages, runs (workflow runs that finished). All but comments when left out.",
88 },
89 "limit": { "type": "integer", "minimum": 1, "maximum": MAX_LIMIT, "description": "The most events to return, newest first: 100 unless you say, at most 200." },
90 },
91 "required": ["workspace"],
92 })
93}
94
95/// The span asked for, in milliseconds, or why it cannot be read.
96pub fn span(input: &Value, now: u64) -> std::result::Result<(u64, u64), String> {
97 let at = |key: &str| -> std::result::Result<Option<u64>, String> {
98 match input.get(key).and_then(Value::as_str).map(str::trim).filter(|text| !text.is_empty()) {
99 None => Ok(None),
100 Some(text) => parse_rfc3339(text).map(Some).ok_or_else(|| format!("{key} must be an RFC 3339 time, such as 2026-10-12T00:00:00Z.")),
101 }
102 };
103 let until = at("until")?.unwrap_or(now).min(now);
104 let since = at("since")?.unwrap_or(until.saturating_sub(DEFAULT_SPAN_MS));
105 if since >= until {
106 return Err("since must be before until.".to_owned());
107 }
108 if until - since > MAX_SPAN_MS {
109 return Err("A span can be at most 31 days. Narrow since and until.".to_owned());
110 }
111 Ok((since, until))
112}
113
114/// The event types the `kinds` asked for read, or the unknown kind.
115pub fn event_types(input: &Value) -> std::result::Result<Vec<&'static str>, String> {
116 // An array, or over REST a comma-separated list.
117 let asked: Vec<String> = match input.get("kinds") {
118 Some(Value::Array(kinds)) => kinds.iter().filter_map(Value::as_str).map(str::to_owned).collect(),
119 Some(Value::String(kinds)) => kinds.split(',').map(str::to_owned).collect(),
120 _ => Vec::new(),
121 }
122 .into_iter()
123 .map(|kind| kind.trim().to_lowercase())
124 .filter(|kind| !kind.is_empty())
125 .collect();
126 let names: Vec<&str> = if asked.is_empty() { DEFAULT_KINDS.to_vec() } else { asked.iter().map(String::as_str).collect() };
127 let mut types = BTreeSet::new();
128 for name in names {
129 let Some((_, kinds)) = KINDS.iter().find(|(kind, _)| *kind == name) else {
130 let known: Vec<&str> = KINDS.iter().map(|(kind, _)| *kind).collect();
131 return Err(format!("There is no kind {name}. Kinds: {}.", known.join(", ")));
132 };
133 types.extend(kinds.iter().copied());
134 }
135 Ok(types.into_iter().collect())
136}
137
138/// The most events asked for: a number, or over REST its text.
139pub fn limit(input: &Value) -> u32 {
140 let asked = match input.get("limit") {
141 Some(Value::String(text)) => text.trim().parse::<u64>().ok(),
142 Some(value) => value.as_u64(),
143 None => None,
144 };
145 asked.map_or(DEFAULT_LIMIT, |n| n.clamp(1, u64::from(MAX_LIMIT)) as u32)
146}
147
148fn text<'a>(data: &'a Value, key: &str) -> Option<&'a str> {
149 data.get(key).and_then(Value::as_str).filter(|text| !text.is_empty())
150}
151
152/// What the reads found, shaped into the answer.
153pub struct Found<'a> {
154 pub workspace: &'a str,
155 pub site: &'a str,
156 pub since: u64,
157 pub until: u64,
158 pub repos: &'a [Repo],
159 /// Newest first, as the events service lists them.
160 pub events: &'a [Event],
161 /// Whether the events service had more than was read.
162 pub more: bool,
163 /// `(repo id, number)` to a pull request's title.
164 pub pull_titles: &'a HashMap<(String, u32), String>,
165 /// Account id to username.
166 pub usernames: &'a HashMap<String, String>,
167}
168
169pub fn shape(found: &Found) -> Value {
170 let site = found.site.trim_end_matches('/');
171 let by_id: HashMap<&str, &Repo> = found.repos.iter().map(|repo| (repo.id.as_str(), repo)).collect();
172 let mut events = Vec::new();
173 let (mut merged, mut releases, mut production, mut landed, mut failed) = (0u32, 0u32, 0u32, 0u32, 0u32);
174 let mut runs_failed = 0u32;
175 for event in found.events {
176 let Some(repo) = event.repo_id.as_deref().and_then(|id| by_id.get(id)) else {
177 continue;
178 };
179 let at = parse_rfc3339(&event.time).unwrap_or_default();
180 if at < found.since || at >= found.until {
181 continue;
182 }
183 let full = format!("{}/{}", repo.namespace, repo.name);
184 let page = format!("{site}/{full}");
185 let data = &event.data;
186 let number = data.get("number").and_then(Value::as_u64).map(|n| n as u32);
187 let actor = event
188 .actor
189 .as_deref()
190 .and_then(|id| found.usernames.get(id).cloned())
191 .or_else(|| data.get("author").and_then(|author| text(author, "username")).map(str::to_owned))
192 .or_else(|| text(data, "triggeredBy").map(str::to_owned));
193 let mut shown = json!({ "kind": event.kind.replace('.', "_"), "time": event.time, "repo": full });
194 if let Some(actor) = actor {
195 shown["actor"] = json!(actor);
196 }
197 match event.kind.as_str() {
198 "pull.opened" | "pull.merged" | "pull.closed" => {
199 let Some(number) = number else { continue };
200 shown["number"] = json!(number);
201 if let Some(title) = found.pull_titles.get(&(repo.id.clone(), number)) {
202 shown["title"] = json!(title);
203 }
204 shown["url"] = json!(format!("{page}/pull/{number}"));
205 if let Some(issue) = data.get("issue").and_then(Value::as_u64) {
206 shown["issue"] = json!(issue);
207 }
208 if event.kind == "pull.merged" {
209 merged += 1;
210 }
211 }
212 "issue.opened" | "issue.closed" => {
213 let Some(number) = number else { continue };
214 shown["number"] = json!(number);
215 if let Some(title) = text(data, "title") {
216 shown["title"] = json!(title);
217 }
218 if let Some(reason) = text(data, "reason") {
219 shown["reason"] = json!(reason);
220 }
221 if let Some(pull) = data.get("resolvedBy").and_then(Value::as_u64) {
222 shown["resolved_by"] = json!(pull);
223 }
224 shown["url"] = json!(format!("{page}/issues/{number}"));
225 }
226 "release.published" => {
227 let tag = text(data, "tagName").unwrap_or_default();
228 shown["tag"] = json!(tag);
229 if let Some(name) = data.get("release").and_then(|release| text(release, "name")) {
230 shown["title"] = json!(name);
231 }
232 shown["url"] = json!(format!("{page}/releases/tag/{tag}"));
233 releases += 1;
234 }
235 "deployment.succeeded" | "deployment.failed" => {
236 let kind = text(data, "kind").unwrap_or("preview");
237 shown["environment"] = json!(kind);
238 if let Some(project) = text(data, "project") {
239 shown["project"] = json!(project);
240 }
241 if let Some(branch) = text(data, "branch") {
242 shown["branch"] = json!(branch);
243 }
244 if let Some(commit) = text(data, "commit") {
245 shown["commit"] = json!(commit.chars().take(12).collect::<String>());
246 }
247 if let Some(error) = text(data, "error") {
248 shown["error"] = json!(error.chars().take(300).collect::<String>());
249 }
250 shown["url"] = json!(match text(data, "path") {
251 Some(path) => format!("{site}{path}"),
252 None => format!("{page}/deployments"),
253 });
254 if event.kind == "deployment.succeeded" && kind == "production" {
255 production += 1;
256 }
257 if event.kind == "deployment.failed" {
258 failed += 1;
259 }
260 }
261 "git.push" => {
262 // Branches only: a tag's push is its release.
263 let Some(branch) = text(data, "ref").and_then(|r| r.strip_prefix("refs/heads/")) else { continue };
264 let default_branch = data.get("defaultBranch").and_then(Value::as_bool) == Some(true);
265 shown["branch"] = json!(branch);
266 shown["default_branch"] = json!(default_branch);
267 if let Some(commits) = data.get("commits").and_then(Value::as_u64) {
268 shown["commits"] = json!(commits);
269 }
270 if let Some(after) = text(data, "after") {
271 shown["commit"] = json!(after.chars().take(12).collect::<String>());
272 shown["url"] = json!(format!("{page}/commit/{after}"));
273 }
274 if default_branch {
275 landed += 1;
276 }
277 }
278 "review.completed" | "comment.created" => {
279 if let Some(number) = number.or_else(|| data.get("issue").and_then(Value::as_u64).map(|n| n as u32)) {
280 shown["number"] = json!(number);
281 shown["url"] = json!(format!("{page}/issues/{number}"));
282 }
283 if let Some(verdict) = text(data, "verdict") {
284 shown["verdict"] = json!(verdict);
285 }
286 }
287 "workflow.completed" => {
288 let conclusion = text(data, "conclusion").unwrap_or_default();
289 if let Some(workflow) = text(data, "workflow") {
290 shown["workflow"] = json!(workflow);
291 }
292 if let Some(run_number) = data.get("number").and_then(Value::as_u64) {
293 shown["run_number"] = json!(run_number);
294 }
295 shown["conclusion"] = json!(conclusion);
296 if let Some(branch) = text(data, "ref").map(|r| r.strip_prefix("refs/heads/").unwrap_or(r)) {
297 shown["branch"] = json!(branch);
298 }
299 if let Some(sha) = text(data, "sha") {
300 shown["commit"] = json!(sha.chars().take(12).collect::<String>());
301 }
302 if let Some(pull) = data.get("pull").and_then(Value::as_u64) {
303 shown["pull"] = json!(pull);
304 }
305 if let Some(run) = text(data, "runId") {
306 shown["run_id"] = json!(run);
307 shown["url"] = json!(format!("{page}/actions/runs/{run}"));
308 }
309 if conclusion == "failure" {
310 runs_failed += 1;
311 }
312 // A repository that deploys itself: its deploy workflow's
313 // runs on the default branch are its production deployments.
314 let on_default = text(data, "ref") == Some(format!("refs/heads/{}", repo.default_branch).as_str());
315 if on_default
316 && text(data, "event") != Some("pull_request")
317 && deploy_workflow(text(data, "workflow").unwrap_or_default(), text(data, "path").unwrap_or_default())
318 {
319 shown["production"] = json!(true);
320 match conclusion {
321 "success" => production += 1,
322 "failure" | "timed_out" => failed += 1,
323 _ => {}
324 }
325 }
326 }
327 "package.published" => {
328 for key in ["ecosystem", "name", "version"] {
329 if let Some(value) = text(data, key) {
330 shown[key] = json!(value);
331 }
332 }
333 }
334 _ => continue,
335 }
336 events.push(shown);
337 }
338 json!({
339 "workspace": found.workspace,
340 "since": rfc3339(found.since),
341 "until": rfc3339(found.until),
342 "repositories": found.repos.len(),
343 "summary": {
344 "merged_pull_requests": merged,
345 "releases": releases,
346 "production_deployments": production,
347 "failed_deployments": failed,
348 "default_branch_pushes": landed,
349 "failed_runs": runs_failed,
350 // The landed commits by where they are in production (`run`
351 // fills it in).
352 "shipped": shipped(&[]),
353 },
354 // What landed on default branches in the span, one line a commit
355 // (`run` fills it in): what shipped where nothing was merged.
356 "landed": [],
357 "events": events,
358 "complete": !found.more && found.repos.len() <= MAX_LIST_REPOS,
359 })
360}
361
362/// Whether a workflow deploys: its name or file says so (Deploy,
363/// deploy.yml, deploy-production.yml), and it is not for previews.
364pub fn deploy_workflow(name: &str, path: &str) -> bool {
365 let file = path.rsplit('/').next().unwrap_or_default();
366 let named = format!("{name} {file}").to_lowercase();
367 named.contains("deploy") && !named.contains("preview")
368}
369
370/// Where a deploy of a commit to production is.
371#[derive(Clone, Copy, Debug, PartialEq, Eq)]
372pub enum DeployState {
373 Succeeded,
374 /// Queued, waiting or running.
375 Running,
376 Failed,
377}
378
379/// One deploy of a commit to production: a run of a repository's deploy
380/// workflow on its default branch, or a project's production deployment.
381#[derive(Clone, Debug, PartialEq)]
382pub struct Deploy {
383 pub sha: String,
384 pub state: DeployState,
385 /// What it was, for a person: "Deploy #311".
386 pub label: String,
387 pub url: String,
388}
389
390/// A run as a deploy to production, when it is one: its deploy workflow's,
391/// on the default branch and not for a pull request. Cancelled and skipped
392/// runs deployed nothing.
393pub fn run_deploy(site: &str, repo: &str, default_branch: &str, run: &WorkflowRun) -> Option<Deploy> {
394 if run.git_ref != format!("refs/heads/{default_branch}") || run.event == "pull_request" || !deploy_workflow(&run.name, &run.path) {
395 return None;
396 }
397 let state = match (run.status.as_str(), run.conclusion.as_deref()) {
398 ("completed", Some("success")) => DeployState::Succeeded,
399 ("completed", Some("failure" | "timed_out")) => DeployState::Failed,
400 ("completed", _) => return None,
401 _ => DeployState::Running,
402 };
403 Some(Deploy {
404 sha: run.sha.clone(),
405 state,
406 label: format!("{} #{}", run.name, run.number),
407 url: format!("{}/{repo}/actions/runs/{}", site.trim_end_matches('/'), run.id),
408 })
409}
410
411/// A repository's deploys to production that the events name: projects'
412/// production deployments, and its deploy workflow's finished runs.
413pub fn event_deploys(site: &str, repo: &Repo, events: &[Event]) -> Vec<Deploy> {
414 let site = site.trim_end_matches('/');
415 let full = format!("{}/{}", repo.namespace, repo.name);
416 let mut found = Vec::new();
417 for event in events.iter().filter(|event| event.repo_id.as_deref() == Some(repo.id.as_str())) {
418 let data = &event.data;
419 match event.kind.as_str() {
420 "deployment.succeeded" | "deployment.failed" if text(data, "kind") == Some("production") => {
421 let Some(sha) = text(data, "commit") else { continue };
422 found.push(Deploy {
423 sha: sha.to_owned(),
424 state: if event.kind == "deployment.succeeded" { DeployState::Succeeded } else { DeployState::Failed },
425 label: format!("{}'s production deployment", text(data, "project").unwrap_or(&repo.name)),
426 url: match text(data, "path") {
427 Some(path) => format!("{site}{path}"),
428 None => format!("{site}/{full}/deployments"),
429 },
430 });
431 }
432 "workflow.completed" => {
433 let run = WorkflowRun {
434 id: text(data, "runId").unwrap_or_default().to_owned(),
435 path: text(data, "path").unwrap_or_default().to_owned(),
436 name: text(data, "workflow").unwrap_or_default().to_owned(),
437 number: data.get("number").and_then(Value::as_u64).unwrap_or_default(),
438 event: text(data, "event").unwrap_or_default().to_owned(),
439 git_ref: text(data, "ref").unwrap_or_default().to_owned(),
440 sha: text(data, "sha").unwrap_or_default().to_owned(),
441 status: "completed".to_owned(),
442 conclusion: text(data, "conclusion").map(str::to_owned),
443 ..WorkflowRun::default()
444 };
445 if run.id.is_empty() || run.sha.is_empty() {
446 continue;
447 }
448 found.extend(run_deploy(site, &full, &repo.default_branch, &run));
449 }
450 _ => {}
451 }
452 }
453 found
454}
455
456/// Where each commit of a default branch's history (newest first) is in
457/// production, given the deploys seen: `live` when a successful deploy was
458/// of it or of a commit after it (which carries it), else `deploying` when
459/// one is under way, `failed` when one failed, or `not_deployed`; with the
460/// deploy that says so. None for every commit when the repository was not
461/// seen deploying at all.
462pub fn production<'a>(log: &[Commit], deploys: &'a [Deploy]) -> Vec<Option<(&'static str, Option<&'a Deploy>)>> {
463 if deploys.is_empty() {
464 return vec![None; log.len()];
465 }
466 let same = |sha: &str, hash: &str| !sha.is_empty() && (hash.starts_with(sha) || sha.starts_with(hash));
467 // Each deploy at its commit's place in the history.
468 let placed: Vec<(usize, &Deploy)> = deploys
469 .iter()
470 .filter_map(|deploy| log.iter().position(|commit| same(&deploy.sha, &commit.hash)).map(|at| (at, deploy)))
471 .collect();
472 // The nearest deploy at or after a commit in the given state: the one
473 // that first carried it.
474 let nearest = |index: usize, state: DeployState| {
475 placed.iter().filter(|(at, deploy)| *at <= index && deploy.state == state).max_by_key(|(at, _)| *at).map(|(_, deploy)| *deploy)
476 };
477 (0..log.len())
478 .map(|index| {
479 Some(if let Some(deploy) = nearest(index, DeployState::Succeeded) {
480 ("live", Some(deploy))
481 } else if let Some(deploy) = nearest(index, DeployState::Running) {
482 ("deploying", Some(deploy))
483 } else if let Some(deploy) = nearest(index, DeployState::Failed) {
484 ("failed", Some(deploy))
485 } else {
486 ("not_deployed", None)
487 })
488 })
489 .collect()
490}
491
492/// One repository's default-branch commits made in the span, newest first,
493/// each as one line: its short sha, the first line of its message, who
494/// wrote it (and their username, when known), when, its page, and where it
495/// is in production when the repository deploys.
496pub fn landed(site: &str, repo: &str, commits: &[Commit], since: u64, until: u64, authors: &Authors, deploys: &[Deploy]) -> Vec<Value> {
497 let site = site.trim_end_matches('/');
498 let states = production(commits, deploys);
499 commits
500 .iter()
501 .zip(states)
502 .filter(|(commit, _)| parse_rfc3339(&commit.authored_at).is_some_and(|at| at >= since && at < until))
503 .map(|(commit, state)| {
504 let subject: String = commit.message.lines().next().unwrap_or_default().trim().chars().take(160).collect();
505 let mut line = json!({
506 "repo": repo,
507 "commit": commit.hash.chars().take(12).collect::<String>(),
508 "subject": subject,
509 "author": commit.author.name,
510 "time": commit.authored_at,
511 "url": format!("{site}/{repo}/commit/{}", commit.hash),
512 });
513 if let Some(username) = commits::username(commit, authors) {
514 line["username"] = json!(username);
515 }
516 if let Some((state, deploy)) = state {
517 line["production"] = json!(state);
518 if let Some(deploy) = deploy {
519 line["deploy"] = json!({ "name": deploy.label, "url": deploy.url });
520 }
521 }
522 line
523 })
524 .collect()
525}
526
527/// How many landed commits are where, for the summary.
528pub fn shipped(lines: &[Value]) -> Value {
529 let count = |state: &str| lines.iter().filter(|line| line["production"] == state).count();
530 json!({
531 "live": count("live"),
532 "deploying": count("deploying"),
533 "failed": count("failed"),
534 "not_deployed": count("not_deployed"),
535 })
536}
537
538pub async fn run(services: &Services, viewer: &Viewer, input: &Value) -> Result<Outcome<Value>> {
539 let workspace = input["workspace"].as_str().unwrap_or_default().trim().to_lowercase();
540 if workspace.is_empty() {
541 return Ok(Outcome::fail(FailureCode::Invalid, "Give the workspace's slug."));
542 }
543 let (since, until) = match span(input, g1t_kit::now_ms()) {
544 Ok(span) => span,
545 Err(problem) => return Ok(Outcome::fail(FailureCode::Invalid, problem)),
546 };
547 let types = match event_types(input) {
548 Ok(types) => types,
549 Err(problem) => return Ok(Outcome::fail(FailureCode::Invalid, problem)),
550 };
551 let limit = limit(input);
552 // The workspace's repositories as the caller may read them; a
553 // workspace they cannot see has none.
554 let repos: Vec<Repo> = g1t_kit::call(
555 &services.repos,
556 "list",
557 &ListReposArgs { viewer: viewer.clone(), query: None, namespace: Some(workspace.clone()), member_only: false },
558 )
559 .await?;
560 let repos: Vec<Repo> = repos.into_iter().filter(|repo| repo.namespace.eq_ignore_ascii_case(&workspace)).collect();
561 let site = services.addresses.site.clone();
562 let none = (HashMap::new(), HashMap::new());
563 if repos.is_empty() {
564 return Ok(Outcome::Ok(shape(&Found {
565 workspace: &workspace,
566 site: &site,
567 since,
568 until,
569 repos: &repos,
570 events: &[],
571 more: false,
572 pull_titles: &none.0,
573 usernames: &none.1,
574 })));
575 }
576 let events: Vec<Event> = g1t_kit::call(
577 &services.events,
578 "list",
579 &ListEventsArgs {
580 repo_ids: repos.iter().take(MAX_LIST_REPOS).map(|repo| repo.id.clone()).collect(),
581 types: types.iter().map(|kind| (*kind).to_owned()).collect(),
582 since: Some(rfc3339(since)),
583 // One past the limit says whether there was more.
584 limit: Some(limit + 1),
585 ..ListEventsArgs::default()
586 },
587 )
588 .await?;
589 let more = events.len() > limit as usize;
590 let events: Vec<Event> = events.into_iter().take(limit as usize).collect();
591 // Titles of the pull requests named, and the names of who did what.
592 let pull_repos: Vec<String> = events
593 .iter()
594 .filter(|event| event.kind.starts_with("pull."))
595 .filter_map(|event| event.repo_id.clone())
596 .collect::<BTreeSet<_>>()
597 .into_iter()
598 .take(MAX_PULLS_FOR_REPOS)
599 .collect();
600 let actors: Vec<String> = events.iter().filter_map(|event| event.actor.clone()).collect::<BTreeSet<_>>().into_iter().collect();
601 let pulls = async {
602 if pull_repos.is_empty() {
603 return Ok(Vec::new());
604 }
605 g1t_kit::call::<_, Vec<RepoPulls>>(
606 &services.work,
607 "pulls_for_repos",
608 &PullsForReposArgs { repo_ids: pull_repos, viewer: viewer.clone(), limit: 100 },
609 )
610 .await
611 };
612 let names = async {
613 if actors.is_empty() {
614 return Ok(HashMap::new());
615 }
616 g1t_kit::call::<_, HashMap<String, String>>(&services.identity, "usernames", &UsernamesArgs { ids: actors }).await
617 };
618 let (pulls, names) = futures_util::future::join(pulls, names).await;
619 // Without titles or names the events still say what happened.
620 let mut pull_titles = HashMap::new();
621 for repo in pulls.unwrap_or_default() {
622 for pull in repo.open.into_iter().chain(repo.closed) {
623 pull_titles.insert((repo.repo_id.clone(), pull.number), pull.title);
624 }
625 }
626 let usernames = names.unwrap_or_default();
627 let mut answer = shape(&Found {
628 workspace: &workspace,
629 site: &site,
630 since,
631 until,
632 repos: &repos,
633 events: &events,
634 more,
635 pull_titles: &pull_titles,
636 usernames: &usernames,
637 });
638 // The repositories whose default branch was pushed to, busiest first:
639 // their commits say what landed.
640 let mut pushed: Vec<(String, usize)> = Vec::new();
641 for event in events.iter().filter(|event| event.kind == "git.push" && event.data.get("defaultBranch").and_then(Value::as_bool) == Some(true)) {
642 let Some(id) = event.repo_id.as_deref() else { continue };
643 match pushed.iter_mut().find(|(known, _)| known == id) {
644 Some((_, count)) => *count += 1,
645 None => pushed.push((id.to_owned(), 1)),
646 }
647 }
648 pushed.sort_by(|a, b| b.1.cmp(&a.1));
649 let by_id: HashMap<&str, &Repo> = repos.iter().map(|repo| (repo.id.as_str(), repo)).collect();
650 // Each one's history, and its default branch's runs in the span: the
651 // deploys among them, finished or not, say what reached production.
652 let logs = pushed.iter().take(LANDED_REPOS).filter_map(|(id, _)| by_id.get(id.as_str()).copied()).map(|repo| async move {
653 let path = RepoPath { namespace: repo.namespace.clone(), name: repo.name.clone() };
654 let log_args = LogArgs { path: path.clone(), viewer: viewer.clone(), git_ref: None, limit: LANDED_LOG };
655 let runs_args = RunsArgs {
656 repo: path,
657 viewer: viewer.clone(),
658 workflow: None,
659 branch: Some(repo.default_branch.clone()),
660 event: None,
661 pull: None,
662 sha: None,
663 status: None,
664 since: Some(rfc3339(since)),
665 until: None,
666 limit: Some(LANDED_RUNS),
667 };
668 let (log, runs) = futures_util::future::join(
669 g1t_kit::call::<_, Outcome<Vec<Commit>>>(&services.repos, "log", &log_args),
670 g1t_kit::call::<_, Outcome<Vec<WorkflowRun>>>(&services.actions, "runs", &runs_args),
671 )
672 .await;
673 (repo, log, runs)
674 });
675 let mut histories = Vec::new();
676 for (repo, log, runs) in futures_util::future::join_all(logs).await {
677 // A history that cannot be read leaves the events to say it.
678 let Ok(Outcome::Ok(commits)) = log else { continue };
679 let full = format!("{}/{}", repo.namespace, repo.name);
680 let mut deploys = event_deploys(&site, repo, &events);
681 if let Ok(Outcome::Ok(runs)) = runs {
682 for deploy in runs.iter().filter_map(|run| run_deploy(&site, &full, &repo.default_branch, run)) {
683 // A run the events named too is the same deploy; the run says
684 // where it is now.
685 deploys.retain(|known| known.url != deploy.url);
686 deploys.push(deploy);
687 }
688 }
689 histories.push((full, commits, deploys));
690 }
691 let all: Vec<&Commit> = histories.iter().flat_map(|(_, commits, _)| commits.iter()).collect();
692 let authors = commits::read_authors(services, &all).await;
693 let mut lines = Vec::new();
694 for (repo, commits, deploys) in &histories {
695 lines.extend(landed(&site, repo, commits, since, until, &authors, deploys));
696 }
697 lines.truncate(LANDED_MAX);
698 answer["summary"]["shipped"] = shipped(&lines);
699 answer["landed"] = json!(lines);
700 Ok(Outcome::Ok(answer))
701}
702
703#[cfg(test)]
704mod tests {
705 use super::*;
706
707 fn repo(id: &str, name: &str) -> Repo {
708 serde_json::from_value(json!({
709 "id": id, "namespace": "acme", "name": name, "description": null, "isPrivate": false,
710 "ownerId": "wsp_1", "defaultBranch": "main", "forkOf": null, "createdAt": "2026-01-01T00:00:00Z",
711 }))
712 .unwrap()
713 }
714
715 fn event(kind: &str, repo: &str, time: &str, actor: &str, data: Value) -> Event {
716 Event {
717 id: format!("evt_{time}"),
718 kind: kind.into(),
719 source: "test".into(),
720 time: time.into(),
721 repo_id: Some(repo.into()),
722 actor: Some(actor.into()),
723 data,
724 }
725 }
726
727 #[test]
728 fn the_span_is_the_last_day_unless_given() {
729 let now = parse_rfc3339("2026-10-12T18:00:00Z").unwrap();
730 let (since, until) = span(&json!({}), now).unwrap();
731 assert_eq!(until, now);
732 assert_eq!(until - since, DEFAULT_SPAN_MS);
733 let (since, _) = span(&json!({ "since": "2026-10-12T00:00:00Z" }), now).unwrap();
734 assert_eq!(since, parse_rfc3339("2026-10-12T00:00:00Z").unwrap());
735 assert!(span(&json!({ "since": "yesterday" }), now).unwrap_err().contains("RFC 3339"));
736 assert!(span(&json!({ "since": "2026-10-13T00:00:00Z" }), now).is_err());
737 assert!(span(&json!({ "since": "2026-08-01T00:00:00Z" }), now).unwrap_err().contains("31 days"));
738 }
739
740 #[test]
741 fn kinds_name_their_events() {
742 assert_eq!(event_types(&json!({ "kinds": ["merged"] })).unwrap(), vec!["pull.merged"]);
743 let all = event_types(&json!({})).unwrap();
744 assert!(all.contains(&"pull.merged") && all.contains(&"git.push") && all.contains(&"release.published"));
745 assert!(!all.contains(&"comment.created"));
746 assert!(event_types(&json!({ "kinds": ["shipped"] })).unwrap_err().contains("Kinds: merged"));
747 // Over REST: one comma-separated text.
748 assert_eq!(event_types(&json!({ "kinds": "merged, releases" })).unwrap(), vec!["pull.merged", "release.published"]);
749 assert_eq!(limit(&json!({ "limit": "500" })), MAX_LIMIT);
750 assert_eq!(limit(&json!({ "limit": 5 })), 5);
751 assert_eq!(limit(&json!({})), DEFAULT_LIMIT);
752 }
753
754 #[test]
755 fn what_landed_is_one_line_a_commit_made_in_the_span() {
756 use g1t_contracts::repos::Signature;
757 let commit = |hash: &str, message: &str, at: &str| Commit {
758 hash: hash.to_owned(),
759 tree_hash: String::new(),
760 message: message.to_owned(),
761 author: Signature { name: "Chase".into(), email: String::new() },
762 parents: Vec::new(),
763 authored_at: at.to_owned(),
764 };
765 let commits = [
766 commit("aaaaaaaaaaaaaaaa", "Ask reads g1t
767
768The long why.", "2026-10-13T05:00:00Z"),
769 commit("bbbbbbbbbbbbbbbb", "Yesterday's change", "2026-10-12T05:00:00Z"),
770 ];
771 let since = parse_rfc3339("2026-10-12T06:00:00Z").unwrap();
772 let until = parse_rfc3339("2026-10-13T06:00:00Z").unwrap();
773 let lines = landed("https://g1t.sh/", "acme/web", &commits, since, until, &Authors::new(), &[]);
774 assert_eq!(lines.len(), 1);
775 assert_eq!(lines[0]["subject"], "Ask reads g1t");
776 assert_eq!(lines[0]["commit"], "aaaaaaaaaaaa");
777 assert_eq!(lines[0]["author"], "Chase");
778 assert_eq!(lines[0]["url"], "https://g1t.sh/acme/web/commit/aaaaaaaaaaaaaaaa");
779 // Nobody known by the address, and no deploys seen: no username made
780 // up from the name, and nothing said about production.
781 assert!(lines[0].get("username").is_none());
782 assert!(lines[0].get("production").is_none());
783 }
784
785 fn by(hash: &str, email: &str, at: &str) -> Commit {
786 Commit {
787 hash: hash.to_owned(),
788 tree_hash: String::new(),
789 message: format!("Change {hash}"),
790 author: g1t_contracts::repos::Signature { name: "Chase Pierce".into(), email: email.into() },
791 parents: Vec::new(),
792 authored_at: at.to_owned(),
793 }
794 }
795
796 fn run(id: &str, name: &str, number: u64, sha: &str, status: &str, conclusion: Option<&str>) -> WorkflowRun {
797 WorkflowRun {
798 id: id.into(),
799 path: ".g1t/workflows/deploy.yml".into(),
800 name: name.into(),
801 number,
802 event: "push".into(),
803 git_ref: "refs/heads/main".into(),
804 sha: sha.into(),
805 status: status.into(),
806 conclusion: conclusion.map(str::to_owned),
807 ..WorkflowRun::default()
808 }
809 }
810
811 #[test]
812 fn what_shipped_is_what_reached_the_default_branch_and_which_of_it_is_live() {
813 // Newest first: e was pushed last and is still deploying, which
814 // carries d, whose own deploy failed; c went out with Deploy #311,
815 // which carried b and a too.
816 let log = [
817 by("eeee0000", "syntaqx@gmail.com", "2026-10-14T13:00:00Z"),
818 by("dddd0000", "syntaqx@gmail.com", "2026-10-14T11:00:00Z"),
819 by("cccc0000", "syntaqx@gmail.com", "2026-10-14T09:00:00Z"),
820 by("bbbb0000", "someone@example.com", "2026-10-14T08:00:00Z"),
821 by("aaaa0000", "syntaqx@gmail.com", "2026-10-13T08:00:00Z"),
822 ];
823 let site = "https://g1t.sh";
824 let runs = [
825 run("run_313", "Deploy", 313, "eeee0000", "in_progress", None),
826 run("run_312", "Deploy", 312, "dddd0000", "completed", Some("failure")),
827 run("run_311", "Deploy", 311, "cccc0000", "completed", Some("success")),
828 // CI is not a deploy; a cancelled deploy deployed nothing; a
829 // deploy of another branch is not production.
830 WorkflowRun { path: ".g1t/workflows/ci.yml".into(), ..run("run_9", "CI", 9, "eeee0000", "completed", Some("success")) },
831 run("run_310", "Deploy", 310, "dddd0000", "completed", Some("cancelled")),
832 WorkflowRun { git_ref: "refs/heads/next".into(), ..run("run_8", "Deploy", 8, "eeee0000", "completed", Some("success")) },
833 ];
834 let deploys: Vec<Deploy> = runs.iter().filter_map(|found| run_deploy(site, "flagon-io/g1t", "main", found)).collect();
835 assert_eq!(deploys.len(), 3);
836 assert_eq!(deploys[2], Deploy {
837 sha: "cccc0000".into(),
838 state: DeployState::Succeeded,
839 label: "Deploy #311".into(),
840 url: "https://g1t.sh/flagon-io/g1t/actions/runs/run_311".into(),
841 });
842 let owners = HashMap::from([(
843 "syntaqx@gmail.com".to_owned(),
844 g1t_contracts::accounts::EmailOwner { id: "usr_1".into(), username: "syntaqx".into(), avatar: None },
845 )]);
846 let authors = commits::authors(&log.iter().collect::<Vec<_>>(), &owners);
847 let since = parse_rfc3339("2026-10-14T06:00:00Z").unwrap();
848 let until = parse_rfc3339("2026-10-14T20:00:00Z").unwrap();
849 let lines = landed(site, "flagon-io/g1t", &log, since, until, &authors, &deploys);
850 let states: Vec<&str> = lines.iter().map(|line| line["production"].as_str().unwrap()).collect();
851 assert_eq!(states, ["deploying", "deploying", "live", "live"], "a is yesterday's");
852 assert_eq!(lines[2]["deploy"], json!({ "name": "Deploy #311", "url": "https://g1t.sh/flagon-io/g1t/actions/runs/run_311" }));
853 assert_eq!(lines[1]["deploy"]["name"], "Deploy #313");
854 // The author by their username; someone unknown by their name alone.
855 assert_eq!(lines[0]["username"], "syntaqx");
856 assert_eq!(lines[0]["author"], "Chase Pierce");
857 assert!(lines[3].get("username").is_none());
858 assert_eq!(shipped(&lines), json!({ "live": 2, "deploying": 2, "failed": 0, "not_deployed": 0 }));
859 // Before #313 started: d failed to deploy, and e is not deployed.
860 let lines = landed(site, "flagon-io/g1t", &log, since, until, &authors, &deploys[1..]);
861 assert_eq!(shipped(&lines), json!({ "live": 2, "deploying": 0, "failed": 1, "not_deployed": 1 }));
862 assert_eq!(lines[1]["deploy"]["name"], "Deploy #312");
863 assert!(lines[0].get("deploy").is_none());
864 // Once #313 succeeds, everything is live; before any deploy of today's
865 // commits, they are not deployed.
866 let done = [Deploy { state: DeployState::Succeeded, ..deploys[0].clone() }];
867 assert!(production(&log, &done).iter().all(|state| state.map(|(state, _)| state) == Some("live")));
868 let old = [Deploy { sha: "ffff9999".into(), ..deploys[2].clone() }];
869 assert!(production(&log, &old).iter().all(|state| state.map(|(state, _)| state) == Some("not_deployed")));
870 assert!(production(&log, &[]).iter().all(Option::is_none));
871 }
872
873 #[test]
874 fn deploy_workflows_are_known_by_name_or_file() {
875 assert!(deploy_workflow("Deploy", ".g1t/workflows/deploy.yml"));
876 assert!(deploy_workflow("Ship it", ".g1t/workflows/deploy-production.yml"));
877 assert!(deploy_workflow("Deploy site", ".g1t/workflows/site.yml"));
878 assert!(!deploy_workflow("CI", ".g1t/workflows/ci.yml"));
879 assert!(!deploy_workflow("Deploy preview", ".g1t/workflows/preview.yml"));
880 }
881
882 #[test]
883 fn what_shipped_reads_as_one_list_with_links_and_names() {
884 let repos = [repo("rep_1", "web"), repo("rep_2", "api")];
885 let events = [
886 event("deployment.succeeded", "rep_1", "2026-10-12T15:00:00Z", "usr_1", json!({ "kind": "production", "project": "web", "commit": "abcdef1234567890", "path": "/acme/web/deployments/dpl_1", "triggeredBy": "ana" })),
887 event("release.published", "rep_2", "2026-10-12T14:00:00Z", "usr_2", json!({ "tagName": "v1.2.0", "release": { "name": "Faster pushes" } })),
888 event("pull.merged", "rep_1", "2026-10-12T13:00:00Z", "usr_1", json!({ "number": 7, "issue": 3 })),
889 event("git.push", "rep_2", "2026-10-12T12:00:00Z", "usr_2", json!({ "ref": "refs/heads/main", "after": "0123456789abcdef", "defaultBranch": true, "commits": 2 })),
890 event("git.push", "rep_2", "2026-10-12T11:00:00Z", "usr_2", json!({ "ref": "refs/tags/v1.2.0", "after": "0123" })),
891 event("workflow.completed", "rep_1", "2026-10-12T11:30:00Z", "usr_1", json!({ "runId": "run_257", "workflow": "Deploy", "number": 257, "conclusion": "failure", "ref": "refs/heads/main", "sha": "fedcba9876543210" })),
892 event("issue.closed", "rep_1", "2026-10-12T10:00:00Z", "usr_1", json!({ "number": 3, "title": "Login fails", "reason": "completed", "resolvedBy": 7 })),
893 // Outside the span, and a repository the caller cannot read.
894 event("pull.merged", "rep_1", "2026-10-11T10:00:00Z", "usr_1", json!({ "number": 6 })),
895 event("pull.merged", "rep_9", "2026-10-12T10:00:00Z", "usr_1", json!({ "number": 1 })),
896 ];
897 let titles = HashMap::from([(("rep_1".to_owned(), 7), "Fix login".to_owned())]);
898 let names = HashMap::from([("usr_1".to_owned(), "ana".to_owned()), ("usr_2".to_owned(), "bo".to_owned())]);
899 let shown = shape(&Found {
900 workspace: "acme",
901 site: "https://g1t.sh/",
902 since: parse_rfc3339("2026-10-12T00:00:00Z").unwrap(),
903 until: parse_rfc3339("2026-10-12T18:00:00Z").unwrap(),
904 repos: &repos,
905 events: &events,
906 more: false,
907 pull_titles: &titles,
908 usernames: &names,
909 });
910 assert_eq!(
911 shown["summary"],
912 json!({
913 "merged_pull_requests": 1, "releases": 1, "production_deployments": 1, "failed_deployments": 1, "default_branch_pushes": 1, "failed_runs": 1,
914 "shipped": { "live": 0, "deploying": 0, "failed": 0, "not_deployed": 0 },
915 })
916 );
917 let listed = shown["events"].as_array().unwrap();
918 assert_eq!(listed.len(), 6, "{listed:#?}");
919 assert_eq!(listed[0]["url"], "https://g1t.sh/acme/web/deployments/dpl_1");
920 assert_eq!(listed[0]["commit"], "abcdef123456");
921 assert_eq!(listed[1]["title"], "Faster pushes");
922 assert_eq!(listed[1]["url"], "https://g1t.sh/acme/api/releases/tag/v1.2.0");
923 assert_eq!(
924 listed[2],
925 json!({ "kind": "pull_merged", "time": "2026-10-12T13:00:00Z", "repo": "acme/web", "actor": "ana", "number": 7, "title": "Fix login", "url": "https://g1t.sh/acme/web/pull/7", "issue": 3 })
926 );
927 assert_eq!(listed[3]["default_branch"], true);
928 assert_eq!(listed[3]["actor"], "bo");
929 assert_eq!(
930 listed[4],
931 json!({ "kind": "workflow_completed", "time": "2026-10-12T11:30:00Z", "repo": "acme/web", "actor": "ana", "workflow": "Deploy", "run_number": 257, "conclusion": "failure", "branch": "main", "commit": "fedcba987654", "run_id": "run_257", "url": "https://g1t.sh/acme/web/actions/runs/run_257", "production": true })
932 );
933 assert_eq!(listed[5]["resolved_by"], 7);
934 assert_eq!(shown["complete"], true);
935 }
936}