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.

git agents cloudflare rust forge mcp

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