1use std::{
2 collections::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::{bstr::ByteSlice, refs::Category, Reference};
12use itertools::{Either, Itertools};
13use rocksdb::WriteBatch;
14use time::{OffsetDateTime, UtcOffset};
15use tracing::{error, info, info_span, instrument, warn};
16
17use crate::database::schema::{
18 commit::Commit,
19 repository::{ArchivedRepository, Repository, RepositoryId},
20 tag::{Tag, TagTree},
21};
22
23pub fn run(scan_path: &Path, repository_list: Option<&Path>, db: &Arc<rocksdb::DB>) {
24 let span = info_span!("index_update");
25 let _entered = span.enter();
26
27 info!("Starting index update");
28
29 update_repository_metadata(scan_path, repository_list, db);
30 update_repository_reflog(scan_path, db.clone());
31 update_repository_tags(scan_path, db.clone());
32
33 info!("Flushing to disk");
34
35 if let Err(error) = db.flush() {
36 error!(%error, "Failed to flush database to disk");
37 }
38
39 info!("Finished index update");
40}
41
42#[instrument(skip(db))]
43fn update_repository_metadata(scan_path: &Path, repository_list: Option<&Path>, db: &rocksdb::DB) {
44 let mut discovered = Vec::new();
45 discover_repositories(scan_path, repository_list, &mut discovered);
46
47 for (repository_path, git_repository) in discovered {
48 let Some(relative) = get_relative_path(scan_path, &repository_path) else {
49 continue;
50 };
51
52 let id = match Repository::open(db, relative) {
53 Ok(v) => v.map_or_else(RepositoryId::new, |v| {
54 RepositoryId(v.get().id.0.to_native())
55 }),
56 Err(error) => {
57 // maybe we could nuke it ourselves, but we need to instantly trigger
58 // a reindex and we could enter into an infinite loop if there's a bug
59 // or something
60 error!(%error, "Failed to open repository index {}, please consider nuking database", relative.display());
61 continue;
62 }
63 };
64
65 let Some(name) = relative.file_name().and_then(OsStr::to_str) else {
66 continue;
67 };
68 let description = std::fs::read(repository_path.join("description")).unwrap_or_default();
69 let description = String::from_utf8(description)
70 .ok()
71 .filter(|v| !v.is_empty());
72
73 let owner = git_repository
74 .config_snapshot()
75 .string("gitweb.owner")
76 .map(|v| v.to_string());
77
78 let res = Repository {
79 id,
80 name: name.to_string(),
81 description,
82 owner,
83 last_modified: {
84 let r =
85 find_last_committed_time(&git_repository).unwrap_or(OffsetDateTime::UNIX_EPOCH);
86 (r.unix_timestamp(), r.offset().whole_seconds())
87 },
88 default_branch: find_default_branch(&git_repository).ok().flatten(),
89 exported: repository_path.join("git-daemon-export-ok").exists(),
90 }
91 .insert(db, relative);
92
93 if let Err(error) = res {
94 warn!(%error, "Failed to insert repository");
95 }
96 }
97}
98
99fn find_default_branch(repo: &gix::Repository) -> Result<Option<String>, anyhow::Error> {
100 if repo.head()?.is_detached() {
101 Ok(None)
102 } else {
103 Ok(Some(
104 repo.head()?
105 .referent_name()
106 .context("HEAD does not point to anything")?
107 .as_bstr()
108 .to_string(),
109 ))
110 }
111}
112
113fn find_last_committed_time(repo: &gix::Repository) -> Result<OffsetDateTime, anyhow::Error> {
114 let mut timestamp = OffsetDateTime::UNIX_EPOCH;
115
116 for reference in repo.references()?.all()? {
117 let Ok(commit) = reference.unwrap().peel_to_commit() else {
118 continue;
119 };
120
121 let committer = commit.committer()?;
122 let offset = UtcOffset::from_whole_seconds(committer.time.offset).unwrap_or(UtcOffset::UTC);
123 let committed_time = OffsetDateTime::from_unix_timestamp(committer.time.seconds)
124 .unwrap_or(OffsetDateTime::UNIX_EPOCH)
125 .to_offset(offset);
126 timestamp = timestamp.max(committed_time);
127 }
128
129 Ok(timestamp)
130}
131
132#[instrument(skip(db))]
133fn update_repository_reflog(scan_path: &Path, db: Arc<rocksdb::DB>) {
134 let repos = match Repository::fetch_all(&db) {
135 Ok(v) => v,
136 Err(error) => {
137 error!(%error, "Failed to read repository index to update reflog, consider deleting database directory");
138 return;
139 }
140 };
141
142 for (relative_path, db_repository) in repos {
143 let Some(git_repository) = open_repo(scan_path, &relative_path, db_repository.get(), &db)
144 else {
145 continue;
146 };
147
148 let references = match git_repository.references() {
149 Ok(v) => v,
150 Err(error) => {
151 error!(%error, "Failed to read references for {relative_path}");
152 continue;
153 }
154 };
155
156 let references = match references.all() {
157 Ok(v) => v,
158 Err(error) => {
159 error!(%error, "Failed to read references for {relative_path}");
160 continue;
161 }
162 };
163
164 let mut valid_references = Vec::new();
165
166 for reference in references {
167 let mut reference = match reference {
168 Ok(v) => v,
169 Err(error) => {
170 error!(%error, "Failed to read reference for {relative_path}");
171 continue;
172 }
173 };
174
175 let reference_name = reference.name();
176 if !matches!(
177 reference_name.category(),
178 Some(Category::Tag | Category::LocalBranch)
179 ) {
180 continue;
181 }
182
183 valid_references.push(reference_name.as_bstr().to_string());
184
185 if let Err(error) = branch_index_update(
186 &mut reference,
187 &relative_path,
188 db_repository.get(),
189 db.clone(),
190 &git_repository,
191 false,
192 ) {
193 error!(%error, "Failed to update reflog for {relative_path}@{:?}", valid_references.last());
194 }
195 }
196
197 if let Err(error) = db_repository.get().replace_heads(&db, &valid_references) {
198 error!(%error, "Failed to update heads");
199 }
200 }
201}
202
203#[instrument(skip(reference, db_repository, db, git_repository))]
204fn branch_index_update(
205 reference: &mut Reference<'_>,
206 relative_path: &str,
207 db_repository: &ArchivedRepository,
208 db: Arc<rocksdb::DB>,
209 git_repository: &gix::Repository,
210 force_reindex: bool,
211) -> Result<(), anyhow::Error> {
212 info!("Refreshing indexes");
213
214 let commit_tree = db_repository.commit_tree(db.clone(), reference.name().as_bstr().to_str()?);
215
216 if force_reindex {
217 commit_tree.drop_commits()?;
218 }
219
220 let commit = reference.peel_to_commit()?;
221
222 let latest_indexed = if let Some(latest_indexed) = commit_tree.fetch_latest_one()? {
223 if commit.id().as_bytes() == latest_indexed.get().hash.as_slice() {
224 info!("No commits since last index");
225 return Ok(());
226 }
227
228 Some(latest_indexed)
229 } else {
230 None
231 };
232
233 // TODO: stop collecting into a vec
234 let revwalk = git_repository
235 .rev_walk([commit.id().detach()])
236 .all()?
237 .collect::<Vec<_>>()
238 .into_iter()
239 .rev();
240
241 let tree_len = commit_tree.len()?;
242 let mut seen = false;
243 let mut i = 0;
244 for revs in &revwalk.chunks(250) {
245 let mut batch = WriteBatch::default();
246
247 for rev in revs {
248 let rev = rev?;
249
250 if let (false, Some(latest_indexed)) = (seen, &latest_indexed) {
251 if rev.id.as_bytes() == latest_indexed.get().hash.as_slice() {
252 seen = true;
253 }
254
255 continue;
256 }
257
258 seen = true;
259
260 if ((i + 1) % 25_000) == 0 {
261 info!("{} commits ingested", i + 1);
262 }
263
264 let commit = rev.object()?;
265 let oid = commit.id;
266 let commit = commit.decode()?;
267 let author = commit.author();
268 let committer = commit.committer();
269
270 Commit::new(oid, &commit, author, committer)?.insert(
271 &commit_tree,
272 tree_len + i,
273 &mut batch,
274 )?;
275 i += 1;
276 }
277
278 commit_tree.update_counter(tree_len + i, &mut batch)?;
279 db.write_without_wal(batch)?;
280 }
281
282 if !seen && !force_reindex {
283 warn!("Detected converged history, forcing reindex");
284
285 return branch_index_update(
286 reference,
287 relative_path,
288 db_repository,
289 db,
290 git_repository,
291 true,
292 );
293 }
294
295 Ok(())
296}
297
298#[instrument(skip(db))]
299fn update_repository_tags(scan_path: &Path, db: Arc<rocksdb::DB>) {
300 let repos = match Repository::fetch_all(&db) {
301 Ok(v) => v,
302 Err(error) => {
303 error!(%error, "Failed to read repository index to update tags, consider deleting database directory");
304 return;
305 }
306 };
307
308 for (relative_path, db_repository) in repos {
309 let Some(git_repository) = open_repo(scan_path, &relative_path, db_repository.get(), &db)
310 else {
311 continue;
312 };
313
314 if let Err(error) = tag_index_scan(
315 &relative_path,
316 db_repository.get(),
317 db.clone(),
318 &git_repository,
319 ) {
320 error!(%error, "Failed to update tags for {relative_path}");
321 }
322 }
323}
324
325#[instrument(skip(db_repository, db, git_repository))]
326fn tag_index_scan(
327 relative_path: &str,
328 db_repository: &ArchivedRepository,
329 db: Arc<rocksdb::DB>,
330 git_repository: &gix::Repository,
331) -> Result<(), anyhow::Error> {
332 let tag_tree = db_repository.tag_tree(db);
333
334 let git_tags: HashSet<_> = git_repository
335 .references()
336 .context("Failed to scan indexes on git repository")?
337 .all()?
338 .filter_map(Result::ok)
339 .filter(|v| v.name().category() == Some(Category::Tag))
340 .map(|v| v.name().as_bstr().to_string())
341 .collect();
342 let indexed_tags: HashSet<String> = tag_tree.list()?.into_iter().collect();
343
344 // insert any git tags that are missing from the index
345 for tag_name in git_tags.difference(&indexed_tags) {
346 tag_index_update(tag_name, git_repository, &tag_tree)?;
347 }
348
349 // remove any extra tags that the index has
350 // TODO: this also needs to check peel_to_tag
351 for tag_name in indexed_tags.difference(&git_tags) {
352 tag_index_delete(tag_name, &tag_tree)?;
353 }
354
355 Ok(())
356}
357
358#[instrument(skip(git_repository, tag_tree))]
359fn tag_index_update(
360 tag_name: &str,
361 git_repository: &gix::Repository,
362 tag_tree: &TagTree,
363) -> Result<(), anyhow::Error> {
364 let mut reference = git_repository
365 .find_reference(tag_name)
366 .context("Failed to read newly discovered tag")?;
367
368 if let Ok(tag) = reference.peel_to_tag() {
369 info!("Inserting newly discovered tag to index");
370
371 Tag::new(tag.tagger()?)?.insert(tag_tree, tag_name)?;
372 }
373
374 Ok(())
375}
376
377#[instrument(skip(tag_tree))]
378fn tag_index_delete(tag_name: &str, tag_tree: &TagTree) -> Result<(), anyhow::Error> {
379 info!("Removing stale tag from index");
380 tag_tree.remove(tag_name)?;
381
382 Ok(())
383}
384
385#[instrument(skip(scan_path, db_repository, db))]
386fn open_repo<P: AsRef<Path> + Debug>(
387 scan_path: &Path,
388 relative_path: P,
389 db_repository: &ArchivedRepository,
390 db: &rocksdb::DB,
391) -> Option<gix::Repository> {
392 match gix::open(scan_path.join(relative_path.as_ref())) {
393 Ok(mut v) => {
394 v.object_cache_size(10 * 1024 * 1024);
395 Some(v)
396 }
397 Err(gix::open::Error::Io(e)) if e.kind() == std::io::ErrorKind::NotFound => {
398 warn!("Repository gone from disk, removing from db");
399
400 if let Err(error) = db_repository.delete(db, relative_path) {
401 warn!(%error, "Failed to delete dangling index");
402 }
403
404 None
405 }
406 Err(error) => {
407 warn!(%error, "Failed to reindex, skipping");
408 None
409 }
410 }
411}
412
413fn get_relative_path<'a>(relative_to: &Path, full_path: &'a Path) -> Option<&'a Path> {
414 full_path.strip_prefix(relative_to).ok()
415}
416
417fn discover_repositories(
418 current: &Path,
419 repository_list: Option<&Path>,
420 discovered_repos: &mut Vec<(PathBuf, gix::Repository)>,
421) {
422 let dirs = if let Some(repo_list) = repository_list {
423 let mut repo_list = match std::fs::File::open(&repo_list) {
424 Ok(v) => BufReader::new(v).lines(),
425 Err(error) => {
426 error!(%error, "Failed to open repository list file");
427 return;
428 }
429 };
430
431 let mut out = Vec::new();
432
433 while let Some(line) = repo_list.next() {
434 let line = match line {
435 Ok(v) => v,
436 Err(error) => {
437 error!(%error, "Failed to read repository list file");
438 return;
439 }
440 };
441
442 out.push(current.join(line));
443 }
444
445 Either::Left(out.into_iter())
446 } else {
447 let current = match std::fs::read_dir(current) {
448 Ok(v) => v,
449 Err(error) => {
450 error!(%error, "Failed to enter repository directory {}", current.display());
451 return;
452 }
453 };
454
455 Either::Right(
456 current
457 .filter_map(Result::ok)
458 .map(|v| v.path())
459 .filter(|path| path.is_dir()),
460 )
461 };
462
463 for dir in dirs {
464 match gix::open_opts(&dir, gix::open::Options::default().open_path_as_is(true)) {
465 Ok(mut repo) => {
466 repo.object_cache_size(10 * 1024 * 1024);
467 discovered_repos.push((dir, repo));
468 }
469 Err(gix::open::Error::NotARepository { .. }) if repository_list.is_none() => {
470 discover_repositories(&dir, None, discovered_repos);
471 }
472
473 Err(gix::open::Error::NotARepository { .. }) => {
474 warn!(
475 "Repository list points to directory which isn't a Git repository: {}",
476 dir.display()
477 );
478 }
479 Err(error) => {
480 warn!(%error, "Failed to open repository {} for indexing", dir.display());
481 }
482 }
483 }
484}