flagon-io/g1t

public

Where people and agents ship software together. The open-source git platform for the whole job: issues, agents, checks and deploys to the edge.

g1t.sh

git agents cloudflare rust forge mcp

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