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