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