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