Sign in

flagon-io/g1t

public

Where people and agents ship software together. The open-source git platform for the whole job: issues, agents, checks and deploys to the edge.

gitagentscloudflarerustforgemcp

g1t/services/actions/src/lib.rs

262 lines10,494 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;
Agents and memory, checks and conflicts, profiles, slug renames, custom domains20mod rename;
GitHub Actions on g1t, part two: running workflows21mod settings;
22mod sync;
23mod trigger;
24mod views;
25
Invite-only launch: sign in with GitHub, repository access and lifecycle, many emails, a new look26use g1t_contracts::access::{self, Capability};
GitHub Actions on g1t, part two: running workflows27use g1t_contracts::events::Event;
28use g1t_contracts::identity::{SlugArgs, Workspace};
29use g1t_contracts::repos::{GetArgs, GetByIdArgs, Repo, RepoPath};
Workspace names and icons, and a component kit for every control30use g1t_contracts::{FailureCode, Membership, Outcome, PrincipalKind, User, Viewer};
GitHub Actions on g1t, part two: running workflows31use g1t_kit::{args, reply, rpc_method};
32use g1t_secrets::Sealer;
33use serde::Deserialize;
34use serde_json::Value;
35use worker::wasm_bindgen::JsValue;
36use worker::{Context, D1Database, Env, Fetcher, MessageBatch, MessageExt, Request, Response, Result, ScheduleContext, ScheduledEvent, event};
37
38/// The most workflow files read from a repository.
39pub const MAX_WORKFLOWS: usize = 50;
40/// Jobs one workspace may have running at once; the rest wait their turn.
41pub const RUNNING_PER_WORKSPACE: u32 = 4;
42/// The longest a job may run, whatever its `timeout-minutes`.
43pub const MAX_TIMEOUT_MINUTES: u32 = 60;
44/// A running job that has said nothing for this long is taken as lost.
45pub const SILENT_MS: u64 = 10 * 60 * 1000;
46pub const SITE: &str = "https://g1t.sh";
47pub const API: &str = "https://api.g1t.sh";
48
49#[derive(Deserialize)]
50struct Count {
51 n: u32,
52}
53
54pub fn optional(value: Option<&str>) -> JsValue {
55 value.map_or(JsValue::NULL, JsValue::from)
56}
57
58pub fn fail<T>(code: FailureCode, message: impl Into<String>) -> Outcome<T> {
59 Outcome::fail(code, message)
60}
61
62/// `owner/name` as a path.
63pub fn repo_path(full_name: &str) -> RepoPath {
64 let (namespace, name) = full_name.split_once('/').unwrap_or((full_name, ""));
65 RepoPath {
66 namespace: namespace.to_owned(),
67 name: name.to_owned(),
68 }
69}
70
71pub struct Actions {
72 db: D1Database,
73 repos: Fetcher,
74 work: Fetcher,
75 identity: Fetcher,
76 runner: Fetcher,
Actions: workflow_run, workflow.completed, artifacts on the run page, Node 2477 events: Fetcher,
Projects: what a workspace builds and runs, first on every page78 /// Projects: a repository's secrets and variables belong to its project.
79 projects: Fetcher,
GitHub Actions on g1t, part two: running workflows80 /// Seals secrets; absent until `ACTIONS_KEY` is set, when secrets
81 /// cannot be saved.
82 sealer: Option<Sealer>,
83}
84
85impl Actions {
86 fn new(env: &Env) -> Result<Self> {
87 Ok(Actions {
88 db: env.d1("DB")?,
89 repos: env.service("REPOS")?,
90 work: env.service("WORK")?,
91 identity: env.service("IDENTITY")?,
92 runner: env.service("RUNNER")?,
Actions: workflow_run, workflow.completed, artifacts on the run page, Node 2493 events: env.service("EVENTS")?,
Projects: what a workspace builds and runs, first on every page94 projects: env.service("PROJECTS")?,
GitHub Actions on g1t, part two: running workflows95 sealer: env.secret("ACTIONS_KEY").ok().and_then(|key| Sealer::new(&key.to_string())),
96 })
97 }
98
99 /// The workspace itself, as the service acts.
100 async fn workspace_actor(&self, slug: &str) -> Result<Option<User>> {
101 let workspace: Option<Workspace> = g1t_kit::call(&self.identity, "get_workspace", &SlugArgs { slug: slug.to_owned() }).await?;
102 Ok(workspace.map(|workspace| User {
103 id: workspace.id,
104 username: workspace.slug.clone(),
105 kind: PrincipalKind::Workspace,
106 verified: true,
Workspace names and icons, and a component kit for every control107 workspaces: vec![Membership::member(workspace.slug)],
108 ..User::default()
GitHub Actions on g1t, part two: running workflows109 }))
110 }
111
112 /// The repository, if the viewer may see it and it is not a pull
113 /// request's working copy.
114 async fn visible_repo(&self, path: &RepoPath, viewer: &Viewer) -> Result<Option<Repo>> {
115 let found: Outcome<Repo> = g1t_kit::call(
116 &self.repos,
117 "get",
118 &GetArgs {
119 path: path.clone(),
120 viewer: viewer.clone(),
121 },
122 )
123 .await?;
124 Ok(found.into_result().ok().filter(|repo| repo.fork_of.is_none()))
125 }
126
127 /// A repository by id, as its workspace sees it.
128 async fn repo_by_id(&self, id: &str) -> Result<Option<(Repo, User)>> {
129 let path: Option<RepoPath> = g1t_kit::call(&self.repos, "path_by_id", &g1t_contracts::repos::PathByIdArgs { id: id.to_owned() }).await?;
130 let Some(path) = path else { return Ok(None) };
131 let Some(actor) = self.workspace_actor(&path.namespace).await? else {
132 return Ok(None);
133 };
134 let found: Outcome<Repo> = g1t_kit::call(
135 &self.repos,
136 "get_by_id",
137 &GetByIdArgs {
138 id: id.to_owned(),
139 viewer: Some(actor.clone()),
140 },
141 )
142 .await?;
143 Ok(found.into_result().ok().filter(|repo| repo.fork_of.is_none()).map(|repo| (repo, actor)))
144 }
145
Invite-only launch: sign in with GitHub, repository access and lifecycle, many emails, a new look146 /// The repository at `path`, when `actor` may do `capability` in it:
147 /// not found when they cannot read it, forbidden when their role falls
148 /// short. Agents never may: people and tokens run and change workflows.
149 async fn may(&self, actor: &User, path: &RepoPath, capability: Capability) -> Result<Outcome<Repo>> {
150 if actor.kind == PrincipalKind::Agent {
151 return Ok(fail(FailureCode::Forbidden, "An agent cannot do that. Ask a person."));
152 }
153 let Some(repo) = self.visible_repo(path, &Some(actor.clone())).await? else {
154 return Ok(fail(FailureCode::NotFound, "There is no such repository."));
155 };
156 if !access::can(Some(actor), &repo, capability) {
157 return Ok(fail(
158 FailureCode::Forbidden,
159 access::needs(capability, &format!("{}/{}", repo.namespace, repo.name)),
160 ));
161 }
162 Ok(Outcome::Ok(repo))
GitHub Actions on g1t, part two: running workflows163 }
164}
165
166/// Unwraps an `Outcome`, or returns its failure from the enclosing method.
167#[macro_export]
168macro_rules! check {
169 ($outcome:expr) => {
170 match $outcome {
171 g1t_contracts::Outcome::Ok(value) => value,
172 g1t_contracts::Outcome::Fail(refused) => return Ok(g1t_contracts::Outcome::Fail(refused)),
173 }
174 };
175}
176
177#[event(fetch)]
178async fn fetch(mut request: Request, env: Env, _ctx: Context) -> Result<Response> {
179 let Some(method) = rpc_method(&request) else {
180 return Response::error("Not found", 404);
181 };
182 let body: Value = request.json().await?;
183 let service = Actions::new(&env)?;
184 match method.as_str() {
185 "workflows" => reply(&service.workflows(args(body)?).await?),
186 "runs" => reply(&service.runs(args(body)?).await?),
187 "run" => reply(&service.run(args(body)?).await?),
188 "logs" => reply(&service.logs(args(body)?).await?),
189 "dispatch" => reply(&service.dispatch(args(body)?).await?),
Sidebar: the panels really slide190 "merge_group" => reply(&service.merge_group(args(body)?).await?),
GitHub Actions on g1t, part two: running workflows191 "cancel" => reply(&service.cancel(args(body)?).await?),
192 "rerun" => reply(&service.rerun(args(body)?).await?),
193 "set_workflow_enabled" => reply(&service.set_workflow_enabled(args(body)?).await?),
194 "settings" => reply(&service.settings(args(body)?).await?),
195 "set_setting" => reply(&service.set_setting(args(body)?).await?),
196 "delete_setting" => reply(&service.delete_setting(args(body)?).await?),
Secrets and variables: one list, rows per environment, for workflows and deployments197 "resolve_settings" => reply(&service.resolve_settings(args(body)?).await?),
GitHub Actions on g1t, part two: running workflows198 "job_spec" => reply(&service.job_spec(args(body)?).await?),
A repository has its own sidebar, as settings do199 "job_auth" => reply(&service.job_auth(args(body)?).await?),
GitHub Actions on g1t, part two: running workflows200 "job_report" => reply(&service.job_report(args(body)?).await?),
201 _ => Response::error("Unknown method", 404),
202 }
203}
204
205/// Events from the bus, on this service's own queue.
206#[event(queue)]
207async fn queue(batch: MessageBatch<Event>, env: Env, _ctx: Context) -> Result<()> {
208 let service = Actions::new(&env)?;
209 for message in batch.messages()? {
Agents and memory, checks and conflicts, profiles, slug renames, custom domains210 // A workspace renamed: its rows move to the slug it has now.
211 if g1t_kit::rename::on_event(&env, &env.d1("DB")?, message.body(), rename::STATEMENTS).await? {
212 message.ack();
213 continue;
214 }
Invite-only launch: sign in with GitHub, repository access and lifecycle, many emails, a new look215 // A repository renamed or transferred: its rows follow its new path.
216 if g1t_kit::transfer::on_event(&env, &env.d1("DB")?, message.body(), rename::TRANSFERRED).await? {
217 message.ack();
218 continue;
219 }
220 // A workspace deleted: what it kept for itself goes.
221 if g1t_kit::deleted::on_event(&env.d1("DB")?, message.body(), rename::DELETED).await? {
222 message.ack();
223 continue;
224 }
225 // A repository purged: every row kept for it goes.
226 if g1t_kit::lifecycle::on_purged(&env.d1("DB")?, message.body(), rename::PURGED).await? {
227 message.ack();
228 continue;
229 }
230 // A repository deleted or archived: its runs stop.
231 if let Some(repo_id) = plan::stops_runs(message.body()) {
232 if let Err(error) = service.stop_runs(&repo_id).await {
233 worker::console_error!("actions: event {} failed: {error}", message.body().id);
234 message.retry();
235 continue;
236 }
237 message.ack();
238 continue;
239 }
GitHub Actions on g1t, part two: running workflows240 if let Err(error) = service.on_event(message.body()).await {
241 worker::console_error!("actions: event {} failed: {error}", message.body().id);
242 message.retry();
243 continue;
244 }
245 message.ack();
246 }
247 Ok(())
248}
249
250/// Every minute: schedules that fire, jobs waiting for room, and jobs
251/// whose sandbox went quiet.
252#[event(scheduled)]
253async fn scheduled(_event: ScheduledEvent, env: Env, _ctx: ScheduleContext) {
254 match Actions::new(&env) {
255 Ok(service) => {
256 if let Err(error) = service.on_minute(g1t_kit::now_ms()).await {
257 worker::console_error!("actions: the sweep failed: {error}");
258 }
259 }
260 Err(error) => worker::console_error!("actions: could not start: {error}"),
261 }
262}