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