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
1,222 linesCodeBlameRaw
1//! Keeping the index current: from events as things change, from queued
2//! jobs for anything bigger than one event should do, and from a backfill
3//! of everything that was there before.
4//!
5//! The cost of a push is capped. A push lists only the files it changed
6//! (up to [`MAX_PUSH_FILES`]; more than that and the repository is compared
7//! whole instead), queues them, and each job reads at most
8//! [`DRAIN_FILES`] files and [`DRAIN_BYTES`] of text before handing the
9//! rest to the next. A repository is indexed up to [`MAX_REPO_FILES`]
10//! files; past that it is marked partial.
11
12use std::collections::{HashMap, HashSet};
13
14use g1t_contracts::events::{Event, ProjectUpdated, WorkMigrated};
15use g1t_contracts::identity::{DirectoryArgs, DirectoryEntry, DirectoryPage, Profile, SlugArgs, UsernameArgs, Workspace};
16use g1t_contracts::repos::{
17 AllIdsArgs, BlobText, ChangedFilesArgs, FileList, IdPage, ListFilesArgs, PathByIdArgs, ReadBlobsArgs, RepoPath, TreeArgs,
18 TreeView,
19};
20use g1t_contracts::time::rfc3339;
21use g1t_contracts::work::{
22 ListPullsArgs, ListWorkItemsArgs, Pull, PullDetail, State, ViewArgs, WorkItem, WorkItemDetail, WorkItemViewArgs,
23};
24use g1t_contracts::{Membership, Outcome, PrincipalKind, User};
25use g1t_kit::now_ms;
26use serde::{Deserialize, Serialize};
27use worker::Result;
28
29use crate::Search;
30use crate::rules::{self, MAX_FILE_BYTES, SKIP_DIRS};
31use crate::sql::{Param, Sql};
32use crate::store;
33
34/// Files a push may change and still be indexed from its own list; more,
35/// and the repository is compared whole.
36pub const MAX_PUSH_FILES: u32 = 300;
37/// Files indexed in one repository, at most.
38pub const MAX_REPO_FILES: u32 = 5000;
39/// Files one job reads, at most.
40pub const DRAIN_FILES: usize = 60;
41/// Text one job reads, at most.
42pub const DRAIN_BYTES: u64 = 3 * 1024 * 1024;
43/// Blobs asked of the repos service at once.
44const READ_GROUP: usize = 20;
45/// Pieces of a file written in one statement.
46const PIECES_PER_STATEMENT: usize = 8;
47/// The most of a work item's or pull request's body indexed.
48const BODY_CHARS: usize = 32 * 1024;
49/// The most of a README indexed.
50const README_CHARS: usize = 4000;
51/// Repositories listed per backfill job.
52const BACKFILL_PAGE: u32 = 100;
53/// Accounts or workspaces per directory job.
54const DIRECTORY_PAGE: u32 = 200;
55/// Work items read one at a time in one job, at most.
56const ITEMS_PER_JOB: usize = 20;
57
58/// Stores a work item or pull request by its key, as it is now.
59pub const UPSERT_ITEM: &str = "INSERT INTO items (kind, key, repo_id, workspace, number, repos, title, body, state, status,
60 author, requested_by, labels, created_at, updated_at)
61 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
62 ON CONFLICT (key) DO UPDATE SET repo_id = excluded.repo_id, workspace = excluded.workspace,
63 number = excluded.number, repos = excluded.repos, title = excluded.title, body = excluded.body,
64 state = excluded.state, status = excluded.status, author = excluded.author,
65 requested_by = excluded.requested_by, labels = excluded.labels, updated_at = excluded.updated_at";
66
67/// Where `park` keeps a backfill page in `meta`, by its key's prefix.
68const PARKED: &str = "parked_job:";
69
70/// The `meta` key a backfill page is parked under while indexing is
71/// paused: one per kind of walk. None for jobs that are not a backfill's.
72pub fn parked_key(job: &Job) -> Option<String> {
73 match job {
74 Job::Backfill { .. } => Some(format!("{PARKED}backfill")),
75 Job::Directory { kind, .. } => Some(format!("{PARKED}directory:{kind}")),
76 _ => None,
77 }
78}
79
80/// Work queued for later, one job per message.
81#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
82#[serde(tag = "type", rename_all = "snake_case")]
83pub enum Job {
84 /// Lists every repository a page at a time, queueing one `repo` job each.
85 Backfill {
86 #[serde(default)]
87 after: Option<String>,
88 },
89 /// Compares a repository's default branch with the index, whole, and
90 /// indexes its pull requests.
91 Repo { repo_id: String },
92 /// Indexes what a push to the default branch changed.
93 Push {
94 repo_id: String,
95 #[serde(default)]
96 before: Option<String>,
97 after: String,
98 },
99 /// Reads the next files waiting for a repository.
100 Drain { repo_id: String },
101 /// Indexes every account or workspace, a page at a time.
102 Directory {
103 kind: String,
104 #[serde(default)]
105 after: Option<String>,
106 },
107 /// Indexes a workspace's work items, as many as one list returns.
108 Workspace { workspace: String },
109 /// Indexes these work items of a workspace, one at a time.
110 Items { workspace: String, numbers: Vec<u32> },
111}
112
113/// What this service reads of the events it handles.
114#[derive(Deserialize)]
115#[serde(rename_all = "camelCase")]
116struct RepoEvent {
117 repo_id: String,
118}
119
120#[derive(Deserialize)]
121#[serde(rename_all = "camelCase")]
122struct PushEvent {
123 repo_id: String,
124 #[serde(default)]
125 before: Option<String>,
126 after: String,
127 #[serde(default)]
128 default_branch: bool,
129}
130
131#[derive(Deserialize)]
132#[serde(rename_all = "camelCase")]
133struct NumberEvent {
134 repo_id: String,
135 number: u32,
136}
137
138#[derive(Deserialize)]
139#[serde(rename_all = "camelCase")]
140struct ItemEvent {
141 item_id: String,
142 workspace: String,
143 number: u32,
144}
145
146#[derive(Deserialize)]
147struct UserEvent {
148 username: String,
149}
150
151#[derive(Deserialize)]
152struct WorkspaceEvent {
153 slug: String,
154}
155
156#[derive(Deserialize)]
157struct Stored {
158 path: String,
159 blob: String,
160}
161
162#[derive(Deserialize)]
163struct Waiting {
164 path: String,
165 blob: Option<String>,
166}
167
168#[derive(Deserialize)]
169struct RepoState {
170 state: String,
171}
172
173/// Where a row of `items` belongs and what it is linked to.
174struct Row<'a> {
175 kind: &'a str,
176 key: &'a str,
177 repo_id: Option<&'a str>,
178 workspace: Option<&'a str>,
179 number: u32,
180 repos: &'a str,
181}
182
183/// What a row of `items` says.
184struct Text<'a> {
185 title: &'a str,
186 body: &'a str,
187 state: &'a str,
188 status: &'a str,
189 /// Who opened it, and for g1t's work, who asked for it.
190 author: &'a str,
191 requested_by: Option<&'a str>,
192 labels: &'a [String],
193 created_at: &'a str,
194 updated_at: &'a str,
195}
196
197fn now() -> String {
198 rfc3339(now_ms())
199}
200
201fn p(text: &str) -> Param {
202 Param::Text(text.to_owned())
203}
204
205/// A commit id that means "nothing": git's forty zeros.
206fn is_null_commit(hash: &str) -> bool {
207 hash.is_empty() || hash.bytes().all(|b| b == b'0')
208}
209
210fn labels_of(labels: &[String]) -> String {
211 let mut out = String::from("|");
212 for label in labels {
213 out.push_str(&label.to_lowercase());
214 out.push('|');
215 }
216 out
217}
218
219fn clip(text: &str, max: usize) -> String {
220 text.chars().take(max).collect()
221}
222
223/// Repository ids between bars, as `items.repos` keeps them: `|rep_1|rep_2|`.
224fn repos_of(ids: &[&str]) -> String {
225 let mut out = String::from("|");
226 for id in ids {
227 out.push_str(id);
228 out.push('|');
229 }
230 out
231}
232
233/// A pull request's key in `items`.
234fn pull_key(repo_id: &str, number: u32) -> String {
235 format!("{repo_id}/{number}")
236}
237
238/// The jobs that index the work items `work.migrated` announces: grouped
239/// by workspace, a few to a job.
240fn migrated_jobs(migrated: &WorkMigrated) -> Vec<Job> {
241 let mut by_workspace: Vec<(String, Vec<u32>)> = Vec::new();
242 for item in &migrated.items {
243 let workspace = item.workspace.to_lowercase();
244 match by_workspace.iter_mut().find(|(slug, _)| *slug == workspace) {
245 Some((_, numbers)) => numbers.push(item.number),
246 None => by_workspace.push((workspace, vec![item.number])),
247 }
248 }
249 let mut jobs = Vec::new();
250 for (workspace, numbers) in by_workspace {
251 for group in numbers.chunks(ITEMS_PER_JOB) {
252 jobs.push(Job::Items { workspace: workspace.clone(), numbers: group.to_vec() });
253 }
254 }
255 jobs
256}
257
258impl Search {
259 /// The workspace itself, through no one: how this service reads a
260 /// repository, private or not, to index it.
261 async fn actor(&self, slug: &str) -> Result<Option<User>> {
262 let slug = slug.to_lowercase();
263 if let Some(found) = self.actors.borrow().get(&slug) {
264 return Ok(found.clone());
265 }
266 let workspace: Option<Workspace> = g1t_kit::call(&self.identity, "get_workspace", &SlugArgs { slug: slug.clone() }).await?;
267 let actor = workspace.map(|workspace| User {
268 id: workspace.id,
269 username: workspace.slug.clone(),
270 kind: PrincipalKind::Workspace,
271 verified: true,
272 workspaces: vec![Membership::member(workspace.slug)],
273 ..User::default()
274 });
275 self.actors.borrow_mut().insert(slug, actor.clone());
276 Ok(actor)
277 }
278
279 async fn path_of(&self, repo_id: &str) -> Result<Option<RepoPath>> {
280 g1t_kit::call(&self.repos, "path_by_id", &PathByIdArgs { id: repo_id.to_owned() }).await
281 }
282
283 /// Pages of `work.migrated` that never arrived, read from the work
284 /// service (`g1t_kit::work_migrated`): each page's work items are read
285 /// again, as the event's would have been. Whether it is done.
286 pub async fn catch_up_on_moves(&self) -> Result<bool> {
287 g1t_kit::work_migrated::catch_up(&self.db, &self.work, |page| async move { self.enqueue(migrated_jobs(&page)).await }).await
288 }
289
290 pub async fn enqueue(&self, jobs: Vec<Job>) -> Result<()> {
291 let queue = self.env.queue("JOBS")?;
292 for batch in jobs.chunks(100) {
293 queue.send_batch(batch.to_vec()).await?;
294 }
295 Ok(())
296 }
297
298 // ---- Repositories --------------------------------------------------------
299
300 /// Reads a repository's details and README again and stores them. Its
301 /// description is its project's, kept from `project.updated`, so it is
302 /// left as it is here.
303 /// `None`, and the repository forgotten, when it is gone or a fork.
304 pub async fn refresh_repo(&self, repo_id: &str) -> Result<Option<TreeView>> {
305 let Some(path) = self.path_of(repo_id).await? else {
306 self.purge(repo_id).await?;
307 return Ok(None);
308 };
309 let Some(actor) = self.actor(&path.namespace).await? else {
310 return Ok(None);
311 };
312 let tree: Outcome<TreeView> = g1t_kit::call(
313 &self.repos,
314 "tree",
315 &TreeArgs { path, viewer: Some(actor), git_ref: None, tree_path: String::new() },
316 )
317 .await?;
318 let Outcome::Ok(tree) = tree else {
319 return Ok(None);
320 };
321 let repo = &tree.repo;
322 let readme = tree.readme.as_ref().and_then(|readme| readme.text.as_deref()).map(|text| clip(text, README_CHARS));
323 let topics = repo.topics.join(" ");
324 let mut statements = vec![store::prepare(
325 &self.db,
326 "INSERT INTO repos (repo_id, namespace, name, topics, readme, private, default_branch, created_at, updated_at)
327 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
328 ON CONFLICT (repo_id) DO UPDATE SET namespace = excluded.namespace, name = excluded.name,
329 topics = excluded.topics, readme = excluded.readme,
330 private = excluded.private, default_branch = excluded.default_branch, updated_at = excluded.updated_at",
331 vec![
332 p(&repo.id),
333 p(&repo.namespace.to_lowercase()),
334 p(&repo.name.to_lowercase()),
335 p(&topics),
336 Param::opt(readme.as_deref()),
337 Param::Int(i64::from(repo.is_private)),
338 p(&repo.default_branch),
339 p(&repo.created_at),
340 p(&now()),
341 ],
342 )?];
343 statements.push(store::prepare(&self.db, "DELETE FROM repo_topics WHERE repo_id = ?", vec![p(&repo.id)])?);
344 for topic in &repo.topics {
345 statements.push(store::prepare(
346 &self.db,
347 "INSERT OR IGNORE INTO repo_topics (topic, repo_id) VALUES (?, ?)",
348 vec![p(topic), p(&repo.id)],
349 )?);
350 }
351 store::run_all(&self.db, statements).await?;
352 Ok(Some(tree))
353 }
354
355 /// Forgets a repository and everything indexed from it.
356 pub async fn purge(&self, repo_id: &str) -> Result<()> {
357 let id = || vec![p(repo_id)];
358 store::run_all(
359 &self.db,
360 vec![
361 store::prepare(&self.db, "DELETE FROM chunks WHERE fid IN (SELECT fid FROM files WHERE repo_id = ?)", id())?,
362 store::prepare(&self.db, "DELETE FROM files WHERE repo_id = ?", id())?,
363 store::prepare(&self.db, "DELETE FROM pending WHERE repo_id = ?", id())?,
364 // Its work items stay: they are their workspace's, now for
365 // its members only.
366 store::prepare(&self.db, "DELETE FROM items WHERE kind = 'pull' AND repo_id = ?", id())?,
367 store::prepare(&self.db, "DELETE FROM repo_topics WHERE repo_id = ?", id())?,
368 store::prepare(&self.db, "DELETE FROM repos WHERE repo_id = ?", id())?,
369 ],
370 )
371 .await
372 }
373
374 /// Queues files to read (or, with no blob, to remove).
375 async fn queue_files(&self, repo_id: &str, files: &[(String, Option<String>)]) -> Result<()> {
376 let at = now();
377 let mut statements = Vec::new();
378 for group in files.chunks(20) {
379 let rows = vec!["(?, ?, ?, ?)"; group.len()].join(", ");
380 let mut params = Vec::with_capacity(group.len() * 4);
381 for (path, blob) in group {
382 params.extend([p(repo_id), p(path), Param::opt(blob.as_deref()), p(&at)]);
383 }
384 statements.push(store::prepare(
385 &self.db,
386 &format!(
387 "INSERT INTO pending (repo_id, path, blob, queued_at) VALUES {rows}
388 ON CONFLICT (repo_id, path) DO UPDATE SET blob = excluded.blob, queued_at = excluded.queued_at"
389 ),
390 params,
391 )?);
392 }
393 store::run_all(&self.db, statements).await
394 }
395
396 fn skip_dirs() -> Vec<String> {
397 SKIP_DIRS.iter().map(|dir| (*dir).to_owned()).collect()
398 }
399
400 /// Compares a repository's default branch with the index, whole, and
401 /// queues what differs.
402 async fn reconcile(&self, repo_id: &str) -> Result<()> {
403 let listed: FileList = g1t_kit::call(
404 &self.repos,
405 "list_files",
406 &ListFilesArgs { repo_id: repo_id.to_owned(), git_ref: None, skip_dirs: Self::skip_dirs(), limit: MAX_REPO_FILES, links: false },
407 )
408 .await?;
409 let stored: HashMap<String, String> = store::all::<Stored>(
410 &self.db,
411 &Sql { text: "SELECT path, blob FROM files WHERE repo_id = ?".into(), params: vec![p(repo_id)] },
412 )
413 .await?
414 .into_iter()
415 .map(|row| (row.path, row.blob))
416 .collect();
417 let mut changes: Vec<(String, Option<String>)> = Vec::new();
418 let present: HashSet<&str> = listed.files.iter().map(|file| file.path.as_str()).collect();
419 for file in &listed.files {
420 if let Some(hash) = &file.hash
421 && stored.get(&file.path) != Some(hash)
422 {
423 changes.push((file.path.clone(), Some(hash.clone())));
424 }
425 }
426 // Files the branch no longer has; unknowable when the list was cut short.
427 if !listed.truncated {
428 for path in stored.keys().filter(|path| !present.contains(path.as_str())) {
429 changes.push((path.clone(), None));
430 }
431 }
432 self.queue_files(repo_id, &changes).await?;
433 store::run_all(
434 &self.db,
435 vec![store::prepare(
436 &self.db,
437 "UPDATE repos SET state = ?, head = ? WHERE repo_id = ?",
438 vec![p(if listed.truncated { "partial" } else { "indexing" }), Param::opt(listed.commit.as_deref()), p(repo_id)],
439 )?],
440 )
441 .await?;
442 self.drain(repo_id).await
443 }
444
445 /// Reads the next files waiting for a repository, within this job's
446 /// allowance, and queues another job for the rest.
447 pub async fn drain(&self, repo_id: &str) -> Result<()> {
448 let waiting = store::all::<Waiting>(
449 &self.db,
450 &Sql {
451 text: "SELECT path, blob FROM pending WHERE repo_id = ? ORDER BY path LIMIT ?".into(),
452 params: vec![p(repo_id), Param::Int(DRAIN_FILES as i64 + 1)],
453 },
454 )
455 .await?;
456 let more = waiting.len() > DRAIN_FILES;
457 let mut statements = Vec::new();
458 let mut done: Vec<&Waiting> = Vec::new();
459 let mut to_read: Vec<&Waiting> = Vec::new();
460 for file in waiting.iter().take(DRAIN_FILES) {
461 match &file.blob {
462 None => {
463 statements.extend(self.remove_file(repo_id, &file.path)?);
464 done.push(file);
465 }
466 Some(blob) => match rules::skip_path(&file.path) {
467 Some(reason) => {
468 statements.extend(self.skip_file(repo_id, &file.path, blob, 0, reason)?);
469 done.push(file);
470 }
471 None => to_read.push(file),
472 },
473 }
474 }
475 let mut bytes = 0u64;
476 let mut stopped = false;
477 for group in to_read.chunks(READ_GROUP) {
478 if bytes >= DRAIN_BYTES {
479 stopped = true;
480 break;
481 }
482 let hashes: Vec<String> = group.iter().filter_map(|file| file.blob.clone()).collect();
483 let texts: Vec<BlobText> = g1t_kit::call(
484 &self.repos,
485 "read_blobs",
486 &ReadBlobsArgs { repo_id: repo_id.to_owned(), hashes, max_bytes: MAX_FILE_BYTES },
487 )
488 .await?;
489 let by_hash: HashMap<&str, &BlobText> = texts.iter().map(|text| (text.hash.as_str(), text)).collect();
490 for file in group {
491 let blob = file.blob.as_deref().unwrap_or_default();
492 let read = by_hash.get(blob);
493 let size = read.map_or(0, |read| read.size);
494 match read.and_then(|read| read.text.as_deref()) {
495 None => {
496 let reason = if size > u64::from(MAX_FILE_BYTES) { "too large" } else { "binary" };
497 statements.extend(self.skip_file(repo_id, &file.path, blob, size, reason)?);
498 }
499 Some(text) => match rules::skip_text(text) {
500 Some(reason) => statements.extend(self.skip_file(repo_id, &file.path, blob, size, reason)?),
501 None => {
502 statements.extend(self.index_file(repo_id, &file.path, blob, text)?);
503 bytes += text.len() as u64;
504 }
505 },
506 }
507 done.push(file);
508 }
509 }
510 // Each queued file leaves the queue only if no newer push replaced it.
511 for file in &done {
512 statements.push(store::prepare(
513 &self.db,
514 "DELETE FROM pending WHERE repo_id = ? AND path = ? AND blob IS ?",
515 vec![p(repo_id), p(&file.path), Param::opt(file.blob.as_deref())],
516 )?);
517 }
518 store::run_all(&self.db, statements).await?;
519 if more || stopped {
520 return self.enqueue(vec![Job::Drain { repo_id: repo_id.to_owned() }]).await;
521 }
522 self.finish(repo_id).await
523 }
524
525 /// Records a repository's size and language once its queue is empty.
526 async fn finish(&self, repo_id: &str) -> Result<()> {
527 let data: Vec<&str> = rules::DATA_LANGUAGES.to_vec();
528 let state = store::all::<RepoState>(
529 &self.db,
530 &Sql { text: "SELECT state FROM repos WHERE repo_id = ?".into(), params: vec![p(repo_id)] },
531 )
532 .await?;
533 let partial = state.first().is_some_and(|row| row.state == "partial");
534 store::run_all(
535 &self.db,
536 vec![store::prepare(
537 &self.db,
538 "UPDATE repos SET
539 files = (SELECT count(*) FROM files WHERE repo_id = ?1 AND skipped IS NULL),
540 bytes = (SELECT COALESCE(sum(bytes), 0) FROM files WHERE repo_id = ?1 AND skipped IS NULL),
541 language = (SELECT language FROM files WHERE repo_id = ?1 AND skipped IS NULL AND language IS NOT NULL
542 AND language NOT IN (SELECT value FROM json_each(?2))
543 GROUP BY language ORDER BY sum(bytes) DESC LIMIT 1),
544 state = ?3
545 WHERE repo_id = ?1",
546 vec![p(repo_id), Param::Text(serde_json::to_string(&data)?), p(if partial { "partial" } else { "done" })],
547 )?],
548 )
549 .await
550 }
551
552 fn remove_file(&self, repo_id: &str, path: &str) -> Result<Vec<worker::D1PreparedStatement>> {
553 Ok(vec![
554 store::prepare(
555 &self.db,
556 "DELETE FROM chunks WHERE fid = (SELECT fid FROM files WHERE repo_id = ? AND path = ?)",
557 vec![p(repo_id), p(path)],
558 )?,
559 store::prepare(&self.db, "DELETE FROM files WHERE repo_id = ? AND path = ?", vec![p(repo_id), p(path)])?,
560 ])
561 }
562
563 fn upsert_file(&self, repo_id: &str, path: &str, blob: &str, bytes: u64, skipped: Option<&str>) -> Result<worker::D1PreparedStatement> {
564 store::prepare(
565 &self.db,
566 "INSERT INTO files (repo_id, path, blob, language, bytes, skipped) VALUES (?, ?, ?, ?, ?, ?)
567 ON CONFLICT (repo_id, path) DO UPDATE SET blob = excluded.blob, language = excluded.language,
568 bytes = excluded.bytes, skipped = excluded.skipped",
569 vec![p(repo_id), p(path), p(blob), Param::opt(rules::language(path)), Param::Int(bytes as i64), Param::opt(skipped)],
570 )
571 }
572
573 /// Records a file as not indexed, and why, so it is not read again
574 /// until its blob changes.
575 fn skip_file(&self, repo_id: &str, path: &str, blob: &str, bytes: u64, reason: &str) -> Result<Vec<worker::D1PreparedStatement>> {
576 Ok(vec![
577 store::prepare(
578 &self.db,
579 "DELETE FROM chunks WHERE fid = (SELECT fid FROM files WHERE repo_id = ? AND path = ?)",
580 vec![p(repo_id), p(path)],
581 )?,
582 self.upsert_file(repo_id, path, blob, bytes, Some(reason))?,
583 ])
584 }
585
586 fn index_file(&self, repo_id: &str, path: &str, blob: &str, text: &str) -> Result<Vec<worker::D1PreparedStatement>> {
587 let mut statements = vec![
588 self.upsert_file(repo_id, path, blob, text.len() as u64, None)?,
589 store::prepare(
590 &self.db,
591 "DELETE FROM chunks WHERE fid = (SELECT fid FROM files WHERE repo_id = ? AND path = ?)",
592 vec![p(repo_id), p(path)],
593 )?,
594 ];
595 let pieces = rules::chunks(text);
596 for group in pieces.chunks(PIECES_PER_STATEMENT) {
597 let rows = vec!["((SELECT fid FROM files WHERE repo_id = ? AND path = ?), ?, ?, ?)"; group.len()].join(", ");
598 let mut params = Vec::with_capacity(group.len() * 5);
599 for (start, content) in group {
600 params.extend([p(repo_id), p(path), Param::Int(i64::from(*start)), p(path), p(content)]);
601 }
602 statements.push(store::prepare(
603 &self.db,
604 &format!("INSERT INTO chunks (fid, start_line, path, content) VALUES {rows}"),
605 params,
606 )?);
607 }
608 Ok(statements)
609 }
610
611 // ---- Work items and pull requests -------------------------------------
612
613 fn work_item_row(&self, item: &WorkItem) -> Result<worker::D1PreparedStatement> {
614 let status = match item.state {
615 State::Open => "open",
616 State::Closed => item.reason.map_or("completed", |reason| reason.as_str()),
617 };
618 let linked: Vec<&str> = item.repos.iter().map(|repo| repo.repo_id.as_str()).collect();
619 self.item_row(
620 Row {
621 kind: "item",
622 key: &item.id,
623 repo_id: item.origin().map(|repo| repo.repo_id.as_str()),
624 workspace: Some(&item.workspace.to_lowercase()),
625 number: item.number,
626 repos: &repos_of(&linked),
627 },
628 Text {
629 title: &item.title,
630 body: &item.body,
631 state: if item.state == State::Open { "open" } else { "closed" },
632 status,
633 author: &item.author.username,
634 requested_by: item.requested_by.as_ref().map(|user| user.username.as_str()),
635 labels: &item.labels,
636 created_at: &item.created_at,
637 updated_at: &item.updated_at,
638 },
639 )
640 }
641
642 fn pull_row(&self, repo_id: &str, pull: &Pull) -> Result<worker::D1PreparedStatement> {
643 self.item_row(
644 Row {
645 kind: "pull",
646 key: &pull_key(repo_id, pull.number),
647 repo_id: Some(repo_id),
648 workspace: None,
649 number: pull.number,
650 repos: "|",
651 },
652 Text {
653 title: &pull.title,
654 body: pull.body.as_deref().unwrap_or_default(),
655 state: if pull.status.is_active() { "open" } else { "closed" },
656 status: pull.status.as_str(),
657 author: &pull.author.username,
658 requested_by: pull.requested_by.as_ref().map(|user| user.username.as_str()),
659 labels: &pull.labels,
660 created_at: &pull.created_at,
661 updated_at: &pull.updated_at,
662 },
663 )
664 }
665
666 fn item_row(&self, row: Row<'_>, text: Text<'_>) -> Result<worker::D1PreparedStatement> {
667 store::prepare(
668 &self.db,
669 UPSERT_ITEM,
670 vec![
671 p(row.kind),
672 p(row.key),
673 Param::opt(row.repo_id),
674 Param::opt(row.workspace),
675 Param::Int(i64::from(row.number)),
676 p(row.repos),
677 p(text.title),
678 p(&clip(text.body, BODY_CHARS)),
679 p(text.state),
680 p(text.status),
681 p(&text.author.to_lowercase()),
682 text.requested_by.map_or(Param::Null, |name| p(&name.to_lowercase())),
683 p(&labels_of(text.labels)),
684 p(text.created_at),
685 p(text.updated_at),
686 ],
687 )
688 }
689
690 async fn known(&self, repo_id: &str) -> Result<bool> {
691 let found = store::count(
692 &self.db,
693 &Sql { text: "SELECT count(*) AS n FROM repos WHERE repo_id = ?".into(), params: vec![p(repo_id)] },
694 )
695 .await?;
696 Ok(found > 0)
697 }
698
699 /// Indexes one pull request as it is now.
700 async fn index_pull(&self, repo_id: &str, number: u32) -> Result<()> {
701 let Some(path) = self.path_of(repo_id).await? else {
702 return Ok(());
703 };
704 let Some(actor) = self.actor(&path.namespace).await? else {
705 return Ok(());
706 };
707 // A pull request on a repository not yet indexed brings the repository in.
708 if !self.known(repo_id).await? && self.refresh_repo(repo_id).await?.is_none() {
709 return Ok(());
710 }
711 let view = ViewArgs { repo: path, number, viewer: Some(actor), after_seq: 0 };
712 let found: Outcome<PullDetail> = g1t_kit::call(&self.work, "get_pull", &view).await?;
713 let Outcome::Ok(detail) = found else { return Ok(()) };
714 store::run_all(&self.db, vec![self.pull_row(repo_id, &detail.pull)?]).await
715 }
716
717 /// Brings in the repository a work item was filed in, if the index
718 /// does not have it yet: who may read the item follows it.
719 async fn ensure_origin(&self, item: &WorkItem) -> Result<()> {
720 if let Some(origin) = item.origin()
721 && !self.known(&origin.repo_id).await?
722 {
723 self.refresh_repo(&origin.repo_id).await?;
724 }
725 Ok(())
726 }
727
728 /// Indexes one work item as it is now. One no longer found is
729 /// forgotten, when its id is known.
730 async fn index_work_item(&self, workspace: &str, number: u32, item_id: Option<&str>) -> Result<()> {
731 if workspace.is_empty() || number == 0 {
732 return Ok(());
733 }
734 let Some(actor) = self.actor(workspace).await? else {
735 return Ok(());
736 };
737 let view = WorkItemViewArgs { workspace: workspace.to_lowercase(), number, viewer: Some(actor) };
738 let found: Outcome<WorkItemDetail> = g1t_kit::call(&self.work, "get_work_item", &view).await?;
739 let statement = match found {
740 Outcome::Ok(detail) => {
741 self.ensure_origin(&detail.item).await?;
742 self.work_item_row(&detail.item)?
743 }
744 Outcome::Fail(_) => match item_id {
745 Some(id) => store::prepare(&self.db, "DELETE FROM items WHERE kind = 'item' AND key = ?", vec![p(id)])?,
746 None => return Ok(()),
747 },
748 };
749 store::run_all(&self.db, vec![statement]).await
750 }
751
752 /// Indexes a workspace's work items, newest first, as many as one list
753 /// returns.
754 async fn index_work_items(&self, workspace: &str) -> Result<()> {
755 let Some(actor) = self.actor(workspace).await? else {
756 return Ok(());
757 };
758 let items: Outcome<Vec<WorkItem>> = g1t_kit::call(
759 &self.work,
760 "list_work_items",
761 &ListWorkItemsArgs {
762 workspace: workspace.to_lowercase(),
763 viewer: Some(actor),
764 repo: None,
765 role: None,
766 state: None,
767 label: None,
768 milestone: None,
769 assignee: None,
770 closed_since: None,
771 horizon: None,
772 limit: None,
773 before: None,
774 },
775 )
776 .await?;
777 let Outcome::Ok(items) = items else { return Ok(()) };
778 let mut statements = Vec::new();
779 for item in items.iter().filter(|item| item.number > 0) {
780 self.ensure_origin(item).await?;
781 statements.push(self.work_item_row(item)?);
782 }
783 store::run_all(&self.db, statements).await
784 }
785
786 /// Indexes a repository's pull requests, newest first, as many as one
787 /// list returns.
788 async fn index_pulls(&self, repo_id: &str, path: &RepoPath) -> Result<()> {
789 let Some(actor) = self.actor(&path.namespace).await? else {
790 return Ok(());
791 };
792 let pulls: Outcome<Vec<Pull>> = g1t_kit::call(
793 &self.work,
794 "list_pulls",
795 &ListPullsArgs { repo: path.clone(), viewer: Some(actor), state: None, label: None, milestone: None, base: None, before: None, merged_since: None },
796 )
797 .await?;
798 let Outcome::Ok(pulls) = pulls else { return Ok(()) };
799 let mut statements = Vec::new();
800 for pull in &pulls {
801 statements.push(self.pull_row(repo_id, pull)?);
802 }
803 store::run_all(&self.db, statements).await
804 }
805
806 // ---- People and workspaces ---------------------------------------------
807
808 fn person_row(&self, kind: &str, reference: &str, entry: &DirectoryEntry) -> Result<worker::D1PreparedStatement> {
809 store::prepare(
810 &self.db,
811 "INSERT INTO people (kind, ref, slug, name, bio, avatar, created_at) VALUES (?, ?, ?, ?, ?, ?, ?)
812 ON CONFLICT (kind, ref) DO UPDATE SET slug = excluded.slug, name = excluded.name, bio = excluded.bio,
813 avatar = excluded.avatar",
814 vec![
815 p(kind),
816 p(reference),
817 p(&entry.slug.to_lowercase()),
818 Param::opt(entry.name.as_deref()),
819 Param::opt(entry.bio.as_deref()),
820 Param::opt(entry.avatar.as_deref()),
821 p(&entry.created_at),
822 ],
823 )
824 }
825
826 async fn index_user(&self, username: &str) -> Result<()> {
827 let username = username.to_lowercase();
828 let profile: Option<Profile> = g1t_kit::call(&self.identity, "profile", &UsernameArgs { username: username.clone() }).await?;
829 let statement = match profile {
830 Some(profile) => self.person_row(
831 "user",
832 &username,
833 &DirectoryEntry {
834 id: String::new(),
835 slug: profile.username,
836 name: profile.name,
837 bio: profile.bio,
838 avatar: profile.avatar,
839 created_at: profile.created_at,
840 },
841 )?,
842 None => store::prepare(&self.db, "DELETE FROM people WHERE kind = 'user' AND ref = ?", vec![p(&username)])?,
843 };
844 store::run_all(&self.db, vec![statement]).await
845 }
846
847 async fn index_workspace(&self, slug: &str) -> Result<()> {
848 let workspace: Option<Workspace> = g1t_kit::call(&self.identity, "get_workspace", &SlugArgs { slug: slug.to_lowercase() }).await?;
849 let Some(workspace) = workspace else {
850 return Ok(());
851 };
852 let entry = DirectoryEntry {
853 id: workspace.id.clone(),
854 slug: workspace.slug,
855 name: Some(workspace.name),
856 bio: workspace.description,
857 avatar: workspace.avatar,
858 created_at: workspace.created_at,
859 };
860 store::run_all(&self.db, vec![self.person_row("workspace", &workspace.id, &entry)?]).await
861 }
862
863 // ---- Events and jobs -----------------------------------------------------
864
865 /// Starts the backfill the first time the service hears anything, so a
866 /// new deployment fills itself without anyone asking.
867 pub async fn ensure_backfill(&self) -> Result<()> {
868 #[derive(Deserialize)]
869 struct Started {
870 #[allow(dead_code)]
871 key: String,
872 }
873 let started = store::all::<Started>(
874 &self.db,
875 &Sql {
876 text: "INSERT INTO meta (key, value) VALUES ('backfill', ?) ON CONFLICT (key) DO NOTHING RETURNING key".into(),
877 params: vec![p(&now())],
878 },
879 )
880 .await?;
881 if !started.is_empty() {
882 self.enqueue(vec![
883 Job::Backfill { after: None },
884 Job::Directory { kind: "user".into(), after: None },
885 Job::Directory { kind: "workspace".into(), after: None },
886 ])
887 .await?;
888 }
889 Ok(())
890 }
891
892 /// A backfill's next page, set aside while indexing is paused across
893 /// g1t (billing's `platform_pause`), to go on from where it was when it
894 /// is resumed (`resume_parked`). Repositories already queued finish.
895 pub async fn park(&self, job: &Job) -> Result<()> {
896 let Some(key) = parked_key(job) else { return Ok(()) };
897 let value = serde_json::to_string(job)?;
898 store::run_all(
899 &self.db,
900 vec![store::prepare(
901 &self.db,
902 "INSERT INTO meta (key, value) VALUES (?1, ?2) ON CONFLICT (key) DO UPDATE SET value = ?2",
903 vec![p(&key), p(&value)],
904 )?],
905 )
906 .await
907 }
908
909 /// Queues the backfill pages `park` set aside. How many it found.
910 pub async fn resume_parked(&self) -> Result<usize> {
911 #[derive(Deserialize)]
912 struct Parked {
913 value: String,
914 }
915 let parked = store::all::<Parked>(
916 &self.db,
917 &Sql { text: format!("DELETE FROM meta WHERE key LIKE '{PARKED}%' RETURNING value"), params: vec![] },
918 )
919 .await?;
920 let jobs: Vec<Job> = parked.iter().filter_map(|row| serde_json::from_str(&row.value).ok()).collect();
921 let found = jobs.len();
922 if found > 0 {
923 self.enqueue(jobs).await?;
924 }
925 Ok(found)
926 }
927
928 pub async fn on_event(&self, event: &Event) -> Result<()> {
929 let data = &event.data;
930 match event.kind.as_str() {
931 "git.push" => {
932 let Ok(push) = serde_json::from_value::<PushEvent>(data.clone()) else { return Ok(()) };
933 if push.default_branch {
934 self.enqueue(vec![Job::Push { repo_id: push.repo_id, before: push.before, after: push.after }]).await?;
935 }
936 }
937 // A transfer changes the repository's path; refreshing reads it
938 // again by id.
939 // Archiving changes nothing searched, but the details are read again.
940 "repo.created"
941 | "repo.updated"
942 | "repo.visibility_changed"
943 | "repo.renamed"
944 | "repo.transferred"
945 | "repo.archived"
946 | "repo.unarchived" => {
947 if let Ok(repo) = serde_json::from_value::<RepoEvent>(data.clone()) {
948 self.refresh_repo(&repo.repo_id).await?;
949 }
950 }
951 // A repository's description is its own project's. Refreshing
952 // first makes sure it is indexed, whichever event comes first.
953 "project.updated" => {
954 let Ok(project) = serde_json::from_value::<ProjectUpdated>(data.clone()) else { return Ok(()) };
955 if project.primary && self.refresh_repo(&project.repo_id).await?.is_some() {
956 let description = vec![Param::opt(project.description.as_deref()), p(&project.repo_id)];
957 let statement = store::prepare(&self.db, "UPDATE repos SET description = ? WHERE repo_id = ?", description)?;
958 store::run_all(&self.db, vec![statement]).await?;
959 }
960 }
961 // A deleted repository is forgotten at once (repos hides it, so a
962 // refresh would forget it too), and purged for good later.
963 "repo.deleted" | "repo.purged" => {
964 if let Ok(repo) = serde_json::from_value::<RepoEvent>(data.clone()) {
965 self.purge(&repo.repo_id).await?;
966 }
967 }
968 // A restored one is indexed again, whole: its code and pull
969 // requests.
970 "repo.restored" => {
971 if let Ok(repo) = serde_json::from_value::<RepoEvent>(data.clone()) {
972 self.enqueue(vec![Job::Repo { repo_id: repo.repo_id }]).await?;
973 }
974 }
975 // Every change to a work item: its text, state, labels and the
976 // repositories it is linked to are all indexed.
977 kind if kind.starts_with("work_item.") => {
978 if let Ok(item) = serde_json::from_value::<ItemEvent>(data.clone()) {
979 self.index_work_item(&item.workspace, item.number, Some(&item.item_id)).await?;
980 }
981 }
982 // Issues moved into work items keep their ids: each is read
983 // again with its workspace and new number. The SQL migration
984 // dropped what was kept of them as issues.
985 "work.migrated" => {
986 if let Ok(migrated) = serde_json::from_value::<WorkMigrated>(data.clone()) {
987 self.enqueue(migrated_jobs(&migrated)).await?;
988 }
989 }
990 "pull.opened" | "pull.ready" | "pull.updated" | "pull.closed" | "pull.reopened"
991 | "pull.converted_to_draft" | "pull.merged" | "pull.labeled" | "pull.unlabeled" => {
992 if let Ok(pull) = serde_json::from_value::<NumberEvent>(data.clone()) {
993 self.index_pull(&pull.repo_id, pull.number).await?;
994 }
995 }
996 // A deleted account's profile is not found, so indexing it
997 // again drops it from results; a restored one comes back.
998 "user.updated" | "user.deleting" | "user.restored" | "user.deleted" => {
999 if let Ok(user) = serde_json::from_value::<UserEvent>(data.clone()) {
1000 self.index_user(&user.username).await?;
1001 }
1002 }
1003 "workspace.updated" => {
1004 if let Ok(workspace) = serde_json::from_value::<WorkspaceEvent>(data.clone()) {
1005 self.index_workspace(&workspace.slug).await?;
1006 }
1007 }
1008 // A deleted workspace drops out of results at once, and comes
1009 // back if staff restore it. Its repositories come and go with
1010 // their own `repo.deleted` and `repo.restored`.
1011 "workspace.deleting" => {
1012 if let Ok(workspace) = serde_json::from_value::<WorkspaceEvent>(data.clone()) {
1013 self.db
1014 .prepare("DELETE FROM people WHERE kind = 'workspace' AND slug = ?1")
1015 .bind(&[workspace.slug.to_lowercase().into()])?
1016 .run()
1017 .await?;
1018 }
1019 }
1020 "workspace.restored" => {
1021 if let Ok(workspace) = serde_json::from_value::<WorkspaceEvent>(data.clone()) {
1022 self.index_workspace(&workspace.slug).await?;
1023 }
1024 }
1025 "workspace.deleted" => {
1026 g1t_kit::deleted::on_event(
1027 &self.db,
1028 event,
1029 &["DELETE FROM people WHERE kind = 'workspace' AND slug = ?1"],
1030 )
1031 .await?;
1032 }
1033 "workspace.renamed" => {
1034 g1t_kit::rename::on_event(
1035 &self.env,
1036 &self.db,
1037 event,
1038 &[
1039 "UPDATE repos SET namespace = ?1 WHERE namespace = ?2",
1040 "UPDATE items SET workspace = ?1 WHERE workspace = ?2",
1041 "UPDATE people SET slug = ?1 WHERE kind = 'workspace' AND slug = ?2",
1042 ],
1043 )
1044 .await?;
1045 }
1046 _ => {}
1047 }
1048 Ok(())
1049 }
1050
1051 pub async fn run_job(&self, job: Job) -> Result<()> {
1052 match job {
1053 Job::Backfill { after } => {
1054 let page: IdPage = g1t_kit::call(&self.repos, "all_ids", &AllIdsArgs { after, limit: BACKFILL_PAGE }).await?;
1055 let mut jobs: Vec<Job> = page.ids.into_iter().map(|repo_id| Job::Repo { repo_id }).collect();
1056 if let Some(next) = page.next {
1057 jobs.push(Job::Backfill { after: Some(next) });
1058 }
1059 self.enqueue(jobs).await
1060 }
1061 Job::Repo { repo_id } => {
1062 let Some(tree) = self.refresh_repo(&repo_id).await? else { return Ok(()) };
1063 let path = RepoPath { namespace: tree.repo.namespace.clone(), name: tree.repo.name.clone() };
1064 if let Some(head) = &tree.head {
1065 store::run_all(
1066 &self.db,
1067 vec![store::prepare(
1068 &self.db,
1069 "UPDATE repos SET pushed_at = COALESCE(pushed_at, ?) WHERE repo_id = ?",
1070 vec![p(&head.authored_at), p(&repo_id)],
1071 )?],
1072 )
1073 .await?;
1074 }
1075 self.index_pulls(&repo_id, &path).await?;
1076 self.reconcile(&repo_id).await
1077 }
1078 Job::Push { repo_id, before, after } => {
1079 if self.refresh_repo(&repo_id).await?.is_none() {
1080 return Ok(());
1081 }
1082 store::run_all(
1083 &self.db,
1084 vec![store::prepare(&self.db, "UPDATE repos SET pushed_at = ? WHERE repo_id = ?", vec![p(&now()), p(&repo_id)])?],
1085 )
1086 .await?;
1087 let Some(before) = before.filter(|before| !is_null_commit(before)) else {
1088 return self.reconcile(&repo_id).await;
1089 };
1090 let changed: FileList = g1t_kit::call(
1091 &self.repos,
1092 "changed_files",
1093 &ChangedFilesArgs {
1094 repo_id: repo_id.clone(),
1095 base: Some(before),
1096 head: after.clone(),
1097 skip_dirs: Self::skip_dirs(),
1098 limit: MAX_PUSH_FILES,
1099 },
1100 )
1101 .await?;
1102 // A push too large to list, or one whose base is unknown
1103 // (a force push over history the store no longer has): compare whole.
1104 if changed.truncated || changed.commit.is_none() {
1105 return self.reconcile(&repo_id).await;
1106 }
1107 let files: Vec<(String, Option<String>)> =
1108 changed.files.into_iter().map(|file| (file.path, file.hash)).collect();
1109 self.queue_files(&repo_id, &files).await?;
1110 store::run_all(
1111 &self.db,
1112 vec![store::prepare(&self.db, "UPDATE repos SET head = ? WHERE repo_id = ?", vec![p(&after), p(&repo_id)])?],
1113 )
1114 .await?;
1115 self.drain(&repo_id).await
1116 }
1117 Job::Drain { repo_id } => self.drain(&repo_id).await,
1118 Job::Directory { kind, after } => {
1119 let page: DirectoryPage = g1t_kit::call(
1120 &self.identity,
1121 "directory",
1122 &DirectoryArgs { kind: kind.clone(), after, limit: DIRECTORY_PAGE },
1123 )
1124 .await?;
1125 let mut statements = Vec::new();
1126 for entry in &page.entries {
1127 let reference = if kind == "workspace" { entry.id.clone() } else { entry.slug.to_lowercase() };
1128 statements.push(self.person_row(&kind, &reference, entry)?);
1129 }
1130 store::run_all(&self.db, statements).await?;
1131 // Each workspace's work items, a job each.
1132 let mut jobs: Vec<Job> = if kind == "workspace" {
1133 page.entries.iter().map(|entry| Job::Workspace { workspace: entry.slug.to_lowercase() }).collect()
1134 } else {
1135 Vec::new()
1136 };
1137 if let Some(next) = page.next {
1138 jobs.push(Job::Directory { kind, after: Some(next) });
1139 }
1140 self.enqueue(jobs).await
1141 }
1142 Job::Workspace { workspace } => self.index_work_items(&workspace).await,
1143 Job::Items { workspace, numbers } => {
1144 for number in numbers {
1145 self.index_work_item(&workspace, number, None).await?;
1146 }
1147 Ok(())
1148 }
1149 }
1150 }
1151}
1152
1153#[cfg(test)]
1154mod tests {
1155 use g1t_contracts::work::PullStatus;
1156
1157 use super::*;
1158
1159 #[test]
1160 fn jobs_are_sent_tagged() {
1161 let job = Job::Push { repo_id: "rep_1".into(), before: None, after: "abc".into() };
1162 let value = serde_json::to_value(&job).unwrap();
1163 assert_eq!(value["type"], "push");
1164 assert_eq!(serde_json::from_value::<Job>(value).unwrap(), job);
1165 let backfill: Job = serde_json::from_str(r#"{"type":"backfill"}"#).unwrap();
1166 assert_eq!(backfill, Job::Backfill { after: None });
1167 }
1168
1169 #[test]
1170 fn only_a_backfills_pages_are_parked_one_per_walk() {
1171 assert_eq!(parked_key(&Job::Backfill { after: Some("rep_9".into()) }).as_deref(), Some("parked_job:backfill"));
1172 assert_eq!(parked_key(&Job::Directory { kind: "user".into(), after: None }).as_deref(), Some("parked_job:directory:user"));
1173 assert_eq!(parked_key(&Job::Repo { repo_id: "rep_1".into() }), None);
1174 assert_eq!(parked_key(&Job::Drain { repo_id: "rep_1".into() }), None);
1175 }
1176
1177 #[test]
1178 fn a_push_from_nothing_is_a_whole_comparison() {
1179 assert!(is_null_commit("0000000000000000000000000000000000000000"));
1180 assert!(is_null_commit(""));
1181 assert!(!is_null_commit("0a1b"));
1182 }
1183
1184 #[test]
1185 fn labels_are_kept_between_bars() {
1186 assert_eq!(labels_of(&["Bug".into(), "good first issue".into()]), "|bug|good first issue|");
1187 assert_eq!(labels_of(&[]), "|");
1188 }
1189
1190 #[test]
1191 fn migrated_items_are_read_again_by_workspace_a_few_at_a_time() {
1192 use g1t_contracts::work::MigratedItem;
1193 let moved = |workspace: &str, number: u32| MigratedItem {
1194 repo_id: "rep_1".into(),
1195 legacy_number: number,
1196 workspace: workspace.into(),
1197 number: number + 100,
1198 item_id: format!("iss_{number}"),
1199 };
1200 let mut items: Vec<MigratedItem> = (1..=25).map(|n| moved("acme", n)).collect();
1201 items.insert(3, moved("Beta", 7));
1202 let jobs = migrated_jobs(&WorkMigrated { page: 1, items, done: false });
1203 assert_eq!(jobs.len(), 3);
1204 assert_eq!(jobs[0], Job::Items { workspace: "acme".into(), numbers: (101..=120).collect() });
1205 assert_eq!(jobs[1], Job::Items { workspace: "acme".into(), numbers: (121..=125).collect() });
1206 assert_eq!(jobs[2], Job::Items { workspace: "beta".into(), numbers: vec![107] });
1207 let value = serde_json::to_value(&jobs[2]).unwrap();
1208 assert_eq!(value, serde_json::json!({ "type": "items", "workspace": "beta", "numbers": [107] }));
1209 }
1210
1211 #[test]
1212 fn keys_and_linked_repositories() {
1213 assert_eq!(pull_key("rep_1", 12), "rep_1/12");
1214 assert_eq!(repos_of(&["rep_1", "rep_2"]), "|rep_1|rep_2|");
1215 assert_eq!(repos_of(&[]), "|");
1216 }
1217
1218 #[test]
1219 fn pull_statuses_have_names() {
1220 assert_eq!(PullStatus::Merged.as_str(), "merged");
1221 }
1222}