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