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