Skip to content
Sign up

flagon-io/g1t

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