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