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