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,897 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 read = |repo_id: String| {
1141 let label = label.clone();
1142 async move {
1143 let query = self
1144 .db
1145 .prepare(format!(
1146 "SELECT {PULL_COLUMNS} FROM pulls WHERE repo_id = ? {filter}
1147 AND (? IS NULL OR EXISTS
1148 (SELECT 1 FROM json_each(pulls.labels) WHERE json_each.value = ?))
1149 AND (? IS NULL OR milestone = ?)
1150 ORDER BY number DESC LIMIT ?"
1151 ))
1152 .bind(&[
1153 repo_id.into(),
1154 optional(&label),
1155 optional(&label),
1156 optional_number(milestone),
1157 optional_number(milestone),
1158 LIST_PAGE.into(),
1159 ])?;
1160 self.timing.db(1, query.all()).await?.results::<PullRow>()
1161 }
1162 };
1163 let (repo, rows) = check!(self.repo_then(&a.repo, &a.viewer, read).await?);
1164 let base = a.base.map(|base| base.trim().to_owned()).filter(|base| !base.is_empty());
1165 Ok(Outcome::Ok(
1166 rows.into_iter()
1167 .map(|row| {
1168 let mut pull = Pull::from(row);
1169 fill_base(&mut pull, &repo);
1170 pull
1171 })
1172 .filter(|pull| base.as_deref().is_none_or(|base| pull.base.as_deref() == Some(base)))
1173 .collect(),
1174 ))
1175 }
1176
1177 /// `pulls_for_repos`: what `list_pulls` gives, open and closed, for many
1178 /// repositories at once: one access check with repos for all of them
1179 /// and one query, instead of two of each per repository.
1180 async fn pulls_for_repos(&self, a: PullsForReposArgs) -> Result<Vec<RepoPulls>> {
1181 let ids: Vec<String> = a.repo_ids.into_iter().take(MAX_PULLS_FOR_REPOS).collect();
1182 if ids.is_empty() {
1183 return Ok(Vec::new());
1184 }
1185 let limit = a.limit.clamp(1, LIST_PAGE);
1186 // The rows are read beside the access check, for every id asked
1187 // about; those of repositories the viewer cannot read are dropped.
1188 let asked = serde_json::to_string(&ids)?;
1189 let check = ReadableArgs { ids, viewer: a.viewer };
1190 let readable = self.timing.rpc(g1t_kit::call::<_, Vec<Repo>>(&self.repos, "readable", &check));
1191 let (readable, rows) = try_join(readable, self.timing.db(1, self.newest_pulls(asked, limit))).await?;
1192 if readable.is_empty() {
1193 return Ok(Vec::new());
1194 }
1195 let mut answer: Vec<RepoPulls> = readable
1196 .iter()
1197 .map(|repo| RepoPulls { repo_id: repo.id.clone(), open: Vec::new(), closed: Vec::new() })
1198 .collect();
1199 for pull in rows.into_iter().map(Pull::from) {
1200 let Some(entry) = answer.iter_mut().find(|entry| entry.repo_id == pull.repo_id) else {
1201 continue;
1202 };
1203 match pull.status {
1204 PullStatus::Draft | PullStatus::Open => entry.open.push(pull),
1205 PullStatus::Merged | PullStatus::Closed => entry.closed.push(pull),
1206 }
1207 }
1208 for entry in &mut answer {
1209 entry.open.sort_by_key(|pull| std::cmp::Reverse(pull.number));
1210 entry.closed.sort_by_key(|pull| std::cmp::Reverse(pull.number));
1211 }
1212 Ok(answer)
1213 }
1214
1215 /// The newest `limit` of each repository's open (draft or open) and
1216 /// closed (merged or closed) pull requests, for the ids in `ids` (JSON).
1217 async fn newest_pulls(&self, ids: String, limit: u32) -> Result<Vec<PullRow>> {
1218 self.db
1219 .prepare(format!(
1220 "SELECT * FROM (
1221 SELECT {PULL_COLUMNS}, ROW_NUMBER() OVER (
1222 PARTITION BY pulls.repo_id, pulls.status IN ('draft', 'open') ORDER BY pulls.number DESC
1223 ) AS place
1224 FROM pulls WHERE pulls.repo_id IN (SELECT value FROM json_each(?1))
1225 ) WHERE place <= ?2"
1226 ))
1227 .bind(&[ids.into(), limit.into()])?
1228 .all()
1229 .await?
1230 .results::<PullRow>()
1231 }
1232
1233 async fn get_pull(&self, a: ViewArgs) -> Result<Outcome<PullDetail>> {
1234 let number = a.number;
1235 // Every row the page and the lifecycle read, in one batch started
1236 // beside the access check; the helpers below read from it.
1237 let namespace = a.repo.namespace.clone();
1238 let read = |repo_id: String| self.prefetch_pull(repo_id, namespace.clone(), number);
1239 let Outcome::Ok((repo, Some(found))) = self.repo_then(&a.repo, &a.viewer, read).await? else {
1240 return Ok(no_pull());
1241 };
1242 let Some(row) = found.first::<PullRow>(prefetch::Slot::Pull)? else {
1243 return Ok(no_pull());
1244 };
1245 let stored = found.first::<StoredBehind>(prefetch::Slot::Pull)?;
1246 // Its work item, as far as the viewer may see it.
1247 let item = found
1248 .held(prefetch::Slot::Item)?
1249 .filter(|held| held.may(a.viewer.as_ref(), Capability::Read))
1250 .map(|held| held.shown(a.viewer.as_ref()));
1251 let mut comments: Vec<Comment> =
1252 found.rows::<CommentRow>(prefetch::Slot::Comments)?.into_iter().map(Comment::from).collect();
1253 let reactions = reactions::attach(&found.rows::<reactions::ReactionRow>(prefetch::Slot::Reactions)?, &row.id, &mut comments);
1254 self.keep_prefetched(Some(found));
1255 let default_branch = repo.default_branch.clone();
1256 let detail = self.pull_detail(repo, Pull::from(row), item, comments, stored, &a.viewer).await;
1257 self.keep_prefetched(None);
1258 // The branch it merges into, named, for whoever reads it.
1259 Ok(match detail? {
1260 Outcome::Ok(mut detail) => {
1261 if detail.pull.base.as_deref().is_none_or(str::is_empty) {
1262 detail.pull.base = Some(default_branch);
1263 }
1264 detail.reactions = reactions;
1265 Outcome::Ok(detail)
1266 }
1267 failed => failed,
1268 })
1269 }
1270
1271 async fn pull_detail(
1272 &self,
1273 repo: Repo,
1274 mut pull: Pull,
1275 item: Option<WorkItem>,
1276 comments: Vec<Comment>,
1277 stored: Option<StoredBehind>,
1278 viewer: &Viewer,
1279 ) -> Result<Outcome<PullDetail>> {
1280 // Whether it is behind, as worked out with its mergeability on the
1281 // last push to either side (mergeability.rs), when that was for
1282 // its head as it is now; otherwise asked of the repos service.
1283 let known_behind = stored.and_then(|stored| stored.for_head(pull.head_commit.as_deref()));
1284 // Worked out on each push; this covers a pull request from before
1285 // that was recorded.
1286 if pull.files.is_empty() && pull.head_commit.is_some() {
1287 pull.files = self.refresh_files(&pull).await?;
1288 }
1289 // Everything else at once: none of it depends on the rest, and each
1290 // is a round trip of its own.
1291 let standing = async {
1292 // Mergeability first: where g1t sees a pull request through, a
1293 // conflict decides its next step.
1294 let behind = async {
1295 match known_behind {
1296 Some(behind) => Ok(behind),
1297 None => {
1298 let behind = self.is_behind(&pull).await?;
1299 // Kept for the next view when the mergeability on
1300 // record is for this head: a pull request from
1301 // before `behind` was kept asks once.
1302 if let Some(head) = pull.head_commit.as_deref()
1303 && pull.status.is_active()
1304 {
1305 self.db
1306 .prepare(
1307 "UPDATE pulls SET behind = ?1
1308 WHERE id = ?2 AND behind IS NULL AND mergeable_key LIKE ?3 || '..%'",
1309 )
1310 .bind(&[u32::from(behind).into(), pull.id.as_str().into(), head.into()])?
1311 .run()
1312 .await?;
1313 }
1314 Ok(behind)
1315 }
1316 }
1317 };
1318 let (merge, behind) = try_join(self.mergeability(&pull), behind).await?;
1319 let lifecycle = self.assess(&pull, &item, behind).await?.map(|(lifecycle, _)| lifecycle);
1320 Ok::<_, worker::Error>((behind, lifecycle, merge))
1321 };
1322 let (((behind, lifecycle, (mergeable, conflicts)), (landing, stalled), comments), (checks, overlaps)) =
1323 try_join(
1324 try_join3(standing, self.landing_state(&pull.id), async { Ok(comments) }),
1325 try_join(self.latest_checks(&pull.id), self.overlaps(&pull)),
1326 )
1327 .await?;
1328 // The rules of the branch it merges into, as they stack, and which
1329 // of them it does not meet yet, for whoever is looking.
1330 let (statuses, settings, gate) = try_join3(
1331 self.statuses(&repo.id, pull.head_commit.as_deref()),
1332 self.settings_on(&repo, &pull),
1333 async {
1334 if pull.status.is_active() {
1335 self.merge_gate_seen(&repo, &pull, viewer.as_ref(), behind).await.map(Some)
1336 } else {
1337 Ok(None)
1338 }
1339 },
1340 )
1341 .await?;
1342 let code_owners = self.pull_code_owners(&pull, &comments, &settings).await?;
1343 // The work items it closes, as its title and description say now;
1344 // none where the repository does not close them that way.
1345 if pull.status.is_active() {
1346 let named = self.items_to_close(&pull, &repo, &settings, false, &[]).await?;
1347 pull.closes = named.close_numbers();
1348 pull.parts = named.part_numbers();
1349 }
1350 let required = if settings.required_checks.is_empty() {
1351 Vec::new()
1352 } else {
1353 let checked = self.with_job_checks(&repo.id, pull.head_commit.as_deref(), statuses.clone()).await?;
1354 required_checks(&settings.required_checks, &checked)
1355 };
1356 Ok(Outcome::Ok(PullDetail {
1357 required_checks: required,
1358 rules: gate.map(|gate| rulesets::merge_rules(&gate.judged, &gate.requirements, pull.base_branch(&repo.default_branch) == repo.default_branch)),
1359 code_owners,
1360 comments,
1361 checks,
1362 overlaps,
1363 behind,
1364 lifecycle,
1365 landing,
1366 stalled,
1367 statuses,
1368 mergeable,
1369 conflicts,
1370 earlier_checks: self.earlier_checks(&pull.id).await?,
1371 reactions: Vec::new(),
1372 item,
1373 pull,
1374 }))
1375 }
1376
1377 /// The pull request, whatever its status, if its repository is not
1378 /// archived and `actor` opened it or may triage its pull requests.
1379 async fn managed_pull(
1380 &self,
1381 actor: &User,
1382 path: &RepoPath,
1383 number: u32,
1384 ) -> Result<Outcome<Pull>> {
1385 let (repo, pull) = check!(self.pull_at(path, number, &Some(actor.clone())).await?);
1386 check!(writable(&repo));
1387 if !pull.is_owned_by(&actor.id) {
1388 check!(allowed(Some(actor), &repo, Capability::Triage));
1389 }
1390 Ok(Outcome::Ok(pull))
1391 }
1392
1393 /// The pull request, if it is still active and `actor` opened it or
1394 /// may triage the repository's pull requests.
1395 async fn manageable_pull(
1396 &self,
1397 actor: &User,
1398 path: &RepoPath,
1399 number: u32,
1400 ) -> Result<Outcome<Pull>> {
1401 let pull = check!(self.managed_pull(actor, path, number).await?);
1402 if !pull.status.is_active() {
1403 return Ok(Outcome::fail(
1404 FailureCode::Conflict,
1405 format!("This pull request is already {}.", pull.status.as_str()),
1406 ));
1407 }
1408 Ok(Outcome::Ok(pull))
1409 }
1410
1411 /// Brings a pull request up to date with the default branch without a
1412 /// sandbox, where the repos service can do that safely. Whoever could
1413 /// have pushed the merge themselves may ask: whoever opened it (or asked
1414 /// g1t for it), for a fork; anyone who may push, for a branch of the
1415 /// repository. When it needs a
1416 /// real merge, says so, naming the conflicting files if a probe found
1417 /// them, and pushes nothing.
1418 async fn catch_up_pull(&self, a: PullActionArgs) -> Result<Outcome<PullBranchUpdate>> {
1419 let (repo, pull) = check!(self.pull_at(&a.repo, a.number, &Some(a.actor.clone())).await?);
1420 check!(writable(&repo));
1421 if !pull.status.is_active() {
1422 return Ok(Outcome::fail(
1423 FailureCode::Conflict,
1424 format!("This pull request is already {}.", pull.status.as_str()),
1425 ));
1426 }
1427 if pull.fork_repo_id.is_some() {
1428 if !pull.is_owned_by(&a.actor.id) {
1429 return Ok(Outcome::fail(
1430 FailureCode::Forbidden,
1431 "Only whoever opened this pull request, or asked g1t for it, can update it.",
1432 ));
1433 }
1434 } else if !pull.from_head_repo() {
1435 check!(allowed(Some(&a.actor), &repo, Capability::Push));
1436 }
1437 // A fork's branch is updated by those who may push to the fork,
1438 // which repos decides.
1439 let updated: Outcome<PullBranchUpdate> = g1t_kit::call(
1440 &self.repos,
1441 "update_pull_branch",
1442 &UpdatePullBranchArgs {
1443 source_id: pull.source_id().to_owned(),
1444 branch: pull.branch.clone(),
1445 number: pull.number,
1446 actor: a.actor,
1447 target_branch: pull.base.clone(),
1448 target_id: pull.target_id(),
1449 method: a.update_method.unwrap_or_default(),
1450 expected_head: a.expected_head.clone().filter(|head| !head.trim().is_empty()),
1451 },
1452 )
1453 .await?;
1454 // A probe that found conflicts says more than "both changed it".
1455 if let Outcome::Ok(PullBranchUpdate::NeedsAgent { .. }) = &updated
1456 && let Some(files) = self.conflicting_files(&pull).await?
1457 && !files.is_empty()
1458 {
1459 return Ok(Outcome::Ok(PullBranchUpdate::NeedsAgent {
1460 reason: NeedsAgentReason::Conflicting,
1461 detail: "Merging it conflicts.".to_owned(),
1462 paths: files,
1463 }));
1464 }
1465 Ok(updated)
1466 }
1467
1468 async fn update_pull(&self, a: UpdatePullArgs) -> Result<Outcome<Pull>> {
1469 let pull = check!(self.manageable_pull(&a.actor, &a.repo, a.number).await?);
1470 // The branch it merges into, its milestone and its labels first:
1471 // each can be refused, and then nothing else changes.
1472 if a.base.is_some() || a.milestone.is_some() || a.labels.is_some() {
1473 let repo = check!(self.repo(&a.repo, &Some(a.actor.clone())).await?);
1474 if let Some(base) = &a.base {
1475 check!(self.change_base(&a.actor, &repo, &pull, base).await?);
1476 }
1477 if let Some(number) = a.milestone {
1478 check!(items::set_milestone(self, &a.actor, &items::Labeled::Pull(repo.clone(), pull.clone()), number).await?);
1479 }
1480 if let Some(labels) = &a.labels {
1481 check!(items::relabel(self, &a.actor, &items::Labeled::Pull(repo.clone(), pull.clone()), labels).await?);
1482 }
1483 }
1484 let assignees = match a.assignees {
1485 Some(names) => Some(check!(self.valid_assignees(names).await?)),
1486 None => None,
1487 };
1488 // Teams, named `workspace/team`, apart from the people.
1489 let (team_names, a_reviewers) = match a.reviewers {
1490 Some(names) => {
1491 let (teams, people): (Vec<String>, Vec<String>) =
1492 names.into_iter().partition(|name| team_reviews::team_name(name).is_some());
1493 (Some(teams), Some(people))
1494 }
1495 None => (None, None),
1496 };
1497 let teams = match team_names {
1498 Some(names) => Some(check!(self.valid_team_reviewers(&a.actor, &a.repo, &pull, names).await?)),
1499 None => None,
1500 };
1501 let reviewers = match a_reviewers {
1502 Some(names) => {
1503 let mut reviewers = check!(self.valid_assignees(names).await?);
1504 // Nobody is asked to review their own, nor what they had g1t make.
1505 reviewers.retain(|name| *name != pull.owner().username);
1506 Some(reviewers)
1507 }
1508 None => None,
1509 };
1510 self.db
1511 .prepare(
1512 "UPDATE pulls
1513 SET assignees = COALESCE(?, assignees), updated_at = ?
1514 WHERE id = ?",
1515 )
1516 .bind(&[
1517 optional(&assignees.as_ref().map(serde_json::to_string).transpose()?),
1518 rfc3339(now_ms()).into(),
1519 pull.id.as_str().into(),
1520 ])?
1521 .run()
1522 .await?;
1523 if let Some(assignees) = &assignees {
1524 self.note_changes(
1525 On::Pull(&pull.repo_id, pull.number),
1526 &a.actor,
1527 &pull.assignees,
1528 assignees,
1529 ("assigned", "unassigned"),
1530 )
1531 .await?;
1532 }
1533 // Who was newly assigned or asked to review, and whose request was
1534 // withdrawn: the inbox tells them, and webhooks say so.
1535 let newly = |after: &[String], before: &[String]| -> Vec<String> {
1536 after.iter().filter(|name| !before.contains(name)).cloned().collect()
1537 };
1538 if let Some(assignees) = &assignees {
1539 let added = newly(assignees, &pull.assignees);
1540 if !added.is_empty() {
1541 self.publish(
1542 "pull.assigned",
1543 &pull.repo_id,
1544 &a.actor,
1545 PullEvent {
1546 assignees: Some(assignees.clone()),
1547 added: Some(added),
1548 ..Self::pull_event(&pull)
1549 },
1550 )
1551 .await?;
1552 }
1553 }
1554 if reviewers.is_some() || teams.is_some() {
1555 let people = reviewers.unwrap_or_else(|| pull.reviewers.clone());
1556 let teams = teams.unwrap_or_else(|| pull.team_reviewers.clone());
1557 self.set_reviewers(&pull, people, teams, Some(&a.actor), false).await?;
1558 }
1559 Ok(match self.pull(&pull.repo_id, pull.number).await? {
1560 Some(pull) => Outcome::Ok(pull),
1561 None => no_pull(),
1562 })
1563 }
1564
1565 /// Points an open pull request at another branch to merge into. Needs
1566 /// the Write role. What it would merge, whether it is behind, its
1567 /// mergeability and its checks are all worked out against the new base.
1568 async fn change_base(&self, actor: &User, repo: &Repo, pull: &Pull, base: &str) -> Result<Outcome<()>> {
1569 check!(allowed(Some(actor), repo, Capability::Push));
1570 let base = base.trim();
1571 if base.is_empty() {
1572 return Ok(Outcome::fail(FailureCode::Invalid, "Name the branch it should merge into."));
1573 }
1574 let before = pull.base_branch(&repo.default_branch).to_owned();
1575 if base == before {
1576 return Ok(Outcome::Ok(()));
1577 }
1578 if !pull.from_other_repo() && pull.branch.as_deref() == Some(base) {
1579 return Ok(Outcome::fail(
1580 FailureCode::Invalid,
1581 format!("A pull request cannot merge {base} into itself. Choose another base."),
1582 ));
1583 }
1584 let exists: Option<String> = g1t_kit::call(
1585 &self.repos,
1586 "head",
1587 &HeadArgs { repo_id: repo.id.clone(), branch: base.to_owned() },
1588 )
1589 .await?;
1590 if exists.is_none() {
1591 return Ok(Outcome::fail(FailureCode::NotFound, format!("There is no branch named {base} to merge into.")));
1592 }
1593 // In the merge queue it was headed for the default branch; it
1594 // leaves the queue for another base.
1595 let left_queue = self
1596 .leave(&pull.repo_id, pull, QueueState::Removed, Some("Its base branch changed."))
1597 .await?;
1598 // Its tested commit is being published to the default branch: a new
1599 // base would describe a merge that is not the one taking place.
1600 if !left_queue && self.publishing(&pull.id).await? {
1601 return Ok(Outcome::fail(FailureCode::Conflict, "This pull request is being published. Wait for landing to finish."));
1602 }
1603 let stored = stored_base(base, repo);
1604 self.db
1605 .prepare(
1606 "UPDATE pulls
1607 SET base_branch = ?, updated_at = ?, land_requested = NULL, land_requested_at = NULL, behind = NULL,
1608 mergeable = NULL, mergeable_key = NULL, conflicts = NULL
1609 WHERE id = ?",
1610 )
1611 .bind(&[optional(&stored), rfc3339(now_ms()).into(), pull.id.as_str().into()])?
1612 .run()
1613 .await?;
1614 self.note(
1615 &pull.repo_id,
1616 pull.number,
1617 actor,
1618 &format!("changed the base branch from `{before}` to `{base}`"),
1619 )
1620 .await?;
1621 self.publish(
1622 "pull.base_changed",
1623 &pull.repo_id,
1624 actor,
1625 PullEvent { base: Some(base.to_owned()), ..Self::pull_event(pull) },
1626 )
1627 .await?;
1628 if left_queue {
1629 self.publish_as(
1630 "queue.changed",
1631 &pull.repo_id,
1632 None,
1633 g1t_contracts::events::QueueChanged { repo_id: pull.repo_id.clone() },
1634 )
1635 .await?;
1636 }
1637 // Its merge ref, and whether it merges cleanly, against the new base.
1638 if let Some(moved) = self.pull_by_id(&pull.id).await? {
1639 self.keep_pull_refs(&moved).await;
1640 if let Err(error) = self.assess_mergeability(&moved).await {
1641 worker::console_warn!("mergeability of {}: {error}", pull.id);
1642 }
1643 }
1644 Ok(Outcome::Ok(()))
1645 }
1646
1647 /// Marks a draft ready for review, or updates the description of one
1648 /// that already is.
1649 async fn ready_pull(&self, a: PullActionArgs) -> Result<Outcome<Pull>> {
1650 let mut pull = check!(self.manageable_pull(&a.actor, &a.repo, a.number).await?);
1651 let summary = Some(a.summary.trim().to_owned()).filter(|summary| !summary.is_empty());
1652 let now = rfc3339(now_ms());
1653 // The work items it closes and is part of, as its description says
1654 // now (closing.rs).
1655 let repo = check!(self.repo(&a.repo, &Some(a.actor.clone())).await?);
1656 let own = self.pull_item_id(&pull.id).await?;
1657 let text = format!("{}\n{}", pull.title, summary.as_deref().or(pull.body.as_deref()).unwrap_or_default());
1658 let named = items::named(self, &repo, own.as_deref(), &text, &[&a.actor]).await?;
1659 self.db
1660 .prepare(
1661 "UPDATE pulls SET status = 'open', body = COALESCE(?, body), updated_at = ?, closes_items = ?, parts = ?
1662 WHERE id = ?",
1663 )
1664 .bind(&[
1665 optional(&summary),
1666 now.as_str().into(),
1667 serde_json::to_string(&named.close_ids())?.into(),
1668 serde_json::to_string(&named.part_ids())?.into(),
1669 pull.id.as_str().into(),
1670 ])?
1671 .run()
1672 .await?;
1673 if pull.status == PullStatus::Draft {
1674 // The head as it is now: the push that came just before may not
1675 // have reached `head_commit` yet, and workflows run on it.
1676 let commit = self.live_head(&pull).await?.or_else(|| pull.head_commit.clone());
1677 self.publish(
1678 "pull.ready",
1679 &pull.repo_id,
1680 &a.actor,
1681 PullEvent {
1682 commit,
1683 ..Self::pull_event(&pull)
1684 },
1685 )
1686 .await?;
1687 }
1688 if pull.status == PullStatus::Draft {
1689 self.note(
1690 &pull.repo_id,
1691 pull.number,
1692 &a.actor,
1693 "marked this ready for review",
1694 )
1695 .await?;
1696 }
1697 pull.status = PullStatus::Open;
1698 pull.body = summary.or(pull.body);
1699 pull.closes = named.close_numbers();
1700 pull.parts = named.part_numbers();
1701 pull.updated_at = now;
1702 // A draft's code owners are asked once it is ready.
1703 self.refresh_code_owners(&pull).await;
1704 if let Some(fresh) = self.pull(&pull.repo_id, pull.number).await? {
1705 pull.reviewers = fresh.reviewers;
1706 pull.team_reviewers = fresh.team_reviewers;
1707 }
1708 Ok(Outcome::Ok(pull))
1709 }
1710
1711 async fn close_pull(&self, a: PullActionArgs) -> Result<Outcome<Pull>> {
1712 let mut pull = check!(self.manageable_pull(&a.actor, &a.repo, a.number).await?);
1713 let now = rfc3339(now_ms());
1714 let closed = self.db
1715 // What it was closed as, so that reopening brings that back.
1716 .prepare("UPDATE pulls SET status = 'closed', closed_from = ?, updated_at = ?, auto_merge = NULL WHERE id = ? AND status <> 'merged'
1717 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)
1718 RETURNING id AS value")
1719 .bind(&[pull.status.as_str().into(), now.as_str().into(), pull.id.as_str().into()])?
1720 .first::<rows::ValueRow>(None)
1721 .await?;
1722 if closed.is_none() {
1723 return Ok(Outcome::fail(FailureCode::Conflict, "This pull request is being published or has already merged."));
1724 }
1725 self.publish(
1726 "pull.closed",
1727 &pull.repo_id,
1728 &a.actor,
1729 Self::pull_event(&pull),
1730 )
1731 .await?;
1732 self.note(
1733 &pull.repo_id,
1734 pull.number,
1735 &a.actor,
1736 "closed this",
1737 )
1738 .await?;
1739 // A closed pull request leaves the merge queue.
1740 if self
1741 .leave(&pull.repo_id, &pull, QueueState::Removed, Some("It was closed."))
1742 .await?
1743 {
1744 self.publish_as(
1745 "queue.changed",
1746 &pull.repo_id,
1747 None,
1748 g1t_contracts::events::QueueChanged {
1749 repo_id: pull.repo_id.clone(),
1750 },
1751 )
1752 .await?;
1753 }
1754 pull.status = PullStatus::Closed;
1755 pull.auto_merge = None;
1756 pull.updated_at = now;
1757 Ok(Outcome::Ok(pull))
1758 }
1759
1760 /// Opens a closed pull request again: as the draft it was, if it was
1761 /// closed as one, else ready for review. A merged one stays merged.
1762 async fn reopen_pull(&self, a: PullActionArgs) -> Result<Outcome<Pull>> {
1763 let mut pull = check!(self.managed_pull(&a.actor, &a.repo, a.number).await?);
1764 if let Some(refusal) = reopen_refusal(pull.status) {
1765 return Ok(Outcome::fail(FailureCode::Conflict, refusal));
1766 }
1767 // The head as it is now, which workflows run on. A branch of the
1768 // repository that was deleted since leaves nothing to reopen; a
1769 // fork removed after the close is made again (repos, `pull.reopened`).
1770 let head = self.live_head(&pull).await?;
1771 if head.is_none() && pull.fork_repo_id.is_none() {
1772 return Ok(Outcome::fail(
1773 FailureCode::Conflict,
1774 format!(
1775 "The branch {} no longer exists. Push it again to reopen this pull request.",
1776 head_repo::source_label(pull.head_repo.as_ref(), pull.branch.as_deref().unwrap_or_default())
1777 ),
1778 ));
1779 }
1780 let closed_from: Option<ClosedFrom> = self
1781 .db
1782 .prepare("SELECT closed_from FROM pulls WHERE id = ?")
1783 .bind(&[pull.id.as_str().into()])?
1784 .first(None)
1785 .await?;
1786 let status = reopened_status(closed_from.and_then(|row| row.closed_from).as_deref());
1787 let now = rfc3339(now_ms());
1788 self.db
1789 .prepare("UPDATE pulls SET status = ?, closed_from = NULL, superseded_by = NULL, updated_at = ? WHERE id = ?")
1790 .bind(&[status.as_str().into(), now.as_str().into(), pull.id.as_str().into()])?
1791 .run()
1792 .await?;
1793 self.publish(
1794 "pull.reopened",
1795 &pull.repo_id,
1796 &a.actor,
1797 PullEvent {
1798 commit: head.or_else(|| pull.head_commit.clone()),
1799 ..Self::pull_event(&pull)
1800 },
1801 )
1802 .await?;
1803 self.note(
1804 &pull.repo_id,
1805 pull.number,
1806 &a.actor,
1807 "reopened this",
1808 )
1809 .await?;
1810 pull.status = status;
1811 pull.superseded_by = None;
1812 pull.updated_at = now;
1813 // Its refs in the repository, and whether it still merges cleanly,
1814 // now that it is open again.
1815 self.keep_pull_refs(&pull).await;
1816 if let Err(error) = self.assess_mergeability(&pull).await {
1817 worker::console_warn!("mergeability of {}: {error}", pull.id);
1818 }
1819 Ok(Outcome::Ok(pull))
1820 }
1821
1822 /// Turns a pull request that is ready for review back into a draft: it
1823 /// cannot be merged until it is marked ready again, and it leaves the
1824 /// merge queue and any merge that was waiting for it to catch up.
1825 async fn convert_pull_to_draft(&self, a: PullActionArgs) -> Result<Outcome<Pull>> {
1826 let mut pull = check!(self.managed_pull(&a.actor, &a.repo, a.number).await?);
1827 if let Some(refusal) = draft_refusal(pull.status) {
1828 return Ok(Outcome::fail(FailureCode::Conflict, refusal));
1829 }
1830 let now = rfc3339(now_ms());
1831 let drafted = self.db
1832 .prepare(
1833 "UPDATE pulls SET status = 'draft', land_requested = NULL, land_requested_at = NULL, updated_at = ?
1834 WHERE id = ? AND status = 'open'
1835 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)
1836 RETURNING id AS value",
1837 )
1838 .bind(&[now.as_str().into(), pull.id.as_str().into()])?
1839 .first::<rows::ValueRow>(None)
1840 .await?;
1841 if drafted.is_none() {
1842 return Ok(Outcome::fail(FailureCode::Conflict, "This pull request changed or is already being published."));
1843 }
1844 self.publish(
1845 "pull.converted_to_draft",
1846 &pull.repo_id,
1847 &a.actor,
1848 Self::pull_event(&pull),
1849 )
1850 .await?;
1851 self.note(
1852 &pull.repo_id,
1853 pull.number,
1854 &a.actor,
1855 "marked this as a draft",
1856 )
1857 .await?;
1858 if self
1859 .leave(&pull.repo_id, &pull, QueueState::Removed, Some("It was marked as a draft."))
1860 .await?
1861 {
1862 self.publish_as(
1863 "queue.changed",
1864 &pull.repo_id,
1865 None,
1866 g1t_contracts::events::QueueChanged {
1867 repo_id: pull.repo_id.clone(),
1868 },
1869 )
1870 .await?;
1871 }
1872 pull.status = PullStatus::Draft;
1873 pull.updated_at = now;
1874 Ok(Outcome::Ok(pull))
1875 }
1876
1877 /// Lands the pull request on the branch it merges into, by the method
1878 /// asked for or the repository's default (merging.rs). Unless told to
1879 /// keep them open, a merge into the default branch lands the work items
1880 /// it closes in this repository: its own and those it names
1881 /// (closing.rs), each closing once nothing else is left to land, and
1882 /// the others still in progress here for its own close as superseded.
1883 async fn merge_pull(&self, a: PullActionArgs) -> Result<Outcome<Pull>> {
1884 let viewer = Some(a.actor.clone());
1885 let (repo, pull) = check!(self.pull_at(&a.repo, a.number, &viewer).await?);
1886 check!(writable(&repo));
1887 match pull.status {
1888 PullStatus::Open => {}
1889 PullStatus::Draft => {
1890 return Ok(Outcome::fail(
1891 FailureCode::Conflict,
1892 "This pull request is still a draft. Mark it ready for review first.",
1893 ));
1894 }
1895 status => {
1896 return Ok(Outcome::fail(
1897 FailureCode::Conflict,
1898 format!("This pull request is already {}.", status.as_str()),
1899 ));
1900 }
1901 }
1902 // Whoever looked at one head merges that one.
1903 if let Some(expected) = a.expected_head.as_deref().map(str::trim).filter(|head| !head.is_empty())
1904 && pull.head_commit.as_deref() != Some(expected)
1905 {
1906 return Ok(Outcome::fail(
1907 FailureCode::Conflict,
1908 "This pull request has new commits since you looked. Review them, then merge again.",
1909 ));
1910 }
1911 let base = pull.base_branch(&repo.default_branch).to_owned();
1912 // How it merges: as asked, or the repository's default, where the
1913 // repository and the rules for its base allow it.
1914 let (choosing, requirements) = self.settings_and_requirements(&repo, &pull).await?;
1915 let method = match merging::choose_method(a.merge_method, &choosing, &requirements, &base) {
1916 Ok(method) => method,
1917 Err(message) => return Ok(Outcome::fail(FailureCode::Invalid, message)),
1918 };
1919 // The rules of the branch it merges into, as they stack: the one
1920 // gate for the merge button, the API, MCP, auto-merge and the queue,
1921 // for a person's pull request and an agent's alike. A bypass counts
1922 // when the merger asks for it, or for g1t when a ruleset lists it.
1923 let gate = self.merge_gate_by(&repo, &pull, Some(&a.actor), a.ignore_checks, true, method).await?;
1924 let bypassable = gate.bypassable();
1925 let gate = if a.bypass_rules || a.actor.is_system() { gate } else { gate.without_bypass() };
1926 let settings = rulesets::overlay(self.settings(&repo.id).await?, &gate.requirements, base == repo.default_branch);
1927 if pull.check_status == Some(CheckStatus::Failed) && !(a.ignore_checks && settings.allow_ignoring_checks) {
1928 return Ok(Outcome::fail(
1929 FailureCode::Conflict,
1930 "It failed in the merge queue; push a fix to try again.",
1931 ));
1932 }
1933 if let Some(refusal) = gate.refusal() {
1934 if access::can(Some(&a.actor), &repo, Capability::Merge) {
1935 self.record_merge_evaluations(&repo, &pull, &gate).await;
1936 }
1937 let offer = if bypassable && !a.bypass_rules {
1938 " You may bypass these rules: merge again and ask to bypass them (bypass_rules)."
1939 } else {
1940 ""
1941 };
1942 return Ok(Outcome::fail(FailureCode::Conflict, format!("{refusal}{offer}")));
1943 }
1944 // Known ahead of time to conflict: neither a merge nor the queue
1945 // would get through, so say what has to be resolved now.
1946 if let Some(files) = self.conflicting_files(&pull).await? {
1947 let named = if files.is_empty() {
1948 String::new()
1949 } else {
1950 format!(" in {}", files.join(", "))
1951 };
1952 return Ok(Outcome::fail(
1953 FailureCode::Conflict,
1954 format!(
1955 "This branch has conflicts with {base}{named} that must be resolved first. Merge {base} into it, fix them and push."
1956 ),
1957 ));
1958 }
1959
1960 // A repository that merges through a queue: it joins the queue, and
1961 // lands once its state together with everything ahead has passed.
1962 // Its state is built by the method asked for, else the merge queue
1963 // rule's, and lands exactly as tested (queue.rs).
1964 if settings.merge_queue {
1965 if !a.actor.verified {
1966 return Ok(Outcome::fail(FailureCode::Forbidden, UNVERIFIED));
1967 }
1968 check!(allowed(Some(&a.actor), &repo, Capability::Merge));
1969 let queued = match a.merge_method {
1970 Some(_) => method,
1971 None => {
1972 let rule = self.queue_rule(&repo.id).await?;
1973 match merging::choose_method(Some(rule.merge_method), &choosing, &requirements, &base) {
1974 Ok(method) => method,
1975 Err(message) => return Ok(Outcome::fail(FailureCode::Invalid, message)),
1976 }
1977 }
1978 };
1979 self.record_merge_evaluations(&repo, &pull, &gate).await;
1980 // Landing honours the same bypass request against the rules as
1981 // they are then (queue.rs `settle`).
1982 let message = (a.commit_title.as_deref(), a.commit_message.as_deref());
1983 return self.enqueue(&repo, &pull, &a.actor, a.keep_item_open, a.bypass_rules, queued, message).await;
1984 }
1985
1986 // The commit it makes, for a merge or a squash: what the merger
1987 // wrote, over the repository's default.
1988 let message = match self
1989 .merge_message_for(&repo, &pull, &settings, method, a.commit_title.as_deref(), a.commit_message.as_deref())
1990 .await?
1991 {
1992 Ok(message) => message,
1993 Err(problem) => return Ok(Outcome::fail(FailureCode::Invalid, problem)),
1994 };
1995
1996 // The branch it merges into has moved under it. A merge, squash or
1997 // rebase makes new commits on its head anyway; a fast-forward needs
1998 // it brought up to date first, and lands when that is done. Unless
1999 // the repository insists on that being dealt with first.
2000 let behind = self.is_behind(&pull).await?;
2001 match merging::when_behind(method, behind, settings.require_up_to_date, &base) {
2002 merging::WhenBehind::Land => {}
2003 merging::WhenBehind::Refuse(refusal) => return Ok(Outcome::fail(FailureCode::Conflict, refusal)),
2004 merging::WhenBehind::CatchUpFirst => {
2005 if !a.actor.verified {
2006 return Ok(Outcome::fail(FailureCode::Forbidden, UNVERIFIED));
2007 }
2008 check!(allowed(Some(&a.actor), &repo, Capability::Merge));
2009 self.record_merge_evaluations(&repo, &pull, &gate).await;
2010 // Brought up to date here, without a sandbox: the merge is
2011 // recorded first, and lands once the push that catches it up
2012 // arrives and the required checks pass on that commit
2013 // (lifecycle.rs `land_if_requested`, `land_when_checked`).
2014 // Merging past failing checks, where that is allowed, carries
2015 // over: it then lands without waiting for them.
2016 self.request_landing(
2017 &pull,
2018 &a.actor,
2019 a.keep_item_open,
2020 a.delete_branch,
2021 a.ignore_checks && settings.allow_ignoring_checks,
2022 )
2023 .await?;
2024 // Bounded: a repository that does not answer gets a clear
2025 // failure back to whoever clicked, not a request that hangs
2026 // until the site gives up on it.
2027 let updated: Option<Outcome<PullBranchUpdate>> = g1t_kit::call_within(
2028 &self.repos,
2029 "update_pull_branch",
2030 &UpdatePullBranchArgs {
2031 source_id: pull.source_id().to_owned(),
2032 branch: pull.branch.clone(),
2033 number: pull.number,
2034 actor: a.actor.clone(),
2035 target_branch: pull.base.clone(),
2036 target_id: pull.target_id(),
2037 // Linear history takes no merge commit on the way.
2038 method: if requirements.linear_history { UpdateMethod::Rebase } else { UpdateMethod::Merge },
2039 expected_head: pull.head_commit.clone(),
2040 },
2041 lifecycle::REPOS_WRITE_WITHIN,
2042 )
2043 .await?;
2044 // A push that still arrives later only brings it up to date.
2045 let updated = updated.unwrap_or_else(lifecycle::repository_unreachable);
2046 match updated {
2047 Outcome::Ok(PullBranchUpdate::Updated { .. }) => return Ok(Outcome::Ok(pull)),
2048 // Moved again in between, or already caught up: land it now.
2049 Outcome::Ok(PullBranchUpdate::UpToDate { .. }) => self.forget_landing(&pull.id).await?,
2050 Outcome::Ok(PullBranchUpdate::NeedsAgent { paths, .. }) => {
2051 self.forget_landing(&pull.id).await?;
2052 let named = if paths.is_empty() { String::new() } else { format!(" ({})", paths.join(", ")) };
2053 return Ok(Outcome::fail(
2054 FailureCode::Conflict,
2055 format!(
2056 "{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."
2057 ),
2058 ));
2059 }
2060 // Only those who can push where its branch is can catch
2061 // it up; anyone may merge it another way.
2062 Outcome::Fail(failure) if failure.code == FailureCode::Forbidden => {
2063 self.forget_landing(&pull.id).await?;
2064 let source = pull.source_path(&a.repo);
2065 return Ok(Outcome::fail(
2066 FailureCode::Conflict,
2067 merging::cannot_fast_forward(&base, &format!("{}/{}", source.namespace, source.name)),
2068 ));
2069 }
2070 Outcome::Fail(failure) => {
2071 self.forget_landing(&pull.id).await?;
2072 return Ok(Outcome::Fail(failure));
2073 }
2074 }
2075 }
2076 }
2077
2078 // Whether the actor may write to the repository is decided by repos.
2079 let head = pull.head_commit.clone();
2080 // Bounded like the catch-up above. Landing the same checked commit
2081 // again is recognised as done (repos `land`), so trying again after
2082 // a land that finished late records it.
2083 let landed: Option<Outcome<Landed>> = g1t_kit::call_within(
2084 &self.repos,
2085 "land",
2086 &LandArgs {
2087 // A pull request from a branch lands from the repository
2088 // itself, or from the fork the branch is in.
2089 source_id: pull.source_id().to_owned(),
2090 branch: pull.branch.clone(),
2091 actor: a.actor.clone(),
2092 target_branch: Some(base.clone()),
2093 expected_commit: head.clone(),
2094 target_id: pull.target_id(),
2095 method: Some(method),
2096 message,
2097 author: merging::squash_author(&pull, method),
2098 sign_off: settings.require_web_commit_sign_off,
2099 pull: Some(pull.number),
2100 },
2101 lifecycle::REPOS_WRITE_WITHIN,
2102 )
2103 .await?;
2104 let landed = check!(landed.unwrap_or_else(lifecycle::repository_unreachable));
2105 self.record_merge_evaluations(&repo, &pull, &gate).await;
2106 let merged = self.record_merge(&repo, pull, &a.actor, a.keep_item_open, landed, None).await?;
2107 // Its branch, where the merger or the repository says it goes.
2108 if merged.status == PullStatus::Merged {
2109 self.delete_branch_after_merge(&repo, &merged, &a.actor, a.delete_branch, head).await?;
2110 }
2111 Ok(Outcome::Ok(self.pull_by_id(&merged.id).await?.unwrap_or(merged)))
2112 }
2113
2114 /// Records a pull request as merged once the default branch holds it:
2115 /// lands the work items it closes (closing.rs), links those it is part
2116 /// of, supersedes the others for its own work item in this repository,
2117 /// and says so. `pushed` names the branch when a push brought its head
2118 /// in, rather than a merge from the site or the queue (pushed.rs).
2119 pub(crate) async fn record_merge(
2120 &self,
2121 repo: &Repo,
2122 mut pull: Pull,
2123 actor: &User,
2124 keep_item_open: bool,
2125 landed: Landed,
2126 pushed: Option<&str>,
2127 ) -> Result<Pull> {
2128 // Only a merge into the default branch lands work: into another
2129 // branch, it has not landed yet. Its own work item and those it
2130 // names land, where the repository closes them that way.
2131 let settings = self.settings(&repo.id).await?;
2132 // Only what whoever merges it, and whoever it is for, may close: an
2133 // outside collaborator's "Fixes #N" closes no work item they cannot see.
2134 let owner = self.pull_owner(&pull).await?;
2135 let mut who: Vec<&User> = vec![actor];
2136 who.extend(owner.as_ref().filter(|owner| owner.id != actor.id));
2137 let named = self.items_to_close(&pull, repo, &settings, keep_item_open, &who).await?;
2138 let own = self.pull_item_id(&pull.id).await?;
2139 // A fast-forward leaves the base at what was landed, which is its
2140 // head; the other methods leave its head as it was.
2141 let head = if landed.method == g1t_contracts::rules::MergeMethod::FastForward {
2142 Some(landed.commit.clone())
2143 } else {
2144 pull.head_commit.clone().or_else(|| Some(landed.commit.clone()))
2145 };
2146 let now = rfc3339(now_ms());
2147 let mut statements = vec![self
2148 .db
2149 .prepare(
2150 "UPDATE pulls
2151 SET status = 'merged', head_commit = ?, merge_base = ?, merged_by = ?,
2152 merged_at = ?, updated_at = ?, merge_method = ?, merge_commit = ?,
2153 auto_merge = NULL, closes_items = ?, parts = ?
2154 WHERE id = ? AND status <> 'merged' RETURNING id AS value",
2155 )
2156 .bind(&[
2157 optional(&head),
2158 optional(&landed.previous),
2159 actor.username.as_str().into(),
2160 now.as_str().into(),
2161 now.as_str().into(),
2162 landed.method.as_str().into(),
2163 landed.commit.as_str().into(),
2164 serde_json::to_string(&named.close_ids())?.into(),
2165 serde_json::to_string(&named.part_ids())?.into(),
2166 pull.id.as_str().into(),
2167 ])?];
2168 // The others for its own work item in this repository were other
2169 // ways to do the same thing.
2170 if let Some(own) = own.as_deref().filter(|own| named.close_ids().iter().any(|id| id == own)) {
2171 statements.push(
2172 self.db
2173 .prepare(
2174 "UPDATE pulls SET status = 'closed', superseded_by = ?, updated_at = ?
2175 WHERE item_id = ? AND repo_id = ? AND id != ? AND status IN ('draft', 'open')",
2176 )
2177 .bind(&[pull.number.into(), now.as_str().into(), own.into(), repo.id.as_str().into(), pull.id.as_str().into()])?,
2178 );
2179 }
2180 let recorded = self.db.batch(statements).await?;
2181 if recorded.first().map(|result| result.results::<rows::ValueRow>()).transpose()?.is_none_or(|rows| rows.is_empty()) {
2182 return Ok(self.pull_by_id(&pull.id).await?.unwrap_or(pull));
2183 }
2184
2185 self.publish(
2186 "pull.merged",
2187 &repo.id,
2188 actor,
2189 PullEvent {
2190 commit: Some(landed.commit.clone()),
2191 ..Self::pull_event(&pull)
2192 },
2193 )
2194 .await?;
2195 let who = (actor.id.as_str(), actor.username.as_str());
2196 let note = match pushed {
2197 Some(branch) => format!("merged this by pushing {} to {branch}", landed.commit),
2198 None => merging::merged_note(landed.method).to_owned(),
2199 };
2200 self.note(&repo.id, pull.number, who, &note).await?;
2201 let landing = items::Landing { pull: Some(pull.number), commit: None, pushed_by: None };
2202 items::land(self, repo, &landing, actor, &named.close_ids(), &named.part_ids(), None, false).await?;
2203 let through_queue = self.landed_through_queue(&pull.id).await?;
2204 self.keep_decision_quietly(
2205 &repo.id,
2206 decisions::NewDecision {
2207 item: own.clone(),
2208 pull: Some(pull.number),
2209 session_id: None,
2210 actor: (!decisions::is_g1t(actor)).then_some(actor),
2211 kind: g1t_contracts::decisions::DecisionKind::Merge,
2212 confidence: decisions::merge_confidence(through_queue, actor),
2213 summary: match pushed {
2214 _ if through_queue => format!("#{} landed through the merge queue.", pull.number),
2215 Some(branch) => format!("{} merged #{} by pushing to {branch}.", actor.username, pull.number),
2216 None => format!("{} merged #{}.", actor.username, pull.number),
2217 },
2218 evidence: serde_json::json!({
2219 "commit": landed.commit,
2220 "previous": landed.previous,
2221 "method": landed.method,
2222 "merge_queue": through_queue,
2223 "pushed": pushed.is_some(),
2224 }),
2225 },
2226 )
2227 .await;
2228 pull.status = PullStatus::Merged;
2229 pull.head_commit = head;
2230 pull.merge_method = Some(landed.method);
2231 pull.merge_commit = Some(landed.commit.clone());
2232 pull.auto_merge = None;
2233 pull.closes = named.close_numbers();
2234 pull.parts = named.part_numbers();
2235 pull.merge_base = landed.previous;
2236 pull.merged_by = Some(actor.username.clone());
2237 pull.merged_at = Some(now.clone());
2238 pull.updated_at = now;
2239 Ok(pull)
2240 }
2241
2242 async fn list_active_pulls(&self, a: ViewerArgs) -> Result<Vec<ActivePull>> {
2243 let Some(viewer) = a.viewer else {
2244 return Ok(Vec::new());
2245 };
2246 // The pull requests and their work items, in one round trip: their
2247 // own, and those g1t made for them (Pull::owner).
2248 let author = [JsValue::from(viewer.id.as_str())];
2249 let found = self
2250 .timing
2251 .db(
2252 2,
2253 self.db.batch(vec![
2254 self.db
2255 .prepare(format!(
2256 "SELECT {PULL_COLUMNS} FROM pulls
2257 WHERE COALESCE(requested_by_id, author_id) = ?1 AND status IN ('draft', 'open')
2258 ORDER BY updated_at DESC LIMIT 50"
2259 ))
2260 .bind(&author)?,
2261 self.db
2262 .prepare(items::select(
2263 "work_items.id IN (
2264 SELECT item_id FROM pulls
2265 WHERE COALESCE(requested_by_id, author_id) = ?1 AND status IN ('draft', 'open') AND item_id IS NOT NULL
2266 ORDER BY updated_at DESC LIMIT 50)",
2267 ))
2268 .bind(&author)?,
2269 ]),
2270 )
2271 .await?;
2272 let (Some(found), Some(item_rows)) = (found.first(), found.get(1)) else {
2273 return Ok(Vec::new());
2274 };
2275 let snapshots = found.results::<Snapshot>()?;
2276 let item_ids: Vec<Option<String>> = found.results::<PullItemRow>()?.into_iter().map(|row| row.item_id).collect();
2277 let pulls: Vec<Pull> = found.results::<PullRow>()?.into_iter().map(Pull::from).collect();
2278 let items: Vec<Held> = items::held_rows(item_rows)?;
2279 let items = &items;
2280 let viewer = &viewer;
2281 // Where each stands: the remembered assessment when there is one,
2282 // and worked out otherwise.
2283 try_join_all(pulls.into_iter().zip(snapshots).zip(item_ids).map(|((pull, snapshot), item_id)| async move {
2284 let item = item_id.and_then(|id| {
2285 items
2286 .iter()
2287 .find(|held| held.item.id == id && held.may(Some(viewer), Capability::Read))
2288 .map(|held| held.clone().shown(Some(viewer)))
2289 });
2290 // Every open pull request has a lifecycle (lifecycle.rs), kept
2291 // fresh by what moves it (`advance`, on each event that can);
2292 // anyone's draft but g1t's has none.
2293 let managed = lifecycle::made_by_g1t(&pull) && snapshot.managed != 0;
2294 let lifecycle = if pull.status == PullStatus::Draft && !managed {
2295 None
2296 } else if let (Some(stage), Some(detail)) = (snapshot.stage, snapshot.stage_detail) {
2297 Some(Lifecycle {
2298 stage,
2299 detail,
2300 revisions: snapshot.revisions,
2301 })
2302 } else {
2303 // Never worked out yet: once, and remembered.
2304 let behind = self.behind_now(&pull).await?;
2305 self.assess(&pull, &item, behind)
2306 .await?
2307 .map(|(lifecycle, _)| lifecycle)
2308 };
2309 Ok::<_, worker::Error>(ActivePull {
2310 pull,
2311 item,
2312 lifecycle,
2313 })
2314 }))
2315 .await
2316 }
2317
2318 // --- Sessions ----------------------------------------------------------
2319
2320 async fn append_session(&self, a: AppendSessionArgs) -> Result<Outcome<Appended>> {
2321 if a.entries.is_empty() {
2322 return Ok(Outcome::Ok(Appended { count: 0 }));
2323 }
2324 if a.entries.len() > MAX_ENTRY_BATCH {
2325 return Ok(Outcome::fail(
2326 FailureCode::Invalid,
2327 format!("Send at most {MAX_ENTRY_BATCH} entries at a time."),
2328 ));
2329 }
2330 let viewer = Some(a.actor.clone());
2331 let (_, pull) = check!(self.pull_at(&a.repo, a.number, &viewer).await?);
2332 if !pull.is_owned_by(&a.actor.id) {
2333 return Ok(Outcome::fail(
2334 FailureCode::Forbidden,
2335 "Only whoever opened a pull request, or asked g1t for it, can record its session.",
2336 ));
2337 }
2338
2339 let now = rfc3339(now_ms());
2340 let count = a.entries.len() as u32;
2341 let mut statements = Vec::with_capacity(a.entries.len() + 1);
2342 for entry in a.entries {
2343 let kind = serde_json::to_value(entry.kind)?;
2344 let text = coordination::shown::clean(&entry.text.chars().take(MAX_ENTRY_CHARS).collect::<String>());
2345 // Each insert takes the next sequence number itself, so two
2346 // writers appending at once cannot collide.
2347 statements.push(
2348 self.db
2349 .prepare(
2350 "INSERT INTO session_entries (pull_id, seq, kind, text, tool, \"commit\", at)
2351 SELECT ?, COALESCE(MAX(seq), 0) + 1, ?, ?, ?, ?, ?
2352 FROM session_entries WHERE pull_id = ?",
2353 )
2354 .bind(&[
2355 pull.id.as_str().into(),
2356 kind.as_str().unwrap_or("note").into(),
2357 text.into(),
2358 optional(&entry.tool),
2359 optional(&entry.commit.or_else(|| pull.head_commit.clone())),
2360 now.as_str().into(),
2361 pull.id.as_str().into(),
2362 ])?,
2363 );
2364 }
2365 statements.push(
2366 self.db
2367 .prepare("UPDATE pulls SET updated_at = ? WHERE id = ?")
2368 .bind(&[now.as_str().into(), pull.id.as_str().into()])?,
2369 );
2370 self.db.batch(statements).await?;
2371 self.publish(
2372 "session.appended",
2373 &pull.repo_id,
2374 &a.actor,
2375 SessionAppended {
2376 pull_id: pull.id.clone(),
2377 repo_id: pull.repo_id.clone(),
2378 number: pull.number,
2379 count,
2380 },
2381 )
2382 .await?;
2383 Ok(Outcome::Ok(Appended { count }))
2384 }
2385
2386 async fn read_session(&self, a: ViewArgs) -> Result<Outcome<Vec<SessionEntry>>> {
2387 let (_, pull) = check!(self.pull_at(&a.repo, a.number, &a.viewer).await?);
2388 let rows = self
2389 .db
2390 .prepare(
2391 "SELECT seq, kind, text, tool, \"commit\", at FROM session_entries
2392 WHERE pull_id = ? AND seq > ? ORDER BY seq LIMIT ?",
2393 )
2394 .bind(&[pull.id.into(), a.after_seq.into(), SESSION_PAGE.into()])?
2395 .all()
2396 .await?
2397 .results::<SessionRow>()?;
2398 Ok(Outcome::Ok(
2399 rows.into_iter().map(SessionEntry::from).collect(),
2400 ))
2401 }
2402
2403 /// A push moves the head of the pull request it concerns: the one whose
2404 /// fork was pushed to, or the one opened from the branch that moved.
2405 async fn on_event(&self, event: &Event) -> Result<()> {
2406 // A new repository: where it is, for its workspace's work items and
2407 // numbers; a workspace's first starts it with the default labels.
2408 if event.kind == "repo.created" {
2409 return items::repo_created(self, &event.data).await;
2410 }
2411 if event.kind == "repo.restored" {
2412 return items::repo_restored(self, &event.data).await;
2413 }
2414 if event.kind == "repo.visibility_changed" || event.kind == "repo.updated" {
2415 return items::repo_visibility(self, &event.data).await;
2416 }
2417 // Pull requests into the branch that became the default merge into
2418 // the default branch, which is stored as none.
2419 if event.kind == "repo.default_branch_changed"
2420 && let (Some(repo_id), Some(to)) = (event.repo_id.as_deref(), event.data["to"].as_str())
2421 {
2422 self.db
2423 .prepare(
2424 "UPDATE pulls SET base_branch = NULL
2425 WHERE repo_id = ? AND base_branch = ? AND status IN ('draft', 'open')",
2426 )
2427 .bind(&[repo_id.into(), to.into()])?
2428 .run()
2429 .await?;
2430 return Ok(());
2431 }
2432 if event.kind != "git.push" {
2433 return Ok(());
2434 }
2435 let (Some(repo_id), Some(after), Some(git_ref)) = (
2436 event.repo_id.as_deref(),
2437 event.data["after"].as_str(),
2438 event.data["ref"].as_str(),
2439 ) else {
2440 return Ok(());
2441 };
2442 let now = rfc3339(now_ms());
2443 // Who moved it, for rules about the most recent push.
2444 let pusher: JsValue = event.actor.as_deref().map_or(JsValue::NULL, Into::into);
2445 // The head moved, so whatever the checks said no longer applies, and
2446 // whatever step g1t was waiting on has been taken.
2447 let moved = "UPDATE pulls
2448 SET head_commit = ?, updated_at = ?, head_pushed_by = ?, head_pushed_at = ?, check_status = NULL, check_run_id = NULL,
2449 working_on = NULL, working_until = NULL, stalled = NULL";
2450 let active = "status IN ('draft', 'open') AND head_commit IS NOT ?";
2451 let returning = "RETURNING id, repo_id, number, status, author_id, author_name, requested_by_id, requested_by_name";
2452 let mut pulls: Vec<MovedRow> = Vec::new();
2453 // A fork carries its pull request on its default branch.
2454 if event.data["defaultBranch"].as_bool() == Some(true) {
2455 pulls.extend(
2456 self.db
2457 .prepare(format!(
2458 "{moved} WHERE fork_repo_id = ? AND {active} {returning}"
2459 ))
2460 .bind(&[
2461 after.into(),
2462 now.as_str().into(),
2463 pusher.clone(),
2464 now.as_str().into(),
2465 repo_id.into(),
2466 after.into(),
2467 ])?
2468 .all()
2469 .await?
2470 .results::<MovedRow>()?,
2471 );
2472 }
2473 if let Some(branch) = git_ref.strip_prefix("refs/heads/") {
2474 // A branch of the repository, and a branch of a fork pull
2475 // requests into other repositories were opened from.
2476 pulls.extend(
2477 self.db
2478 .prepare(format!(
2479 "{moved} WHERE {} AND {active} {returning}",
2480 head_repo::ON_BRANCH
2481 ))
2482 .bind(&[
2483 after.into(),
2484 now.as_str().into(),
2485 pusher.clone(),
2486 now.as_str().into(),
2487 repo_id.into(),
2488 repo_id.into(),
2489 branch.into(),
2490 after.into(),
2491 ])?
2492 .all()
2493 .await?
2494 .results::<MovedRow>()?,
2495 );
2496 }
2497 // What each now changes, so overlaps show while the work is under way.
2498 let mut moved_pulls = Vec::new();
2499 for moved in &pulls {
2500 if let Some(mut pull) = self.pull_by_id(&moved.id).await? {
2501 pull.files = self.refresh_files(&pull).await?;
2502 // Owners of files it now changes are asked too.
2503 self.refresh_code_owners(&pull).await;
2504 // A new overlap with another pull request is announced.
2505 self.announce_overlaps(&pull, event.actor.clone()).await;
2506 // Auto-merge someone else turned on stops (automerge.rs).
2507 self.auto_merge_after_push(&pull, event.actor.as_deref()).await?;
2508 moved_pulls.push(pull);
2509 }
2510 }
2511 // The harness sessions that pushed the branch, or that its new
2512 // commits name in trailers, are tied to its pull requests.
2513 if let Some(branch) = git_ref.strip_prefix("refs/heads/") {
2514 let commits = event.data["commits"].as_u64().and_then(|n| u32::try_from(n).ok());
2515 self.link_push(repo_id, branch, after, commits, &moved_pulls).await;
2516 }
2517 // A merge that was waiting for this push to bring it up to date.
2518 for moved in &pulls {
2519 self.land_if_requested(&moved.id).await?;
2520 }
2521 // The pull requests into the branch whose heads it brought in are
2522 // merged, and the work items its commits name landed (pushed.rs).
2523 if let Some(branch) = git_ref.strip_prefix("refs/heads/") {
2524 self.after_branch_push(event, repo_id, branch, after).await;
2525 }
2526 // Whether each still merges cleanly, and, when a default branch
2527 // moved, every open pull request into it.
2528 let moved_ids: Vec<String> = pulls.iter().map(|pull| pull.id.clone()).collect();
2529 let into = match event.data["defaultBranch"].as_bool() {
2530 Some(true) => mergeability::Moved::DefaultBranch,
2531 _ => match git_ref.strip_prefix("refs/heads/") {
2532 Some(branch) => mergeability::Moved::Branch(branch),
2533 None => mergeability::Moved::Nothing,
2534 },
2535 };
2536 self.after_push(repo_id, into, &moved_ids).await;
2537 // A draft is announced when it is marked ready instead.
2538 for pull in pulls
2539 .into_iter()
2540 .filter(|pull| pull.status == PullStatus::Open)
2541 {
2542 let item = moved_pulls.iter().find(|moved| moved.id == pull.id).and_then(|moved| moved.item);
2543 self.publish_as(
2544 "pull.updated",
2545 &pull.repo_id,
2546 event.actor.clone(),
2547 PullEvent {
2548 author: Some(g1t_contracts::credentials::Principal { id: pull.author_id, username: pull.author_name }),
2549 requested_by: pull
2550 .requested_by_id
2551 .zip(pull.requested_by_name)
2552 .map(|(id, username)| g1t_contracts::credentials::Principal { id, username }),
2553 pull_id: pull.id,
2554 repo_id: pull.repo_id.clone(),
2555 number: pull.number,
2556 item,
2557 commit: Some(after.to_owned()),
2558 ..PullEvent::default()
2559 }
2560 .carrying(&event.data),
2561 )
2562 .await?;
2563 }
2564 Ok(())
2565 }
2566}
2567
2568/// An event's data that carries on what caused the event it follows from.
2569trait Carrying: serde::Serialize + Sized {
2570 /// As JSON, marked as a workflow job's doing when `cause` was.
2571 fn carrying(self, cause: &serde_json::Value) -> serde_json::Value {
2572 g1t_contracts::events::carried(self, cause)
2573 }
2574}
2575
2576impl Carrying for PullEvent {}
2577
2578fn service(env: &Env) -> Result<Work> {
2579 Ok(Work {
2580 db: env.d1("DB")?,
2581 identity: env.service("IDENTITY")?,
2582 repos: env.service("REPOS")?,
2583 events: env.service("EVENTS")?,
2584 actions: env.service("ACTIONS")?,
2585 integrations: env.service("INTEGRATIONS").ok(),
2586 timing: g1t_kit::d1::Timing::default(),
2587 prefetched: std::cell::RefCell::new(None),
2588 known_repos: std::cell::RefCell::new(std::collections::HashMap::new()),
2589 })
2590}
2591
2592#[event(fetch)]
2593async fn fetch(mut request: Request, env: Env, _ctx: Context) -> Result<Response> {
2594 let Some(method) = rpc_method(&request) else {
2595 return Response::error("Not found", 404);
2596 };
2597 // A replica near the caller when it asks for one (crates/kit/src/d1.rs).
2598 let (db, served) = g1t_kit::d1::open(&env, "DB", &request)?;
2599 let body: serde_json::Value = request.json().await?;
2600 let mut work = service(&env)?;
2601 work.db = db;
2602
2603 let answered = match method.as_str() {
2604 // Work items (items/) and moving issues into them (items/migration.rs).
2605 "open_work_item" => reply(&work.open_work_item(args(body)?).await?),
2606 "list_work_items" => reply(&work.list_work_items(args(body)?).await?),
2607 "get_work_item" => reply(&work.get_work_item(args(body)?).await?),
2608 "update_work_item" => reply(&work.update_work_item(args(body)?).await?),
2609 "close_work_item" => reply(&work.close_work_item(args(body)?).await?),
2610 "reopen_work_item" => reply(&work.reopen_work_item(args(body)?).await?),
2611 "link_work_item" => reply(&work.link_work_item(args(body)?).await?),
2612 "unlink_work_item" => reply(&work.unlink_work_item(args(body)?).await?),
2613 "resolve_ref" => reply(&work.resolve_ref(args(body)?).await?),
2614 "legacy_item_map" => reply(&work.legacy_item_map(args(body)?).await?),
2615 "migrate_work_items" => {
2616 let _: MigrateWorkItemsArgs = args(body)?;
2617 reply(&work.migrate_work_items().await?)
2618 }
2619 "list_labels" => reply(&work.list_labels(args(body)?).await?),
2620 "save_label" => reply(&work.save_label(args(body)?).await?),
2621 "delete_label" => reply(&work.delete_label(args(body)?).await?),
2622 "add_default_labels" => reply(&work.add_default_labels(args(body)?).await?),
2623 "set_labels" => reply(&work.set_labels(args(body)?).await?),
2624 "list_milestones" => reply(&work.list_milestones(args(body)?).await?),
2625 "get_milestone" => reply(&work.get_milestone(args(body)?).await?),
2626 "save_milestone" => reply(&work.save_milestone(args(body)?).await?),
2627 "delete_milestone" => reply(&work.delete_milestone(args(body)?).await?),
2628 "counts" => reply(&work.counts(args(body)?).await?),
2629 "add_comment" => reply(&work.add_comment(args(body)?).await?),
2630 "upload_attachment" => reply(&work.upload_attachment(&env, args(body)?).await?),
2631 "edit_comment" => reply(&work.edit_comment(args(body)?).await?),
2632 "delete_comment" => reply(&work.delete_comment(args(body)?).await?),
2633 "react" => reply(&work.react(args(body)?).await?),
2634 "commit_comments" => reply(&work.commit_comments(args(body)?).await?),
2635 "closed_by_commit" => reply(&work.closed_by_commit(args(body)?).await?),
2636 "add_commit_comment" => reply(&work.add_commit_comment(args(body)?).await?),
2637 "edit_commit_comment" => reply(&work.edit_commit_comment(args(body)?).await?),
2638 "delete_commit_comment" => reply(&work.delete_commit_comment(args(body)?).await?),
2639 "seen_checks" => reply(&work.seen_checks(args(body)?).await?),
2640 "report_checks" => reply(&work.report_checks(args(body)?).await?),
2641 "set_commit_status" => reply(&work.set_commit_status(args(body)?).await?),
2642 "create_commit_status" => reply(&work.create_commit_status(args(body)?).await?),
2643 "commit_statuses" => reply(&work.commit_statuses(args(body)?).await?),
2644 "combined_status" => reply(&work.combined_status(args(body)?).await?),
2645 "create_check_run" => reply(&work.create_check_run(args(body)?).await?),
2646 "update_check_run" => reply(&work.update_check_run(args(body)?).await?),
2647 "get_check_run" => reply(&work.get_check_run(args(body)?).await?),
2648 "check_run_annotations" => reply(&work.check_run_annotations(args(body)?).await?),
2649 "ref_check_runs" => reply(&work.ref_check_runs(args(body)?).await?),
2650 "ref_check_suites" => reply(&work.ref_check_suites(args(body)?).await?),
2651 "get_check_suite" => reply(&work.get_check_suite(args(body)?).await?),
2652 "rerequest_check_run" => reply(&work.rerequest_check_run(args(body)?).await?),
2653 "rerequest_check_suite" => reply(&work.rerequest_check_suite(args(body)?).await?),
2654 "request_check_action" => reply(&work.request_check_action(args(body)?).await?),
2655 "commit_checks" => reply(&work.commit_checks(args(body)?).await?),
2656 "advance" => reply(&work.advance(args(body)?).await?),
2657 "managed_pulls" => reply(&work.managed_pulls(args(body)?).await?),
2658 "queue" => reply(&work.queue(args(body)?).await?),
2659 "queue_build" => reply(&work.queue_build(args(body)?).await?),
2660 "report_queue" => reply(&work.report_queue(args(body)?).await?),
2661 "remove_from_queue" => reply(&work.remove_from_queue(args(body)?).await?),
2662 "locate_pull" => reply(&work.locate_pull(args(body)?).await?),
2663 "inbox_subject" => reply(&work.inbox_subject(args(body)?).await?),
2664 "get_settings" => reply(&work.get_settings(args(body)?).await?),
2665 "codeowners_errors" => reply(&work.codeowners_errors(args(body)?).await?),
2666 "update_settings" => reply(&work.update_settings(args(body)?).await?),
2667 "open_pull" => reply(&work.open_pull(args(body)?).await?),
2668 "list_pulls" => reply(&work.list_pulls(args(body)?).await?),
2669 "delivery_insights" => reply(&work.delivery_insights(args(body)?).await?),
2670 "workspace_signals" => reply(&work.workspace_signals(args(body)?).await?),
2671 "pulls_for_repos" => reply(&work.pulls_for_repos(args(body)?).await?),
2672 "get_pull" => reply(&work.get_pull(args(body)?).await?),
2673 "update_pull" => reply(&work.update_pull(args(body)?).await?),
2674 "catch_up_pull" => reply(&work.catch_up_pull(args(body)?).await?),
2675 "ready_pull" => reply(&work.ready_pull(args(body)?).await?),
2676 "close_pull" => reply(&work.close_pull(args(body)?).await?),
2677 "reopen_pull" => reply(&work.reopen_pull(args(body)?).await?),
2678 "convert_pull_to_draft" => reply(&work.convert_pull_to_draft(args(body)?).await?),
2679 "merge_pull" => reply(&work.merge_pull(args(body)?).await?),
2680 // How it may merge, auto-merge and its branch (merging.rs, automerge.rs).
2681 "merge_choices" => reply(&work.merge_choices(args(body)?).await?),
2682 "set_auto_merge" => reply(&work.set_auto_merge(args(body)?).await?),
2683 "delete_head_branch" => reply(&work.delete_head_branch(args(body)?).await?),
2684 "restore_head_branch" => reply(&work.restore_head_branch(args(body)?).await?),
2685 "list_active_pulls" => reply(&work.list_active_pulls(args(body)?).await?),
2686 "by_author" => reply(&work.by_author(args(body)?).await?),
2687 "contributions" => reply(&work.contributions(args(body)?).await?),
2688 "append_session" => reply(&work.append_session(args(body)?).await?),
2689 "read_session" => reply(&work.read_session(args(body)?).await?),
2690 // Decision records (decisions.rs).
2691 "record_decision" => reply(&work.record_decision(args(body)?).await?),
2692 "list_decisions" => reply(&work.list_decisions(args(body)?).await?),
2693 // Coordinating work: harness sessions, claims, dependencies and a
2694 // work item's whole context (coordination/).
2695 "start_session" => reply(&work.start_session(args(body)?).await?),
2696 "update_session" => reply(&work.update_session(args(body)?).await?),
2697 "end_session" => reply(&work.end_session(args(body)?).await?),
2698 "title_session" => reply(&work.title_session(args(body)?).await?),
2699 "get_work_session" => reply(&work.get_work_session(args(body)?).await?),
2700 "list_work_sessions" => reply(&work.list_work_sessions(args(body)?).await?),
2701 "list_workspace_sessions" => reply(&work.list_workspace_sessions(args(body)?).await?),
2702 "get_workspace_session" => reply(&work.get_workspace_session(args(body)?).await?),
2703 "pull_sessions" => reply(&work.pull_sessions(args(body)?).await?),
2704 "record_session_entries" => reply(&work.record_session_entries(args(body)?).await?),
2705 "claim_work_item" => reply(&work.claim_work_item(args(body)?).await?),
2706 "release_claim" => reply(&work.release_claim(args(body)?).await?),
2707 "hand_off_work_item" => reply(&work.hand_off_work_item(args(body)?).await?),
2708 "list_claims" => reply(&work.list_claims(args(body)?).await?),
2709 "get_work" => reply(&work.get_work(args(body)?).await?),
2710 // Runs of sandboxes (runs.rs).
2711 "open_run" => reply(&work.open_run(args(body)?).await?),
2712 "report_run" => reply(&work.report_run(args(body)?).await?),
2713 "list_runs" => reply(&work.list_runs(args(body)?).await?),
2714 "get_run" => reply(&work.get_run(args(body)?).await?),
2715 "list_sessions" => reply(&work.list_sessions(args(body)?).await?),
2716 "get_session" => reply(&work.get_session(args(body)?).await?),
2717 // What a sandbox may reach (guardrails.rs).
2718 "get_guardrails" => reply(&work.get_guardrails(args(body)?).await?),
2719 "update_guardrails" => reply(&work.update_guardrails(args(body)?).await?),
2720 "run_guardrails" => reply(&work.run_guardrails(args(body)?).await?),
2721 // The runs whose sandboxes stop with their repository (retired.rs).
2722 "runs_in_repo" => reply(&work.runs_in_repo(args(body)?).await?),
2723 "start_mergecheck" => reply(&work.start_mergecheck(args(body)?).await?),
2724 "report_mergecheck" => reply(&work.report_mergecheck(args(body)?).await?),
2725 // Rulesets (rulesets.rs): kept here, enforced here on merge and by
2726 // repos on push.
2727 "list_rulesets" => reply(&work.list_rulesets(args(body)?).await?),
2728 "get_ruleset" => reply(&work.get_ruleset(args(body)?).await?),
2729 "save_ruleset" => reply(&work.save_ruleset(args(body)?).await?),
2730 "delete_ruleset" => reply(&work.delete_ruleset(args(body)?).await?),
2731 "effective_rules" => reply(&work.effective_rules(args(body)?).await?),
2732 "rule_evaluations" => reply(&work.rule_evaluations(args(body)?).await?),
2733 "ref_rules" => reply(&work.ref_rules(args(body)?).await?),
2734 "record_evaluations" => reply(&work.record_evaluations(args(body)?).await?),
2735 "set_requires_pull_request" => reply(&work.set_requires_pull_request(args(body)?).await?),
2736 _ => Response::error("Unknown method", 404),
2737 };
2738 served.finish_timed(answered, &work.timing)
2739}
2740
2741/// The cron trigger: moves on the job that moves issues into work items
2742/// (items/migration.rs), until it is done, and settles the sessions'
2743/// titles that are due (coordination/titles.rs).
2744#[event(scheduled)]
2745async fn scheduled(_event: ScheduledEvent, env: Env, _ctx: ScheduleContext) {
2746 let moved = async {
2747 let work = service(&env)?;
2748 work.migrate_work_items().await
2749 };
2750 if let Err(error) = moved.await {
2751 worker::console_error!("moving issues into work items: {error}");
2752 }
2753 let settled = async {
2754 let work = service(&env)?;
2755 work.settle_titles().await
2756 };
2757 if let Err(error) = settled.await {
2758 worker::console_error!("settling session titles: {error}");
2759 }
2760}
2761
2762/// Events from the bus, delivered on this service's own queue.
2763#[event(queue)]
2764async fn queue(batch: MessageBatch<Event>, env: Env, _ctx: Context) -> Result<()> {
2765 let work = service(&env)?;
2766 for message in batch.messages()? {
2767 // A workspace renamed: its runs, work items, labels, milestones and
2768 // numbers move to the slug it has now.
2769 let renamed = [runs::RENAMED, guardrails::RENAMED, rulesets::RENAMED, items::RENAMED, attachments::RENAMED].concat();
2770 if g1t_kit::rename::on_event(&env, &env.d1("DB")?, message.body(), &renamed).await? {
2771 message.ack();
2772 continue;
2773 }
2774 // A repository renamed or transferred: its runs and guardrails
2775 // follow, and where it is is recorded for work items and numbers.
2776 let transferred = [runs::TRANSFERRED, guardrails::TRANSFERRED, head_repo::TRANSFERRED, items::TRANSFERRED].concat();
2777 if g1t_kit::transfer::on_event(&env, &env.d1("DB")?, message.body(), &transferred).await? {
2778 message.ack();
2779 continue;
2780 }
2781 // A workspace deleted: what it kept for itself goes, its work items'
2782 // attachments' bytes first.
2783 if message.body().kind == "workspace.deleted"
2784 && let Some(slug) = message.body().data["slug"].as_str()
2785 {
2786 attachments::purge_workspace(&env, &work.db, slug).await?;
2787 }
2788 let deleted = [guardrails::DELETED, items::DELETED, attachments::DELETED].concat();
2789 if g1t_kit::deleted::on_event(&env.d1("DB")?, message.body(), &deleted).await? {
2790 message.ack();
2791 continue;
2792 }
2793 // An account purged: what it wrote shows as ghost (ghost.rs).
2794 let ghost = ghost::statements();
2795 let ghost: Vec<&str> = ghost.iter().map(String::as_str).collect();
2796 if g1t_kit::user_deleted::on_event(&env.d1("DB")?, message.body(), &ghost).await? {
2797 message.ack();
2798 continue;
2799 }
2800 // A repository purged: its attachments' bytes go before their rows do.
2801 if let Some(g1t_kit::lifecycle::Lifecycle::Purged(purged)) = g1t_kit::lifecycle::read(message.body()) {
2802 attachments::purge(&env, &work.db, &purged.repo_id).await?;
2803 }
2804 // A repository deleted, archived or purged, or a branch renamed (retired.rs).
2805 if work.on_retired(message.body()).await? {
2806 message.ack();
2807 continue;
2808 }
2809 work.on_event(message.body()).await?;
2810 message.ack();
2811 }
2812 Ok(())
2813}
2814
2815/// The rules that once read a pull request's author read its owner now:
2816/// whoever asked g1t for it, or its author. For each, the person who asked
2817/// is held to what an author was, and g1t's agent (a token it works with)
2818/// gains nothing by being the author.
2819#[cfg(test)]
2820mod owner_rules {
2821 use super::*;
2822 use crate::rows::stored::{ASKER, G1T, pull};
2823 use g1t_contracts::identity::AGENT_ID;
2824
2825 const SOMEONE: &str = "usr_2";
2826
2827 #[test]
2828 fn no_self_approval() {
2829 // add_comment refuses a verdict on one that is theirs.
2830 let made = pull(G1T, Some(ASKER));
2831 assert!(made.is_owned_by(ASKER.0), "the person who asked cannot approve it");
2832 assert!(!made.is_owned_by(AGENT_ID), "g1t's review agent still gives its verdict");
2833 assert!(!made.is_owned_by(SOMEONE));
2834 }
2835
2836 #[test]
2837 fn what_an_author_could_do_without_a_role() {
2838 // manageable_pull (update, ready, close), catch_up_pull on a fork,
2839 // and append_session: theirs to do.
2840 let made = pull(G1T, Some(ASKER));
2841 assert!(made.is_owned_by(ASKER.0));
2842 assert!(!made.is_owned_by(SOMEONE), "anyone else still needs the role");
2843 assert!(!made.is_owned_by(AGENT_ID), "being its author gives g1t's tokens nothing more");
2844 }
2845
2846 #[test]
2847 fn nobody_is_asked_to_review_what_they_asked_for() {
2848 // update_pull drops the owner from the reviewers asked.
2849 let made = pull(G1T, Some(ASKER));
2850 let mut reviewers = vec!["syntaqx".to_owned(), "ana".to_owned()];
2851 reviewers.retain(|name| *name != made.owner().username);
2852 assert_eq!(reviewers, ["ana"]);
2853 }
2854
2855 #[test]
2856 fn sandboxes_act_as_whoever_asked() {
2857 // MergecheckJob and the merge queue's job
2858 // carry who the sandbox's credential acts for: a real account.
2859 let made = pull(G1T, Some(ASKER));
2860 assert_eq!(made.owner().id, ASKER.0);
2861 let acts_as = made.requested_by.unwrap_or(made.author);
2862 assert_eq!(acts_as.id, ASKER.0);
2863 // g1t's own work, which nobody asked for, acts as g1t, as before.
2864 let own = pull(("g1t", "g1t"), None);
2865 assert_eq!(own.requested_by.unwrap_or(own.author).id, "g1t");
2866 }
2867
2868 #[test]
2869 fn events_name_g1t_and_whoever_asked() {
2870 let made = pull(G1T, Some(ASKER));
2871 let event = serde_json::to_value(Work::pull_event(&made)).unwrap();
2872 assert_eq!(event["author"], serde_json::json!({ "id": AGENT_ID, "username": "g1t" }));
2873 assert_eq!(event["requestedBy"], serde_json::json!({ "id": "usr_1", "username": "syntaqx" }));
2874 let own = serde_json::to_value(Work::pull_event(&pull(ASKER, None))).unwrap();
2875 assert!(own.get("requestedBy").is_none());
2876 }
2877
2878 #[test]
2879 fn only_a_closed_pull_request_reopens_and_only_an_open_one_turns_draft() {
2880 assert_eq!(reopen_refusal(PullStatus::Closed), None);
2881 assert!(reopen_refusal(PullStatus::Merged).is_some(), "a merge cannot be undone");
2882 assert!(reopen_refusal(PullStatus::Open).is_some());
2883 assert!(reopen_refusal(PullStatus::Draft).is_some());
2884 assert_eq!(draft_refusal(PullStatus::Open), None);
2885 assert!(draft_refusal(PullStatus::Draft).is_some());
2886 assert!(draft_refusal(PullStatus::Closed).is_some());
2887 assert!(draft_refusal(PullStatus::Merged).is_some());
2888 }
2889
2890 #[test]
2891 fn a_pull_request_reopens_as_what_it_was_closed_as() {
2892 assert_eq!(reopened_status(Some("draft")), PullStatus::Draft);
2893 assert_eq!(reopened_status(Some("open")), PullStatus::Open);
2894 // Closed before this was recorded, or by a merge of another.
2895 assert_eq!(reopened_status(None), PullStatus::Open);
2896 }
2897}