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
1,552 linesCodeBlameRaw
1//! What starts a run: an event on the bus, a schedule, or someone running a
2//! workflow by hand. Each finds the workflows that want it, at the commit
3//! the event is about, and checks their filters.
4
5use g1t_actions::events::{RunInfo, github_events};
6use g1t_actions::workflow::{self, Trigger, Workflow};
7use g1t_contracts::access::{self, Capability};
8use g1t_contracts::actions::{DispatchArgs, RepositoryDispatchArgs, WorkflowRun};
9use g1t_contracts::events::{Event, caused_by_job};
10use g1t_contracts::identity::{AGENT_ID, AGENT_NAME, UsernamesArgs};
11use g1t_contracts::repos::{Commit, CompareArgs, Comparison, LogArgs, Repo, RepoPath};
12use g1t_contracts::work::{IssueDetail, PullDetail, ViewArgs};
13use g1t_contracts::{FailureCode, Outcome, User, new_id};
14use g1t_contracts::time::rfc3339;
15use g1t_kit::now_ms;
16use serde_json::{Map, Value, json};
17use worker::Result;
18
19use crate::plan::NewRun;
20use crate::sync::{Read, WorkflowRow};
21use crate::{API, Actions, SITE, check, fail, payload};
22
23/// How many commits a push's log reads: its `commits`, and how far back
24/// `before` is looked for to tell a forced push.
25const PUSH_LOG: u32 = 20;
26
27/// What an event is about, worked out once for every workflow it starts.
28struct Subject {
29 /// Where the workflow files are read, and at which commit.
30 source: RepoPath,
31 source_ref: Option<String>,
32 git_ref: String,
33 sha: String,
34 head_ref: Option<String>,
35 base_ref: Option<String>,
36 pull: Option<u32>,
37 /// The branch or tag for `branches`/`tags` filters; for pull requests,
38 /// the branch they merge into.
39 filter_ref: String,
40 /// The files it changes, for `paths` filters; `None` until needed.
41 paths: Option<Vec<String>>,
42 /// For a push, what to compare to find the files.
43 compare: Option<(Option<String>, String)>,
44 payload: Value,
45 title: String,
46 trusted: bool,
47 /// Why its runs wait for approval first: a pull request from outside,
48 /// by the repository's approval policy (protection.rs).
49 approval: Option<String>,
50}
51
52/// Whether a deployment's `ref` is a commit's full hash rather than a
53/// branch or tag.
54fn is_commit(name: &str) -> bool {
55 name.len() == 40 && name.chars().all(|c| c.is_ascii_hexdigit())
56}
57
58/// What a ref made or deleted is, as `create` and `delete` name it:
59/// `branch` or `tag`, and its short name. `None` for any other ref, and
60/// for the merge queue's own branches, which come and go with every
61/// queued change.
62fn branch_or_tag(git_ref: &str) -> Option<(&'static str, &str)> {
63 if git_ref.starts_with("refs/heads/g1t-queue/") {
64 return None;
65 }
66 match (git_ref.strip_prefix("refs/heads/"), git_ref.strip_prefix("refs/tags/")) {
67 (Some(branch), _) => Some(("branch", branch)),
68 (_, Some(tag)) => Some(("tag", tag)),
69 _ => None,
70 }
71}
72
73/// Events GitHub takes no `branches`, `tags` or `paths` filters on.
74fn unfiltered(event_name: &str) -> bool {
75 matches!(event_name, "create" | "delete")
76}
77
78/// Whether an `*.assigned` event added nobody: g1t says so for a change
79/// that only took people off.
80fn nobody_added(data: &Value) -> bool {
81 data["added"].as_array().is_some_and(Vec::is_empty)
82}
83
84/// Whether whoever a pull request is for is trusted without asking
85/// identity: g1t's agent in work nobody asked it for, or someone whose
86/// role here is known to allow pushing.
87fn trusted_outright(owner: &User, repo: &Repo) -> bool {
88 owner.id == AGENT_ID || access::can(Some(owner), repo, Capability::Push)
89}
90
91impl Actions {
92 async fn username(&self, id: Option<&str>) -> Result<Option<String>> {
93 let Some(id) = id else { return Ok(None) };
94 if id == AGENT_ID {
95 return Ok(Some(AGENT_NAME.to_owned()));
96 }
97 let names: std::collections::HashMap<String, String> =
98 g1t_kit::call(&self.identity, "usernames", &UsernamesArgs { ids: vec![id.to_owned()] }).await?;
99 Ok(names.get(id).cloned())
100 }
101
102 /// Whether whoever a pull request is for (Pull::owner: whoever asked
103 /// g1t for it, or its author) could push to the repository, so its
104 /// runs get the secrets and a token. Anyone else's, a reader's included
105 /// (who may open one on a private repository too), runs without them.
106 /// A change g1t made for someone is trusted as they are.
107 async fn insider(&self, owner: &User, repo: &Repo, ws: &User) -> Result<bool> {
108 if trusted_outright(owner, repo) {
109 return Ok(true);
110 }
111 // Stored authors carry no memberships or grants: ask identity, as
112 // the workspace (which may see anyone's permission).
113 let permission: Outcome<access::PermissionInfo> = g1t_kit::call(
114 &self.identity,
115 "collaborator_permission",
116 &access::CollaboratorPermissionArgs {
117 viewer: Some(ws.clone()),
118 path: RepoPath { namespace: repo.namespace.clone(), name: repo.name.clone() },
119 username: owner.username.clone(),
120 },
121 )
122 .await?;
123 Ok(permission
124 .into_result()
125 .ok()
126 .and_then(|info| info.role)
127 .is_some_and(|role| access::allows(role, Capability::Push)))
128 }
129
130 async fn commits(&self, repo: &Repo, actor: &User, after: &str, before: Option<&str>) -> Result<Vec<Commit>> {
131 Ok(self.pushed_commits(repo, actor, after, before).await?.0)
132 }
133
134 /// The commits a push added, oldest first, and whether it was forced:
135 /// `before` is not in `after`'s history. Known only when that history
136 /// is read whole (it is shorter than the [`PUSH_LOG`] commits read);
137 /// a longer one says not forced.
138 async fn pushed_commits(&self, repo: &Repo, actor: &User, after: &str, before: Option<&str>) -> Result<(Vec<Commit>, bool)> {
139 let log: Outcome<Vec<Commit>> = g1t_kit::call(
140 &self.repos,
141 "log",
142 &LogArgs {
143 path: RepoPath {
144 namespace: repo.namespace.clone(),
145 name: repo.name.clone(),
146 },
147 viewer: Some(actor.clone()),
148 git_ref: Some(after.to_owned()),
149 limit: PUSH_LOG,
150 },
151 )
152 .await?;
153 let mut commits: Vec<Commit> = log.into_result().unwrap_or_default();
154 let whole = commits.len() < PUSH_LOG as usize;
155 let mut forced = false;
156 if let Some(before) = before.filter(|before| *before != payload::ZERO) {
157 match commits.iter().position(|commit| commit.hash == before) {
158 Some(at) => commits.truncate(at),
159 None => forced = whole && !commits.is_empty(),
160 }
161 }
162 // GitHub lists them oldest first, with the head commit last.
163 commits.reverse();
164 Ok((commits, forced))
165 }
166
167 /// The message of the commit `sha` in `path`, when it can be read.
168 async fn head_message(&self, path: &RepoPath, actor: &User, sha: &str) -> Result<Option<String>> {
169 Ok(self.commit_at(path, actor, sha).await?.map(|commit| commit.message))
170 }
171
172 /// The commit `sha` in `path`, when it can be read.
173 async fn commit_at(&self, path: &RepoPath, actor: &User, sha: &str) -> Result<Option<Commit>> {
174 let log: Outcome<Vec<Commit>> = g1t_kit::call(
175 &self.repos,
176 "log",
177 &LogArgs { path: path.clone(), viewer: Some(actor.clone()), git_ref: Some(sha.to_owned()), limit: 1 },
178 )
179 .await?;
180 Ok(log.into_result().ok().and_then(|commits| commits.into_iter().next()))
181 }
182
183 async fn changed_paths(&self, repo: &Repo, actor: &User, base: Option<String>, head: String) -> Result<Vec<String>> {
184 let compared: Outcome<Comparison> = g1t_kit::call(
185 &self.repos,
186 "compare",
187 &CompareArgs {
188 repo_id: repo.id.clone(),
189 viewer: Some(actor.clone()),
190 base,
191 head: Some(head),
192 base_branch: None,
193 },
194 )
195 .await?;
196 Ok(compared.into_result().map(|c| c.files.into_iter().map(|f| f.path).collect()).unwrap_or_default())
197 }
198
199 async fn default_head(&self, repo: &Repo) -> Result<Option<String>> {
200 self.branch_head(repo, &repo.default_branch).await
201 }
202
203 /// The commit `branch` points to now.
204 async fn branch_head(&self, repo: &Repo, branch: &str) -> Result<Option<String>> {
205 g1t_kit::call(
206 &self.repos,
207 "head",
208 &g1t_contracts::repos::HeadArgs {
209 repo_id: repo.id.clone(),
210 branch: branch.to_owned(),
211 },
212 )
213 .await
214 }
215
216 /// Whether `name` is one of the repository's branches.
217 async fn is_branch(&self, repo: &Repo, ws: &User, name: &str) -> Result<bool> {
218 let branches: Outcome<Vec<g1t_contracts::repos::Branch>> = g1t_kit::call(
219 &self.repos,
220 "branches",
221 &g1t_contracts::repos::BranchesArgs {
222 path: Self::repo_path(repo),
223 viewer: Some(ws.clone()),
224 },
225 )
226 .await?;
227 Ok(branches.into_result().unwrap_or_default().iter().any(|branch| branch.name == name))
228 }
229
230 fn repo_path(repo: &Repo) -> RepoPath {
231 RepoPath {
232 namespace: repo.namespace.clone(),
233 name: repo.name.clone(),
234 }
235 }
236
237 /// The subject of an event of `kind`, as GitHub's `event_name`.
238 async fn subject(&self, event: &Event, event_name: &str, action: Option<&str>, repo: &Repo, ws: &User, sender: &str) -> Result<Option<Subject>> {
239 let path = Self::repo_path(repo);
240 let data = &event.data;
241 let on_default = |sha: String, payload: Value, title: String, pull: Option<u32>| Subject {
242 source: path.clone(),
243 source_ref: None,
244 git_ref: format!("refs/heads/{}", repo.default_branch),
245 sha,
246 head_ref: None,
247 base_ref: None,
248 pull,
249 filter_ref: format!("refs/heads/{}", repo.default_branch),
250 paths: None,
251 compare: None,
252 payload,
253 title,
254 trusted: true,
255 approval: None,
256 };
257 let view = |number: u32| ViewArgs {
258 repo: path.clone(),
259 number,
260 viewer: Some(ws.clone()),
261 after_seq: 0,
262 };
263 Ok(match event_name {
264 "push" => {
265 let (Some(git_ref), Some(after)) = (data["ref"].as_str(), data["after"].as_str()) else {
266 return Ok(None);
267 };
268 // The merge queue's states run merge_group workflows, not push ones.
269 if git_ref.starts_with("refs/heads/g1t-queue/") {
270 return Ok(None);
271 }
272 let before = data["before"].as_str();
273 let (commits, forced) = self.pushed_commits(repo, ws, after, before).await?;
274 let title = commits.last().map(|c| c.message.lines().next().unwrap_or_default().to_owned()).unwrap_or_default();
275 let payload = payload::push(repo, git_ref, before, after, &commits, sender, forced);
276 Some(Subject {
277 source: path.clone(),
278 source_ref: Some(after.to_owned()),
279 git_ref: git_ref.to_owned(),
280 sha: after.to_owned(),
281 head_ref: None,
282 base_ref: None,
283 pull: None,
284 filter_ref: git_ref.to_owned(),
285 paths: None,
286 compare: Some((before.map(str::to_owned), after.to_owned())),
287 payload,
288 title,
289 trusted: true,
290 approval: None,
291 })
292 }
293 // A branch or tag was made: its own commit, as on GitHub.
294 "create" => {
295 let (Some(git_ref), Some(after)) = (data["ref"].as_str(), data["after"].as_str()) else {
296 return Ok(None);
297 };
298 if git_ref.starts_with("refs/heads/g1t-queue/") || !data["before"].is_null() {
299 return Ok(None);
300 }
301 let Some((ref_type, name)) = branch_or_tag(git_ref) else { return Ok(None) };
302 let payload = json!({
303 "ref": name,
304 "ref_type": ref_type,
305 "master_branch": repo.default_branch,
306 "description": repo.description,
307 "pusher_type": "user",
308 "repository": payload::repository(repo),
309 "sender": payload::user(sender),
310 });
311 Some(Subject {
312 source: path.clone(),
313 source_ref: Some(after.to_owned()),
314 git_ref: git_ref.to_owned(),
315 sha: after.to_owned(),
316 head_ref: None,
317 base_ref: None,
318 pull: None,
319 filter_ref: git_ref.to_owned(),
320 paths: None,
321 compare: None,
322 payload,
323 title: format!("Created {ref_type} {name}"),
324 trusted: true,
325 approval: None,
326 })
327 }
328 // A branch or tag was deleted: on the default branch's latest
329 // commit, as on GitHub, since what was deleted has no commit to
330 // run on now.
331 "delete" => {
332 let Some(git_ref) = data["ref"].as_str() else { return Ok(None) };
333 let Some((ref_type, name)) = branch_or_tag(git_ref) else { return Ok(None) };
334 let Some(sha) = self.default_head(repo).await? else { return Ok(None) };
335 let payload = json!({
336 "ref": name,
337 "ref_type": ref_type,
338 "pusher_type": "user",
339 "repository": payload::repository(repo),
340 "sender": payload::user(sender),
341 });
342 Some(on_default(sha, payload, format!("Deleted {ref_type} {name}"), None))
343 }
344 "pull_request" | "pull_request_target" | "pull_request_review" => {
345 let Some(number) = data["number"].as_u64().map(|n| n as u32) else { return Ok(None) };
346 // Assignees changed but nobody was added: someone was only
347 // taken off, which is not `assigned`.
348 if action == Some("assigned") && nobody_added(data) {
349 return Ok(None);
350 }
351 let detail: Outcome<PullDetail> = g1t_kit::call(&self.work, "get_pull", &view(number)).await?;
352 let Outcome::Ok(detail) = detail else { return Ok(None) };
353 let pull = &detail.pull;
354 let base_ref = pull.base_branch(&repo.default_branch).to_owned();
355 let mut payload = json!({
356 "action": action,
357 "number": pull.number,
358 "pull_request": payload::pull(repo, pull),
359 "repository": payload::repository(repo),
360 "sender": payload::user(sender),
361 });
362 payload::changed(repo, &mut payload, data);
363 // `base.sha` is the base branch's tip now, as GitHub gives
364 // it, not where the two last agreed.
365 if let Some(tip) = self.branch_head(repo, &base_ref).await? {
366 payload["pull_request"]["base"]["sha"] = json!(tip);
367 }
368 // A merge's commit, once it has landed.
369 if pull.status == g1t_contracts::work::PullStatus::Merged
370 && let Some(commit) = data["commit"].as_str()
371 {
372 payload["pull_request"]["merge_commit_sha"] = json!(commit);
373 }
374 if event_name == "pull_request_review" {
375 let review = detail.comments.iter().rev().find(|c| c.verdict.is_some() && !c.advisory);
376 payload["review"] = payload::review(repo, pull, review);
377 }
378 let trusted = self.insider(pull.owner(), repo, ws).await?;
379 // A pull request from outside may wait for approval before
380 // its head's code runs. `pull_request_target` runs the
381 // base's code, and a merged one's run the commit it landed
382 // as, so neither waits.
383 let approval = if event_name != "pull_request_target" && action != Some("closed") {
384 self.approval_needed(repo, pull.owner(), ws).await?
385 } else {
386 None
387 };
388 let head_ref = payload::head_ref(pull);
389 if event_name == "pull_request_target" {
390 // In the base's context: its workflows, its head.
391 let Some(sha) = self.default_head(repo).await? else { return Ok(None) };
392 let mut subject = on_default(sha, payload, pull.title.clone(), Some(pull.number));
393 subject.head_ref = Some(head_ref);
394 subject.base_ref = Some(base_ref.clone());
395 subject.filter_ref = format!("refs/heads/{base_ref}");
396 subject.paths = Some(pull.files.iter().map(|f| f.path.clone()).collect());
397 return Ok(Some(subject));
398 }
399 // A merged pull request's run is on the commit it landed as,
400 // in the repository; otherwise on its head, where that is.
401 let landed = match (action, data["commit"].as_str()) {
402 (Some("closed"), Some(commit)) => Some(commit.to_owned()),
403 _ => None,
404 };
405 let sha = match (&landed, data["commit"].as_str(), &pull.head_commit) {
406 (Some(commit), _, _) => commit.clone(),
407 (None, Some(commit), _) => commit.to_owned(),
408 (None, None, Some(head)) => head.clone(),
409 (None, None, None) => return Ok(None),
410 };
411 let source = match landed {
412 Some(_) => path.clone(),
413 None => pull.fork.clone().unwrap_or_else(|| path.clone()),
414 };
415 Some(Subject {
416 source,
417 source_ref: Some(sha.clone()),
418 git_ref: format!("refs/pull/{}/merge", pull.number),
419 sha,
420 head_ref: Some(head_ref),
421 base_ref: Some(base_ref.clone()),
422 pull: Some(pull.number),
423 // `branches` filters on pull requests name the base.
424 filter_ref: format!("refs/heads/{base_ref}"),
425 paths: Some(pull.files.iter().map(|f| f.path.clone()).collect()),
426 compare: None,
427 payload,
428 title: pull.title.clone(),
429 trusted,
430 approval,
431 })
432 }
433 "workflow_run" => {
434 // A run of a workflow_run workflow does not start another,
435 // so two such workflows cannot set each other off.
436 if data["event"].as_str() == Some("workflow_run") {
437 return Ok(None);
438 }
439 let Some(sha) = self.default_head(repo).await? else { return Ok(None) };
440 let head_branch = data["ref"].as_str().unwrap_or_default().trim_start_matches("refs/heads/").to_owned();
441 let name = data["workflow"].as_str().unwrap_or_default();
442 // The followed run's row, for what the event does not carry.
443 let row = match data["runId"].as_str() {
444 Some(id) => self.db.prepare("SELECT * FROM runs WHERE id = ?").bind(&[id.into()])?.first::<crate::plan::RunRow>(None).await?,
445 None => None,
446 };
447 let head_commit = match data["sha"].as_str() {
448 Some(head) => self.commit_at(&path, ws, head).await?.filter(|commit| commit.hash == head),
449 None => None,
450 };
451 let followed = payload::FollowedRun {
452 data,
453 workflow_id: row.as_ref().map(|r| r.workflow_id.as_str()),
454 attempt: row.as_ref().map(|r| r.attempt),
455 created_at: row.as_ref().map(|r| r.created_at.as_str()),
456 started_at: row.as_ref().and_then(|r| r.started_at.as_deref()),
457 updated_at: row.as_ref().and_then(|r| r.finished_at.as_deref()),
458 actor: row.as_ref().and_then(|r| r.actor.as_deref()),
459 triggering_actor: row.as_ref().and_then(|r| r.triggering_actor.as_deref()),
460 title: row.as_ref().map(|r| r.title.as_str()),
461 head_commit: head_commit.as_ref(),
462 };
463 let payload = json!({
464 "action": "completed",
465 "workflow_run": payload::workflow_run(repo, &followed),
466 "workflow": {
467 "id": followed.workflow_id,
468 "node_id": followed.workflow_id,
469 "name": name,
470 "path": data["path"],
471 "state": "active",
472 "html_url": format!("{SITE}/{}/{}/actions", repo.namespace, repo.name),
473 },
474 "repository": payload::repository(repo),
475 "sender": payload::user(sender),
476 });
477 let mut subject = on_default(sha, payload, format!("After {name}"), None);
478 // Branch filters apply to the branch the followed run was on.
479 subject.filter_ref = format!("refs/heads/{head_branch}");
480 Some(subject)
481 }
482 "issues" | "issue_comment" => {
483 let Some(number) = data["number"].as_u64().map(|n| n as u32) else { return Ok(None) };
484 if action == Some("assigned") && nobody_added(data) {
485 return Ok(None);
486 }
487 let Some(sha) = self.default_head(repo).await? else { return Ok(None) };
488 let issue: Outcome<IssueDetail> = g1t_kit::call(&self.work, "get_issue", &view(number)).await?;
489 let (issue_json, comments, title, on_pull) = match issue {
490 Outcome::Ok(detail) => (payload::issue(repo, &detail.issue), detail.comments, detail.issue.title.clone(), false),
491 Outcome::Fail(_) => {
492 let pull: Outcome<PullDetail> = g1t_kit::call(&self.work, "get_pull", &view(number)).await?;
493 let Outcome::Ok(detail) = pull else { return Ok(None) };
494 (payload::pull_as_issue(repo, &detail.pull), detail.comments, detail.pull.title.clone(), true)
495 }
496 };
497 let mut payload = json!({
498 "action": action,
499 "issue": issue_json,
500 "repository": payload::repository(repo),
501 "sender": payload::user(sender),
502 });
503 payload::changed(repo, &mut payload, data);
504 if event_name == "issue_comment" {
505 let comment_id = data["commentId"].as_str();
506 if action == Some("deleted") {
507 // Gone by now: as the event kept it.
508 match data.get("comment").filter(|kept| kept.is_object()) {
509 Some(kept) => payload["comment"] = payload::deleted_comment(repo, number, kept, on_pull),
510 None => return Ok(None),
511 }
512 } else {
513 let comment = comments.iter().find(|c| Some(c.id.as_str()) == comment_id);
514 // A new comment is the newest; an edited one must be found.
515 let comment = if action == Some("created") { comment.or(comments.last()) } else { comment };
516 match comment {
517 Some(comment) => payload["comment"] = payload::comment(repo, number, comment, on_pull),
518 None => return Ok(None),
519 }
520 }
521 // `edited`: what the body was before.
522 if let Some(changes) = data.get("changes").filter(|changes| changes.is_object()) {
523 payload["changes"] = changes.clone();
524 }
525 }
526 Some(on_default(sha, payload, title, on_pull.then_some(number)))
527 }
528 // A release: on its tag, at the commit the tag named.
529 "release" => {
530 let release = &data["release"];
531 let Some(tag) = data["tagName"].as_str().or_else(|| release["tagName"].as_str()) else { return Ok(None) };
532 let sha = match release["target"].as_str().filter(|target| !target.is_empty()) {
533 Some(target) => target.to_owned(),
534 None => match self.default_head(repo).await? {
535 Some(head) => head,
536 None => return Ok(None),
537 },
538 };
539 let git_ref = format!("refs/tags/{tag}");
540 let mut payload = json!({
541 "action": action,
542 "release": payload::release(repo, release),
543 "repository": payload::repository(repo),
544 "sender": payload::user(sender),
545 });
546 if let Some(changes) = data.get("changes").filter(|changes| changes.is_object()) {
547 payload["changes"] = changes.clone();
548 }
549 let name = release["name"].as_str().filter(|name| !name.is_empty()).unwrap_or(tag);
550 Some(Subject {
551 source: path.clone(),
552 source_ref: Some(sha.clone()),
553 git_ref: git_ref.clone(),
554 sha,
555 head_ref: None,
556 base_ref: None,
557 pull: None,
558 filter_ref: git_ref,
559 paths: None,
560 compare: None,
561 payload,
562 title: format!("Release {name} {}", action.unwrap_or("changed")),
563 trusted: true,
564 approval: None,
565 })
566 }
567 // A deployment, or a new status of one: at the commit deployed,
568 // on the branch or tag it names (none for a bare commit).
569 "deployment" | "deployment_status" => {
570 let deployment = &data["deployment"];
571 let Some(sha) = deployment["sha"].as_str().filter(|sha| !sha.is_empty()).map(str::to_owned) else { return Ok(None) };
572 let named = deployment["ref"].as_str().unwrap_or_default();
573 let git_ref = if named.is_empty() || named == sha || is_commit(named) {
574 String::new()
575 } else if named.starts_with("refs/") {
576 named.to_owned()
577 } else if self.is_branch(repo, ws, named).await? {
578 format!("refs/heads/{named}")
579 } else {
580 format!("refs/tags/{named}")
581 };
582 let environment = deployment["environment"].as_str().unwrap_or_default();
583 let mut payload = json!({
584 "action": "created",
585 "deployment": payload::deployment(repo, deployment),
586 "repository": payload::repository(repo),
587 "sender": payload::user(sender),
588 });
589 let title = if event_name == "deployment_status" {
590 let status = &data["deploymentStatus"];
591 payload["deployment_status"] = payload::deployment_status(repo, status, deployment);
592 format!("Deployment to {environment}: {}", status["state"].as_str().unwrap_or("changed"))
593 } else {
594 format!("Deployment to {environment}")
595 };
596 Some(Subject {
597 source: path.clone(),
598 source_ref: Some(sha.clone()),
599 filter_ref: git_ref.clone(),
600 git_ref,
601 sha,
602 head_ref: None,
603 base_ref: None,
604 pull: None,
605 paths: None,
606 compare: None,
607 payload,
608 title,
609 trusted: true,
610 approval: None,
611 })
612 }
613 _ => None,
614 })
615 }
616
617 /// A push keeps the repository's schedules going (`run_schedules`).
618 /// Written at most once a day, and only on scheduled workflows.
619 async fn note_push(&self, repo_id: &str) -> Result<()> {
620 let at = now_ms();
621 self.db
622 .prepare("UPDATE workflows SET pushed_at = ? WHERE repo_id = ? AND crons != '[]' AND (pushed_at IS NULL OR pushed_at < ?)")
623 .bind(&[rfc3339(at).into(), repo_id.into(), rfc3339(at.saturating_sub(24 * 60 * 60 * 1000)).into()])?
624 .run()
625 .await?;
626 Ok(())
627 }
628
629 pub async fn on_event(&self, event: &Event) -> Result<()> {
630 let Some(repo_id) = event.repo_id.as_deref() else { return Ok(()) };
631 let mut mapped = github_events(&event.kind);
632 // A new branch or tag is also `create`.
633 if event.kind == "git.push" && event.data["before"].is_null() {
634 mapped.push(("create", None));
635 }
636 let pushed_default = event.kind == "git.push" && event.data["defaultBranch"].as_bool() == Some(true);
637 if event.kind == "git.push" {
638 self.note_push(repo_id).await?;
639 }
640 if mapped.is_empty() && !pushed_default {
641 return Ok(());
642 }
643 let Some((repo, ws)) = self.repo_by_id(repo_id).await? else { return Ok(()) };
644 if pushed_default {
645 self.sync(&repo, &ws).await?;
646 }
647 // A mirror runs only what its state says (mirrored.rs).
648 let copied_in = event.kind == "git.push" && event.data["mirrored"].as_bool() == Some(true);
649 let policy = crate::mirrored::policy(repo.mirror.as_ref(), copied_in);
650 if !policy.runs {
651 return Ok(());
652 }
653 // What a workflow job's own token did starts no workflows, as on
654 // GitHub, so a workflow cannot set itself off; only
655 // `workflow_dispatch` and `repository_dispatch` do.
656 if let Some(run) = caused_by_job(&event.data) {
657 worker::console_log!("actions: {} {} came from run {run}'s token; no workflows start for it", event.kind, event.id);
658 return Ok(());
659 }
660 let sender = self.username(event.actor.as_deref()).await?.unwrap_or_else(|| repo.namespace.clone());
661 for (event_name, action) in mapped {
662 // Issues and comments start the default branch's workflows,
663 // which the synced table lists: when none listens, nothing is
664 // read from git. Agents make many of these events.
665 // Deployments' statuses are as frequent (every g1t.page build
666 // reports several), so they look there first too.
667 if matches!(event_name, "issues" | "issue_comment" | "deployment" | "deployment_status" | "delete")
668 && self.listens(repo_id, event_name).await? == Some(false)
669 {
670 continue;
671 }
672 let Some(mut subject) = self.subject(event, event_name, action, &repo, &ws, &sender).await? else {
673 continue;
674 };
675 // `[skip ci]` in the head commit's message, as on GitHub.
676 if matches!(event_name, "push" | "pull_request") {
677 let message = match subject.payload["head_commit"]["message"].as_str() {
678 Some(message) => Some(message.to_owned()),
679 None => self.head_message(&subject.source, &ws, &subject.sha).await?,
680 };
681 if message.as_deref().is_some_and(g1t_actions::events::skips_ci) {
682 worker::console_log!("actions: {} {} at {} asks to skip its runs", event.kind, event.id, subject.sha);
683 continue;
684 }
685 }
686 // A change to workflows made on the remote, by someone g1t
687 // knows nothing of, waits before it may use this repository's
688 // secrets.
689 if copied_in && matches!(event_name, "push" | "create") {
690 if subject.paths.is_none() {
691 let (base, head) = subject.compare.clone().unwrap_or((None, subject.sha.clone()));
692 subject.paths = Some(self.changed_paths(&repo, &ws, base, head).await?);
693 }
694 if crate::mirrored::touches_workflows(subject.paths.as_deref().unwrap_or_default())
695 && let Some(mirror) = &repo.mirror
696 {
697 subject.approval = subject.approval.take().or_else(|| Some(crate::mirrored::copied_workflows(&mirror.remote)));
698 }
699 }
700 // A pull request from a fork reads the fork's files.
701 let read = if subject.source == Self::repo_path(&repo) {
702 self.read_for(&repo, &ws, subject.source_ref.as_deref(), policy.github).await?
703 } else {
704 // And this repository's managed workflows: they are its own,
705 // whatever the fork's files say (managed.rs).
706 let mut read = self.read_workflows(&subject.source, &ws, subject.source_ref.as_deref()).await?;
707 let managed = self.managed_beside(&repo.id, &read.files).await?;
708 read.files.extend(managed);
709 read
710 };
711 // A pull request's head runs each workflow once, however many
712 // events say it is there (marked ready, and pushed).
713 let key = match subject.pull {
714 Some(number) if event_name.starts_with("pull_request") && event_name != "pull_request_review" => {
715 // Reopened or made a draft again runs anew, at a head
716 // that may have run before.
717 let phase = match action {
718 Some("closed") => "closed".to_owned(),
719 Some(again @ ("reopened" | "converted_to_draft")) => format!("{again}:{}", event.id),
720 _ => "open".to_owned(),
721 };
722 format!("{event_name}:{number}:{}:{phase}", subject.sha)
723 }
724 _ if event_name == "create" => format!("{}:create", event.id),
725 _ => event.id.clone(),
726 };
727 self.start_matching(&repo, &ws, read, &mut subject, event_name, action, &key, event.actor.as_deref(), &sender)
728 .await?;
729 }
730 Ok(())
731 }
732
733 #[allow(clippy::too_many_arguments)]
734 async fn start_matching(
735 &self,
736 repo: &Repo,
737 ws: &User,
738 read: Read,
739 subject: &mut Subject,
740 event_name: &str,
741 action: Option<&str>,
742 event_key: &str,
743 actor_id: Option<&str>,
744 sender: &str,
745 ) -> Result<()> {
746 for file in read.files {
747 let parsed = workflow::parse(&file.source);
748 let workflow = match parsed {
749 Ok(workflow) => workflow,
750 Err(problem) => {
751 // A push shows a broken workflow as a failed run, as GitHub does.
752 if event_name == "push" && file.source.contains("on") {
753 self.record_invalid(repo, &file.path, &file.source, subject, event_key, actor_id, sender, &problem)
754 .await?;
755 }
756 continue;
757 }
758 };
759 let Some(trigger) = workflow.trigger(event_name) else { continue };
760 // workflow_run follows the workflows it names.
761 if event_name == "workflow_run" {
762 let followed = subject.payload["workflow_run"]["name"].as_str().unwrap_or_default();
763 if !trigger.workflows.iter().any(|name| name == followed) {
764 continue;
765 }
766 }
767 if !trigger.wants_type(action) || !self.passes(repo, ws, trigger, subject, event_name).await? {
768 continue;
769 }
770 if self.disabled(&repo.id, &file.path).await? {
771 continue;
772 }
773 // On a mirror the remote may deploy too: a workflow that deploys
774 // waits for approval (mirrored.rs).
775 let held = crate::mirrored::policy(repo.mirror.as_ref(), false)
776 .hold
777 .zip(crate::mirrored::deploys_to(&workflow))
778 .map(|(remote, environment)| crate::mirrored::held(&remote, &environment));
779 self.create_run(NewRun {
780 repo: repo.clone(),
781 path: file.path,
782 source: file.source,
783 info: self.run_info(repo, &workflow, event_name, subject, sender, actor_id),
784 workflow,
785 action: action.map(str::to_owned),
786 pull: subject.pull,
787 title: subject.title.clone(),
788 inputs: Map::new(),
789 event_key: event_key.to_owned(),
790 actor_id: actor_id.map(str::to_owned),
791 actor: Some(sender.to_owned()),
792 trusted: subject.trusted,
793 approval: subject.approval.clone().or(held),
794 })
795 .await?;
796 }
797 Ok(())
798 }
799
800 /// Whether the branch, tag and path filters let the event through.
801 async fn passes(&self, repo: &Repo, ws: &User, trigger: &Trigger, subject: &mut Subject, event_name: &str) -> Result<bool> {
802 // GitHub takes no filters on these: a workflow with `branches` on
803 // `create` or `delete` still runs for every branch and tag.
804 if unfiltered(event_name) {
805 return Ok(true);
806 }
807 let git_ref = subject.filter_ref.as_str();
808 if let Some(tag) = git_ref.strip_prefix("refs/tags/") {
809 // A tag push runs a workflow that filters tags, or filters nothing.
810 if trigger.tags.is_set() {
811 if !trigger.tags.allows(tag) {
812 return Ok(false);
813 }
814 } else if trigger.branches.is_set() {
815 return Ok(false);
816 }
817 // Paths are not checked for tags, as on GitHub.
818 return Ok(true);
819 }
820 let branch = git_ref.strip_prefix("refs/heads/").unwrap_or(git_ref);
821 if trigger.branches.is_set() {
822 if !trigger.branches.allows(branch) {
823 return Ok(false);
824 }
825 } else if event_name == "push" && trigger.tags.is_set() {
826 return Ok(false);
827 }
828 if trigger.paths.is_set() {
829 if subject.paths.is_none() {
830 let (base, head) = subject.compare.clone().unwrap_or((None, subject.sha.clone()));
831 subject.paths = Some(self.changed_paths(repo, ws, base, head).await?);
832 }
833 if !trigger.paths.allows_paths(subject.paths.as_deref().unwrap_or_default()) {
834 return Ok(false);
835 }
836 }
837 Ok(true)
838 }
839
840 async fn disabled(&self, repo_id: &str, path: &str) -> Result<bool> {
841 let row = self
842 .db
843 .prepare("SELECT * FROM workflows WHERE repo_id = ? AND path = ?")
844 .bind(&[repo_id.into(), path.into()])?
845 .first::<WorkflowRow>(None)
846 .await?;
847 Ok(row.is_some_and(|row| row.state == "disabled"))
848 }
849
850 fn run_info(&self, repo: &Repo, workflow: &Workflow, event_name: &str, subject: &Subject, sender: &str, actor_id: Option<&str>) -> RunInfo {
851 RunInfo {
852 repository: format!("{}/{}", repo.namespace, repo.name),
853 repository_id: repo.id.clone(),
854 default_branch: repo.default_branch.clone(),
855 event_name: event_name.to_owned(),
856 event: subject.payload.clone(),
857 git_ref: subject.git_ref.clone(),
858 sha: subject.sha.clone(),
859 head_ref: subject.head_ref.clone(),
860 base_ref: subject.base_ref.clone(),
861 actor: sender.to_owned(),
862 actor_id: actor_id.unwrap_or_default().to_owned(),
863 triggering_actor: sender.to_owned(),
864 run_id: String::new(),
865 run_number: 0,
866 run_attempt: 1,
867 workflow: workflow.name.clone().unwrap_or_default(),
868 workflow_path: String::new(),
869 server_url: SITE.to_owned(),
870 api_url: API.to_owned(),
871 }
872 }
873
874 #[allow(clippy::too_many_arguments)]
875 async fn record_invalid(
876 &self,
877 repo: &Repo,
878 path: &str,
879 source: &str,
880 subject: &Subject,
881 event_key: &str,
882 actor_id: Option<&str>,
883 sender: &str,
884 problem: &str,
885 ) -> Result<()> {
886 let row = self.workflow_row(repo, path, path, source).await?;
887 self.record_failed_run(&row, subject.git_ref.as_str(), &subject.sha, event_key, actor_id, sender, problem).await
888 }
889
890 /// Scheduled workflows that run this minute, on the default branch: at
891 /// most every five minutes (`cron::Schedule::runs_at`). Not in a
892 /// repository with no push for [`crate::SCHEDULE_IDLE_MS`], nor for an
893 /// hour after billing refused one of the workflow's scheduled jobs
894 /// ([`crate::SCHEDULE_REFUSED_MS`]): no run is made only to be refused.
895 pub async fn run_schedules(&self, minute: u64) -> Result<()> {
896 let rows = self
897 .db
898 .prepare(
899 "SELECT * FROM workflows WHERE state = 'active' AND crons != '[]' AND error IS NULL
900 AND COALESCE(pushed_at, updated_at) >= ? AND (schedule_refused_until IS NULL OR schedule_refused_until <= ?)",
901 )
902 .bind(&[rfc3339(minute.saturating_sub(crate::SCHEDULE_IDLE_MS)).into(), rfc3339(minute).into()])?
903 .all()
904 .await?
905 .results::<WorkflowRow>()?;
906 for row in rows {
907 let crons: Vec<String> = serde_json::from_str(&row.crons).unwrap_or_default();
908 let Some(cron) = crons.iter().find(|cron| g1t_actions::cron::Schedule::parse(cron).is_ok_and(|s| s.runs_at(minute))) else {
909 continue;
910 };
911 let Ok(workflow) = workflow::parse(&row.source) else { continue };
912 // Schedules wait while a repository is archived, or a mirror runs
913 // nothing; a deleted one is not found.
914 let Some((repo, _ws)) = self
915 .repo_by_id(&row.repo_id)
916 .await?
917 .filter(|(repo, _)| !repo.archived() && crate::mirrored::policy(repo.mirror.as_ref(), false).runs)
918 else {
919 continue;
920 };
921 let Some(sha) = self.default_head(&repo).await? else { continue };
922 let payload = json!({ "schedule": cron, "repository": payload::repository(&repo), "workflow": row.path });
923 let mut subject = Subject {
924 source: Self::repo_path(&repo),
925 source_ref: None,
926 git_ref: format!("refs/heads/{}", repo.default_branch),
927 sha,
928 head_ref: None,
929 base_ref: None,
930 pull: None,
931 filter_ref: String::new(),
932 paths: None,
933 compare: None,
934 payload,
935 title: format!("Scheduled: {cron}"),
936 trusted: true,
937 approval: None,
938 };
939 subject.filter_ref = subject.git_ref.clone();
940 let info = self.run_info(&repo, &workflow, "schedule", &subject, &repo.namespace, None);
941 self.create_run(NewRun {
942 repo: repo.clone(),
943 path: row.path.clone(),
944 source: row.source.clone(),
945 workflow,
946 info,
947 action: None,
948 pull: None,
949 title: subject.title.clone(),
950 inputs: Map::new(),
951 event_key: format!("schedule:{minute}"),
952 actor_id: None,
953 actor: None,
954 trusted: true,
955 approval: None,
956 })
957 .await?;
958 }
959 Ok(())
960 }
961
962 /// `dispatch`: someone with the Write role runs a workflow that has
963 /// `workflow_dispatch`.
964 pub async fn dispatch(&self, a: DispatchArgs) -> Result<Outcome<WorkflowRun>> {
965 let repo = check!(self.may(&a.actor, &a.repo, Capability::Run).await?);
966 if repo.archived() {
967 return Ok(fail(FailureCode::Forbidden, g1t_contracts::repos::archived_message(&repo.namespace, &repo.name)));
968 }
969 let Some(ws) = self.workspace_actor(&repo.namespace).await? else {
970 return Ok(fail(FailureCode::NotFound, "There is no such workspace."));
971 };
972 let git_ref = a.git_ref.clone().unwrap_or_else(|| repo.default_branch.clone());
973 let full_ref = if git_ref.starts_with("refs/") {
974 git_ref.clone()
975 } else {
976 // A branch if there is one by that name, otherwise a tag.
977 let branches: Outcome<Vec<g1t_contracts::repos::Branch>> = g1t_kit::call(
978 &self.repos,
979 "branches",
980 &g1t_contracts::repos::BranchesArgs {
981 path: Self::repo_path(&repo),
982 viewer: Some(ws.clone()),
983 },
984 )
985 .await?;
986 let is_branch = branches.into_result().unwrap_or_default().iter().any(|branch| branch.name == git_ref);
987 format!("refs/{}/{git_ref}", if is_branch { "heads" } else { "tags" })
988 };
989 let short = full_ref.trim_start_matches("refs/heads/").trim_start_matches("refs/tags/").to_owned();
990 let policy = crate::mirrored::policy(repo.mirror.as_ref(), false);
991 if let Some(mirror) = repo.mirror.as_ref().filter(|_| !policy.runs) {
992 return Ok(fail(FailureCode::Forbidden, crate::mirrored::refused(&repo.namespace, &repo.name, mirror)));
993 }
994 let read = self.read_for(&repo, &ws, Some(&short), policy.github).await?;
995 let Some(sha) = read.head.clone() else {
996 return Ok(fail(FailureCode::NotFound, format!("There is no branch or tag called {short}.")));
997 };
998 // A workflow is named by its file (`build.yml`), its path, or its id
999 // (`wfl_…`), which stands for the path it was read from.
1000 let by_id = if a.workflow.starts_with("wfl_") {
1001 self.db
1002 .prepare("SELECT * FROM workflows WHERE repo_id = ? AND id = ?")
1003 .bind(&[repo.id.as_str().into(), a.workflow.as_str().into()])?
1004 .first::<WorkflowRow>(None)
1005 .await?
1006 .map(|row| row.path)
1007 } else {
1008 None
1009 };
1010 let named = by_id.as_deref().unwrap_or(&a.workflow);
1011 let wanted = named.trim_start_matches(".g1t/workflows/");
1012 let Some(file) = read.files.iter().find(|file| {
1013 file.path.rsplit('/').next() == Some(wanted) || file.path == named
1014 }) else {
1015 return Ok(fail(FailureCode::NotFound, format!("There is no workflow {wanted} on {short}.")));
1016 };
1017 let workflow = match workflow::parse(&file.source) {
1018 Ok(workflow) => workflow,
1019 Err(problem) => return Ok(fail(FailureCode::Invalid, format!("The workflow does not read: {problem}"))),
1020 };
1021 let Some(trigger) = workflow.trigger("workflow_dispatch") else {
1022 return Ok(fail(FailureCode::Invalid, "That workflow cannot be run by hand: it has no `workflow_dispatch` trigger."));
1023 };
1024 // `type: environment` names one of the repository's environments.
1025 let environments = if trigger.inputs.values().any(|spec| spec.get("type").and_then(Value::as_str) == Some("environment")) {
1026 self.environment_list(&repo.id).await?.into_iter().map(|env| env.name).collect()
1027 } else {
1028 Vec::new()
1029 };
1030 let inputs = check!(dispatch_inputs(trigger, &a.inputs, &environments));
1031 let payload = json!({
1032 // As GitHub gives them here: text, `true` and `false` included.
1033 // The `inputs` context keeps booleans and numbers.
1034 "inputs": event_inputs(&inputs),
1035 "ref": full_ref,
1036 "repository": payload::repository(&repo),
1037 "sender": payload::user(&a.actor.username),
1038 "workflow": file.path,
1039 });
1040 let subject = Subject {
1041 source: Self::repo_path(&repo),
1042 source_ref: Some(sha.clone()),
1043 git_ref: full_ref.clone(),
1044 sha,
1045 head_ref: None,
1046 base_ref: None,
1047 pull: None,
1048 filter_ref: full_ref,
1049 paths: None,
1050 compare: None,
1051 payload,
1052 title: format!("{} run by {}", workflow.display_name(&file.path), a.actor.username),
1053 trusted: true,
1054 approval: None,
1055 };
1056 let info = self.run_info(&repo, &workflow, "workflow_dispatch", &subject, &a.actor.username, Some(&a.actor.id));
1057 let created = self
1058 .create_run(NewRun {
1059 repo: repo.clone(),
1060 path: file.path.clone(),
1061 source: file.source.clone(),
1062 workflow,
1063 info,
1064 action: None,
1065 pull: None,
1066 title: subject.title.clone(),
1067 inputs,
1068 event_key: format!("dispatch:{}", new_id("dsp", now_ms())),
1069 actor_id: Some(a.actor.id.clone()),
1070 actor: Some(a.actor.username.clone()),
1071 trusted: true,
1072 approval: None,
1073 })
1074 .await?;
1075 match created {
1076 Some(id) => self.run_summary(&id).await,
1077 None => Ok(fail(FailureCode::Conflict, "It did not start.")),
1078 }
1079 }
1080
1081 /// `repository_dispatch`: an outside event, by name, starts the default
1082 /// branch's workflows that run `on: repository_dispatch` with that type
1083 /// (or with no `types`). A workflow job's token may send one: this,
1084 /// with `workflow_dispatch`, is how a workflow starts another.
1085 pub async fn repository_dispatch(&self, a: RepositoryDispatchArgs) -> Result<Outcome<u32>> {
1086 let repo = check!(self.may(&a.actor, &a.repo, Capability::Push).await?);
1087 if repo.archived() {
1088 return Ok(fail(FailureCode::Forbidden, g1t_contracts::repos::archived_message(&repo.namespace, &repo.name)));
1089 }
1090 let event_type = a.event_type.trim().to_owned();
1091 if event_type.is_empty() || event_type.chars().count() > 100 {
1092 return Ok(fail(FailureCode::Invalid, "event_type is 1 to 100 characters."));
1093 }
1094 let client_payload = match a.client_payload {
1095 Value::Null => json!({}),
1096 Value::Object(map) if map.len() <= 10 => Value::Object(map),
1097 Value::Object(_) => return Ok(fail(FailureCode::Invalid, "client_payload has at most 10 top-level properties.")),
1098 _ => return Ok(fail(FailureCode::Invalid, "client_payload is a JSON object.")),
1099 };
1100 if serde_json::to_string(&client_payload).map_or(0, |text| text.len()) > 64 * 1024 {
1101 return Ok(fail(FailureCode::Invalid, "client_payload is at most 64 KB."));
1102 }
1103 let Some(ws) = self.workspace_actor(&repo.namespace).await? else {
1104 return Ok(fail(FailureCode::NotFound, "There is no such workspace."));
1105 };
1106 let policy = crate::mirrored::policy(repo.mirror.as_ref(), false);
1107 if let Some(mirror) = repo.mirror.as_ref().filter(|_| !policy.runs) {
1108 return Ok(fail(FailureCode::Forbidden, crate::mirrored::refused(&repo.namespace, &repo.name, mirror)));
1109 }
1110 let read = self.read_for(&repo, &ws, Some(&repo.default_branch), policy.github).await?;
1111 let Some(sha) = read.head.clone() else {
1112 return Ok(fail(FailureCode::NotFound, "The repository has no default branch to run on yet."));
1113 };
1114 let git_ref = format!("refs/heads/{}", repo.default_branch);
1115 let payload = json!({
1116 "action": event_type,
1117 "branch": repo.default_branch,
1118 "client_payload": client_payload,
1119 "repository": payload::repository(&repo),
1120 "sender": payload::user(&a.actor.username),
1121 });
1122 let key = format!("repository_dispatch:{}", new_id("dsp", now_ms()));
1123 let mut started = 0u32;
1124 for file in read.files {
1125 let Ok(workflow) = workflow::parse(&file.source) else { continue };
1126 let Some(trigger) = workflow.trigger("repository_dispatch") else { continue };
1127 if !trigger.wants_type(Some(&event_type)) || self.disabled(&repo.id, &file.path).await? {
1128 continue;
1129 }
1130 let subject = Subject {
1131 source: Self::repo_path(&repo),
1132 source_ref: Some(sha.clone()),
1133 git_ref: git_ref.clone(),
1134 sha: sha.clone(),
1135 head_ref: None,
1136 base_ref: None,
1137 pull: None,
1138 filter_ref: git_ref.clone(),
1139 paths: None,
1140 compare: None,
1141 payload: payload.clone(),
1142 title: event_type.clone(),
1143 trusted: true,
1144 approval: None,
1145 };
1146 let info = self.run_info(&repo, &workflow, "repository_dispatch", &subject, &a.actor.username, Some(&a.actor.id));
1147 let created = self
1148 .create_run(NewRun {
1149 repo: repo.clone(),
1150 path: file.path.clone(),
1151 source: file.source.clone(),
1152 workflow,
1153 info,
1154 action: Some(event_type.clone()),
1155 pull: None,
1156 title: subject.title.clone(),
1157 inputs: Map::new(),
1158 event_key: key.clone(),
1159 actor_id: Some(a.actor.id.clone()),
1160 actor: Some(a.actor.username.clone()),
1161 trusted: true,
1162 approval: None,
1163 })
1164 .await?;
1165 if created.is_some() {
1166 started += 1;
1167 }
1168 }
1169 Ok(Outcome::Ok(started))
1170 }
1171}
1172
1173#[derive(serde::Deserialize)]
1174#[serde(rename_all = "camelCase")]
1175pub struct MergeGroupArgs {
1176 pub repo_id: String,
1177 pub entry: String,
1178 pub sha: String,
1179 pub head_ref: String,
1180 #[serde(default)]
1181 pub base_sha: Option<String>,
1182 pub number: u32,
1183 #[serde(default)]
1184 pub ahead: Vec<u32>,
1185}
1186
1187impl Actions {
1188 /// `merge_group`: the merge queue built a state and its checks passed.
1189 /// Starts the workflows that run `on: merge_group` on it, as GitHub's
1190 /// queue does, and says how many started; the queue waits for their
1191 /// statuses on that commit.
1192 pub async fn merge_group(&self, a: MergeGroupArgs) -> Result<Outcome<Value>> {
1193 let Some((repo, ws)) = self.repo_by_id(&a.repo_id).await? else {
1194 return Ok(Outcome::Ok(json!({ "runs": 0 })));
1195 };
1196 let read = self.read_workflows(&Self::repo_path(&repo), &ws, Some(&a.sha)).await?;
1197 let head_commit = self.commits(&repo, &ws, &a.sha, None).await?.pop();
1198 let payload = json!({
1199 "action": "checks_requested",
1200 "merge_group": {
1201 "head_sha": a.sha,
1202 "head_ref": a.head_ref,
1203 "base_sha": a.base_sha,
1204 "base_ref": format!("refs/heads/{}", repo.default_branch),
1205 "head_commit": head_commit.as_ref().map(|c| payload::commit(&repo, c)),
1206 },
1207 "repository": payload::repository(&repo),
1208 "sender": payload::user(&repo.namespace),
1209 });
1210 let mut started = 0u32;
1211 for file in read.files {
1212 let Ok(workflow) = workflow::parse(&file.source) else { continue };
1213 let Some(trigger) = workflow.trigger("merge_group") else { continue };
1214 if !trigger.wants_type(Some("checks_requested")) || self.disabled(&repo.id, &file.path).await? {
1215 continue;
1216 }
1217 // Branch filters on merge_group name the branch it merges into.
1218 if trigger.branches.is_set() && !trigger.branches.allows(&repo.default_branch) {
1219 continue;
1220 }
1221 let ahead = if a.ahead.is_empty() {
1222 String::new()
1223 } else {
1224 format!(" after {}", a.ahead.iter().map(|n| format!("#{n}")).collect::<Vec<_>>().join(", "))
1225 };
1226 let subject = Subject {
1227 source: Self::repo_path(&repo),
1228 source_ref: Some(a.sha.clone()),
1229 git_ref: a.head_ref.clone(),
1230 sha: a.sha.clone(),
1231 head_ref: None,
1232 base_ref: Some(repo.default_branch.clone()),
1233 pull: Some(a.number),
1234 filter_ref: format!("refs/heads/{}", repo.default_branch),
1235 paths: None,
1236 compare: None,
1237 payload: payload.clone(),
1238 title: format!("Merge queue: #{}{ahead}", a.number),
1239 trusted: true,
1240 approval: None,
1241 };
1242 let info = self.run_info(&repo, &workflow, "merge_group", &subject, &repo.namespace, None);
1243 let created = self
1244 .create_run(NewRun {
1245 repo: repo.clone(),
1246 path: file.path.clone(),
1247 source: file.source.clone(),
1248 workflow,
1249 info,
1250 action: Some("checks_requested".to_owned()),
1251 pull: Some(a.number),
1252 title: subject.title.clone(),
1253 inputs: Map::new(),
1254 event_key: format!("merge_group:{}:{}", a.entry, a.sha),
1255 actor_id: None,
1256 actor: None,
1257 trusted: true,
1258 approval: None,
1259 })
1260 .await?;
1261 if created.is_some() {
1262 started += 1;
1263 }
1264 }
1265 Ok(Outcome::Ok(json!({ "runs": started })))
1266 }
1267}
1268
1269/// The inputs of a manual run: what was given, checked against the
1270/// workflow's declared inputs, with their defaults filled in. An input the
1271/// workflow does not declare is refused, as GitHub refuses it.
1272/// `environments` are the repository's, for `type: environment`.
1273fn dispatch_inputs(trigger: &Trigger, given: &Map<String, Value>, environments: &[String]) -> Outcome<Map<String, Value>> {
1274 let unexpected: Vec<&String> = given.keys().filter(|name| !trigger.inputs.contains_key(*name)).collect();
1275 if !unexpected.is_empty() {
1276 let names = serde_json::to_string(&unexpected).unwrap_or_default();
1277 let declared = if trigger.inputs.is_empty() {
1278 "The workflow declares no inputs.".to_owned()
1279 } else {
1280 format!("The workflow declares {}.", trigger.inputs.keys().map(|name| format!("`{name}`")).collect::<Vec<_>>().join(", "))
1281 };
1282 return fail(FailureCode::Invalid, format!("Unexpected inputs provided: {names}. {declared}"));
1283 }
1284 let mut inputs = Map::new();
1285 for (name, spec) in &trigger.inputs {
1286 let kind = spec.get("type").and_then(Value::as_str).unwrap_or("string");
1287 let value = given.get(name).cloned().or_else(|| spec.get("default").cloned());
1288 let required = spec.get("required").and_then(Value::as_bool).unwrap_or(false);
1289 let value = match value {
1290 Some(Value::Null) | None if required => return fail(FailureCode::Invalid, format!("The input `{name}` is required.")),
1291 Some(Value::Null) | None => match kind {
1292 "boolean" => Value::Bool(false),
1293 _ => Value::String(String::new()),
1294 },
1295 Some(value) => match kind {
1296 "boolean" => Value::Bool(match &value {
1297 Value::Bool(flag) => *flag,
1298 Value::String(text) if text.trim().eq_ignore_ascii_case("true") => true,
1299 Value::String(text) if text.trim().eq_ignore_ascii_case("false") || text.trim().is_empty() => false,
1300 _ => return fail(FailureCode::Invalid, format!("The input `{name}` is a boolean: true or false.")),
1301 }),
1302 "number" => match &value {
1303 Value::Number(_) => value,
1304 Value::String(text) => match number(text) {
1305 Some(number) => Value::Number(number),
1306 None => return fail(FailureCode::Invalid, format!("The input `{name}` is a number.")),
1307 },
1308 _ => return fail(FailureCode::Invalid, format!("The input `{name}` is a number.")),
1309 },
1310 "choice" => {
1311 let text = g1t_actions::expr::to_text(&value);
1312 let options: Vec<String> =
1313 spec.get("options").and_then(Value::as_array).map(|o| o.iter().map(g1t_actions::expr::to_text).collect()).unwrap_or_default();
1314 if !options.is_empty() && !options.contains(&text) {
1315 return fail(FailureCode::Invalid, format!("The input `{name}` is one of {}.", options.join(", ")));
1316 }
1317 Value::String(text)
1318 }
1319 "environment" => {
1320 let text = g1t_actions::expr::to_text(&value);
1321 let wanted = text.trim().to_ascii_lowercase();
1322 if !wanted.is_empty() && !environments.iter().any(|env| env.eq_ignore_ascii_case(&wanted)) {
1323 return fail(
1324 FailureCode::Invalid,
1325 format!("The input `{name}` names an environment, and this repository has none called `{}`.", text.trim()),
1326 );
1327 }
1328 Value::String(text)
1329 }
1330 _ => Value::String(g1t_actions::expr::to_text(&value)),
1331 },
1332 };
1333 inputs.insert(name.clone(), value);
1334 }
1335 Outcome::Ok(inputs)
1336}
1337
1338/// A number input given as text: a whole number stays one.
1339fn number(text: &str) -> Option<serde_json::Number> {
1340 let text = text.trim();
1341 match text.parse::<i64>() {
1342 Ok(whole) => Some(whole.into()),
1343 Err(_) => text.parse::<f64>().ok().filter(|n| n.is_finite()).and_then(serde_json::Number::from_f64),
1344 }
1345}
1346
1347/// A manual run's inputs as `github.event.inputs` gives them: every one as
1348/// text, as GitHub's `workflow_dispatch` payload has them.
1349fn event_inputs(inputs: &Map<String, Value>) -> Map<String, Value> {
1350 inputs.iter().map(|(name, value)| (name.clone(), Value::String(g1t_actions::expr::to_text(value)))).collect()
1351}
1352
1353#[cfg(test)]
1354mod tests {
1355 use super::*;
1356 use g1t_contracts::work::{Pull, g1t_author};
1357
1358 #[test]
1359 fn every_event_a_workflow_runs_on_is_sent_to_this_queue() {
1360 for kind in g1t_contracts::webhooks::EVENT_TYPES {
1361 if !github_events(kind).is_empty() {
1362 assert!(
1363 g1t_contracts::subscribers::routed("SUBSCRIBER_ACTIONS", kind),
1364 "{kind} starts workflows but is not routed to actions (g1t_contracts::subscribers)"
1365 );
1366 }
1367 }
1368 }
1369 use g1t_contracts::{Membership, PrincipalKind};
1370
1371 fn repo() -> Repo {
1372 serde_json::from_value(json!({
1373 "id": "rep_1", "namespace": "acme", "name": "web", "description": null, "isPrivate": true,
1374 "ownerId": "ws_1", "defaultBranch": "main", "forkOf": null, "createdAt": ""
1375 }))
1376 .unwrap()
1377 }
1378
1379 fn person(id: &str, username: &str) -> User {
1380 User { id: id.into(), username: username.into(), kind: PrincipalKind::User, ..User::default() }
1381 }
1382
1383 fn made_for(asker: User) -> Pull {
1384 serde_json::from_value(json!({
1385 "id": "pr_1", "repoId": "rep_1", "number": 14, "issue": 12, "title": "Fix it", "body": null,
1386 "agent": "g1t", "runtime": "hosted", "status": "open",
1387 "fork": { "namespace": "pulls", "name": "pr_1" }, "forkRepoId": "rep_f",
1388 "branch": null, "headCommit": "abc", "mergeBase": null, "mergedBy": null, "mergedAt": null,
1389 "supersededBy": null, "checkStatus": null,
1390 "author": g1t_author(), "requestedBy": asker,
1391 "createdAt": "", "updatedAt": ""
1392 }))
1393 .unwrap()
1394 }
1395
1396 #[test]
1397 fn g1t_s_change_for_someone_is_trusted_as_they_are() {
1398 // Stored people carry no memberships, so identity is asked about
1399 // them; being g1t's change gives it nothing more.
1400 let pull = made_for(person("usr_2", "ana"));
1401 assert!(!trusted_outright(pull.owner(), &repo()));
1402 // Someone known to be able to push is trusted at once.
1403 let mut member = person("usr_1", "syntaqx");
1404 member.workspaces.push(Membership::member("acme"));
1405 let pull = made_for(member);
1406 assert!(trusted_outright(pull.owner(), &repo()));
1407 }
1408
1409 #[test]
1410 fn the_payload_names_g1t_as_its_user_and_who_asked_for_it() {
1411 let pull = made_for(person("usr_1", "syntaqx"));
1412 let event = payload::pull(&repo(), &pull);
1413 assert_eq!(event["user"]["login"], "g1t");
1414 assert_eq!(event["user"]["type"], "Bot");
1415 assert_eq!(event["requested_by"]["login"], "syntaqx");
1416 assert_eq!(event["requested_by"]["type"], "User");
1417 let as_issue = payload::pull_as_issue(&repo(), &pull);
1418 assert_eq!(as_issue["user"]["login"], "g1t");
1419 assert_eq!(as_issue["requested_by"]["login"], "syntaqx");
1420 }
1421
1422 #[test]
1423 fn releases_deployments_and_deleted_comments_read_as_githubs() {
1424 let release = payload::release(
1425 &repo(),
1426 &json!({ "id": "rel_1", "tagName": "v1.2.0", "target": "abc", "name": null, "body": "Notes", "draft": false,
1427 "prerelease": true, "author": "ana", "createdAt": "2026-10-08T00:00:00Z", "publishedAt": "2026-10-08T00:00:00Z" }),
1428 );
1429 assert_eq!(release["tag_name"], "v1.2.0");
1430 assert_eq!(release["name"], "v1.2.0");
1431 assert_eq!(release["prerelease"], true);
1432 assert_eq!(release["author"]["login"], "ana");
1433 assert_eq!(release["html_url"], "https://g1t.sh/acme/web/releases/tag/v1.2.0");
1434 let deployment = json!({ "id": "dep_1", "sha": "abc", "ref": "main", "environment": "staging", "creator": "ana",
1435 "production_environment": false, "created_at": "t", "updated_at": "t" });
1436 let status = payload::deployment_status(&repo(), &json!({ "id": "dst_1", "state": "success", "environment_url": "https://s.example", "log_url": null, "creator": "g1t", "created_at": "t" }), &deployment);
1437 assert_eq!(status["state"], "success");
1438 assert_eq!(status["environment"], "staging");
1439 assert_eq!(status["environment_url"], "https://s.example");
1440 assert_eq!(payload::deployment(&repo(), &deployment)["payload"], json!({}));
1441 let gone = payload::deleted_comment(&repo(), 7, &json!({ "id": "cmt_1", "body": "hi", "author": { "id": "usr_1", "username": "bo" }, "createdAt": "t" }), true);
1442 assert_eq!(gone["user"]["login"], "bo");
1443 assert_eq!(gone["html_url"], "https://g1t.sh/acme/web/pull/7#cmt_1");
1444 assert!(is_commit("0123456789abcdef0123456789abcdef01234567"));
1445 assert!(!is_commit("main"));
1446 }
1447
1448 #[test]
1449 fn a_pull_request_names_its_own_base_labels_and_milestone() {
1450 let mut pull = made_for(person("usr_1", "syntaqx"));
1451 let event = payload::pull(&repo(), &pull);
1452 assert_eq!(event["base"]["ref"], repo().default_branch);
1453 pull.base = Some("release/1.x".into());
1454 pull.labels = vec!["bug".into()];
1455 pull.milestone = Some(g1t_contracts::work::MilestoneRef { number: 2, title: "1.1".into() });
1456 let event = payload::pull(&repo(), &pull);
1457 assert_eq!(event["base"]["ref"], "release/1.x");
1458 assert_eq!(event["labels"][0]["name"], "bug");
1459 assert!(event["labels"][0]["id"].is_u64());
1460 assert_eq!(event["milestone"]["title"], "1.1");
1461 let mut labeled = serde_json::json!({ "action": "labeled" });
1462 payload::changed(&repo(), &mut labeled, &serde_json::json!({ "label": { "name": "bug", "color": "d73a4a" } }));
1463 assert_eq!(labeled["label"]["color"], "d73a4a");
1464 let mut assigned = serde_json::json!({ "action": "assigned" });
1465 payload::changed(&repo(), &mut assigned, &serde_json::json!({ "assignees": ["ana", "bo"], "added": ["bo"] }));
1466 assert_eq!(assigned["assignee"]["login"], "bo");
1467 assert!(nobody_added(&serde_json::json!({ "assignees": ["ana"], "added": [] })));
1468 assert!(!nobody_added(&serde_json::json!({ "assignees": ["ana"], "added": ["ana"] })));
1469 }
1470
1471 fn dispatchable(yaml: &str) -> Trigger {
1472 workflow::parse(yaml).unwrap().trigger("workflow_dispatch").unwrap().clone()
1473 }
1474
1475 fn given(value: Value) -> Map<String, Value> {
1476 value.as_object().unwrap().clone()
1477 }
1478
1479 const DISPATCH: &str = "on:
1480 workflow_dispatch:
1481 inputs:
1482 dry-run: { type: boolean, default: false }
1483 count: { type: number, default: 2 }
1484 target: { type: environment }
1485 level: { type: choice, options: [low, high], default: low }
1486 note: { required: false }
1487jobs:
1488 a:
1489 runs-on: x
1490 steps: [{ run: 'true' }]";
1491
1492 #[test]
1493 fn an_input_the_workflow_does_not_declare_is_refused() {
1494 let trigger = dispatchable(DISPATCH);
1495 let Outcome::Fail(failure) = dispatch_inputs(&trigger, &given(json!({ "dry-run": true, "colour": "red" })), &[]) else {
1496 panic!("an undeclared input was taken");
1497 };
1498 assert_eq!(failure.code, FailureCode::Invalid);
1499 assert!(failure.message.starts_with(r#"Unexpected inputs provided: ["colour"]."#), "{}", failure.message);
1500 assert!(failure.message.contains("`dry-run`"));
1501 let bare = dispatchable("on: workflow_dispatch\njobs:\n a:\n runs-on: x\n steps: [{ run: 'true' }]");
1502 let Outcome::Fail(failure) = dispatch_inputs(&bare, &given(json!({ "x": "1" })), &[]) else { panic!() };
1503 assert!(failure.message.contains("declares no inputs"));
1504 assert!(matches!(dispatch_inputs(&bare, &Map::new(), &[]), Outcome::Ok(inputs) if inputs.is_empty()));
1505 }
1506
1507 #[test]
1508 fn inputs_are_typed_and_the_event_has_them_as_text() {
1509 let trigger = dispatchable(DISPATCH);
1510 let environments = vec!["staging".to_owned(), "production".to_owned()];
1511 let Outcome::Ok(inputs) = dispatch_inputs(&trigger, &given(json!({ "dry-run": "true", "count": "3", "target": "Staging" })), &environments)
1512 else {
1513 panic!("refused");
1514 };
1515 assert_eq!(inputs["dry-run"], json!(true));
1516 assert_eq!(inputs["count"], json!(3));
1517 assert_eq!(inputs["level"], json!("low"));
1518 assert_eq!(inputs["note"], json!(""));
1519 let event = event_inputs(&inputs);
1520 assert_eq!(event["dry-run"], json!("true"));
1521 assert_eq!(event["count"], json!("3"));
1522 assert_eq!(event["target"], json!("Staging"));
1523 let Outcome::Ok(defaults) = dispatch_inputs(&trigger, &Map::new(), &environments) else { panic!() };
1524 assert_eq!(event_inputs(&defaults)["dry-run"], json!("false"));
1525 assert_eq!(defaults["count"], json!(2));
1526 let refused = |value: Value| match dispatch_inputs(&trigger, &given(value), &environments) {
1527 Outcome::Fail(failure) => failure.message,
1528 Outcome::Ok(_) => panic!("taken"),
1529 };
1530 assert!(refused(json!({ "target": "qa" })).contains("none called `qa`"));
1531 assert!(refused(json!({ "dry-run": "yes" })).contains("true or false"));
1532 assert!(refused(json!({ "count": "many" })).contains("is a number"));
1533 assert!(refused(json!({ "level": "mid" })).contains("low, high"));
1534 }
1535
1536 #[test]
1537 fn create_and_delete_take_no_filters_and_name_their_ref() {
1538 assert!(unfiltered("create") && unfiltered("delete"));
1539 assert!(!unfiltered("push") && !unfiltered("pull_request"));
1540 assert_eq!(branch_or_tag("refs/heads/feature/x"), Some(("branch", "feature/x")));
1541 assert_eq!(branch_or_tag("refs/tags/v1.0"), Some(("tag", "v1.0")));
1542 assert_eq!(branch_or_tag("refs/heads/g1t-queue/main/3"), None);
1543 assert_eq!(branch_or_tag("refs/notes/commits"), None);
1544 // A workflow can name `delete`. It takes no filters, as GitHub's
1545 // schema says, so writing one is refused when the workflow is read.
1546 assert!(workflow::parse("on:\n delete:\n branches: [release/*]\njobs:\n a:\n runs-on: x\n steps: [{ run: 'true' }]").is_err());
1547 let workflow = workflow::parse("on:\n delete:\njobs:\n a:\n runs-on: x\n steps: [{ run: 'true' }]").unwrap();
1548 let trigger = workflow.trigger("delete").unwrap();
1549 assert!(trigger.wants_type(None));
1550 assert!(!workflow.notes.iter().any(|note| note.severity == workflow::Severity::Unsupported));
1551 }
1552}