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
394 linesCodeBlameRaw
1//! Moving repositories' issues into their workspaces as work items: the
2//! part of migrations/0049_work_items.sql that needs to know which
3//! workspace each repository is in, which only repos does.
4//!
5//! The job pages through repos' repositories (`repo_creators`, then
6//! `visibility`), recording each in `repo_workspaces`, a few pages a run,
7//! resuming where it stopped (`work_jobs.cursor`). Once every page is in,
8//! one batch ([`finish_sql`]) gives every moved issue its workspace and a
9//! number above every number its workspace's repositories used, in the
10//! order the issues were opened; starts each workspace's sequence after
11//! them; merges the repositories' labels into the workspace's by name
12//! (the first made keeps its color and description); renumbers their
13//! milestones in the workspace and moves what was in each; and empties the
14//! per-repository counters. Then it announces where each issue went as
15//! `work.migrated`, a page at a time, for the services that kept issue
16//! numbers. It runs from the cron trigger and from `migrate_work_items`,
17//! and does nothing once done.
18//!
19//! After it, `repo_workspaces` is kept by events: repo.created,
20//! repo.renamed and repo.transferred ([`TRANSFERRED`]),
21//! repo.visibility_changed, and workspace.renamed and workspace.deleted
22//! ([`RENAMED`], [`DELETED`]).
23
24use g1t_contracts::events::WorkMigrated;
25use g1t_contracts::repos::{AllIdsArgs, CreatorPage, RepoVisibility, VisibilityArgs};
26use g1t_contracts::time::rfc3339;
27use g1t_contracts::work::*;
28use g1t_kit::now_ms;
29use serde::Deserialize;
30use worker::Result;
31
32use crate::Work;
33use crate::optional;
34use crate::rows::NumberRow;
35
36/// Repositories asked for at a time, and pages read in one run.
37const PAGE: u32 = 100;
38const PAGES_PER_RUN: u32 = 10;
39/// Moved issues announced in one `work.migrated` event: well under a
40/// queue message's size limit.
41const ANNOUNCED_PER_EVENT: u32 = 250;
42
43/// Records a repository (`?1`) as in workspace `?2` at path `?3`,
44/// private (`?4`) or not, and names its links by that path.
45pub(crate) const RECORD_REPO: [&str; 2] = [
46 super::links::REMEMBER_REPO,
47 "UPDATE work_item_repos SET path = ?3 WHERE repo_id = ?1 AND path IS NOT ?3",
48];
49
50/// The workspace an issue whose repository repos no longer knows was in,
51/// from what else this database kept about that repository (`o.repo_id`).
52const FALLBACK_WORKSPACE: &str = "COALESCE(
53 (SELECT a.workspace FROM agent_runs a WHERE a.repo_id = o.repo_id ORDER BY a.created_at DESC LIMIT 1),
54 (SELECT e.workspace FROM rule_evaluations e WHERE e.repo_id = o.repo_id ORDER BY e.id DESC LIMIT 1),
55 (SELECT s.workspace FROM rulesets s WHERE s.repo_id = o.repo_id LIMIT 1))";
56
57/// What finishing does, in one batch, `?1` the time (see the module's
58/// notes). Each statement changes nothing run again, and those that
59/// would are guarded by the job not being done yet.
60pub(crate) fn finish_sql() -> Vec<String> {
61 let pending = "(SELECT done_at FROM work_jobs WHERE name = 'work_items') IS NULL";
62 vec![
63 // Each moved issue's workspace: its repository's.
64 format!(
65 "UPDATE work_items SET workspace = (
66 SELECT lower(COALESCE(rw.workspace, {FALLBACK_WORKSPACE}))
67 FROM work_item_repos o LEFT JOIN repo_workspaces rw ON rw.repo_id = o.repo_id
68 WHERE o.item_id = work_items.id AND o.role = 'origin')
69 WHERE workspace IS NULL AND number IS NULL"
70 ),
71 // Numbered above every number its workspace used, oldest first.
72 "UPDATE work_items SET number = n.number
73 FROM (
74 SELECT i.id AS id, f.floor + ROW_NUMBER() OVER (PARTITION BY i.workspace ORDER BY i.created_at, i.id) AS number
75 FROM work_items i JOIN (
76 SELECT w.workspace AS workspace, MAX(
77 COALESCE((SELECT MAX(c.last) FROM counters c JOIN repo_workspaces rw ON rw.repo_id = c.repo_id
78 WHERE rw.workspace = w.workspace), 0),
79 COALESCE((SELECT wc.last FROM work_counters wc WHERE wc.workspace = w.workspace), 0),
80 COALESCE((SELECT MAX(x.number) FROM work_items x WHERE x.workspace = w.workspace), 0),
81 COALESCE((SELECT MAX(p.number) FROM pulls p JOIN repo_workspaces rw ON rw.repo_id = p.repo_id
82 WHERE rw.workspace = w.workspace), 0)) AS floor
83 FROM (SELECT DISTINCT workspace FROM work_items WHERE number IS NULL AND workspace IS NOT NULL) w
84 ) f ON f.workspace = i.workspace
85 WHERE i.number IS NULL AND i.workspace IS NOT NULL
86 ) n
87 WHERE work_items.id = n.id"
88 .to_owned(),
89 // Each workspace's sequence goes on after everything it used.
90 "INSERT INTO work_counters (workspace, last)
91 SELECT rw.workspace, MAX(c.last) FROM counters c JOIN repo_workspaces rw ON rw.repo_id = c.repo_id
92 GROUP BY rw.workspace
93 ON CONFLICT (workspace) DO UPDATE SET last = MAX(work_counters.last, excluded.last)"
94 .to_owned(),
95 "INSERT INTO work_counters (workspace, last)
96 SELECT workspace, MAX(number) FROM work_items WHERE workspace IS NOT NULL AND number IS NOT NULL
97 GROUP BY workspace
98 ON CONFLICT (workspace) DO UPDATE SET last = MAX(work_counters.last, excluded.last)"
99 .to_owned(),
100 // Labels merge by name; the first made keeps its color and words.
101 "INSERT INTO labels (workspace, name, color, description, created_at)
102 SELECT rw.workspace, l.name, l.color, l.description, l.created_at
103 FROM repo_labels l JOIN repo_workspaces rw ON rw.repo_id = l.repo_id
104 WHERE true ORDER BY l.created_at, l.repo_id
105 ON CONFLICT (workspace, name) DO UPDATE
106 SET color = excluded.color, description = excluded.description, created_at = excluded.created_at
107 WHERE excluded.created_at < labels.created_at"
108 .to_owned(),
109 // Milestones are numbered on after the workspace's own, oldest
110 // first, and what was in each moves with it.
111 format!(
112 "INSERT INTO milestone_moves (repo_id, old_number, workspace, number)
113 SELECT m.repo_id, m.number, rw.workspace,
114 COALESCE((SELECT MAX(x.number) FROM milestones x WHERE x.workspace = rw.workspace), 0)
115 + ROW_NUMBER() OVER (PARTITION BY rw.workspace ORDER BY m.created_at, m.repo_id, m.number)
116 FROM repo_milestones m JOIN repo_workspaces rw ON rw.repo_id = m.repo_id
117 WHERE {pending}
118 AND NOT EXISTS (SELECT 1 FROM milestone_moves mm WHERE mm.repo_id = m.repo_id AND mm.old_number = m.number)"
119 ),
120 "INSERT INTO milestones (workspace, number, origin_repo_id, title, description, due_on, state, created_at, updated_at, closed_at)
121 SELECT mm.workspace, mm.number, m.repo_id, m.title, m.description, m.due_on, m.state, m.created_at, m.updated_at, m.closed_at
122 FROM milestone_moves mm JOIN repo_milestones m ON m.repo_id = mm.repo_id AND m.number = mm.old_number
123 WHERE true
124 ON CONFLICT DO NOTHING"
125 .to_owned(),
126 format!(
127 "UPDATE work_items SET milestone = (
128 SELECT mm.number FROM milestone_moves mm JOIN work_item_repos o ON o.repo_id = mm.repo_id
129 WHERE o.item_id = work_items.id AND o.role = 'origin' AND mm.old_number = work_items.milestone)
130 WHERE milestone IS NOT NULL AND {pending}
131 AND id IN (SELECT item_id FROM work_item_repos WHERE legacy_number IS NOT NULL)"
132 ),
133 format!(
134 "UPDATE pulls SET milestone = (
135 SELECT mm.number FROM milestone_moves mm WHERE mm.repo_id = pulls.repo_id AND mm.old_number = pulls.milestone)
136 WHERE milestone IS NOT NULL AND {pending}"
137 ),
138 // What the repositories kept for themselves is the workspace's now.
139 format!("DELETE FROM counters WHERE {pending}"),
140 format!("DELETE FROM repo_labels WHERE {pending}"),
141 format!("DELETE FROM repo_milestones WHERE {pending}"),
142 "UPDATE work_jobs SET done_at = ?1, cursor = NULL WHERE name = 'work_items' AND done_at IS NULL".to_owned(),
143 ]
144}
145
146/// A workspace renamed (`g1t_kit::rename`, `?1` its slug now, `?2` before):
147/// its work items, labels, milestones, numbers and repositories follow.
148pub(crate) const RENAMED: &[&str] = &[
149 "UPDATE OR IGNORE work_items SET workspace = ?1 WHERE workspace = ?2",
150 "INSERT INTO labels (workspace, name, color, description, created_at)
151 SELECT ?1, name, color, description, created_at FROM labels WHERE workspace = ?2
152 ON CONFLICT DO NOTHING",
153 "DELETE FROM labels WHERE workspace = ?2",
154 "UPDATE OR IGNORE milestones SET workspace = ?1 WHERE workspace = ?2",
155 "UPDATE milestone_moves SET workspace = ?1 WHERE workspace = ?2",
156 "INSERT INTO work_counters (workspace, last) SELECT ?1, last FROM work_counters WHERE workspace = ?2
157 ON CONFLICT (workspace) DO UPDATE SET last = MAX(work_counters.last, excluded.last)",
158 "DELETE FROM work_counters WHERE workspace = ?2",
159 "UPDATE repo_workspaces SET workspace = ?1, path = ?1 || substr(path, length(?2) + 1)
160 WHERE workspace = ?2 AND substr(path, 1, length(?2) + 1) = ?2 || '/'",
161 "UPDATE work_item_repos SET path = ?1 || substr(path, length(?2) + 1) WHERE substr(path, 1, length(?2) + 1) = ?2 || '/'",
162];
163
164/// A repository renamed or transferred (`g1t_kit::transfer`: `?1` its path
165/// now, `?3` its workspace now, `?5` its id): where it is now, and in the
166/// workspace it moved to, numbers go on after its pull requests', so a new
167/// one never takes a number it has. Its work items stay in their own
168/// workspace, linked to it.
169pub(crate) const TRANSFERRED: &[&str] = &[
170 "INSERT INTO repo_workspaces (repo_id, workspace, path, private) VALUES (?5, ?3, ?1, 1)
171 ON CONFLICT (repo_id) DO UPDATE SET workspace = ?3, path = ?1",
172 "UPDATE work_item_repos SET path = ?1 WHERE repo_id = ?5",
173 "INSERT INTO work_counters (workspace, last)
174 SELECT ?3, MAX(number) FROM pulls WHERE repo_id = ?5 GROUP BY repo_id
175 ON CONFLICT (workspace) DO UPDATE SET last = MAX(work_counters.last, excluded.last)",
176];
177
178/// A workspace deleted (`g1t_kit::deleted`, `?1` its slug): its work
179/// items go, with everything kept for them, and its labels, milestones
180/// and numbers.
181pub(crate) const DELETED: &[&str] = &[
182 "UPDATE pulls SET item_id = NULL WHERE item_id IN (SELECT id FROM work_items WHERE workspace = ?1)",
183 "DELETE FROM comments WHERE item_id IN (SELECT id FROM work_items WHERE workspace = ?1)",
184 "DELETE FROM reactions WHERE item_id IN (SELECT id FROM work_items WHERE workspace = ?1)",
185 "DELETE FROM claims WHERE item_id IN (SELECT id FROM work_items WHERE workspace = ?1)",
186 "DELETE FROM decisions WHERE item_id IN (SELECT id FROM work_items WHERE workspace = ?1)",
187 "UPDATE sessions SET item_id = NULL WHERE item_id IN (SELECT id FROM work_items WHERE workspace = ?1)",
188 "DELETE FROM session_links WHERE item_id IN (SELECT id FROM work_items WHERE workspace = ?1)",
189 "DELETE FROM work_item_repos WHERE item_id IN (SELECT id FROM work_items WHERE workspace = ?1)",
190 "DELETE FROM work_items WHERE workspace = ?1",
191 "DELETE FROM labels WHERE workspace = ?1",
192 "DELETE FROM milestones WHERE workspace = ?1",
193 "DELETE FROM work_counters WHERE workspace = ?1",
194 "DELETE FROM repo_workspaces WHERE workspace = ?1",
195];
196
197/// The default labels for workspace `?1` (name `?2`, color `?3`,
198/// description `?4`, at `?5`), unless it has labels already.
199const SEED_LABEL: &str = "INSERT OR IGNORE INTO labels (workspace, name, color, description, created_at)
200 SELECT ?1, ?2, ?3, ?4, ?5 WHERE NOT EXISTS (SELECT 1 FROM labels WHERE workspace = ?1 AND created_at < ?5)";
201
202#[derive(Deserialize)]
203#[serde(rename_all = "camelCase")]
204struct Created {
205 repo_id: String,
206 namespace: String,
207 name: String,
208 #[serde(default = "private")]
209 is_private: bool,
210}
211
212fn private() -> bool {
213 true
214}
215
216#[derive(Deserialize)]
217struct JobRow {
218 #[serde(default)]
219 cursor: Option<String>,
220 #[serde(default)]
221 done_at: Option<String>,
222}
223
224impl Work {
225 /// `migrate_work_items`, and the cron trigger: see the module's notes.
226 pub(crate) async fn migrate_work_items(&self) -> Result<WorkMigration> {
227 let job = self
228 .db
229 .prepare("SELECT cursor, done_at FROM work_jobs WHERE name = 'work_items'")
230 .first::<JobRow>(None)
231 .await?;
232 let Some(job) = job else {
233 return Ok(WorkMigration { done: true, ..WorkMigration::default() });
234 };
235 if job.done_at.is_some() {
236 return Ok(WorkMigration { done: true, ..WorkMigration::default() });
237 }
238 let mut cursor = job.cursor;
239 let mut recorded = 0;
240 for _ in 0..PAGES_PER_RUN {
241 let page: CreatorPage =
242 g1t_kit::call(&self.repos, "repo_creators", &AllIdsArgs { after: cursor.clone(), limit: PAGE }).await?;
243 recorded += self.record_page(&page).await?;
244 match page.next {
245 Some(next) => {
246 self.db
247 .prepare("UPDATE work_jobs SET cursor = ? WHERE name = 'work_items'")
248 .bind(&[next.as_str().into()])?
249 .run()
250 .await?;
251 cursor = Some(next);
252 }
253 None => return self.finish(recorded).await,
254 }
255 }
256 Ok(WorkMigration { done: false, repos: recorded, ..WorkMigration::default() })
257 }
258
259 /// Records one page of repositories: where each is, and whether it is
260 /// private.
261 async fn record_page(&self, page: &CreatorPage) -> Result<u32> {
262 if page.repos.is_empty() {
263 return Ok(0);
264 }
265 let paths: Vec<String> = page.repos.iter().map(|repo| format!("{}/{}", repo.namespace, repo.name)).collect();
266 let seen: Vec<RepoVisibility> = g1t_kit::call(&self.repos, "visibility", &VisibilityArgs { paths: paths.clone() }).await?;
267 let mut statements = Vec::new();
268 for (repo, path) in page.repos.iter().zip(&paths) {
269 let private = seen.iter().find(|known| known.path.eq_ignore_ascii_case(path)).is_none_or(|known| known.is_private);
270 let binds = [
271 repo.id.as_str().into(),
272 repo.namespace.to_lowercase().into(),
273 path.as_str().into(),
274 u32::from(private).into(),
275 ];
276 for sql in RECORD_REPO {
277 statements.push(self.db.prepare(sql).bind(&binds[..g1t_kit::rename::parameters(sql)])?);
278 }
279 }
280 self.db.batch(statements).await?;
281 Ok(page.repos.len() as u32)
282 }
283
284 /// Numbers every moved issue in its workspace and moves labels and
285 /// milestones, in one batch, then announces where each issue went.
286 async fn finish(&self, recorded: u32) -> Result<WorkMigration> {
287 let now = rfc3339(now_ms());
288 let before = self.unnumbered().await?;
289 let statements = finish_sql()
290 .iter()
291 .map(|sql| {
292 let binds = [now.as_str().into()];
293 self.db.prepare(sql).bind(&binds[..g1t_kit::rename::parameters(sql)])
294 })
295 .collect::<Result<Vec<_>>>()?;
296 self.db.batch(statements).await?;
297 let unplaced = self.unnumbered().await?;
298 if unplaced > 0 {
299 worker::console_warn!("{unplaced} issues were not moved: their repositories are gone, and nothing says where they were");
300 }
301 self.announce_moves().await?;
302 Ok(WorkMigration { done: true, repos: recorded, numbered: before.saturating_sub(unplaced), unplaced })
303 }
304
305 async fn unnumbered(&self) -> Result<u32> {
306 Ok(self
307 .db
308 .prepare("SELECT count(*) AS n FROM work_items WHERE number IS NULL")
309 .first::<NumberRow>(None)
310 .await?
311 .map_or(0, |row| row.n))
312 }
313
314 /// `work.migrated`, a page at a time.
315 async fn announce_moves(&self) -> Result<()> {
316 let mut after = None;
317 let mut page = 0;
318 loop {
319 let found = self.legacy_item_map(LegacyItemMapArgs { after: after.clone(), limit: ANNOUNCED_PER_EVENT }).await?;
320 page += 1;
321 let done = found.next.is_none();
322 let data = serde_json::to_value(WorkMigrated { page, items: found.items, done })?;
323 self.publish_on("work.migrated", None, None, data).await?;
324 match found.next {
325 Some(next) => after = Some(next),
326 None => return Ok(()),
327 }
328 }
329 }
330
331 /// `repo.created`: where the repository is, and for a workspace's
332 /// first, the default labels.
333 pub(super) async fn repo_created(&self, data: &serde_json::Value) -> Result<()> {
334 let Ok(created) = serde_json::from_value::<Created>(data.clone()) else {
335 return Ok(());
336 };
337 let workspace = created.namespace.to_lowercase();
338 let path = format!("{}/{}", created.namespace, created.name);
339 let binds = [
340 created.repo_id.as_str().into(),
341 workspace.as_str().into(),
342 path.as_str().into(),
343 u32::from(created.is_private).into(),
344 ];
345 let mut statements = RECORD_REPO
346 .iter()
347 .map(|sql| self.db.prepare(*sql).bind(&binds[..g1t_kit::rename::parameters(sql)]))
348 .collect::<Result<Vec<_>>>()?;
349 let now = rfc3339(now_ms());
350 for (name, color, description) in DEFAULT_LABELS {
351 statements.push(self.db.prepare(SEED_LABEL).bind(&[
352 workspace.as_str().into(),
353 name.into(),
354 color.into(),
355 description.into(),
356 now.as_str().into(),
357 ])?);
358 }
359 self.db.batch(statements).await?;
360 Ok(())
361 }
362
363 /// `repo.visibility_changed`: whether its work items are public.
364 pub(super) async fn repo_visibility(&self, data: &serde_json::Value) -> Result<()> {
365 let (Some(repo_id), Some(private)) = (data["repoId"].as_str(), data["isPrivate"].as_bool()) else {
366 return Ok(());
367 };
368 self.db
369 .prepare("UPDATE repo_workspaces SET private = ? WHERE repo_id = ?")
370 .bind(&[u32::from(private).into(), repo_id.into()])?
371 .run()
372 .await?;
373 Ok(())
374 }
375
376 /// `legacy_item_map`: a page of where repositories' old issues went.
377 pub(crate) async fn legacy_item_map(&self, a: LegacyItemMapArgs) -> Result<LegacyItemPage> {
378 let limit = a.limit.clamp(1, 500);
379 let items = self
380 .db
381 .prepare(
382 "SELECT r.repo_id, r.legacy_number, i.workspace, i.number, i.id AS item_id
383 FROM work_item_repos r JOIN work_items i ON i.id = r.item_id
384 WHERE r.legacy_number IS NOT NULL AND i.number IS NOT NULL AND (?1 IS NULL OR i.id > ?1)
385 ORDER BY i.id LIMIT ?2",
386 )
387 .bind(&[optional(&a.after), limit.into()])?
388 .all()
389 .await?
390 .results::<MigratedItem>()?;
391 let next = (items.len() as u32 == limit).then(|| items.last().map(|item| item.item_id.clone())).flatten();
392 Ok(LegacyItemPage { items, next, moved: self.items_moved().await? })
393 }
394}