1use std::{
2 collections::{BTreeMap, HashSet},
3 ffi::OsStr,
4 fmt::Debug,
5 io::{BufRead, BufReader},
6 path::{Path, PathBuf},
7 sync::Arc,
8};
9
10use anyhow::Context;
11use gix::{
12 ObjectId, Reference, Url,
13 bstr::{BStr, ByteSlice},
14 objs::tree::EntryKind,
15 refs::Category,
16 url::Scheme,
17};
18use itertools::{Either, Itertools};
19use rocksdb::WriteBatch;
20use time::{OffsetDateTime, UtcOffset};
21use tracing::{error, info, info_span, instrument, warn};
22use xxhash_rust::xxh3::Xxh3;
23
24use crate::{
25 database::schema::{
26 commit::Commit,
27 repository::{ArchivedRepository, Repository, RepositoryId},
28 tag::{Tag, TagTree},
29 },
30 git::{PathVisitor, PathVisitorHandler},
31};
32
33use super::schema::tree::{SortedTree, SortedTreeItem, Tree, TreeItem, TreeItemKind, TreeKey};
34
35pub fn run(scan_path: &Path, repository_list: Option<&Path>, db: &Arc<rocksdb::DB>) {
36 let span = info_span!("index_update");
37 let _entered = span.enter();
38
39 info!("Starting index update");
40
41 update_repository_metadata(scan_path, repository_list, db);
42 update_repository_reflog(scan_path, db.clone());
43 update_repository_tags(scan_path, db.clone());
44
45 info!("Flushing to disk");
46
47 if let Err(error) = db.flush() {
48 error!(%error, "Failed to flush database to disk");
49 }
50
51 info!("Finished index update");
52}
53
54#[instrument(skip(db))]
55fn update_repository_metadata(scan_path: &Path, repository_list: Option<&Path>, db: &rocksdb::DB) {
56 let mut discovered = Vec::new();
57 discover_repositories(scan_path, repository_list, &mut discovered);
58
59 for (repository_path, git_repository) in discovered {
60 let Some(relative) = get_relative_path(scan_path, &repository_path) else {
61 continue;
62 };
63
64 let id = match Repository::open(db, relative) {
65 Ok(v) => v.map_or_else(RepositoryId::new, |v| {
66 RepositoryId(v.get().id.0.to_native())
67 }),
68 Err(error) => {
69 // maybe we could nuke it ourselves, but we need to instantly trigger
70 // a reindex and we could enter into an infinite loop if there's a bug
71 // or something
72 error!(%error, "Failed to open repository index {}, please consider nuking database", relative.display());
73 continue;
74 }
75 };
76
77 let Some(name) = relative.file_name().and_then(OsStr::to_str) else {
78 continue;
79 };
80 let description = std::fs::read(repository_path.join("description")).unwrap_or_default();
81 let description = String::from_utf8(description)
82 .ok()
83 .filter(|v| !v.is_empty());
84
85 let owner = git_repository
86 .config_snapshot()
87 .string("gitweb.owner")
88 .map(|v| v.to_string());
89
90 let res = Repository {
91 id,
92 name: name.to_string(),
93 description,
94 owner,
95 last_modified: {
96 let r =
97 find_last_committed_time(&git_repository).unwrap_or(OffsetDateTime::UNIX_EPOCH);
98 (r.unix_timestamp(), r.offset().whole_seconds())
99 },
100 default_branch: find_default_branch(&git_repository).ok().flatten(),
101 exported: repository_path.join("git-daemon-export-ok").exists(),
102 }
103 .insert(db, relative);
104
105 if let Err(error) = res {
106 warn!(%error, "Failed to insert repository");
107 }
108 }
109}
110
111fn find_default_branch(repo: &gix::Repository) -> Result<Option<String>, anyhow::Error> {
112 if repo.head()?.is_detached() {
113 Ok(None)
114 } else {
115 Ok(Some(
116 repo.head()?
117 .referent_name()
118 .context("HEAD does not point to anything")?
119 .as_bstr()
120 .to_string(),
121 ))
122 }
123}
124
125fn find_last_committed_time(repo: &gix::Repository) -> Result<OffsetDateTime, anyhow::Error> {
126 let mut timestamp = OffsetDateTime::UNIX_EPOCH;
127
128 for reference in repo.references()?.all()? {
129 let Ok(commit) = reference.unwrap().peel_to_commit() else {
130 continue;
131 };
132
133 let committer = commit.committer()?;
134 let offset = UtcOffset::from_whole_seconds(committer.time.offset).unwrap_or(UtcOffset::UTC);
135 let committed_time = OffsetDateTime::from_unix_timestamp(committer.time.seconds)
136 .unwrap_or(OffsetDateTime::UNIX_EPOCH)
137 .to_offset(offset);
138 timestamp = timestamp.max(committed_time);
139 }
140
141 Ok(timestamp)
142}
143
144#[instrument(skip(db))]
145fn update_repository_reflog(scan_path: &Path, db: Arc<rocksdb::DB>) {
146 let repos = match Repository::fetch_all(&db) {
147 Ok(v) => v,
148 Err(error) => {
149 error!(%error, "Failed to read repository index to update reflog, consider deleting database directory");
150 return;
151 }
152 };
153
154 for (relative_path, db_repository) in repos {
155 let Some(git_repository) = open_repo(scan_path, &relative_path, db_repository.get(), &db)
156 else {
157 continue;
158 };
159
160 let references = match git_repository.references() {
161 Ok(v) => v,
162 Err(error) => {
163 error!(%error, "Failed to read references for {relative_path}");
164 continue;
165 }
166 };
167
168 let references = match references.all() {
169 Ok(v) => v,
170 Err(error) => {
171 error!(%error, "Failed to read references for {relative_path}");
172 continue;
173 }
174 };
175
176 let submodules = match git_repository.submodules() {
177 Ok(submodules) => submodules
178 .into_iter()
179 .flatten()
180 .filter_map(|v| Some((v.name().to_path_lossy().to_path_buf(), v.url().ok()?)))
181 .collect::<BTreeMap<_, _>>(),
182 Err(error) => {
183 error!(%error, "Failed to read submodules for {relative_path}");
184 continue;
185 }
186 };
187
188 let mut valid_references = Vec::new();
189
190 for reference in references {
191 let mut reference = match reference {
192 Ok(v) => v,
193 Err(error) => {
194 error!(%error, "Failed to read reference for {relative_path}");
195 continue;
196 }
197 };
198
199 let reference_name = reference.name();
200 if !matches!(
201 reference_name.category(),
202 Some(Category::Tag | Category::LocalBranch)
203 ) {
204 continue;
205 }
206
207 valid_references.push(reference_name.as_bstr().to_string());
208
209 if let Err(error) = branch_index_update(
210 &mut reference,
211 &relative_path,
212 db_repository.get(),
213 db.clone(),
214 &git_repository,
215 false,
216 &submodules,
217 ) {
218 error!(%error, "Failed to update reflog for {relative_path}@{:?}", valid_references.last());
219 }
220 }
221
222 if let Err(error) = db_repository.get().replace_heads(&db, &valid_references) {
223 error!(%error, "Failed to update heads");
224 }
225 }
226}
227
228#[instrument(skip(reference, db_repository, db, git_repository))]
229fn branch_index_update(
230 reference: &mut Reference<'_>,
231 relative_path: &str,
232 db_repository: &ArchivedRepository,
233 db: Arc<rocksdb::DB>,
234 git_repository: &gix::Repository,
235 force_reindex: bool,
236 submodules: &BTreeMap<PathBuf, Url>,
237) -> Result<(), anyhow::Error> {
238 info!("Refreshing indexes");
239
240 let commit_tree = db_repository.commit_tree(db.clone(), reference.name().as_bstr().to_str()?);
241
242 if force_reindex {
243 commit_tree.drop_commits()?;
244 }
245
246 let commit = reference.peel_to_commit()?;
247
248 let latest_indexed = if let Some(latest_indexed) = commit_tree.fetch_latest_one()? {
249 if commit.id().as_bytes() == latest_indexed.get().hash.as_slice() {
250 info!("No commits since last index");
251 return Ok(());
252 }
253
254 Some(latest_indexed)
255 } else {
256 None
257 };
258
259 // TODO: stop collecting into a vec
260 let revwalk = git_repository
261 .rev_walk([commit.id().detach()])
262 .all()?
263 .collect::<Vec<_>>()
264 .into_iter()
265 .rev();
266
267 let mut hasher = Xxh3::new();
268
269 let tree_len = commit_tree.len()?;
270 let mut seen = false;
271 let mut i = 0;
272 for revs in &revwalk.chunks(250) {
273 let mut batch = WriteBatch::default();
274
275 for rev in revs {
276 let rev = rev?;
277
278 if let (false, Some(latest_indexed)) = (seen, &latest_indexed) {
279 if rev.id.as_bytes() == latest_indexed.get().hash.as_slice() {
280 seen = true;
281 }
282
283 continue;
284 }
285
286 seen = true;
287
288 if ((i + 1) % 25_000) == 0 {
289 info!("{} commits ingested", i + 1);
290 }
291
292 let commit = rev.object()?;
293 let oid = commit.id;
294 let commit = commit.decode()?;
295 let author = commit.author();
296 let committer = commit.committer();
297
298 let tree = git_repository.find_tree(commit.tree())?;
299 let tree_id = index_tree(&db, &mut batch, &tree, &mut hasher, submodules)?;
300
301 Commit::new(oid, &commit, author, committer, tree_id)?.insert(
302 &commit_tree,
303 tree_len + i,
304 &mut batch,
305 )?;
306
307 i += 1;
308 }
309
310 commit_tree.update_counter(tree_len + i, &mut batch)?;
311 db.write_without_wal(batch)?;
312 }
313
314 if !seen && !force_reindex {
315 warn!("Detected converged history, forcing reindex");
316
317 return branch_index_update(
318 reference,
319 relative_path,
320 db_repository,
321 db,
322 git_repository,
323 true,
324 submodules,
325 );
326 }
327
328 Ok(())
329}
330
331fn index_tree(
332 database: &rocksdb::DB,
333 batch: &mut WriteBatch,
334 tree: &gix::Tree<'_>,
335 hasher: &mut Xxh3,
336 submodules: &BTreeMap<PathBuf, Url>,
337) -> Result<u64, anyhow::Error> {
338 hasher.reset();
339 tree.traverse()
340 .breadthfirst(&mut PathVisitor::new(TreeHasherVisitor { hasher }))?;
341 let digest = hasher.digest();
342
343 if !TreeItem::contains(database, digest)? {
344 let mut sorted_out = SortedTree::default();
345
346 tree.traverse()
347 .breadthfirst(&mut PathVisitor::new(TreeItemIndexerVisitor {
348 buffer: Vec::new(),
349 digest,
350 database,
351 batch,
352 submodules,
353 sorted: &mut sorted_out,
354 }))?;
355
356 sorted_out.insert(digest, database, batch)?;
357 }
358
359 Tree {
360 indexed_tree_id: digest,
361 }
362 .insert(database, batch, tree.id)?;
363
364 Ok(digest)
365}
366
367/// Walks the entire tree and hashes all the (path, mode)s so trees can be deduplicated.
368///
369/// Note: unlike git's tree oid, this does not take into account blob contents.
370struct TreeHasherVisitor<'a> {
371 hasher: &'a mut Xxh3,
372}
373
374impl PathVisitorHandler for TreeHasherVisitor<'_> {
375 fn visit(
376 &mut self,
377 entry: &gix::objs::tree::EntryRef<'_>,
378 path: &BStr,
379 ) -> gix::traverse::tree::visit::Action {
380 self.hasher.update(path);
381 self.hasher.update(&entry.mode.to_ne_bytes());
382 gix::traverse::tree::visit::Action::Continue
383 }
384}
385
386struct TreeItemIndexerVisitor<'a> {
387 digest: u64,
388 buffer: Vec<u8>,
389 sorted: &'a mut SortedTree,
390 database: &'a rocksdb::DB,
391 batch: &'a mut WriteBatch,
392 submodules: &'a BTreeMap<PathBuf, Url>,
393}
394
395impl PathVisitorHandler for TreeItemIndexerVisitor<'_> {
396 fn visit(
397 &mut self,
398 entry: &gix::objs::tree::EntryRef<'_>,
399 path: &BStr,
400 ) -> gix::traverse::tree::visit::Action {
401 let kind = match entry.mode.kind() {
402 EntryKind::Blob | EntryKind::BlobExecutable | EntryKind::Link => TreeItemKind::File,
403 EntryKind::Commit => {
404 let Some(mut url) = self
405 .submodules
406 .get(&path.to_path_lossy().into_owned())
407 .cloned()
408 else {
409 return gix::traverse::tree::visit::Action::Continue;
410 };
411
412 if matches!(url.scheme, Scheme::Git | Scheme::Ssh) {
413 url.scheme = Scheme::Https;
414 }
415
416 TreeItemKind::Submodule(match entry.oid.to_owned() {
417 ObjectId::Sha1(oid) => super::schema::tree::Submodule {
418 url: url.to_string(),
419 oid,
420 },
421 })
422 }
423 EntryKind::Tree => TreeItemKind::Tree,
424 };
425
426 // TODO: this could be more optimised doing a recursive DFS to not have to keep diving deep into self.sorted
427 if !entry.mode.is_tree() {
428 let mut tree = &mut *self.sorted;
429 let mut offset = 0;
430 for end in memchr::memchr_iter(b'/', path) {
431 let path = &path[offset..end];
432 offset = end + 1;
433
434 tree = match tree
435 .0
436 .entry(TreeKey(path.to_string()))
437 .or_insert(SortedTreeItem::Directory(SortedTree::default()))
438 {
439 SortedTreeItem::Directory(dir) => dir,
440 SortedTreeItem::File => panic!("a file is somehow not in a directory"),
441 };
442 }
443 tree.0
444 .insert(TreeKey(path[offset..].to_string()), SortedTreeItem::File);
445 }
446
447 TreeItem {
448 mode: entry.mode.0,
449 kind,
450 }
451 .insert(
452 &mut self.buffer,
453 self.digest,
454 path,
455 self.database,
456 self.batch,
457 )
458 .expect("failed to insert TreeItem");
459
460 gix::traverse::tree::visit::Action::Continue
461 }
462}
463
464#[instrument(skip(db))]
465fn update_repository_tags(scan_path: &Path, db: Arc<rocksdb::DB>) {
466 let repos = match Repository::fetch_all(&db) {
467 Ok(v) => v,
468 Err(error) => {
469 error!(%error, "Failed to read repository index to update tags, consider deleting database directory");
470 return;
471 }
472 };
473
474 for (relative_path, db_repository) in repos {
475 let Some(git_repository) = open_repo(scan_path, &relative_path, db_repository.get(), &db)
476 else {
477 continue;
478 };
479
480 let submodules = match git_repository.submodules() {
481 Ok(submodules) => submodules
482 .into_iter()
483 .flatten()
484 .filter_map(|v| Some((v.name().to_path_lossy().to_path_buf(), v.url().ok()?)))
485 .collect::<BTreeMap<_, _>>(),
486 Err(error) => {
487 error!(%error, "Failed to read submodules for {relative_path}");
488 continue;
489 }
490 };
491
492 if let Err(error) = tag_index_scan(
493 &relative_path,
494 db_repository.get(),
495 db.clone(),
496 &git_repository,
497 &submodules,
498 ) {
499 error!(%error, "Failed to update tags for {relative_path}");
500 }
501 }
502}
503
504#[instrument(skip(db_repository, db, git_repository))]
505fn tag_index_scan(
506 relative_path: &str,
507 db_repository: &ArchivedRepository,
508 db: Arc<rocksdb::DB>,
509 git_repository: &gix::Repository,
510 submodules: &BTreeMap<PathBuf, Url>,
511) -> Result<(), anyhow::Error> {
512 let tag_tree = db_repository.tag_tree(db);
513
514 let git_tags: HashSet<_> = git_repository
515 .references()
516 .context("Failed to scan indexes on git repository")?
517 .all()?
518 .filter_map(Result::ok)
519 .filter(|v| v.name().category() == Some(Category::Tag))
520 .map(|v| v.name().as_bstr().to_string())
521 .collect();
522 let indexed_tags: HashSet<String> = tag_tree.list()?.into_iter().collect();
523
524 // insert any git tags that are missing from the index
525 for tag_name in git_tags.difference(&indexed_tags) {
526 tag_index_update(tag_name, git_repository, &tag_tree, submodules)?;
527 }
528
529 // remove any extra tags that the index has
530 // TODO: this also needs to check peel_to_tag
531 for tag_name in indexed_tags.difference(&git_tags) {
532 tag_index_delete(tag_name, &tag_tree)?;
533 }
534
535 Ok(())
536}
537
538#[instrument(skip(git_repository, tag_tree))]
539fn tag_index_update(
540 tag_name: &str,
541 git_repository: &gix::Repository,
542 tag_tree: &TagTree,
543 submodules: &BTreeMap<PathBuf, Url>,
544) -> Result<(), anyhow::Error> {
545 let mut reference = git_repository
546 .find_reference(tag_name)
547 .context("Failed to read newly discovered tag")?;
548
549 let tree_id = if let Ok(tree) = reference.peel_to_tree() {
550 let mut batch = WriteBatch::default();
551 let tree_id = index_tree(
552 &tag_tree.db,
553 &mut batch,
554 &tree,
555 &mut Xxh3::new(),
556 submodules,
557 )?;
558 tag_tree.db.write_without_wal(batch)?;
559 Some(tree_id)
560 } else {
561 None
562 };
563
564 if let Ok(tag) = reference.peel_to_tag() {
565 info!("Inserting newly discovered tag to index");
566
567 Tag::new(tag.tagger()?, tree_id)?.insert(tag_tree, tag_name)?;
568 }
569
570 Ok(())
571}
572
573#[instrument(skip(tag_tree))]
574fn tag_index_delete(tag_name: &str, tag_tree: &TagTree) -> Result<(), anyhow::Error> {
575 info!("Removing stale tag from index");
576 tag_tree.remove(tag_name)?;
577
578 Ok(())
579}
580
581#[instrument(skip(scan_path, db_repository, db))]
582fn open_repo<P: AsRef<Path> + Debug>(
583 scan_path: &Path,
584 relative_path: P,
585 db_repository: &ArchivedRepository,
586 db: &rocksdb::DB,
587) -> Option<gix::Repository> {
588 match gix::open(scan_path.join(relative_path.as_ref())) {
589 Ok(mut v) => {
590 v.object_cache_size(10 * 1024 * 1024);
591 Some(v)
592 }
593 Err(gix::open::Error::Io(e)) if e.kind() == std::io::ErrorKind::NotFound => {
594 warn!("Repository gone from disk, removing from db");
595
596 if let Err(error) = db_repository.delete(db, relative_path) {
597 warn!(%error, "Failed to delete dangling index");
598 }
599
600 None
601 }
602 Err(error) => {
603 warn!(%error, "Failed to reindex, skipping");
604 None
605 }
606 }
607}
608
609fn get_relative_path<'a>(relative_to: &Path, full_path: &'a Path) -> Option<&'a Path> {
610 full_path.strip_prefix(relative_to).ok()
611}
612
613fn discover_repositories(
614 current: &Path,
615 repository_list: Option<&Path>,
616 discovered_repos: &mut Vec<(PathBuf, gix::Repository)>,
617) {
618 let dirs = if let Some(repo_list) = repository_list {
619 let repo_list = match std::fs::File::open(repo_list) {
620 Ok(v) => BufReader::new(v).lines(),
621 Err(error) => {
622 error!(%error, "Failed to open repository list file");
623 return;
624 }
625 };
626
627 let mut out = Vec::new();
628
629 for line in repo_list {
630 let line = match line {
631 Ok(v) => v,
632 Err(error) => {
633 error!(%error, "Failed to read repository list file");
634 return;
635 }
636 };
637
638 out.push(current.join(line));
639 }
640
641 Either::Left(out.into_iter())
642 } else {
643 let current = match std::fs::read_dir(current) {
644 Ok(v) => v,
645 Err(error) => {
646 error!(%error, "Failed to enter repository directory {}", current.display());
647 return;
648 }
649 };
650
651 Either::Right(
652 current
653 .filter_map(Result::ok)
654 .map(|v| v.path())
655 .filter(|path| path.is_dir()),
656 )
657 };
658
659 for dir in dirs {
660 match gix::open_opts(&dir, gix::open::Options::default().open_path_as_is(true)) {
661 Ok(mut repo) => {
662 repo.object_cache_size(10 * 1024 * 1024);
663 discovered_repos.push((dir, repo));
664 }
665 Err(gix::open::Error::NotARepository { .. }) if repository_list.is_none() => {
666 discover_repositories(&dir, None, discovered_repos);
667 }
668
669 Err(gix::open::Error::NotARepository { .. }) => {
670 warn!(
671 "Repository list points to directory which isn't a Git repository: {}",
672 dir.display()
673 );
674 }
675 Err(error) => {
676 warn!(%error, "Failed to open repository {} for indexing", dir.display());
677 }
678 }
679 }
680}