Skip to content
Sign up

flagon-io/g1t

public
0
279 linesCodeBlameRaw

Pick any line to see why it is the way it is: the commit, the pull request and issue it came from, and what the agent was thinking.

The statement is a month at a time, a line per kind of charge1//! 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
Fewer Artifacts reads: the store is asked for a handle only when needed, objects are kept in the isolate, and issue events read no workflows nobody listens for43/// 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
The statement is a month at a time, a line per kind of charge56impl 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> {
Merge branch 'mirroring' into artifacts-mode60 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
Merge branch 'claude/cool-newton-7nm8r9' (#36): real GitHub Actions on g1t, managed deploy workflows, g1t/deploy and the Sysbox runner 0.2.0. The merge keeps both sides' webhook events and API operations, says agent images need tar, guards the job engine's stop to Unix so the runner builds on Windows again, and keeps the moves tests' previews from going idle as the calendar passes their fixture's date.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).
Merge branch 'mirroring' into artifacts-mode68 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(),
The statement is a month at a time, a line per kind of charge72 };
Merge branch 'claude/cool-newton-7nm8r9' (#36): real GitHub Actions on g1t, managed deploy workflows, g1t/deploy and the Sysbox runner 0.2.0. The merge keeps both sides' webhook events and API operations, says agent images need tar, guards the job engine's stop to Unix so the runner builds on Windows again, and keeps the moves tests' previews from going idle as the calendar passes their fixture's date.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)
Merge branch 'mirroring' into artifacts-mode86 }
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());
The statement is a month at a time, a line per kind of charge90 let mut files = Vec::new();
Merge branch 'mirroring' into artifacts-mode91 let mut found_head = None;
92 for folder in folders {
93 let tree: Outcome<TreeView> = g1t_kit::call(
The statement is a month at a time, a line per kind of charge94 &self.repos,
Merge branch 'mirroring' into artifacts-mode95 "tree",
96 &TreeArgs {
The statement is a month at a time, a line per kind of charge97 path: path.clone(),
98 viewer: viewer.clone(),
Merge branch 'mirroring' into artifacts-mode99 git_ref: git_ref.map(str::to_owned),
100 tree_path: (*folder).to_owned(),
The statement is a month at a time, a line per kind of charge101 },
102 )
103 .await?;
Merge branch 'mirroring' into artifacts-mode104 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 }
The statement is a month at a time, a line per kind of charge132 }
133 }
Merge branch 'mirroring' into artifacts-mode134 Ok(Read { files, head: found_head })
The statement is a month at a time, a line per kind of charge135 }
136
137 /// Keeps the `workflows` table in step with the default branch.
138 pub async fn sync(&self, repo: &Repo, actor: &User) -> Result<()> {
Merge branch 'mirroring' into artifacts-mode139 let github = crate::mirrored::policy(repo.mirror.as_ref(), false).github;
140 let read = self.read_for(repo, actor, None, github).await?;
The statement is a month at a time, a line per kind of charge141 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
Fewer Artifacts reads: the store is asked for a handle only when needed, objects are kept in the isolate, and issue events read no workflows nobody listens for192 .prepare("UPDATE workflows SET crons = '[]', error = ? WHERE id = ?")
193 .bind(&[GONE.into(), row.id.as_str().into()])?,
The statement is a month at a time, a line per kind of charge194 );
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
Fewer Artifacts reads: the store is asked for a handle only when needed, objects are kept in the isolate, and issue events read no workflows nobody listens for205 /// 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
The statement is a month at a time, a line per kind of charge226 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}
Fewer Artifacts reads: the store is asked for a handle only when needed, objects are kept in the isolate, and issue events read no workflows nobody listens for265
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}

This file's history is long; its oldest lines are credited to the oldest commit read.