flagon-io/g1t

public

Where people and agents ship software together. The open-source git platform for the whole job: issues, agents, checks and deploys to the edge.

g1t.sh

git agents cloudflare rust forge mcp

3,137 linesCode BlameRaw

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