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,602 linesCode BlameRaw

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