Sign in

flagon-io/g1t

public

g1t/services/work/src/lib.rs

1,110 lines39,455 bytesCodeBlame

Pick any line to see why it is the way it is: the commit, the pull request and issue it came from, and what the agent was thinking.

Issues and pull requests replace intents and attempts1//! The work service: issues, pull requests, comments and sessions.
Work service in Rust, with RFC 3339 timestamps2//!
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 rows;
8
9use g1t_contracts::events::{
Issues and pull requests replace intents and attempts10 CommentCreated, Delivered, IssueEvent, NewEvent, PullEvent, SessionAppended,
Work service in Rust, with RFC 3339 timestamps11};
Issues and pull requests replace intents and attempts12use g1t_contracts::repos::{ForkArgs, GetArgs, LandArgs, Landed, Repo, RepoPath};
Work service in Rust, with RFC 3339 timestamps13use g1t_contracts::time::rfc3339;
14use g1t_contracts::work::*;
15use g1t_contracts::{FailureCode, Outcome, User, Viewer, new_id};
16use g1t_kit::{args, js, now_ms, reply, rpc_method};
17use serde::Serialize;
18use worker::wasm_bindgen::JsValue;
19use worker::{
20 Context, D1Database, Env, Fetcher, MessageBatch, MessageExt, Request, Response, Result, event,
21};
22
Issues and pull requests replace intents and attempts23use rows::{CommentRow, IssueRow, NumberRow, PullRow, SessionRow, ValueRow};
Work service in Rust, with RFC 3339 timestamps24
25const SOURCE: &str = "work";
26const MAX_ENTRY_BATCH: usize = 200;
27const MAX_ENTRY_CHARS: usize = 64_000;
Issues and pull requests replace intents and attempts28const MAX_TITLE_CHARS: usize = 200;
Work service in Rust, with RFC 3339 timestamps29const SESSION_PAGE: u32 = 500;
Issues and pull requests replace intents and attempts30const LIST_PAGE: u32 = 100;
Work service in Rust, with RFC 3339 timestamps31const UNVERIFIED: &str = "Confirm your email address first. Check your inbox, or resend the link from the banner on g1t.sh.";
32
Issues and pull requests replace intents and attempts33const ISSUE_COLUMNS: &str = "issues.*,
34 (SELECT count(*) FROM pulls WHERE pulls.issue_id = issues.id) AS pull_count,
35 (SELECT count(*) FROM comments
36 WHERE comments.repo_id = issues.repo_id AND comments.number = issues.number) AS comment_count";
Work service in Rust, with RFC 3339 timestamps37
Issues and pull requests replace intents and attempts38fn no_issue<T>() -> Outcome<T> {
39 Outcome::fail(FailureCode::NotFound, "Issue not found.")
Work service in Rust, with RFC 3339 timestamps40}
41
Issues and pull requests replace intents and attempts42fn no_pull<T>() -> Outcome<T> {
43 Outcome::fail(FailureCode::NotFound, "Pull request not found.")
Work service in Rust, with RFC 3339 timestamps44}
45
46fn optional(value: &Option<String>) -> JsValue {
47 value.as_deref().map_or(JsValue::NULL, JsValue::from)
48}
49
Issues and pull requests replace intents and attempts50fn optional_number(value: Option<u32>) -> JsValue {
51 value.map_or(JsValue::NULL, JsValue::from)
52}
53
54/// The lowercase name a `State` is stored and sent as.
55fn state_name(state: Option<State>) -> Option<&'static str> {
56 state.map(|state| match state {
57 State::Open => "open",
58 State::Closed => "closed",
59 })
60}
61
62/// A trimmed title, or why it cannot be used.
63fn valid_title(title: &str) -> std::result::Result<&str, &'static str> {
64 let title = title.trim();
65 if title.is_empty() {
66 Err("A title is required.")
67 } else if title.chars().count() > MAX_TITLE_CHARS {
68 Err("That title is too long.")
69 } else {
70 Ok(title)
71 }
72}
73
74/// Unwraps an `Outcome`, returning its failure from the enclosing method.
75macro_rules! check {
76 ($outcome:expr) => {
77 match $outcome {
78 Outcome::Ok(value) => value,
79 Outcome::Fail(failure) => return Ok(Outcome::Fail(failure)),
80 }
81 };
82}
83
Work service in Rust, with RFC 3339 timestamps84struct Work {
85 db: D1Database,
86 repos: Fetcher,
87 /// The events service, an RPC stub.
88 events: JsValue,
89}
90
91impl Work {
Issues and pull requests replace intents and attempts92 async fn publish<T: Serialize>(
93 &self,
94 kind: &'static str,
95 repo_id: &str,
96 actor: &User,
97 data: T,
98 ) -> Result<()> {
99 let event = NewEvent {
100 kind,
101 source: SOURCE,
102 repo_id: Some(repo_id.to_owned()),
103 actor: Some(actor.id.clone()),
104 data,
105 };
Work service in Rust, with RFC 3339 timestamps106 js::call(&self.events, "publish", &[js::to_js(&[event])?]).await?;
107 Ok(())
108 }
109
Issues and pull requests replace intents and attempts110 /// The repository, if the viewer may see it. Whether they may is
111 /// decided by the repos service.
112 async fn repo(&self, path: &RepoPath, viewer: &Viewer) -> Result<Outcome<Repo>> {
Work service in Rust, with RFC 3339 timestamps113 g1t_kit::call(
114 &self.repos,
115 "get",
116 &GetArgs {
117 path: path.clone(),
118 viewer: viewer.clone(),
119 },
120 )
121 .await
122 }
123
Issues and pull requests replace intents and attempts124 /// The next number in the repository's sequence. Taking it is one
125 /// statement, so concurrent opens cannot be given the same number.
126 async fn next_number(&self, repo_id: &str) -> Result<u32> {
127 let row = self
128 .db
129 .prepare(
130 "INSERT INTO counters (repo_id, last) VALUES (?, 1)
131 ON CONFLICT (repo_id) DO UPDATE SET last = last + 1
132 RETURNING last AS n",
133 )
134 .bind(&[repo_id.into()])?
135 .first::<NumberRow>(None)
136 .await?;
137 row.map(|row| row.n)
138 .ok_or_else(|| worker::Error::RustError("no number was assigned".into()))
Work service in Rust, with RFC 3339 timestamps139 }
140
Issues and pull requests replace intents and attempts141 async fn issue(&self, repo_id: &str, number: u32) -> Result<Option<Issue>> {
Work service in Rust, with RFC 3339 timestamps142 Ok(self
143 .db
Issues and pull requests replace intents and attempts144 .prepare(format!(
145 "SELECT {ISSUE_COLUMNS} FROM issues WHERE repo_id = ? AND number = ?"
146 ))
147 .bind(&[repo_id.into(), number.into()])?
148 .first::<IssueRow>(None)
Work service in Rust, with RFC 3339 timestamps149 .await?
Issues and pull requests replace intents and attempts150 .map(Issue::from))
Work service in Rust, with RFC 3339 timestamps151 }
152
Issues and pull requests replace intents and attempts153 async fn pull(&self, repo_id: &str, number: u32) -> Result<Option<Pull>> {
Work service in Rust, with RFC 3339 timestamps154 Ok(self
155 .db
Issues and pull requests replace intents and attempts156 .prepare("SELECT * FROM pulls WHERE repo_id = ? AND number = ?")
157 .bind(&[repo_id.into(), number.into()])?
158 .first::<PullRow>(None)
159 .await?
160 .map(Pull::from))
161 }
162
163 async fn comments(&self, repo_id: &str, number: u32) -> Result<Vec<Comment>> {
164 let rows = self
165 .db
166 .prepare(
167 "SELECT * FROM comments WHERE repo_id = ? AND number = ? ORDER BY id LIMIT 500",
168 )
169 .bind(&[repo_id.into(), number.into()])?
170 .all()
Work service in Rust, with RFC 3339 timestamps171 .await?
Issues and pull requests replace intents and attempts172 .results::<CommentRow>()?;
173 Ok(rows.into_iter().map(Comment::from).collect())
Work service in Rust, with RFC 3339 timestamps174 }
175
Issues and pull requests replace intents and attempts176 /// The repository and one of its issues, as seen by `viewer`.
177 async fn issue_at(
178 &self,
179 path: &RepoPath,
180 number: u32,
181 viewer: &Viewer,
182 ) -> Result<Outcome<(Repo, Issue)>> {
183 let Outcome::Ok(repo) = self.repo(path, viewer).await? else {
184 return Ok(no_issue());
Work service in Rust, with RFC 3339 timestamps185 };
Issues and pull requests replace intents and attempts186 Ok(match self.issue(&repo.id, number).await? {
187 Some(issue) => Outcome::Ok((repo, issue)),
188 None => no_issue(),
189 })
190 }
191
192 /// The repository and one of its pull requests, as seen by `viewer`.
193 async fn pull_at(
194 &self,
195 path: &RepoPath,
196 number: u32,
197 viewer: &Viewer,
198 ) -> Result<Outcome<(Repo, Pull)>> {
199 let Outcome::Ok(repo) = self.repo(path, viewer).await? else {
200 return Ok(no_pull());
201 };
202 Ok(match self.pull(&repo.id, number).await? {
203 Some(pull) => Outcome::Ok((repo, pull)),
204 None => no_pull(),
205 })
206 }
207
208 fn issue_event(issue: &Issue) -> IssueEvent {
209 IssueEvent {
210 issue_id: issue.id.clone(),
211 repo_id: issue.repo_id.clone(),
212 number: issue.number,
213 ..IssueEvent::default()
Work service in Rust, with RFC 3339 timestamps214 }
215 }
216
Issues and pull requests replace intents and attempts217 fn pull_event(pull: &Pull) -> PullEvent {
218 PullEvent {
219 pull_id: pull.id.clone(),
220 repo_id: pull.repo_id.clone(),
221 number: pull.number,
222 issue: pull.issue,
223 ..PullEvent::default()
Work service in Rust, with RFC 3339 timestamps224 }
225 }
226
Issues and pull requests replace intents and attempts227 // --- Issues ------------------------------------------------------------
228
229 async fn open_issue(&self, a: OpenIssueArgs) -> Result<Outcome<Issue>> {
Work service in Rust, with RFC 3339 timestamps230 if !a.actor.verified {
231 return Ok(Outcome::fail(FailureCode::Forbidden, UNVERIFIED));
232 }
Issues and pull requests replace intents and attempts233 let title = match valid_title(&a.title) {
234 Ok(title) => title,
235 Err(message) => return Ok(Outcome::fail(FailureCode::Invalid, message)),
236 };
237 let Some(labels) = normalize_labels(&a.labels) else {
Work service in Rust, with RFC 3339 timestamps238 return Ok(Outcome::fail(
239 FailureCode::Invalid,
Issues and pull requests replace intents and attempts240 "An issue can have up to 10 labels of up to 40 characters each.",
Work service in Rust, with RFC 3339 timestamps241 ));
242 };
Issues and pull requests replace intents and attempts243 let repo = check!(self.repo(&a.repo, &Some(a.actor.clone())).await?);
Work service in Rust, with RFC 3339 timestamps244 let checks: Vec<&str> = a
245 .checks
246 .iter()
247 .map(|check| check.trim())
248 .filter(|check| !check.is_empty())
249 .collect();
250
251 let now = now_ms();
Issues and pull requests replace intents and attempts252 let id = new_id("iss", now);
253 let number = self.next_number(&repo.id).await?;
254 let timestamp = rfc3339(now);
Work service in Rust, with RFC 3339 timestamps255 self.db
256 .prepare(
Issues and pull requests replace intents and attempts257 "INSERT INTO issues
258 (id, repo_id, number, title, body, labels, checks, author_id, author_name,
259 created_at, updated_at)
260 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
Work service in Rust, with RFC 3339 timestamps261 )
262 .bind(&[
263 id.as_str().into(),
264 repo.id.as_str().into(),
Issues and pull requests replace intents and attempts265 number.into(),
Work service in Rust, with RFC 3339 timestamps266 title.into(),
Issues and pull requests replace intents and attempts267 a.body.trim().into(),
268 serde_json::to_string(&labels)?.into(),
Work service in Rust, with RFC 3339 timestamps269 serde_json::to_string(&checks)?.into(),
270 a.actor.id.as_str().into(),
271 a.actor.username.as_str().into(),
Issues and pull requests replace intents and attempts272 timestamp.as_str().into(),
273 timestamp.as_str().into(),
Work service in Rust, with RFC 3339 timestamps274 ])?
275 .run()
276 .await?;
Issues and pull requests replace intents and attempts277 let Some(issue) = self.issue(&repo.id, number).await? else {
278 return Ok(no_issue());
Work service in Rust, with RFC 3339 timestamps279 };
Issues and pull requests replace intents and attempts280 self.publish(
281 "issue.opened",
282 &repo.id,
283 &a.actor,
284 IssueEvent {
285 title: Some(issue.title.clone()),
286 ..Self::issue_event(&issue)
Work service in Rust, with RFC 3339 timestamps287 },
Issues and pull requests replace intents and attempts288 )
Work service in Rust, with RFC 3339 timestamps289 .await?;
Issues and pull requests replace intents and attempts290 Ok(Outcome::Ok(issue))
Work service in Rust, with RFC 3339 timestamps291 }
292
Issues and pull requests replace intents and attempts293 async fn list_issues(&self, a: ListIssuesArgs) -> Result<Outcome<Vec<Issue>>> {
294 let repo = check!(self.repo(&a.repo, &a.viewer).await?);
295 let state = state_name(a.state).map_or(JsValue::NULL, JsValue::from);
296 let label = a
297 .label
298 .map(|label| label.trim().to_lowercase())
299 .filter(|label| !label.is_empty());
Work service in Rust, with RFC 3339 timestamps300 let rows = self
301 .db
302 .prepare(format!(
Issues and pull requests replace intents and attempts303 "SELECT {ISSUE_COLUMNS} FROM issues
304 WHERE repo_id = ? AND (? IS NULL OR state = ?)
305 AND (? IS NULL OR EXISTS
306 (SELECT 1 FROM json_each(issues.labels) WHERE json_each.value = ?))
307 ORDER BY number DESC LIMIT ?"
Work service in Rust, with RFC 3339 timestamps308 ))
Issues and pull requests replace intents and attempts309 .bind(&[
310 repo.id.into(),
311 state.clone(),
312 state,
313 optional(&label),
314 optional(&label),
315 LIST_PAGE.into(),
316 ])?
Work service in Rust, with RFC 3339 timestamps317 .all()
318 .await?
Issues and pull requests replace intents and attempts319 .results::<IssueRow>()?;
320 Ok(Outcome::Ok(rows.into_iter().map(Issue::from).collect()))
Work service in Rust, with RFC 3339 timestamps321 }
322
Issues and pull requests replace intents and attempts323 async fn get_issue(&self, a: ViewArgs) -> Result<Outcome<IssueDetail>> {
324 let (repo, issue) = check!(self.issue_at(&a.repo, a.number, &a.viewer).await?);
325 let pulls = self
Work service in Rust, with RFC 3339 timestamps326 .db
Issues and pull requests replace intents and attempts327 .prepare("SELECT * FROM pulls WHERE issue_id = ? ORDER BY number")
328 .bind(&[issue.id.as_str().into()])?
Work service in Rust, with RFC 3339 timestamps329 .all()
330 .await?
Issues and pull requests replace intents and attempts331 .results::<PullRow>()?;
332 Ok(Outcome::Ok(IssueDetail {
333 comments: self.comments(&repo.id, issue.number).await?,
334 pulls: pulls.into_iter().map(Pull::from).collect(),
335 issue,
Work service in Rust, with RFC 3339 timestamps336 }))
337 }
338
Issues and pull requests replace intents and attempts339 /// The issue, if `actor` wrote it or belongs to the repository's workspace.
340 async fn manageable_issue(
341 &self,
342 actor: &User,
343 path: &RepoPath,
344 number: u32,
345 ) -> Result<Outcome<Issue>> {
346 let (repo, issue) = check!(self.issue_at(path, number, &Some(actor.clone())).await?);
347 if issue.author.id != actor.id && !actor.is_member(&repo.namespace) {
Work service in Rust, with RFC 3339 timestamps348 return Ok(Outcome::fail(
349 FailureCode::Forbidden,
Issues and pull requests replace intents and attempts350 "Only the author or a member of the workspace can change an issue.",
Work service in Rust, with RFC 3339 timestamps351 ));
352 }
Issues and pull requests replace intents and attempts353 Ok(Outcome::Ok(issue))
354 }
355
356 async fn update_issue(&self, a: UpdateIssueArgs) -> Result<Outcome<Issue>> {
357 let issue = check!(self.manageable_issue(&a.actor, &a.repo, a.number).await?);
358 let title = match a.title.as_deref().map(valid_title) {
359 Some(Err(message)) => return Ok(Outcome::fail(FailureCode::Invalid, message)),
360 Some(Ok(title)) => Some(title.to_owned()),
361 None => None,
362 };
363 let labels = match a.labels.as_deref().map(normalize_labels) {
364 Some(None) => {
365 return Ok(Outcome::fail(
366 FailureCode::Invalid,
367 "An issue can have up to 10 labels of up to 40 characters each.",
368 ));
369 }
370 Some(Some(labels)) => Some(serde_json::to_string(&labels)?),
371 None => None,
372 };
373 let body = a.body.map(|body| body.trim().to_owned());
374 self.db
375 .prepare(
376 "UPDATE issues
377 SET title = COALESCE(?, title), body = COALESCE(?, body),
378 labels = COALESCE(?, labels), updated_at = ?
379 WHERE id = ?",
380 )
381 .bind(&[
382 optional(&title),
383 optional(&body),
384 optional(&labels),
385 rfc3339(now_ms()).into(),
386 issue.id.as_str().into(),
387 ])?
388 .run()
389 .await?;
390 let Some(issue) = self.issue(&issue.repo_id, issue.number).await? else {
391 return Ok(no_issue());
392 };
393 self.publish(
394 "issue.updated",
395 &issue.repo_id,
396 &a.actor,
397 Self::issue_event(&issue),
398 )
399 .await?;
400 Ok(Outcome::Ok(issue))
401 }
402
403 async fn close_issue(&self, a: IssueActionArgs) -> Result<Outcome<Issue>> {
404 let mut issue = check!(self.manageable_issue(&a.actor, &a.repo, a.number).await?);
405 if issue.state == State::Closed {
Work service in Rust, with RFC 3339 timestamps406 return Ok(Outcome::fail(
407 FailureCode::Conflict,
Issues and pull requests replace intents and attempts408 "This issue is already closed.",
Work service in Rust, with RFC 3339 timestamps409 ));
410 }
Issues and pull requests replace intents and attempts411 let reason = a.reason.unwrap_or(IssueReason::Completed);
412 let now = rfc3339(now_ms());
Work service in Rust, with RFC 3339 timestamps413 self.db
Issues and pull requests replace intents and attempts414 .prepare(
415 "UPDATE issues SET state = 'closed', reason = ?, closed_at = ?, updated_at = ?
416 WHERE id = ?",
417 )
418 .bind(&[
419 reason.as_str().into(),
420 now.as_str().into(),
421 now.as_str().into(),
422 issue.id.as_str().into(),
423 ])?
Work service in Rust, with RFC 3339 timestamps424 .run()
425 .await?;
Issues and pull requests replace intents and attempts426 self.publish(
427 "issue.closed",
428 &issue.repo_id,
429 &a.actor,
430 IssueEvent {
431 reason: Some(reason.as_str()),
432 ..Self::issue_event(&issue)
Work service in Rust, with RFC 3339 timestamps433 },
Issues and pull requests replace intents and attempts434 )
Work service in Rust, with RFC 3339 timestamps435 .await?;
Issues and pull requests replace intents and attempts436 issue.state = State::Closed;
437 issue.reason = Some(reason);
438 issue.closed_at = Some(now.clone());
439 issue.updated_at = now;
440 Ok(Outcome::Ok(issue))
441 }
442
443 async fn reopen_issue(&self, a: IssueActionArgs) -> Result<Outcome<Issue>> {
444 let mut issue = check!(self.manageable_issue(&a.actor, &a.repo, a.number).await?);
445 if issue.state == State::Open {
446 return Ok(Outcome::fail(
447 FailureCode::Conflict,
448 "This issue is already open.",
449 ));
450 }
451 let now = rfc3339(now_ms());
452 self.db
453 .prepare(
454 "UPDATE issues
455 SET state = 'open', reason = NULL, resolved_by = NULL, closed_at = NULL,
456 updated_at = ?
457 WHERE id = ?",
458 )
459 .bind(&[now.as_str().into(), issue.id.as_str().into()])?
460 .run()
461 .await?;
462 self.publish(
463 "issue.reopened",
464 &issue.repo_id,
465 &a.actor,
466 Self::issue_event(&issue),
467 )
468 .await?;
469 issue.state = State::Open;
470 issue.reason = None;
471 issue.resolved_by = None;
472 issue.closed_at = None;
473 issue.updated_at = now;
474 Ok(Outcome::Ok(issue))
Work service in Rust, with RFC 3339 timestamps475 }
476
Issues and pull requests replace intents and attempts477 /// The default labels, then every other label in use on the repository.
478 async fn list_labels(&self, a: ViewArgs) -> Result<Outcome<Vec<String>>> {
479 let repo = check!(self.repo(&a.repo, &a.viewer).await?);
480 let used = self
481 .db
482 .prepare(
483 "SELECT DISTINCT json_each.value AS value
484 FROM issues, json_each(issues.labels)
485 WHERE issues.repo_id = ? ORDER BY 1 LIMIT 200",
486 )
487 .bind(&[repo.id.into()])?
488 .all()
489 .await?
490 .results::<ValueRow>()?;
491 let mut labels: Vec<String> = DEFAULT_LABELS.iter().map(|label| (*label).into()).collect();
492 for row in used {
493 if !labels.contains(&row.value) {
494 labels.push(row.value);
495 }
496 }
497 Ok(Outcome::Ok(labels))
498 }
499
500 async fn counts(&self, a: ViewArgs) -> Result<Outcome<Counts>> {
501 let repo = check!(self.repo(&a.repo, &a.viewer).await?);
502 let counts = self
503 .db
504 .prepare(
505 "SELECT
506 (SELECT count(*) FROM issues WHERE repo_id = ? AND state = 'open') AS issues,
507 (SELECT count(*) FROM pulls
508 WHERE repo_id = ? AND status IN ('draft', 'open')) AS pulls",
509 )
510 .bind(&[repo.id.as_str().into(), repo.id.as_str().into()])?
511 .first::<Counts>(None)
512 .await?;
513 Ok(Outcome::Ok(counts.unwrap_or(Counts {
514 issues: 0,
515 pulls: 0,
516 })))
517 }
518
519 // --- Comments ----------------------------------------------------------
520
521 async fn add_comment(&self, a: AddCommentArgs) -> Result<Outcome<Comment>> {
Work service in Rust, with RFC 3339 timestamps522 if !a.actor.verified {
523 return Ok(Outcome::fail(FailureCode::Forbidden, UNVERIFIED));
524 }
Issues and pull requests replace intents and attempts525 let body = a.body.trim();
526 if body.is_empty() {
527 return Ok(Outcome::fail(
528 FailureCode::Invalid,
529 "A comment cannot be empty.",
530 ));
531 }
532 if body.chars().count() > MAX_ENTRY_CHARS {
533 return Ok(Outcome::fail(
534 FailureCode::Invalid,
535 "That comment is too long.",
536 ));
537 }
538 let repo = check!(self.repo(&a.repo, &Some(a.actor.clone())).await?);
539 // The number names an issue or a pull request, never both.
540 let table = if self.issue(&repo.id, a.number).await?.is_some() {
541 "issues"
542 } else if self.pull(&repo.id, a.number).await?.is_some() {
543 "pulls"
544 } else {
Work service in Rust, with RFC 3339 timestamps545 return Ok(Outcome::fail(
Issues and pull requests replace intents and attempts546 FailureCode::NotFound,
547 "No issue or pull request has that number.",
Work service in Rust, with RFC 3339 timestamps548 ));
Issues and pull requests replace intents and attempts549 };
550
551 let now = now_ms();
552 let comment = Comment {
553 id: new_id("cmt", now),
554 author: a.actor.clone(),
555 body: body.to_owned(),
556 created_at: rfc3339(now),
557 };
558 self.db
559 .batch(vec![
560 self.db
561 .prepare(
562 "INSERT INTO comments
563 (id, repo_id, number, author_id, author_name, body, created_at)
564 VALUES (?, ?, ?, ?, ?, ?, ?)",
565 )
566 .bind(&[
567 comment.id.as_str().into(),
568 repo.id.as_str().into(),
569 a.number.into(),
570 a.actor.id.as_str().into(),
571 a.actor.username.as_str().into(),
572 body.into(),
573 comment.created_at.as_str().into(),
574 ])?,
575 self.db
576 .prepare(format!(
577 "UPDATE {table} SET updated_at = ? WHERE repo_id = ? AND number = ?"
578 ))
579 .bind(&[
580 comment.created_at.as_str().into(),
581 repo.id.as_str().into(),
582 a.number.into(),
583 ])?,
584 ])
585 .await?;
586 self.publish(
587 "comment.created",
588 &repo.id,
589 &a.actor,
590 CommentCreated {
591 comment_id: comment.id.clone(),
592 repo_id: repo.id.clone(),
593 number: a.number,
594 },
595 )
596 .await?;
597 Ok(Outcome::Ok(comment))
598 }
599
600 // --- Pull requests -----------------------------------------------------
601
602 async fn open_pull(&self, a: OpenPullArgs) -> Result<Outcome<Pull>> {
603 if !a.actor.verified {
604 return Ok(Outcome::fail(FailureCode::Forbidden, UNVERIFIED));
Work service in Rust, with RFC 3339 timestamps605 }
Issues and pull requests replace intents and attempts606 let repo = check!(self.repo(&a.repo, &Some(a.actor.clone())).await?);
607 let issue = match a.issue {
608 Some(number) => match self.issue(&repo.id, number).await? {
609 Some(issue) if issue.state == State::Open => Some(issue),
610 Some(_) => {
611 return Ok(Outcome::fail(
612 FailureCode::Conflict,
613 "This issue is closed.",
614 ));
615 }
616 None => return Ok(no_issue()),
617 },
618 None => None,
619 };
620 // A pull request for an issue takes the issue's title unless given one.
621 let title = match (a.title.trim(), &issue) {
622 ("", Some(issue)) => issue.title.clone(),
623 (title, _) => match valid_title(title) {
624 Ok(title) => title.to_owned(),
625 Err(message) => return Ok(Outcome::fail(FailureCode::Invalid, message)),
626 },
627 };
Work service in Rust, with RFC 3339 timestamps628 let agent = match a.agent.trim() {
629 "" => "agent",
630 agent => agent,
631 };
632 let runtime = match a.runtime {
Issues and pull requests replace intents and attempts633 Runtime::Hosted => "hosted",
634 Runtime::External => "external",
Work service in Rust, with RFC 3339 timestamps635 };
636
637 let now = now_ms();
Issues and pull requests replace intents and attempts638 let id = new_id("pr", now);
Work service in Rust, with RFC 3339 timestamps639 let fork: Outcome<Repo> = g1t_kit::call(
640 &self.repos,
Issues and pull requests replace intents and attempts641 "fork_for_pull",
Work service in Rust, with RFC 3339 timestamps642 &ForkArgs {
Issues and pull requests replace intents and attempts643 source_id: repo.id.clone(),
644 pull_id: id.clone(),
Work service in Rust, with RFC 3339 timestamps645 actor: a.actor.clone(),
646 },
647 )
648 .await?;
Issues and pull requests replace intents and attempts649 let fork = check!(fork);
Work service in Rust, with RFC 3339 timestamps650
Issues and pull requests replace intents and attempts651 let number = self.next_number(&repo.id).await?;
Work service in Rust, with RFC 3339 timestamps652 let timestamp = rfc3339(now);
653 self.db
654 .prepare(
Issues and pull requests replace intents and attempts655 "INSERT INTO pulls
656 (id, repo_id, number, issue_id, issue_number, title, agent, runtime,
657 fork_repo_id, fork_namespace, fork_name, author_id, author_name,
Work service in Rust, with RFC 3339 timestamps658 created_at, updated_at)
Issues and pull requests replace intents and attempts659 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
Work service in Rust, with RFC 3339 timestamps660 )
661 .bind(&[
662 id.as_str().into(),
Issues and pull requests replace intents and attempts663 repo.id.as_str().into(),
664 number.into(),
665 optional(&issue.as_ref().map(|issue| issue.id.clone())),
666 optional_number(issue.as_ref().map(|issue| issue.number)),
667 title.into(),
Work service in Rust, with RFC 3339 timestamps668 agent.into(),
669 runtime.into(),
670 fork.id.into(),
671 fork.namespace.into(),
672 fork.name.into(),
673 a.actor.id.as_str().into(),
674 a.actor.username.as_str().into(),
675 timestamp.as_str().into(),
676 timestamp.as_str().into(),
677 ])?
678 .run()
679 .await?;
Issues and pull requests replace intents and attempts680 let Some(pull) = self.pull(&repo.id, number).await? else {
681 return Ok(no_pull());
Work service in Rust, with RFC 3339 timestamps682 };
Issues and pull requests replace intents and attempts683 self.publish(
684 "pull.opened",
685 &repo.id,
686 &a.actor,
687 PullEvent {
688 agent: Some(pull.agent.clone()),
689 ..Self::pull_event(&pull)
Work service in Rust, with RFC 3339 timestamps690 },
Issues and pull requests replace intents and attempts691 )
Work service in Rust, with RFC 3339 timestamps692 .await?;
Issues and pull requests replace intents and attempts693 Ok(Outcome::Ok(pull))
Work service in Rust, with RFC 3339 timestamps694 }
695
Issues and pull requests replace intents and attempts696 async fn list_pulls(&self, a: ListPullsArgs) -> Result<Outcome<Vec<Pull>>> {
697 let repo = check!(self.repo(&a.repo, &a.viewer).await?);
698 let filter = match a.state {
699 Some(State::Open) => "AND status IN ('draft', 'open')",
700 Some(State::Closed) => "AND status IN ('merged', 'closed')",
701 None => "",
Work service in Rust, with RFC 3339 timestamps702 };
Issues and pull requests replace intents and attempts703 let rows = self
704 .db
705 .prepare(format!(
706 "SELECT * FROM pulls WHERE repo_id = ? {filter} ORDER BY number DESC LIMIT ?"
707 ))
708 .bind(&[repo.id.into(), LIST_PAGE.into()])?
709 .all()
710 .await?
711 .results::<PullRow>()?;
712 Ok(Outcome::Ok(rows.into_iter().map(Pull::from).collect()))
713 }
714
715 async fn get_pull(&self, a: ViewArgs) -> Result<Outcome<PullDetail>> {
716 let (repo, pull) = check!(self.pull_at(&a.repo, a.number, &a.viewer).await?);
717 let issue = match pull.issue {
718 Some(number) => self.issue(&repo.id, number).await?,
719 None => None,
Work service in Rust, with RFC 3339 timestamps720 };
Issues and pull requests replace intents and attempts721 Ok(Outcome::Ok(PullDetail {
722 comments: self.comments(&repo.id, pull.number).await?,
723 issue,
724 pull,
725 }))
Work service in Rust, with RFC 3339 timestamps726 }
727
Issues and pull requests replace intents and attempts728 /// The pull request, if it is still active and `actor` opened it or
729 /// belongs to the repository's workspace.
730 async fn manageable_pull(
Work service in Rust, with RFC 3339 timestamps731 &self,
Issues and pull requests replace intents and attempts732 actor: &User,
733 path: &RepoPath,
734 number: u32,
735 ) -> Result<Outcome<Pull>> {
736 let (repo, pull) = check!(self.pull_at(path, number, &Some(actor.clone())).await?);
737 if pull.author.id != actor.id && !actor.is_member(&repo.namespace) {
Work service in Rust, with RFC 3339 timestamps738 return Ok(Outcome::fail(
Issues and pull requests replace intents and attempts739 FailureCode::Forbidden,
740 "Only whoever opened a pull request, or a member of the workspace, can change it.",
741 ));
742 }
743 if !pull.status.is_active() {
744 return Ok(Outcome::fail(
Work service in Rust, with RFC 3339 timestamps745 FailureCode::Conflict,
Issues and pull requests replace intents and attempts746 format!("This pull request is already {}.", pull.status.as_str()),
Work service in Rust, with RFC 3339 timestamps747 ));
748 }
Issues and pull requests replace intents and attempts749 Ok(Outcome::Ok(pull))
750 }
751
752 /// Marks a draft ready for review, or updates the description of one
753 /// that already is.
754 async fn ready_pull(&self, a: PullActionArgs) -> Result<Outcome<Pull>> {
755 let mut pull = check!(self.manageable_pull(&a.actor, &a.repo, a.number).await?);
Work service in Rust, with RFC 3339 timestamps756 let summary = Some(a.summary.trim().to_owned()).filter(|summary| !summary.is_empty());
757 let now = rfc3339(now_ms());
758 self.db
759 .prepare(
Issues and pull requests replace intents and attempts760 "UPDATE pulls SET status = 'open', body = COALESCE(?, body), updated_at = ?
Work service in Rust, with RFC 3339 timestamps761 WHERE id = ?",
762 )
763 .bind(&[
764 optional(&summary),
765 now.as_str().into(),
Issues and pull requests replace intents and attempts766 pull.id.as_str().into(),
Work service in Rust, with RFC 3339 timestamps767 ])?
768 .run()
769 .await?;
Issues and pull requests replace intents and attempts770 if pull.status == PullStatus::Draft {
771 self.publish(
772 "pull.ready",
773 &pull.repo_id,
774 &a.actor,
775 Self::pull_event(&pull),
776 )
777 .await?;
778 }
779 pull.status = PullStatus::Open;
780 pull.body = summary.or(pull.body);
781 pull.updated_at = now;
782 Ok(Outcome::Ok(pull))
783 }
784
785 async fn close_pull(&self, a: PullActionArgs) -> Result<Outcome<Pull>> {
786 let mut pull = check!(self.manageable_pull(&a.actor, &a.repo, a.number).await?);
787 let now = rfc3339(now_ms());
788 self.db
789 .prepare("UPDATE pulls SET status = 'closed', updated_at = ? WHERE id = ?")
790 .bind(&[now.as_str().into(), pull.id.as_str().into()])?
791 .run()
792 .await?;
793 self.publish(
794 "pull.closed",
795 &pull.repo_id,
796 &a.actor,
797 Self::pull_event(&pull),
798 )
Work service in Rust, with RFC 3339 timestamps799 .await?;
Issues and pull requests replace intents and attempts800 pull.status = PullStatus::Closed;
801 pull.updated_at = now;
802 Ok(Outcome::Ok(pull))
Work service in Rust, with RFC 3339 timestamps803 }
804
Issues and pull requests replace intents and attempts805 /// Lands the pull request on the repository's default branch. Unless
806 /// told to keep it open, that resolves the issue it was for: the issue
807 /// closes naming this pull request, and the others still in progress
808 /// for it close as superseded.
809 async fn merge_pull(&self, a: PullActionArgs) -> Result<Outcome<Pull>> {
Work service in Rust, with RFC 3339 timestamps810 let viewer = Some(a.actor.clone());
Issues and pull requests replace intents and attempts811 let (repo, mut pull) = check!(self.pull_at(&a.repo, a.number, &viewer).await?);
812 match pull.status {
813 PullStatus::Open => {}
814 PullStatus::Draft => {
815 return Ok(Outcome::fail(
816 FailureCode::Conflict,
817 "This pull request is still a draft. Mark it ready for review first.",
818 ));
819 }
820 status => {
821 return Ok(Outcome::fail(
822 FailureCode::Conflict,
823 format!("This pull request is already {}.", status.as_str()),
824 ));
825 }
Work service in Rust, with RFC 3339 timestamps826 }
Issues and pull requests replace intents and attempts827 let issue = match pull.issue {
828 Some(number) if !a.keep_issue_open => self
829 .issue(&repo.id, number)
830 .await?
831 .filter(|issue| issue.state == State::Open),
832 _ => None,
Work service in Rust, with RFC 3339 timestamps833 };
834
Issues and pull requests replace intents and attempts835 // Whether the actor may write to the repository is decided by repos.
Work service in Rust, with RFC 3339 timestamps836 let landed: Outcome<Landed> = g1t_kit::call(
837 &self.repos,
838 "land",
839 &LandArgs {
Issues and pull requests replace intents and attempts840 fork_id: pull.fork_repo_id.clone(),
Work service in Rust, with RFC 3339 timestamps841 actor: a.actor.clone(),
842 },
843 )
844 .await?;
Issues and pull requests replace intents and attempts845 let landed = check!(landed);
Work service in Rust, with RFC 3339 timestamps846
847 let now = rfc3339(now_ms());
Issues and pull requests replace intents and attempts848 let mut statements = vec![
849 self.db
850 .prepare(
851 "UPDATE pulls
852 SET status = 'merged', head_commit = ?, merge_base = ?, merged_by = ?,
853 merged_at = ?, updated_at = ?
854 WHERE id = ?",
855 )
856 .bind(&[
857 landed.commit.as_str().into(),
858 optional(&landed.previous),
859 a.actor.username.as_str().into(),
860 now.as_str().into(),
861 now.as_str().into(),
862 pull.id.as_str().into(),
863 ])?,
864 ];
865 if let Some(issue) = &issue {
866 statements.push(
Work service in Rust, with RFC 3339 timestamps867 self.db
868 .prepare(
Issues and pull requests replace intents and attempts869 "UPDATE issues
870 SET state = 'closed', reason = 'completed', resolved_by = ?,
871 closed_at = ?, updated_at = ?
Work service in Rust, with RFC 3339 timestamps872 WHERE id = ?",
873 )
874 .bind(&[
Issues and pull requests replace intents and attempts875 pull.number.into(),
876 now.as_str().into(),
Work service in Rust, with RFC 3339 timestamps877 now.as_str().into(),
Issues and pull requests replace intents and attempts878 issue.id.as_str().into(),
Work service in Rust, with RFC 3339 timestamps879 ])?,
Issues and pull requests replace intents and attempts880 );
881 statements.push(
Work service in Rust, with RFC 3339 timestamps882 self.db
Issues and pull requests replace intents and attempts883 .prepare(
884 "UPDATE pulls SET status = 'closed', superseded_by = ?, updated_at = ?
885 WHERE issue_id = ? AND id != ? AND status IN ('draft', 'open')",
886 )
887 .bind(&[
888 pull.number.into(),
889 now.as_str().into(),
890 issue.id.as_str().into(),
891 pull.id.as_str().into(),
892 ])?,
893 );
894 }
895 self.db.batch(statements).await?;
896
897 self.publish(
898 "pull.merged",
899 &repo.id,
900 &a.actor,
901 PullEvent {
Work service in Rust, with RFC 3339 timestamps902 commit: Some(landed.commit.clone()),
Issues and pull requests replace intents and attempts903 ..Self::pull_event(&pull)
Work service in Rust, with RFC 3339 timestamps904 },
Issues and pull requests replace intents and attempts905 )
Work service in Rust, with RFC 3339 timestamps906 .await?;
Issues and pull requests replace intents and attempts907 if let Some(issue) = &issue {
908 self.publish(
909 "issue.closed",
910 &repo.id,
911 &a.actor,
912 IssueEvent {
913 reason: Some(IssueReason::Completed.as_str()),
914 resolved_by: Some(pull.number),
915 ..Self::issue_event(issue)
916 },
917 )
918 .await?;
919 }
Work service in Rust, with RFC 3339 timestamps920
Issues and pull requests replace intents and attempts921 pull.status = PullStatus::Merged;
922 pull.head_commit = Some(landed.commit);
923 pull.merge_base = landed.previous;
924 pull.merged_by = Some(a.actor.username);
925 pull.merged_at = Some(now.clone());
926 pull.updated_at = now;
927 Ok(Outcome::Ok(pull))
Work service in Rust, with RFC 3339 timestamps928 }
929
Issues and pull requests replace intents and attempts930 async fn list_active_pulls(&self, a: ViewerArgs) -> Result<Vec<ActivePull>> {
Work service in Rust, with RFC 3339 timestamps931 let Some(viewer) = a.viewer else {
932 return Ok(Vec::new());
933 };
934 let rows = self
935 .db
936 .prepare(
Issues and pull requests replace intents and attempts937 "SELECT * FROM pulls
938 WHERE author_id = ? AND status IN ('draft', 'open')
Work service in Rust, with RFC 3339 timestamps939 ORDER BY updated_at DESC LIMIT 50",
940 )
941 .bind(&[viewer.id.into()])?
942 .all()
943 .await?
Issues and pull requests replace intents and attempts944 .results::<PullRow>()?;
Work service in Rust, with RFC 3339 timestamps945 let mut active = Vec::with_capacity(rows.len());
Issues and pull requests replace intents and attempts946 for pull in rows.into_iter().map(Pull::from) {
947 let issue = match pull.issue {
948 Some(number) => self.issue(&pull.repo_id, number).await?,
949 None => None,
950 };
951 active.push(ActivePull { pull, issue });
Work service in Rust, with RFC 3339 timestamps952 }
953 Ok(active)
954 }
955
Issues and pull requests replace intents and attempts956 // --- Sessions ----------------------------------------------------------
957
Work service in Rust, with RFC 3339 timestamps958 async fn append_session(&self, a: AppendSessionArgs) -> Result<Outcome<Appended>> {
959 if a.entries.is_empty() {
960 return Ok(Outcome::Ok(Appended { count: 0 }));
961 }
962 if a.entries.len() > MAX_ENTRY_BATCH {
963 return Ok(Outcome::fail(
964 FailureCode::Invalid,
965 format!("Send at most {MAX_ENTRY_BATCH} entries at a time."),
966 ));
967 }
Issues and pull requests replace intents and attempts968 let viewer = Some(a.actor.clone());
969 let (_, pull) = check!(self.pull_at(&a.repo, a.number, &viewer).await?);
970 if pull.author.id != a.actor.id {
971 return Ok(Outcome::fail(
972 FailureCode::Forbidden,
973 "Only whoever opened a pull request can record its session.",
974 ));
975 }
Work service in Rust, with RFC 3339 timestamps976
977 let now = rfc3339(now_ms());
978 let count = a.entries.len() as u32;
979 let mut statements = Vec::with_capacity(a.entries.len() + 1);
980 for entry in a.entries {
981 let kind = serde_json::to_value(entry.kind)?;
982 let text: String = entry.text.chars().take(MAX_ENTRY_CHARS).collect();
983 // Each insert takes the next sequence number itself, so two
984 // writers appending at once cannot collide.
985 statements.push(
986 self.db
987 .prepare(
Issues and pull requests replace intents and attempts988 "INSERT INTO session_entries (pull_id, seq, kind, text, tool, \"commit\", at)
Work service in Rust, with RFC 3339 timestamps989 SELECT ?, COALESCE(MAX(seq), 0) + 1, ?, ?, ?, ?, ?
Issues and pull requests replace intents and attempts990 FROM session_entries WHERE pull_id = ?",
Work service in Rust, with RFC 3339 timestamps991 )
992 .bind(&[
Issues and pull requests replace intents and attempts993 pull.id.as_str().into(),
Work service in Rust, with RFC 3339 timestamps994 kind.as_str().unwrap_or("note").into(),
995 text.into(),
996 optional(&entry.tool),
Issues and pull requests replace intents and attempts997 optional(&entry.commit.or_else(|| pull.head_commit.clone())),
Work service in Rust, with RFC 3339 timestamps998 now.as_str().into(),
Issues and pull requests replace intents and attempts999 pull.id.as_str().into(),
Work service in Rust, with RFC 3339 timestamps1000 ])?,
1001 );
1002 }
1003 statements.push(
1004 self.db
Issues and pull requests replace intents and attempts1005 .prepare("UPDATE pulls SET updated_at = ? WHERE id = ?")
1006 .bind(&[now.as_str().into(), pull.id.as_str().into()])?,
Work service in Rust, with RFC 3339 timestamps1007 );
1008 self.db.batch(statements).await?;
Issues and pull requests replace intents and attempts1009 self.publish(
1010 "session.appended",
1011 &pull.repo_id,
1012 &a.actor,
1013 SessionAppended {
1014 pull_id: pull.id.clone(),
1015 repo_id: pull.repo_id.clone(),
1016 number: pull.number,
Work service in Rust, with RFC 3339 timestamps1017 count,
1018 },
Issues and pull requests replace intents and attempts1019 )
Work service in Rust, with RFC 3339 timestamps1020 .await?;
1021 Ok(Outcome::Ok(Appended { count }))
1022 }
1023
Issues and pull requests replace intents and attempts1024 async fn read_session(&self, a: ViewArgs) -> Result<Outcome<Vec<SessionEntry>>> {
1025 let (_, pull) = check!(self.pull_at(&a.repo, a.number, &a.viewer).await?);
Work service in Rust, with RFC 3339 timestamps1026 let rows = self
1027 .db
1028 .prepare(
1029 "SELECT seq, kind, text, tool, \"commit\", at FROM session_entries
Issues and pull requests replace intents and attempts1030 WHERE pull_id = ? AND seq > ? ORDER BY seq LIMIT ?",
Work service in Rust, with RFC 3339 timestamps1031 )
Issues and pull requests replace intents and attempts1032 .bind(&[pull.id.into(), a.after_seq.into(), SESSION_PAGE.into()])?
Work service in Rust, with RFC 3339 timestamps1033 .all()
1034 .await?
1035 .results::<SessionRow>()?;
1036 Ok(Outcome::Ok(
1037 rows.into_iter().map(SessionEntry::from).collect(),
1038 ))
1039 }
1040
Issues and pull requests replace intents and attempts1041 /// A push to a pull request's fork moves that pull request's head.
Work service in Rust, with RFC 3339 timestamps1042 async fn on_event(&self, event: &Delivered) -> Result<()> {
1043 if event.kind != "git.push" {
1044 return Ok(());
1045 }
1046 let (Some(repo_id), Some(after)) = (event.repo_id.as_deref(), event.data["after"].as_str())
1047 else {
1048 return Ok(());
1049 };
1050 self.db
Issues and pull requests replace intents and attempts1051 .prepare(
1052 "UPDATE pulls SET head_commit = ?, updated_at = ?
1053 WHERE fork_repo_id = ? AND status IN ('draft', 'open')",
1054 )
Work service in Rust, with RFC 3339 timestamps1055 .bind(&[after.into(), rfc3339(now_ms()).into(), repo_id.into()])?
1056 .run()
1057 .await?;
1058 Ok(())
1059 }
1060}
1061
1062fn service(env: &Env) -> Result<Work> {
1063 Ok(Work {
1064 db: env.d1("DB")?,
1065 repos: env.service("REPOS")?,
1066 events: js::binding(env, "EVENTS")?,
1067 })
1068}
1069
1070#[event(fetch)]
1071async fn fetch(mut request: Request, env: Env, _ctx: Context) -> Result<Response> {
1072 let Some(method) = rpc_method(&request) else {
1073 return Response::error("Not found", 404);
1074 };
1075 let body: serde_json::Value = request.json().await?;
1076 let work = service(&env)?;
1077
1078 match method.as_str() {
Issues and pull requests replace intents and attempts1079 "open_issue" => reply(&work.open_issue(args(body)?).await?),
1080 "list_issues" => reply(&work.list_issues(args(body)?).await?),
1081 "get_issue" => reply(&work.get_issue(args(body)?).await?),
1082 "update_issue" => reply(&work.update_issue(args(body)?).await?),
1083 "close_issue" => reply(&work.close_issue(args(body)?).await?),
1084 "reopen_issue" => reply(&work.reopen_issue(args(body)?).await?),
1085 "list_labels" => reply(&work.list_labels(args(body)?).await?),
1086 "counts" => reply(&work.counts(args(body)?).await?),
1087 "add_comment" => reply(&work.add_comment(args(body)?).await?),
1088 "open_pull" => reply(&work.open_pull(args(body)?).await?),
1089 "list_pulls" => reply(&work.list_pulls(args(body)?).await?),
1090 "get_pull" => reply(&work.get_pull(args(body)?).await?),
1091 "ready_pull" => reply(&work.ready_pull(args(body)?).await?),
1092 "close_pull" => reply(&work.close_pull(args(body)?).await?),
1093 "merge_pull" => reply(&work.merge_pull(args(body)?).await?),
1094 "list_active_pulls" => reply(&work.list_active_pulls(args(body)?).await?),
Work service in Rust, with RFC 3339 timestamps1095 "append_session" => reply(&work.append_session(args(body)?).await?),
1096 "read_session" => reply(&work.read_session(args(body)?).await?),
1097 _ => Response::error("Unknown method", 404),
1098 }
1099}
1100
1101/// Events from the bus, delivered on this service's own queue.
1102#[event(queue)]
1103async fn queue(batch: MessageBatch<Delivered>, env: Env, _ctx: Context) -> Result<()> {
1104 let work = service(&env)?;
1105 for message in batch.messages()? {
1106 work.on_event(message.body()).await?;
1107 message.ack();
1108 }
1109 Ok(())
1110}