flagon-io/g1t

public

1	//! Mirroring: a repository's links to copies of it on other hosts (see
2	//! `g1t_contracts::mirrors` for the model).
3	//!
4	//! Every linked repository has exactly one leader. A mirror follows a
5	//! remote that leads: it stands by as an exact, read-only copy, and nothing
6	//! runs on it. Someone can turn on CI failover (the remote keeps the
7	//! code, g1t runs its workflows) or take over (g1t leads for a while).
8	//! A takeover is handed back ref by ref: what only g1t changed is
9	//! pushed, a protected branch goes as a pull request, and a branch both
10	//! sides changed waits for a person's decision. A mirror can also be
11	//! moved to g1t for good, after which g1t no longer tracks the remote.
12	//! A repository g1t leads can be mirrored to any number of followers.
13	//!
14	//! This service keeps the links (`remotes`), each host's health
15	//! (`remote_hosts`) and a takeover's starting point (`remote_refs`), and
16	//! decides. The repos service keeps each repository's `RepoMirror`, set
17	//! here with `set_mirror`, so pushes and merges are refused or allowed
18	//! without asking. Git itself moves through the repos service (`mirror`,
19	//! `mirror_refs`, `mirror_apply`).
20	//!
21	//! Hosts are adapters: [`RemoteProvider`] names them, and the few places a
22	//! host's own API is used (credentials, whether a branch is protected,
23	//! opening a pull request) match on it. GitHub goes through g1t's GitHub
24	//! App; another g1t or any git host takes a username and token. A host
25	//! that sends no webhook is polled.
26	//!
27	//! Nothing happens on its own unless someone asked for it in the link's
28	//! settings: by default an unreachable remote is only shown on the
29	//! repository, and a takeover starts when someone starts it.
30	//!
31	//! While g1t runs a GitHub mirror's workflows (CI failover, a takeover, CI
32	//! kept warm), the statuses set on its commits go back to the commit on
33	//! GitHub as `g1t / <context>` (`on_status`), unless the link's
34	//! `report_checks` is off. GitHub's refusals that last, such as the App
35	//! lacking the Commit statuses permission, are kept on the link
36	//! (`checks_error`) for its page.
37
38	use std::collections::{BTreeMap, BTreeSet, HashMap};
39
40	use g1t_contracts::access::{self, Capability};
41	use g1t_contracts::checks::StatusEvent;
42	use g1t_contracts::events::{NewEvent, Publish};
43	use g1t_contracts::identity::{ListMembersArgs, Member};
44	use g1t_contracts::mirrors::*;
45	use g1t_contracts::repos::{
46	GetByIdArgs, MirrorApplied, MirrorApplyArgs, MirrorArgs, MirrorDirection, MirrorRefs, MirrorRefsArgs, Mirrored, RefMove,
47	Repo, SetMirrorArgs,
48	};
49	use g1t_contracts::time::rfc3339;
50	use g1t_contracts::{FailureCode, Outcome, Role, User, new_id};
51	use g1t_kit::{args, now_ms, reply};
52	use g1t_secrets::Sealer;
53	use serde::Deserialize;
54	use serde_json::Value;
55	use worker::wasm_bindgen::JsValue;
56	use worker::{Context, D1Database, Env, Fetcher, Method, Response, Result};
57
58	use crate::github::GithubApp;
59	use crate::http;
60
61	const SOURCE: &str = "integrations";
62	/// A host is unreachable after this many failed checks in a row, spanning
63	/// at least [`DOWN_AFTER_MS`].
64	const DOWN_CHECKS: u32 = 3;
65	const DOWN_AFTER_MS: u64 = 2 * 60 * 1000;
66	/// And reachable again after this many good ones, spanning [`UP_AFTER_MS`].
67	const UP_CHECKS: u32 = 3;
68	const UP_AFTER_MS: u64 = 5 * 60 * 1000;
69	/// How often a remote with no webhook is asked for news.
70	const POLL_MS: u64 = 5 * 60 * 1000;
71	/// Where g1t's commits go when the remote will not take them directly.
72	const HANDBACK_PREFIX: &str = "refs/heads/g1t/handback/";
73	/// The most branches asked about protection in one plan.
74	const MAX_PROTECTION_CHECKS: usize = 20;
75
76	fn fail<T>(code: FailureCode, message: impl Into<String>) -> Outcome<T> {
77	Outcome::fail(code, message)
78	}
79
80	fn null_or(value: Option<&str>) -> JsValue {
81	value.map_or(JsValue::NULL, Into::into)
82	}
83
84	// --- Pure parts, tested below ----------------------------------------------
85
86	/// The host of an https address, lowercased: `github.com`.
87	pub fn host_of(url: &str) -> String {
88	url.trim()
89	.trim_start_matches("https://")
90	.trim_start_matches("http://")
91	.split(['/', '?', '#'])
92	.next()
93	.unwrap_or_default()
94	.rsplit('@')
95	.next()
96	.unwrap_or_default()
97	.to_lowercase()
98	}
99
100	/// A remote as people name it: `github.com/acme/web`.
101	pub fn display_name(url: &str) -> String {
102	let rest = url
103	.trim()
104	.trim_start_matches("https://")
105	.trim_start_matches("http://")
106	.trim_end_matches('/');
107	let rest = rest.strip_suffix(".git").unwrap_or(rest);
108	let rest = rest.rsplit_once('@').map_or(rest, \|(_, after)\| after);
109	let (host, path) = rest.split_once('/').unwrap_or((rest, ""));
110	if path.is_empty() { host.to_lowercase() } else { format!("{}/{path}", host.to_lowercase()) }
111	}
112
113	/// A clone address: https, with `.git` added when the host leaves it off,
114	/// and no credentials in it.
115	pub fn clean_clone_url(url: &str) -> Option<String> {
116	let url = url.trim().trim_end_matches('/');
117	let rest = url.strip_prefix("https://")?;
118	if rest.contains('@') \|\| rest.contains(' ') \|\| !rest.contains('/') \|\| rest.starts_with('/') {
119	return None;
120	}
121	Some(if url.ends_with(".git") { url.to_owned() } else { format!("{url}.git") })
122	}
123
124	/// The web address of a clone address.
125	pub fn web_url(clone_url: &str) -> String {
126	clone_url.strip_suffix(".git").unwrap_or(clone_url).to_owned()
127	}
128
129	const CHECKS_PREFIX: &str = "g1t / ";
130	/// The longest description GitHub keeps on a status.
131	const GITHUB_DESCRIPTION_CHARS: usize = 140;
132	/// What the mirror's page says when the GitHub App cannot set statuses.
133	pub const CHECKS_PERMISSION: &str = "g1t can't report checks to GitHub: the GitHub App needs the Commit statuses permission (read and write). An owner of the GitHub account can accept it in the App's settings on GitHub.";
134
135	/// What g1t's checks are called on GitHub: under `g1t / `, so they read
136	/// apart from GitHub's own (`g1t / CI / push`). A context already named
137	/// for g1t (`g1t / deploy`) keeps its name.
138	pub fn github_context(context: &str) -> String {
139	let context = context.trim();
140	if context.starts_with(CHECKS_PREFIX) { context.to_owned() } else { format!("{CHECKS_PREFIX}{context}") }
141	}
142
143	/// A g1t commit status's state as GitHub takes it. The two name the same
144	/// four; anything else is not sent.
145	pub fn github_state(state: &str) -> Option<&'static str> {
146	Some(match state {
147	"pending" => "pending",
148	"success" => "success",
149	"failure" => "failure",
150	"error" => "error",
151	_ => return None,
152	})
153	}
154
155	/// Whether a link reports g1t's checks to its remote now: a mirror of a
156	/// GitHub repository, with the setting on, while g1t runs its workflows
157	/// (CI failover, a takeover or its hand-back, or CI kept warm). A mirror
158	/// standing by quietly runs nothing, so it has nothing to report.
159	pub fn reports_checks(provider: RemoteProvider, leads: bool, state: RemoteState, settings: &MirrorSettings) -> bool {
160	if provider != RemoteProvider::Github \|\| !leads \|\| !settings.report_checks {
161	return false;
162	}
163	match state {
164	RemoteState::Standby => settings.keep_ci_warm,
165	RemoteState::Ci \| RemoteState::Takeover \| RemoteState::HandingBack => true,
166	RemoteState::Following \| RemoteState::Stuck => false,
167	}
168	}
169
170	/// Where a check reported to GitHub links: the status's own page on g1t,
171	/// or the commit's when it has none.
172	pub fn status_target(site: &str, repo: &str, sha: &str, target: Option<&str>) -> String {
173	let site = site.trim_end_matches('/');
174	match target.map(str::trim).filter(\|url\| !url.is_empty()) {
175	Some(url) if url.starts_with("https://") \|\| url.starts_with("http://") => url.to_owned(),
176	Some(path) if path.starts_with('/') => format!("{site}{path}"),
177	_ => format!("{site}/{repo}/commit/{sha}"),
178	}
179	}
180
181	/// A status's description cut to what GitHub keeps.
182	pub fn github_description(description: Option<&str>) -> Option<String> {
183	let text = description.map(str::trim).filter(\|text\| !text.is_empty())?;
184	if text.chars().count() <= GITHUB_DESCRIPTION_CHARS {
185	return Some(text.to_owned());
186	}
187	let cut: String = text.chars().take(GITHUB_DESCRIPTION_CHARS - 1).collect();
188	Some(format!("{}…", cut.trim_end()))
189	}
190
191	/// Whether `sha` is a whole commit id, as GitHub's statuses take it.
192	pub fn whole_commit(sha: &str) -> bool {
193	sha.len() == 40 && sha.bytes().all(\|byte\| byte.is_ascii_hexdigit())
194	}
195
196	/// What one report to GitHub leaves on the link.
197	#[derive(Clone, Debug, PartialEq, Eq)]
198	pub enum Reported {
199	/// It went through: any recorded problem is over.
200	Clear,
201	/// Nothing to say: the commit is not on GitHub (made on g1t during a
202	/// takeover), the commit has all the statuses GitHub keeps, or GitHub
203	/// is busy or limiting. The next status tries again.
204	Keep,
205	/// A lasting problem, shown on the mirror's page until a report goes
206	/// through.
207	Problem(String),
208	}
209
210	/// Reads GitHub's answer to `POST /repos/{owner}/{repo}/statuses/{sha}`.
211	/// `message` is GitHub's own words, `problem` them said for people.
212	pub fn reported(status: u16, message: &str, problem: &str) -> Reported {
213	match status {
214	200..=299 => Reported::Clear,
215	403 if message.to_lowercase().contains("rate limit") => Reported::Keep,
216	403 => Reported::Problem(CHECKS_PERMISSION.to_owned()),
217	422 \| 429 \| 500..=599 => Reported::Keep,
218	_ => Reported::Problem(format!("g1t can't report checks to GitHub: {problem}")),
219	}
220	}
221
222	/// How a host's health moves with one more check. `now` in milliseconds.
223	#[derive(Clone, Debug, Default, PartialEq, Eq)]
224	pub struct HostHealth {
225	pub failures: u32,
226	pub successes: u32,
227	/// When the current run of failures or successes began.
228	pub streak_ms: u64,
229	pub unreachable_since: Option<u64>,
230	}
231
232	#[derive(Clone, Copy, Debug, PartialEq, Eq)]
233	pub enum Change {
234	None,
235	WentDown,
236	CameBack,
237	}
238
239	impl HostHealth {
240	pub fn reachable(&self) -> bool {
241	self.unreachable_since.is_none()
242	}
243
244	pub fn checked(&self, answered: bool, now: u64) -> (HostHealth, Change) {
245	let mut next = self.clone();
246	if answered {
247	if next.successes == 0 {
248	next.streak_ms = now;
249	}
250	next.successes += 1;
251	next.failures = 0;
252	if next.unreachable_since.is_some()
253	&& next.successes >= UP_CHECKS
254	&& now.saturating_sub(next.streak_ms) >= UP_AFTER_MS
255	{
256	next.unreachable_since = None;
257	return (next, Change::CameBack);
258	}
259	} else {
260	if next.failures == 0 {
261	next.streak_ms = now;
262	}
263	next.failures += 1;
264	next.successes = 0;
265	if next.unreachable_since.is_none()
266	&& next.failures >= DOWN_CHECKS
267	&& now.saturating_sub(next.streak_ms) >= DOWN_AFTER_MS
268	{
269	next.unreachable_since = Some(now);
270	return (next, Change::WentDown);
271	}
272	}
273	(next, Change::None)
274	}
275	}
276
277	/// The branch name in a ref: `main` for `refs/heads/main`.
278	fn branch_of(git_ref: &str) -> Option<&str> {
279	git_ref.strip_prefix("refs/heads/")
280	}
281
282	/// The moves that carry out a hand-back plan, and the refs that go as pull
283	/// requests. `theirs_all` is every ref the remote has, so a hand-back
284	/// branch made before is moved from where it is.
285	pub fn hand_back_moves(plan: &HandbackPlan, theirs_all: &BTreeMap<String, String>) -> (Vec<RefMove>, Vec<String>) {
286	let mut moves = Vec::new();
287	let mut pull_requests = Vec::new();
288	for item in &plan.refs {
289	let push = \|moves: &mut Vec<RefMove>\| {
290	moves.push(RefMove {
291	direction: MirrorDirection::Push,
292	git_ref: item.git_ref.clone(),
293	to: None,
294	old: item.theirs.clone(),
295	new: item.ours.clone(),
296	})
297	};
298	let fetch = \|moves: &mut Vec<RefMove>\| {
299	moves.push(RefMove {
300	direction: MirrorDirection::Pull,
301	git_ref: item.git_ref.clone(),
302	to: None,
303	old: item.ours.clone(),
304	new: item.theirs.clone(),
305	})
306	};
307	let decision = match item.action {
308	RefAction::Diverged => item.decision,
309	RefAction::PullRequest => Some(RefDecision::PullRequest),
310	_ => None,
311	};
312	match (item.action, decision) {
313	(RefAction::Same, _) => {}
314	(RefAction::Push, _) \| (RefAction::Diverged, Some(RefDecision::KeepOurs)) => push(&mut moves),
315	(RefAction::Fetch, _) \| (RefAction::Diverged, Some(RefDecision::KeepTheirs)) => fetch(&mut moves),
316	(_, Some(RefDecision::PullRequest)) => {
317	let Some(branch) = branch_of(&item.git_ref) else {
318	// Only branches can be pulled; a tag goes as it is.
319	push(&mut moves);
320	continue;
321	};
322	if let Some(ours) = &item.ours {
323	let target = format!("{HANDBACK_PREFIX}{branch}");
324	moves.push(RefMove {
325	direction: MirrorDirection::Push,
326	git_ref: item.git_ref.clone(),
327	to: Some(target.clone()),
328	old: theirs_all.get(&target).cloned(),
329	new: Some(ours.clone()),
330	});
331	pull_requests.push(item.git_ref.clone());
332	}
333	// g1t follows the remote's branch; its commits are on the
334	// hand-back branch and kept under refs/g1t/replaced/.
335	fetch(&mut moves);
336	}
337	(RefAction::Diverged, None) \| (RefAction::PullRequest, _) => {}
338	}
339	}
340	(moves, pull_requests)
341	}
342
343	/// Builds the plan for a hand-back from both sides' refs and what each ref
344	/// was when the takeover began. `protected` names branches the remote
345	/// protects.
346	pub fn plan_refs(
347	bases: &BTreeMap<String, (Option<String>, Option<RefDecision>)>,
348	ours: &BTreeMap<String, String>,
349	theirs: &BTreeMap<String, String>,
350	protected: &BTreeSet<String>,
351	) -> Vec<RefPlan> {
352	let names: BTreeSet<&String> = bases.keys().chain(ours.keys()).chain(theirs.keys()).collect();
353	names
354	.into_iter()
355	.filter(\|name\| !name.starts_with(HANDBACK_PREFIX))
356	.filter(\|name\| name.starts_with("refs/heads/") \|\| name.starts_with("refs/tags/"))
357	.filter_map(\|name\| {
358	let (base, decision) = bases.get(name).cloned().unwrap_or((None, None));
359	let ours = ours.get(name).cloned();
360	let theirs = theirs.get(name).cloned();
361	// A ref g1t never had and the remote made since is copied in; a
362	// ref neither had at the start nor has now is nothing.
363	let action = ref_action(base.as_deref(), ours.as_deref(), theirs.as_deref(), protected.contains(name));
364	(action != RefAction::Same).then(\|\| RefPlan {
365	git_ref: name.clone(),
366	base,
367	ours,
368	theirs,
369	action,
370	decision: if action == RefAction::Diverged { decision } else { None },
371	})
372	})
373	.collect()
374	}
375
376	// --- Rows -------------------------------------------------------------------
377
378	#[derive(Clone, Debug, Deserialize)]
379	pub(crate) struct RemoteRow {
380	pub id: String,
381	pub repo_id: String,
382	pub workspace: String,
383	pub repo: String,
384	pub provider: String,
385	pub role: String,
386	pub name: String,
387	pub url: String,
388	pub clone_url: String,
389	pub connection_id: Option<String>,
390	pub username: Option<String>,
391	pub credential: Option<String>,
392	pub state: String,
393	pub state_since: String,
394	pub state_by: Option<String>,
395	pub settings: String,
396	pub synced_at: Option<String>,
397	pub last_error: Option<String>,
398	/// Why the last check could not be reported to the remote.
399	#[serde(default)]
400	pub checks_error: Option<String>,
401	pub created_at: String,
402	}
403
404	impl RemoteRow {
405	fn provider(&self) -> RemoteProvider {
406	RemoteProvider::parse(&self.provider).unwrap_or(RemoteProvider::Git)
407	}
408
409	fn leads(&self) -> bool {
410	self.role == RemoteRole::Leader.as_str()
411	}
412
413	fn state(&self) -> RemoteState {
414	RemoteState::parse(&self.state)
415	}
416
417	fn settings(&self) -> MirrorSettings {
418	serde_json::from_str(&self.settings).unwrap_or_default()
419	}
420
421	fn host(&self) -> String {
422	host_of(&self.clone_url)
423	}
424
425	/// `owner/name` on GitHub.
426	fn full_name(&self) -> &str {
427	self.url.trim_start_matches("https://github.com/")
428	}
429
430	/// What the repos service keeps for a mirror of this remote.
431	fn repo_mirror(&self) -> Option<RepoMirror> {
432	if !self.leads() {
433	return None;
434	}
435	let settings = self.settings();
436	Some(RepoMirror {
437	state: self.state().mirror().unwrap_or_default(),
438	remote: self.name.clone(),
439	url: self.url.clone(),
440	since: self.state_since.clone(),
441	warm: settings.keep_ci_warm,
442	github_workflows: settings.github_workflows,
443	hold_deploys: settings.hold_deploys,
444	})
445	}
446
447	fn contract(&self, health: Option<&HostRow>) -> Remote {
448	Remote {
449	id: self.id.clone(),
450	repo_id: self.repo_id.clone(),
451	repo: self.repo.clone(),
452	provider: self.provider(),
453	role: if self.leads() { RemoteRole::Leader } else { RemoteRole::Follower },
454	name: self.name.clone(),
455	url: self.url.clone(),
456	state: self.state(),
457	state_since: self.state_since.clone(),
458	state_by: self.state_by.clone(),
459	reachable: health.is_none_or(\|health\| health.unreachable_since.is_none()),
460	unreachable_since: health.and_then(\|health\| health.unreachable_since.clone()),
461	synced_at: self.synced_at.clone(),
462	last_error: self.last_error.clone(),
463	checks_problem: self.checks_error.clone(),
464	settings: self.settings(),
465	created_at: self.created_at.clone(),
466	}
467	}
468
469	fn link(&self) -> String {
470	format!("/{}/settings/mirroring", self.repo)
471	}
472	}
473
474	#[derive(Clone, Debug, Default, Deserialize)]
475	pub(crate) struct HostRow {
476	pub host: String,
477	pub failures: u32,
478	pub successes: u32,
479	pub streak_ms: f64,
480	pub unreachable_since: Option<String>,
481	}
482
483	impl HostRow {
484	fn health(&self) -> HostHealth {
485	HostHealth {
486	failures: self.failures,
487	successes: self.successes,
488	streak_ms: self.streak_ms as u64,
489	unreachable_since: self.unreachable_since.as_deref().and_then(crate::github::parse_time),
490	}
491	}
492	}
493
494	/// Who is acting: a person, or g1t on its own (an automatic takeover or
495	/// hand-back), which needs no role.
496	enum By<'a> {
497	Person(&'a User),
498	G1t,
499	}
500
501	impl By<'_> {
502	fn name(&self) -> String {
503	match self {
504	By::Person(user) => user.username.clone(),
505	By::G1t => "g1t".to_owned(),
506	}
507	}
508	}
509
510	pub struct Mirrors {
511	db: D1Database,
512	sealer: Option<Sealer>,
513	repos: Fetcher,
514	identity: Fetcher,
515	events: Option<Fetcher>,
516	github: GithubApp,
517	/// g1t's own address, which checks reported to a remote link to.
518	site_url: String,
519	}
520
521	impl Mirrors {
522	pub fn new(env: &Env) -> Result<Self> {
523	Ok(Mirrors {
524	site_url: env.var("SITE_URL").map(\|value\| value.to_string()).unwrap_or_else(\|_\| "https://g1t.sh".to_owned()),
525	db: env.d1("DB")?,
526	sealer: env.secret("INTEGRATIONS_KEY").ok().and_then(\|key\| Sealer::new(&key.to_string())),
527	repos: env.service("REPOS")?,
528	identity: env.service("IDENTITY")?,
529	events: env.service("EVENTS").ok(),
530	github: GithubApp::new(env)?,
531	})
532	}
533
534	// --- Reading --------------------------------------------------------------
535
536	async fn rows(&self, repo_id: &str) -> Result<Vec<RemoteRow>> {
537	self.db
538	.prepare("SELECT * FROM remotes WHERE repo_id = ? ORDER BY role, created_at")
539	.bind(&[repo_id.into()])?
540	.all()
541	.await?
542	.results::<RemoteRow>()
543	}
544
545	async fn row(&self, id: &str) -> Result<Option<RemoteRow>> {
546	self.db.prepare("SELECT * FROM remotes WHERE id = ?").bind(&[id.into()])?.first::<RemoteRow>(None).await
547	}
548
549	async fn leader(&self, repo_id: &str) -> Result<Option<RemoteRow>> {
550	self.db
551	.prepare("SELECT * FROM remotes WHERE repo_id = ? AND role = 'leader'")
552	.bind(&[repo_id.into()])?
553	.first::<RemoteRow>(None)
554	.await
555	}
556
557	async fn hosts(&self) -> Result<HashMap<String, HostRow>> {
558	Ok(self
559	.db
560	.prepare("SELECT * FROM remote_hosts")
561	.all()
562	.await?
563	.results::<HostRow>()?
564	.into_iter()
565	.map(\|row\| (row.host.clone(), row))
566	.collect())
567	}
568
569	async fn repo(&self, repo_id: &str, viewer: Option<&User>) -> Result<Option<Repo>> {
570	let viewer = viewer.cloned().or_else(\|\| Some(User::system("g1t")));
571	let found: Outcome<Repo> =
572	g1t_kit::call(&self.repos, "get_by_id", &GetByIdArgs { id: repo_id.to_owned(), viewer }).await?;
573	Ok(found.into_result().ok())
574	}
575
576	/// Why `actor` may not change `repo`'s links, if they may not.
577	fn refused<T>(actor: &User, repo: &Repo, capability: Capability) -> Option<Outcome<T>> {
578	let target = access::RepoRef { id: &repo.id, namespace: &repo.namespace, private: repo.is_private };
579	match access::check(Some(actor), target, capability) {
580	Ok(()) => None,
581	Err(access::Denied::NotFound) => Some(fail(FailureCode::NotFound, "Repository not found.")),
582	Err(access::Denied::Forbidden) => {
583	Some(fail(FailureCode::Forbidden, access::needs(capability, &format!("{}/{}", repo.namespace, repo.name))))
584	}
585	}
586	}
587
588	/// The repository and its leader, for someone changing them.
589	async fn leader_for(&self, actor: &User, repo_id: &str) -> Result<std::result::Result<(Repo, RemoteRow), Outcome<MirrorView>>> {
590	let Some(repo) = self.repo(repo_id, Some(actor)).await? else {
591	return Ok(Err(fail(FailureCode::NotFound, "Repository not found.")));
592	};
593	if let Some(refused) = Self::refused(actor, &repo, Capability::ManageIntegrations) {
594	return Ok(Err(refused));
595	}
596	let Some(row) = self.leader(repo_id).await? else {
597	return Ok(Err(fail(FailureCode::NotFound, format!("{}/{} is not a mirror.", repo.namespace, repo.name))));
598	};
599	Ok(Ok((repo, row)))
600	}
601
602	async fn view_of(&self, repo_id: &str, can_manage: bool, plan: Option<HandbackPlan>) -> Result<MirrorView> {
603	let hosts = self.hosts().await?;
604	Ok(MirrorView {
605	remotes: self.rows(repo_id).await?.iter().map(\|row\| row.contract(hosts.get(&row.host()))).collect(),
606	plan,
607	can_manage,
608	notes: Vec::new(),
609	})
610	}
611
612	pub async fn view(&self, a: MirrorViewArgs) -> Result<Outcome<MirrorView>> {
613	let Some(repo) = self.repo(&a.repo_id, a.viewer.as_ref()).await? else {
614	return Ok(fail(FailureCode::NotFound, "Repository not found."));
615	};
616	let can_manage = a
617	.viewer
618	.as_ref()
619	.is_some_and(\|viewer\| Self::refused::<()>(viewer, &repo, Capability::ManageIntegrations).is_none());
620	Ok(Outcome::Ok(self.view_of(&repo.id, can_manage, None).await?))
621	}
622
623	pub async fn briefs(&self, a: MirrorBriefsArgs) -> Result<Vec<RemoteBrief>> {
624	let ids: Vec<&String> = a.repo_ids.iter().take(200).collect();
625	if ids.is_empty() {
626	return Ok(Vec::new());
627	}
628	let marks = vec!["?"; ids.len()].join(", ");
629	let bind: Vec<JsValue> = ids.iter().map(\|id\| id.as_str().into()).collect();
630	let rows = self
631	.db
632	.prepare(format!("SELECT * FROM remotes WHERE repo_id IN ({marks}) ORDER BY role, created_at"))
633	.bind(&bind)?
634	.all()
635	.await?
636	.results::<RemoteRow>()?;
637	let hosts = self.hosts().await?;
638	Ok(rows
639	.iter()
640	.map(\|row\| RemoteBrief {
641	repo_id: row.repo_id.clone(),
642	role: if row.leads() { RemoteRole::Leader } else { RemoteRole::Follower },
643	name: row.name.clone(),
644	state: row.state(),
645	reachable: hosts.get(&row.host()).is_none_or(\|host\| host.unreachable_since.is_none()),
646	synced_at: row.synced_at.clone(),
647	})
648	.collect())
649	}
650
651	// --- Talking to hosts -----------------------------------------------------
652
653	/// The user and token a remote is opened with.
654	async fn credential(&self, row: &RemoteRow) -> Result<std::result::Result<(Option<String>, String), String>> {
655	match row.provider() {
656	RemoteProvider::Github => {
657	let Some(installation) = row.connection_id.as_deref().and_then(\|id\| id.parse::<u64>().ok()) else {
658	return Ok(Err("This link has no GitHub installation.".to_owned()));
659	};
660	Ok(self.github.installation_token(installation).await?.map(\|token\| (None, token)))
661	}
662	RemoteProvider::G1t \| RemoteProvider::Git => {
663	let (Some(sealer), Some(sealed)) = (&self.sealer, row.credential.as_deref()) else {
664	return Ok(Err("This link has no token. Add one in its settings.".to_owned()));
665	};
666	match sealer.open(sealed, &format!("rmt:{}", row.id)) {
667	Some(token) => Ok(Ok((row.username.clone(), token))),
668	None => Ok(Err("This link's token could not be read. Add it again.".to_owned())),
669	}
670	}
671	}
672	}
673
674	/// Which of `branches` the remote protects. Hosts that cannot say
675	/// protect none: a refused push still goes as a pull request.
676	async fn protected(&self, row: &RemoteRow, token: &str, branches: &[String]) -> Result<BTreeSet<String>> {
677	let mut protected = BTreeSet::new();
678	if row.provider() != RemoteProvider::Github {
679	return Ok(protected);
680	}
681	for git_ref in branches.iter().take(MAX_PROTECTION_CHECKS) {
682	let Some(branch) = branch_of(git_ref) else { continue };
683	let Ok(answer) = self
684	.github
685	.api(Method::Get, &format!("/repos/{}/branches/{}", row.full_name(), branch), token, None)
686	.await
687	else {
688	continue;
689	};
690	if answer.ok() && answer.json()["protected"].as_bool() == Some(true) {
691	protected.insert(git_ref.clone());
692	}
693	}
694	Ok(protected)
695	}
696
697	/// Opens a pull request on the remote from a hand-back branch. Returns
698	/// a line for people: where it is, or how to open it by hand.
699	async fn open_pull_request(&self, row: &RemoteRow, token: &str, git_ref: &str, repo: &str) -> Result<String> {
700	let branch = branch_of(git_ref).unwrap_or(git_ref);
701	let head = format!("g1t/handback/{branch}");
702	let by_hand = format!("{branch}: g1t's commits are on {head} on {}. Open a pull request from it into {branch}.", row.name);
703	if row.provider() != RemoteProvider::Github {
704	return Ok(by_hand);
705	}
706	let body = serde_json::json!({
707	"title": format!("Changes made on g1t while {} was unreachable", row.name),
708	"head": head,
709	"base": branch,
710	"body": format!(
711	"g1t led [{repo}](https://g1t.sh/{repo}) while this repository could not be reached. These are the commits made there on `{branch}`.\n\nThe branch is protected here, so they come as a pull request rather than a push."
712	),
713	"maintainer_can_modify": true,
714	});
715	let Ok(answer) = self.github.api(Method::Post, &format!("/repos/{}/pulls", row.full_name()), token, Some(body)).await
716	else {
717	return Ok(by_hand);
718	};
719	let said = answer.json();
720	Ok(if answer.ok() {
721	format!("{branch}: opened {}", said["html_url"].as_str().unwrap_or("a pull request"))
722	} else if answer.status == 422 && said.to_string().contains("already exists") {
723	format!("{branch}: the pull request from {head} was already open and now has the new commits.")
724	} else {
725	format!("{by_hand} ({})", answer.problem("GitHub"))
726	})
727	}
728
729	async fn repos_refs(&self, row: &RemoteRow, username: Option<String>, token: String) -> Result<Outcome<MirrorRefs>> {
730	g1t_kit::call(
731	&self.repos,
732	"mirror_refs",
733	&MirrorRefsArgs { repo_id: row.repo_id.clone(), url: row.clone_url.clone(), token, username },
734	)
735	.await
736	}
737
738	/// g1t's own refs, asking nothing of the remote: a takeover starts while
739	/// it is away, when not even a token can be had from it.
740	async fn our_refs(&self, row: &RemoteRow) -> Result<Outcome<MirrorRefs>> {
741	g1t_kit::call(
742	&self.repos,
743	"mirror_refs",
744	&MirrorRefsArgs { repo_id: row.repo_id.clone(), url: String::new(), token: String::new(), username: None },
745	)
746	.await
747	}
748
749	async fn apply(&self, row: &RemoteRow, username: Option<String>, token: String, moves: Vec<RefMove>) -> Result<Outcome<MirrorApplied>> {
750	g1t_kit::call(
751	&self.repos,
752	"mirror_apply",
753	&MirrorApplyArgs { repo_id: row.repo_id.clone(), url: row.clone_url.clone(), token, username, moves },
754	)
755	.await
756	}
757
758	/// Copies refs in the link's direction: a mirror catches up, a follower
759	/// is pushed to. `Err` is a reason, already recorded on the link.
760	async fn sync(&self, row: &RemoteRow) -> Result<std::result::Result<Mirrored, String>> {
761	let direction = if row.leads() { MirrorDirection::Pull } else { MirrorDirection::Push };
762	let (username, token) = match self.credential(row).await? {
763	Ok(credential) => credential,
764	Err(reason) => {
765	self.note(row, Some(&reason)).await?;
766	return Ok(Err(reason));
767	}
768	};
769	let done: Outcome<Mirrored> = g1t_kit::call(
770	&self.repos,
771	"mirror",
772	&MirrorArgs { repo_id: row.repo_id.clone(), url: row.clone_url.clone(), token, username, direction },
773	)
774	.await?;
775	match done {
776	Outcome::Ok(mirrored) => {
777	self.note(row, None).await?;
778	self.host_checked(&row.host(), true, None).await?;
779	Ok(Ok(mirrored))
780	}
781	Outcome::Fail(failure) => {
782	let unreachable = failure.code == FailureCode::Unavailable;
783	if unreachable {
784	self.host_checked(&row.host(), false, Some(&failure.message)).await?;
785	} else {
786	self.note(row, Some(&failure.message)).await?;
787	}
788	if !row.leads() && !unreachable {
789	self.set_state(row, RemoteState::Stuck, None).await?;
790	}
791	Ok(Err(failure.message))
792	}
793	}
794	}
795
796	async fn note(&self, row: &RemoteRow, problem: Option<&str>) -> Result<()> {
797	let statement = match problem {
798	Some(problem) => self
799	.db
800	.prepare("UPDATE remotes SET last_error = ?2 WHERE id = ?1")
801	.bind(&[row.id.as_str().into(), problem.into()])?,
802	None => self
803	.db
804	.prepare("UPDATE remotes SET last_error = NULL, synced_at = ?2 WHERE id = ?1")
805	.bind(&[row.id.as_str().into(), rfc3339(now_ms()).into()])?,
806	};
807	statement.run().await?;
808	if problem.is_none() && row.state() == RemoteState::Stuck {
809	self.set_state(row, RemoteState::Following, None).await?;
810	}
811	Ok(())
812	}
813
814	// --- State ------------------------------------------------------------------
815
816	/// Records a link's state, and tells the repos service what a mirror
817	/// is now. A link the repos service has not heard of yet is retried by
818	/// the cron (`recorded`).
819	async fn set_state(&self, row: &RemoteRow, state: RemoteState, by: Option<&str>) -> Result<RemoteRow> {
820	let now = rfc3339(now_ms());
821	self.db
822	.prepare("UPDATE remotes SET state = ?2, state_since = ?3, state_by = ?4, recorded = 0 WHERE id = ?1")
823	.bind(&[row.id.as_str().into(), state.as_str().into(), now.as_str().into(), null_or(by)])?
824	.run()
825	.await?;
826	let row = RemoteRow {
827	state: state.as_str().to_owned(),
828	state_since: now,
829	state_by: by.map(str::to_owned),
830	..row.clone()
831	};
832	self.record(&row).await?;
833	Ok(row)
834	}
835
836	/// Tells the repos service a repository's mirror, as this link has it.
837	async fn record(&self, row: &RemoteRow) -> Result<()> {
838	if !row.leads() {
839	return Ok(());
840	}
841	let done: Outcome<Repo> = g1t_kit::call(
842	&self.repos,
843	"set_mirror",
844	&SetMirrorArgs { repo_id: row.repo_id.clone(), mirror: row.repo_mirror() },
845	)
846	.await?;
847	if done.into_result().is_ok() {
848	self.db.prepare("UPDATE remotes SET recorded = 1 WHERE id = ?").bind(&[row.id.as_str().into()])?.run().await?;
849	}
850	Ok(())
851	}
852
853	async fn publish(&self, kind: &'static str, row: &RemoteRow, by: Option<&str>, title: String, detail: Option<String>, notify: Vec<String>) {
854	let Some(events) = &self.events else { return };
855	let event = NewEvent {
856	kind,
857	source: SOURCE,
858	repo_id: Some(row.repo_id.clone()),
859	actor: None,
860	data: MirrorEvent {
861	repo_id: row.repo_id.clone(),
862	repo: row.repo.clone(),
863	remote_id: row.id.clone(),
864	remote: row.name.clone(),
865	state: (kind != "mirror.moved_in").then(\|\| row.state.clone()),
866	from: None,
867	by: by.map(str::to_owned),
868	title,
869	detail,
870	notify,
871	link: row.link(),
872	},
873	};
874	let sent: Result<Value> = g1t_kit::call(events, "publish", &Publish { events: vec![event] }).await;
875	if let Err(error) = sent {
876	worker::console_error!("mirrors: publishing {kind} for {} failed: {error}", row.repo);
877	}
878	}
879
880	/// The workspace's owners, when the link asks for the inbox.
881	async fn told(&self, row: &RemoteRow, except: Option<&str>) -> Vec<String> {
882	if row.settings().notify != Notify::Inbox {
883	return Vec::new();
884	}
885	let members: Result<Outcome<Vec<Member>>> = g1t_kit::call(
886	&self.identity,
887	"list_members",
888	&ListMembersArgs { slug: row.workspace.clone(), viewer: Some(User::system(&row.workspace)) },
889	)
890	.await;
891	match members {
892	Ok(Outcome::Ok(members)) => members
893	.into_iter()
894	.filter(\|member\| member.role == Role::Owner)
895	.map(\|member\| member.username)
896	.filter(\|name\| except.is_none_or(\|except\| !name.eq_ignore_ascii_case(except)))
897	.collect(),
898	_ => Vec::new(),
899	}
900	}
901
902	// --- Takeover, CI failover, hand-back, moving in ------------------------------
903
904	pub async fn take_over(&self, a: MirrorActArgs) -> Result<Outcome<MirrorView>> {
905	let (_, row) = match self.leader_for(&a.actor, &a.repo_id).await? {
906	Ok(found) => found,
907	Err(refused) => return Ok(refused),
908	};
909	self.start_takeover(row, By::Person(&a.actor)).await
910	}
911
912	async fn start_takeover(&self, row: RemoteRow, by: By<'_>) -> Result<Outcome<MirrorView>> {
913	match row.state() {
914	RemoteState::Takeover => return Ok(Outcome::Ok(self.view_of(&row.repo_id, true, None).await?)),
915	RemoteState::HandingBack => {
916	return Ok(fail(FailureCode::Conflict, "It is being handed back. Wait for that to finish, or for it to stop."));
917	}
918	_ => {}
919	}
920	// The starting point: what g1t holds now, which is what it last
921	// copied from the remote.
922	let refs = match self.our_refs(&row).await? {
923	Outcome::Ok(refs) => refs,
924	Outcome::Fail(failure) => return Ok(Outcome::Fail(failure)),
925	};
926	let mut statements = vec![self.db.prepare("DELETE FROM remote_refs WHERE remote_id = ?").bind(&[row.id.as_str().into()])?];
927	for (name, hash) in &refs.ours {
928	statements.push(
929	self.db
930	.prepare("INSERT INTO remote_refs (remote_id, ref, base) VALUES (?, ?, ?)")
931	.bind(&[row.id.as_str().into(), name.as_str().into(), hash.as_str().into()])?,
932	);
933	}
934	self.db.batch(statements).await?;
935	let name = by.name();
936	let row = self.set_state(&row, RemoteState::Takeover, Some(&name)).await?;
937	let notify = self.told(&row, Some(&name)).await;
938	let title = match by {
939	By::G1t => format!("g1t took over {} while {} is unreachable", row.repo, row.name),
940	By::Person(_) => format!("{name} took over {} from {}", row.repo, row.name),
941	};
942	self.publish("mirror.state_changed", &row, Some(&name), title, None, notify).await;
943	Ok(Outcome::Ok(self.view_of(&row.repo_id, true, None).await?))
944	}
945
946	pub async fn ci(&self, a: MirrorCiArgs) -> Result<Outcome<MirrorView>> {
947	let (_, row) = match self.leader_for(&a.actor, &a.repo_id).await? {
948	Ok(found) => found,
949	Err(refused) => return Ok(refused),
950	};
951	let next = match (row.state(), a.on) {
952	(RemoteState::Standby, true) => RemoteState::Ci,
953	(RemoteState::Ci, false) => RemoteState::Standby,
954	(RemoteState::Ci, true) \| (RemoteState::Standby, false) => {
955	return Ok(Outcome::Ok(self.view_of(&row.repo_id, true, None).await?));
956	}
957	_ => return Ok(fail(FailureCode::Conflict, "g1t leads this repository now: its workflows already run here.")),
958	};
959	let row = self.set_state(&row, next, Some(&a.actor.username)).await?;
960	let title = if a.on {
961	format!("{} started running {}'s workflows on g1t", a.actor.username, row.name)
962	} else {
963	format!("{} ended CI failover for {}", a.actor.username, row.repo)
964	};
965	self.publish("mirror.state_changed", &row, Some(&a.actor.username), title, None, Vec::new()).await;
966	Ok(Outcome::Ok(self.view_of(&row.repo_id, true, None).await?))
967	}
968
969	/// What handing back would do now.
970	async fn plan(&self, row: &RemoteRow) -> Result<Outcome<HandbackPlan>> {
971	let (username, token) = match self.credential(row).await? {
972	Ok(credential) => credential,
973	Err(reason) => return Ok(fail(FailureCode::Conflict, reason)),
974	};
975	let refs = match self.repos_refs(row, username, token.clone()).await? {
976	Outcome::Ok(refs) => refs,
977	Outcome::Fail(failure) => return Ok(Outcome::Fail(failure)),
978	};
979	let bases = self.bases(row).await?;
980	let Some(theirs) = refs.theirs else {
981	// The remote is away: what g1t changed, as it would go.
982	let refs = plan_refs(&bases, &refs.ours, &bases_as_theirs(&bases), &BTreeSet::new());
983	return Ok(Outcome::Ok(HandbackPlan::new(refs, false)));
984	};
985	let candidates: Vec<String> = plan_refs(&bases, &refs.ours, &theirs, &BTreeSet::new())
986	.into_iter()
987	.filter(\|plan\| plan.action == RefAction::Push && plan.ours.is_some() && plan.theirs.is_some())
988	.map(\|plan\| plan.git_ref)
989	.collect();
990	let protected = self.protected(row, &token, &candidates).await?;
991	Ok(Outcome::Ok(HandbackPlan::new(plan_refs(&bases, &refs.ours, &theirs, &protected), true)))
992	}
993
994	async fn bases(&self, row: &RemoteRow) -> Result<BTreeMap<String, (Option<String>, Option<RefDecision>)>> {
995	#[derive(Deserialize)]
996	struct Base {
997	#[serde(rename = "ref")]
998	git_ref: String,
999	base: Option<String>,
1000	decision: Option<String>,
1001	}
1002	Ok(self
1003	.db
1004	.prepare("SELECT ref, base, decision FROM remote_refs WHERE remote_id = ?")
1005	.bind(&[row.id.as_str().into()])?
1006	.all()
1007	.await?
1008	.results::<Base>()?
1009	.into_iter()
1010	.map(\|base\| {
1011	let decision = base.decision.and_then(\|text\| serde_json::from_value(Value::String(text)).ok());
1012	(base.git_ref, (base.base, decision))
1013	})
1014	.collect())
1015	}
1016
1017	pub async fn hand_back_plan(&self, a: MirrorActArgs) -> Result<Outcome<MirrorView>> {
1018	let (_, row) = match self.leader_for(&a.actor, &a.repo_id).await? {
1019	Ok(found) => found,
1020	Err(refused) => return Ok(refused),
1021	};
1022	if !matches!(row.state(), RemoteState::Takeover \| RemoteState::HandingBack) {
1023	return Ok(fail(FailureCode::Conflict, "g1t has not taken over, so there is nothing to hand back."));
1024	}
1025	let plan = match self.plan(&row).await? {
1026	Outcome::Ok(plan) => plan,
1027	Outcome::Fail(failure) => return Ok(Outcome::Fail(failure)),
1028	};
1029	Ok(Outcome::Ok(self.view_of(&row.repo_id, true, Some(plan)).await?))
1030	}
1031
1032	pub async fn hand_back(&self, a: MirrorHandBackArgs) -> Result<Outcome<MirrorView>> {
1033	let (_, row) = match self.leader_for(&a.actor, &a.repo_id).await? {
1034	Ok(found) => found,
1035	Err(refused) => return Ok(refused),
1036	};
1037	let mut statements = Vec::new();
1038	for (git_ref, decision) in &a.decisions {
1039	let decision = serde_json::to_value(decision).ok().and_then(\|value\| value.as_str().map(str::to_owned));
1040	statements.push(
1041	self.db
1042	.prepare(
1043	"INSERT INTO remote_refs (remote_id, ref, decision) VALUES (?1, ?2, ?3)
1044	ON CONFLICT (remote_id, ref) DO UPDATE SET decision = excluded.decision",
1045	)
1046	.bind(&[row.id.as_str().into(), git_ref.as_str().into(), null_or(decision.as_deref())])?,
1047	);
1048	}
1049	if !statements.is_empty() {
1050	self.db.batch(statements).await?;
1051	}
1052	self.finish_takeover(row, By::Person(&a.actor)).await
1053	}
1054
1055	/// Hands a takeover back: freezes the repository, moves each ref as the
1056	/// plan says, and returns to standing by. If anything is refused, g1t
1057	/// keeps the lead and says why, so nobody is stuck.
1058	async fn finish_takeover(&self, row: RemoteRow, by: By<'_>) -> Result<Outcome<MirrorView>> {
1059	if !matches!(row.state(), RemoteState::Takeover \| RemoteState::HandingBack) {
1060	return Ok(fail(FailureCode::Conflict, "g1t has not taken over, so there is nothing to hand back."));
1061	}
1062	let plan = match self.plan(&row).await? {
1063	Outcome::Ok(plan) => plan,
1064	Outcome::Fail(failure) => return Ok(Outcome::Fail(failure)),
1065	};
1066	if !plan.reachable {
1067	return Ok(fail(FailureCode::Unavailable, format!("{} is not answering yet. Hand back once it is.", row.name)));
1068	}
1069	if !plan.ready {
1070	return Ok(Outcome::Fail(undecided(&plan)));
1071	}
1072	let name = by.name();
1073	// Nothing moves on g1t while it goes back.
1074	let row = self.set_state(&row, RemoteState::HandingBack, Some(&name)).await?;
1075	let plan = match self.plan(&row).await? {
1076	Outcome::Ok(plan) if plan.ready => plan,
1077	Outcome::Ok(plan) => {
1078	let row = self.set_state(&row, RemoteState::Takeover, row.state_by.as_deref()).await?;
1079	let _ = row;
1080	return Ok(Outcome::Fail(undecided(&plan)));
1081	}
1082	Outcome::Fail(failure) => {
1083	self.set_state(&row, RemoteState::Takeover, Some(&name)).await?;
1084	return Ok(Outcome::Fail(failure));
1085	}
1086	};
1087	let (username, token) = match self.credential(&row).await? {
1088	Ok(credential) => credential,
1089	Err(reason) => {
1090	self.set_state(&row, RemoteState::Takeover, Some(&name)).await?;
1091	return Ok(fail(FailureCode::Conflict, reason));
1092	}
1093	};
1094	let theirs_all = match self.repos_refs(&row, username.clone(), token.clone()).await? {
1095	Outcome::Ok(refs) => refs.theirs.unwrap_or_default(),
1096	Outcome::Fail(_) => BTreeMap::new(),
1097	};
1098	let (moves, mut pull_requests) = hand_back_moves(&plan, &theirs_all);
1099	let mut problems = Vec::new();
1100	if !moves.is_empty() {
1101	let applied = match self.apply(&row, username.clone(), token.clone(), moves).await? {
1102	Outcome::Ok(applied) => applied,
1103	Outcome::Fail(failure) => {
1104	self.set_state(&row, RemoteState::Takeover, Some(&name)).await?;
1105	self.note(&row, Some(&failure.message)).await?;
1106	return Ok(Outcome::Fail(failure));
1107	}
1108	};
1109	// A push the remote refused (a protected branch it did not say
1110	// was protected) goes as a pull request instead.
1111	let refused: Vec<RefPlan> = applied
1112	.moved
1113	.iter()
1114	.filter(\|moved\| moved.direction == MirrorDirection::Push && moved.problem.is_some())
1115	.filter_map(\|moved\| {
1116	plan.refs
1117	.iter()
1118	.find(\|item\| item.git_ref == moved.git_ref && branch_of(&item.git_ref).is_some() && item.ours.is_some())
1119	.cloned()
1120	})
1121	.collect();
1122	for moved in applied.moved.iter().filter(\|moved\| moved.problem.is_some()) {
1123	if !refused.iter().any(\|item\| item.git_ref == moved.git_ref) {
1124	problems.push(format!("{}: {}", moved.git_ref, moved.problem.as_deref().unwrap_or_default()));
1125	}
1126	}
1127	if !refused.is_empty() {
1128	let retry = HandbackPlan::new(
1129	refused
1130	.into_iter()
1131	.map(\|item\| RefPlan { action: RefAction::PullRequest, ..item })
1132	.collect(),
1133	true,
1134	);
1135	let (moves, more) = hand_back_moves(&retry, &theirs_all);
1136	match self.apply(&row, username.clone(), token.clone(), moves).await? {
1137	Outcome::Ok(applied) => {
1138	for moved in applied.moved.iter().filter(\|moved\| moved.problem.is_some()) {
1139	problems.push(format!("{}: {}", moved.git_ref, moved.problem.as_deref().unwrap_or_default()));
1140	}
1141	pull_requests.extend(more);
1142	}
1143	Outcome::Fail(failure) => problems.push(failure.message),
1144	}
1145	}
1146	}
1147	if !problems.is_empty() {
1148	let reason = format!("Some branches did not go back: {}", problems.join("; "));
1149	let row = self.set_state(&row, RemoteState::Takeover, Some(&name)).await?;
1150	self.note(&row, Some(&reason)).await?;
1151	return Ok(fail(FailureCode::Conflict, format!("{reason}. g1t still leads; try again or decide differently.")));
1152	}
1153	let mut notes = Vec::new();
1154	for git_ref in &pull_requests {
1155	notes.push(self.open_pull_request(&row, &token, git_ref, &row.repo).await?);
1156	}
1157	self.db.prepare("DELETE FROM remote_refs WHERE remote_id = ?").bind(&[row.id.as_str().into()])?.run().await?;
1158	let row = self.set_state(&row, RemoteState::Standby, Some(&name)).await?;
1159	// Anything else the remote has, g1t now follows.
1160	if let Err(reason) = self.sync(&row).await? {
1161	notes.push(format!("Catching up after handing back: {reason}"));
1162	}
1163	let notify = self.told(&row, Some(&name)).await;
1164	let detail = (!notes.is_empty()).then(\|\| notes.join("\n"));
1165	self.publish(
1166	"mirror.state_changed",
1167	&row,
1168	Some(&name),
1169	format!("{} was handed back to {}", row.repo, row.name),
1170	detail,
1171	notify,
1172	)
1173	.await;
1174	let mut view = self.view_of(&row.repo_id, true, None).await?;
1175	view.notes = notes;
1176	Ok(Outcome::Ok(view))
1177	}
1178
1179	pub async fn move_in(&self, a: MirrorMoveInArgs) -> Result<Outcome<MirrorView>> {
1180	let (_, row) = match self.leader_for(&a.actor, &a.repo_id).await? {
1181	Ok(found) => found,
1182	Err(refused) => return Ok(refused),
1183	};
1184	if row.state() == RemoteState::HandingBack {
1185	return Ok(fail(FailureCode::Conflict, "It is being handed back. Move it to g1t once that is done."));
1186	}
1187	let done: Outcome<Repo> =
1188	g1t_kit::call(&self.repos, "set_mirror", &SetMirrorArgs { repo_id: row.repo_id.clone(), mirror: None }).await?;
1189	if let Outcome::Fail(failure) = done {
1190	return Ok(Outcome::Fail(failure));
1191	}
1192	self.db.prepare("DELETE FROM remote_refs WHERE remote_id = ?").bind(&[row.id.as_str().into()])?.run().await?;
1193	let mut notes = Vec::new();
1194	if a.keep_remote_updated {
1195	let now = rfc3339(now_ms());
1196	self.db
1197	.prepare(
1198	"UPDATE remotes SET role = 'follower', state = 'following', state_since = ?2, state_by = ?3, recorded = 1
1199	WHERE id = ?1",
1200	)
1201	.bind(&[row.id.as_str().into(), now.as_str().into(), a.actor.username.as_str().into()])?
1202	.run()
1203	.await?;
1204	if let Some(follower) = self.row(&row.id).await?
1205	&& let Err(reason) = self.sync(&follower).await?
1206	{
1207	notes.push(format!("{} could not be brought up to date yet: {reason}", row.name));
1208	}
1209	} else {
1210	self.db.prepare("DELETE FROM remotes WHERE id = ?").bind(&[row.id.as_str().into()])?.run().await?;
1211	}
1212	if row.provider() == RemoteProvider::Github {
1213	let mode = if a.keep_remote_updated { "push" } else { "import" };
1214	self.db
1215	.prepare("UPDATE github_repos SET mode = ? WHERE repo_id = ?")
1216	.bind(&[mode.into(), row.repo_id.as_str().into()])?
1217	.run()
1218	.await?;
1219	}
1220	let title = if a.keep_remote_updated {
1221	format!("{} moved {} to g1t; {} now follows it", a.actor.username, row.repo, row.name)
1222	} else {
1223	format!("{} moved {} to g1t and stopped tracking {}", a.actor.username, row.repo, row.name)
1224	};
1225	self.publish("mirror.moved_in", &row, Some(&a.actor.username), title, None, Vec::new()).await;
1226	let mut view = self.view_of(&row.repo_id, true, None).await?;
1227	view.notes = notes;
1228	Ok(Outcome::Ok(view))
1229	}
1230
1231	pub async fn sync_now(&self, a: MirrorActArgs) -> Result<Outcome<MirrorView>> {
1232	let Some(repo) = self.repo(&a.repo_id, Some(&a.actor)).await? else {
1233	return Ok(fail(FailureCode::NotFound, "Repository not found."));
1234	};
1235	// Syncing brings commits in, as pushing does.
1236	if let Some(refused) = Self::refused(&a.actor, &repo, Capability::Push) {
1237	return Ok(refused);
1238	}
1239	let rows = self.rows(&repo.id).await?;
1240	if rows.is_empty() {
1241	return Ok(fail(FailureCode::NotFound, "This repository is not linked to another host."));
1242	}
1243	let mut problems = Vec::new();
1244	for row in &rows {
1245	if row.leads() && !row.state().mirror().is_some_and(MirrorState::follows) {
1246	continue;
1247	}
1248	if let Err(reason) = self.sync(row).await? {
1249	problems.push(format!("{}: {reason}", row.name));
1250	}
1251	}
1252	if !problems.is_empty() {
1253	return Ok(fail(FailureCode::Conflict, problems.join("; ")));
1254	}
1255	Ok(Outcome::Ok(self.view_of(&repo.id, true, None).await?))
1256	}
1257
1258	pub async fn settings(&self, a: MirrorSettingsArgs) -> Result<Outcome<Remote>> {
1259	let Some(row) = self.row(&a.remote_id).await? else {
1260	return Ok(fail(FailureCode::NotFound, "That link was not found."));
1261	};
1262	let Some(repo) = self.repo(&row.repo_id, Some(&a.actor)).await? else {
1263	return Ok(fail(FailureCode::NotFound, "Repository not found."));
1264	};
1265	if let Some(refused) = Self::refused(&a.actor, &repo, Capability::ManageIntegrations) {
1266	return Ok(refused);
1267	}
1268	let (least, most) = TAKE_OVER_AFTER_MINUTES;
1269	if a.settings.take_over_after.is_some_and(\|minutes\| minutes < least \|\| minutes > most) {
1270	return Ok(fail(FailureCode::Invalid, format!("Take over after between {least} minutes and {} hours.", most / 60)));
1271	}
1272	let settings = serde_json::to_string(&a.settings).unwrap_or_else(\|_\| "{}".to_owned());
1273	self.db
1274	.prepare("UPDATE remotes SET settings = ?2, recorded = 0 WHERE id = ?1")
1275	.bind(&[row.id.as_str().into(), settings.as_str().into()])?
1276	.run()
1277	.await?;
1278	let row = RemoteRow { settings, ..row };
1279	self.record(&row).await?;
1280	let hosts = self.hosts().await?;
1281	Ok(Outcome::Ok(row.contract(hosts.get(&row.host()))))
1282	}
1283
1284	pub async fn add(&self, a: MirrorAddArgs) -> Result<Outcome<Remote>> {
1285	if a.provider == RemoteProvider::Github {
1286	return Ok(fail(FailureCode::Invalid, "Link GitHub repositories through the GitHub App, from New → Import from GitHub."));
1287	}
1288	let Some(repo) = self.repo(&a.repo_id, Some(&a.actor)).await? else {
1289	return Ok(fail(FailureCode::NotFound, "Repository not found."));
1290	};
1291	if let Some(refused) = Self::refused(&a.actor, &repo, Capability::ManageIntegrations) {
1292	return Ok(refused);
1293	}
1294	let Some(clone_url) = clean_clone_url(&a.url) else {
1295	return Ok(fail(FailureCode::Invalid, "Give the remote's https address, such as https://g1t.sh/acme/web.git."));
1296	};
1297	let Some(token) = a.token.as_deref().map(str::trim).filter(\|token\| !token.is_empty()) else {
1298	return Ok(fail(FailureCode::Invalid, "Give a token that can read and push to the remote."));
1299	};
1300	let Some(sealer) = &self.sealer else {
1301	return Ok(fail(FailureCode::Conflict, "Tokens cannot be kept on this g1t: INTEGRATIONS_KEY is not set."));
1302	};
1303	if a.role == RemoteRole::Leader && self.leader(&repo.id).await?.is_some() {
1304	return Ok(fail(FailureCode::Conflict, "This repository already mirrors a remote. A repository follows one leader."));
1305	}
1306	let now = now_ms();
1307	let id = new_id("rmt", now);
1308	let row = RemoteRow {
1309	id: id.clone(),
1310	repo_id: repo.id.clone(),
1311	workspace: repo.namespace.clone(),
1312	repo: format!("{}/{}", repo.namespace, repo.name),
1313	provider: a.provider.as_str().to_owned(),
1314	role: a.role.as_str().to_owned(),
1315	name: display_name(&clone_url),
1316	url: web_url(&clone_url),
1317	clone_url: clone_url.clone(),
1318	connection_id: None,
1319	username: a.username.map(\|name\| name.trim().to_owned()).filter(\|name\| !name.is_empty()),
1320	credential: Some(sealer.seal(token, &format!("rmt:{id}"))),
1321	state: if a.role == RemoteRole::Leader { "standby" } else { "following" }.to_owned(),
1322	state_since: rfc3339(now),
1323	state_by: Some(a.actor.username.clone()),
1324	settings: "{}".to_owned(),
1325	synced_at: None,
1326	last_error: None,
1327	checks_error: None,
1328	created_at: rfc3339(now),
1329	};
1330	if row.name.eq_ignore_ascii_case(&format!("g1t.sh/{}", row.repo)) {
1331	return Ok(fail(FailureCode::Invalid, "That is this repository."));
1332	}
1333	// A leader fills an empty repository; one with history of its own
1334	// would lose it.
1335	if a.role == RemoteRole::Leader {
1336	let refs = match self.repos_refs(&row, row.username.clone(), token.to_owned()).await? {
1337	Outcome::Ok(refs) => refs,
1338	Outcome::Fail(failure) => return Ok(Outcome::Fail(failure)),
1339	};
1340	if !refs.ours.is_empty() {
1341	return Ok(fail(
1342	FailureCode::Conflict,
1343	"Only an empty repository can become a mirror: it is filled from the remote. Make a new repository for it.",
1344	));
1345	}
1346	if let Some(reason) = refs.unreachable {
1347	return Ok(fail(FailureCode::Unavailable, reason));
1348	}
1349	}
1350	self.db
1351	.prepare(
1352	"INSERT INTO remotes
1353	(id, repo_id, workspace, repo, provider, role, name, url, clone_url, username, credential,
1354	state, state_since, state_by, settings, created_by, created_at)
1355	VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, '{}', ?15, ?13)",
1356	)
1357	.bind(&[
1358	row.id.as_str().into(),
1359	row.repo_id.as_str().into(),
1360	row.workspace.as_str().into(),
1361	row.repo.as_str().into(),
1362	row.provider.as_str().into(),
1363	row.role.as_str().into(),
1364	row.name.as_str().into(),
1365	row.url.as_str().into(),
1366	row.clone_url.as_str().into(),
1367	null_or(row.username.as_deref()),
1368	null_or(row.credential.as_deref()),
1369	row.state.as_str().into(),
1370	row.state_since.as_str().into(),
1371	a.actor.username.as_str().into(),
1372	a.actor.id.as_str().into(),
1373	])?
1374	.run()
1375	.await?;
1376	self.record(&row).await?;
1377	let synced = self.sync(&row).await?;
1378	let row = self.row(&id).await?.unwrap_or(row);
1379	if let Err(reason) = synced {
1380	let hosts = self.hosts().await?;
1381	let mut remote = row.contract(hosts.get(&row.host()));
1382	remote.last_error = Some(reason);
1383	return Ok(Outcome::Ok(remote));
1384	}
1385	let hosts = self.hosts().await?;
1386	Ok(Outcome::Ok(row.contract(hosts.get(&row.host()))))
1387	}
1388
1389	pub async fn remove(&self, a: MirrorRemoveArgs) -> Result<Outcome<bool>> {
1390	let Some(row) = self.row(&a.remote_id).await? else {
1391	return Ok(Outcome::Ok(false));
1392	};
1393	let Some(repo) = self.repo(&row.repo_id, Some(&a.actor)).await? else {
1394	return Ok(fail(FailureCode::NotFound, "Repository not found."));
1395	};
1396	if let Some(refused) = Self::refused(&a.actor, &repo, Capability::ManageIntegrations) {
1397	return Ok(refused);
1398	}
1399	if row.leads() && matches!(row.state(), RemoteState::Takeover \| RemoteState::HandingBack) {
1400	return Ok(fail(
1401	FailureCode::Conflict,
1402	"g1t leads this repository for now. Hand it back, or move it to g1t, before unlinking.",
1403	));
1404	}
1405	if row.leads() {
1406	let done: Outcome<Repo> =
1407	g1t_kit::call(&self.repos, "set_mirror", &SetMirrorArgs { repo_id: row.repo_id.clone(), mirror: None }).await?;
1408	if let Outcome::Fail(failure) = done {
1409	return Ok(Outcome::Fail(failure));
1410	}
1411	}
1412	self.db.prepare("DELETE FROM remote_refs WHERE remote_id = ?").bind(&[row.id.as_str().into()])?.run().await?;
1413	self.db.prepare("DELETE FROM remotes WHERE id = ?").bind(&[row.id.as_str().into()])?.run().await?;
1414	if row.provider() == RemoteProvider::Github {
1415	self.db
1416	.prepare("UPDATE github_repos SET mode = 'import' WHERE repo_id = ?")
1417	.bind(&[row.repo_id.as_str().into()])?
1418	.run()
1419	.await?;
1420	}
1421	Ok(Outcome::Ok(true))
1422	}
1423
1424	/// A remote g1t can no longer reach for good (uninstalled, deleted). A
1425	/// mirror standing by becomes an ordinary repository with what it has,
1426	/// since it can no longer follow; a follower is let go; a takeover
1427	/// keeps going and is told why, so nothing is lost.
1428	pub async fn link_gone(&self, id: &str, why: &str) -> Result<()> {
1429	let Some(row) = self.row(id).await? else { return Ok(()) };
1430	if row.leads() && matches!(row.state(), RemoteState::Takeover \| RemoteState::HandingBack) {
1431	let reason = format!("{} {why}. g1t keeps leading; move it to g1t to keep it here.", row.name);
1432	return self.note(&row, Some(&reason)).await;
1433	}
1434	if row.leads() {
1435	let _: Outcome<Repo> =
1436	g1t_kit::call(&self.repos, "set_mirror", &SetMirrorArgs { repo_id: row.repo_id.clone(), mirror: None }).await?;
1437	}
1438	self.db.prepare("DELETE FROM remote_refs WHERE remote_id = ?").bind(&[row.id.as_str().into()])?.run().await?;
1439	self.db.prepare("DELETE FROM remotes WHERE id = ?").bind(&[row.id.as_str().into()])?.run().await?;
1440	let title = if row.leads() {
1441	format!("{} stopped mirroring {}: it {why}. g1t keeps what it has.", row.repo, row.name)
1442	} else {
1443	format!("{} stopped pushing to {}: it {why}.", row.repo, row.name)
1444	};
1445	let notify = self.told(&row, None).await;
1446	self.publish("mirror.moved_in", &row, Some("g1t"), title, None, notify).await;
1447	Ok(())
1448	}
1449
1450	// --- What hosts and g1t say -----------------------------------------------
1451
1452	/// A push on GitHub, from the App's webhook.
1453	pub async fn on_github_push(&self, payload: &Value) -> Result<()> {
1454	let Some(id) = payload["repository"]["id"].as_u64() else { return Ok(()) };
1455	let rows = self
1456	.db
1457	.prepare("SELECT * FROM remotes WHERE provider = 'github' AND external_id = ?")
1458	.bind(&[id.to_string().into()])?
1459	.all()
1460	.await?
1461	.results::<RemoteRow>()?;
1462	for row in rows {
1463	if let Err(error) = self.on_remote_push(&row, payload).await {
1464	worker::console_error!("mirrors: a push on {} was not followed: {error}", row.name);
1465	}
1466	}
1467	Ok(())
1468	}
1469
1470	/// A push on a remote: a mirror standing by catches up; during a
1471	/// takeover nothing moves (the hand-back will see it); a follower takes
1472	/// in fast-forwards, or overwrites, as its settings say.
1473	async fn on_remote_push(&self, row: &RemoteRow, payload: &Value) -> Result<()> {
1474	if row.leads() {
1475	if row.state().mirror().is_some_and(MirrorState::follows)
1476	&& let Err(reason) = self.sync(row).await?
1477	{
1478	worker::console_log!("mirrors: {} did not catch up with {}: {reason}", row.repo, row.name);
1479	}
1480	return Ok(());
1481	}
1482	let (Some(git_ref), Some(after)) = (payload["ref"].as_str(), payload["after"].as_str()) else {
1483	return Ok(());
1484	};
1485	let before = payload["before"].as_str().filter(\|hash\| !hash.chars().all(\|c\| c == '0'));
1486	let deleted = payload["deleted"].as_bool() == Some(true) \|\| after.chars().all(\|c\| c == '0');
1487	let (username, token) = match self.credential(row).await? {
1488	Ok(credential) => credential,
1489	Err(reason) => return self.note(row, Some(&reason)).await,
1490	};
1491	let refs = match self.repos_refs(row, username.clone(), token.clone()).await? {
1492	Outcome::Ok(refs) => refs,
1493	Outcome::Fail(failure) => return self.note(row, Some(&failure.message)).await,
1494	};
1495	let ours = refs.ours.get(git_ref).map(String::as_str);
1496	// g1t's own push coming back: nothing to do.
1497	if ours == Some(after) \|\| (deleted && ours.is_none()) {
1498	return Ok(());
1499	}
1500	if row.settings().remote_pushes == RemotePushes::Overwrite {
1501	let _ = self.sync(row).await?;
1502	return Ok(());
1503	}
1504	let forced = payload["forced"].as_bool() == Some(true);
1505	if !deleted && !forced && before == ours {
1506	let moves = vec![RefMove {
1507	direction: MirrorDirection::Pull,
1508	git_ref: git_ref.to_owned(),
1509	to: None,
1510	old: ours.map(str::to_owned),
1511	new: Some(after.to_owned()),
1512	}];
1513	if let Outcome::Ok(applied) = self.apply(row, username, token, moves).await?
1514	&& applied.moved.iter().all(\|moved\| moved.problem.is_none())
1515	{
1516	return self.note(row, None).await;
1517	}
1518	}
1519	let branch = branch_of(git_ref).unwrap_or(git_ref);
1520	let reason = format!(
1521	"{branch} changed on {} in a way g1t cannot follow. Sync to push g1t's {branch} over it, or bring that change into g1t first.",
1522	row.name
1523	);
1524	self.note(row, Some(&reason)).await?;
1525	self.set_state(row, RemoteState::Stuck, None).await?;
1526	Ok(())
1527	}
1528
1529	/// A push on g1t: each follower is sent it. A stuck follower waits for
1530	/// someone to sync it, so a change made on it is not pushed over.
1531	pub async fn on_event(&self, event: &g1t_contracts::events::Event) -> Result<()> {
1532	if event.kind != "git.push" {
1533	return Ok(());
1534	}
1535	let Some(repo_id) = event.repo_id.as_deref() else { return Ok(()) };
1536	for row in self.rows(repo_id).await? {
1537	if row.leads() \|\| row.state() != RemoteState::Following {
1538	continue;
1539	}
1540	if let Err(reason) = self.sync(&row).await? {
1541	worker::console_log!("mirrors: {} not pushed to {}: {reason}", row.repo, row.name);
1542	}
1543	}
1544	Ok(())
1545	}
1546
1547	/// A status set on a commit (`status.created`): a workflow run, a
1548	/// deploy. A mirror of a GitHub repository whose workflows g1t runs
1549	/// reports it to the commit there, once, as `g1t / <context>`. What
1550	/// GitHub refuses for good is kept on the link for its page; nothing is
1551	/// sent again until the next status.
1552	pub async fn on_status(&self, event: &g1t_contracts::events::Event) -> Result<()> {
1553	let Some(repo_id) = event.repo_id.as_deref() else { return Ok(()) };
1554	let Ok(status) = serde_json::from_value::<StatusEvent>(event.data.clone()) else { return Ok(()) };
1555	let (Some(state), true) = (github_state(&status.state), whole_commit(&status.sha)) else { return Ok(()) };
1556	let Some(row) = self.leader(repo_id).await? else { return Ok(()) };
1557	if !reports_checks(row.provider(), row.leads(), row.state(), &row.settings()) {
1558	return Ok(());
1559	}
1560	let Some(installation) = row.connection_id.as_deref().and_then(\|id\| id.parse::<u64>().ok()) else {
1561	let problem = "g1t can't report checks to GitHub: this link has no GitHub installation.".to_owned();
1562	return self.checks_noted(&row, Reported::Problem(problem)).await;
1563	};
1564	let token = match self.github.installation_token(installation).await? {
1565	Ok(token) => token,
1566	Err(reason) => {
1567	return self.checks_noted(&row, Reported::Problem(format!("g1t can't report checks to GitHub: {reason}"))).await;
1568	}
1569	};
1570	let mut body = serde_json::json!({
1571	"state": state,
1572	"context": github_context(&status.context),
1573	"target_url": status_target(&self.site_url, &row.repo, &status.sha, status.target_url.as_deref()),
1574	});
1575	if let Some(description) = github_description(status.description.as_deref()) {
1576	body["description"] = Value::String(description);
1577	}
1578	let path = format!("/repos/{}/statuses/{}", row.full_name(), status.sha);
1579	let answer = match self.github.api(Method::Post, &path, &token, Some(body)).await {
1580	Ok(answer) => answer,
1581	Err(error) => {
1582	worker::console_log!("mirrors: {} check not reported to {}: {error}", row.repo, row.name);
1583	return Ok(());
1584	}
1585	};
1586	let said = answer.json();
1587	let verdict = reported(answer.status, said["message"].as_str().unwrap_or_default(), &answer.problem("GitHub"));
1588	if verdict == Reported::Keep && !answer.ok() {
1589	worker::console_log!("mirrors: {} check not reported to {}: {}", row.repo, row.name, answer.problem("GitHub"));
1590	}
1591	self.checks_noted(&row, verdict).await
1592	}
1593
1594	/// Keeps what the last report to a remote left, writing only a change.
1595	async fn checks_noted(&self, row: &RemoteRow, verdict: Reported) -> Result<()> {
1596	let next = match verdict {
1597	Reported::Keep => return Ok(()),
1598	Reported::Clear => None,
1599	Reported::Problem(problem) => Some(problem),
1600	};
1601	if next == row.checks_error {
1602	return Ok(());
1603	}
1604	self.db
1605	.prepare("UPDATE remotes SET checks_error = ?2 WHERE id = ?1")
1606	.bind(&[row.id.as_str().into(), null_or(next.as_deref())])?
1607	.run()
1608	.await?;
1609	Ok(())
1610	}
1611
1612	/// Records one check of a host, and acts when it goes down or comes back.
1613	async fn host_checked(&self, host: &str, answered: bool, problem: Option<&str>) -> Result<()> {
1614	let now = now_ms();
1615	let row = self
1616	.db
1617	.prepare("SELECT * FROM remote_hosts WHERE host = ?")
1618	.bind(&[host.into()])?
1619	.first::<HostRow>(None)
1620	.await?
1621	.unwrap_or_else(\|\| HostRow { host: host.to_owned(), ..HostRow::default() });
1622	let (next, change) = row.health().checked(answered, now);
1623	self.db
1624	.prepare(
1625	"INSERT INTO remote_hosts (host, failures, successes, streak_ms, unreachable_since, checked_ms, last_problem)
1626	VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)
1627	ON CONFLICT (host) DO UPDATE SET failures = excluded.failures, successes = excluded.successes,
1628	streak_ms = excluded.streak_ms, unreachable_since = excluded.unreachable_since,
1629	checked_ms = excluded.checked_ms, last_problem = COALESCE(excluded.last_problem, remote_hosts.last_problem)",
1630	)
1631	.bind(&[
1632	host.into(),
1633	next.failures.into(),
1634	next.successes.into(),
1635	(next.streak_ms as f64).into(),
1636	null_or(next.unreachable_since.map(rfc3339).as_deref()),
1637	(now as f64).into(),
1638	null_or(problem),
1639	])?
1640	.run()
1641	.await?;
1642	if change == Change::None {
1643	return Ok(());
1644	}
1645	let rows = self
1646	.db
1647	.prepare("SELECT * FROM remotes WHERE role = 'leader'")
1648	.all()
1649	.await?
1650	.results::<RemoteRow>()?;
1651	for row in rows.into_iter().filter(\|row\| row.host() == host) {
1652	let (kind, title) = match change {
1653	Change::WentDown => ("mirror.unreachable", format!("{} is not answering. {} keeps its copy.", row.name, row.repo)),
1654	_ => ("mirror.reachable", format!("{} answers again", row.name)),
1655	};
1656	let notify = self.told(&row, None).await;
1657	let detail = (change == Change::WentDown && row.state().mirror().is_some_and(MirrorState::follows))
1658	.then(\|\| format!("Take over {} to keep working on g1t until it is back.", row.repo));
1659	self.publish(kind, &row, None, title, detail, notify).await;
1660	}
1661	Ok(())
1662	}
1663
1664	// --- Every minute -----------------------------------------------------------
1665
1666	pub async fn on_minute(&self) -> Result<()> {
1667	let now = now_ms();
1668	// Mirrors the repos service has not heard of yet.
1669	let unrecorded = self
1670	.db
1671	.prepare("SELECT * FROM remotes WHERE recorded = 0 LIMIT 50")
1672	.all()
1673	.await?
1674	.results::<RemoteRow>()?;
1675	for row in &unrecorded {
1676	if row.leads() {
1677	self.record(row).await?;
1678	} else {
1679	self.db.prepare("UPDATE remotes SET recorded = 1 WHERE id = ?").bind(&[row.id.as_str().into()])?.run().await?;
1680	}
1681	}
1682	let leaders = self
1683	.db
1684	.prepare("SELECT * FROM remotes WHERE role = 'leader'")
1685	.all()
1686	.await?
1687	.results::<RemoteRow>()?;
1688	// One check per host a mirror follows.
1689	let mut by_host: BTreeMap<String, &RemoteRow> = BTreeMap::new();
1690	for row in &leaders {
1691	by_host.entry(row.host()).or_insert(row);
1692	}
1693	for (host, row) in &by_host {
1694	let answered = self.probe(row).await?;
1695	self.host_checked(host, answered.is_ok(), answered.err().as_deref()).await?;
1696	}
1697	let hosts = self.hosts().await?;
1698	for row in &leaders {
1699	let health = hosts.get(&row.host()).map(HostRow::health).unwrap_or_default();
1700	let settings = row.settings();
1701	match row.state() {
1702	// Taking over on its own, only when asked to.
1703	RemoteState::Standby \| RemoteState::Ci => {
1704	if let (Some(minutes), Some(since)) = (settings.take_over_after, health.unreachable_since)
1705	&& now.saturating_sub(since) >= u64::from(minutes) * 60_000
1706	{
1707	if let Outcome::Fail(failure) = self.start_takeover(row.clone(), By::G1t).await? {
1708	worker::console_log!("mirrors: {} not taken over: {}", row.repo, failure.message);
1709	}
1710	continue;
1711	}
1712	// Hosts with no webhook are polled.
1713	if row.provider() != RemoteProvider::Github && health.reachable() {
1714	let polled = self
1715	.db
1716	.prepare("UPDATE remotes SET polled_ms = ?2 WHERE id = ?1 AND polled_ms < ?3 RETURNING id")
1717	.bind(&[row.id.as_str().into(), (now as f64).into(), (now.saturating_sub(POLL_MS) as f64).into()])?
1718	.first::<Value>(None)
1719	.await?;
1720	if polled.is_some() {
1721	let _ = self.sync(row).await?;
1722	}
1723	}
1724	}
1725	// Handing back on its own once the remote is back and it is
1726	// clean, when asked to.
1727	RemoteState::Takeover if health.reachable() && settings.hand_back == HandBack::WhenClean && row.state_by.as_deref() == Some("g1t") => {
1728	if let Outcome::Ok(plan) = self.plan(row).await?
1729	&& plan.clean()
1730	&& let Outcome::Fail(failure) = self.finish_takeover(row.clone(), By::G1t).await?
1731	{
1732	worker::console_log!("mirrors: {} not handed back: {}", row.repo, failure.message);
1733	}
1734	}
1735	_ => {}
1736	}
1737	}
1738	Ok(())
1739	}
1740
1741	/// Whether a remote's host answers `info/refs`: any answer but a server
1742	/// error counts, since a refused credential is the link's problem, not
1743	/// the host's.
1744	async fn probe(&self, row: &RemoteRow) -> Result<std::result::Result<(), String>> {
1745	let (username, token) = match self.credential(row).await? {
1746	Ok(credential) => credential,
1747	// No credential: ask without one; the host still answers.
1748	Err(_) => (None, String::new()),
1749	};
1750	let authorization = if token.is_empty() {
1751	String::new()
1752	} else {
1753	let user = username.unwrap_or_else(\|\| "x-access-token".to_owned());
1754	format!("Basic {}", base64(&format!("{user}:{token}")))
1755	};
1756	let url = format!("{}/info/refs?service=git-upload-pack", row.clone_url);
1757	let mut headers = vec![("accept", "/")];
1758	if !authorization.is_empty() {
1759	headers.push(("authorization", authorization.as_str()));
1760	}
1761	match http::send(Method::Get, &url, &headers, None).await {
1762	Ok(answer) if answer.status < 500 && answer.status != 429 && answer.status != 408 => Ok(Ok(())),
1763	Ok(answer) => Ok(Err(format!("{} answered {}.", row.host(), answer.status))),
1764	Err(_) => Ok(Err(format!("{} could not be reached.", row.host()))),
1765	}
1766	}
1767	}
1768
1769	fn base64(text: &str) -> String {
1770	const ALPHABET: &[u8; 64] = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/";
1771	let bytes = text.as_bytes();
1772	let mut out = String::with_capacity(bytes.len().div_ceil(3) * 4);
1773	for chunk in bytes.chunks(3) {
1774	let n = (chunk[0] as u32) << 16 \| (chunk.get(1).unwrap_or(&0) as u32) << 8 \| chunk.get(2).unwrap_or(&0) as u32;
1775	for (i, shift) in [18, 12, 6, 0].into_iter().enumerate() {
1776	out.push(if i <= chunk.len() { ALPHABET[(n >> shift) as usize & 63] as char } else { '=' });
1777	}
1778	}
1779	out
1780	}
1781
1782	/// When the remote is away, the plan assumes it is where the takeover
1783	/// started.
1784	fn bases_as_theirs(bases: &BTreeMap<String, (Option<String>, Option<RefDecision>)>) -> BTreeMap<String, String> {
1785	bases.iter().filter_map(\|(name, (base, _))\| base.clone().map(\|base\| (name.clone(), base))).collect()
1786	}
1787
1788	fn undecided(plan: &HandbackPlan) -> g1t_contracts::Failure {
1789	let waiting: Vec<&str> = plan
1790	.refs
1791	.iter()
1792	.filter(\|item\| item.action == RefAction::Diverged && item.decision.is_none())
1793	.map(\|item\| branch_of(&item.git_ref).unwrap_or(&item.git_ref))
1794	.collect();
1795	g1t_contracts::Failure {
1796	code: FailureCode::Conflict,
1797	message: format!(
1798	"Changed on both sides: {}. Decide for each whether to keep g1t's, keep the remote's, or send g1t's as a pull request.",
1799	waiting.join(", ")
1800	),
1801	}
1802	}
1803
1804	/// Answers the `mirror_*` methods; `None` for any other.
1805	pub async fn route(method: &str, body: &Value, env: &Env, _ctx: &Context) -> Option<Result<Response>> {
1806	if !method.starts_with("mirror_") {
1807	return None;
1808	}
1809	Some(handle(method, body.clone(), env).await)
1810	}
1811
1812	async fn handle(method: &str, body: Value, env: &Env) -> Result<Response> {
1813	let mirrors = Mirrors::new(env)?;
1814	match method {
1815	"mirror_view" => reply(&mirrors.view(args(body)?).await?),
1816	"mirror_briefs" => reply(&mirrors.briefs(args(body)?).await?),
1817	"mirror_take_over" => reply(&mirrors.take_over(args(body)?).await?),
1818	"mirror_ci" => reply(&mirrors.ci(args(body)?).await?),
1819	"mirror_hand_back_plan" => reply(&mirrors.hand_back_plan(args(body)?).await?),
1820	"mirror_hand_back" => reply(&mirrors.hand_back(args(body)?).await?),
1821	"mirror_move_in" => reply(&mirrors.move_in(args(body)?).await?),
1822	"mirror_sync" => reply(&mirrors.sync_now(args(body)?).await?),
1823	"mirror_settings" => reply(&mirrors.settings(args(body)?).await?),
1824	"mirror_add" => reply(&mirrors.add(args(body)?).await?),
1825	"mirror_remove" => reply(&mirrors.remove(args(body)?).await?),
1826	_ => Response::error("Unknown method", 404),
1827	}
1828	}
1829
1830	#[cfg(test)]
1831	mod tests {
1832	use super::*;
1833
1834	#[test]
1835	fn remotes_are_named_as_people_say_them() {
1836	assert_eq!(host_of("https://GitHub.com/acme/web.git"), "github.com");
1837	assert_eq!(host_of("https://user@git.acme.internal/x"), "git.acme.internal");
1838	assert_eq!(display_name("https://github.com/acme/web.git"), "github.com/acme/web");
1839	assert_eq!(display_name("https://g1t.sh/acme/web/"), "g1t.sh/acme/web");
1840	assert_eq!(clean_clone_url("https://g1t.sh/acme/web").as_deref(), Some("https://g1t.sh/acme/web.git"));
1841	assert_eq!(clean_clone_url("https://git.example/a.git/").as_deref(), Some("https://git.example/a.git"));
1842	assert_eq!(clean_clone_url("http://git.example/a"), None, "https only");
1843	assert_eq!(clean_clone_url("https://me:secret@git.example/a"), None, "no credentials in the address");
1844	assert_eq!(clean_clone_url("https://git.example"), None);
1845	assert_eq!(web_url("https://github.com/acme/web.git"), "https://github.com/acme/web");
1846	}
1847
1848	#[test]
1849	fn a_host_goes_down_slowly_and_comes_back_slower() {
1850	let minute = 60_000;
1851	let mut health = HostHealth::default();
1852	let mut changes = Vec::new();
1853	for at in [0, minute, 2 * minute] {
1854	let (next, change) = health.checked(false, at);
1855	health = next;
1856	changes.push(change);
1857	}
1858	assert_eq!(changes, [Change::None, Change::None, Change::WentDown], "three failures over two minutes");
1859	assert!(!health.reachable());
1860	// One blip does not bring it back, nor do three quick answers.
1861	let (next, change) = health.checked(true, 3 * minute);
1862	assert_eq!(change, Change::None);
1863	let (next, _) = next.checked(true, 3 * minute + 1);
1864	let (next, change) = next.checked(true, 3 * minute + 2);
1865	assert_eq!(change, Change::None, "three answers in a moment are not five minutes");
1866	let (back, change) = next.checked(true, 8 * minute);
1867	assert_eq!(change, Change::CameBack);
1868	assert!(back.reachable());
1869	// A failure in between starts the count again.
1870	let (flaky, _) = HostHealth::default().checked(false, 0);
1871	let (flaky, _) = flaky.checked(true, minute);
1872	let (flaky, _) = flaky.checked(false, 2 * minute);
1873	let (_, change) = flaky.checked(false, 3 * minute);
1874	assert_eq!(change, Change::None);
1875	}
1876
1877	fn refs(pairs: &[(&str, &str)]) -> BTreeMap<String, String> {
1878	pairs.iter().map(\|(name, hash)\| (name.to_string(), hash.to_string())).collect()
1879	}
1880
1881	#[test]
1882	fn a_hand_back_plan_reads_each_ref_against_where_the_takeover_began() {
1883	let bases: BTreeMap<_, _> = [
1884	("refs/heads/main", "a"),
1885	("refs/heads/fix", "f"),
1886	("refs/heads/docs", "d"),
1887	("refs/heads/quiet", "q"),
1888	]
1889	.into_iter()
1890	.map(\|(name, base)\| (name.to_owned(), (Some(base.to_owned()), None)))
1891	.collect();
1892	let ours = refs(&[
1893	("refs/heads/main", "a2"),
1894	("refs/heads/fix", "f2"),
1895	("refs/heads/docs", "d2"),
1896	("refs/heads/quiet", "q"),
1897	("refs/heads/agent", "n"),
1898	]);
1899	let theirs = refs(&[
1900	("refs/heads/main", "a"),
1901	("refs/heads/fix", "f"),
1902	("refs/heads/docs", "d3"),
1903	("refs/heads/quiet", "q"),
1904	("refs/heads/g1t/handback/old", "x"),
1905	]);
1906	let protected = BTreeSet::from(["refs/heads/main".to_owned()]);
1907	let plan = plan_refs(&bases, &ours, &theirs, &protected);
1908	let actions: Vec<(&str, RefAction)> = plan.iter().map(\|item\| (item.git_ref.as_str(), item.action)).collect();
1909	assert_eq!(actions, [
1910	("refs/heads/agent", RefAction::Push),
1911	("refs/heads/docs", RefAction::Diverged),
1912	("refs/heads/fix", RefAction::Push),
1913	("refs/heads/main", RefAction::PullRequest),
1914	]);
1915	let plan = HandbackPlan::new(plan, true);
1916	assert!(!plan.ready, "docs needs a decision");
1917
1918	let (moves, pull_requests) = hand_back_moves(&plan, &theirs);
1919	assert_eq!(pull_requests, ["refs/heads/main"]);
1920	let summary: Vec<(MirrorDirection, &str, Option<&str>)> =
1921	moves.iter().map(\|m\| (m.direction, m.git_ref.as_str(), m.to.as_deref())).collect();
1922	assert_eq!(summary, [
1923	(MirrorDirection::Push, "refs/heads/agent", None),
1924	(MirrorDirection::Push, "refs/heads/fix", None),
1925	(MirrorDirection::Push, "refs/heads/main", Some("refs/heads/g1t/handback/main")),
1926	(MirrorDirection::Pull, "refs/heads/main", None),
1927	], "an undecided ref does not move");
1928	let main_back = &moves[3];
1929	assert_eq!((main_back.old.as_deref(), main_back.new.as_deref()), (Some("a2"), Some("a")), "g1t follows the remote's main");
1930	let push_fix = &moves[1];
1931	assert_eq!((push_fix.old.as_deref(), push_fix.new.as_deref()), (Some("f"), Some("f2")), "only from where the remote is");
1932	}
1933
1934	#[test]
1935	fn decisions_settle_a_diverged_ref() {
1936	let plan = \|decision\| {
1937	HandbackPlan::new(
1938	vec![RefPlan {
1939	git_ref: "refs/heads/docs".into(),
1940	base: Some("d".into()),
1941	ours: Some("d2".into()),
1942	theirs: Some("d3".into()),
1943	action: RefAction::Diverged,
1944	decision: Some(decision),
1945	}],
1946	true,
1947	)
1948	};
1949	let theirs = refs(&[("refs/heads/g1t/handback/docs", "old")]);
1950	let (moves, _) = hand_back_moves(&plan(RefDecision::KeepOurs), &theirs);
1951	assert_eq!((moves[0].direction, moves[0].old.as_deref(), moves[0].new.as_deref()), (MirrorDirection::Push, Some("d3"), Some("d2")));
1952	let (moves, _) = hand_back_moves(&plan(RefDecision::KeepTheirs), &theirs);
1953	assert_eq!((moves[0].direction, moves[0].old.as_deref(), moves[0].new.as_deref()), (MirrorDirection::Pull, Some("d2"), Some("d3")));
1954	let (moves, pull_requests) = hand_back_moves(&plan(RefDecision::PullRequest), &theirs);
1955	assert_eq!(pull_requests, ["refs/heads/docs"]);
1956	assert_eq!(moves[0].old.as_deref(), Some("old"), "an old hand-back branch is moved from where it is");
1957	assert_eq!(moves.len(), 2);
1958	}
1959
1960	#[test]
1961	fn checks_are_named_for_g1t_on_github() {
1962	assert_eq!(github_context("CI / push"), "g1t / CI / push");
1963	assert_eq!(github_context("g1t / deploy"), "g1t / deploy", "already g1t's");
1964	assert_eq!(github_context(" Lint / pull_request "), "g1t / Lint / pull_request");
1965	assert_eq!(github_context("g1t/deploy"), "g1t / g1t/deploy", "only the exact prefix counts");
1966	for state in ["pending", "success", "failure", "error"] {
1967	assert_eq!(github_state(state), Some(state));
1968	}
1969	assert_eq!(github_state("cancelled"), None);
1970	assert_eq!(github_state(""), None);
1971	}
1972
1973	#[test]
1974	fn only_a_github_mirror_running_workflows_reports_checks() {
1975	let on = MirrorSettings::default();
1976	let warm = MirrorSettings { keep_ci_warm: true, ..MirrorSettings::default() };
1977	let off = MirrorSettings { report_checks: false, keep_ci_warm: true, ..MirrorSettings::default() };
1978	let github = RemoteProvider::Github;
1979	assert!(!reports_checks(github, true, RemoteState::Standby, &on), "a quiet standby runs nothing");
1980	assert!(reports_checks(github, true, RemoteState::Standby, &warm), "CI kept warm");
1981	for state in [RemoteState::Ci, RemoteState::Takeover, RemoteState::HandingBack] {
1982	assert!(reports_checks(github, true, state, &on), "{state:?}");
1983	assert!(!reports_checks(github, true, state, &off), "{state:?} turned off");
1984	assert!(!reports_checks(RemoteProvider::G1t, true, state, &on), "{state:?} on another g1t");
1985	assert!(!reports_checks(RemoteProvider::Git, true, state, &on), "{state:?} on a git host");
1986	}
1987	assert!(!reports_checks(github, false, RemoteState::Following, &warm), "a follower is g1t's own");
1988	assert!(!reports_checks(github, true, RemoteState::Stuck, &warm));
1989	}
1990
1991	#[test]
1992	fn a_reported_check_links_back_to_g1t() {
1993	let sha = "a".repeat(40);
1994	assert_eq!(
1995	status_target("https://g1t.sh", "acme/web", &sha, Some("https://g1t.sh/acme/web/actions/runs/run_1")),
1996	"https://g1t.sh/acme/web/actions/runs/run_1"
1997	);
1998	assert_eq!(status_target("https://g1t.sh/", "acme/web", &sha, Some("/acme/web/deployments")), "https://g1t.sh/acme/web/deployments");
1999	assert_eq!(status_target("https://g1t.sh", "acme/web", &sha, None), format!("https://g1t.sh/acme/web/commit/{sha}"));
2000	assert_eq!(status_target("https://g1t.sh", "acme/web", &sha, Some(" ")), format!("https://g1t.sh/acme/web/commit/{sha}"));
2001	assert!(whole_commit(&sha) && !whole_commit("abc123") && !whole_commit(&"z".repeat(40)));
2002	assert_eq!(github_description(Some("CI passed")).as_deref(), Some("CI passed"));
2003	assert_eq!(github_description(Some(" ")), None);
2004	let long = github_description(Some(&"word ".repeat(60))).unwrap();
2005	assert!(long.chars().count() <= 140 && long.ends_with('…'));
2006	}
2007
2008	#[test]
2009	fn what_github_answers_is_kept_only_when_it_lasts() {
2010	assert_eq!(reported(201, "", ""), Reported::Clear);
2011	assert_eq!(reported(403, "Resource not accessible by integration", "x"), Reported::Problem(CHECKS_PERMISSION.to_owned()));
2012	assert_eq!(reported(403, "API rate limit exceeded for installation", "x"), Reported::Keep);
2013	assert_eq!(reported(422, "No commit found for SHA: abc", "x"), Reported::Keep, "a commit made on g1t");
2014	assert_eq!(reported(502, "", "x"), Reported::Keep);
2015	assert_eq!(
2016	reported(404, "Not Found", "GitHub has nothing there (Not Found)."),
2017	Reported::Problem("g1t can't report checks to GitHub: GitHub has nothing there (Not Found).".to_owned())
2018	);
2019	}
2020	}