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