flagon-io/g1t

public

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