flagon-io/g1t

public

1	//! The work service: issues, pull requests, comments and sessions.
2	//!
3	//! Other services reach it over `POST /rpc/<method>`; see
4	//! `g1t_contracts::work` for the methods and their arguments. It also
5	//! consumes its queue of events from the bus.
6
7	mod authored;
8	mod checks;
9	mod ghost;
10	mod codeowners;
11	mod commit_checks;
12	mod contributions;
13	mod coordination;
14	mod decisions;
15	mod guardrails;
16	mod inbox;
17	mod integration;
18	mod labels;
19	mod lifecycle;
20	mod mergeability;
21	mod milestones;
22	mod prefetch;
23	mod queue;
24	mod retired;
25	mod reviews;
26	mod rows;
27	mod rulesets;
28	mod runs;
29	mod settings;
30	mod statuses;
31	mod team_reviews;
32
33	use g1t_contracts::events::{
34	ChangedFrom, CommentChanges, CommentCreated, CommentDeleted, CommentEdited, DeletedComment, Event, IssueEvent,
35	NewEvent, Publish, PullEvent, SessionAppended,
36	};
37	use g1t_contracts::identity::UsernameArgs;
38	use g1t_contracts::repos::{
39	ForkArgs, GetArgs, HeadArgs, LandArgs, Landed, NeedsAgentReason, PullBranchUpdate, ReadableArgs, Repo, RepoPath,
40	UpdatePullBranchArgs,
41	};
42	use g1t_contracts::access::{self, Capability, Denied};
43	use g1t_contracts::time::rfc3339;
44	use g1t_contracts::work::*;
45	use futures_util::future::{try_join, try_join3, try_join_all};
46	use g1t_contracts::{FailureCode, Outcome, User, Viewer, new_id};
47	use g1t_kit::{args, now_ms, reply, rpc_method};
48	use serde::Serialize;
49	use worker::wasm_bindgen::JsValue;
50	use worker::{
51	Context, D1Database, Env, Fetcher, MessageBatch, MessageExt, Request, Response, Result, event,
52	};
53
54	use retired::writable;
55	use rows::{CommentRow, IssueRow, MovedRow, NumberRow, PULL_COLUMNS, PullRow, SessionRow, Snapshot};
56
57	pub(crate) const SOURCE: &str = "work";
58	const MAX_ENTRY_BATCH: usize = 200;
59	pub(crate) const MAX_ENTRY_CHARS: usize = 64_000;
60	const MAX_TITLE_CHARS: usize = 200;
61	const SESSION_PAGE: u32 = 500;
62	const LIST_PAGE: u32 = 100;
63	const MAX_ASSIGNEES: usize = 10;
64	pub(crate) const UNVERIFIED: &str = "Confirm your email address first. Check your inbox, or resend the link from the banner on g1t.sh.";
65
66	const ISSUE_COLUMNS: &str = "issues.*,
67	(SELECT count(*) FROM pulls WHERE pulls.issue_id = issues.id) AS pull_count,
68	(SELECT agent FROM pulls
69	WHERE pulls.issue_id = issues.id AND pulls.status IN ('draft', 'open')
70	AND pulls.fork_repo_id IS NOT NULL
71	ORDER BY pulls.number DESC LIMIT 1) AS agent,
72	(SELECT count(*) FROM comments
73	WHERE comments.repo_id = issues.repo_id AND comments.number = issues.number
74	AND comments.kind = 'comment') AS comment_count,
75	(SELECT title FROM milestones
76	WHERE milestones.repo_id = issues.repo_id AND milestones.number = issues.milestone) AS milestone_title";
77
78	fn no_issue<T>() -> Outcome<T> {
79	Outcome::fail(FailureCode::NotFound, "Issue not found.")
80	}
81
82	fn no_pull<T>() -> Outcome<T> {
83	Outcome::fail(FailureCode::NotFound, "Pull request not found.")
84	}
85
86	/// Whether a pull request was behind when its mergeability was last
87	/// worked out, and for which pair of commits (mergeability.rs).
88	#[derive(serde::Deserialize)]
89	struct StoredBehind {
90	#[serde(default)]
91	behind: Option<u8>,
92	#[serde(default)]
93	mergeable_key: Option<String>,
94	}
95
96	impl StoredBehind {
97	/// The stored answer, if it was worked out for `head`.
98	fn for_head(&self, head: Option<&str>) -> Option<bool> {
99	let (worked_for, _) = self.mergeable_key.as_deref()?.split_once("..")?;
100	(Some(worked_for) == head).then_some(self.behind? != 0)
101	}
102	}
103
104	/// Refuses `actor` unless their role on `repo` has `capability`: not found
105	/// when they cannot read it, forbidden with the role it needs otherwise.
106	pub(crate) fn allowed(actor: Option<&User>, repo: &Repo, capability: Capability) -> Outcome<()> {
107	match access::check(actor, repo, capability) {
108	Ok(()) => Outcome::Ok(()),
109	Err(Denied::NotFound) => Outcome::fail(FailureCode::NotFound, "Repository not found."),
110	Err(Denied::Forbidden) => Outcome::fail(
111	FailureCode::Forbidden,
112	access::needs(capability, &format!("{}/{}", repo.namespace, repo.name)),
113	),
114	}
115	}
116
117	fn optional(value: &Option<String>) -> JsValue {
118	value.as_deref().map_or(JsValue::NULL, JsValue::from)
119	}
120
121	fn optional_number(value: Option<u32>) -> JsValue {
122	value.map_or(JsValue::NULL, JsValue::from)
123	}
124
125	/// Names the branch a pull request merges into when it is the default
126	/// branch, which is stored as none so that it follows a change of default.
127	pub(crate) fn fill_base(pull: &mut Pull, repo: &Repo) {
128	if pull.base.as_deref().is_none_or(str::is_empty) {
129	pull.base = Some(repo.default_branch.clone());
130	}
131	}
132
133	/// The branch a pull request is stored as merging into: none for the
134	/// default branch.
135	fn stored_base(base: &str, repo: &Repo) -> Option<String> {
136	let base = base.trim();
137	(!base.is_empty() && base != repo.default_branch).then(\|\| base.to_owned())
138	}
139
140	/// The lowercase name a `State` is stored and sent as.
141	fn state_name(state: Option<State>) -> Option<&'static str> {
142	state.map(\|state\| match state {
143	State::Open => "open",
144	State::Closed => "closed",
145	})
146	}
147
148	/// A trimmed title, or why it cannot be used.
149	fn valid_title(title: &str) -> std::result::Result<&str, &'static str> {
150	let title = title.trim();
151	if title.is_empty() {
152	Err("A title is required.")
153	} else if title.chars().count() > MAX_TITLE_CHARS {
154	Err("That title is too long.")
155	} else {
156	Ok(title)
157	}
158	}
159
160	/// What a closed pull request was when it was closed.
161	#[derive(serde::Deserialize)]
162	struct ClosedFrom {
163	#[serde(default)]
164	closed_from: Option<String>,
165	}
166
167	/// Why a pull request in `status` cannot be reopened, if it cannot: only a
168	/// closed one can, never a merged one.
169	fn reopen_refusal(status: PullStatus) -> Option<&'static str> {
170	match status {
171	PullStatus::Closed => None,
172	PullStatus::Merged => Some("This pull request was merged; it cannot be reopened."),
173	PullStatus::Draft \| PullStatus::Open => Some("This pull request is already open."),
174	}
175	}
176
177	/// What a closed pull request is reopened as: the draft it was, when it
178	/// was closed as one, else ready for review.
179	fn reopened_status(closed_from: Option<&str>) -> PullStatus {
180	if closed_from == Some("draft") { PullStatus::Draft } else { PullStatus::Open }
181	}
182
183	/// Why a pull request in `status` cannot be turned into a draft, if it
184	/// cannot: only one that is open, ready for review, can.
185	fn draft_refusal(status: PullStatus) -> Option<&'static str> {
186	match status {
187	PullStatus::Open => None,
188	PullStatus::Draft => Some("This pull request is already a draft."),
189	PullStatus::Merged => Some("This pull request is already merged."),
190	PullStatus::Closed => Some("This pull request is closed. Reopen it first."),
191	}
192	}
193
194	/// Unwraps an `Outcome`, returning its failure from the enclosing method.
195	macro_rules! check {
196	($outcome:expr) => {
197	match $outcome {
198	Outcome::Ok(value) => value,
199	Outcome::Fail(failure) => return Ok(Outcome::Fail(failure)),
200	}
201	};
202	}
203
204	struct Work {
205	db: D1Database,
206	identity: Fetcher,
207	repos: Fetcher,
208	events: Fetcher,
209	/// GitHub Actions: runs a merge queue's `merge_group` workflows.
210	actions: Fetcher,
211	/// Where this request's time went, for its `Server-Timing`.
212	timing: g1t_kit::d1::Timing,
213	/// A pull request's rows read in one batch for this request
214	/// (prefetch.rs), which the helpers below read instead of the database.
215	prefetched: std::cell::RefCell<Option<std::rc::Rc<prefetch::Prefetched>>>,
216	/// Repositories read in this request, by id, for rules that need one
217	/// where only a pull request is at hand (rulesets.rs `repo_for`).
218	known_repos: std::cell::RefCell<std::collections::HashMap<String, Repo>>,
219	}
220
221	impl Work {
222	async fn publish<T: Serialize>(
223	&self,
224	kind: &'static str,
225	repo_id: &str,
226	actor: &User,
227	data: T,
228	) -> Result<()> {
229	// What a workflow job's token did is marked, so it starts no
230	// workflows (`g1t_contracts::events::CAUSED_BY_JOB`).
231	self.publish_as(kind, repo_id, Some(actor.id.clone()), g1t_contracts::events::marked(data, Some(actor)))
232	.await
233	}
234
235	/// A pull request's owner (whoever asked g1t for it, or its author) as
236	/// a viewer who can read its repository and source. Stored people carry
237	/// no memberships, so a private repository would otherwise look missing
238	/// to them. The membership given reads and nothing more: it is for
239	/// looking, never for acting.
240	pub(crate) async fn owner_viewer(&self, pull: &Pull) -> Result<Viewer> {
241	let path: Option<RepoPath> = g1t_kit::call(
242	&self.repos,
243	"path_by_id",
244	&g1t_contracts::repos::PathByIdArgs { id: pull.repo_id.clone() },
245	)
246	.await?;
247	let mut owner = pull.owner().clone();
248	if let Some(path) = path
249	&& !owner.is_member(&path.namespace.to_lowercase())
250	{
251	owner.workspaces.push(g1t_contracts::Membership {
252	base_permission: Some(access::BasePermission::Read),
253	..g1t_contracts::Membership::member(path.namespace.to_lowercase())
254	});
255	}
256	Ok(Some(owner))
257	}
258
259	/// Publishes an event caused by `actor`, or by g1t itself.
260	async fn publish_as<T: Serialize>(
261	&self,
262	kind: &'static str,
263	repo_id: &str,
264	actor: Option<String>,
265	data: T,
266	) -> Result<()> {
267	let event = NewEvent {
268	kind,
269	source: SOURCE,
270	repo_id: Some(repo_id.to_owned()),
271	actor,
272	data,
273	};
274	g1t_kit::call(
275	&self.events,
276	"publish",
277	&Publish {
278	events: vec![event],
279	},
280	)
281	.await
282	}
283
284	/// The repository, if the viewer may see it. Whether they may is
285	/// decided by the repos service.
286	async fn repo(&self, path: &RepoPath, viewer: &Viewer) -> Result<Outcome<Repo>> {
287	let found: Outcome<Repo> = self
288	.timing
289	.rpc(g1t_kit::call(
290	&self.repos,
291	"get",
292	&GetArgs {
293	path: path.clone(),
294	viewer: viewer.clone(),
295	},
296	))
297	.await?;
298	if let Outcome::Ok(repo) = &found {
299	self.known_repos.borrow_mut().insert(repo.id.clone(), repo.clone());
300	}
301	Ok(found)
302	}
303
304	/// The next number in the repository's sequence. Taking it is one
305	/// statement, so concurrent opens cannot be given the same number.
306	async fn next_number(&self, repo_id: &str) -> Result<u32> {
307	let row = self
308	.db
309	.prepare(
310	"INSERT INTO counters (repo_id, last) VALUES (?, 1)
311	ON CONFLICT (repo_id) DO UPDATE SET last = last + 1
312	RETURNING last AS n",
313	)
314	.bind(&[repo_id.into()])?
315	.first::<NumberRow>(None)
316	.await?;
317	row.map(\|row\| row.n)
318	.ok_or_else(\|\| worker::Error::RustError("no number was assigned".into()))
319	}
320
321	async fn issue(&self, repo_id: &str, number: u32) -> Result<Option<Issue>> {
322	Ok(self
323	.db
324	.prepare(format!(
325	"SELECT {ISSUE_COLUMNS} FROM issues WHERE repo_id = ? AND number = ?"
326	))
327	.bind(&[repo_id.into(), number.into()])?
328	.first::<IssueRow>(None)
329	.await?
330	.map(Issue::from))
331	}
332
333	async fn pull(&self, repo_id: &str, number: u32) -> Result<Option<Pull>> {
334	Ok(self
335	.db
336	.prepare(format!("SELECT {PULL_COLUMNS} FROM pulls WHERE repo_id = ? AND number = ?"))
337	.bind(&[repo_id.into(), number.into()])?
338	.first::<PullRow>(None)
339	.await?
340	.map(Pull::from))
341	}
342
343	/// The repository and one of its issues, as seen by `viewer`.
344	async fn issue_at(
345	&self,
346	path: &RepoPath,
347	number: u32,
348	viewer: &Viewer,
349	) -> Result<Outcome<(Repo, Issue)>> {
350	let Outcome::Ok(repo) = self.repo(path, viewer).await? else {
351	return Ok(no_issue());
352	};
353	Ok(match self.issue(&repo.id, number).await? {
354	Some(issue) => Outcome::Ok((repo, issue)),
355	None => no_issue(),
356	})
357	}
358
359	/// The repository and one of its pull requests, as seen by `viewer`.
360	async fn pull_at(
361	&self,
362	path: &RepoPath,
363	number: u32,
364	viewer: &Viewer,
365	) -> Result<Outcome<(Repo, Pull)>> {
366	let Outcome::Ok(repo) = self.repo(path, viewer).await? else {
367	return Ok(no_pull());
368	};
369	Ok(match self.pull(&repo.id, number).await? {
370	Some(pull) => Outcome::Ok((repo, pull)),
371	None => no_pull(),
372	})
373	}
374
375	/// Records something that happened to an issue or a pull request, so
376	/// that it shows in the conversation where it happened. `text` is what
377	/// `author` did, as the rest of a sentence starting with their name.
378	pub(crate) async fn note(
379	&self,
380	repo_id: &str,
381	number: u32,
382	author: (&str, &str),
383	text: &str,
384	) -> Result<()> {
385	let now = now_ms();
386	self.db
387	.prepare(
388	"INSERT INTO comments
389	(id, repo_id, number, author_id, author_name, body, kind, created_at)
390	VALUES (?, ?, ?, ?, ?, ?, 'event', ?)",
391	)
392	.bind(&[
393	new_id("cmt", now).into(),
394	repo_id.into(),
395	number.into(),
396	author.0.into(),
397	author.1.into(),
398	text.into(),
399	rfc3339(now).into(),
400	])?
401	.run()
402	.await?;
403	Ok(())
404	}
405
406	/// Notes who was added to and removed from a list of people, such as
407	/// "assigned ana" or "requested a review from g1t".
408	async fn note_changes(
409	&self,
410	repo_id: &str,
411	number: u32,
412	actor: &User,
413	before: &[String],
414	after: &[String],
415	(added, removed): (&str, &str),
416	) -> Result<()> {
417	let joined = \|names: Vec<&String>\| {
418	names
419	.into_iter()
420	.map(String::as_str)
421	.collect::<Vec<_>>()
422	.join(", ")
423	};
424	let new: Vec<&String> = after.iter().filter(\|name\| !before.contains(name)).collect();
425	let gone: Vec<&String> = before.iter().filter(\|name\| !after.contains(name)).collect();
426	let who = (actor.id.as_str(), actor.username.as_str());
427	if !new.is_empty() {
428	// Taking something on oneself reads better said that way.
429	let text = if added == "assigned" && new == [&actor.username] {
430	"self-assigned this".to_owned()
431	} else {
432	format!("{added} {}", joined(new))
433	};
434	self.note(repo_id, number, who, &text).await?;
435	}
436	if !gone.is_empty() {
437	self.note(repo_id, number, who, &format!("{removed} {}", joined(gone)))
438	.await?;
439	}
440	Ok(())
441	}
442
443	fn issue_event(issue: &Issue) -> IssueEvent {
444	IssueEvent {
445	issue_id: issue.id.clone(),
446	repo_id: issue.repo_id.clone(),
447	number: issue.number,
448	author: Some((&issue.author).into()),
449	requested_by: issue.requested_by.as_ref().map(Into::into),
450	..IssueEvent::default()
451	}
452	}
453
454	/// The commit a pull request's change is at in git right now: its
455	/// fork's default branch, or its branch.
456	async fn live_head(&self, pull: &Pull) -> Result<Option<String>> {
457	g1t_kit::call(
458	&self.repos,
459	"head",
460	&HeadArgs {
461	repo_id: pull.fork_repo_id.clone().unwrap_or_else(\|\| pull.repo_id.clone()),
462	branch: pull.branch.clone().unwrap_or_default(),
463	},
464	)
465	.await
466	}
467
468	fn pull_event(pull: &Pull) -> PullEvent {
469	PullEvent {
470	pull_id: pull.id.clone(),
471	repo_id: pull.repo_id.clone(),
472	number: pull.number,
473	author: Some((&pull.author).into()),
474	requested_by: pull.requested_by.as_ref().map(Into::into),
475	issue: pull.issue,
476	..PullEvent::default()
477	}
478	}
479
480	// --- Issues ------------------------------------------------------------
481
482	async fn open_issue(&self, a: OpenIssueArgs) -> Result<Outcome<Issue>> {
483	if !a.actor.verified {
484	return Ok(Outcome::fail(FailureCode::Forbidden, UNVERIFIED));
485	}
486	let title = match valid_title(&a.title) {
487	Ok(title) => title,
488	Err(message) => return Ok(Outcome::fail(FailureCode::Invalid, message)),
489	};
490	let Some(labels) = normalize_labels(&a.labels) else {
491	return Ok(Outcome::fail(
492	FailureCode::Invalid,
493	format!("An issue can have up to {MAX_LABELS} labels of up to {MAX_LABEL_CHARS} characters each."),
494	));
495	};
496	let repo = check!(self.repo(&a.repo, &Some(a.actor.clone())).await?);
497	check!(writable(&repo));
498	// Labels it does not have yet are made for someone who may triage;
499	// anyone else picks from those there are.
500	let colors = check!(self.ensure_labels(&a.actor, &repo, &labels).await?);
501	let milestone = match a.milestone.filter(\|number\| *number > 0) {
502	Some(number) => {
503	check!(allowed(Some(&a.actor), &repo, Capability::Triage));
504	check!(self.milestone_ref(&repo.id, number).await?)
505	}
506	None => None,
507	};
508	// Commands given the old way are words now: added to the body
509	// under "Definition of done". What has to pass to merge is
510	// the branch's required checks.
511	let body = with_definition_of_done(&a.body, &commands_pass(&a.checks));
512
513	let now = now_ms();
514	let id = new_id("iss", now);
515	let number = self.next_number(&repo.id).await?;
516	let timestamp = rfc3339(now);
517	let (author, requested_by) = authorship(&a.actor);
518	self.db
519	.prepare(
520	"INSERT INTO issues
521	(id, repo_id, number, title, body, labels, checks, author_id, author_name,
522	requested_by_id, requested_by_name, created_at, updated_at, milestone)
523	VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
524	)
525	.bind(&[
526	id.as_str().into(),
527	repo.id.as_str().into(),
528	number.into(),
529	title.into(),
530	body.into(),
531	serde_json::to_string(&labels)?.into(),
532	"[]".into(),
533	author.id.as_str().into(),
534	author.username.as_str().into(),
535	optional(&requested_by.as_ref().map(\|user\| user.id.clone())),
536	optional(&requested_by.as_ref().map(\|user\| user.username.clone())),
537	timestamp.as_str().into(),
538	timestamp.as_str().into(),
539	optional_number(milestone.as_ref().map(\|milestone\| milestone.number)),
540	])?
541	.run()
542	.await?;
543	let Some(issue) = self.issue(&repo.id, number).await? else {
544	return Ok(no_issue());
545	};
546	self.publish(
547	"issue.opened",
548	&repo.id,
549	&a.actor,
550	IssueEvent {
551	title: Some(issue.title.clone()),
552	..Self::issue_event(&issue)
553	},
554	)
555	.await?;
556	// Opened with labels and a milestone: each is said, as it would be
557	// if they were added afterwards, without notes in the conversation.
558	for label in &issue.labels {
559	let color = colors.iter().find(\|(name, _)\| name == label).map_or_else(\|\| label_color_for(label), \|(_, c)\| c.clone());
560	let label = Some(g1t_contracts::events::EventLabel { name: label.clone(), color });
561	self.publish("issue.labeled", &repo.id, &a.actor, IssueEvent { label, ..Self::issue_event(&issue) })
562	.await?;
563	}
564	if let Some(milestone) = milestone {
565	self.publish(
566	"issue.milestoned",
567	&repo.id,
568	&a.actor,
569	IssueEvent { milestone: Some(milestone), ..Self::issue_event(&issue) },
570	)
571	.await?;
572	}
573	Ok(Outcome::Ok(issue))
574	}
575
576	async fn list_issues(&self, a: ListIssuesArgs) -> Result<Outcome<Vec<Issue>>> {
577	let state = state_name(a.state);
578	let label = a
579	.label
580	.map(\|label\| label.split_whitespace().collect::<Vec<_>>().join(" ").to_lowercase())
581	.filter(\|label\| !label.is_empty());
582	let milestone = a.milestone;
583	let list = \|repo_id: String\| {
584	let label = label.clone();
585	async move {
586	let state = state.map_or(JsValue::NULL, JsValue::from);
587	let query = self
588	.db
589	.prepare(format!(
590	"SELECT {ISSUE_COLUMNS} FROM issues
591	WHERE repo_id = ? AND (? IS NULL OR state = ?)
592	AND (? IS NULL OR EXISTS
593	(SELECT 1 FROM json_each(issues.labels) WHERE json_each.value = ?))
594	AND (? IS NULL OR milestone = ?)
595	ORDER BY number DESC LIMIT ?"
596	))
597	.bind(&[
598	repo_id.into(),
599	state.clone(),
600	state,
601	optional(&label),
602	optional(&label),
603	optional_number(milestone),
604	optional_number(milestone),
605	LIST_PAGE.into(),
606	])?;
607	self.timing.db(1, query.all()).await?.results::<IssueRow>()
608	}
609	};
610	let (_, rows) = check!(self.repo_then(&a.repo, &a.viewer, list).await?);
611	Ok(Outcome::Ok(rows.into_iter().map(Issue::from).collect()))
612	}
613
614	/// An issue, the pull requests for it and its comments: one batch,
615	/// started beside the access check (prefetch.rs).
616	async fn get_issue(&self, a: ViewArgs) -> Result<Outcome<IssueDetail>> {
617	let number = a.number;
618	let read = \|repo_id: String\| async move {
619	let key = [JsValue::from(repo_id.as_str()), JsValue::from(number)];
620	let statements = vec![
621	self.db
622	.prepare(format!("SELECT {ISSUE_COLUMNS} FROM issues WHERE repo_id = ?1 AND number = ?2"))
623	.bind(&key)?,
624	self.db
625	.prepare(format!(
626	"SELECT {PULL_COLUMNS} FROM pulls
627	WHERE issue_id = (SELECT id FROM issues WHERE repo_id = ?1 AND number = ?2)
628	ORDER BY number"
629	))
630	.bind(&key)?,
631	self.db
632	.prepare("SELECT * FROM comments WHERE repo_id = ?1 AND number = ?2 ORDER BY id LIMIT 500")
633	.bind(&key)?,
634	];
635	let results = self.timing.db(3, self.db.batch(statements)).await?;
636	let rows = \|index: usize\| results.get(index).ok_or_else(\|\| worker::Error::RustError("short batch".into()));
637	Ok::<_, worker::Error>((
638	rows(0)?.results::<IssueRow>()?.into_iter().next().map(Issue::from),
639	rows(1)?.results::<PullRow>()?.into_iter().map(Pull::from).collect::<Vec<_>>(),
640	rows(2)?.results::<CommentRow>()?.into_iter().map(Comment::from).collect::<Vec<_>>(),
641	))
642	};
643	let Outcome::Ok((_, (Some(issue), pulls, comments))) = self.repo_then(&a.repo, &a.viewer, read).await? else {
644	return Ok(no_issue());
645	};
646	Ok(Outcome::Ok(IssueDetail { comments, pulls, issue }))
647	}
648
649	/// The issue, if `actor` wrote it or may triage the repository's issues.
650	async fn manageable_issue(
651	&self,
652	actor: &User,
653	path: &RepoPath,
654	number: u32,
655	) -> Result<Outcome<Issue>> {
656	let (repo, issue) = check!(self.issue_at(path, number, &Some(actor.clone())).await?);
657	check!(writable(&repo));
658	if issue.owner().id != actor.id {
659	check!(allowed(Some(actor), &repo, Capability::Triage));
660	}
661	Ok(Outcome::Ok(issue))
662	}
663
664	async fn update_issue(&self, a: UpdateIssueArgs) -> Result<Outcome<Issue>> {
665	let issue = check!(self.manageable_issue(&a.actor, &a.repo, a.number).await?);
666	let title = match a.title.as_deref().map(valid_title) {
667	Some(Err(message)) => return Ok(Outcome::fail(FailureCode::Invalid, message)),
668	Some(Ok(title)) => Some(title.to_owned()),
669	None => None,
670	};
671	if a.labels.as_deref().is_some_and(\|labels\| normalize_labels(labels).is_none()) {
672	return Ok(Outcome::fail(
673	FailureCode::Invalid,
674	format!("An issue can have up to {MAX_LABELS} labels of up to {MAX_LABEL_CHARS} characters each."),
675	));
676	}
677	let assignees = match a.assignees {
678	Some(names) => Some(check!(self.valid_assignees(names).await?)),
679	None => None,
680	};
681	// What it is blocked by: checked before anything changes.
682	let blocked_by = match &a.blocked_by {
683	Some(asked) => Some(check!(self.checked_blocked_by(&issue.repo_id, issue.number, asked).await?)),
684	None => None,
685	};
686	// Its labels and milestone first: either can be refused, and then
687	// nothing else changes.
688	if a.milestone.is_some() \|\| a.labels.is_some() {
689	let repo = check!(self.repo(&a.repo, &Some(a.actor.clone())).await?);
690	if let Some(number) = a.milestone {
691	check!(self.set_milestone(&a.actor, &repo, &labels::Item::Issue(issue.clone()), number).await?);
692	}
693	if let Some(labels) = &a.labels {
694	check!(self.relabel(&a.actor, &repo, &labels::Item::Issue(issue.clone()), labels).await?);
695	}
696	}
697	let assigned = assignees.as_ref().map(serde_json::to_string).transpose()?;
698	let blocked = blocked_by.as_ref().map(serde_json::to_string).transpose()?;
699	let body = a.body.map(\|body\| body.trim().to_owned());
700	self.db
701	.prepare(
702	"UPDATE issues
703	SET title = COALESCE(?, title), body = COALESCE(?, body),
704	assignees = COALESCE(?, assignees), blocked_by = COALESCE(?, blocked_by),
705	updated_at = ?
706	WHERE id = ?",
707	)
708	.bind(&[
709	optional(&title),
710	optional(&body),
711	optional(&assigned),
712	optional(&blocked),
713	rfc3339(now_ms()).into(),
714	issue.id.as_str().into(),
715	])?
716	.run()
717	.await?;
718	let before = issue.assignees.clone();
719	let blocked_before = issue.blocked_by.clone();
720	let Some(issue) = self.issue(&issue.repo_id, issue.number).await? else {
721	return Ok(no_issue());
722	};
723	self.publish(
724	"issue.updated",
725	&issue.repo_id,
726	&a.actor,
727	Self::issue_event(&issue),
728	)
729	.await?;
730	if blocked_by.is_some() {
731	self.blocked_by_changed(&issue, &blocked_before, &a.actor).await?;
732	}
733	if let Some(assignees) = assignees {
734	self.note_changes(
735	&issue.repo_id,
736	issue.number,
737	&a.actor,
738	&before,
739	&assignees,
740	("assigned", "unassigned"),
741	)
742	.await?;
743	let added: Vec<String> = assignees.iter().filter(\|name\| !before.contains(name)).cloned().collect();
744	self.publish(
745	"issue.assigned",
746	&issue.repo_id,
747	&a.actor,
748	IssueEvent {
749	assignees: Some(assignees),
750	added: Some(added),
751	..Self::issue_event(&issue)
752	},
753	)
754	.await?;
755	}
756	Ok(Outcome::Ok(issue))
757	}
758
759	/// Usernames as given, tidied, if each names an account.
760	async fn valid_assignees(&self, names: Vec<String>) -> Result<Outcome<Vec<String>>> {
761	let mut assignees: Vec<String> = Vec::new();
762	for name in names {
763	let name = name.trim().trim_start_matches('@').to_lowercase();
764	if name.is_empty() \|\| assignees.contains(&name) {
765	continue;
766	}
767	if assignees.len() == MAX_ASSIGNEES {
768	return Ok(Outcome::fail(
769	FailureCode::Invalid,
770	format!("An issue can be assigned to at most {MAX_ASSIGNEES} people."),
771	));
772	}
773	let account: Viewer = g1t_kit::call(
774	&self.identity,
775	"user_by_username",
776	&UsernameArgs {
777	username: name.clone(),
778	},
779	)
780	.await?;
781	if account.is_none() {
782	return Ok(Outcome::fail(
783	FailureCode::Invalid,
784	format!("There is no account named {name}."),
785	));
786	}
787	assignees.push(name);
788	}
789	Ok(Outcome::Ok(assignees))
790	}
791
792	async fn close_issue(&self, a: IssueActionArgs) -> Result<Outcome<Issue>> {
793	let mut issue = check!(self.manageable_issue(&a.actor, &a.repo, a.number).await?);
794	if issue.state == State::Closed {
795	return Ok(Outcome::fail(
796	FailureCode::Conflict,
797	"This issue is already closed.",
798	));
799	}
800	let reason = a.reason.unwrap_or(IssueReason::Completed);
801	let now = rfc3339(now_ms());
802	self.db
803	.prepare(
804	"UPDATE issues SET state = 'closed', reason = ?, closed_at = ?, updated_at = ?
805	WHERE id = ?",
806	)
807	.bind(&[
808	reason.as_str().into(),
809	now.as_str().into(),
810	now.as_str().into(),
811	issue.id.as_str().into(),
812	])?
813	.run()
814	.await?;
815	self.publish(
816	"issue.closed",
817	&issue.repo_id,
818	&a.actor,
819	IssueEvent {
820	reason: Some(reason.as_str()),
821	..Self::issue_event(&issue)
822	},
823	)
824	.await?;
825	self.note(
826	&issue.repo_id,
827	issue.number,
828	(&a.actor.id, &a.actor.username),
829	match reason {
830	IssueReason::Completed => "closed this as completed",
831	IssueReason::NotPlanned => "closed this as not planned",
832	},
833	)
834	.await?;
835	// Its claims end, and the issues it blocks hear (coordination/).
836	self.issue_resolved(&issue.repo_id, issue.number, &a.actor, true, None).await;
837	issue.state = State::Closed;
838	issue.reason = Some(reason);
839	issue.closed_at = Some(now.clone());
840	issue.updated_at = now;
841	Ok(Outcome::Ok(issue))
842	}
843
844	async fn reopen_issue(&self, a: IssueActionArgs) -> Result<Outcome<Issue>> {
845	let mut issue = check!(self.manageable_issue(&a.actor, &a.repo, a.number).await?);
846	if issue.state == State::Open {
847	return Ok(Outcome::fail(
848	FailureCode::Conflict,
849	"This issue is already open.",
850	));
851	}
852	let now = rfc3339(now_ms());
853	self.db
854	.prepare(
855	"UPDATE issues
856	SET state = 'open', reason = NULL, resolved_by = NULL, closed_at = NULL,
857	updated_at = ?
858	WHERE id = ?",
859	)
860	.bind(&[now.as_str().into(), issue.id.as_str().into()])?
861	.run()
862	.await?;
863	self.publish(
864	"issue.reopened",
865	&issue.repo_id,
866	&a.actor,
867	Self::issue_event(&issue),
868	)
869	.await?;
870	self.note(
871	&issue.repo_id,
872	issue.number,
873	(&a.actor.id, &a.actor.username),
874	"reopened this",
875	)
876	.await?;
877	issue.state = State::Open;
878	issue.reason = None;
879	issue.resolved_by = None;
880	issue.closed_at = None;
881	issue.updated_at = now;
882	Ok(Outcome::Ok(issue))
883	}
884
885
886	async fn counts(&self, a: ViewArgs) -> Result<Outcome<Counts>> {
887	let read = \|repo_id: String\| async move {
888	let query = self
889	.db
890	.prepare(
891	"SELECT
892	(SELECT count(*) FROM issues WHERE repo_id = ?1 AND state = 'open') AS issues,
893	(SELECT count(*) FROM pulls
894	WHERE repo_id = ?1 AND status IN ('draft', 'open')) AS pulls",
895	)
896	.bind(&[repo_id.into()])?;
897	self.timing.db(1, query.first::<Counts>(None)).await
898	};
899	let (_, counts) = check!(self.repo_then(&a.repo, &a.viewer, read).await?);
900	Ok(Outcome::Ok(counts.unwrap_or(Counts {
901	issues: 0,
902	pulls: 0,
903	})))
904	}
905
906	// --- Comments ----------------------------------------------------------
907
908	async fn add_comment(&self, a: AddCommentArgs) -> Result<Outcome<Comment>> {
909	if !a.actor.verified {
910	return Ok(Outcome::fail(FailureCode::Forbidden, UNVERIFIED));
911	}
912	let body = a.body.trim();
913	// An approval speaks for itself; anything else has to say something.
914	if body.is_empty() && a.verdict != Some(Verdict::Approve) {
915	return Ok(Outcome::fail(
916	FailureCode::Invalid,
917	"A comment cannot be empty.",
918	));
919	}
920	let path = a
921	.path
922	.as_deref()
923	.map(str::trim)
924	.filter(\|path\| !path.is_empty());
925	let line = a.line.filter(\|line\| *line > 0 && path.is_some());
926	if body.chars().count() > MAX_ENTRY_CHARS {
927	return Ok(Outcome::fail(
928	FailureCode::Invalid,
929	"That comment is too long.",
930	));
931	}
932	let repo = check!(self.repo(&a.repo, &Some(a.actor.clone())).await?);
933	check!(writable(&repo));
934	// The number names an issue or a pull request, never both.
935	let mut pull_id = None;
936	let mut pull_issue = None;
937	let table = if self.issue(&repo.id, a.number).await?.is_some() {
938	if path.is_some() \|\| a.verdict.is_some() {
939	return Ok(Outcome::fail(
940	FailureCode::Invalid,
941	"Only a pull request can be reviewed or commented on by line.",
942	));
943	}
944	"issues"
945	} else if let Some(pull) = self.pull(&repo.id, a.number).await? {
946	if a.verdict.is_some() && pull.is_owned_by(&a.actor.id) {
947	return Ok(Outcome::fail(
948	FailureCode::Forbidden,
949	"You cannot approve or request changes on your own pull request.",
950	));
951	}
952	pull_id = Some(pull.id.clone());
953	pull_issue = pull.issue;
954	"pulls"
955	} else {
956	return Ok(Outcome::fail(
957	FailureCode::NotFound,
958	"No issue or pull request has that number.",
959	));
960	};
961
962	let now = now_ms();
963	let comment = Comment {
964	kind: CommentKind::Comment,
965	id: new_id("cmt", now),
966	author: a.actor.clone(),
967	body: body.to_owned(),
968	path: path.map(str::to_owned),
969	line,
970	verdict: a.verdict,
971	created_at: rfc3339(now),
972	edited_at: None,
973	agent: None,
974	acting_for: None,
975	advisory: false,
976	};
977	self.db
978	.batch(vec![
979	self.db
980	.prepare(
981	"INSERT INTO comments
982	(id, repo_id, number, author_id, author_name, body, path, line,
983	verdict, created_at)
984	VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
985	)
986	.bind(&[
987	comment.id.as_str().into(),
988	repo.id.as_str().into(),
989	a.number.into(),
990	a.actor.id.as_str().into(),
991	a.actor.username.as_str().into(),
992	body.into(),
993	optional(&comment.path),
994	optional_number(line),
995	a.verdict
996	.map_or(JsValue::NULL, \|verdict\| verdict.as_str().into()),
997	comment.created_at.as_str().into(),
998	])?,
999	self.db
1000	.prepare(format!(
1001	"UPDATE {table} SET updated_at = ? WHERE repo_id = ? AND number = ?"
1002	))
1003	.bind(&[
1004	comment.created_at.as_str().into(),
1005	repo.id.as_str().into(),
1006	a.number.into(),
1007	])?,
1008	])
1009	.await?;
1010	self.publish(
1011	"comment.created",
1012	&repo.id,
1013	&a.actor,
1014	CommentCreated {
1015	comment_id: comment.id.clone(),
1016	repo_id: repo.id.clone(),
1017	number: a.number,
1018	pull_id,
1019	verdict: a.verdict,
1020	..CommentCreated::default()
1021	},
1022	)
1023	.await?;
1024	// An approval is a decision: a person's is approved.
1025	if table == "pulls" && a.verdict == Some(Verdict::Approve) {
1026	let said = if body.is_empty() { String::new() } else { format!(": {}", body.chars().take(300).collect::<String>()) };
1027	self.keep_decision_quietly(
1028	&repo.id,
1029	decisions::NewDecision {
1030	issue: pull_issue,
1031	pull: Some(a.number),
1032	session_id: None,
1033	actor: Some(&a.actor),
1034	kind: g1t_contracts::decisions::DecisionKind::Approval,
1035	confidence: decisions::approval_confidence(&a.actor),
1036	summary: format!("{} approved #{}{said}", a.actor.username, a.number),
1037	evidence: serde_json::json!({ "comment": comment.id }),
1038	},
1039	)
1040	.await;
1041	}
1042	Ok(Outcome::Ok(comment))
1043	}
1044
1045	/// A comment in the repository, with the repository and the pull
1046	/// request it is on (if it is on one), when `actor` may edit it, or
1047	/// with `deleting`, delete it (`may_change_comment`).
1048	async fn changeable_comment(
1049	&self,
1050	a: &CommentActionArgs,
1051	deleting: bool,
1052	) -> Result<Outcome<(Repo, CommentRow, Option<String>)>> {
1053	let repo = check!(self.repo(&a.repo, &Some(a.actor.clone())).await?);
1054	check!(writable(&repo));
1055	let Some(row) = self
1056	.db
1057	.prepare("SELECT * FROM comments WHERE id = ? AND repo_id = ?")
1058	.bind(&[a.comment_id.trim().into(), repo.id.as_str().into()])?
1059	.first::<CommentRow>(None)
1060	.await?
1061	else {
1062	return Ok(Outcome::fail(FailureCode::NotFound, "Comment not found."));
1063	};
1064	let maintains = access::can(Some(&a.actor), &repo, Capability::ManageSettings);
1065	// What an agent wrote as itself is answered for by whoever it acted for.
1066	let author_id = row.answerable_id().to_owned();
1067	if let Err(refusal) = may_change_comment(
1068	row.kind,
1069	row.has_verdict(),
1070	&author_id,
1071	&a.actor.id,
1072	maintains,
1073	deleting,
1074	) {
1075	// Someone who may change it otherwise was refused for what it is.
1076	let code = if author_id == a.actor.id \|\| maintains { FailureCode::Conflict } else { FailureCode::Forbidden };
1077	return Ok(Outcome::fail(code, refusal));
1078	}
1079	let pull_id = self.pull(&repo.id, row.number).await?.map(\|pull\| pull.id);
1080	Ok(Outcome::Ok((repo, row, pull_id)))
1081	}
1082
1083	/// Changes the text of a comment: its author's to do, or a
1084	/// maintainer's. Publishes `comment.edited` with what it said before.
1085	async fn edit_comment(&self, a: CommentActionArgs) -> Result<Outcome<Comment>> {
1086	let body = a.body.trim().to_owned();
1087	if body.chars().count() > MAX_ENTRY_CHARS {
1088	return Ok(Outcome::fail(FailureCode::Invalid, "That comment is too long."));
1089	}
1090	let (repo, row, pull_id) = check!(self.changeable_comment(&a, false).await?);
1091	// An approval speaks for itself; anything else has to say something.
1092	if body.is_empty() && row.verdict != Some(Verdict::Approve) && row.agent_verdict.as_deref() != Some("approve") {
1093	return Ok(Outcome::fail(FailureCode::Invalid, "A comment cannot be empty."));
1094	}
1095	let number = row.number;
1096	let mut comment = Comment::from(row);
1097	if comment.body == body {
1098	return Ok(Outcome::Ok(comment));
1099	}
1100	let now = rfc3339(now_ms());
1101	self.db
1102	.prepare("UPDATE comments SET body = ?, edited_at = ? WHERE id = ?")
1103	.bind(&[body.as_str().into(), now.as_str().into(), comment.id.as_str().into()])?
1104	.run()
1105	.await?;
1106	let before = std::mem::replace(&mut comment.body, body);
1107	comment.edited_at = Some(now);
1108	self.publish(
1109	"comment.edited",
1110	&repo.id,
1111	&a.actor,
1112	CommentEdited {
1113	comment_id: comment.id.clone(),
1114	repo_id: repo.id.clone(),
1115	number,
1116	pull_id,
1117	changes: CommentChanges { body: ChangedFrom { from: before } },
1118	},
1119	)
1120	.await?;
1121	Ok(Outcome::Ok(comment))
1122	}
1123
1124	/// Deletes a comment: its author's to do, or a maintainer's. A review
1125	/// that gave a verdict stays. Publishes `comment.deleted` with the
1126	/// comment as it was.
1127	async fn delete_comment(&self, a: CommentActionArgs) -> Result<Outcome<bool>> {
1128	let (repo, row, pull_id) = check!(self.changeable_comment(&a, true).await?);
1129	self.db
1130	.prepare("DELETE FROM comments WHERE id = ?")
1131	.bind(&[row.id.as_str().into()])?
1132	.run()
1133	.await?;
1134	let number = row.number;
1135	let comment = Comment::from(row);
1136	self.publish(
1137	"comment.deleted",
1138	&repo.id,
1139	&a.actor,
1140	CommentDeleted {
1141	comment_id: comment.id.clone(),
1142	repo_id: repo.id.clone(),
1143	number,
1144	pull_id,
1145	comment: DeletedComment {
1146	id: comment.id,
1147	body: comment.body,
1148	author: (&comment.author).into(),
1149	created_at: comment.created_at,
1150	path: comment.path,
1151	line: comment.line,
1152	},
1153	},
1154	)
1155	.await?;
1156	Ok(Outcome::Ok(true))
1157	}
1158
1159	// --- Pull requests -----------------------------------------------------
1160
1161	async fn open_pull(&self, a: OpenPullArgs) -> Result<Outcome<Pull>> {
1162	if !a.actor.verified {
1163	return Ok(Outcome::fail(FailureCode::Forbidden, UNVERIFIED));
1164	}
1165	let repo = check!(self.repo(&a.repo, &Some(a.actor.clone())).await?);
1166	check!(writable(&repo));
1167	let issue = match a.issue {
1168	Some(number) => match self.issue(&repo.id, number).await? {
1169	Some(issue) if issue.state == State::Open => Some(issue),
1170	Some(_) => {
1171	return Ok(Outcome::fail(
1172	FailureCode::Conflict,
1173	"This issue is closed.",
1174	));
1175	}
1176	None => return Ok(no_issue()),
1177	},
1178	None => None,
1179	};
1180	// A pull request for an issue takes the issue's title unless given one.
1181	let title = match (a.title.trim(), &issue) {
1182	("", Some(issue)) => issue.title.clone(),
1183	(title, _) => match valid_title(title) {
1184	Ok(title) => title.to_owned(),
1185	Err(message) => return Ok(Outcome::fail(FailureCode::Invalid, message)),
1186	},
1187	};
1188	// Unnamed, the change is its author's, unless an agent opened it.
1189	let agent = match a.agent.trim() {
1190	"" if g1t_contracts::rules::is_agent(&a.actor) => "agent",
1191	"" => a.actor.username.as_str(),
1192	agent => agent,
1193	};
1194	let runtime = match a.runtime {
1195	Runtime::Hosted => "hosted",
1196	Runtime::External => "external",
1197	};
1198
1199	let now = now_ms();
1200	let id = new_id("pr", now);
1201	let branch = a
1202	.branch
1203	.as_deref()
1204	.map(str::trim)
1205	.filter(\|branch\| !branch.is_empty());
1206	// The branch it merges into: the default branch unless another is
1207	// asked for, which has to exist.
1208	let base = a.base.as_deref().and_then(\|base\| stored_base(base, &repo));
1209	if let Some(base) = &base {
1210	if branch == Some(base.as_str()) {
1211	return Ok(Outcome::fail(
1212	FailureCode::Invalid,
1213	format!("A pull request cannot merge {base} into itself. Choose another base."),
1214	));
1215	}
1216	let exists: Option<String> = g1t_kit::call(
1217	&self.repos,
1218	"head",
1219	&HeadArgs { repo_id: repo.id.clone(), branch: base.clone() },
1220	)
1221	.await?;
1222	if exists.is_none() {
1223	return Ok(Outcome::fail(FailureCode::NotFound, format!("There is no branch named {base} to merge into.")));
1224	}
1225	}
1226	let base_name = base.clone().unwrap_or_else(\|\| repo.default_branch.clone());
1227	// The change is on a branch already pushed to the repository, or
1228	// will be made in a fork created for this pull request.
1229	let (fork, head) = match branch {
1230	Some(branch) => {
1231	if branch == base_name {
1232	return Ok(Outcome::fail(
1233	FailureCode::Invalid,
1234	format!("Choose a branch other than {branch}."),
1235	));
1236	}
1237	let head: Option<String> = g1t_kit::call(
1238	&self.repos,
1239	"head",
1240	&HeadArgs {
1241	repo_id: repo.id.clone(),
1242	branch: branch.to_owned(),
1243	},
1244	)
1245	.await?;
1246	let Some(head) = head else {
1247	return Ok(Outcome::fail(
1248	FailureCode::NotFound,
1249	format!("There is no branch named {branch}. Push it first."),
1250	));
1251	};
1252	// One open pull request for each branch and base.
1253	let existing = self
1254	.db
1255	.prepare(
1256	"SELECT number AS n FROM pulls
1257	WHERE repo_id = ? AND source_branch = ? AND status IN ('draft', 'open')
1258	AND base_branch IS ?",
1259	)
1260	.bind(&[repo.id.as_str().into(), branch.into(), optional(&base)])?
1261	.first::<NumberRow>(None)
1262	.await?;
1263	if let Some(existing) = existing {
1264	return Ok(Outcome::fail(
1265	FailureCode::Conflict,
1266	format!("Pull request #{} is already open from {branch} into {base_name}.", existing.n),
1267	));
1268	}
1269	(None, Some(head))
1270	}
1271	None => {
1272	let fork: Outcome<Repo> = g1t_kit::call(
1273	&self.repos,
1274	"fork_for_pull",
1275	&ForkArgs {
1276	source_id: repo.id.clone(),
1277	pull_id: id.clone(),
1278	actor: a.actor.clone(),
1279	},
1280	)
1281	.await?;
1282	(Some(check!(fork)), None)
1283	}
1284	};
1285	// A branch already holds the work, so its pull request is ready for
1286	// review from the start; one with a fork starts as a draft.
1287	// Asked for as a draft, it waits until it is marked ready.
1288	let status = if branch.is_some() && !a.draft { "open" } else { "draft" };
1289	let body = Some(a.body.trim().to_owned()).filter(\|body\| !body.is_empty());
1290
1291	let number = self.next_number(&repo.id).await?;
1292	let timestamp = rfc3339(now);
1293	let (author, requested_by) = authorship(&a.actor);
1294	self.db
1295	.prepare(
1296	"INSERT INTO pulls
1297	(id, repo_id, number, issue_id, issue_number, title, body, agent, runtime,
1298	status, fork_repo_id, fork_namespace, fork_name, source_branch, head_commit,
1299	author_id, author_name, requested_by_id, requested_by_name, created_at, updated_at,
1300	base_branch)
1301	VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
1302	)
1303	.bind(&[
1304	id.as_str().into(),
1305	repo.id.as_str().into(),
1306	number.into(),
1307	optional(&issue.as_ref().map(\|issue\| issue.id.clone())),
1308	optional_number(issue.as_ref().map(\|issue\| issue.number)),
1309	title.into(),
1310	optional(&body),
1311	agent.into(),
1312	runtime.into(),
1313	status.into(),
1314	optional(&fork.as_ref().map(\|fork\| fork.id.clone())),
1315	optional(&fork.as_ref().map(\|fork\| fork.namespace.clone())),
1316	optional(&fork.as_ref().map(\|fork\| fork.name.clone())),
1317	optional(&branch.map(str::to_owned)),
1318	optional(&head),
1319	author.id.as_str().into(),
1320	author.username.as_str().into(),
1321	optional(&requested_by.as_ref().map(\|user\| user.id.clone())),
1322	optional(&requested_by.as_ref().map(\|user\| user.username.clone())),
1323	timestamp.as_str().into(),
1324	timestamp.as_str().into(),
1325	optional(&base),
1326	])?
1327	.run()
1328	.await?;
1329	let Some(mut pull) = self.pull(&repo.id, number).await? else {
1330	return Ok(no_pull());
1331	};
1332	fill_base(&mut pull, &repo);
1333	if let Some(issue) = pull.issue {
1334	let text = format!("opened #{} for this", pull.number);
1335	self.note(&repo.id, issue, (&a.actor.id, &a.actor.username), &text)
1336	.await?;
1337	}
1338	// Its head and merge refs in the repository, before its events go out.
1339	self.keep_pull_refs(&pull).await;
1340	self.publish(
1341	"pull.opened",
1342	&repo.id,
1343	&a.actor,
1344	PullEvent {
1345	agent: Some(pull.agent.clone()),
1346	base: pull.base.clone(),
1347	..Self::pull_event(&pull)
1348	},
1349	)
1350	.await?;
1351	// Its code owners asked to review (codeowners.rs).
1352	self.refresh_code_owners(&pull).await;
1353	let pull = self.pull(&repo.id, number).await?.unwrap_or(pull);
1354	// A branch pushed before may overlap others from the start.
1355	self.announce_overlaps(&pull, Some(a.actor.id.clone())).await;
1356	Ok(Outcome::Ok(pull))
1357	}
1358
1359	async fn list_pulls(&self, a: ListPullsArgs) -> Result<Outcome<Vec<Pull>>> {
1360	let filter = match a.state {
1361	Some(State::Open) => "AND status IN ('draft', 'open')",
1362	Some(State::Closed) => "AND status IN ('merged', 'closed')",
1363	None => "",
1364	};
1365	let label = a
1366	.label
1367	.map(\|label\| label.split_whitespace().collect::<Vec<_>>().join(" ").to_lowercase())
1368	.filter(\|label\| !label.is_empty());
1369	let milestone = a.milestone;
1370	let read = \|repo_id: String\| {
1371	let label = label.clone();
1372	async move {
1373	let query = self
1374	.db
1375	.prepare(format!(
1376	"SELECT {PULL_COLUMNS} FROM pulls WHERE repo_id = ? {filter}
1377	AND (? IS NULL OR EXISTS
1378	(SELECT 1 FROM json_each(pulls.labels) WHERE json_each.value = ?))
1379	AND (? IS NULL OR milestone = ?)
1380	ORDER BY number DESC LIMIT ?"
1381	))
1382	.bind(&[
1383	repo_id.into(),
1384	optional(&label),
1385	optional(&label),
1386	optional_number(milestone),
1387	optional_number(milestone),
1388	LIST_PAGE.into(),
1389	])?;
1390	self.timing.db(1, query.all()).await?.results::<PullRow>()
1391	}
1392	};
1393	let (repo, rows) = check!(self.repo_then(&a.repo, &a.viewer, read).await?);
1394	let base = a.base.map(\|base\| base.trim().to_owned()).filter(\|base\| !base.is_empty());
1395	Ok(Outcome::Ok(
1396	rows.into_iter()
1397	.map(\|row\| {
1398	let mut pull = Pull::from(row);
1399	fill_base(&mut pull, &repo);
1400	pull
1401	})
1402	.filter(\|pull\| base.as_deref().is_none_or(\|base\| pull.base.as_deref() == Some(base)))
1403	.collect(),
1404	))
1405	}
1406
1407	/// `pulls_for_repos`: what `list_pulls` gives, open and closed, for many
1408	/// repositories at once: one access check with repos for all of them
1409	/// and one query, instead of two of each per repository.
1410	async fn pulls_for_repos(&self, a: PullsForReposArgs) -> Result<Vec<RepoPulls>> {
1411	let ids: Vec<String> = a.repo_ids.into_iter().take(MAX_PULLS_FOR_REPOS).collect();
1412	if ids.is_empty() {
1413	return Ok(Vec::new());
1414	}
1415	let limit = a.limit.clamp(1, LIST_PAGE);
1416	// The rows are read beside the access check, for every id asked
1417	// about; those of repositories the viewer cannot read are dropped.
1418	let asked = serde_json::to_string(&ids)?;
1419	let check = ReadableArgs { ids, viewer: a.viewer };
1420	let readable = self.timing.rpc(g1t_kit::call::<_, Vec<Repo>>(&self.repos, "readable", &check));
1421	let (readable, rows) = try_join(readable, self.timing.db(1, self.newest_pulls(asked, limit))).await?;
1422	if readable.is_empty() {
1423	return Ok(Vec::new());
1424	}
1425	let mut answer: Vec<RepoPulls> = readable
1426	.iter()
1427	.map(\|repo\| RepoPulls { repo_id: repo.id.clone(), open: Vec::new(), closed: Vec::new() })
1428	.collect();
1429	for pull in rows.into_iter().map(Pull::from) {
1430	let Some(entry) = answer.iter_mut().find(\|entry\| entry.repo_id == pull.repo_id) else {
1431	continue;
1432	};
1433	match pull.status {
1434	PullStatus::Draft \| PullStatus::Open => entry.open.push(pull),
1435	PullStatus::Merged \| PullStatus::Closed => entry.closed.push(pull),
1436	}
1437	}
1438	for entry in &mut answer {
1439	entry.open.sort_by_key(\|pull\| std::cmp::Reverse(pull.number));
1440	entry.closed.sort_by_key(\|pull\| std::cmp::Reverse(pull.number));
1441	}
1442	Ok(answer)
1443	}
1444
1445	/// The newest `limit` of each repository's open (draft or open) and
1446	/// closed (merged or closed) pull requests, for the ids in `ids` (JSON).
1447	async fn newest_pulls(&self, ids: String, limit: u32) -> Result<Vec<PullRow>> {
1448	self.db
1449	.prepare(format!(
1450	"SELECT * FROM (
1451	SELECT {PULL_COLUMNS}, ROW_NUMBER() OVER (
1452	PARTITION BY pulls.repo_id, pulls.status IN ('draft', 'open') ORDER BY pulls.number DESC
1453	) AS place
1454	FROM pulls WHERE pulls.repo_id IN (SELECT value FROM json_each(?1))
1455	) WHERE place <= ?2"
1456	))
1457	.bind(&[ids.into(), limit.into()])?
1458	.all()
1459	.await?
1460	.results::<PullRow>()
1461	}
1462
1463	async fn get_pull(&self, a: ViewArgs) -> Result<Outcome<PullDetail>> {
1464	let number = a.number;
1465	// Every row the page and the lifecycle read, in one batch started
1466	// beside the access check; the helpers below read from it.
1467	let namespace = a.repo.namespace.clone();
1468	let read = \|repo_id: String\| self.prefetch_pull(repo_id, namespace.clone(), number);
1469	let Outcome::Ok((repo, Some(found))) = self.repo_then(&a.repo, &a.viewer, read).await? else {
1470	return Ok(no_pull());
1471	};
1472	let Some(row) = found.first::<PullRow>(prefetch::Slot::Pull)? else {
1473	return Ok(no_pull());
1474	};
1475	let stored = found.first::<StoredBehind>(prefetch::Slot::Pull)?;
1476	let issue = found.first::<IssueRow>(prefetch::Slot::Issue)?.map(Issue::from);
1477	let comments: Vec<Comment> =
1478	found.rows::<CommentRow>(prefetch::Slot::Comments)?.into_iter().map(Comment::from).collect();
1479	self.keep_prefetched(Some(found));
1480	let default_branch = repo.default_branch.clone();
1481	let detail = self.pull_detail(repo, Pull::from(row), issue, comments, stored, &a.viewer).await;
1482	self.keep_prefetched(None);
1483	// The branch it merges into, named, for whoever reads it.
1484	Ok(match detail? {
1485	Outcome::Ok(mut detail) => {
1486	if detail.pull.base.as_deref().is_none_or(str::is_empty) {
1487	detail.pull.base = Some(default_branch);
1488	}
1489	Outcome::Ok(detail)
1490	}
1491	failed => failed,
1492	})
1493	}
1494
1495	async fn pull_detail(
1496	&self,
1497	repo: Repo,
1498	mut pull: Pull,
1499	issue: Option<Issue>,
1500	comments: Vec<Comment>,
1501	stored: Option<StoredBehind>,
1502	viewer: &Viewer,
1503	) -> Result<Outcome<PullDetail>> {
1504	// Whether it is behind, as worked out with its mergeability on the
1505	// last push to either side (mergeability.rs), when that was for
1506	// its head as it is now; otherwise asked of the repos service.
1507	let known_behind = stored.and_then(\|stored\| stored.for_head(pull.head_commit.as_deref()));
1508	// Worked out on each push; this covers a pull request from before
1509	// that was recorded.
1510	if pull.files.is_empty() && pull.head_commit.is_some() {
1511	pull.files = self.refresh_files(&pull).await?;
1512	}
1513	// Everything else at once: none of it depends on the rest, and each
1514	// is a round trip of its own.
1515	let standing = async {
1516	// Mergeability first: where g1t sees a pull request through, a
1517	// conflict decides its next step.
1518	let behind = async {
1519	match known_behind {
1520	Some(behind) => Ok(behind),
1521	None => {
1522	let behind = self.is_behind(&repo.id, &pull).await?;
1523	// Kept for the next view when the mergeability on
1524	// record is for this head: a pull request from
1525	// before `behind` was kept asks once.
1526	if let Some(head) = pull.head_commit.as_deref()
1527	&& pull.status.is_active()
1528	{
1529	self.db
1530	.prepare(
1531	"UPDATE pulls SET behind = ?1
1532	WHERE id = ?2 AND behind IS NULL AND mergeable_key LIKE ?3 \|\| '..%'",
1533	)
1534	.bind(&[u32::from(behind).into(), pull.id.as_str().into(), head.into()])?
1535	.run()
1536	.await?;
1537	}
1538	Ok(behind)
1539	}
1540	}
1541	};
1542	let (merge, behind) = try_join(self.mergeability(&pull), behind).await?;
1543	let lifecycle = self.assess(&pull, &issue, behind).await?.map(\|(lifecycle, _)\| lifecycle);
1544	Ok::<_, worker::Error>((behind, lifecycle, merge))
1545	};
1546	let (((behind, lifecycle, (mergeable, conflicts)), (landing, stalled), comments), (checks, overlaps)) =
1547	try_join(
1548	try_join3(standing, self.landing_state(&pull.id), async { Ok(comments) }),
1549	try_join(self.latest_checks(&pull.id), self.overlaps(&pull)),
1550	)
1551	.await?;
1552	// The rules of the branch it merges into, as they stack, and which
1553	// of them it does not meet yet, for whoever is looking.
1554	let (statuses, settings, gate) = try_join3(
1555	self.statuses(&repo.id, pull.head_commit.as_deref()),
1556	self.settings_on(&repo, &pull),
1557	async {
1558	if pull.status.is_active() {
1559	self.merge_gate(&repo, &pull, viewer.as_ref(), false, true).await.map(Some)
1560	} else {
1561	Ok(None)
1562	}
1563	},
1564	)
1565	.await?;
1566	let code_owners = self.pull_code_owners(&pull, &comments, &settings).await?;
1567	let required = if settings.required_checks.is_empty() {
1568	Vec::new()
1569	} else {
1570	let checked = self.with_job_checks(&repo.id, pull.head_commit.as_deref(), statuses.clone()).await?;
1571	required_checks(&settings.required_checks, &checked)
1572	};
1573	Ok(Outcome::Ok(PullDetail {
1574	required_checks: required,
1575	rules: gate.map(\|gate\| rulesets::merge_rules(&gate.judged, &gate.requirements, pull.base_branch(&repo.default_branch) == repo.default_branch)),
1576	code_owners,
1577	comments,
1578	checks,
1579	overlaps,
1580	behind,
1581	lifecycle,
1582	landing,
1583	stalled,
1584	statuses,
1585	mergeable,
1586	conflicts,
1587	earlier_checks: self.earlier_checks(&pull.id).await?,
1588	issue,
1589	pull,
1590	}))
1591	}
1592
1593	/// The pull request, whatever its status, if its repository is not
1594	/// archived and `actor` opened it or may triage its pull requests.
1595	async fn managed_pull(
1596	&self,
1597	actor: &User,
1598	path: &RepoPath,
1599	number: u32,
1600	) -> Result<Outcome<Pull>> {
1601	let (repo, pull) = check!(self.pull_at(path, number, &Some(actor.clone())).await?);
1602	check!(writable(&repo));
1603	if !pull.is_owned_by(&actor.id) {
1604	check!(allowed(Some(actor), &repo, Capability::Triage));
1605	}
1606	Ok(Outcome::Ok(pull))
1607	}
1608
1609	/// The pull request, if it is still active and `actor` opened it or
1610	/// may triage the repository's pull requests.
1611	async fn manageable_pull(
1612	&self,
1613	actor: &User,
1614	path: &RepoPath,
1615	number: u32,
1616	) -> Result<Outcome<Pull>> {
1617	let pull = check!(self.managed_pull(actor, path, number).await?);
1618	if !pull.status.is_active() {
1619	return Ok(Outcome::fail(
1620	FailureCode::Conflict,
1621	format!("This pull request is already {}.", pull.status.as_str()),
1622	));
1623	}
1624	Ok(Outcome::Ok(pull))
1625	}
1626
1627	/// Brings a pull request up to date with the default branch without a
1628	/// sandbox, where the repos service can do that safely. Whoever could
1629	/// have pushed the merge themselves may ask: whoever opened it (or asked
1630	/// g1t for it), for a fork; anyone who may push, for a branch of the
1631	/// repository. When it needs a
1632	/// real merge, says so, naming the conflicting files if a probe found
1633	/// them, and pushes nothing.
1634	async fn catch_up_pull(&self, a: PullActionArgs) -> Result<Outcome<PullBranchUpdate>> {
1635	let (repo, pull) = check!(self.pull_at(&a.repo, a.number, &Some(a.actor.clone())).await?);
1636	check!(writable(&repo));
1637	if !pull.status.is_active() {
1638	return Ok(Outcome::fail(
1639	FailureCode::Conflict,
1640	format!("This pull request is already {}.", pull.status.as_str()),
1641	));
1642	}
1643	if pull.fork_repo_id.is_some() {
1644	if !pull.is_owned_by(&a.actor.id) {
1645	return Ok(Outcome::fail(
1646	FailureCode::Forbidden,
1647	"Only whoever opened this pull request, or asked g1t for it, can update it.",
1648	));
1649	}
1650	} else {
1651	check!(allowed(Some(&a.actor), &repo, Capability::Push));
1652	}
1653	let updated: Outcome<PullBranchUpdate> = g1t_kit::call(
1654	&self.repos,
1655	"update_pull_branch",
1656	&UpdatePullBranchArgs {
1657	source_id: pull.fork_repo_id.clone().unwrap_or_else(\|\| repo.id.clone()),
1658	branch: pull.branch.clone(),
1659	number: pull.number,
1660	actor: a.actor,
1661	target_branch: pull.base.clone(),
1662	},
1663	)
1664	.await?;
1665	// A probe that found conflicts says more than "both changed it".
1666	if let Outcome::Ok(PullBranchUpdate::NeedsAgent { .. }) = &updated
1667	&& let Some(files) = self.conflicting_files(&pull).await?
1668	&& !files.is_empty()
1669	{
1670	return Ok(Outcome::Ok(PullBranchUpdate::NeedsAgent {
1671	reason: NeedsAgentReason::Conflicting,
1672	detail: "Merging it conflicts.".to_owned(),
1673	paths: files,
1674	}));
1675	}
1676	Ok(updated)
1677	}
1678
1679	async fn update_pull(&self, a: UpdatePullArgs) -> Result<Outcome<Pull>> {
1680	let pull = check!(self.manageable_pull(&a.actor, &a.repo, a.number).await?);
1681	// The branch it merges into, its milestone and its labels first:
1682	// each can be refused, and then nothing else changes.
1683	if a.base.is_some() \|\| a.milestone.is_some() \|\| a.labels.is_some() {
1684	let repo = check!(self.repo(&a.repo, &Some(a.actor.clone())).await?);
1685	if let Some(base) = &a.base {
1686	check!(self.change_base(&a.actor, &repo, &pull, base).await?);
1687	}
1688	if let Some(number) = a.milestone {
1689	check!(self.set_milestone(&a.actor, &repo, &labels::Item::Pull(pull.clone()), number).await?);
1690	}
1691	if let Some(labels) = &a.labels {
1692	check!(self.relabel(&a.actor, &repo, &labels::Item::Pull(pull.clone()), labels).await?);
1693	}
1694	}
1695	let assignees = match a.assignees {
1696	Some(names) => Some(check!(self.valid_assignees(names).await?)),
1697	None => None,
1698	};
1699	// Teams, named `workspace/team`, apart from the people.
1700	let (team_names, a_reviewers) = match a.reviewers {
1701	Some(names) => {
1702	let (teams, people): (Vec<String>, Vec<String>) =
1703	names.into_iter().partition(\|name\| team_reviews::team_name(name).is_some());
1704	(Some(teams), Some(people))
1705	}
1706	None => (None, None),
1707	};
1708	let teams = match team_names {
1709	Some(names) => Some(check!(self.valid_team_reviewers(&a.actor, &a.repo, &pull, names).await?)),
1710	None => None,
1711	};
1712	let reviewers = match a_reviewers {
1713	Some(names) => {
1714	let mut reviewers = check!(self.valid_assignees(names).await?);
1715	// Nobody is asked to review their own, nor what they had g1t make.
1716	reviewers.retain(\|name\| *name != pull.owner().username);
1717	Some(reviewers)
1718	}
1719	None => None,
1720	};
1721	self.db
1722	.prepare(
1723	"UPDATE pulls
1724	SET assignees = COALESCE(?, assignees), updated_at = ?
1725	WHERE id = ?",
1726	)
1727	.bind(&[
1728	optional(&assignees.as_ref().map(serde_json::to_string).transpose()?),
1729	rfc3339(now_ms()).into(),
1730	pull.id.as_str().into(),
1731	])?
1732	.run()
1733	.await?;
1734	if let Some(assignees) = &assignees {
1735	self.note_changes(
1736	&pull.repo_id,
1737	pull.number,
1738	&a.actor,
1739	&pull.assignees,
1740	assignees,
1741	("assigned", "unassigned"),
1742	)
1743	.await?;
1744	}
1745	// Who was newly assigned or asked to review, and whose request was
1746	// withdrawn: the inbox tells them, and webhooks say so.
1747	let newly = \|after: &[String], before: &[String]\| -> Vec<String> {
1748	after.iter().filter(\|name\| !before.contains(name)).cloned().collect()
1749	};
1750	if let Some(assignees) = &assignees {
1751	let added = newly(assignees, &pull.assignees);
1752	if !added.is_empty() {
1753	self.publish(
1754	"pull.assigned",
1755	&pull.repo_id,
1756	&a.actor,
1757	PullEvent {
1758	assignees: Some(assignees.clone()),
1759	added: Some(added),
1760	..Self::pull_event(&pull)
1761	},
1762	)
1763	.await?;
1764	}
1765	}
1766	if reviewers.is_some() \|\| teams.is_some() {
1767	let people = reviewers.unwrap_or_else(\|\| pull.reviewers.clone());
1768	let teams = teams.unwrap_or_else(\|\| pull.team_reviewers.clone());
1769	self.set_reviewers(&pull, people, teams, Some(&a.actor), false).await?;
1770	}
1771	Ok(match self.pull(&pull.repo_id, pull.number).await? {
1772	Some(pull) => Outcome::Ok(pull),
1773	None => no_pull(),
1774	})
1775	}
1776
1777	/// Points an open pull request at another branch to merge into. Needs
1778	/// the Write role. What it would merge, whether it is behind, its
1779	/// mergeability and its checks are all worked out against the new base.
1780	async fn change_base(&self, actor: &User, repo: &Repo, pull: &Pull, base: &str) -> Result<Outcome<()>> {
1781	check!(allowed(Some(actor), repo, Capability::Push));
1782	let base = base.trim();
1783	if base.is_empty() {
1784	return Ok(Outcome::fail(FailureCode::Invalid, "Name the branch it should merge into."));
1785	}
1786	let before = pull.base_branch(&repo.default_branch).to_owned();
1787	if base == before {
1788	return Ok(Outcome::Ok(()));
1789	}
1790	if pull.fork_repo_id.is_none() && pull.branch.as_deref() == Some(base) {
1791	return Ok(Outcome::fail(
1792	FailureCode::Invalid,
1793	format!("A pull request cannot merge {base} into itself. Choose another base."),
1794	));
1795	}
1796	let exists: Option<String> = g1t_kit::call(
1797	&self.repos,
1798	"head",
1799	&HeadArgs { repo_id: repo.id.clone(), branch: base.to_owned() },
1800	)
1801	.await?;
1802	if exists.is_none() {
1803	return Ok(Outcome::fail(FailureCode::NotFound, format!("There is no branch named {base} to merge into.")));
1804	}
1805	// In the merge queue it was headed for the default branch; it
1806	// leaves the queue for another base.
1807	let left_queue = self
1808	.leave(&pull.repo_id, pull, QueueState::Removed, Some("Its base branch changed."))
1809	.await?;
1810	let stored = stored_base(base, repo);
1811	self.db
1812	.prepare(
1813	"UPDATE pulls
1814	SET base_branch = ?, updated_at = ?, land_requested = NULL, land_requested_at = NULL, behind = NULL,
1815	mergeable = NULL, mergeable_key = NULL, conflicts = NULL
1816	WHERE id = ?",
1817	)
1818	.bind(&[optional(&stored), rfc3339(now_ms()).into(), pull.id.as_str().into()])?
1819	.run()
1820	.await?;
1821	self.note(
1822	&pull.repo_id,
1823	pull.number,
1824	(&actor.id, &actor.username),
1825	&format!("changed the base branch from `{before}` to `{base}`"),
1826	)
1827	.await?;
1828	self.publish(
1829	"pull.base_changed",
1830	&pull.repo_id,
1831	actor,
1832	PullEvent { base: Some(base.to_owned()), ..Self::pull_event(pull) },
1833	)
1834	.await?;
1835	if left_queue {
1836	self.publish_as(
1837	"queue.changed",
1838	&pull.repo_id,
1839	None,
1840	g1t_contracts::events::QueueChanged { repo_id: pull.repo_id.clone() },
1841	)
1842	.await?;
1843	}
1844	// Its merge ref, and whether it merges cleanly, against the new base.
1845	if let Some(moved) = self.pull_by_id(&pull.id).await? {
1846	self.keep_pull_refs(&moved).await;
1847	if let Err(error) = self.assess_mergeability(&moved).await {
1848	worker::console_warn!("mergeability of {}: {error}", pull.id);
1849	}
1850	}
1851	Ok(Outcome::Ok(()))
1852	}
1853
1854	/// Marks a draft ready for review, or updates the description of one
1855	/// that already is.
1856	async fn ready_pull(&self, a: PullActionArgs) -> Result<Outcome<Pull>> {
1857	let mut pull = check!(self.manageable_pull(&a.actor, &a.repo, a.number).await?);
1858	let summary = Some(a.summary.trim().to_owned()).filter(\|summary\| !summary.is_empty());
1859	let now = rfc3339(now_ms());
1860	self.db
1861	.prepare(
1862	"UPDATE pulls SET status = 'open', body = COALESCE(?, body), updated_at = ?
1863	WHERE id = ?",
1864	)
1865	.bind(&[
1866	optional(&summary),
1867	now.as_str().into(),
1868	pull.id.as_str().into(),
1869	])?
1870	.run()
1871	.await?;
1872	if pull.status == PullStatus::Draft {
1873	// The head as it is now: the push that came just before may not
1874	// have reached `head_commit` yet, and workflows run on it.
1875	let commit = self.live_head(&pull).await?.or_else(\|\| pull.head_commit.clone());
1876	self.publish(
1877	"pull.ready",
1878	&pull.repo_id,
1879	&a.actor,
1880	PullEvent {
1881	commit,
1882	..Self::pull_event(&pull)
1883	},
1884	)
1885	.await?;
1886	}
1887	if pull.status == PullStatus::Draft {
1888	self.note(
1889	&pull.repo_id,
1890	pull.number,
1891	(&a.actor.id, &a.actor.username),
1892	"marked this ready for review",
1893	)
1894	.await?;
1895	}
1896	pull.status = PullStatus::Open;
1897	pull.body = summary.or(pull.body);
1898	pull.updated_at = now;
1899	// A draft's code owners are asked once it is ready.
1900	self.refresh_code_owners(&pull).await;
1901	if let Some(fresh) = self.pull(&pull.repo_id, pull.number).await? {
1902	pull.reviewers = fresh.reviewers;
1903	pull.team_reviewers = fresh.team_reviewers;
1904	}
1905	Ok(Outcome::Ok(pull))
1906	}
1907
1908	async fn close_pull(&self, a: PullActionArgs) -> Result<Outcome<Pull>> {
1909	let mut pull = check!(self.manageable_pull(&a.actor, &a.repo, a.number).await?);
1910	let now = rfc3339(now_ms());
1911	self.db
1912	// What it was closed as, so that reopening brings that back.
1913	.prepare("UPDATE pulls SET status = 'closed', closed_from = ?, updated_at = ? WHERE id = ?")
1914	.bind(&[pull.status.as_str().into(), now.as_str().into(), pull.id.as_str().into()])?
1915	.run()
1916	.await?;
1917	self.publish(
1918	"pull.closed",
1919	&pull.repo_id,
1920	&a.actor,
1921	Self::pull_event(&pull),
1922	)
1923	.await?;
1924	self.note(
1925	&pull.repo_id,
1926	pull.number,
1927	(&a.actor.id, &a.actor.username),
1928	"closed this",
1929	)
1930	.await?;
1931	// A closed pull request leaves the merge queue.
1932	if self
1933	.leave(&pull.repo_id, &pull, QueueState::Removed, Some("It was closed."))
1934	.await?
1935	{
1936	self.publish_as(
1937	"queue.changed",
1938	&pull.repo_id,
1939	None,
1940	g1t_contracts::events::QueueChanged {
1941	repo_id: pull.repo_id.clone(),
1942	},
1943	)
1944	.await?;
1945	}
1946	pull.status = PullStatus::Closed;
1947	pull.updated_at = now;
1948	Ok(Outcome::Ok(pull))
1949	}
1950
1951	/// Opens a closed pull request again: as the draft it was, if it was
1952	/// closed as one, else ready for review. A merged one stays merged.
1953	async fn reopen_pull(&self, a: PullActionArgs) -> Result<Outcome<Pull>> {
1954	let mut pull = check!(self.managed_pull(&a.actor, &a.repo, a.number).await?);
1955	if let Some(refusal) = reopen_refusal(pull.status) {
1956	return Ok(Outcome::fail(FailureCode::Conflict, refusal));
1957	}
1958	// The head as it is now, which workflows run on. A branch of the
1959	// repository that was deleted since leaves nothing to reopen; a
1960	// fork removed after the close is made again (repos, `pull.reopened`).
1961	let head = self.live_head(&pull).await?;
1962	if head.is_none() && pull.fork_repo_id.is_none() {
1963	return Ok(Outcome::fail(
1964	FailureCode::Conflict,
1965	format!(
1966	"The branch {} no longer exists. Push it again to reopen this pull request.",
1967	pull.branch.as_deref().unwrap_or_default()
1968	),
1969	));
1970	}
1971	let closed_from: Option<ClosedFrom> = self
1972	.db
1973	.prepare("SELECT closed_from FROM pulls WHERE id = ?")
1974	.bind(&[pull.id.as_str().into()])?
1975	.first(None)
1976	.await?;
1977	let status = reopened_status(closed_from.and_then(\|row\| row.closed_from).as_deref());
1978	let now = rfc3339(now_ms());
1979	self.db
1980	.prepare("UPDATE pulls SET status = ?, closed_from = NULL, superseded_by = NULL, updated_at = ? WHERE id = ?")
1981	.bind(&[status.as_str().into(), now.as_str().into(), pull.id.as_str().into()])?
1982	.run()
1983	.await?;
1984	self.publish(
1985	"pull.reopened",
1986	&pull.repo_id,
1987	&a.actor,
1988	PullEvent {
1989	commit: head.or_else(\|\| pull.head_commit.clone()),
1990	..Self::pull_event(&pull)
1991	},
1992	)
1993	.await?;
1994	self.note(
1995	&pull.repo_id,
1996	pull.number,
1997	(&a.actor.id, &a.actor.username),
1998	"reopened this",
1999	)
2000	.await?;
2001	pull.status = status;
2002	pull.superseded_by = None;
2003	pull.updated_at = now;
2004	// Its refs in the repository, and whether it still merges cleanly,
2005	// now that it is open again.
2006	self.keep_pull_refs(&pull).await;
2007	if let Err(error) = self.assess_mergeability(&pull).await {
2008	worker::console_warn!("mergeability of {}: {error}", pull.id);
2009	}
2010	Ok(Outcome::Ok(pull))
2011	}
2012
2013	/// Turns a pull request that is ready for review back into a draft: it
2014	/// cannot be merged until it is marked ready again, and it leaves the
2015	/// merge queue and any merge that was waiting for it to catch up.
2016	async fn convert_pull_to_draft(&self, a: PullActionArgs) -> Result<Outcome<Pull>> {
2017	let mut pull = check!(self.managed_pull(&a.actor, &a.repo, a.number).await?);
2018	if let Some(refusal) = draft_refusal(pull.status) {
2019	return Ok(Outcome::fail(FailureCode::Conflict, refusal));
2020	}
2021	let now = rfc3339(now_ms());
2022	self.db
2023	.prepare(
2024	"UPDATE pulls SET status = 'draft', land_requested = NULL, land_requested_at = NULL, updated_at = ?
2025	WHERE id = ?",
2026	)
2027	.bind(&[now.as_str().into(), pull.id.as_str().into()])?
2028	.run()
2029	.await?;
2030	self.publish(
2031	"pull.converted_to_draft",
2032	&pull.repo_id,
2033	&a.actor,
2034	Self::pull_event(&pull),
2035	)
2036	.await?;
2037	self.note(
2038	&pull.repo_id,
2039	pull.number,
2040	(&a.actor.id, &a.actor.username),
2041	"marked this as a draft",
2042	)
2043	.await?;
2044	if self
2045	.leave(&pull.repo_id, &pull, QueueState::Removed, Some("It was marked as a draft."))
2046	.await?
2047	{
2048	self.publish_as(
2049	"queue.changed",
2050	&pull.repo_id,
2051	None,
2052	g1t_contracts::events::QueueChanged {
2053	repo_id: pull.repo_id.clone(),
2054	},
2055	)
2056	.await?;
2057	}
2058	pull.status = PullStatus::Draft;
2059	pull.updated_at = now;
2060	Ok(Outcome::Ok(pull))
2061	}
2062
2063	/// Lands the pull request on the repository's default branch. Unless
2064	/// told to keep it open, that resolves the issue it was for: the issue
2065	/// closes naming this pull request, and the others still in progress
2066	/// for it close as superseded.
2067	async fn merge_pull(&self, a: PullActionArgs) -> Result<Outcome<Pull>> {
2068	let viewer = Some(a.actor.clone());
2069	let (repo, pull) = check!(self.pull_at(&a.repo, a.number, &viewer).await?);
2070	check!(writable(&repo));
2071	match pull.status {
2072	PullStatus::Open => {}
2073	PullStatus::Draft => {
2074	return Ok(Outcome::fail(
2075	FailureCode::Conflict,
2076	"This pull request is still a draft. Mark it ready for review first.",
2077	));
2078	}
2079	status => {
2080	return Ok(Outcome::fail(
2081	FailureCode::Conflict,
2082	format!("This pull request is already {}.", status.as_str()),
2083	));
2084	}
2085	}
2086	let base = pull.base_branch(&repo.default_branch).to_owned();
2087	// The rules of the branch it merges into, as they stack: the one
2088	// gate for the merge button, the API, MCP, auto-merge and the queue,
2089	// for a person's pull request and an agent's alike. A bypass counts
2090	// when the merger asks for it, or for g1t when a ruleset lists it.
2091	let gate = self.merge_gate(&repo, &pull, Some(&a.actor), a.ignore_checks, true).await?;
2092	let bypassable = gate.bypassable();
2093	let gate = if a.bypass_rules \|\| a.actor.is_system() { gate } else { gate.without_bypass() };
2094	let settings = rulesets::overlay(self.settings(&repo.id).await?, &gate.requirements, base == repo.default_branch);
2095	if pull.check_status == Some(CheckStatus::Failed) && !(a.ignore_checks && settings.allow_ignoring_checks) {
2096	return Ok(Outcome::fail(
2097	FailureCode::Conflict,
2098	"It failed in the merge queue; push a fix to try again.",
2099	));
2100	}
2101	if let Some(refusal) = gate.refusal() {
2102	if access::can(Some(&a.actor), &repo, Capability::Merge) {
2103	self.record_merge_evaluations(&repo, &pull, &gate).await;
2104	}
2105	let offer = if bypassable && !a.bypass_rules {
2106	" You may bypass these rules: merge again and ask to bypass them (bypass_rules)."
2107	} else {
2108	""
2109	};
2110	return Ok(Outcome::fail(FailureCode::Conflict, format!("{refusal}{offer}")));
2111	}
2112	// Known ahead of time to conflict: neither a merge nor the queue
2113	// would get through, so say what has to be resolved now.
2114	if let Some(files) = self.conflicting_files(&pull).await? {
2115	let named = if files.is_empty() {
2116	String::new()
2117	} else {
2118	format!(" in {}", files.join(", "))
2119	};
2120	return Ok(Outcome::fail(
2121	FailureCode::Conflict,
2122	format!(
2123	"This branch has conflicts with {base}{named} that must be resolved first. Merge {base} into it, fix them and push."
2124	),
2125	));
2126	}
2127
2128	// A repository that merges through a queue: it joins the queue, and
2129	// lands once its state together with everything ahead has passed.
2130	if settings.merge_queue {
2131	if !a.actor.verified {
2132	return Ok(Outcome::fail(FailureCode::Forbidden, UNVERIFIED));
2133	}
2134	check!(allowed(Some(&a.actor), &repo, Capability::Merge));
2135	self.record_merge_evaluations(&repo, &pull, &gate).await;
2136	return self.enqueue(&repo, &pull, &a.actor, a.keep_issue_open).await;
2137	}
2138
2139	// The default branch has moved under it. Unless the repository
2140	// insists on that being dealt with first, bring it up to date and
2141	// land it when that is done.
2142	if self.is_behind(&repo.id, &pull).await? {
2143	if settings.require_up_to_date {
2144	return Ok(Outcome::fail(
2145	FailureCode::Conflict,
2146	format!(
2147	"{base} has moved since this pull request was made, and this repository requires pull requests to be up to date before they merge. Catch up with {base} first."
2148	),
2149	));
2150	}
2151	if !a.actor.verified {
2152	return Ok(Outcome::fail(FailureCode::Forbidden, UNVERIFIED));
2153	}
2154	check!(allowed(Some(&a.actor), &repo, Capability::Merge));
2155	self.record_merge_evaluations(&repo, &pull, &gate).await;
2156	// Brought up to date here, without a sandbox: the merge is
2157	// recorded first, and lands when the push that catches it up
2158	// arrives (lifecycle.rs `land_if_requested`).
2159	self.request_landing(&pull, &a.actor, a.keep_issue_open)
2160	.await?;
2161	let updated: Outcome<PullBranchUpdate> = g1t_kit::call(
2162	&self.repos,
2163	"update_pull_branch",
2164	&UpdatePullBranchArgs {
2165	source_id: pull.fork_repo_id.clone().unwrap_or_else(\|\| repo.id.clone()),
2166	branch: pull.branch.clone(),
2167	number: pull.number,
2168	actor: a.actor.clone(),
2169	target_branch: pull.base.clone(),
2170	},
2171	)
2172	.await?;
2173	match updated {
2174	Outcome::Ok(PullBranchUpdate::Updated { .. }) => return Ok(Outcome::Ok(pull)),
2175	// Moved again in between, or already caught up: land it now.
2176	Outcome::Ok(PullBranchUpdate::UpToDate { .. }) => self.forget_landing(&pull.id).await?,
2177	Outcome::Ok(PullBranchUpdate::NeedsAgent { paths, .. }) => {
2178	self.forget_landing(&pull.id).await?;
2179	let named = if paths.is_empty() { String::new() } else { format!(" ({})", paths.join(", ")) };
2180	return Ok(Outcome::fail(
2181	FailureCode::Conflict,
2182	format!(
2183	"{base} has moved since this pull request was made, and both changed the same files{named}. Merge {base} into it, resolve any conflicts, push, and merge again."
2184	),
2185	));
2186	}
2187	Outcome::Fail(failure) => {
2188	self.forget_landing(&pull.id).await?;
2189	return Ok(Outcome::Fail(failure));
2190	}
2191	}
2192	}
2193
2194	// Whether the actor may write to the repository is decided by repos.
2195	let landed: Outcome<Landed> = g1t_kit::call(
2196	&self.repos,
2197	"land",
2198	&LandArgs {
2199	// A pull request from a branch lands from the repository itself.
2200	source_id: pull.fork_repo_id.clone().unwrap_or_else(\|\| repo.id.clone()),
2201	branch: pull.branch.clone(),
2202	actor: a.actor.clone(),
2203	target_branch: Some(base.clone()),
2204	},
2205	)
2206	.await?;
2207	let landed = check!(landed);
2208	self.record_merge_evaluations(&repo, &pull, &gate).await;
2209	Ok(Outcome::Ok(
2210	self.record_merge(&repo, pull, &a.actor, a.keep_issue_open, landed)
2211	.await?,
2212	))
2213	}
2214
2215	/// Records a pull request as merged once the default branch holds it:
2216	/// closes its issue, supersedes the others for it, and says so.
2217	pub(crate) async fn record_merge(
2218	&self,
2219	repo: &Repo,
2220	mut pull: Pull,
2221	actor: &User,
2222	keep_issue_open: bool,
2223	landed: Landed,
2224	) -> Result<Pull> {
2225	// Only a merge into the default branch resolves the issue: into
2226	// another branch, the work has not landed yet.
2227	let keep_issue_open = keep_issue_open \|\| !pull.targets_default(&repo.default_branch);
2228	let issue = match pull.issue {
2229	Some(number) if !keep_issue_open => self
2230	.issue(&repo.id, number)
2231	.await?
2232	.filter(\|issue\| issue.state == State::Open),
2233	_ => None,
2234	};
2235	let now = rfc3339(now_ms());
2236	let mut statements = vec![
2237	self.db
2238	.prepare(
2239	"UPDATE pulls
2240	SET status = 'merged', head_commit = ?, merge_base = ?, merged_by = ?,
2241	merged_at = ?, updated_at = ?
2242	WHERE id = ?",
2243	)
2244	.bind(&[
2245	landed.commit.as_str().into(),
2246	optional(&landed.previous),
2247	actor.username.as_str().into(),
2248	now.as_str().into(),
2249	now.as_str().into(),
2250	pull.id.as_str().into(),
2251	])?,
2252	];
2253	if let Some(issue) = &issue {
2254	statements.push(
2255	self.db
2256	.prepare(
2257	"UPDATE issues
2258	SET state = 'closed', reason = 'completed', resolved_by = ?,
2259	closed_at = ?, updated_at = ?
2260	WHERE id = ?",
2261	)
2262	.bind(&[
2263	pull.number.into(),
2264	now.as_str().into(),
2265	now.as_str().into(),
2266	issue.id.as_str().into(),
2267	])?,
2268	);
2269	statements.push(
2270	self.db
2271	.prepare(
2272	"UPDATE pulls SET status = 'closed', superseded_by = ?, updated_at = ?
2273	WHERE issue_id = ? AND id != ? AND status IN ('draft', 'open')",
2274	)
2275	.bind(&[
2276	pull.number.into(),
2277	now.as_str().into(),
2278	issue.id.as_str().into(),
2279	pull.id.as_str().into(),
2280	])?,
2281	);
2282	}
2283	self.db.batch(statements).await?;
2284
2285	self.publish(
2286	"pull.merged",
2287	&repo.id,
2288	actor,
2289	PullEvent {
2290	commit: Some(landed.commit.clone()),
2291	..Self::pull_event(&pull)
2292	},
2293	)
2294	.await?;
2295	if let Some(issue) = &issue {
2296	self.publish(
2297	"issue.closed",
2298	&repo.id,
2299	actor,
2300	IssueEvent {
2301	reason: Some(IssueReason::Completed.as_str()),
2302	resolved_by: Some(pull.number),
2303	..Self::issue_event(issue)
2304	},
2305	)
2306	.await?;
2307	}
2308
2309	let who = (actor.id.as_str(), actor.username.as_str());
2310	self.note(&repo.id, pull.number, who, "merged this").await?;
2311	if let Some(issue) = &issue {
2312	self.note(
2313	&repo.id,
2314	issue.number,
2315	who,
2316	&format!("closed this by merging #{}", pull.number),
2317	)
2318	.await?;
2319	}
2320	let through_queue = self.landed_through_queue(&pull.id).await?;
2321	self.keep_decision_quietly(
2322	&repo.id,
2323	decisions::NewDecision {
2324	issue: pull.issue,
2325	pull: Some(pull.number),
2326	session_id: None,
2327	actor: (!decisions::is_g1t(actor)).then_some(actor),
2328	kind: g1t_contracts::decisions::DecisionKind::Merge,
2329	confidence: decisions::merge_confidence(through_queue, actor),
2330	summary: if through_queue {
2331	format!("#{} landed through the merge queue.", pull.number)
2332	} else {
2333	format!("{} merged #{}.", actor.username, pull.number)
2334	},
2335	evidence: serde_json::json!({
2336	"commit": landed.commit,
2337	"previous": landed.previous,
2338	"merge_queue": through_queue,
2339	}),
2340	},
2341	)
2342	.await;
2343	// A merge into the default branch clears its issue for the issues
2344	// it blocks, and ends the claims on it when it closed it.
2345	if let Some(number) = pull.issue
2346	&& pull.targets_default(&repo.default_branch)
2347	{
2348	self.issue_resolved(&repo.id, number, actor, issue.is_some(), Some(pull.number)).await;
2349	}
2350	pull.status = PullStatus::Merged;
2351	pull.head_commit = Some(landed.commit.clone());
2352	pull.merge_base = landed.previous;
2353	pull.merged_by = Some(actor.username.clone());
2354	pull.merged_at = Some(now.clone());
2355	pull.updated_at = now;
2356	Ok(pull)
2357	}
2358
2359	async fn list_active_pulls(&self, a: ViewerArgs) -> Result<Vec<ActivePull>> {
2360	let Some(viewer) = a.viewer else {
2361	return Ok(Vec::new());
2362	};
2363	// The pull requests and their issues, in one round trip: their own,
2364	// and those g1t made for them (Pull::owner).
2365	let author = [JsValue::from(viewer.id.as_str())];
2366	let found = self
2367	.timing
2368	.db(
2369	2,
2370	self.db.batch(vec![
2371	self.db
2372	.prepare(format!(
2373	"SELECT {PULL_COLUMNS} FROM pulls
2374	WHERE COALESCE(requested_by_id, author_id) = ?1 AND status IN ('draft', 'open')
2375	ORDER BY updated_at DESC LIMIT 50"
2376	))
2377	.bind(&author)?,
2378	self.db
2379	.prepare(format!(
2380	"SELECT {ISSUE_COLUMNS} FROM issues WHERE issues.id IN (
2381	SELECT issue_id FROM pulls
2382	WHERE COALESCE(requested_by_id, author_id) = ?1 AND status IN ('draft', 'open') AND issue_id IS NOT NULL
2383	ORDER BY updated_at DESC LIMIT 50)"
2384	))
2385	.bind(&author)?,
2386	]),
2387	)
2388	.await?;
2389	let (Some(found), Some(issues)) = (found.first(), found.get(1)) else {
2390	return Ok(Vec::new());
2391	};
2392	let snapshots = found.results::<Snapshot>()?;
2393	let pulls: Vec<Pull> = found.results::<PullRow>()?.into_iter().map(Pull::from).collect();
2394	let issues: Vec<Issue> = issues.results::<IssueRow>()?.into_iter().map(Issue::from).collect();
2395	let issues = &issues;
2396	// Where each stands: the remembered assessment when there is one,
2397	// and worked out otherwise.
2398	try_join_all(pulls.into_iter().zip(snapshots).map(\|(pull, snapshot)\| async move {
2399	let issue = pull.issue.and_then(\|number\| {
2400	issues
2401	.iter()
2402	.find(\|issue\| issue.repo_id == pull.repo_id && issue.number == number)
2403	.cloned()
2404	});
2405	// Only a pull request g1t is seeing through has a lifecycle.
2406	let lifecycle = if !lifecycle::made_by_g1t(&pull) \|\| snapshot.managed == 0 {
2407	None
2408	} else if let (Some(stage), Some(detail)) = (snapshot.stage, snapshot.stage_detail) {
2409	Some(Lifecycle {
2410	stage,
2411	detail,
2412	revisions: snapshot.revisions,
2413	})
2414	} else {
2415	let behind = self.is_behind(&pull.repo_id, &pull).await?;
2416	self.assess(&pull, &issue, behind)
2417	.await?
2418	.map(\|(lifecycle, _)\| lifecycle)
2419	};
2420	Ok::<_, worker::Error>(ActivePull {
2421	pull,
2422	issue,
2423	lifecycle,
2424	})
2425	}))
2426	.await
2427	}
2428
2429	// --- Sessions ----------------------------------------------------------
2430
2431	async fn append_session(&self, a: AppendSessionArgs) -> Result<Outcome<Appended>> {
2432	if a.entries.is_empty() {
2433	return Ok(Outcome::Ok(Appended { count: 0 }));
2434	}
2435	if a.entries.len() > MAX_ENTRY_BATCH {
2436	return Ok(Outcome::fail(
2437	FailureCode::Invalid,
2438	format!("Send at most {MAX_ENTRY_BATCH} entries at a time."),
2439	));
2440	}
2441	let viewer = Some(a.actor.clone());
2442	let (_, pull) = check!(self.pull_at(&a.repo, a.number, &viewer).await?);
2443	if !pull.is_owned_by(&a.actor.id) {
2444	return Ok(Outcome::fail(
2445	FailureCode::Forbidden,
2446	"Only whoever opened a pull request, or asked g1t for it, can record its session.",
2447	));
2448	}
2449
2450	let now = rfc3339(now_ms());
2451	let count = a.entries.len() as u32;
2452	let mut statements = Vec::with_capacity(a.entries.len() + 1);
2453	for entry in a.entries {
2454	let kind = serde_json::to_value(entry.kind)?;
2455	let text: String = entry.text.chars().take(MAX_ENTRY_CHARS).collect();
2456	// Each insert takes the next sequence number itself, so two
2457	// writers appending at once cannot collide.
2458	statements.push(
2459	self.db
2460	.prepare(
2461	"INSERT INTO session_entries (pull_id, seq, kind, text, tool, \"commit\", at)
2462	SELECT ?, COALESCE(MAX(seq), 0) + 1, ?, ?, ?, ?, ?
2463	FROM session_entries WHERE pull_id = ?",
2464	)
2465	.bind(&[
2466	pull.id.as_str().into(),
2467	kind.as_str().unwrap_or("note").into(),
2468	text.into(),
2469	optional(&entry.tool),
2470	optional(&entry.commit.or_else(\|\| pull.head_commit.clone())),
2471	now.as_str().into(),
2472	pull.id.as_str().into(),
2473	])?,
2474	);
2475	}
2476	statements.push(
2477	self.db
2478	.prepare("UPDATE pulls SET updated_at = ? WHERE id = ?")
2479	.bind(&[now.as_str().into(), pull.id.as_str().into()])?,
2480	);
2481	self.db.batch(statements).await?;
2482	self.publish(
2483	"session.appended",
2484	&pull.repo_id,
2485	&a.actor,
2486	SessionAppended {
2487	pull_id: pull.id.clone(),
2488	repo_id: pull.repo_id.clone(),
2489	number: pull.number,
2490	count,
2491	},
2492	)
2493	.await?;
2494	Ok(Outcome::Ok(Appended { count }))
2495	}
2496
2497	async fn read_session(&self, a: ViewArgs) -> Result<Outcome<Vec<SessionEntry>>> {
2498	let (_, pull) = check!(self.pull_at(&a.repo, a.number, &a.viewer).await?);
2499	let rows = self
2500	.db
2501	.prepare(
2502	"SELECT seq, kind, text, tool, \"commit\", at FROM session_entries
2503	WHERE pull_id = ? AND seq > ? ORDER BY seq LIMIT ?",
2504	)
2505	.bind(&[pull.id.into(), a.after_seq.into(), SESSION_PAGE.into()])?
2506	.all()
2507	.await?
2508	.results::<SessionRow>()?;
2509	Ok(Outcome::Ok(
2510	rows.into_iter().map(SessionEntry::from).collect(),
2511	))
2512	}
2513
2514	/// A push moves the head of the pull request it concerns: the one whose
2515	/// fork was pushed to, or the one opened from the branch that moved.
2516	async fn on_event(&self, event: &Event) -> Result<()> {
2517	// A new repository starts with the default labels.
2518	if event.kind == "repo.created"
2519	&& let Some(repo_id) = event.repo_id.as_deref()
2520	{
2521	return self.seed_labels(repo_id).await;
2522	}
2523	// Pull requests into the branch that became the default merge into
2524	// the default branch, which is stored as none.
2525	if event.kind == "repo.default_branch_changed"
2526	&& let (Some(repo_id), Some(to)) = (event.repo_id.as_deref(), event.data["to"].as_str())
2527	{
2528	self.db
2529	.prepare(
2530	"UPDATE pulls SET base_branch = NULL
2531	WHERE repo_id = ? AND base_branch = ? AND status IN ('draft', 'open')",
2532	)
2533	.bind(&[repo_id.into(), to.into()])?
2534	.run()
2535	.await?;
2536	return Ok(());
2537	}
2538	if event.kind != "git.push" {
2539	return Ok(());
2540	}
2541	let (Some(repo_id), Some(after), Some(git_ref)) = (
2542	event.repo_id.as_deref(),
2543	event.data["after"].as_str(),
2544	event.data["ref"].as_str(),
2545	) else {
2546	return Ok(());
2547	};
2548	let now = rfc3339(now_ms());
2549	// Who moved it, for rules about the most recent push.
2550	let pusher: JsValue = event.actor.as_deref().map_or(JsValue::NULL, Into::into);
2551	// The head moved, so whatever the checks said no longer applies, and
2552	// whatever step g1t was waiting on has been taken.
2553	let moved = "UPDATE pulls
2554	SET head_commit = ?, updated_at = ?, head_pushed_by = ?, head_pushed_at = ?, check_status = NULL, check_run_id = NULL,
2555	working_on = NULL, working_until = NULL, stalled = NULL";
2556	let active = "status IN ('draft', 'open') AND head_commit IS NOT ?";
2557	let returning =
2558	"RETURNING id, repo_id, number, issue_number, status, author_id, author_name, requested_by_id, requested_by_name";
2559	let mut pulls: Vec<MovedRow> = Vec::new();
2560	// A fork carries its pull request on its default branch.
2561	if event.data["defaultBranch"].as_bool() == Some(true) {
2562	pulls.extend(
2563	self.db
2564	.prepare(format!(
2565	"{moved} WHERE fork_repo_id = ? AND {active} {returning}"
2566	))
2567	.bind(&[
2568	after.into(),
2569	now.as_str().into(),
2570	pusher.clone(),
2571	now.as_str().into(),
2572	repo_id.into(),
2573	after.into(),
2574	])?
2575	.all()
2576	.await?
2577	.results::<MovedRow>()?,
2578	);
2579	}
2580	if let Some(branch) = git_ref.strip_prefix("refs/heads/") {
2581	pulls.extend(
2582	self.db
2583	.prepare(format!(
2584	"{moved} WHERE repo_id = ? AND source_branch = ? AND {active} {returning}"
2585	))
2586	.bind(&[
2587	after.into(),
2588	now.as_str().into(),
2589	pusher.clone(),
2590	now.as_str().into(),
2591	repo_id.into(),
2592	branch.into(),
2593	after.into(),
2594	])?
2595	.all()
2596	.await?
2597	.results::<MovedRow>()?,
2598	);
2599	}
2600	// What each now changes, so overlaps show while the work is under way.
2601	let mut moved_pulls = Vec::new();
2602	for moved in &pulls {
2603	if let Some(mut pull) = self.pull_by_id(&moved.id).await? {
2604	pull.files = self.refresh_files(&pull).await?;
2605	// Owners of files it now changes are asked too.
2606	self.refresh_code_owners(&pull).await;
2607	// A new overlap with another pull request is announced.
2608	self.announce_overlaps(&pull, event.actor.clone()).await;
2609	moved_pulls.push(pull);
2610	}
2611	}
2612	// The harness sessions that pushed the branch, or that its new
2613	// commits name in trailers, are tied to its pull requests.
2614	if let Some(branch) = git_ref.strip_prefix("refs/heads/") {
2615	let commits = event.data["commits"].as_u64().and_then(\|n\| u32::try_from(n).ok());
2616	self.link_push(repo_id, branch, after, commits, &moved_pulls).await;
2617	}
2618	// A merge that was waiting for this push to bring it up to date.
2619	for moved in &pulls {
2620	self.land_if_requested(&moved.id).await?;
2621	}
2622	// Whether each still merges cleanly, and, when a default branch
2623	// moved, every open pull request into it.
2624	let moved_ids: Vec<String> = pulls.iter().map(\|pull\| pull.id.clone()).collect();
2625	let into = match event.data["defaultBranch"].as_bool() {
2626	Some(true) => mergeability::Moved::DefaultBranch,
2627	_ => match git_ref.strip_prefix("refs/heads/") {
2628	Some(branch) => mergeability::Moved::Branch(branch),
2629	None => mergeability::Moved::Nothing,
2630	},
2631	};
2632	self.after_push(repo_id, into, &moved_ids).await;
2633	// A draft is announced when it is marked ready instead.
2634	for pull in pulls
2635	.into_iter()
2636	.filter(\|pull\| pull.status == PullStatus::Open)
2637	{
2638	self.publish_as(
2639	"pull.updated",
2640	&pull.repo_id,
2641	event.actor.clone(),
2642	PullEvent {
2643	author: Some(g1t_contracts::credentials::Principal { id: pull.author_id, username: pull.author_name }),
2644	requested_by: pull
2645	.requested_by_id
2646	.zip(pull.requested_by_name)
2647	.map(\|(id, username)\| g1t_contracts::credentials::Principal { id, username }),
2648	pull_id: pull.id,
2649	repo_id: pull.repo_id.clone(),
2650	number: pull.number,
2651	issue: pull.issue_number,
2652	commit: Some(after.to_owned()),
2653	..PullEvent::default()
2654	}
2655	.carrying(&event.data),
2656	)
2657	.await?;
2658	}
2659	Ok(())
2660	}
2661	}
2662
2663	/// An event's data that carries on what caused the event it follows from.
2664	trait Carrying: serde::Serialize + Sized {
2665	/// As JSON, marked as a workflow job's doing when `cause` was.
2666	fn carrying(self, cause: &serde_json::Value) -> serde_json::Value {
2667	g1t_contracts::events::carried(self, cause)
2668	}
2669	}
2670
2671	impl Carrying for PullEvent {}
2672
2673	fn service(env: &Env) -> Result<Work> {
2674	Ok(Work {
2675	db: env.d1("DB")?,
2676	identity: env.service("IDENTITY")?,
2677	repos: env.service("REPOS")?,
2678	events: env.service("EVENTS")?,
2679	actions: env.service("ACTIONS")?,
2680	timing: g1t_kit::d1::Timing::default(),
2681	prefetched: std::cell::RefCell::new(None),
2682	known_repos: std::cell::RefCell::new(std::collections::HashMap::new()),
2683	})
2684	}
2685
2686	#[event(fetch)]
2687	async fn fetch(mut request: Request, env: Env, _ctx: Context) -> Result<Response> {
2688	let Some(method) = rpc_method(&request) else {
2689	return Response::error("Not found", 404);
2690	};
2691	// A replica near the caller when it asks for one (crates/kit/src/d1.rs).
2692	let (db, served) = g1t_kit::d1::open(&env, "DB", &request)?;
2693	let body: serde_json::Value = request.json().await?;
2694	let mut work = service(&env)?;
2695	work.db = db;
2696
2697	let answered = match method.as_str() {
2698	"open_issue" => reply(&work.open_issue(args(body)?).await?),
2699	"list_issues" => reply(&work.list_issues(args(body)?).await?),
2700	"get_issue" => reply(&work.get_issue(args(body)?).await?),
2701	"update_issue" => reply(&work.update_issue(args(body)?).await?),
2702	"close_issue" => reply(&work.close_issue(args(body)?).await?),
2703	"reopen_issue" => reply(&work.reopen_issue(args(body)?).await?),
2704	"list_labels" => reply(&work.list_labels(args(body)?).await?),
2705	"save_label" => reply(&work.save_label(args(body)?).await?),
2706	"delete_label" => reply(&work.delete_label(args(body)?).await?),
2707	"add_default_labels" => reply(&work.add_default_labels(args(body)?).await?),
2708	"set_labels" => reply(&work.set_labels(args(body)?).await?),
2709	"list_milestones" => reply(&work.list_milestones(args(body)?).await?),
2710	"get_milestone" => reply(&work.get_milestone(args(body)?).await?),
2711	"save_milestone" => reply(&work.save_milestone(args(body)?).await?),
2712	"delete_milestone" => reply(&work.delete_milestone(args(body)?).await?),
2713	"counts" => reply(&work.counts(args(body)?).await?),
2714	"add_comment" => reply(&work.add_comment(args(body)?).await?),
2715	"edit_comment" => reply(&work.edit_comment(args(body)?).await?),
2716	"delete_comment" => reply(&work.delete_comment(args(body)?).await?),
2717	"seen_checks" => reply(&work.seen_checks(args(body)?).await?),
2718	"report_checks" => reply(&work.report_checks(args(body)?).await?),
2719	"set_commit_status" => reply(&work.set_commit_status(args(body)?).await?),
2720	"create_commit_status" => reply(&work.create_commit_status(args(body)?).await?),
2721	"commit_statuses" => reply(&work.commit_statuses(args(body)?).await?),
2722	"combined_status" => reply(&work.combined_status(args(body)?).await?),
2723	"create_check_run" => reply(&work.create_check_run(args(body)?).await?),
2724	"update_check_run" => reply(&work.update_check_run(args(body)?).await?),
2725	"get_check_run" => reply(&work.get_check_run(args(body)?).await?),
2726	"check_run_annotations" => reply(&work.check_run_annotations(args(body)?).await?),
2727	"ref_check_runs" => reply(&work.ref_check_runs(args(body)?).await?),
2728	"ref_check_suites" => reply(&work.ref_check_suites(args(body)?).await?),
2729	"get_check_suite" => reply(&work.get_check_suite(args(body)?).await?),
2730	"rerequest_check_run" => reply(&work.rerequest_check_run(args(body)?).await?),
2731	"rerequest_check_suite" => reply(&work.rerequest_check_suite(args(body)?).await?),
2732	"request_check_action" => reply(&work.request_check_action(args(body)?).await?),
2733	"commit_checks" => reply(&work.commit_checks(args(body)?).await?),
2734	"advance" => reply(&work.advance(args(body)?).await?),
2735	"managed_pulls" => reply(&work.managed_pulls(args(body)?).await?),
2736	"queue" => reply(&work.queue(args(body)?).await?),
2737	"queue_build" => reply(&work.queue_build(args(body)?).await?),
2738	"report_queue" => reply(&work.report_queue(args(body)?).await?),
2739	"remove_from_queue" => reply(&work.remove_from_queue(args(body)?).await?),
2740	"locate_pull" => reply(&work.locate_pull(args(body)?).await?),
2741	"inbox_subject" => reply(&work.inbox_subject(args(body)?).await?),
2742	"get_settings" => reply(&work.get_settings(args(body)?).await?),
2743	"codeowners_errors" => reply(&work.codeowners_errors(args(body)?).await?),
2744	"update_settings" => reply(&work.update_settings(args(body)?).await?),
2745	"open_pull" => reply(&work.open_pull(args(body)?).await?),
2746	"list_pulls" => reply(&work.list_pulls(args(body)?).await?),
2747	"pulls_for_repos" => reply(&work.pulls_for_repos(args(body)?).await?),
2748	"get_pull" => reply(&work.get_pull(args(body)?).await?),
2749	"update_pull" => reply(&work.update_pull(args(body)?).await?),
2750	"catch_up_pull" => reply(&work.catch_up_pull(args(body)?).await?),
2751	"ready_pull" => reply(&work.ready_pull(args(body)?).await?),
2752	"close_pull" => reply(&work.close_pull(args(body)?).await?),
2753	"reopen_pull" => reply(&work.reopen_pull(args(body)?).await?),
2754	"convert_pull_to_draft" => reply(&work.convert_pull_to_draft(args(body)?).await?),
2755	"merge_pull" => reply(&work.merge_pull(args(body)?).await?),
2756	"list_active_pulls" => reply(&work.list_active_pulls(args(body)?).await?),
2757	"by_author" => reply(&work.by_author(args(body)?).await?),
2758	"contributions" => reply(&work.contributions(args(body)?).await?),
2759	"append_session" => reply(&work.append_session(args(body)?).await?),
2760	"read_session" => reply(&work.read_session(args(body)?).await?),
2761	// Decision records (decisions.rs).
2762	"record_decision" => reply(&work.record_decision(args(body)?).await?),
2763	"list_decisions" => reply(&work.list_decisions(args(body)?).await?),
2764	// Coordinating work: harness sessions, claims, dependencies and an
2765	// issue's whole context (coordination/).
2766	"start_session" => reply(&work.start_session(args(body)?).await?),
2767	"update_session" => reply(&work.update_session(args(body)?).await?),
2768	"end_session" => reply(&work.end_session(args(body)?).await?),
2769	"get_work_session" => reply(&work.get_work_session(args(body)?).await?),
2770	"list_work_sessions" => reply(&work.list_work_sessions(args(body)?).await?),
2771	"record_session_entries" => reply(&work.record_session_entries(args(body)?).await?),
2772	"claim_issue" => reply(&work.claim_issue(args(body)?).await?),
2773	"release_claim" => reply(&work.release_claim(args(body)?).await?),
2774	"hand_off_issue" => reply(&work.hand_off_issue(args(body)?).await?),
2775	"list_claims" => reply(&work.list_claims(args(body)?).await?),
2776	"get_work" => reply(&work.get_work(args(body)?).await?),
2777	// Runs of sandboxes (runs.rs).
2778	"open_run" => reply(&work.open_run(args(body)?).await?),
2779	"report_run" => reply(&work.report_run(args(body)?).await?),
2780	"list_runs" => reply(&work.list_runs(args(body)?).await?),
2781	"get_run" => reply(&work.get_run(args(body)?).await?),
2782	"list_sessions" => reply(&work.list_sessions(args(body)?).await?),
2783	"get_session" => reply(&work.get_session(args(body)?).await?),
2784	// What a sandbox may reach (guardrails.rs).
2785	"get_guardrails" => reply(&work.get_guardrails(args(body)?).await?),
2786	"update_guardrails" => reply(&work.update_guardrails(args(body)?).await?),
2787	"run_guardrails" => reply(&work.run_guardrails(args(body)?).await?),
2788	// The runs whose sandboxes stop with their repository (retired.rs).
2789	"runs_in_repo" => reply(&work.runs_in_repo(args(body)?).await?),
2790	"start_mergecheck" => reply(&work.start_mergecheck(args(body)?).await?),
2791	"report_mergecheck" => reply(&work.report_mergecheck(args(body)?).await?),
2792	// Rulesets (rulesets.rs): kept here, enforced here on merge and by
2793	// repos on push.
2794	"list_rulesets" => reply(&work.list_rulesets(args(body)?).await?),
2795	"get_ruleset" => reply(&work.get_ruleset(args(body)?).await?),
2796	"save_ruleset" => reply(&work.save_ruleset(args(body)?).await?),
2797	"delete_ruleset" => reply(&work.delete_ruleset(args(body)?).await?),
2798	"effective_rules" => reply(&work.effective_rules(args(body)?).await?),
2799	"rule_evaluations" => reply(&work.rule_evaluations(args(body)?).await?),
2800	"ref_rules" => reply(&work.ref_rules(args(body)?).await?),
2801	"record_evaluations" => reply(&work.record_evaluations(args(body)?).await?),
2802	"set_requires_pull_request" => reply(&work.set_requires_pull_request(args(body)?).await?),
2803	_ => Response::error("Unknown method", 404),
2804	};
2805	served.finish_timed(answered, &work.timing)
2806	}
2807
2808	/// Events from the bus, delivered on this service's own queue.
2809	#[event(queue)]
2810	async fn queue(batch: MessageBatch<Event>, env: Env, _ctx: Context) -> Result<()> {
2811	let work = service(&env)?;
2812	for message in batch.messages()? {
2813	// A workspace renamed: its runs move to the slug it has now.
2814	if g1t_kit::rename::on_event(&env, &env.d1("DB")?, message.body(), &[runs::RENAMED, guardrails::RENAMED, rulesets::RENAMED].concat()).await? {
2815	message.ack();
2816	continue;
2817	}
2818	// A repository renamed or transferred: its runs and guardrails
2819	// follow.
2820	if g1t_kit::transfer::on_event(&env, &env.d1("DB")?, message.body(), &[runs::TRANSFERRED, guardrails::TRANSFERRED].concat()).await? {
2821	message.ack();
2822	continue;
2823	}
2824	// A workspace deleted: what it kept for itself goes.
2825	if g1t_kit::deleted::on_event(&env.d1("DB")?, message.body(), guardrails::DELETED).await? {
2826	message.ack();
2827	continue;
2828	}
2829	// An account purged: what it wrote shows as ghost (ghost.rs).
2830	let ghost = ghost::statements();
2831	let ghost: Vec<&str> = ghost.iter().map(String::as_str).collect();
2832	if g1t_kit::user_deleted::on_event(&env.d1("DB")?, message.body(), &ghost).await? {
2833	message.ack();
2834	continue;
2835	}
2836	// A repository deleted, archived or purged, or a branch renamed (retired.rs).
2837	if work.on_retired(message.body()).await? {
2838	message.ack();
2839	continue;
2840	}
2841	work.on_event(message.body()).await?;
2842	message.ack();
2843	}
2844	Ok(())
2845	}
2846
2847	/// The rules that once read a pull request's author read its owner now:
2848	/// whoever asked g1t for it, or its author. For each, the person who asked
2849	/// is held to what an author was, and g1t's agent (a token it works with)
2850	/// gains nothing by being the author.
2851	#[cfg(test)]
2852	mod owner_rules {
2853	use super::*;
2854	use crate::rows::stored::{ASKER, G1T, pull};
2855	use g1t_contracts::identity::AGENT_ID;
2856
2857	const SOMEONE: &str = "usr_2";
2858
2859	#[test]
2860	fn no_self_approval() {
2861	// add_comment refuses a verdict on one that is theirs.
2862	let made = pull(G1T, Some(ASKER));
2863	assert!(made.is_owned_by(ASKER.0), "the person who asked cannot approve it");
2864	assert!(!made.is_owned_by(AGENT_ID), "g1t's review agent still gives its verdict");
2865	assert!(!made.is_owned_by(SOMEONE));
2866	}
2867
2868	#[test]
2869	fn what_an_author_could_do_without_a_role() {
2870	// manageable_pull (update, ready, close), catch_up_pull on a fork,
2871	// and append_session: theirs to do.
2872	let made = pull(G1T, Some(ASKER));
2873	assert!(made.is_owned_by(ASKER.0));
2874	assert!(!made.is_owned_by(SOMEONE), "anyone else still needs the role");
2875	assert!(!made.is_owned_by(AGENT_ID), "being its author gives g1t's tokens nothing more");
2876	}
2877
2878	#[test]
2879	fn nobody_is_asked_to_review_what_they_asked_for() {
2880	// update_pull drops the owner from the reviewers asked.
2881	let made = pull(G1T, Some(ASKER));
2882	let mut reviewers = vec!["syntaqx".to_owned(), "ana".to_owned()];
2883	reviewers.retain(\|name\| *name != made.owner().username);
2884	assert_eq!(reviewers, ["ana"]);
2885	}
2886
2887	#[test]
2888	fn sandboxes_act_as_whoever_asked() {
2889	// MergecheckJob and the merge queue's job
2890	// carry who the sandbox's credential acts for: a real account.
2891	let made = pull(G1T, Some(ASKER));
2892	assert_eq!(made.owner().id, ASKER.0);
2893	let acts_as = made.requested_by.unwrap_or(made.author);
2894	assert_eq!(acts_as.id, ASKER.0);
2895	// g1t's own work, which nobody asked for, acts as g1t, as before.
2896	let own = pull(("g1t", "g1t"), None);
2897	assert_eq!(own.requested_by.unwrap_or(own.author).id, "g1t");
2898	}
2899
2900	#[test]
2901	fn events_name_g1t_and_whoever_asked() {
2902	let made = pull(G1T, Some(ASKER));
2903	let event = serde_json::to_value(Work::pull_event(&made)).unwrap();
2904	assert_eq!(event["author"], serde_json::json!({ "id": AGENT_ID, "username": "g1t" }));
2905	assert_eq!(event["requestedBy"], serde_json::json!({ "id": "usr_1", "username": "syntaqx" }));
2906	let own = serde_json::to_value(Work::pull_event(&pull(ASKER, None))).unwrap();
2907	assert!(own.get("requestedBy").is_none());
2908	}
2909
2910	#[test]
2911	fn only_a_closed_pull_request_reopens_and_only_an_open_one_turns_draft() {
2912	assert_eq!(reopen_refusal(PullStatus::Closed), None);
2913	assert!(reopen_refusal(PullStatus::Merged).is_some(), "a merge cannot be undone");
2914	assert!(reopen_refusal(PullStatus::Open).is_some());
2915	assert!(reopen_refusal(PullStatus::Draft).is_some());
2916	assert_eq!(draft_refusal(PullStatus::Open), None);
2917	assert!(draft_refusal(PullStatus::Draft).is_some());
2918	assert!(draft_refusal(PullStatus::Closed).is_some());
2919	assert!(draft_refusal(PullStatus::Merged).is_some());
2920	}
2921
2922	#[test]
2923	fn a_pull_request_reopens_as_what_it_was_closed_as() {
2924	assert_eq!(reopened_status(Some("draft")), PullStatus::Draft);
2925	assert_eq!(reopened_status(Some("open")), PullStatus::Open);
2926	// Closed before this was recorded, or by a merge of another.
2927	assert_eq!(reopened_status(None), PullStatus::Open);
2928	}
2929	}