Skip to content
Sign up

flagon-io/g1t

public
00

Where people and agents ship software together. The open-source git platform for the whole job: issues, agents, checks and deploys to the edge.

g1t.sh
gitagentscloudflarerustforgemcp
3,525 linesCodeBlameRaw
1//! 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
8mod about;
9mod archive_links;
10mod backups;
11mod blame;
12mod catch_up;
13mod coalesce;
14mod commit_file;
15mod contributors;
16mod copy;
17mod diff;
18mod drift;
19mod fallback;
20mod forks;
21mod git_http;
22mod git_ops;
23mod hidden_refs;
24mod import;
25mod land;
26mod languages;
27mod last_commits;
28mod license;
29mod lifecycle;
30mod limits;
31mod listing;
32mod merge_methods;
33mod meters;
34mod moves;
35mod namespaces;
36mod network;
37mod pack_cache;
38mod pack_limits;
39mod push_checks;
40mod push_feedback;
41mod push_commits;
42mod refs;
43mod refs_cache;
44mod registry;
45mod resilience;
46mod rule_facts;
47mod rules;
48mod run_access;
49mod secret_scan;
50mod shards;
51mod shared;
52mod signatures;
53mod stats;
54mod store;
55mod test_merge;
56mod transfer;
57mod workflow_gate;
58
59use g1t_contracts::events::{
60 Event, GitDeleted, GitPush, NewEvent, Publish, RepoCreated, RepoForked, RepoUpdated, WorkspaceDeleted,
61 WorkspaceDeleting, WorkspaceRenamed, WorkspaceRestored,
62};
63use g1t_contracts::access::{self, Capability};
64use g1t_contracts::repos::*;
65use g1t_contracts::time::rfc3339;
66use g1t_contracts::{FailureCode, Outcome, PrincipalKind, Role, User, Viewer, is_valid_repo_name, new_id};
67use g1t_kit::{args, now_ms, reply, rpc_method};
68use std::collections::{HashMap, HashSet, VecDeque};
69use std::rc::Rc;
70
71use serde::Serialize;
72use worker::{
73 Context, Env, Fetcher, MessageBatch, MessageExt, Method, Request, Response, Result, ScheduleContext, ScheduledEvent,
74 event,
75};
76
77use registry::{Registry, can_read, can_write, store_key};
78use store::{ArtifactsStore, GitRepo, GitStore, Scope};
79
80/// Namespace that holds every pull request's fork: `pulls/<pull id>`.
81pub(crate) const PULLS_NAMESPACE: &str = "pulls";
82const MAX_TEXT_BYTES: usize = 512 * 1024;
83/// How far back a pull request may have forked and still be landed.
84const MAX_ANCESTRY: u32 = 1000;
85/// The highest limit on how many branches and tags one push may change
86/// (`UpdateArgs::max_ref_updates`).
87const MAX_REF_UPDATES_LIMIT: u32 = 1000;
88/// How many forks up a pull request's source is looked for in the fork
89/// network of the repository it merges into.
90const MAX_FORK_DEPTH: usize = 10;
91/// The most files one `compare` asked for by path reads.
92const MAX_COMPARED_PATHS: usize = 50;
93/// The most tags a repository's Tags page reads and lists.
94const MAX_TAGS_READ: usize = 100;
95/// Branch heads measured in one `branch_drift` call.
96const MAX_DRIFT_HEADS: usize = 100;
97/// How long a `last_commits` walk asked without a budget runs before it
98/// stops and keeps its progress for the next call. The site asks from a
99/// waitUntil, which may run 30 s past its response.
100const LAST_COMMITS_WALK_MS: u64 = 20_000;
101/// How long a path's history may walk when the caller gives no budget.
102const PATH_LOG_WALK_MS: u64 = 10_000;
103/// How long a blame walks before it answers with what it has: a page waits
104/// on it. A blame cut short is kept only briefly (blame.rs, `cache_control`).
105const BLAME_WALK_MS: u64 = 6_000;
106
107/// One path segment, percent-encoded for a cache key.
108fn urlencoding_segment(segment: &str) -> String {
109 segment
110 .bytes()
111 .map(|b| if b.is_ascii_alphanumeric() || b"-._~".contains(&b) { (b as char).to_string() } else { format!("%{b:02X}") })
112 .collect()
113}
114const MAX_DESCRIPTION_CHARS: usize = 200;
115pub(crate) const SOURCE: &str = "repos";
116pub(crate) const UNVERIFIED: &str = "Confirm your email address first. Check your inbox, or resend the link from the banner on g1t.sh.";
117
118pub(crate) fn not_found<T>() -> Outcome<T> {
119 Outcome::fail(FailureCode::NotFound, "Repository not found.")
120}
121
122/// Decoded text, or `None` when the file is too large or looks binary.
123/// Whether a ref is a full commit hash rather than a branch name.
124fn is_commit_hash(git_ref: &str) -> bool {
125 git_ref.len() == 40 && git_ref.bytes().all(|b| b.is_ascii_hexdigit())
126}
127
128fn text_of(bytes: Vec<u8>) -> Option<String> {
129 if bytes.len() > MAX_TEXT_BYTES || bytes.contains(&0) {
130 return None;
131 }
132 Some(String::from_utf8_lossy(&bytes).into_owned())
133}
134
135fn is_readme(name: &str) -> bool {
136 matches!(
137 name.to_lowercase().as_str(),
138 "readme" | "readme.md" | "readme.markdown" | "readme.txt"
139 )
140}
141
142/// Whether `ancestor` is reachable from the newest commit in `history`.
143///
144/// `history` is the first-parent chain, which is all the store lists; a fork
145/// that merged the target branch in has the target's head on a second
146/// parent, so the walk follows every parent.
147async fn descends_from<R: GitRepo>(repo: &R, history: &[Commit], ancestor: &str) -> Result<bool> {
148 let known: HashMap<&str, &[String]> = history
149 .iter()
150 .map(|commit| (commit.hash.as_str(), commit.parents.as_slice()))
151 .collect();
152 let mut seen = HashSet::new();
153 let mut queue: Vec<String> = history
154 .first()
155 .map(|c| c.hash.clone())
156 .into_iter()
157 .collect();
158 while let Some(hash) = queue.pop() {
159 if hash == ancestor {
160 return Ok(true);
161 }
162 if !seen.insert(hash.clone()) || seen.len() > MAX_ANCESTRY as usize {
163 continue;
164 }
165 match known.get(hash.as_str()) {
166 Some(parents) => queue.extend(parents.iter().cloned()),
167 None => queue.extend(repo.parents(&hash).await?.unwrap_or_default()),
168 }
169 }
170 Ok(false)
171}
172
173/// The commit closest to the newest in `history` that is also in `shared`:
174/// where a fork and the repository it came from last agreed.
175async fn nearest_ancestor_in<R: GitRepo>(
176 repo: &R,
177 history: &[Commit],
178 shared: &HashSet<String>,
179) -> Result<Option<String>> {
180 let known: HashMap<&str, &[String]> = history
181 .iter()
182 .map(|commit| (commit.hash.as_str(), commit.parents.as_slice()))
183 .collect();
184 let mut seen = HashSet::new();
185 let mut queue: VecDeque<String> = history
186 .first()
187 .map(|c| c.hash.clone())
188 .into_iter()
189 .collect();
190 while let Some(hash) = queue.pop_front() {
191 if shared.contains(&hash) {
192 return Ok(Some(hash));
193 }
194 if !seen.insert(hash.clone()) || seen.len() > MAX_ANCESTRY as usize {
195 continue;
196 }
197 match known.get(hash.as_str()) {
198 Some(parents) => queue.extend(parents.iter().cloned()),
199 None => queue.extend(repo.parents(&hash).await?.unwrap_or_default()),
200 }
201 }
202 Ok(None)
203}
204
205thread_local! {
206 /// Targets' sides of mergeability, by head (coalesce.rs).
207 static TARGETS: std::cell::RefCell<coalesce::Memo<coalesce::TargetKey, Rc<coalesce::TargetSide>>> =
208 std::cell::RefCell::new(coalesce::Memo::new(coalesce::TARGET_TTL_MS, 32));
209 /// What targets changed between two trees.
210 static THEIRS: std::cell::RefCell<coalesce::Memo<coalesce::TheirsKey, (Vec<String>, bool)>> =
211 std::cell::RefCell::new(coalesce::Memo::new(coalesce::THEIRS_TTL_MS, 256));
212 /// What repositories hold, as read for a push's first request, for the
213 /// same push's second: a push's POST does not wait on the database.
214 static HELD: std::cell::RefCell<coalesce::Memo<String, u64>> =
215 std::cell::RefCell::new(coalesce::Memo::new(60_000, 512));
216}
217
218pub(crate) struct Repos<S: GitStore> {
219 registry: Registry,
220 store: S,
221 events: Fetcher,
222 /// Asked during a push which secrets have been allowed.
223 security: Option<Fetcher>,
224 /// Asked whether a workspace is on a plan, for its private storage.
225 billing: Option<Fetcher>,
226 /// Says which rulesets hold for a change to a branch or tag, and keeps
227 /// how they judged it (rules.rs). `None` where it is not deployed: the
228 /// old protection flag then holds on push.
229 work: Option<Fetcher>,
230 /// Told when a repository moves, for the tokens of agents at work on it.
231 identity: Option<Fetcher>,
232 /// What a free workspace's private repositories may hold.
233 free_private_bytes: i64,
234 /// Days a pull request's working copy is kept after it settles (forks.rs).
235 pub(crate) fork_days: u64,
236 /// The most a repository may hold (pack_limits.rs), and what happens
237 /// to a push too large to scan.
238 repo_limit: u64,
239 large_pushes: git_http::LargePushes,
240 /// Which git store namespace new repositories go in (shards.rs), and
241 /// the most each should hold (`GITSTORE_NAMESPACE_LIMITS`).
242 placement: shards::Placement,
243 limits: HashMap<String, u64>,
244 /// What isolates share: answers that list refs (refs_cache.rs).
245 shared: Option<Rc<shared::Shared>>,
246 /// Packs for fresh clones (pack_cache.rs); `None` without the bucket.
247 packs: Option<Rc<pack_cache::Packs>>,
248 /// What this request started that need not hold up its answer
249 /// (store.rs), handed to its `waitUntil` once it has answered.
250 deferred: Rc<store::Deferred>,
251}
252
253impl<S: GitStore> Repos<S> {
254 /// Records that the refs of the repository with this id changed, once
255 /// they have, so that the answers kept that list them go stale (see
256 /// refs_cache.rs). Everything that changes a repository's refs calls
257 /// this after it (`every_ref_writer_records_the_change` checks). A
258 /// failure is logged: the change itself happened, and what was kept
259 /// expires within `refs_cache::TTL_SECONDS` regardless.
260 pub(crate) async fn refs_moved(&self, repo_id: &str) {
261 if let Err(error) = self.registry.refs_moved(repo_id).await {
262 worker::console_error!("refs of {repo_id} changed but not recorded: {error}");
263 }
264 }
265
266 pub(crate) async fn publish<T: Serialize>(&self, event: NewEvent<T>) -> Result<()> {
267 g1t_kit::call(
268 &self.events,
269 "publish",
270 &Publish {
271 events: vec![event],
272 },
273 )
274 .await
275 }
276
277 /// Whether the viewer may read `repo`. A pull request's fork of a
278 /// private repository can be read by everyone who can read that
279 /// repository, so its members can review and check out the change, as
280 /// well as by whoever opened the pull request.
281 async fn may_read(&self, repo: &Repo, viewer: &Viewer) -> Result<bool> {
282 if can_read(repo, viewer) {
283 return Ok(true);
284 }
285 let Some(source_id) = &repo.fork_of else {
286 return Ok(false);
287 };
288 Ok(self
289 .registry
290 .by_id(source_id)
291 .await?
292 .is_some_and(|source| can_read(&source, viewer)))
293 }
294
295 /// `repo`, if there is one and the viewer may read it.
296 async fn visible(&self, repo: Option<Repo>, viewer: &Viewer) -> Result<Option<Repo>> {
297 Ok(match repo {
298 Some(repo) if self.may_read(&repo, viewer).await? => Some(repo),
299 _ => None,
300 })
301 }
302
303 /// Resolves a repo the viewer may read; private repos look missing.
304 pub(crate) async fn readable(&self, path: &RepoPath, viewer: &Viewer) -> Result<Option<Repo>> {
305 self.visible(self.registry.by_path(path).await?, viewer)
306 .await
307 }
308
309 async fn get(&self, a: GetArgs) -> Result<Outcome<Repo>> {
310 Ok(match self.readable(&a.path, &a.viewer).await? {
311 Some(repo) => Outcome::Ok(self.with_parent(repo, &a.viewer).await?),
312 None => not_found(),
313 })
314 }
315
316 async fn get_by_id(&self, a: GetByIdArgs) -> Result<Outcome<Repo>> {
317 Ok(match self.visible(self.registry.by_id(&a.id).await?, &a.viewer).await? {
318 Some(repo) => Outcome::Ok(self.with_parent(repo, &a.viewer).await?),
319 None => not_found(),
320 })
321 }
322
323 async fn update(&self, a: UpdateArgs) -> Result<Outcome<Repo>> {
324 let viewer = Some(a.actor.clone());
325 let Some(repo) = self.readable(&a.path, &viewer).await? else {
326 return Ok(not_found());
327 };
328 // Its details take Maintain; its protection, Admin; who can see it,
329 // Admin and the member privileges (below). See g1t_contracts::access.
330 let protection_changes = a.protected.is_some_and(|protected| protected != repo.protected);
331 let forking_changes = a.allow_forking.is_some_and(|allow| allow != repo.allow_forking);
332 // 0 clears the limit.
333 let max_ref_updates = match a.max_ref_updates {
334 Some(0) => None,
335 Some(limit) if limit > MAX_REF_UPDATES_LIMIT => {
336 return Ok(Outcome::fail(
337 FailureCode::Invalid,
338 format!("A push can be limited to at most {MAX_REF_UPDATES_LIMIT} branches and tags; use 0 for no limit."),
339 ));
340 }
341 Some(limit) => Some(limit),
342 None => repo.max_ref_updates,
343 };
344 let limit_changes = max_ref_updates != repo.max_ref_updates;
345 let details_change = a.description.is_some() || a.website.is_some() || a.topics.is_some();
346 let mut needed = Vec::new();
347 if details_change || !(protection_changes || forking_changes || limit_changes) {
348 needed.push(Capability::ManageSettings);
349 }
350 if protection_changes {
351 needed.push(Capability::ManageProtection);
352 }
353 // Who may copy a private repository elsewhere is an admin's to say.
354 // So is how much one push may change.
355 if forking_changes || limit_changes {
356 needed.push(Capability::Administer);
357 }
358 let full_name = format!("{}/{}", repo.namespace, repo.name);
359 if repo.fork_of.is_some() {
360 return Ok(Outcome::fail(FailureCode::Forbidden, access::needs(Capability::ManageSettings, &full_name)));
361 }
362 if let Some(missing) = needed.into_iter().find(|capability| !registry::can(&repo, &viewer, *capability)) {
363 return Ok(Outcome::fail(FailureCode::Forbidden, access::needs(missing, &full_name)));
364 }
365 if !a.actor.verified {
366 return Ok(Outcome::fail(FailureCode::Forbidden, UNVERIFIED));
367 }
368 if let Some((code, message)) = lifecycle::archived_refusal(&repo) {
369 return Ok(Outcome::fail(code, message));
370 }
371 let description = match a.description {
372 Some(text) => Some(
373 text.trim()
374 .chars()
375 .take(MAX_DESCRIPTION_CHARS)
376 .collect::<String>(),
377 )
378 .filter(|text| !text.is_empty()),
379 None => repo.description.clone(),
380 };
381 let website = match a.website.as_deref() {
382 Some(text) => match clean_website(text) {
383 Ok(website) => website,
384 Err(reason) => return Ok(Outcome::fail(FailureCode::Invalid, reason)),
385 },
386 None => repo.website.clone(),
387 };
388 // Who can see it is an owner's to change, and a free workspace's
389 // storage may not take it private: see lifecycle.rs.
390 let wants_private = a.is_private.filter(|private| *private != repo.is_private);
391 if wants_private.is_some()
392 && let Err((code, message)) = lifecycle::admin_only(
393 lifecycle::Asker::on(&a.actor, &repo),
394 &repo.namespace,
395 "change the visibility of",
396 Capability::ChangeVisibility,
397 )
398 {
399 return Ok(Outcome::fail(code, message));
400 }
401 if let Some(private) = wants_private
402 && let Some(why) = lifecycle::visibility_refusal(&a.actor, &repo, private)
403 {
404 return Ok(Outcome::fail(FailureCode::Forbidden, why));
405 }
406 let is_private = repo.is_private;
407 let protected = a.protected.unwrap_or(repo.protected);
408 let topics = match &a.topics {
409 Some(topics) => match clean_topics(topics) {
410 Ok(topics) => topics,
411 Err(reason) => return Ok(Outcome::fail(FailureCode::Invalid, reason)),
412 },
413 None => repo.topics.clone(),
414 };
415 let allow_forking = a.allow_forking.unwrap_or(repo.allow_forking);
416 self.registry
417 .update(&repo.id, description.as_deref(), protected, &topics, website.as_deref(), allow_forking, max_ref_updates)
418 .await?;
419 let updated = Repo {
420 description,
421 is_private,
422 protected,
423 topics,
424 website,
425 allow_forking,
426 max_ref_updates,
427 ..repo
428 };
429 // Whether the default branch takes only pull requests is now its
430 // branch protection ruleset's to say (work's rulesets.rs).
431 if let (Some(protected), Some(work)) = (a.protected, &self.work) {
432 #[derive(Serialize)]
433 struct RequirePullRequest<'a> {
434 repo: &'a Repo,
435 protected: bool,
436 actor: &'a User,
437 }
438 let set: Result<Outcome<bool>> =
439 g1t_kit::call(work, "set_requires_pull_request", &RequirePullRequest { repo: &updated, protected, actor: &a.actor }).await;
440 match set {
441 Ok(Outcome::Ok(_)) => {}
442 Ok(Outcome::Fail(failure)) => return Ok(Outcome::Fail(failure)),
443 Err(error) => return Err(error),
444 }
445 }
446 if let Some(private) = wants_private {
447 return self.change_visibility(updated, private, &a.actor, a.surface).await;
448 }
449 let visibility_changed = false;
450 // Search and anything else that shows the repository hears of it;
451 // a change of visibility is announced on its own as well, so that
452 // what was public stops being shown at once.
453 self.publish(NewEvent {
454 kind: "repo.updated",
455 source: SOURCE,
456 repo_id: Some(updated.id.clone()),
457 actor: Some(a.actor.id.clone()),
458 data: RepoUpdated {
459 repo_id: updated.id.clone(),
460 namespace: updated.namespace.clone(),
461 name: updated.name.clone(),
462 is_private,
463 visibility_changed,
464 },
465 })
466 .await?;
467 Ok(Outcome::Ok(updated))
468 }
469
470 /// The repository with this id, if it is not a fork, and its store.
471 async fn stored(&self, repo_id: &str) -> Result<Option<S::Repo>> {
472 match self.registry.by_id(repo_id).await? {
473 Some(repo) if repo.fork_of.is_none() => Ok(Some(self.store.open(&store_key(&repo)).await?)),
474 _ => Ok(None),
475 }
476 }
477
478 async fn list_files(&self, a: ListFilesArgs) -> Result<FileList> {
479 let Some(repo) = self.registry.by_id(&a.repo_id).await?.filter(|repo| repo.fork_of.is_none()) else {
480 return Ok(FileList::default());
481 };
482 let git = self.store.open(&store_key(&repo)).await?;
483 let head = a.git_ref.unwrap_or_else(|| repo.default_branch.clone());
484 listing::list(&git, None, &head, &a.skip_dirs, a.limit, a.links).await
485 }
486
487 /// A link to download a branch, tag or commit as an archive, good for
488 /// a few minutes with no session or token (archive_links.rs).
489 async fn archive_link(&self, a: ArchiveLinkArgs, sealer: Option<&g1t_secrets::Sealer>) -> Result<Outcome<ArchiveLink>> {
490 let Some(repo) = self.readable(&a.path, &a.viewer).await? else {
491 return Ok(not_found());
492 };
493 let Some(sealer) = sealer else {
494 return Ok(Outcome::fail(FailureCode::Invalid, "Archive downloads are not set up on this installation."));
495 };
496 let git = self.read_git(&repo).await?;
497 let git_ref = a.git_ref.filter(|git_ref| !git_ref.is_empty()).unwrap_or_else(|| repo.default_branch.clone());
498 let Some(head) = git.log(&git_ref, 1).await?.into_iter().next() else {
499 return Ok(Outcome::fail(FailureCode::NotFound, "No such branch, tag or commit."));
500 };
501 let (token, expires) = archive_links::seal(sealer, &repo.id, &head.hash, now_ms());
502 Ok(Outcome::Ok(ArchiveLink { repo, commit: head.hash, token, expires_at: rfc3339(expires) }))
503 }
504
505 /// What an archive link's token lets its holder download.
506 async fn open_archive_token(&self, a: OpenArchiveTokenArgs, sealer: Option<&g1t_secrets::Sealer>) -> Result<Option<ArchiveGrant>> {
507 let Some(sealer) = sealer else {
508 return Ok(None);
509 };
510 let Some(repo) = self.registry.by_path(&a.path).await? else {
511 return Ok(None);
512 };
513 let Some(commit) = archive_links::open(sealer, &repo.id, &a.token, now_ms()) else {
514 return Ok(None);
515 };
516 Ok(Some(ArchiveGrant { repo_id: repo.id, namespace: repo.namespace, name: repo.name, commit }))
517 }
518
519 async fn changed_files(&self, a: ChangedFilesArgs) -> Result<FileList> {
520 let Some(git) = self.stored(&a.repo_id).await? else {
521 return Ok(FileList::default());
522 };
523 listing::list(&git, a.base.as_deref(), &a.head, &a.skip_dirs, a.limit, false).await
524 }
525
526 /// Branches and tags with their commits, for g1t's own services.
527 async fn refs_of(&self, a: RefsArgs) -> Result<Option<RepoRefs>> {
528 let Some(repo) = self.registry.by_id(&a.repo_id).await?.filter(|repo| repo.fork_of.is_none()) else {
529 return Ok(None);
530 };
531 let git = self.store.open(&store_key(&repo)).await?;
532 let access = git.access(Scope::Read).await?;
533 let refs = refs::heads_and_tags(refs::all(&access).await?)
534 .into_iter()
535 .map(|(name, commit)| GitRefEntry { name, commit })
536 .collect();
537 Ok(Some(RepoRefs { repo, refs }))
538 }
539
540 async fn raw_file(&self, a: RawFileArgs) -> Result<Option<RawFile>> {
541 use base64::Engine;
542 let Some(git) = self.stored(&a.repo_id).await? else {
543 return Ok(None);
544 };
545 Ok(git
546 .read_file(&a.git_ref, &a.path)
547 .await?
548 .filter(|bytes| bytes.len() <= a.max_bytes as usize)
549 .map(|bytes| RawFile { size: bytes.len() as u64, data: base64::engine::general_purpose::STANDARD.encode(bytes) }))
550 }
551
552 async fn raw_blobs(&self, a: RawBlobsArgs) -> Result<Vec<RawBlob>> {
553 use base64::Engine;
554 let Some(git) = self.stored(&a.repo_id).await? else {
555 return Ok(Vec::new());
556 };
557 let hashes: Vec<&String> = a.hashes.iter().take(MAX_READ_BLOBS).collect();
558 let mut out = Vec::with_capacity(hashes.len());
559 // A few at a time, as listing::read does: each is a round trip.
560 for group in hashes.chunks(8) {
561 let read = futures_util::future::try_join_all(group.iter().map(|hash| git.read_blob(hash))).await?;
562 for (hash, bytes) in group.iter().zip(read) {
563 let size = bytes.as_ref().map_or(0, |bytes| bytes.len() as u64);
564 let data = bytes
565 .filter(|bytes| bytes.len() <= a.max_bytes as usize)
566 .map(|bytes| base64::engine::general_purpose::STANDARD.encode(bytes));
567 out.push(RawBlob { hash: (*hash).clone(), size, data });
568 }
569 }
570 Ok(out)
571 }
572
573 async fn read_blobs(&self, a: ReadBlobsArgs) -> Result<Vec<BlobText>> {
574 let Some(git) = self.stored(&a.repo_id).await? else {
575 return Ok(Vec::new());
576 };
577 listing::read(&git, &a.hashes, a.max_bytes.min(MAX_TEXT_BYTES as u32)).await
578 }
579
580 async fn create(&self, a: CreateArgs) -> Result<Outcome<Repo>> {
581 if !a.owner.verified {
582 return Ok(Outcome::fail(FailureCode::Forbidden, UNVERIFIED));
583 }
584 let name = a.name.trim().to_lowercase();
585 if !is_valid_repo_name(&name) {
586 return Ok(Outcome::fail(
587 FailureCode::Invalid,
588 "Use letters, digits, dots, hyphens and underscores only.",
589 ));
590 }
591 let namespace = a.namespace.trim().to_lowercase();
592 if namespace.is_empty() {
593 return Ok(Outcome::fail(
594 FailureCode::Invalid,
595 "Say which workspace to create the repository in.",
596 ));
597 }
598 let Some(role) = a.owner.role_in(&namespace) else {
599 return Ok(Outcome::fail(
600 FailureCode::Forbidden,
601 "You are not a member of that workspace.",
602 ));
603 };
604 // Who may create which: the workspace's member privileges. A
605 // workspace's own token acts as an owner would.
606 let role = if a.owner.kind == PrincipalKind::Workspace { Role::Owner } else { role };
607 if let Some(why) = a.owner.privileges_in(&namespace).creation_refusal(role, a.is_private, &namespace) {
608 return Ok(Outcome::fail(FailureCode::Forbidden, why));
609 }
610 let path = RepoPath { namespace, name };
611 match self.registry.by_path_any(&path).await? {
612 Some((_, None)) => {
613 return Ok(Outcome::fail(
614 FailureCode::Conflict,
615 "That workspace already has a repository with that name.",
616 ));
617 }
618 Some((_, Some(_))) => {
619 return Ok(Outcome::fail(
620 FailureCode::Conflict,
621 format!(
622 "{}/{} was deleted recently and can still be restored, so its name is taken. Restore it, or delete it permanently from the workspace's Recently deleted list.",
623 path.namespace, path.name
624 ),
625 ));
626 }
627 None => {}
628 }
629 // With a credential (a GitHub App installation's token), everything
630 // is copied: every branch and tag. See copy.rs.
631 let mut credentialed = None;
632 if let (Some(url), Some(token)) = (a.import_url.as_deref(), a.import_token.as_deref()) {
633 let Some(url) = import::clean_url(url) else {
634 return Ok(Outcome::fail(FailureCode::Invalid, "That is not an https repository address."));
635 };
636 let source = copy::Endpoint::github(&url, token);
637 match copy::probe(&source).await? {
638 Ok(advertised) => credentialed = Some((source, advertised)),
639 Err(reason) => return Ok(Outcome::fail(FailureCode::Invalid, reason)),
640 }
641 }
642 // An import is fetched before anything is created, so that an
643 // address that does not work leaves nothing behind.
644 let mut imported = None;
645 if let Some(url) = a
646 .import_url
647 .as_deref()
648 .map(str::trim)
649 .filter(|url| !url.is_empty() && credentialed.is_none())
650 {
651 let Some(url) = import::clean_url(url) else {
652 return Ok(Outcome::fail(
653 FailureCode::Invalid,
654 "Give the https address of a public repository, such as https://github.com/owner/repo.",
655 ));
656 };
657 let remote = match import::discover(&url).await? {
658 Ok(remote) => remote,
659 Err(reason) => return Ok(Outcome::fail(FailureCode::Invalid, reason)),
660 };
661 imported = Some((remote, url));
662 }
663 let now = now_ms();
664 let repo = Repo {
665 id: new_id("rep", now),
666 namespace: path.namespace,
667 name: path.name,
668 description: a
669 .description
670 .map(|text| text.trim().to_owned())
671 .filter(|text| !text.is_empty()),
672 is_private: a.is_private,
673 owner_id: a.owner.id.clone(),
674 default_branch: imported
675 .as_ref()
676 .map(|(remote, _)| remote.branch.clone())
677 .or_else(|| credentialed.as_ref().and_then(|(_, advertised)| advertised.default_branch()))
678 .unwrap_or_else(|| "main".to_owned()),
679 fork_of: None,
680 parent_id: None,
681 forked_from: None,
682 parent: None,
683 allow_forking: false,
684 max_ref_updates: None,
685 protected: false,
686 created_at: rfc3339(now),
687 topics: Vec::new(),
688 website: None,
689 archived_at: None,
690 };
691 let namespace = match self.place(&repo).await? {
692 Ok(namespace) => namespace,
693 Err(unplaced) => return Ok(Outcome::fail(FailureCode::Conflict, unplaced.message())),
694 };
695 self.registry
696 .claim_store_key(&repo, namespace.as_deref(), &self.store.default_namespace())
697 .await?;
698 self.store
699 .create(
700 &store_key(&repo),
701 repo.description.as_deref(),
702 &repo.default_branch,
703 )
704 .await?;
705 self.registry.insert(&repo).await?;
706 // A repository that was transferred away from this path stops
707 // redirecting here.
708 self.registry
709 .drop_redirect(&RepoPath {
710 namespace: repo.namespace.clone(),
711 name: repo.name.clone(),
712 })
713 .await?;
714 // Every branch and tag the import made, announced as pushes.
715 let mut pushed: Vec<(String, String)> = Vec::new();
716 // A public repository, read with no credential: every branch and
717 // tag is copied too, the default branch the one its HEAD names.
718 if let Some((_, url)) = imported {
719 let access = self
720 .store
721 .open(&store_key(&repo))
722 .await?
723 .access(Scope::Write)
724 .await?;
725 let target = copy::Endpoint::bearer(&access.remote, &access.token);
726 let copied = copy::copy(&copy::Endpoint::anonymous(&url), &target).await?;
727 self.refs_moved(&repo.id).await;
728 match copied {
729 Ok(copied) => pushed = copy::import_pushes(&copied.updated, &repo.default_branch),
730 Err(reason) => {
731 self.registry.remove(&repo.id).await?;
732 return Ok(Outcome::fail(
733 FailureCode::Invalid,
734 format!("The repository could not be stored: {reason}"),
735 ));
736 }
737 }
738 }
739 if let Some((source, _)) = credentialed {
740 let access = self
741 .store
742 .open(&store_key(&repo))
743 .await?
744 .access(Scope::Write)
745 .await?;
746 let target = copy::Endpoint::bearer(&access.remote, &access.token);
747 let copied = copy::copy(&source, &target).await?;
748 self.refs_moved(&repo.id).await;
749 match copied {
750 Ok(copied) => pushed = copy::import_pushes(&copied.updated, &repo.default_branch),
751 Err(reason) => {
752 self.registry.remove(&repo.id).await?;
753 return Ok(Outcome::fail(
754 FailureCode::Invalid,
755 format!("The repository could not be copied: {reason}"),
756 ));
757 }
758 }
759 }
760 // Whoever creates a repository is an Admin of it, as a role given
761 // to them on it, whatever the workspace's base permission.
762 if a.owner.kind == PrincipalKind::User {
763 self.grant_creator(&repo, &a.owner.id).await;
764 }
765 self.publish(NewEvent {
766 kind: "repo.created",
767 source: SOURCE,
768 repo_id: Some(repo.id.clone()),
769 actor: Some(a.owner.id),
770 data: RepoCreated {
771 repo_id: repo.id.clone(),
772 namespace: repo.namespace.clone(),
773 name: repo.name.clone(),
774 is_private: repo.is_private,
775 parent_id: None,
776 },
777 })
778 .await?;
779 for (git_ref, head) in &pushed {
780 self.publish_push(&repo, git_ref, None, head, None, None).await?;
781 }
782 Ok(Outcome::Ok(repo))
783 }
784
785 /// Makes `user_id` an Admin of `repo`, the repository they created.
786 pub(crate) async fn grant_creator(&self, repo: &Repo, user_id: &str) {
787 let Some(identity) = &self.identity else { return };
788 let granted: Result<bool> = g1t_kit::call(
789 identity,
790 "grant_creator",
791 &g1t_contracts::members::GrantCreatorArgs {
792 repo_id: repo.id.clone(),
793 namespace: repo.namespace.clone(),
794 name: repo.name.clone(),
795 user_id: user_id.to_owned(),
796 },
797 )
798 .await;
799 if let Err(error) = granted {
800 worker::console_error!("creator of {} not given Admin: {error}", repo.id);
801 }
802 }
803
804 /// Where a workspace keeps its data, asked of identity only when an EU
805 /// namespace is configured: without one, every workspace's
806 /// repositories go anywhere and identity is never asked.
807 async fn residency_of(&self, workspace: &str) -> Result<shards::Residency> {
808 if self.placement.eu.is_none() {
809 return Ok(shards::Residency::Anywhere);
810 }
811 let Some(identity) = &self.identity else {
812 return Ok(shards::Residency::Anywhere);
813 };
814 let residency: Option<g1t_contracts::identity::DataResidency> = g1t_kit::call(
815 identity,
816 "workspace_residency",
817 &g1t_contracts::identity::SlugArgs { slug: workspace.to_owned() },
818 )
819 .await?;
820 Ok(match residency {
821 Some(g1t_contracts::identity::DataResidency::Eu) => shards::Residency::Eu,
822 _ => shards::Residency::Anywhere,
823 })
824 }
825
826 /// How each bound namespace stands (namespaces.rs).
827 async fn standings(&self) -> Result<Vec<namespaces::Standing>> {
828 let bound = self.store.namespaces();
829 let default = self.store.default_namespace();
830 let now = now_ms();
831 let config = namespaces::Configured {
832 bound: &bound,
833 default: &default,
834 placement: &self.placement,
835 limits: &self.limits,
836 on_fallback: &|namespace| self.store.on_fallback(&shards::compose(Some(namespace), "x", &default)),
837 writable: &|namespace| self.store.writable(namespace),
838 breaker_open: &|namespace| resilience::open_now(namespace, now),
839 };
840 let (held, recent) = futures_util::future::join(namespaces::held(&self.registry.db), namespaces::recent(&self.registry.db, now)).await;
841 Ok(namespaces::standings(&config, &held?, &recent?))
842 }
843
844 /// The namespace a new repository goes in (shards.rs): its workspace's
845 /// residency, then how each namespace stands, read only when there is
846 /// a choice to make. `Ok(None)` for the default.
847 async fn place(&self, repo: &Repo) -> Result<std::result::Result<Option<String>, shards::Unplaced>> {
848 let residency = self.residency_of(&repo.namespace).await?;
849 let bound = self.store.namespaces();
850 let loads = if residency == shards::Residency::Anywhere && !self.placement.needs_loads(&bound) {
851 // One namespace to choose from at most: nothing to read.
852 bound
853 .iter()
854 .map(|namespace| shards::Load {
855 namespace: namespace.clone(),
856 bound: true,
857 writable: self.store.writable(namespace),
858 ..shards::Load::default()
859 })
860 .collect()
861 } else {
862 let default = self.store.default_namespace();
863 let now = now_ms();
864 let config = namespaces::Configured {
865 bound: &bound,
866 default: &default,
867 placement: &self.placement,
868 limits: &self.limits,
869 on_fallback: &|namespace| self.store.on_fallback(&shards::compose(Some(namespace), "x", &default)),
870 writable: &|namespace| self.store.writable(namespace),
871 breaker_open: &|namespace| resilience::open_now(namespace, now),
872 };
873 namespaces::loads(&self.registry.db, &config, now).await?
874 };
875 Ok(self.placement.choose(&repo.id, residency, &loads))
876 }
877
878 /// `storage_options`: what a workspace may choose about where its
879 /// repositories are kept.
880 fn storage_options(&self) -> StorageOptions {
881 let bound = self.store.namespaces();
882 StorageOptions { eu_available: self.placement.eu_available(&bound, |namespace| self.store.writable(namespace)) }
883 }
884
885 async fn tree(&self, a: TreeArgs) -> Result<Outcome<TreeView>> {
886 let Some(repo) = self.readable(&a.path, &a.viewer).await? else {
887 return Ok(not_found());
888 };
889 let git = self.read_git(&repo).await?;
890 let git_ref = a
891 .git_ref
892 .clone()
893 .unwrap_or_else(|| repo.default_branch.clone());
894
895 let Some(head) = git.log(&git_ref, 1).await?.into_iter().next() else {
896 // An unknown ref is an error; a repo with no commits is just empty.
897 if a.git_ref.is_some() {
898 return Ok(Outcome::fail(
899 FailureCode::NotFound,
900 "No such branch, tag or commit.",
901 ));
902 }
903 return Ok(Outcome::Ok(TreeView {
904 repo,
905 git_ref,
906 path: a.tree_path,
907 head: None,
908 entries: Vec::new(),
909 readme: None,
910 }));
911 };
912
913 let no_directory = || Outcome::fail(FailureCode::NotFound, "No such directory.");
914 let mut entries = git.read_tree(&head.tree_hash).await?;
915 for segment in a.tree_path.split('/').filter(|segment| !segment.is_empty()) {
916 let next = entries.as_ref().and_then(|entries| {
917 entries
918 .iter()
919 .find(|entry| entry.name == segment && entry.kind == EntryKind::Tree)
920 });
921 let Some(next) = next else {
922 return Ok(no_directory());
923 };
924 entries = git.read_tree(&next.hash).await?;
925 }
926 let Some(mut entries) = entries else {
927 return Ok(no_directory());
928 };
929 // Directories first, then by name.
930 entries.sort_by(|a, b| {
931 (b.kind == EntryKind::Tree)
932 .cmp(&(a.kind == EntryKind::Tree))
933 .then_with(|| a.name.cmp(&b.name))
934 });
935
936 let readme_entry = entries
937 .iter()
938 .find(|entry| entry.kind == EntryKind::Blob && is_readme(&entry.name));
939 let readme = match readme_entry {
940 Some(entry) => git.read_blob(&entry.hash).await?.map(|bytes| Readme {
941 name: entry.name.clone(),
942 text: text_of(bytes),
943 }),
944 None => None,
945 };
946 Ok(Outcome::Ok(TreeView {
947 repo,
948 git_ref,
949 path: a.tree_path,
950 head: Some(head),
951 entries,
952 readme,
953 }))
954 }
955
956 async fn blob(&self, a: BlobArgs) -> Result<Outcome<BlobView>> {
957 let Some(repo) = self.readable(&a.path, &a.viewer).await? else {
958 return Ok(not_found());
959 };
960 let bytes = if a.file_path.is_empty() {
961 None
962 } else {
963 let git = self.read_git(&repo).await?;
964 git.read_file(&a.git_ref, &a.file_path).await?
965 };
966 let Some(bytes) = bytes else {
967 return Ok(Outcome::fail(FailureCode::NotFound, "No such file."));
968 };
969 Ok(Outcome::Ok(BlobView {
970 repo,
971 git_ref: a.git_ref,
972 path: a.file_path,
973 size: bytes.len() as u64,
974 text: text_of(bytes),
975 }))
976 }
977
978 async fn blame(&self, a: BlameArgs) -> Result<Outcome<Blame>> {
979 let Some(repo) = self.readable(&a.path, &a.viewer).await? else {
980 return Ok(not_found());
981 };
982 let git = self.read_git(&repo).await?;
983 let git_ref = a.git_ref.unwrap_or_else(|| repo.default_branch.clone());
984 // The commit the ref names, so the answer can be kept by it: a
985 // blame of a commit and path never changes. Most blames are of a
986 // commit already (the site links files by commit, and crawlers
987 // follow those links), which needs no read.
988 let head = if store::is_commit_hash(&git_ref) {
989 git_ref
990 } else {
991 match git.log(&git_ref, 1).await?.into_iter().next() {
992 Some(head) => head.hash,
993 None => return Ok(not_found()),
994 }
995 };
996 let url = blame::cache_url(&repo.id, &head, &a.file_path);
997 if let Ok(Some(mut kept)) = worker::Cache::default().get(url.as_str(), false).await
998 && let Ok(found) = kept.json::<Blame>().await
999 {
1000 return Ok(Outcome::Ok(found));
1001 }
1002 let started = now_ms();
1003 let out_of_time = move || now_ms().saturating_sub(started) > BLAME_WALK_MS;
1004 let Some(blamed) = blame::blame(&git, &head, &a.file_path, &out_of_time, blame::MAX_READS).await? else {
1005 return Ok(not_found());
1006 };
1007 // Kept once the answer has gone.
1008 if let Ok(mut response) = worker::Response::from_json(&blamed.blame) {
1009 let _ = response.headers_mut().set("cache-control", blame::cache_control(&blamed));
1010 self.deferred.spawn(async move {
1011 let _ = worker::Cache::default().put(url.as_str(), response).await;
1012 });
1013 }
1014 Ok(Outcome::Ok(blamed.blame))
1015 }
1016
1017 async fn log(&self, a: LogArgs) -> Result<Outcome<Vec<Commit>>> {
1018 let Some(repo) = self.readable(&a.path, &a.viewer).await? else {
1019 return Ok(not_found());
1020 };
1021 let git = self.read_git(&repo).await?;
1022 let git_ref = a.git_ref.unwrap_or_else(|| repo.default_branch.clone());
1023 Ok(Outcome::Ok(git.log(&git_ref, a.limit).await?))
1024 }
1025
1026 /// The commits that changed a file or directory, newest first: the
1027 /// first-parent chain walked within `PATH_LOG_WALK_MS` (or the caller's
1028 /// budget) and `last_commits::MAX_READS`.
1029 async fn path_log(&self, a: g1t_contracts::repos::PathLogArgs) -> Result<Outcome<g1t_contracts::repos::PathLog>> {
1030 let Some(repo) = self.readable(&a.path, &a.viewer).await? else {
1031 return Ok(not_found());
1032 };
1033 let git = self.read_git(&repo).await?;
1034 let git_ref = a.git_ref.unwrap_or_else(|| repo.default_branch.clone());
1035 let Some(head) = git.log(&git_ref, 1).await?.into_iter().next() else {
1036 return Ok(Outcome::fail(FailureCode::NotFound, "No such branch, tag or commit."));
1037 };
1038 let started = worker::Date::now().as_millis();
1039 let budget = a.budget_ms.unwrap_or(PATH_LOG_WALK_MS);
1040 let out_of_time = move || worker::Date::now().as_millis().saturating_sub(started) > budget;
1041 let limit = a.limit.clamp(1, 100) as usize;
1042 let (commits, complete) = last_commits::path_log(&git, &head.hash, &a.file_path, limit, &out_of_time, last_commits::MAX_READS).await?;
1043 Ok(Outcome::Ok(g1t_contracts::repos::PathLog { commits, complete }))
1044 }
1045
1046 /// Which commit last changed each entry of a directory. A finished
1047 /// answer is kept in this colo's cache by repository, head commit and
1048 /// path: a commit's history never changes, so it is good for as long as
1049 /// it is kept. Every walk also keeps its progress by ref and path
1050 /// (last_commits.rs), so the next call goes on from it: from where it
1051 /// stopped, or for a new head, only back to the old one.
1052 async fn last_commits(&self, a: g1t_contracts::repos::LastCommitsArgs) -> Result<Outcome<g1t_contracts::repos::LastCommits>> {
1053 let Some(repo) = self.readable(&a.path, &a.viewer).await? else {
1054 return Ok(not_found());
1055 };
1056 let git = self.read_git(&repo).await?;
1057 let git_ref = a.git_ref.unwrap_or_else(|| repo.default_branch.clone());
1058 let Some(head) = git.log(&git_ref, 1).await?.into_iter().next() else {
1059 return Ok(Outcome::fail(FailureCode::NotFound, "No such branch, tag or commit."));
1060 };
1061 // v2: v1 kept walks cut short at 300 commits as if finished.
1062 let key = format!(
1063 "https://last-commits.g1t.internal/v2/{}/{}/{}",
1064 repo.id,
1065 head.hash,
1066 a.tree_path.split('/').map(urlencoding_segment).collect::<Vec<_>>().join("/")
1067 );
1068 if let Ok(Some(mut kept)) = worker::Cache::default().get(key.as_str(), false).await
1069 && let Ok(found) = kept.json::<g1t_contracts::repos::LastCommits>().await
1070 {
1071 return Ok(Outcome::Ok(found));
1072 }
1073 let memo = last_commits::Memo::new(&repo.id, &git_ref, &a.tree_path);
1074 let shared = self.shared.clone();
1075 let progress = memo.get(shared.as_deref()).await;
1076 // Asked with a budget, the walk stops past it; without one, past
1077 // LAST_COMMITS_WALK_MS, well before the caller's waitUntil ends.
1078 let started = worker::Date::now().as_millis();
1079 let budget = a.budget_ms.unwrap_or(LAST_COMMITS_WALK_MS);
1080 let out_of_time = move || worker::Date::now().as_millis().saturating_sub(started) > budget;
1081 let walk = last_commits::last_commits(&git, &head.hash, &a.tree_path, progress, &out_of_time, last_commits::MAX_READS).await?;
1082 let settled = walk.progress.settled();
1083 // What the walk found is kept once the answer has gone: the caller
1084 // waits on neither write.
1085 if walk.reads > 0 {
1086 let progress = walk.progress.clone();
1087 self.deferred.spawn(async move { memo.keep(shared.as_deref(), &progress).await });
1088 }
1089 let found = g1t_contracts::repos::LastCommits { complete: walk.progress.complete(), entries: walk.progress.found };
1090 if settled && let Ok(mut response) = worker::Response::from_json(&found) {
1091 let _ = response.headers_mut().set("cache-control", "max-age=604800");
1092 self.deferred.spawn(async move {
1093 let _ = worker::Cache::default().put(key.as_str(), response).await;
1094 });
1095 }
1096 Ok(Outcome::Ok(found))
1097 }
1098
1099 /// How far each branch head has moved from the default branch's head,
1100 /// in one call (drift.rs). Each count is kept in this colo's cache by
1101 /// repository and the pair of hashes, for good: neither history can
1102 /// change. A head that moved is the only one walked. A failed read is
1103 /// not kept, and "too far to count" only for a day.
1104 async fn branch_drift(&self, a: BranchDriftArgs) -> Result<Outcome<BranchDrifts>> {
1105 let Some(repo) = self.readable(&a.path, &a.viewer).await? else {
1106 return Ok(not_found());
1107 };
1108 if !store::is_commit_hash(&a.base) {
1109 return Ok(Outcome::fail(FailureCode::Invalid, "The default branch's head is a full commit hash."));
1110 }
1111 let heads: Vec<String> = a.heads.into_iter().take(MAX_DRIFT_HEADS).collect();
1112 let git = self.read_git(&repo).await?;
1113 // v2: answers kept before 2026-10-09 said "no count" for every
1114 // branch whose default branch took a merge since it left (drift.rs).
1115 let key = |head: &str| format!("https://drift.g1t.internal/v2/{}/{}/{head}", repo.id, a.base);
1116 let cache = worker::Cache::default();
1117 let (base, kept) = futures_util::future::join(
1118 git.log(&a.base, 1),
1119 futures_util::future::join_all(heads.iter().map(|head| {
1120 let (cache, url) = (&cache, key(head));
1121 async move {
1122 if !store::is_commit_hash(head) {
1123 return None;
1124 }
1125 let mut found = cache.get(url.as_str(), false).await.ok()??;
1126 found.json::<BranchDrift>().await.ok()
1127 }
1128 })),
1129 )
1130 .await;
1131 let missing: Vec<String> = heads
1132 .iter()
1133 .zip(&kept)
1134 .filter(|(head, kept)| kept.is_none() && store::is_commit_hash(head))
1135 .map(|(head, _)| head.clone())
1136 .collect();
1137 let measured = drift::measure(&git, &a.base, &missing).await;
1138 let mut fresh: HashMap<String, BranchDrift> = HashMap::new();
1139 for (head, found) in missing.into_iter().zip(measured) {
1140 let answer = BranchDrift { head: head.clone(), commit: found.commit, drift: found.drift };
1141 if found.settled
1142 && let Ok(mut response) = worker::Response::from_json(&answer)
1143 {
1144 // A count is kept for good; "too far to count" for a day, so
1145 // a change to how far a walk goes reaches it.
1146 let max_age = if answer.drift.is_some() { "public, max-age=31536000, immutable" } else { "public, max-age=86400" };
1147 let _ = response.headers_mut().set("cache-control", max_age);
1148 let _ = cache.put(key(&head).as_str(), response).await;
1149 }
1150 fresh.insert(head, answer);
1151 }
1152 let branches = heads
1153 .iter()
1154 .zip(kept)
1155 .map(|(head, kept)| {
1156 kept.or_else(|| fresh.get(head).cloned())
1157 .unwrap_or_else(|| BranchDrift { head: head.clone(), commit: None, drift: None })
1158 })
1159 .collect();
1160 Ok(Outcome::Ok(BranchDrifts { base: base.ok().and_then(|log| log.into_iter().next()), branches }))
1161 }
1162
1163 /// The repository's tags, newest commit first, at most 100.
1164 async fn tags(&self, a: g1t_contracts::repos::TagsArgs) -> Result<Outcome<Vec<g1t_contracts::repos::Tag>>> {
1165 let Some(repo) = self.readable(&a.path, &a.viewer).await? else {
1166 return Ok(not_found());
1167 };
1168 // Kept until the refs move, as the branch list is (store.rs): listing
1169 // the refs is a round trip to the store on every call otherwise.
1170 let version = refs_cache::usable(registry::refs_state(&repo.id), now_ms()).filter(|_| !self.store.on_fallback(&store_key(&repo)));
1171 let kept_at = version.map(|version| format!("https://tags.g1t.internal/{}/{version}", repo.id));
1172 if let Some(url) = &kept_at
1173 && let Ok(Some(mut kept)) = worker::Cache::default().get(url.as_str(), false).await
1174 && let Ok(tags) = kept.json::<Vec<g1t_contracts::repos::Tag>>().await
1175 {
1176 return Ok(Outcome::Ok(tags));
1177 }
1178 let (tags, complete) = self.read_tags(&repo).await?;
1179 if complete
1180 && let Some(url) = &kept_at
1181 && let Ok(mut response) = worker::Response::from_json(&tags)
1182 {
1183 let _ = response.headers_mut().set("cache-control", "public, max-age=300");
1184 let _ = worker::Cache::default().put(url.as_str(), response).await;
1185 }
1186 Ok(Outcome::Ok(tags))
1187 }
1188
1189 /// The tags, and whether every one's commit was read (only then kept).
1190 async fn read_tags(&self, repo: &Repo) -> Result<(Vec<g1t_contracts::repos::Tag>, bool)> {
1191 let git = self.store.open(&store_key(repo)).await?;
1192 let access = git.access(Scope::Read).await?;
1193 let named: Vec<(String, String)> = refs::heads_and_tags(refs::all(&access).await?)
1194 .into_iter()
1195 .filter_map(|(name, hash)| name.strip_prefix("refs/tags/").map(|tag| (tag.to_owned(), hash)))
1196 .collect();
1197 let read = self.read_git(repo).await?;
1198 let commits = futures_util::future::join_all(named.iter().take(MAX_TAGS_READ).map(|(_, hash)| read.log(hash, 1))).await;
1199 let complete = commits.iter().all(Result::is_ok);
1200 let mut tags: Vec<g1t_contracts::repos::Tag> = named
1201 .into_iter()
1202 .zip(commits.into_iter().map(|found| found.ok().and_then(|list| list.into_iter().next())).chain(std::iter::repeat(None)))
1203 .map(|((name, _), commit)| g1t_contracts::repos::Tag { name, commit })
1204 .collect();
1205 tags.sort_by(|a, b| {
1206 let at = |tag: &g1t_contracts::repos::Tag| tag.commit.as_ref().map(|c| c.authored_at.clone()).unwrap_or_default();
1207 at(b).cmp(&at(a)).then_with(|| b.name.cmp(&a.name))
1208 });
1209 tags.truncate(MAX_TAGS_READ);
1210 Ok((tags, complete))
1211 }
1212
1213 /// The repository's branches, default branch first.
1214 async fn branches(&self, a: BranchesArgs) -> Result<Outcome<Vec<Branch>>> {
1215 let Some(repo) = self.readable(&a.path, &a.viewer).await? else {
1216 return Ok(not_found());
1217 };
1218 let mut branches = self.read_git(&repo).await?.branches().await?;
1219 branches.sort_by_key(|branch| branch.name != repo.default_branch);
1220 Ok(Outcome::Ok(branches))
1221 }
1222
1223 /// Whether `source` is a fork of `ancestor_id`: its parent, or its
1224 /// parent's, and so on, at most [`MAX_FORK_DEPTH`] repositories up. A
1225 /// detached fork still counts as one of what it was made from
1226 /// (`forked_from`), so pull requests opened from it still merge.
1227 pub(crate) async fn forked_from(&self, source: &Repo, ancestor_id: &str) -> Result<bool> {
1228 let up = |repo: &Repo| repo.parent_id.clone().or_else(|| repo.forked_from.clone());
1229 let mut parent = up(source);
1230 for _ in 0..MAX_FORK_DEPTH {
1231 let Some(id) = parent else {
1232 return Ok(false);
1233 };
1234 if id == ancestor_id {
1235 return Ok(true);
1236 }
1237 parent = self.registry.by_id(&id).await?.and_then(|repo| up(&repo));
1238 }
1239 Ok(false)
1240 }
1241
1242 /// The repository a pull request whose change is in `source` merges
1243 /// into. `target_id`, when given and not `source` itself, must be the
1244 /// repository a working copy was made from or one `source` is a fork of
1245 /// ([`forked_from`](Self::forked_from)); otherwise none. Without it, a
1246 /// working copy merges into the repository it was made from, and a
1247 /// branch into its own.
1248 pub(crate) async fn pull_target(&self, source: &Repo, target_id: Option<&str>) -> Result<Option<Repo>> {
1249 match target_id.filter(|id| *id != source.id) {
1250 Some(id) if source.fork_of.as_deref() == Some(id) || self.forked_from(source, id).await? => {
1251 self.registry.by_id(id).await
1252 }
1253 Some(_) => Ok(None),
1254 None => match &source.fork_of {
1255 Some(id) => self.registry.by_id(id).await,
1256 None => Ok(Some(source.clone())),
1257 },
1258 }
1259 }
1260
1261 /// Whether a pull request's source lacks commits that the branch it
1262 /// would merge into has.
1263 async fn behind(&self, a: BehindArgs) -> Result<bool> {
1264 let Some(source) = self.registry.by_id(&a.source_id).await? else {
1265 return Ok(false);
1266 };
1267 let target = self.pull_target(&source, a.target_id.as_deref()).await?;
1268 let Some(target) = target else {
1269 return Ok(false);
1270 };
1271 let branch = a.branch.unwrap_or_else(|| target.default_branch.clone());
1272 let target_branch = a.target_branch.unwrap_or_else(|| target.default_branch.clone());
1273 let target_head = self
1274 .read_git(&target)
1275 .await?
1276 .log(&target_branch, 1)
1277 .await?
1278 .into_iter()
1279 .next()
1280 .map(|commit| commit.hash);
1281 let Some(target_head) = target_head else {
1282 return Ok(false);
1283 };
1284 let source_git = self.read_git(&source).await?;
1285 let history = source_git.log(&branch, MAX_ANCESTRY).await?;
1286 if history.is_empty() {
1287 return Ok(false);
1288 }
1289 Ok(!descends_from(&source_git, &history, &target_head).await?)
1290 }
1291
1292 /// The files a pull request's source and the default branch it would
1293 /// merge into each changed since they last agreed. Where the two lists
1294 /// share no file, the merge cannot conflict; where they do, it may.
1295 async fn divergence(&self, a: BehindArgs) -> Result<Option<Divergence>> {
1296 let Some(source) = self.registry.by_id(&a.source_id).await? else {
1297 return Ok(None);
1298 };
1299 let target = self.pull_target(&source, a.target_id.as_deref()).await?;
1300 let Some(target) = target else {
1301 return Ok(None);
1302 };
1303 let branch = a.branch.unwrap_or_else(|| target.default_branch.clone());
1304 let target_branch = a.target_branch.unwrap_or_else(|| target.default_branch.clone());
1305 let source_git = self.read_git(&source).await?;
1306 let target_git = self.read_git(&target).await?;
1307 // The target's side is the same for every pull request into it, and
1308 // worked out once per head (coalesce.rs).
1309 let (history, side) = futures_util::future::try_join(
1310 source_git.log(&branch, MAX_ANCESTRY),
1311 self.target_side(&target, &target_git, &target_branch),
1312 )
1313 .await?;
1314 let target_history = &side.history;
1315 let (Some(head), Some(base)) = (history.first(), target_history.first()) else {
1316 return Ok(None);
1317 };
1318 let behind = !descends_from(&source_git, &history, &base.hash).await?;
1319 let merge_base = nearest_ancestor_in(&source_git, &history, &side.shared).await?;
1320 let mut divergence = Divergence {
1321 head: head.hash.clone(),
1322 base: base.hash.clone(),
1323 merge_base: merge_base.clone(),
1324 behind,
1325 ..Divergence::default()
1326 };
1327 let merge_base_tree = match &merge_base {
1328 Some(hash) => target_history
1329 .iter()
1330 .find(|commit| commit.hash == *hash)
1331 .map(|commit| commit.tree_hash.clone()),
1332 None => None,
1333 };
1334 let Some(merge_base_tree) = merge_base_tree else {
1335 // No common history to compare from: say nothing is known.
1336 divergence.truncated = true;
1337 return Ok(Some(divergence));
1338 };
1339 let (ours, truncated_ours) =
1340 diff::changed_paths(&source_git, Some(&merge_base_tree), &head.tree_hash).await?;
1341 divergence.ours = ours;
1342 divergence.truncated = truncated_ours;
1343 if behind {
1344 let now = now_ms();
1345 let key = (target.id.clone(), merge_base_tree.clone(), base.tree_hash.clone());
1346 let (theirs, truncated_theirs) = match THEIRS.with(|memo| memo.borrow().get(&key, now)) {
1347 Some(kept) => kept,
1348 None => {
1349 let found = diff::changed_paths(&target_git, Some(&merge_base_tree), &base.tree_hash).await?;
1350 THEIRS.with(|memo| memo.borrow_mut().put(key, found.clone(), now));
1351 found
1352 }
1353 };
1354 divergence.theirs = theirs;
1355 divergence.truncated |= truncated_theirs;
1356 }
1357 Ok(Some(divergence))
1358 }
1359
1360 /// A target branch's history from its head, worked out once per head
1361 /// for every pull request asking about it (coalesce.rs). The head is
1362 /// read under the refs version; the history by its hash, which the
1363 /// object cache keeps for good.
1364 async fn target_side<R: GitRepo>(&self, target: &Repo, git: &R, branch: &str) -> Result<Rc<coalesce::TargetSide>> {
1365 let now = now_ms();
1366 let key = refs_cache::usable(registry::refs_state(&target.id), now)
1367 .map(|version| (target.id.clone(), branch.to_owned(), version));
1368 if let Some(key) = &key
1369 && let Some(side) = TARGETS.with(|memo| memo.borrow().get(key, now))
1370 {
1371 return Ok(side);
1372 }
1373 let history = match git.log(branch, 1).await?.first() {
1374 Some(head) => git.log(&head.hash, MAX_ANCESTRY).await?,
1375 None => Vec::new(),
1376 };
1377 let side = coalesce::TargetSide::new(history);
1378 if let Some(key) = key {
1379 TARGETS.with(|memo| memo.borrow_mut().put(key, side.clone(), now));
1380 }
1381 Ok(side)
1382 }
1383
1384 async fn head(&self, a: HeadArgs) -> Result<Option<String>> {
1385 let Some(repo) = self.registry.by_id(&a.repo_id).await? else {
1386 return Ok(None);
1387 };
1388 let branch = if a.branch.is_empty() { &repo.default_branch } else { &a.branch };
1389 let git = self.read_git(&repo).await?;
1390 Ok(git
1391 .log(branch, 1)
1392 .await?
1393 .into_iter()
1394 .next()
1395 .map(|commit| commit.hash))
1396 }
1397
1398 async fn delete_branch(&self, a: DeleteBranchArgs) -> Result<Outcome<bool>> {
1399 if !deletable_branch(&a.branch, a.head.as_deref()) {
1400 return Ok(Outcome::fail(
1401 FailureCode::Forbidden,
1402 "Only branches g1t made for itself can be deleted this way.",
1403 ));
1404 }
1405 let Some(repo) = self.registry.by_id(&a.repo_id).await? else {
1406 return Ok(not_found());
1407 };
1408 if a.branch == repo.default_branch {
1409 return Ok(Outcome::fail(FailureCode::Forbidden, "The default branch is never deleted."));
1410 }
1411 let repo = match self.unpaused(repo).await? {
1412 Ok(repo) => repo,
1413 Err((code, message)) => return Ok(Outcome::fail(code, message)),
1414 };
1415 self.live(&repo).await?;
1416 let git = self.store.open(&store_key(&repo)).await?;
1417 let Some(old) = git
1418 .branches()
1419 .await?
1420 .into_iter()
1421 .find(|branch| branch.name == a.branch)
1422 .map(|branch| branch.hash)
1423 else {
1424 return Ok(Outcome::Ok(false));
1425 };
1426 // Moved since the caller looked: someone else's commits are on it.
1427 if a.head.as_deref().is_some_and(|head| head != old) {
1428 return Ok(Outcome::fail(FailureCode::Conflict, format!("{} moved, so it was left alone.", a.branch)));
1429 }
1430 let access = git.access(Scope::Write).await?;
1431 let deleted = land::delete_ref(&access, &a.branch, &old).await?;
1432 self.refs_moved(&repo.id).await;
1433 if let Err(reason) = deleted {
1434 return Ok(Outcome::fail(
1435 FailureCode::Conflict,
1436 format!("{} could not be deleted: {reason}", a.branch),
1437 ));
1438 }
1439 Ok(Outcome::Ok(true))
1440 }
1441
1442 async fn fork_for_pull(&self, a: ForkArgs) -> Result<Outcome<Repo>> {
1443 let viewer = Some(a.actor.clone());
1444 let Some(source) = self
1445 .registry
1446 .by_id(&a.source_id)
1447 .await?
1448 .filter(|repo| can_read(repo, &viewer))
1449 else {
1450 return Ok(not_found());
1451 };
1452 if let Some((code, message)) = lifecycle::read_only_refusal(&source) {
1453 return Ok(Outcome::fail(code, message));
1454 }
1455 // Its working copy is made in its namespace: not while it moves.
1456 let source = match self.unpaused(source).await? {
1457 Ok(source) => source,
1458 Err((code, message)) => return Ok(Outcome::fail(code, message)),
1459 };
1460 let now = now_ms();
1461 let fork = Repo {
1462 id: new_id("rep", now),
1463 namespace: PULLS_NAMESPACE.to_owned(),
1464 name: a.pull_id.clone(),
1465 description: None,
1466 // A fork is exactly as visible as the repo it came from.
1467 is_private: source.is_private,
1468 owner_id: a.actor.id.clone(),
1469 default_branch: source.default_branch.clone(),
1470 fork_of: Some(source.id.clone()),
1471 parent_id: None,
1472 forked_from: None,
1473 parent: None,
1474 allow_forking: false,
1475 max_ref_updates: None,
1476 protected: false,
1477 created_at: rfc3339(now),
1478 topics: Vec::new(),
1479 website: None,
1480 archived_at: None,
1481 };
1482 // Artifacts forks within a namespace: the copy goes where its
1483 // repository is.
1484 let (namespace, _) = store::locate(&store_key(&source));
1485 self.registry
1486 .claim_store_key(&fork, Some(&namespace), &self.store.default_namespace())
1487 .await?;
1488 self.store
1489 .open(&store_key(&source))
1490 .await?
1491 .fork(&store_key(&fork), true)
1492 .await?;
1493 self.registry.insert(&fork).await?;
1494 self.publish(NewEvent {
1495 kind: "repo.forked",
1496 source: SOURCE,
1497 repo_id: Some(source.id.clone()),
1498 actor: Some(a.actor.id),
1499 data: RepoForked {
1500 repo_id: fork.id.clone(),
1501 source_repo_id: source.id,
1502 pull_id: Some(a.pull_id),
1503 },
1504 })
1505 .await?;
1506 Ok(Outcome::Ok(fork))
1507 }
1508
1509 async fn git_access(&self, a: GitAccessArgs) -> Result<Outcome<GitAccess>> {
1510 let found = self.registry.by_path(&a.path).await?;
1511 Ok(match self.authorize_git(&a.path, &a.viewer, a.service, found).await? {
1512 Outcome::Ok(repo) => {
1513 self.live(&repo).await?;
1514 let write = a.service == GitService::ReceivePack;
1515 if write {
1516 // A push with this credential would not pass through
1517 // here, so nothing that lists the refs is kept until it
1518 // has expired (see refs_cache.rs).
1519 let until = now_ms() + store::CREDENTIAL_LIFE_MS + 60_000;
1520 if let Err(error) = self.registry.refs_open(&repo.id, until).await {
1521 // Before the column exists nothing is kept anyway.
1522 if registry::refs_state(&repo.id).is_some() {
1523 return Err(error);
1524 }
1525 }
1526 }
1527 let scope = if write { Scope::Write } else { Scope::Read };
1528 Outcome::Ok(self.store.handout(&store_key(&repo), scope).await?)
1529 }
1530 Outcome::Fail(failure) => Outcome::Fail(failure),
1531 })
1532 }
1533
1534 /// The repository at `path` (`found`, as just read), if the viewer may
1535 /// use `service` on it: fetch from it, or push to it. A push to a path
1536 /// with nothing there makes the repository, in a workspace the pusher
1537 /// belongs to.
1538 async fn authorize_git(
1539 &self,
1540 path: &RepoPath,
1541 viewer: &Viewer,
1542 service: GitService,
1543 found: Option<Repo>,
1544 ) -> Result<Outcome<Repo>> {
1545 let mut a = GitAccessArgs {
1546 path: path.clone(),
1547 viewer: viewer.clone(),
1548 service,
1549 };
1550 let write = a.service == GitService::ReceivePack;
1551 // An access token: pushing needs code:write, reading a private
1552 // repository code:read. A public repository reads as it would for
1553 // anyone. Which repositories a token reaches is its owner's, checked
1554 // below as for anyone.
1555 if let Some(access) = a.viewer.as_ref().and_then(|user| user.token.as_deref()).cloned() {
1556 // A workflow job's token, and a deploy key, reach their own
1557 // repository only; a job's also the working copies of that
1558 // repository's pull requests, where their heads are.
1559 let name = format!("{}/{}", path.namespace, path.name);
1560 let source = match found.as_ref().and_then(|repo| repo.fork_of.as_deref()) {
1561 Some(source_id) if g1t_contracts::scopes::decide_repo(&access, &name).is_some() => self
1562 .registry
1563 .by_id(source_id)
1564 .await?
1565 .map(|source| format!("{}/{}", source.namespace, source.name)),
1566 _ => None,
1567 };
1568 let public = found.as_ref().is_some_and(|repo| !repo.is_private);
1569 if let Some(why) = git_token_refusal(&access, &name, source.as_deref(), write, public, found.is_some()) {
1570 return Ok(Outcome::fail(FailureCode::Forbidden, format!("{why}\n")));
1571 }
1572 if !write && !access.allows(g1t_contracts::scopes::Scope::CodeRead) {
1573 a.viewer = None;
1574 }
1575 }
1576
1577 // Anonymous callers are asked to authenticate whether or not the repo
1578 // exists, so private repos cannot be told apart from missing ones.
1579 let denied = || match &a.viewer {
1580 Some(_) => not_found(),
1581 None => Outcome::fail(FailureCode::Unauthenticated, "Authentication required."),
1582 };
1583 // A run's token works through the API only: its sandbox has its
1584 // own way to push, to its own pull request.
1585 if a.viewer.as_ref().is_some_and(|user| user.kind == PrincipalKind::Agent) {
1586 return Ok(Outcome::fail(
1587 FailureCode::Forbidden,
1588 "A run's token cannot be used with git this way.",
1589 ));
1590 }
1591 if let (true, Some(user)) = (write, &a.viewer)
1592 && !user.verified
1593 {
1594 return Ok(Outcome::fail(FailureCode::Forbidden, UNVERIFIED));
1595 }
1596 let repo = match found {
1597 Some(repo) => {
1598 let allowed = if write {
1599 can_write(&repo, &a.viewer)
1600 } else {
1601 self.may_read(&repo, &a.viewer).await?
1602 };
1603 if !allowed {
1604 return Ok(denied());
1605 }
1606 // An archived repository, or a pull request's copy of one,
1607 // is read-only.
1608 if write {
1609 let archived = match &repo.fork_of {
1610 Some(source) => self.registry.by_id(source).await?,
1611 None => Some(repo.clone()),
1612 };
1613 match archived {
1614 Some(source) => {
1615 if let Some((code, message)) = lifecycle::read_only_refusal(&source) {
1616 return Ok(Outcome::fail(code, format!("{message}\n")));
1617 }
1618 }
1619 // The repository it was copied from is deleted.
1620 None => return Ok(denied()),
1621 }
1622 }
1623 repo
1624 }
1625 None => {
1626 // Push to create, in a workspace the pusher belongs to.
1627 let owner = a
1628 .viewer
1629 .as_ref()
1630 .filter(|user| write && user.is_member(&a.path.namespace.to_lowercase()));
1631 let Some(owner) = owner else {
1632 return Ok(denied());
1633 };
1634 let created = self.create(push_to_create(owner, &a.path)).await?;
1635 match created {
1636 Outcome::Ok(repo) => repo,
1637 Outcome::Fail(failure) => return Ok(Outcome::Fail(failure)),
1638 }
1639 }
1640 };
1641 // A push, or a credential to push with, waits while the repository
1642 // moves between namespaces (moves.rs), and goes to where it is now.
1643 if write {
1644 return Ok(match self.unpaused(repo).await? {
1645 Ok(repo) => Outcome::Ok(repo),
1646 Err((code, message)) => Outcome::fail(code, format!("{message}\n")),
1647 });
1648 }
1649 Ok(Outcome::Ok(repo))
1650 }
1651
1652 async fn land(&self, a: LandArgs) -> Result<Outcome<Landed>> {
1653 let actor: Viewer = Some(a.actor.clone());
1654 let Some(source) = self.registry.by_id(&a.source_id).await? else {
1655 return Ok(not_found());
1656 };
1657 // A working copy lands on the repository it came from, a fork's
1658 // branch on the repository named, and a branch on its own.
1659 let target = self.pull_target(&source, a.target_id.as_deref()).await?;
1660 let Some(target) = target.filter(|repo| can_read(repo, &actor)) else {
1661 return Ok(not_found());
1662 };
1663 if !registry::can(&target, &actor, Capability::Merge) {
1664 return Ok(Outcome::fail(
1665 FailureCode::Forbidden,
1666 access::needs(Capability::Merge, &format!("{}/{}", target.namespace, target.name)),
1667 ));
1668 }
1669 if !a.actor.verified {
1670 return Ok(Outcome::fail(FailureCode::Forbidden, UNVERIFIED));
1671 }
1672 if let Some((code, message)) = lifecycle::read_only_refusal(&target) {
1673 return Ok(Outcome::fail(code, message));
1674 }
1675 // Moving between namespaces: wait for it (moves.rs). Both are read
1676 // again once it is done, for their new keys.
1677 let (source, target) = match (self.unpaused(source).await?, self.unpaused(target).await?) {
1678 (Ok(source), Ok(target)) => (source, target),
1679 (Err((code, message)), _) | (_, Err((code, message))) => return Ok(Outcome::fail(code, message)),
1680 };
1681
1682 let branch = &a.target_branch.clone().unwrap_or_else(|| target.default_branch.clone());
1683 let from_fork = source.id != target.id;
1684 let source_branch = match a.branch.clone() {
1685 Some(name) if !from_fork && name == *branch => {
1686 return Ok(Outcome::fail(
1687 FailureCode::Invalid,
1688 format!("{branch} cannot be merged into itself."),
1689 ));
1690 }
1691 Some(name) => name,
1692 // A working copy holds the change on the branch it merges into.
1693 None if source.fork_of.is_some() => branch.clone(),
1694 None => {
1695 return Ok(Outcome::fail(
1696 FailureCode::Invalid,
1697 "Say which branch to merge.",
1698 ));
1699 }
1700 };
1701
1702 self.live(&source).await?;
1703 let source_git = self.store.open(&store_key(&source)).await?;
1704 let target_git = self.store.open(&store_key(&target)).await?;
1705 let history = source_git.log(&source_branch, MAX_ANCESTRY).await?;
1706 let Some(new) = history.first().map(|commit| commit.hash.clone()) else {
1707 return Ok(Outcome::fail(
1708 FailureCode::Conflict,
1709 "This pull request has no commits to merge.",
1710 ));
1711 };
1712 let Some(new) = land::checked_commit(&new, a.expected_commit.as_deref()) else {
1713 return Ok(Outcome::fail(FailureCode::Conflict, "The source branch changed after it was checked. Test it again before landing."));
1714 };
1715 // Every method but fast-forward makes commits (merge_methods.rs).
1716 let method = a.method.unwrap_or(g1t_contracts::rules::MergeMethod::FastForward);
1717 if method != g1t_contracts::rules::MergeMethod::FastForward {
1718 return self.land_by(method, &a, &target, &source_git, &target_git, branch, &history, new, from_fork).await;
1719 }
1720 let old = target_git
1721 .log(branch, 1)
1722 .await?
1723 .into_iter()
1724 .next()
1725 .map(|commit| commit.hash);
1726
1727 if old.as_deref() == Some(new.as_str()) {
1728 return Ok(Outcome::Ok(Landed {
1729 commit: new,
1730 previous: None,
1731 method,
1732 rebased: Vec::new(),
1733 }));
1734 }
1735 if a.expected_commit.is_some() && old.is_some() {
1736 // A prior attempt may have published this exact commit and stopped
1737 // before recording the merge; another change can already follow it.
1738 let target_history = target_git.log(branch, MAX_ANCESTRY).await?;
1739 if descends_from(&target_git, &target_history, &new).await? {
1740 return Ok(Outcome::Ok(Landed { commit: new, previous: None, method, rebased: Vec::new() }));
1741 }
1742 }
1743 // Moving the branch to a commit that does not descend from its
1744 // current head would discard whatever landed in between.
1745 if let Some(old) = &old
1746 && !descends_from(&source_git, &history, old).await?
1747 {
1748 let remedy = if from_fork {
1749 format!("Pull {branch} into the pull request's fork, push, and merge again.")
1750 } else {
1751 format!("Merge {branch} into {source_branch}, push, and merge again.")
1752 };
1753 return Ok(Outcome::fail(
1754 FailureCode::Conflict,
1755 format!("{branch} has moved since this pull request was opened. {remedy}"),
1756 ));
1757 }
1758
1759 // For a branch the objects are already in the target; sending them
1760 // again is harmless and keeps one way of moving a ref.
1761 let source_access = source_git.access(Scope::Read).await?;
1762 let target_access = target_git.access(Scope::Write).await?;
1763 let pushed =
1764 land::fast_forward(&source_access, &target_access, branch, old.as_deref(), &new)
1765 .await?;
1766 self.refs_moved(&target.id).await;
1767 if let Err(reason) = pushed {
1768 // Most often another pull request landed between the check and the push.
1769 return Ok(Outcome::fail(
1770 FailureCode::Conflict,
1771 format!("{branch} could not be updated: {reason}"),
1772 ));
1773 }
1774 self.publish_push(
1775 &target,
1776 &format!("refs/heads/{branch}"),
1777 old.as_deref(),
1778 &new,
1779 Some(&a.actor),
1780 None,
1781 )
1782 .await?;
1783 Ok(Outcome::Ok(Landed {
1784 commit: new,
1785 previous: old,
1786 method,
1787 rebased: Vec::new(),
1788 }))
1789 }
1790
1791 async fn compare(&self, a: CompareArgs) -> Result<Outcome<Comparison>> {
1792 let found = self.registry.by_id(&a.repo_id).await?;
1793 let repo = match self.visible(found.clone(), &a.viewer).await? {
1794 Some(repo) => Some(repo),
1795 // A pull request from a private fork is reviewed by those who
1796 // read the repository it merges into, whether or not they can
1797 // read the fork: its head, a commit that repository holds as
1798 // well (`refs/pull/<n>/head`), compared with it, from a base it
1799 // holds. Nothing else of the fork is read this way.
1800 None => match (found, a.target_id.as_deref(), a.head.as_deref()) {
1801 (Some(fork), Some(target_id), Some(head))
1802 if is_commit_hash(head) && a.base.as_deref().is_none_or(is_commit_hash) =>
1803 {
1804 match self.visible(self.registry.by_id(target_id).await?, &a.viewer).await? {
1805 Some(target) if self.forked_from(&fork, target_id).await? => {
1806 let target_git = self.read_git(&target).await?;
1807 let mut held = !target_git.log(head, 1).await?.is_empty();
1808 if let Some(base) = a.base.as_deref() {
1809 held = held && !target_git.log(base, 1).await?.is_empty();
1810 }
1811 held.then_some(fork)
1812 }
1813 _ => None,
1814 }
1815 }
1816 _ => None,
1817 },
1818 };
1819 let Some(repo) = repo else {
1820 return Ok(not_found());
1821 };
1822 let git = self.read_git(&repo).await?;
1823 let head_ref = a.head.as_deref().unwrap_or(&repo.default_branch);
1824 // A pull request into another branch is compared from where it
1825 // left that branch.
1826 let base_branch = a.base_branch.clone();
1827 // The head's history is only searched when the base is worked out
1828 // from another branch.
1829 let depth = if a.base.is_some() || is_commit_hash(head_ref) { 1 } else { MAX_ANCESTRY };
1830 let history = git.log(head_ref, depth).await?;
1831 let Some(head) = history.first() else {
1832 return Ok(Outcome::fail(
1833 FailureCode::Conflict,
1834 "There are no commits to compare.",
1835 ));
1836 };
1837
1838 // Where the head's history meets the default branch of `against`,
1839 // or the branch asked for.
1840 let shared_with = async |against: &Repo| -> Result<Option<String>> {
1841 let against_git = self.read_git(against).await?;
1842 let branch = base_branch.as_deref().unwrap_or(&against.default_branch);
1843 let shared: HashSet<String> = against_git
1844 .log(branch, MAX_ANCESTRY)
1845 .await?
1846 .into_iter()
1847 .map(|commit| commit.hash)
1848 .collect();
1849 nearest_ancestor_in(&git, &history, &shared).await
1850 };
1851 // A fork's branch, compared from the repository it would merge into.
1852 let target = match a.target_id.as_deref().filter(|id| a.base.is_none() && *id != repo.id) {
1853 Some(id) if self.forked_from(&repo, id).await? => {
1854 match self.visible(self.registry.by_id(id).await?, &a.viewer).await? {
1855 Some(target) => Some(target),
1856 None => return Ok(not_found()),
1857 }
1858 }
1859 Some(_) => return Ok(not_found()),
1860 None => None,
1861 };
1862 let base = match (a.base, &repo.fork_of) {
1863 (Some(base), _) => Some(base),
1864 (None, _) if let Some(target) = &target => shared_with(target).await?,
1865 // A fork is compared with the last commit it shares with the
1866 // repository it came from.
1867 (None, Some(target_id)) => match self.registry.by_id(target_id).await? {
1868 Some(target) => shared_with(&target).await?,
1869 None => None,
1870 },
1871 // A branch, with the point where it left the default branch.
1872 // A single commit, with its first parent.
1873 (None, None) if is_commit_hash(head_ref) => head.parents.first().cloned(),
1874 (None, None) if head_ref != base_branch.as_deref().unwrap_or(&repo.default_branch) => {
1875 shared_with(&repo).await?
1876 }
1877 (None, None) => head.parents.first().cloned(),
1878 };
1879 let base_tree = match &base {
1880 Some(base) => git
1881 .log(base, 1)
1882 .await?
1883 .into_iter()
1884 .next()
1885 .map(|commit| commit.tree_hash),
1886 None => None,
1887 };
1888 // Only a few files of it, when asked: how the files a large
1889 // comparison lists without their lines are read, as a page reaches them.
1890 let only = a.paths.as_deref().map(|paths| &paths[..paths.len().min(MAX_COMPARED_PATHS)]);
1891 let (files, truncated) =
1892 diff::compare_trees(&git, base_tree.as_deref(), &head.tree_hash, only).await?;
1893 Ok(Outcome::Ok(Comparison {
1894 base,
1895 head: head.hash.clone(),
1896 files,
1897 truncated,
1898 }))
1899 }
1900
1901 /// Reports that `git_ref` of `repo` (a full ref) now points to `after`,
1902 /// moved by `actor` (marked when that was a workflow job's token).
1903 /// `commits` is how many the move brought, when the caller knows: one
1904 /// for a file committed on the site or a branch brought up to date.
1905 async fn publish_push(
1906 &self,
1907 repo: &Repo,
1908 git_ref: &str,
1909 before: Option<&str>,
1910 after: &str,
1911 actor: Option<&User>,
1912 commits: Option<u32>,
1913 ) -> Result<()> {
1914 let caused_by_job = actor.and_then(g1t_contracts::events::job_run_of).map(str::to_owned);
1915 self.publish_git_push(repo, git_ref, before, after, actor.map(|user| user.id.clone()), false, caused_by_job, commits)
1916 .await
1917 }
1918
1919 /// `publish_push`, saying whether the push reached the store without
1920 /// being scanned for secrets first.
1921 #[allow(clippy::too_many_arguments)]
1922 async fn publish_git_push(
1923 &self,
1924 repo: &Repo,
1925 git_ref: &str,
1926 before: Option<&str>,
1927 after: &str,
1928 actor: Option<String>,
1929 unscanned: bool,
1930 caused_by_job: Option<String>,
1931 commits: Option<u32>,
1932 ) -> Result<()> {
1933 self.publish(NewEvent {
1934 kind: "git.push",
1935 source: SOURCE,
1936 repo_id: Some(repo.id.clone()),
1937 actor,
1938 data: GitPush {
1939 repo_id: repo.id.clone(),
1940 git_ref: git_ref.to_owned(),
1941 before: before.map(str::to_owned),
1942 after: after.to_owned(),
1943 default_branch: git_ref.strip_prefix("refs/heads/")
1944 == Some(repo.default_branch.as_str()),
1945 unscanned,
1946 caused_by_job,
1947 commits,
1948 },
1949 })
1950 .await
1951 }
1952
1953 /// Git over HTTPS. Only what decides the answer happens before it:
1954 /// the repository, who is asking and whether they may, the free
1955 /// workspace limits, push protection, and the store's own answer. The
1956 /// audit entry and what a push changed are recorded once git has its
1957 /// answer. Each answer says how long its steps took (`Server-Timing`).
1958 async fn git_http(&self, request: Request, env: &Env, ctx: &Context) -> Result<Response> {
1959 let mut timing = git_http::Timing::start();
1960 let Some(git) = git_http::parse(&request.url()?) else {
1961 return Response::error("Not found", 404);
1962 };
1963 let response = match self.answer_git(request, &git, env, ctx, &mut timing).await {
1964 Ok(response) => response,
1965 // The git store is busy: git hears when to try again.
1966 Err(error) => match resilience::busy(&error.to_string()) {
1967 Some(busy) => git_http::busy_response(busy)?,
1968 None => return Err(error),
1969 },
1970 };
1971 timing.apply(response)
1972 }
1973
1974 async fn answer_git(
1975 &self,
1976 request: Request,
1977 git: &git_http::GitRequest,
1978 env: &Env,
1979 ctx: &Context,
1980 timing: &mut git_http::Timing,
1981 ) -> Result<Response> {
1982 let write = git.service == GitService::ReceivePack;
1983 let get = request.method() == Method::Get;
1984 let identity = env.service("IDENTITY")?;
1985 // The repository and the caller's credentials, at once. A fetch may
1986 // go by the row as read a moment ago, for the same clone's next
1987 // request; a push always reads it. Anonymous callers cost nothing.
1988 let lookup = async {
1989 if write {
1990 self.registry.by_path(&git.path).await
1991 } else {
1992 self.registry.by_path_recent(&git.path).await
1993 }
1994 };
1995 let (found, viewer) =
1996 futures_util::future::join(lookup, git_http::viewer(&request, &identity)).await;
1997 let mut found = found?;
1998 timing.mark("repo");
1999 // A workspace alias staff set (identity's aliases.rs: `g1t` for
2000 // `flagon-io`) is answered in place, as the repository under the
2001 // workspace's slug: pushes and some clients do not follow
2002 // redirects. Everything after this sees only the workspace's slug.
2003 let aliased = match found {
2004 Some(_) => None,
2005 None => git_http::aliased(git, &identity).await?,
2006 };
2007 if let Some(aliased) = &aliased {
2008 found = if write {
2009 self.registry.by_path(&aliased.path).await?
2010 } else {
2011 self.registry.by_path_recent(&aliased.path).await?
2012 };
2013 timing.mark("alias");
2014 }
2015 let git = aliased.as_ref().unwrap_or(git);
2016 if found.is_none() {
2017 // A workspace that was renamed: git follows a redirect when it
2018 // first asks for refs, and uses the new address from then on.
2019 // A repository transferred to another workspace: the same, to
2020 // its new path. Fetches and pushes both follow either.
2021 let url = request.url()?;
2022 let (renamed, moved) = futures_util::future::join(
2023 git_http::renamed(&url, &identity),
2024 self.registry.resolve_moved(&git.path),
2025 )
2026 .await;
2027 timing.mark("moved");
2028 if let Some(location) = renamed? {
2029 return git_http::moved(&location, get);
2030 }
2031 if let Some(now) = moved?
2032 && let Some(location) = git_http::transferred(&url, &now)
2033 {
2034 return git_http::moved(&location, get);
2035 }
2036 }
2037 let viewer = viewer?;
2038 // A person who has not confirmed their email address can do
2039 // nothing with git until they do: told so, not asked to sign in.
2040 if viewer.as_ref().is_some_and(g1t_contracts::User::awaits_confirmation) {
2041 let site = request.url()?.origin().ascii_serialization();
2042 return git_http::refuse(Outcome::<()>::fail(
2043 FailureCode::Forbidden,
2044 g1t_contracts::accounts::confirm_email_first(&site),
2045 ));
2046 }
2047 // A run credential is checked against its grants, then acts as the
2048 // person it works for. See run_access.rs.
2049 let (request, viewer, audit) = match self.admit_git(request, git, viewer, found.as_ref()).await? {
2050 run_access::Admitted::Go { request, viewer, entry } => (request, viewer, entry),
2051 run_access::Admitted::Refused(response) => return Ok(response),
2052 };
2053 let mut after = AfterGit {
2054 audit,
2055 status: 0,
2056 message: None,
2057 push: None,
2058 };
2059 let repo = match self.authorize_git(&git.path, &viewer, git.service, found).await? {
2060 Outcome::Ok(repo) => repo,
2061 refused => {
2062 let why = match &refused {
2063 Outcome::Fail(failure) => Some(failure.message.trim_end().to_owned()),
2064 Outcome::Ok(_) => None,
2065 };
2066 let response = git_http::refuse(refused)?;
2067 after.ended(response.status_code(), why);
2068 after.spawn(env, ctx);
2069 return Ok(response);
2070 }
2071 };
2072 // A pull request's working copy removed after it closed is made
2073 // again before git uses it (forks.rs).
2074 self.live(&repo).await?;
2075 timing.mark("access");
2076 // Clones check out the default branch g1t keeps, which can have
2077 // changed since the store made the repository.
2078 let default_branch = repo.fork_of.is_none().then(|| repo.default_branch.clone());
2079 let key = store_key(&repo);
2080 let scope = if write { Scope::Write } else { Scope::Read };
2081 let mut request = request;
2082 let protocol = refs_cache::protocol(request.headers().get("git-protocol")?.as_deref());
2083 // A fetch's POST is read here, to tell an `ls-refs` from a fetch of
2084 // objects; the store would have it read in full anyway.
2085 let body = if !write && !get { Some(request.bytes().await?) } else { None };
2086 // What it asks the store, for the meters (meters.rs).
2087 let call = git_ops::classify(git.service, git.endpoint, get, body.as_deref());
2088 // Answers kept from the usual store may name refs the fallback
2089 // store does not have (fallback.rs): none are used, or kept.
2090 let fallback = self.store.on_fallback(&key);
2091 // An answer that lists refs may have been kept: see refs_cache.rs.
2092 let kept_key = refs_cache::kind(git, get, protocol, body.as_deref())
2093 .filter(|_| !fallback)
2094 .zip(refs_cache::usable(registry::refs_state(&repo.id), now_ms()))
2095 .map(|(kind, version)| {
2096 refs_cache::Key::new(&repo.id, version, default_branch.as_deref(), protocol, &kind)
2097 });
2098 // A fresh clone's pack may have been kept too: see pack_cache.rs.
2099 // Under the same refs version, so never across a change to them.
2100 let pack_key = self
2101 .packs
2102 .as_ref()
2103 .filter(|_| !fallback)
2104 .and_then(|_| {
2105 let encoding = request.headers().get("content-encoding").ok().flatten();
2106 pack_cache::cacheable(git, get, protocol, encoding.as_deref(), body.as_deref())
2107 })
2108 .zip(refs_cache::usable(registry::refs_state(&repo.id), now_ms()))
2109 .map(|(normalized, version)| pack_cache::Key::new(&repo.id, version, &normalized));
2110 // A kept answer and the free workspace limits, with a kept
2111 // credential and, for a push, what the repository holds looked up
2112 // alongside. A kept answer goes back without waiting for the
2113 // credential, which it does not need.
2114 let pushing = write && !get;
2115 let ((answer, pack, (limited, held)), kept_access) = {
2116 let shared = self.shared.as_deref();
2117 let answer_and_limits = std::pin::pin!(futures_util::future::join3(
2118 async {
2119 match &kept_key {
2120 Some(kept_key) => refs_cache::get(shared, kept_key).await,
2121 None => None,
2122 }
2123 },
2124 async {
2125 match (&pack_key, self.packs.as_deref()) {
2126 (Some(pack_key), Some(packs)) => pack_cache::get(packs, pack_key).await,
2127 _ => None,
2128 }
2129 },
2130 futures_util::future::join(self.git_limits(call, git, &repo, env), async {
2131 if pushing { Some(self.held(&repo).await) } else { None }
2132 }),
2133 ));
2134 let kept_access = std::pin::pin!(self.store.kept_access(&key, scope));
2135 match futures_util::future::select(answer_and_limits, kept_access).await {
2136 futures_util::future::Either::Left((first, kept_access)) => {
2137 let answered = first.0.is_some() || first.1.is_some() || matches!(first.2.0, Ok(Some(_)) | Err(_));
2138 (first, if answered { None } else { kept_access.await })
2139 }
2140 futures_util::future::Either::Right((kept_access, first)) => (first.await, kept_access),
2141 }
2142 };
2143 timing.mark("kept");
2144 // A kept pack first: it never reaches the store, so it is never an
2145 // operation, and a free workspace past its operation cap still gets
2146 // it. (The other limits are a push's, and a pack is only a fetch.)
2147 if let Some(kept) = pack {
2148 timing.note("pack", "hit");
2149 let sent = body.as_ref().map_or(0, |body| body.len() as u64);
2150 meters::record(pack_cache::HIT, &key, sent, kept.size);
2151 after.ended(200, None);
2152 after.spawn(env, ctx);
2153 return kept.response();
2154 }
2155 if let Some((response, status, message)) = limited? {
2156 after.ended(status, Some(message.to_owned()));
2157 after.spawn(env, ctx);
2158 return Ok(response);
2159 }
2160 if pack_key.is_some() {
2161 timing.note("pack", "miss");
2162 }
2163 if let (Some((entry, found)), Some(kept_key)) = (answer, &kept_key) {
2164 timing.note("refs", found.as_str());
2165 if found == refs_cache::Found::Shared {
2166 let (kept_key, entry) = (kept_key.clone(), entry.clone());
2167 ctx.wait_until(async move { refs_cache::keep_in_colo(&kept_key, &entry).await });
2168 }
2169 // Never reached the store: never an operation.
2170 meters::record(call.cached_meter(), &key, 0, entry.body.len() as u64);
2171 after.ended(200, None);
2172 after.spawn(env, ctx);
2173 return entry.response();
2174 }
2175 if kept_key.is_some() {
2176 timing.note("refs", "miss");
2177 }
2178 // An anonymous fetch the store is about to answer: an operation for
2179 // the repository's workspace, so limited per repository (limits.rs).
2180 if limits::counts_as_anonymous_fetch(call, viewer.is_none())
2181 && g1t_kit::limits::check(env, limits::ANONYMOUS_FETCH, repo.id.clone()).await.limited()
2182 {
2183 let response = limits::too_many_anonymous_fetches(&format!("{}/{}", repo.namespace, repo.name))?;
2184 after.ended(429, Some("Too many anonymous fetches.".to_owned()));
2185 after.spawn(env, ctx);
2186 return Ok(response);
2187 }
2188 // The store's credential: one made a moment ago, here or in another
2189 // isolate (see store.rs), or a new one.
2190 let access = match kept_access {
2191 Some((access, from)) => {
2192 timing.note("cred", from.as_str());
2193 access
2194 }
2195 None => {
2196 let access = self.store.mint_access(&key, scope).await?;
2197 timing.mark("mint");
2198 timing.note("cred", "mint");
2199 access
2200 }
2201 };
2202 // Should the store turn a kept credential down, a fetch's first
2203 // request is tried again with a new one; the requests after it then
2204 // have that one too.
2205 let again = if get { Some(request.clone()?) } else { None };
2206 // A push's checks: workflow files, the rules of the branches and
2207 // tags it changes, saying which rule and why (rules.rs), and push
2208 // protection, which refuses a push that adds a secret
2209 // (secret_scan.rs). See push_checks.rs.
2210 let checks = async |head: &[u8], whole: bool, scan: bool| self.check_receive(&repo, viewer.as_ref(), head, whole, scan).await;
2211 // What a push may bring (pack_limits.rs): the repository's size is
2212 // its own and its pull requests' working copies'.
2213 let limits = if let Some(held) = held {
2214 git_http::PushLimits {
2215 held,
2216 repo_limit: self.repo_limit,
2217 large: self.large_pushes,
2218 ..git_http::PushLimits::default()
2219 }
2220 } else {
2221 git_http::PushLimits::default()
2222 };
2223 let mut outcome = git_http::forward(
2224 request,
2225 body,
2226 git,
2227 &access,
2228 checks,
2229 default_branch.as_deref(),
2230 limits,
2231 timing,
2232 )
2233 .await?;
2234 let turned_down = matches!(
2235 &outcome,
2236 git_http::Push::Forwarded(forwarded) if matches!(forwarded.response.status_code(), 401 | 403)
2237 );
2238 if turned_down {
2239 self.store.forget_access(&key).await;
2240 if let Some(again) = again {
2241 let access = self.store.mint_access(&key, scope).await?;
2242 outcome = git_http::forward(
2243 again,
2244 None,
2245 git,
2246 &access,
2247 async |_: &[u8], _: bool, _: bool| Ok(push_checks::Checks::clear()),
2248 default_branch.as_deref(),
2249 git_http::PushLimits::default(),
2250 timing,
2251 )
2252 .await?;
2253 }
2254 }
2255 let forwarded =
2256 match outcome {
2257 git_http::Push::Forwarded(forwarded) => forwarded,
2258 git_http::Push::Refused(response) => {
2259 after.ended(403, Some("The push was declined by rules.".to_owned()));
2260 after.spawn(env, ctx);
2261 return Ok(response);
2262 }
2263 git_http::Push::Blocked(response) => {
2264 after.ended(403, Some("The push adds a secret.".to_owned()));
2265 after.spawn(env, ctx);
2266 return Ok(response);
2267 }
2268 git_http::Push::Declined(response, reason) => {
2269 after.ended(403, Some(format!("The push was declined: {reason}.")));
2270 after.spawn(env, ctx);
2271 return Ok(response);
2272 }
2273 };
2274 if forwarded.from_store {
2275 let received = forwarded
2276 .response
2277 .headers()
2278 .get("content-length")?
2279 .and_then(|length| length.parse().ok())
2280 .unwrap_or(0);
2281 meters::record(call.meter(), &key, forwarded.sent, received);
2282 }
2283 timing.mark("store");
2284 let mut response = forwarded.response;
2285 let status = response.status_code();
2286 if write && !get {
2287 // A push: the store has moved its refs once it has answered in
2288 // full, so the answer is read before the change is recorded, and
2289 // only then goes back. Whoever fetches after it sees the push.
2290 let headers = response.headers().clone();
2291 headers.delete("content-length")?;
2292 let mut report = response.bytes().await?;
2293 self.refs_moved(&repo.id).await;
2294 timing.mark("refs");
2295 if status == 200 {
2296 push_feedback::add(&mut report, self.security.as_ref(), &repo, &viewer).await;
2297 }
2298 response = Response::from_bytes(report)?.with_headers(headers).with_status(status);
2299 } else if let (Some(kept_key), 200) = (&kept_key, status) {
2300 // A miss: this answer is kept for the next to ask.
2301 let headers = response.headers().clone();
2302 headers.delete("content-length")?;
2303 let body = response.bytes().await?;
2304 if let Some(content_type) = headers.get("content-type")? {
2305 let entry = refs_cache::Entry { content_type, body: body.clone() };
2306 if entry.keepable() {
2307 let shared = self.shared.clone();
2308 let kept_key = kept_key.clone();
2309 ctx.wait_until(async move { refs_cache::keep(shared.as_deref(), &kept_key, &entry).await });
2310 }
2311 }
2312 response = Response::from_bytes(body)?.with_headers(headers).with_status(status);
2313 } else if let (Some(pack_key), Some(packs), true) = (&pack_key, &self.packs, forwarded.from_store) {
2314 // A fresh clone the bucket did not have: counted, and its pack
2315 // kept as it streams to git, when it is a whole one.
2316 meters::record(pack_cache::MISS, &key, forwarded.sent, 0);
2317 // Past the repository's limit on writes to the bucket, the pack
2318 // goes to git without being kept (limits.rs).
2319 if status == 200 && !g1t_kit::limits::check(env, limits::PACK_FILL, repo.id.clone()).await.limited() {
2320 let store_key = key.clone();
2321 let measured = Box::new(move |bytes: u64| meters::record_bytes(pack_cache::MISS, &store_key, 0, bytes));
2322 let (teed, filling) = pack_cache::tee(response, packs.clone(), pack_key, measured)?;
2323 response = teed;
2324 if let Some(filling) = filling {
2325 let pack_key = pack_key.clone();
2326 ctx.wait_until(async move {
2327 let filled = filling.await;
2328 if !matches!(filled, pack_cache::Filled::Kept { .. } | pack_cache::Filled::Abandoned) {
2329 worker::console_warn!("pack {} not kept: {filled:?}", pack_key.as_str());
2330 }
2331 });
2332 }
2333 }
2334 }
2335 after.ended(status, None);
2336 if status == 200 && (forwarded.pack_bytes > 0 || !forwarded.pushed.is_empty() || !forwarded.deleted.is_empty()) {
2337 after.push = Some(PushDone {
2338 repo,
2339 pushed: forwarded.pushed,
2340 deleted: forwarded.deleted,
2341 pack_bytes: forwarded.pack_bytes,
2342 caused_by_job: viewer.as_ref().and_then(g1t_contracts::events::job_run_of).map(str::to_owned),
2343 // The calendar credits people, never a job's, an agent's or a workspace's token.
2344 credit: viewer
2345 .as_ref()
2346 .filter(|user| user.kind == PrincipalKind::User && g1t_contracts::events::job_run_of(user).is_none())
2347 .map(|user| user.id.clone()),
2348 actor: viewer.map(|user: User| user.id),
2349 unscanned: forwarded.unscanned,
2350 });
2351 }
2352 after.spawn(env, ctx);
2353 Ok(response)
2354 }
2355
2356 /// The answer for a request a free workspace's limits stop, or a push
2357 /// to a full repository, with its status and reason for the audit log;
2358 /// `None` to go on.
2359 ///
2360 /// A clone, fetch or push is a git operation, which the git store
2361 /// charges g1t for: counted for billing once the answer has gone back
2362 /// (meters.rs), and a free workspace far past its share is slowed down
2363 /// rather than charged (see git_ops.rs). Whether it is past it is
2364 /// decided from counts this isolate already holds: the database is not
2365 /// asked on the way. A free workspace is never charged for private
2366 /// storage: once its private repositories hold the free amount, pushes
2367 /// to them stop, checked when a push begins so that git shows the
2368 /// reason. So do pushes to a repository at the store's size limit.
2369 async fn git_limits(
2370 &self,
2371 call: git_ops::GitCall,
2372 git: &git_http::GitRequest,
2373 repo: &Repo,
2374 env: &Env,
2375 ) -> Result<Option<(Response, u16, &'static str)>> {
2376 let namespace = git.path.namespace.to_lowercase();
2377 if meters::mapping_now().billable(call.meter()) > 0.0 {
2378 let now = now_ms();
2379 let hour = git_ops::hour_key(&rfc3339(now));
2380 let limits = git_ops::Limits::from_env(env);
2381 if let Some((month, hour_ops)) = git_ops::standing(&namespace, &hour, now)
2382 && git_ops::slow_down(month + 1, hour_ops + 1, limits.free_cap, limits.hourly)
2383 && git_ops::is_free_kept(env.service("BILLING").ok().as_ref(), &namespace).await
2384 {
2385 return Ok(Some((
2386 git_ops::too_many(&namespace, limits.free_cap, limits.hourly)?,
2387 429,
2388 "Too many git operations this hour.",
2389 )));
2390 }
2391 }
2392 if git.service == GitService::ReceivePack && git.endpoint == "info/refs" {
2393 let held = self.held(repo).await;
2394 if held >= self.repo_limit {
2395 let message = format!(
2396 "{}/{} holds about {}, the most a repository may hold on g1t, so it takes no more pushes. Delete what you no longer need, or split it: https://docs.g1t.sh/guides/git/#size-limits\n",
2397 repo.namespace,
2398 repo.name,
2399 pack_limits::megabytes(held)
2400 );
2401 return Ok(Some((Response::error(message, 403)?, 403, "The repository is full.")));
2402 }
2403 }
2404 if git.service == GitService::ReceivePack && git.endpoint == "info/refs" && repo.is_private {
2405 let free = git_ops::free_private_bytes(env);
2406 let held = self.registry.private_bytes(&namespace).await.unwrap_or(0);
2407 if git_ops::storage_full(held, free)
2408 && git_ops::is_free(env.service("BILLING").ok().as_ref(), &namespace).await
2409 {
2410 return Ok(Some((
2411 git_ops::storage_full_response(&namespace, held, free)?,
2412 403,
2413 "Free private storage is full.",
2414 )));
2415 }
2416 }
2417 if git.service == GitService::ReceivePack && git.endpoint == "info/refs" && !repo.is_private {
2418 let free = git_ops::free_public_bytes(env);
2419 if free > 0 {
2420 let held = self.registry.public_bytes(&namespace).await.unwrap_or(0);
2421 if git_ops::public_storage_full(held, free)
2422 && git_ops::is_free(env.service("BILLING").ok().as_ref(), &namespace).await
2423 {
2424 return Ok(Some((
2425 git_ops::public_storage_full_response(&namespace, held, free)?,
2426 403,
2427 "Free public storage is full.",
2428 )));
2429 }
2430 }
2431 }
2432 Ok(None)
2433 }
2434
2435 /// What a repository and its pull requests' working copies hold, as
2436 /// g1t counts it: read for a push's first request, kept a minute for
2437 /// the rest of it.
2438 async fn held(&self, repo: &Repo) -> u64 {
2439 let root = repo.fork_of.clone().unwrap_or_else(|| repo.id.clone());
2440 let now = now_ms();
2441 if let Some(held) = HELD.with(|held| held.borrow().get(&root, now)) {
2442 return held;
2443 }
2444 let held = self.registry.stored_bytes(&root).await.unwrap_or(0).max(0) as u64;
2445 HELD.with(|kept| kept.borrow_mut().put(root, held, now));
2446 held
2447 }
2448
2449 /// A `git.deleted` event for each branch or tag a push deleted that is
2450 /// gone from the store: the store can refuse one ref and accept
2451 /// another. As many as a push announces of tags and branches it moves.
2452 async fn publish_deleted(
2453 &self,
2454 repo: &Repo,
2455 deleted: &[(String, String)],
2456 actor: Option<&str>,
2457 caused_by_job: Option<&str>,
2458 ) -> Result<()> {
2459 let stored = self.store.open(&store_key(repo)).await?;
2460 let is_branch = |git_ref: &str| git_ref.starts_with("refs/heads/");
2461 let tags = deleted.iter().filter(|(git_ref, _)| !is_branch(git_ref)).count();
2462 let announced = deleted
2463 .iter()
2464 .filter(|(git_ref, _)| is_branch(git_ref))
2465 .take(MAX_PUSH_BRANCH_EVENTS)
2466 .chain(deleted.iter().filter(|(git_ref, _)| !is_branch(git_ref) && tags <= MAX_PUSH_TAG_EVENTS));
2467 for (git_ref, before) in announced {
2468 let named = git_ref.strip_prefix("refs/heads/").unwrap_or(git_ref);
2469 let gone = match stored.log(named, 1).await {
2470 Ok(head) => head.is_empty(),
2471 Err(error) => {
2472 worker::console_error!("whether {git_ref} of {} was deleted is not known: {error}", repo.name);
2473 false
2474 }
2475 };
2476 if !gone {
2477 continue;
2478 }
2479 self.publish(NewEvent {
2480 kind: "git.deleted",
2481 source: SOURCE,
2482 repo_id: Some(repo.id.clone()),
2483 actor: actor.map(str::to_owned),
2484 data: GitDeleted {
2485 repo_id: repo.id.clone(),
2486 git_ref: git_ref.clone(),
2487 before: before.clone(),
2488 caused_by_job: caused_by_job.map(str::to_owned),
2489 },
2490 })
2491 .await?;
2492 }
2493 Ok(())
2494 }
2495
2496 /// What a push changed, recorded once git has its answer.
2497 async fn record_push(&self, push: PushDone) -> Result<()> {
2498 let PushDone {
2499 repo,
2500 pushed,
2501 deleted,
2502 pack_bytes,
2503 actor,
2504 unscanned,
2505 caused_by_job,
2506 credit,
2507 } = push;
2508 // What the push stored, for billing's storage meter. A failure only
2509 // leaves the count short.
2510 if pack_bytes > 0
2511 && let Err(error) = self.registry.add_stored_bytes(&repo, pack_bytes).await
2512 {
2513 worker::console_error!("stored bytes for {} not counted: {error}", repo.name);
2514 }
2515 if !deleted.is_empty() {
2516 self.publish_deleted(&repo, &deleted, actor.as_deref(), caused_by_job.as_deref()).await?;
2517 }
2518 if pushed.is_empty() {
2519 return Ok(());
2520 }
2521 // Artifacts' own push notifications are per repository, which does
2522 // not fit a repo per pull request, so the front end reports pushes
2523 // itself: one event for each branch that moved.
2524 let stored = self.store.open(&store_key(&repo)).await?;
2525 let announced = announced_refs(&pushed, &repo.default_branch);
2526 if announced.len() < pushed.len() {
2527 worker::console_log!(
2528 "push to {}: {} of {} refs announced",
2529 repo.name,
2530 announced.len(),
2531 pushed.len()
2532 );
2533 }
2534 for pushed in announced {
2535 // The store can refuse one ref and accept another, so each
2536 // branch is checked against where it actually is. A tag the
2537 // store cannot read back is taken as pushed.
2538 let moved = match pushed.branch() {
2539 Some(branch) => stored
2540 .log(branch, 1)
2541 .await?
2542 .first()
2543 .is_some_and(|commit| commit.hash == pushed.after),
2544 None => stored.log(&pushed.git_ref, 1).await.map_or(true, |head| {
2545 head.first().is_none_or(|commit| commit.hash == pushed.after)
2546 }),
2547 };
2548 // How many commits a branch push brought (push_commits.rs), for
2549 // the event and the person's contribution calendar. A log the
2550 // store cannot read leaves the event without a count and the
2551 // day short.
2552 let commits = match (moved, pushed.branch()) {
2553 (true, Some(_)) => match stored.log(&pushed.after, push_commits::MOST_PER_PUSH + 1).await {
2554 Ok(log) => Some(push_commits::new_commits(&log, pushed.before.as_deref())),
2555 Err(error) => {
2556 worker::console_error!("commits pushed to {} not counted: {error}", repo.name);
2557 None
2558 }
2559 },
2560 _ => None,
2561 };
2562 if let (Some(commits), Some(user_id), Some(branch)) = (commits, credit.as_deref(), pushed.branch())
2563 && push_commits::counts(branch, &repo.default_branch)
2564 && let Err(error) =
2565 push_commits::record(&self.registry.db, user_id, &repo.id, &rfc3339(now_ms())[..10], commits).await
2566 {
2567 worker::console_error!("commits pushed to {} not credited: {error}", repo.name);
2568 }
2569 if moved {
2570 self.publish_git_push(
2571 &repo,
2572 &pushed.git_ref,
2573 pushed.before.as_deref(),
2574 &pushed.after,
2575 actor.clone(),
2576 unscanned,
2577 caused_by_job.clone(),
2578 commits,
2579 )
2580 .await?;
2581 }
2582 }
2583 Ok(())
2584 }
2585}
2586
2587/// The most tags one push announces: a push of more (`git push --tags`
2588/// into a new repository, say) announces none of them, so it starts no
2589/// workflows or package reads, one per tag.
2590const MAX_PUSH_TAG_EVENTS: usize = 3;
2591/// The most branches one push announces. A push of more announces only
2592/// the default branch, if it moved, which the rest of g1t reads from.
2593const MAX_PUSH_BRANCH_EVENTS: usize = 1000;
2594
2595/// The refs of a push that are announced with a `git.push` event each.
2596/// Every ref is stored whatever this says; only the events are capped.
2597fn announced_refs<'a>(pushed: &'a [git_http::Pushed], default_branch: &str) -> Vec<&'a git_http::Pushed> {
2598 let (branches, tags): (Vec<&git_http::Pushed>, Vec<&git_http::Pushed>) =
2599 pushed.iter().partition(|pushed| pushed.branch().is_some());
2600 let mut announced = if branches.len() > MAX_PUSH_BRANCH_EVENTS {
2601 branches.into_iter().filter(|pushed| pushed.branch() == Some(default_branch)).collect()
2602 } else {
2603 branches
2604 };
2605 if tags.len() <= MAX_PUSH_TAG_EVENTS {
2606 announced.extend(tags);
2607 }
2608 announced
2609}
2610
2611/// How long after a push's answer its About may still be reading: the
2612/// waitUntil it runs in ends 30 s after the answer, and what it worked out
2613/// must still be written.
2614const PUSH_ABOUT_BY_MS: u64 = 24_000;
2615/// Less time left than this after recording a push, and its About is left
2616/// for the next view to start.
2617const MIN_PUSH_ABOUT_MS: u64 = 5_000;
2618
2619/// How long a push's About may read, `elapsed` after the push's answer:
2620/// what is left before [`PUSH_ABOUT_BY_MS`], or `None` when that is too
2621/// little to be worth starting.
2622fn push_about_budget(elapsed: u64) -> Option<u64> {
2623 let left = PUSH_ABOUT_BY_MS.saturating_sub(elapsed);
2624 (left >= MIN_PUSH_ABOUT_MS).then_some(left)
2625}
2626
2627/// The repository whose About a push should work out: one that has an
2628/// About, when the push moved its default branch.
2629fn about_after_push(push: &PushDone) -> Option<Repo> {
2630 let moved_default = push.pushed.iter().any(|pushed| pushed.branch() == Some(push.repo.default_branch.as_str()));
2631 (stats::has_about(&push.repo) && moved_default).then(|| push.repo.clone())
2632}
2633
2634/// A push the store accepted, to be recorded once git has its answer.
2635struct PushDone {
2636 repo: Repo,
2637 pushed: Vec<git_http::Pushed>,
2638 /// The branches and tags it asked to delete, each with the commit it
2639 /// pointed to.
2640 deleted: Vec<(String, String)>,
2641 pack_bytes: u64,
2642 actor: Option<String>,
2643 /// Too large to scan for secrets before it was stored.
2644 unscanned: bool,
2645 /// The run whose job's token pushed, if one did: its push starts no
2646 /// workflows.
2647 caused_by_job: Option<String>,
2648 /// The person credited with the push's commits on their contribution
2649 /// calendar: whoever pushed, when that is a person (push_commits.rs).
2650 credit: Option<String>,
2651}
2652
2653/// What a git request leaves for after its answer: its audit entry, with
2654/// how the request ended, and what a push changed.
2655struct AfterGit {
2656 audit: Option<Box<run_access::GitEntry>>,
2657 status: u16,
2658 message: Option<String>,
2659 push: Option<PushDone>,
2660}
2661
2662impl AfterGit {
2663 fn ended(&mut self, status: u16, message: Option<String>) {
2664 self.status = status;
2665 self.message = message;
2666 }
2667
2668 /// Does the work once the response is on its way. A failure is logged:
2669 /// git has already been told how its request went.
2670 fn spawn(self, env: &Env, ctx: &Context) {
2671 if self.audit.is_none() && self.push.is_none() {
2672 return;
2673 }
2674 let env = env.clone();
2675 // A push that moved the default branch has its About worked out
2676 // now, so the next view finds it current rather than starting the
2677 // run itself (about.rs, `refresh`).
2678 let about = self.push.as_ref().and_then(about_after_push);
2679 let started = now_ms();
2680 ctx.wait_until(async move {
2681 let repos = match service(&env) {
2682 Ok(repos) => repos,
2683 Err(error) => {
2684 worker::console_error!("git request not recorded: {error}");
2685 return;
2686 }
2687 };
2688 repos.finish_git(self.audit, self.status, self.message).await;
2689 if let Some(push) = self.push
2690 && let Err(error) = repos.record_push(push).await
2691 {
2692 worker::console_error!("push not recorded: {error}");
2693 }
2694 repos.deferred.settle().await;
2695 if let Some(repo) = about
2696 && let Some(budget) = push_about_budget(now_ms().saturating_sub(started))
2697 {
2698 repos.refresh(&repo, budget).await;
2699 repos.deferred.settle().await;
2700 }
2701 });
2702 }
2703}
2704
2705fn service(env: &Env) -> Result<Repos<ArtifactsStore>> {
2706 let shared = shared::Shared::from_env(env).map(Rc::new);
2707 let deferred = Rc::new(store::Deferred::default());
2708 Ok(Repos {
2709 registry: Registry { db: env.d1("DB")? },
2710 store: ArtifactsStore::new(env, shared.clone(), deferred.clone())?,
2711 deferred,
2712 shared,
2713 packs: pack_cache::Packs::from_env(env).map(Rc::new),
2714 events: env.service("EVENTS")?,
2715 security: env.service("SECURITY").ok(),
2716 billing: env.service("BILLING").ok(),
2717 identity: env.service("IDENTITY").ok(),
2718 work: env.service("WORK").ok(),
2719 free_private_bytes: git_ops::free_private_bytes(env),
2720 fork_days: forks::retention_days(env),
2721 repo_limit: env
2722 .var("REPO_STORAGE_LIMIT_BYTES")
2723 .ok()
2724 .and_then(|value| value.to_string().parse().ok())
2725 .unwrap_or(pack_limits::DEFAULT_REPO_LIMIT_BYTES),
2726 large_pushes: git_http::LargePushes::from_var(env.var("LARGE_PUSHES").ok().map(|value| value.to_string()).as_deref()),
2727 placement: shards::Placement::from_vars(
2728 env.var("GITSTORE_NEW_REPOS").ok().map(|value| value.to_string()).as_deref(),
2729 env.var("GITSTORE_EU_NAMESPACE").ok().map(|value| value.to_string()).as_deref(),
2730 ),
2731 limits: shards::limits(env.var("GITSTORE_NAMESPACE_LIMITS").ok().map(|value| value.to_string()).as_deref()),
2732 })
2733}
2734
2735/// Writes what this isolate metered once the answer has gone back, every
2736/// few seconds at most: now, or once it is due, waiting in this request's
2737/// `wait_until` so nothing counted is left for a request that may never
2738/// come (meters.rs).
2739fn flush_later(env: &Env, ctx: &Context) {
2740 let Some(wait) = meters::plan_flush() else {
2741 return;
2742 };
2743 if let Ok(db) = env.d1("DB") {
2744 ctx.wait_until(async move { meters::flush_after(&db, wait).await });
2745 }
2746}
2747
2748/// `/backups/<job id>/parts/<number>`: the job and the part's number.
2749fn backup_part_path(path: &str) -> Option<(String, u16)> {
2750 let rest = path.strip_prefix("/backups/")?;
2751 let (job, number) = rest.split_once("/parts/")?;
2752 let number = number.parse::<u16>().ok()?;
2753 (!job.is_empty() && !job.contains('/')).then(|| (job.to_owned(), number))
2754}
2755
2756fn backups_off<T>() -> Outcome<T> {
2757 Outcome::fail(FailureCode::Conflict, "Backups are off on this installation: it has no storage for them.")
2758}
2759
2760/// One part of a backup's bundle, with the job's token in its header.
2761async fn backup_part(request: &mut Request, env: &Env, repos: &Repos<ArtifactsStore>, job_id: String, number: u16) -> Result<Response> {
2762 let Some(blobs) = backups::storage(env) else {
2763 return reply(&backups_off::<()>());
2764 };
2765 let token = request.headers().get(g1t_contracts::backups::TOKEN_HEADER)?.unwrap_or_default();
2766 let bytes = request.bytes().await?;
2767 let job = g1t_contracts::backups::BackupJobArgs { job_id, token };
2768 reply(&backups::part(&repos.registry.db, &blobs, &job, number, bytes).await?)
2769}
2770
2771#[cfg(test)]
2772mod backup_path_tests {
2773 use super::backup_part_path;
2774
2775 #[test]
2776 fn a_part_is_named_by_its_job_and_number() {
2777 assert_eq!(backup_part_path("/backups/bkp_1/parts/3"), Some(("bkp_1".to_owned(), 3)));
2778 assert_eq!(backup_part_path("/backups/bkp_1/parts/x"), None);
2779 assert_eq!(backup_part_path("/backups//parts/1"), None);
2780 assert_eq!(backup_part_path("/acme/rocket.git/info/refs"), None);
2781 }
2782}
2783
2784/// Read methods whose answer is an `Outcome`: when the git store is busy,
2785/// the site is told so in words instead of failing the page.
2786const OUTCOME_READS: [&str; 7] = ["tree", "blob", "log", "branches", "blame", "compare", "branch_drift"];
2787
2788#[event(fetch)]
2789async fn fetch(mut request: Request, env: Env, ctx: Context) -> Result<Response> {
2790 let mut repos = service(&env)?;
2791 // A part of a backup's bundle, as the API passes it on from the
2792 // sandbox: bytes, not JSON (backups.rs).
2793 if request.method() == Method::Put
2794 && let Some((job_id, number)) = backup_part_path(&request.path())
2795 {
2796 let answered = backup_part(&mut request, &env, &repos, job_id, number).await;
2797 repos.deferred.hand_over(&ctx);
2798 flush_later(&env, &ctx);
2799 return answered;
2800 }
2801 let Some(method) = rpc_method(&request) else {
2802 let answered = repos.git_http(request, &env, &ctx).await;
2803 repos.deferred.hand_over(&ctx);
2804 flush_later(&env, &ctx);
2805 return answered;
2806 };
2807 let started = now_ms();
2808 // A replica near the caller when it asks for one (crates/kit/src/d1.rs).
2809 // Git over HTTPS above always reads the primary.
2810 let (db, served) = g1t_kit::d1::open(&env, "DB", &request)?;
2811 repos.registry.db = db;
2812 let body: serde_json::Value = request.json().await?;
2813
2814 let answered = async { match method.as_str() {
2815 "get" => reply(&repos.get(args(body)?).await?),
2816 "get_by_id" => reply(&repos.get_by_id(args(body)?).await?),
2817 "readable" => {
2818 let a: ReadableArgs = args(body)?;
2819 reply(&repos.registry.readable(&a.ids, &a.viewer).await?)
2820 }
2821 // Work's contribution calendar: the commits a person pushed each
2822 // day, in repositories the viewer may read (push_commits.rs).
2823 "commit_days" => {
2824 let a: g1t_contracts::repos::CommitDaysArgs = args(body)?;
2825 reply(&push_commits::days(&repos.registry, &a.user_id, &a.since, &a.viewer).await?)
2826 }
2827 "public_namespaces" => {
2828 let a: PublicNamespacesArgs = args(body)?;
2829 reply(&repos.registry.public_namespaces(&a.owner_id).await?)
2830 }
2831 "path_by_id" => {
2832 let a: PathByIdArgs = args(body)?;
2833 reply(
2834 &repos
2835 .registry
2836 .by_id(&a.id)
2837 .await?
2838 .filter(|repo| repo.fork_of.is_none())
2839 .map(|repo| RepoPath {
2840 namespace: repo.namespace,
2841 name: repo.name,
2842 }),
2843 )
2844 }
2845 "list" => {
2846 let a: ListArgs = args(body)?;
2847 reply(
2848 &repos
2849 .registry
2850 .list(
2851 &a.viewer,
2852 a.query.as_deref(),
2853 a.namespace.as_deref(),
2854 a.member_only,
2855 )
2856 .await?,
2857 )
2858 }
2859 "create" => reply(&repos.create(args(body)?).await?),
2860 "transfer" => reply(&repos.transfer(args(body)?).await?),
2861 // A repository's lifecycle: see lifecycle.rs.
2862 "delete" => reply(&repos.delete(args(body)?).await?),
2863 "deleted" => reply(&repos.deleted(args(body)?).await?),
2864 "restore" => reply(&repos.restore(args(body)?).await?),
2865 "purge" => reply(&repos.purge(args(body)?).await?),
2866 "purge_due" => reply(&repos.purge_due(args(body)?).await?),
2867 "rename" => reply(&repos.rename(args(body)?).await?),
2868 "archive" => reply(&repos.archive(args(body)?).await?),
2869 "set_visibility" => reply(&repos.set_visibility(args(body)?).await?),
2870 "set_default_branch" => reply(&repos.set_default_branch(args(body)?).await?),
2871 "rename_branch" => reply(&repos.rename_branch(args(body)?).await?),
2872 "create_branch" => reply(&repos.create_branch(args(body)?).await?),
2873 "remove_branch" => reply(&repos.remove_branch(args(body)?).await?),
2874 "resolve_branch" => reply(&repos.resolve_branch(args(body)?).await?),
2875 "status_by_id" => reply(&repos.status_by_id(args(body)?).await?),
2876 "resolve_path" => {
2877 let a: ResolvePathArgs = args(body)?;
2878 reply(&repos.registry.resolve_moved(&a.path).await?)
2879 }
2880 "namespace_count" => {
2881 let a: NamespaceCountArgs = args(body)?;
2882 reply(&repos.registry.count_in(&a.namespace).await?)
2883 }
2884 "update" => reply(&repos.update(args(body)?).await?),
2885 "tree" => reply(&repos.tree(args(body)?).await?),
2886 "blob" => reply(&repos.blob(args(body)?).await?),
2887 "log" => reply(&repos.log(args(body)?).await?),
2888 "blame" => reply(&repos.blame(args(body)?).await?),
2889 "fork_for_pull" => reply(&repos.fork_for_pull(args(body)?).await?),
2890 // Forks people make into their workspaces: see network.rs.
2891 "fork" => reply(&repos.fork(args(body)?).await?),
2892 "forks" => reply(&repos.forks(args(body)?).await?),
2893 "sync_fork" => reply(&repos.sync_fork(args(body)?).await?),
2894 "detach_fork" => reply(&repos.detach_fork(args(body)?).await?),
2895 "git_access" => reply(&repos.git_access(args(body)?).await?),
2896 "branches" => reply(&repos.branches(args(body)?).await?),
2897 "last_commits" => reply(&repos.last_commits(args(body)?).await?),
2898 "path_log" => reply(&repos.path_log(args(body)?).await?),
2899 "branch_drift" => reply(&repos.branch_drift(args(body)?).await?),
2900 "tags" => reply(&repos.tags(args(body)?).await?),
2901 // The About: what the Files page shows beside the files (about.rs).
2902 // What is kept behind the head is worked out again after the answer.
2903 "about" => {
2904 let (answer, refresh) = repos.about(args(body)?).await?;
2905 about::refresh_later(&env, &ctx, refresh);
2906 reply(&answer)
2907 }
2908 "languages" => {
2909 let (answer, refresh) = repos.languages(args(body)?).await?;
2910 about::refresh_later(&env, &ctx, refresh);
2911 reply(&answer)
2912 }
2913 "contributors" => {
2914 let (answer, refresh) = repos.contributors(args(body)?).await?;
2915 about::refresh_later(&env, &ctx, refresh);
2916 reply(&answer)
2917 }
2918 "license" => {
2919 let (answer, refresh) = repos.license(args(body)?).await?;
2920 about::refresh_later(&env, &ctx, refresh);
2921 reply(&answer)
2922 }
2923 "stars" => reply(&repos.stars(args(body)?).await?),
2924 "star" => reply(&repos.star(args(body)?).await?),
2925 "stargazers" => reply(&repos.stargazers(args(body)?).await?),
2926 "starred" => reply(&repos.starred(args(body)?).await?),
2927 "releases" => reply(&repos.releases(args(body)?).await?),
2928 "release" => reply(&repos.release(args(body)?).await?),
2929 "create_release" => reply(&repos.create_release(args(body)?).await?),
2930 "update_release" => reply(&repos.update_release(args(body)?).await?),
2931 "delete_release" => reply(&repos.delete_release(args(body)?).await?),
2932 "head" => reply(&repos.head(args(body)?).await?),
2933 "behind" => reply(&repos.behind(args(body)?).await?),
2934 "divergence" => reply(&repos.divergence(args(body)?).await?),
2935 "land" => reply(&repos.land(args(body)?).await?),
2936 "build_queue_step" => reply(&repos.build_queue_step(args(body)?).await?),
2937 "update_pull_branch" => reply(&repos.update_pull_branch(args(body)?).await?),
2938 // Services only: a pull request's test merge commit (test_merge.rs).
2939 "pull_merge_ref" => reply(&repos.pull_merge_ref(args(body)?).await?),
2940 "delete_branch" => reply(&repos.delete_branch(args(body)?).await?),
2941 "commit_file" => reply(&repos.commit_file(args(body)?).await?),
2942 "compare" => reply(&repos.compare(args(body)?).await?),
2943 // Services only: a pull request's commits, as rules look at them (rules.rs).
2944 "inspect_commits" => reply(&repos.inspect_commits(args(body)?).await?),
2945 "scan_history" => reply(&repos.scan_history(args(body)?).await?),
2946 "find_lockfiles" => reply(&repos.find_lockfiles(args(body)?).await?),
2947 "match_pattern" => reply(&repos.match_pattern(args(body)?).await?),
2948 "check_secret" => reply(&repos.check_secret(args(body)?).await?),
2949 "list_files" => reply(&repos.list_files(args(body)?).await?),
2950 // The API's tarball and zipball links, and the site opening them (archive_links.rs).
2951 "archive_link" => {
2952 let sealer = env.secret("REPOS_KEY").ok().and_then(|key| g1t_secrets::Sealer::new(&key.to_string()));
2953 reply(&repos.archive_link(args(body)?, sealer.as_ref()).await?)
2954 }
2955 "open_archive_token" => {
2956 let sealer = env.secret("REPOS_KEY").ok().and_then(|key| g1t_secrets::Sealer::new(&key.to_string()));
2957 reply(&repos.open_archive_token(args(body)?, sealer.as_ref()).await?)
2958 }
2959 "changed_files" => reply(&repos.changed_files(args(body)?).await?),
2960 "read_blobs" => reply(&repos.read_blobs(args(body)?).await?),
2961 // Services only: what the Composer registry builds packages from.
2962 "refs" => reply(&repos.refs_of(args(body)?).await?),
2963 "raw_file" => reply(&repos.raw_file(args(body)?).await?),
2964 "raw_blobs" => reply(&repos.raw_blobs(args(body)?).await?),
2965 "visibility" => {
2966 let a: g1t_contracts::repos::VisibilityArgs = args(body)?;
2967 reply(&repos.registry.visibility(&a.paths).await?)
2968 }
2969 "storage" => reply(&repos.registry.storage().await?),
2970 "git_operations" => {
2971 let a: GitOperationsArgs = args(body)?;
2972 reply(&git_ops::totals(&repos.registry.db, &a.month, a.since.as_deref(), a.namespace.as_deref().map(str::to_lowercase).as_deref()).await?)
2973 }
2974 // Identity, once: who created each repository (members.rs there).
2975 "repo_creators" => {
2976 let a: AllIdsArgs = args(body)?;
2977 let limit = a.limit.clamp(1, 500);
2978 let repos = repos.registry.creators_after(a.after.as_deref(), limit).await?;
2979 let next = (repos.len() == limit as usize).then(|| repos.last().map(|repo| repo.id.clone())).flatten();
2980 reply(&CreatorPage { repos, next })
2981 }
2982 "all_ids" => {
2983 let a: AllIdsArgs = args(body)?;
2984 let limit = a.limit.clamp(1, 500);
2985 let ids = repos.registry.ids_after(a.after.as_deref(), limit).await?;
2986 let next = (ids.len() == limit as usize).then(|| ids.last().cloned()).flatten();
2987 reply(&IdPage { ids, next })
2988 }
2989 // The raw meters of the git store, for reconciling with Cloudflare
2990 // (meters.rs, scripts/ops/gitstore-usage.mjs).
2991 "artifacts_usage" => {
2992 let a: meters::UsageArgs = args(body)?;
2993 reply(&meters::usage(&repos.registry.db, &a).await?)
2994 }
2995 "operation_mapping" => reply(&meters::read_mapping(&repos.registry.db).await?),
2996 // Billing: the workspace each pull request's working copy is counted
2997 // for, so Cloudflare's own count of `pulls--<id>` shares out too.
2998 "pull_owners" => {
2999 #[derive(serde::Deserialize)]
3000 struct PullOwnersArgs {
3001 pulls: Vec<String>,
3002 }
3003 let a: PullOwnersArgs = args(body)?;
3004 let pulls: Vec<String> = a.pulls.into_iter().take(500).collect();
3005 reply(&serde_json::json!({ "owners": meters::pull_owners(&repos.registry.db, &pulls).await? }))
3006 }
3007 // Services only: which meters are operations, changed without a deploy.
3008 "set_operation_mapping" => {
3009 let row: meters::MappingRow = args(body)?;
3010 meters::set_mapping(&repos.registry.db, &row, &rfc3339(now_ms())).await?;
3011 reply(&meters::read_mapping(&repos.registry.db).await?)
3012 }
3013 // Backups (backups.rs): the runner's sweep claims queued ones, and
3014 // each sandbox, through the API, asks for its job and says how it went.
3015 "claim_backups" => {
3016 let a: g1t_contracts::backups::ClaimBackupsArgs = args(body)?;
3017 let blobs = backups::storage(&env);
3018 reply(&backups::claim(&repos.registry.db, blobs.as_ref(), &a, now_ms()).await?)
3019 }
3020 "backup_spec" => match backups::storage(&env) {
3021 Some(blobs) => {
3022 let a: g1t_contracts::backups::BackupJobArgs = args(body)?;
3023 let every = backups::Settings::from_env(&env).full_every;
3024 reply(&backups::spec(&repos.registry, &blobs, &repos.store, &a, every, now_ms()).await?)
3025 }
3026 None => reply(&backups_off::<bool>()),
3027 },
3028 "backup_complete" => match backups::storage(&env) {
3029 Some(blobs) => reply(&backups::complete(&repos.registry, &blobs, &args(body)?, now_ms()).await?),
3030 None => reply(&backups_off::<bool>()),
3031 },
3032 "backup_fail" => match backups::storage(&env) {
3033 Some(blobs) => reply(&backups::fail(&repos.registry.db, &blobs, &args(body)?).await?),
3034 None => reply(&backups_off::<bool>()),
3035 },
3036 // How the git store has been answering, for the status page.
3037 "store_health" => {
3038 let a: meters::HealthArgs = args(body)?;
3039 reply(&meters::health(&repos.registry.db, &a).await?)
3040 }
3041 // Where repositories may be kept, for a workspace's settings.
3042 "storage_options" => reply(&repos.storage_options()),
3043 // Services and operators only: how each namespace stands, and
3044 // moving a repository between them (namespaces.rs, moves.rs).
3045 "namespaces" => reply(&repos.standings().await?),
3046 "move_repository" => reply(&repos.move_repository(args(body)?).await?),
3047 "repository_moves" => {
3048 let a: moves::ListMovesArgs = args(body)?;
3049 reply(&repos.registry.moves(a.limit.unwrap_or(50)).await?)
3050 }
3051 _ => Response::error("Unknown method", 404),
3052 } }
3053 .await;
3054 // The git store is busy: said in words, with when to try again.
3055 let answered = match answered {
3056 Err(error) => match resilience::busy(&error.to_string()) {
3057 Some(busy) if OUTCOME_READS.contains(&method.as_str()) => {
3058 reply(&Outcome::<()>::fail(FailureCode::Conflict, busy.message().trim()))
3059 }
3060 Some(busy) => {
3061 let response = Response::error(busy.message(), 503)?;
3062 response.headers().set("retry-after", &busy.retry_after.to_string())?;
3063 Ok(response)
3064 }
3065 None => Err(error),
3066 },
3067 answered => answered,
3068 };
3069 repos.deferred.hand_over(&ctx);
3070 flush_later(&env, &ctx);
3071 slow_answer(&method, now_ms().saturating_sub(started));
3072 served.finish(answered)
3073}
3074
3075/// RPC answers this slow are logged with how long the answer took.
3076const SLOW_RPC_MS: u64 = 1_000;
3077
3078/// Logs an RPC answer that took [`SLOW_RPC_MS`] or more, as a JSON line
3079/// Workers Logs keeps as fields (`rpc`, `answered_ms`). An invocation's
3080/// wall time there also counts what `ctx.wait_until` does after the
3081/// answer, which here includes waiting up to `meters::FLUSH_EVERY_MS` to
3082/// write the meters (`flush_later`), so it reads as seconds for answers
3083/// that took milliseconds; `answered_ms` is the time the caller waited,
3084/// as the `svc` entry of the answer's `server-timing` says.
3085fn slow_answer(method: &str, took: u64) {
3086 if took >= SLOW_RPC_MS {
3087 worker::console_log!("{}", serde_json::json!({ "rpc": method, "answered_ms": took }));
3088 }
3089}
3090
3091/// The nightly cron in wrangler.jsonc: tonight's backups are queued.
3092const BACKUP_CRON: &str = "53 2 * * *";
3093
3094/// The hourly sweep: deleted repositories whose time to be restored has
3095/// passed are purged. See lifecycle.rs. And, at [`BACKUP_CRON`], the
3096/// repositories whose refs moved are queued for a backup (backups.rs).
3097#[event(scheduled)]
3098async fn scheduled(event: ScheduledEvent, env: Env, _ctx: ScheduleContext) {
3099 let repos = match service(&env) {
3100 Ok(repos) => repos,
3101 Err(error) => {
3102 worker::console_error!("repos: the sweep could not start: {error}");
3103 return;
3104 }
3105 };
3106 if event.cron() == BACKUP_CRON {
3107 let Some(blobs) = backups::storage(&env) else { return };
3108 match backups::nightly(&repos.registry.db, &blobs, backups::Settings::from_env(&env), now_ms()).await {
3109 Ok(night) => worker::console_log!("repos: queued {} backups, removed {} of purged repositories", night.queued, night.pruned),
3110 Err(error) => worker::console_error!("repos: backups could not be queued: {error}"),
3111 }
3112 return;
3113 }
3114 match repos.purge_due(PurgeDueArgs::default()).await {
3115 Ok(0) => {}
3116 Ok(count) => worker::console_log!("repos: purged {count} deleted repositories"),
3117 Err(error) => worker::console_error!("repos: the purge sweep failed: {error}"),
3118 }
3119 // Pull requests' working copies whose time has come (forks.rs).
3120 match repos.retire_due().await {
3121 Ok(0) => {}
3122 Ok(count) => worker::console_log!("repos: removed {count} pull request working copies"),
3123 Err(error) => worker::console_error!("repos: the working copy sweep failed: {error}"),
3124 }
3125 // Private forks whose makers can no longer read where they came from
3126 // (network.rs).
3127 match repos.check_private_forks().await {
3128 Ok(0) => {}
3129 Ok(count) => worker::console_log!("repos: deleted {count} private forks their makers can no longer read the source of"),
3130 Err(error) => worker::console_error!("repos: the private fork sweep failed: {error}"),
3131 }
3132 // Repositories moving between namespaces, and old copies (moves.rs).
3133 match repos.run_moves().await {
3134 Ok(0) => {}
3135 Ok(count) => worker::console_log!("repos: moved {count} repositories between namespaces"),
3136 Err(error) => worker::console_error!("repos: the move sweep failed: {error}"),
3137 }
3138 meters::flush(&repos.registry.db).await;
3139}
3140
3141/// Events from the bus. A workspace's rename: its repositories move to the
3142/// workspace's current slug, asked of identity by id, so a repeated or late
3143/// delivery lands in the same place; their git store keys stay as they
3144/// were. A workspace's deletion: its repositories are deleted with it,
3145/// restored with it, or purged with it.
3146#[event(queue)]
3147async fn queue(batch: MessageBatch<Event>, env: Env, ctx: Context) -> Result<()> {
3148 let registry = Registry { db: env.d1("DB")? };
3149 let identity = env.service("IDENTITY")?;
3150 let handled = handle_events(&batch, &env, &registry, &identity).await;
3151 flush_later(&env, &ctx);
3152 handled
3153}
3154
3155/// Each event is acknowledged or retried on its own, so one that fails is
3156/// tried again without the others before and after it running twice.
3157async fn handle_events(batch: &MessageBatch<Event>, env: &Env, registry: &Registry, identity: &Fetcher) -> Result<()> {
3158 for message in batch.messages()? {
3159 match handle_event(message.body(), env, registry, identity).await {
3160 Ok(()) => message.ack(),
3161 Err(error) => {
3162 worker::console_error!("repos: event {} failed: {error}", message.body().id);
3163 message.retry();
3164 }
3165 }
3166 }
3167 Ok(())
3168}
3169
3170async fn handle_event(event: &Event, env: &Env, registry: &Registry, identity: &Fetcher) -> Result<()> {
3171 // A pull request merged, closed or reopened: its working copy is
3172 // kept or let go (forks.rs).
3173 if let Some(change) = forks::pull_change(&event.kind) {
3174 let Some(pull_id) = forks::pull_id_of(&event.data) else {
3175 worker::console_error!("{} {} names no pull request", event.kind, event.id);
3176 return Ok(());
3177 };
3178 let repos = service(env)?;
3179 match change {
3180 forks::PullChange::Settled => repos.pull_settled(&pull_id).await?,
3181 forks::PullChange::Reopened => repos.pull_reopened(&pull_id).await?,
3182 }
3183 return Ok(());
3184 }
3185 // A workspace deleted, restored or purged: its repositories go with
3186 // it, come back with it, or are purged with it (lifecycle.rs).
3187 if event.kind == "workspace.deleting" {
3188 match serde_json::from_value::<WorkspaceDeleting>(event.data.clone()) {
3189 Ok(deleting) => service(env)?.delete_with_workspace(&deleting, &protected_workspaces(env)).await?,
3190 Err(_) => worker::console_error!("workspace.deleting {} could not be read", event.id),
3191 }
3192 return Ok(());
3193 }
3194 if event.kind == "workspace.restored" {
3195 match serde_json::from_value::<WorkspaceRestored>(event.data.clone()) {
3196 Ok(restored) => service(env)?.restore_with_workspace(&restored).await?,
3197 Err(_) => worker::console_error!("workspace.restored {} could not be read", event.id),
3198 }
3199 return Ok(());
3200 }
3201 if event.kind == "workspace.deleted" {
3202 match serde_json::from_value::<WorkspaceDeleted>(event.data.clone()) {
3203 Ok(deleted) => service(env)?.purge_workspace(&deleted, &protected_workspaces(env)).await?,
3204 Err(_) => worker::console_error!("workspace.deleted {} could not be read", event.id),
3205 }
3206 return Ok(());
3207 }
3208 // An account deleted or restored: kept contributors that name it
3209 // (or ghost, for a restore) are worked out again, so it shows as
3210 // ghost for its 30 days, and as itself again if restored.
3211 if let Some(needle) = stats::shown_differently(&event.kind, &event.data) {
3212 let db = env.d1("DB")?;
3213 if let Err(error) = stats::rework_naming(&db, &needle).await {
3214 worker::console_error!("{} {}: contributors not marked to be counted again: {error}", event.kind, event.id);
3215 }
3216 return Ok(());
3217 }
3218 if event.kind != "workspace.renamed" {
3219 return Ok(());
3220 }
3221 let Ok(renamed) = serde_json::from_value::<WorkspaceRenamed>(event.data.clone()) else {
3222 worker::console_error!("workspace.renamed {} could not be read", event.id);
3223 return Ok(());
3224 };
3225 let names: HashMap<String, String> = g1t_kit::call(
3226 identity,
3227 "usernames",
3228 &g1t_contracts::identity::UsernamesArgs {
3229 ids: vec![renamed.workspace_id.clone()],
3230 },
3231 )
3232 .await?;
3233 let current = names
3234 .get(&renamed.workspace_id)
3235 .cloned()
3236 .unwrap_or_else(|| renamed.to.clone());
3237 let left = registry
3238 .rename_namespace(&renamed.stale_slugs(&current), &current)
3239 .await?;
3240 if left > 0 {
3241 worker::console_error!(
3242 "{left} repositories stayed under {} or {}: {current} already has repositories of the same names",
3243 renamed.from,
3244 renamed.to
3245 );
3246 }
3247 Ok(())
3248}
3249
3250/// The workspaces whose repositories never go with a deletion, whatever is
3251/// published: `PROTECTED_WORKSPACES` if set here, and Flagon's always.
3252fn protected_workspaces(env: &Env) -> Vec<String> {
3253 let configured = env.var("PROTECTED_WORKSPACES").ok().map(|v| v.to_string());
3254 g1t_contracts::identity::protected_names(configured.as_deref())
3255}
3256
3257/// What a token's own limits say about git on the repository `repo`
3258/// (`owner/name`), before anyone's role is asked: why it is refused, or
3259/// `None`. `source` is the repository a pull request's working copy at
3260/// `repo` belongs to, which a workflow job's token reaches too. `public`
3261/// is whether anyone may read it, and `exists` whether there is one.
3262///
3263/// A job's token and a deploy key reach their own repository only. Its
3264/// scopes decide the rest: `code:read` to read a private repository,
3265/// `code:write` to push, which a read-only deploy key never has. A deploy
3266/// key never makes a repository by pushing to an empty address.
3267pub(crate) fn git_token_refusal(
3268 access: &g1t_contracts::scopes::TokenAccess,
3269 repo: &str,
3270 source: Option<&str>,
3271 write: bool,
3272 public: bool,
3273 exists: bool,
3274) -> Option<String> {
3275 if let Some(refused) = g1t_contracts::scopes::decide_repo(access, repo)
3276 && !source.is_some_and(|source| access.reaches(source))
3277 {
3278 return Some(refused.reason.unwrap_or_default());
3279 }
3280 let decision = g1t_contracts::scopes::decide_git(access, write, public);
3281 if !decision.allowed {
3282 return Some(decision.reason.unwrap_or_default());
3283 }
3284 if access.deploy_key.is_some() && !exists {
3285 return Some(format!("This deploy key is for {repo}, which is not there any more."));
3286 }
3287 // A token narrowed to chosen repositories, or to another workspace, never
3288 // makes one outside them by pushing to an empty address.
3289 if write
3290 && !exists
3291 && let Some(reach) = &access.reach
3292 {
3293 let namespace = repo.split('/').next().unwrap_or_default();
3294 let all = reach.repositories == g1t_contracts::scopes::RepositorySelection::All;
3295 if !all || !reach.workspace.as_deref().is_some_and(|w| w.eq_ignore_ascii_case(namespace)) {
3296 return Some(format!(
3297 "This access token cannot create {repo}: it reaches only the repositories chosen for it. Create the repository on g1t.sh, then add it to the token."
3298 ));
3299 }
3300 }
3301 None
3302}
3303
3304/// The repository a push to a path that does not exist yet creates: private,
3305/// so nothing pushed by mistake is published. An owner makes it public on
3306/// purpose (`POST /repos/{owner}/{repo}/visibility`).
3307fn push_to_create(owner: &User, path: &RepoPath) -> CreateArgs {
3308 CreateArgs {
3309 owner: owner.clone(),
3310 namespace: path.namespace.clone(),
3311 name: path.name.clone(),
3312 description: None,
3313 is_private: true,
3314 import_url: None,
3315 import_token: None,
3316 }
3317}
3318
3319/// Whether `delete_branch` may remove `branch`: one of g1t's own
3320/// (`g1t-…`), or one whose tip the caller names, such as a dependency
3321/// update's branch after its pull request closed.
3322fn deletable_branch(branch: &str, head: Option<&str>) -> bool {
3323 !branch.is_empty() && (branch.starts_with(G1T_BRANCH_PREFIX) || head.is_some_and(|head| !head.is_empty()))
3324}
3325
3326#[cfg(test)]
3327mod delete_branch_tests {
3328 use super::deletable_branch;
3329
3330 #[test]
3331 fn only_g1t_branches_or_a_named_tip_are_deleted() {
3332 assert!(deletable_branch("g1t-queue-12", None));
3333 assert!(!deletable_branch("g1t/security/sharp-0.35.5", None));
3334 assert!(deletable_branch("g1t/security/sharp-0.35.5", Some("abc123")));
3335 assert!(!deletable_branch("feature", Some("")));
3336 assert!(!deletable_branch("", Some("abc123")));
3337 }
3338}
3339
3340#[cfg(test)]
3341mod announced_refs_tests {
3342 use super::*;
3343
3344 fn pushed(git_ref: String) -> git_http::Pushed {
3345 git_http::Pushed { git_ref, before: None, after: "abc".into() }
3346 }
3347
3348 fn refs(announced: Vec<&git_http::Pushed>) -> Vec<&str> {
3349 announced.into_iter().map(|pushed| pushed.git_ref.as_str()).collect()
3350 }
3351
3352 #[test]
3353 fn a_few_tags_are_announced_and_many_are_not() {
3354 let few: Vec<_> = (1..=3).map(|n| pushed(format!("refs/tags/v{n}"))).chain([pushed("refs/heads/main".into())]).collect();
3355 assert_eq!(refs(announced_refs(&few, "main")), ["refs/heads/main", "refs/tags/v1", "refs/tags/v2", "refs/tags/v3"]);
3356 let many: Vec<_> = (1..=10_000).map(|n| pushed(format!("refs/tags/v{n}"))).chain([pushed("refs/heads/main".into())]).collect();
3357 assert_eq!(refs(announced_refs(&many, "main")), ["refs/heads/main"]);
3358 }
3359
3360 #[test]
3361 fn past_the_branch_cap_only_the_default_branch_is_announced() {
3362 let at_cap: Vec<_> = (0..MAX_PUSH_BRANCH_EVENTS).map(|n| pushed(format!("refs/heads/b{n}"))).collect();
3363 assert_eq!(announced_refs(&at_cap, "main").len(), MAX_PUSH_BRANCH_EVENTS);
3364 let over: Vec<_> = (0..=MAX_PUSH_BRANCH_EVENTS)
3365 .map(|n| pushed(format!("refs/heads/b{n}")))
3366 .chain([pushed("refs/heads/main".into())])
3367 .collect();
3368 assert_eq!(refs(announced_refs(&over, "main")), ["refs/heads/main"]);
3369 assert!(announced_refs(&over, "trunk").is_empty());
3370 }
3371}
3372
3373#[cfg(test)]
3374mod push_about_tests {
3375 use super::*;
3376
3377 fn push(refs: &[&str], fork_of: Option<&str>) -> PushDone {
3378 let repo: Repo = serde_json::from_value(serde_json::json!({
3379 "id": "rep_1", "namespace": "acme", "name": "web", "description": null, "isPrivate": false,
3380 "ownerId": "usr_1", "defaultBranch": "main", "forkOf": fork_of, "createdAt": ""
3381 }))
3382 .unwrap();
3383 PushDone {
3384 repo,
3385 pushed: refs.iter().map(|git_ref| git_http::Pushed { git_ref: (*git_ref).into(), before: None, after: "abc".into() }).collect(),
3386 deleted: Vec::new(),
3387 pack_bytes: 0,
3388 actor: None,
3389 unscanned: false,
3390 caused_by_job: None,
3391 credit: None,
3392 }
3393 }
3394
3395 #[test]
3396 fn a_push_to_the_default_branch_works_out_the_about() {
3397 assert!(about_after_push(&push(&["refs/heads/main"], None)).is_some());
3398 assert!(about_after_push(&push(&["refs/heads/topic", "refs/heads/main"], None)).is_some());
3399 assert!(about_after_push(&push(&["refs/heads/topic"], None)).is_none());
3400 assert!(about_after_push(&push(&["refs/tags/main"], None)).is_none());
3401 // A pull request's working copy has no About.
3402 assert!(about_after_push(&push(&["refs/heads/main"], Some("rep_0"))).is_none());
3403 }
3404
3405 #[test]
3406 fn a_push_about_reads_only_within_what_is_left() {
3407 assert_eq!(push_about_budget(0), Some(PUSH_ABOUT_BY_MS));
3408 assert_eq!(push_about_budget(4_000), Some(PUSH_ABOUT_BY_MS - 4_000));
3409 assert_eq!(push_about_budget(PUSH_ABOUT_BY_MS - MIN_PUSH_ABOUT_MS), Some(MIN_PUSH_ABOUT_MS));
3410 assert_eq!(push_about_budget(PUSH_ABOUT_BY_MS - MIN_PUSH_ABOUT_MS + 1), None);
3411 assert_eq!(push_about_budget(60_000), None);
3412 }
3413}
3414
3415#[cfg(test)]
3416mod push_to_create_tests {
3417 use super::*;
3418
3419 #[test]
3420 fn a_pushed_repository_starts_private() {
3421 let owner: User = serde_json::from_value(serde_json::json!({ "id": "usr_1", "username": "ada" })).unwrap();
3422 let args = push_to_create(&owner, &RepoPath { namespace: "acme".into(), name: "site".into() });
3423 assert!(args.is_private);
3424 assert_eq!((args.namespace.as_str(), args.name.as_str()), ("acme", "site"));
3425 }
3426
3427 fn token(repositories: &str, workspace: Option<&str>) -> g1t_contracts::scopes::TokenAccess {
3428 serde_json::from_value(serde_json::json!({
3429 "token_id": "tok_1",
3430 "scopes": ["code:read", "code:write"],
3431 "fine_grained": { "workspace": workspace, "repositories": repositories, "repo_ids": ["rep_plugins"] },
3432 }))
3433 .unwrap()
3434 }
3435
3436 #[test]
3437 fn a_token_for_chosen_repositories_never_creates_one_by_pushing() {
3438 let why = git_token_refusal(&token("selected", Some("acme")), "acme/new", None, true, false, false).unwrap();
3439 assert!(why.contains("cannot create acme/new"), "{why}");
3440 // It still pushes to a repository that is there; its selection is
3441 // the role check's.
3442 assert_eq!(git_token_refusal(&token("selected", Some("acme")), "acme/plugins", None, true, false, true), None);
3443 }
3444
3445 #[test]
3446 fn a_token_for_all_of_a_workspace_creates_there_and_nowhere_else() {
3447 assert_eq!(git_token_refusal(&token("all", Some("acme")), "Acme/new", None, true, false, false), None);
3448 assert!(git_token_refusal(&token("all", Some("acme")), "other/new", None, true, false, false).is_some());
3449 assert!(git_token_refusal(&token("public", Some("acme")), "acme/new", None, true, false, false).is_some());
3450 assert!(git_token_refusal(&token("all", None), "acme/new", None, true, false, false).is_some());
3451 }
3452
3453 #[test]
3454 fn a_token_without_a_reach_creates_as_its_owner_would() {
3455 let wide: g1t_contracts::scopes::TokenAccess =
3456 serde_json::from_value(serde_json::json!({ "token_id": "tok_1", "scopes": ["code:write"] })).unwrap();
3457 assert_eq!(git_token_refusal(&wide, "acme/new", None, true, false, false), None);
3458 }
3459}
3460
3461#[cfg(test)]
3462mod deploy_key_git_tests {
3463 use super::*;
3464 use g1t_contracts::deploy_keys;
3465
3466 fn key(read_only: bool) -> User {
3467 deploy_keys::principal("wsp_acme", "acme", deploy_keys::access("dk_1", "CI", "acme/rocket", read_only))
3468 }
3469
3470 fn rocket(private: bool) -> Repo {
3471 serde_json::from_value(serde_json::json!({
3472 "id": "rep_rocket",
3473 "namespace": "acme",
3474 "name": "rocket",
3475 "description": null,
3476 "isPrivate": private,
3477 "ownerId": "usr_owner",
3478 "defaultBranch": "main",
3479 "forkOf": null,
3480 "protected": false,
3481 "createdAt": "",
3482 }))
3483 .unwrap()
3484 }
3485
3486 fn refusal(user: &User, repo: &str, write: bool, exists: bool) -> Option<String> {
3487 git_token_refusal(user.token.as_deref().unwrap(), repo, None, write, false, exists)
3488 }
3489
3490 #[test]
3491 fn a_read_only_deploy_key_clones_its_repository_and_never_pushes() {
3492 let user = key(true);
3493 assert_eq!(refusal(&user, "acme/rocket", false, true), None);
3494 assert!(refusal(&user, "acme/rocket", true, true).unwrap().contains("read-only"));
3495 // Its role is a workspace token's: it reads a private repository.
3496 assert!(registry::can_read(&rocket(true), &Some(user)));
3497 }
3498
3499 #[test]
3500 fn a_deploy_key_with_write_access_pushes_to_its_repository() {
3501 let user = key(false);
3502 assert_eq!(refusal(&user, "acme/rocket", true, true), None);
3503 assert!(registry::can_write(&rocket(true), &Some(user)));
3504 }
3505
3506 #[test]
3507 fn a_deploy_key_reaches_no_other_repository() {
3508 let user = key(false);
3509 for other in ["acme/booster", "other/rocket"] {
3510 for write in [false, true] {
3511 let why = refusal(&user, other, write, true).expect(other);
3512 assert!(why.contains("deploy key is for acme/rocket"), "{why}");
3513 }
3514 }
3515 // Not even a pull request's working copy of another repository.
3516 let token = user.token.as_deref().unwrap();
3517 assert!(git_token_refusal(token, "pulls/pr_1", Some("acme/booster"), false, false, true).is_some());
3518 }
3519
3520 #[test]
3521 fn a_deploy_key_never_creates_a_repository() {
3522 let why = refusal(&key(false), "acme/rocket", true, false).unwrap();
3523 assert!(why.contains("not there"), "{why}");
3524 }
3525}