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