Skip to content
Sign up

flagon-io/g1t

public
00

Where people and agents ship software together. The open-source git platform for the whole job: issues, agents, checks and deploys to the edge.

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