Sign in

flagon-io/g1t

public

g1t/services/actions/src/views.rs

226 lines9,249 bytesCodeBlame

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

GitHub Actions on g1t, part two: running workflows1//! Reading: workflows, runs, a run's jobs, and a job's log.
2
3use g1t_actions::workflow::{self, Severity};
Invite-only launch: sign in with GitHub, repository access and lifecycle, many emails, a new look4use g1t_contracts::access::Capability;
GitHub Actions on g1t, part two: running workflows5use g1t_contracts::actions::{
6 Annotation, Job, JobLog, LogChunk, LogsArgs, RunArgs, RunDetail, RunsArgs, SetWorkflowEnabledArgs, StepState, Workflow, WorkflowNote,
7 WorkflowRun, WorkflowsArgs,
8};
9use g1t_contracts::{FailureCode, Outcome};
10use serde::Deserialize;
11use worker::Result;
12
13use crate::plan::{JobRow, RunRow};
14use crate::sync::WorkflowRow;
15use crate::{Actions, check, fail};
16
17const RUNS_SHOWN: u32 = 50;
18
19fn notes(source: &str) -> Vec<WorkflowNote> {
20 workflow::parse(source)
21 .map(|w| {
22 w.notes
23 .into_iter()
24 .map(|note| WorkflowNote {
25 severity: match note.severity {
26 Severity::Info => "info",
27 Severity::Warning => "warning",
28 Severity::Unsupported => "unsupported",
29 }
30 .to_owned(),
31 job: note.job,
32 message: note.message,
33 })
34 .collect()
35 })
36 .unwrap_or_default()
37}
38
39fn job_view(row: JobRow) -> Job {
40 let needs = row.needs();
41 Job {
42 id: row.id,
43 run_id: row.run_id,
44 key: row.key,
45 name: row.name,
46 needs,
47 status: row.status,
48 conclusion: row.conclusion,
49 steps: serde_json::from_str::<Vec<StepState>>(&row.steps).unwrap_or_default(),
50 annotations: serde_json::from_str::<Vec<Annotation>>(&row.annotations).unwrap_or_default(),
51 reason: row.reason,
52 started_at: row.started_at,
53 finished_at: row.finished_at,
Fast pages, required checks on the branch, self-hosted runners, honest incidents54 self_hosted: row.labels.is_some(),
55 runner: row.runner_name,
GitHub Actions on g1t, part two: running workflows56 }
57}
58
59impl Actions {
60 async fn summary(&self, row: &WorkflowRow) -> Result<Workflow> {
61 let last_run = self
62 .db
63 .prepare("SELECT * FROM runs WHERE workflow_id = ? ORDER BY id DESC LIMIT 1")
64 .bind(&[row.id.as_str().into()])?
65 .first::<RunRow>(None)
66 .await?
67 .map(|run| run.summary());
68 let parsed = workflow::parse(&row.source).ok();
69 Ok(Workflow {
70 id: row.id.clone(),
71 path: row.path.clone(),
72 name: row.name.clone(),
73 events: serde_json::from_str(&row.events).unwrap_or_default(),
74 state: row.state.clone(),
75 error: row.error.clone(),
76 notes: notes(&row.source),
77 dispatch: parsed
78 .as_ref()
79 .filter(|_| row.error.is_none())
80 .and_then(|w| w.trigger("workflow_dispatch"))
81 .map(|t| serde_json::Value::Object(t.inputs.clone())),
82 last_run,
83 })
84 }
85
86 pub async fn workflows(&self, a: WorkflowsArgs) -> Result<Outcome<Vec<Workflow>>> {
87 let Some(repo) = self.visible_repo(&a.repo, &a.viewer).await? else {
88 return Ok(fail(FailureCode::NotFound, "There is no such repository."));
89 };
90 if !self.synced(&repo.id).await?
91 && let Some(ws) = self.workspace_actor(&repo.namespace).await?
92 {
93 self.sync(&repo, &ws).await?;
94 }
One CI in the workflow list: hide files from before .github stopped being read, and gone files a live one replaced95 // A file that is gone stays listed while it has runs, unless one in
96 // the folder now has its name: then it would read as a second copy.
97 // Files outside the folder are from before g1t stopped reading
98 // `.github`; their runs stay under All workflows.
GitHub Actions on g1t, part two: running workflows99 let rows = self
100 .db
101 .prepare(
One CI in the workflow list: hide files from before .github stopped being read, and gone files a live one replaced102 "SELECT * FROM workflows w WHERE w.repo_id = ?1 AND w.path LIKE ?2
103 AND (w.error IS NULL OR w.error NOT LIKE 'Its file is%'
104 OR (w.id IN (SELECT workflow_id FROM runs WHERE repo_id = ?1)
105 AND NOT EXISTS (SELECT 1 FROM workflows o WHERE o.repo_id = ?1 AND o.id <> w.id
106 AND o.name = w.name AND (o.error IS NULL OR o.error NOT LIKE 'Its file is%'))))
107 ORDER BY w.name",
GitHub Actions on g1t, part two: running workflows108 )
One CI in the workflow list: hide files from before .github stopped being read, and gone files a live one replaced109 .bind(&[repo.id.as_str().into(), format!("{}/%", g1t_actions::workflow::FOLDER).into()])?
GitHub Actions on g1t, part two: running workflows110 .all()
111 .await?
112 .results::<WorkflowRow>()?;
113 let mut out = Vec::with_capacity(rows.len());
114 for row in &rows {
115 out.push(self.summary(row).await?);
116 }
117 Ok(Outcome::Ok(out))
118 }
119
120 pub async fn runs(&self, a: RunsArgs) -> Result<Outcome<Vec<WorkflowRun>>> {
121 let Some(repo) = self.visible_repo(&a.repo, &a.viewer).await? else {
122 return Ok(fail(FailureCode::NotFound, "There is no such repository."));
123 };
124 let mut sql = "SELECT * FROM runs WHERE repo_id = ?".to_owned();
125 let mut binds: Vec<worker::wasm_bindgen::JsValue> = vec![repo.id.as_str().into()];
126 if let Some(workflow) = &a.workflow {
127 sql.push_str(" AND (workflow_id = ? OR path = ? OR path = ?)");
128 binds.push(workflow.as_str().into());
129 binds.push(workflow.as_str().into());
130 binds.push(format!("{}/{workflow}", g1t_actions::workflow::FOLDER).into());
131 }
132 if let Some(branch) = &a.branch {
133 sql.push_str(" AND (git_ref = ? OR json_extract(info, '$.headRef') = ?)");
134 binds.push(format!("refs/heads/{branch}").into());
135 binds.push(branch.as_str().into());
136 }
137 if let Some(event) = &a.event {
138 sql.push_str(" AND event = ?");
139 binds.push(event.as_str().into());
140 }
141 if let Some(pull) = a.pull {
142 sql.push_str(" AND pull = ?");
143 binds.push(pull.into());
144 }
145 if let Some(sha) = &a.sha {
146 sql.push_str(" AND sha = ?");
147 binds.push(sha.as_str().into());
148 }
149 sql.push_str(" ORDER BY id DESC LIMIT ?");
150 binds.push(a.limit.unwrap_or(RUNS_SHOWN).clamp(1, 100).into());
151 let rows = self.db.prepare(sql).bind(&binds)?.all().await?.results::<RunRow>()?;
152 Ok(Outcome::Ok(rows.iter().map(RunRow::summary).collect()))
153 }
154
155 pub async fn run(&self, a: RunArgs) -> Result<Outcome<RunDetail>> {
156 if self.visible_repo(&a.repo, &a.viewer).await?.is_none() {
157 return Ok(fail(FailureCode::NotFound, "There is no such repository."));
158 }
159 let run = check!(self.run_in(&a.repo, &a.id).await?);
160 let jobs = self.job_rows(&run.id).await?.into_iter().map(job_view).collect();
161 Ok(Outcome::Ok(RunDetail {
162 notes: notes(&run.source),
163 run: run.summary(),
164 jobs,
165 }))
166 }
167
168 pub async fn logs(&self, a: LogsArgs) -> Result<Outcome<JobLog>> {
169 if self.visible_repo(&a.repo, &a.viewer).await?.is_none() {
170 return Ok(fail(FailureCode::NotFound, "There is no such repository."));
171 }
172 let job = self
173 .db
174 .prepare("SELECT jobs.* FROM jobs JOIN runs ON runs.id = jobs.run_id WHERE jobs.id = ? AND lower(runs.repo) = lower(?)")
175 .bind(&[a.job.as_str().into(), format!("{}/{}", a.repo.namespace, a.repo.name).into()])?
176 .first::<JobRow>(None)
177 .await?;
178 let Some(job) = job else {
179 return Ok(fail(FailureCode::NotFound, "No such job."));
180 };
181 #[derive(Deserialize)]
182 struct Row {
183 seq: u64,
184 step: u32,
185 text: String,
186 }
187 let chunks = self
188 .db
189 .prepare("SELECT seq, step, text FROM logs WHERE job_id = ? AND seq > ? ORDER BY seq LIMIT 500")
190 .bind(&[job.id.as_str().into(), (a.after as f64).into()])?
191 .all()
192 .await?
193 .results::<Row>()?
194 .into_iter()
195 .map(|row| LogChunk { seq: row.seq, step: row.step, text: row.text })
196 .collect();
197 Ok(Outcome::Ok(JobLog {
198 chunks,
199 done: job.status == "completed",
200 }))
201 }
202
203 pub async fn set_workflow_enabled(&self, a: SetWorkflowEnabledArgs) -> Result<Outcome<Workflow>> {
Invite-only launch: sign in with GitHub, repository access and lifecycle, many emails, a new look204 if let Outcome::Fail(refused) = self.may(&a.actor, &a.repo, Capability::ManageSettings).await? {
GitHub Actions on g1t, part two: running workflows205 return Ok(Outcome::Fail(refused));
206 }
GitHub Actions on g1t, part three: .g1t/workflows, the pages, the docs207 let wanted = a.workflow.trim_start_matches(".g1t/workflows/");
GitHub Actions on g1t, part two: running workflows208 let row = self
209 .db
210 .prepare("SELECT * FROM workflows WHERE lower(repo) = lower(?) AND (id = ? OR path = ?)")
211 .bind(&[
212 format!("{}/{}", a.repo.namespace, a.repo.name).into(),
213 a.workflow.as_str().into(),
214 format!("{}/{wanted}", g1t_actions::workflow::FOLDER).into(),
215 ])?
216 .first::<WorkflowRow>(None)
217 .await?;
218 let Some(row) = row else {
219 return Ok(fail(FailureCode::NotFound, "No such workflow."));
220 };
221 let state = if a.enabled { "active" } else { "disabled" };
222 self.db.prepare("UPDATE workflows SET state = ? WHERE id = ?").bind(&[state.into(), row.id.as_str().into()])?.run().await?;
223 let row = WorkflowRow { state: state.to_owned(), ..row };
224 Ok(Outcome::Ok(self.summary(&row).await?))
225 }
226}