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