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