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