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.

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