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.

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