flagon-io/g1t

public

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