services/events/src/inbox.rs · flagon-io/g1t

//! | `secret_scanning_alert.summary`, `code_scanning_alert.summary`, `vulnerability_alert.summary` | the workspace's owners and security managers, whoever pushed a refused push, and watchers who chose security and may see findings; one thread per repository and alert type | security_alert | error for critical or high (told at once), info for the rest (a digest, unless the person chose at once) |

26

27

28

//!

29

//! Watchers of a repository at `all` (or `custom`, for the kinds they

30

//! chose) hear of every issue and pull request opened, commented on,

31

//! closed, reopened or merged, and of deployments. Anyone who ignores the

32

//! thread or the repository hears of nothing on it; anyone who

33

//! unsubscribed hears only of what is asked of them.

34

//!

35

//! Nobody is told of what they did themselves, though an outcome they set

36

//! off (checks, a workflow, a deployment, g1t's work) is theirs to hear of.

37

//! g1t is never told. A failure here is logged and the batch goes on: the

38

//! bus never waits on the inbox, so an item can be missed, but nothing else

39

//! is held up.

40

41

use std::collections::{HashMap, HashSet};

42

43

use g1t_contracts::credentials::Principal;

44

use g1t_contracts::events::{Event, WorkspaceRenamed};

45

use g1t_contracts::identity::{AGENT_ID, UsernamesArgs};

46

use g1t_contracts::inbox::*;

47

use g1t_contracts::repos::{PathByIdArgs, ReadableArgs, Repo, RepoPath};

48

use g1t_contracts::security::SeverityCounts;

49

use g1t_contracts::security_suite::{AlertType, SecuritySummary};

50

use g1t_contracts::time::rfc3339;

51

use g1t_contracts::{new_id, system};

52

use g1t_kit::now_ms;

53

use serde::Deserialize;

54

use worker::wasm_bindgen::JsValue;

55

use worker::{D1Database, D1PreparedStatement, Fetcher, Result};

56

57

use crate::subscriptions::{self, Audience};

58

59

/// Items marked done are kept this long, then removed.

60

pub const DONE_DAYS: u32 = 30;

61

/// No item is kept longer than this, unless it was saved.

62

pub const MAX_DAYS: u32 = 180;

63

/// Unread warnings shown ahead of everything else on the first page.

64

const MAX_RANKED: u32 = 20;

65

const MAX_TITLE: usize = 200;

66

const MAX_BODY: usize = 300;

67

68

/// What the inbox asks about an event before deciding who is told.

69

#[derive(Debug, PartialEq, Eq)]

70

pub struct Wanted {

71

pub repo_id: String,

72

/// The issue or pull request, when the event names one.

73

pub number: Option<u32>,

74

pub comment_id: Option<String>,

75

}

76

77

/// One person to tell, and what.

78

#[derive(Debug, PartialEq, Eq)]

79

pub struct Notice {

80

pub username: String,

81

pub reason: Reason,

82

pub severity: Severity,

pub title: String,

pub body: String,

}

/// Who did it, as far as is known.

88

#[derive(Debug, Default)]

89

pub struct Actor {

90

pub id: Option<String>,

91

pub username: Option<String>,

}

impl Actor {

fn is(&self, person: &Principal) -> bool {

96

self.id.as_deref().is_some_and(|id| id == person.id) || self.is_named(&person.username)

97

}

98

99

fn is_named(&self, username: &str) -> bool {

100

self.username.as_deref().is_some_and(|name| name.eq_ignore_ascii_case(username))

101

}

102

103

/// A person's name, for a title: never g1t's ids.

104

fn name(&self) -> Option<&str> {

105

self.username.as_deref().filter(|name| !is_g1t(name))

}

}

pub(crate) fn is_g1t(username: &str) -> bool {

110

username.eq_ignore_ascii_case(system::USERNAME) || username.eq_ignore_ascii_case("g1t-agent")

111

}

112

113

pub(crate) fn is_g1t_id(id: &str) -> bool {

114

system::is_system_id(id) || id == AGENT_ID

115

}

116

117

/// Events that end what an agent was waiting on a person for: it picked

118

/// back up, its head moved, a merge was asked for, or it is over.

119

const RESUMES: [&str; 5] = ["pull.resumed", "pull.updated", "pull.merge_requested", "pull.merged", "pull.closed"];

120

121

/// The issue or pull request an event names, and what to read for it.

122

/// None for events the inbox does not tell anyone of.

123

pub fn wants(event: &Event) -> Option<Wanted> {

124

let data = &event.data;

125

let text = |key: &str| data[key].as_str().filter(|value| !value.is_empty()).map(str::to_owned);

126

let number = |key: &str| data[key].as_u64().and_then(|n| u32::try_from(n).ok());

127

let repo_id = text("repoId").or_else(|| event.repo_id.clone())?;

128

let on = |number: Option<u32>, comment_id: Option<String>| {

129

Some(Wanted {

130

repo_id: repo_id.clone(),

number,

comment_id,

})

};

match event.kind.as_str() {

136

137

138

| "issue.assigned" => on(Some(number("number")?), None),

139

"checks.completed" => match data["status"].as_str() {

140

Some("failed" | "errored") => on(Some(number("number")?), None),

141

_ => None,

142

},

143

"review.completed" => match data["verdict"].as_str() {

144

Some("approve" | "request_changes") => on(Some(number("number")?), None),

145

_ => None,

146

},

147

"workflow.completed" => match data["conclusion"].as_str() {

148

Some("failure") => on(number("pull"), None),

149

_ => None,

150

},

151

"deployment.failed" | "deployment.succeeded" => on(number("number"), None),

152

// A workflow run's jobs wait for their environment's reviewers.

153

"deployment.review_requested" => on(None, None),

154

"comment.created" => on(Some(number("number")?), Some(text("commentId")?)),

155

// Security alerts are threads of their own, not of an issue.

156

kind if SECURITY_EVENTS.contains(&kind) => on(None, None),

157

// So is a repository's mirror, told to whom its settings name.

158

kind if MIRROR_EVENTS.contains(&kind) => on(None, None),

_ => None,

}

}

/// The security service's events the inbox tells people of: each scan's

164

/// summary of what it opened and fixed (one thread per repository and

165

/// alert type, brought up to date by the next scan; never one item per

166

/// alert), push protection bypasses asked for and decided, vulnerability

167

/// reports, and advisories published. The per-alert `.created`, `.fixed`

168

/// and `.dismissed` events are for webhooks.

169

pub const SECURITY_EVENTS: [&str; 7] = [

170

"secret_scanning_alert.summary",

171

"code_scanning_alert.summary",

172

"vulnerability_alert.summary",

173

"secret_scanning.bypass_requested",

174

"secret_scanning.bypass_reviewed",

175

"security_advisory.reported",

176

"security_advisory.published",

177

];

178

179

/// The integrations service's mirroring events (see

180

/// `g1t_contracts::mirrors::MirrorEvent`). People are told only when the

181

/// link's settings ask for the inbox, and the event names them.

182

pub const MIRROR_EVENTS: [&str; 4] = ["mirror.unreachable", "mirror.reachable", "mirror.state_changed", "mirror.moved_in"];

183

184

/// Where an event's items go: which thread, what it is, and where it is.

185

#[derive(Clone, Debug, PartialEq, Eq)]

186

pub struct Thread {

187

pub key: String,

188

pub kind: Option<SubjectKind>,

189

pub number: Option<u32>,

190

pub run_id: Option<String>,

191

/// A path, for a subject with a page of its own.

192

pub link: Option<String>,

193

}

194

195

/// The thread key of an issue or pull request.

196

pub fn numbered_thread(repo_id: &str, number: u32) -> String {

197

format!("{repo_id}#{number}")

198

}

199

200

/// The thread an event's items go to.

201

pub fn thread_of(event: &Event, wanted: &Wanted, subject: Option<&InboxSubject>) -> Thread {

202

let data = &event.data;

203

let text = |key: &str| data[key].as_str().filter(|value| !value.is_empty()).map(str::to_owned);

204

match event.kind.as_str() {

205

// A project's production, or one pull request's preview.

206

"deployment.failed" | "deployment.succeeded" => {

207

let which = wanted.number.map_or_else(|| "production".to_owned(), |number| number.to_string());

208

Thread {

209

key: format!("{}/deploy/{}/{which}", wanted.repo_id, text("projectId").unwrap_or_default()),

210

kind: Some(SubjectKind::Deploy),

211

number: wanted.number,

212

run_id: text("deploymentId"),

link: text("path"),

}

}

// One run's deployment to one environment, waiting for review.

217

"deployment.review_requested" => Thread {

key: format!(

"{}/review/{}/{}",

wanted.repo_id,

text("runId").unwrap_or_default(),

222

text("environment").unwrap_or_default()

223

),

224

kind: Some(SubjectKind::Run),

225

number: None,

226

run_id: text("runId"),

227

link: text("link"),

228

},

229

// A workflow on a branch: its next failure bumps the same thread.

230

"workflow.completed" if subject.is_none() => {

231

let branch = text("ref").unwrap_or_default();

232

let branch = branch.strip_prefix("refs/heads/").unwrap_or(&branch).to_owned();

233

let workflow = text("path").or_else(|| text("workflow")).unwrap_or_default();

234

Thread {

235

key: format!("{}/run/{workflow}@{branch}", wanted.repo_id),

236

kind: Some(SubjectKind::Run),

237

number: None,

238

run_id: text("runId"),

link: None,

}

}

// A repository's mirror: one thread, its settings page the link.

243

kind if MIRROR_EVENTS.contains(&kind) => Thread {

244

key: format!("{}/mirror", wanted.repo_id),

kind: None,

number: None,

run_id: None,

link: text("link"),

},

// A repository's alerts of one type (`rep_1/security/vulnerability`),

251

// its reports, one advisory, or one bypass request: the page is the link.

252

kind if SECURITY_EVENTS.contains(&kind) => {

253

let which = match kind {

254

"security_advisory.reported" => "reports".to_owned(),

255

"security_advisory.published" => format!("advisories/{}", text("id").unwrap_or_default()),

256

_ if kind.ends_with(".summary") => text("alertType").unwrap_or_else(|| "alerts".to_owned()),

257

_ => text("requestId").or_else(|| text("alertId")).unwrap_or_else(|| event.id.clone()),

258

};

259

Thread {

260

key: format!("{}/security/{which}", wanted.repo_id),

kind: None,

number: None,

run_id: None,

link: text("link"),

}

}

_ => match wanted.number {

268

Some(number) => Thread {

269

key: numbered_thread(&wanted.repo_id, number),

270

kind: subject.and_then(|subject| subject.kind),

271

number: Some(number),

run_id: None,

link: None,

},

None => Thread {

key: format!("event/{}", event.id),

kind: None,

number: None,

run_id: None,

link: None,

},

},

}

}

/// The issue or pull request whose agent threads an event closes: what an

287

/// agent was waiting on a person for is over.

288

pub fn resolves(event: &Event) -> Option<String> {

289

if !RESUMES.contains(&event.kind.as_str()) {

290

return None;

291

}

292

let repo_id = event.data["repoId"].as_str().map(str::to_owned).or_else(|| event.repo_id.clone())?;

293

let number = event.data["number"].as_u64().and_then(|n| u32::try_from(n).ok())?;

294

Some(numbered_thread(&repo_id, number))

295

}

296

297

/// Collects who is told, each once with the most specific reason, never

298

/// the actor, never g1t, and never anyone ignoring the thread.

299

struct Told<'a> {

300

actor: &'a Actor,

301

audience: &'a Audience,

302

notices: Vec<Notice>,

}

impl Told<'_> {

/// `asked`: something asked of the person directly, told even when

307

/// they unsubscribed from the thread.

308

fn tell(&mut self, username: &str, reason: Reason, severity: Severity, title: &str, body: &str, asked: bool) {

309

let username = username.trim().trim_start_matches('@').to_lowercase();

310

if username.is_empty()

311

|| is_g1t(&username)

312

|| self.actor.is_named(&username)

313

|| self.audience.ignores(&username)

314

|| (!asked && self.audience.unsubscribed(&username))

{

return;

}

let notice = Notice {

username,

reason,

severity,

title: clip(title, MAX_TITLE),

323

body: clip(body, MAX_BODY),

324

};

325

match self.notices.iter_mut().find(|told| told.username == notice.username) {

326

Some(told) if notice.reason.rank() < told.reason.rank() => *told = notice,

327

Some(_) => {}

328

None => self.notices.push(notice),

}

}

/// A person known by id as well as name, such as an author.

333

fn tell_person(&mut self, person: &Principal, reason: Reason, severity: Severity, title: &str, body: &str, asked: bool) {

334

if is_g1t_id(&person.id) || self.actor.is(person) {

335

return;

336

}

337

self.tell(&person.username, reason, severity, title, body, asked);

338

}

339

340

/// Everyone subscribed to an issue or pull request: its owner and

341

/// author, its assignees and reviewers, and whoever subscribed by

342

/// commenting, being mentioned or by hand. `reason` overrides why

343

/// each is told, as a state change does.

344

fn tell_subscribed(&mut self, subject: &InboxSubject, reason: Option<Reason>, severity: Severity, title: &str, body: &str) {

345

let why = |own: Reason| reason.unwrap_or(own);

346

self.tell_person(subject.owner(), why(Reason::Author), severity, title, body, false);

347

self.tell_person(&subject.author, why(Reason::Author), severity, title, body, false);

348

for name in &subject.assignees {

349

self.tell(name, why(Reason::Assign), severity, title, body, false);

350

}

351

for name in &subject.reviewers {

352

self.tell(name, why(Reason::ReviewRequested), severity, title, body, false);

353

}

354

let subscribed: Vec<(String, Reason)> = self.audience.subscribed().collect();

355

for (name, own) in subscribed {

356

self.tell(&name, why(own), severity, title, body, false);

}

}

/// Everyone watching the repository for this kind of activity.

361

fn tell_watchers(&mut self, kind: &str, severity: Severity, title: &str, body: &str) {

362

let watching: Vec<String> = self.audience.watching(kind).collect();

363

for name in watching {

364

self.tell(&name, Reason::Subscribed, severity, title, body, false);

}

}

}

fn clip(text: &str, max: usize) -> String {

370

let text = text.trim();

371

if text.chars().count() <= max {

372

return text.to_owned();

373

}

374

let cut: String = text.chars().take(max - 1).collect();

375

format!("{}…", cut.trim_end())

376

}

377

378

/// The kind of activity a watcher chooses: `issues` or `pulls`.

379

fn activity_of(subject: &InboxSubject) -> &'static str {

380

match subject.kind {

381

Some(SubjectKind::Pull) => "pulls",

_ => "issues",

}

}

/// The names in an event's list field, such as the reviewers just asked.

387

/// The teams an event asked to review: each `workspace/slug` with the

388

/// people it tells.

389

fn teams_asked(data: &serde_json::Value) -> Vec<(String, Vec<String>)> {

data["teams"]

.as_array()

.map(|teams| {

teams

.iter()

.filter_map(|team| Some((team["team"].as_str()?.to_owned(), names(team, "notified"))))

.collect()

})

.unwrap_or_default()

}

fn names(data: &serde_json::Value, key: &str) -> Vec<String> {

402

data[key]

403

.as_array()

404

.map(|names| names.iter().filter_map(|name| name.as_str().map(str::to_owned)).collect())

.unwrap_or_default()

}

/// Who is told of `event`, in `repo` (`owner/name`), given what it names

409

/// and who follows it. `actor` is who caused it; outcomes nobody chose

410

/// (checks, workflows, deployments, a review, an agent finishing or

411

/// stopping) are told whoever caused them.

412

pub fn notices(event: &Event, repo: &str, actor: &Actor, subject: Option<&InboxSubject>, audience: &Audience) -> Vec<Notice> {

413

let nobody = Actor::default();

414

let data = &event.data;

415

// The issue or pull request, as titles name it: `acme/rocket#12`.

416

let at = match data["number"].as_u64().or(data["pull"].as_u64()) {

417

Some(number) if subject.is_some() => format!("{repo}#{number}"),

418

_ => repo.to_owned(),

419

};

420

let outcome = matches!(

421

event.kind.as_str(),

422

"checks.completed"

423

| "workflow.completed"

| "review.completed"

| "pull.ready"

| "agent.asked"

| "pull.stalled"

| "deployment.failed"

429

| "deployment.succeeded"

430

| "deployment.review_requested"

431

) || SECURITY_EVENTS.contains(&event.kind.as_str())

432

|| MIRROR_EVENTS.contains(&event.kind.as_str());

433

let mut told = Told {

434

actor: if outcome { &nobody } else { actor },

audience,

notices: Vec::new(),

};

// Who did it, as titles name them.

439

let who = actor.name().unwrap_or("g1t");

440

441

match (event.kind.as_str(), subject) {

442

("agent.asked" | "pull.stalled", Some(pull)) => {

443

let (title, body) = if event.kind == "agent.asked" {

444

(format!("An agent is waiting on {at}"), pull.title.clone())

445

} else {

446

let detail = data["detail"].as_str().map(str::trim).filter(|detail| !detail.is_empty());

447

(format!("g1t stopped on {at} and needs you"), detail.unwrap_or(&pull.title).to_owned())

448

};

449

told.tell_person(pull.owner(), Reason::Agent, Severity::Warning, &title, &body, true);

450

if let Some(issue) = &pull.issue {

451

told.tell_person(issue.owner(), Reason::Agent, Severity::Warning, &title, &body, true);

452

for name in &issue.assignees {

453

told.tell(name, Reason::Agent, Severity::Warning, &title, &body, true);

}

}

}

("pull.review_requested", Some(pull)) => {

458

// Asked because the CODEOWNERS file says they own what changed.

459

let owned = data["codeOwners"].as_bool() == Some(true);

460

let title = if owned {

461

format!("{at} changes files you own")

462

} else {

463

format!("{who} asked you to review {at}")

464

};

465

for name in names(data, "reviewers") {

466

told.tell(&name, Reason::ReviewRequested, Severity::Warning, &title, &pull.title, true);

467

}

468

for (team, people) in teams_asked(data) {

469

let title = if owned {

470

format!("{at} changes files @{team} owns")

471

} else {

472

format!("{who} asked @{team} to review {at}")

473

};

474

for name in people {

475

told.tell(&name, Reason::ReviewRequested, Severity::Warning, &title, &pull.title, true);

}

}

}

("issue.assigned" | "pull.assigned", Some(on)) => {

480

let title = format!("{who} assigned you to {at}");

481

for name in names(data, "added") {

482

told.tell(&name, Reason::Assign, Severity::Info, &title, &on.title, true);

483

}

484

}

485

("issue.opened" | "pull.opened", Some(on)) => {

486

let title = format!("{who} opened {at}");

487

for name in &on.assignees {

488

told.tell(name, Reason::Assign, Severity::Info, &format!("{who} assigned you to {at}"), &on.title, true);

489

}

490

for name in &on.reviewers {

491

told.tell(name, Reason::ReviewRequested, Severity::Warning, &format!("{who} asked you to review {at}"), &on.title, true);

492

}

493

for name in &on.mentions {

494

told.tell(name, Reason::Mention, Severity::Info, &format!("{who} mentioned you on {at}"), &on.title, true);

495

}

496

for team in &on.team_mentions {

497

let title = format!("{who} mentioned @{} on {at}", team.team);

498

for name in &team.members {

499

told.tell(name, Reason::TeamMention, Severity::Info, &title, &on.title, true);

500

}

501

}

502

told.tell_watchers(activity_of(on), Severity::Info, &title, &on.title);

503

}

504

("checks.completed", Some(pull)) => {

505

let title = match data["status"].as_str() {

506

Some("errored") => format!("Checks could not run on {at}"),

507

_ => format!("Checks failed on {at}"),

508

};

509

told.tell_person(pull.owner(), Reason::CiActivity, Severity::Error, &title, &pull.title, true);

510

}

511

("workflow.completed", subject) => {

512

let workflow = data["workflow"].as_str().filter(|name| !name.is_empty()).unwrap_or("A workflow");

513

match subject {

514

Some(pull) => {

515

let title = format!("{workflow} failed on {at}");

516

told.tell_person(pull.owner(), Reason::CiActivity, Severity::Error, &title, &pull.title, true);

517

}

518

None => {

519

// Not on a pull request: whoever pushed the commit it ran on.

520

let branch = data["ref"].as_str().unwrap_or_default();

521

let branch = branch.strip_prefix("refs/heads/").unwrap_or(branch);

522

let title = format!("{workflow} failed on {branch} in {repo}");

523

let body = format!("Run {} at {}", data["number"], short(data["sha"].as_str().unwrap_or_default()));

524

if let Some(id) = &actor.id

525

&& !is_g1t_id(id)

526

&& let Some(name) = &actor.username

527

{

528

told.tell(name, Reason::CiActivity, Severity::Error, &title, &body, true);

}

}

}

}

("deployment.failed" | "deployment.succeeded", subject) => {

534

let failed = event.kind == "deployment.failed";

535

let project = data["project"].as_str().filter(|name| !name.is_empty()).unwrap_or(repo);

536

let what = match subject {

537

Some(_) => format!("The preview of {at}"),

538

None => format!("Production of {project}"),

539

};

540

let (title, severity) = if failed {

541

(format!("{what} failed to deploy"), Severity::Error)

542

} else {

543

(format!("{what} is live"), Severity::Success)

544

};

545

let body = match (failed, data["error"].as_str().map(str::trim).filter(|error| !error.is_empty())) {

546

(true, Some(error)) => error.to_owned(),

547

_ => match subject {

548

Some(pull) => pull.title.clone(),

549

None => format!("Commit {}", short(data["commit"].as_str().unwrap_or_default())),

550

},

551

};

552

// A success is news to whoever answers for it only after a failure.

553

if failed || data["recovered"].as_bool() == Some(true) {

554

let title = if failed { title.clone() } else { format!("{what} is live again") };

555

match subject {

556

Some(pull) => told.tell_person(pull.owner(), Reason::CiActivity, severity, &title, &body, true),

None => {

let pusher = actor

.id

.as_deref()

.filter(|id| !is_g1t_id(id))

562

.and(actor.username.as_deref())

563

.or_else(|| data["triggeredBy"].as_str());

564

if let Some(name) = pusher {

565

told.tell(name, Reason::CiActivity, severity, &title, &body, true);

}

}

}

}

told.tell_watchers("deployments", severity, &title, &body);

571

}

572

("review.completed", Some(pull)) => {

573

let (title, severity) = match data["verdict"].as_str() {

574

Some("approve") => (format!("g1t approved {at}"), Severity::Success),

575

_ => (format!("g1t asked for changes on {at}"), Severity::Info),

576

};

577

told.tell_person(pull.owner(), Reason::Author, severity, &title, &pull.title, true);

578

}

579

("pull.ready", Some(pull)) => {

580

// A change g1t made is ready: the agent's run is over.

581

if let Some(owner) = pull.requested_by.as_ref().filter(|_| is_g1t_id(&pull.author.id) || is_g1t(&pull.author.username)) {

582

let title = format!("g1t finished {at}");

583

told.tell_person(owner, Reason::Author, Severity::Success, &title, &pull.title, true);

584

}

585

}

586

("pull.merged" | "pull.closed" | "issue.closed" | "issue.reopened", Some(on)) => {

587

let (verb, severity) = match event.kind.as_str() {

588

"pull.merged" => ("merged", Severity::Success),

589

"issue.reopened" => ("reopened", Severity::Info),

590

_ => ("closed", Severity::Info),

591

};

592

let title = match (actor.name(), data["resolvedBy"].as_u64()) {

593

(_, Some(pull)) if event.kind == "issue.closed" => format!("{at} was closed by #{pull}"),

594

(Some(name), _) => format!("{name} {verb} {at}"),

595

(None, _) => format!("{at} was {verb}"),

596

};

597

told.tell_subscribed(on, Some(Reason::StateChange), severity, &title, &on.title);

598

told.tell_watchers(activity_of(on), severity, &title, &on.title);

599

}

600

("deployment.review_requested", _) => {

601

let environment = data["environment"].as_str().unwrap_or("an environment");

602

let workflow = data["workflow"].as_str().unwrap_or("A workflow");

603

let title = format!("{workflow} is waiting for your review to deploy to {environment} in {repo}");

604

let body = data["title"].as_str().unwrap_or_default();

605

// Those the actions service names: the environment's reviewers.

606

for name in names(data, "notify") {

607

told.tell(&name, Reason::ReviewRequested, Severity::Warning, &title, body, true);

608

}

609

}

610

(kind, None) if MIRROR_EVENTS.contains(&kind) => {

611

let severity = match kind {

612

"mirror.unreachable" => Severity::Warning,

613

"mirror.state_changed" if data["state"].as_str() == Some("takeover") => Severity::Warning,

614

"mirror.state_changed" if data["state"].as_str() == Some("standby") => Severity::Success,

615

_ => Severity::Info,

616

};

617

let title = data["title"].as_str().unwrap_or("A mirror changed");

618

let body = data["detail"].as_str().unwrap_or_default();

619

for name in names(data, "notify") {

620

told.tell(&name, Reason::StateChange, severity, title, body, true);

621

}

622

}

623

(kind, None) if SECURITY_EVENTS.contains(&kind) => {

624

let what = data["title"].as_str().unwrap_or("A security alert");

625

match kind {

626

"secret_scanning.bypass_requested" => {

627

let title = format!("{who} asked to bypass push protection in {repo}");

628

for name in names(data, "notify") {

629

told.tell(&name, Reason::SecurityAlert, Severity::Warning, &title, what, true);

630

}

631

}

632

"secret_scanning.bypass_reviewed" => {

633

let verdict = data["state"].as_str().unwrap_or("reviewed");

634

let title = format!("Your request to bypass push protection in {repo} was {verdict}");

635

for name in names(data, "notify") {

636

told.tell(&name, Reason::SecurityAlert, Severity::Info, &title, what, true);

637

}

638

}

639

// Something to act on: the owners and security managers, at once.

640

"security_advisory.reported" => {

641

let title = format!("A vulnerability was reported privately to {repo}");

642

for name in names(data, "notify") {

643

told.tell(&name, Reason::SecurityAlert, Severity::Warning, &title, what, true);

644

}

645

}

646

// Published for everyone: watchers who chose security.

647

"security_advisory.published" => {

648

let title = format!("{repo} published a security advisory");

649

told.tell_watchers("security", Severity::Info, &title, what);

650

}

651

// A scan's summary. One that opened nothing is no news: it

652

// brings the thread up to date instead (deliver_one). Whether

653

// what is not urgent is told now or in a digest is each

654

// person's choice, applied there too.

655

_ => {

656

let Some(summary) = security_summary(event) else {

657

return told.notices;

658

};

659

if summary.opened_count == 0 {

660

return told.notices;

661

}

662

let severity = if summary.urgent() { Severity::Error } else { Severity::Info };

663

let (title, pushed) = summary_titles(repo, &summary);

664

let body = summary.latest.clone().unwrap_or_default();

665

// Whoever pushed a refused push must act: told at once, whatever was found.

666

if let Some(pusher) = &summary.pusher {

667

told.tell(pusher, Reason::SecurityAlert, Severity::Error, &pushed, &body, true);

668

}

669

for name in &summary.notify {

670

told.tell(name, Reason::SecurityAlert, severity, &title, &body, true);

671

}

672

// Watchers who chose security alerts, if they may see findings:

673

// the service lists who may (`members`), as findings are never

674

// shown to someone who cannot change the code.

675

let members = names(data, "members");

676

let watching: Vec<String> = told.audience.watching("security").collect();

677

for name in watching.iter().filter(|name| members.iter().any(|member| member.eq_ignore_ascii_case(name))) {

678

told.tell(name, Reason::SecurityAlert, severity, &title, &body, false);

}

}

}

}

("comment.created", Some(on)) => {

684

let Some(comment) = on.comment.as_ref().filter(|comment| !comment.event) else {

685

return Vec::new();

686

};

687

// Whoever wrote it is the actor, whatever the event says. One of

688

// the workspace's agents is shown by its name, marked as an

689

// agent; its handle is not a person's, so nobody is left out

690

// for sharing it, and whoever it acted for hears of it as of

691

// anything they set off.

692

let writer = match &comment.agent {

693

Some(agent) => Actor { id: Some(agent.id.clone()), username: None },

694

None => Actor {

695

id: Some(comment.author.id.clone()),

696

username: Some(comment.author.username.clone()),

697

},

698

};

699

told.actor = &writer;

700

let who = match &comment.agent {

701

Some(agent) => format!("{} (agent)", agent.display_name),

702

None => comment.author.username.clone(),

703

};

704

let advisory = if comment.advisory { " (advisory)" } else { "" };

705

let body = if comment.excerpt.is_empty() { &on.title } else { &comment.excerpt };

706

for name in &comment.mentions {

707

let title = format!("{who} mentioned you on {at}");

708

told.tell(name, Reason::Mention, Severity::Info, &title, body, true);

709

}

710

for team in &comment.team_mentions {

711

let title = format!("{who} mentioned @{} on {at}", team.team);

712

for name in &team.members {

713

told.tell(name, Reason::TeamMention, Severity::Info, &title, body, true);

714

}

715

}

716

let (severity, title) = match comment.verdict.as_deref() {

717

Some("approve") => (Severity::Success, format!("{who} approved {at}{advisory}")),

718

Some("request_changes") => (Severity::Info, format!("{who} asked for changes on {at}{advisory}")),

719

_ if comment.advisory => (Severity::Info, format!("{who} reviewed {at}{advisory}")),

720

_ => (Severity::Info, format!("{who} commented on {at}")),

721

};

722

// An approval or a request for changes (or an agent's review)

723

// is the owner's to hear of whatever they chose; the rest, as

724

// they subscribed.

725

if comment.verdict.is_some() || comment.advisory {

726

told.tell_person(on.owner(), Reason::Author, severity, &title, body, true);

727

}

728

told.tell_subscribed(on, None, severity, &title, body);

729

told.tell_watchers(activity_of(on), severity, &title, body);

return told.notices;

}

_ => {}

}

told.notices

}

/// Who an event subscribes to its issue or pull request without asking,

738

/// and why: whoever commented, whoever they mentioned, the people assigned

739

/// and the reviewers asked. Never g1t.

740

pub fn subscribes(event: &Event, subject: Option<&InboxSubject>) -> Vec<(String, Reason)> {

741

let mut people: Vec<(String, Reason)> = Vec::new();

742

let mut add = |name: &str, reason: Reason| {

743

let name = name.trim().trim_start_matches('@').to_lowercase();

744

if !name.is_empty() && !is_g1t(&name) && !people.iter().any(|(had, _)| *had == name) {

745

people.push((name, reason));

746

}

747

};

748

match event.kind.as_str() {

749

"comment.created" => {

750

if let Some(comment) = subject.and_then(|subject| subject.comment.as_ref()).filter(|comment| !comment.event) {

751

// An agent is no person to subscribe; its handle may be someone's name.

752

if !is_g1t_id(&comment.author.id) && comment.agent.is_none() {

753

add(&comment.author.username, Reason::Comment);

754

}

755

for name in &comment.mentions {

756

add(name, Reason::Mention);

757

}

758

for team in &comment.team_mentions {

759

for name in &team.members {

760

add(name, Reason::TeamMention);

}

}

}

}

"issue.opened" | "pull.opened" => {

766

if let Some(on) = subject {

767

for name in &on.mentions {

768

add(name, Reason::Mention);

769

}

770

for team in &on.team_mentions {

771

for name in &team.members {

772

add(name, Reason::TeamMention);

}

}

}

}

"issue.assigned" | "pull.assigned" => {

778

for name in names(&event.data, "added") {

779

add(&name, Reason::Assign);

780

}

781

}

782

"pull.review_requested" => {

783

for name in names(&event.data, "reviewers") {

784

add(&name, Reason::ReviewRequested);

785

}

786

for (_, people) in teams_asked(&event.data) {

787

for name in people {

788

add(&name, Reason::ReviewRequested);

}

}

}

_ => {}

}

people

}

fn short(sha: &str) -> &str {

798

sha.get(..7).unwrap_or(sha)

799

}

800

801

// --- Security summaries ------------------------------------------------------

802

803

/// The scan summary a `<type>_alert.summary` event carries.

804

pub fn security_summary(event: &Event) -> Option<SecuritySummary> {

805

if !event.kind.ends_with("_alert.summary") {

806

return None;

807

}

808

serde_json::from_value(event.data.clone()).ok()

809

}

810

811

/// "1 critical, 2 high": the counts that are not zero, worst first.

812

pub fn breakdown(counts: &SeverityCounts) -> String {

813

[("critical", counts.critical), ("high", counts.high), ("medium", counts.medium), ("low", counts.low), ("unknown", counts.unknown)]

814

.into_iter()

815

.filter(|(_, n)| *n > 0)

816

.map(|(name, n)| format!("{n} {name}"))

.collect::<Vec<_>>()

.join(", ")

}

/// "vulnerability" or "vulnerabilities", as a count wants.

822

fn alerts_called(alert_type: &str, n: u32) -> &'static str {

823

match (AlertType::parse(alert_type), n) {

824

(Some(AlertType::SecretScanning), 1) => "secret",

825

(Some(AlertType::SecretScanning), _) => "secrets",

826

(Some(AlertType::CodeScanning), 1) => "code scanning alert",

827

(Some(AlertType::CodeScanning), _) => "code scanning alerts",

828

(Some(AlertType::Vulnerability), 1) => "vulnerability",

829

(Some(AlertType::Vulnerability), _) => "vulnerabilities",

830

(None, 1) => "security alert",

831

(None, _) => "security alerts",

}

}

/// What a scan's item says to those who look after the repository ("3 new

836

/// vulnerabilities in acme/rocket: 1 critical, 2 high"), and to whoever

837

/// pushed a push that was refused.

838

pub fn summary_titles(repo: &str, summary: &SecuritySummary) -> (String, String) {

839

let n = summary.opened_count;

840

let counts = breakdown(&summary.new_by_severity);

841

let title = if counts.is_empty() {

842

format!("{n} new {} in {repo}", alerts_called(&summary.alert_type, n))

843

} else {

844

format!("{n} new {} in {repo}: {counts}", alerts_called(&summary.alert_type, n))

845

};

846

let pushed = match summary.blocked {

847

0 | 1 => format!("A push to {repo} was blocked: it adds a secret"),

848

n => format!("A push to {repo} was blocked: it adds {n} secrets"),

};

(title, pushed)

}

/// What a scan that opened nothing makes the thread say: how many alerts

854

/// are open now ("2 open vulnerabilities in acme/rocket: 2 high"), or that

855

/// none is, with what changed as the line of history.

856

pub fn refreshed_text(repo: &str, summary: &SecuritySummary) -> (String, String) {

857

let open = summary.open_total();

858

let mut changes = Vec::new();

859

if summary.fixed_count > 0 {

860

changes.push(format!("{} fixed", summary.fixed_count));

861

}

862

if summary.dismissed_count > 0 {

863

changes.push(format!("{} dismissed", summary.dismissed_count));

864

}

865

let title = if open == 0 {

866

format!("No open {} in {repo}", alerts_called(&summary.alert_type, 2))

867

} else {

868

format!("{open} open {} in {repo}: {}", alerts_called(&summary.alert_type, open), breakdown(&summary.open))

869

};

870

(title, changes.join(", "))

871

}

872

873

/// Where a scan summary's notice goes for one person.

874

#[derive(Clone, Copy, Debug, PartialEq, Eq)]

875

pub enum Delivery {

876

/// Their inbox, now.

877

Now,

878

/// Their next security digest.

Digest,

/// Nowhere.

Never,

}

/// The policy: critical and high alerts, and a refused push, reach the

885

/// people the service names (the workspace's owners and security managers,

886

/// the pusher) at once whatever they chose, and watchers at once unless

887

/// they turned security alerts off. Everything else waits for the person's

888

/// digest, or comes at once or never, as they chose.

889

pub fn security_delivery(summary: &SecuritySummary, username: &str, settings: &InboxSettings) -> Delivery {

890

if summary.urgent() {

891

let named = summary.notify.iter().any(|name| name.eq_ignore_ascii_case(username))

892

|| summary.pusher.as_deref().is_some_and(|pusher| pusher.eq_ignore_ascii_case(username));

893

return if named || settings.security_alerts { Delivery::Now } else { Delivery::Never };

894

}

895

match settings.security_digest {

896

DigestCadence::AtOnce => Delivery::Now,

897

DigestCadence::Daily | DigestCadence::Weekly => Delivery::Digest,

898

DigestCadence::Off => Delivery::Never,

}

}

/// Brings every person's thread about a repository's alerts of one type up

903

/// to date after a scan that opened nothing: the title says what is open

904

/// now, a line of history says what changed, and a thread with nothing

905

/// left open is done. Nobody is woken: what was read stays read, and the

906

/// thread keeps its place.

907

async fn refresh_security_thread(db: &D1Database, thread: &str, repo: &str, summary: &SecuritySummary, event: &Event) -> Result<()> {

908

let (title, body) = refreshed_text(repo, summary);

909

let now = now_ms();

910

let at = rfc3339(now);

911

let settled = i32::from(summary.open_total() == 0);

912

db.batch(vec![

913

db.prepare(

914

"UPDATE inbox_items SET event_id = ?1, event = ?2, title = ?3, body = ?4, updated_at = ?5, activity = activity + 1,

915

done_at = CASE WHEN ?6 = 1 THEN COALESCE(done_at, ?5) ELSE done_at END,

916

read_at = CASE WHEN ?6 = 1 THEN COALESCE(read_at, ?5) ELSE read_at END

917

WHERE thread = ?7

918

AND NOT EXISTS (SELECT 1 FROM inbox_activity a WHERE a.event_id = ?1 AND a.username = inbox_items.username)",

919

)

920

.bind(&[

921

event.id.as_str().into(),

922

event.kind.as_str().into(),

923

title.as_str().into(),

924

body.as_str().into(),

at.as_str().into(),

settled.into(),

thread.into(),

])?,

db.prepare(

"INSERT OR IGNORE INTO inbox_activity (id, item_id, username, event_id, event, reason, severity, title, body, actor, created_at)

931

SELECT ?1 || ':' || username, id, username, ?2, ?3, 'security_alert', 'info', ?4, ?5, NULL, ?6 FROM inbox_items WHERE thread = ?7",

932

)

933

.bind(&[

934

new_id("ntf", now).into(),

935

event.id.as_str().into(),

936

event.kind.as_str().into(),

937

title.as_str().into(),

938

body.as_str().into(),

939

event.time.as_str().into(),

thread.into(),

])?,

])

.await?;

Ok(())

}

/// Where an item is on g1t.sh.

948

pub fn url(repo: Option<&str>, subject: Option<SubjectKind>, number: Option<u32>, run_id: Option<&str>, link: Option<&str>) -> String {

949

if let Some(link) = link.filter(|link| link.starts_with('/')) {

950

return link.to_owned();

951

}

952

let Some(repo) = repo else {

953

return "/inbox".to_owned();

954

};

955

match (subject, number, run_id) {

956

(Some(SubjectKind::Pull), Some(number), _) => format!("/{repo}/pull/{number}"),

957

(Some(SubjectKind::Issue), Some(number), _) => format!("/{repo}/issues/{number}"),

958

(Some(SubjectKind::Run), _, Some(run)) => format!("/{repo}/actions/runs/{run}"),

959

(Some(SubjectKind::Deploy), Some(number), _) => format!("/{repo}/pull/{number}"),

960

_ => format!("/{repo}"),

}

}

// --- Writing ---------------------------------------------------------------

965

966

/// The services the inbox reads from as events arrive.

967

pub struct Sources<'a> {

968

pub work: &'a Fetcher,

969

pub repos: &'a Fetcher,

970

pub identity: &'a Fetcher,

971

/// The notify service, told of each new item for live toasts and pushes; None, nobody is.

972

pub notify: Option<&'a Fetcher>,

973

}

974

975

/// One notice written, and whether it was news (not a redelivery): what

/// may be emailed.

struct Written {

notice: Notice,

repo_id: String,

url: String,

event_id: String,

}

/// Writes the items a batch from the bus calls for. Never fails the batch:

985

/// what cannot be worked out is logged and left.

986

pub async fn deliver(db: &D1Database, sources: &Sources<'_>, events: &[Event]) {

987

if let Err(error) = resolve(db, events).await {

988

worker::console_error!("inbox: agent threads not closed: {error}");

989

}

990

// A workspace's own notices, about no repository (workspace_notices).

991

for event in events.iter().filter(|event| WORKSPACE_EVENTS.contains(&event.kind.as_str())) {

992

if let Err(error) = deliver_workspace(db, event).await {

993

worker::console_error!("inbox: {} {} not delivered: {error}", event.kind, event.id);

994

} else if let Some(notify) = sources.notify {

995

// Notify: the feed tells each person once per event, redelivered or not.

996

let (_, told) = workspace_notices(event, None);

997

let link = event.data["link"].as_str().filter(|link| link.starts_with('/')).unwrap_or("/inbox");

998

let live = told.iter().map(|notice| (notice.username.clone(), live_notification(&event.id, notice, link, &event.time))).collect();

999

tell_live(notify, live).await;

1000

for notice in &told {

1001

tell_inbox_count(db, notify, &notice.username).await;

}

}

}

let wanted: Vec<(&Event, Wanted)> = events

1006

.iter()

1007

.filter_map(|event| wants(event).map(|wanted| (event, wanted)))

1008

.collect();

1009

let created: Vec<&Event> = events.iter().filter(|event| event.kind == "repo.created").collect();

1010

if wanted.is_empty() && created.is_empty() {

1011

return;

1012

}

1013

// Everyone who caused one, named in one call.

1014

let ids: Vec<String> = wanted

1015

.iter()

1016

.map(|(event, _)| *event)

1017

.chain(created.iter().copied())

1018

.filter_map(|event| event.actor.clone())

1019

.filter(|id| !is_g1t_id(id))

1020

.collect::<HashSet<_>>()

1021

.into_iter()

1022

.collect();

1023

let names: HashMap<String, String> = if ids.is_empty() {

1024

HashMap::new()

1025

} else {

1026

g1t_kit::call(sources.identity, "usernames", &UsernamesArgs { ids })

1027

.await

1028

.unwrap_or_else(|error| {

1029

worker::console_error!("inbox: could not name who acted: {error}");

HashMap::new()

})

};

for event in created {

1034

if let Err(error) = subscriptions::watch_created(db, event, &names).await {

1035

worker::console_error!("inbox: {} {} not watched: {error}", event.kind, event.id);

1036

}

1037

}

1038

let mut paths: HashMap<String, Option<RepoPath>> = HashMap::new();

1039

let mut watchers: HashMap<String, Vec<subscriptions::Watcher>> = HashMap::new();

1040

let mut written: Vec<Written> = Vec::new();

1041

for (event, wanted) in wanted {

1042

match deliver_one(db, sources, &names, &mut paths, &mut watchers, event, wanted).await {

1043

Ok(mut news) => written.append(&mut news),

1044

Err(error) => worker::console_error!("inbox: {} {} not delivered: {error}", event.kind, event.id),

1045

}

1046

}

1047

// Notify: every item that was news, live in the person's tabs (services/notify).

1048

if let Some(notify) = sources.notify {

1049

let live = written

1050

.iter()

1051

.map(|item| (item.notice.username.clone(), live_notification(&item.event_id, &item.notice, &item.url, &rfc3339(now_ms()))))

1052

.collect::<Vec<_>>();

1053

tell_live(notify, live).await;

1054

let people: HashSet<&str> = written.iter().map(|item| item.notice.username.as_str()).collect();

1055

for username in people {

1056

tell_inbox_count(db, notify, username).await;

1057

}

1058

}

1059

if let Err(error) = email(db, sources.identity, written).await {

1060

worker::console_error!("inbox: emails not sent: {error}");

}

}

/// Events about a workspace rather than a repository, each naming who to

1065

/// tell (`notify`), what to say (`title`, `body`) and where it is (`link`):

1066

/// identity's personal access token approvals.

1067

///

1068

1069

/// | --- | --- | --- | --- |

///

/// An invitation's item for the person invited is done once it is

1077

/// accepted, declined or revoked (`workspace_invitation.revoked`, which

1078

/// tells nobody).

1079

pub const WORKSPACE_EVENTS: [&str; 5] = [

1080

"token.approval_requested",

1081

"token.approval_reviewed",

1082

"workspace_invitation.created",

1083

"workspace_invitation.accepted",

1084

"workspace_invitation.declined",

1085

];

1086

1087

/// Events that close what a workspace invitation asked of its person.

1088

pub const INVITATION_ANSWERED: [&str; 3] =

1089

["workspace_invitation.accepted", "workspace_invitation.declined", "workspace_invitation.revoked"];

1090

1091

/// The notices a workspace event calls for, and the thread they go to.

1092

pub fn workspace_notices(event: &Event, actor: Option<&str>) -> (String, Vec<Notice>) {

1093

let data = &event.data;

1094

let text = |key: &str| data[key].as_str().unwrap_or_default().to_owned();

1095

let (reason, severity) = match event.kind.as_str() {

1096

"token.approval_requested" | "workspace_invitation.created" => (Reason::ReviewRequested, Severity::Warning),

1097

"workspace_invitation.accepted" => (Reason::Author, Severity::Success),

1098

_ => (Reason::Author, Severity::Info),

1099

};

1100

// An invitation names its own thread; a token approval's is the token's.

1101

let thread = match data["thread"].as_str().filter(|thread| !thread.is_empty()) {

1102

Some(thread) => thread.to_owned(),

1103

None => format!("workspace:{}/token/{}", text("workspace"), text("tokenId")),

1104

};

1105

let notices = names(data, "notify")

1106

.into_iter()

1107

.map(|name| name.to_lowercase())

1108

.filter(|name| actor.is_none_or(|actor| !actor.eq_ignore_ascii_case(name)) && !is_g1t(name))

1109

.collect::<HashSet<_>>()

1110

.into_iter()

1111

.map(|username| Notice { username, reason, severity, title: text("title"), body: text("body") })

.collect();

(thread, notices)

}

/// Writes a workspace event's notices: a thread each, with no repository.

1117

async fn deliver_workspace(db: &D1Database, event: &Event) -> Result<()> {

1118

let (thread, told) = workspace_notices(event, None);

if told.is_empty() {

return Ok(());

}

let now = now_ms();

let workspace = event.data["workspace"].as_str().unwrap_or_default().to_lowercase();

1124

let link = event.data["link"].as_str().filter(|link| link.starts_with('/'));

1125

let mut statements = Vec::new();

1126

for notice in &told {

1127

let username = notice.username.as_str();

1128

statements.push(db.prepare(BUMP).bind(&[

1129

new_id("ntf", now).into(),

1130

username.into(),

1131

thread.as_str().into(),

1132

event.id.as_str().into(),

1133

event.kind.as_str().into(),

1134

notice.reason.as_str().into(),

1135

notice.severity.as_str().into(),

1136

notice.title.as_str().into(),

1137

notice.body.as_str().into(),

1138

workspace.as_str().into(),

JsValue::NULL,

JsValue::NULL,

JsValue::NULL,

JsValue::NULL,

JsValue::NULL,

link.map_or(JsValue::NULL, JsValue::from),

1145

JsValue::NULL,

1146

event.time.as_str().into(),

1147

new_id("ntf", now).into(),

])?);

statements.push(

db.prepare(

"INSERT OR IGNORE INTO inbox_activity (id, item_id, username, event_id, event, reason, severity, title, body, actor, created_at)

1152

SELECT ?1, id, ?2, ?4, ?5, ?6, ?7, ?8, ?9, NULL, ?10 FROM inbox_items WHERE username = ?2 AND thread = ?3",

1153

)

1154

.bind(&[

1155

new_id("ntf", now).into(),

1156

username.into(),

1157

thread.as_str().into(),

1158

event.id.as_str().into(),

1159

event.kind.as_str().into(),

1160

notice.reason.as_str().into(),

1161

notice.severity.as_str().into(),

1162

notice.title.as_str().into(),

1163

notice.body.as_str().into(),

1164

event.time.as_str().into(),

])?,

);

}

db.batch(statements).await?;

Ok(())

}

/// Closes what an agent was waiting on a person for once it is over.

1173

async fn resolve(db: &D1Database, events: &[Event]) -> Result<()> {

1174

let now = rfc3339(now_ms());

1175

let mut statements = Vec::new();

1176

for thread in events.iter().filter_map(resolves) {

1177

statements.push(

1178

db.prepare(

1179

"UPDATE inbox_items SET done_at = ?1, read_at = COALESCE(read_at, ?1)

1180

WHERE thread = ?2 AND reason = 'agent' AND done_at IS NULL",

1181

)

1182

.bind(&[now.as_str().into(), thread.into()])?,

1183

);

1184

}

1185

// A workspace invitation answered or revoked no longer waits on its person.

1186

for event in events.iter().filter(|event| INVITATION_ANSWERED.contains(&event.kind.as_str())) {

1187

let Some(thread) = event.data["thread"].as_str().filter(|thread| thread.starts_with("invitation:")) else {

continue;

};

statements.push(

db.prepare(

"UPDATE inbox_items SET done_at = ?1, read_at = COALESCE(read_at, ?1)

1193

WHERE thread = ?2 AND reason = 'review_requested' AND done_at IS NULL",

1194

)

1195

.bind(&[now.as_str().into(), thread.into()])?,

1196

);

1197

}

1198

// Reviews no longer asked for are no longer waiting.

1199

for event in events.iter().filter(|event| event.kind == "pull.review_request_removed") {

1200

let (Some(repo_id), Some(number)) = (

1201

event.data["repoId"].as_str().map(str::to_owned).or_else(|| event.repo_id.clone()),

1202

event.data["number"].as_u64(),

) else {

continue;

};

for name in names(&event.data, "reviewers") {

1207

statements.push(

1208

db.prepare(

1209

"UPDATE inbox_items SET done_at = ?1, read_at = COALESCE(read_at, ?1)

1210

WHERE thread = ?2 AND username = ?3 AND reason = 'review_requested' AND done_at IS NULL",

)

.bind(&[

now.as_str().into(),

format!("{repo_id}#{number}").into(),

1215

name.to_lowercase().into(),

])?,

);

}

}

if !statements.is_empty() {

1221

db.batch(statements).await?;

}

Ok(())

}

async fn deliver_one(

1227

db: &D1Database,

1228

sources: &Sources<'_>,

1229

names: &HashMap<String, String>,

1230

paths: &mut HashMap<String, Option<RepoPath>>,

1231

watchers: &mut HashMap<String, Vec<subscriptions::Watcher>>,

1232

event: &Event,

1233

wanted: Wanted,

1234

) -> Result<Vec<Written>> {

1235

if !paths.contains_key(&wanted.repo_id) {

1236

let path: Option<RepoPath> = g1t_kit::call(

sources.repos,

"path_by_id",

&PathByIdArgs {

id: wanted.repo_id.clone(),

},

)

.await?;

paths.insert(wanted.repo_id.clone(), path);

1245

}

1246

// A repository that is gone tells nobody.

1247

let Some(path) = paths.get(&wanted.repo_id).cloned().flatten() else {

1248

return Ok(Vec::new());

1249

};

1250

let subject: Option<InboxSubject> = match wanted.number {

1251

Some(number) => {

1252

let found: Option<InboxSubject> = g1t_kit::call(

sources.work,

"inbox_subject",

&InboxSubjectArgs {

repo_id: wanted.repo_id.clone(),

1257

number,

1258

comment_id: wanted.comment_id.clone(),

},

)

.await?;

if found.is_none() {

return Ok(Vec::new());

}

found

}

None => None,

};

let actor = Actor {

id: event.actor.clone(),

1271

username: match event.actor.as_deref() {

1272

Some(id) if is_g1t_id(id) => Some(system::USERNAME.to_owned()),

1273

Some(id) => names.get(id).map(|name| name.to_lowercase()),

None => None,

},

};

let thread = thread_of(event, &wanted, subject.as_ref());

1278

if !watchers.contains_key(&wanted.repo_id) {

1279

watchers.insert(wanted.repo_id.clone(), subscriptions::watchers(db, &wanted.repo_id).await?);

1280

}

1281

let audience = Audience {

1282

subscriptions: match thread.kind {

1283

Some(SubjectKind::Issue | SubjectKind::Pull) => subscriptions::of_thread(db, &thread.key).await?,

1284

_ => Vec::new(),

1285

},

1286

watchers: watchers.get(&wanted.repo_id).cloned().unwrap_or_default(),

1287

};

1288

let repo = format!("{}/{}", path.namespace, path.name).to_lowercase();

1289

let mut told = notices(event, &repo, &actor, subject.as_ref(), &audience);

1290

let mut statements = subscriptions::auto_subscribe(db, &thread.key, &wanted.repo_id, &subscribes(event, subject.as_ref()), &event.time)?;

1291

// A scan's summary: one that opened nothing brings the thread up to date

1292

// without waking anyone, and what is not urgent waits for each person's

1293

// digest, or comes at once or never, as they chose (security_delivery).

1294

if let Some(summary) = security_summary(event) {

1295

if summary.opened_count == 0 {

1296

refresh_security_thread(db, &thread.key, &repo, &summary, event).await?;

1297

return Ok(Vec::new());

1298

}

1299

let people: Vec<String> = told.iter().map(|notice| notice.username.clone()).collect();

1300

let settings = subscriptions::settings_of(db, &people).await?;

1301

let now = now_ms();

1302

let mut kept = Vec::with_capacity(told.len());

1303

for notice in told {

1304

let setting = settings.get(&notice.username).cloned().unwrap_or_default();

1305

match security_delivery(&summary, &notice.username, &setting) {

1306

Delivery::Now => kept.push(notice),

1307

Delivery::Digest => statements.push(crate::security_digest::queue(db, &notice.username, &path, &summary, &event.id, now)?),

1308

Delivery::Never => {}

}

}

told = kept;

}

if told.is_empty() {

if !statements.is_empty() {

1315

db.batch(statements).await?;

1316

}

1317

return Ok(Vec::new());

1318

}

1319

let now = now_ms();

1320

let link = thread.link.as_deref();

1321

let item_url = url(Some(&repo), thread.kind, thread.number, thread.run_id.as_deref(), link);

1322

// Three statements a notice: its thread brought up (unless this event

1323

// was told before), a line of history, and the history kept short.

1324

let first = statements.len();

1325

for notice in &told {

1326

let username = notice.username.as_str();

1327

let values: Vec<JsValue> = vec![

1328

new_id("ntf", now).into(),

1329

username.into(),

1330

thread.key.as_str().into(),

1331

event.id.as_str().into(),

1332

event.kind.as_str().into(),

1333

notice.reason.as_str().into(),

1334

notice.severity.as_str().into(),

1335

notice.title.as_str().into(),

1336

notice.body.as_str().into(),

1337

path.namespace.to_lowercase().into(),

1338

wanted.repo_id.as_str().into(),

1339

repo.as_str().into(),

1340

thread.kind.map_or(JsValue::NULL, |kind| kind.as_str().into()),

1341

thread.number.map_or(JsValue::NULL, JsValue::from),

1342

thread.run_id.as_deref().map_or(JsValue::NULL, JsValue::from),

1343

link.map_or(JsValue::NULL, JsValue::from),

1344

actor.username.as_deref().map_or(JsValue::NULL, JsValue::from),

1345

event.time.as_str().into(),

1346

// Same prefix as item ids, so old and new sort by time together.

1347

new_id("ntf", now).into(),

1348

];

1349

statements.push(db.prepare(BUMP).bind(&values)?);

1350

statements.push(

1351

db.prepare(

1352

"INSERT OR IGNORE INTO inbox_activity (id, item_id, username, event_id, event, reason, severity, title, body, actor, created_at)

1353

SELECT ?1, id, ?2, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11 FROM inbox_items WHERE username = ?2 AND thread = ?3

RETURNING username",

)

.bind(&[

new_id("ntf", now).into(),

1358

username.into(),

1359

thread.key.as_str().into(),

1360

event.id.as_str().into(),

1361

event.kind.as_str().into(),

1362

notice.reason.as_str().into(),

1363

notice.severity.as_str().into(),

1364

notice.title.as_str().into(),

1365

notice.body.as_str().into(),

1366

actor.username.as_deref().map_or(JsValue::NULL, JsValue::from),

1367

event.time.as_str().into(),

])?,

);

statements.push(

db.prepare(

"DELETE FROM inbox_activity

1373

WHERE item_id = (SELECT id FROM inbox_items WHERE username = ?1 AND thread = ?2)

1374

AND id NOT IN (

1375

SELECT a.id FROM inbox_activity a

1376

WHERE a.item_id = (SELECT id FROM inbox_items WHERE username = ?1 AND thread = ?2)

1377

ORDER BY a.id DESC LIMIT ?3)",

1378

)

1379

.bind(&[username.into(), thread.key.as_str().into(), MAX_ACTIVITY.into()])?,

1380

);

1381

}

1382

let results = db.batch(statements).await?;

1383

#[derive(Deserialize)]

struct Told {

username: String,

}

// A line of history written means the event is news to that person.

1388

let mut news = HashSet::new();

1389

for (at, _) in told.iter().enumerate() {

1390

if let Some(result) = results.get(first + at * 3 + 1)

1391

&& let Ok(rows) = result.results::<Told>()

1392

{

1393

news.extend(rows.into_iter().map(|row| row.username));

}

}

Ok(told

.into_iter()

.filter(|notice| news.contains(&notice.username))

1399

.map(|notice| Written {

1400

notice,

1401

repo_id: wanted.repo_id.clone(),

1402

url: item_url.clone(),

1403

event_id: event.id.clone(),

})

.collect())

}

/// Brings a person's thread up with new activity, or starts it. `?1` id,

1409

/// `?2` username, `?3` thread, `?4` event id, `?5` event type, `?6` reason,

1410

/// `?7` severity, `?8` title, `?9` body, `?10` workspace, `?11` repo id,

1411

/// `?12` repo, `?13` subject, `?14` number, `?15` run id, `?16` link, `?17`

1412

/// actor, `?18` time, `?19` the time-sortable id lists order by. Nothing

1413

/// happens when the person was already told of this event. While the

1414

/// thread is unread its most urgent severity is kept; done or not, it

1415

/// comes back to the inbox. A snooze stands.

1416

pub(crate) const BUMP: &str = "INSERT INTO inbox_items (id, username, thread, event_id, event, reason, severity, title, body,

1417

workspace, repo_id, repo, subject, number, run_id, link, actor, created_at, updated_at, bumped, activity)

1418

SELECT ?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16, ?17, ?18, ?18, ?19, 1

1419

WHERE NOT EXISTS (SELECT 1 FROM inbox_activity WHERE event_id = ?4 AND username = ?2)

1420

ON CONFLICT (username, thread) DO UPDATE SET

1421

event_id = excluded.event_id,

1422

event = excluded.event,

1423

reason = excluded.reason,

1424

severity = CASE

1425

WHEN inbox_items.read_at IS NULL AND inbox_items.done_at IS NULL

1426

AND (CASE inbox_items.severity WHEN 'warning' THEN 0 WHEN 'error' THEN 1 WHEN 'success' THEN 2 ELSE 3 END)

1427

< (CASE excluded.severity WHEN 'warning' THEN 0 WHEN 'error' THEN 1 WHEN 'success' THEN 2 ELSE 3 END)

1428

THEN inbox_items.severity ELSE excluded.severity END,

1429

title = excluded.title,

1430

body = excluded.body,

1431

workspace = excluded.workspace,

1432

repo = excluded.repo,

1433

subject = COALESCE(excluded.subject, inbox_items.subject),

1434

run_id = COALESCE(excluded.run_id, inbox_items.run_id),

1435

link = COALESCE(excluded.link, inbox_items.link),

1436

actor = excluded.actor,

1437

updated_at = excluded.updated_at,

1438

bumped = excluded.bumped,

1439

activity = inbox_items.activity + 1,

read_at = NULL,

done_at = NULL";

/// What the notify service is told of an inbox item (`FeedNotification` in

1444

/// @g1t/contracts): its kind from why the person was told, the workspace

1445

/// from where it links, and the item's title and first line.

1446

pub fn live_notification(event_id: &str, notice: &Notice, url: &str, at: &str) -> serde_json::Value {

1447

let kind = match notice.reason {

1448

Reason::Agent => "agent_waiting",

1449

Reason::ReviewRequested => "approval",

1450

Reason::Mention | Reason::TeamMention => "mention",

1451

// A critical or high alert, a refused push, a report or a bypass to

1452

// review: toasted like a mention. A digest, or what came at once by

1453

// choice, is an ordinary inbox item.

1454

Reason::SecurityAlert if notice.severity != Severity::Info => "security",

1455

_ => "inbox",

1456

};

1457

let path = url.split(['?', '#']).next().unwrap_or_default();

1458

let mut segments = path.trim_start_matches('/').split('/');

1459

let workspace = segments.next().unwrap_or_default().to_lowercase();

1460

let repo = segments.next().map(|name| format!("{workspace}/{name}"));

1461

let name = repo.unwrap_or_else(|| if workspace.is_empty() { "g1t".to_owned() } else { workspace.clone() });

1462

serde_json::json!({

1463

"id": format!("inbox:{event_id}"),

1464

"kind": kind,

1465

"workspace": workspace,

1466

"title": notice.title,

1467

"body": clip(&notice.body, 140),

1468

"href": if url.starts_with('/') { url } else { "/inbox" },

1469

"actor": { "kind": "system", "id": name, "name": name },

"created_at": at,

})

}

/// What the notify service is told when a person's inbox count changes

1475

/// (its `set_inbox`): the count as it now is, never a difference, so a

1476

/// call repeated or out of order still ends right.

1477

pub fn inbox_count_update(username: &str, unread: u32) -> serde_json::Value {

1478

serde_json::json!({ "username": username.to_lowercase(), "unread": unread })

1479

}

1480

1481

/// Tells the notify service a person's unread count as it now is, so every

1482

/// tab of theirs shows it at once: after items arrive, and after marks made

1483

/// anywhere (the site, the API, MCP). Logged and left when it fails.

1484

pub async fn tell_inbox_count(db: &D1Database, notify: &Fetcher, username: &str) {

1485

let counted = counts(db, InboxCountsArgs { username: username.to_owned() }).await;

1486

let told: Result<serde_json::Value> = match counted {

1487

Ok(counts) => g1t_kit::call(notify, "set_inbox", &inbox_count_update(username, counts.unread)).await,

1488

Err(error) => Err(error),

1489

};

1490

if let Err(error) = told {

1491

worker::console_error!("inbox: live count not sent: {error}");

}

}

/// Hands items to the notify service, one call each. Logged and left when it

1496

/// cannot be reached: the inbox and email do not wait on it.

1497

pub(crate) async fn tell_live(notify: &Fetcher, items: Vec<(String, serde_json::Value)>) {

1498

#[derive(serde::Serialize)]

1499

struct NotifyArgs {

1500

username: String,

1501

notification: serde_json::Value,

1502

}

1503

for (username, notification) in items {

1504

let told: Result<serde_json::Value> = g1t_kit::call(notify, "notify", &NotifyArgs { username, notification }).await;

1505

if let Err(error) = told {

1506

worker::console_error!("inbox: live notification not sent: {error}");

}

}

}

/// Emails what was news to people who asked to be emailed for its reason.

1512

/// Identity sends each, only to a confirmed address and only if the person

1513

/// can still read the repository.

1514

async fn email(db: &D1Database, identity: &Fetcher, written: Vec<Written>) -> Result<()> {

1515

if written.is_empty() {

1516

return Ok(());

1517

}

1518

let people: Vec<String> = written

1519

.iter()

1520

.map(|item| item.notice.username.clone())

1521

.collect::<HashSet<_>>()

1522

.into_iter()

1523

.collect();

1524

let settings = subscriptions::settings_of(db, &people).await?;

1525

for item in written {

1526

let wants = settings.get(&item.notice.username).map_or(&DEFAULT_EMAIL[..], |settings| &settings.email[..]);

1527

if !wants.contains(&item.notice.reason) {

1528

continue;

1529

}

1530

let sent: Result<bool> = g1t_kit::call(

identity,

"notify_by_email",

&NotifyByEmailArgs {

username: item.notice.username.clone(),

1535

repo_id: item.repo_id.clone(),

1536

subject: item.notice.title.clone(),

1537

intro: item.notice.body.clone(),

1538

quote: None,

1539

path: item.url.clone(),

1540

reason: item.notice.reason,

},

)

.await;

if let Err(error) = sent {

1545

worker::console_error!("inbox: email to {} not sent: {error}", item.notice.username);

}

}

Ok(())

}

// --- Reading and changing ----------------------------------------------------

1552

1553

#[derive(Deserialize)]

1554

pub(crate) struct Row {

1555

pub(crate) id: String,

reason: String,

severity: String,

title: String,

body: String,

event: Option<String>,

1561

workspace: Option<String>,

1562

pub(crate) repo_id: Option<String>,

1563

repo: Option<String>,

1564

subject: Option<String>,

1565

number: Option<f64>,

1566

run_id: Option<String>,

1567

link: Option<String>,

1568

actor: Option<String>,

1569

activity: Option<f64>,

1570

created_at: String,

1571

updated_at: Option<String>,

1572

read_at: Option<String>,

1573

done_at: Option<String>,

1574

saved: f64,

1575

snoozed_until: Option<String>,

1576

bumped: Option<String>,

}

impl Row {

pub(crate) fn into_item(self) -> InboxItem {

1581

let subject = self.subject.as_deref().and_then(SubjectKind::parse);

1582

let number = self.number.map(|n| n as u32);

1583

InboxItem {

1584

url: url(self.repo.as_deref(), subject, number, self.run_id.as_deref(), self.link.as_deref()),

1585

id: self.id,

1586

reason: Reason::parse(&self.reason).unwrap_or(Reason::Subscribed),

1587

severity: Severity::parse(&self.severity).unwrap_or(Severity::Info),

title: self.title,

body: self.body,

event: self.event,

repo: self.repo,

workspace: self.workspace,

subject,

number,

actor: self.actor,

count: self.activity.map_or(1, |n| n as u32),

1597

updated_at: self.updated_at.unwrap_or_else(|| self.created_at.clone()),

1598

created_at: self.created_at,

1599

read_at: self.read_at,

1600

done_at: self.done_at,

1601

saved: self.saved != 0.0,

1602

snoozed_until: self.snoozed_until,

}

}

}

pub(crate) const COLUMNS: &str = "id, reason, severity, title, body, event, workspace, repo_id, repo, subject, number, run_id,

1608

link, actor, activity, created_at, updated_at, read_at, done_at, saved, snoozed_until, bumped";

1609

1610

/// The conditions that pick a view's items, after `username = ?1`; `?2` is now.

1611

fn view_filter(view: InboxView) -> &'static str {

1612

match view {

1613

InboxView::Inbox => "done_at IS NULL AND (snoozed_until IS NULL OR snoozed_until <= ?2)",

1614

InboxView::Saved => "saved = 1",

1615

InboxView::Done => "done_at IS NOT NULL",

}

}

/// Whether a list ranks unread warnings first: the inbox itself, unfiltered.

1620

fn ranked(a: &ListInboxArgs) -> bool {

1621

a.view == InboxView::Inbox

1622

&& a.severity.is_none()

1623

&& a.reason.is_none()

1624

&& !a.participating

1625

&& a.repo_id.is_none()

1626

&& !a.unread

1627

&& a.since.is_none()

1628

&& a.updated_before.is_none()

1629

}

1630

1631

/// The conditions a list's filters add, and the values they bind, numbered

1632

/// after the `bound` values already given.

1633

fn filters(a: &ListInboxArgs, bound: usize) -> (Vec<String>, Vec<String>) {

1634

let mut conditions = Vec::new();

1635

let mut values: Vec<String> = Vec::new();

1636

let mut bind = |value: &str, condition: &str| {

1637

values.push(value.to_owned());

1638

conditions.push(condition.replace('?', &format!("?{}", bound + values.len())));

1639

};

1640

if let Some(severity) = a.severity {

1641

bind(severity.as_str(), "severity = ?");

1642

}

1643

if let Some(reason) = a.reason {

1644

bind(reason.as_str(), "reason = ?");

1645

}

1646

if let Some(repo_id) = &a.repo_id {

1647

bind(repo_id, "repo_id = ?");

1648

}

1649

if let Some(since) = &a.since {

1650

bind(since.trim(), "COALESCE(updated_at, created_at) >= ?");

1651

}

1652

if let Some(before) = &a.updated_before {

1653

bind(before.trim(), "COALESCE(updated_at, created_at) < ?");

1654

}

1655

if a.participating {

1656

conditions.push("reason NOT IN ('manual', 'subscribed')".to_owned());

1657

}

1658

if a.unread {

1659

conditions.push("read_at IS NULL".to_owned());

}

(conditions, values)

}

pub async fn list(db: &D1Database, repos: &Fetcher, a: ListInboxArgs) -> Result<InboxPage> {

1665

let Some(viewer) = &a.viewer else {

1666

return Ok(InboxPage::default());

1667

};

1668

let username = viewer.username.to_lowercase();

1669

let now = rfc3339(now_ms());

1670

let limit = a.limit.unwrap_or(DEFAULT_INBOX_PAGE).clamp(1, MAX_INBOX_PAGE);

1671

let mut values: Vec<JsValue> = vec![username.as_str().into(), now.as_str().into()];

1672

let mut conditions = vec!["username = ?1".to_owned(), view_filter(a.view).to_owned()];

1673

let (filtered, bound) = filters(&a, values.len());

1674

conditions.extend(filtered);

1675

values.extend(bound.iter().map(|value| JsValue::from(value.as_str())));

1676

let ranked = ranked(&a);

1677

// Unread warnings lead the first page, and are left out of the rest.

1678

let leading = "severity = 'warning' AND read_at IS NULL";

1679

let mut rest = conditions.clone();

1680

let mut rest_values = values.clone();

1681

if ranked {

1682

rest.push(format!("NOT ({leading})"));

1683

}

1684

if let Some(before) = &a.before {

1685

rest_values.push(before.as_str().into());

1686

rest.push(format!("bumped < ?{}", rest_values.len()));

1687

}

1688

rest_values.push((limit + 1).into());

1689

let order = if a.view == InboxView::Done { "done_at DESC, bumped DESC" } else { "bumped DESC" };

1690

let mut statements = vec![

1691

db.prepare(format!(

1692

"SELECT {COLUMNS} FROM inbox_items WHERE {} ORDER BY {order} LIMIT ?{}",

rest.join(" AND "),

rest_values.len()

))

.bind(&rest_values)?,

1697

];

1698

if ranked && a.before.is_none() {

1699

statements.push(

1700

db.prepare(format!(

1701

"SELECT {COLUMNS} FROM inbox_items WHERE {} AND {leading} ORDER BY bumped DESC LIMIT ?3",

1702

conditions.join(" AND ")

1703

))

1704

.bind(&[username.as_str().into(), now.as_str().into(), MAX_RANKED.into()])?,

1705

);

1706

}

1707

let results = db.batch(statements).await?;

1708

let mut rows = results.first().map(|result| result.results::<Row>()).transpose()?.unwrap_or_default();

1709

let next = if rows.len() > limit as usize {

1710

rows.truncate(limit as usize);

1711

rows.last().map(|row| row.bumped.clone().unwrap_or_else(|| row.id.clone()))

} else {

None

};

let mut items: Vec<Row> = results.get(1).map(|result| result.results::<Row>()).transpose()?.unwrap_or_default();

1716

items.extend(rows);

1717

let items = readable_only(db, repos, &a.viewer, &username, items).await?;

1718

Ok(InboxPage {

1719

items: items.into_iter().map(Row::into_item).collect(),

next,

})

}

/// Rows about repositories the viewer can still read; the rest are

1725

/// removed from their inbox as they are found.

1726

pub(crate) async fn readable_only(db: &D1Database, repos: &Fetcher, viewer: &g1t_contracts::Viewer, username: &str, mut rows: Vec<Row>) -> Result<Vec<Row>> {

1727

let ids: Vec<String> = rows

1728

.iter()

1729

.filter_map(|row| row.repo_id.clone())

1730

.collect::<HashSet<_>>()

.into_iter()

.collect();

if ids.is_empty() {

return Ok(rows);

}

let readable: Vec<Repo> = g1t_kit::call(

repos,

"readable",

&ReadableArgs {

ids: ids.clone(),

viewer: viewer.clone(),

},

)

.await?;

let readable: HashSet<String> = readable.into_iter().map(|repo| repo.id).collect();

1746

let gone: Vec<String> = ids.into_iter().filter(|id| !readable.contains(id)).collect();

1747

if !gone.is_empty() {

1748

forget(db, username, &gone).await?;

1749

rows.retain(|row| row.repo_id.as_ref().is_none_or(|id| !gone.contains(id)));

}

Ok(rows)

}

/// Removes a person's items about repositories they cannot read.

1755

async fn forget(db: &D1Database, username: &str, repo_ids: &[String]) -> Result<()> {

1756

let marks = vec!["?"; repo_ids.len()].join(", ");

1757

let mut values: Vec<JsValue> = vec![username.into()];

1758

values.extend(repo_ids.iter().map(|id| JsValue::from(id.as_str())));

1759

db.batch(vec![

1760

db.prepare(format!(

1761

"DELETE FROM inbox_activity WHERE item_id IN (SELECT id FROM inbox_items WHERE username = ? AND repo_id IN ({marks}))"

1762

))

1763

.bind(&values)?,

1764

db.prepare(format!("DELETE FROM inbox_items WHERE username = ? AND repo_id IN ({marks})"))

.bind(&values)?,

])

.await?;

Ok(())

}

#[derive(Deserialize)]

struct CountRow {

severity: String,

n: f64,

}

pub async fn counts(db: &D1Database, a: InboxCountsArgs) -> Result<InboxCounts> {

1778

let rows = db

1779

.prepare(

1780

"SELECT severity, count(*) AS n FROM inbox_items

1781

WHERE username = ? AND read_at IS NULL AND done_at IS NULL

1782

AND (snoozed_until IS NULL OR snoozed_until <= ?)

1783

GROUP BY severity",

1784

)

1785

.bind(&[a.username.to_lowercase().into(), rfc3339(now_ms()).into()])?

1786

.all()

1787

.await?

1788

.results::<CountRow>()?;

1789

let mut counts = InboxCounts::default();

1790

for row in rows {

1791

let n = row.n as u32;

1792

counts.unread += n;

1793

match Severity::parse(&row.severity) {

1794

Some(Severity::Error) => counts.error += n,

1795

Some(Severity::Warning) => counts.warning += n,

1796

Some(Severity::Success) => counts.success += n,

1797

Some(Severity::Info) | None => counts.info += n,

}

}

Ok(counts)

}

/// The change a mark makes, as a `SET` clause; `?1` is now.

1804

fn mark_change(mark: InboxMark) -> &'static str {

1805

match mark {

1806

InboxMark::Read => "read_at = COALESCE(read_at, ?1)",

1807

InboxMark::Unread => "read_at = NULL",

1808

InboxMark::Done => "done_at = ?1, read_at = COALESCE(read_at, ?1)",

1809

InboxMark::Undone => "done_at = NULL",

1810

InboxMark::Save => "saved = 1",

1811

InboxMark::Unsave => "saved = 0",

1812

InboxMark::Snooze => "snoozed_until = ?2, read_at = COALESCE(read_at, ?1)",

1813

InboxMark::Unsnooze => "snoozed_until = NULL",

}

}

#[derive(Deserialize)]

struct IdRow {

#[allow(dead_code)]

id: String,

}

/// Changes the person's own items. Returns how many changed.

1824

pub async fn mark(db: &D1Database, a: MarkInboxArgs) -> Result<u32> {

1825

let now = rfc3339(now_ms());

1826

let until = match (a.mark, a.until.as_deref().map(str::trim)) {

1827

// Times compare as text (`g1t_contracts::time`); a snooze is for later.

1828

(InboxMark::Snooze, Some(until)) if until.len() == now.len() && until > now.as_str() => until.to_owned(),

1829

(InboxMark::Snooze, _) => return Ok(0),

1830

_ => String::new(),

1831

};

1832

let mut values: Vec<JsValue> = vec![now.as_str().into(), until.as_str().into(), a.username.to_lowercase().into()];

1833

let target = if a.all && a.ids.is_empty() {

1834

let mut target = "done_at IS NULL".to_owned();

1835

let mut bind = |values: &mut Vec<JsValue>, value: &str, condition: &str| {

1836

values.push(value.into());

1837

target.push_str(&format!(" AND {}", condition.replace('?', &format!("?{}", values.len()))));

1838

};

1839

if let Some(severity) = a.severity {

1840

bind(&mut values, severity.as_str(), "severity = ?");

1841

}

1842

if let Some(repo_id) = &a.repo_id {

1843

bind(&mut values, repo_id, "repo_id = ?");

1844

}

1845

if let Some(last_read_at) = a.last_read_at.as_deref().map(str::trim).filter(|at| !at.is_empty()) {

1846

bind(&mut values, last_read_at, "COALESCE(updated_at, created_at) <= ?");

}

target

} else {

let ids: Vec<&String> = a.ids.iter().take(MAX_MARK).collect();

if ids.is_empty() {

return Ok(0);

}

let first = values.len() + 1;

1855

values.extend(ids.iter().map(|id| JsValue::from(id.as_str())));

1856

let marks: Vec<String> = (first..values.len() + 1).map(|at| format!("?{at}")).collect();

1857

format!("id IN ({})", marks.join(", "))

};

let changed = db

.prepare(format!(

"UPDATE inbox_items SET {} WHERE username = ?3 AND {target} RETURNING id",

mark_change(a.mark)

))

.bind(&values)?

.all()

.await?

.results::<IdRow>()?;

1868

Ok(changed.len() as u32)

1869

}

1870

1871

#[derive(Deserialize)]

struct ActivityRow {

reason: String,

severity: String,

title: String,

body: String,

event: Option<String>,

1878

actor: Option<String>,

created_at: String,

}

/// One of the viewer's threads, with its history and their subscription.

1883

pub async fn thread(db: &D1Database, repos: &Fetcher, work: &Fetcher, a: ThreadArgs) -> Result<Option<InboxThread>> {

1884

let Some(viewer) = &a.viewer else {

1885

return Ok(None);

1886

};

1887

let username = viewer.username.to_lowercase();

1888

let row = db

1889

.prepare(format!("SELECT {COLUMNS} FROM inbox_items WHERE id = ? AND username = ?"))

1890

.bind(&[a.id.as_str().into(), username.as_str().into()])?

1891

.first::<Row>(None)

1892

.await?;

1893

let Some(row) = readable_only(db, repos, &a.viewer, &username, row.into_iter().collect()).await?.pop() else {

return Ok(None);

};

let activity = db

.prepare(

"SELECT reason, severity, title, body, event, actor, created_at FROM inbox_activity

1899

WHERE item_id = ? ORDER BY id DESC LIMIT ?",

1900

)

1901

.bind(&[row.id.as_str().into(), MAX_ACTIVITY.into()])?

1902

.all()

1903

.await?

1904

.results::<ActivityRow>()?

1905

.into_iter()

1906

.map(|row| InboxActivity {

1907

reason: Reason::parse(&row.reason).unwrap_or(Reason::Subscribed),

1908

severity: Severity::parse(&row.severity).unwrap_or(Severity::Info),

title: row.title,

body: row.body,

event: row.event,

actor: row.actor,

created_at: row.created_at,

1914

})

1915

.collect();

1916

let item = row.into_item();

1917

let subscription = match (item.subject, item.number) {

1918

(Some(SubjectKind::Issue | SubjectKind::Pull), Some(_)) => {

1919

subscriptions::subscription(

db,

work,

SubscriptionArgs {

viewer: a.viewer.clone(),

1924

id: Some(item.id.clone()),

1925

..SubscriptionArgs::default()

},

)

.await?

}

_ => None,

};

Ok(Some(InboxThread {

item,

activity,

subscription,

}))

}

// --- Keeping up ------------------------------------------------------------------

1940

1941

/// Moves rows with renamed workspaces and repositories, and drops those

1942

/// of purged repositories, deleted workspaces and purged accounts.

1943

pub async fn follow(db: &D1Database, events: &[Event]) -> Result<()> {

1944

let mut statements = Vec::new();

1945

for event in events {

1946

let text = |key: &str| event.data[key].as_str().unwrap_or_default().to_lowercase();

1947

match event.kind.as_str() {

1948

"workspace.renamed" => {

1949

let Ok(renamed) = serde_json::from_value::<WorkspaceRenamed>(event.data.clone()) else {

1950

continue;

1951

};

1952

let (from, to) = (renamed.from.to_lowercase(), renamed.to.to_lowercase());

if from == to {

continue;

}

statements.push(

db.prepare(

"UPDATE inbox_items SET repo = ?2 || substr(repo, length(?1) + 1), workspace = ?2,

1959

link = CASE WHEN link LIKE '/' || ?1 || '/%' THEN '/' || ?2 || substr(link, length(?1) + 2) ELSE link END

1960

WHERE workspace = ?1",

1961

)

1962

.bind(&[from.as_str().into(), to.as_str().into()])?,

1963

);

1964

statements.push(

1965

db.prepare("UPDATE inbox_watching SET repo = ?2 || substr(repo, length(?1) + 1) WHERE repo LIKE ?1 || '/%'")

1966

.bind(&[from.as_str().into(), to.as_str().into()])?,

);

}

"repo.renamed" => {

let path = format!("{}/{}", text("namespace"), text("to"));

1971

for table in ["inbox_items", "inbox_watching"] {

1972

statements.push(

1973

db.prepare(format!("UPDATE {table} SET repo = ? WHERE repo_id = ?"))

1974

.bind(&[path.as_str().into(), text("repoId").into()])?,

);

}

}

"repo.transferred" => {

1979

let path = format!("{}/{}", text("to"), text("name"));

1980

statements.push(

1981

db.prepare("UPDATE inbox_items SET repo = ?, workspace = ? WHERE repo_id = ?")

1982

.bind(&[path.as_str().into(), text("to").into(), text("repoId").into()])?,

1983

);

1984

statements.push(

1985

db.prepare("UPDATE inbox_watching SET repo = ? WHERE repo_id = ?")

1986

.bind(&[path.as_str().into(), text("repoId").into()])?,

);

}

"repo.purged" => {

for sql in [

"DELETE FROM inbox_activity WHERE item_id IN (SELECT id FROM inbox_items WHERE repo_id = ?)",

1992

"DELETE FROM inbox_items WHERE repo_id = ?",

1993

"DELETE FROM inbox_subscriptions WHERE repo_id = ?",

1994

"DELETE FROM inbox_watching WHERE repo_id = ?",

1995

] {

1996

statements.push(db.prepare(sql).bind(&[text("repoId").into()])?);

1997

}

1998

}

1999

"workspace.deleted" => {

2000

statements.push(

2001

db.prepare("DELETE FROM inbox_items WHERE workspace = ?")

2002

.bind(&[text("slug").into()])?,

2003

);

2004

statements.push(

2005

db.prepare("DELETE FROM inbox_watching WHERE repo LIKE ? || '/%'")

2006

.bind(&[text("slug").into()])?,

2007

);

2008

}

2009

// An account purged: its inbox, what it watched and its

2010

// settings go with it (identity's account_deletion.rs).

2011

"user.deleted" => {

2012

let username = text("username");

2013

if username.is_empty() {

continue;

}

for sql in [

"DELETE FROM inbox_activity WHERE username = ?",

2018

"DELETE FROM inbox_items WHERE username = ?",

2019

"DELETE FROM inbox_subscriptions WHERE username = ?",

2020

"DELETE FROM inbox_watching WHERE username = ?",

2021

"DELETE FROM inbox_settings WHERE username = ?",

2022

] {

2023

statements.push(db.prepare(sql).bind(&[username.as_str().into()])?);

}

}

_ => {}

}

}

if !statements.is_empty() {

2030

db.batch(statements).await?;

}

Ok(())

}

/// Removes items done more than [`DONE_DAYS`] ago, and any not saved older

2036

/// than [`MAX_DAYS`], with their history. Returns how many went.

2037

pub async fn purge(db: &D1Database, now: u64) -> Result<u32> {

2038

let done = crate::audit::keep_from(now, DONE_DAYS);

2039

let oldest = crate::audit::keep_from(now, MAX_DAYS);

2040

let statements: Vec<D1PreparedStatement> = vec![

2041

db.prepare("DELETE FROM inbox_items WHERE saved = 0 AND done_at < ?")

2042

.bind(&[done.into()])?,

2043

db.prepare("DELETE FROM inbox_items WHERE saved = 0 AND COALESCE(updated_at, created_at) < ?")

2044

.bind(&[oldest.into()])?,

2045

db.prepare("DELETE FROM inbox_activity WHERE NOT EXISTS (SELECT 1 FROM inbox_items WHERE inbox_items.id = inbox_activity.item_id)"),

2046

];

2047

let results = db.batch(statements).await?;

2048

let mut removed = 0;

2049

for result in results.into_iter().take(2) {

2050

removed += result.meta()?.and_then(|meta| meta.changes).unwrap_or(0) as u32;

}

Ok(removed)

}

#[cfg(test)]

mod tests {

use super::*;

use crate::subscriptions::{State, Subscription, Watcher};

2059

use serde_json::json;

2060

2061

fn event(kind: &str, actor: Option<&str>, data: serde_json::Value) -> Event {

Event {

id: "evt_1".into(),

kind: kind.into(),

source: "work".into(),

2066

time: "2026-10-07T12:00:00.000Z".into(),

2067

repo_id: Some("rep_1".into()),

2068

actor: actor.map(str::to_owned),

data,

}

}

fn person(id: &str, username: &str) -> Principal {

2074

Principal {

2075

id: id.into(),

2076

username: username.into(),

}

}

fn actor(id: &str, username: &str) -> Actor {

2081

Actor {

2082

id: Some(id.into()),

2083

username: Some(username.into()),

}

}

/// A pull request ana opened, assigned to bo, for an issue cy filed and dee is assigned.

2088

fn pull() -> InboxSubject {

2089

InboxSubject {

2090

kind: Some(SubjectKind::Pull),

2091

title: "Add the inbox".into(),

2092

author: person("usr_ana", "ana"),

2093

assignees: vec!["bo".into()],

2094

issue: Some(Box::new(InboxSubject {

2095

kind: Some(SubjectKind::Issue),

2096

title: "An inbox".into(),

2097

author: person("usr_cy", "cy"),

2098

assignees: vec!["dee".into()],

2099

..InboxSubject::default()

2100

})),

2101

..InboxSubject::default()

}

}

/// A change g1t made for ana.

2106

fn g1t_pull() -> InboxSubject {

2107

InboxSubject {

2108

author: person(AGENT_ID, "g1t"),

2109

requested_by: Some(person("usr_ana", "ana")),

..pull()

}

}

fn nobody() -> Audience {

Audience::default()

}

fn told(notices: &[Notice]) -> Vec<(&str, Reason, Severity)> {

2119

notices

2120

.iter()

2121

.map(|notice| (notice.username.as_str(), notice.reason, notice.severity))

.collect()

}

fn comment(author: Principal, body: &str, mentions: &[&str]) -> InboxSubject {

2126

InboxSubject {

2127

comment: Some(InboxComment {

2128

author,

2129

excerpt: body.into(),

2130

mentions: mentions.iter().map(|name| (*name).to_owned()).collect(),

2131

..InboxComment::default()

}),

..pull()

}

}

fn subscribed(rows: &[(&str, State, Option<Reason>)]) -> Audience {

Audience {

subscriptions: rows

.iter()

.map(|(name, state, reason)| Subscription {

2142

username: (*name).into(),

state: *state,

reason: *reason,

})

.collect(),

watchers: Vec::new(),

}

}

fn watched(rows: &[(&str, WatchLevel, &[&str])]) -> Audience {

2152

Audience {

2153

subscriptions: Vec::new(),

2154

watchers: rows

2155

.iter()

2156

.map(|(name, level, events)| Watcher {

2157

username: (*name).into(),

2158

level: *level,

2159

events: events.iter().map(|kind| (*kind).to_owned()).collect(),

})

.collect(),

}

}

#[test]

fn only_events_that_tell_someone_are_read() {

2167

let asked = |kind: &str, data| wants(&event(kind, None, data));

2168

assert_eq!(

2169

asked("checks.completed", json!({ "repoId": "rep_1", "number": 4, "status": "failed" })),

2170

Some(Wanted { repo_id: "rep_1".into(), number: Some(4), comment_id: None })

2171

);

2172

assert_eq!(asked("checks.completed", json!({ "repoId": "rep_1", "number": 4, "status": "passed" })), None);

2173

assert_eq!(asked("review.completed", json!({ "repoId": "rep_1", "number": 4 })), None);

2174

assert_eq!(asked("workflow.completed", json!({ "repoId": "rep_1", "conclusion": "success" })), None);

2175

assert_eq!(

2176

asked("workflow.completed", json!({ "repoId": "rep_1", "conclusion": "failure" })),

2177

Some(Wanted { repo_id: "rep_1".into(), number: None, comment_id: None })

2178

);

2179

assert_eq!(

2180

asked("comment.created", json!({ "repoId": "rep_1", "number": 2, "commentId": "cmt_1" })).map(|w| w.comment_id),

2181

Some(Some("cmt_1".into()))

2182

);

2183

assert_eq!(asked("git.push", json!({ "repoId": "rep_1" })), None);

2184

for kind in ["issue.opened", "pull.review_requested", "pull.stalled", "issue.assigned", "pull.closed", "issue.reopened"] {

2185

assert!(asked(kind, json!({ "repoId": "rep_1", "number": 1 })).is_some(), "{kind}");

2186

}

2187

assert_eq!(asked("deployment.failed", json!({ "repoId": "rep_1" })).map(|w| w.number), Some(None));

}

#[test]

fn a_waiting_or_stopped_agent_needs_the_pull_requests_and_the_issues_people_first() {

2192

let asked = event("agent.asked", Some("usr_agent"), json!({ "number": 7 }));

2193

let notices_ = notices(&asked, "acme/rocket", &Actor::default(), Some(&pull()), &nobody());

assert_eq!(

told(&notices_),

vec![

("ana", Reason::Agent, Severity::Warning),

2198

("cy", Reason::Agent, Severity::Warning),

2199

("dee", Reason::Agent, Severity::Warning),

2200

]

2201

);

2202

assert_eq!(notices_[0].title, "An agent is waiting on acme/rocket#7");

2203

assert_eq!(notices_[0].body, "Add the inbox");

2204

// Stopping says why; even someone who unsubscribed hears of it.

2205

let stalled = event("pull.stalled", None, json!({ "number": 7, "detail": "Its checks could not be run." }));

2206

let audience = subscribed(&[("ana", State::Unsubscribed, None)]);

2207

let notices_ = notices(&stalled, "acme/rocket", &Actor::default(), Some(&g1t_pull()), &audience);

2208

assert_eq!(notices_[0].username, "ana");

2209

assert_eq!(notices_[0].title, "g1t stopped on acme/rocket#7 and needs you");

2210

assert_eq!(notices_[0].body, "Its checks could not be run.");

2211

// Ignoring the thread silences even that.

2212

let audience = subscribed(&[("ana", State::Ignored, None)]);

2213

let notices_ = notices(&stalled, "acme/rocket", &Actor::default(), Some(&g1t_pull()), &audience);

2214

assert!(!notices_.iter().any(|notice| notice.username == "ana"));

}

#[test]

fn whatever_an_agent_waits_on_closes_when_it_goes_on() {

2219

for kind in RESUMES {

2220

assert_eq!(resolves(&event(kind, None, json!({ "repoId": "rep_1", "number": 7 }))), Some("rep_1#7".into()));

2221

}

2222

assert_eq!(resolves(&event("comment.created", None, json!({ "repoId": "rep_1", "number": 7 }))), None);

}

#[test]

fn a_deployments_reviewers_are_told() {

2227

let asked = event(

2228

"deployment.review_requested",

2229

Some("usr_ana"),

2230

json!({

2231

"repoId": "rep_1", "runId": "run_7", "environment": "production", "workflow": "Deploy",

2232

"title": "Ship it", "notify": ["cy", "ana"], "link": "/acme/rocket/actions/runs/run_7"

2233

}),

2234

);

2235

let wanted = wants(&asked).unwrap();

2236

assert_eq!(wanted.number, None);

2237

let thread = thread_of(&asked, &wanted, None);

2238

assert_eq!(thread.key, "rep_1/review/run_7/production");

2239

assert_eq!(thread.link.as_deref(), Some("/acme/rocket/actions/runs/run_7"));

2240

let told = notices(&asked, "acme/rocket", &actor("usr_ana", "ana"), None, &nobody());

2241

let names: Vec<&str> = told.iter().map(|notice| notice.username.as_str()).collect();

2242

// Whoever started the run is told too, when they review it.

2243

assert_eq!(names, ["cy", "ana"]);

2244

assert!(told[0].title.contains("waiting for your review to deploy to production"));

}

#[test]

fn teams_asked_to_review_tell_the_people_they_name() {

2249

let requested = event(

2250

"pull.review_requested",

2251

Some("usr_ana"),

2252

json!({ "number": 7, "reviewers": ["cy"], "teams": [

2253

{ "team": "acme/backend", "notified": ["cy", "dee", "ana"], "assigned": ["cy"] }

2254

] }),

2255

);

2256

let notices_ = notices(&requested, "acme/rocket", &actor("usr_ana", "ana"), Some(&pull()), &nobody());

2257

// cy was picked and is told as a reviewer; dee through the team;

2258

// never whoever asked.

2259

assert_eq!(

2260

told(&notices_),

2261

vec![("cy", Reason::ReviewRequested, Severity::Warning), ("dee", Reason::ReviewRequested, Severity::Warning)]

2262

);

2263

assert_eq!(notices_[0].title, "ana asked you to review acme/rocket#7");

2264

assert_eq!(notices_[1].title, "ana asked @acme/backend to review acme/rocket#7");

2265

assert_eq!(

2266

subscribes(&requested, Some(&pull())),

2267

vec![("cy".into(), Reason::ReviewRequested), ("dee".into(), Reason::ReviewRequested), ("ana".into(), Reason::ReviewRequested)]

2268

);

2269

// Asked by the CODEOWNERS file.

2270

let owned = event(

2271

"pull.review_requested",

2272

None,

2273

json!({ "number": 7, "reviewers": ["bo"], "codeOwners": true, "teams": [{ "team": "acme/docs", "notified": ["wren"], "assigned": [] }] }),

2274

);

2275

let notices_ = notices(&owned, "acme/rocket", &Actor::default(), Some(&pull()), &nobody());

2276

assert_eq!(notices_[0].title, "acme/rocket#7 changes files you own");

2277

assert_eq!(notices_[1].title, "acme/rocket#7 changes files @acme/docs owns");

}

#[test]

fn a_team_mention_tells_its_people_once_and_a_name_wins() {

2282

let mut on = comment(person("usr_bo", "bo"), "cc @ana @acme/backend", &["ana"]);

2283

if let Some(comment) = on.comment.as_mut() {

2284

comment.team_mentions = vec![TeamMentioned {

2285

team: "acme/backend".into(),

2286

members: vec!["ana".into(), "cy".into(), "bo".into()],

2287

}];

2288

}

2289

let created = event("comment.created", Some("usr_bo"), json!({ "number": 7, "commentId": "cmt_1" }));

2290

let notices_ = notices(&created, "acme/rocket", &actor("usr_bo", "bo"), Some(&on), &nobody());

2291

2292

assert_eq!(mentioned, vec![("ana", Reason::Mention), ("cy", Reason::TeamMention)]);

2293

assert_eq!(

2294

notices_.iter().find(|n| n.username == "cy").unwrap().title,

2295

"bo mentioned @acme/backend on acme/rocket#7"

2296

);

2297

let subscribed = subscribes(&created, Some(&on));

2298

assert!(subscribed.contains(&("cy".into(), Reason::TeamMention)));

2299

assert!(subscribed.contains(&("ana".into(), Reason::Mention)));

}

#[test]

fn a_description_tells_the_people_and_teams_it_mentions() {

2304

let mut on = pull();

2305

on.mentions = vec!["dee".into()];

2306

on.team_mentions = vec![TeamMentioned {

2307

team: "acme/web".into(),

2308

members: vec!["eve".into()],

2309

}];

2310

let opened = event("pull.opened", Some("usr_ana"), json!({ "number": 7 }));

2311

let notices_ = notices(&opened, "acme/rocket", &actor("usr_ana", "ana"), Some(&on), &nobody());

2312

assert!(told(&notices_).contains(&("dee", Reason::Mention, Severity::Info)));

2313

assert!(told(&notices_).contains(&("eve", Reason::TeamMention, Severity::Info)));

2314

assert_eq!(

2315

subscribes(&opened, Some(&on)),

2316

vec![("dee".into(), Reason::Mention), ("eve".into(), Reason::TeamMention)]

);

}

#[test]

fn reviewers_and_assignees_asked_are_told_never_whoever_asked() {

2322

let requested = event("pull.review_requested", Some("usr_ana"), json!({ "number": 7, "reviewers": ["bo", "g1t", "ana"] }));

2323

let notices_ = notices(&requested, "acme/rocket", &actor("usr_ana", "ana"), Some(&pull()), &nobody());

2324

assert_eq!(told(&notices_), vec![("bo", Reason::ReviewRequested, Severity::Warning)]);

2325

assert_eq!(notices_[0].title, "ana asked you to review acme/rocket#7");

2326

let assigned = event("issue.assigned", Some("usr_ana"), json!({ "number": 3, "added": ["ana", "eve"] }));

2327

let notices_ = notices(&assigned, "acme/rocket", &actor("usr_ana", "ana"), Some(&pull()), &nobody());

2328

assert_eq!(told(&notices_), vec![("eve", Reason::Assign, Severity::Info)]);

2329

assert_eq!(notices_[0].title, "ana assigned you to acme/rocket#3");

2330

// Both subscribe whoever they name, never g1t.

2331

assert_eq!(subscribes(&requested, Some(&pull())), vec![("bo".into(), Reason::ReviewRequested), ("ana".into(), Reason::ReviewRequested)]);

2332

assert_eq!(subscribes(&assigned, Some(&pull())), vec![("ana".into(), Reason::Assign), ("eve".into(), Reason::Assign)]);

}

#[test]

fn failures_go_to_whoever_answers_for_the_change() {

2337

let failed = event("checks.completed", None, json!({ "number": 7, "status": "failed" }));

2338

assert_eq!(

2339

told(&notices(&failed, "acme/rocket", &Actor::default(), Some(&pull()), &nobody())),

2340

vec![("ana", Reason::CiActivity, Severity::Error)]

2341

);

2342

// g1t's change is the person's who asked for it, never g1t's.

2343

let notices_ = notices(&failed, "acme/rocket", &Actor::default(), Some(&g1t_pull()), &nobody());

2344

assert_eq!(told(&notices_), vec![("ana", Reason::CiActivity, Severity::Error)]);

2345

let errored = event("checks.completed", None, json!({ "number": 7, "status": "errored" }));

2346

assert_eq!(

2347

notices(&errored, "acme/rocket", &Actor::default(), Some(&pull()), &nobody())[0].title,

2348

"Checks could not run on acme/rocket#7"

);

}

#[test]

fn a_workflow_that_fails_tells_its_pull_requests_owner_even_if_they_pushed() {

2354

let failed = event("workflow.completed", Some("usr_ana"), json!({ "pull": 7, "workflow": "CI", "conclusion": "failure" }));

2355

let notices_ = notices(&failed, "acme/rocket", &actor("usr_ana", "ana"), Some(&pull()), &nobody());

2356

assert_eq!(told(&notices_), vec![("ana", Reason::CiActivity, Severity::Error)]);

2357

assert_eq!(notices_[0].title, "CI failed on acme/rocket#7");

2358

// On a branch: whoever pushed.

2359

let pushed = event(

2360

"workflow.completed",

2361

Some("usr_bo"),

2362

json!({ "workflow": "Deploy", "path": ".g1t/workflows/deploy.yml", "conclusion": "failure", "ref": "refs/heads/main", "number": 12, "sha": "abcdef0123", "runId": "run_9" }),

2363

);

2364

let notices_ = notices(&pushed, "acme/rocket", &actor("usr_bo", "bo"), None, &nobody());

2365

assert_eq!(told(&notices_), vec![("bo", Reason::CiActivity, Severity::Error)]);

2366

assert_eq!(notices_[0].title, "Deploy failed on main in acme/rocket");

2367

assert_eq!(notices_[0].body, "Run 12 at abcdef0");

2368

// Nobody to tell when g1t pushed.

2369

assert!(notices(&pushed, "acme/rocket", &actor("g1t", "g1t"), None, &nobody()).is_empty());

2370

// Every failure of a workflow on a branch is one thread.

2371

let wanted = wants(&pushed).unwrap();

2372

let thread = thread_of(&pushed, &wanted, None);

2373

assert_eq!(thread.key, "rep_1/run/.g1t/workflows/deploy.yml@main");

2374

assert_eq!(thread.run_id.as_deref(), Some("run_9"));

}

#[test]

fn deployments_tell_whoever_answers_for_them_and_watchers() {

let failed = event(

"deployment.failed",

Some("usr_bo"),

json!({ "repoId": "rep_1", "projectId": "prj_1", "project": "rocket", "deploymentId": "dpl_1", "error": "The build failed.", "path": "/acme/rocket/deployments/dpl_1" }),

2383

);

2384

let audience = watched(&[("cy", WatchLevel::Custom, &["deployments"]), ("dee", WatchLevel::Custom, &["issues"]), ("eve", WatchLevel::All, &[])]);

2385

let notices_ = notices(&failed, "acme/rocket", &actor("usr_bo", "bo"), None, &audience);

assert_eq!(

told(&notices_),

vec![

("bo", Reason::CiActivity, Severity::Error),

2390

("cy", Reason::Subscribed, Severity::Error),

2391

("eve", Reason::Subscribed, Severity::Error),

2392

]

2393

);

2394

assert_eq!(notices_[0].title, "Production of rocket failed to deploy");

2395

assert_eq!(notices_[0].body, "The build failed.");

2396

let thread = thread_of(&failed, &wants(&failed).unwrap(), None);

2397

assert_eq!(thread.key, "rep_1/deploy/prj_1/production");

2398

assert_eq!(url(Some("acme/rocket"), thread.kind, None, None, thread.link.as_deref()), "/acme/rocket/deployments/dpl_1");

2399

// A success is news to the owner only after a failure.

2400

let live = event("deployment.succeeded", Some("usr_bo"), json!({ "repoId": "rep_1", "projectId": "prj_1", "project": "rocket", "commit": "abcdef0123" }));

2401

assert!(notices(&live, "acme/rocket", &actor("usr_bo", "bo"), None, &nobody()).is_empty());

2402

let recovered = event("deployment.succeeded", Some("usr_bo"), json!({ "repoId": "rep_1", "projectId": "prj_1", "project": "rocket", "recovered": true }));

2403

let notices_ = notices(&recovered, "acme/rocket", &actor("usr_bo", "bo"), None, &nobody());

2404

assert_eq!(told(&notices_), vec![("bo", Reason::CiActivity, Severity::Success)]);

2405

assert_eq!(notices_[0].title, "Production of rocket is live again");

2406

// A preview's is the pull request's owner's.

2407

let preview = event("deployment.failed", None, json!({ "repoId": "rep_1", "projectId": "prj_1", "number": 7, "triggeredBy": "g1t" }));

2408

assert_eq!(told(&notices(&preview, "acme/rocket", &Actor::default(), Some(&pull()), &nobody())), vec![("ana", Reason::CiActivity, Severity::Error)]);

2409

assert_eq!(thread_of(&preview, &wants(&preview).unwrap(), Some(&pull())).key, "rep_1/deploy/prj_1/7");

}

#[test]

fn a_mirror_tells_only_the_people_its_settings_name() {

2414

let down = event(

2415

"mirror.unreachable",

2416

None,

2417

json!({

2418

"repoId": "rep_1", "repo": "acme/rocket", "remoteId": "rmt_1", "remote": "github.com/acme/rocket",

2419

"state": "standby", "title": "github.com/acme/rocket is not answering. acme/rocket keeps its copy.",

2420

"detail": "Take over acme/rocket to keep working on g1t until it is back.",

2421

"notify": ["ana"], "link": "/acme/rocket/settings/mirroring"

2422

}),

2423

);

2424

assert_eq!(wants(&down), Some(Wanted { repo_id: "rep_1".into(), number: None, comment_id: None }));

2425

// Watchers of everything hear nothing: only those named.

2426

let audience = watched(&[("eve", WatchLevel::All, &[])]);

2427

let notices_ = notices(&down, "acme/rocket", &Actor::default(), None, &audience);

2428

assert_eq!(told(&notices_), vec![("ana", Reason::StateChange, Severity::Warning)]);

2429

assert!(notices_[0].body.contains("Take over"));

2430

let thread = thread_of(&down, &wants(&down).unwrap(), None);

2431

assert_eq!(thread.key, "rep_1/mirror");

2432

assert_eq!(url(Some("acme/rocket"), thread.kind, None, None, thread.link.as_deref()), "/acme/rocket/settings/mirroring");

2433

// The banner-only default names nobody, so nobody is told.

2434

let quiet = event("mirror.unreachable", None, json!({ "repoId": "rep_1", "title": "x", "link": "/x" }));

2435

assert!(notices(&quiet, "acme/rocket", &Actor::default(), None, &audience).is_empty());

2436

let back = event("mirror.state_changed", None, json!({ "repoId": "rep_1", "state": "standby", "title": "handed back", "notify": ["ana"], "link": "/x" }));

2437

assert_eq!(told(&notices(&back, "acme/rocket", &Actor::default(), None, &audience)), vec![("ana", Reason::StateChange, Severity::Success)]);

2438

}

2439

2440

/// A scan's summary, as the security service publishes it.

2441

fn summary(alert_type: &str, opened: &[&str], new: (u32, u32, u32, u32), open: (u32, u32, u32, u32), pusher: Option<&str>) -> Event {

2442

let counts = |(critical, high, medium, low): (u32, u32, u32, u32)| json!({ "critical": critical, "high": high, "medium": medium, "low": low, "unknown": 0 });

2443

let worst = [(new.0, "critical"), (new.1, "high"), (new.2, "medium"), (new.3, "low")].into_iter().find(|(n, _)| *n > 0).map(|(_, name)| name);

2444

let page = match alert_type {

2445

"secret_scanning" => "secret-scanning",

2446

"code_scanning" => "code-scanning",

2447

_ => "vulnerabilities",

2448

};

2449

event(

2450

&format!("{alert_type}_alert.summary"),

2451

None,

2452

json!({

2453

"repoId": "rep_1", "alertType": alert_type, "scan": if pusher.is_some() { "push" } else { "dependencies" },

2454

"opened": opened, "openedCount": opened.len(), "fixedCount": 0, "dismissedCount": 0,

2455

"newBySeverity": counts(new), "open": counts(open), "worst": worst,

2456

"latest": if opened.is_empty() { serde_json::Value::Null } else { json!("lodash 4.17.20 in package-lock.json: GHSA-1") },

2457

"link": format!("/acme/rocket/security/{page}"),

2458

"pusher": pusher, "blocked": if pusher.is_some() { opened.len() } else { 0 },

2459

"notify": ["ana"], "members": ["ana", "bo", "cy"]

}),

)

}

#[test]

fn a_scan_is_one_item_per_repository_and_alert_type() {

2466

let found = summary("vulnerability", &["vul_3", "vul_2", "vul_1"], (1, 2, 0, 0), (1, 2, 0, 0), None);

2467

assert_eq!(wants(&found), Some(Wanted { repo_id: "rep_1".into(), number: None, comment_id: None }));

2468

// cy watches for security alerts and is a member; eve watches

2469

// everything but cannot see findings; dee watches issues only.

2470

let audience = watched(&[

2471

("cy", WatchLevel::Custom, &["security"]),

2472

("dee", WatchLevel::Custom, &["issues"]),

2473

("eve", WatchLevel::All, &[]),

2474

]);

2475

let notices_ = notices(&found, "acme/rocket", &Actor::default(), None, &audience);

2476

assert_eq!(told(&notices_), vec![("ana", Reason::SecurityAlert, Severity::Error), ("cy", Reason::SecurityAlert, Severity::Error)]);

2477

assert_eq!(notices_[0].title, "3 new vulnerabilities in acme/rocket: 1 critical, 2 high");

2478

assert_eq!(notices_[0].body, "lodash 4.17.20 in package-lock.json: GHSA-1");

2479

// The same thread for every scan of this type, however many alerts.

2480

let thread = thread_of(&found, &wants(&found).unwrap(), None);

2481

assert_eq!(thread.key, "rep_1/security/vulnerability");

2482

assert_eq!(url(Some("acme/rocket"), thread.kind, None, None, thread.link.as_deref()), "/acme/rocket/security/vulnerabilities");

2483

// Medium and low are news, but not urgent: an info item, or the digest.

2484

let quiet = summary("code_scanning", &["cod_1"], (0, 0, 1, 0), (0, 0, 1, 0), None);

2485

let notices_ = notices(&quiet, "acme/rocket", &Actor::default(), None, &audience);

2486

assert_eq!(told(&notices_), vec![("ana", Reason::SecurityAlert, Severity::Info), ("cy", Reason::SecurityAlert, Severity::Info)]);

2487

assert_eq!(notices_[0].title, "1 new code scanning alert in acme/rocket: 1 medium");

2488

// A scan that opened nothing tells nobody: the thread is brought up to date instead.

2489

let fixed = summary("vulnerability", &[], (0, 0, 0, 0), (0, 1, 0, 0), None);

2490

assert!(notices(&fixed, "acme/rocket", &Actor::default(), None, &audience).is_empty());

2491

let (title, body) = refreshed_text("acme/rocket", &SecuritySummary { fixed_count: 2, ..security_summary(&fixed).unwrap() });

2492

assert_eq!(title, "1 open vulnerability in acme/rocket: 1 high");

2493

assert_eq!(body, "2 fixed");

2494

let settled = SecuritySummary { fixed_count: 1, dismissed_count: 1, open: SeverityCounts::default(), ..security_summary(&fixed).unwrap() };

2495

assert_eq!(refreshed_text("acme/rocket", &settled), ("No open vulnerabilities in acme/rocket".to_owned(), "1 fixed, 1 dismissed".to_owned()));

2496

// Per-alert events are for webhooks.

2497

assert_eq!(wants(&event("vulnerability_alert.created", None, json!({ "repoId": "rep_1" }))), None);

2498

assert_eq!(wants(&event("code_scanning_alert.fixed", None, json!({ "repoId": "rep_1" }))), None);

}

#[test]

fn a_refused_push_tells_the_pusher_at_once_and_the_owners_once() {

2503

let blocked = summary("secret_scanning", &["sec_1", "sec_2"], (2, 0, 0, 0), (2, 0, 0, 0), Some("bo"));

2504

let notices_ = notices(&blocked, "acme/rocket", &Actor::default(), None, &nobody());

2505

assert_eq!(told(&notices_), vec![("bo", Reason::SecurityAlert, Severity::Error), ("ana", Reason::SecurityAlert, Severity::Error)]);

2506

assert_eq!(notices_[0].title, "A push to acme/rocket was blocked: it adds 2 secrets");

2507

assert_eq!(notices_[1].title, "2 new secrets in acme/rocket: 2 critical");

2508

assert_eq!(thread_of(&blocked, &wants(&blocked).unwrap(), None).key, "rep_1/security/secret_scanning");

2509

// Whatever bo chose, the push is theirs to fix; ana, an owner, is told whatever she chose.

2510

let summary_ = security_summary(&blocked).unwrap();

2511

let off = InboxSettings { security_alerts: false, security_digest: DigestCadence::Off, ..InboxSettings::default() };

2512

assert_eq!(security_delivery(&summary_, "bo", &off), Delivery::Now);

2513

assert_eq!(security_delivery(&summary_, "ana", &off), Delivery::Now);

2514

// A watcher who turned security alerts off hears of nothing urgent.

2515

assert_eq!(security_delivery(&summary_, "cy", &off), Delivery::Never);

2516

assert_eq!(security_delivery(&summary_, "cy", &InboxSettings::default()), Delivery::Now);

}

#[test]

fn what_is_not_urgent_waits_for_the_digest_unless_the_person_chose_otherwise() {

2521

let medium = security_summary(&summary("vulnerability", &["vul_1"], (0, 0, 1, 0), (0, 0, 1, 0), None)).unwrap();

2522

assert!(!medium.urgent());

2523

let choose = |cadence| InboxSettings { security_digest: cadence, ..InboxSettings::default() };

2524

// Owners too: a medium alert is for the digest, as they chose.

2525

assert_eq!(security_delivery(&medium, "ana", &choose(DigestCadence::Daily)), Delivery::Digest);

2526

assert_eq!(security_delivery(&medium, "cy", &choose(DigestCadence::Weekly)), Delivery::Digest);

2527

assert_eq!(security_delivery(&medium, "cy", &choose(DigestCadence::AtOnce)), Delivery::Now);

2528

assert_eq!(security_delivery(&medium, "ana", &choose(DigestCadence::Off)), Delivery::Never);

2529

let high = security_summary(&summary("vulnerability", &["vul_1"], (0, 1, 0, 0), (0, 1, 0, 0), None)).unwrap();

2530

assert_eq!(security_delivery(&high, "ana", &choose(DigestCadence::Off)), Delivery::Now);

2531

assert_eq!(breakdown(&SeverityCounts { critical: 1, low: 3, ..SeverityCounts::default() }), "1 critical, 3 low");

2532

assert_eq!(breakdown(&SeverityCounts::default()), "");

}

#[test]

fn bypass_requests_reports_and_advisories_tell_who_they_name() {

2537

let audience = watched(&[("cy", WatchLevel::Custom, &["security"]), ("eve", WatchLevel::All, &[])]);

2538

// A bypass request goes to the reviewers named, never to watchers.

2539

let requested = event(

2540

"secret_scanning.bypass_requested",

2541

Some("usr_bo"),

2542

json!({ "repoId": "rep_1", "alertId": "sec_1", "requestId": "byp_1", "title": "An AWS access key in a.env", "notify": ["ana"], "link": "/acme/-/security/bypass-requests" }),

2543

);

2544

let notices_ = notices(&requested, "acme/rocket", &actor("usr_bo", "bo"), None, &audience);

2545

assert_eq!(told(&notices_), vec![("ana", Reason::SecurityAlert, Severity::Warning)]);

2546

assert_eq!(notices_[0].title, "bo asked to bypass push protection in acme/rocket");

2547

assert_eq!(thread_of(&requested, &wants(&requested).unwrap(), None).key, "rep_1/security/byp_1");

2548

// A private report is for the owners and security managers, at once, with no reporter named.

2549

let reported = event(

2550

"security_advisory.reported",

2551

None,

2552

json!({ "repoId": "rep_1", "id": "vrp_1", "title": "XSS in the search box", "state": "new", "link": "/acme/rocket/security/reports", "notify": ["ana"], "members": ["ana", "cy"] }),

2553

);

2554

let notices_ = notices(&reported, "acme/rocket", &Actor::default(), None, &audience);

2555

assert_eq!(told(&notices_), vec![("ana", Reason::SecurityAlert, Severity::Warning)]);

2556

assert_eq!(notices_[0].title, "A vulnerability was reported privately to acme/rocket");

2557

assert_eq!(thread_of(&reported, &wants(&reported).unwrap(), None).key, "rep_1/security/reports");

2558

// A published advisory is for watchers who chose security; it is public, so no members list is needed.

2559

let published = event(

2560

"security_advisory.published",

2561

Some("usr_ana"),

2562

json!({ "repoId": "rep_1", "id": "adv_1", "g1tsaId": "G1TSA-2026-abcd-efgh", "title": "XSS in the search box", "severity": "high", "state": "published", "link": "/acme/rocket/security/advisories/G1TSA-2026-abcd-efgh" }),

2563

);

2564

let notices_ = notices(&published, "acme/rocket", &actor("usr_ana", "ana"), None, &audience);

2565

assert_eq!(told(&notices_), vec![("cy", Reason::Subscribed, Severity::Info), ("eve", Reason::Subscribed, Severity::Info)]);

2566

assert_eq!(thread_of(&published, &wants(&published).unwrap(), None).key, "rep_1/security/advisories/adv_1");

}

#[test]

fn nobody_hears_of_what_they_did_themselves() {

2571

let merged = event("pull.merged", Some("usr_ana"), json!({ "number": 7 }));

2572

let by_ana = notices(&merged, "acme/rocket", &actor("usr_ana", "ana"), Some(&pull()), &nobody());

2573

// The assignee still hears; ana, who merged, does not.

2574

assert_eq!(told(&by_ana), vec![("bo", Reason::StateChange, Severity::Success)]);

2575

let by_bo = notices(&merged, "acme/rocket", &actor("usr_bo", "bo"), Some(&pull()), &nobody());

2576

assert_eq!(told(&by_bo), vec![("ana", Reason::StateChange, Severity::Success)]);

2577

assert_eq!(by_bo[0].title, "bo merged acme/rocket#7");

2578

let by_queue = notices(&merged, "acme/rocket", &Actor::default(), Some(&pull()), &nobody());

2579

assert_eq!(by_queue[0].title, "acme/rocket#7 was merged");

}

#[test]

fn closing_and_reopening_tells_everyone_subscribed_and_watchers() {

2584

let closed = event("issue.closed", Some("usr_bo"), json!({ "number": 3, "reason": "completed" }));

2585

let issue = InboxSubject {

2586

kind: Some(SubjectKind::Issue),

2587

title: "Crash".into(),

2588

author: person("usr_cy", "cy"),

2589

assignees: vec!["bo".into()],

2590

..InboxSubject::default()

2591

};

2592

let mut audience = subscribed(&[("eve", State::Subscribed, Some(Reason::Comment)), ("fay", State::Unsubscribed, None)]);

2593

audience.watchers = watched(&[("gus", WatchLevel::All, &[]), ("hal", WatchLevel::Custom, &["pulls"])]).watchers;

2594

let notices_ = notices(&closed, "acme/rocket", &actor("usr_bo", "bo"), Some(&issue), &audience);

assert_eq!(

told(&notices_),

vec![

("cy", Reason::StateChange, Severity::Info),

2599

("eve", Reason::StateChange, Severity::Info),

2600

("gus", Reason::Subscribed, Severity::Info),

2601

]

2602

);

2603

assert_eq!(notices_[0].title, "bo closed acme/rocket#3");

2604

let by_pull = event("issue.closed", None, json!({ "number": 3, "resolvedBy": 9 }));

2605

assert_eq!(notices(&by_pull, "acme/rocket", &Actor::default(), Some(&issue), &nobody())[0].title, "acme/rocket#3 was closed by #9");

2606

let reopened = event("issue.reopened", Some("usr_cy"), json!({ "number": 3 }));

2607

assert_eq!(told(&notices(&reopened, "acme/rocket", &actor("usr_cy", "cy"), Some(&issue), &nobody())), vec![("bo", Reason::StateChange, Severity::Info)]);

}

#[test]

fn opening_tells_who_it_names_and_watchers_of_its_kind() {

2612

let opened = event("pull.opened", Some("usr_ana"), json!({ "number": 7 }));

2613

let subject = InboxSubject { reviewers: vec!["cy".into(), "g1t".into()], ..pull() };

2614

let audience = watched(&[("bo", WatchLevel::All, &[]), ("dee", WatchLevel::Custom, &["issues"]), ("eve", WatchLevel::Custom, &["pulls"]), ("fay", WatchLevel::Participating, &[])]);

2615

let notices_ = notices(&opened, "acme/rocket", &actor("usr_ana", "ana"), Some(&subject), &audience);

assert_eq!(

told(&notices_),

vec![

("bo", Reason::Assign, Severity::Info),

2620

("cy", Reason::ReviewRequested, Severity::Warning),

2621

("eve", Reason::Subscribed, Severity::Info),

2622

]

2623

);

2624

assert_eq!(notices_[2].title, "ana opened acme/rocket#7");

}

#[test]

fn ignoring_a_repository_silences_it_and_unsubscribing_keeps_only_what_is_asked() {

2629

let created = event("comment.created", Some("usr_bo"), json!({ "number": 7, "commentId": "cmt_1" }));

2630

let on = comment(person("usr_bo", "bo"), "@cy have a look", &["cy"]);

2631

let audience = watched(&[("cy", WatchLevel::Ignore, &[]), ("ana", WatchLevel::All, &[])]);

2632

assert!(notices(&created, "acme/rocket", &actor("usr_bo", "bo"), Some(&on), &audience).iter().all(|notice| notice.username != "cy"));

2633

let audience = subscribed(&[("cy", State::Unsubscribed, None), ("ana", State::Unsubscribed, None)]);

2634

let notices_ = notices(&created, "acme/rocket", &actor("usr_bo", "bo"), Some(&on), &audience);

2635

// cy was mentioned, which is asked of them; ana unsubscribed from the conversation.

2636

assert_eq!(told(&notices_), vec![("cy", Reason::Mention, Severity::Info)]);

}

#[test]

fn g1t_finishing_or_reviewing_tells_the_person_it_worked_for() {

2641

// The agent acts as the person it works for: still an outcome they hear of.

2642

let ready = event("pull.ready", Some("usr_ana"), json!({ "number": 7 }));

2643

let notices_ = notices(&ready, "acme/rocket", &actor("usr_ana", "ana"), Some(&g1t_pull()), &nobody());

2644

assert_eq!(told(&notices_), vec![("ana", Reason::Author, Severity::Success)]);

2645

assert_eq!(notices_[0].title, "g1t finished acme/rocket#7");

2646

// A person's draft marked ready is not news to them.

2647

assert!(notices(&ready, "acme/rocket", &actor("usr_ana", "ana"), Some(&pull()), &nobody()).is_empty());

2648

2649

let approve = event("review.completed", None, json!({ "number": 7, "verdict": "approve" }));

2650

assert_eq!(

2651

told(&notices(&approve, "acme/rocket", &Actor::default(), Some(&pull()), &nobody())),

2652

vec![("ana", Reason::Author, Severity::Success)]

2653

);

2654

let changes = event("review.completed", None, json!({ "number": 7, "verdict": "request_changes" }));

2655

assert_eq!(

2656

told(&notices(&changes, "acme/rocket", &Actor::default(), Some(&pull()), &nobody())),

2657

vec![("ana", Reason::Author, Severity::Info)]

);

}

#[test]

fn an_agent_s_review_is_shown_by_its_name_and_marked_advisory() {

2663

// Margo (@margo), acting for bo, asks for changes on ana's pull request.

2664

let created = event("comment.created", Some("usr_bo"), json!({ "number": 7, "commentId": "cmt_1" }));

2665

let mut on = comment(person("agt_1", "ana"), "Trim the name.", &[]);

2666

if let Some(said) = on.comment.as_mut() {

2667

said.agent = Some(g1t_contracts::work::AgentRef {

2668

id: "agt_1".into(),

2669

handle: "ana".into(),

2670

display_name: "Margo".into(),

2671

avatar_seed: "margo".into(),

2672

});

2673

said.acting_for = Some(person("usr_bo", "bo"));

2674

said.verdict = Some("request_changes".into());

2675

said.advisory = true;

2676

}

2677

let notices_ = notices(&created, "acme/rocket", &actor("usr_bo", "bo"), Some(&on), &nobody());

2678

// Its handle sharing ana's name leaves nobody out: ana owns it and hears.

2679

assert!(told(&notices_).contains(&("ana", Reason::Author, Severity::Info)));

2680

assert_eq!(notices_[0].title, "Margo (agent) asked for changes on acme/rocket#7 (advisory)");

2681

// An agent subscribes nobody by its handle.

2682

assert!(subscribes(&created, Some(&on)).iter().all(|(name, _)| name != "ana"));

}

#[test]

fn comments_tell_those_mentioned_then_everyone_subscribed_never_the_writer() {

2687

let created = event("comment.created", Some("usr_bo"), json!({ "number": 7, "commentId": "cmt_1" }));

2688

let on = comment(person("usr_bo", "bo"), "@cy @bo have a look", &["cy", "bo", "g1t"]);

2689

let audience = subscribed(&[("eve", State::Subscribed, Some(Reason::Comment)), ("fay", State::Subscribed, Some(Reason::Manual))]);

2690

let notices_ = notices(&created, "acme/rocket", &actor("usr_bo", "bo"), Some(&on), &audience);

assert_eq!(

told(&notices_),

vec![

("cy", Reason::Mention, Severity::Info),

2695

("ana", Reason::Author, Severity::Info),

2696

("eve", Reason::Comment, Severity::Info),

2697

("fay", Reason::Manual, Severity::Info),

2698

]

2699

);

2700

assert_eq!(notices_[0].title, "bo mentioned you on acme/rocket#7");

2701

assert_eq!(notices_[1].title, "bo commented on acme/rocket#7");

2702

assert_eq!(notices_[0].body, "@cy @bo have a look");

2703

// Mentioned and the owner: told once, as mentioned.

2704

let on = comment(person("usr_bo", "bo"), "@ana", &["ana"]);

2705

assert_eq!(

2706

told(&notices(&created, "acme/rocket", &actor("usr_bo", "bo"), Some(&on), &nobody())),

2707

vec![("ana", Reason::Mention, Severity::Info)]

2708

);

2709

// The owner's own comment tells the assignee, not the owner.

2710

let on = comment(person("usr_ana", "ana"), "thanks", &[]);

2711

assert_eq!(

2712

told(&notices(&created, "acme/rocket", &actor("usr_ana", "ana"), Some(&on), &nobody())),

2713

vec![("bo", Reason::Assign, Severity::Info)]

2714

);

2715

// Something that happened, not something written, tells nobody.

2716

let mut on = comment(person("usr_bo", "bo"), "assigned cy", &[]);

2717

on.comment.as_mut().unwrap().event = true;

2718

assert!(notices(&created, "acme/rocket", &actor("usr_bo", "bo"), Some(&on), &nobody()).is_empty());

2719

// An approval is good news.

2720

let mut on = comment(person("usr_cy", "cy"), "", &[]);

2721

on.comment.as_mut().unwrap().verdict = Some("approve".into());

2722

let notices_ = notices(&created, "acme/rocket", &actor("usr_cy", "cy"), Some(&on), &nobody());

2723

assert_eq!(told(&notices_)[0], ("ana", Reason::Author, Severity::Success));

2724

assert_eq!(notices_[0].body, "Add the inbox");

2725

// Writing and being mentioned subscribe, never g1t.

2726

let on = comment(person("usr_bo", "bo"), "@cy", &["cy"]);

2727

assert_eq!(subscribes(&created, Some(&on)), vec![("bo".into(), Reason::Comment), ("cy".into(), Reason::Mention)]);

2728

let on = comment(person(AGENT_ID, "g1t"), "done", &[]);

2729

assert!(subscribes(&created, Some(&on)).is_empty());

}

#[test]

fn issues_and_pull_requests_are_one_thread_each() {

2734

let created = event("comment.created", Some("usr_bo"), json!({ "repoId": "rep_1", "number": 7, "commentId": "cmt_1" }));

2735

let wanted = wants(&created).unwrap();

2736

let thread = thread_of(&created, &wanted, Some(&pull()));

2737

assert_eq!(thread.key, "rep_1#7");

2738

assert_eq!(thread.kind, Some(SubjectKind::Pull));

2739

let merged = event("pull.merged", None, json!({ "repoId": "rep_1", "number": 7 }));

2740

assert_eq!(thread_of(&merged, &wants(&merged).unwrap(), Some(&pull())).key, thread.key);

}

#[test]

fn items_link_to_what_they_are_about() {

2745

assert_eq!(url(Some("acme/rocket"), Some(SubjectKind::Pull), Some(7), None, None), "/acme/rocket/pull/7");

2746

assert_eq!(url(Some("acme/rocket"), Some(SubjectKind::Issue), Some(3), None, None), "/acme/rocket/issues/3");

2747

assert_eq!(url(Some("acme/rocket"), Some(SubjectKind::Run), None, Some("run_1"), None), "/acme/rocket/actions/runs/run_1");

2748

assert_eq!(url(Some("acme/rocket"), Some(SubjectKind::Deploy), None, None, Some("/acme/site/deployments/dpl_1")), "/acme/site/deployments/dpl_1");

2749

assert_eq!(url(Some("acme/rocket"), None, None, None, None), "/acme/rocket");

2750

assert_eq!(url(None, None, None, None, None), "/inbox");

}

#[test]

fn long_titles_are_cut_to_a_line() {

2755

let long = "x".repeat(400);

2756

assert_eq!(clip(&long, MAX_TITLE).chars().count(), MAX_TITLE);

2757

assert_eq!(clip(" short ", MAX_TITLE), "short");

}

#[test]

fn marks_change_only_what_they_say() {

2762

assert_eq!(mark_change(InboxMark::Read), "read_at = COALESCE(read_at, ?1)");

2763

assert!(mark_change(InboxMark::Done).contains("done_at = ?1"));

2764

assert_eq!(mark_change(InboxMark::Unsnooze), "snoozed_until = NULL");

2765

assert!(ranked(&ListInboxArgs::default()));

2766

assert!(!ranked(&ListInboxArgs { reason: Some(Reason::Mention), ..ListInboxArgs::default() }));

2767

assert!(!ranked(&ListInboxArgs { participating: true, ..ListInboxArgs::default() }));

}

#[test]

fn filters_bind_in_order_after_what_is_bound() {

2772

let a = ListInboxArgs {

2773

reason: Some(Reason::Mention),

2774

repo_id: Some("rep_1".into()),

2775

participating: true,

2776

unread: true,

2777

..ListInboxArgs::default()

2778

};

2779

// Two values come first: the username and now.

2780

let (conditions, values) = filters(&a, 2);

2781

assert_eq!(

2782

conditions,

2783

vec!["reason = ?3", "repo_id = ?4", "reason NOT IN ('manual', 'subscribed')", "read_at IS NULL"]

2784

);

2785

assert_eq!(values, vec!["mention", "rep_1"]);

}

#[test]

fn the_bump_keeps_whatever_is_most_urgent_while_unread() {

2790

assert!(BUMP.contains("WHERE NOT EXISTS (SELECT 1 FROM inbox_activity WHERE event_id = ?4 AND username = ?2)"));

2791

assert!(BUMP.contains("ON CONFLICT (username, thread) DO UPDATE"));

2792

assert!(BUMP.contains("activity = inbox_items.activity + 1"));

2793

// Its numbered parameters run from 1 to 19.

2794

for at in 1..=19 {

2795

assert!(BUMP.contains(&format!("?{at}")), "?{at}");

2796

}

2797

assert!(!BUMP.contains("?20"));

}

#[test]

fn live_notifications_say_what_and_where() {

2802

let notice = Notice {

2803

username: "ana".into(),

2804

reason: Reason::Agent,

2805

severity: Severity::Warning,

2806

title: "g1t needs you on acme/api#4".into(),

2807

body: "Which database?".into(),

2808

};

2809

let live = live_notification("evt_1", &notice, "/Acme/api/pull/4#c1", "2026-10-08T00:00:00Z");

2810

assert_eq!(live["id"], "inbox:evt_1");

2811

assert_eq!(live["kind"], "agent_waiting");

2812

assert_eq!(live["workspace"], "acme");

2813

assert_eq!(live["href"], "/Acme/api/pull/4#c1");

2814

assert_eq!(live["actor"]["name"], "acme/api");

2815

let mention = Notice { reason: Reason::TeamMention, ..notice };

2816

assert_eq!(live_notification("evt_2", &mention, "https://elsewhere", "t")["href"], "/inbox");

2817

assert_eq!(live_notification("evt_2", &mention, "/x", "t")["kind"], "mention");

}

#[test]

fn inbox_counts_are_told_as_they_are_by_username() {

2822

assert_eq!(inbox_count_update("Ana", 3), serde_json::json!({ "username": "ana", "unread": 3 }));

2823

assert_eq!(inbox_count_update("bo", 0)["unread"], 0);

}

#[test]

fn token_approvals_tell_the_people_named_about_no_repository() {

2828

let mut asked = event(

2829

"token.approval_requested",

2830

Some("usr_ana"),

2831

json!({ "workspace": "acme", "tokenId": "tok_1", "notify": ["Bo", "cy", "bo", "g1t"], "title": "ana asks", "body": "ci: contents: write", "link": "/acme/-/settings/tokens" }),

2832

);

2833

asked.repo_id = None;

2834

let (thread, told) = workspace_notices(&asked, None);

2835

assert_eq!(thread, "workspace:acme/token/tok_1");

2836

let mut names: Vec<&str> = told.iter().map(|notice| notice.username.as_str()).collect();

2837

names.sort();

2838

assert_eq!(names, ["bo", "cy"], "each once, lowercased, never g1t");

2839

assert!(told.iter().all(|notice| notice.reason == Reason::ReviewRequested && notice.severity == Severity::Warning));

2840

assert!(wants(&asked).is_none(), "about no repository");

2841

let reviewed = event("token.approval_reviewed", None, json!({ "workspace": "acme", "tokenId": "tok_1", "notify": ["ana"], "title": "approved" }));

2842

let (_, told) = workspace_notices(&reviewed, None);

2843

assert_eq!(told[0].reason, Reason::Author);

2844

assert!(WORKSPACE_EVENTS.contains(&"token.approval_reviewed"));

}

#[test]

fn a_workspace_invitation_asks_its_person_and_tells_whoever_sent_it_the_answer() {

2849

let mut sent = event(

2850

"workspace_invitation.created",

2851

Some("usr_owner"),

2852

json!({ "workspace": "flagon-io", "invitationId": "inv_1", "thread": "invitation:inv_1", "notify": ["daweazl"], "title": "@syntaqx invited you to join Flagon, Inc.", "link": "/invitations" }),

2853

);

2854

sent.repo_id = None;

2855

let (thread, told) = workspace_notices(&sent, None);

2856

assert_eq!(thread, "invitation:inv_1");

2857

assert_eq!(told.len(), 1);

2858

assert_eq!((told[0].username.as_str(), told[0].reason, told[0].severity), ("daweazl", Reason::ReviewRequested, Severity::Warning));

2859

let accepted = event("workspace_invitation.accepted", None, json!({ "workspace": "flagon-io", "thread": "invitation:inv_1", "notify": ["syntaqx"] }));

2860

let (thread, told) = workspace_notices(&accepted, None);

2861

assert_eq!(thread, "invitation:inv_1");

2862

assert_eq!((told[0].reason, told[0].severity), (Reason::Author, Severity::Success));

2863

let declined = event("workspace_invitation.declined", None, json!({ "workspace": "flagon-io", "thread": "invitation:inv_1", "notify": ["syntaqx"] }));

2864

assert_eq!(workspace_notices(&declined, None).1[0].severity, Severity::Info);

2865

// Answered or revoked, the person's item is done; a revocation tells nobody.

2866

for kind in ["workspace_invitation.accepted", "workspace_invitation.declined", "workspace_invitation.revoked"] {

2867

assert!(INVITATION_ANSWERED.contains(&kind));

2868

}

2869

assert!(!WORKSPACE_EVENTS.contains(&"workspace_invitation.revoked"));

2870

}

2871

}