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.

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