Skip to content
Sign up

flagon-io/g1t

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