Sign in

flagon-io/g1t

public

g1t/services/repos/src/lib.rs

644 lines23,053 bytesCodeBlame

Pick any line to see why it is the way it is: the commit, the pull request and issue it came from, and what the agent was thinking.

Rust repos service with shipping; pull requests kept in the model1//! The repos service: repository metadata, contents, forks, landing, and
2//! git over HTTPS.
3//!
4//! Other services reach it over `POST /rpc/<method>`; see
5//! `g1t_contracts::repos` for the methods and their arguments. Any other
6//! request is treated as git's smart HTTP protocol.
7
Diffs on attempts; hosted agent presented as the g1t agent8mod diff;
Rust repos service with shipping; pull requests kept in the model9mod git_http;
10mod land;
11mod registry;
12mod store;
13
14use g1t_contracts::events::{GitPush, NewEvent, RepoCreated, RepoForked};
15use g1t_contracts::repos::*;
RFC 3339 timestamps in identity and repos16use g1t_contracts::time::rfc3339;
Rust repos service with shipping; pull requests kept in the model17use g1t_contracts::{
18 FailureCode, Outcome, User, Viewer, is_valid_namespace, is_valid_repo_name, new_id,
19};
20use g1t_kit::{args, js, now_ms, reply, rpc_method};
Diffs on attempts; hosted agent presented as the g1t agent21use std::collections::{HashMap, HashSet, VecDeque};
Rust repos service with shipping; pull requests kept in the model22
23use serde::Serialize;
24use worker::wasm_bindgen::JsValue;
25use worker::{Context, Env, Request, Response, Result, event};
26
27use registry::{Registry, can_read, can_write, store_key};
28use store::{ArtifactsStore, GitRepo, GitStore, Scope};
29
30/// Namespace that holds every attempt's fork: `attempts/<attempt id>`.
31const ATTEMPTS_NAMESPACE: &str = "attempts";
32const MAX_TEXT_BYTES: usize = 512 * 1024;
33/// How far back an attempt may have forked and still be landed.
34const MAX_ANCESTRY: u32 = 1000;
35const SOURCE: &str = "repos";
36const UNVERIFIED: &str = "Confirm your email address first. Check your inbox, or resend the link from the banner on g1t.sh.";
37
38fn not_found<T>() -> Outcome<T> {
39 Outcome::fail(FailureCode::NotFound, "Repository not found.")
40}
41
42/// Decoded text, or `None` when the file is too large or looks binary.
43fn text_of(bytes: Vec<u8>) -> Option<String> {
44 if bytes.len() > MAX_TEXT_BYTES || bytes.contains(&0) {
45 return None;
46 }
47 Some(String::from_utf8_lossy(&bytes).into_owned())
48}
49
50fn is_readme(name: &str) -> bool {
51 matches!(
52 name.to_lowercase().as_str(),
53 "readme" | "readme.md" | "readme.markdown" | "readme.txt"
54 )
55}
56
57/// Whether `ancestor` is reachable from the newest commit in `history`.
58///
59/// `history` is the first-parent chain, which is all the store lists; a fork
60/// that merged the target branch in has the target's head on a second
61/// parent, so the walk follows every parent.
62async fn descends_from<R: GitRepo>(repo: &R, history: &[Commit], ancestor: &str) -> Result<bool> {
63 let known: HashMap<&str, &[String]> = history
64 .iter()
65 .map(|commit| (commit.hash.as_str(), commit.parents.as_slice()))
66 .collect();
67 let mut seen = HashSet::new();
68 let mut queue: Vec<String> = history
69 .first()
70 .map(|c| c.hash.clone())
71 .into_iter()
72 .collect();
73 while let Some(hash) = queue.pop() {
74 if hash == ancestor {
75 return Ok(true);
76 }
77 if !seen.insert(hash.clone()) || seen.len() > MAX_ANCESTRY as usize {
78 continue;
79 }
80 match known.get(hash.as_str()) {
81 Some(parents) => queue.extend(parents.iter().cloned()),
82 None => queue.extend(repo.parents(&hash).await?.unwrap_or_default()),
83 }
84 }
85 Ok(false)
86}
87
Diffs on attempts; hosted agent presented as the g1t agent88/// The commit closest to the newest in `history` that is also in `shared`:
89/// where a fork and the repository it came from last agreed.
90async fn nearest_ancestor_in<R: GitRepo>(
91 repo: &R,
92 history: &[Commit],
93 shared: &HashSet<String>,
94) -> Result<Option<String>> {
95 let known: HashMap<&str, &[String]> = history
96 .iter()
97 .map(|commit| (commit.hash.as_str(), commit.parents.as_slice()))
98 .collect();
99 let mut seen = HashSet::new();
100 let mut queue: VecDeque<String> = history
101 .first()
102 .map(|c| c.hash.clone())
103 .into_iter()
104 .collect();
105 while let Some(hash) = queue.pop_front() {
106 if shared.contains(&hash) {
107 return Ok(Some(hash));
108 }
109 if !seen.insert(hash.clone()) || seen.len() > MAX_ANCESTRY as usize {
110 continue;
111 }
112 match known.get(hash.as_str()) {
113 Some(parents) => queue.extend(parents.iter().cloned()),
114 None => queue.extend(repo.parents(&hash).await?.unwrap_or_default()),
115 }
116 }
117 Ok(None)
118}
119
Rust repos service with shipping; pull requests kept in the model120struct Repos<S: GitStore> {
121 registry: Registry,
122 store: S,
123 /// The events service, an RPC stub.
124 events: JsValue,
125}
126
127impl<S: GitStore> Repos<S> {
128 async fn publish<T: Serialize>(&self, event: NewEvent<T>) -> Result<()> {
129 js::call(&self.events, "publish", &[js::to_js(&[event])?]).await?;
130 Ok(())
131 }
132
133 /// Resolves a repo the viewer may read; private repos look missing.
134 async fn readable(&self, path: &RepoPath, viewer: &Viewer) -> Result<Option<Repo>> {
135 Ok(self
136 .registry
137 .by_path(path)
138 .await?
139 .filter(|repo| can_read(repo, viewer)))
140 }
141
142 async fn get(&self, a: GetArgs) -> Result<Outcome<Repo>> {
143 Ok(self
144 .readable(&a.path, &a.viewer)
145 .await?
146 .map_or_else(not_found, Outcome::Ok))
147 }
148
149 async fn get_by_id(&self, a: GetByIdArgs) -> Result<Outcome<Repo>> {
150 Ok(self
151 .registry
152 .by_id(&a.id)
153 .await?
154 .filter(|repo| can_read(repo, &a.viewer))
155 .map_or_else(not_found, Outcome::Ok))
156 }
157
158 async fn create(&self, a: CreateArgs) -> Result<Outcome<Repo>> {
159 if !a.owner.verified {
160 return Ok(Outcome::fail(FailureCode::Forbidden, UNVERIFIED));
161 }
162 let name = a.name.trim().to_lowercase();
163 if !is_valid_repo_name(&name) {
164 return Ok(Outcome::fail(
165 FailureCode::Invalid,
166 "Use letters, digits, dots, hyphens and underscores only.",
167 ));
168 }
169 if !is_valid_namespace(&a.owner.username) {
170 return Ok(Outcome::fail(
171 FailureCode::Invalid,
172 "This account cannot own repositories.",
173 ));
174 }
175 let path = RepoPath {
176 namespace: a.owner.username.clone(),
177 name,
178 };
179 if self.registry.by_path(&path).await?.is_some() {
180 return Ok(Outcome::fail(
181 FailureCode::Conflict,
182 "You already have a repository with that name.",
183 ));
184 }
185 let now = now_ms();
186 let repo = Repo {
187 id: new_id("rep", now),
188 namespace: path.namespace,
189 name: path.name,
190 description: a
191 .description
192 .map(|text| text.trim().to_owned())
193 .filter(|text| !text.is_empty()),
194 is_private: a.is_private,
195 owner_id: a.owner.id.clone(),
196 default_branch: "main".to_owned(),
197 fork_of: None,
RFC 3339 timestamps in identity and repos198 created_at: rfc3339(now),
Rust repos service with shipping; pull requests kept in the model199 };
200 self.store
201 .create(
202 &store_key(&repo),
203 repo.description.as_deref(),
204 &repo.default_branch,
205 )
206 .await?;
207 self.registry.insert(&repo).await?;
208 self.publish(NewEvent {
209 kind: "repo.created",
210 source: SOURCE,
211 repo_id: Some(repo.id.clone()),
212 actor: Some(a.owner.id),
213 data: RepoCreated {
214 repo_id: repo.id.clone(),
215 namespace: repo.namespace.clone(),
216 name: repo.name.clone(),
217 is_private: repo.is_private,
218 },
219 })
220 .await?;
221 Ok(Outcome::Ok(repo))
222 }
223
224 async fn tree(&self, a: TreeArgs) -> Result<Outcome<TreeView>> {
225 let Some(repo) = self.readable(&a.path, &a.viewer).await? else {
226 return Ok(not_found());
227 };
228 let git = self.store.open(&store_key(&repo)).await?;
229 let git_ref = a
230 .git_ref
231 .clone()
232 .unwrap_or_else(|| repo.default_branch.clone());
233
234 let Some(head) = git.log(&git_ref, 1).await?.into_iter().next() else {
235 // An unknown ref is an error; a repo with no commits is just empty.
236 if a.git_ref.is_some() {
237 return Ok(Outcome::fail(
238 FailureCode::NotFound,
239 "No such branch, tag or commit.",
240 ));
241 }
242 return Ok(Outcome::Ok(TreeView {
243 repo,
244 git_ref,
245 path: a.tree_path,
246 head: None,
247 entries: Vec::new(),
248 readme: None,
249 }));
250 };
251
252 let no_directory = || Outcome::fail(FailureCode::NotFound, "No such directory.");
253 let mut entries = git.read_tree(&head.tree_hash).await?;
254 for segment in a.tree_path.split('/').filter(|segment| !segment.is_empty()) {
255 let next = entries.as_ref().and_then(|entries| {
256 entries
257 .iter()
258 .find(|entry| entry.name == segment && entry.kind == EntryKind::Tree)
259 });
260 let Some(next) = next else {
261 return Ok(no_directory());
262 };
263 entries = git.read_tree(&next.hash).await?;
264 }
265 let Some(mut entries) = entries else {
266 return Ok(no_directory());
267 };
268 // Directories first, then by name.
269 entries.sort_by(|a, b| {
270 (b.kind == EntryKind::Tree)
271 .cmp(&(a.kind == EntryKind::Tree))
272 .then_with(|| a.name.cmp(&b.name))
273 });
274
275 let readme_entry = entries
276 .iter()
277 .find(|entry| entry.kind == EntryKind::Blob && is_readme(&entry.name));
278 let readme = match readme_entry {
279 Some(entry) => git.read_blob(&entry.hash).await?.map(|bytes| Readme {
280 name: entry.name.clone(),
281 text: text_of(bytes),
282 }),
283 None => None,
284 };
285 Ok(Outcome::Ok(TreeView {
286 repo,
287 git_ref,
288 path: a.tree_path,
289 head: Some(head),
290 entries,
291 readme,
292 }))
293 }
294
295 async fn blob(&self, a: BlobArgs) -> Result<Outcome<BlobView>> {
296 let Some(repo) = self.readable(&a.path, &a.viewer).await? else {
297 return Ok(not_found());
298 };
299 let bytes = if a.file_path.is_empty() {
300 None
301 } else {
302 let git = self.store.open(&store_key(&repo)).await?;
303 git.read_file(&a.git_ref, &a.file_path).await?
304 };
305 let Some(bytes) = bytes else {
306 return Ok(Outcome::fail(FailureCode::NotFound, "No such file."));
307 };
308 Ok(Outcome::Ok(BlobView {
309 repo,
310 git_ref: a.git_ref,
311 path: a.file_path,
312 size: bytes.len() as u64,
313 text: text_of(bytes),
314 }))
315 }
316
317 async fn log(&self, a: LogArgs) -> Result<Outcome<Vec<Commit>>> {
318 let Some(repo) = self.readable(&a.path, &a.viewer).await? else {
319 return Ok(not_found());
320 };
321 let git = self.store.open(&store_key(&repo)).await?;
322 let git_ref = a.git_ref.unwrap_or_else(|| repo.default_branch.clone());
323 Ok(Outcome::Ok(git.log(&git_ref, a.limit).await?))
324 }
325
326 async fn fork_for_attempt(&self, a: ForkArgs) -> Result<Outcome<Repo>> {
327 let viewer = Some(a.actor.clone());
328 let Some(source) = self
329 .registry
330 .by_id(&a.source_id)
331 .await?
332 .filter(|repo| can_read(repo, &viewer))
333 else {
334 return Ok(not_found());
335 };
336 let now = now_ms();
337 let fork = Repo {
338 id: new_id("rep", now),
339 namespace: ATTEMPTS_NAMESPACE.to_owned(),
340 name: a.attempt_id.clone(),
341 description: None,
342 // A fork is exactly as visible as the repo it came from.
343 is_private: source.is_private,
344 owner_id: a.actor.id.clone(),
345 default_branch: source.default_branch.clone(),
346 fork_of: Some(source.id.clone()),
RFC 3339 timestamps in identity and repos347 created_at: rfc3339(now),
Rust repos service with shipping; pull requests kept in the model348 };
349 self.store
350 .open(&store_key(&source))
351 .await?
352 .fork(&store_key(&fork))
353 .await?;
354 self.registry.insert(&fork).await?;
355 self.publish(NewEvent {
356 kind: "repo.forked",
357 source: SOURCE,
358 repo_id: Some(source.id.clone()),
359 actor: Some(a.actor.id),
360 data: RepoForked {
361 repo_id: fork.id.clone(),
362 source_repo_id: source.id,
363 attempt_id: a.attempt_id,
364 },
365 })
366 .await?;
367 Ok(Outcome::Ok(fork))
368 }
369
370 async fn git_access(&self, a: GitAccessArgs) -> Result<Outcome<GitAccess>> {
371 let write = a.service == GitService::ReceivePack;
372 // Anonymous callers are asked to authenticate whether or not the repo
373 // exists, so private repos cannot be told apart from missing ones.
374 let denied = || match &a.viewer {
375 Some(_) => not_found(),
376 None => Outcome::fail(FailureCode::Unauthenticated, "Authentication required."),
377 };
378 if let (true, Some(user)) = (write, &a.viewer)
379 && !user.verified
380 {
381 return Ok(Outcome::fail(FailureCode::Forbidden, UNVERIFIED));
382 }
383
384 let repo = match self.registry.by_path(&a.path).await? {
385 Some(repo) => {
386 let allowed = if write {
387 can_write(&repo, &a.viewer)
388 } else {
389 can_read(&repo, &a.viewer)
390 };
391 if !allowed {
392 return Ok(denied());
393 }
394 repo
395 }
396 None => {
397 // Push to create, in the pusher's own namespace only.
398 let owner = a
399 .viewer
400 .as_ref()
401 .filter(|user| write && user.username == a.path.namespace.to_lowercase());
402 let Some(owner) = owner else {
403 return Ok(denied());
404 };
405 let created = self
406 .create(CreateArgs {
407 owner: owner.clone(),
408 name: a.path.name.clone(),
409 description: None,
410 is_private: false,
411 })
412 .await?;
413 match created {
414 Outcome::Ok(repo) => repo,
415 Outcome::Fail(failure) => return Ok(Outcome::Fail(failure)),
416 }
417 }
418 };
419 let git = self.store.open(&store_key(&repo)).await?;
420 let scope = if write { Scope::Write } else { Scope::Read };
421 Ok(Outcome::Ok(git.access(scope).await?))
422 }
423
424 async fn land(&self, a: LandArgs) -> Result<Outcome<Landed>> {
425 let actor: Viewer = Some(a.actor.clone());
426 let Some(fork) = self.registry.by_id(&a.fork_id).await? else {
427 return Ok(not_found());
428 };
429 let target = match &fork.fork_of {
430 Some(id) => self.registry.by_id(id).await?,
431 None => None,
432 };
433 let Some(target) = target.filter(|repo| can_read(repo, &actor)) else {
434 return Ok(not_found());
435 };
436 if !can_write(&target, &actor) {
437 return Ok(Outcome::fail(
438 FailureCode::Forbidden,
439 "Only the repository's owner can land an attempt.",
440 ));
441 }
442 if !a.actor.verified {
443 return Ok(Outcome::fail(FailureCode::Forbidden, UNVERIFIED));
444 }
445
446 let branch = &target.default_branch;
447 let fork_git = self.store.open(&store_key(&fork)).await?;
448 let target_git = self.store.open(&store_key(&target)).await?;
449 let history = fork_git.log(branch, MAX_ANCESTRY).await?;
450 let Some(new) = history.first().map(|commit| commit.hash.clone()) else {
451 return Ok(Outcome::fail(
452 FailureCode::Conflict,
453 "This attempt has no commits to land.",
454 ));
455 };
456 let old = target_git
457 .log(branch, 1)
458 .await?
459 .into_iter()
460 .next()
461 .map(|commit| commit.hash);
462
463 if old.as_deref() == Some(new.as_str()) {
Diffs on attempts; hosted agent presented as the g1t agent464 return Ok(Outcome::Ok(Landed {
465 commit: new,
466 previous: None,
467 }));
Rust repos service with shipping; pull requests kept in the model468 }
469 // Moving the branch to a commit that does not descend from its
470 // current head would discard whatever landed in between.
471 if let Some(old) = &old
472 && !descends_from(&fork_git, &history, old).await?
473 {
474 return Ok(Outcome::fail(
475 FailureCode::Conflict,
476 format!(
477 "{branch} has moved since this attempt started. Pull {branch} into the attempt's fork, push, and land again."
478 ),
479 ));
480 }
481
482 let source_access = fork_git.access(Scope::Read).await?;
483 let target_access = target_git.access(Scope::Write).await?;
484 let pushed =
485 land::fast_forward(&source_access, &target_access, branch, old.as_deref(), &new)
486 .await?;
487 if let Err(reason) = pushed {
488 // Most often another attempt landed between the check and the push.
489 return Ok(Outcome::fail(
490 FailureCode::Conflict,
491 format!("{branch} could not be updated: {reason}"),
492 ));
493 }
494 self.publish_push(&target, &new, Some(a.actor.id)).await?;
Diffs on attempts; hosted agent presented as the g1t agent495 Ok(Outcome::Ok(Landed {
496 commit: new,
497 previous: old,
498 }))
499 }
500
501 async fn compare(&self, a: CompareArgs) -> Result<Outcome<Comparison>> {
502 let Some(repo) = self
503 .registry
504 .by_id(&a.repo_id)
505 .await?
506 .filter(|repo| can_read(repo, &a.viewer))
507 else {
508 return Ok(not_found());
509 };
510 let git = self.store.open(&store_key(&repo)).await?;
511 let history = git.log(&repo.default_branch, MAX_ANCESTRY).await?;
512 let Some(head) = history.first() else {
513 return Ok(Outcome::fail(
514 FailureCode::Conflict,
515 "This repository has no commits yet.",
516 ));
517 };
518
519 let base = match (a.base, &repo.fork_of) {
520 (Some(base), _) => Some(base),
521 // A fork is compared with the last commit it shares with the
522 // repository it came from.
523 (None, Some(target_id)) => match self.registry.by_id(target_id).await? {
524 Some(target) => {
525 let target_git = self.store.open(&store_key(&target)).await?;
526 let shared: HashSet<String> = target_git
527 .log(&target.default_branch, MAX_ANCESTRY)
528 .await?
529 .into_iter()
530 .map(|commit| commit.hash)
531 .collect();
532 nearest_ancestor_in(&git, &history, &shared).await?
533 }
534 None => None,
535 },
536 (None, None) => head.parents.first().cloned(),
537 };
538 let base_tree = match &base {
539 Some(base) => git
540 .log(base, 1)
541 .await?
542 .into_iter()
543 .next()
544 .map(|commit| commit.tree_hash),
545 None => None,
546 };
547 let (files, truncated) =
548 diff::compare_trees(&git, base_tree.as_deref(), &head.tree_hash).await?;
549 Ok(Outcome::Ok(Comparison {
550 base,
551 head: head.hash.clone(),
552 files,
553 truncated,
554 }))
Rust repos service with shipping; pull requests kept in the model555 }
556
557 async fn publish_push(&self, repo: &Repo, after: &str, actor: Option<String>) -> Result<()> {
558 self.publish(NewEvent {
559 kind: "git.push",
560 source: SOURCE,
561 repo_id: Some(repo.id.clone()),
562 actor,
563 data: GitPush {
564 repo_id: repo.id.clone(),
565 git_ref: format!("refs/heads/{}", repo.default_branch),
566 after: after.to_owned(),
567 },
568 })
569 .await
570 }
571
572 /// Git over HTTPS.
573 async fn git_http(&self, request: Request, env: &Env) -> Result<Response> {
574 let Some(git) = git_http::parse(&request.url()?) else {
575 return Response::error("Not found", 404);
576 };
577 let viewer = git_http::viewer(&request, &env.service("IDENTITY")?).await?;
578 let access = self
579 .git_access(GitAccessArgs {
580 path: git.path.clone(),
581 viewer: viewer.clone(),
582 service: git.service,
583 })
584 .await?;
585 let access = match access {
586 Outcome::Ok(access) => access,
587 refused => return git_http::refuse(refused),
588 };
589 let response = git_http::forward(request, &git, &access).await?;
590
591 // Artifacts' own push notifications are per repository, which does
592 // not fit a repo per attempt, so the front end reports pushes itself.
593 let pushed = git.endpoint == "git-receive-pack" && response.status_code() == 200;
594 if pushed && let Some(repo) = self.registry.by_path(&git.path).await? {
595 let head = self
596 .store
597 .open(&store_key(&repo))
598 .await?
599 .log(&repo.default_branch, 1)
600 .await?;
601 if let Some(head) = head.first() {
602 self.publish_push(&repo, &head.hash, viewer.map(|user: User| user.id))
603 .await?;
604 }
605 }
606 Ok(response)
607 }
608}
609
610#[event(fetch)]
611async fn fetch(mut request: Request, env: Env, _ctx: Context) -> Result<Response> {
612 let repos = Repos {
613 registry: Registry { db: env.d1("DB")? },
614 store: ArtifactsStore::new(&env)?,
615 events: js::binding(&env, "EVENTS")?,
616 };
617 let Some(method) = rpc_method(&request) else {
618 return repos.git_http(request, &env).await;
619 };
620 let body: serde_json::Value = request.json().await?;
621
622 match method.as_str() {
623 "get" => reply(&repos.get(args(body)?).await?),
624 "get_by_id" => reply(&repos.get_by_id(args(body)?).await?),
625 "list" => {
626 let a: ListArgs = args(body)?;
627 reply(
628 &repos
629 .registry
630 .list(&a.viewer, a.query.as_deref(), a.namespace.as_deref())
631 .await?,
632 )
633 }
634 "create" => reply(&repos.create(args(body)?).await?),
635 "tree" => reply(&repos.tree(args(body)?).await?),
636 "blob" => reply(&repos.blob(args(body)?).await?),
637 "log" => reply(&repos.log(args(body)?).await?),
638 "fork_for_attempt" => reply(&repos.fork_for_attempt(args(body)?).await?),
639 "git_access" => reply(&repos.git_access(args(body)?).await?),
640 "land" => reply(&repos.land(args(body)?).await?),
Diffs on attempts; hosted agent presented as the g1t agent641 "compare" => reply(&repos.compare(args(body)?).await?),
Rust repos service with shipping; pull requests kept in the model642 _ => Response::error("Unknown method", 404),
643 }
644}