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