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