flagon-io/g1t

public

Where people and agents ship software together. The open-source git platform for the whole job: issues, agents, checks and deploys to the edge.

g1t.sh

git agents cloudflare rust forge mcp

2,991 linesCode BlameRaw

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