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