Skip to content
Sign up

flagon-io/g1t

public
00

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

gitagentscloudflarerustforgemcp
1,300 linesCodeBlameRaw
1//! The security service's tables in D1.
2
3use std::collections::HashMap;
4
5use g1t_contracts::security::{
6 AlertActivity, DismissReason, NewSecret, ScanState, SecretCounts, SecretFinding, SecretStatus, SecurityUpdate,
7 SeverityCounts, UpdateState, VersionUpdatesState, VulnStatus, Vulnerability,
8};
9use g1t_contracts::time::rfc3339;
10use g1t_contracts::new_id;
11use g1t_kit::now_ms;
12use serde::Deserialize;
13use worker::wasm_bindgen::JsValue;
14use worker::{D1Database, Result};
15
16pub fn optional(value: Option<&str>) -> JsValue {
17 value.map_or(JsValue::NULL, JsValue::from)
18}
19
20pub fn now() -> String {
21 rfc3339(now_ms())
22}
23
24/// A repository purged (`repo.purged`): every row kept for it, `?1` its
25/// id. Advisories are shared by every repository, so they stay.
26pub const PURGED: &[&str] = &[
27 "DELETE FROM secrets WHERE repo_id = ?1",
28 "DELETE FROM vulnerabilities WHERE repo_id = ?1",
29 "DELETE FROM upgrades WHERE repo_id = ?1",
30 "DELETE FROM updates WHERE repo_id = ?1",
31 "DELETE FROM alert_activity WHERE repo_id = ?1",
32 "DELETE FROM push_scans WHERE repo_id = ?1",
33 "DELETE FROM repos WHERE repo_id = ?1",
34];
35
36#[derive(Clone, Deserialize)]
37pub struct RepoRow {
38 pub repo_id: String,
39 pub namespace: String,
40 pub name: String,
41 pub upkeep: i64,
42 pub history: String,
43 pub history_cursor: Option<String>,
44 pub history_commits: i64,
45 pub history_finished_at: Option<String>,
46 pub deps_scanned_at: Option<String>,
47 pub deps_error: Option<String>,
48 pub lockfiles: String,
49 #[serde(default)]
50 pub version_updates: Option<String>,
51}
52
53impl RepoRow {
54 pub fn scan_state(&self) -> ScanState {
55 ScanState {
56 history: self.history.clone(),
57 commits_scanned: self.history_commits.max(0) as u32,
58 history_finished_at: self.history_finished_at.clone(),
59 dependencies_scanned_at: self.deps_scanned_at.clone(),
60 dependencies_error: self.deps_error.clone(),
61 lockfiles: serde_json::from_str(&self.lockfiles).unwrap_or_default(),
62 }
63 }
64
65 /// What `.g1t/dependencies.yml` said when it was last read.
66 pub fn version_updates(&self) -> VersionUpdatesState {
67 self.version_updates
68 .as_deref()
69 .and_then(|json| serde_json::from_str(json).ok())
70 .unwrap_or_default()
71 }
72}
73
74#[derive(Deserialize)]
75struct SecretRow {
76 id: String,
77 repo_id: String,
78 kind: String,
79 path: String,
80 line: i64,
81 commit_hash: String,
82 preview: String,
83 status: String,
84 source: String,
85 found_by: Option<String>,
86 found_at: String,
87 decided_by: Option<String>,
88 reason: Option<String>,
89 decided_at: Option<String>,
90 dismiss_reason: Option<String>,
91 test_value: Option<String>,
92 // The security suite's (migration 0004).
93 #[serde(default)]
94 validity: Option<String>,
95 #[serde(default)]
96 validity_checked_at: Option<String>,
97 #[serde(default)]
98 bypass_reason: Option<String>,
99 #[serde(default)]
100 bypass_comment: Option<String>,
101 #[serde(default)]
102 bypassed_by: Option<String>,
103 #[serde(default)]
104 bypassed_at: Option<String>,
105 #[serde(default)]
106 bypass_approved_by: Option<String>,
107 #[serde(default)]
108 pattern_id: Option<String>,
109 #[serde(default)]
110 pattern_name: Option<String>,
111 #[serde(default)]
112 location_count: Option<i64>,
113}
114
115/// Secrets with how many places each was found.
116/// Rewrites the work items kept by the old issue numbers of their
117/// repositories to their numbers in the workspace, `?1` a page of
118/// `work.migrated`'s items as JSON. Each new number is above every old one
119/// of its workspace, so a page heard twice changes nothing more.
120pub const WORK_MIGRATED: [&str; 6] = [
121 "UPDATE updates SET item = m.number FROM moved_items m WHERE updates.repo_id = m.repo_id AND updates.item = m.legacy",
122 "UPDATE update_pulls SET item = m.number FROM moved_items m WHERE update_pulls.repo_id = m.repo_id AND update_pulls.item = m.legacy",
123 "UPDATE upgrades SET item = m.number FROM moved_items m WHERE upgrades.repo_id = m.repo_id AND upgrades.item = m.legacy",
124 "UPDATE code_alerts SET item = m.number FROM moved_items m WHERE code_alerts.repo_id = m.repo_id AND code_alerts.item = m.legacy",
125 "UPDATE alert_fixes SET item = m.number FROM moved_items m WHERE alert_fixes.repo_id = m.repo_id AND alert_fixes.item = m.legacy",
126 // Only an update that needed code changes named a work item; the
127 // other steps name pull requests, which kept their numbers.
128 "UPDATE alert_activity SET number = m.number FROM moved_items m
129 WHERE alert_activity.repo_id = m.repo_id AND alert_activity.number = m.legacy AND alert_activity.action = 'update_needs_code'",
130];
131
132/// `work.migrated`'s items (`?1`), as the rows [`WORK_MIGRATED`] reads.
133const MOVED_ITEMS: &str = "WITH moved_items AS (
134 SELECT json_extract(value, '$.repoId') AS repo_id, json_extract(value, '$.legacyNumber') AS legacy,
135 json_extract(value, '$.number') AS number
136 FROM json_each(?1))";
137
138/// One of [`WORK_MIGRATED`], ready to run.
139pub fn work_migrated_sql(statement: &str) -> String {
140 format!("{MOVED_ITEMS}
141{statement}")
142}
143
144const SECRET_SELECT: &str =
145 "SELECT s.*, (SELECT count(*) FROM secret_locations l WHERE l.secret_id = s.id) AS location_count FROM secrets s";
146
147impl From<SecretRow> for SecretFinding {
148 fn from(row: SecretRow) -> Self {
149 let status = SecretStatus::parse(&row.status).unwrap_or(SecretStatus::Open);
150 let bypass = match (row.bypass_reason.as_deref().and_then(g1t_contracts::security_suite::BypassReason::parse), row.bypassed_by, row.bypassed_at) {
151 (Some(reason), Some(by), Some(at)) => Some(g1t_contracts::security_suite::Bypass {
152 reason,
153 comment: row.bypass_comment,
154 by,
155 at,
156 approved_by: row.bypass_approved_by,
157 }),
158 _ => None,
159 };
160 SecretFinding {
161 label: crate::secret_alerts::label_of(&row.kind, row.pattern_name.as_deref()),
162 validity: row.validity,
163 validity_checked_at: row.validity_checked_at,
164 bypass,
165 pattern_id: row.pattern_id,
166 pattern_name: row.pattern_name,
167 locations: row.location_count.unwrap_or(0).max(0) as u32,
168 id: row.id,
169 repo_id: row.repo_id,
170 kind: row.kind,
171 path: row.path,
172 line: row.line.max(0) as u32,
173 commit: row.commit_hash,
174 preview: row.preview,
175 status,
176 source: row.source,
177 found_by: row.found_by,
178 found_at: row.found_at,
179 decided_by: row.decided_by,
180 reason: row.reason,
181 decided_at: row.decided_at,
182 dismissed_reason: row.dismiss_reason.as_deref().and_then(DismissReason::parse),
183 test_value: row.test_value,
184 state: status.state(),
185 }
186 }
187}
188
189#[derive(Deserialize)]
190pub struct VulnRow {
191 pub id: String,
192 pub repo_id: String,
193 pub ecosystem: String,
194 pub package: String,
195 pub version: String,
196 pub manifest: String,
197 pub osv_id: String,
198 pub advisory: String,
199 pub summary: String,
200 pub severity: String,
201 pub fixed_version: Option<String>,
202 pub status: String,
203 pub found_at: String,
204 pub fixed_at: Option<String>,
205 pub item: Option<i64>,
206 #[serde(default)]
207 pub dismiss_reason: Option<String>,
208 #[serde(default)]
209 pub dismiss_comment: Option<String>,
210 #[serde(default)]
211 pub dismissed_by: Option<String>,
212 #[serde(default)]
213 pub dismissed_at: Option<String>,
214}
215
216impl From<VulnRow> for Vulnerability {
217 fn from(row: VulnRow) -> Self {
218 let status = VulnStatus::parse(&row.status).unwrap_or(VulnStatus::Open);
219 let dismissed = status == VulnStatus::Dismissed;
220 Vulnerability {
221 id: row.id,
222 repo_id: row.repo_id,
223 ecosystem: row.ecosystem,
224 package: row.package,
225 version: row.version,
226 manifest: row.manifest,
227 advisory: row.advisory,
228 osv_id: row.osv_id,
229 summary: row.summary,
230 severity: row.severity,
231 fixed_version: row.fixed_version,
232 status,
233 item: row.item.map(|number| number as u32),
234 found_at: row.found_at,
235 fixed_at: row.fixed_at,
236 state: status.state(),
237 dismissed_by: row.dismissed_by.filter(|_| dismissed),
238 dismissed_reason: row.dismiss_reason.as_deref().and_then(DismissReason::parse).filter(|_| dismissed),
239 dismissed_comment: row.dismiss_comment.filter(|_| dismissed),
240 dismissed_at: row.dismissed_at.filter(|_| dismissed),
241 update: None,
242 }
243 }
244}
245
246/// A security update, as stored.
247#[derive(Clone, Deserialize)]
248pub struct UpdateRow {
249 pub repo_id: String,
250 pub ecosystem: String,
251 pub package: String,
252 pub target: String,
253 pub state: String,
254 pub branch: Option<String>,
255 pub pull: Option<i64>,
256 pub item: Option<i64>,
257 pub error: Option<String>,
258 pub updated_at: String,
259}
260
261impl UpdateRow {
262 pub fn state(&self) -> UpdateState {
263 UpdateState::parse(&self.state).unwrap_or(UpdateState::Failed)
264 }
265
266 pub fn pull(&self) -> Option<u32> {
267 self.pull.map(|n| n.max(0) as u32)
268 }
269
270 pub fn item(&self) -> Option<u32> {
271 self.item.map(|n| n.max(0) as u32)
272 }
273
274 pub fn to_contract(&self) -> SecurityUpdate {
275 SecurityUpdate {
276 state: self.state(),
277 target: self.target.clone(),
278 branch: self.branch.clone(),
279 pull: self.pull(),
280 item: self.item(),
281 error: self.error.clone(),
282 updated_at: self.updated_at.clone(),
283 }
284 }
285}
286
287#[derive(Deserialize)]
288struct ActivityRow {
289 id: String,
290 alert_id: String,
291 action: String,
292 actor: Option<String>,
293 reason: Option<String>,
294 comment: Option<String>,
295 number: Option<i64>,
296 at: String,
297}
298
299impl From<ActivityRow> for AlertActivity {
300 fn from(row: ActivityRow) -> Self {
301 AlertActivity {
302 id: row.id,
303 alert_id: row.alert_id,
304 action: row.action,
305 actor: row.actor,
306 reason: row.reason.as_deref().and_then(DismissReason::parse),
307 comment: row.comment,
308 number: row.number.map(|n| n.max(0) as u32),
309 at: row.at,
310 }
311 }
312}
313
314/// A push too large to scan before it was stored, scanned after it landed.
315#[derive(Clone, Deserialize)]
316pub struct PushScanRow {
317 pub id: String,
318 pub repo_id: String,
319 pub git_ref: String,
320 pub head: String,
321 pub base: Option<String>,
322 pub cursor: Option<String>,
323 pub pusher: Option<String>,
324 pub pages: i64,
325}
326
327/// One action on an alert, to record.
328pub struct Activity<'a> {
329 pub alert_id: &'a str,
330 pub action: &'a str,
331 pub actor: Option<&'a str>,
332 pub reason: Option<DismissReason>,
333 pub comment: Option<&'a str>,
334 pub number: Option<u32>,
335}
336
337#[derive(Deserialize)]
338pub struct UpgradeRow {
339 pub item: i64,
340}
341
342#[derive(Deserialize)]
343struct CountRow {
344 severity: String,
345 n: i64,
346}
347
348#[derive(Deserialize)]
349struct SecretCountRow {
350 status: String,
351 test: i64,
352 n: i64,
353}
354
355#[derive(Deserialize)]
356struct NumberRow {
357 n: i64,
358}
359
360#[derive(Deserialize)]
361struct FingerprintRow {
362 fingerprint: String,
363 id: String,
364 status: String,
365}
366
367#[derive(Deserialize)]
368struct BodyRow {
369 body: String,
370 fetched_at: String,
371}
372
373pub struct Store {
374 pub db: D1Database,
375}
376
377const VULN_COLUMNS: &str = "v.id, v.repo_id, v.ecosystem, v.package, v.version, v.manifest, v.osv_id, v.advisory,
378 v.summary, v.severity, v.fixed_version, v.status, v.found_at, v.fixed_at, v.dismiss_reason, v.dismiss_comment,
379 v.dismissed_by, v.dismissed_at, COALESCE(n.item, u.item) AS item";
380
381/// The work item an upgrade has: the security update's, when it went to
382/// g1t, or one opened before security updates.
383const VULN_JOINS: &str = "LEFT JOIN upgrades u ON u.repo_id = v.repo_id AND u.ecosystem = v.ecosystem AND u.package = v.package
384 LEFT JOIN updates n ON n.repo_id = v.repo_id AND n.ecosystem = v.ecosystem AND n.package = v.package";
385
386impl Store {
387 pub async fn repo(&self, repo_id: &str) -> Result<Option<RepoRow>> {
388 self.db
389 .prepare("SELECT * FROM repos WHERE repo_id = ?")
390 .bind(&[repo_id.into()])?
391 .first::<RepoRow>(None)
392 .await
393 }
394
395 /// Records a repository the first time it is seen, and keeps its
396 /// address current. Returns its row.
397 pub async fn register(&self, repo_id: &str, namespace: &str, name: &str) -> Result<RepoRow> {
398 self.db
399 .prepare(
400 "INSERT INTO repos (repo_id, namespace, name, created_at) VALUES (?1, ?2, ?3, ?4)
401 ON CONFLICT (repo_id) DO UPDATE SET namespace = ?2, name = ?3",
402 )
403 .bind(&[repo_id.into(), namespace.into(), name.into(), now().into()])?
404 .run()
405 .await?;
406 self.repo(repo_id)
407 .await?
408 .ok_or_else(|| worker::Error::RustError("the repository was not recorded".into()))
409 }
410
411 /// A repository's path changed: transferred or renamed.
412 pub async fn moved(&self, repo_id: &str, namespace: &str, name: &str) -> Result<()> {
413 self.db
414 .prepare("UPDATE repos SET namespace = ?, name = ? WHERE repo_id = ?")
415 .bind(&[namespace.into(), name.into(), repo_id.into()])?
416 .run()
417 .await?;
418 Ok(())
419 }
420
421 pub async fn rename_namespace(&self, stale: &[String], current: &str) -> Result<()> {
422 for slug in stale {
423 self.db
424 .prepare("UPDATE repos SET namespace = ? WHERE namespace = ?")
425 .bind(&[current.into(), slug.as_str().into()])?
426 .run()
427 .await?;
428 }
429 Ok(())
430 }
431
432 pub async fn in_namespace(&self, namespace: &str) -> Result<Vec<RepoRow>> {
433 self.db
434 .prepare("SELECT * FROM repos WHERE namespace = ? ORDER BY name")
435 .bind(&[namespace.into()])?
436 .all()
437 .await?
438 .results::<RepoRow>()
439 }
440
441 /// A repository purged: everything found in it goes. `?1` its id.
442 pub async fn purge(&self, repo_id: &str) -> Result<()> {
443 let mut batch = Vec::with_capacity(PURGED.len());
444 for sql in PURGED.iter().chain(crate::update_store::PURGED) {
445 batch.push(self.db.prepare(*sql).bind(&[repo_id.into()])?);
446 }
447 self.db.batch(batch).await?;
448 Ok(())
449 }
450
451 /// `work.migrated`: what was kept by old issue numbers names the work
452 /// items they became.
453 pub async fn work_migrated(&self, items: &[g1t_contracts::work::MigratedItem]) -> Result<()> {
454 if items.is_empty() {
455 return Ok(());
456 }
457 let page = serde_json::to_string(items)?;
458 let mut batch = Vec::with_capacity(WORK_MIGRATED.len());
459 for sql in WORK_MIGRATED {
460 batch.push(self.db.prepare(work_migrated_sql(sql)).bind(&[page.as_str().into()])?);
461 }
462 self.db.batch(batch).await?;
463 Ok(())
464 }
465
466 /// Records that a repository's daily dependency read was skipped, and
467 /// why, so the sweep moves on to others until the next day.
468 pub async fn skip_dependencies(&self, repo_id: &str, why: &str) -> Result<()> {
469 self.db
470 .prepare("UPDATE repos SET deps_scanned_at = ?, deps_error = ? WHERE repo_id = ?")
471 .bind(&[now().into(), why.into(), repo_id.into()])?
472 .run()
473 .await?;
474 Ok(())
475 }
476
477 /// Repositories whose history still has to be scanned, oldest first.
478 pub async fn unfinished_histories(&self, limit: u32) -> Result<Vec<RepoRow>> {
479 self.db
480 .prepare("SELECT * FROM repos WHERE history IN ('pending', 'running', 'stopped') ORDER BY created_at LIMIT ?")
481 .bind(&[limit.into()])?
482 .all()
483 .await?
484 .results::<RepoRow>()
485 }
486
487 /// Repositories whose dependencies were last read before `before`.
488 pub async fn stale_dependencies(&self, before: &str, limit: u32) -> Result<Vec<RepoRow>> {
489 self.db
490 .prepare(
491 "SELECT * FROM repos WHERE deps_scanned_at IS NULL OR deps_scanned_at < ?
492 ORDER BY deps_scanned_at LIMIT ?",
493 )
494 .bind(&[before.into(), limit.into()])?
495 .all()
496 .await?
497 .results::<RepoRow>()
498 }
499
500 pub async fn set_history(&self, repo_id: &str, state: &str, cursor: Option<&str>, commits: u32) -> Result<()> {
501 let finished = (state == "done").then(now);
502 self.db
503 .prepare(
504 "UPDATE repos SET history = ?, history_cursor = ?, history_commits = history_commits + ?,
505 history_finished_at = COALESCE(?, history_finished_at)
506 WHERE repo_id = ?",
507 )
508 .bind(&[state.into(), optional(cursor), commits.into(), optional(finished.as_deref()), repo_id.into()])?
509 .run()
510 .await?;
511 Ok(())
512 }
513
514 pub async fn restart_history(&self, repo_id: &str) -> Result<()> {
515 self.db
516 .prepare(
517 "UPDATE repos SET history = 'pending', history_cursor = NULL, history_commits = 0,
518 history_finished_at = NULL WHERE repo_id = ?",
519 )
520 .bind(&[repo_id.into()])?
521 .run()
522 .await?;
523 Ok(())
524 }
525
526 pub async fn set_dependencies_scanned(
527 &self,
528 repo_id: &str,
529 commit: Option<&str>,
530 lockfiles: &[String],
531 error: Option<&str>,
532 ) -> Result<()> {
533 self.db
534 .prepare("UPDATE repos SET deps_scanned_at = ?, deps_commit = ?, lockfiles = ?, deps_error = ? WHERE repo_id = ?")
535 .bind(&[now().into(), optional(commit), serde_json::to_string(lockfiles)?.into(), optional(error), repo_id.into()])?
536 .run()
537 .await?;
538 Ok(())
539 }
540
541 pub async fn set_version_updates(&self, repo_id: &str, state: &VersionUpdatesState) -> Result<()> {
542 self.db
543 .prepare("UPDATE repos SET version_updates = ? WHERE repo_id = ?")
544 .bind(&[serde_json::to_string(state)?.into(), repo_id.into()])?
545 .run()
546 .await?;
547 Ok(())
548 }
549
550 pub async fn set_upkeep(&self, repo_id: &str, enabled: bool, by: &str) -> Result<()> {
551 self.db
552 .prepare("UPDATE repos SET upkeep = ?, upkeep_by = ?, upkeep_at = ? WHERE repo_id = ?")
553 .bind(&[i32::from(enabled).into(), by.into(), now().into(), repo_id.into()])?
554 .run()
555 .await?;
556 Ok(())
557 }
558
559 // --- Secrets ------------------------------------------------------------
560
561 pub async fn secrets(&self, repo_id: &str) -> Result<Vec<SecretFinding>> {
562 let rows = self
563 .db
564 .prepare(format!(
565 "{SECRET_SELECT} WHERE s.repo_id = ?
566 ORDER BY CASE status WHEN 'open' THEN 0 WHEN 'blocked' THEN 1 WHEN 'allowed' THEN 2 ELSE 3 END,
567 test_value IS NOT NULL, found_at DESC LIMIT 500"
568 ))
569 .bind(&[repo_id.into()])?
570 .all()
571 .await?
572 .results::<SecretRow>()?;
573 Ok(rows.into_iter().map(SecretFinding::from).collect())
574 }
575
576 pub async fn secret(&self, repo_id: &str, id: &str) -> Result<Option<SecretFinding>> {
577 Ok(self
578 .db
579 .prepare(format!("{SECRET_SELECT} WHERE s.repo_id = ? AND s.id = ?"))
580 .bind(&[repo_id.into(), id.into()])?
581 .first::<SecretRow>(None)
582 .await?
583 .map(SecretFinding::from))
584 }
585
586 /// The findings already known for these fingerprints: fingerprint, id
587 /// and status.
588 pub async fn known(&self, repo_id: &str, fingerprints: &[String]) -> Result<Vec<(String, String, String)>> {
589 let mut known = Vec::new();
590 for chunk in fingerprints.chunks(90) {
591 let marks = vec!["?"; chunk.len()].join(", ");
592 let mut binds: Vec<JsValue> = vec![repo_id.into()];
593 binds.extend(chunk.iter().map(|fingerprint| JsValue::from(fingerprint.as_str())));
594 let rows = self
595 .db
596 .prepare(format!("SELECT fingerprint, id, status FROM secrets WHERE repo_id = ? AND fingerprint IN ({marks})"))
597 .bind(&binds)?
598 .all()
599 .await?
600 .results::<FingerprintRow>()?;
601 known.extend(rows.into_iter().map(|row| (row.fingerprint, row.id, row.status)));
602 }
603 Ok(known)
604 }
605
606 /// Records secrets not seen before in this repository, with `status`.
607 /// A secret already known keeps its record and its decision.
608 pub async fn add_secrets(
609 &self,
610 repo_id: &str,
611 secrets: &[NewSecret],
612 status: SecretStatus,
613 source: &str,
614 found_by: Option<&str>,
615 ) -> Result<()> {
616 if secrets.is_empty() {
617 return Ok(());
618 }
619 let now_ms = now_ms();
620 let found_at = rfc3339(now_ms);
621 let statements = secrets
622 .iter()
623 .map(|secret| {
624 self.db
625 .prepare(
626 "INSERT INTO secrets (id, repo_id, fingerprint, kind, path, line, commit_hash, preview,
627 status, source, found_by, found_at, test_value)
628 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
629 ON CONFLICT (repo_id, fingerprint) DO UPDATE SET test_value = excluded.test_value",
630 )
631 .bind(&[
632 new_id("sec", now_ms).into(),
633 repo_id.into(),
634 secret.fingerprint.as_str().into(),
635 secret.kind.as_str().into(),
636 secret.path.as_str().into(),
637 secret.line.into(),
638 secret.commit.as_str().into(),
639 secret.preview.as_str().into(),
640 status.as_str().into(),
641 source.into(),
642 optional(found_by),
643 found_at.as_str().into(),
644 optional(secret.test_value.as_deref()),
645 ])
646 })
647 .collect::<Result<Vec<_>>>()?;
648 self.db.batch(statements).await?;
649 Ok(())
650 }
651
652 /// A secret found in history that was only ever blocked has landed
653 /// after all (it was allowed, then pushed): it is open now unless
654 /// someone allowed it.
655 pub async fn landed(&self, repo_id: &str, fingerprints: &[String]) -> Result<()> {
656 for chunk in fingerprints.chunks(90) {
657 let marks = vec!["?"; chunk.len()].join(", ");
658 let mut binds: Vec<JsValue> = vec![repo_id.into()];
659 binds.extend(chunk.iter().map(|fingerprint| JsValue::from(fingerprint.as_str())));
660 self.db
661 .prepare(format!(
662 "UPDATE secrets SET status = 'open', source = 'history'
663 WHERE repo_id = ? AND status = 'blocked' AND fingerprint IN ({marks})"
664 ))
665 .bind(&binds)?
666 .run()
667 .await?;
668 }
669 Ok(())
670 }
671
672 /// Dismisses a secret: allowed, or resolved when it was revoked.
673 pub async fn dismiss_secret(&self, repo_id: &str, id: &str, reason: DismissReason, by: &str, comment: Option<&str>) -> Result<()> {
674 self.db
675 .prepare(
676 "UPDATE secrets SET status = ?, decided_by = ?, reason = ?, decided_at = ?, dismiss_reason = ?
677 WHERE repo_id = ? AND id = ?",
678 )
679 .bind(&[
680 reason.secret_status().as_str().into(),
681 by.into(),
682 optional(comment),
683 now().into(),
684 reason.as_str().into(),
685 repo_id.into(),
686 id.into(),
687 ])?
688 .run()
689 .await?;
690 Ok(())
691 }
692
693 /// Opens a secret again, as `status` (open, or blocked for one that
694 /// never landed).
695 pub async fn reopen_secret(&self, repo_id: &str, id: &str, status: SecretStatus) -> Result<()> {
696 self.db
697 .prepare(
698 "UPDATE secrets SET status = ?, decided_by = NULL, reason = NULL, decided_at = NULL, dismiss_reason = NULL
699 WHERE repo_id = ? AND id = ?",
700 )
701 .bind(&[status.as_str().into(), repo_id.into(), id.into()])?
702 .run()
703 .await?;
704 Ok(())
705 }
706
707 // --- Vulnerabilities ----------------------------------------------------
708
709 pub async fn vulnerabilities(&self, repo_id: &str) -> Result<Vec<Vulnerability>> {
710 let rows = self
711 .db
712 .prepare(format!(
713 "SELECT {VULN_COLUMNS} FROM vulnerabilities v
714 {VULN_JOINS}
715 WHERE v.repo_id = ?
716 ORDER BY CASE v.status WHEN 'open' THEN 0 WHEN 'dismissed' THEN 1 ELSE 2 END,
717 CASE v.severity WHEN 'critical' THEN 0 WHEN 'high' THEN 1 WHEN 'medium' THEN 2 WHEN 'low' THEN 3 ELSE 4 END,
718 v.package LIMIT 1000"
719 ))
720 .bind(&[repo_id.into()])?
721 .all()
722 .await?
723 .results::<VulnRow>()?;
724 let updates: HashMap<(String, String), UpdateRow> = self
725 .updates(repo_id)
726 .await?
727 .into_iter()
728 .map(|row| ((row.ecosystem.clone(), row.package.clone()), row))
729 .collect();
730 Ok(rows
731 .into_iter()
732 .map(|row| {
733 let update = updates.get(&(row.ecosystem.clone(), row.package.clone())).map(UpdateRow::to_contract);
734 Vulnerability { update, ..Vulnerability::from(row) }
735 })
736 .collect())
737 }
738
739 pub async fn vulnerability(&self, repo_id: &str, id: &str) -> Result<Option<Vulnerability>> {
740 let row = self
741 .db
742 .prepare(format!("SELECT {VULN_COLUMNS} FROM vulnerabilities v {VULN_JOINS} WHERE v.repo_id = ? AND v.id = ?"))
743 .bind(&[repo_id.into(), id.into()])?
744 .first::<VulnRow>(None)
745 .await?;
746 let Some(row) = row else { return Ok(None) };
747 let update = self.update(repo_id, &row.ecosystem, &row.package).await?.map(|row| row.to_contract());
748 Ok(Some(Vulnerability { update, ..Vulnerability::from(row) }))
749 }
750
751 pub async fn dismiss_vulnerability(
752 &self,
753 repo_id: &str,
754 id: &str,
755 reason: DismissReason,
756 by: &str,
757 comment: Option<&str>,
758 ) -> Result<()> {
759 self.db
760 .prepare(
761 "UPDATE vulnerabilities SET status = CASE status WHEN 'fixed' THEN 'fixed' ELSE 'dismissed' END,
762 dismiss_reason = ?, dismiss_comment = ?, dismissed_by = ?, dismissed_at = ?
763 WHERE repo_id = ? AND id = ?",
764 )
765 .bind(&[reason.as_str().into(), optional(comment), by.into(), now().into(), repo_id.into(), id.into()])?
766 .run()
767 .await?;
768 Ok(())
769 }
770
771 pub async fn reopen_vulnerability(&self, repo_id: &str, id: &str) -> Result<()> {
772 self.db
773 .prepare(
774 "UPDATE vulnerabilities SET status = CASE status WHEN 'dismissed' THEN 'open' ELSE status END,
775 dismiss_reason = NULL, dismiss_comment = NULL, dismissed_by = NULL, dismissed_at = NULL
776 WHERE repo_id = ? AND id = ?",
777 )
778 .bind(&[repo_id.into(), id.into()])?
779 .run()
780 .await?;
781 Ok(())
782 }
783
784 /// The ids of a package's open vulnerabilities.
785 pub async fn open_ids(&self, repo_id: &str, ecosystem: &str, package: &str) -> Result<Vec<String>> {
786 #[derive(Deserialize)]
787 struct IdRow {
788 id: String,
789 }
790 Ok(self
791 .db
792 .prepare("SELECT id FROM vulnerabilities WHERE repo_id = ? AND ecosystem = ? AND package = ? AND status = 'open'")
793 .bind(&[repo_id.into(), ecosystem.into(), package.into()])?
794 .all()
795 .await?
796 .results::<IdRow>()?
797 .into_iter()
798 .map(|row| row.id)
799 .collect())
800 }
801
802 pub async fn open_vulnerabilities(&self, repo_id: &str) -> Result<Vec<VulnRow>> {
803 self.db
804 .prepare(format!(
805 "SELECT {VULN_COLUMNS} FROM vulnerabilities v
806 {VULN_JOINS}
807 WHERE v.repo_id = ? AND v.status = 'open'"
808 ))
809 .bind(&[repo_id.into()])?
810 .all()
811 .await?
812 .results::<VulnRow>()
813 }
814
815 /// Every vulnerability alert of a repository: open, fixed and dismissed.
816 pub async fn all_vulnerabilities(&self, repo_id: &str) -> Result<Vec<VulnRow>> {
817 self.db
818 .prepare(format!("SELECT {VULN_COLUMNS} FROM vulnerabilities v {VULN_JOINS} WHERE v.repo_id = ?"))
819 .bind(&[repo_id.into()])?
820 .all()
821 .await?
822 .results::<VulnRow>()
823 }
824
825 /// Replaces what is known about a repository's dependencies with what a
826 /// scan found: new findings open, findings no longer true fixed, and
827 /// findings that came back open again, or dismissed again when someone
828 /// had dismissed them.
829 pub async fn replace_vulnerabilities(&self, repo_id: &str, found: &[Vulnerability]) -> Result<()> {
830 let now_ms = now_ms();
831 let now = rfc3339(now_ms);
832 let mut statements = vec![self
833 .db
834 .prepare(
835 "UPDATE vulnerabilities SET status = 'fixed', fixed_at = ? WHERE repo_id = ? AND status IN ('open', 'dismissed')",
836 )
837 .bind(&[now.as_str().into(), repo_id.into()])?];
838 for vuln in found {
839 statements.push(
840 self.db
841 .prepare(
842 "INSERT INTO vulnerabilities (id, repo_id, ecosystem, package, version, manifest, osv_id, advisory,
843 summary, severity, fixed_version, status, found_at)
844 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'open', ?)
845 ON CONFLICT (repo_id, ecosystem, package, version, manifest, osv_id) DO UPDATE SET
846 status = CASE WHEN vulnerabilities.dismiss_reason IS NULL THEN 'open' ELSE 'dismissed' END,
847 fixed_at = NULL, advisory = excluded.advisory, summary = excluded.summary,
848 severity = excluded.severity, fixed_version = excluded.fixed_version",
849 )
850 .bind(&[
851 new_id("vul", now_ms).into(),
852 repo_id.into(),
853 vuln.ecosystem.as_str().into(),
854 vuln.package.as_str().into(),
855 vuln.version.as_str().into(),
856 vuln.manifest.as_str().into(),
857 vuln.osv_id.as_str().into(),
858 vuln.advisory.as_str().into(),
859 vuln.summary.as_str().into(),
860 vuln.severity.as_str().into(),
861 optional(vuln.fixed_version.as_deref()),
862 now.as_str().into(),
863 ])?,
864 );
865 }
866 // D1 runs a batch as one transaction, so readers never see the
867 // moment between marking everything fixed and opening it again.
868 self.db.batch(statements).await?;
869 Ok(())
870 }
871
872 pub async fn counts(&self, repo_id: &str) -> Result<(SeverityCounts, u32, u32)> {
873 let rows = self
874 .db
875 .prepare("SELECT severity, count(*) AS n FROM vulnerabilities WHERE repo_id = ? AND status = 'open' GROUP BY severity")
876 .bind(&[repo_id.into()])?
877 .all()
878 .await?
879 .results::<CountRow>()?;
880 let mut counts = SeverityCounts::default();
881 let mut vulns = 0;
882 for row in rows {
883 let n = row.n.max(0) as u32;
884 vulns += n;
885 match row.severity.as_str() {
886 "critical" => counts.critical += n,
887 "high" => counts.high += n,
888 "medium" => counts.medium += n,
889 "low" => counts.low += n,
890 _ => counts.unknown += n,
891 }
892 }
893 let secrets = self.secret_counts(repo_id).await?;
894 // A secret in the history that looks real is the worst thing a
895 // repository can hold. One stopped at a push never landed, and a
896 // likely test value is no danger: neither is counted.
897 counts.critical += secrets.open;
898 Ok((counts, secrets.open + secrets.blocked, vulns))
899 }
900
901 pub async fn secret_counts(&self, repo_id: &str) -> Result<SecretCounts> {
902 let rows = self
903 .db
904 .prepare(
905 "SELECT status, (test_value IS NOT NULL) AS test, count(*) AS n FROM secrets WHERE repo_id = ?
906 GROUP BY status, test",
907 )
908 .bind(&[repo_id.into()])?
909 .all()
910 .await?
911 .results::<SecretCountRow>()?;
912 Ok(secret_counts(rows.iter().map(|row| (row.status.as_str(), row.test != 0, row.n.max(0) as u32))))
913 }
914
915 // --- Activity -----------------------------------------------------------
916
917 pub async fn record(&self, repo_id: &str, activity: &[Activity<'_>]) -> Result<()> {
918 if activity.is_empty() {
919 return Ok(());
920 }
921 let now_ms = now_ms();
922 let at = rfc3339(now_ms);
923 let statements = activity
924 .iter()
925 .map(|item| {
926 self.db
927 .prepare(
928 "INSERT INTO alert_activity (id, repo_id, alert_id, action, actor, reason, comment, number, at)
929 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
930 )
931 .bind(&[
932 new_id("act", now_ms).into(),
933 repo_id.into(),
934 item.alert_id.into(),
935 item.action.into(),
936 optional(item.actor),
937 optional(item.reason.map(DismissReason::as_str)),
938 optional(item.comment),
939 item.number.map_or(JsValue::NULL, JsValue::from),
940 at.as_str().into(),
941 ])
942 })
943 .collect::<Result<Vec<_>>>()?;
944 self.db.batch(statements).await?;
945 Ok(())
946 }
947
948 /// Each of `alert_ids`' latest action and comment, for the alerts that
949 /// have any.
950 pub async fn latest_activity(&self, repo_id: &str, alert_ids: &[String]) -> Result<HashMap<String, (String, Option<String>)>> {
951 #[derive(Deserialize)]
952 struct Latest {
953 alert_id: String,
954 action: String,
955 comment: Option<String>,
956 }
957 if alert_ids.is_empty() {
958 return Ok(HashMap::new());
959 }
960 let marks = vec!["?"; alert_ids.len()].join(", ");
961 let mut binds: Vec<JsValue> = vec![repo_id.into()];
962 binds.extend(alert_ids.iter().map(|id| JsValue::from(id.as_str())));
963 let rows = self
964 .db
965 .prepare(format!(
966 "SELECT alert_id, action, comment FROM alert_activity WHERE repo_id = ? AND alert_id IN ({marks})
967 ORDER BY at DESC, id DESC"
968 ))
969 .bind(&binds)?
970 .all()
971 .await?
972 .results::<Latest>()?;
973 let mut latest = HashMap::new();
974 for row in rows {
975 latest.entry(row.alert_id).or_insert((row.action, row.comment));
976 }
977 Ok(latest)
978 }
979
980 /// What happened to a repository's alerts, newest first.
981 pub async fn activity(&self, repo_id: &str, limit: u32) -> Result<Vec<AlertActivity>> {
982 Ok(self
983 .db
984 .prepare("SELECT * FROM alert_activity WHERE repo_id = ? ORDER BY at DESC, id DESC LIMIT ?")
985 .bind(&[repo_id.into(), limit.into()])?
986 .all()
987 .await?
988 .results::<ActivityRow>()?
989 .into_iter()
990 .map(AlertActivity::from)
991 .collect())
992 }
993
994 // --- Security updates ---------------------------------------------------
995
996 pub async fn updates(&self, repo_id: &str) -> Result<Vec<UpdateRow>> {
997 self.db
998 .prepare("SELECT * FROM updates WHERE repo_id = ?")
999 .bind(&[repo_id.into()])?
1000 .all()
1001 .await?
1002 .results::<UpdateRow>()
1003 }
1004
1005 pub async fn update(&self, repo_id: &str, ecosystem: &str, package: &str) -> Result<Option<UpdateRow>> {
1006 self.db
1007 .prepare("SELECT * FROM updates WHERE repo_id = ? AND ecosystem = ? AND package = ?")
1008 .bind(&[repo_id.into(), ecosystem.into(), package.into()])?
1009 .first::<UpdateRow>(None)
1010 .await
1011 }
1012
1013 pub async fn update_by_branch(&self, repo_id: &str, branch: &str) -> Result<Option<UpdateRow>> {
1014 self.db
1015 .prepare("SELECT * FROM updates WHERE repo_id = ? AND branch = ?")
1016 .bind(&[repo_id.into(), branch.into()])?
1017 .first::<UpdateRow>(None)
1018 .await
1019 }
1020
1021 pub async fn update_by_pull(&self, repo_id: &str, pull: u32) -> Result<Option<UpdateRow>> {
1022 self.db
1023 .prepare("SELECT * FROM updates WHERE repo_id = ? AND pull = ?")
1024 .bind(&[repo_id.into(), pull.into()])?
1025 .first::<UpdateRow>(None)
1026 .await
1027 }
1028
1029 /// Updates asked of a sandbox before `before` that never pushed.
1030 /// Updates left in `needs_code`, which g1t opened a work item for before
1031 /// it stopped doing that.
1032 pub async fn needs_code_updates(&self, limit: u32) -> Result<Vec<UpdateRow>> {
1033 self.db
1034 .prepare("SELECT * FROM updates WHERE state = 'needs_code' ORDER BY updated_at LIMIT ?")
1035 .bind(&[limit.into()])?
1036 .all()
1037 .await?
1038 .results::<UpdateRow>()
1039 }
1040
1041 pub async fn stalled_updates(&self, before: &str, limit: u32) -> Result<Vec<UpdateRow>> {
1042 self.db
1043 .prepare("SELECT * FROM updates WHERE state = 'requested' AND updated_at < ? ORDER BY updated_at LIMIT ?")
1044 .bind(&[before.into(), limit.into()])?
1045 .all()
1046 .await?
1047 .results::<UpdateRow>()
1048 }
1049
1050 /// Asks for a security update: a sandbox is making the change on
1051 /// `branch`. An open pull request for an older version is kept until
1052 /// the new one opens, which supersedes it.
1053 pub async fn request_update(&self, repo_id: &str, ecosystem: &str, package: &str, target: &str, branch: &str) -> Result<()> {
1054 let now = now();
1055 self.db
1056 .prepare(
1057 "INSERT INTO updates (repo_id, ecosystem, package, target, state, branch, requested_at, updated_at)
1058 VALUES (?1, ?2, ?3, ?4, 'requested', ?5, ?6, ?6)
1059 ON CONFLICT (repo_id, ecosystem, package) DO UPDATE SET
1060 target = ?4, state = 'requested', branch = ?5, error = NULL, item = NULL,
1061 pull = CASE WHEN updates.state = 'open' THEN updates.pull ELSE NULL END,
1062 requested_at = ?6, updated_at = ?6",
1063 )
1064 .bind(&[repo_id.into(), ecosystem.into(), package.into(), target.into(), branch.into(), now.into()])?
1065 .run()
1066 .await?;
1067 Ok(())
1068 }
1069
1070 pub async fn set_update(
1071 &self,
1072 row: &UpdateRow,
1073 state: UpdateState,
1074 pull: Option<u32>,
1075 item: Option<u32>,
1076 error: Option<&str>,
1077 ) -> Result<()> {
1078 self.db
1079 .prepare(
1080 "UPDATE updates SET state = ?, pull = ?, item = ?, error = ?, updated_at = ?
1081 WHERE repo_id = ? AND ecosystem = ? AND package = ?",
1082 )
1083 .bind(&[
1084 state.as_str().into(),
1085 pull.map_or(JsValue::NULL, JsValue::from),
1086 item.map_or(JsValue::NULL, JsValue::from),
1087 optional(error),
1088 now().into(),
1089 row.repo_id.as_str().into(),
1090 row.ecosystem.as_str().into(),
1091 row.package.as_str().into(),
1092 ])?
1093 .run()
1094 .await?;
1095 Ok(())
1096 }
1097
1098 // --- Pushes scanned after they landed -----------------------------------
1099
1100 pub async fn add_push_scan(&self, repo_id: &str, git_ref: &str, head: &str, base: Option<&str>, pusher: Option<&str>) -> Result<String> {
1101 let id = new_id("psc", now_ms());
1102 self.db
1103 .prepare(
1104 "INSERT INTO push_scans (id, repo_id, git_ref, head, base, pusher, created_at) VALUES (?, ?, ?, ?, ?, ?, ?)",
1105 )
1106 .bind(&[id.as_str().into(), repo_id.into(), git_ref.into(), head.into(), optional(base), optional(pusher), now().into()])?
1107 .run()
1108 .await?;
1109 Ok(id)
1110 }
1111
1112 pub async fn push_scan(&self, id: &str) -> Result<Option<PushScanRow>> {
1113 self.db
1114 .prepare("SELECT * FROM push_scans WHERE id = ?")
1115 .bind(&[id.into()])?
1116 .first::<PushScanRow>(None)
1117 .await
1118 }
1119
1120 pub async fn pending_push_scans(&self, limit: u32) -> Result<Vec<PushScanRow>> {
1121 self.db
1122 .prepare("SELECT * FROM push_scans WHERE state = 'pending' ORDER BY created_at LIMIT ?")
1123 .bind(&[limit.into()])?
1124 .all()
1125 .await?
1126 .results::<PushScanRow>()
1127 }
1128
1129 /// Where a push's scan stands after a page: its next page, or done.
1130 pub async fn advance_push_scan(&self, id: &str, next: Option<&str>, commits: u32, found: u32) -> Result<()> {
1131 let done = next.is_none();
1132 self.db
1133 .prepare(
1134 "UPDATE push_scans SET cursor = ?, state = ?, commits = commits + ?, pages = pages + 1, found = found + ?,
1135 finished_at = CASE WHEN ? THEN ? ELSE finished_at END
1136 WHERE id = ?",
1137 )
1138 .bind(&[
1139 optional(next),
1140 if done { "done" } else { "pending" }.into(),
1141 commits.into(),
1142 found.into(),
1143 done.into(),
1144 now().into(),
1145 id.into(),
1146 ])?
1147 .run()
1148 .await?;
1149 Ok(())
1150 }
1151
1152 // --- Upgrades -----------------------------------------------------------
1153
1154 /// The work item opened for a package before security updates, if any.
1155 pub async fn upgrade(&self, repo_id: &str, ecosystem: &str, package: &str) -> Result<Option<UpgradeRow>> {
1156 self.db
1157 .prepare("SELECT item FROM upgrades WHERE repo_id = ? AND ecosystem = ? AND package = ?")
1158 .bind(&[repo_id.into(), ecosystem.into(), package.into()])?
1159 .first::<UpgradeRow>(None)
1160 .await
1161 }
1162
1163 // --- Advisories and usage -----------------------------------------------
1164
1165 /// OSV's record of a vulnerability, if one was fetched after `after`.
1166 pub async fn advisory(&self, osv_id: &str, after: &str) -> Result<Option<serde_json::Value>> {
1167 let row = self
1168 .db
1169 .prepare("SELECT body, fetched_at FROM advisories WHERE osv_id = ?")
1170 .bind(&[osv_id.into()])?
1171 .first::<BodyRow>(None)
1172 .await?;
1173 Ok(row
1174 .filter(|row| row.fetched_at.as_str() >= after)
1175 .and_then(|row| serde_json::from_str(&row.body).ok()))
1176 }
1177
1178 pub async fn keep_advisory(&self, osv_id: &str, body: &serde_json::Value) -> Result<()> {
1179 self.db
1180 .prepare(
1181 "INSERT INTO advisories (osv_id, body, fetched_at) VALUES (?1, ?2, ?3)
1182 ON CONFLICT (osv_id) DO UPDATE SET body = ?2, fetched_at = ?3",
1183 )
1184 .bind(&[osv_id.into(), body.to_string().into(), now().into()])?
1185 .run()
1186 .await?;
1187 Ok(())
1188 }
1189
1190 /// Adds to a workspace's scanning this month; returns the month
1191 /// (`YYYY-MM`) with what it has cost so far, in millionths of a dollar,
1192 /// so the total is reported under the month it was counted in.
1193 pub async fn meter(&self, workspace: &str, reads: u32, commits: u32, osv_queries: u32, cost_micros: i64) -> Result<(String, i64)> {
1194 let month = now()[..7].to_owned();
1195 self.db
1196 .prepare(
1197 "INSERT INTO usage (workspace, month, reads, commits, osv_queries, cost_micros) VALUES (?1, ?2, ?3, ?4, ?5, ?6)
1198 ON CONFLICT (workspace, month) DO UPDATE SET
1199 reads = reads + ?3, commits = commits + ?4, osv_queries = osv_queries + ?5,
1200 cost_micros = cost_micros + ?6",
1201 )
1202 .bind(&[
1203 workspace.into(),
1204 month.as_str().into(),
1205 reads.into(),
1206 commits.into(),
1207 osv_queries.into(),
1208 JsValue::from(cost_micros as f64),
1209 ])?
1210 .run()
1211 .await?;
1212 let total = self
1213 .db
1214 .prepare("SELECT cost_micros AS n FROM usage WHERE workspace = ? AND month = ?")
1215 .bind(&[workspace.into(), month.as_str().into()])?
1216 .first::<NumberRow>(None)
1217 .await?
1218 .map_or(0, |row| row.n);
1219 Ok((month, total))
1220 }
1221}
1222
1223/// Secret alerts counted by where they stand, from (status, looks like a
1224/// test value, how many).
1225pub fn secret_counts<'a>(rows: impl Iterator<Item = (&'a str, bool, u32)>) -> SecretCounts {
1226 let mut counts = SecretCounts::default();
1227 for (status, test, n) in rows {
1228 match (SecretStatus::parse(status), test) {
1229 (Some(SecretStatus::Open | SecretStatus::Blocked), true) => counts.test_values += n,
1230 (Some(SecretStatus::Open), false) => counts.open += n,
1231 (Some(SecretStatus::Blocked), false) => counts.blocked += n,
1232 (Some(SecretStatus::Allowed), _) => counts.dismissed += n,
1233 (Some(SecretStatus::Resolved), _) => counts.fixed += n,
1234 (None, _) => {}
1235 }
1236 }
1237 counts
1238}
1239
1240#[cfg(test)]
1241mod tests {
1242 use super::*;
1243
1244 /// The migrations, in order, from `migrations/`.
1245 fn migrations() -> Vec<(String, String)> {
1246 let dir = std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("migrations");
1247 let mut names: Vec<String> =
1248 std::fs::read_dir(&dir).unwrap().map(|entry| entry.unwrap().file_name().into_string().unwrap()).collect();
1249 names.sort();
1250 names.into_iter().map(|name| (name.clone(), std::fs::read_to_string(dir.join(&name)).unwrap())).collect()
1251 }
1252
1253 #[test]
1254 fn old_issue_numbers_become_their_work_items_numbers() {
1255 let db = rusqlite::Connection::open_in_memory().unwrap();
1256 let (before, after): (Vec<_>, Vec<_>) = migrations().into_iter().partition(|(name, _)| name.as_str() < "0007");
1257 for (name, sql) in &before {
1258 db.execute_batch(sql).unwrap_or_else(|error| panic!("{name}: {error}"));
1259 }
1260 db.execute_batch(
1261 "INSERT INTO updates (repo_id, ecosystem, package, target, state, pull, issue, requested_at, updated_at)
1262 VALUES ('rep_1', 'npm', 'a', '2', 'needs_code', NULL, 5, '', ''), ('rep_2', 'npm', 'a', '2', 'needs_code', NULL, 5, '', '');
1263 INSERT INTO upgrades (repo_id, ecosystem, package, number, target, opened_at) VALUES ('rep_1', 'npm', 'b', 3, '2', '');
1264 INSERT INTO alert_activity (id, repo_id, alert_id, action, number, at)
1265 VALUES ('a1', 'rep_1', 'vul_1', 'update_needs_code', 5, ''), ('a2', 'rep_1', 'vul_1', 'update_opened', 5, '');",
1266 )
1267 .unwrap();
1268 for (name, sql) in &after {
1269 db.execute_batch(sql).unwrap_or_else(|error| panic!("{name}: {error}"));
1270 }
1271 let page = serde_json::json!([
1272 { "repoId": "rep_1", "legacyNumber": 5, "workspace": "acme", "number": 41, "itemId": "iss_5" },
1273 { "repoId": "rep_1", "legacyNumber": 3, "workspace": "acme", "number": 40, "itemId": "iss_3" },
1274 ])
1275 .to_string();
1276 // Heard twice, as a queue may deliver it.
1277 for _ in 0..2 {
1278 for statement in WORK_MIGRATED {
1279 db.execute(&work_migrated_sql(statement), [&page]).unwrap();
1280 }
1281 }
1282 let item = |sql: &str| db.query_row(sql, [], |row| row.get::<_, Option<i64>>(0)).unwrap();
1283 assert_eq!(item("SELECT item FROM updates WHERE repo_id = 'rep_1'"), Some(41));
1284 // Another repository's number 5 was another issue.
1285 assert_eq!(item("SELECT item FROM updates WHERE repo_id = 'rep_2'"), Some(5));
1286 assert_eq!(item("SELECT item FROM upgrades WHERE repo_id = 'rep_1'"), Some(40));
1287 assert_eq!(item("SELECT number FROM alert_activity WHERE id = 'a1'"), Some(41));
1288 // A pull request kept its number.
1289 assert_eq!(item("SELECT number FROM alert_activity WHERE id = 'a2'"), Some(5));
1290 }
1291
1292 #[test]
1293 fn only_real_secrets_in_history_are_open() {
1294 let counts = secret_counts(
1295 [("open", false, 2), ("open", true, 3), ("blocked", false, 1), ("blocked", true, 4), ("allowed", true, 5), ("resolved", false, 6)]
1296 .into_iter(),
1297 );
1298 assert_eq!((counts.open, counts.blocked, counts.test_values, counts.dismissed, counts.fixed), (2, 1, 7, 5, 6));
1299 }
1300}