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
622 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, DeletedRepoPage, RepoVisibility, VisibilityArgs};
26use g1t_contracts::time::rfc3339;
27use g1t_contracts::work::*;
28use g1t_kit::now_ms;
29use serde::Deserialize;
30use worker::Result;
31use worker::wasm_bindgen::JsValue;
32
33use crate::Work;
34use crate::optional;
35use crate::rows::NumberRow;
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 super::links::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): [`place_sql`], each workspace's floor, and the job done.
60pub(crate) fn finish_sql() -> Vec<String> {
61 let mut statements = place_sql();
62 statements.push("UPDATE work_jobs SET done_at = ?1, cursor = NULL WHERE name = 'work_items' AND done_at IS NULL".to_owned());
63 statements
64}
65
66/// Placing what the repositories recorded in `repo_workspaces` kept for
67/// themselves, in one batch, `?1` the time: their moved issues' workspaces
68/// and numbers, labels, milestones and numbers become their workspaces'.
69/// Only what is still theirs is moved, and then dropped, so run again it
70/// changes nothing; a repository recorded later (one deleted while the job
71/// ran, then restored, or the job for those, [`DELETED_REPOS_JOB`]) has
72/// its own moved then, and nothing of anyone else's twice.
73pub(crate) fn place_sql() -> Vec<String> {
74 let recorded = "SELECT repo_id FROM repo_workspaces";
75 // A repository's milestones are still its own: not moved yet.
76 let unmoved = "SELECT repo_id FROM repo_milestones WHERE repo_id IN (SELECT repo_id FROM repo_workspaces)";
77 vec![
78 // Each moved issue's workspace: its repository's.
79 format!(
80 "UPDATE work_items SET workspace = (
81 SELECT lower(COALESCE(rw.workspace, {FALLBACK_WORKSPACE}))
82 FROM work_item_repos o LEFT JOIN repo_workspaces rw ON rw.repo_id = o.repo_id
83 WHERE o.item_id = work_items.id AND o.role = 'origin')
84 WHERE workspace IS NULL AND number IS NULL"
85 ),
86 // Numbered above every number its workspace used, oldest first.
87 "UPDATE work_items SET number = n.number
88 FROM (
89 SELECT i.id AS id, f.floor + ROW_NUMBER() OVER (PARTITION BY i.workspace ORDER BY i.created_at, i.id) AS number
90 FROM work_items i JOIN (
91 SELECT w.workspace AS workspace, MAX(
92 COALESCE((SELECT MAX(c.last) FROM counters c JOIN repo_workspaces rw ON rw.repo_id = c.repo_id
93 WHERE rw.workspace = w.workspace), 0),
94 COALESCE((SELECT wc.last FROM work_counters wc WHERE wc.workspace = w.workspace), 0),
95 COALESCE((SELECT MAX(x.number) FROM work_items x WHERE x.workspace = w.workspace), 0),
96 COALESCE((SELECT MAX(p.number) FROM pulls p JOIN repo_workspaces rw ON rw.repo_id = p.repo_id
97 WHERE rw.workspace = w.workspace), 0)) AS floor
98 FROM (SELECT DISTINCT workspace FROM work_items WHERE number IS NULL AND workspace IS NOT NULL) w
99 ) f ON f.workspace = i.workspace
100 WHERE i.number IS NULL AND i.workspace IS NOT NULL
101 ) n
102 WHERE work_items.id = n.id"
103 .to_owned(),
104 // Each workspace's sequence goes on after everything it used.
105 "INSERT INTO work_counters (workspace, last)
106 SELECT rw.workspace, MAX(c.last) FROM counters c JOIN repo_workspaces rw ON rw.repo_id = c.repo_id
107 GROUP BY rw.workspace
108 ON CONFLICT (workspace) DO UPDATE SET last = MAX(work_counters.last, excluded.last)"
109 .to_owned(),
110 "INSERT INTO work_counters (workspace, last)
111 SELECT workspace, MAX(number) FROM work_items WHERE workspace IS NOT NULL AND number IS NOT NULL
112 GROUP BY workspace
113 ON CONFLICT (workspace) DO UPDATE SET last = MAX(work_counters.last, excluded.last)"
114 .to_owned(),
115 // Labels merge by name, whatever its case (names are lowercase as
116 // work items carry them); the first made keeps its color and words.
117 "INSERT INTO labels (workspace, name, color, description, created_at)
118 SELECT rw.workspace, lower(l.name), l.color, l.description, l.created_at
119 FROM repo_labels l JOIN repo_workspaces rw ON rw.repo_id = l.repo_id
120 WHERE true ORDER BY l.created_at, l.repo_id
121 ON CONFLICT (workspace, name) DO UPDATE
122 SET color = excluded.color, description = excluded.description, created_at = excluded.created_at
123 WHERE excluded.created_at < labels.created_at"
124 .to_owned(),
125 // Milestones are numbered on after the workspace's own, oldest
126 // first, and what was in each moves with it.
127 "INSERT INTO milestone_moves (repo_id, old_number, workspace, number)
128 SELECT m.repo_id, m.number, rw.workspace,
129 COALESCE((SELECT MAX(x.number) FROM milestones x WHERE x.workspace = rw.workspace), 0)
130 + ROW_NUMBER() OVER (PARTITION BY rw.workspace ORDER BY m.created_at, m.repo_id, m.number)
131 FROM repo_milestones m JOIN repo_workspaces rw ON rw.repo_id = m.repo_id
132 WHERE NOT EXISTS (SELECT 1 FROM milestone_moves mm WHERE mm.repo_id = m.repo_id AND mm.old_number = m.number)"
133 .to_owned(),
134 "INSERT INTO milestones (workspace, number, origin_repo_id, title, description, due_on, state, created_at, updated_at, closed_at)
135 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
136 FROM milestone_moves mm JOIN repo_milestones m ON m.repo_id = mm.repo_id AND m.number = mm.old_number
137 WHERE true
138 ON CONFLICT DO NOTHING"
139 .to_owned(),
140 // Only what was in a milestone of a repository moving now: a
141 // repository with none keeps what its work items say.
142 format!(
143 "UPDATE work_items SET milestone = (
144 SELECT mm.number FROM milestone_moves mm JOIN work_item_repos o ON o.repo_id = mm.repo_id
145 WHERE o.item_id = work_items.id AND o.role = 'origin' AND mm.old_number = work_items.milestone)
146 WHERE milestone IS NOT NULL
147 AND id IN (SELECT item_id FROM work_item_repos WHERE legacy_number IS NOT NULL AND role = 'origin'
148 AND repo_id IN ({unmoved}))"
149 ),
150 format!(
151 "UPDATE pulls SET milestone = (
152 SELECT mm.number FROM milestone_moves mm WHERE mm.repo_id = pulls.repo_id AND mm.old_number = pulls.milestone)
153 WHERE milestone IS NOT NULL AND repo_id IN ({unmoved})"
154 ),
155 // Every number they used is below the workspace's floor, and its
156 // sequence goes on above it.
157 "UPDATE work_counters SET floor = MAX(floor,
158 COALESCE((SELECT MAX(c.last) FROM counters c JOIN repo_workspaces rw ON rw.repo_id = c.repo_id
159 WHERE rw.workspace = work_counters.workspace), 0))
160 WHERE true"
161 .to_owned(),
162 RAISE_FLOORS.to_owned(),
163 "UPDATE work_counters SET last = floor WHERE last < floor".to_owned(),
164 // What the repositories kept for themselves is the workspace's now.
165 format!("DELETE FROM counters WHERE repo_id IN ({recorded})"),
166 format!("DELETE FROM repo_labels WHERE repo_id IN ({recorded})"),
167 format!("DELETE FROM repo_milestones WHERE repo_id IN ({recorded})"),
168 ]
169}
170
171/// The job, after the first, that records repositories deleted but still
172/// restorable ([`Work::migrate_deleted_repos`]): the first job's paging
173/// left them out, and their work items with no path, private, and their
174/// labels and old numbers unmerged.
175pub(crate) const DELETED_REPOS_JOB: &str = "deleted_repos";
176
177/// The job after that ([`Work::mark_deleted_repos`]): marks each
178/// repository deleted but not purged as deleted in `repo_workspaces`
179/// ([`super::links::MARK_DELETED`]), for those deleted before that was
180/// kept. Run again, it changes nothing.
181pub(crate) const DELETED_MARKS_JOB: &str = "deleted_marks";
182
183/// Every label a workspace's work items or pull requests carry that the
184/// workspace does not have, as a label (gray, no description) made at
185/// `?1`: one a repository had that was never merged. Run again, it adds
186/// nothing.
187pub(crate) const CARRIED_LABELS: [&str; 2] = [
188 "INSERT INTO labels (workspace, name, color, description, created_at)
189 SELECT DISTINCT i.workspace, j.value, 'ededed', '', ?1 FROM work_items i, json_each(i.labels) j
190 WHERE i.workspace IS NOT NULL AND j.value <> ''
191 ON CONFLICT DO NOTHING",
192 "INSERT INTO labels (workspace, name, color, description, created_at)
193 SELECT DISTINCT rw.workspace, j.value, 'ededed', '', ?1 FROM pulls p JOIN repo_workspaces rw ON rw.repo_id = p.repo_id, json_each(p.labels) j
194 WHERE j.value <> ''
195 ON CONFLICT DO NOTHING",
196];
197
198/// A workspace renamed (`g1t_kit::rename`, `?1` its slug now, `?2` before):
199/// its work items, labels, milestones, numbers and repositories follow.
200pub(crate) const RENAMED: &[&str] = &[
201 "UPDATE OR IGNORE work_items SET workspace = ?1 WHERE workspace = ?2",
202 "INSERT INTO labels (workspace, name, color, description, created_at)
203 SELECT ?1, name, color, description, created_at FROM labels WHERE workspace = ?2
204 ON CONFLICT DO NOTHING",
205 "DELETE FROM labels WHERE workspace = ?2",
206 "UPDATE OR IGNORE milestones SET workspace = ?1 WHERE workspace = ?2",
207 "UPDATE milestone_moves SET workspace = ?1 WHERE workspace = ?2",
208 "INSERT INTO work_counters (workspace, last) SELECT ?1, last FROM work_counters WHERE workspace = ?2
209 ON CONFLICT (workspace) DO UPDATE SET last = MAX(work_counters.last, excluded.last)",
210 "DELETE FROM work_counters WHERE workspace = ?2",
211 "UPDATE repo_workspaces SET workspace = ?1, path = ?1 || substr(path, length(?2) + 1)
212 WHERE workspace = ?2 AND substr(path, 1, length(?2) + 1) = ?2 || '/'",
213 "UPDATE work_item_repos SET path = ?1 || substr(path, length(?2) + 1) WHERE substr(path, 1, length(?2) + 1) = ?2 || '/'",
214];
215
216/// A repository renamed or transferred (`g1t_kit::transfer`: `?1` its path
217/// now, `?3` its workspace now, `?5` its id): where it is now, and in the
218/// workspace it moved to, numbers go on after its pull requests' and its
219/// old issues', so a new one never takes a number it has. Those numbers
220/// were never that workspace's, so its floor rises above them too: `#N`
221/// read in the workspace's other repositories names none of them
222/// (refs.rs). A rename (`?4`, its workspace before, the same) changes no
223/// numbers. Its work items stay in their own workspace, linked to it.
224pub(crate) const TRANSFERRED: &[&str] = &[
225 "INSERT INTO repo_workspaces (repo_id, workspace, path, private) VALUES (?5, lower(?3), ?1, 1)
226 ON CONFLICT (repo_id) DO UPDATE SET workspace = lower(?3), path = ?1",
227 "UPDATE work_item_repos SET path = ?1 WHERE repo_id = ?5",
228 "INSERT INTO work_counters (workspace, last, floor)
229 SELECT lower(?3), MAX(n), MAX(n) FROM (
230 SELECT MAX(number) AS n FROM pulls WHERE repo_id = ?5
231 UNION ALL SELECT MAX(legacy_number) FROM work_item_repos WHERE repo_id = ?5)
232 WHERE lower(?3) <> lower(?4)
233 HAVING MAX(n) IS NOT NULL
234 ON CONFLICT (workspace) DO UPDATE SET last = MAX(work_counters.last, excluded.last), floor = MAX(work_counters.floor, excluded.floor)",
235];
236
237/// Raises each workspace's floor (`work_counters.floor`) to the highest
238/// number any of its repositories used on its own: a pull request opened
239/// before work items moved (by when the job finished, or `?1`, now, while
240/// it finishes) or an old issue's number. Run again, it changes nothing.
241/// migrations/0052_work_floors.sql does the same.
242pub(crate) const RAISE_FLOORS: &str = "UPDATE work_counters SET floor = MAX(floor,
243 COALESCE((SELECT MAX(p.number) FROM pulls p JOIN repo_workspaces rw ON rw.repo_id = p.repo_id
244 WHERE rw.workspace = work_counters.workspace
245 AND p.created_at <= COALESCE((SELECT done_at FROM work_jobs WHERE name = 'work_items'), ?1)), 0),
246 COALESCE((SELECT MAX(l.legacy_number) FROM work_item_repos l JOIN repo_workspaces rw ON rw.repo_id = l.repo_id
247 WHERE rw.workspace = work_counters.workspace), 0))
248 WHERE true";
249
250/// A workspace deleted (`g1t_kit::deleted`, `?1` its slug): its work
251/// items go, with everything kept for them, and its labels, milestones
252/// and numbers.
253pub(crate) const DELETED: &[&str] = &[
254 "UPDATE pulls SET item_id = NULL WHERE item_id IN (SELECT id FROM work_items WHERE workspace = ?1)",
255 "DELETE FROM comments WHERE item_id IN (SELECT id FROM work_items WHERE workspace = ?1)",
256 "DELETE FROM reactions WHERE item_id IN (SELECT id FROM work_items WHERE workspace = ?1)",
257 "DELETE FROM claims WHERE item_id IN (SELECT id FROM work_items WHERE workspace = ?1)",
258 "DELETE FROM decisions WHERE item_id IN (SELECT id FROM work_items WHERE workspace = ?1)",
259 "UPDATE sessions SET item_id = NULL WHERE item_id IN (SELECT id FROM work_items WHERE workspace = ?1)",
260 "DELETE FROM session_links WHERE item_id IN (SELECT id FROM work_items WHERE workspace = ?1)",
261 "DELETE FROM work_item_repos WHERE item_id IN (SELECT id FROM work_items WHERE workspace = ?1)",
262 "DELETE FROM work_items WHERE workspace = ?1",
263 "DELETE FROM labels WHERE workspace = ?1",
264 "DELETE FROM milestones WHERE workspace = ?1",
265 "DELETE FROM work_counters WHERE workspace = ?1",
266 "DELETE FROM repo_workspaces WHERE workspace = ?1",
267];
268
269/// The default labels for workspace `?1` (name `?2`, color `?3`,
270/// description `?4`, at `?5`), unless it has labels already.
271const SEED_LABEL: &str = "INSERT OR IGNORE INTO labels (workspace, name, color, description, created_at)
272 SELECT ?1, ?2, ?3, ?4, ?5 WHERE NOT EXISTS (SELECT 1 FROM labels WHERE workspace = ?1 AND created_at < ?5)";
273
274#[derive(Deserialize)]
275#[serde(rename_all = "camelCase")]
276struct Created {
277 repo_id: String,
278 namespace: String,
279 name: String,
280 #[serde(default = "private")]
281 is_private: bool,
282}
283
284fn private() -> bool {
285 true
286}
287
288#[derive(Deserialize)]
289struct JobRow {
290 #[serde(default)]
291 cursor: Option<String>,
292 #[serde(default)]
293 done_at: Option<String>,
294}
295
296impl Work {
297 /// `migrate_work_items`, and the cron trigger: see the module's notes.
298 pub(crate) async fn migrate_work_items(&self) -> Result<WorkMigration> {
299 let job = self
300 .db
301 .prepare("SELECT cursor, done_at FROM work_jobs WHERE name = 'work_items'")
302 .first::<JobRow>(None)
303 .await?;
304 let Some(job) = job else {
305 return Ok(WorkMigration { done: true, ..WorkMigration::default() });
306 };
307 if job.done_at.is_some() {
308 let deleted = self.migrate_deleted_repos().await?;
309 if !deleted.done {
310 return Ok(deleted);
311 }
312 return self.mark_deleted_repos().await;
313 }
314 let mut cursor = job.cursor;
315 let mut recorded = 0;
316 for _ in 0..PAGES_PER_RUN {
317 let page: CreatorPage =
318 g1t_kit::call(&self.repos, "repo_creators", &AllIdsArgs { after: cursor.clone(), limit: PAGE }).await?;
319 recorded += self.record_page(&page).await?;
320 match page.next {
321 Some(next) => {
322 self.db
323 .prepare("UPDATE work_jobs SET cursor = ? WHERE name = 'work_items'")
324 .bind(&[next.as_str().into()])?
325 .run()
326 .await?;
327 cursor = Some(next);
328 }
329 None => return self.finish(recorded).await,
330 }
331 }
332 Ok(WorkMigration { done: false, repos: recorded, ..WorkMigration::default() })
333 }
334
335 /// The job after the first ([`DELETED_REPOS_JOB`]): records each
336 /// repository deleted but not purged, a few pages a run, then places
337 /// what it kept ([`place_sql`]) and gives its work items' labels to
338 /// the workspace ([`CARRIED_LABELS`]). Done once; run again, it
339 /// changes nothing.
340 async fn migrate_deleted_repos(&self) -> Result<WorkMigration> {
341 let job = self
342 .db
343 .prepare("SELECT cursor, done_at FROM work_jobs WHERE name = ?")
344 .bind(&[DELETED_REPOS_JOB.into()])?
345 .first::<JobRow>(None)
346 .await?;
347 let Some(job) = job.filter(|job| job.done_at.is_none()) else {
348 return Ok(WorkMigration { done: true, ..WorkMigration::default() });
349 };
350 let mut cursor = job.cursor;
351 let mut recorded = 0;
352 for _ in 0..PAGES_PER_RUN {
353 let page: DeletedRepoPage =
354 g1t_kit::call(&self.repos, "deleted_repos", &AllIdsArgs { after: cursor.clone(), limit: PAGE }).await?;
355 let repos: Vec<(String, String, String, bool)> = page
356 .repos
357 .iter()
358 .map(|repo| (repo.id.clone(), repo.namespace.clone(), repo.name.clone(), repo.is_private))
359 .collect();
360 recorded += self.record_repos(&repos).await?;
361 // Recording one says it is there; these are deleted.
362 self.mark_deleted(&page).await?;
363 match page.next {
364 Some(next) => {
365 self.db
366 .prepare("UPDATE work_jobs SET cursor = ? WHERE name = ?")
367 .bind(&[next.as_str().into(), DELETED_REPOS_JOB.into()])?
368 .run()
369 .await?;
370 cursor = Some(next);
371 }
372 None => {
373 self.place(true).await?;
374 self.db
375 .prepare("UPDATE work_jobs SET done_at = ?, cursor = NULL WHERE name = ? AND done_at IS NULL")
376 .bind(&[rfc3339(now_ms()).into(), DELETED_REPOS_JOB.into()])?
377 .run()
378 .await?;
379 return Ok(WorkMigration { done: true, repos: recorded, ..WorkMigration::default() });
380 }
381 }
382 }
383 Ok(WorkMigration { done: false, repos: recorded, ..WorkMigration::default() })
384 }
385
386 /// The job after [`Self::migrate_deleted_repos`] ([`DELETED_MARKS_JOB`]):
387 /// marks each repository deleted but not purged as deleted, a few
388 /// pages a run. Done once; run again, it changes nothing.
389 async fn mark_deleted_repos(&self) -> Result<WorkMigration> {
390 let job = self
391 .db
392 .prepare("SELECT cursor, done_at FROM work_jobs WHERE name = ?")
393 .bind(&[DELETED_MARKS_JOB.into()])?
394 .first::<JobRow>(None)
395 .await?;
396 let Some(job) = job.filter(|job| job.done_at.is_none()) else {
397 return Ok(WorkMigration { done: true, ..WorkMigration::default() });
398 };
399 let mut cursor = job.cursor;
400 let mut marked = 0;
401 for _ in 0..PAGES_PER_RUN {
402 let page: DeletedRepoPage =
403 g1t_kit::call(&self.repos, "deleted_repos", &AllIdsArgs { after: cursor.clone(), limit: PAGE }).await?;
404 marked += self.mark_deleted(&page).await?;
405 let done = page.next.is_none();
406 self.db
407 .prepare("UPDATE work_jobs SET cursor = ?, done_at = ? WHERE name = ? AND done_at IS NULL")
408 .bind(&[
409 optional(&page.next),
410 if done { rfc3339(now_ms()).into() } else { JsValue::NULL },
411 DELETED_MARKS_JOB.into(),
412 ])?
413 .run()
414 .await?;
415 if done {
416 return Ok(WorkMigration { done: true, repos: marked, ..WorkMigration::default() });
417 }
418 cursor = page.next;
419 }
420 Ok(WorkMigration { done: false, repos: marked, ..WorkMigration::default() })
421 }
422
423 /// Marks a page of deleted repositories as deleted where they are recorded.
424 async fn mark_deleted(&self, page: &DeletedRepoPage) -> Result<u32> {
425 if page.repos.is_empty() {
426 return Ok(0);
427 }
428 let now = rfc3339(now_ms());
429 let statements = page
430 .repos
431 .iter()
432 .map(|repo| self.db.prepare(super::links::MARK_DELETED).bind(&[repo.id.as_str().into(), now.as_str().into()]))
433 .collect::<Result<Vec<_>>>()?;
434 self.db.batch(statements).await?;
435 Ok(page.repos.len() as u32)
436 }
437
438 /// Records repositories (id, workspace, name, private) where they are.
439 async fn record_repos(&self, repos: &[(String, String, String, bool)]) -> Result<u32> {
440 if repos.is_empty() {
441 return Ok(0);
442 }
443 let mut statements = Vec::new();
444 for (id, namespace, name, private) in repos {
445 let path = format!("{namespace}/{name}");
446 let binds = [id.as_str().into(), namespace.to_lowercase().into(), path.as_str().into(), u32::from(*private).into()];
447 for sql in RECORD_REPO {
448 statements.push(self.db.prepare(sql).bind(&binds[..g1t_kit::rename::parameters(sql)])?);
449 }
450 }
451 self.db.batch(statements).await?;
452 Ok(repos.len() as u32)
453 }
454
455 /// [`place_sql`], and with `carried`, [`CARRIED_LABELS`], in one batch.
456 async fn place(&self, carried: bool) -> Result<()> {
457 let now = rfc3339(now_ms());
458 let mut sql = place_sql();
459 if carried {
460 sql.extend(CARRIED_LABELS.iter().map(|sql| (*sql).to_owned()));
461 }
462 let statements = sql
463 .iter()
464 .map(|sql| {
465 let binds = [now.as_str().into()];
466 self.db.prepare(sql).bind(&binds[..g1t_kit::rename::parameters(sql)])
467 })
468 .collect::<Result<Vec<_>>>()?;
469 self.db.batch(statements).await?;
470 Ok(())
471 }
472
473 /// `repo.restored`: where the repository is again, and, once work items
474 /// have moved, whatever of its own is still to place (a repository
475 /// deleted while the job ran).
476 pub(super) async fn repo_restored(&self, data: &serde_json::Value) -> Result<()> {
477 let Ok(restored) = serde_json::from_value::<Created>(data.clone()) else {
478 return Ok(());
479 };
480 self.record_repos(&[(restored.repo_id, restored.namespace, restored.name, restored.is_private)]).await?;
481 if self.items_moved().await? {
482 self.place(false).await?;
483 }
484 Ok(())
485 }
486
487 /// Records one page of repositories: where each is, and whether it is
488 /// private.
489 async fn record_page(&self, page: &CreatorPage) -> Result<u32> {
490 if page.repos.is_empty() {
491 return Ok(0);
492 }
493 let paths: Vec<String> = page.repos.iter().map(|repo| format!("{}/{}", repo.namespace, repo.name)).collect();
494 let seen: Vec<RepoVisibility> = g1t_kit::call(&self.repos, "visibility", &VisibilityArgs { paths: paths.clone() }).await?;
495 let mut statements = Vec::new();
496 for (repo, path) in page.repos.iter().zip(&paths) {
497 let private = seen.iter().find(|known| known.path.eq_ignore_ascii_case(path)).is_none_or(|known| known.is_private);
498 let binds = [
499 repo.id.as_str().into(),
500 repo.namespace.to_lowercase().into(),
501 path.as_str().into(),
502 u32::from(private).into(),
503 ];
504 for sql in RECORD_REPO {
505 statements.push(self.db.prepare(sql).bind(&binds[..g1t_kit::rename::parameters(sql)])?);
506 }
507 }
508 self.db.batch(statements).await?;
509 Ok(page.repos.len() as u32)
510 }
511
512 /// Numbers every moved issue in its workspace and moves labels and
513 /// milestones, in one batch, then announces where each issue went.
514 async fn finish(&self, recorded: u32) -> Result<WorkMigration> {
515 let now = rfc3339(now_ms());
516 let before = self.unnumbered().await?;
517 let statements = finish_sql()
518 .iter()
519 .map(|sql| {
520 let binds = [now.as_str().into()];
521 self.db.prepare(sql).bind(&binds[..g1t_kit::rename::parameters(sql)])
522 })
523 .collect::<Result<Vec<_>>>()?;
524 self.db.batch(statements).await?;
525 let unplaced = self.unnumbered().await?;
526 if unplaced > 0 {
527 worker::console_warn!("{unplaced} issues were not moved: their repositories are gone, and nothing says where they were");
528 }
529 self.announce_moves().await?;
530 Ok(WorkMigration { done: true, repos: recorded, numbered: before.saturating_sub(unplaced), unplaced })
531 }
532
533 async fn unnumbered(&self) -> Result<u32> {
534 Ok(self
535 .db
536 .prepare("SELECT count(*) AS n FROM work_items WHERE number IS NULL")
537 .first::<NumberRow>(None)
538 .await?
539 .map_or(0, |row| row.n))
540 }
541
542 /// `work.migrated`, a page at a time.
543 async fn announce_moves(&self) -> Result<()> {
544 let mut after = None;
545 let mut page = 0;
546 loop {
547 let found = self.legacy_item_map(LegacyItemMapArgs { after: after.clone(), limit: ANNOUNCED_PER_EVENT }).await?;
548 page += 1;
549 let done = found.next.is_none();
550 let data = serde_json::to_value(WorkMigrated { page, items: found.items, done })?;
551 self.publish_on("work.migrated", None, None, data).await?;
552 match found.next {
553 Some(next) => after = Some(next),
554 None => return Ok(()),
555 }
556 }
557 }
558
559 /// `repo.created`: where the repository is, and for a workspace's
560 /// first, the default labels.
561 pub(super) async fn repo_created(&self, data: &serde_json::Value) -> Result<()> {
562 let Ok(created) = serde_json::from_value::<Created>(data.clone()) else {
563 return Ok(());
564 };
565 let workspace = created.namespace.to_lowercase();
566 let path = format!("{}/{}", created.namespace, created.name);
567 let binds = [
568 created.repo_id.as_str().into(),
569 workspace.as_str().into(),
570 path.as_str().into(),
571 u32::from(created.is_private).into(),
572 ];
573 let mut statements = RECORD_REPO
574 .iter()
575 .map(|sql| self.db.prepare(*sql).bind(&binds[..g1t_kit::rename::parameters(sql)]))
576 .collect::<Result<Vec<_>>>()?;
577 let now = rfc3339(now_ms());
578 for (name, color, description) in DEFAULT_LABELS {
579 statements.push(self.db.prepare(SEED_LABEL).bind(&[
580 workspace.as_str().into(),
581 name.into(),
582 color.into(),
583 description.into(),
584 now.as_str().into(),
585 ])?);
586 }
587 self.db.batch(statements).await?;
588 Ok(())
589 }
590
591 /// `repo.visibility_changed`: whether its work items are public.
592 pub(super) async fn repo_visibility(&self, data: &serde_json::Value) -> Result<()> {
593 let (Some(repo_id), Some(private)) = (data["repoId"].as_str(), data["isPrivate"].as_bool()) else {
594 return Ok(());
595 };
596 self.db
597 .prepare("UPDATE repo_workspaces SET private = ? WHERE repo_id = ?")
598 .bind(&[u32::from(private).into(), repo_id.into()])?
599 .run()
600 .await?;
601 Ok(())
602 }
603
604 /// `legacy_item_map`: a page of where repositories' old issues went.
605 pub(crate) async fn legacy_item_map(&self, a: LegacyItemMapArgs) -> Result<LegacyItemPage> {
606 let limit = a.limit.clamp(1, 500);
607 let items = self
608 .db
609 .prepare(
610 "SELECT r.repo_id, r.legacy_number, i.workspace, i.number, i.id AS item_id
611 FROM work_item_repos r JOIN work_items i ON i.id = r.item_id
612 WHERE r.legacy_number IS NOT NULL AND i.number IS NOT NULL AND (?1 IS NULL OR i.id > ?1)
613 ORDER BY i.id LIMIT ?2",
614 )
615 .bind(&[optional(&a.after), limit.into()])?
616 .all()
617 .await?
618 .results::<MigratedItem>()?;
619 let next = (items.len() as u32 == limit).then(|| items.last().map(|item| item.item_id.clone())).flatten();
620 Ok(LegacyItemPage { items, next, moved: self.items_moved().await? })
621 }
622}