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

//! What leaves nothing for a person to do settles their item about it

30

//! (marks it done): see "Settling" below for which event settles what.

31

//! A merge or a close reaches each thread's history but settles it at

32

//! once, so it is neither left unread nor pushed.

33

//!

34

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

35

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

36

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

37

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

38

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

39

//!

40

//! A work item belongs to a workspace and is one thread (`item:<id>`),

41

//! however many repositories it is linked to. Watchers of each of them

42

//! hear of it; whoever ignores its origin hears of nothing on it. Its

43

//! items are kept with its workspace and its origin, if it has one, and

44

//! are shown to the workspace's members and to whoever can read the

45

//! origin (the event's own repository, never the first of `repoIds`).

46

//! Those who hear of it only by watching another repository it is linked

47

//! to are told only when they may read it: members of its workspace, or

48

//! anyone when its origin is public.

49

//!

50

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

51

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

52

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

53

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

54

//! is held up.

55

56

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

57

58

use g1t_contracts::credentials::Principal;

59

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

60

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

61

use g1t_contracts::inbox::*;

62

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

63

use g1t_contracts::security::SeverityCounts;

64

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

65

use g1t_contracts::time::rfc3339;

66

use g1t_contracts::{new_id, system};

67

use g1t_kit::now_ms;

68

use serde::Deserialize;

69

use worker::wasm_bindgen::JsValue;

70

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

71

72

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

73

74

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

75

pub const DONE_DAYS: u32 = 30;

76

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

77

pub const MAX_DAYS: u32 = 180;

78

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

79

const MAX_RANKED: u32 = 20;

80

const MAX_TITLE: usize = 200;

81

const MAX_BODY: usize = 300;

82

83

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

84

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

85

pub struct Wanted {

86

/// The repository; for a work item, its origin, if it has one.

87

pub repo_id: Option<String>,

88

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

89

pub number: Option<u32>,

90

pub comment_id: Option<String>,

91

/// When the event is about a work item.

92

pub item: Option<WantedItem>,

93

}

94

95

/// The work item an event is about.

96

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

97

pub struct WantedItem {

98

pub id: String,

99

/// The workspace's slug.

100

pub workspace: String,

101

/// Every repository it is linked to, its origin first.

102

pub repo_ids: Vec<String>,

}

impl Wanted {

/// The repository's id, or empty for a work item without an origin.

107

fn repo(&self) -> &str {

108

self.repo_id.as_deref().unwrap_or_default()

}

}

/// One person to tell, and what.

113

#[derive(Debug, PartialEq, Eq)]

114

pub struct Notice {

115

pub username: String,

116

pub reason: Reason,

117

pub severity: Severity,

pub title: String,

pub body: String,

}

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

123

#[derive(Debug, Default)]

124

pub struct Actor {

125

pub id: Option<String>,

126

pub username: Option<String>,

}

impl Actor {

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

131

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

132

}

133

134

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

135

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

136

}

137

138

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

139

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

140

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

}

}

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

145

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

146

}

147

148

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

149

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

150

}

151

152

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

153

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

154

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

155

156

/// The work item an event's data names (`itemId`, with its workspace and

157

/// linked repositories), if any.

158

fn item_of(data: &serde_json::Value) -> Option<WantedItem> {

159

let id = data["itemId"].as_str().filter(|id| !id.is_empty())?;

160

Some(WantedItem {

161

id: id.to_owned(),

162

workspace: data["workspace"].as_str().unwrap_or_default().to_lowercase(),

163

repo_ids: names(data, "repoIds"),

})

}

/// The work item events the inbox tells people of.

168

const ITEM_TOLD: [&str; 4] = ["work_item.opened", "work_item.closed", "work_item.reopened", "work_item.assigned"];

169

170

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

171

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

172

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

173

let data = &event.data;

174

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

175

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

176

// A work item's events, and comments on one: it may have no origin.

177

// Its origin is the event's own repository and nothing else: never a

178

// `repoId` in its data, nor the first of `repoIds`, which may be a

179

// repository it is only linked to, whose readers may not see it.

180

let on_item = event.kind == "comment.created" && data["itemId"].is_string();

181

if ITEM_TOLD.contains(&event.kind.as_str()) || on_item {

182

return Some(Wanted {

183

repo_id: event.repo_id.clone().filter(|id| !id.is_empty()),

184

number: Some(number("number")?),

185

comment_id: if on_item { Some(text("commentId")?) } else { None },

186

item: Some(item_of(data)?),

187

});

188

}

189

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

190

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

191

Some(Wanted {

192

repo_id: Some(repo_id.clone()),

number,

comment_id,

item: None,

})

};

match event.kind.as_str() {

199

"pull.stalled" | "pull.merged" | "pull.closed" | "pull.opened" | "pull.auto_merge_disabled"

200

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

201

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

202

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

203

_ => None,

204

},

205

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

206

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

207

_ => None,

208

},

209

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

210

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

211

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

212

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

213

// A commit is a thread of its own, its link the commit's page.

214

"commit_comment.created" => on(None, None),

215

// Security alerts are threads of their own, not of a work item.

216

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

_ => None,

}

}

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

222

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

223

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

224

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

225

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

226

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

227

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

228

"secret_scanning_alert.summary",

229

"code_scanning_alert.summary",

230

"vulnerability_alert.summary",

231

"secret_scanning.bypass_requested",

232

"secret_scanning.bypass_reviewed",

233

"security_advisory.reported",

234

"security_advisory.published",

235

];

236

237

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

238

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

239

pub struct Thread {

240

pub key: String,

241

pub kind: Option<SubjectKind>,

242

pub number: Option<u32>,

243

pub run_id: Option<String>,

244

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

245

pub link: Option<String>,

246

}

247

248

/// The thread key of a pull request.

249

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

250

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

251

}

252

253

/// What a work item's thread key starts with: `item:<item_id>`.

254

pub const ITEM_THREAD: &str = "item:";

255

256

/// The thread key of a work item.

257

pub fn item_thread(item_id: &str) -> String {

258

format!("{ITEM_THREAD}{item_id}")

259

}

260

261

/// Whether people subscribe to a thread: a work item's or a pull request's.

262

pub fn subscribable(thread: &str) -> bool {

263

thread.starts_with(ITEM_THREAD) || thread.contains('#')

264

}

265

266

/// The thread of a workflow on a branch: `<repo_id>/run/<workflow>@<branch>`.

267

pub fn run_thread(repo_id: &str, data: &serde_json::Value) -> String {

268

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

269

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

270

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

271

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

272

format!("{repo_id}/run/{workflow}@{branch}")

273

}

274

275

/// The thread of a project's production, or of one pull request's preview:

276

/// `<repo_id>/deploy/<project_id>/<production|number>`.

277

pub fn deploy_thread(repo_id: &str, data: &serde_json::Value, number: Option<u32>) -> String {

278

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

279

format!("{repo_id}/deploy/{}/{which}", data["projectId"].as_str().unwrap_or_default())

280

}

281

282

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

283

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

284

let data = &event.data;

285

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

286

match event.kind.as_str() {

287

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

288

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

289

Thread {

290

key: deploy_thread(wanted.repo(), data, wanted.number),

291

kind: Some(SubjectKind::Deploy),

292

number: wanted.number,

293

run_id: text("deploymentId"),

link: text("path"),

}

}

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

298

"deployment.review_requested" => Thread {

key: format!(

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

wanted.repo(),

text("runId").unwrap_or_default(),

303

text("environment").unwrap_or_default()

304

),

305

kind: Some(SubjectKind::Run),

306

number: None,

307

run_id: text("runId"),

308

link: text("link"),

309

},

310

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

311

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

312

Thread {

313

key: run_thread(wanted.repo(), data),

314

kind: Some(SubjectKind::Run),

315

number: None,

316

run_id: text("runId"),

link: None,

}

}

// One commit's comments (work/src/commit_comments.rs).

321

"commit_comment.created" => Thread {

322

key: format!("{}/commit/{}", wanted.repo(), text("commit").unwrap_or_default()),

kind: None,

number: None,

run_id: None,

link: text("link"),

},

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

329

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

330

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

331

let which = match kind {

332

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

333

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

334

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

335

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

336

};

337

Thread {

338

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

kind: None,

number: None,

run_id: None,

link: text("link"),

}

}

// A work item: one thread, wherever it is linked.

346

_ if wanted.item.is_some() => Thread {

347

key: item_thread(wanted.item.as_ref().map_or("", |item| item.id.as_str())),

348

kind: Some(SubjectKind::Item),

349

number: wanted.number,

run_id: None,

link: None,

},

_ => match wanted.number {

354

Some(number) => Thread {

355

key: numbered_thread(wanted.repo(), number),

356

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

357

number: Some(number),

run_id: None,

link: None,

},

None => Thread {

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

kind: None,

number: None,

run_id: None,

link: None,

},

},

}

}

// --- Settling ----------------------------------------------------------------

373

//

374

// An item is settled (marked done, and read) by the event that leaves

375

// nothing for its person to do about it, so the inbox, its count, Home's

376

// Needs you and `g1t status` show only what still waits on someone. A

377

// settled item stays in Done, with its history, until it is purged; new

378

// activity on its thread brings it back.

379

//

380

// | Item | Settled by |

381

// | --- | --- |

382

// | g1t stopped and needs you (`agent`) | `pull.resumed`, `pull.updated`, `pull.merge_requested`, `pull.merged`, `pull.closed` |

383

// | Anything on a work item or pull request | `work_item.closed`, `pull.merged`, `pull.closed`, after it is told (`closes`) |

384

// | Your own item on a thread | your comment or review on it (`handled_by`) |

385

// | A review asked of you | `pull.review_request_removed` naming you or your team |

386

// | Checks failed on a pull request (`ci_activity`) | `checks.completed` passed on its head |

387

// | A workflow failed on a branch | the next `workflow.completed` success of that workflow on that branch |

388

// | A deployment failed | `deployment.succeeded` of that project's production, or that preview |

389

// | A deployment waits for your review | `workflow.completed` of its run, however it ended |

390

// | A push protection bypass to review | `secret_scanning.bypass_reviewed` |

391

// | A token waits for your approval | `token.approval_reviewed` |

392

// | A workspace invitation | `workspace_invitation.accepted`, `.declined`, `.revoked` |

393

// | A scan's alerts | the next scan's summary with none open (`refresh_security_thread`) |

394

395

/// Which threads a settle reaches.

396

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

pub enum Scope {

/// One thread.

Thread(String),

/// Every thread matching a GLOB pattern, such as one pull request's

401

/// previews in every project: `rep_1/deploy/*/7`.

Matching(String),

}

/// Items an event settles.

406

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

407

pub struct Settle {

408

pub scope: Scope,

409

/// One person's item, or everyone's.

410

pub username: Option<String>,

411

/// Only items whose latest activity was for one of these reasons; any when empty.

412

pub reasons: Vec<Reason>,

413

/// Only items whose latest activity was one of these events; any when empty.

414

pub events: Vec<&'static str>,

415

/// Why, in a few words (`settled_reason`): "CI passed in run 330", "Merged".

416

pub why: Option<String>,

417

/// Who settled it, when a person or g1t did (`settled_by`): a username, or g1t.

418

pub by: Option<String>,

419

}

420

421

/// An item a settle reached: whose, and its thread.

422

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

423

pub struct Settled {

424

pub username: String,

pub thread: String,

}

impl Settle {

fn thread(thread: impl Into<String>) -> Self {

430

Settle {

431

scope: Scope::Thread(thread.into()),

username: None,

reasons: Vec::new(),

events: Vec::new(),

why: None,

by: None,

}

}

fn matching(pattern: impl Into<String>) -> Self {

441

Settle {

442

scope: Scope::Matching(pattern.into()),

..Settle::thread("")

}

}

fn because(mut self, reasons: &[Reason]) -> Self {

448

self.reasons = reasons.to_vec();

self

}

fn after(mut self, events: &[&'static str]) -> Self {

453

self.events = events.to_vec();

self

}

fn of(mut self, username: &str) -> Self {

458

self.username = Some(username.trim().trim_start_matches('@').to_lowercase());

self

}

/// Says why it settled, and who settled it when someone did.

463

fn saying(mut self, why: impl Into<String>, by: Option<&str>) -> Self {

464

self.why = Some(why.into());

465

self.by = by.map(|name| name.trim().trim_start_matches('@').to_lowercase()).filter(|name| !name.is_empty());

self

}

/// The statement that settles the items, as of `now`, naming whose it

470

/// settled and their threads.

471

fn statement(&self, db: &D1Database, now: &str) -> Result<D1PreparedStatement> {

472

let mut values: Vec<JsValue> = vec![

473

now.into(),

474

self.why.as_deref().map_or(JsValue::NULL, JsValue::from),

475

self.by.as_deref().map_or(JsValue::NULL, JsValue::from),

476

];

477

let mut conditions = vec!["done_at IS NULL".to_owned()];

478

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

479

values.push(value.into());

480

format!("?{}", values.len())

481

};

482

match &self.scope {

483

Scope::Thread(thread) => conditions.push(format!("thread = {}", bind(&mut values, thread))),

484

Scope::Matching(pattern) => conditions.push(format!("thread GLOB {}", bind(&mut values, pattern))),

485

}

486

if let Some(username) = &self.username {

487

conditions.push(format!("username = {}", bind(&mut values, username)));

488

}

489

if !self.reasons.is_empty() {

490

let marks: Vec<String> = self.reasons.iter().map(|reason| bind(&mut values, reason.as_str())).collect();

491

conditions.push(format!("reason IN ({})", marks.join(", ")));

492

}

493

if !self.events.is_empty() {

494

let marks: Vec<String> = self.events.iter().map(|kind| bind(&mut values, kind)).collect();

495

conditions.push(format!("event IN ({})", marks.join(", ")));

496

}

497

db.prepare(format!(

498

"UPDATE inbox_items SET done_at = ?1, read_at = COALESCE(read_at, ?1), settled_reason = ?2, settled_by = ?3

499

WHERE {} RETURNING username, thread",

500

conditions.join(" AND ")

))

.bind(&values)

}

}

/// The work item or pull request an event names, as its thread.

507

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

508

if let Some(item) = item_of(&event.data) {

509

return Some(item_thread(&item.id));

510

}

511

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

512

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

513

Some(numbered_thread(&repo_id, number))

514

}

515

516

/// Why an agent's wait is over, by the event that ended it (`RESUMES`).

517

fn resumed_because(kind: &str) -> &'static str {

518

match kind {

519

"pull.resumed" => "g1t went on",

520

"pull.updated" => "New commits were pushed",

521

"pull.merge_requested" => "Merge was asked for",

522

"pull.merged" => "Merged",

_ => "Closed",

}

}

/// What an event settles before its own notices are written: what it

528

/// supersedes. A success after a failure, a review given or no longer

529

/// asked for, a request answered. `actor` is who caused it, by username

530

/// (g1t for g1t), when known: each settle says why, and who did.

531

pub fn settles(event: &Event, actor: Option<&str>) -> Vec<Settle> {

532

let data = &event.data;

533

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

534

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

535

let mut settled = Vec::new();

536

let kind = event.kind.as_str();

537

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

538

if RESUMES.contains(&kind)

539

&& let Some(thread) = numbered(event)

540

{

541

settled.push(Settle::thread(thread).because(&[Reason::Agent]).saying(resumed_because(kind), actor));

542

}

543

match kind {

544

// Reviews no longer asked for, of the people or the teams named.

545

"pull.review_request_removed" => {

546

if let Some(thread) = numbered(event) {

547

let people = names(data, "reviewers").into_iter().chain(teams_asked(data).into_iter().flat_map(|(_, people)| people));

548

for name in people.collect::<HashSet<_>>() {

549

settled.push(

550

Settle::thread(thread.clone())

551

.because(&[Reason::ReviewRequested])

552

.of(&name)

553

.saying("Review no longer asked of you", actor),

);

}

}

}

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

559

_ if INVITATION_ANSWERED.contains(&kind) => {

560

if let Some(thread) = text("thread").filter(|thread| thread.starts_with("invitation:")) {

561

let why = match kind {

562

"workspace_invitation.accepted" => "Invitation accepted",

563

"workspace_invitation.declined" => "Invitation declined",

564

_ => "Invitation revoked",

565

};

566

settled.push(Settle::thread(thread).because(&[Reason::ReviewRequested]).saying(why, actor));

567

}

568

}

569

// A token approved, denied or revoked no longer waits on the owners.

570

"token.approval_reviewed" => {

571

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

572

settled.push(Settle::thread(thread).because(&[Reason::ReviewRequested]).saying("Token request decided", actor));

573

}

574

// Every check on its head passes now (or none fails): the failure

575

// its owner was told of is behind them.

576

"checks.completed" if data["status"].as_str() == Some("passed") => {

577

if let Some(thread) = numbered(event) {

578

let why = match text("commit") {

579

Some(commit) => format!("Checks passed on {}", short(&commit)),

580

None => "Checks passed".to_owned(),

581

};

582

settled.push(Settle::thread(thread).because(&[Reason::CiActivity]).saying(why, None));

583

}

584

}

585

"workflow.completed" => {

586

if let Some(repo_id) = &repo_id {

587

let workflow = text("workflow").unwrap_or_else(|| "The workflow".to_owned());

588

let run = data["number"].as_u64().map(|n| format!(" in run {n}")).unwrap_or_default();

589

// The workflow passes on the branch again: its failures there are over.

590

if data["conclusion"].as_str() == Some("success") && data["pull"].as_u64().is_none() {

591

settled.push(Settle::thread(run_thread(repo_id, data)).saying(format!("{workflow} passed{run}"), None));

592

}

593

// Its run is over, however it ended: no deployment of it waits for review.

594

if let Some(id) = text("runId") {

595

settled.push(

596

Settle::matching(format!("{repo_id}/review/{id}/*"))

597

.because(&[Reason::ReviewRequested])

598

.saying(format!("{workflow} finished{run}"), None),

);

}

}

}

// It deployed: the failure before it is over.

604

"deployment.succeeded" => {

605

if let Some(repo_id) = &repo_id {

606

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

607

let what = match number {

608

Some(n) => format!("Preview of #{n}"),

609

None => text("project").map_or_else(|| "Production".to_owned(), |project| format!("Production of {project}")),

610

};

611

let commit = text("commit").map(|commit| format!(" at {}", short(&commit))).unwrap_or_default();

612

let by = actor.map(str::to_owned).or_else(|| text("triggeredBy"));

613

settled.push(

614

Settle::thread(deploy_thread(repo_id, data, number))

615

.after(&["deployment.failed"])

616

.saying(format!("{what} deployed{commit}"), by.as_deref()),

);

}

}

// A bypass decided no longer waits on its reviewers.

621

"secret_scanning.bypass_reviewed" => {

622

if let (Some(repo_id), Some(which)) = (&repo_id, text("requestId").or_else(|| text("alertId"))) {

623

settled.push(

624

Settle::thread(format!("{repo_id}/security/{which}"))

625

.because(&[Reason::SecurityAlert])

626

.saying("Bypass request decided", actor),

);

}

}

_ => {}

}

settled

}

/// The events after which nothing is left to do about a work item or pull

636

/// request: every item about it is settled once its own notices are

637

/// written, so they reach each thread's history but nobody's inbox, and

638

/// are not pushed. What landed shows on Home instead.

639

pub const CLOSES: [&str; 3] = ["pull.merged", "pull.closed", "work_item.closed"];

640

641

/// What an event settles once its own notices are written: everything

642

/// about a work item or pull request that is over, a pull request's

643

/// previews with it. `actor` is who merged or closed it, by username.

644

pub fn closes(event: &Event, actor: Option<&str>) -> Vec<Settle> {

645

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

646

return Vec::new();

647

}

648

let Some(thread) = numbered(event) else {

649

return Vec::new();

650

};

651

let why = match event.kind.as_str() {

652

"pull.merged" => "Merged",

653

"work_item.closed" if event.data["reason"].as_str() == Some("not_planned") => "Closed as not planned",

654

_ => "Closed",

655

};

656

let mut settled = vec![Settle::thread(thread).saying(why, actor)];

657

if event.kind.starts_with("pull.")

658

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

659

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

660

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

661

)

662

{

663

settled.push(Settle::matching(format!("{repo_id}/deploy/*/{number}")).saying(why, actor));

}

settled

}

/// The item a person's own comment or review settles: theirs on that

669

/// thread, which they have answered. Never an agent's, nor g1t's.

670

pub fn handled_by(event: &Event, subject: Option<&InboxSubject>) -> Option<Settle> {

671

if event.kind != "comment.created" {

672

return None;

673

}

674

let comment = subject?.comment.as_ref().filter(|comment| !comment.event && comment.agent.is_none())?;

675

if is_g1t_id(&comment.author.id) || is_g1t(&comment.author.username) {

676

return None;

677

}

678

let why = if comment.verdict.is_some() { "Reviewed" } else { "Commented" };

679

Some(Settle::thread(numbered(event)?).of(&comment.author.username).saying(why, Some(&comment.author.username)))

680

}

681

682

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

683

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

684

struct Told<'a> {

685

actor: &'a Actor,

686

audience: &'a Audience,

687

notices: Vec<Notice>,

}

impl Told<'_> {

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

692

/// they unsubscribed from the thread.

693

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

694

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

695

if username.is_empty()

696

|| is_g1t(&username)

697

|| self.actor.is_named(&username)

698

|| self.audience.ignores(&username)

699

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

{

return;

}

let notice = Notice {

username,

reason,

severity,

title: clip(title, MAX_TITLE),

708

body: clip(body, MAX_BODY),

709

};

710

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

711

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

712

Some(_) => {}

713

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

}

}

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

718

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

719

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

720

return;

721

}

722

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

723

}

724

725

/// Everyone subscribed to a work item or pull request: its owner and

726

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

727

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

728

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

729

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

730

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

731

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

732

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

733

for name in &subject.assignees {

734

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

735

}

736

for name in &subject.reviewers {

737

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

738

}

739

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

740

for (name, own) in subscribed {

741

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

}

}

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

746

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

747

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

748

for name in watching {

749

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

}

}

}

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

755

let text = text.trim();

756

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

757

return text.to_owned();

758

}

759

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

760

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

761

}

762

763

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

764

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

765

match subject.kind {

766

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

_ => "work_items",

}

}

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

772

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

773

/// people it tells.

774

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> {

787

data[key]

788

.as_array()

789

.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`, or for a work item its

794

/// workspace), given what it names and who follows it. `actor` is who

795

/// caused it; outcomes nobody chose (checks, workflows, deployments, g1t

796

/// stopping) are told whoever caused them.

797

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

798

let nobody = Actor::default();

799

let data = &event.data;

800

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

801

// or `acme#12` for a work item.

802

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

803

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

804

_ => repo.to_owned(),

805

};

806

let outcome = matches!(

807

event.kind.as_str(),

808

"checks.completed"

809

| "workflow.completed"

810

| "pull.stalled"

811

| "deployment.failed"

812

| "deployment.succeeded"

813

| "deployment.review_requested"

814

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

815

let mut told = Told {

816

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

audience,

notices: Vec::new(),

};

// Who did it, as titles name them.

821

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

822

823

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

824

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

825

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

826

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

827

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

828

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

829

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

830

for name in &item.assignees {

831

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

}

}

}

// Whoever turned auto-merge on hears that it is off, unless they

836

// turned it off themselves: it will not merge without them now.

837

// Not when g1t handed it to the merge queue, which lands it: there

838

// is nothing for them to do.

839

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

840

if data["queued"].as_bool() == Some(true) {

841

return told.notices;

842

}

843

if let Some(by) = data["autoMergeBy"].as_str() {

844

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

845

let title = format!("Auto-merge was turned off on {at}");

846

told.tell(by, Reason::StateChange, Severity::Warning, &title, detail.unwrap_or(&pull.title), true);

847

}

848

}

849

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

850

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

851

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

852

let title = if owned {

853

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

854

} else {

855

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

856

};

857

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

858

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

859

}

860

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

861

let title = if owned {

862

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

863

} else {

864

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

865

};

866

for name in people {

867

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

}

}

}

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

872

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

873

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

874

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

875

}

876

}

877

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

878

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

879

for name in &on.assignees {

880

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

881

}

882

for name in &on.reviewers {

883

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

884

}

885

for name in &on.mentions {

886

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

887

}

888

for team in &on.team_mentions {

889

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

890

for name in &team.members {

891

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

892

}

893

}

894

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

895

}

896

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

897

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

898

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

899

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

900

};

901

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

902

}

903

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

904

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

905

match subject {

906

Some(pull) => {

907

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

908

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

909

}

910

None => {

911

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

912

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

913

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

914

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

915

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

916

if let Some(id) = &actor.id

917

&& !is_g1t_id(id)

918

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

919

{

920

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

}

}

}

}

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

926

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

927

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

928

let what = match subject {

929

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

930

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

931

};

932

let (title, severity) = if failed {

933

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

934

} else {

935

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

936

};

937

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

938

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

939

_ => match subject {

940

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

941

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

942

},

943

};

944

// A failure is whoever answers for it's to fix. A success after

945

// one tells them nothing: it settles the failure's item instead

946

// (`settles`), so there is nothing left in their inbox about it.

947

if failed {

948

match subject {

949

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))

955

.and(actor.username.as_deref())

956

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

957

if let Some(name) = pusher {

958

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

}

}

}

}

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

964

}

965

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

966

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

967

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

968

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

969

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

970

};

971

let title = match (actor.name(), landed_by(data)) {

972

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

973

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

974

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

975

};

976

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

977

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

978

}

979

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

980

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

981

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

982

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

983

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

984

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

985

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

986

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

987

}

988

}

989

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

990

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

991

match kind {

992

"secret_scanning.bypass_requested" => {

993

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

994

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

995

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

996

}

997

}

998

"secret_scanning.bypass_reviewed" => {

999

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

1000

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

1001

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

1002

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

1003

}

1004

}

1005

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

1006

"security_advisory.reported" => {

1007

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

1008

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

1009

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

1010

}

1011

}

1012

// Published for everyone: watchers who chose security.

1013

"security_advisory.published" => {

1014

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

1015

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

1016

}

1017

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

1018

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

1019

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

1020

// person's choice, applied there too.

1021

_ => {

1022

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

1023

return told.notices;

1024

};

1025

if summary.opened_count == 0 {

1026

return told.notices;

1027

}

1028

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

1029

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

1030

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

1031

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

1032

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

1033

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

1034

}

1035

for name in &summary.notify {

1036

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

1037

}

1038

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

1039

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

1040

// shown to someone who cannot change the code.

1041

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

1042

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

1043

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

1044

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

}

}

}

}

// The work service names who: the people and teams mentioned, and

1050

// the commit's author when they have an account.

1051

("commit_comment.created", None) => {

1052

let short = data["commit"].as_str().map(short).unwrap_or_default();

1053

let at = format!("{repo}@{short}");

1054

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

1055

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

1056

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

1057

}

1058

for team in data["teamMentions"].as_array().into_iter().flatten() {

1059

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

1060

for name in names(team, "members") {

1061

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

1062

}

1063

}

1064

if let Some(author) = data["commitAuthor"].as_str() {

1065

told.tell(author, Reason::Author, Severity::Info, &format!("{who} commented on your commit {at}"), body, false);

1066

}

1067

}

1068

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

1069

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

1070

return Vec::new();

1071

};

1072

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

1073

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

1074

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

1075

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

1076

// anything they set off.

1077

let writer = match &comment.agent {

1078

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

1079

None => Actor {

1080

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

1081

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

1082

},

1083

};

1084

told.actor = &writer;

1085

let who = match &comment.agent {

1086

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

1087

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

1088

};

1089

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

1090

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

1091

for name in &comment.mentions {

1092

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

1093

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

1094

}

1095

for team in &comment.team_mentions {

1096

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

1097

for name in &team.members {

1098

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

1099

}

1100

}

1101

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

1102

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

1103

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

1104

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

1105

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

1106

};

1107

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

1108

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

1109

// they subscribed.

1110

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

1111

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

1112

}

1113

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

1114

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

return told.notices;

}

_ => {}

}

told.notices

}

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

1123

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

1124

/// and the reviewers asked. Never g1t.

1125

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

1126

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

1127

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

1128

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

1129

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

1130

people.push((name, reason));

1131

}

1132

};

1133

match event.kind.as_str() {

1134

"comment.created" => {

1135

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

1136

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

1137

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

1138

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

1139

}

1140

for name in &comment.mentions {

1141

add(name, Reason::Mention);

1142

}

1143

for team in &comment.team_mentions {

1144

for name in &team.members {

1145

add(name, Reason::TeamMention);

}

}

}

}

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

1151

if let Some(on) = subject {

1152

for name in &on.mentions {

1153

add(name, Reason::Mention);

1154

}

1155

for team in &on.team_mentions {

1156

for name in &team.members {

1157

add(name, Reason::TeamMention);

}

}

}

}

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

1163

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

1164

add(&name, Reason::Assign);

1165

}

1166

}

1167

"pull.review_requested" => {

1168

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

1169

add(&name, Reason::ReviewRequested);

1170

}

1171

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

1172

for name in people {

1173

add(&name, Reason::ReviewRequested);

}

}

}

_ => {}

}

people

}

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

1183

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

1184

}

1185

1186

/// What landed a work item, as a close says it: the pull request

1187

/// (`acme/rocket#9`) or the commit (`acme/rocket@1a2b3c4`), from its

1188

/// `resolution`.

1189

fn landed_by(data: &serde_json::Value) -> Option<String> {

1190

let resolution = &data["resolution"];

1191

let repo = resolution["repo"].as_str().filter(|repo| !repo.is_empty())?;

1192

match (resolution["pull"].as_u64(), resolution["commit"].as_str()) {

1193

(Some(pull), _) => Some(format!("{repo}#{pull}")),

1194

(None, Some(commit)) => Some(format!("{repo}@{}", short(commit))),

_ => None,

}

}

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

1200

1201

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

1202

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

1203

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

1204

return None;

1205

}

1206

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

1207

}

1208

1209

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

1210

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

1211

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

1212

.into_iter()

1213

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

1214

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

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

.join(", ")

}

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

1220

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

1221

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

1222

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

1223

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

1224

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

1225

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

1226

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

1227

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

1228

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

1229

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

}

}

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

1234

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

1235

/// pushed a push that was refused.

1236

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

1237

let n = summary.opened_count;

1238

let counts = breakdown(&summary.new_by_severity);

1239

let title = if counts.is_empty() {

1240

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

1241

} else {

1242

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

1243

};

1244

let pushed = match summary.blocked {

1245

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

1246

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

1252

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

1253

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

1254

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

1255

let open = summary.open_total();

1256

let mut changes = Vec::new();

1257

if summary.fixed_count > 0 {

1258

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

1259

}

1260

if summary.dismissed_count > 0 {

1261

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

1262

}

1263

let title = if open == 0 {

1264

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

1265

} else {

1266

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

1267

};

1268

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

1269

}

1270

1271

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

1272

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

1273

pub enum Delivery {

1274

/// Their inbox, now.

1275

Now,

1276

/// Their next security digest.

Digest,

/// Nowhere.

Never,

}

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

1283

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

1284

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

1285

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

1286

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

1287

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

1288

if summary.urgent() {

1289

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

1290

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

1291

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

1292

}

1293

match settings.security_digest {

1294

DigestCadence::AtOnce => Delivery::Now,

1295

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

1296

DigestCadence::Off => Delivery::Never,

}

}

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

1301

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

1302

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

1303

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

1304

/// thread keeps its place.

1305

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

1306

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

1307

let now = now_ms();

1308

let at = rfc3339(now);

1309

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

1310

db.batch(vec![

1311

db.prepare(

1312

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

1313

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

1314

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

1315

settled_reason = CASE WHEN ?6 = 1 AND done_at IS NULL THEN 'Nothing left open after the latest scan' ELSE settled_reason END,

1316

settled_by = CASE WHEN ?6 = 1 AND done_at IS NULL THEN NULL ELSE settled_by END

1317

WHERE thread = ?7

1318

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

1319

)

1320

.bind(&[

1321

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

1322

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

1323

title.as_str().into(),

1324

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)

1331

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

1332

)

1333

.bind(&[

1334

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

1335

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

1336

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

1337

title.as_str().into(),

1338

body.as_str().into(),

1339

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

thread.into(),

])?,

])

.await?;

Ok(())

}

/// Where an item is on g1t.sh: a work item's in its workspace, the rest in

1348

/// their repository.

1349

pub fn url(

1350

repo: Option<&str>,

1351

workspace: Option<&str>,

1352

subject: Option<SubjectKind>,

1353

number: Option<u32>,

1354

run_id: Option<&str>,

1355

link: Option<&str>,

1356

) -> String {

1357

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

1358

return link.to_owned();

1359

}

1360

if let (Some(SubjectKind::Item), Some(workspace), Some(number)) = (subject, workspace.filter(|slug| !slug.is_empty()), number) {

1361

return format!("/{workspace}/-/work/{number}");

1362

}

1363

let Some(repo) = repo else {

1364

return "/inbox".to_owned();

1365

};

1366

match (subject, number, run_id) {

1367

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

1368

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

1369

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

1370

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

}

}

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

1375

1376

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

1377

pub struct Sources<'a> {

1378

pub work: &'a Fetcher,

1379

pub repos: &'a Fetcher,

1380

pub identity: &'a Fetcher,

1381

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

1382

pub notify: Option<&'a Fetcher>,

1383

}

1384

1385

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

/// may be emailed.

struct Written {

notice: Notice,

/// What its email is checked against: its repository (a work item's

1390

/// origin, which it may not have), and a work item's workspace, whose

1391

/// members may see it.

1392

repo_id: Option<String>,

1393

workspace: Option<String>,

1394

url: String,

1395

event_id: String,

1396

/// Its event leaves nothing to do (`CLOSES`): settled once written.

1397

closing: bool,

1398

/// The thread it is on, which notify keeps with what it replays.

thread: String,

}

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

1403

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

1404

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

1405

// Everyone who caused one, named in one call: notices and settles say who.

1406

let names = actor_names(sources, events).await;

1407

let actor_of = |event: &Event| -> Option<String> {

1408

match event.actor.as_deref() {

1409

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

1410

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

None => None,

}

};

// Whose items were settled, and on which threads, to tell once at the end.

1415

let mut settled: HashSet<Settled> = HashSet::new();

1416

// What the batch supersedes goes first, so its own notices start afresh.

1417

let superseded: Vec<Settle> = events.iter().flat_map(|event| settles(event, actor_of(event).as_deref())).collect();

1418

match settle(db, &superseded).await {

1419

Ok(items) => settled.extend(items),

1420

Err(error) => worker::console_error!("inbox: items not settled: {error}"),

1421

}

1422

deliver_events(db, sources, events, &names, &mut settled).await;

1423

// What is over goes last, after its own notices reached each thread's history.

1424

let over: Vec<Settle> = events.iter().flat_map(|event| closes(event, actor_of(event).as_deref())).collect();

1425

match settle(db, &over).await {

1426

Ok(items) => settled.extend(items),

1427

Err(error) => worker::console_error!("inbox: closed threads not settled: {error}"),

1428

}

1429

if let Some(notify) = sources.notify {

1430

let mut threads: HashMap<&str, Vec<String>> = HashMap::new();

1431

for item in &settled {

1432

threads.entry(item.username.as_str()).or_default().push(item.thread.clone());

1433

}

1434

for (username, threads) in threads {

1435

tell_inbox_settled(db, notify, username, threads).await;

}

}

}

/// The usernames of whoever caused the events that tell, watch or settle

1441

/// anything, by account id, in one call. g1t is not looked up.

1442

async fn actor_names(sources: &Sources<'_>, events: &[Event]) -> HashMap<String, String> {

1443

let ids: Vec<String> = events

1444

.iter()

1445

.filter(|event| {

1446

wants(event).is_some() || event.kind == "repo.created" || !settles(event, None).is_empty() || !closes(event, None).is_empty()

1447

})

1448

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

1449

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

1450

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

.into_iter()

.collect();

if ids.is_empty() {

return HashMap::new();

1455

}

1456

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

1457

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

HashMap::new()

})

}

async fn deliver_events(db: &D1Database, sources: &Sources<'_>, events: &[Event], names: &HashMap<String, String>, settled: &mut HashSet<Settled>) {

1463

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

1464

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

1465

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

1466

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

1467

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

1468

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

1469

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

1470

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

1471

let live = told

1472

.iter()

1473

.map(|notice| (notice.username.clone(), live_notification(&event.id, notice, link, &event.time, Some(&thread))))

1474

.collect();

1475

tell_live(notify, live).await;

1476

for notice in &told {

1477

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

}

}

}

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

1482

.iter()

1483

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

1484

.collect();

1485

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

1486

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

1487

return;

1488

}

1489

for event in created {

1490

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

1491

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

1492

}

1493

}

1494

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

1495

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

1496

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

1497

for (event, wanted) in wanted {

1498

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

1499

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

1500

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

1501

}

1502

}

1503

// Notify: every item that was news, live in the person's tabs

1504

// (services/notify), but for what is over, which is settled at once:

1505

// nobody is pushed a notification that is already done.

1506

if let Some(notify) = sources.notify {

1507

let live = written

1508

.iter()

1509

.filter(|item| !item.closing)

1510

.map(|item| {

1511

let live = live_notification(&item.event_id, &item.notice, &item.url, &rfc3339(now_ms()), Some(&item.thread));

1512

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

1513

})

1514

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

1515

tell_live(notify, live).await;

1516

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

1517

for username in people {

1518

if !settled.iter().any(|item| item.username == username) {

1519

tell_inbox_count(db, notify, username).await;

}

}

}

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

1524

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

}

}

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

1529

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

1530

/// identity's personal access token approvals.

1531

///

1532

1533

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

///

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

1541

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

1542

/// tells nobody).

1543

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

1544

"token.approval_requested",

1545

"token.approval_reviewed",

1546

"workspace_invitation.created",

1547

"workspace_invitation.accepted",

1548

"workspace_invitation.declined",

1549

];

1550

1551

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

1552

pub const INVITATION_ANSWERED: [&str; 3] =

1553

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

1554

1555

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

1556

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

1557

let data = &event.data;

1558

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

1559

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

1560

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

1561

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

1562

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

1563

};

1564

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

1565

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

1566

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

1567

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

1568

};

1569

let notices = names(data, "notify")

1570

.into_iter()

1571

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

1572

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

1573

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

1574

.into_iter()

1575

.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.

1581

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

1582

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();

1588

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

1589

let mut statements = Vec::new();

1590

for notice in &told {

1591

let username = notice.username.as_str();

1592

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

1593

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

1594

username.into(),

1595

thread.as_str().into(),

1596

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

1597

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

1598

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

1599

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

1600

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

1601

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

1602

workspace.as_str().into(),

JsValue::NULL,

JsValue::NULL,

JsValue::NULL,

JsValue::NULL,

JsValue::NULL,

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

1609

JsValue::NULL,

1610

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

1611

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)

1616

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

1617

)

1618

.bind(&[

1619

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

1620

username.into(),

1621

thread.as_str().into(),

1622

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

1623

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

1624

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

1625

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

1626

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

1627

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

1628

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

])?,

);

}

db.batch(statements).await?;

Ok(())

}

/// Settles items (see [`settles`]), in one batch. Returns the items it

1637

/// settled, by person and thread, so their counts can be told again and

1638

/// notify can stop replaying them.

1639

async fn settle(db: &D1Database, settled: &[Settle]) -> Result<HashSet<Settled>> {

1640

if settled.is_empty() {

1641

return Ok(HashSet::new());

1642

}

1643

let now = rfc3339(now_ms());

1644

let statements = settled.iter().map(|settle| settle.statement(db, &now)).collect::<Result<Vec<_>>>()?;

1645

let mut items = HashSet::new();

1646

for result in db.batch(statements).await? {

1647

items.extend(result.results::<Settled>()?);

}

Ok(items)

}

#[allow(clippy::too_many_arguments)]

1653

async fn deliver_one(

1654

db: &D1Database,

1655

sources: &Sources<'_>,

1656

names: &HashMap<String, String>,

1657

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

1658

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

1659

event: &Event,

1660

wanted: Wanted,

1661

settled: &mut HashSet<Settled>,

1662

) -> Result<Vec<Written>> {

1663

if let Some(repo_id) = &wanted.repo_id

1664

&& !paths.contains_key(repo_id)

1665

{

1666

let path: Option<RepoPath> = g1t_kit::call(sources.repos, "path_by_id", &PathByIdArgs { id: repo_id.clone() }).await?;

1667

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

1668

}

1669

let path = wanted.repo_id.as_ref().and_then(|repo_id| paths.get(repo_id).cloned().flatten());

1670

// A repository that is gone tells nobody; a work item stays in its

1671

// workspace when its origin goes.

1672

if path.is_none() && wanted.item.is_none() {

1673

return Ok(Vec::new());

1674

}

1675

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

1676

Some(number) => {

1677

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

sources.work,

"inbox_subject",

&InboxSubjectArgs {

repo_id: wanted.repo_id.clone(),

1682

number,

1683

item_id: wanted.item.as_ref().map(|item| item.id.clone()),

1684

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(),

1697

username: match event.actor.as_deref() {

1698

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

1699

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

None => None,

},

};

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

1704

// Whoever commented has answered what was on the thread for them.

1705

if let Some(handled) = handled_by(event, subject.as_ref()) {

1706

settled.extend(settle(db, &[handled]).await?);

1707

}

1708

// Who watches it: its repository's watchers, or a work item's linked

1709

// repositories', its origin's first.

1710

let linked: Vec<&str> = wanted.item.as_ref().map_or_else(Vec::new, |item| {

item.repo_ids

.iter()

.map(String::as_str)

.filter(|id| wanted.repo_id.as_deref() != Some(*id))

1715

.collect()

1716

});

1717

for repo_id in wanted.repo_id.iter().map(String::as_str).chain(linked.iter().copied()) {

1718

if !watchers.contains_key(repo_id) {

1719

watchers.insert(repo_id.to_owned(), subscriptions::watchers(db, repo_id).await?);

1720

}

1721

}

1722

let own = wanted.repo_id.as_ref().and_then(|id| watchers.get(id)).cloned().unwrap_or_default();

1723

let others: Vec<Vec<subscriptions::Watcher>> = linked.iter().map(|id| watchers.get(*id).cloned().unwrap_or_default()).collect();

1724

let (watching, elsewhere) = subscriptions::linked_watchers(own, &others);

1725

let audience = Audience {

1726

subscriptions: match thread.kind {

1727

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

_ => Vec::new(),

},

watchers: watching,

};

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

1733

// Where titles say it is: a work item's workspace, else the repository.

1734

let workspace = match &wanted.item {

1735

Some(item) => item.workspace.clone(),

1736

None => path.as_ref().map(|path| path.namespace.to_lowercase()).unwrap_or_default(),

1737

};

1738

let place = match &wanted.item {

1739

Some(item) => item.workspace.clone(),

1740

None => repo.clone().unwrap_or_default(),

1741

};

1742

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

1743

// Who hears of a work item only by watching another repository it is

1744

// linked to is told only if they may read it: a member of its

1745

// workspace, or anyone when its origin is public. Nothing of it (title,

1746

// body, push, email, count) reaches anyone else.

1747

if let Some(item) = &wanted.item {

1748

let hearsay: Vec<String> = told

1749

.iter()

1750

.filter(|notice| only_elsewhere(notice, &elsewhere))

1751

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

1752

.collect();

1753

if !hearsay.is_empty() {

1754

let asked = MembersAmongArgs { workspace: item.workspace.clone(), usernames: hearsay.clone() };

1755

let members: HashSet<String> = g1t_kit::call::<_, Vec<String>>(sources.identity, "members_among", &asked)

1756

.await

1757

.unwrap_or_else(|error| {

1758

worker::console_error!("inbox: could not tell who is a member of {}: {error}", item.workspace);

Vec::new()

})

.into_iter()

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

1763

.collect();

1764

let public = match &wanted.repo_id {

1765

Some(origin) if hearsay.iter().any(|name| !members.contains(name)) => {

1766

crate::live::public_of(sources.repos, vec![origin.clone()]).await.contains(origin)

}

_ => false,

};

told.retain(|notice| may_hear(notice, &elsewhere, &members, public));

}

}

let mut statements =

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

1775

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

1776

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

1777

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

1778

if let (Some(summary), Some(path)) = (security_summary(event), &path) {

1779

if summary.opened_count == 0 {

1780

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

1781

return Ok(Vec::new());

1782

}

1783

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

1784

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

1785

let now = now_ms();

1786

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

1787

for notice in told {

1788

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

1789

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

1790

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

1791

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

1792

Delivery::Never => {}

}

}

told = kept;

}

if told.is_empty() {

if !statements.is_empty() {

1799

db.batch(statements).await?;

1800

}

1801

return Ok(Vec::new());

1802

}

1803

let now = now_ms();

1804

let link = thread.link.as_deref();

1805

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

1806

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

1807

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

1808

let first = statements.len();

1809

for notice in &told {

1810

let username = notice.username.as_str();

1811

let values: Vec<JsValue> = vec![

1812

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

1813

username.into(),

1814

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

1815

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

1816

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

1817

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

1818

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

1819

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

1820

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

1821

workspace.as_str().into(),

1822

wanted.repo_id.as_deref().map_or(JsValue::NULL, JsValue::from),

1823

repo.as_deref().map_or(JsValue::NULL, JsValue::from),

1824

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

1825

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

1826

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

1827

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

1828

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

1829

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

1830

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

1831

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

1832

];

1833

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

1834

statements.push(

1835

db.prepare(

1836

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

1837

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(),

1842

username.into(),

1843

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

1844

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

1845

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

1846

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

1847

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

1848

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

1849

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

1850

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

1851

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

])?,

);

statements.push(

db.prepare(

"DELETE FROM inbox_activity

1857

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

1858

AND id NOT IN (

1859

SELECT a.id FROM inbox_activity a

1860

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

1861

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

1862

)

1863

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

1864

);

1865

}

1866

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

1867

#[derive(Deserialize)]

struct Told {

username: String,

}

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

1872

let mut news = HashSet::new();

1873

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

1874

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

1875

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

1876

{

1877

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

}

}

Ok(told

.into_iter()

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

1883

.map(|notice| Written {

1884

notice,

1885

repo_id: wanted.repo_id.clone(),

1886

workspace: wanted.item.as_ref().map(|item| item.workspace.clone()),

1887

url: item_url.clone(),

1888

event_id: event.id.clone(),

1889

closing: CLOSES.contains(&event.kind.as_str()),

1890

thread: thread.key.clone(),

})

.collect())

}

/// Whether a notice is someone's only by watching a repository a work item

1896

/// is linked to other than its origin: whether they may read it is not

1897

/// known from that.

1898

fn only_elsewhere(notice: &Notice, elsewhere: &HashSet<String>) -> bool {

1899

notice.reason == Reason::Subscribed && elsewhere.contains(&notice.username)

1900

}

1901

1902

/// Whether a work item's notice may be written: anyone told for a reason

1903

/// of their own, or who watches its origin; one who only watches another

1904

/// repository it is linked to (`elsewhere`) when they may read it, as a

1905

/// member of its workspace (`members`) or because its origin is public.

1906

/// An item with no origin is its workspace members' alone.

1907

fn may_hear(notice: &Notice, elsewhere: &HashSet<String>, members: &HashSet<String>, origin_public: bool) -> bool {

1908

!only_elsewhere(notice, elsewhere) || origin_public || members.contains(&notice.username)

1909

}

1910

1911

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

1912

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

1913

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

1914

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

1915

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

1916

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

1917

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

1918

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

1919

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

1920

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

1921

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

1922

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

1923

ON CONFLICT (username, thread) DO UPDATE SET

1924

event_id = excluded.event_id,

1925

event = excluded.event,

1926

reason = excluded.reason,

1927

severity = CASE

1928

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

1929

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

1930

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

1931

THEN inbox_items.severity ELSE excluded.severity END,

1932

title = excluded.title,

1933

body = excluded.body,

1934

workspace = excluded.workspace,

1935

repo = excluded.repo,

1936

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

1937

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

1938

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

1939

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

1940

actor = excluded.actor,

1941

updated_at = excluded.updated_at,

1942

bumped = excluded.bumped,

1943

activity = inbox_items.activity + 1,

1944

read_at = NULL,

1945

done_at = NULL,

1946

settled_reason = NULL,

1947

settled_by = NULL";

1948

1949

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

1950

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

1951

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

1952

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

1953

let kind = match notice.reason {

1954

Reason::ReviewRequested => "approval",

1955

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

1956

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

1957

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

1958

// choice, is an ordinary inbox item.

1959

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

1960

_ => "inbox",

1961

};

1962

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

1963

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

1964

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

1965

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

1966

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

1967

serde_json::json!({

1968

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

1969

"kind": kind,

1970

"workspace": workspace,

1971

"title": notice.title,

1972

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

1973

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

1974

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

1975

"inbox_thread": thread,

"created_at": at,

})

}

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

1981

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

1982

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

1983

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

1984

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

1985

}

1986

1987

/// What the notify service is told when some of a person's items settled

1988

/// (its `set_inbox` with `settled`): the count as it now is, and the

1989

/// threads that settled, whose notifications it then stops replaying to

1990

/// tabs that open later and drops from open tabs' lists.

1991

pub fn inbox_settled_update(username: &str, unread: u32, threads: &[String]) -> serde_json::Value {

1992

let mut threads: Vec<&str> = threads.iter().map(String::as_str).collect();

1993

threads.sort_unstable();

1994

threads.dedup();

1995

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

1996

}

1997

1998

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

1999

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

2000

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

2001

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

2002

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

2003

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

2004

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

2005

Err(error) => Err(error),

2006

};

2007

if let Err(error) = told {

2008

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

}

}

/// Tells the notify service a person's count after items of theirs

2013

/// settled, with the threads that did. Logged and left when it fails.

2014

pub async fn tell_inbox_settled(db: &D1Database, notify: &Fetcher, username: &str, threads: Vec<String>) {

2015

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

2016

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

2017

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

2018

Err(error) => Err(error),

2019

};

2020

if let Err(error) = told {

2021

worker::console_error!("inbox: settled items not told: {error}");

}

}

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

2026

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

2027

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

2028

#[derive(serde::Serialize)]

2029

struct NotifyArgs {

2030

username: String,

2031

notification: serde_json::Value,

2032

}

2033

for (username, notification) in items {

2034

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

2035

if let Err(error) = told {

2036

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

}

}

}

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

2042

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

2043

/// can still see it: they can read its repository, or are a member of a

2044

/// work item's workspace (all one filed with no repository has).

2045

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

2046

let written: Vec<Written> = written.into_iter().filter(|item| item.repo_id.is_some() || item.workspace.is_some()).collect();

2047

if written.is_empty() {

2048

return Ok(());

2049

}

2050

let people: Vec<String> = written

2051

.iter()

2052

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

2053

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

2054

.into_iter()

2055

.collect();

2056

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

2057

for item in written {

2058

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

2059

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

2060

continue;

2061

}

2062

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

identity,

"notify_by_email",

&NotifyByEmailArgs {

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

2067

repo_id: item.repo_id.clone(),

2068

workspace: item.workspace.clone(),

2069

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

2070

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

2071

quote: None,

2072

path: item.url.clone(),

2073

reason: item.notice.reason,

},

)

.await;

if let Err(error) = sent {

2078

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

}

}

Ok(())

}

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

2085

2086

#[derive(Deserialize)]

2087

pub(crate) struct Row {

2088

pub(crate) id: String,

2089

#[serde(default)]

2090

thread: Option<String>,

reason: String,

severity: String,

title: String,

body: String,

event: Option<String>,

2096

workspace: Option<String>,

2097

pub(crate) repo_id: Option<String>,

2098

repo: Option<String>,

2099

subject: Option<String>,

2100

number: Option<f64>,

2101

run_id: Option<String>,

2102

link: Option<String>,

2103

actor: Option<String>,

2104

activity: Option<f64>,

2105

created_at: String,

2106

updated_at: Option<String>,

2107

read_at: Option<String>,

2108

done_at: Option<String>,

2109

saved: f64,

2110

snoozed_until: Option<String>,

2111

bumped: Option<String>,

2112

settled_reason: Option<String>,

2113

settled_by: Option<String>,

}

impl Row {

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

2118

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

2119

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

2120

// A work item whose thread still has its issue's number, until the

2121

// work service's numbers arrive (work_items.rs): the old link,

2122

// which leads to it.

2123

let legacy = subject == Some(SubjectKind::Item) && self.thread.as_deref().is_some_and(|thread| !thread.starts_with(ITEM_THREAD));

2124

let link = match (legacy, &self.repo, number) {

2125

(true, Some(repo), Some(number)) => Some(format!("/{repo}/issues/{number}")),

2126

_ => self.link.clone(),

2127

};

2128

InboxItem {

2129

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

2130

id: self.id,

2131

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

2132

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),

2142

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

2143

created_at: self.created_at,

2144

read_at: self.read_at,

2145

done_at: self.done_at,

2146

saved: self.saved != 0.0,

2147

snoozed_until: self.snoozed_until,

2148

settled_reason: self.settled_reason,

2149

settled_by: self.settled_by,

}

}

}

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

2155

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

2156

settled_reason, settled_by";

2157

2158

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

2159

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

2160

match view {

2161

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

2162

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

2163

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

}

}

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

2168

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

2169

a.view == InboxView::Inbox

2170

&& a.severity.is_none()

2171

&& a.reason.is_none()

2172

&& !a.participating

2173

&& a.repo_id.is_none()

2174

&& !a.unread

2175

&& a.since.is_none()

2176

&& a.updated_before.is_none()

2177

}

2178

2179

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

2180

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

2181

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

2182

let mut conditions = Vec::new();

2183

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

2184

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

2185

values.push(value.to_owned());

2186

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

2187

};

2188

if let Some(severity) = a.severity {

2189

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

2190

}

2191

if let Some(reason) = a.reason {

2192

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

2193

}

2194

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

2195

bind(repo_id, "repo_id = ?");

2196

}

2197

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

2198

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

2199

}

2200

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

2201

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

2202

}

2203

if a.participating {

2204

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

2205

}

2206

if a.unread {

2207

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

}

(conditions, values)

}

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

2213

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

2214

return Ok(InboxPage::default());

2215

};

2216

let username = viewer.username.to_lowercase();

2217

let now = rfc3339(now_ms());

2218

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

2219

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

2220

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

2221

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

2222

conditions.extend(filtered);

2223

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

2224

let ranked = ranked(&a);

2225

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

2226

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

2227

let mut rest = conditions.clone();

2228

let mut rest_values = values.clone();

2229

if ranked {

2230

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

2231

}

2232

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

2233

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

2234

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

2235

}

2236

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

2237

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

2238

let mut statements = vec![

2239

db.prepare(format!(

2240

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

rest.join(" AND "),

rest_values.len()

))

.bind(&rest_values)?,

2245

];

2246

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

2247

statements.push(

2248

db.prepare(format!(

2249

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

2250

conditions.join(" AND ")

2251

))

2252

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

2253

);

2254

}

2255

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

2256

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

2257

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

2258

rows.truncate(limit as usize);

2259

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();

2264

items.extend(rows);

2265

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

2266

Ok(InboxPage {

2267

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

next,

})

}

/// The workspaces whose work items the viewer sees all of: those they are

2273

/// a member of, lowercased.

2274

fn member_of(viewer: &g1t_contracts::Viewer) -> Vec<String> {

2275

let Some(user) = viewer else {

return Vec::new();

};

user.workspaces

.iter()

.filter(|membership| g1t_contracts::access::workspace_role(user, &membership.slug).is_some())

2281

.map(|membership| membership.slug.to_lowercase())

.collect()

}

impl Row {

/// A work item of one of these workspaces, which a member sees

2287

/// whatever its origin.

2288

fn seen_as_member(&self, members: &[String]) -> bool {

2289

self.subject.as_deref() == Some(SubjectKind::Item.as_str())

2290

&& self.workspace.as_deref().is_some_and(|slug| members.iter().any(|member| member.eq_ignore_ascii_case(slug)))

}

}

/// Rows the viewer may still see: about repositories they can read, and

2295

/// work items of workspaces they belong to or whose origin they can read.

2296

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

2297

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

2298

let members = member_of(viewer);

2299

// A work item outside the viewer's workspaces with no origin is not theirs to see.

2300

let unseen: Vec<String> = rows

2301

.iter()

2302

.filter(|row| row.subject.as_deref() == Some(SubjectKind::Item.as_str()) && row.repo_id.is_none() && !row.seen_as_member(&members))

2303

.map(|row| row.id.clone())

2304

.collect();

2305

let ids: Vec<String> = rows

2306

.iter()

2307

.filter(|row| !row.seen_as_member(&members))

2308

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

2309

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

2310

.into_iter()

2311

.collect();

2312

let gone: Vec<String> = if ids.is_empty() {

2313

Vec::new()

2314

} else {

2315

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();

2325

ids.into_iter().filter(|id| !readable.contains(id)).collect()

2326

};

2327

if !gone.is_empty() || !unseen.is_empty() {

2328

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

2329

rows.retain(|row| {

2330

!unseen.contains(&row.id) && (row.seen_as_member(&members) || row.repo_id.as_ref().is_none_or(|id| !gone.contains(id)))

});

}

Ok(rows)

}

/// Removes a person's items about repositories they cannot read (but for

2337

/// work items of the workspaces they are a member of), and the items by

2338

/// id in `unseen`.

2339

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

2340

let which = "username = ?1 AND (

2341

(repo_id IN (SELECT value FROM json_each(?2))

2342

AND NOT (subject = 'item' AND workspace IN (SELECT value FROM json_each(?3))))

2343

OR id IN (SELECT value FROM json_each(?4)))";

2344

let values: Vec<JsValue> = vec![

2345

username.into(),

2346

serde_json::to_string(repo_ids)?.into(),

2347

serde_json::to_string(members)?.into(),

2348

serde_json::to_string(unseen)?.into(),

2349

];

2350

db.batch(vec![

2351

db.prepare(format!("DELETE FROM inbox_activity WHERE item_id IN (SELECT id FROM inbox_items WHERE {which})"))

2352

.bind(&values)?,

2353

db.prepare(format!("DELETE FROM inbox_items WHERE {which}")).bind(&values)?,

])

.await?;

Ok(())

}

#[derive(Deserialize)]

struct CountRow {

severity: String,

n: f64,

}

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

2366

let rows = db

2367

.prepare(

2368

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

2369

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

2370

AND (snoozed_until IS NULL OR snoozed_until <= ?)

2371

GROUP BY severity",

2372

)

2373

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

2374

.all()

2375

.await?

2376

.results::<CountRow>()?;

2377

let mut counts = InboxCounts::default();

2378

for row in rows {

2379

let n = row.n as u32;

2380

counts.unread += n;

2381

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

2382

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

2383

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

2384

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

2385

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

}

}

Ok(counts)

}

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

2392

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

2393

match mark {

2394

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

2395

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

2396

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

2397

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

2398

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

2399

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

2400

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

2401

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

}

}

#[derive(Deserialize)]

struct IdRow {

#[allow(dead_code)]

id: String,

}

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

2412

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

2413

let now = rfc3339(now_ms());

2414

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

2415

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

2416

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

2417

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

2418

_ => String::new(),

2419

};

2420

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

2421

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

2422

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

2423

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

2424

values.push(value.into());

2425

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

2426

};

2427

if let Some(severity) = a.severity {

2428

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

2429

}

2430

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

2431

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

2432

}

2433

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

2434

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;

2443

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

2444

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

2445

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>()?;

2456

Ok(changed.len() as u32)

2457

}

2458

2459

#[derive(Deserialize)]

struct ActivityRow {

reason: String,

severity: String,

title: String,

body: String,

event: Option<String>,

2466

actor: Option<String>,

created_at: String,

}

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

2471

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

2472

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

2473

return Ok(None);

2474

};

2475

let username = viewer.username.to_lowercase();

2476

let row = db

2477

.prepare(format!("SELECT {COLUMNS} FROM inbox_items WHERE id = ? AND username = ?"))

2478

.bind(&[a.id.as_str().into(), username.as_str().into()])?

2479

.first::<Row>(None)

2480

.await?;

2481

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

2487

WHERE item_id = ? ORDER BY id DESC LIMIT ?",

2488

)

2489

.bind(&[row.id.as_str().into(), MAX_ACTIVITY.into()])?

2490

.all()

2491

.await?

2492

.results::<ActivityRow>()?

2493

.into_iter()

2494

.map(|row| InboxActivity {

2495

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

2496

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,

2502

})

2503

.collect();

2504

let item = row.into_item();

2505

let subscription = match (item.subject, item.number) {

2506

(Some(SubjectKind::Item | SubjectKind::Pull), Some(_)) => {

2507

subscriptions::subscription(

db,

work,

SubscriptionArgs {

viewer: a.viewer.clone(),

2512

id: Some(item.id.clone()),

2513

..SubscriptionArgs::default()

},

)

.await?

}

_ => None,

};

Ok(Some(InboxThread {

item,

activity,

subscription,

}))

}

// --- Keeping up ------------------------------------------------------------------

2528

2529

/// Moves rows with renamed workspaces and repositories, drops those of

2530

/// purged repositories, deleted workspaces and purged accounts, and

2531

/// rewrites what was kept by issue number as issues become work items

2532

/// (work_items.rs).

2533

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

2534

let mut statements = Vec::new();

2535

for event in events {

2536

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

2537

match event.kind.as_str() {

2538

"workspace.renamed" => {

2539

let Ok(renamed) = serde_json::from_value::<WorkspaceRenamed>(event.data.clone()) else {

2540

continue;

2541

};

2542

let (from, to) = (renamed.from.to_lowercase(), renamed.to.to_lowercase());

if from == to {

continue;

}

// A work item's origin may be in another workspace by now.

2547

statements.push(

2548

db.prepare(

2549

"UPDATE inbox_items SET workspace = ?2,

2550

repo = CASE WHEN repo LIKE ?1 || '/%' THEN ?2 || substr(repo, length(?1) + 1) ELSE repo END,

2551

link = CASE WHEN link LIKE '/' || ?1 || '/%' THEN '/' || ?2 || substr(link, length(?1) + 2) ELSE link END

2552

WHERE workspace = ?1",

2553

)

2554

.bind(&[from.as_str().into(), to.as_str().into()])?,

2555

);

2556

statements.push(

2557

db.prepare("UPDATE inbox_items SET repo = ?2 || substr(repo, length(?1) + 1) WHERE workspace IS NOT ?2 AND repo LIKE ?1 || '/%'")

2558

.bind(&[from.as_str().into(), to.as_str().into()])?,

2559

);

2560

statements.push(

2561

db.prepare("UPDATE inbox_watching SET repo = ?2 || substr(repo, length(?1) + 1) WHERE repo LIKE ?1 || '/%'")

2562

.bind(&[from.as_str().into(), to.as_str().into()])?,

);

}

"repo.renamed" => {

let path = format!("{}/{}", text("namespace"), text("to"));

2567

for table in ["inbox_items", "inbox_watching"] {

2568

statements.push(

2569

db.prepare(format!("UPDATE {table} SET repo = ? WHERE repo_id = ?"))

2570

.bind(&[path.as_str().into(), text("repoId").into()])?,

);

}

}

"repo.transferred" => {

2575

let path = format!("{}/{}", text("to"), text("name"));

2576

// A work item stays in its workspace when its origin moves.

2577

statements.push(

2578

db.prepare(

2579

"UPDATE inbox_items SET repo = ?1, workspace = CASE WHEN subject = 'item' THEN workspace ELSE ?2 END

2580

WHERE repo_id = ?3",

2581

)

2582

.bind(&[path.as_str().into(), text("to").into(), text("repoId").into()])?,

2583

);

2584

statements.push(

2585

db.prepare("UPDATE inbox_watching SET repo = ? WHERE repo_id = ?")

2586

.bind(&[path.as_str().into(), text("repoId").into()])?,

2587

);

2588

}

2589

// Its work items stay in their workspace, with no origin now.

2590

"repo.purged" => {

2591

for sql in [

2592

"UPDATE inbox_items SET repo_id = NULL, repo = NULL WHERE repo_id = ?1 AND subject = 'item'",

2593

"UPDATE inbox_subscriptions SET repo_id = NULL WHERE repo_id = ?1 AND thread LIKE 'item:%'",

2594

"DELETE FROM inbox_activity WHERE item_id IN (SELECT id FROM inbox_items WHERE repo_id = ?1)",

2595

"DELETE FROM inbox_items WHERE repo_id = ?1",

2596

"DELETE FROM inbox_subscriptions WHERE repo_id = ?1",

2597

"DELETE FROM inbox_watching WHERE repo_id = ?1",

2598

] {

2599

statements.push(db.prepare(sql).bind(&[text("repoId").into()])?);

2600

}

2601

}

2602

"workspace.deleted" => {

2603

statements.push(

2604

db.prepare("DELETE FROM inbox_items WHERE workspace = ?")

2605

.bind(&[text("slug").into()])?,

2606

);

2607

statements.push(

2608

db.prepare("DELETE FROM inbox_watching WHERE repo LIKE ? || '/%'")

2609

.bind(&[text("slug").into()])?,

2610

);

2611

}

2612

// Issues moved into work items: what was kept by their numbers follows.

2613

"work.migrated" => statements.extend(crate::work_items::moved(db, event)?),

2614

// An account purged: its inbox, what it watched and its

2615

// settings go with it (identity's account_deletion.rs).

2616

"user.deleted" => {

2617

let username = text("username");

2618

if username.is_empty() {

continue;

}

for sql in [

"DELETE FROM inbox_activity WHERE username = ?",

2623

"DELETE FROM inbox_items WHERE username = ?",

2624

"DELETE FROM inbox_subscriptions WHERE username = ?",

2625

"DELETE FROM inbox_watching WHERE username = ?",

2626

"DELETE FROM inbox_settings WHERE username = ?",

2627

] {

2628

statements.push(db.prepare(sql).bind(&[username.as_str().into()])?);

}

}

_ => {}

}

}

if !statements.is_empty() {

2635

db.batch(statements).await?;

}

Ok(())

}

/// Removes items done more than [`DONE_DAYS`] ago, and any not saved older

2641

/// than [`MAX_DAYS`], with their history. Returns how many went.

2642

pub async fn purge(db: &D1Database, now: u64) -> Result<u32> {

2643

let done = crate::audit::keep_from(now, DONE_DAYS);

2644

let oldest = crate::audit::keep_from(now, MAX_DAYS);

2645

let statements: Vec<D1PreparedStatement> = vec![

2646

db.prepare("DELETE FROM inbox_items WHERE saved = 0 AND done_at < ?")

2647

.bind(&[done.into()])?,

2648

db.prepare("DELETE FROM inbox_items WHERE saved = 0 AND COALESCE(updated_at, created_at) < ?")

2649

.bind(&[oldest.into()])?,

2650

db.prepare("DELETE FROM inbox_activity WHERE NOT EXISTS (SELECT 1 FROM inbox_items WHERE inbox_items.id = inbox_activity.item_id)"),

2651

];

2652

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

2653

let mut removed = 0;

2654

for result in results.into_iter().take(2) {

2655

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};

2664

use serde_json::json;

2665

2666

fn event(kind: &str, actor: Option<&str>, data: serde_json::Value) -> Event {

Event {

id: "evt_1".into(),

kind: kind.into(),

source: "work".into(),

2671

time: "2026-10-07T12:00:00.000Z".into(),

2672

repo_id: Some("rep_1".into()),

2673

actor: actor.map(str::to_owned),

data,

}

}

fn person(id: &str, username: &str) -> Principal {

2679

Principal {

2680

id: id.into(),

2681

username: username.into(),

}

}

fn actor(id: &str, username: &str) -> Actor {

2686

Actor {

2687

id: Some(id.into()),

2688

username: Some(username.into()),

}

}

/// A pull request ana opened, assigned to bo, for a work item cy filed and dee is assigned.

2693

fn pull() -> InboxSubject {

2694

InboxSubject {

2695

kind: Some(SubjectKind::Pull),

2696

title: "Add the inbox".into(),

2697

author: person("usr_ana", "ana"),

2698

assignees: vec!["bo".into()],

2699

item: Some(Box::new(InboxSubject {

2700

kind: Some(SubjectKind::Item),

2701

title: "An inbox".into(),

2702

author: person("usr_cy", "cy"),

2703

assignees: vec!["dee".into()],

2704

..InboxSubject::default()

2705

})),

2706

..InboxSubject::default()

}

}

/// A change g1t made for ana.

2711

fn g1t_pull() -> InboxSubject {

2712

InboxSubject {

2713

author: person(AGENT_ID, "g1t"),

2714

requested_by: Some(person("usr_ana", "ana")),

..pull()

}

}

fn nobody() -> Audience {

Audience::default()

}

fn told(notices: &[Notice]) -> Vec<(&str, Reason, Severity)> {

2724

notices

2725

.iter()

2726

.map(|notice| (notice.username.as_str(), notice.reason, notice.severity))

.collect()

}

fn comment(author: Principal, body: &str, mentions: &[&str]) -> InboxSubject {

2731

InboxSubject {

2732

comment: Some(InboxComment {

2733

author,

2734

excerpt: body.into(),

2735

mentions: mentions.iter().map(|name| (*name).to_owned()).collect(),

2736

..InboxComment::default()

}),

..pull()

}

}

fn subscribed(rows: &[(&str, State, Option<Reason>)]) -> Audience {

Audience {

subscriptions: rows

.iter()

.map(|(name, state, reason)| Subscription {

2747

username: (*name).into(),

state: *state,

reason: *reason,

})

.collect(),

watchers: Vec::new(),

}

}

fn watched(rows: &[(&str, WatchLevel, &[&str])]) -> Audience {

2757

Audience {

2758

subscriptions: Vec::new(),

2759

watchers: rows

2760

.iter()

2761

.map(|(name, level, events)| Watcher {

2762

username: (*name).into(),

2763

level: *level,

2764

events: events.iter().map(|kind| (*kind).to_owned()).collect(),

})

.collect(),

}

}

#[test]

fn only_events_that_tell_someone_are_read() {

2772

let asked = |kind: &str, data| wants(&event(kind, None, data));

2773

assert_eq!(

2774

asked("checks.completed", json!({ "repoId": "rep_1", "number": 4, "status": "failed" })),

2775

Some(Wanted { repo_id: Some("rep_1".into()), number: Some(4), ..Wanted::default() })

2776

);

2777

assert_eq!(asked("checks.completed", json!({ "repoId": "rep_1", "number": 4, "status": "passed" })), None);

2778

assert_eq!(asked("review.completed", json!({ "repoId": "rep_1", "number": 4, "verdict": "approve" })), None);

2779

assert_eq!(asked("agent.asked", json!({ "repoId": "rep_1", "number": 4 })), None);

2780

assert_eq!(asked("workflow.completed", json!({ "repoId": "rep_1", "conclusion": "success" })), None);

2781

assert_eq!(

2782

asked("workflow.completed", json!({ "repoId": "rep_1", "conclusion": "failure" })),

2783

Some(Wanted { repo_id: Some("rep_1".into()), ..Wanted::default() })

2784

);

2785

assert_eq!(

2786

asked("comment.created", json!({ "repoId": "rep_1", "number": 2, "commentId": "cmt_1" })).map(|w| w.comment_id),

2787

Some(Some("cmt_1".into()))

2788

);

2789

assert_eq!(asked("git.push", json!({ "repoId": "rep_1" })), None);

2790

for kind in ["pull.review_requested", "pull.stalled", "pull.closed"] {

2791

assert!(asked(kind, json!({ "repoId": "rep_1", "number": 1 })).is_some(), "{kind}");

2792

}

2793

assert_eq!(asked("deployment.failed", json!({ "repoId": "rep_1" })).map(|w| w.number), Some(None));

}

#[test]

fn a_work_items_events_are_read_by_its_id_with_or_without_an_origin() {

2798

let item = json!({ "itemId": "wki_1", "workspace": "Acme", "number": 40, "repoIds": ["rep_1", "rep_2"] });

2799

for kind in ITEM_TOLD {

2800

let wanted = wants(&event(kind, None, item.clone())).unwrap_or_else(|| panic!("{kind}"));

2801

assert_eq!(wanted.repo_id.as_deref(), Some("rep_1"));

2802

assert_eq!(wanted.number, Some(40));

2803

assert_eq!(

2804

wanted.item,

2805

Some(WantedItem { id: "wki_1".into(), workspace: "acme".into(), repo_ids: vec!["rep_1".into(), "rep_2".into()] })

2806

);

2807

}

2808

// Filed in the workspace: no origin.

2809

let filed = Event { repo_id: None, ..event("work_item.opened", None, item.clone()) };

2810

assert_eq!(wants(&filed).map(|wanted| wanted.repo_id), Some(None));

2811

// A comment on one names it; one on a pull request does not.

2812

let comment = json!({ "itemId": "wki_1", "workspace": "acme", "number": 40, "commentId": "cmt_1", "repoIds": [] });

2813

let wanted = wants(&Event { repo_id: None, ..event("comment.created", None, comment) }).unwrap();

2814

assert_eq!((wanted.comment_id.as_deref(), wanted.item.map(|item| item.id)), (Some("cmt_1"), Some("wki_1".into())));

2815

// Without its id there is nothing to read.

2816

assert_eq!(wants(&event("work_item.opened", None, json!({ "number": 40 }))), None);

2817

// What the inbox does not tell of.

2818

assert_eq!(wants(&event("work_item.labeled", None, item)), None);

}

#[test]

fn a_work_item_is_one_thread_in_its_workspace() {

2823

let data = json!({ "itemId": "wki_1", "workspace": "acme", "number": 40, "repoIds": ["rep_1"] });

2824

let opened = event("work_item.opened", None, data.clone());

2825

let wanted = wants(&opened).unwrap();

2826

let thread = thread_of(&opened, &wanted, None);

2827

assert_eq!(thread.key, "item:wki_1");

2828

assert_eq!(thread.kind, Some(SubjectKind::Item));

2829

assert_eq!(url(Some("acme/rocket"), Some("acme"), thread.kind, thread.number, None, None), "/acme/-/work/40");

2830

// Closing it settles everything on it, and says what landed it.

let closed = event(

"work_item.closed",

None,

json!({ "itemId": "wki_1", "workspace": "acme", "number": 40, "resolution": { "repo": "acme/api", "pull": 9 } }),

2835

);

2836

assert_eq!(closes(&closed, Some("bo")), vec![Settle::thread("item:wki_1").saying("Closed", Some("bo"))]);

2837

let item = InboxSubject { kind: Some(SubjectKind::Item), title: "Ship it".into(), author: person("usr_cy", "cy"), ..InboxSubject::default() };

2838

assert_eq!(notices(&closed, "acme", &Actor::default(), Some(&item), &nobody())[0].title, "acme#40 was closed by acme/api#9");

2839

// A comment on it is handled on its thread.

2840

let comment = event("comment.created", None, json!({ "itemId": "wki_1", "workspace": "acme", "number": 40, "commentId": "cmt_1" }));

2841

let on = InboxSubject {

2842

comment: Some(InboxComment { author: person("usr_bo", "bo"), ..InboxComment::default() }),

2843

..item

2844

};

2845

assert_eq!(handled_by(&comment, Some(&on)).map(|settle| settle.scope), Some(Scope::Thread("item:wki_1".into())));

2846

assert!(subscribable("item:wki_1") && subscribable("rep_1#7") && !subscribable("rep_1/run/ci@main"));

}

#[test]

fn a_work_items_origin_is_its_events_repository_alone() {

2851

// Filed in the workspace and linked to rep_2: no origin, whatever

2852

// its data's `repoId` or the first of its `repoIds` say.

2853

let data = json!({ "itemId": "wki_1", "workspace": "acme", "number": 40, "commentId": "cmt_1", "repoId": "rep_2", "repoIds": ["rep_2"] });

2854

for kind in ["work_item.opened", "comment.created"] {

2855

let wanted = wants(&Event { repo_id: None, ..event(kind, None, data.clone()) }).unwrap_or_else(|| panic!("{kind}"));

2856

assert_eq!(wanted.repo_id, None, "{kind}");

2857

}

2858

// With an origin, the event's repository, not its data's.

2859

let wanted = wants(&event("comment.created", None, data)).unwrap();

2860

assert_eq!(wanted.repo_id.as_deref(), Some("rep_1"));

}

#[test]

fn who_only_watches_a_linked_repository_hears_of_a_work_item_they_may_read() {

2865

let notice = |username: &str, reason: Reason| Notice {

2866

username: username.into(),

2867

reason,

2868

severity: Severity::Info,

2869

title: "acme#40 Secret plans".into(),

2870

body: String::new(),

2871

};

2872

let set = |names: &[&str]| names.iter().map(|name| (*name).to_owned()).collect::<HashSet<String>>();

2873

let elsewhere = set(&["cy", "dee"]);

2874

let members = set(&["dee"]);

2875

// Its origin's watchers and whoever is told for a reason of their own.

2876

assert!(may_hear(&notice("ana", Reason::Subscribed), &elsewhere, &members, false));

2877

assert!(may_hear(&notice("cy", Reason::Mention), &elsewhere, &members, false));

2878

// A member of its workspace may read it.

2879

assert!(may_hear(&notice("dee", Reason::Subscribed), &elsewhere, &members, false));

2880

// Anyone else only when its origin is public.

2881

assert!(!may_hear(&notice("cy", Reason::Subscribed), &elsewhere, &members, false));

2882

assert!(may_hear(&notice("cy", Reason::Subscribed), &elsewhere, &members, true));

}

#[test]

fn a_work_items_watchers_are_its_origins_then_its_other_repositories() {

2887

let watcher = |name: &str, level: WatchLevel| Watcher { username: name.into(), level, events: Vec::new() };

2888

let own = vec![watcher("ana", WatchLevel::All), watcher("bo", WatchLevel::Ignore)];

2889

let others = vec![vec![watcher("bo", WatchLevel::All), watcher("cy", WatchLevel::All), watcher("dee", WatchLevel::Ignore)]];

2890

let (watchers, elsewhere) = subscriptions::linked_watchers(own, &others);

2891

let names: Vec<(&str, WatchLevel)> = watchers.iter().map(|w| (w.username.as_str(), w.level)).collect();

2892

// Ignoring the origin stands; ignoring another repository says nothing.

2893

assert_eq!(names, vec![("ana", WatchLevel::All), ("bo", WatchLevel::Ignore), ("cy", WatchLevel::All)]);

2894

assert_eq!(elsewhere.into_iter().collect::<Vec<_>>(), vec!["cy".to_owned()]);

}

#[test]

fn a_stopped_pull_request_needs_its_people_and_its_work_items_people_first() {

2899

let stalled = event("pull.stalled", None, json!({ "number": 7 }));

2900

let notices_ = notices(&stalled, "acme/rocket", &Actor::default(), Some(&pull()), &nobody());

assert_eq!(

told(&notices_),

vec![

("ana", Reason::Agent, Severity::Warning),

2905

("cy", Reason::Agent, Severity::Warning),

2906

("dee", Reason::Agent, Severity::Warning),

2907

]

2908

);

2909

assert_eq!(notices_[0].title, "g1t stopped on acme/rocket#7 and needs you");

2910

assert_eq!(notices_[0].body, "Add the inbox");

2911

// Stopping says why; even someone who unsubscribed hears of it.

2912

let stalled = event("pull.stalled", None, json!({ "number": 7, "detail": "Its checks could not be run." }));

2913

let audience = subscribed(&[("ana", State::Unsubscribed, None)]);

2914

let notices_ = notices(&stalled, "acme/rocket", &Actor::default(), Some(&g1t_pull()), &audience);

2915

assert_eq!(notices_[0].username, "ana");

2916

assert_eq!(notices_[0].title, "g1t stopped on acme/rocket#7 and needs you");

2917

assert_eq!(notices_[0].body, "Its checks could not be run.");

2918

// Ignoring the thread silences even that.

2919

let audience = subscribed(&[("ana", State::Ignored, None)]);

2920

let notices_ = notices(&stalled, "acme/rocket", &Actor::default(), Some(&g1t_pull()), &audience);

2921

assert!(!notices_.iter().any(|notice| notice.username == "ana"));

}

#[test]

fn whoever_turned_on_auto_merge_hears_it_is_off() {

2926

let off = event("pull.auto_merge_disabled", None, json!({ "number": 7, "autoMergeBy": "bo", "detail": "Its branch conflicts with main." }));

2927

let notices_ = notices(&off, "acme/rocket", &Actor::default(), Some(&pull()), &nobody());

2928

assert_eq!(told(&notices_), vec![("bo", Reason::StateChange, Severity::Warning)]);

2929

assert_eq!(notices_[0].title, "Auto-merge was turned off on acme/rocket#7");

2930

assert_eq!(notices_[0].body, "Its branch conflicts with main.");

2931

// Turning it off yourself tells no one.

2932

let notices_ = notices(&off, "acme/rocket", &actor("usr_bo", "bo"), Some(&pull()), &nobody());

2933

assert!(notices_.is_empty());

}

#[test]

fn g1t_handing_a_pull_request_to_the_merge_queue_tells_no_one() {

2938

let off = event(

2939

"pull.auto_merge_disabled",

2940

None,

2941

json!({ "number": 7, "autoMergeBy": "bo", "detail": "It joined the merge queue.", "queued": true }),

2942

);

2943

assert!(notices(&off, "acme/rocket", &Actor::default(), Some(&pull()), &nobody()).is_empty());

}

#[test]

fn whatever_an_agent_waits_on_closes_when_it_goes_on() {

2948

for kind in RESUMES {

2949

let settled = settles(&event(kind, None, json!({ "repoId": "rep_1", "number": 7 })), Some("ana"));

2950

let agent = settled.iter().find(|settle| settle.reasons == [Reason::Agent]).unwrap_or_else(|| panic!("{kind}"));

2951

assert_eq!(agent.scope, Scope::Thread("rep_1#7".into()));

2952

assert_eq!(agent.why.as_deref(), Some(resumed_because(kind)));

2953

assert_eq!(agent.by.as_deref(), Some("ana"));

2954

}

2955

assert_eq!(resumed_because("pull.updated"), "New commits were pushed");

2956

assert!(settles(&event("comment.created", None, json!({ "repoId": "rep_1", "number": 7 })), None).is_empty());

}

#[test]

fn a_merge_or_a_close_settles_everything_about_it() {

2961

let merged = event("pull.merged", Some("usr_ana"), json!({ "repoId": "rep_1", "number": 7 }));

2962

assert_eq!(

2963

closes(&merged, Some("Ana")),

2964

vec![

2965

Settle::thread("rep_1#7").saying("Merged", Some("ana")),

2966

Settle::matching("rep_1/deploy/*/7").saying("Merged", Some("ana")),

2967

]

2968

);

2969

let closed = event("pull.closed", None, json!({ "repoId": "rep_1", "number": 7 }));

2970

assert_eq!(closes(&closed, None).len(), 2);

2971

assert_eq!(closes(&closed, None)[0].why.as_deref(), Some("Closed"));

2972

assert_eq!(closes(&closed, None)[0].by, None);

2973

// A work item has no previews.

2974

let item = event("work_item.closed", None, json!({ "itemId": "wki_1", "workspace": "acme", "number": 3, "reason": "not_planned" }));

2975

assert_eq!(closes(&item, Some("bo")), vec![Settle::thread("item:wki_1").saying("Closed as not planned", Some("bo"))]);

2976

// Reopening is something to hear of, and settles nothing.

2977

assert!(closes(&event("work_item.reopened", None, json!({ "itemId": "wki_1", "number": 3 })), None).is_empty());

2978

assert!(settles(&event("work_item.reopened", None, json!({ "itemId": "wki_1", "number": 3 })), None).is_empty());

}

#[test]

fn a_workflow_passing_on_a_branch_settles_its_failures_there() {

2983

let data = |conclusion: &str| {

2984

json!({ "repoId": "rep_1", "runId": "run_9", "workflow": "Deploy", "path": ".g1t/workflows/deploy.yml", "ref": "refs/heads/main", "conclusion": conclusion, "number": 328 })

2985

};

2986

let failed = event("workflow.completed", Some("usr_bo"), data("failure"));

2987

let thread = thread_of(&failed, &wants(&failed).unwrap(), None).key;

2988

assert_eq!(thread, "rep_1/run/.g1t/workflows/deploy.yml@main");

2989

let passed = event("workflow.completed", Some("usr_bo"), data("success"));

2990

assert!(wants(&passed).is_none(), "a success tells no one");

2991

// Whoever started the run did not settle anything: the run passing did.

2992

assert_eq!(

2993

settles(&passed, Some("bo")),

2994

vec![

2995

Settle::thread(thread.clone()).saying("Deploy passed in run 328", None),

2996

Settle::matching("rep_1/review/run_9/*").because(&[Reason::ReviewRequested]).saying("Deploy finished in run 328", None),

2997

]

2998

);

2999

// A failure or a cancelled run settles no failure, but ends what its deployments waited for.

3000

for conclusion in ["failure", "cancelled"] {

3001

assert_eq!(

3002

settles(&event("workflow.completed", None, data(conclusion)), None),

3003

vec![Settle::matching("rep_1/review/run_9/*").because(&[Reason::ReviewRequested]).saying("Deploy finished in run 328", None)]

3004

);

3005

}

3006

// On a pull request, its checks passing settles the failure instead.

3007

let on_pull = event("workflow.completed", None, json!({ "repoId": "rep_1", "runId": "run_9", "conclusion": "success", "pull": 7 }));

3008

assert!(!settles(&on_pull, None).iter().any(|settle| settle.scope == Scope::Thread(thread.clone())));

}

#[test]

fn checks_passing_settle_the_failure_its_owner_was_told_of() {

3013

let passed = event("checks.completed", None, json!({ "repoId": "rep_1", "number": 7, "status": "passed", "commit": "abcdef0123456" }));

3014

assert_eq!(

3015

settles(&passed, None),

3016

vec![Settle::thread("rep_1#7").because(&[Reason::CiActivity]).saying("Checks passed on abcdef0", None)]

3017

);

3018

let failed = event("checks.completed", None, json!({ "repoId": "rep_1", "number": 7, "status": "failed" }));

3019

assert!(settles(&failed, None).is_empty());

}

#[test]

fn a_deployment_settles_the_failure_before_it() {

3024

let live = event("deployment.succeeded", Some("usr_bo"), json!({ "repoId": "rep_1", "projectId": "prj_1", "project": "web", "commit": "1234567890" }));

3025

assert_eq!(

3026

settles(&live, Some("bo")),

3027

vec![Settle::thread("rep_1/deploy/prj_1/production").after(&["deployment.failed"]).saying("Production of web deployed at 1234567", Some("bo"))]

3028

);

3029

// Started by g1t or a workflow: the deployment names who.

3030

let preview = event("deployment.succeeded", None, json!({ "repoId": "rep_1", "projectId": "prj_1", "number": 7, "triggeredBy": "g1t" }));

3031

assert_eq!(

3032

settles(&preview, None),

3033

vec![Settle::thread("rep_1/deploy/prj_1/7").after(&["deployment.failed"]).saying("Preview of #7 deployed", Some("g1t"))]

);

}

#[test]

fn requests_answered_settle_what_they_asked() {

3039

let removed = event(

3040

"pull.review_request_removed",

3041

None,

3042

json!({ "repoId": "rep_1", "number": 7, "reviewers": ["Bo"], "teams": [{ "team": "acme/web", "notified": ["cy"] }] }),

3043

);

3044

let mut people: Vec<Option<String>> = settles(&removed, Some("ana")).into_iter().map(|settle| settle.username).collect();

3045

people.sort();

3046

assert_eq!(people, vec![Some("bo".to_owned()), Some("cy".to_owned())]);

3047

assert!(settles(&removed, Some("ana")).iter().all(|settle| settle.why.as_deref() == Some("Review no longer asked of you")));

3048

let token = event("token.approval_reviewed", None, json!({ "workspace": "acme", "tokenId": "tok_1" }));

3049

assert_eq!(

3050

settles(&token, Some("ana")),

3051

vec![Settle::thread("workspace:acme/token/tok_1").because(&[Reason::ReviewRequested]).saying("Token request decided", Some("ana"))]

3052

);

3053

let bypass = event("secret_scanning.bypass_reviewed", None, json!({ "repoId": "rep_1", "requestId": "byp_1" }));

3054

assert_eq!(

3055

settles(&bypass, None),

3056

vec![Settle::thread("rep_1/security/byp_1").because(&[Reason::SecurityAlert]).saying("Bypass request decided", None)]

3057

);

3058

let invitation = event("workspace_invitation.declined", None, json!({ "thread": "invitation:inv_1" }));

3059

assert_eq!(

3060

settles(&invitation, Some("cy")),

3061

vec![Settle::thread("invitation:inv_1").because(&[Reason::ReviewRequested]).saying("Invitation declined", Some("cy"))]

);

}

#[test]

fn your_own_comment_settles_your_item_on_the_thread() {

3067

let commented = event("comment.created", Some("usr_bo"), json!({ "repoId": "rep_1", "number": 7, "commentId": "cmt_1" }));

3068

let by_bo = comment(person("usr_bo", "Bo"), "Looks good", &[]);

3069

assert_eq!(handled_by(&commented, Some(&by_bo)), Some(Settle::thread("rep_1#7").of("bo").saying("Commented", Some("bo"))));

3070

let mut review = comment(person("usr_bo", "Bo"), "Ship it", &[]);

3071

review.comment.as_mut().unwrap().verdict = Some("approve".into());

3072

assert_eq!(handled_by(&commented, Some(&review)).and_then(|settle| settle.why), Some("Reviewed".to_owned()));

3073

// Not g1t's, nor an agent's, nor a note the service wrote.

3074

let by_g1t = comment(person(AGENT_ID, "g1t"), "Done", &[]);

3075

assert_eq!(handled_by(&commented, Some(&by_g1t)), None);

3076

let mut note = comment(person("usr_bo", "bo"), "", &[]);

3077

note.comment.as_mut().unwrap().event = true;

3078

assert_eq!(handled_by(&commented, Some(&note)), None);

3079

assert_eq!(handled_by(&event("pull.merged", None, json!({ "repoId": "rep_1", "number": 7 })), Some(&by_bo)), None);

}

#[test]

fn a_deployments_reviewers_are_told() {

3084

let asked = event(

3085

"deployment.review_requested",

3086

Some("usr_ana"),

3087

json!({

3088

"repoId": "rep_1", "runId": "run_7", "environment": "production", "workflow": "Deploy",

3089

"title": "Ship it", "notify": ["cy", "ana"], "link": "/acme/rocket/actions/runs/run_7"

3090

}),

3091

);

3092

let wanted = wants(&asked).unwrap();

3093

assert_eq!(wanted.number, None);

3094

let thread = thread_of(&asked, &wanted, None);

3095

assert_eq!(thread.key, "rep_1/review/run_7/production");

3096

assert_eq!(thread.link.as_deref(), Some("/acme/rocket/actions/runs/run_7"));

3097

let told = notices(&asked, "acme/rocket", &actor("usr_ana", "ana"), None, &nobody());

3098

let names: Vec<&str> = told.iter().map(|notice| notice.username.as_str()).collect();

3099

// Whoever started the run is told too, when they review it.

3100

assert_eq!(names, ["cy", "ana"]);

3101

assert!(told[0].title.contains("waiting for your review to deploy to production"));

}

#[test]

fn teams_asked_to_review_tell_the_people_they_name() {

3106

let requested = event(

3107

"pull.review_requested",

3108

Some("usr_ana"),

3109

json!({ "number": 7, "reviewers": ["cy"], "teams": [

3110

{ "team": "acme/backend", "notified": ["cy", "dee", "ana"], "assigned": ["cy"] }

3111

] }),

3112

);

3113

let notices_ = notices(&requested, "acme/rocket", &actor("usr_ana", "ana"), Some(&pull()), &nobody());

3114

// cy was picked and is told as a reviewer; dee through the team;

3115

// never whoever asked.

3116

assert_eq!(

3117

told(&notices_),

3118

vec![("cy", Reason::ReviewRequested, Severity::Warning), ("dee", Reason::ReviewRequested, Severity::Warning)]

3119

);

3120

assert_eq!(notices_[0].title, "ana asked you to review acme/rocket#7");

3121

assert_eq!(notices_[1].title, "ana asked @acme/backend to review acme/rocket#7");

3122

assert_eq!(

3123

subscribes(&requested, Some(&pull())),

3124

vec![("cy".into(), Reason::ReviewRequested), ("dee".into(), Reason::ReviewRequested), ("ana".into(), Reason::ReviewRequested)]

3125

);

3126

// Asked by the CODEOWNERS file.

3127

let owned = event(

3128

"pull.review_requested",

3129

None,

3130

json!({ "number": 7, "reviewers": ["bo"], "codeOwners": true, "teams": [{ "team": "acme/docs", "notified": ["wren"], "assigned": [] }] }),

3131

);

3132

let notices_ = notices(&owned, "acme/rocket", &Actor::default(), Some(&pull()), &nobody());

3133

assert_eq!(notices_[0].title, "acme/rocket#7 changes files you own");

3134

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() {

3139

let mut on = comment(person("usr_bo", "bo"), "cc @ana @acme/backend", &["ana"]);

3140

if let Some(comment) = on.comment.as_mut() {

3141

comment.team_mentions = vec![TeamMentioned {

3142

team: "acme/backend".into(),

3143

members: vec!["ana".into(), "cy".into(), "bo".into()],

3144

}];

3145

}

3146

let created = event("comment.created", Some("usr_bo"), json!({ "number": 7, "commentId": "cmt_1" }));

3147

let notices_ = notices(&created, "acme/rocket", &actor("usr_bo", "bo"), Some(&on), &nobody());

3148

3149

assert_eq!(mentioned, vec![("ana", Reason::Mention), ("cy", Reason::TeamMention)]);

3150

assert_eq!(

3151

notices_.iter().find(|n| n.username == "cy").unwrap().title,

3152

"bo mentioned @acme/backend on acme/rocket#7"

3153

);

3154

let subscribed = subscribes(&created, Some(&on));

3155

assert!(subscribed.contains(&("cy".into(), Reason::TeamMention)));

3156

assert!(subscribed.contains(&("ana".into(), Reason::Mention)));

}

#[test]

fn a_description_tells_the_people_and_teams_it_mentions() {

3161

let mut on = pull();

3162

on.mentions = vec!["dee".into()];

3163

on.team_mentions = vec![TeamMentioned {

3164

team: "acme/web".into(),

3165

members: vec!["eve".into()],

3166

}];

3167

let opened = event("pull.opened", Some("usr_ana"), json!({ "number": 7 }));

3168

let notices_ = notices(&opened, "acme/rocket", &actor("usr_ana", "ana"), Some(&on), &nobody());

3169

assert!(told(&notices_).contains(&("dee", Reason::Mention, Severity::Info)));

3170

assert!(told(&notices_).contains(&("eve", Reason::TeamMention, Severity::Info)));

3171

assert_eq!(

3172

subscribes(&opened, Some(&on)),

3173

vec![("dee".into(), Reason::Mention), ("eve".into(), Reason::TeamMention)]

);

}

#[test]

fn reviewers_and_assignees_asked_are_told_never_whoever_asked() {

3179

let requested = event("pull.review_requested", Some("usr_ana"), json!({ "number": 7, "reviewers": ["bo", "g1t", "ana"] }));

3180

let notices_ = notices(&requested, "acme/rocket", &actor("usr_ana", "ana"), Some(&pull()), &nobody());

3181

assert_eq!(told(&notices_), vec![("bo", Reason::ReviewRequested, Severity::Warning)]);

3182

assert_eq!(notices_[0].title, "ana asked you to review acme/rocket#7");

3183

let assigned = event("work_item.assigned", Some("usr_ana"), json!({ "number": 3, "added": ["ana", "eve"] }));

3184

let notices_ = notices(&assigned, "acme/rocket", &actor("usr_ana", "ana"), Some(&pull()), &nobody());

3185

assert_eq!(told(&notices_), vec![("eve", Reason::Assign, Severity::Info)]);

3186

assert_eq!(notices_[0].title, "ana assigned you to acme/rocket#3");

3187

// Both subscribe whoever they name, never g1t.

3188

assert_eq!(subscribes(&requested, Some(&pull())), vec![("bo".into(), Reason::ReviewRequested), ("ana".into(), Reason::ReviewRequested)]);

3189

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() {

3194

let failed = event("checks.completed", None, json!({ "number": 7, "status": "failed" }));

3195

assert_eq!(

3196

told(&notices(&failed, "acme/rocket", &Actor::default(), Some(&pull()), &nobody())),

3197

vec![("ana", Reason::CiActivity, Severity::Error)]

3198

);

3199

// g1t's change is the person's who asked for it, never g1t's.

3200

let notices_ = notices(&failed, "acme/rocket", &Actor::default(), Some(&g1t_pull()), &nobody());

3201

assert_eq!(told(&notices_), vec![("ana", Reason::CiActivity, Severity::Error)]);

3202

let errored = event("checks.completed", None, json!({ "number": 7, "status": "errored" }));

3203

assert_eq!(

3204

notices(&errored, "acme/rocket", &Actor::default(), Some(&pull()), &nobody())[0].title,

3205

"Checks could not run on acme/rocket#7"

);

}

#[test]

fn a_workflow_that_fails_tells_its_pull_requests_owner_even_if_they_pushed() {

3211

let failed = event("workflow.completed", Some("usr_ana"), json!({ "pull": 7, "workflow": "CI", "conclusion": "failure" }));

3212

let notices_ = notices(&failed, "acme/rocket", &actor("usr_ana", "ana"), Some(&pull()), &nobody());

3213

assert_eq!(told(&notices_), vec![("ana", Reason::CiActivity, Severity::Error)]);

3214

assert_eq!(notices_[0].title, "CI failed on acme/rocket#7");

3215

// On a branch: whoever pushed.

3216

let pushed = event(

3217

"workflow.completed",

3218

Some("usr_bo"),

3219

json!({ "workflow": "Deploy", "path": ".g1t/workflows/deploy.yml", "conclusion": "failure", "ref": "refs/heads/main", "number": 12, "sha": "abcdef0123", "runId": "run_9" }),

3220

);

3221

let notices_ = notices(&pushed, "acme/rocket", &actor("usr_bo", "bo"), None, &nobody());

3222

assert_eq!(told(&notices_), vec![("bo", Reason::CiActivity, Severity::Error)]);

3223

assert_eq!(notices_[0].title, "Deploy failed on main in acme/rocket");

3224

assert_eq!(notices_[0].body, "Run 12 at abcdef0");

3225

// Nobody to tell when g1t pushed.

3226

assert!(notices(&pushed, "acme/rocket", &actor("g1t", "g1t"), None, &nobody()).is_empty());

3227

// Every failure of a workflow on a branch is one thread.

3228

let wanted = wants(&pushed).unwrap();

3229

let thread = thread_of(&pushed, &wanted, None);

3230

assert_eq!(thread.key, "rep_1/run/.g1t/workflows/deploy.yml@main");

3231

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" }),

3240

);

3241

let audience = watched(&[("cy", WatchLevel::Custom, &["deployments"]), ("dee", WatchLevel::Custom, &["work_items"]), ("eve", WatchLevel::All, &[])]);

3242

let notices_ = notices(&failed, "acme/rocket", &actor("usr_bo", "bo"), None, &audience);

assert_eq!(

told(&notices_),

vec![

("bo", Reason::CiActivity, Severity::Error),

3247

("cy", Reason::Subscribed, Severity::Error),

3248

("eve", Reason::Subscribed, Severity::Error),

3249

]

3250

);

3251

assert_eq!(notices_[0].title, "Production of rocket failed to deploy");

3252

assert_eq!(notices_[0].body, "The build failed.");

3253

let thread = thread_of(&failed, &wants(&failed).unwrap(), None);

3254

assert_eq!(thread.key, "rep_1/deploy/prj_1/production");

3255

assert_eq!(url(Some("acme/rocket"), None, thread.kind, None, None, thread.link.as_deref()), "/acme/rocket/deployments/dpl_1");

3256

// A success tells the owner nothing, even after a failure: it

3257

// settles the failure (settles). Watchers who chose deployments hear of it.

3258

let live = event("deployment.succeeded", Some("usr_bo"), json!({ "repoId": "rep_1", "projectId": "prj_1", "project": "rocket", "commit": "abcdef0123" }));

3259

assert!(notices(&live, "acme/rocket", &actor("usr_bo", "bo"), None, &nobody()).is_empty());

3260

let recovered = event("deployment.succeeded", Some("usr_bo"), json!({ "repoId": "rep_1", "projectId": "prj_1", "project": "rocket", "recovered": true }));

3261

assert!(notices(&recovered, "acme/rocket", &actor("usr_bo", "bo"), None, &nobody()).is_empty());

3262

let notices_ = notices(&recovered, "acme/rocket", &actor("usr_bo", "bo"), None, &audience);

3263

assert_eq!(told(&notices_), vec![("cy", Reason::Subscribed, Severity::Success), ("eve", Reason::Subscribed, Severity::Success)]);

3264

assert_eq!(notices_[0].title, "Production of rocket is live");

3265

// A preview's is the pull request's owner's.

3266

let preview = event("deployment.failed", None, json!({ "repoId": "rep_1", "projectId": "prj_1", "number": 7, "triggeredBy": "g1t" }));

3267

assert_eq!(told(&notices(&preview, "acme/rocket", &Actor::default(), Some(&pull()), &nobody())), vec![("ana", Reason::CiActivity, Severity::Error)]);

3268

assert_eq!(thread_of(&preview, &wants(&preview).unwrap(), Some(&pull())).key, "rep_1/deploy/prj_1/7");

3269

}

3270

3271

/// A scan's summary, as the security service publishes it.

3272

fn summary(alert_type: &str, opened: &[&str], new: (u32, u32, u32, u32), open: (u32, u32, u32, u32), pusher: Option<&str>) -> Event {

3273

let counts = |(critical, high, medium, low): (u32, u32, u32, u32)| json!({ "critical": critical, "high": high, "medium": medium, "low": low, "unknown": 0 });

3274

let worst = [(new.0, "critical"), (new.1, "high"), (new.2, "medium"), (new.3, "low")].into_iter().find(|(n, _)| *n > 0).map(|(_, name)| name);

3275

let page = match alert_type {

3276

"secret_scanning" => "secret-scanning",

3277

"code_scanning" => "code-scanning",

3278

_ => "vulnerabilities",

3279

};

3280

event(

3281

&format!("{alert_type}_alert.summary"),

3282

None,

3283

json!({

3284

"repoId": "rep_1", "alertType": alert_type, "scan": if pusher.is_some() { "push" } else { "dependencies" },

3285

"opened": opened, "openedCount": opened.len(), "fixedCount": 0, "dismissedCount": 0,

3286

"newBySeverity": counts(new), "open": counts(open), "worst": worst,

3287

"latest": if opened.is_empty() { serde_json::Value::Null } else { json!("lodash 4.17.20 in package-lock.json: GHSA-1") },

3288

"link": format!("/acme/rocket/security/{page}"),

3289

"pusher": pusher, "blocked": if pusher.is_some() { opened.len() } else { 0 },

3290

"notify": ["ana"], "members": ["ana", "bo", "cy"]

}),

)

}

#[test]

fn a_scan_is_one_item_per_repository_and_alert_type() {

3297

let found = summary("vulnerability", &["vul_3", "vul_2", "vul_1"], (1, 2, 0, 0), (1, 2, 0, 0), None);

3298

assert_eq!(wants(&found), Some(Wanted { repo_id: Some("rep_1".into()), ..Wanted::default() }));

3299

// cy watches for security alerts and is a member; eve watches

3300

// everything but cannot see findings; dee watches work items only.

3301

let audience = watched(&[

3302

("cy", WatchLevel::Custom, &["security"]),

3303

("dee", WatchLevel::Custom, &["work_items"]),

3304

("eve", WatchLevel::All, &[]),

3305

]);

3306

let notices_ = notices(&found, "acme/rocket", &Actor::default(), None, &audience);

3307

assert_eq!(told(&notices_), vec![("ana", Reason::SecurityAlert, Severity::Error), ("cy", Reason::SecurityAlert, Severity::Error)]);

3308

assert_eq!(notices_[0].title, "3 new vulnerabilities in acme/rocket: 1 critical, 2 high");

3309

assert_eq!(notices_[0].body, "lodash 4.17.20 in package-lock.json: GHSA-1");

3310

// The same thread for every scan of this type, however many alerts.

3311

let thread = thread_of(&found, &wants(&found).unwrap(), None);

3312

assert_eq!(thread.key, "rep_1/security/vulnerability");

3313

assert_eq!(url(Some("acme/rocket"), None, thread.kind, None, None, thread.link.as_deref()), "/acme/rocket/security/vulnerabilities");

3314

// Medium and low are news, but not urgent: an info item, or the digest.

3315

let quiet = summary("code_scanning", &["cod_1"], (0, 0, 1, 0), (0, 0, 1, 0), None);

3316

let notices_ = notices(&quiet, "acme/rocket", &Actor::default(), None, &audience);

3317

assert_eq!(told(&notices_), vec![("ana", Reason::SecurityAlert, Severity::Info), ("cy", Reason::SecurityAlert, Severity::Info)]);

3318

assert_eq!(notices_[0].title, "1 new code scanning alert in acme/rocket: 1 medium");

3319

// A scan that opened nothing tells nobody: the thread is brought up to date instead.

3320

let fixed = summary("vulnerability", &[], (0, 0, 0, 0), (0, 1, 0, 0), None);

3321

assert!(notices(&fixed, "acme/rocket", &Actor::default(), None, &audience).is_empty());

3322

let (title, body) = refreshed_text("acme/rocket", &SecuritySummary { fixed_count: 2, ..security_summary(&fixed).unwrap() });

3323

assert_eq!(title, "1 open vulnerability in acme/rocket: 1 high");

3324

assert_eq!(body, "2 fixed");

3325

let settled = SecuritySummary { fixed_count: 1, dismissed_count: 1, open: SeverityCounts::default(), ..security_summary(&fixed).unwrap() };

3326

assert_eq!(refreshed_text("acme/rocket", &settled), ("No open vulnerabilities in acme/rocket".to_owned(), "1 fixed, 1 dismissed".to_owned()));

3327

// Per-alert events are for webhooks.

3328

assert_eq!(wants(&event("vulnerability_alert.created", None, json!({ "repoId": "rep_1" }))), None);

3329

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() {

3334

let blocked = summary("secret_scanning", &["sec_1", "sec_2"], (2, 0, 0, 0), (2, 0, 0, 0), Some("bo"));

3335

let notices_ = notices(&blocked, "acme/rocket", &Actor::default(), None, &nobody());

3336

assert_eq!(told(&notices_), vec![("bo", Reason::SecurityAlert, Severity::Error), ("ana", Reason::SecurityAlert, Severity::Error)]);

3337

assert_eq!(notices_[0].title, "A push to acme/rocket was blocked: it adds 2 secrets");

3338

assert_eq!(notices_[1].title, "2 new secrets in acme/rocket: 2 critical");

3339

assert_eq!(thread_of(&blocked, &wants(&blocked).unwrap(), None).key, "rep_1/security/secret_scanning");

3340

// Whatever bo chose, the push is theirs to fix; ana, an owner, is told whatever she chose.

3341

let summary_ = security_summary(&blocked).unwrap();

3342

let off = InboxSettings { security_alerts: false, security_digest: DigestCadence::Off, ..InboxSettings::default() };

3343

assert_eq!(security_delivery(&summary_, "bo", &off), Delivery::Now);

3344

assert_eq!(security_delivery(&summary_, "ana", &off), Delivery::Now);

3345

// A watcher who turned security alerts off hears of nothing urgent.

3346

assert_eq!(security_delivery(&summary_, "cy", &off), Delivery::Never);

3347

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() {

3352

let medium = security_summary(&summary("vulnerability", &["vul_1"], (0, 0, 1, 0), (0, 0, 1, 0), None)).unwrap();

3353

assert!(!medium.urgent());

3354

let choose = |cadence| InboxSettings { security_digest: cadence, ..InboxSettings::default() };

3355

// Owners too: a medium alert is for the digest, as they chose.

3356

assert_eq!(security_delivery(&medium, "ana", &choose(DigestCadence::Daily)), Delivery::Digest);

3357

assert_eq!(security_delivery(&medium, "cy", &choose(DigestCadence::Weekly)), Delivery::Digest);

3358

assert_eq!(security_delivery(&medium, "cy", &choose(DigestCadence::AtOnce)), Delivery::Now);

3359

assert_eq!(security_delivery(&medium, "ana", &choose(DigestCadence::Off)), Delivery::Never);

3360

let high = security_summary(&summary("vulnerability", &["vul_1"], (0, 1, 0, 0), (0, 1, 0, 0), None)).unwrap();

3361

assert_eq!(security_delivery(&high, "ana", &choose(DigestCadence::Off)), Delivery::Now);

3362

assert_eq!(breakdown(&SeverityCounts { critical: 1, low: 3, ..SeverityCounts::default() }), "1 critical, 3 low");

3363

assert_eq!(breakdown(&SeverityCounts::default()), "");

}

#[test]

fn bypass_requests_reports_and_advisories_tell_who_they_name() {

3368

let audience = watched(&[("cy", WatchLevel::Custom, &["security"]), ("eve", WatchLevel::All, &[])]);

3369

// A bypass request goes to the reviewers named, never to watchers.

3370

let requested = event(

3371

"secret_scanning.bypass_requested",

3372

Some("usr_bo"),

3373

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" }),

3374

);

3375

let notices_ = notices(&requested, "acme/rocket", &actor("usr_bo", "bo"), None, &audience);

3376

assert_eq!(told(&notices_), vec![("ana", Reason::SecurityAlert, Severity::Warning)]);

3377

assert_eq!(notices_[0].title, "bo asked to bypass push protection in acme/rocket");

3378

assert_eq!(thread_of(&requested, &wants(&requested).unwrap(), None).key, "rep_1/security/byp_1");

3379

// A private report is for the owners and security managers, at once, with no reporter named.

3380

let reported = event(

3381

"security_advisory.reported",

3382

None,

3383

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"] }),

3384

);

3385

let notices_ = notices(&reported, "acme/rocket", &Actor::default(), None, &audience);

3386

assert_eq!(told(&notices_), vec![("ana", Reason::SecurityAlert, Severity::Warning)]);

3387

assert_eq!(notices_[0].title, "A vulnerability was reported privately to acme/rocket");

3388

assert_eq!(thread_of(&reported, &wants(&reported).unwrap(), None).key, "rep_1/security/reports");

3389

// A published advisory is for watchers who chose security; it is public, so no members list is needed.

3390

let published = event(

3391

"security_advisory.published",

3392

Some("usr_ana"),

3393

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" }),

3394

);

3395

let notices_ = notices(&published, "acme/rocket", &actor("usr_ana", "ana"), None, &audience);

3396

assert_eq!(told(&notices_), vec![("cy", Reason::Subscribed, Severity::Info), ("eve", Reason::Subscribed, Severity::Info)]);

3397

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() {

3402

let merged = event("pull.merged", Some("usr_ana"), json!({ "number": 7 }));

3403

let by_ana = notices(&merged, "acme/rocket", &actor("usr_ana", "ana"), Some(&pull()), &nobody());

3404

// The assignee still hears; ana, who merged, does not.

3405

assert_eq!(told(&by_ana), vec![("bo", Reason::StateChange, Severity::Success)]);

3406

let by_bo = notices(&merged, "acme/rocket", &actor("usr_bo", "bo"), Some(&pull()), &nobody());

3407

assert_eq!(told(&by_bo), vec![("ana", Reason::StateChange, Severity::Success)]);

3408

assert_eq!(by_bo[0].title, "bo merged acme/rocket#7");

3409

let by_queue = notices(&merged, "acme/rocket", &Actor::default(), Some(&pull()), &nobody());

3410

assert_eq!(by_queue[0].title, "acme/rocket#7 was merged");

}

#[test]

fn closing_and_reopening_tells_everyone_subscribed_and_watchers() {

3415

let closed = event("work_item.closed", Some("usr_bo"), json!({ "number": 3, "reason": "completed" }));

3416

let item = InboxSubject {

3417

kind: Some(SubjectKind::Item),

3418

title: "Crash".into(),

3419

author: person("usr_cy", "cy"),

3420

assignees: vec!["bo".into()],

3421

..InboxSubject::default()

3422

};

3423

let mut audience = subscribed(&[("eve", State::Subscribed, Some(Reason::Comment)), ("fay", State::Unsubscribed, None)]);

3424

audience.watchers = watched(&[("gus", WatchLevel::All, &[]), ("hal", WatchLevel::Custom, &["pulls"])]).watchers;

3425

let notices_ = notices(&closed, "acme", &actor("usr_bo", "bo"), Some(&item), &audience);

assert_eq!(

told(&notices_),

vec![

("cy", Reason::StateChange, Severity::Info),

3430

("eve", Reason::StateChange, Severity::Info),

3431

("gus", Reason::Subscribed, Severity::Info),

3432

]

3433

);

3434

assert_eq!(notices_[0].title, "bo closed acme#3");

3435

let by_pull = event("work_item.closed", None, json!({ "number": 3, "resolution": { "repo": "acme/rocket", "commit": "1a2b3c4d5e" } }));

3436

assert_eq!(notices(&by_pull, "acme", &Actor::default(), Some(&item), &nobody())[0].title, "acme#3 was closed by acme/rocket@1a2b3c4");

3437

let reopened = event("work_item.reopened", Some("usr_cy"), json!({ "number": 3 }));

3438

assert_eq!(told(&notices(&reopened, "acme", &actor("usr_cy", "cy"), Some(&item), &nobody())), vec![("bo", Reason::StateChange, Severity::Info)]);

}

#[test]

fn opening_tells_who_it_names_and_watchers_of_its_kind() {

3443

let opened = event("pull.opened", Some("usr_ana"), json!({ "number": 7 }));

3444

let subject = InboxSubject { reviewers: vec!["cy".into(), "g1t".into()], ..pull() };

3445

let audience = watched(&[("bo", WatchLevel::All, &[]), ("dee", WatchLevel::Custom, &["work_items"]), ("eve", WatchLevel::Custom, &["pulls"]), ("fay", WatchLevel::Participating, &[])]);

3446

let notices_ = notices(&opened, "acme/rocket", &actor("usr_ana", "ana"), Some(&subject), &audience);

assert_eq!(

told(&notices_),

vec![

("bo", Reason::Assign, Severity::Info),

3451

("cy", Reason::ReviewRequested, Severity::Warning),

3452

("eve", Reason::Subscribed, Severity::Info),

3453

]

3454

);

3455

assert_eq!(notices_[2].title, "ana opened acme/rocket#7");

}

#[test]

fn ignoring_a_repository_silences_it_and_unsubscribing_keeps_only_what_is_asked() {

3460

let created = event("comment.created", Some("usr_bo"), json!({ "number": 7, "commentId": "cmt_1" }));

3461

let on = comment(person("usr_bo", "bo"), "@cy have a look", &["cy"]);

3462

let audience = watched(&[("cy", WatchLevel::Ignore, &[]), ("ana", WatchLevel::All, &[])]);

3463

assert!(notices(&created, "acme/rocket", &actor("usr_bo", "bo"), Some(&on), &audience).iter().all(|notice| notice.username != "cy"));

3464

let audience = subscribed(&[("cy", State::Unsubscribed, None), ("ana", State::Unsubscribed, None)]);

3465

let notices_ = notices(&created, "acme/rocket", &actor("usr_bo", "bo"), Some(&on), &audience);

3466

// cy was mentioned, which is asked of them; ana unsubscribed from the conversation.

3467

assert_eq!(told(&notices_), vec![("cy", Reason::Mention, Severity::Info)]);

}

#[test]

fn an_agent_s_review_is_shown_by_its_name_and_marked_advisory() {

3472

// Margo (@margo), acting for bo, asks for changes on ana's pull request.

3473

let created = event("comment.created", Some("usr_bo"), json!({ "number": 7, "commentId": "cmt_1" }));

3474

let mut on = comment(person("agt_1", "ana"), "Trim the name.", &[]);

3475

if let Some(said) = on.comment.as_mut() {

3476

said.agent = Some(g1t_contracts::work::AgentRef {

3477

id: "agt_1".into(),

3478

handle: "ana".into(),

3479

display_name: "Margo".into(),

3480

avatar_seed: "margo".into(),

3481

});

3482

said.acting_for = Some(person("usr_bo", "bo"));

3483

said.verdict = Some("request_changes".into());

3484

said.advisory = true;

3485

}

3486

let notices_ = notices(&created, "acme/rocket", &actor("usr_bo", "bo"), Some(&on), &nobody());

3487

// Its handle sharing ana's name leaves nobody out: ana owns it and hears.

3488

assert!(told(&notices_).contains(&("ana", Reason::Author, Severity::Info)));

3489

assert_eq!(notices_[0].title, "Margo (agent) asked for changes on acme/rocket#7 (advisory)");

3490

// An agent subscribes nobody by its handle.

3491

assert!(subscribes(&created, Some(&on)).iter().all(|(name, _)| name != "ana"));

}

#[test]

fn comments_tell_those_mentioned_then_everyone_subscribed_never_the_writer() {

3496

let created = event("comment.created", Some("usr_bo"), json!({ "number": 7, "commentId": "cmt_1" }));

3497

let on = comment(person("usr_bo", "bo"), "@cy @bo have a look", &["cy", "bo", "g1t"]);

3498

let audience = subscribed(&[("eve", State::Subscribed, Some(Reason::Comment)), ("fay", State::Subscribed, Some(Reason::Manual))]);

3499

let notices_ = notices(&created, "acme/rocket", &actor("usr_bo", "bo"), Some(&on), &audience);

assert_eq!(

told(&notices_),

vec![

("cy", Reason::Mention, Severity::Info),

3504

("ana", Reason::Author, Severity::Info),

3505

("eve", Reason::Comment, Severity::Info),

3506

("fay", Reason::Manual, Severity::Info),

3507

]

3508

);

3509

assert_eq!(notices_[0].title, "bo mentioned you on acme/rocket#7");

3510

assert_eq!(notices_[1].title, "bo commented on acme/rocket#7");

3511

assert_eq!(notices_[0].body, "@cy @bo have a look");

3512

// Mentioned and the owner: told once, as mentioned.

3513

let on = comment(person("usr_bo", "bo"), "@ana", &["ana"]);

3514

assert_eq!(

3515

told(&notices(&created, "acme/rocket", &actor("usr_bo", "bo"), Some(&on), &nobody())),

3516

vec![("ana", Reason::Mention, Severity::Info)]

3517

);

3518

// The owner's own comment tells the assignee, not the owner.

3519

let on = comment(person("usr_ana", "ana"), "thanks", &[]);

3520

assert_eq!(

3521

told(&notices(&created, "acme/rocket", &actor("usr_ana", "ana"), Some(&on), &nobody())),

3522

vec![("bo", Reason::Assign, Severity::Info)]

3523

);

3524

// Something that happened, not something written, tells nobody.

3525

let mut on = comment(person("usr_bo", "bo"), "assigned cy", &[]);

3526

on.comment.as_mut().unwrap().event = true;

3527

assert!(notices(&created, "acme/rocket", &actor("usr_bo", "bo"), Some(&on), &nobody()).is_empty());

3528

// An approval is good news.

3529

let mut on = comment(person("usr_cy", "cy"), "", &[]);

3530

on.comment.as_mut().unwrap().verdict = Some("approve".into());

3531

let notices_ = notices(&created, "acme/rocket", &actor("usr_cy", "cy"), Some(&on), &nobody());

3532

assert_eq!(told(&notices_)[0], ("ana", Reason::Author, Severity::Success));

3533

assert_eq!(notices_[0].body, "Add the inbox");

3534

// Writing and being mentioned subscribe, never g1t.

3535

let on = comment(person("usr_bo", "bo"), "@cy", &["cy"]);

3536

assert_eq!(subscribes(&created, Some(&on)), vec![("bo".into(), Reason::Comment), ("cy".into(), Reason::Mention)]);

3537

let on = comment(person(AGENT_ID, "g1t"), "done", &[]);

3538

assert!(subscribes(&created, Some(&on)).is_empty());

}

#[test]

fn a_commit_comment_tells_who_it_mentions_and_the_commit_s_author() {

3543

let created = event(

3544

"commit_comment.created",

3545

Some("usr_bo"),

3546

json!({

3547

"repoId": "rep_1", "commit": "abcdef1234567890", "link": "/acme/rocket/commit/abcdef1234567890",

3548

"excerpt": "@cy why this?", "mentions": ["cy", "bo"], "commitAuthor": "ana",

3549

"teamMentions": [{ "team": "acme/web", "members": ["dee", "cy"] }],

3550

}),

3551

);

3552

let wanted = wants(&created).unwrap();

3553

assert_eq!(wanted.number, None);

3554

let thread = thread_of(&created, &wanted, None);

3555

assert_eq!(thread.key, "rep_1/commit/abcdef1234567890");

3556

assert_eq!(url(Some("acme/rocket"), None, thread.kind, thread.number, None, thread.link.as_deref()), "/acme/rocket/commit/abcdef1234567890");

3557

let notices_ = notices(&created, "acme/rocket", &actor("usr_bo", "bo"), None, &nobody());

assert_eq!(

told(&notices_),

vec![

("cy", Reason::Mention, Severity::Info),

3562

("dee", Reason::TeamMention, Severity::Info),

3563

("ana", Reason::Author, Severity::Info),

3564

]

3565

);

3566

assert_eq!(notices_[0].title, "bo mentioned you on acme/rocket@abcdef1");

3567

assert_eq!(notices_[2].title, "bo commented on your commit acme/rocket@abcdef1");

3568

assert_eq!(notices_[0].body, "@cy why this?");

}

#[test]

fn pull_requests_are_one_thread_each() {

3573

let created = event("comment.created", Some("usr_bo"), json!({ "repoId": "rep_1", "number": 7, "commentId": "cmt_1" }));

3574

let wanted = wants(&created).unwrap();

3575

let thread = thread_of(&created, &wanted, Some(&pull()));

3576

assert_eq!(thread.key, "rep_1#7");

3577

assert_eq!(thread.kind, Some(SubjectKind::Pull));

3578

let merged = event("pull.merged", None, json!({ "repoId": "rep_1", "number": 7 }));

3579

assert_eq!(thread_of(&merged, &wants(&merged).unwrap(), Some(&pull())).key, thread.key);

}

#[test]

fn items_link_to_what_they_are_about() {

3584

assert_eq!(url(Some("acme/rocket"), None, Some(SubjectKind::Pull), Some(7), None, None), "/acme/rocket/pull/7");

3585

assert_eq!(url(Some("acme/rocket"), Some("acme"), Some(SubjectKind::Item), Some(3), None, None), "/acme/-/work/3");

3586

assert_eq!(url(None, Some("acme"), Some(SubjectKind::Item), Some(3), None, None), "/acme/-/work/3");

3587

assert_eq!(url(Some("acme/rocket"), None, Some(SubjectKind::Run), None, Some("run_1"), None), "/acme/rocket/actions/runs/run_1");

3588

assert_eq!(url(Some("acme/rocket"), None, Some(SubjectKind::Deploy), None, None, Some("/acme/site/deployments/dpl_1")), "/acme/site/deployments/dpl_1");

3589

assert_eq!(url(Some("acme/rocket"), None, None, None, None, None), "/acme/rocket");

3590

assert_eq!(url(None, None, None, None, None, None), "/inbox");

}

#[test]

fn long_titles_are_cut_to_a_line() {

3595

let long = "x".repeat(400);

3596

assert_eq!(clip(&long, MAX_TITLE).chars().count(), MAX_TITLE);

3597

assert_eq!(clip(" short ", MAX_TITLE), "short");

}

#[test]

fn marks_change_only_what_they_say() {

3602

assert_eq!(mark_change(InboxMark::Read), "read_at = COALESCE(read_at, ?1)");

3603

assert!(mark_change(InboxMark::Done).contains("done_at = ?1"));

3604

assert_eq!(mark_change(InboxMark::Unsnooze), "snoozed_until = NULL");

3605

assert!(ranked(&ListInboxArgs::default()));

3606

assert!(!ranked(&ListInboxArgs { reason: Some(Reason::Mention), ..ListInboxArgs::default() }));

3607

assert!(!ranked(&ListInboxArgs { participating: true, ..ListInboxArgs::default() }));

}

#[test]

fn filters_bind_in_order_after_what_is_bound() {

3612

let a = ListInboxArgs {

3613

reason: Some(Reason::Mention),

3614

repo_id: Some("rep_1".into()),

3615

participating: true,

3616

unread: true,

3617

..ListInboxArgs::default()

3618

};

3619

// Two values come first: the username and now.

3620

let (conditions, values) = filters(&a, 2);

3621

assert_eq!(

3622

conditions,

3623

vec!["reason = ?3", "repo_id = ?4", "reason NOT IN ('manual', 'subscribed')", "read_at IS NULL"]

3624

);

3625

assert_eq!(values, vec!["mention", "rep_1"]);

}

#[test]

fn the_bump_keeps_whatever_is_most_urgent_while_unread() {

3630

assert!(BUMP.contains("WHERE NOT EXISTS (SELECT 1 FROM inbox_activity WHERE event_id = ?4 AND username = ?2)"));

3631

assert!(BUMP.contains("ON CONFLICT (username, thread) DO UPDATE"));

3632

assert!(BUMP.contains("activity = inbox_items.activity + 1"));

3633

// Its numbered parameters run from 1 to 19.

3634

for at in 1..=19 {

3635

assert!(BUMP.contains(&format!("?{at}")), "?{at}");

3636

}

3637

assert!(!BUMP.contains("?20"));

}

#[test]

fn live_notifications_say_what_and_where() {

3642

let notice = Notice {

3643

username: "ana".into(),

3644

reason: Reason::Agent,

3645

severity: Severity::Warning,

3646

title: "g1t needs you on acme/api#4".into(),

3647

body: "Which database?".into(),

3648

};

3649

let live = live_notification("evt_1", &notice, "/Acme/api/pull/4#c1", "2026-10-08T00:00:00Z", Some("rep_1#4"));

3650

assert_eq!(live["id"], "inbox:evt_1");

3651

// A stop that needs a person is an ordinary inbox item.

3652

assert_eq!(live["kind"], "inbox");

3653

assert_eq!(live["workspace"], "acme");

3654

assert_eq!(live["href"], "/Acme/api/pull/4#c1");

3655

assert_eq!(live["actor"]["name"], "acme/api");

3656

// Notify keeps the thread, to stop replaying it once it settles.

3657

assert_eq!(live["inbox_thread"], "rep_1#4");

3658

let mention = Notice { reason: Reason::TeamMention, ..notice };

3659

assert_eq!(live_notification("evt_2", &mention, "https://elsewhere", "t", None)["href"], "/inbox");

3660

assert_eq!(live_notification("evt_2", &mention, "/x", "t", None)["kind"], "mention");

}

#[test]

fn inbox_counts_are_told_as_they_are_by_username() {

3665

assert_eq!(inbox_count_update("Ana", 3), serde_json::json!({ "username": "ana", "unread": 3 }));

3666

assert_eq!(inbox_count_update("bo", 0)["unread"], 0);

3667

// What settled is told with the count, each thread once.

3668

let threads = vec!["rep_1#7".to_owned(), "rep_1/deploy/*/7".to_owned(), "rep_1#7".to_owned()];

3669

assert_eq!(

3670

inbox_settled_update("Ana", 2, &threads),

3671

serde_json::json!({ "username": "ana", "unread": 2, "settled": ["rep_1#7", "rep_1/deploy/*/7"] })

);

}

#[test]

fn a_settle_says_why_and_who_and_clears_on_new_activity() {

3677

let settle = Settle::thread("rep_1#7").saying("Merged", Some("@Ana"));

3678

assert_eq!(settle.why.as_deref(), Some("Merged"));

3679

assert_eq!(settle.by.as_deref(), Some("ana"));

3680

assert_eq!(Settle::thread("rep_1#7").saying("Checks passed", Some(" ")).by, None);

3681

// New activity brings a thread back with no reason left on it, and so does marking it by hand.

3682

assert!(BUMP.contains("settled_reason = NULL"));

3683

assert!(mark_change(InboxMark::Done).contains("settled_reason = NULL"));

3684

assert!(mark_change(InboxMark::Undone).contains("settled_reason = NULL"));

}

#[test]

fn token_approvals_tell_the_people_named_about_no_repository() {

3689

let mut asked = event(

3690

"token.approval_requested",

3691

Some("usr_ana"),

3692

json!({ "workspace": "acme", "tokenId": "tok_1", "notify": ["Bo", "cy", "bo", "g1t"], "title": "ana asks", "body": "ci: contents: write", "link": "/acme/-/settings/tokens" }),

3693

);

3694

asked.repo_id = None;

3695

let (thread, told) = workspace_notices(&asked, None);

3696

assert_eq!(thread, "workspace:acme/token/tok_1");

3697

let mut names: Vec<&str> = told.iter().map(|notice| notice.username.as_str()).collect();

3698

names.sort();

3699

assert_eq!(names, ["bo", "cy"], "each once, lowercased, never g1t");

3700

assert!(told.iter().all(|notice| notice.reason == Reason::ReviewRequested && notice.severity == Severity::Warning));

3701

assert!(wants(&asked).is_none(), "about no repository");

3702

let reviewed = event("token.approval_reviewed", None, json!({ "workspace": "acme", "tokenId": "tok_1", "notify": ["ana"], "title": "approved" }));

3703

let (_, told) = workspace_notices(&reviewed, None);

3704

assert_eq!(told[0].reason, Reason::Author);

3705

assert!(WORKSPACE_EVENTS.contains(&"token.approval_reviewed"));

}

#[test]

fn a_workspace_invitation_asks_its_person_and_tells_whoever_sent_it_the_answer() {

3710

let mut sent = event(

3711

"workspace_invitation.created",

3712

Some("usr_owner"),

3713

json!({ "workspace": "flagon-io", "invitationId": "inv_1", "thread": "invitation:inv_1", "notify": ["daweazl"], "title": "@syntaqx invited you to join Flagon, Inc.", "link": "/invitations" }),

3714

);

3715

sent.repo_id = None;

3716

let (thread, told) = workspace_notices(&sent, None);

3717

assert_eq!(thread, "invitation:inv_1");

3718

assert_eq!(told.len(), 1);

3719

assert_eq!((told[0].username.as_str(), told[0].reason, told[0].severity), ("daweazl", Reason::ReviewRequested, Severity::Warning));

3720

let accepted = event("workspace_invitation.accepted", None, json!({ "workspace": "flagon-io", "thread": "invitation:inv_1", "notify": ["syntaqx"] }));

3721

let (thread, told) = workspace_notices(&accepted, None);

3722

assert_eq!(thread, "invitation:inv_1");

3723

assert_eq!((told[0].reason, told[0].severity), (Reason::Author, Severity::Success));

3724

let declined = event("workspace_invitation.declined", None, json!({ "workspace": "flagon-io", "thread": "invitation:inv_1", "notify": ["syntaqx"] }));

3725

assert_eq!(workspace_notices(&declined, None).1[0].severity, Severity::Info);

3726

// Answered or revoked, the person's item is done; a revocation tells nobody.

3727

for kind in ["workspace_invitation.accepted", "workspace_invitation.declined", "workspace_invitation.revoked"] {

3728

assert!(INVITATION_ANSWERED.contains(&kind));

3729

}

3730

assert!(!WORKSPACE_EVENTS.contains(&"workspace_invitation.revoked"));

3731

}

3732

}