| 1 | //! 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 | //! - [`protection`] keeps a repository's workflow settings and its |
| 14 | //! environments' protection rules, holds jobs at those rules until they |
| 15 | //! pass, and lets pull requests' runs from outside wait for approval. |
| 16 | //! - [`cache`] lists `actions/cache` entries, which the API keeps in R2. |
| 17 | //! - [`runners`] keeps self-hosted runners, hands them jobs (and agent |
| 18 | //! work from the runner service) when they ask, and hears back. |
| 19 | //! |
| 20 | //! The service acts as the repository's workspace: it reads what the |
| 21 | //! workspace can read, and a job's `GITHUB_TOKEN` is a token of the |
| 22 | //! workspace's that reaches the job's repository only, with the scopes its |
| 23 | //! `permissions:` give it, revoked when the job ends. |
| 24 | |
| 25 | mod artifacts; |
| 26 | mod cache; |
| 27 | mod payload; |
| 28 | mod plan; |
| 29 | mod protection; |
| 30 | mod rename; |
| 31 | pub mod runtime; |
| 32 | mod runners; |
| 33 | mod settings; |
| 34 | mod sync; |
| 35 | mod trigger; |
| 36 | mod views; |
| 37 | |
| 38 | use g1t_contracts::access::{self, Capability}; |
| 39 | use g1t_contracts::events::Event; |
| 40 | use g1t_contracts::identity::{SlugArgs, Workspace}; |
| 41 | use g1t_contracts::repos::{GetArgs, GetByIdArgs, Repo, RepoPath}; |
| 42 | use g1t_contracts::{FailureCode, Membership, Outcome, PrincipalKind, User, Viewer}; |
| 43 | use g1t_kit::{args, reply, rpc_method}; |
| 44 | use g1t_secrets::Sealer; |
| 45 | use serde::Deserialize; |
| 46 | use serde_json::Value; |
| 47 | use worker::wasm_bindgen::JsValue; |
| 48 | use worker::{Context, D1Database, Env, Fetcher, MessageBatch, MessageExt, Request, Response, Result, ScheduleContext, ScheduledEvent, event}; |
| 49 | |
| 50 | /// The most workflow files read from a repository. |
| 51 | pub const MAX_WORKFLOWS: usize = 50; |
| 52 | /// Jobs one workspace may have running at once; the rest wait their turn. |
| 53 | pub const RUNNING_PER_WORKSPACE: u32 = 4; |
| 54 | /// The longest a job may run, whatever its `timeout-minutes`. |
| 55 | pub const MAX_TIMEOUT_MINUTES: u32 = 60; |
| 56 | /// The longest a job on a self-hosted runner may run: the machine is the |
| 57 | /// workspace's own, and its time costs nothing. |
| 58 | pub const SELF_HOSTED_MAX_TIMEOUT_MINUTES: u32 = 24 * 60; |
| 59 | /// A running job that has said nothing for this long is taken as lost. |
| 60 | pub const SILENT_MS: u64 = 10 * 60 * 1000; |
| 61 | pub const SITE: &str = "https://g1t.sh"; |
| 62 | pub const API: &str = "https://api.g1t.sh"; |
| 63 | |
| 64 | #[derive(Deserialize)] |
| 65 | struct Count { |
| 66 | n: u32, |
| 67 | } |
| 68 | |
| 69 | pub fn optional(value: Option<&str>) -> JsValue { |
| 70 | value.map_or(JsValue::NULL, JsValue::from) |
| 71 | } |
| 72 | |
| 73 | pub fn fail<T>(code: FailureCode, message: impl Into<String>) -> Outcome<T> { |
| 74 | Outcome::fail(code, message) |
| 75 | } |
| 76 | |
| 77 | /// `owner/name` as a path. |
| 78 | pub fn repo_path(full_name: &str) -> RepoPath { |
| 79 | let (namespace, name) = full_name.split_once('/').unwrap_or((full_name, "")); |
| 80 | RepoPath { |
| 81 | namespace: namespace.to_owned(), |
| 82 | name: name.to_owned(), |
| 83 | } |
| 84 | } |
| 85 | |
| 86 | pub struct Actions { |
| 87 | db: D1Database, |
| 88 | repos: Fetcher, |
| 89 | work: Fetcher, |
| 90 | identity: Fetcher, |
| 91 | runner: Fetcher, |
| 92 | events: Fetcher, |
| 93 | /// Projects: a repository's secrets and variables belong to its project. |
| 94 | projects: Fetcher, |
| 95 | /// Billing: self-hosted runners' time, recorded at $0, and the |
| 96 | /// cache's storage. |
| 97 | billing: Fetcher, |
| 98 | /// Deployments: a job with an `environment:` deploys to it, and its |
| 99 | /// run's deployment is recorded there (plan.rs `report_deployment`). |
| 100 | deployments: Fetcher, |
| 101 | /// Where the API keeps cache entries, for deleting evicted ones. |
| 102 | cache: Option<worker::Bucket>, |
| 103 | /// Seals secrets; absent until `ACTIONS_KEY` is set, when secrets |
| 104 | /// cannot be saved. |
| 105 | sealer: Option<Sealer>, |
| 106 | /// Signs the toolkit's blob links (runtime.rs), from `ACTIONS_KEY`. |
| 107 | blob_key: Option<String>, |
| 108 | } |
| 109 | |
| 110 | impl Actions { |
| 111 | fn new(env: &Env) -> Result<Self> { |
| 112 | Ok(Actions { |
| 113 | db: env.d1("DB")?, |
| 114 | repos: env.service("REPOS")?, |
| 115 | work: env.service("WORK")?, |
| 116 | identity: env.service("IDENTITY")?, |
| 117 | runner: env.service("RUNNER")?, |
| 118 | events: env.service("EVENTS")?, |
| 119 | projects: env.service("PROJECTS")?, |
| 120 | billing: env.service("BILLING")?, |
| 121 | deployments: env.service("DEPLOYMENTS")?, |
| 122 | cache: env.bucket("ACTIONS_CACHE").ok(), |
| 123 | sealer: env.secret("ACTIONS_KEY").ok().and_then(|key| Sealer::new(&key.to_string())), |
| 124 | blob_key: env.secret("ACTIONS_KEY").ok().map(|key| runtime::blob_key(&key.to_string())), |
| 125 | }) |
| 126 | } |
| 127 | |
| 128 | /// The workspace itself, as the service acts. |
| 129 | async fn workspace_actor(&self, slug: &str) -> Result<Option<User>> { |
| 130 | let workspace: Option<Workspace> = g1t_kit::call(&self.identity, "get_workspace", &SlugArgs { slug: slug.to_owned() }).await?; |
| 131 | Ok(workspace.map(|workspace| User { |
| 132 | id: workspace.id, |
| 133 | username: workspace.slug.clone(), |
| 134 | kind: PrincipalKind::Workspace, |
| 135 | verified: true, |
| 136 | workspaces: vec![Membership::member(workspace.slug)], |
| 137 | ..User::default() |
| 138 | })) |
| 139 | } |
| 140 | |
| 141 | /// The repository, if the viewer may see it and it is not a pull |
| 142 | /// request's working copy. |
| 143 | async fn visible_repo(&self, path: &RepoPath, viewer: &Viewer) -> Result<Option<Repo>> { |
| 144 | let found: Outcome<Repo> = g1t_kit::call( |
| 145 | &self.repos, |
| 146 | "get", |
| 147 | &GetArgs { |
| 148 | path: path.clone(), |
| 149 | viewer: viewer.clone(), |
| 150 | }, |
| 151 | ) |
| 152 | .await?; |
| 153 | Ok(found.into_result().ok().filter(|repo| repo.fork_of.is_none())) |
| 154 | } |
| 155 | |
| 156 | /// A repository by id, as its workspace sees it. |
| 157 | async fn repo_by_id(&self, id: &str) -> Result<Option<(Repo, User)>> { |
| 158 | let path: Option<RepoPath> = g1t_kit::call(&self.repos, "path_by_id", &g1t_contracts::repos::PathByIdArgs { id: id.to_owned() }).await?; |
| 159 | let Some(path) = path else { return Ok(None) }; |
| 160 | let Some(actor) = self.workspace_actor(&path.namespace).await? else { |
| 161 | return Ok(None); |
| 162 | }; |
| 163 | let found: Outcome<Repo> = g1t_kit::call( |
| 164 | &self.repos, |
| 165 | "get_by_id", |
| 166 | &GetByIdArgs { |
| 167 | id: id.to_owned(), |
| 168 | viewer: Some(actor.clone()), |
| 169 | }, |
| 170 | ) |
| 171 | .await?; |
| 172 | Ok(found.into_result().ok().filter(|repo| repo.fork_of.is_none()).map(|repo| (repo, actor))) |
| 173 | } |
| 174 | |
| 175 | /// The repository at `path`, when `actor` may do `capability` in it: |
| 176 | /// not found when they cannot read it, forbidden when their role falls |
| 177 | /// short. Agents never may: people and tokens run and change workflows. |
| 178 | async fn may(&self, actor: &User, path: &RepoPath, capability: Capability) -> Result<Outcome<Repo>> { |
| 179 | if actor.kind == PrincipalKind::Agent { |
| 180 | return Ok(fail(FailureCode::Forbidden, "An agent cannot do that. Ask a person.")); |
| 181 | } |
| 182 | let Some(repo) = self.visible_repo(path, &Some(actor.clone())).await? else { |
| 183 | return Ok(fail(FailureCode::NotFound, "There is no such repository.")); |
| 184 | }; |
| 185 | if !access::can(Some(actor), &repo, capability) { |
| 186 | return Ok(fail( |
| 187 | FailureCode::Forbidden, |
| 188 | access::needs(capability, &format!("{}/{}", repo.namespace, repo.name)), |
| 189 | )); |
| 190 | } |
| 191 | Ok(Outcome::Ok(repo)) |
| 192 | } |
| 193 | } |
| 194 | |
| 195 | /// Unwraps an `Outcome`, or returns its failure from the enclosing method. |
| 196 | #[macro_export] |
| 197 | macro_rules! check { |
| 198 | ($outcome:expr) => { |
| 199 | match $outcome { |
| 200 | g1t_contracts::Outcome::Ok(value) => value, |
| 201 | g1t_contracts::Outcome::Fail(refused) => return Ok(g1t_contracts::Outcome::Fail(refused)), |
| 202 | } |
| 203 | }; |
| 204 | } |
| 205 | |
| 206 | #[event(fetch)] |
| 207 | async fn fetch(mut request: Request, env: Env, _ctx: Context) -> Result<Response> { |
| 208 | let Some(method) = rpc_method(&request) else { |
| 209 | return Response::error("Not found", 404); |
| 210 | }; |
| 211 | let body: Value = request.json().await?; |
| 212 | let service = Actions::new(&env)?; |
| 213 | match method.as_str() { |
| 214 | "workflows" => reply(&service.workflows(args(body)?).await?), |
| 215 | "runs" => reply(&service.runs(args(body)?).await?), |
| 216 | "run" => reply(&service.run(args(body)?).await?), |
| 217 | "logs" => reply(&service.logs(args(body)?).await?), |
| 218 | "dispatch" => reply(&service.dispatch(args(body)?).await?), |
| 219 | "repository_dispatch" => reply(&service.repository_dispatch(args(body)?).await?), |
| 220 | "approve_run" => reply(&service.approve_run(args(body)?).await?), |
| 221 | "pending_deployments" => reply(&service.pending_deployments(args(body)?).await?), |
| 222 | "review_deployments" => reply(&service.review_deployments(args(body)?).await?), |
| 223 | "actions_settings" => reply(&service.actions_settings(args(body)?).await?), |
| 224 | "set_actions_settings" => reply(&service.set_actions_settings(args(body)?).await?), |
| 225 | "workspace_actions_settings" => reply(&service.workspace_actions_settings(args(body)?).await?), |
| 226 | "set_workspace_actions_settings" => reply(&service.set_workspace_actions_settings(args(body)?).await?), |
| 227 | "environments" => reply(&service.environments(args(body)?).await?), |
| 228 | "set_environment" => reply(&service.set_environment(args(body)?).await?), |
| 229 | "delete_environment" => reply(&service.delete_environment(args(body)?).await?), |
| 230 | "merge_group" => reply(&service.merge_group(args(body)?).await?), |
| 231 | "cancel" => reply(&service.cancel(args(body)?).await?), |
| 232 | "rerun" => reply(&service.rerun(args(body)?).await?), |
| 233 | "set_workflow_enabled" => reply(&service.set_workflow_enabled(args(body)?).await?), |
| 234 | "settings" => reply(&service.settings(args(body)?).await?), |
| 235 | "set_setting" => reply(&service.set_setting(args(body)?).await?), |
| 236 | "delete_setting" => reply(&service.delete_setting(args(body)?).await?), |
| 237 | "resolve_settings" => reply(&service.resolve_settings(args(body)?).await?), |
| 238 | "job_spec" => reply(&service.job_spec(args(body)?).await?), |
| 239 | "job_auth" => reply(&service.job_auth(args(body)?).await?), |
| 240 | "check_runs" => reply(&service.check_runs(args(body)?).await?), |
| 241 | "job_report" => reply(&service.job_report(args(body)?).await?), |
| 242 | // actions/cache, through the API with the job's token. |
| 243 | "cache_lookup" => reply(&service.cache_lookup(args(body)?).await?), |
| 244 | "cache_reserve" => reply(&service.cache_reserve(args(body)?).await?), |
| 245 | "cache_commit" => reply(&service.cache_commit(args(body)?).await?), |
| 246 | "cache_abort" => reply(&service.cache_abort(args(body)?).await?), |
| 247 | "cache_upload" => reply(&service.cache_upload(args(body)?).await?), |
| 248 | // Artifacts: a job's side, with its token or its runtime token. |
| 249 | "artifact_reserve" => reply(&service.artifact_reserve(args(body)?).await?), |
| 250 | "artifact_commit" => reply(&service.artifact_commit(args(body)?).await?), |
| 251 | "artifact_abort" => reply(&service.artifact_abort(args(body)?).await?), |
| 252 | "job_artifacts" => reply(&service.job_artifacts(args(body)?).await?), |
| 253 | "job_artifact" => reply(&service.job_artifact(args(body)?).await?), |
| 254 | "job_delete_artifact" => reply(&service.job_delete_artifact(args(body)?).await?), |
| 255 | // Artifacts: people's side. |
| 256 | "artifacts" => reply(&service.artifacts(args(body)?).await?), |
| 257 | "artifact" => reply(&service.artifact(args(body)?).await?), |
| 258 | "artifact_download" => reply(&service.artifact_download(args(body)?).await?), |
| 259 | "delete_artifact" => reply(&service.delete_artifact(args(body)?).await?), |
| 260 | "artifact_retention" => reply(&service.artifact_retention(args(body)?).await?), |
| 261 | // The toolkit's protocols: the runtime token, OIDC, blob links. |
| 262 | "runtime_auth" => reply(&service.runtime_auth(args(body)?).await?), |
| 263 | "oidc_claims" => reply(&service.oidc_claims(args(body)?).await?), |
| 264 | "blob_sign" => reply(&service.blob_sign(args(body)?).await?), |
| 265 | "blob_open" => reply(&service.blob_open(args(body)?).await?), |
| 266 | "blob_part" => reply(&service.blob_part(args(body)?).await?), |
| 267 | "blob_parts" => reply(&service.blob_parts(args(body)?).await?), |
| 268 | "blob_done" => reply(&service.blob_done(args(body)?).await?), |
| 269 | // Self-hosted runners: people's side. |
| 270 | "runners" => reply(&service.runners(args(body)?).await?), |
| 271 | "create_registration_token" => reply(&service.create_registration_token(args(body)?).await?), |
| 272 | "remove_runner" => reply(&service.remove_runner(args(body)?).await?), |
| 273 | "runner_groups" => reply(&service.runner_groups(args(body)?).await?), |
| 274 | "set_runner_group" => reply(&service.set_runner_group(args(body)?).await?), |
| 275 | "delete_runner_group" => reply(&service.delete_runner_group(args(body)?).await?), |
| 276 | "runner_settings" => reply(&service.runner_settings(args(body)?).await?), |
| 277 | "set_runner_settings" => reply(&service.set_runner_settings(args(body)?).await?), |
| 278 | // The runner's own side, through the API with its credential. |
| 279 | "runner_register" => reply(&service.runner_register(args(body)?).await?), |
| 280 | "runner_poll" => reply(&service.runner_poll(args(body)?).await?), |
| 281 | "runner_finished" => reply(&service.runner_finished(args(body)?).await?), |
| 282 | "runner_remove_self" => reply(&service.runner_remove_self(args(body)?).await?), |
| 283 | // Agent work, from the runner service. |
| 284 | "runner_route" => reply(&service.runner_route(args(body)?).await?), |
| 285 | "stuck_jobs" => reply(&service.stuck_jobs(args(body)?).await?), |
| 286 | "enqueue_task" => reply(&service.enqueue_task(args(body)?).await?), |
| 287 | "cancel_task" => reply(&service.cancel_task(args(body)?).await?), |
| 288 | _ => Response::error("Unknown method", 404), |
| 289 | } |
| 290 | } |
| 291 | |
| 292 | /// Events from the bus, on this service's own queue. |
| 293 | #[event(queue)] |
| 294 | async fn queue(batch: MessageBatch<Event>, env: Env, _ctx: Context) -> Result<()> { |
| 295 | let service = Actions::new(&env)?; |
| 296 | for message in batch.messages()? { |
| 297 | // A workspace renamed: its rows move to the slug it has now. |
| 298 | if g1t_kit::rename::on_event(&env, &env.d1("DB")?, message.body(), rename::STATEMENTS).await? { |
| 299 | message.ack(); |
| 300 | continue; |
| 301 | } |
| 302 | // A repository renamed or transferred: its rows follow its new path. |
| 303 | if g1t_kit::transfer::on_event(&env, &env.d1("DB")?, message.body(), rename::TRANSFERRED).await? { |
| 304 | message.ack(); |
| 305 | continue; |
| 306 | } |
| 307 | // A workspace deleted: what it kept for itself goes. |
| 308 | if g1t_kit::deleted::on_event(&env.d1("DB")?, message.body(), rename::DELETED).await? { |
| 309 | message.ack(); |
| 310 | continue; |
| 311 | } |
| 312 | // A repository purged: every row kept for it goes. |
| 313 | if g1t_kit::lifecycle::on_purged(&env.d1("DB")?, message.body(), rename::PURGED).await? { |
| 314 | message.ack(); |
| 315 | continue; |
| 316 | } |
| 317 | // A repository deleted or archived: its runs stop. |
| 318 | if let Some(repo_id) = plan::stops_runs(message.body()) { |
| 319 | if let Err(error) = service.stop_runs(&repo_id).await { |
| 320 | worker::console_error!("actions: event {} failed: {error}", message.body().id); |
| 321 | message.retry(); |
| 322 | continue; |
| 323 | } |
| 324 | message.ack(); |
| 325 | continue; |
| 326 | } |
| 327 | if let Err(error) = service.on_event(message.body()).await { |
| 328 | worker::console_error!("actions: event {} failed: {error}", message.body().id); |
| 329 | message.retry(); |
| 330 | continue; |
| 331 | } |
| 332 | message.ack(); |
| 333 | } |
| 334 | Ok(()) |
| 335 | } |
| 336 | |
| 337 | /// Every minute: schedules that fire, jobs waiting for room, and jobs |
| 338 | /// whose sandbox went quiet. |
| 339 | #[event(scheduled)] |
| 340 | async fn scheduled(_event: ScheduledEvent, env: Env, _ctx: ScheduleContext) { |
| 341 | match Actions::new(&env) { |
| 342 | Ok(service) => { |
| 343 | if let Err(error) = service.on_minute(g1t_kit::now_ms()).await { |
| 344 | worker::console_error!("actions: the sweep failed: {error}"); |
| 345 | } |
| 346 | } |
| 347 | Err(error) => worker::console_error!("actions: could not start: {error}"), |
| 348 | } |
| 349 | } |