Sign in

flagon-io/g1t

public

g1t/services/actions/src/lib.rs

214 lines8,245 bytesCodeBlame

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

GitHub Actions on g1t, part two: running workflows1//! The actions service: a repository's GitHub Actions workflows, run on
2//! g1t as they are. See `g1t_contracts::actions` for the methods and
3//! `g1t_actions` for how workflows, expressions and filters are read.
4//!
5//! - [`sync`] reads workflow files, at the default branch for the list and
6//! at an event's own commit for its runs.
7//! - [`trigger`] turns events, schedules and manual runs into runs.
8//! - [`plan`] moves a run's jobs along: each waits for the jobs it needs,
9//! is skipped or expanded into its matrix, queued, started in a sandbox
10//! when the workspace has room, and finished by the sandbox's report.
11//! - [`payload`] builds the webhook-shaped `github.event`.
12//! - [`settings`] keeps secrets and variables.
13//!
14//! The service acts as the repository's workspace: it reads what the
15//! workspace can read, and a job's `GITHUB_TOKEN` is a short-lived token of
16//! the workspace's.
17
18mod payload;
19mod plan;
20mod settings;
21mod sync;
22mod trigger;
23mod views;
24
25use g1t_contracts::events::Event;
26use g1t_contracts::identity::{SlugArgs, Workspace};
27use g1t_contracts::repos::{GetArgs, GetByIdArgs, Repo, RepoPath};
28use g1t_contracts::{FailureCode, Membership, Outcome, PrincipalKind, Role, User, Viewer};
29use g1t_kit::{args, reply, rpc_method};
30use g1t_secrets::Sealer;
31use serde::Deserialize;
32use serde_json::Value;
33use worker::wasm_bindgen::JsValue;
34use worker::{Context, D1Database, Env, Fetcher, MessageBatch, MessageExt, Request, Response, Result, ScheduleContext, ScheduledEvent, event};
35
36/// The most workflow files read from a repository.
37pub const MAX_WORKFLOWS: usize = 50;
38/// Jobs one workspace may have running at once; the rest wait their turn.
39pub const RUNNING_PER_WORKSPACE: u32 = 4;
40/// The longest a job may run, whatever its `timeout-minutes`.
41pub const MAX_TIMEOUT_MINUTES: u32 = 60;
42/// A running job that has said nothing for this long is taken as lost.
43pub const SILENT_MS: u64 = 10 * 60 * 1000;
44pub const SITE: &str = "https://g1t.sh";
45pub const API: &str = "https://api.g1t.sh";
46
47#[derive(Deserialize)]
48struct Count {
49 n: u32,
50}
51
52pub fn optional(value: Option<&str>) -> JsValue {
53 value.map_or(JsValue::NULL, JsValue::from)
54}
55
56pub fn fail<T>(code: FailureCode, message: impl Into<String>) -> Outcome<T> {
57 Outcome::fail(code, message)
58}
59
60/// `owner/name` as a path.
61pub fn repo_path(full_name: &str) -> RepoPath {
62 let (namespace, name) = full_name.split_once('/').unwrap_or((full_name, ""));
63 RepoPath {
64 namespace: namespace.to_owned(),
65 name: name.to_owned(),
66 }
67}
68
69pub struct Actions {
70 db: D1Database,
71 repos: Fetcher,
72 work: Fetcher,
73 identity: Fetcher,
74 runner: Fetcher,
Actions: workflow_run, workflow.completed, artifacts on the run page, Node 2475 events: Fetcher,
GitHub Actions on g1t, part two: running workflows76 /// Seals secrets; absent until `ACTIONS_KEY` is set, when secrets
77 /// cannot be saved.
78 sealer: Option<Sealer>,
79}
80
81impl Actions {
82 fn new(env: &Env) -> Result<Self> {
83 Ok(Actions {
84 db: env.d1("DB")?,
85 repos: env.service("REPOS")?,
86 work: env.service("WORK")?,
87 identity: env.service("IDENTITY")?,
88 runner: env.service("RUNNER")?,
Actions: workflow_run, workflow.completed, artifacts on the run page, Node 2489 events: env.service("EVENTS")?,
GitHub Actions on g1t, part two: running workflows90 sealer: env.secret("ACTIONS_KEY").ok().and_then(|key| Sealer::new(&key.to_string())),
91 })
92 }
93
94 /// The workspace itself, as the service acts.
95 async fn workspace_actor(&self, slug: &str) -> Result<Option<User>> {
96 let workspace: Option<Workspace> = g1t_kit::call(&self.identity, "get_workspace", &SlugArgs { slug: slug.to_owned() }).await?;
97 Ok(workspace.map(|workspace| User {
98 id: workspace.id,
99 username: workspace.slug.clone(),
100 kind: PrincipalKind::Workspace,
101 verified: true,
102 workspaces: vec![Membership {
103 slug: workspace.slug,
104 role: Role::Member,
105 }],
106 }))
107 }
108
109 /// The repository, if the viewer may see it and it is not a pull
110 /// request's working copy.
111 async fn visible_repo(&self, path: &RepoPath, viewer: &Viewer) -> Result<Option<Repo>> {
112 let found: Outcome<Repo> = g1t_kit::call(
113 &self.repos,
114 "get",
115 &GetArgs {
116 path: path.clone(),
117 viewer: viewer.clone(),
118 },
119 )
120 .await?;
121 Ok(found.into_result().ok().filter(|repo| repo.fork_of.is_none()))
122 }
123
124 /// A repository by id, as its workspace sees it.
125 async fn repo_by_id(&self, id: &str) -> Result<Option<(Repo, User)>> {
126 let path: Option<RepoPath> = g1t_kit::call(&self.repos, "path_by_id", &g1t_contracts::repos::PathByIdArgs { id: id.to_owned() }).await?;
127 let Some(path) = path else { return Ok(None) };
128 let Some(actor) = self.workspace_actor(&path.namespace).await? else {
129 return Ok(None);
130 };
131 let found: Outcome<Repo> = g1t_kit::call(
132 &self.repos,
133 "get_by_id",
134 &GetByIdArgs {
135 id: id.to_owned(),
136 viewer: Some(actor.clone()),
137 },
138 )
139 .await?;
140 Ok(found.into_result().ok().filter(|repo| repo.fork_of.is_none()).map(|repo| (repo, actor)))
141 }
142
143 /// Refuses anyone but a member of the repository's workspace.
144 fn member(actor: &User, repo: &RepoPath) -> Option<Outcome<()>> {
145 (actor.kind == PrincipalKind::Agent || !actor.is_member(&repo.namespace.to_lowercase()))
146 .then(|| fail(FailureCode::Forbidden, format!("Only members of {} can do that.", repo.namespace)))
147 }
148}
149
150/// Unwraps an `Outcome`, or returns its failure from the enclosing method.
151#[macro_export]
152macro_rules! check {
153 ($outcome:expr) => {
154 match $outcome {
155 g1t_contracts::Outcome::Ok(value) => value,
156 g1t_contracts::Outcome::Fail(refused) => return Ok(g1t_contracts::Outcome::Fail(refused)),
157 }
158 };
159}
160
161#[event(fetch)]
162async fn fetch(mut request: Request, env: Env, _ctx: Context) -> Result<Response> {
163 let Some(method) = rpc_method(&request) else {
164 return Response::error("Not found", 404);
165 };
166 let body: Value = request.json().await?;
167 let service = Actions::new(&env)?;
168 match method.as_str() {
169 "workflows" => reply(&service.workflows(args(body)?).await?),
170 "runs" => reply(&service.runs(args(body)?).await?),
171 "run" => reply(&service.run(args(body)?).await?),
172 "logs" => reply(&service.logs(args(body)?).await?),
173 "dispatch" => reply(&service.dispatch(args(body)?).await?),
174 "cancel" => reply(&service.cancel(args(body)?).await?),
175 "rerun" => reply(&service.rerun(args(body)?).await?),
176 "set_workflow_enabled" => reply(&service.set_workflow_enabled(args(body)?).await?),
177 "settings" => reply(&service.settings(args(body)?).await?),
178 "set_setting" => reply(&service.set_setting(args(body)?).await?),
179 "delete_setting" => reply(&service.delete_setting(args(body)?).await?),
180 "job_spec" => reply(&service.job_spec(args(body)?).await?),
A repository has its own sidebar, as settings do181 "job_auth" => reply(&service.job_auth(args(body)?).await?),
GitHub Actions on g1t, part two: running workflows182 "job_report" => reply(&service.job_report(args(body)?).await?),
183 _ => Response::error("Unknown method", 404),
184 }
185}
186
187/// Events from the bus, on this service's own queue.
188#[event(queue)]
189async fn queue(batch: MessageBatch<Event>, env: Env, _ctx: Context) -> Result<()> {
190 let service = Actions::new(&env)?;
191 for message in batch.messages()? {
192 if let Err(error) = service.on_event(message.body()).await {
193 worker::console_error!("actions: event {} failed: {error}", message.body().id);
194 message.retry();
195 continue;
196 }
197 message.ack();
198 }
199 Ok(())
200}
201
202/// Every minute: schedules that fire, jobs waiting for room, and jobs
203/// whose sandbox went quiet.
204#[event(scheduled)]
205async fn scheduled(_event: ScheduledEvent, env: Env, _ctx: ScheduleContext) {
206 match Actions::new(&env) {
207 Ok(service) => {
208 if let Err(error) = service.on_minute(g1t_kit::now_ms()).await {
209 worker::console_error!("actions: the sweep failed: {error}");
210 }
211 }
212 Err(error) => worker::console_error!("actions: could not start: {error}"),
213 }
214}