Sign in

flagon-io/g1t

public

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

g1t/services/security/src/store.rs

660 lines24,216 bytesCodeBlame
1//! The security service's tables in D1.
2
3use g1t_contracts::security::{
4 NewSecret, ScanState, SecretFinding, SecretStatus, SeverityCounts, VulnStatus, Vulnerability,
5};
6use g1t_contracts::time::rfc3339;
7use g1t_contracts::new_id;
8use g1t_kit::now_ms;
9use g1t_scan::secrets::SecretKind;
10use serde::Deserialize;
11use worker::wasm_bindgen::JsValue;
12use worker::{D1Database, Result};
13
14pub fn optional(value: Option<&str>) -> JsValue {
15 value.map_or(JsValue::NULL, JsValue::from)
16}
17
18pub fn now() -> String {
19 rfc3339(now_ms())
20}
21
22/// A repository purged (`repo.purged`): every row kept for it, `?1` its
23/// id. Advisories are shared by every repository, so they stay.
24pub const PURGED: &[&str] = &[
25 "DELETE FROM secrets WHERE repo_id = ?1",
26 "DELETE FROM vulnerabilities WHERE repo_id = ?1",
27 "DELETE FROM upgrades WHERE repo_id = ?1",
28 "DELETE FROM repos WHERE repo_id = ?1",
29];
30
31#[derive(Clone, Deserialize)]
32pub struct RepoRow {
33 pub repo_id: String,
34 pub namespace: String,
35 pub name: String,
36 pub upkeep: i64,
37 pub history: String,
38 pub history_cursor: Option<String>,
39 pub history_commits: i64,
40 pub history_finished_at: Option<String>,
41 pub deps_scanned_at: Option<String>,
42 pub deps_error: Option<String>,
43 pub lockfiles: String,
44}
45
46impl RepoRow {
47 pub fn scan_state(&self) -> ScanState {
48 ScanState {
49 history: self.history.clone(),
50 commits_scanned: self.history_commits.max(0) as u32,
51 history_finished_at: self.history_finished_at.clone(),
52 dependencies_scanned_at: self.deps_scanned_at.clone(),
53 dependencies_error: self.deps_error.clone(),
54 lockfiles: serde_json::from_str(&self.lockfiles).unwrap_or_default(),
55 }
56 }
57}
58
59#[derive(Deserialize)]
60struct SecretRow {
61 id: String,
62 repo_id: String,
63 kind: String,
64 path: String,
65 line: i64,
66 commit_hash: String,
67 preview: String,
68 status: String,
69 source: String,
70 found_by: Option<String>,
71 found_at: String,
72 decided_by: Option<String>,
73 reason: Option<String>,
74 decided_at: Option<String>,
75}
76
77impl From<SecretRow> for SecretFinding {
78 fn from(row: SecretRow) -> Self {
79 SecretFinding {
80 label: SecretKind::parse(&row.kind).map_or("a secret", |kind| kind.label()).to_owned(),
81 id: row.id,
82 repo_id: row.repo_id,
83 kind: row.kind,
84 path: row.path,
85 line: row.line.max(0) as u32,
86 commit: row.commit_hash,
87 preview: row.preview,
88 status: SecretStatus::parse(&row.status).unwrap_or(SecretStatus::Open),
89 source: row.source,
90 found_by: row.found_by,
91 found_at: row.found_at,
92 decided_by: row.decided_by,
93 reason: row.reason,
94 decided_at: row.decided_at,
95 }
96 }
97}
98
99#[derive(Deserialize)]
100pub struct VulnRow {
101 pub id: String,
102 pub repo_id: String,
103 pub ecosystem: String,
104 pub package: String,
105 pub version: String,
106 pub manifest: String,
107 pub osv_id: String,
108 pub advisory: String,
109 pub summary: String,
110 pub severity: String,
111 pub fixed_version: Option<String>,
112 pub status: String,
113 pub found_at: String,
114 pub fixed_at: Option<String>,
115 pub number: Option<i64>,
116}
117
118impl From<VulnRow> for Vulnerability {
119 fn from(row: VulnRow) -> Self {
120 Vulnerability {
121 id: row.id,
122 repo_id: row.repo_id,
123 ecosystem: row.ecosystem,
124 package: row.package,
125 version: row.version,
126 manifest: row.manifest,
127 advisory: row.advisory,
128 osv_id: row.osv_id,
129 summary: row.summary,
130 severity: row.severity,
131 fixed_version: row.fixed_version,
132 status: if row.status == "fixed" { VulnStatus::Fixed } else { VulnStatus::Open },
133 issue: row.number.map(|number| number as u32),
134 found_at: row.found_at,
135 fixed_at: row.fixed_at,
136 }
137 }
138}
139
140#[derive(Deserialize)]
141pub struct UpgradeRow {
142 pub number: i64,
143}
144
145#[derive(Deserialize)]
146struct CountRow {
147 severity: String,
148 n: i64,
149}
150
151#[derive(Deserialize)]
152struct NumberRow {
153 n: i64,
154}
155
156#[derive(Deserialize)]
157struct FingerprintRow {
158 fingerprint: String,
159 id: String,
160 status: String,
161}
162
163#[derive(Deserialize)]
164struct BodyRow {
165 body: String,
166 fetched_at: String,
167}
168
169pub struct Store {
170 pub db: D1Database,
171}
172
173const VULN_COLUMNS: &str = "v.id, v.repo_id, v.ecosystem, v.package, v.version, v.manifest, v.osv_id, v.advisory,
174 v.summary, v.severity, v.fixed_version, v.status, v.found_at, v.fixed_at, u.number AS number";
175
176impl Store {
177 pub async fn repo(&self, repo_id: &str) -> Result<Option<RepoRow>> {
178 self.db
179 .prepare("SELECT * FROM repos WHERE repo_id = ?")
180 .bind(&[repo_id.into()])?
181 .first::<RepoRow>(None)
182 .await
183 }
184
185 /// Records a repository the first time it is seen, and keeps its
186 /// address current. Returns its row.
187 pub async fn register(&self, repo_id: &str, namespace: &str, name: &str) -> Result<RepoRow> {
188 self.db
189 .prepare(
190 "INSERT INTO repos (repo_id, namespace, name, created_at) VALUES (?1, ?2, ?3, ?4)
191 ON CONFLICT (repo_id) DO UPDATE SET namespace = ?2, name = ?3",
192 )
193 .bind(&[repo_id.into(), namespace.into(), name.into(), now().into()])?
194 .run()
195 .await?;
196 self.repo(repo_id)
197 .await?
198 .ok_or_else(|| worker::Error::RustError("the repository was not recorded".into()))
199 }
200
201 /// A repository's path changed: transferred or renamed.
202 pub async fn moved(&self, repo_id: &str, namespace: &str, name: &str) -> Result<()> {
203 self.db
204 .prepare("UPDATE repos SET namespace = ?, name = ? WHERE repo_id = ?")
205 .bind(&[namespace.into(), name.into(), repo_id.into()])?
206 .run()
207 .await?;
208 Ok(())
209 }
210
211 pub async fn rename_namespace(&self, stale: &[String], current: &str) -> Result<()> {
212 for slug in stale {
213 self.db
214 .prepare("UPDATE repos SET namespace = ? WHERE namespace = ?")
215 .bind(&[current.into(), slug.as_str().into()])?
216 .run()
217 .await?;
218 }
219 Ok(())
220 }
221
222 pub async fn in_namespace(&self, namespace: &str) -> Result<Vec<RepoRow>> {
223 self.db
224 .prepare("SELECT * FROM repos WHERE namespace = ? ORDER BY name")
225 .bind(&[namespace.into()])?
226 .all()
227 .await?
228 .results::<RepoRow>()
229 }
230
231 /// A repository purged: everything found in it goes. `?1` its id.
232 pub async fn purge(&self, repo_id: &str) -> Result<()> {
233 let mut batch = Vec::with_capacity(PURGED.len());
234 for sql in PURGED {
235 batch.push(self.db.prepare(*sql).bind(&[repo_id.into()])?);
236 }
237 self.db.batch(batch).await?;
238 Ok(())
239 }
240
241 /// Records that a repository's daily dependency read was skipped, and
242 /// why, so the sweep moves on to others until the next day.
243 pub async fn skip_dependencies(&self, repo_id: &str, why: &str) -> Result<()> {
244 self.db
245 .prepare("UPDATE repos SET deps_scanned_at = ?, deps_error = ? WHERE repo_id = ?")
246 .bind(&[now().into(), why.into(), repo_id.into()])?
247 .run()
248 .await?;
249 Ok(())
250 }
251
252 /// Repositories whose history still has to be scanned, oldest first.
253 pub async fn unfinished_histories(&self, limit: u32) -> Result<Vec<RepoRow>> {
254 self.db
255 .prepare("SELECT * FROM repos WHERE history IN ('pending', 'running', 'stopped') ORDER BY created_at LIMIT ?")
256 .bind(&[limit.into()])?
257 .all()
258 .await?
259 .results::<RepoRow>()
260 }
261
262 /// Repositories whose dependencies were last read before `before`.
263 pub async fn stale_dependencies(&self, before: &str, limit: u32) -> Result<Vec<RepoRow>> {
264 self.db
265 .prepare(
266 "SELECT * FROM repos WHERE deps_scanned_at IS NULL OR deps_scanned_at < ?
267 ORDER BY deps_scanned_at LIMIT ?",
268 )
269 .bind(&[before.into(), limit.into()])?
270 .all()
271 .await?
272 .results::<RepoRow>()
273 }
274
275 pub async fn set_history(&self, repo_id: &str, state: &str, cursor: Option<&str>, commits: u32) -> Result<()> {
276 let finished = (state == "done").then(now);
277 self.db
278 .prepare(
279 "UPDATE repos SET history = ?, history_cursor = ?, history_commits = history_commits + ?,
280 history_finished_at = COALESCE(?, history_finished_at)
281 WHERE repo_id = ?",
282 )
283 .bind(&[state.into(), optional(cursor), commits.into(), optional(finished.as_deref()), repo_id.into()])?
284 .run()
285 .await?;
286 Ok(())
287 }
288
289 pub async fn restart_history(&self, repo_id: &str) -> Result<()> {
290 self.db
291 .prepare(
292 "UPDATE repos SET history = 'pending', history_cursor = NULL, history_commits = 0,
293 history_finished_at = NULL WHERE repo_id = ?",
294 )
295 .bind(&[repo_id.into()])?
296 .run()
297 .await?;
298 Ok(())
299 }
300
301 pub async fn set_dependencies_scanned(
302 &self,
303 repo_id: &str,
304 commit: Option<&str>,
305 lockfiles: &[String],
306 error: Option<&str>,
307 ) -> Result<()> {
308 self.db
309 .prepare("UPDATE repos SET deps_scanned_at = ?, deps_commit = ?, lockfiles = ?, deps_error = ? WHERE repo_id = ?")
310 .bind(&[now().into(), optional(commit), serde_json::to_string(lockfiles)?.into(), optional(error), repo_id.into()])?
311 .run()
312 .await?;
313 Ok(())
314 }
315
316 pub async fn set_upkeep(&self, repo_id: &str, enabled: bool, by: &str) -> Result<()> {
317 self.db
318 .prepare("UPDATE repos SET upkeep = ?, upkeep_by = ?, upkeep_at = ? WHERE repo_id = ?")
319 .bind(&[i32::from(enabled).into(), by.into(), now().into(), repo_id.into()])?
320 .run()
321 .await?;
322 Ok(())
323 }
324
325 // --- Secrets ------------------------------------------------------------
326
327 pub async fn secrets(&self, repo_id: &str) -> Result<Vec<SecretFinding>> {
328 let rows = self
329 .db
330 .prepare(
331 "SELECT * FROM secrets WHERE repo_id = ?
332 ORDER BY CASE status WHEN 'open' THEN 0 WHEN 'blocked' THEN 1 WHEN 'allowed' THEN 2 ELSE 3 END,
333 found_at DESC LIMIT 500",
334 )
335 .bind(&[repo_id.into()])?
336 .all()
337 .await?
338 .results::<SecretRow>()?;
339 Ok(rows.into_iter().map(SecretFinding::from).collect())
340 }
341
342 pub async fn secret(&self, repo_id: &str, id: &str) -> Result<Option<SecretFinding>> {
343 Ok(self
344 .db
345 .prepare("SELECT * FROM secrets WHERE repo_id = ? AND id = ?")
346 .bind(&[repo_id.into(), id.into()])?
347 .first::<SecretRow>(None)
348 .await?
349 .map(SecretFinding::from))
350 }
351
352 /// The findings already known for these fingerprints: fingerprint, id
353 /// and status.
354 pub async fn known(&self, repo_id: &str, fingerprints: &[String]) -> Result<Vec<(String, String, String)>> {
355 let mut known = Vec::new();
356 for chunk in fingerprints.chunks(90) {
357 let marks = vec!["?"; chunk.len()].join(", ");
358 let mut binds: Vec<JsValue> = vec![repo_id.into()];
359 binds.extend(chunk.iter().map(|fingerprint| JsValue::from(fingerprint.as_str())));
360 let rows = self
361 .db
362 .prepare(format!("SELECT fingerprint, id, status FROM secrets WHERE repo_id = ? AND fingerprint IN ({marks})"))
363 .bind(&binds)?
364 .all()
365 .await?
366 .results::<FingerprintRow>()?;
367 known.extend(rows.into_iter().map(|row| (row.fingerprint, row.id, row.status)));
368 }
369 Ok(known)
370 }
371
372 /// Records secrets not seen before in this repository, with `status`.
373 /// A secret already known keeps its record and its decision.
374 pub async fn add_secrets(
375 &self,
376 repo_id: &str,
377 secrets: &[NewSecret],
378 status: SecretStatus,
379 source: &str,
380 found_by: Option<&str>,
381 ) -> Result<()> {
382 if secrets.is_empty() {
383 return Ok(());
384 }
385 let now_ms = now_ms();
386 let found_at = rfc3339(now_ms);
387 let statements = secrets
388 .iter()
389 .map(|secret| {
390 self.db
391 .prepare(
392 "INSERT INTO secrets (id, repo_id, fingerprint, kind, path, line, commit_hash, preview,
393 status, source, found_by, found_at)
394 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
395 ON CONFLICT (repo_id, fingerprint) DO NOTHING",
396 )
397 .bind(&[
398 new_id("sec", now_ms).into(),
399 repo_id.into(),
400 secret.fingerprint.as_str().into(),
401 secret.kind.as_str().into(),
402 secret.path.as_str().into(),
403 secret.line.into(),
404 secret.commit.as_str().into(),
405 secret.preview.as_str().into(),
406 status.as_str().into(),
407 source.into(),
408 optional(found_by),
409 found_at.as_str().into(),
410 ])
411 })
412 .collect::<Result<Vec<_>>>()?;
413 self.db.batch(statements).await?;
414 Ok(())
415 }
416
417 /// A secret found in history that was only ever blocked has landed
418 /// after all (it was allowed, then pushed): it is open now unless
419 /// someone allowed it.
420 pub async fn landed(&self, repo_id: &str, fingerprints: &[String]) -> Result<()> {
421 for chunk in fingerprints.chunks(90) {
422 let marks = vec!["?"; chunk.len()].join(", ");
423 let mut binds: Vec<JsValue> = vec![repo_id.into()];
424 binds.extend(chunk.iter().map(|fingerprint| JsValue::from(fingerprint.as_str())));
425 self.db
426 .prepare(format!(
427 "UPDATE secrets SET status = 'open', source = 'history'
428 WHERE repo_id = ? AND status = 'blocked' AND fingerprint IN ({marks})"
429 ))
430 .bind(&binds)?
431 .run()
432 .await?;
433 }
434 Ok(())
435 }
436
437 pub async fn decide(&self, repo_id: &str, id: &str, status: SecretStatus, by: &str, reason: Option<&str>) -> Result<()> {
438 let reopened = status == SecretStatus::Open;
439 self.db
440 .prepare("UPDATE secrets SET status = ?, decided_by = ?, reason = ?, decided_at = ? WHERE repo_id = ? AND id = ?")
441 .bind(&[
442 status.as_str().into(),
443 optional((!reopened).then_some(by)),
444 optional(reason.filter(|_| !reopened)),
445 optional((!reopened).then(now).as_deref()),
446 repo_id.into(),
447 id.into(),
448 ])?
449 .run()
450 .await?;
451 Ok(())
452 }
453
454 // --- Vulnerabilities ----------------------------------------------------
455
456 pub async fn vulnerabilities(&self, repo_id: &str) -> Result<Vec<Vulnerability>> {
457 let rows = self
458 .db
459 .prepare(format!(
460 "SELECT {VULN_COLUMNS} FROM vulnerabilities v
461 LEFT JOIN upgrades u ON u.repo_id = v.repo_id AND u.ecosystem = v.ecosystem AND u.package = v.package
462 WHERE v.repo_id = ?
463 ORDER BY CASE v.status WHEN 'open' THEN 0 ELSE 1 END,
464 CASE v.severity WHEN 'critical' THEN 0 WHEN 'high' THEN 1 WHEN 'medium' THEN 2 WHEN 'low' THEN 3 ELSE 4 END,
465 v.package LIMIT 1000"
466 ))
467 .bind(&[repo_id.into()])?
468 .all()
469 .await?
470 .results::<VulnRow>()?;
471 Ok(rows.into_iter().map(Vulnerability::from).collect())
472 }
473
474 pub async fn open_vulnerabilities(&self, repo_id: &str) -> Result<Vec<VulnRow>> {
475 self.db
476 .prepare(format!(
477 "SELECT {VULN_COLUMNS} FROM vulnerabilities v
478 LEFT JOIN upgrades u ON u.repo_id = v.repo_id AND u.ecosystem = v.ecosystem AND u.package = v.package
479 WHERE v.repo_id = ? AND v.status = 'open'"
480 ))
481 .bind(&[repo_id.into()])?
482 .all()
483 .await?
484 .results::<VulnRow>()
485 }
486
487 /// Replaces what is known about a repository's dependencies with what a
488 /// scan found: new findings open, findings no longer true fixed, and
489 /// findings that came back open again.
490 pub async fn replace_vulnerabilities(&self, repo_id: &str, found: &[Vulnerability]) -> Result<()> {
491 let now_ms = now_ms();
492 let now = rfc3339(now_ms);
493 let mut statements = vec![self
494 .db
495 .prepare("UPDATE vulnerabilities SET status = 'fixed', fixed_at = ? WHERE repo_id = ? AND status = 'open'")
496 .bind(&[now.as_str().into(), repo_id.into()])?];
497 for vuln in found {
498 statements.push(
499 self.db
500 .prepare(
501 "INSERT INTO vulnerabilities (id, repo_id, ecosystem, package, version, manifest, osv_id, advisory,
502 summary, severity, fixed_version, status, found_at)
503 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'open', ?)
504 ON CONFLICT (repo_id, ecosystem, package, version, manifest, osv_id) DO UPDATE SET
505 status = 'open', fixed_at = NULL, advisory = excluded.advisory, summary = excluded.summary,
506 severity = excluded.severity, fixed_version = excluded.fixed_version",
507 )
508 .bind(&[
509 new_id("vul", now_ms).into(),
510 repo_id.into(),
511 vuln.ecosystem.as_str().into(),
512 vuln.package.as_str().into(),
513 vuln.version.as_str().into(),
514 vuln.manifest.as_str().into(),
515 vuln.osv_id.as_str().into(),
516 vuln.advisory.as_str().into(),
517 vuln.summary.as_str().into(),
518 vuln.severity.as_str().into(),
519 optional(vuln.fixed_version.as_deref()),
520 now.as_str().into(),
521 ])?,
522 );
523 }
524 // D1 runs a batch as one transaction, so readers never see the
525 // moment between marking everything fixed and opening it again.
526 self.db.batch(statements).await?;
527 Ok(())
528 }
529
530 pub async fn counts(&self, repo_id: &str) -> Result<(SeverityCounts, u32, u32)> {
531 let rows = self
532 .db
533 .prepare("SELECT severity, count(*) AS n FROM vulnerabilities WHERE repo_id = ? AND status = 'open' GROUP BY severity")
534 .bind(&[repo_id.into()])?
535 .all()
536 .await?
537 .results::<CountRow>()?;
538 let mut counts = SeverityCounts::default();
539 let mut vulns = 0;
540 for row in rows {
541 let n = row.n.max(0) as u32;
542 vulns += n;
543 match row.severity.as_str() {
544 "critical" => counts.critical += n,
545 "high" => counts.high += n,
546 "medium" => counts.medium += n,
547 "low" => counts.low += n,
548 _ => counts.unknown += n,
549 }
550 }
551 let secrets = self
552 .db
553 .prepare("SELECT count(*) AS n FROM secrets WHERE repo_id = ? AND status IN ('open', 'blocked')")
554 .bind(&[repo_id.into()])?
555 .first::<NumberRow>(None)
556 .await?
557 .map_or(0, |row| row.n.max(0) as u32);
558 counts.critical += secrets;
559 Ok((counts, secrets, vulns))
560 }
561
562 // --- Upgrades -----------------------------------------------------------
563
564 pub async fn upgrade(&self, repo_id: &str, ecosystem: &str, package: &str) -> Result<Option<UpgradeRow>> {
565 self.db
566 .prepare("SELECT number FROM upgrades WHERE repo_id = ? AND ecosystem = ? AND package = ?")
567 .bind(&[repo_id.into(), ecosystem.into(), package.into()])?
568 .first::<UpgradeRow>(None)
569 .await
570 }
571
572 pub async fn record_upgrade(
573 &self,
574 repo_id: &str,
575 ecosystem: &str,
576 package: &str,
577 number: u32,
578 target: &str,
579 assigned: bool,
580 note: Option<&str>,
581 ) -> Result<()> {
582 self.db
583 .prepare(
584 "INSERT INTO upgrades (repo_id, ecosystem, package, number, target, opened_at, assigned, note)
585 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)
586 ON CONFLICT (repo_id, ecosystem, package) DO UPDATE SET
587 number = ?4, target = ?5, opened_at = ?6, assigned = ?7, note = ?8",
588 )
589 .bind(&[
590 repo_id.into(),
591 ecosystem.into(),
592 package.into(),
593 number.into(),
594 target.into(),
595 now().into(),
596 i32::from(assigned).into(),
597 optional(note),
598 ])?
599 .run()
600 .await?;
601 Ok(())
602 }
603
604 // --- Advisories and usage -----------------------------------------------
605
606 /// OSV's record of a vulnerability, if one was fetched after `after`.
607 pub async fn advisory(&self, osv_id: &str, after: &str) -> Result<Option<serde_json::Value>> {
608 let row = self
609 .db
610 .prepare("SELECT body, fetched_at FROM advisories WHERE osv_id = ?")
611 .bind(&[osv_id.into()])?
612 .first::<BodyRow>(None)
613 .await?;
614 Ok(row
615 .filter(|row| row.fetched_at.as_str() >= after)
616 .and_then(|row| serde_json::from_str(&row.body).ok()))
617 }
618
619 pub async fn keep_advisory(&self, osv_id: &str, body: &serde_json::Value) -> Result<()> {
620 self.db
621 .prepare(
622 "INSERT INTO advisories (osv_id, body, fetched_at) VALUES (?1, ?2, ?3)
623 ON CONFLICT (osv_id) DO UPDATE SET body = ?2, fetched_at = ?3",
624 )
625 .bind(&[osv_id.into(), body.to_string().into(), now().into()])?
626 .run()
627 .await?;
628 Ok(())
629 }
630
631 /// Adds to a workspace's scanning this month; returns what the month
632 /// has cost so far, in millionths of a dollar.
633 pub async fn meter(&self, workspace: &str, reads: u32, commits: u32, osv_queries: u32, cost_micros: i64) -> Result<i64> {
634 let month = now()[..7].to_owned();
635 self.db
636 .prepare(
637 "INSERT INTO usage (workspace, month, reads, commits, osv_queries, cost_micros) VALUES (?1, ?2, ?3, ?4, ?5, ?6)
638 ON CONFLICT (workspace, month) DO UPDATE SET
639 reads = reads + ?3, commits = commits + ?4, osv_queries = osv_queries + ?5,
640 cost_micros = cost_micros + ?6",
641 )
642 .bind(&[
643 workspace.into(),
644 month.as_str().into(),
645 reads.into(),
646 commits.into(),
647 osv_queries.into(),
648 JsValue::from(cost_micros as f64),
649 ])?
650 .run()
651 .await?;
652 Ok(self
653 .db
654 .prepare("SELECT cost_micros AS n FROM usage WHERE workspace = ? AND month = ?")
655 .bind(&[workspace.into(), month.into()])?
656 .first::<NumberRow>(None)
657 .await?
658 .map_or(0, |row| row.n))
659 }
660}