Skip to content
Sign up

flagon-io/g1t

public
0
621 linesCodeBlameRaw
1//! Reading: workflows, runs (any attempt), a run's jobs, a job's log and
2//! its steps' summaries.
3
4use g1t_actions::workflow::{self, Severity};
5use g1t_contracts::access::Capability;
6use g1t_contracts::checks::{ActionsChecksArgs, MAX_COMMITS};
7use g1t_contracts::actions::{
8 Annotation, Job, JobLog, JobLogText, JobLogTextArgs, JobSummary, LogChunk, LogsArgs, MAX_RUN_LOG_BYTES, RunArgs, RunAttempt, RunDetail,
9 RunLogsArgs, RunsArgs, SetWorkflowEnabledArgs, StepState, StepSummary, SummariesArgs, Workflow, WorkflowNote, WorkflowRun, WorkflowsArgs,
10};
11use g1t_contracts::{FailureCode, Outcome};
12use serde::Deserialize;
13use worker::Result;
14
15use crate::plan::{JobRow, RunRow};
16use crate::sync::WorkflowRow;
17use crate::{Actions, check, fail};
18
19const RUNS_SHOWN: u32 = 50;
20
21fn notes(source: &str) -> Vec<WorkflowNote> {
22 workflow::parse(source)
23 .map(|w| {
24 w.notes
25 .into_iter()
26 .map(|note| WorkflowNote {
27 severity: match note.severity {
28 Severity::Info => "info",
29 Severity::Warning => "warning",
30 Severity::Unsupported => "unsupported",
31 }
32 .to_owned(),
33 job: note.job,
34 message: note.message,
35 })
36 .collect()
37 })
38 .unwrap_or_default()
39}
40
41/// A job of an earlier attempt, as it ended (migration 0009).
42#[derive(Clone, Deserialize)]
43pub struct AttemptJobRow {
44 pub id: String,
45 pub run_id: String,
46 pub log_id: String,
47 pub key: String,
48 pub name: String,
49 pub needs: String,
50 pub status: String,
51 pub conclusion: Option<String>,
52 pub steps: String,
53 pub annotations: String,
54 pub reason: Option<String>,
55 pub environment: Option<String>,
56 pub labels: Option<String>,
57 pub runner_name: Option<String>,
58 pub started_at: Option<String>,
59 pub finished_at: Option<String>,
60}
61
62fn attempt_job_view(row: AttemptJobRow) -> Job {
63 Job {
64 needs: serde_json::from_str(&row.needs).unwrap_or_default(),
65 id: row.id,
66 run_id: row.run_id,
67 key: row.key,
68 name: row.name,
69 status: row.status,
70 conclusion: row.conclusion,
71 steps: serde_json::from_str::<Vec<StepState>>(&row.steps).unwrap_or_default(),
72 annotations: serde_json::from_str::<Vec<Annotation>>(&row.annotations).unwrap_or_default(),
73 reason: row.reason,
74 started_at: row.started_at,
75 finished_at: row.finished_at,
76 environment: row.environment,
77 self_hosted: row.labels.is_some(),
78 runner: row.runner_name,
79 cancelling: false,
80 environment_url: None,
81 uses: None,
82 }
83}
84
85/// An earlier attempt of a run, as it ended.
86#[derive(Clone, Deserialize)]
87struct AttemptRow {
88 attempt: u64,
89 conclusion: Option<String>,
90 actor: Option<String>,
91 debug: u32,
92 started_at: Option<String>,
93 finished_at: Option<String>,
94}
95
96/// One attempt's jobs, each with where its logs and summary are kept.
97struct AttemptJobs {
98 jobs: Vec<Job>,
99 log_ids: Vec<String>,
100}
101
102fn job_view(row: JobRow) -> Job {
103 let needs = row.needs();
104 let cancelling = row.cancel_requested_at.is_some() && row.status != "completed";
105 let uses = row.call().filter(|call| call["role"] == "caller").and_then(|call| call["path"].as_str().map(str::to_owned));
106 Job {
107 uses,
108 environment_url: None,
109 cancelling,
110 id: row.id,
111 run_id: row.run_id,
112 key: row.key,
113 name: row.name,
114 needs,
115 status: row.status,
116 conclusion: row.conclusion,
117 steps: serde_json::from_str::<Vec<StepState>>(&row.steps).unwrap_or_default(),
118 annotations: serde_json::from_str::<Vec<Annotation>>(&row.annotations).unwrap_or_default(),
119 reason: row.reason,
120 started_at: row.started_at,
121 finished_at: row.finished_at,
122 environment: row.environment,
123 self_hosted: row.labels.is_some(),
124 runner: row.runner_name,
125 }
126}
127
128impl Actions {
129 async fn summary(&self, row: &WorkflowRow) -> Result<Workflow> {
130 let last_run = self
131 .db
132 .prepare("SELECT * FROM runs WHERE workflow_id = ? ORDER BY id DESC LIMIT 1")
133 .bind(&[row.id.as_str().into()])?
134 .first::<RunRow>(None)
135 .await?
136 .map(|run| run.summary());
137 let parsed = workflow::parse(&row.source).ok();
138 let managed_by = if g1t_actions::managed::is_managed(&row.path) {
139 let workspace = row.repo.split_once('/').map_or(row.repo.as_str(), |(workspace, _)| workspace);
140 self.managed_by(workspace, &row.path, &row.source).await?
141 } else {
142 None
143 };
144 Ok(Workflow {
145 id: row.id.clone(),
146 path: row.path.clone(),
147 name: row.name.clone(),
148 events: serde_json::from_str(&row.events).unwrap_or_default(),
149 state: row.state.clone(),
150 error: row.error.clone(),
151 notes: notes(&row.source),
152 dispatch: parsed
153 .as_ref()
154 .filter(|_| row.error.is_none())
155 .and_then(|w| w.trigger("workflow_dispatch"))
156 .map(|t| serde_json::Value::Object(t.inputs.clone())),
157 last_run,
158 managed_by,
159 })
160 }
161
162 pub async fn workflows(&self, a: WorkflowsArgs) -> Result<Outcome<Vec<Workflow>>> {
163 let Some(repo) = self.visible_repo(&a.repo, &a.viewer).await? else {
164 return Ok(fail(FailureCode::NotFound, "There is no such repository."));
165 };
166 if !self.synced(&repo.id).await?
167 && let Some(ws) = self.workspace_actor(&repo.namespace).await?
168 {
169 self.sync(&repo, &ws).await?;
170 }
171 // A file that is gone stays listed while it has runs, unless one in
172 // the folder now has its name: then it would read as a second copy.
173 // Files outside the folder are from before g1t stopped reading
174 // `.github`; their runs stay under All workflows. Managed
175 // workflows (managed.rs) are listed with the folder's.
176 let rows = self
177 .db
178 .prepare(
179 "SELECT * FROM workflows w WHERE w.repo_id = ?1 AND (w.path LIKE ?2 OR w.path LIKE 'managed/%')
180 AND (w.error IS NULL OR w.error NOT LIKE 'Its file is%'
181 OR (w.id IN (SELECT workflow_id FROM runs WHERE repo_id = ?1)
182 AND NOT EXISTS (SELECT 1 FROM workflows o WHERE o.repo_id = ?1 AND o.id <> w.id
183 AND o.name = w.name AND (o.error IS NULL OR o.error NOT LIKE 'Its file is%'))))
184 ORDER BY w.name",
185 )
186 .bind(&[repo.id.as_str().into(), format!("{}/%", g1t_actions::workflow::FOLDER).into()])?
187 .all()
188 .await?
189 .results::<WorkflowRow>()?;
190 let mut out = Vec::with_capacity(rows.len());
191 for row in &rows {
192 out.push(self.summary(row).await?);
193 }
194 Ok(Outcome::Ok(out))
195 }
196
197 pub async fn runs(&self, a: RunsArgs) -> Result<Outcome<Vec<WorkflowRun>>> {
198 let Some(repo) = self.visible_repo(&a.repo, &a.viewer).await? else {
199 return Ok(fail(FailureCode::NotFound, "There is no such repository."));
200 };
201 let mut sql = "SELECT * FROM runs WHERE repo_id = ?".to_owned();
202 let mut binds: Vec<worker::wasm_bindgen::JsValue> = vec![repo.id.as_str().into()];
203 if let Some(workflow) = &a.workflow {
204 sql.push_str(" AND (workflow_id = ? OR path = ? OR path = ?)");
205 binds.push(workflow.as_str().into());
206 binds.push(workflow.as_str().into());
207 binds.push(format!("{}/{workflow}", g1t_actions::workflow::FOLDER).into());
208 }
209 if let Some(branch) = &a.branch {
210 sql.push_str(" AND (git_ref = ? OR json_extract(info, '$.headRef') = ?)");
211 binds.push(format!("refs/heads/{branch}").into());
212 binds.push(branch.as_str().into());
213 }
214 if let Some(event) = &a.event {
215 sql.push_str(" AND event = ?");
216 binds.push(event.as_str().into());
217 }
218 if let Some(pull) = a.pull {
219 sql.push_str(" AND pull = ?");
220 binds.push(pull.into());
221 }
222 if let Some(sha) = &a.sha {
223 sql.push_str(" AND sha = ?");
224 binds.push(sha.as_str().into());
225 }
226 sql.push_str(" ORDER BY id DESC LIMIT ?");
227 binds.push(a.limit.unwrap_or(RUNS_SHOWN).clamp(1, 100).into());
228 let rows = self.db.prepare(sql).bind(&binds)?.all().await?.results::<RunRow>()?;
229 Ok(Outcome::Ok(rows.iter().map(RunRow::summary).collect()))
230 }
231
232 /// Every attempt of a run, oldest first, the current one last.
233 async fn attempts_of(&self, run: &RunRow) -> Result<Vec<RunAttempt>> {
234 let mut attempts: Vec<RunAttempt> = self
235 .db
236 .prepare("SELECT attempt, conclusion, actor, debug, started_at, finished_at FROM run_attempts WHERE run_id = ? ORDER BY attempt")
237 .bind(&[run.id.as_str().into()])?
238 .all()
239 .await?
240 .results::<AttemptRow>()?
241 .into_iter()
242 .filter(|row| row.attempt < run.attempt)
243 .map(|row| RunAttempt {
244 attempt: row.attempt,
245 status: "completed".to_owned(),
246 conclusion: row.conclusion,
247 actor: row.actor,
248 debug: row.debug != 0,
249 started_at: row.started_at,
250 finished_at: row.finished_at,
251 })
252 .collect();
253 attempts.push(RunAttempt {
254 attempt: run.attempt,
255 status: run.status.clone(),
256 conclusion: run.conclusion.clone().filter(|_| run.status == "completed"),
257 actor: run.triggering_actor.clone().or_else(|| run.actor.clone()),
258 debug: run.debug != 0,
259 started_at: run.started_at.clone(),
260 finished_at: run.finished_at.clone(),
261 });
262 Ok(attempts)
263 }
264
265 /// The jobs of one attempt: the live ones for the current attempt, the
266 /// kept ones for an earlier attempt. None for an attempt it never had.
267 async fn attempt_jobs(&self, run: &RunRow, attempt: Option<u64>) -> Result<Option<AttemptJobs>> {
268 match attempt.filter(|n| *n != run.attempt) {
269 None => {
270 // A job that deploys: where to, as its `environment.url` reads.
271 let jobs: Vec<Job> = self
272 .job_rows(&run.id)
273 .await?
274 .into_iter()
275 .map(|row| {
276 let environment_url = row.environment.as_ref().and_then(|_| crate::plan::deploys_to(run, &row)).and_then(|env| env.url);
277 Job { environment_url, ..job_view(row) }
278 })
279 .collect();
280 let log_ids = jobs.iter().map(|job| job.id.clone()).collect();
281 Ok(Some(AttemptJobs { jobs, log_ids }))
282 }
283 Some(n) if n == 0 || n > run.attempt => Ok(None),
284 Some(n) => {
285 let rows = self
286 .db
287 .prepare("SELECT * FROM job_attempts WHERE run_id = ? AND attempt = ? ORDER BY rowid")
288 .bind(&[run.id.as_str().into(), (n as f64).into()])?
289 .all()
290 .await?
291 .results::<AttemptJobRow>()?;
292 let log_ids = rows.iter().map(|row| row.log_id.clone()).collect();
293 Ok(Some(AttemptJobs { jobs: rows.into_iter().map(attempt_job_view).collect(), log_ids }))
294 }
295 }
296 }
297
298 pub async fn run(&self, a: RunArgs) -> Result<Outcome<RunDetail>> {
299 if self.visible_repo(&a.repo, &a.viewer).await?.is_none() {
300 return Ok(fail(FailureCode::NotFound, "There is no such repository."));
301 }
302 let run = check!(self.run_in(&a.repo, &a.id).await?);
303 let Some(AttemptJobs { jobs, .. }) = self.attempt_jobs(&run, a.attempt).await? else {
304 return Ok(fail(FailureCode::NotFound, format!("This run has no attempt {}.", a.attempt.unwrap_or_default())));
305 };
306 let attempts = self.attempts_of(&run).await?;
307 let mut summary = run.summary();
308 // An earlier attempt: as it ended, and nothing waits for it.
309 if let Some(earlier) = a.attempt.filter(|n| *n != run.attempt).and_then(|n| attempts.iter().find(|at| at.attempt == n)) {
310 summary.attempt = earlier.attempt;
311 summary.status = "completed".to_owned();
312 summary.conclusion = earlier.conclusion.clone();
313 summary.started_at = earlier.started_at.clone();
314 summary.finished_at = earlier.finished_at.clone();
315 return Ok(Outcome::Ok(RunDetail {
316 notes: notes(&run.source),
317 run: summary,
318 jobs,
319 approval: run.approval(),
320 pending_deployments: Vec::new(),
321 attempts,
322 }));
323 }
324 let pending_deployments = self.pending_for(&run, &a.viewer).await?;
325 summary.status = shown_status(&summary.status, jobs.iter().map(|job| job.status.as_str()));
326 Ok(Outcome::Ok(RunDetail {
327 notes: notes(&run.source),
328 run: summary,
329 jobs,
330 approval: run.approval(),
331 pending_deployments,
332 attempts,
333 }))
334 }
335
336 /// A job of the repository's runs by the id a run's page gives it: a
337 /// live job, or one of an earlier attempt. Its name, steps, where its
338 /// logs and summary are kept, and whether it has finished.
339 async fn job_for_log(&self, repo: &g1t_contracts::repos::RepoPath, id: &str) -> Result<Option<(Job, String, bool)>> {
340 let full = format!("{}/{}", repo.namespace, repo.name);
341 let live = self
342 .db
343 .prepare("SELECT jobs.* FROM jobs JOIN runs ON runs.id = jobs.run_id WHERE jobs.id = ? AND lower(runs.repo) = lower(?)")
344 .bind(&[id.into(), full.as_str().into()])?
345 .first::<JobRow>(None)
346 .await?;
347 if let Some(job) = live {
348 let done = job.status == "completed";
349 let view = job_view(job);
350 let log_id = view.id.clone();
351 return Ok(Some((view, log_id, done)));
352 }
353 let kept = self
354 .db
355 .prepare(
356 "SELECT job_attempts.* FROM job_attempts JOIN runs ON runs.id = job_attempts.run_id
357 WHERE job_attempts.id = ? AND lower(runs.repo) = lower(?)",
358 )
359 .bind(&[id.into(), full.as_str().into()])?
360 .first::<AttemptJobRow>(None)
361 .await?;
362 Ok(kept.map(|row| {
363 let log_id = row.log_id.clone();
364 (attempt_job_view(row), log_id, true)
365 }))
366 }
367
368 async fn log_chunks(&self, log_id: &str, after: u64, limit: u32) -> Result<Vec<LogChunk>> {
369 #[derive(Deserialize)]
370 struct Row {
371 seq: u64,
372 step: u32,
373 text: String,
374 }
375 Ok(self
376 .db
377 .prepare("SELECT seq, step, text FROM logs WHERE job_id = ? AND seq > ? ORDER BY seq LIMIT ?")
378 .bind(&[log_id.into(), (after as f64).into(), limit.into()])?
379 .all()
380 .await?
381 .results::<Row>()?
382 .into_iter()
383 .map(|row| LogChunk { seq: row.seq, step: row.step, text: row.text })
384 .collect())
385 }
386
387 /// Every chunk of a job's log, a page at a time.
388 async fn whole_log(&self, log_id: &str) -> Result<Vec<LogChunk>> {
389 let mut chunks: Vec<LogChunk> = Vec::new();
390 loop {
391 let after = chunks.last().map_or(0, |chunk| chunk.seq);
392 let page = self.log_chunks(log_id, after, 500).await?;
393 let more = page.len() == 500;
394 chunks.extend(page);
395 if !more {
396 return Ok(chunks);
397 }
398 }
399 }
400
401 pub async fn logs(&self, a: LogsArgs) -> Result<Outcome<JobLog>> {
402 if self.visible_repo(&a.repo, &a.viewer).await?.is_none() {
403 return Ok(fail(FailureCode::NotFound, "There is no such repository."));
404 }
405 let Some((_, log_id, done)) = self.job_for_log(&a.repo, &a.job).await? else {
406 return Ok(fail(FailureCode::NotFound, "No such job."));
407 };
408 let chunks = self.log_chunks(&log_id, a.after, 500).await?;
409 Ok(Outcome::Ok(JobLog { chunks, done }))
410 }
411
412 /// `job_log_text`: a job's whole log, to download.
413 pub async fn job_log_text(&self, a: JobLogTextArgs) -> Result<Outcome<JobLogText>> {
414 if self.visible_repo(&a.repo, &a.viewer).await?.is_none() {
415 return Ok(fail(FailureCode::NotFound, "There is no such repository."));
416 }
417 let Some((job, log_id, done)) = self.job_for_log(&a.repo, &a.job).await? else {
418 return Ok(fail(FailureCode::NotFound, "No such job."));
419 };
420 let chunks = self.whole_log(&log_id).await?;
421 Ok(Outcome::Ok(JobLogText { job_id: job.id, name: job.name, steps: job.steps, chunks, done, omitted: false }))
422 }
423
424 /// `run_logs`: every job's whole log for one attempt, to download as one
425 /// archive.
426 pub async fn run_logs(&self, a: RunLogsArgs) -> Result<Outcome<Vec<JobLogText>>> {
427 if self.visible_repo(&a.repo, &a.viewer).await?.is_none() {
428 return Ok(fail(FailureCode::NotFound, "There is no such repository."));
429 }
430 let run = check!(self.run_in(&a.repo, &a.id).await?);
431 let Some(AttemptJobs { jobs, log_ids }) = self.attempt_jobs(&run, a.attempt).await? else {
432 return Ok(fail(FailureCode::NotFound, format!("This run has no attempt {}.", a.attempt.unwrap_or_default())));
433 };
434 let mut out = Vec::with_capacity(jobs.len());
435 let mut total = 0usize;
436 for (job, log_id) in jobs.into_iter().zip(log_ids) {
437 let done = job.status == "completed";
438 let omitted = total >= MAX_RUN_LOG_BYTES;
439 let chunks = if omitted { Vec::new() } else { self.whole_log(&log_id).await? };
440 total += chunks.iter().map(|chunk| chunk.text.len()).sum::<usize>();
441 out.push(JobLogText { job_id: job.id, name: job.name, steps: job.steps, chunks, done, omitted });
442 }
443 Ok(Outcome::Ok(out))
444 }
445
446 /// `summaries`: what each job's steps wrote to `$GITHUB_STEP_SUMMARY`.
447 pub async fn summaries(&self, a: SummariesArgs) -> Result<Outcome<Vec<JobSummary>>> {
448 if self.visible_repo(&a.repo, &a.viewer).await?.is_none() {
449 return Ok(fail(FailureCode::NotFound, "There is no such repository."));
450 }
451 let run = check!(self.run_in(&a.repo, &a.id).await?);
452 let Some(AttemptJobs { jobs, log_ids }) = self.attempt_jobs(&run, a.attempt).await? else {
453 return Ok(fail(FailureCode::NotFound, format!("This run has no attempt {}.", a.attempt.unwrap_or_default())));
454 };
455 #[derive(Deserialize)]
456 struct Row {
457 job_id: String,
458 step: u32,
459 markdown: String,
460 }
461 let rows = self
462 .db
463 .prepare("SELECT job_id, step, markdown FROM job_summaries WHERE job_id IN (SELECT value FROM json_each(?)) ORDER BY step")
464 .bind(&[serde_json::to_string(&log_ids)?.into()])?
465 .all()
466 .await?
467 .results::<Row>()?;
468 Ok(Outcome::Ok(
469 jobs.into_iter()
470 .zip(log_ids)
471 .filter_map(|(job, log_id)| {
472 let steps: Vec<StepSummary> = rows
473 .iter()
474 .filter(|row| row.job_id == log_id)
475 .map(|row| StepSummary { step: row.step, markdown: row.markdown.clone() })
476 .collect();
477 (!steps.is_empty()).then_some(JobSummary { job_id: job.id, name: job.name, steps })
478 })
479 .collect(),
480 ))
481 }
482
483 pub async fn set_workflow_enabled(&self, a: SetWorkflowEnabledArgs) -> Result<Outcome<Workflow>> {
484 let repo = match self.may(&a.actor, &a.repo, Capability::ManageSettings).await? {
485 Outcome::Ok(repo) => repo,
486 Outcome::Fail(refused) => return Ok(Outcome::Fail(refused)),
487 };
488 let wanted = a.workflow.trim_start_matches(".g1t/workflows/");
489 let row = self
490 .db
491 .prepare("SELECT * FROM workflows WHERE lower(repo) = lower(?) AND (id = ? OR path = ?)")
492 .bind(&[
493 format!("{}/{}", a.repo.namespace, a.repo.name).into(),
494 a.workflow.as_str().into(),
495 format!("{}/{wanted}", g1t_actions::workflow::FOLDER).into(),
496 ])?
497 .first::<WorkflowRow>(None)
498 .await?;
499 let Some(row) = row else {
500 return Ok(fail(FailureCode::NotFound, "No such workflow."));
501 };
502 // A workspace's required workflow (managed.rs) stays on.
503 if !a.enabled && self.required_here(&repo, &row.path).await? {
504 return Ok(fail(
505 FailureCode::Forbidden,
506 format!("{} is required by {}: only its owners can change where it runs, under its Settings, Actions.", row.name, repo.namespace),
507 ));
508 }
509 let state = if a.enabled { "active" } else { "disabled" };
510 self.db.prepare("UPDATE workflows SET state = ? WHERE id = ?").bind(&[state.into(), row.id.as_str().into()])?.run().await?;
511 let row = WorkflowRow { state: state.to_owned(), ..row };
512 Ok(Outcome::Ok(self.summary(&row).await?))
513 }
514
515 /// `check_runs`: workflow runs with their jobs, by commits, by a run's
516 /// id or by one of its jobs' ids, newest first, for the work service to
517 /// show as check suites and check runs. It decides who may see them.
518 pub async fn check_runs(&self, a: ActionsChecksArgs) -> Result<Vec<RunDetail>> {
519 let runs: Vec<RunRow> = if let Some(job) = &a.job_id {
520 self.db
521 .prepare("SELECT runs.* FROM runs JOIN jobs ON jobs.run_id = runs.id WHERE jobs.id = ? AND runs.repo_id = ?")
522 .bind(&[job.as_str().into(), a.repo_id.as_str().into()])?
523 .all()
524 .await?
525 .results::<RunRow>()?
526 } else if let Some(run) = &a.run_id {
527 self.db
528 .prepare("SELECT * FROM runs WHERE id = ? AND repo_id = ?")
529 .bind(&[run.as_str().into(), a.repo_id.as_str().into()])?
530 .all()
531 .await?
532 .results::<RunRow>()?
533 } else if a.shas.is_empty() {
534 Vec::new()
535 } else {
536 let shas: Vec<&String> = a.shas.iter().take(MAX_COMMITS).collect();
537 let marks = vec!["?"; shas.len()].join(", ");
538 let mut binds: Vec<worker::wasm_bindgen::JsValue> = vec![a.repo_id.as_str().into()];
539 binds.extend(shas.iter().map(|sha| sha.as_str().into()));
540 binds.push(CHECK_RUNS_LIMIT.into());
541 self.db
542 .prepare(format!("SELECT * FROM runs WHERE repo_id = ? AND sha IN ({marks}) ORDER BY id DESC LIMIT ?"))
543 .bind(&binds)?
544 .all()
545 .await?
546 .results::<RunRow>()?
547 };
548 if runs.is_empty() {
549 return Ok(Vec::new());
550 }
551 let marks = vec!["?"; runs.len()].join(", ");
552 let ids: Vec<worker::wasm_bindgen::JsValue> = runs.iter().map(|run| run.id.as_str().into()).collect();
553 let jobs = self
554 .db
555 .prepare(format!("SELECT * FROM jobs WHERE run_id IN ({marks}) ORDER BY rowid"))
556 .bind(&ids)?
557 .all()
558 .await?
559 .results::<JobRow>()?;
560 let mut by_run: std::collections::HashMap<String, Vec<Job>> = std::collections::HashMap::new();
561 for job in jobs {
562 by_run.entry(job.run_id.clone()).or_default().push(job_view(job));
563 }
564 Ok(runs
565 .iter()
566 .map(|run| RunDetail {
567 run: run.summary(),
568 jobs: by_run.remove(&run.id).unwrap_or_default(),
569 notes: Vec::new(),
570 approval: run.approval(),
571 pending_deployments: Vec::new(),
572 attempts: Vec::new(),
573 })
574 .collect())
575 }
576}
577
578/// The most runs `check_runs` reads for a set of commits.
579const CHECK_RUNS_LIMIT: u32 = 200;
580
581/// How the run's current attempt stands on its page: as its row says, and
582/// `waiting`, as on GitHub, while a job is held at an environment's rules.
583/// A job queued for a runner, or waiting for the jobs it needs, leaves a
584/// run that has started `in_progress`: it is still going.
585fn shown_status<'a>(status: &str, mut jobs: impl Iterator<Item = &'a str>) -> String {
586 if matches!(status, "queued" | "in_progress") && jobs.any(|job| job == "pending") {
587 "waiting".to_owned()
588 } else {
589 status.to_owned()
590 }
591}
592
593#[cfg(test)]
594mod tests {
595 use super::*;
596
597 fn shown(status: &str, jobs: &[&str]) -> String {
598 shown_status(status, jobs.iter().copied())
599 }
600
601 #[test]
602 fn a_job_queued_for_a_self_hosted_runner_keeps_its_run_going() {
603 // A re-run of failed jobs: the job that failed now succeeded, and the
604 // one that needs it waits for a runner with labels no runner has.
605 assert_eq!(shown("in_progress", &["completed", "queued"]), "in_progress");
606 assert_eq!(shown("queued", &["waiting", "waiting"]), "queued");
607 }
608
609 #[test]
610 fn a_job_held_at_an_environment_makes_its_run_wait() {
611 assert_eq!(shown("in_progress", &["completed", "pending"]), "waiting");
612 assert_eq!(shown("queued", &["pending"]), "waiting");
613 }
614
615 #[test]
616 fn a_finished_or_held_run_reads_as_its_row_says() {
617 assert_eq!(shown("completed", &["completed"]), "completed");
618 assert_eq!(shown("action_required", &["waiting"]), "action_required");
619 assert_eq!(shown("pending", &["pending"]), "pending");
620 }
621}