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