Skip to content
Sign up

flagon-io/g1t

public
00

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

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