Skip to content
Sign up

flagon-io/g1t

public
00

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

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