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