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