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.

gitagentscloudflarerustforgemcp
295 linesCodeBlameRaw
1//! Reading workflow files: at any commit for a run, and from the default
2//! branch into the `workflows` table, which lists them, holds their
3//! schedules and remembers which are turned off.
4
5use g1t_actions::workflow::{self, FOLDER};
6use g1t_contracts::new_id;
7use g1t_contracts::repos::{BlobArgs, BlobView, EntryKind, Repo, RepoPath, TreeArgs, TreeView};
8use g1t_contracts::time::rfc3339;
9use g1t_contracts::{Outcome, User};
10use g1t_kit::now_ms;
11use serde::Deserialize;
12use worker::Result;
13
14use futures_util::{StreamExt, stream};
15
16use crate::{Actions, Count, MAX_WORKFLOWS, optional};
17
18/// Workflow files read from repos at once: under the six connections a
19/// Worker keeps open.
20const BLOBS_AT_ONCE: usize = 6;
21
22/// One workflow file as read at a commit.
23#[derive(Clone)]
24pub struct WorkflowFile {
25 pub path: String,
26 pub source: String,
27 /// A workspace's required workflow (managed.rs): no one in the
28 /// repository can turn it off.
29 pub required: bool,
30}
31
32/// The files at a commit, and the commit the ref resolved to.
33#[derive(Clone)]
34pub struct Read {
35 pub files: Vec<WorkflowFile>,
36 pub head: Option<String>,
37}
38
39#[derive(Deserialize)]
40pub struct WorkflowRow {
41 pub id: String,
42 pub repo_id: String,
43 pub repo: String,
44 pub path: String,
45 pub name: String,
46 pub source: String,
47 pub events: String,
48 pub crons: String,
49 pub error: Option<String>,
50 pub state: String,
51 pub updated_at: String,
52}
53
54/// What a workflow's row says once its file has left the default branch.
55pub const GONE: &str = "Its file is no longer on the default branch.";
56
57/// Whether a synced workflow could start on `event`: it lists the event,
58/// or it did not parse (and is still on the branch), so reading it again
59/// reports why.
60pub fn could_start(events: &str, error: Option<&str>, event: &str) -> bool {
61 match error {
62 Some(error) => error != GONE,
63 None => serde_json::from_str::<Vec<String>>(events).is_ok_and(|events| events.iter().any(|e| e == event)),
64 }
65}
66
67impl Actions {
68 /// The workflow files of `path` as of `git_ref` (the default branch
69 /// when absent).
70 pub async fn read_workflows(&self, path: &RepoPath, actor: &User, git_ref: Option<&str>) -> Result<Read> {
71 self.read_folders(path, actor, git_ref, &[FOLDER]).await
72 }
73
74 /// The workflow files a repository runs: `.g1t/workflows`, with the
75 /// managed workflows installed in it and its workspace's that target
76 /// it, as managed.rs puts them together.
77 pub async fn read_for(&self, repo: &Repo, actor: &User, git_ref: Option<&str>) -> Result<Read> {
78 let path = RepoPath {
79 namespace: repo.namespace.clone(),
80 name: repo.name.clone(),
81 };
82 let mut read = self.read_workflows(&path, actor, git_ref).await?;
83 // And the managed workflows beside them (managed.rs).
84 read.files = self.with_managed(repo, read.files).await?;
85 Ok(read)
86 }
87
88 async fn read_folders(&self, path: &RepoPath, actor: &User, git_ref: Option<&str>, folders: &[&str]) -> Result<Read> {
89 let viewer = Some(actor.clone());
90 let mut files = Vec::new();
91 let mut found_head = None;
92 for folder in folders {
93 let tree: Outcome<TreeView> = g1t_kit::call(
94 &self.repos,
95 "tree",
96 &TreeArgs {
97 path: path.clone(),
98 viewer: viewer.clone(),
99 git_ref: git_ref.map(str::to_owned),
100 tree_path: (*folder).to_owned(),
101 },
102 )
103 .await?;
104 let (entries, head, resolved) = match tree {
105 Outcome::Ok(tree) => (tree.entries, tree.head.map(|commit| commit.hash), tree.git_ref),
106 // No folder: no workflows from it.
107 Outcome::Fail(_) => continue,
108 };
109 let at = head.clone().unwrap_or(resolved);
110 found_head = found_head.or(head);
111 let file_paths: Vec<String> = entries
112 .into_iter()
113 .filter(|entry| matches!(entry.kind, EntryKind::Blob | EntryKind::Exec))
114 .filter(|entry| entry.name.ends_with(".yml") || entry.name.ends_with(".yaml"))
115 .take(MAX_WORKFLOWS.saturating_sub(files.len()))
116 .map(|entry| format!("{folder}/{}", entry.name))
117 .collect();
118 // A few at a time, in the tree's order: one after another, a
119 // repository's dozen workflows took 4 s of every push.
120 let repos = &self.repos;
121 let blobs: Vec<Result<Outcome<BlobView>>> = stream::iter(file_paths.iter().map(|file_path| {
122 let args = BlobArgs {
123 path: path.clone(),
124 viewer: viewer.clone(),
125 git_ref: at.clone(),
126 file_path: file_path.clone(),
127 };
128 async move { g1t_kit::call(repos, "blob", &args).await }
129 }))
130 .buffered(BLOBS_AT_ONCE)
131 .collect()
132 .await;
133 for (file_path, blob) in file_paths.into_iter().zip(blobs) {
134 if let Outcome::Ok(BlobView { text: Some(source), .. }) = blob? {
135 files.push(WorkflowFile { path: file_path, source, required: false });
136 }
137 }
138 }
139 Ok(Read { files, head: found_head })
140 }
141
142 /// Keeps the `workflows` table in step with the default branch, and
143 /// gives back the files it read there (`on_event` starts a default
144 /// branch push's runs from them rather than reading them again).
145 pub async fn sync(&self, repo: &Repo, actor: &User) -> Result<Read> {
146 let read = self.read_for(repo, actor, None).await?;
147 let full_name = format!("{}/{}", repo.namespace, repo.name);
148 let now = rfc3339(now_ms());
149 let mut statements = Vec::new();
150 for file in &read.files {
151 let parsed = workflow::parse(&file.source);
152 let (name, error, events, crons) = match &parsed {
153 Ok(parsed) => (
154 parsed.display_name(&file.path),
155 None,
156 parsed.triggers.iter().map(|t| t.event.clone()).collect::<Vec<_>>(),
157 parsed.trigger("schedule").map(|t| t.crons.clone()).unwrap_or_default(),
158 ),
159 Err(problem) => (file.path.clone(), Some(problem.clone()), Vec::new(), Vec::new()),
160 };
161 statements.push(
162 self.db
163 .prepare(
164 "INSERT INTO workflows (id, repo_id, repo, path, name, source, events, crons, error, updated_at)
165 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
166 ON CONFLICT (repo_id, path) DO UPDATE SET
167 repo = excluded.repo, name = excluded.name, source = excluded.source,
168 events = excluded.events, crons = excluded.crons, error = excluded.error,
169 updated_at = excluded.updated_at",
170 )
171 .bind(&[
172 new_id("wfl", now_ms()).into(),
173 repo.id.as_str().into(),
174 full_name.as_str().into(),
175 file.path.as_str().into(),
176 name.into(),
177 file.source.as_str().into(),
178 serde_json::to_string(&events)?.into(),
179 serde_json::to_string(&crons)?.into(),
180 optional(error.as_deref()),
181 now.as_str().into(),
182 ])?,
183 );
184 }
185 // A workflow whose file is gone keeps its runs, but no longer runs
186 // on schedule or by hand: it is listed only while it has runs.
187 let kept: Vec<String> = read.files.iter().map(|file| file.path.clone()).collect();
188 let existing = self
189 .db
190 .prepare("SELECT * FROM workflows WHERE repo_id = ?")
191 .bind(&[repo.id.as_str().into()])?
192 .all()
193 .await?
194 .results::<WorkflowRow>()?;
195 for row in existing.iter().filter(|row| !kept.contains(&row.path)) {
196 statements.push(
197 self.db
198 .prepare("UPDATE workflows SET crons = '[]', error = ? WHERE id = ?")
199 .bind(&[GONE.into(), row.id.as_str().into()])?,
200 );
201 }
202 // A workspace's required workflow cannot stay turned off, even one
203 // turned off while it was a default.
204 let required: Vec<&str> = read.files.iter().filter(|file| file.required).map(|file| file.path.as_str()).collect();
205 if !required.is_empty() {
206 statements.push(
207 self.db
208 .prepare("UPDATE workflows SET state = 'active' WHERE repo_id = ? AND path IN (SELECT value FROM json_each(?))")
209 .bind(&[repo.id.as_str().into(), serde_json::to_string(&required)?.into()])?,
210 );
211 }
212 statements.push(
213 self.db
214 .prepare("INSERT OR REPLACE INTO synced (repo_id, at) VALUES (?, ?)")
215 .bind(&[repo.id.as_str().into(), now.into()])?,
216 );
217 self.db.batch(statements).await?;
218 Ok(read)
219 }
220
221 /// Whether any of the repository's default-branch workflows could
222 /// start on `event`, from the synced table; None before the first sync.
223 pub async fn listens(&self, repo_id: &str, event: &str) -> Result<Option<bool>> {
224 #[derive(Deserialize)]
225 struct Row {
226 events: String,
227 error: Option<String>,
228 }
229 if !self.synced(repo_id).await? {
230 return Ok(None);
231 }
232 let rows = self
233 .db
234 .prepare("SELECT events, error FROM workflows WHERE repo_id = ? AND state = 'active'")
235 .bind(&[repo_id.into()])?
236 .all()
237 .await?
238 .results::<Row>()?;
239 Ok(Some(rows.iter().any(|row| could_start(&row.events, row.error.as_deref(), event))))
240 }
241
242 pub async fn synced(&self, repo_id: &str) -> Result<bool> {
243 Ok(self
244 .db
245 .prepare("SELECT COUNT(*) AS n FROM synced WHERE repo_id = ?")
246 .bind(&[repo_id.into()])?
247 .first::<Count>(None)
248 .await?
249 .is_some_and(|count| count.n > 0))
250 }
251
252 /// The row for a workflow file, made if it is new (a file that exists
253 /// only on a branch still gets its runs counted and listed).
254 pub async fn workflow_row(&self, repo: &Repo, path: &str, name: &str, source: &str) -> Result<WorkflowRow> {
255 let now = rfc3339(now_ms());
256 self.db
257 .prepare(
258 "INSERT INTO workflows (id, repo_id, repo, path, name, source, events, error, updated_at)
259 VALUES (?, ?, ?, ?, ?, ?, '[]', 'Its file is not on the default branch.', ?)
260 ON CONFLICT (repo_id, path) DO NOTHING",
261 )
262 .bind(&[
263 new_id("wfl", now_ms()).into(),
264 repo.id.as_str().into(),
265 format!("{}/{}", repo.namespace, repo.name).into(),
266 path.into(),
267 name.into(),
268 source.into(),
269 now.into(),
270 ])?
271 .run()
272 .await?;
273 self.db
274 .prepare("SELECT * FROM workflows WHERE repo_id = ? AND path = ?")
275 .bind(&[repo.id.as_str().into(), path.into()])?
276 .first::<WorkflowRow>(None)
277 .await?
278 .ok_or_else(|| worker::Error::RustError("the workflow was not recorded".into()))
279 }
280}
281
282#[cfg(test)]
283mod tests {
284 use super::*;
285
286 #[test]
287 fn a_synced_workflow_starts_on_the_events_it_lists_or_when_broken() {
288 assert!(could_start(r#"["push","issues"]"#, None, "issues"));
289 assert!(!could_start(r#"["push","pull_request"]"#, None, "issue_comment"));
290 // A file that does not parse is read again, so its error shows.
291 assert!(could_start("[]", Some("bad yaml"), "issues"));
292 // A file gone from the branch starts nothing.
293 assert!(!could_start("[]", Some(GONE), "issues"));
294 }
295}