1use std::{
2 collections::HashSet,
3 ffi::OsStr,
4 fmt::Debug,
5 path::{Path, PathBuf},
6 sync::Arc,
7};
8
9use anyhow::Context;
10use gix::{bstr::ByteSlice, refs::Category, Reference};
11use ini::Ini;
12use itertools::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, 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, 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, db: &rocksdb::DB) {
44 let mut discovered = Vec::new();
45 discover_repositories(scan_path, &mut discovered);
46
47 for repository in discovered {
48 let Some(relative) = get_relative_path(scan_path, &repository) 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.join("description")).unwrap_or_default();
69 let description = String::from_utf8(description)
70 .ok()
71 .filter(|v| !v.is_empty());
72
73 let repository_path = scan_path.join(relative);
74
75 let mut git_repository = match gix::open(repository_path.clone()) {
76 Ok(v) => v,
77 Err(error) => {
78 warn!(%error, "Failed to open repository {} to update metadata, skipping", relative.display());
79 continue;
80 }
81 };
82
83 git_repository.object_cache_size(10 * 1024 * 1024);
84
85 let res = Repository {
86 id,
87 name: name.to_string(),
88 description,
89 owner: find_gitweb_owner(repository_path.as_path()),
90 last_modified: {
91 let r =
92 find_last_committed_time(&git_repository).unwrap_or(OffsetDateTime::UNIX_EPOCH);
93 (r.unix_timestamp(), r.offset().whole_seconds())
94 },
95 default_branch: find_default_branch(&git_repository).ok().flatten(),
96 }
97 .insert(db, relative);
98
99 if let Err(error) = res {
100 warn!(%error, "Failed to insert repository");
101 }
102 }
103}
104
105fn find_default_branch(repo: &gix::Repository) -> Result<Option<String>, anyhow::Error> {
106 Ok(Some(repo.head()?.name().as_bstr().to_string()))
107}
108
109fn find_last_committed_time(repo: &gix::Repository) -> Result<OffsetDateTime, anyhow::Error> {
110 let mut timestamp = OffsetDateTime::UNIX_EPOCH;
111
112 for reference in repo.references()?.all()? {
113 let Ok(commit) = reference.unwrap().peel_to_commit() else {
114 continue;
115 };
116
117 let committer = commit.committer()?;
118 let mut committed_time = OffsetDateTime::from_unix_timestamp(committer.time.seconds)
119 .unwrap_or(OffsetDateTime::UNIX_EPOCH);
120
121 if let Ok(offset) = UtcOffset::from_whole_seconds(committer.time.offset) {
122 committed_time = committed_time.to_offset(offset);
123 }
124
125 if committed_time > timestamp {
126 timestamp = committed_time;
127 }
128 }
129
130 Ok(timestamp)
131}
132
133#[instrument(skip(db))]
134fn update_repository_reflog(scan_path: &Path, db: Arc<rocksdb::DB>) {
135 let repos = match Repository::fetch_all(&db) {
136 Ok(v) => v,
137 Err(error) => {
138 error!(%error, "Failed to read repository index to update reflog, consider deleting database directory");
139 return;
140 }
141 };
142
143 for (relative_path, db_repository) in repos {
144 let Some(git_repository) = open_repo(scan_path, &relative_path, db_repository.get(), &db)
145 else {
146 continue;
147 };
148
149 let references = match git_repository.references() {
150 Ok(v) => v,
151 Err(error) => {
152 error!(%error, "Failed to read references for {relative_path}");
153 continue;
154 }
155 };
156
157 let references = match references.all() {
158 Ok(v) => v,
159 Err(error) => {
160 error!(%error, "Failed to read references for {relative_path}");
161 continue;
162 }
163 };
164
165 let mut valid_references = Vec::new();
166
167 for reference in references {
168 let mut reference = match reference {
169 Ok(v) => v,
170 Err(error) => {
171 error!(%error, "Failed to read reference for {relative_path}");
172 continue;
173 }
174 };
175
176 let reference_name = reference.name();
177 if !matches!(
178 reference_name.category(),
179 Some(Category::Tag | Category::LocalBranch)
180 ) {
181 continue;
182 }
183
184 valid_references.push(reference_name.as_bstr().to_string());
185
186 if let Err(error) = branch_index_update(
187 &mut reference,
188 &relative_path,
189 db_repository.get(),
190 db.clone(),
191 &git_repository,
192 false,
193 ) {
194 error!(%error, "Failed to update reflog for {relative_path}@{:?}", valid_references.last());
195 }
196 }
197
198 if let Err(error) = db_repository.get().replace_heads(&db, &valid_references) {
199 error!(%error, "Failed to update heads");
200 }
201 }
202}
203
204#[instrument(skip(reference, db_repository, db, git_repository))]
205fn branch_index_update(
206 reference: &mut Reference<'_>,
207 relative_path: &str,
208 db_repository: &ArchivedRepository,
209 db: Arc<rocksdb::DB>,
210 git_repository: &gix::Repository,
211 force_reindex: bool,
212) -> Result<(), anyhow::Error> {
213 info!("Refreshing indexes");
214
215 let commit_tree = db_repository.commit_tree(db.clone(), reference.name().as_bstr().to_str()?);
216
217 if force_reindex {
218 commit_tree.drop_commits()?;
219 }
220
221 let commit = reference.peel_to_commit()?;
222
223 let latest_indexed = if let Some(latest_indexed) = commit_tree.fetch_latest_one()? {
224 if commit.id().as_bytes() == latest_indexed.get().hash.as_slice() {
225 info!("No commits since last index");
226 return Ok(());
227 }
228
229 Some(latest_indexed)
230 } else {
231 None
232 };
233
234 // TODO: stop collecting into a vec
235 let revwalk = git_repository
236 .rev_walk([commit.id().detach()])
237 .all()?
238 .collect::<Vec<_>>()
239 .into_iter()
240 .rev();
241
242 let tree_len = commit_tree.len()?;
243 let mut seen = false;
244 let mut i = 0;
245 for revs in &revwalk.chunks(250) {
246 let mut batch = WriteBatch::default();
247
248 for rev in revs {
249 let rev = rev?;
250
251 if let (false, Some(latest_indexed)) = (seen, &latest_indexed) {
252 if rev.id.as_bytes() == latest_indexed.get().hash.as_slice() {
253 seen = true;
254 }
255
256 continue;
257 }
258
259 seen = true;
260
261 if ((i + 1) % 25_000) == 0 {
262 info!("{} commits ingested", i + 1);
263 }
264
265 let commit = rev.object()?;
266 let author = commit.author()?;
267 let committer = commit.committer()?;
268
269 Commit::new(&commit, author, committer)?.insert(
270 &commit_tree,
271 tree_len + i,
272 &mut batch,
273 )?;
274 i += 1;
275 }
276
277 commit_tree.update_counter(tree_len + i, &mut batch)?;
278 db.write_without_wal(batch)?;
279 }
280
281 if !seen && !force_reindex {
282 warn!("Detected converged history, forcing reindex");
283
284 return branch_index_update(
285 reference,
286 relative_path,
287 db_repository,
288 db,
289 git_repository,
290 true,
291 );
292 }
293
294 Ok(())
295}
296
297#[instrument(skip(db))]
298fn update_repository_tags(scan_path: &Path, db: Arc<rocksdb::DB>) {
299 let repos = match Repository::fetch_all(&db) {
300 Ok(v) => v,
301 Err(error) => {
302 error!(%error, "Failed to read repository index to update tags, consider deleting database directory");
303 return;
304 }
305 };
306
307 for (relative_path, db_repository) in repos {
308 let Some(git_repository) = open_repo(scan_path, &relative_path, db_repository.get(), &db)
309 else {
310 continue;
311 };
312
313 if let Err(error) = tag_index_scan(
314 &relative_path,
315 db_repository.get(),
316 db.clone(),
317 &git_repository,
318 ) {
319 error!(%error, "Failed to update tags for {relative_path}");
320 }
321 }
322}
323
324#[instrument(skip(db_repository, db, git_repository))]
325fn tag_index_scan(
326 relative_path: &str,
327 db_repository: &ArchivedRepository,
328 db: Arc<rocksdb::DB>,
329 git_repository: &gix::Repository,
330) -> Result<(), anyhow::Error> {
331 let tag_tree = db_repository.tag_tree(db);
332
333 let git_tags: HashSet<_> = git_repository
334 .references()
335 .context("Failed to scan indexes on git repository")?
336 .all()?
337 .filter_map(Result::ok)
338 .filter(|v| v.name().category() == Some(Category::Tag))
339 .map(|v| v.name().as_bstr().to_string())
340 .collect();
341 let indexed_tags: HashSet<String> = tag_tree.list()?.into_iter().collect();
342
343 // insert any git tags that are missing from the index
344 for tag_name in git_tags.difference(&indexed_tags) {
345 tag_index_update(tag_name, git_repository, &tag_tree)?;
346 }
347
348 // remove any extra tags that the index has
349 // TODO: this also needs to check peel_to_tag
350 for tag_name in indexed_tags.difference(&git_tags) {
351 tag_index_delete(tag_name, &tag_tree)?;
352 }
353
354 Ok(())
355}
356
357#[instrument(skip(git_repository, tag_tree))]
358fn tag_index_update(
359 tag_name: &str,
360 git_repository: &gix::Repository,
361 tag_tree: &TagTree,
362) -> Result<(), anyhow::Error> {
363 let mut reference = git_repository
364 .find_reference(tag_name)
365 .context("Failed to read newly discovered tag")?;
366
367 if let Ok(tag) = reference.peel_to_tag() {
368 info!("Inserting newly discovered tag to index");
369
370 Tag::new(tag.tagger()?)?.insert(tag_tree, tag_name)?;
371 }
372
373 Ok(())
374}
375
376#[instrument(skip(tag_tree))]
377fn tag_index_delete(tag_name: &str, tag_tree: &TagTree) -> Result<(), anyhow::Error> {
378 info!("Removing stale tag from index");
379 tag_tree.remove(tag_name)?;
380
381 Ok(())
382}
383
384#[instrument(skip(scan_path, db_repository, db))]
385fn open_repo<P: AsRef<Path> + Debug>(
386 scan_path: &Path,
387 relative_path: P,
388 db_repository: &ArchivedRepository,
389 db: &rocksdb::DB,
390) -> Option<gix::Repository> {
391 match gix::open(scan_path.join(relative_path.as_ref())) {
392 Ok(mut v) => {
393 v.object_cache_size(10 * 1024 * 1024);
394 Some(v)
395 }
396 Err(gix::open::Error::Io(e)) if e.kind() == std::io::ErrorKind::NotFound => {
397 warn!("Repository gone from disk, removing from db");
398
399 if let Err(error) = db_repository.delete(db, relative_path) {
400 warn!(%error, "Failed to delete dangling index");
401 }
402
403 None
404 }
405 Err(error) => {
406 warn!(%error, "Failed to reindex, skipping");
407 None
408 }
409 }
410}
411
412fn get_relative_path<'a>(relative_to: &Path, full_path: &'a Path) -> Option<&'a Path> {
413 full_path.strip_prefix(relative_to).ok()
414}
415
416fn discover_repositories(current: &Path, discovered_repos: &mut Vec<PathBuf>) {
417 let current = match std::fs::read_dir(current) {
418 Ok(v) => v,
419 Err(error) => {
420 error!(%error, "Failed to enter repository directory {}", current.display());
421 return;
422 }
423 };
424
425 let dirs = current
426 .filter_map(Result::ok)
427 .map(|v| v.path())
428 .filter(|path| path.is_dir());
429
430 for dir in dirs {
431 if dir.join("packed-refs").is_file() {
432 // we've hit what looks like a bare git repo, lets take it
433 discovered_repos.push(dir);
434 } else {
435 // probably not a bare git repo, lets recurse deeper
436 discover_repositories(&dir, discovered_repos);
437 }
438 }
439}
440
441fn find_gitweb_owner(repository_path: &Path) -> Option<String> {
442 // Load the Git config file and attempt to extract the owner from the "gitweb" section.
443 // If the owner is not found, an empty string is returned.
444 Ini::load_from_file(repository_path.join("config"))
445 .ok()?
446 .section_mut(Some("gitweb"))
447 .and_then(|section| section.remove("owner"))
448}