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,220 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 before: None,
772 },
773 )
774 .await?;
775 let Outcome::Ok(items) = items else { return Ok(()) };
776 let mut statements = Vec::new();
777 for item in items.iter().filter(|item| item.number > 0) {
778 self.ensure_origin(item).await?;
779 statements.push(self.work_item_row(item)?);
780 }
781 store::run_all(&self.db, statements).await
782 }
783
784 /// Indexes a repository's pull requests, newest first, as many as one
785 /// list returns.
786 async fn index_pulls(&self, repo_id: &str, path: &RepoPath) -> Result<()> {
787 let Some(actor) = self.actor(&path.namespace).await? else {
788 return Ok(());
789 };
790 let pulls: Outcome<Vec<Pull>> = g1t_kit::call(
791 &self.work,
792 "list_pulls",
793 &ListPullsArgs { repo: path.clone(), viewer: Some(actor), state: None, label: None, milestone: None, base: None, before: None, merged_since: None },
794 )
795 .await?;
796 let Outcome::Ok(pulls) = pulls else { return Ok(()) };
797 let mut statements = Vec::new();
798 for pull in &pulls {
799 statements.push(self.pull_row(repo_id, pull)?);
800 }
801 store::run_all(&self.db, statements).await
802 }
803
804 // ---- People and workspaces ---------------------------------------------
805
806 fn person_row(&self, kind: &str, reference: &str, entry: &DirectoryEntry) -> Result<worker::D1PreparedStatement> {
807 store::prepare(
808 &self.db,
809 "INSERT INTO people (kind, ref, slug, name, bio, avatar, created_at) VALUES (?, ?, ?, ?, ?, ?, ?)
810 ON CONFLICT (kind, ref) DO UPDATE SET slug = excluded.slug, name = excluded.name, bio = excluded.bio,
811 avatar = excluded.avatar",
812 vec![
813 p(kind),
814 p(reference),
815 p(&entry.slug.to_lowercase()),
816 Param::opt(entry.name.as_deref()),
817 Param::opt(entry.bio.as_deref()),
818 Param::opt(entry.avatar.as_deref()),
819 p(&entry.created_at),
820 ],
821 )
822 }
823
824 async fn index_user(&self, username: &str) -> Result<()> {
825 let username = username.to_lowercase();
826 let profile: Option<Profile> = g1t_kit::call(&self.identity, "profile", &UsernameArgs { username: username.clone() }).await?;
827 let statement = match profile {
828 Some(profile) => self.person_row(
829 "user",
830 &username,
831 &DirectoryEntry {
832 id: String::new(),
833 slug: profile.username,
834 name: profile.name,
835 bio: profile.bio,
836 avatar: profile.avatar,
837 created_at: profile.created_at,
838 },
839 )?,
840 None => store::prepare(&self.db, "DELETE FROM people WHERE kind = 'user' AND ref = ?", vec![p(&username)])?,
841 };
842 store::run_all(&self.db, vec![statement]).await
843 }
844
845 async fn index_workspace(&self, slug: &str) -> Result<()> {
846 let workspace: Option<Workspace> = g1t_kit::call(&self.identity, "get_workspace", &SlugArgs { slug: slug.to_lowercase() }).await?;
847 let Some(workspace) = workspace else {
848 return Ok(());
849 };
850 let entry = DirectoryEntry {
851 id: workspace.id.clone(),
852 slug: workspace.slug,
853 name: Some(workspace.name),
854 bio: workspace.description,
855 avatar: workspace.avatar,
856 created_at: workspace.created_at,
857 };
858 store::run_all(&self.db, vec![self.person_row("workspace", &workspace.id, &entry)?]).await
859 }
860
861 // ---- Events and jobs -----------------------------------------------------
862
863 /// Starts the backfill the first time the service hears anything, so a
864 /// new deployment fills itself without anyone asking.
865 pub async fn ensure_backfill(&self) -> Result<()> {
866 #[derive(Deserialize)]
867 struct Started {
868 #[allow(dead_code)]
869 key: String,
870 }
871 let started = store::all::<Started>(
872 &self.db,
873 &Sql {
874 text: "INSERT INTO meta (key, value) VALUES ('backfill', ?) ON CONFLICT (key) DO NOTHING RETURNING key".into(),
875 params: vec![p(&now())],
876 },
877 )
878 .await?;
879 if !started.is_empty() {
880 self.enqueue(vec![
881 Job::Backfill { after: None },
882 Job::Directory { kind: "user".into(), after: None },
883 Job::Directory { kind: "workspace".into(), after: None },
884 ])
885 .await?;
886 }
887 Ok(())
888 }
889
890 /// A backfill's next page, set aside while indexing is paused across
891 /// g1t (billing's `platform_pause`), to go on from where it was when it
892 /// is resumed (`resume_parked`). Repositories already queued finish.
893 pub async fn park(&self, job: &Job) -> Result<()> {
894 let Some(key) = parked_key(job) else { return Ok(()) };
895 let value = serde_json::to_string(job)?;
896 store::run_all(
897 &self.db,
898 vec![store::prepare(
899 &self.db,
900 "INSERT INTO meta (key, value) VALUES (?1, ?2) ON CONFLICT (key) DO UPDATE SET value = ?2",
901 vec![p(&key), p(&value)],
902 )?],
903 )
904 .await
905 }
906
907 /// Queues the backfill pages `park` set aside. How many it found.
908 pub async fn resume_parked(&self) -> Result<usize> {
909 #[derive(Deserialize)]
910 struct Parked {
911 value: String,
912 }
913 let parked = store::all::<Parked>(
914 &self.db,
915 &Sql { text: format!("DELETE FROM meta WHERE key LIKE '{PARKED}%' RETURNING value"), params: vec![] },
916 )
917 .await?;
918 let jobs: Vec<Job> = parked.iter().filter_map(|row| serde_json::from_str(&row.value).ok()).collect();
919 let found = jobs.len();
920 if found > 0 {
921 self.enqueue(jobs).await?;
922 }
923 Ok(found)
924 }
925
926 pub async fn on_event(&self, event: &Event) -> Result<()> {
927 let data = &event.data;
928 match event.kind.as_str() {
929 "git.push" => {
930 let Ok(push) = serde_json::from_value::<PushEvent>(data.clone()) else { return Ok(()) };
931 if push.default_branch {
932 self.enqueue(vec![Job::Push { repo_id: push.repo_id, before: push.before, after: push.after }]).await?;
933 }
934 }
935 // A transfer changes the repository's path; refreshing reads it
936 // again by id.
937 // Archiving changes nothing searched, but the details are read again.
938 "repo.created"
939 | "repo.updated"
940 | "repo.visibility_changed"
941 | "repo.renamed"
942 | "repo.transferred"
943 | "repo.archived"
944 | "repo.unarchived" => {
945 if let Ok(repo) = serde_json::from_value::<RepoEvent>(data.clone()) {
946 self.refresh_repo(&repo.repo_id).await?;
947 }
948 }
949 // A repository's description is its own project's. Refreshing
950 // first makes sure it is indexed, whichever event comes first.
951 "project.updated" => {
952 let Ok(project) = serde_json::from_value::<ProjectUpdated>(data.clone()) else { return Ok(()) };
953 if project.primary && self.refresh_repo(&project.repo_id).await?.is_some() {
954 let description = vec![Param::opt(project.description.as_deref()), p(&project.repo_id)];
955 let statement = store::prepare(&self.db, "UPDATE repos SET description = ? WHERE repo_id = ?", description)?;
956 store::run_all(&self.db, vec![statement]).await?;
957 }
958 }
959 // A deleted repository is forgotten at once (repos hides it, so a
960 // refresh would forget it too), and purged for good later.
961 "repo.deleted" | "repo.purged" => {
962 if let Ok(repo) = serde_json::from_value::<RepoEvent>(data.clone()) {
963 self.purge(&repo.repo_id).await?;
964 }
965 }
966 // A restored one is indexed again, whole: its code and pull
967 // requests.
968 "repo.restored" => {
969 if let Ok(repo) = serde_json::from_value::<RepoEvent>(data.clone()) {
970 self.enqueue(vec![Job::Repo { repo_id: repo.repo_id }]).await?;
971 }
972 }
973 // Every change to a work item: its text, state, labels and the
974 // repositories it is linked to are all indexed.
975 kind if kind.starts_with("work_item.") => {
976 if let Ok(item) = serde_json::from_value::<ItemEvent>(data.clone()) {
977 self.index_work_item(&item.workspace, item.number, Some(&item.item_id)).await?;
978 }
979 }
980 // Issues moved into work items keep their ids: each is read
981 // again with its workspace and new number. The SQL migration
982 // dropped what was kept of them as issues.
983 "work.migrated" => {
984 if let Ok(migrated) = serde_json::from_value::<WorkMigrated>(data.clone()) {
985 self.enqueue(migrated_jobs(&migrated)).await?;
986 }
987 }
988 "pull.opened" | "pull.ready" | "pull.updated" | "pull.closed" | "pull.reopened"
989 | "pull.converted_to_draft" | "pull.merged" | "pull.labeled" | "pull.unlabeled" => {
990 if let Ok(pull) = serde_json::from_value::<NumberEvent>(data.clone()) {
991 self.index_pull(&pull.repo_id, pull.number).await?;
992 }
993 }
994 // A deleted account's profile is not found, so indexing it
995 // again drops it from results; a restored one comes back.
996 "user.updated" | "user.deleting" | "user.restored" | "user.deleted" => {
997 if let Ok(user) = serde_json::from_value::<UserEvent>(data.clone()) {
998 self.index_user(&user.username).await?;
999 }
1000 }
1001 "workspace.updated" => {
1002 if let Ok(workspace) = serde_json::from_value::<WorkspaceEvent>(data.clone()) {
1003 self.index_workspace(&workspace.slug).await?;
1004 }
1005 }
1006 // A deleted workspace drops out of results at once, and comes
1007 // back if staff restore it. Its repositories come and go with
1008 // their own `repo.deleted` and `repo.restored`.
1009 "workspace.deleting" => {
1010 if let Ok(workspace) = serde_json::from_value::<WorkspaceEvent>(data.clone()) {
1011 self.db
1012 .prepare("DELETE FROM people WHERE kind = 'workspace' AND slug = ?1")
1013 .bind(&[workspace.slug.to_lowercase().into()])?
1014 .run()
1015 .await?;
1016 }
1017 }
1018 "workspace.restored" => {
1019 if let Ok(workspace) = serde_json::from_value::<WorkspaceEvent>(data.clone()) {
1020 self.index_workspace(&workspace.slug).await?;
1021 }
1022 }
1023 "workspace.deleted" => {
1024 g1t_kit::deleted::on_event(
1025 &self.db,
1026 event,
1027 &["DELETE FROM people WHERE kind = 'workspace' AND slug = ?1"],
1028 )
1029 .await?;
1030 }
1031 "workspace.renamed" => {
1032 g1t_kit::rename::on_event(
1033 &self.env,
1034 &self.db,
1035 event,
1036 &[
1037 "UPDATE repos SET namespace = ?1 WHERE namespace = ?2",
1038 "UPDATE items SET workspace = ?1 WHERE workspace = ?2",
1039 "UPDATE people SET slug = ?1 WHERE kind = 'workspace' AND slug = ?2",
1040 ],
1041 )
1042 .await?;
1043 }
1044 _ => {}
1045 }
1046 Ok(())
1047 }
1048
1049 pub async fn run_job(&self, job: Job) -> Result<()> {
1050 match job {
1051 Job::Backfill { after } => {
1052 let page: IdPage = g1t_kit::call(&self.repos, "all_ids", &AllIdsArgs { after, limit: BACKFILL_PAGE }).await?;
1053 let mut jobs: Vec<Job> = page.ids.into_iter().map(|repo_id| Job::Repo { repo_id }).collect();
1054 if let Some(next) = page.next {
1055 jobs.push(Job::Backfill { after: Some(next) });
1056 }
1057 self.enqueue(jobs).await
1058 }
1059 Job::Repo { repo_id } => {
1060 let Some(tree) = self.refresh_repo(&repo_id).await? else { return Ok(()) };
1061 let path = RepoPath { namespace: tree.repo.namespace.clone(), name: tree.repo.name.clone() };
1062 if let Some(head) = &tree.head {
1063 store::run_all(
1064 &self.db,
1065 vec![store::prepare(
1066 &self.db,
1067 "UPDATE repos SET pushed_at = COALESCE(pushed_at, ?) WHERE repo_id = ?",
1068 vec![p(&head.authored_at), p(&repo_id)],
1069 )?],
1070 )
1071 .await?;
1072 }
1073 self.index_pulls(&repo_id, &path).await?;
1074 self.reconcile(&repo_id).await
1075 }
1076 Job::Push { repo_id, before, after } => {
1077 if self.refresh_repo(&repo_id).await?.is_none() {
1078 return Ok(());
1079 }
1080 store::run_all(
1081 &self.db,
1082 vec![store::prepare(&self.db, "UPDATE repos SET pushed_at = ? WHERE repo_id = ?", vec![p(&now()), p(&repo_id)])?],
1083 )
1084 .await?;
1085 let Some(before) = before.filter(|before| !is_null_commit(before)) else {
1086 return self.reconcile(&repo_id).await;
1087 };
1088 let changed: FileList = g1t_kit::call(
1089 &self.repos,
1090 "changed_files",
1091 &ChangedFilesArgs {
1092 repo_id: repo_id.clone(),
1093 base: Some(before),
1094 head: after.clone(),
1095 skip_dirs: Self::skip_dirs(),
1096 limit: MAX_PUSH_FILES,
1097 },
1098 )
1099 .await?;
1100 // A push too large to list, or one whose base is unknown
1101 // (a force push over history the store no longer has): compare whole.
1102 if changed.truncated || changed.commit.is_none() {
1103 return self.reconcile(&repo_id).await;
1104 }
1105 let files: Vec<(String, Option<String>)> =
1106 changed.files.into_iter().map(|file| (file.path, file.hash)).collect();
1107 self.queue_files(&repo_id, &files).await?;
1108 store::run_all(
1109 &self.db,
1110 vec![store::prepare(&self.db, "UPDATE repos SET head = ? WHERE repo_id = ?", vec![p(&after), p(&repo_id)])?],
1111 )
1112 .await?;
1113 self.drain(&repo_id).await
1114 }
1115 Job::Drain { repo_id } => self.drain(&repo_id).await,
1116 Job::Directory { kind, after } => {
1117 let page: DirectoryPage = g1t_kit::call(
1118 &self.identity,
1119 "directory",
1120 &DirectoryArgs { kind: kind.clone(), after, limit: DIRECTORY_PAGE },
1121 )
1122 .await?;
1123 let mut statements = Vec::new();
1124 for entry in &page.entries {
1125 let reference = if kind == "workspace" { entry.id.clone() } else { entry.slug.to_lowercase() };
1126 statements.push(self.person_row(&kind, &reference, entry)?);
1127 }
1128 store::run_all(&self.db, statements).await?;
1129 // Each workspace's work items, a job each.
1130 let mut jobs: Vec<Job> = if kind == "workspace" {
1131 page.entries.iter().map(|entry| Job::Workspace { workspace: entry.slug.to_lowercase() }).collect()
1132 } else {
1133 Vec::new()
1134 };
1135 if let Some(next) = page.next {
1136 jobs.push(Job::Directory { kind, after: Some(next) });
1137 }
1138 self.enqueue(jobs).await
1139 }
1140 Job::Workspace { workspace } => self.index_work_items(&workspace).await,
1141 Job::Items { workspace, numbers } => {
1142 for number in numbers {
1143 self.index_work_item(&workspace, number, None).await?;
1144 }
1145 Ok(())
1146 }
1147 }
1148 }
1149}
1150
1151#[cfg(test)]
1152mod tests {
1153 use g1t_contracts::work::PullStatus;
1154
1155 use super::*;
1156
1157 #[test]
1158 fn jobs_are_sent_tagged() {
1159 let job = Job::Push { repo_id: "rep_1".into(), before: None, after: "abc".into() };
1160 let value = serde_json::to_value(&job).unwrap();
1161 assert_eq!(value["type"], "push");
1162 assert_eq!(serde_json::from_value::<Job>(value).unwrap(), job);
1163 let backfill: Job = serde_json::from_str(r#"{"type":"backfill"}"#).unwrap();
1164 assert_eq!(backfill, Job::Backfill { after: None });
1165 }
1166
1167 #[test]
1168 fn only_a_backfills_pages_are_parked_one_per_walk() {
1169 assert_eq!(parked_key(&Job::Backfill { after: Some("rep_9".into()) }).as_deref(), Some("parked_job:backfill"));
1170 assert_eq!(parked_key(&Job::Directory { kind: "user".into(), after: None }).as_deref(), Some("parked_job:directory:user"));
1171 assert_eq!(parked_key(&Job::Repo { repo_id: "rep_1".into() }), None);
1172 assert_eq!(parked_key(&Job::Drain { repo_id: "rep_1".into() }), None);
1173 }
1174
1175 #[test]
1176 fn a_push_from_nothing_is_a_whole_comparison() {
1177 assert!(is_null_commit("0000000000000000000000000000000000000000"));
1178 assert!(is_null_commit(""));
1179 assert!(!is_null_commit("0a1b"));
1180 }
1181
1182 #[test]
1183 fn labels_are_kept_between_bars() {
1184 assert_eq!(labels_of(&["Bug".into(), "good first issue".into()]), "|bug|good first issue|");
1185 assert_eq!(labels_of(&[]), "|");
1186 }
1187
1188 #[test]
1189 fn migrated_items_are_read_again_by_workspace_a_few_at_a_time() {
1190 use g1t_contracts::work::MigratedItem;
1191 let moved = |workspace: &str, number: u32| MigratedItem {
1192 repo_id: "rep_1".into(),
1193 legacy_number: number,
1194 workspace: workspace.into(),
1195 number: number + 100,
1196 item_id: format!("iss_{number}"),
1197 };
1198 let mut items: Vec<MigratedItem> = (1..=25).map(|n| moved("acme", n)).collect();
1199 items.insert(3, moved("Beta", 7));
1200 let jobs = migrated_jobs(&WorkMigrated { page: 1, items, done: false });
1201 assert_eq!(jobs.len(), 3);
1202 assert_eq!(jobs[0], Job::Items { workspace: "acme".into(), numbers: (101..=120).collect() });
1203 assert_eq!(jobs[1], Job::Items { workspace: "acme".into(), numbers: (121..=125).collect() });
1204 assert_eq!(jobs[2], Job::Items { workspace: "beta".into(), numbers: vec![107] });
1205 let value = serde_json::to_value(&jobs[2]).unwrap();
1206 assert_eq!(value, serde_json::json!({ "type": "items", "workspace": "beta", "numbers": [107] }));
1207 }
1208
1209 #[test]
1210 fn keys_and_linked_repositories() {
1211 assert_eq!(pull_key("rep_1", 12), "rep_1/12");
1212 assert_eq!(repos_of(&["rep_1", "rep_2"]), "|rep_1|rep_2|");
1213 assert_eq!(repos_of(&[]), "|");
1214 }
1215
1216 #[test]
1217 fn pull_statuses_have_names() {
1218 assert_eq!(PullStatus::Merged.as_str(), "merged");
1219 }
1220}