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.

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