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