flagon-io/g1t

public

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