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