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