| 380 | 380 | | .then(|| (day[..10].to_owned(), workspace.to_lowercase(), slug(meter), count)) |
| 381 | 381 | | } |
| 382 | 382 | | |
| 383 | | − | /// A price meter's units from raw counts and `billable_units` weights: |
| 384 | | − | /// what g1t charges for, from what was counted. |
| 385 | | − | pub(crate) fn billable(raw: &[(String, f64)], weights: &BTreeMap<String, f64>) -> f64 { |
| 386 | | − | raw.iter().map(|(meter, count)| count * weights.get(meter).copied().unwrap_or(0.0)).sum() |
| 383 | + | /// The repos service's operation mapping, as `artifacts_usage` returns it |
| 384 | + | /// beside the rows: for each raw meter (slugged), how many operations it |
| 385 | + | /// is to Cloudflare (`cost_operations`) and to the customer |
| 386 | + | /// (`billable_operations`). Repos owns this mapping |
| 387 | + | /// (`set_operation_mapping`); billing only reads it. |
| 388 | + | pub(crate) fn operation_mapping(body: &Value) -> BTreeMap<String, (f64, f64)> { |
| 389 | + | body["mapping"] |
| 390 | + | .as_array() |
| 391 | + | .map(|rows| { |
| 392 | + | rows.iter() |
| 393 | + | .filter_map(|r| { |
| 394 | + | let meter = r["meter"].as_str()?; |
| 395 | + | Some((slug(meter), (r["cost_operations"].as_f64().unwrap_or(0.0), r["billable_operations"].as_f64().unwrap_or(0.0)))) |
| 396 | + | }) |
| 397 | + | .collect() |
| 398 | + | }) |
| 399 | + | .unwrap_or_default() |
| 400 | + | } |
| 401 | + | |
| 402 | + | /// Raw counts weighted by one column of the mapping: what Cloudflare |
| 403 | + | /// should count (`cost`), or what customers are charged for. |
| 404 | + | pub(crate) fn weighted(raw: &[(String, f64)], mapping: &BTreeMap<String, (f64, f64)>, cost: bool) -> f64 { |
| 405 | + | raw.iter() |
| 406 | + | .map(|(meter, count)| count * mapping.get(meter).map_or(0.0, |(c, b)| if cost { *c } else { *b })) |
| 407 | + | .sum() |
| 387 | 408 | | } |
| 388 | 409 | | |
| 389 | 410 | | impl Billing { |
| ⋯ |
| 507 | 528 | | .collect()) |
| 508 | 529 | | } |
| 509 | 530 | | |
| 510 | | − | /// g1t's own counts for the days: raw Artifacts meters from the repos |
| 511 | | − | /// service's `artifacts_usage` when it answers, and git operations, |
| 512 | | − | /// from those raw meters and `billable_units` weights when both exist, |
| 513 | | − | /// else from its `git_operations`. |
| 531 | + | /// g1t's own counts for the days, all from the repos service, which |
| 532 | + | /// owns the mapping from raw meters to operations (`operation_mapping`): |
| 533 | + | /// |
| 534 | + | /// - `git_operations`: what customers are charged for, as repos counts |
| 535 | + | /// it (`git_operations`, already through its mapping). |
| 536 | + | /// - `cost_operations`: what g1t expects Cloudflare to bill, from the |
| 537 | + | /// raw meters (`artifacts_usage`) and the mapping's cost column. |
| 538 | + | /// - `artifacts_<meter>`: each raw meter. |
| 514 | 539 | | pub(crate) async fn count_own(&self, since: &str, until: &str) -> Result<()> { |
| 515 | 540 | | let Some(repos) = &self.repos else { return Ok(()) }; |
| 516 | 541 | | let days = days_between(since, until); |
| 517 | 542 | | let fetched_at = rfc3339(now_ms()); |
| 518 | 543 | | let mut rows: Vec<(String, String, String, f64)> = Vec::new(); |
| 519 | 544 | | |
| 520 | | − | // Raw meters, while the repos service may not have them yet. |
| 521 | | − | let raw: Vec<(String, String, String, f64)> = |
| 522 | | − | match g1t_kit::call::<_, Value>(repos, "artifacts_usage", &json!({ "from": since, "to": until })).await { |
| 523 | | − | Ok(Value::Array(list)) => list.iter().filter_map(raw_usage_row).collect(), |
| 524 | | − | Ok(other) => other["rows"].as_array().map(|l| l.iter().filter_map(raw_usage_row).collect()).unwrap_or_default(), |
| 525 | | − | Err(_) => Vec::new(), |
| 526 | | − | }; |
| 527 | | − | #[derive(Deserialize)] |
| 528 | | − | struct Weight { |
| 529 | | − | raw_meter: String, |
| 530 | | − | weight: f64, |
| 531 | | − | } |
| 532 | | − | let weights: BTreeMap<String, f64> = self |
| 533 | | − | .db |
| 534 | | − | .prepare("SELECT raw_meter, weight FROM billable_units WHERE price_meter = 'git_operations'") |
| 535 | | − | .all() |
| 536 | | − | .await? |
| 537 | | − | .results::<Weight>()? |
| 538 | | − | .into_iter() |
| 539 | | − | .map(|w| (slug(&w.raw_meter), w.weight)) |
| 540 | | − | .collect(); |
| 541 | | − | for (day, workspace, meter, count) in &raw { |
| 542 | | − | rows.push((day.clone(), format!("artifacts_{meter}"), workspace.clone(), *count)); |
| 543 | | − | } |
| 544 | | − | if !raw.is_empty() && !weights.is_empty() { |
| 545 | + | // Raw meters and repos' mapping; skipped while repos does not answer. |
| 546 | + | if let Ok(body) = g1t_kit::call::<_, Value>(repos, "artifacts_usage", &json!({ "from": since, "to": until })).await { |
| 547 | + | let mapping = operation_mapping(&body); |
| 545 | 548 | | let mut by: BTreeMap<(String, String), Vec<(String, f64)>> = BTreeMap::new(); |
| 546 | | − | for (day, workspace, meter, count) in &raw { |
| 547 | | − | by.entry((day.clone(), workspace.clone())).or_default().push((meter.clone(), *count)); |
| 549 | + | for (day, workspace, meter, count) in body["rows"].as_array().map(|l| l.iter().filter_map(raw_usage_row).collect::<Vec<_>>()).unwrap_or_default() { |
| 550 | + | by.entry((day, workspace)).or_default().push((meter, count)); |
| 548 | 551 | | } |
| 549 | 552 | | for ((day, workspace), counts) in by { |
| 550 | | − | rows.push((day, "git_operations".to_owned(), workspace, billable(&counts, &weights))); |
| 551 | | − | } |
| 552 | | − | } else { |
| 553 | | − | let mut cumulative = Vec::with_capacity(days.len()); |
| 554 | | − | for day in &days { |
| 555 | | − | let list: Vec<WorkspaceGitOperations> = g1t_kit::call( |
| 556 | | − | repos, |
| 557 | | − | "git_operations", |
| 558 | | − | &GitOperationsArgs { month: day[..7].to_owned(), since: Some(format!("{day}T00")), namespace: None }, |
| 559 | | − | ) |
| 560 | | − | .await?; |
| 561 | | − | cumulative.push(list.into_iter().map(|w| (w.namespace.to_lowercase(), w.operations)).collect::<BTreeMap<_, _>>()); |
| 553 | + | // Several stores (namespaces) can give the same meter. |
| 554 | + | let mut merged: BTreeMap<String, f64> = BTreeMap::new(); |
| 555 | + | for (meter, count) in &counts { |
| 556 | + | *merged.entry(meter.clone()).or_default() += count; |
| 557 | + | } |
| 558 | + | for (meter, count) in &merged { |
| 559 | + | rows.push((day.clone(), format!("artifacts_{meter}"), workspace.clone(), *count)); |
| 560 | + | } |
| 561 | + | if !mapping.is_empty() { |
| 562 | + | rows.push((day, "cost_operations".to_owned(), workspace, weighted(&counts, &mapping, true))); |
| 563 | + | } |
| 562 | 564 | | } |
| 563 | | − | for (day, workspace, count) in daily_from_cumulative(&days, &cumulative) { |
| 564 | | − | rows.push((day, "git_operations".to_owned(), workspace, count as f64)); |
| 565 | | − | } |
| 566 | 565 | | } |
| 566 | + | // What customers are charged for, as repos counts it through its mapping. |
| 567 | + | let mut cumulative = Vec::with_capacity(days.len()); |
| 568 | + | for day in &days { |
| 569 | + | let list: Vec<WorkspaceGitOperations> = g1t_kit::call( |
| 570 | + | repos, |
| 571 | + | "git_operations", |
| 572 | + | &GitOperationsArgs { month: day[..7].to_owned(), since: Some(format!("{day}T00")), namespace: None }, |
| 573 | + | ) |
| 574 | + | .await?; |
| 575 | + | cumulative.push(list.into_iter().map(|w| (w.namespace.to_lowercase(), w.operations)).collect::<BTreeMap<_, _>>()); |
| 576 | + | } |
| 577 | + | for (day, workspace, count) in daily_from_cumulative(&days, &cumulative) { |
| 578 | + | rows.push((day, "git_operations".to_owned(), workspace, count as f64)); |
| 579 | + | } |
| 567 | 580 | | |
| 568 | 581 | | // Each day's counts replace what was there. |
| 569 | 582 | | self.db |
| ⋯ |
| 757 | 770 | | } |
| 758 | 771 | | |
| 759 | 772 | | #[test] |
| 760 | | − | fn raw_meters_become_billable_units_by_their_weights() { |
| 761 | | − | let row = raw_usage_row(&json!({ "day": "2026-10-15", "namespace": "Acme", "meter": "upload_pack", "count": 12 })).unwrap(); |
| 762 | | − | assert_eq!(row, ("2026-10-15".into(), "acme".into(), "upload_pack".into(), 12.0)); |
| 773 | + | fn raw_meters_are_weighted_by_the_repos_mapping() { |
| 774 | + | // As repos' artifacts_usage answers: rows, and its operation_mapping. |
| 775 | + | let body = json!({ |
| 776 | + | "rows": [ |
| 777 | + | { "day": "2026-10-15", "store": "g1t", "workspace": "Acme", "meter": "git.fetch", "count": 12, "bytes_in": 0, "bytes_out": 0 }, |
| 778 | + | { "day": "2026-10-15", "store": "g1t", "workspace": "acme", "meter": "git.receive_pack", "count": 3, "bytes_in": 0, "bytes_out": 0 }, |
| 779 | + | { "day": "2026-10-15", "store": "g1t", "workspace": "acme", "meter": "binding.read_blob", "count": 400, "bytes_in": 0, "bytes_out": 0 } |
| 780 | + | ], |
| 781 | + | "mapping": [ |
| 782 | + | { "meter": "git.fetch", "cost_operations": 1, "billable_operations": 1 }, |
| 783 | + | { "meter": "git.receive_pack", "cost_operations": 1, "billable_operations": 1 }, |
| 784 | + | { "meter": "binding.read_blob", "cost_operations": 0, "billable_operations": 0 } |
| 785 | + | ], |
| 786 | + | "truncated": false |
| 787 | + | }); |
| 788 | + | let row = raw_usage_row(&body["rows"][0]).unwrap(); |
| 789 | + | assert_eq!(row, ("2026-10-15".into(), "acme".into(), "git_fetch".into(), 12.0)); |
| 763 | 790 | | assert!(raw_usage_row(&json!({ "day": "2026-10-15", "count": 1 })).is_none()); |
| 764 | | − | let weights: BTreeMap<String, f64> = [("upload_pack".to_string(), 1.0), ("receive_pack".to_string(), 1.0), ("binding_read".to_string(), 0.0)].into(); |
| 765 | | − | let raw = vec![("upload_pack".to_string(), 12.0), ("receive_pack".to_string(), 3.0), ("binding_read".to_string(), 400.0), ("ls_refs".to_string(), 9.0)]; |
| 766 | | − | assert_eq!(billable(&raw, &weights), 15.0); |
| 767 | | − | // Cloudflare turns out to count binding reads: one row changes, and |
| 768 | | − | // so does what is counted from then on. |
| 769 | | − | let mut weights = weights; |
| 770 | | − | weights.insert("binding_read".into(), 1.0); |
| 771 | | − | assert_eq!(billable(&raw, &weights), 415.0); |
| 791 | + | let mut mapping = operation_mapping(&body); |
| 792 | + | let raw: Vec<(String, f64)> = body["rows"].as_array().unwrap().iter().filter_map(raw_usage_row).map(|r| (r.2, r.3)).collect(); |
| 793 | + | assert_eq!(weighted(&raw, &mapping, true), 15.0); |
| 794 | + | assert_eq!(weighted(&raw, &mapping, false), 15.0); |
| 795 | + | // Cloudflare turns out to bill binding reads: repos changes one row |
| 796 | + | // (set_operation_mapping), and the bill g1t expects follows. |
| 797 | + | mapping.insert("binding_read_blob".into(), (1.0, 0.0)); |
| 798 | + | assert_eq!(weighted(&raw, &mapping, true), 415.0); |
| 799 | + | assert_eq!(weighted(&raw, &mapping, false), 15.0); |
| 800 | + | assert!(operation_mapping(&json!({})).is_empty()); |
| 772 | 801 | | } |
| 773 | 802 | | |
| 774 | 803 | | #[test] |