alice.pds.demo.boltless.dev
This repository has no description
1use std::net::SocketAddr;
2use std::path::PathBuf;
3use std::pin::Pin;
4
5use gix::bstr::ByteSlice as _;
6use gix::revision::plumbing::Spec as RevSpec;
7use gix::revision::walk::Sorting;
8use tempfile::TempDir;
9use tokio::sync::mpsc;
10use tokio_stream::wrappers::ReceiverStream;
11use tokio_stream::Stream;
12use tonic::transport::Server;
13use tonic::{Request, Response, Status};
14use tracing::{debug, info};
15
16use crate::diff::{self, prepare_interdiff};
17use crate::protocol::v1::diff_request::ComparisonType;
18use crate::protocol::v1::git_mirror_service_server::{GitMirrorService, GitMirrorServiceServer};
19use crate::protocol::v1::{
20 BlobChunk, CommitLogRequest, CommitLogResponse, DiffRequest, FileDiff, GetBlobRequest,
21 GitCommit, GitSignature, InterdiffRequest, MergeCheckRequest, MergeCheckResponse,
22 ResolveRevRequest, ResolveRevResponse,
23};
24
25type FileDiffStream = Pin<Box<dyn Stream<Item = Result<FileDiff, Status>> + Send>>;
26type BlobChunkStream = Pin<Box<dyn Stream<Item = Result<BlobChunk, Status>> + Send>>;
27type CommitLogResponseStream =
28 Pin<Box<dyn Stream<Item = Result<CommitLogResponse, Status>> + Send>>;
29
30const BLOB_CHUNK_SIZE: usize = 64 * 1024;
31const COMMIT_LOG_BATCH_SIZE: usize = 128;
32
33struct GitMirror {
34 repo_base: PathBuf,
35}
36
37/// A [`gix::Repository`] backed by a temporary directory that is removed on drop. Derefs to
38/// `gix::Repository`, so it is used just like one; keeping it alive keeps the scratch dir alive.
39struct TempRepository {
40 repo: gix::Repository,
41 // Declared AFTER `repo` so `repo` drops first: any file handles into the scratch dir close
42 // before the dir itself is removed (Rust drops struct fields in declaration order).
43 _dir: TempDir,
44}
45
46impl TempRepository {
47 fn new(repo: gix::Repository, dir: TempDir) -> Self {
48 Self { repo, _dir: dir }
49 }
50}
51
52impl std::ops::Deref for TempRepository {
53 type Target = gix::Repository;
54 fn deref(&self) -> &gix::Repository {
55 &self.repo
56 }
57}
58
59impl GitMirror {
60 fn open_repo(&self, did: &str) -> Result<gix::Repository, Status> {
61 if !did.starts_with("did:") {
62 return Err(Status::invalid_argument("repo must be a DID"));
63 }
64 gix::open(self.repo_base.join(did))
65 .map_err(|e| Status::not_found(format!("repo not found: {e}")))
66 }
67
68 /// Build a throwaway bare repo whose `objects/info/alternates` points read-only at each of the
69 /// given git repositories, so a single `gix::Repository` can see objects from all of them
70 /// without ever mutating them.
71 fn open_scratch(&self, dids: &[&str]) -> Result<TempRepository, Status> {
72 let mut seen: Vec<&str> = Vec::new();
73 let mut object_dirs = Vec::new();
74 for &did in dids {
75 if seen.contains(&did) {
76 continue
77 }
78 if !did.starts_with("did:") {
79 return Err(Status::invalid_argument("repo must be a DID"));
80 }
81 seen.push(did);
82 let object_dir = std::fs::canonicalize(self.repo_base.join(did).join("objects"))
83 .map_err(|_| Status::not_found(format!("repo not found: {did}")))?;
84 object_dirs.push(object_dir);
85 }
86 if object_dirs.is_empty() {
87 return Err(Status::internal("open_scratch requires at least one repo"));
88 }
89
90 let scratch = tempfile::tempdir().map_err(|e| Status::internal(e.to_string()))?;
91 gix::init_bare(scratch.path()).map_err(|e| Status::internal(e.to_string()))?;
92
93 let info_dir = scratch.path().join("objects").join("info");
94 std::fs::create_dir_all(&info_dir).map_err(|e| Status::internal(e.to_string()))?;
95 let alternates = object_dirs
96 .iter()
97 .map(|p| p.display().to_string())
98 .collect::<Vec<_>>()
99 .join("\n");
100 std::fs::write(info_dir.join("alternates"), format!("{alternates}\n"))
101 .map_err(|e| Status::internal(e.to_string()))?;
102
103 let repo = gix::open(scratch.path())
104 .map_err(|e| Status::internal(e.to_string()))?
105 .with_object_memory();
106 Ok(TempRepository::new(repo, scratch))
107 }
108}
109
110#[tonic::async_trait]
111impl GitMirrorService for GitMirror {
112 type DiffStream = FileDiffStream;
113
114 async fn diff(
115 &self,
116 request: Request<DiffRequest>,
117 ) -> Result<Response<Self::DiffStream>, Status> {
118 let req = request.into_inner();
119 let head = req
120 .head
121 .ok_or_else(|| Status::invalid_argument("head is required"))?;
122 let base = req.base;
123 debug!(
124 base_repo = ?base.as_ref().map(|b| b.repo.as_str()),
125 base_commit = ?base.as_ref().map(|b| b.commit.as_bstr()),
126 head_repo = %head.repo,
127 head_commit = %head.commit.as_bstr(),
128 "diff request",
129 );
130 let scratch = match &base {
131 Some(base) => self.open_scratch(&[head.repo.as_str(), base.repo.as_str()])?,
132 None => self.open_scratch(&[head.repo.as_str()])?,
133 };
134 let head_commit_id = find_commit_by_sha(&scratch, &head.commit)?;
135 let base_commit_id = base
136 .as_ref()
137 .map(|b| find_commit_by_sha(&scratch, &b.commit))
138 .transpose()?;
139 let comparison_type =
140 ComparisonType::try_from(req.comparison_type).unwrap_or(ComparisonType::Unspecified);
141
142 let (tx, rx) = mpsc::channel::<Result<FileDiff, Status>>(16);
143 tokio::task::spawn_blocking(move || {
144 let repo = &*scratch;
145
146 let run = || -> anyhow::Result<()> {
147 let head_commit = repo.find_commit(head_commit_id)?;
148 let old = match base_commit_id {
149 Some(base_commit_id) => match comparison_type {
150 ComparisonType::OnlyInHead => {
151 repo.find_tree(repo.find_commit(base_commit_id)?.tree_id()?)?
152 }
153 ComparisonType::Unspecified | ComparisonType::Intersection => {
154 let merge_base =
155 repo.merge_base(base_commit_id, head_commit_id)?.detach();
156 repo.find_tree(repo.find_commit(merge_base)?.tree_id()?)?
157 }
158 },
159 None => match head_commit.parent_ids().next() {
160 Some(parent_id) => {
161 repo.find_tree(repo.find_commit(parent_id.detach())?.tree_id()?)?
162 }
163 None => repo.empty_tree(),
164 },
165 };
166 let new = repo.find_tree(head_commit.tree_id()?)?;
167 for item in diff::diff(repo, &old, &new, false)? {
168 if tx.blocking_send(Ok(item?.into())).is_err() {
169 break;
170 }
171 }
172 Ok(())
173 };
174 if let Err(e) = run() {
175 let _ = tx.blocking_send(Err(Status::internal(e.to_string())));
176 }
177 });
178
179 Ok(Response::new(Box::pin(ReceiverStream::new(rx))))
180 }
181
182 type InterdiffStream = FileDiffStream;
183
184 async fn interdiff(
185 &self,
186 request: Request<InterdiffRequest>,
187 ) -> Result<Response<Self::InterdiffStream>, Status> {
188 let req = request.into_inner();
189 let repo = self.open_repo(&req.repo)?;
190 let from_base_id = resolve_commit_oid(&repo, &req.from_base)?;
191 let from_head_id = resolve_commit_oid(&repo, &req.from_head)?;
192 let to_base_id = resolve_commit_oid(&repo, &req.to_base)?;
193 let to_head_id = resolve_commit_oid(&repo, &req.to_head)?;
194 let safe = repo.into_sync();
195
196 let (tx, rx) = mpsc::channel::<Result<FileDiff, Status>>(16);
197 tokio::task::spawn_blocking(move || {
198 let repo = safe.to_thread_local();
199 let run = || -> anyhow::Result<()> {
200 let to_head = repo.find_commit(to_head_id)?;
201 let to_head_tree = to_head.tree()?;
202 let rebased_tree =
203 prepare_interdiff(&repo, (from_base_id, from_head_id), to_base_id)?;
204 for item in diff::diff(&repo, &rebased_tree, &to_head_tree, true)? {
205 if tx.blocking_send(Ok(item?.into())).is_err() {
206 break;
207 }
208 }
209 Ok(())
210 };
211 if let Err(e) = run() {
212 let _ = tx.blocking_send(Err(Status::internal(e.to_string())));
213 }
214 });
215
216 Ok(Response::new(Box::pin(ReceiverStream::new(rx))))
217 }
218
219 type GetBlobStream = BlobChunkStream;
220
221 async fn get_blob(
222 &self,
223 request: Request<GetBlobRequest>,
224 ) -> Result<Response<Self::GetBlobStream>, Status> {
225 let req = request.into_inner();
226 let repo = self.open_repo(&req.repo)?;
227 let oid = gix::ObjectId::from_hex(req.oid.as_bytes())
228 .map_err(|e| Status::invalid_argument(format!("bad oid '{}': {e}", req.oid)))?;
229 let safe = repo.into_sync();
230
231 let (tx, rx) = mpsc::channel::<Result<BlobChunk, Status>>(16);
232 tokio::task::spawn_blocking(move || {
233 let repo = safe.to_thread_local();
234 let run = || -> anyhow::Result<()> {
235 let blob = repo.find_object(oid)?.try_into_blob()?;
236 for chunk in blob.data.chunks(BLOB_CHUNK_SIZE) {
237 let msg = BlobChunk {
238 data: chunk.to_vec(),
239 };
240 if tx.blocking_send(Ok(msg)).is_err() {
241 break;
242 }
243 }
244 Ok(())
245 };
246 if let Err(e) = run() {
247 let _ = tx.blocking_send(Err(Status::internal(e.to_string())));
248 }
249 });
250
251 Ok(Response::new(Box::pin(ReceiverStream::new(rx))))
252 }
253
254 type CommitLogStream = CommitLogResponseStream;
255
256 async fn commit_log(
257 &self,
258 request: Request<CommitLogRequest>,
259 ) -> Result<Response<Self::CommitLogStream>, Status> {
260 let req = request.into_inner();
261 debug!(
262 repo = %req.repo,
263 all_refs = req.all_refs,
264 ranges = ?req.ranges,
265 after = ?req.after,
266 before = ?req.before,
267 max_commits = req.max_commits,
268 skip = req.skip,
269 "commit_log request",
270 );
271 if req.all_refs != req.ranges.is_empty() {
272 return Err(Status::invalid_argument(
273 "specify exactly one of `ranges` or `all_refs`",
274 ));
275 }
276 let repo = self.open_repo(&req.repo)?;
277 let safe = repo.into_sync();
278
279 let (tx, rx) = mpsc::channel::<Result<CommitLogResponse, Status>>(16);
280 tokio::task::spawn_blocking(move || {
281 let repo = safe.to_thread_local();
282 let run = || -> anyhow::Result<()> {
283 let after = req.after.map(|t| t.seconds);
284 let before = req.before.map(|t| t.seconds);
285 let max_commits = req.max_commits as usize;
286 let mut skip = req.skip as usize;
287 let mut sent = 0usize;
288 let mut batch = Vec::with_capacity(COMMIT_LOG_BATCH_SIZE);
289
290 for info in commit_log_walk(&repo, &req)? {
291 let info = info?;
292 let commit_time = info.commit_time();
293 if before.is_some_and(|b| commit_time > b) {
294 continue;
295 }
296 if after.is_some_and(|a| commit_time < a) {
297 break;
298 }
299 if skip > 0 {
300 skip -= 1;
301 continue;
302 }
303 if max_commits != 0 && sent >= max_commits {
304 break;
305 }
306 batch.push(GitCommit::try_from(&info.object()?)?);
307 sent += 1;
308 if batch.len() == COMMIT_LOG_BATCH_SIZE {
309 let msg = CommitLogResponse {
310 commits: std::mem::take(&mut batch),
311 };
312 if tx.blocking_send(Ok(msg)).is_err() {
313 return Ok(()); // client hung up
314 }
315 }
316 }
317 if !batch.is_empty() {
318 let _ = tx.blocking_send(Ok(CommitLogResponse { commits: batch }));
319 }
320 Ok(())
321 };
322 if let Err(e) = run() {
323 let _ = tx.blocking_send(Err(Status::internal(e.to_string())));
324 }
325 });
326
327 Ok(Response::new(Box::pin(ReceiverStream::new(rx))))
328 }
329
330 async fn merge_check(
331 &self,
332 request: Request<MergeCheckRequest>,
333 ) -> Result<Response<MergeCheckResponse>, Status> {
334 let req = request.into_inner();
335 let target = req
336 .target
337 .ok_or_else(|| Status::invalid_argument("target is required"))?;
338 let source = req
339 .source
340 .ok_or_else(|| Status::invalid_argument("source is required"))?;
341 debug!(
342 target_repo = %target.repo,
343 target_commit = %target.commit.as_bstr(),
344 source_repo = %source.repo,
345 source_commit = %source.commit.as_bstr(),
346 "merge_check request",
347 );
348
349 let scratch = self.open_scratch(&[target.repo.as_str(), source.repo.as_str()])?;
350
351 let response =
352 tokio::task::spawn_blocking(move || -> Result<MergeCheckResponse, Status> {
353 let target_oid = find_commit_by_sha(&scratch, &target.commit)?;
354 let source_oid = find_commit_by_sha(&scratch, &source.commit)?;
355 crate::merge::merge_check(&scratch, target_oid, source_oid)
356 .map_err(|e| Status::internal(e.to_string()))
357 })
358 .await
359 .map_err(|e| Status::internal(format!("merge task panicked: {e}")))??;
360
361 Ok(Response::new(response))
362 }
363
364 async fn resolve_rev(
365 &self,
366 request: Request<ResolveRevRequest>,
367 ) -> Result<Response<ResolveRevResponse>, Status> {
368 let req = request.into_inner();
369 debug!(repo = %req.repo, rev = %req.rev.as_bstr(), "resolve_rev request");
370 let repo = self.open_repo(&req.repo)?;
371 let commit = tokio::task::spawn_blocking(move || -> Result<String, Status> {
372 Ok(resolve_simple_rev(&repo, &req.rev)?.to_string())
373 })
374 .await
375 .map_err(|e| Status::internal(format!("resolve task panicked: {e}")))??;
376 Ok(Response::new(ResolveRevResponse { commit }))
377 }
378}
379
380fn resolve_commit_oid(repo: &gix::Repository, spec: &[u8]) -> Result<gix::ObjectId, Status> {
381 let commit = repo
382 .rev_parse_single(spec.as_bstr())
383 .map_err(|e| {
384 Status::invalid_argument(format!("bad rev '{}': {e}", String::from_utf8_lossy(spec)))
385 })?
386 .object()
387 .map_err(|e| Status::internal(e.to_string()))?
388 .peel_to_commit()
389 .map_err(|e| Status::internal(e.to_string()))?;
390 Ok(commit.id)
391}
392
393fn resolve_simple_rev(repo: &gix::Repository, rev: &[u8]) -> Result<gix::ObjectId, Status> {
394 let unsupported = rev == b"HEAD"
395 || rev == b"@"
396 || rev.windows(2).any(|w| w == b"@{")
397 || rev.iter().any(|&b| matches!(b, b'~' | b'^' | b':'));
398 if unsupported {
399 return Err(Status::invalid_argument(format!(
400 "unsupported rev '{}': only a branch, tag, or commit oid is allowed",
401 rev.as_bstr()
402 )));
403 }
404 resolve_commit_oid(repo, rev)
405}
406
407fn find_commit_by_sha(repo: &gix::Repository, sha: &[u8]) -> Result<gix::ObjectId, Status> {
408 let oid = gix::ObjectId::from_hex(sha).map_err(|e| {
409 Status::invalid_argument(format!("bad commit sha '{}': {e}", sha.as_bstr()))
410 })?;
411 repo.find_commit(oid)
412 .map_err(|_| Status::not_found(format!("commit not found: {oid}")))?;
413 Ok(oid)
414}
415
416fn commit_log_walk<'repo>(
417 repo: &'repo gix::Repository,
418 req: &CommitLogRequest,
419) -> anyhow::Result<gix::revision::Walk<'repo>> {
420 let mut tips = Vec::new();
421 let mut hidden = Vec::new();
422
423 if req.all_refs {
424 for r in repo.references()?.all()? {
425 let mut r = r.map_err(|e| anyhow::anyhow!(e))?;
426 if let Ok(commit) = r.peel_to_commit() {
427 tips.push(commit.id);
428 }
429 }
430 } else {
431 for range in &req.ranges {
432 let revspec = repo.rev_parse(range.as_bstr())?;
433 let spec = revspec.detach();
434 match spec {
435 RevSpec::Include(id) => tips.push(id),
436 RevSpec::Range { from, to } => {
437 tips.push(to);
438 hidden.push(from);
439 }
440 _ => {
441 anyhow::bail!("The spec isn't currently supported: {spec:?}")
442 }
443 }
444 }
445 }
446
447 Ok(repo
448 .rev_walk(tips)
449 .sorting(Sorting::ByCommitTime(Default::default()))
450 .with_hidden(hidden)
451 .all()?)
452}
453
454impl TryFrom<&gix::Commit<'_>> for GitCommit {
455 type Error = anyhow::Error;
456
457 fn try_from(commit: &gix::Commit<'_>) -> anyhow::Result<Self> {
458 let decoded = commit.decode()?;
459 Ok(GitCommit {
460 oid: commit.id.to_string(),
461 author: Some(decoded.author()?.try_into()?),
462 committer: Some(decoded.committer()?.try_into()?),
463 message: decoded.message.to_vec(),
464 parents: decoded
465 .parents
466 .into_iter()
467 .map(|id| id.to_string())
468 .collect(),
469 extra_headers: decoded
470 .extra_headers
471 .iter()
472 .map(|(k, v)| (k.to_string(), v.to_string()))
473 .collect(),
474 })
475 }
476}
477
478impl TryFrom<gix::actor::SignatureRef<'_>> for GitSignature {
479 type Error = anyhow::Error;
480
481 fn try_from(sig: gix::actor::SignatureRef<'_>) -> anyhow::Result<Self> {
482 let time = sig.time()?;
483 Ok(GitSignature {
484 name: sig.name.to_vec(),
485 email: sig.email.to_vec(),
486 date: Some(prost_types::Timestamp {
487 seconds: time.seconds,
488 nanos: 0,
489 }),
490 })
491 }
492}
493
494pub async fn serve(addr: SocketAddr, repo_base: PathBuf) -> anyhow::Result<()> {
495 let service = GitMirror { repo_base };
496 info!(addr = %addr, "gitmirror gRPC server listening");
497
498 Server::builder()
499 .add_service(GitMirrorServiceServer::new(service))
500 .serve(addr)
501 .await?;
502
503 Ok(())
504}
505
506#[cfg(test)]
507mod tests {
508 use super::*;
509
510 #[test]
511 fn temp_repository_removes_scratch_dir_on_drop() {
512 let dir = tempfile::tempdir().unwrap();
513 gix::init_bare(dir.path()).unwrap();
514 let repo = gix::open(dir.path()).unwrap().with_object_memory();
515 let temp = TempRepository::new(repo, dir);
516 let path = temp.git_dir().to_path_buf();
517 assert!(path.exists());
518 // Derefs to gix::Repository.
519 assert!(temp.object_hash() == gix::hash::Kind::Sha1);
520 drop(temp);
521 assert!(!path.exists(), "scratch dir should be gone after drop");
522 }
523}