flagon-io/g1t

public

2,595 lines111,495 bytesCode Blame

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