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