Skip to content
Sign up

flagon-io/g1t

public
0
385 linesCodeBlameRaw

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.

Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays1//! 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.
Merge branch 'worktree-agent-a3abfcce648e87dca'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.
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays16//! - [`cache`] lists `actions/cache` entries, which the API keeps in R2.
Merge remote-tracking branch 'origin/claude/peaceful-franklin-7qtkwx'17//! - [`reach`] finds the actions and reusable workflows of other
18//! repositories on g1t, and [`registry`] resolves those from GitHub to
19//! commits and keeps their archives, for jobs to download from g1t.
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays20//! - [`runners`] keeps self-hosted runners, hands them jobs (and agent
21//! work from the runner service) when they ask, and hears back.
22//!
23//! The service acts as the repository's workspace: it reads what the
Merge branch 'worktree-agent-a3abfcce648e87dca'24//! workspace can read, and a job's `GITHUB_TOKEN` is a token of the
25//! workspace's that reaches the job's repository only, with the scopes its
26//! `permissions:` give it, revoked when the job ends.
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays27
The artifacts service is services/artifacts, the Worker g1t-artifacts, bound as ARTIFACTS by the API, the site and the agents; its live rooms move to it with a Durable Object transfer from g1t-docs-service, and its database, bucket, indexes and queue keep their names. The git store's binding and settings are GITSTORE, its ops scripts gitstore-*, and workflow run artifacts keep their compatible API under run_artifacts modules. The deploy tool puts a Worker that has never deployed before the Workers in its stage that bind to it, and the deploy guide gives the cutover runbook.28mod run_artifacts;
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays29mod cache;
A repository can run a managed workflow: one installed by g1t's own services without a file in the repository. It is listed, started, run and re-run like any workflow in `.g1t/workflows/`, under the path `managed/<name>.yml`, and a workflow of the same `name:` in the repository runs in its place, so moving one into the repository is committing its source. The Actions page says a managed workflow is managed by g1t rather than linking to a file. Nothing installs one yet; the deploy workflow deployments will install, which calls g1t's reusable deploy workflow with a project's settings as its inputs and reports as `g1t / deploy`, is written and tested but not installed.30mod managed;
Merge branch 'mirroring' into artifacts-mode31mod mirrored;
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays32mod payload;
33mod plan;
Merge branch 'worktree-agent-a3abfcce648e87dca'34mod protection;
Merge Actions: cross-repo workflows and actions, release and deployment triggers, step timeouts35mod reach;
Merge remote-tracking branch 'origin/claude/peaceful-franklin-7qtkwx'36mod registry;
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays37mod rename;
Actions: OIDC tokens, the toolkit's cache and artifact services, and artifacts in R238pub mod runtime;
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays39mod runners;
40mod settings;
41mod sync;
42mod trigger;
43mod views;
44
45use g1t_contracts::access::{self, Capability};
46use g1t_contracts::events::Event;
47use g1t_contracts::identity::{SlugArgs, Workspace};
48use g1t_contracts::repos::{GetArgs, GetByIdArgs, Repo, RepoPath};
49use g1t_contracts::{FailureCode, Membership, Outcome, PrincipalKind, User, Viewer};
50use g1t_kit::{args, reply, rpc_method};
51use g1t_secrets::Sealer;
52use serde::Deserialize;
53use serde_json::Value;
54use worker::wasm_bindgen::JsValue;
55use worker::{Context, D1Database, Env, Fetcher, MessageBatch, MessageExt, Request, Response, Result, ScheduleContext, ScheduledEvent, event};
56
57/// The most workflow files read from a repository.
58pub const MAX_WORKFLOWS: usize = 50;
59/// Jobs one workspace may have running at once; the rest wait their turn.
60pub const RUNNING_PER_WORKSPACE: u32 = 4;
What runs differently on g1t's machines is said before it surprises anyone, and four things now behave as on GitHub. A commit whose message says `[skip ci]`, `[ci skip]`, `[no ci]`, `[skip actions]` or `[actions skip]`, or ends with `skip-checks: true`, starts no `push` or `pull_request` workflows. A `pull_request_review` event's `review.state` is `approved` or `changes_requested`, not `approve` or `requestchanges`. A self-hosted job that names no `timeout-minutes` runs for 360 minutes, not 60. A JavaScript action's `::debug::` lines show on a debug run, as `run:` steps' do. The workflow's page now warns when a job asks for more than the 60 minutes g1t's machines give it, when an Arm label will run on x64, and when a matrix on a job that calls a reusable workflow will run it once; it no longer says artifacts are kept up to 60 MB (they are kept up to 5 GiB), nor that `setup-node` with `cache:` runs without its cache (the toolkit's cache works). The desktop release's two hosted jobs ask for the 60 minutes they get. The Actions and self-hosted runners guides say so.61/// The longest a job on g1t's own machines may run, whatever its
62/// `timeout-minutes`.
63pub const MAX_TIMEOUT_MINUTES: u32 = g1t_actions::workflow::HOSTED_MAX_TIMEOUT_MINUTES;
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays64/// The longest a job on a self-hosted runner may run: the machine is the
65/// workspace's own, and its time costs nothing.
66pub const SELF_HOSTED_MAX_TIMEOUT_MINUTES: u32 = 24 * 60;
67/// A running job that has said nothing for this long is taken as lost.
68pub const SILENT_MS: u64 = 10 * 60 * 1000;
Merge branch 'worktree-agent-ab9543c492a7ed481' into spend-guardrails69/// Schedules pause in a repository that has had no push for this long; the
70/// next push resumes them.
71pub const SCHEDULE_IDLE_MS: u64 = 60 * 24 * 60 * 60 * 1000;
72/// How long a workflow's schedule waits after billing refused to start a
73/// job of one of its scheduled runs, before it tries again.
74pub const SCHEDULE_REFUSED_MS: u64 = 60 * 60 * 1000;
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays75pub const SITE: &str = "https://g1t.sh";
76pub const API: &str = "https://api.g1t.sh";
77
78#[derive(Deserialize)]
79struct Count {
80 n: u32,
81}
82
83pub fn optional(value: Option<&str>) -> JsValue {
84 value.map_or(JsValue::NULL, JsValue::from)
85}
86
87pub fn fail<T>(code: FailureCode, message: impl Into<String>) -> Outcome<T> {
88 Outcome::fail(code, message)
89}
90
91/// `owner/name` as a path.
92pub fn repo_path(full_name: &str) -> RepoPath {
93 let (namespace, name) = full_name.split_once('/').unwrap_or((full_name, ""));
94 RepoPath {
95 namespace: namespace.to_owned(),
96 name: name.to_owned(),
97 }
98}
99
100pub struct Actions {
101 db: D1Database,
102 repos: Fetcher,
103 work: Fetcher,
104 identity: Fetcher,
105 runner: Fetcher,
106 events: Fetcher,
107 /// Projects: a repository's secrets and variables belong to its project.
108 projects: Fetcher,
109 /// Billing: self-hosted runners' time, recorded at $0, and the
110 /// cache's storage.
111 billing: Fetcher,
Merge branch 'main' into worktree-agent-a69aeabc4b0deeb97112 /// Deployments: a job with an `environment:` deploys to it, and its
113 /// run's deployment is recorded there (plan.rs `report_deployment`).
114 deployments: Fetcher,
Merge branch 'worktree-agent-a5c94df323642630e'115 /// Notify, optional: pages open on a repository hear its runs' jobs
116 /// start and finish (plan.rs `tell_live`). Without it they catch up
117 /// on their own.
118 notify: Option<Fetcher>,
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays119 /// Where the API keeps cache entries, for deleting evicted ones.
120 cache: Option<worker::Bucket>,
121 /// Seals secrets; absent until `ACTIONS_KEY` is set, when secrets
122 /// cannot be saved.
123 sealer: Option<Sealer>,
Actions: OIDC tokens, the toolkit's cache and artifact services, and artifacts in R2124 /// Signs the toolkit's blob links (runtime.rs), from `ACTIONS_KEY`.
125 blob_key: Option<String>,
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays126}
127
128impl Actions {
129 fn new(env: &Env) -> Result<Self> {
130 Ok(Actions {
131 db: env.d1("DB")?,
132 repos: env.service("REPOS")?,
133 work: env.service("WORK")?,
134 identity: env.service("IDENTITY")?,
135 runner: env.service("RUNNER")?,
136 events: env.service("EVENTS")?,
137 projects: env.service("PROJECTS")?,
138 billing: env.service("BILLING")?,
Merge branch 'main' into worktree-agent-a69aeabc4b0deeb97139 deployments: env.service("DEPLOYMENTS")?,
Merge branch 'worktree-agent-a5c94df323642630e'140 notify: env.service("NOTIFY").ok(),
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays141 cache: env.bucket("ACTIONS_CACHE").ok(),
142 sealer: env.secret("ACTIONS_KEY").ok().and_then(|key| Sealer::new(&key.to_string())),
Actions: OIDC tokens, the toolkit's cache and artifact services, and artifacts in R2143 blob_key: env.secret("ACTIONS_KEY").ok().map(|key| runtime::blob_key(&key.to_string())),
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays144 })
145 }
146
147 /// The workspace itself, as the service acts.
148 async fn workspace_actor(&self, slug: &str) -> Result<Option<User>> {
149 let workspace: Option<Workspace> = g1t_kit::call(&self.identity, "get_workspace", &SlugArgs { slug: slug.to_owned() }).await?;
150 Ok(workspace.map(|workspace| User {
151 id: workspace.id,
152 username: workspace.slug.clone(),
153 kind: PrincipalKind::Workspace,
154 verified: true,
155 workspaces: vec![Membership::member(workspace.slug)],
156 ..User::default()
157 }))
158 }
159
160 /// The repository, if the viewer may see it and it is not a pull
161 /// request's working copy.
162 async fn visible_repo(&self, path: &RepoPath, viewer: &Viewer) -> Result<Option<Repo>> {
163 let found: Outcome<Repo> = g1t_kit::call(
164 &self.repos,
165 "get",
166 &GetArgs {
167 path: path.clone(),
168 viewer: viewer.clone(),
169 },
170 )
171 .await?;
172 Ok(found.into_result().ok().filter(|repo| repo.fork_of.is_none()))
173 }
174
175 /// A repository by id, as its workspace sees it.
176 async fn repo_by_id(&self, id: &str) -> Result<Option<(Repo, User)>> {
177 let path: Option<RepoPath> = g1t_kit::call(&self.repos, "path_by_id", &g1t_contracts::repos::PathByIdArgs { id: id.to_owned() }).await?;
178 let Some(path) = path else { return Ok(None) };
179 let Some(actor) = self.workspace_actor(&path.namespace).await? else {
180 return Ok(None);
181 };
182 let found: Outcome<Repo> = g1t_kit::call(
183 &self.repos,
184 "get_by_id",
185 &GetByIdArgs {
186 id: id.to_owned(),
187 viewer: Some(actor.clone()),
188 },
189 )
190 .await?;
191 Ok(found.into_result().ok().filter(|repo| repo.fork_of.is_none()).map(|repo| (repo, actor)))
192 }
193
194 /// The repository at `path`, when `actor` may do `capability` in it:
195 /// not found when they cannot read it, forbidden when their role falls
196 /// short. Agents never may: people and tokens run and change workflows.
197 async fn may(&self, actor: &User, path: &RepoPath, capability: Capability) -> Result<Outcome<Repo>> {
198 if actor.kind == PrincipalKind::Agent {
199 return Ok(fail(FailureCode::Forbidden, "An agent cannot do that. Ask a person."));
200 }
201 let Some(repo) = self.visible_repo(path, &Some(actor.clone())).await? else {
202 return Ok(fail(FailureCode::NotFound, "There is no such repository."));
203 };
204 if !access::can(Some(actor), &repo, capability) {
205 return Ok(fail(
206 FailureCode::Forbidden,
207 access::needs(capability, &format!("{}/{}", repo.namespace, repo.name)),
208 ));
209 }
210 Ok(Outcome::Ok(repo))
211 }
212}
213
214/// Unwraps an `Outcome`, or returns its failure from the enclosing method.
215#[macro_export]
216macro_rules! check {
217 ($outcome:expr) => {
218 match $outcome {
219 g1t_contracts::Outcome::Ok(value) => value,
220 g1t_contracts::Outcome::Fail(refused) => return Ok(g1t_contracts::Outcome::Fail(refused)),
221 }
222 };
223}
224
225#[event(fetch)]
226async fn fetch(mut request: Request, env: Env, _ctx: Context) -> Result<Response> {
227 let Some(method) = rpc_method(&request) else {
228 return Response::error("Not found", 404);
229 };
230 let body: Value = request.json().await?;
231 let service = Actions::new(&env)?;
232 match method.as_str() {
233 "workflows" => reply(&service.workflows(args(body)?).await?),
234 "runs" => reply(&service.runs(args(body)?).await?),
235 "run" => reply(&service.run(args(body)?).await?),
236 "logs" => reply(&service.logs(args(body)?).await?),
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)237 "job_log_text" => reply(&service.job_log_text(args(body)?).await?),
238 "run_logs" => reply(&service.run_logs(args(body)?).await?),
239 "summaries" => reply(&service.summaries(args(body)?).await?),
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays240 "dispatch" => reply(&service.dispatch(args(body)?).await?),
Merge branch 'worktree-agent-a3abfcce648e87dca'241 "repository_dispatch" => reply(&service.repository_dispatch(args(body)?).await?),
242 "approve_run" => reply(&service.approve_run(args(body)?).await?),
243 "pending_deployments" => reply(&service.pending_deployments(args(body)?).await?),
244 "review_deployments" => reply(&service.review_deployments(args(body)?).await?),
245 "actions_settings" => reply(&service.actions_settings(args(body)?).await?),
246 "set_actions_settings" => reply(&service.set_actions_settings(args(body)?).await?),
247 "workspace_actions_settings" => reply(&service.workspace_actions_settings(args(body)?).await?),
248 "set_workspace_actions_settings" => reply(&service.set_workspace_actions_settings(args(body)?).await?),
249 "environments" => reply(&service.environments(args(body)?).await?),
250 "set_environment" => reply(&service.set_environment(args(body)?).await?),
251 "delete_environment" => reply(&service.delete_environment(args(body)?).await?),
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays252 "merge_group" => reply(&service.merge_group(args(body)?).await?),
253 "cancel" => reply(&service.cancel(args(body)?).await?),
254 "rerun" => reply(&service.rerun(args(body)?).await?),
255 "set_workflow_enabled" => reply(&service.set_workflow_enabled(args(body)?).await?),
256 "settings" => reply(&service.settings(args(body)?).await?),
257 "set_setting" => reply(&service.set_setting(args(body)?).await?),
258 "delete_setting" => reply(&service.delete_setting(args(body)?).await?),
259 "resolve_settings" => reply(&service.resolve_settings(args(body)?).await?),
A repository can run a managed workflow: one installed by g1t's own services without a file in the repository. It is listed, started, run and re-run like any workflow in `.g1t/workflows/`, under the path `managed/<name>.yml`, and a workflow of the same `name:` in the repository runs in its place, so moving one into the repository is committing its source. The Actions page says a managed workflow is managed by g1t rather than linking to a file. Nothing installs one yet; the deploy workflow deployments will install, which calls g1t's reusable deploy workflow with a project's settings as its inputs and reports as `g1t / deploy`, is written and tested but not installed.260 // Managed workflows: g1t's services' side only.
261 "set_managed_workflow" => reply(&service.set_managed_workflow(args(body)?).await?),
262 "remove_managed_workflow" => reply(&service.remove_managed_workflow(args(body)?).await?),
A project can build with a workflow. Turning on **Build with a workflow** under Settings, Deployments installs a managed workflow named `g1t` that calls g1t's reusable deploy workflow (`g1t/workflows/.g1t/workflows/deploy.yml@v1`, published from `workflows/` by the actions release) with the project's settings, and pushes and pull requests then start it rather than a build of Deployments' own: the same previews and production, at the same addresses, reported as `g1t / deploy`, now the check of the workflow's job, listed on the Actions page with its runs and logs. A pull request from a fork runs it with a read-only token, as on GitHub, and still gets its preview without the project's build secrets: a read-only job token may open a preview build only from a run of a managed workflow. Fork pull requests run the repository's managed workflows too. Its jobs are Actions time, at cost plus 20% on g1t's machines and free on a self-hosted runner. A workflow named `g1t` in `.g1t/workflows/` takes its place. The deployments and Actions guides say so.263 "set_managed_deploy" => reply(&service.set_managed_deploy(args(body)?).await?),
Merge GitHub pull request #8 (claude/cool-newton-7nm8r9): Docker deployments on g1t.page behind a switch that stays off until Cloudflare's container APIs are checked live, workflows a workspace manages in its repositories, tarball and zipball downloads, refs/pull/* kept to g1t alone, framework detection for g1t.page builds, and a workflow's event payload matching GitHub's.264 // A workspace's managed workflows: its owners'.
265 "workspace_managed_workflows" => reply(&service.workspace_managed_workflows(args(body)?).await?),
266 "set_workspace_managed_workflow" => reply(&service.set_workspace_managed_workflow(args(body)?).await?),
267 "remove_workspace_managed_workflow" => reply(&service.remove_workspace_managed_workflow(args(body)?).await?),
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays268 "job_spec" => reply(&service.job_spec(args(body)?).await?),
269 "job_auth" => reply(&service.job_auth(args(body)?).await?),
Merge Actions: cross-repo workflows and actions, release and deployment triggers, step timeouts270 "job_action" => reply(&service.job_action(args(body)?).await?),
Merge remote-tracking branch 'origin/claude/peaceful-franklin-7qtkwx'271 "job_action_archive" => reply(&service.job_action_archive(args(body)?).await?),
Merge checks: statuses and check runs on every commit272 "check_runs" => reply(&service.check_runs(args(body)?).await?),
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays273 "job_report" => reply(&service.job_report(args(body)?).await?),
274 // actions/cache, through the API with the job's token.
275 "cache_lookup" => reply(&service.cache_lookup(args(body)?).await?),
276 "cache_reserve" => reply(&service.cache_reserve(args(body)?).await?),
277 "cache_commit" => reply(&service.cache_commit(args(body)?).await?),
278 "cache_abort" => reply(&service.cache_abort(args(body)?).await?),
Actions: OIDC tokens, the toolkit's cache and artifact services, and artifacts in R2279 "cache_upload" => reply(&service.cache_upload(args(body)?).await?),
280 // Artifacts: a job's side, with its token or its runtime token.
281 "artifact_reserve" => reply(&service.artifact_reserve(args(body)?).await?),
282 "artifact_commit" => reply(&service.artifact_commit(args(body)?).await?),
283 "artifact_abort" => reply(&service.artifact_abort(args(body)?).await?),
284 "job_artifacts" => reply(&service.job_artifacts(args(body)?).await?),
285 "job_artifact" => reply(&service.job_artifact(args(body)?).await?),
286 "job_delete_artifact" => reply(&service.job_delete_artifact(args(body)?).await?),
287 // Artifacts: people's side.
288 "artifacts" => reply(&service.artifacts(args(body)?).await?),
289 "artifact" => reply(&service.artifact(args(body)?).await?),
290 "artifact_download" => reply(&service.artifact_download(args(body)?).await?),
291 "delete_artifact" => reply(&service.delete_artifact(args(body)?).await?),
292 "artifact_retention" => reply(&service.artifact_retention(args(body)?).await?),
293 // The toolkit's protocols: the runtime token, OIDC, blob links.
294 "runtime_auth" => reply(&service.runtime_auth(args(body)?).await?),
295 "oidc_claims" => reply(&service.oidc_claims(args(body)?).await?),
296 "blob_sign" => reply(&service.blob_sign(args(body)?).await?),
297 "blob_open" => reply(&service.blob_open(args(body)?).await?),
298 "blob_part" => reply(&service.blob_part(args(body)?).await?),
299 "blob_parts" => reply(&service.blob_parts(args(body)?).await?),
300 "blob_done" => reply(&service.blob_done(args(body)?).await?),
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays301 // Self-hosted runners: people's side.
302 "runners" => reply(&service.runners(args(body)?).await?),
303 "create_registration_token" => reply(&service.create_registration_token(args(body)?).await?),
304 "remove_runner" => reply(&service.remove_runner(args(body)?).await?),
305 "runner_groups" => reply(&service.runner_groups(args(body)?).await?),
306 "set_runner_group" => reply(&service.set_runner_group(args(body)?).await?),
307 "delete_runner_group" => reply(&service.delete_runner_group(args(body)?).await?),
308 "runner_settings" => reply(&service.runner_settings(args(body)?).await?),
309 "set_runner_settings" => reply(&service.set_runner_settings(args(body)?).await?),
Merge the desktop app's runner: it starts with the app for the signed-in person, sandboxed by default and running nothing where no sandbox is available yet, registers through the site, keeps itself up to date, and says everywhere what it is doing, why it is not, and how to fix it.310 // The desktop app's runners, as each app says they are.
311 "report_desktop" => reply(&service.report_desktop(args(body)?).await?),
312 "desktop_runners" => reply(&service.desktop_runners(args(body)?).await?),
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays313 // The runner's own side, through the API with its credential.
314 "runner_register" => reply(&service.runner_register(args(body)?).await?),
315 "runner_poll" => reply(&service.runner_poll(args(body)?).await?),
316 "runner_finished" => reply(&service.runner_finished(args(body)?).await?),
317 "runner_remove_self" => reply(&service.runner_remove_self(args(body)?).await?),
318 // Agent work, from the runner service.
319 "runner_route" => reply(&service.runner_route(args(body)?).await?),
320 "stuck_jobs" => reply(&service.stuck_jobs(args(body)?).await?),
The Runners page shows the machines a workspace's agents and workflow jobs run on, in Workspace under Compute: g1t's cloud beside your own runners, what each is running now with a link to it, what's waiting, this month's machine time and its cost (your own runners' free), where agent work and workflow jobs run, adding a runner, and groups; runner_activity in Actions says what runs where, work handed to your runners keeps its agent run, and the self-hosted runners guide says how.321 "runner_activity" => reply(&service.runner_activity(args(body)?).await?),
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays322 "enqueue_task" => reply(&service.enqueue_task(args(body)?).await?),
323 "cancel_task" => reply(&service.cancel_task(args(body)?).await?),
324 _ => Response::error("Unknown method", 404),
325 }
326}
327
328/// Events from the bus, on this service's own queue.
329#[event(queue)]
330async fn queue(batch: MessageBatch<Event>, env: Env, _ctx: Context) -> Result<()> {
331 let service = Actions::new(&env)?;
332 for message in batch.messages()? {
333 // A workspace renamed: its rows move to the slug it has now.
334 if g1t_kit::rename::on_event(&env, &env.d1("DB")?, message.body(), rename::STATEMENTS).await? {
335 message.ack();
336 continue;
337 }
338 // A repository renamed or transferred: its rows follow its new path.
339 if g1t_kit::transfer::on_event(&env, &env.d1("DB")?, message.body(), rename::TRANSFERRED).await? {
340 message.ack();
341 continue;
342 }
343 // A workspace deleted: what it kept for itself goes.
344 if g1t_kit::deleted::on_event(&env.d1("DB")?, message.body(), rename::DELETED).await? {
345 message.ack();
346 continue;
347 }
348 // A repository purged: every row kept for it goes.
349 if g1t_kit::lifecycle::on_purged(&env.d1("DB")?, message.body(), rename::PURGED).await? {
350 message.ack();
351 continue;
352 }
353 // A repository deleted or archived: its runs stop.
354 if let Some(repo_id) = plan::stops_runs(message.body()) {
355 if let Err(error) = service.stop_runs(&repo_id).await {
356 worker::console_error!("actions: event {} failed: {error}", message.body().id);
357 message.retry();
358 continue;
359 }
360 message.ack();
361 continue;
362 }
363 if let Err(error) = service.on_event(message.body()).await {
364 worker::console_error!("actions: event {} failed: {error}", message.body().id);
365 message.retry();
366 continue;
367 }
368 message.ack();
369 }
370 Ok(())
371}
372
373/// Every minute: schedules that fire, jobs waiting for room, and jobs
374/// whose sandbox went quiet.
375#[event(scheduled)]
376async fn scheduled(_event: ScheduledEvent, env: Env, _ctx: ScheduleContext) {
377 match Actions::new(&env) {
378 Ok(service) => {
379 if let Err(error) = service.on_minute(g1t_kit::now_ms()).await {
380 worker::console_error!("actions: the sweep failed: {error}");
381 }
382 }
383 Err(error) => worker::console_error!("actions: could not start: {error}"),
384 }
385}

This file's history is long; its oldest lines are credited to the oldest commit read.