Sign in

flagon-io/g1t

public

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

gitagentscloudflarerustforgemcp

g1t/services/actions/src/views.rs

216 lines8,505 bytesCodeBlame
1//! Reading: workflows, runs, a run's jobs, and a job's log.
2
3use g1t_actions::workflow::{self, Severity};
4use g1t_contracts::actions::{
5 Annotation, Job, JobLog, LogChunk, LogsArgs, RunArgs, RunDetail, RunsArgs, SetWorkflowEnabledArgs, StepState, Workflow, WorkflowNote,
6 WorkflowRun, WorkflowsArgs,
7};
8use g1t_contracts::{FailureCode, Outcome};
9use serde::Deserialize;
10use worker::Result;
11
12use crate::plan::{JobRow, RunRow};
13use crate::sync::WorkflowRow;
14use crate::{Actions, check, fail};
15
16const RUNS_SHOWN: u32 = 50;
17
18fn notes(source: &str) -> Vec<WorkflowNote> {
19 workflow::parse(source)
20 .map(|w| {
21 w.notes
22 .into_iter()
23 .map(|note| WorkflowNote {
24 severity: match note.severity {
25 Severity::Info => "info",
26 Severity::Warning => "warning",
27 Severity::Unsupported => "unsupported",
28 }
29 .to_owned(),
30 job: note.job,
31 message: note.message,
32 })
33 .collect()
34 })
35 .unwrap_or_default()
36}
37
38fn job_view(row: JobRow) -> Job {
39 let needs = row.needs();
40 Job {
41 id: row.id,
42 run_id: row.run_id,
43 key: row.key,
44 name: row.name,
45 needs,
46 status: row.status,
47 conclusion: row.conclusion,
48 steps: serde_json::from_str::<Vec<StepState>>(&row.steps).unwrap_or_default(),
49 annotations: serde_json::from_str::<Vec<Annotation>>(&row.annotations).unwrap_or_default(),
50 reason: row.reason,
51 started_at: row.started_at,
52 finished_at: row.finished_at,
53 }
54}
55
56impl Actions {
57 async fn summary(&self, row: &WorkflowRow) -> Result<Workflow> {
58 let last_run = self
59 .db
60 .prepare("SELECT * FROM runs WHERE workflow_id = ? ORDER BY id DESC LIMIT 1")
61 .bind(&[row.id.as_str().into()])?
62 .first::<RunRow>(None)
63 .await?
64 .map(|run| run.summary());
65 let parsed = workflow::parse(&row.source).ok();
66 Ok(Workflow {
67 id: row.id.clone(),
68 path: row.path.clone(),
69 name: row.name.clone(),
70 events: serde_json::from_str(&row.events).unwrap_or_default(),
71 state: row.state.clone(),
72 error: row.error.clone(),
73 notes: notes(&row.source),
74 dispatch: parsed
75 .as_ref()
76 .filter(|_| row.error.is_none())
77 .and_then(|w| w.trigger("workflow_dispatch"))
78 .map(|t| serde_json::Value::Object(t.inputs.clone())),
79 last_run,
80 })
81 }
82
83 pub async fn workflows(&self, a: WorkflowsArgs) -> Result<Outcome<Vec<Workflow>>> {
84 let Some(repo) = self.visible_repo(&a.repo, &a.viewer).await? else {
85 return Ok(fail(FailureCode::NotFound, "There is no such repository."));
86 };
87 if !self.synced(&repo.id).await?
88 && let Some(ws) = self.workspace_actor(&repo.namespace).await?
89 {
90 self.sync(&repo, &ws).await?;
91 }
92 let rows = self
93 .db
94 .prepare(
95 "SELECT * FROM workflows WHERE repo_id = ?
96 AND (error IS NULL OR error NOT LIKE 'Its file is%' OR id IN (SELECT workflow_id FROM runs WHERE repo_id = ?))
97 ORDER BY name",
98 )
99 .bind(&[repo.id.as_str().into(), repo.id.as_str().into()])?
100 .all()
101 .await?
102 .results::<WorkflowRow>()?;
103 let mut out = Vec::with_capacity(rows.len());
104 for row in &rows {
105 out.push(self.summary(row).await?);
106 }
107 Ok(Outcome::Ok(out))
108 }
109
110 pub async fn runs(&self, a: RunsArgs) -> Result<Outcome<Vec<WorkflowRun>>> {
111 let Some(repo) = self.visible_repo(&a.repo, &a.viewer).await? else {
112 return Ok(fail(FailureCode::NotFound, "There is no such repository."));
113 };
114 let mut sql = "SELECT * FROM runs WHERE repo_id = ?".to_owned();
115 let mut binds: Vec<worker::wasm_bindgen::JsValue> = vec![repo.id.as_str().into()];
116 if let Some(workflow) = &a.workflow {
117 sql.push_str(" AND (workflow_id = ? OR path = ? OR path = ?)");
118 binds.push(workflow.as_str().into());
119 binds.push(workflow.as_str().into());
120 binds.push(format!("{}/{workflow}", g1t_actions::workflow::FOLDER).into());
121 }
122 if let Some(branch) = &a.branch {
123 sql.push_str(" AND (git_ref = ? OR json_extract(info, '$.headRef') = ?)");
124 binds.push(format!("refs/heads/{branch}").into());
125 binds.push(branch.as_str().into());
126 }
127 if let Some(event) = &a.event {
128 sql.push_str(" AND event = ?");
129 binds.push(event.as_str().into());
130 }
131 if let Some(pull) = a.pull {
132 sql.push_str(" AND pull = ?");
133 binds.push(pull.into());
134 }
135 if let Some(sha) = &a.sha {
136 sql.push_str(" AND sha = ?");
137 binds.push(sha.as_str().into());
138 }
139 sql.push_str(" ORDER BY id DESC LIMIT ?");
140 binds.push(a.limit.unwrap_or(RUNS_SHOWN).clamp(1, 100).into());
141 let rows = self.db.prepare(sql).bind(&binds)?.all().await?.results::<RunRow>()?;
142 Ok(Outcome::Ok(rows.iter().map(RunRow::summary).collect()))
143 }
144
145 pub async fn run(&self, a: RunArgs) -> Result<Outcome<RunDetail>> {
146 if self.visible_repo(&a.repo, &a.viewer).await?.is_none() {
147 return Ok(fail(FailureCode::NotFound, "There is no such repository."));
148 }
149 let run = check!(self.run_in(&a.repo, &a.id).await?);
150 let jobs = self.job_rows(&run.id).await?.into_iter().map(job_view).collect();
151 Ok(Outcome::Ok(RunDetail {
152 notes: notes(&run.source),
153 run: run.summary(),
154 jobs,
155 }))
156 }
157
158 pub async fn logs(&self, a: LogsArgs) -> Result<Outcome<JobLog>> {
159 if self.visible_repo(&a.repo, &a.viewer).await?.is_none() {
160 return Ok(fail(FailureCode::NotFound, "There is no such repository."));
161 }
162 let job = self
163 .db
164 .prepare("SELECT jobs.* FROM jobs JOIN runs ON runs.id = jobs.run_id WHERE jobs.id = ? AND lower(runs.repo) = lower(?)")
165 .bind(&[a.job.as_str().into(), format!("{}/{}", a.repo.namespace, a.repo.name).into()])?
166 .first::<JobRow>(None)
167 .await?;
168 let Some(job) = job else {
169 return Ok(fail(FailureCode::NotFound, "No such job."));
170 };
171 #[derive(Deserialize)]
172 struct Row {
173 seq: u64,
174 step: u32,
175 text: String,
176 }
177 let chunks = self
178 .db
179 .prepare("SELECT seq, step, text FROM logs WHERE job_id = ? AND seq > ? ORDER BY seq LIMIT 500")
180 .bind(&[job.id.as_str().into(), (a.after as f64).into()])?
181 .all()
182 .await?
183 .results::<Row>()?
184 .into_iter()
185 .map(|row| LogChunk { seq: row.seq, step: row.step, text: row.text })
186 .collect();
187 Ok(Outcome::Ok(JobLog {
188 chunks,
189 done: job.status == "completed",
190 }))
191 }
192
193 pub async fn set_workflow_enabled(&self, a: SetWorkflowEnabledArgs) -> Result<Outcome<Workflow>> {
194 if let Some(Outcome::Fail(refused)) = Self::member(&a.actor, &a.repo) {
195 return Ok(Outcome::Fail(refused));
196 }
197 let wanted = a.workflow.trim_start_matches(".g1t/workflows/");
198 let row = self
199 .db
200 .prepare("SELECT * FROM workflows WHERE lower(repo) = lower(?) AND (id = ? OR path = ?)")
201 .bind(&[
202 format!("{}/{}", a.repo.namespace, a.repo.name).into(),
203 a.workflow.as_str().into(),
204 format!("{}/{wanted}", g1t_actions::workflow::FOLDER).into(),
205 ])?
206 .first::<WorkflowRow>(None)
207 .await?;
208 let Some(row) = row else {
209 return Ok(fail(FailureCode::NotFound, "No such workflow."));
210 };
211 let state = if a.enabled { "active" } else { "disabled" };
212 self.db.prepare("UPDATE workflows SET state = ? WHERE id = ?").bind(&[state.into(), row.id.as_str().into()])?.run().await?;
213 let row = WorkflowRow { state: state.to_owned(), ..row };
214 Ok(Outcome::Ok(self.summary(&row).await?))
215 }
216}