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.

gitagentscloudflarerustforgemcp
301 linesCodeBlameRaw
1//! Repositories that are archived, deleted or purged, and branches
2//! renamed: what issues, pull requests and runs do about each.
3//!
4//! An archived repository is read-only: its issues and pull requests are
5//! locked and nothing new starts on it. A deleted one looks missing (repos
6//! hides it) and keeps its rows for a restore. A purged one is gone, and
7//! every row kept for it goes with it.
8
9use g1t_contracts::events::{BranchRenamed, Event, RepoArchived, RepoDeleted};
10use g1t_contracts::repos::{Repo, RepoStatus, StatusByIdArgs};
11use g1t_contracts::{FailureCode, Outcome};
12use g1t_contracts::time::rfc3339;
13use g1t_kit::now_ms;
14use serde::{Deserialize, Serialize};
15use worker::Result;
16
17use crate::Work;
18
19/// `Ok` when `repo` may be changed; refused, saying why, when it is
20/// archived.
21pub(crate) fn writable(repo: &Repo) -> Outcome<()> {
22 match repo.read_only_reason() {
23 Some(reason) => Outcome::fail(FailureCode::Forbidden, reason),
24 None => Outcome::Ok(()),
25 }
26}
27
28/// `repo` as found, unless it is archived: then refused, as [`writable`]
29/// says. For the steps g1t starts by itself, which stop on a refusal.
30pub(crate) fn unless_archived(repo: Outcome<Repo>) -> Outcome<Repo> {
31 match repo {
32 Outcome::Ok(repo) => match writable(&repo) {
33 Outcome::Ok(()) => Outcome::Ok(repo),
34 Outcome::Fail(failure) => Outcome::Fail(failure),
35 },
36 failed => failed,
37 }
38}
39
40/// `repo.purged`: every row kept for the repository, `?1` its id. Rows
41/// kept by pull request go first, while the pull requests still name them.
42pub(crate) const PURGED: &[&str] = &[
43 "DELETE FROM session_entries WHERE pull_id IN (SELECT id FROM pulls WHERE repo_id = ?1)",
44 "DELETE FROM session_entries WHERE session_id IN (SELECT id FROM sessions WHERE repo_id = ?1)",
45 "DELETE FROM sessions WHERE repo_id = ?1",
46 "DELETE FROM claims WHERE repo_id = ?1",
47 "DELETE FROM overlap_pairs WHERE repo_id = ?1",
48 "DELETE FROM check_runs WHERE pull_id IN (SELECT id FROM pulls WHERE repo_id = ?1)",
49 "DELETE FROM review_runs WHERE pull_id IN (SELECT id FROM pulls WHERE repo_id = ?1)",
50 "DELETE FROM agent_messages WHERE repo_id = ?1 OR pull_id IN (SELECT id FROM pulls WHERE repo_id = ?1)",
51 "DELETE FROM queue_entries WHERE repo_id = ?1",
52 "DELETE FROM queue_probes WHERE repo_id = ?1",
53 "DELETE FROM decisions WHERE repo_id = ?1",
54 "DELETE FROM agent_mentions WHERE repo_id = ?1",
55 "DELETE FROM agent_rules WHERE repo_id = ?1",
56 "DELETE FROM pull_confidence WHERE repo_id = ?1",
57 "DELETE FROM run_confidence WHERE repo_id = ?1",
58 "DELETE FROM confidence_rules WHERE repo_id = ?1",
59 "DELETE FROM agent_runs WHERE repo_id = ?1",
60 "DELETE FROM commit_statuses WHERE repo_id = ?1",
61 crate::commit_comments::PURGED,
62 "DELETE FROM commit_check_annotations WHERE run_id IN (SELECT id FROM commit_check_runs WHERE repo_id = ?1)",
63 "DELETE FROM commit_check_runs WHERE repo_id = ?1",
64 "DELETE FROM commit_check_suites WHERE repo_id = ?1",
65 "DELETE FROM plans WHERE repo_id = ?1",
66 "DELETE FROM comments WHERE repo_id = ?1",
67 "DELETE FROM pulls WHERE repo_id = ?1",
68 "DELETE FROM issues WHERE repo_id = ?1",
69 "DELETE FROM labels WHERE repo_id = ?1",
70 "DELETE FROM milestones WHERE repo_id = ?1",
71 "DELETE FROM counters WHERE repo_id = ?1",
72 "DELETE FROM repo_settings WHERE repo_id = ?1",
73 // Notes kept when g1t had project memory, no longer read or written.
74 "DELETE FROM memories WHERE scope = 'project' AND scope_key = ?1",
75 "DELETE FROM guardrails WHERE scope <> 'workspace' AND scope_key = ?1",
76];
77
78/// What a run stopped because its repository was archived or deleted says
79/// as its last step; [`Work::runs_in_repo`] finds such runs by it.
80const STOPPED_STEP: &str = "Stopped: the repository was archived or deleted.";
81
82/// `repo.deleted` and `repo.archived`: the repository's runs that
83/// are queued or running stop, `?1` the step they end on, `?2` the time,
84/// `?3` its id.
85const STOP_RUNS: &str = "UPDATE agent_runs SET status = 'stopped', step = ?1, finished_at = ?2, updated_at = ?2
86 WHERE repo_id = ?3 AND status IN ('queued', 'running')";
87
88/// `runs_in_repo` (internal, for the runner): the agent runs of a
89/// repository whose sandboxes should be stopped: those queued or running,
90/// and those work stopped in the last hour because the repository was
91/// archived or deleted. Takes `{ "repoId" }`; returns `[RepoRun]`.
92#[derive(Deserialize)]
93#[serde(rename_all = "camelCase")]
94pub(crate) struct RunsInRepoArgs {
95 pub repo_id: String,
96}
97
98#[derive(Debug, Deserialize, Serialize)]
99#[serde(rename_all = "camelCase")]
100pub(crate) struct RepoRun {
101 pub run_id: String,
102 pub sandbox: String,
103 pub kind: String,
104 pub status: String,
105 pub pull_id: Option<String>,
106}
107
108/// The repository, by id and path, that `event` says was deleted or
109/// archived (not unarchived): its runs stop.
110fn stopping(event: &Event) -> Option<(String, String, String)> {
111 let data = event.data.clone();
112 match event.kind.as_str() {
113 "repo.deleted" => serde_json::from_value::<RepoDeleted>(data).ok().map(|e| (e.repo_id, e.namespace, e.name)),
114 "repo.archived" => serde_json::from_value::<RepoArchived>(data)
115 .ok()
116 .filter(|e| e.archived)
117 .map(|e| (e.repo_id, e.namespace, e.name)),
118 _ => None,
119 }
120}
121
122/// `branch.renamed`: open pull requests from `?2` of repository `?1`
123/// come from `?3` now: those in it from its own branch, and those opened
124/// from it as a fork of another (`head_repo_id`). A working copy carries
125/// its change on its own default branch, so it names none.
126const BRANCH_RENAMED: &str = "UPDATE pulls SET source_branch = ?3
127 WHERE ((repo_id = ?1 AND head_repo_id IS NULL) OR head_repo_id = ?1)
128 AND source_branch = ?2 AND fork_repo_id IS NULL
129 AND status IN ('draft', 'open')";
130
131/// `branch.renamed`: open pull requests into `?2` merge into `?3` now.
132/// Those into the default branch name none, and follow it as they are.
133const BASE_RENAMED: &str = "UPDATE pulls SET base_branch = ?3
134 WHERE repo_id = ?1 AND base_branch = ?2 AND status IN ('draft', 'open')";
135
136impl Work {
137 /// Whether work may start on the repository: neither archived nor
138 /// deleted. When repos cannot say, it is taken as active, and the
139 /// repos calls that follow (which hide deleted repositories) decide.
140 pub(crate) async fn repo_active(&self, repo_id: &str) -> Result<bool> {
141 let status: Result<RepoStatus> = g1t_kit::call(
142 &self.repos,
143 "status_by_id",
144 &StatusByIdArgs {
145 id: repo_id.to_owned(),
146 },
147 )
148 .await;
149 Ok(match status {
150 Ok(status) => status.active(),
151 Err(error) => {
152 worker::console_error!("status_by_id {repo_id}: {error}");
153 true
154 }
155 })
156 }
157
158 /// See [`RunsInRepoArgs`].
159 pub(crate) async fn runs_in_repo(&self, a: RunsInRepoArgs) -> Result<Vec<RepoRun>> {
160 let hour_ago = rfc3339(now_ms().saturating_sub(60 * 60 * 1000));
161 self.db
162 .prepare(
163 "SELECT id AS run_id, sandbox, kind, status, pull_id FROM agent_runs
164 WHERE repo_id = ?1 AND (status IN ('queued', 'running')
165 OR (status = 'stopped' AND step = ?2 AND finished_at >= ?3))
166 ORDER BY created_at LIMIT 200",
167 )
168 .bind(&[a.repo_id.as_str().into(), STOPPED_STEP.into(), hour_ago.as_str().into()])?
169 .all()
170 .await?
171 .results::<RepoRun>()
172 }
173
174 /// `repo.deleted` and `repo.archived`: its runs stop (the runner stops
175 /// their sandboxes, see [`RunsInRepoArgs`]). Rows are kept, for a restore.
176 async fn stop_runs_in(&self, repo_id: &str) -> Result<()> {
177 let now = rfc3339(now_ms());
178 self.db
179 .prepare(STOP_RUNS)
180 .bind(&[STOPPED_STEP.into(), now.as_str().into(), repo_id.into()])?
181 .run()
182 .await?;
183 Ok(())
184 }
185
186 /// `repo.purged`, `repo.deleted`, `repo.archived` and `branch.renamed`.
187 /// Says whether `event` was one this handles completely.
188 pub(crate) async fn on_retired(&self, event: &Event) -> Result<bool> {
189 if g1t_kit::lifecycle::on_purged(&self.db, event, PURGED).await? {
190 return Ok(true);
191 }
192 if let Some((repo_id, _, _)) = stopping(event) {
193 self.stop_runs_in(&repo_id).await?;
194 return Ok(true);
195 }
196 if event.kind != "branch.renamed" {
197 return Ok(false);
198 }
199 let Ok(renamed) = serde_json::from_value::<BranchRenamed>(event.data.clone()) else {
200 worker::console_error!("branch.renamed {} could not be read", event.id);
201 return Ok(true);
202 };
203 if renamed.from == renamed.to {
204 return Ok(true);
205 }
206 let names = [
207 renamed.repo_id.as_str().into(),
208 renamed.from.as_str().into(),
209 renamed.to.as_str().into(),
210 ];
211 self.db
212 .batch(vec![
213 self.db.prepare(BRANCH_RENAMED).bind(&names)?,
214 self.db.prepare(BASE_RENAMED).bind(&names)?,
215 ])
216 .await?;
217 Ok(true)
218 }
219}
220
221#[cfg(test)]
222mod tests {
223 use super::*;
224
225 fn repo(archived_at: Option<&str>) -> Repo {
226 Repo {
227 id: "rep_1".into(),
228 namespace: "acme".into(),
229 name: "web".into(),
230 description: None,
231 is_private: false,
232 owner_id: "usr_1".into(),
233 default_branch: "main".into(),
234 fork_of: None,
235 parent_id: None,
236 forked_from: None,
237 parent: None,
238 allow_forking: false,
239 max_ref_updates: None,
240 protected: false,
241 created_at: "2026-10-05T00:00:00Z".into(),
242 topics: Vec::new(),
243 website: None,
244 archived_at: archived_at.map(str::to_owned),
245 }
246 }
247
248 #[test]
249 fn an_archived_repository_refuses_changes() {
250 assert!(matches!(writable(&repo(None)), Outcome::Ok(())));
251 match writable(&repo(Some("2026-10-05T00:00:00Z"))) {
252 Outcome::Fail(failure) => {
253 assert_eq!(failure.code, FailureCode::Forbidden);
254 assert!(failure.message.contains("acme/web is archived"));
255 }
256 Outcome::Ok(()) => panic!("an archived repository was writable"),
257 }
258 }
259
260 #[test]
261 fn purging_takes_only_the_repository_id() {
262 for sql in PURGED {
263 assert_eq!(g1t_kit::transfer::parameters(sql), 1, "{sql}");
264 }
265 // Rows kept by pull request go before the pull requests do.
266 let pulls = PURGED.iter().position(|sql| sql.starts_with("DELETE FROM pulls")).unwrap();
267 for table in ["session_entries", "check_runs", "review_runs", "agent_messages"] {
268 let at = PURGED.iter().position(|sql| sql.contains(&format!("FROM {table} "))).unwrap();
269 assert!(at < pulls, "{table}");
270 }
271 }
272
273 fn event(kind: &str, data: serde_json::Value) -> Event {
274 Event {
275 id: "evt_1".into(),
276 kind: kind.into(),
277 source: "repos".into(),
278 time: "2026-10-05T00:00:00Z".into(),
279 repo_id: Some("rep_1".into()),
280 actor: None,
281 data,
282 }
283 }
284
285 #[test]
286 fn deleting_or_archiving_stops_runs_and_unarchiving_does_not() {
287 let named = Some(("rep_1".to_owned(), "acme".to_owned(), "web".to_owned()));
288 let data = serde_json::json!({ "repoId": "rep_1", "namespace": "acme", "name": "web", "archived": true });
289 assert_eq!(stopping(&event("repo.deleted", data.clone())), named);
290 assert_eq!(stopping(&event("repo.archived", data)), named);
291 let unarchived = serde_json::json!({ "repoId": "rep_1", "namespace": "acme", "name": "web", "archived": false });
292 assert_eq!(stopping(&event("repo.unarchived", unarchived.clone())), None);
293 assert_eq!(stopping(&event("repo.archived", unarchived)), None);
294 assert_eq!(STOP_RUNS.matches('?').count(), 4);
295 }
296
297 #[test]
298 fn renamed_branches_bind_three_names() {
299 assert_eq!(g1t_kit::transfer::parameters(BRANCH_RENAMED), 3);
300 }
301}