flagon-io/g1t

public

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