Sign in

flagon-io/g1t

public

g1t/services/billing/src/margin.rs

1,605 lines72,164 bytesCodeBlame

Pick any line to see why it is the way it is: the commit, the pull request and issue it came from, and what the agent was thinking.

Git storage hardened, pages in tens of milliseconds, honest security alerts, and costs reconciled daily1//! What g1t earns on each thing it sells, measured against what
2//! Cloudflare actually charged for it.
3//!
4//! Once a day, after `costs` has read Cloudflare's bill, the reconciler
5//! puts three figures side by side for every day and each of g1t's
6//! products (a "bucket": sandboxes, deployments, git, repository storage,
7//! …):
8//!
9//! 1. **What Cloudflare charged**: the day's cost lines `cost_map` gives
10//! the bucket.
11//! 2. **What g1t's meters recorded**: the cost on the ledger's entries for
12//! it (the price book's cost at the time) and, where a mapping names
13//! one, g1t's own count of the same units (git operations).
14//! 3. **What customers were charged**: the entries' value at price, before
15//! the plan's included usage, a trial or a pool paid part of it; and of
16//! that, what workspaces paid. Month-end meters (git, storage, scans,
17//! embeddings, the actions cache) come from daily snapshots of what they
18//! had come to (`pending_days`). The plan's price is the `platform`
19//! bucket's: the plan pays for running g1t.
20//!
21//! From those: margin per product (value against cost) and for all of g1t
22//! (money in against every cost); drift (counts or costs that disagree past
23//! a mapping's threshold, and leaks: cost with no revenue, or a Cloudflare
24//! meter no one mapped); each workspace's cost, Cloudflare's figure shared
25//! out by each workspace's own meters; and price proposals when a unit's
26//! real cost has moved (`pricing`). Alerts go to staff by email and as a
27//! banner in sudo. See docs/BILLING_OPERATIONS.md.
28
29use std::collections::{BTreeMap, BTreeSet};
30
31use g1t_contracts::billing::*;
32use g1t_contracts::{FailureCode, Outcome, new_id};
33use g1t_contracts::time::rfc3339;
34use g1t_kit::now_ms;
35use serde::{Deserialize, Serialize};
36use worker::wasm_bindgen::JsValue;
37use worker::{Env, Result};
38
39use crate::Billing;
40use crate::costs::{self, ARTIFACTS_OPERATIONS, DAY_MS, Rule, SOURCE_ARTIFACTS, SOURCE_BILLABLE, UNMAPPED};
41
42/// Buckets that are the cost of running g1t, paid by the plan rather than
43/// sold by the unit: never a leak for having no revenue of their own.
44pub(crate) const OVERHEAD: [&str; 1] = ["platform"];
45/// Buckets Cloudflare does not bill: their cost is g1t's own figure.
46pub(crate) const NOT_CLOUDFLARE: [&str; 1] = ["models"];
47/// The days drift is judged over.
48const DRIFT_DAYS: u64 = 7;
49/// The days a workspace's cost is set against its revenue.
50const ANOMALY_DAYS: u64 = 30;
51/// The days a unit's cost is measured over.
52const MEASURE_DAYS: u64 = 30;
53/// Fewer of g1t's units than this say nothing about cost per unit.
54const MIN_UNITS: f64 = 1_000.0;
55/// An open alert is emailed again after this long.
56const REMIND_MS: u64 = 7 * DAY_MS;
57
58// ---------------------------------------------------------------------
59// The arithmetic, apart from the database so it can be tested.
60// ---------------------------------------------------------------------
61
62/// One of g1t's products on one day.
63#[derive(Clone, Debug, Default, PartialEq)]
64pub(crate) struct ProductDay {
65 pub day: String,
66 pub bucket: String,
67 /// What Cloudflare charged g1t, in millionths of a dollar.
68 pub cf_cost_micros: i64,
69 /// What g1t's meters recorded it cost (the price book's cost).
70 pub own_cost_micros: i64,
71 /// What customers were charged for it at price, before what paid.
72 pub value_micros: i64,
73 /// Of that, what workspaces paid themselves.
74 pub cash_micros: i64,
75 /// Units Cloudflare counted and units g1t counted, where a mapping
76 /// says they are the same units.
77 pub cf_quantity: f64,
78 pub own_quantity: f64,
79}
80
81impl ProductDay {
82 /// What it cost: Cloudflare's figure where Cloudflare bills it, else
83 /// g1t's own (models are billed by their providers, through the gateway).
84 pub fn cost(&self) -> i64 {
85 if NOT_CLOUDFLARE.contains(&self.bucket.as_str()) { self.own_cost_micros } else { self.cf_cost_micros }
86 }
87}
88
89/// A line of Cloudflare's bill, as stored.
90#[derive(Clone, Debug, Deserialize)]
91pub(crate) struct LineRow {
92 pub day: String,
93 pub source: String,
94 pub product: String,
95 pub meter: String,
96 pub quantity: f64,
97 pub cost_usd: f64,
98}
99
100/// A count of g1t's own, as stored.
101#[derive(Clone, Debug, Deserialize)]
102pub(crate) struct OwnRow {
103 pub day: String,
104 pub meter: String,
105 pub workspace: String,
106 pub quantity: f64,
107}
108
109/// What a workspace was charged for one key on one day.
110#[derive(Clone, Debug, Default, PartialEq)]
111pub(crate) struct UsageRow {
112 pub day: String,
113 pub workspace: String,
114 /// A ledger task (or `builds`), a month-end source, or `plan`.
115 pub key: String,
116 pub value: i64,
117 pub cash: i64,
118 pub cost: i64,
119}
120
121/// One workspace's share of a product's cost on one day.
122#[derive(Clone, Debug, PartialEq)]
123pub(crate) struct WorkspaceDay {
124 pub day: String,
125 pub workspace: String,
126 pub bucket: String,
127 pub cost: i64,
128 pub revenue: i64,
129}
130
131fn micros(dollars: f64) -> i64 {
132 (dollars * 1_000_000.0).round() as i64
133}
134
135/// Puts the day's bill, g1t's counts and what customers were charged side
136/// by side, a row per day and bucket, and shares each bucket's cost out
137/// to workspaces.
138pub(crate) fn fold(
139 rules: &[Rule],
140 revenue_map: &BTreeMap<String, String>,
141 lines: &[LineRow],
142 own: &[OwnRow],
143 usage: &[UsageRow],
144) -> (Vec<ProductDay>, Vec<WorkspaceDay>) {
145 let mut days: BTreeMap<(String, String), ProductDay> = BTreeMap::new();
146 let entry = |day: &str, bucket: &str| -> ProductDay {
147 ProductDay { day: day.to_owned(), bucket: bucket.to_owned(), ..ProductDay::default() }
148 };
149 // Which of g1t's own meters count each bucket's units.
150 let mut own_meters: BTreeMap<&str, BTreeSet<&str>> = BTreeMap::new();
151 for rule in rules {
152 if let Some(meter) = &rule.own_meter {
153 own_meters.entry(rule.bucket.as_str()).or_default().insert(meter.as_str());
154 }
155 }
156 let mut events: BTreeMap<(String, String), f64> = BTreeMap::new();
157 for line in lines {
158 let rule = costs::classify(rules, &line.product, &line.meter);
159 let bucket = rule.map_or(UNMAPPED, |r| r.bucket.as_str());
160 let key = (line.day.clone(), bucket.to_owned());
161 if line.source == SOURCE_ARTIFACTS {
162 // What Artifacts counted: operations only, and only where the
163 // bill does not count them itself.
164 if ARTIFACTS_OPERATIONS.contains(&line.meter.as_str()) {
165 *events.entry(key).or_default() += line.quantity;
166 }
167 continue;
168 }
169 let row = days.entry(key.clone()).or_insert_with(|| entry(&key.0, &key.1));
170 row.cf_cost_micros += micros(line.cost_usd);
171 if line.source == SOURCE_BILLABLE && rule.is_some_and(|r| r.own_meter.is_some()) {
172 row.cf_quantity += line.quantity;
173 }
174 }
175 for (key, quantity) in events {
176 let row = days.entry(key.clone()).or_insert_with(|| entry(&key.0, &key.1));
177 if row.cf_quantity == 0.0 {
178 row.cf_quantity = quantity;
179 }
180 }
181 // g1t's own counts of the same units, by bucket and by workspace.
182 let mut own_by: BTreeMap<(String, String), Vec<(String, f64)>> = BTreeMap::new();
183 // Cloudflare's own count by workspace, where it gives one
184 // (`cloudflare_<bucket>`): the best way to share its cost.
185 let mut cf_by: BTreeMap<(String, String), Vec<(String, f64)>> = BTreeMap::new();
186 for count in own {
187 if let Some(bucket) = count.meter.strip_prefix("cloudflare_") {
188 cf_by.entry((count.day.clone(), bucket.to_owned())).or_default().push((count.workspace.clone(), count.quantity));
189 continue;
190 }
191 for (bucket, meters) in &own_meters {
192 if meters.contains(count.meter.as_str()) {
193 let key = (count.day.clone(), (*bucket).to_owned());
194 days.entry(key.clone()).or_insert_with(|| entry(&key.0, &key.1)).own_quantity += count.quantity;
195 own_by.entry(key).or_default().push((count.workspace.clone(), count.quantity));
196 }
197 }
198 }
199 // What customers were charged.
200 let bucket_of = |key: &str| revenue_map.get(key).cloned().unwrap_or_else(|| "models".to_owned());
201 let mut value_by: BTreeMap<(String, String), Vec<(String, f64)>> = BTreeMap::new();
202 let mut cost_by: BTreeMap<(String, String), Vec<(String, f64)>> = BTreeMap::new();
203 let mut revenue: BTreeMap<(String, String, String), i64> = BTreeMap::new();
204 let mut active: BTreeMap<String, Vec<(String, f64)>> = BTreeMap::new();
205 for u in usage {
206 let bucket = bucket_of(&u.key);
207 let key = (u.day.clone(), bucket.clone());
208 let row = days.entry(key.clone()).or_insert_with(|| entry(&key.0, &key.1));
209 row.own_cost_micros += u.cost;
210 row.value_micros += u.value;
211 row.cash_micros += u.cash;
212 value_by.entry(key.clone()).or_default().push((u.workspace.clone(), u.value as f64));
213 cost_by.entry(key).or_default().push((u.workspace.clone(), u.cost as f64));
214 *revenue.entry((u.day.clone(), u.workspace.clone(), bucket)).or_default() += u.cash;
215 active.entry(u.day.clone()).or_default().push((u.workspace.clone(), u.value.max(u.cost) as f64));
216 }
217 // Each bucket's cost shared out: by Cloudflare's own count per
218 // workspace, else by g1t's own count of its units, else
219 // by what each workspace was charged for it, else by what its usage
220 // cost; running g1t, and what no one mapped, by each workspace's share
221 // of all usage that day.
222 let mut shares: BTreeMap<(String, String, String), i64> = BTreeMap::new();
223 for ((day, bucket), row) in &days {
224 let key = (day.clone(), bucket.clone());
225 let weigh = |m: &BTreeMap<(String, String), Vec<(String, f64)>>| m.get(&key).filter(|w| w.iter().any(|(_, v)| *v > 0.0)).cloned();
226 let weights = if OVERHEAD.contains(&bucket.as_str()) || bucket == UNMAPPED {
227 active.get(day).cloned()
228 } else {
229 weigh(&cf_by).or_else(|| weigh(&own_by)).or_else(|| weigh(&value_by)).or_else(|| weigh(&cost_by)).or_else(|| active.get(day).cloned())
230 };
231 for (workspace, micros) in attribute(row.cost(), &weights.unwrap_or_default()) {
232 *shares.entry((day.clone(), workspace, bucket.clone())).or_default() += micros;
233 }
234 }
235 let keys: BTreeSet<(String, String, String)> = shares.keys().chain(revenue.keys()).cloned().collect();
236 let workspaces = keys
237 .into_iter()
238 .map(|(day, workspace, bucket)| WorkspaceDay {
239 cost: shares.get(&(day.clone(), workspace.clone(), bucket.clone())).copied().unwrap_or(0),
240 revenue: revenue.get(&(day.clone(), workspace.clone(), bucket.clone())).copied().unwrap_or(0),
241 day,
242 workspace,
243 bucket,
244 })
245 .collect();
246 (days.into_values().collect(), workspaces)
247}
248
249/// A month-end source's day, from the snapshots of what it had come to:
250/// each day's figure less the day before's in the same month (the first
251/// day of a month, or the first snapshot, is its own).
252pub(crate) fn pending_deltas(snapshots: &[(String, String, String, i64, i64)]) -> Vec<UsageRow> {
253 // (day, workspace, source, cost, charge), any order.
254 let mut sorted = snapshots.to_vec();
255 sorted.sort_by(|a, b| (&a.1, &a.2, &a.0).cmp(&(&b.1, &b.2, &b.0)));
256 let mut out = Vec::new();
257 let mut previous: Option<&(String, String, String, i64, i64)> = None;
258 for snap in &sorted {
259 let (day, workspace, source, cost, charge) = snap;
260 let (before_cost, before_charge) = match previous {
261 Some(p) if p.1 == *workspace && p.2 == *source && p.0[..7] == day[..7] => (p.3, p.4),
262 _ => (0, 0),
263 };
264 let (cost, charge) = ((cost - before_cost).max(0), (charge - before_charge).max(0));
265 if cost > 0 || charge > 0 {
266 out.push(UsageRow { day: day.clone(), workspace: workspace.clone(), key: source.clone(), value: charge, cash: charge, cost });
267 }
268 previous = Some(snap);
269 }
270 out
271}
272
273/// Margin as a share of what was charged, in percent; None when nothing was.
274pub(crate) fn margin_percent(revenue_micros: i64, cost_micros: i64) -> Option<f64> {
275 (revenue_micros > 0).then(|| (revenue_micros - cost_micros) as f64 * 100.0 / revenue_micros as f64)
276}
277
278/// How far `ours` is from `theirs`, in percent of theirs; None when theirs
279/// is nothing.
280pub(crate) fn delta_percent(ours: f64, theirs: f64) -> Option<f64> {
281 (theirs > 0.0).then(|| (ours - theirs) * 100.0 / theirs)
282}
283
284#[derive(Clone, Copy, Debug, PartialEq, Eq)]
285pub(crate) enum DriftKind {
286 /// g1t counted a different number of units than Cloudflare did.
287 Count,
288 /// What Cloudflare charged differs from what the price book says the
289 /// same usage cost.
290 Cost,
291 /// Cloudflare charged for something nothing charges customers for.
292 Leak,
293}
294
295impl DriftKind {
296 pub fn as_str(self) -> &'static str {
297 match self {
298 DriftKind::Count => "count",
299 DriftKind::Cost => "cost",
300 DriftKind::Leak => "leak",
301 }
302 }
303}
304
305#[derive(Clone, Debug, PartialEq)]
306pub(crate) struct Drift {
307 pub bucket: String,
308 pub kind: DriftKind,
309 pub ours: f64,
310 pub cloudflare: f64,
311 pub delta_percent: Option<f64>,
312}
313
314/// Drift over a window for one bucket: counts more than `threshold`
315/// percent apart, a bill that far from the price book's cost of the same
316/// usage, and cost with nothing charged for it. Under `min_cost_micros`
317/// in all, cost says nothing.
318pub(crate) fn drifts(bucket: &str, days: &[ProductDay], threshold: f64, counted: bool, min_cost_micros: i64) -> Vec<Drift> {
319 let overhead = OVERHEAD.contains(&bucket);
320 let sum = |f: &dyn Fn(&ProductDay) -> f64| days.iter().map(f).sum::<f64>();
321 let cf_cost = sum(&|d| d.cf_cost_micros as f64);
322 let own_cost = sum(&|d| d.own_cost_micros as f64);
323 let value = sum(&|d| d.value_micros as f64);
324 let (cf_quantity, own_quantity) = (sum(&|d| d.cf_quantity), sum(&|d| d.own_quantity));
325 let mut out = Vec::new();
326 if counted && cf_quantity > 0.0 {
327 let delta = delta_percent(own_quantity, cf_quantity);
328 if delta.is_some_and(|d| d.abs() > threshold) {
329 out.push(Drift { bucket: bucket.into(), kind: DriftKind::Count, ours: own_quantity, cloudflare: cf_quantity, delta_percent: delta });
330 }
331 }
332 let enough = cf_cost.max(own_cost) >= min_cost_micros as f64;
333 if enough && !overhead && cf_cost > 0.0 && own_cost > 0.0 && !NOT_CLOUDFLARE.contains(&bucket) {
334 let delta = delta_percent(own_cost, cf_cost);
335 if delta.is_some_and(|d| d.abs() > threshold) {
336 out.push(Drift { bucket: bucket.into(), kind: DriftKind::Cost, ours: own_cost, cloudflare: cf_cost, delta_percent: delta });
337 }
338 }
339 if !overhead && cf_cost >= min_cost_micros as f64 && value <= 0.0 {
340 out.push(Drift { bucket: bucket.into(), kind: DriftKind::Leak, ours: value, cloudflare: cf_cost, delta_percent: None });
341 }
342 out
343}
344
345/// When the last `days` in a row (each with enough cost to say something)
346/// were all under the floor: the first of them and the worst margin.
347/// Each item is a day's (day, revenue, cost).
348pub(crate) fn breach(series: &[(String, i64, i64)], floor_percent: f64, days: usize, min_cost_micros: i64) -> Option<(String, f64)> {
349 if days == 0 || series.len() < days {
350 return None;
351 }
352 let tail = &series[series.len() - days..];
353 let mut worst = f64::INFINITY;
354 for (_, revenue, cost) in tail {
355 if *cost < min_cost_micros {
356 return None;
357 }
358 let margin = margin_percent(*revenue, *cost).unwrap_or(-100.0);
359 if margin >= floor_percent {
360 return None;
361 }
362 worst = worst.min(margin);
363 }
364 Some((tail[0].0.clone(), worst))
365}
366
367/// `total` shared out in proportion to `weights`, in whole millionths that
368/// add up to it exactly (largest remainder first). Nothing to share, or no
369/// weight, shares nothing.
370pub(crate) fn attribute(total: i64, weights: &[(String, f64)]) -> Vec<(String, i64)> {
371 let mut merged: BTreeMap<String, f64> = BTreeMap::new();
372 for (key, w) in weights {
373 *merged.entry(key.clone()).or_default() += w.max(0.0);
374 }
375 let sum: f64 = merged.values().sum();
376 if total <= 0 || sum <= 0.0 {
377 return Vec::new();
378 }
379 let mut shares: Vec<(String, i64, f64)> = merged
380 .into_iter()
381 .map(|(key, w)| {
382 let exact = total as f64 * w / sum;
383 (key, exact.floor() as i64, exact - exact.floor())
384 })
385 .collect();
386 let mut left = total - shares.iter().map(|s| s.1).sum::<i64>();
387 let mut order: Vec<usize> = (0..shares.len()).collect();
388 order.sort_by(|a, b| shares[*b].2.total_cmp(&shares[*a].2).then(shares[*a].0.cmp(&shares[*b].0)));
389 for index in order {
390 if left <= 0 {
391 break;
392 }
393 shares[index].1 += 1;
394 left -= 1;
395 }
396 shares.into_iter().filter(|s| s.1 > 0).map(|(key, micros, _)| (key, micros)).collect()
397}
398
399/// Workspaces that cost g1t more than `factor` times what they paid, with
400/// at least `floor_micros` of cost: each (workspace, cost, revenue), the
401/// biggest gap first.
402pub(crate) fn anomalies(rows: &[(String, i64, i64)], factor: f64, floor_micros: i64) -> Vec<(String, i64, i64)> {
403 let mut out: Vec<(String, i64, i64)> = rows
404 .iter()
405 .filter(|(_, cost, revenue)| *cost >= floor_micros && *cost as f64 > *revenue as f64 * factor)
406 .cloned()
407 .collect();
408 out.sort_by(|a, b| (b.1 - b.2).cmp(&(a.1 - a.2)).then(a.0.cmp(&b.0)));
409 out
410}
411
412/// Cloudflare's marginal rate for one of its units: the median over the
413/// charged days of cost over quantity, in dollars. None while the included
414/// amounts still cover it. Each item is a day's (quantity, cost).
415pub(crate) fn billed_rate(days: &[(f64, f64)]) -> Option<f64> {
416 let mut rates: Vec<f64> = days.iter().filter(|(q, c)| *q > 0.0 && *c > 0.0).map(|(q, c)| c / q).collect();
417 if rates.is_empty() {
418 return None;
419 }
420 rates.sort_by(f64::total_cmp);
421 Some(rates[rates.len() / 2])
422}
423
424/// What one of g1t's units costs, from Cloudflare's rate per its own unit
425/// and how many of Cloudflare's units each of g1t's took: if Cloudflare
426/// counts three operations for every git operation g1t counts, a git
427/// operation costs three of Cloudflare's. None without enough of g1t's
428/// units to say.
429pub(crate) fn derived_unit_cost(rate_per_cf_unit: f64, cf_units: f64, own_units: f64) -> Option<f64> {
430 (own_units >= MIN_UNITS && cf_units > 0.0 && rate_per_cf_unit > 0.0).then(|| rate_per_cf_unit * cf_units / own_units)
431}
432
433/// How many units a price is per: `1,000 operations` → 1,000, `million
434/// requests` → 1,000,000, `second` → 1.
435pub(crate) fn unit_size(unit: &str) -> f64 {
436 let first = unit.split_whitespace().next().unwrap_or_default().replace(',', "");
437 match first.as_str() {
438 "million" => 1_000_000.0,
439 "thousand" => 1_000.0,
440 n => n.parse().unwrap_or(1.0),
441 }
442}
443
444fn day_before(day: &str, days: u64) -> String {
445 let ms = g1t_contracts::time::parse_rfc3339(&format!("{day}T00:00:00Z")).unwrap_or(0);
446 rfc3339(ms.saturating_sub(days * DAY_MS))[..10].to_owned()
447}
448
449/// Dollars to the cent from a dollar up, finer below: `$17.02`, `$0.063`.
450fn dollars(micros: i64) -> String {
451 if micros.abs() >= 1_000_000 {
452 let cents = (micros as f64 / 10_000.0).round() as i64;
453 format!("{}${}.{:02}", if cents < 0 { "-" } else { "" }, cents.abs() / 100, cents.abs() % 100)
454 } else {
455 crate::features::dollars(micros)
456 }
457}
458
459/// The days a plan payment is spread over.
460const PLAN_DAYS: u64 = 30;
461
462/// `micros` paid on `day` spread evenly over `days` days from it, in
463/// whole micros that add up to it (the first days take the remainder).
464pub(crate) fn spread(day: &str, micros: i64, days: u64) -> Vec<(String, i64)> {
465 if micros <= 0 || days == 0 {
466 return Vec::new();
467 }
468 let start = g1t_contracts::time::parse_rfc3339(&format!("{}T00:00:00Z", &day[..10.min(day.len())])).unwrap_or(0);
469 let each = micros / days as i64;
470 let rest = micros % days as i64;
471 (0..days)
472 .map(|n| (rfc3339(start + n * DAY_MS)[..10].to_owned(), each + i64::from((n as i64) < rest)))
473 .collect()
474}
475
476// ---------------------------------------------------------------------
477// The daily run, and what sudo reads.
478// ---------------------------------------------------------------------
479
480#[derive(Serialize)]
481struct Mail<'a> {
482 to: &'a str,
483 from: &'a str,
484 subject: &'a str,
485 text: String,
486 html: String,
487}
488
489fn escape(text: &str) -> String {
490 text.replace('&', "&amp;").replace('<', "&lt;").replace('>', "&gt;").replace('"', "&quot;")
491}
492
493/// Emails staff through Cloudflare Email Sending, the `EMAIL` binding.
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays494pub(crate) async fn email_staff(env: &Env, to: &str, subject: &str, lines: &[String]) -> Result<()> {
Git storage hardened, pages in tens of milliseconds, honest security alerts, and costs reconciled daily495 let link = "https://sudo.g1t.sh/costs";
496 let text = format!("{}\n\nCosts & margin: {link}\n\nSent by g1t-billing's margin guard (COSTS_ALERT_EMAIL).\n", lines.join("\n\n"));
497 let mut html = String::from("<div style=\"font-family:system-ui,sans-serif;max-width:560px;margin:0 auto;padding:24px 16px;color:#16150f\">");
498 for line in lines {
499 html.push_str(&format!("<p style=\"font-size:15px;line-height:1.6\">{}</p>", escape(line)));
500 }
501 html.push_str(&format!(
502 "<p><a href=\"{link}\">Open Costs &amp; margin in sudo</a></p><p style=\"font-size:13px;color:#6e6a5e\">Sent by g1t-billing's margin guard (COSTS_ALERT_EMAIL).</p></div>"
503 ));
504 let mail = Mail { to, from: "g1t <noreply@g1t.sh>", subject, text, html };
505 let binding = g1t_kit::js::binding(env, "EMAIL")?;
506 g1t_kit::js::call(&binding, "send", &[g1t_kit::js::to_js(&mail)?]).await?;
507 Ok(())
508}
509
510#[derive(Deserialize)]
511struct AlertRow {
512 id: String,
513 kind: String,
514 subject: String,
515 detail: String,
516 since: String,
517 opened_at: String,
518 emailed_at: Option<String>,
519}
520
521impl From<AlertRow> for MarginAlert {
522 fn from(r: AlertRow) -> Self {
523 MarginAlert { id: r.id, kind: r.kind, subject: r.subject, detail: r.detail, since: r.since, opened_at: r.opened_at, emailed_at: r.emailed_at }
524 }
525}
526
527#[derive(Deserialize)]
528struct MarginRow {
529 day: String,
530 bucket: String,
531 cf_cost_micros: i64,
532 own_cost_micros: i64,
533 value_micros: i64,
534 cash_micros: i64,
535 cf_quantity: f64,
536 own_quantity: f64,
537}
538
539impl From<MarginRow> for ProductDay {
540 fn from(r: MarginRow) -> Self {
541 ProductDay {
542 day: r.day,
543 bucket: r.bucket,
544 cf_cost_micros: r.cf_cost_micros,
545 own_cost_micros: r.own_cost_micros,
546 value_micros: r.value_micros,
547 cash_micros: r.cash_micros,
548 cf_quantity: r.cf_quantity,
549 own_quantity: r.own_quantity,
550 }
551 }
552}
553
554impl Billing {
555 /// The day's work: read Cloudflare's bill and g1t's own counts,
556 /// reconcile, look for drift, measure unit costs, apply prices whose
557 /// day has come, and raise or clear alerts.
558 pub(crate) async fn costs_daily(&self, env: &Env, keeper: &crate::keeper::Keeper) -> Result<CostsRun> {
559 let mut run = CostsRun::default();
560 let (since, until) = match self.read_cloudflare(keeper, &mut run.problems).await? {
561 Some((since, until, lines)) => {
562 run.lines = lines;
563 (since, until)
564 }
565 // Without the bill, still reconcile what g1t knows itself, over
566 // the same days the bill would be read for.
567 None => {
568 #[derive(Deserialize)]
569 struct Last {
570 day: Option<String>,
571 }
572 let last = self.db.prepare("SELECT MAX(day) AS day FROM margin_days").first::<Last>(None).await?.and_then(|l| l.day);
573 costs::window(last.as_deref(), now_ms())
574 }
575 };
576 if let Err(error) = self.count_own(&since, &until).await {
577 run.problems.push(format!("g1t's own counts could not be read: {error}"));
578 }
579 self.snapshot_pending(&until).await?;
580 run.days = self.reconcile_range(&since, &until).await?;
581 let drift = self.find_drift(&until).await?;
582 run.proposals = self.measure_units(&until).await?;
583 self.apply_due_versions().await?;
584 run.alerts = self.raise_alerts(env, &until, &drift).await?;
585 if let Some(identity) = &self.identity
586 && let Err(error) = self.tell_owners_of_rises(identity).await
587 {
588 run.problems.push(format!("owners could not be told of a price rise: {error}"));
589 }
590 for problem in &run.problems {
591 worker::console_warn!("costs: {problem}");
592 }
593 Ok(run)
594 }
595
596 /// What each month-end source had come to by the end of `day`.
597 async fn snapshot_pending(&self, day: &str) -> Result<()> {
598 self.db
599 .prepare(
600 "INSERT INTO pending_days (day, workspace, source, cost_micros, charge_micros)
601 SELECT ?1, workspace, source, COALESCE(cost_micros, 0), COALESCE(charge_micros, 0) FROM pending_usage WHERE month = ?2
602 ON CONFLICT (day, workspace, source) DO UPDATE SET cost_micros = excluded.cost_micros, charge_micros = excluded.charge_micros",
603 )
604 .bind(&[day.into(), day[..7].into()])?
605 .run()
606 .await?;
607 Ok(())
608 }
609
610 /// What customers were charged on the days, by workspace and key.
611 async fn usage_rows(&self, since: &str, until: &str) -> Result<Vec<UsageRow>> {
612 #[derive(Deserialize)]
613 struct Row {
614 day: String,
615 workspace: String,
616 key: String,
617 internal: i64,
618 own_provider: i64,
619 cash: Option<i64>,
620 drawn: Option<i64>,
621 cost: Option<i64>,
622 }
623 let charged_here = crate::storage::CHARGED_HERE.iter().map(|s| format!("'{s}'")).collect::<Vec<_>>().join(", ");
624 let end = format!("{until}T23:59:59.999Z");
625 let rows = self
626 .db
627 .prepare(format!(
628 "SELECT substr(created_at, 1, 10) AS day, workspace,
629 CASE WHEN task = 'deployments' AND reference LIKE 'deploy/%' THEN 'builds' ELSE COALESCE(task, 'other') END AS key,
630 CASE WHEN workspace IN ({internal}) THEN 1 ELSE 0 END AS internal,
631 CASE WHEN billed_to = 'workspace' THEN 1 ELSE 0 END AS own_provider,
632 -SUM(amount_micros) AS cash,
633 SUM(COALESCE(credit_micros, 0) + COALESCE(trial_micros, 0) + COALESCE(oss_micros, 0) + COALESCE(given_micros, 0)) AS drawn,
634 SUM(COALESCE(cost_micros, 0)) AS cost
635 FROM ledger
636 WHERE kind = 'usage' AND created_at >= ?1 AND created_at <= ?2 AND COALESCE(task, '') NOT IN ({charged_here})
637 GROUP BY 1, 2, 3, 4, 5",
638 internal = crate::sales::INTERNAL_SQL
639 ))
640 .bind(&[since.into(), end.as_str().into()])?
641 .all()
642 .await?
643 .results::<Row>()?;
644 let mut out: Vec<UsageRow> = rows
645 .into_iter()
646 .map(|r| {
647 // A workspace's own model provider was paid there: no cost
648 // to g1t. g1t's own workspaces are valued at price.
649 let cost = if r.own_provider == 1 { 0 } else { r.cost.unwrap_or(0) };
650 let cash = r.cash.unwrap_or(0);
651 let value = if r.internal == 1 { crate::credits::with_margin(cost, self.margin_percent) } else { cash + r.drawn.unwrap_or(0) };
652 UsageRow { day: r.day, workspace: r.workspace, key: r.key, value, cash, cost }
653 })
654 .collect();
655 // Month-end sources, from their daily snapshots.
656 #[derive(Deserialize)]
657 struct Snap {
658 day: String,
659 workspace: String,
660 source: String,
661 cost_micros: i64,
662 charge_micros: i64,
663 }
664 let snaps = self
665 .db
666 .prepare("SELECT day, workspace, source, cost_micros, charge_micros FROM pending_days WHERE day >= ?1 AND day <= ?2")
667 .bind(&[day_before(since, 1).into(), until.into()])?
668 .all()
669 .await?
670 .results::<Snap>()?
671 .into_iter()
672 .filter(|s| crate::storage::CHARGED_HERE.contains(&s.source.as_str()) || s.source == "domains")
673 .map(|s| (s.day, s.workspace, s.source, s.cost_micros, s.charge_micros))
674 .collect::<Vec<_>>();
675 out.extend(pending_deltas(&snaps).into_iter().filter(|u| u.day.as_str() >= since));
676 // The plan's price, spread over the 30 days it pays for, so a month's
677 // payment does not read as one very good day and 29 bad ones.
678 #[derive(Deserialize)]
679 struct Plan {
680 day: String,
681 workspace: String,
682 micros: Option<i64>,
683 }
684 let plans = self
685 .db
686 .prepare(
687 "SELECT substr(paid_at, 1, 10) AS day, workspace, SUM(amount_micros) AS micros FROM plan_payments
688 WHERE paid_at >= ?1 AND paid_at <= ?2 GROUP BY 1, 2",
689 )
690 .bind(&[day_before(since, PLAN_DAYS - 1).into(), end.as_str().into()])?
691 .all()
692 .await?
693 .results::<Plan>()?;
694 for p in plans {
695 for (day, micros) in spread(&p.day, p.micros.unwrap_or(0), PLAN_DAYS) {
696 if day.as_str() >= since && day.as_str() <= until {
697 out.push(UsageRow { day, workspace: p.workspace.clone(), key: "plan".into(), value: micros, cash: micros, cost: 0 });
698 }
699 }
700 }
701 Ok(out)
702 }
703
704 /// Reconciles the days and writes `margin_days` and `workspace_costs`.
705 async fn reconcile_range(&self, since: &str, until: &str) -> Result<u32> {
706 let rules = self.rules().await?;
707 #[derive(Deserialize)]
708 struct Map {
709 key: String,
710 bucket: String,
711 }
712 let revenue_map: BTreeMap<String, String> = self
713 .db
714 .prepare("SELECT key, bucket FROM revenue_map")
715 .all()
716 .await?
717 .results::<Map>()?
718 .into_iter()
719 .map(|m| (m.key, m.bucket))
720 .collect();
721 let lines = self
722 .db
723 .prepare("SELECT day, source, product, meter, quantity, cost_usd FROM cost_lines WHERE day >= ?1 AND day <= ?2")
724 .bind(&[since.into(), until.into()])?
725 .all()
726 .await?
727 .results::<LineRow>()?;
728 let own = self
729 .db
730 .prepare("SELECT day, meter, workspace, quantity FROM own_counts WHERE day >= ?1 AND day <= ?2")
731 .bind(&[since.into(), until.into()])?
732 .all()
733 .await?
734 .results::<OwnRow>()?;
735 let usage = self.usage_rows(since, until).await?;
736 let (days, workspaces) = fold(&rules, &revenue_map, &lines, &own, &usage);
737 let now = rfc3339(now_ms());
738 self.db
739 .batch(vec![
740 self.db.prepare("DELETE FROM margin_days WHERE day >= ?1 AND day <= ?2").bind(&[since.into(), until.into()])?,
741 self.db.prepare("DELETE FROM workspace_costs WHERE day >= ?1 AND day <= ?2").bind(&[since.into(), until.into()])?,
742 ])
743 .await?;
744 for chunk in days.chunks(50) {
745 let mut statements = Vec::with_capacity(chunk.len());
746 for d in chunk {
747 statements.push(
748 self.db
749 .prepare(
750 "INSERT OR REPLACE INTO margin_days (day, bucket, cf_cost_micros, own_cost_micros, value_micros, cash_micros, cf_quantity, own_quantity, computed_at)
751 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
752 )
753 .bind(&[
754 d.day.as_str().into(),
755 d.bucket.as_str().into(),
756 (d.cf_cost_micros as f64).into(),
757 (d.own_cost_micros as f64).into(),
758 (d.value_micros as f64).into(),
759 (d.cash_micros as f64).into(),
760 d.cf_quantity.into(),
761 d.own_quantity.into(),
762 now.as_str().into(),
763 ])?,
764 );
765 }
766 self.db.batch(statements).await?;
767 }
768 for chunk in workspaces.chunks(50) {
769 let mut statements = Vec::with_capacity(chunk.len());
770 for w in chunk {
771 statements.push(
772 self.db
773 .prepare("INSERT OR REPLACE INTO workspace_costs (day, workspace, bucket, cost_micros, revenue_micros) VALUES (?, ?, ?, ?, ?)")
774 .bind(&[
775 w.day.as_str().into(),
776 w.workspace.as_str().into(),
777 w.bucket.as_str().into(),
778 (w.cost as f64).into(),
779 (w.revenue as f64).into(),
780 ])?,
781 );
782 }
783 self.db.batch(statements).await?;
784 }
785 Ok(costs::days_between(since, until).len() as u32)
786 }
787
788 async fn margin_days(&self, since: &str, until: &str) -> Result<Vec<ProductDay>> {
789 Ok(self
790 .db
791 .prepare("SELECT * FROM margin_days WHERE day >= ?1 AND day <= ?2 ORDER BY day, bucket")
792 .bind(&[since.into(), until.into()])?
793 .all()
794 .await?
795 .results::<MarginRow>()?
796 .into_iter()
797 .map(ProductDay::from)
798 .collect())
799 }
800
801 /// Drift over the last week, written to `cost_drift` (replacing the
802 /// last run's), with unmapped Cloudflare meters as leaks.
803 async fn find_drift(&self, until: &str) -> Result<Vec<(Drift, String)>> {
804 let since = day_before(until, DRIFT_DAYS - 1);
805 let settings = self.cost_settings().await?;
806 let rules = self.rules().await?;
807 let days = self.margin_days(&since, until).await?;
808 let mut by: BTreeMap<String, Vec<ProductDay>> = BTreeMap::new();
809 for d in days {
810 by.entry(d.bucket.clone()).or_default().push(d);
811 }
812 let mut found = Vec::new();
813 for (bucket, days) in &by {
814 let bucket_rules: Vec<&Rule> = rules.iter().filter(|r| &r.bucket == bucket).collect();
815 let threshold = bucket_rules.iter().map(|r| r.drift_percent).fold(f64::INFINITY, f64::min);
816 let threshold = if threshold.is_finite() { threshold } else { 10.0 };
817 let counted = bucket_rules.iter().any(|r| r.own_meter.is_some());
818 for drift in drifts(bucket, days, threshold, counted, settings.min_daily_cost_micros) {
819 let title = costs::bucket_title(bucket);
820 let detail = match drift.kind {
821 DriftKind::Count => format!(
One operation mapping, owned by repos; billing reads it instead of keeping its own822 "{title}: g1t counted {}, Cloudflare {} over the last {DRIFT_DAYS} days ({:+.1}%). Customers are charged for what g1t counts; check what Cloudflare counts as a unit and change the repos service's operation_mapping (set_operation_mapping).",
Git storage hardened, pages in tens of milliseconds, honest security alerts, and costs reconciled daily823 crate::features::thousands(drift.ours.max(0.0).round() as u64),
824 crate::features::thousands(drift.cloudflare.max(0.0).round() as u64),
825 drift.delta_percent.unwrap_or(0.0)
826 ),
827 DriftKind::Cost => format!(
828 "{title}: Cloudflare charged {} over the last {DRIFT_DAYS} days; the price book's cost of the same usage is {} ({:+.1}%). A price may be stale: see the proposals.",
829 dollars(drift.cloudflare as i64),
830 dollars(drift.ours as i64),
831 drift.delta_percent.unwrap_or(0.0)
832 ),
833 DriftKind::Leak if bucket == UNMAPPED => {
834 format!("Cloudflare charged {} for meters no mapping claims. Map them on Costs & margin.", dollars(drift.cloudflare as i64))
835 }
836 DriftKind::Leak => format!(
837 "{title}: Cloudflare charged {} over the last {DRIFT_DAYS} days and customers were charged nothing for it.",
838 dollars(drift.cloudflare as i64)
839 ),
840 };
841 found.push((drift, detail));
842 }
843 }
844 let now = rfc3339(now_ms());
845 let mut statements = vec![self.db.prepare("DELETE FROM cost_drift")];
846 for (drift, detail) in &found {
847 statements.push(
848 self.db
849 .prepare("INSERT OR REPLACE INTO cost_drift (bucket, kind, ours, cloudflare, delta_percent, detail, found_at) VALUES (?, ?, ?, ?, ?, ?, ?)")
850 .bind(&[
851 drift.bucket.as_str().into(),
852 drift.kind.as_str().into(),
853 drift.ours.into(),
854 drift.cloudflare.into(),
855 drift.delta_percent.map_or(JsValue::NULL, JsValue::from),
856 detail.as_str().into(),
857 now.as_str().into(),
858 ])?,
859 );
860 }
861 self.db.batch(statements).await?;
862 Ok(found)
863 }
864
865 /// Unit costs from the bill for mappings that scale to g1t's own count
866 /// (git operations), proposed to the price book.
867 async fn measure_units(&self, until: &str) -> Result<u32> {
868 #[derive(Deserialize)]
869 struct Scaled {
870 product: String,
871 meter: String,
872 price_meter: String,
873 own_meter: String,
874 unit: Option<String>,
875 }
876 let scaled = self
877 .db
878 .prepare(
879 "SELECT m.product, m.meter, m.price_meter, m.own_meter, p.unit FROM cost_map m LEFT JOIN prices p ON p.meter = m.price_meter
880 WHERE m.scale_to_own = 1 AND m.price_meter IS NOT NULL AND m.own_meter IS NOT NULL",
881 )
882 .all()
883 .await?
884 .results::<Scaled>()?;
885 let since = day_before(until, MEASURE_DAYS - 1);
886 let rules = self.rules().await?;
887 let mut proposed = 0;
888 for s in scaled {
889 #[derive(Deserialize)]
890 struct Day {
891 product: String,
892 meter: String,
893 quantity: f64,
894 cost_usd: f64,
895 }
896 let lines = self
897 .db
898 .prepare("SELECT product, meter, quantity, cost_usd FROM cost_lines WHERE source = ?1 AND product = ?2 AND day >= ?3 AND day <= ?4")
899 .bind(&[SOURCE_BILLABLE.into(), s.product.as_str().into(), since.as_str().into(), until.into()])?
900 .all()
901 .await?
902 .results::<Day>()?;
903 // Only the lines this very mapping claims.
904 let mine: Vec<(f64, f64)> = lines
905 .iter()
906 .filter(|l| costs::classify(&rules, &l.product, &l.meter).is_some_and(|r| r.product == s.product && r.meter == s.meter))
907 .map(|l| (l.quantity, l.cost_usd))
908 .collect();
909 let Some(rate) = billed_rate(&mine) else { continue };
910 let cf_units: f64 = mine.iter().map(|(q, _)| q).sum();
911 #[derive(Deserialize)]
912 struct Own {
913 total: Option<f64>,
914 }
915 let own_units = self
916 .db
917 .prepare("SELECT SUM(quantity) AS total FROM own_counts WHERE meter = ?1 AND day >= ?2 AND day <= ?3")
918 .bind(&[s.own_meter.as_str().into(), since.as_str().into(), until.into()])?
919 .first::<Own>(None)
920 .await?
921 .and_then(|o| o.total)
922 .unwrap_or(0.0);
923 let Some(per_unit) = derived_unit_cost(rate, cf_units, own_units) else { continue };
924 let size = unit_size(s.unit.as_deref().unwrap_or("1"));
925 let measured = per_unit * size * 1_000_000.0;
926 let reason = format!(
927 "Cloudflare billed ${:.4} per 1,000 of its units and counted {:.2} of them for each one g1t counted over the last {MEASURE_DAYS} days ({} against {})",
928 rate * 1000.0,
929 cf_units / own_units,
930 crate::features::thousands(cf_units.round() as u64),
931 crate::features::thousands(own_units.round() as u64)
932 );
933 if self.propose(&s.price_meter, measured, &reason, "reconciler").await?.is_some() {
934 proposed += 1;
935 }
936 }
937 Ok(proposed)
938 }
939
940 /// Opens, updates and closes margin alerts, and emails staff about new
941 /// ones (and open ones each week).
942 async fn raise_alerts(&self, env: &Env, until: &str, drift: &[(Drift, String)]) -> Result<u32> {
943 let settings = self.cost_settings().await?;
944 let since = day_before(until, u64::from(settings.alert_days.max(1)) - 1);
945 let days = self.margin_days(&since, until).await?;
946 let mut conditions: Vec<(String, String, String, String)> = Vec::new();
947 // Each product under the floor.
948 let mut by: BTreeMap<String, Vec<(String, i64, i64)>> = BTreeMap::new();
949 let mut all: BTreeMap<String, (i64, i64)> = BTreeMap::new();
950 for d in &days {
951 let overall = all.entry(d.day.clone()).or_default();
952 overall.0 += d.cash_micros;
953 overall.1 += d.cost();
954 if !OVERHEAD.contains(&d.bucket.as_str()) && d.bucket != UNMAPPED {
955 by.entry(d.bucket.clone()).or_default().push((d.day.clone(), d.value_micros, d.cost()));
956 }
957 }
958 let floor = settings.margin_floor_percent;
959 let n = settings.alert_days as usize;
960 for (bucket, series) in &by {
961 if let Some((from, worst)) = breach(series, floor, n, settings.min_daily_cost_micros) {
962 conditions.push((
963 "margin".into(),
964 bucket.clone(),
965 format!("{}: margin under {floor:.0}% for {n} days running, as low as {worst:.1}%.", costs::bucket_title(bucket)),
966 from,
967 ));
968 }
969 }
970 let series: Vec<(String, i64, i64)> = all.into_iter().map(|(day, (revenue, cost))| (day, revenue, cost)).collect();
971 if let Some((from, worst)) = breach(&series, floor, n, settings.min_daily_cost_micros) {
972 conditions.push((
973 "overall".into(),
974 "g1t".into(),
975 format!("All of g1t: money in against Cloudflare's bill under {floor:.0}% for {n} days running, as low as {worst:.1}%."),
976 from,
977 ));
978 }
979 for (d, detail) in drift {
980 let kind = if d.kind == DriftKind::Leak { "leak" } else { "drift" };
981 conditions.push((kind.into(), format!("{}:{}", d.bucket, d.kind.as_str()), detail.clone(), until.to_owned()));
982 }
983 // Workspaces costing more than they pay.
984 for (workspace, cost, revenue) in self.workspace_anomalies(until, &settings).await? {
985 conditions.push((
986 "workspace".into(),
987 workspace.clone(),
988 format!("{workspace} cost g1t {} on Cloudflare over {ANOMALY_DAYS} days and paid {}.", dollars(cost), dollars(revenue)),
989 day_before(until, ANOMALY_DAYS - 1),
990 ));
991 }
992
993 let open = self
994 .db
995 .prepare("SELECT * FROM margin_alerts WHERE resolved_at IS NULL")
996 .all()
997 .await?
998 .results::<AlertRow>()?;
999 let now = now_ms();
1000 let stamp = rfc3339(now);
1001 let mut to_email: Vec<String> = Vec::new();
1002 let mut kept: BTreeSet<String> = BTreeSet::new();
1003 for (kind, subject, detail, from) in &conditions {
1004 match open.iter().find(|a| &a.kind == kind && &a.subject == subject) {
1005 Some(alert) => {
1006 kept.insert(alert.id.clone());
1007 self.db
1008 .prepare("UPDATE margin_alerts SET detail = ? WHERE id = ?")
1009 .bind(&[detail.as_str().into(), alert.id.as_str().into()])?
1010 .run()
1011 .await?;
1012 let stale = alert
1013 .emailed_at
1014 .as_deref()
1015 .and_then(g1t_contracts::time::parse_rfc3339)
1016 .is_none_or(|at| now.saturating_sub(at) >= REMIND_MS);
1017 if stale && kind != "workspace" {
1018 to_email.push(format!("Still open: {detail}"));
1019 kept.insert(format!("email:{}", alert.id));
1020 }
1021 }
1022 None => {
1023 let id = new_id("mal", now);
1024 self.db
1025 .prepare("INSERT INTO margin_alerts (id, kind, subject, detail, since, opened_at) VALUES (?, ?, ?, ?, ?, ?)")
1026 .bind(&[id.as_str().into(), kind.as_str().into(), subject.as_str().into(), detail.as_str().into(), from.as_str().into(), stamp.as_str().into()])?
1027 .run()
1028 .await?;
1029 kept.insert(id.clone());
1030 to_email.push(detail.clone());
1031 kept.insert(format!("email:{id}"));
1032 }
1033 }
1034 }
1035 for alert in &open {
1036 if !kept.contains(&alert.id) {
1037 self.db
1038 .prepare("UPDATE margin_alerts SET resolved_at = ? WHERE id = ?")
1039 .bind(&[stamp.as_str().into(), alert.id.as_str().into()])?
1040 .run()
1041 .await?;
1042 }
1043 }
1044 let to = env.var("COSTS_ALERT_EMAIL").map(|v| v.to_string()).unwrap_or_default();
1045 if !to_email.is_empty() && !to.trim().is_empty() {
1046 let subject = format!("[g1t costs] {} margin alert{}", to_email.len(), if to_email.len() == 1 { "" } else { "s" });
1047 match email_staff(env, to.trim(), &subject, &to_email).await {
1048 Ok(()) => {
1049 for marker in kept.iter().filter_map(|k| k.strip_prefix("email:")) {
1050 self.db
1051 .prepare("UPDATE margin_alerts SET emailed_at = ? WHERE id = ?")
1052 .bind(&[stamp.as_str().into(), marker.into()])?
1053 .run()
1054 .await?;
1055 }
1056 }
1057 Err(error) => worker::console_error!("could not email the margin alerts: {error}"),
1058 }
1059 }
1060 Ok(conditions.len() as u32)
1061 }
1062
1063 /// Workspaces costing g1t more than they pay over 30 days, not g1t's own.
1064 async fn workspace_anomalies(&self, until: &str, settings: &CostSettings) -> Result<Vec<(String, i64, i64)>> {
1065 #[derive(Deserialize)]
1066 struct Row {
1067 workspace: String,
1068 cost: Option<i64>,
1069 revenue: Option<i64>,
1070 }
1071 let rows = self
1072 .db
1073 .prepare(format!(
1074 "SELECT workspace, SUM(cost_micros) AS cost, SUM(revenue_micros) AS revenue FROM workspace_costs
1075 WHERE day >= ?1 AND day <= ?2 AND workspace NOT IN ({}) GROUP BY workspace",
1076 crate::sales::INTERNAL_SQL
1077 ))
1078 .bind(&[day_before(until, ANOMALY_DAYS - 1).into(), until.into()])?
1079 .all()
1080 .await?
1081 .results::<Row>()?;
1082 let rows: Vec<(String, i64, i64)> = rows.into_iter().map(|r| (r.workspace, r.cost.unwrap_or(0), r.revenue.unwrap_or(0))).collect();
1083 Ok(anomalies(&rows, settings.anomaly_factor, settings.anomaly_floor_micros))
1084 }
1085
1086 /// For Reach out: workspaces with an open cost-over-revenue alert,
1087 /// each with its detail and cost.
1088 pub(crate) async fn costing_more_than_they_pay(&self) -> Result<Vec<(String, String, i64)>> {
1089 let alerts = self
1090 .db
1091 .prepare("SELECT * FROM margin_alerts WHERE resolved_at IS NULL AND kind = 'workspace' ORDER BY opened_at DESC LIMIT 50")
1092 .all()
1093 .await?
1094 .results::<AlertRow>()?;
1095 let mut out = Vec::new();
1096 for alert in alerts {
1097 #[derive(Deserialize)]
1098 struct Cost {
1099 cost: Option<i64>,
1100 }
1101 let cost = self
1102 .db
1103 .prepare("SELECT SUM(cost_micros) AS cost FROM workspace_costs WHERE workspace = ? AND day >= ?")
1104 .bind(&[alert.subject.as_str().into(), alert.since.as_str().into()])?
1105 .first::<Cost>(None)
1106 .await?
1107 .and_then(|c| c.cost)
1108 .unwrap_or(0);
1109 out.push((alert.subject, alert.detail, cost));
1110 }
1111 Ok(out)
1112 }
1113
1114 /// `admin_cost_alerts`: what sudo's banner says.
1115 pub(crate) async fn admin_cost_alerts(&self, _: AdminCostAlertsArgs) -> Result<Vec<MarginAlert>> {
1116 Ok(self
1117 .db
1118 .prepare("SELECT * FROM margin_alerts WHERE resolved_at IS NULL ORDER BY opened_at DESC LIMIT 50")
1119 .all()
1120 .await?
1121 .results::<AlertRow>()?
1122 .into_iter()
1123 .map(MarginAlert::from)
1124 .collect())
1125 }
1126
1127 /// `admin_run_costs`: the daily run, now.
1128 pub(crate) async fn admin_run_costs(&self, env: &Env, a: AdminRunCostsArgs) -> Result<Outcome<CostsRun>> {
1129 let keeper = crate::keeper::Keeper::from_env(env);
1130 let run = self.costs_daily(env, &keeper).await?;
1131 if !a.by.is_empty() {
1132 self.audit(
1133 "costs",
1134 "costs_run",
1135 &format!("{} lines, {} days, {} proposals, {} alerts", run.lines, run.days, run.proposals, run.alerts),
1136 &a.by,
1137 )
1138 .await?;
1139 }
1140 Ok(Outcome::Ok(run))
1141 }
1142
1143 /// `admin_set_cost_mapping`.
1144 pub(crate) async fn admin_set_cost_mapping(&self, a: AdminSetCostMappingArgs) -> Result<Outcome<CostMapping>> {
1145 let product = costs::slug(&a.product);
1146 let meter = if a.meter.trim() == "*" { "*".to_owned() } else { costs::slug(&a.meter) };
1147 if product.is_empty() || meter.is_empty() {
1148 return Ok(Outcome::fail(FailureCode::Invalid, "Name Cloudflare's product and a meter (or * for all of it)."));
1149 }
1150 let now = rfc3339(now_ms());
1151 if a.remove {
1152 self.db
1153 .prepare("DELETE FROM cost_map WHERE product = ? AND meter = ?")
1154 .bind(&[product.as_str().into(), meter.as_str().into()])?
1155 .run()
1156 .await?;
1157 self.audit("costs", "cost_mapping_removed", &format!("{product}/{meter}"), &a.by).await?;
1158 return Ok(Outcome::Ok(CostMapping {
1159 product,
1160 meter,
1161 bucket: String::new(),
1162 price_meter: None,
1163 own_meter: None,
1164 scale_to_own: false,
1165 drift_percent: 0.0,
1166 note: String::new(),
1167 updated_at: now,
1168 updated_by: a.by,
1169 }));
1170 }
1171 let bucket = costs::slug(&a.bucket);
1172 if bucket.is_empty() {
1173 return Ok(Outcome::fail(FailureCode::Invalid, "Say which of g1t's products it is a cost of."));
1174 }
1175 let clean = |v: Option<String>| v.map(|v| v.trim().to_owned()).filter(|v| !v.is_empty());
1176 let (price_meter, own_meter) = (clean(a.price_meter), clean(a.own_meter));
1177 let drift = a.drift_percent.filter(|d| d.is_finite() && *d > 0.0).unwrap_or(10.0);
1178 self.db
1179 .prepare(
1180 "INSERT INTO cost_map (product, meter, bucket, price_meter, own_meter, scale_to_own, drift_percent, note, updated_at, updated_by)
1181 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)
1182 ON CONFLICT (product, meter) DO UPDATE SET bucket = ?3, price_meter = ?4, own_meter = ?5, scale_to_own = ?6,
1183 drift_percent = ?7, note = ?8, updated_at = ?9, updated_by = ?10",
1184 )
1185 .bind(&[
1186 product.as_str().into(),
1187 meter.as_str().into(),
1188 bucket.as_str().into(),
1189 crate::optional(price_meter.as_deref()),
1190 crate::optional(own_meter.as_deref()),
1191 i32::from(a.scale_to_own).into(),
1192 drift.into(),
1193 a.note.trim().into(),
1194 now.as_str().into(),
1195 a.by.as_str().into(),
1196 ])?
1197 .run()
1198 .await?;
1199 self.audit("costs", "cost_mapping", &format!("{product}/{meter} → {bucket}"), &a.by).await?;
1200 Ok(Outcome::Ok(CostMapping {
1201 product,
1202 meter,
1203 bucket,
1204 price_meter,
1205 own_meter,
1206 scale_to_own: a.scale_to_own,
1207 drift_percent: drift,
1208 note: a.note.trim().to_owned(),
1209 updated_at: now,
1210 updated_by: a.by,
1211 }))
1212 }
1213
1214 /// `admin_costs`: the Costs & margin page.
1215 pub(crate) async fn admin_costs(&self, a: AdminCostsArgs, configured: bool) -> Result<CostsReport> {
1216 let until = rfc3339(now_ms())[..10].to_owned();
1217 let span = u64::from(a.days.unwrap_or(30).clamp(7, 90));
1218 let since = day_before(&until, span - 1);
1219 let days = self.margin_days(&since, &until).await?;
1220 let rules = self.rules().await?;
1221
1222 let mut products: BTreeMap<String, ProductMargin> = BTreeMap::new();
1223 let mut overall = OverallMargin::default();
1224 for d in &days {
1225 let p = products.entry(d.bucket.clone()).or_insert_with(|| ProductMargin {
1226 bucket: d.bucket.clone(),
1227 title: costs::bucket_title(&d.bucket),
1228 cost_source: if NOT_CLOUDFLARE.contains(&d.bucket.as_str()) { "ledger" } else { "cloudflare" }.into(),
1229 overhead: OVERHEAD.contains(&d.bucket.as_str()),
1230 ..ProductMargin::default()
1231 });
1232 p.cf_cost_micros += d.cf_cost_micros;
1233 p.own_cost_micros += d.own_cost_micros;
1234 p.value_micros += d.value_micros;
1235 p.cost_micros += d.cost();
1236 overall.cost_micros += d.cost();
1237 if d.bucket == "platform" {
1238 overall.plans_micros += d.cash_micros;
1239 } else {
1240 overall.usage_micros += d.cash_micros;
1241 }
1242 }
1243 for p in products.values_mut() {
1244 p.margin_micros = p.value_micros - p.cost_micros;
1245 p.margin_percent = margin_percent(p.value_micros, p.cost_micros);
1246 }
1247 let revenue = overall.usage_micros + overall.plans_micros;
1248 overall.margin_micros = revenue - overall.cost_micros;
1249 overall.margin_percent = margin_percent(revenue, overall.cost_micros);
1250 let mut products: Vec<ProductMargin> = products.into_values().collect();
1251 products.sort_by_key(|p| std::cmp::Reverse(p.cost_micros.max(p.value_micros)));
1252
1253 #[derive(Deserialize)]
1254 struct DriftRow {
1255 bucket: String,
1256 kind: String,
1257 ours: f64,
1258 cloudflare: f64,
1259 delta_percent: Option<f64>,
1260 detail: String,
1261 found_at: String,
1262 }
1263 let drift = self
1264 .db
1265 .prepare("SELECT * FROM cost_drift ORDER BY kind, bucket")
1266 .all()
1267 .await?
1268 .results::<DriftRow>()?
1269 .into_iter()
1270 .map(|r| CostDrift {
1271 title: costs::bucket_title(&r.bucket),
1272 bucket: r.bucket,
1273 kind: r.kind,
1274 ours: r.ours,
1275 cloudflare: r.cloudflare,
1276 delta_percent: r.delta_percent,
1277 detail: r.detail,
1278 found_at: r.found_at,
1279 })
1280 .collect();
1281
1282 #[derive(Deserialize)]
1283 struct Top {
1284 workspace: String,
1285 cost: Option<i64>,
1286 revenue: Option<i64>,
1287 internal: i64,
1288 }
1289 let top_workspaces = self
1290 .db
1291 .prepare(format!(
1292 "SELECT workspace, SUM(cost_micros) AS cost, SUM(revenue_micros) AS revenue,
1293 CASE WHEN workspace IN ({}) THEN 1 ELSE 0 END AS internal
1294 FROM workspace_costs WHERE day >= ?1 AND day <= ?2 GROUP BY workspace ORDER BY cost DESC LIMIT 15",
1295 crate::sales::INTERNAL_SQL
1296 ))
1297 .bind(&[since.as_str().into(), until.as_str().into()])?
1298 .all()
1299 .await?
1300 .results::<Top>()?
1301 .into_iter()
1302 .map(|t| WorkspaceCost { workspace: t.workspace, cost_micros: t.cost.unwrap_or(0), revenue_micros: t.revenue.unwrap_or(0), internal: t.internal == 1 })
1303 .collect();
1304
1305 #[derive(Deserialize)]
1306 struct Summary {
1307 source: String,
1308 product: String,
1309 meter: String,
1310 raw_name: String,
1311 unit: String,
1312 quantity: f64,
1313 cost_usd: f64,
1314 }
1315 let lines = self
1316 .db
1317 .prepare(
1318 "SELECT source, product, meter, MAX(raw_name) AS raw_name, MAX(unit) AS unit, SUM(quantity) AS quantity, SUM(cost_usd) AS cost_usd
1319 FROM cost_lines WHERE day >= ?1 AND day <= ?2 GROUP BY source, product, meter ORDER BY cost_usd DESC, product, meter LIMIT 200",
1320 )
1321 .bind(&[since.as_str().into(), until.as_str().into()])?
1322 .all()
1323 .await?
1324 .results::<Summary>()?
1325 .into_iter()
1326 .map(|l| CostLineSummary {
1327 bucket: costs::classify(&rules, &l.product, &l.meter).map(|r| r.bucket.clone()),
1328 product: l.product,
1329 meter: l.meter,
1330 raw_name: l.raw_name,
1331 unit: l.unit,
1332 source: l.source,
1333 quantity: l.quantity,
1334 cost_micros: micros(l.cost_usd),
1335 })
1336 .collect();
1337
1338 #[derive(Deserialize)]
1339 struct MapRow {
1340 product: String,
1341 meter: String,
1342 bucket: String,
1343 price_meter: Option<String>,
1344 own_meter: Option<String>,
1345 scale_to_own: i64,
1346 drift_percent: f64,
1347 note: String,
1348 updated_at: String,
1349 updated_by: String,
1350 }
1351 let mappings = self
1352 .db
1353 .prepare("SELECT * FROM cost_map ORDER BY product, meter")
1354 .all()
1355 .await?
1356 .results::<MapRow>()?
1357 .into_iter()
1358 .map(|m| CostMapping {
1359 product: m.product,
1360 meter: m.meter,
1361 bucket: m.bucket,
1362 price_meter: m.price_meter,
1363 own_meter: m.own_meter,
1364 scale_to_own: m.scale_to_own == 1,
1365 drift_percent: m.drift_percent,
1366 note: m.note,
1367 updated_at: m.updated_at,
1368 updated_by: m.updated_by,
1369 })
1370 .collect();
1371
1372 #[derive(Deserialize)]
1373 struct Fetched {
1374 at: Option<String>,
1375 }
1376 let fetched_at = self.db.prepare("SELECT MAX(fetched_at) AS at FROM cost_lines").first::<Fetched>(None).await?.and_then(|f| f.at);
1377
1378 Ok(CostsReport {
1379 configured,
1380 fetched_at,
1381 days: days
1382 .iter()
1383 .map(|d| CostDay {
1384 day: d.day.clone(),
1385 bucket: d.bucket.clone(),
1386 cf_cost_micros: d.cf_cost_micros,
1387 own_cost_micros: d.own_cost_micros,
1388 value_micros: d.value_micros,
1389 cash_micros: d.cash_micros,
1390 })
1391 .collect(),
1392 since,
1393 until,
1394 products,
1395 overall,
1396 drift,
1397 alerts: self.admin_cost_alerts(AdminCostAlertsArgs {}).await?,
1398 proposals: self.proposals().await?,
1399 versions: self.versions().await?,
1400 top_workspaces,
1401 lines,
1402 mappings,
1403 settings: self.cost_settings().await?,
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays1404 caps: self.spend_caps().await?,
Git storage hardened, pages in tens of milliseconds, honest security alerts, and costs reconciled daily1405 })
1406 }
1407}
1408
1409#[cfg(test)]
1410mod tests {
1411 use super::*;
1412
1413 fn rule(product: &str, meter: &str, bucket: &str, own: Option<&str>) -> Rule {
1414 Rule { product: product.into(), meter: meter.into(), bucket: bucket.into(), price_meter: None, own_meter: own.map(Into::into), drift_percent: 10.0 }
1415 }
1416
1417 fn rules() -> Vec<Rule> {
1418 vec![
1419 rule("containers", "*", "sandboxes", None),
1420 rule("workers", "*", "platform", None),
1421 rule("artifacts", "*", "git", Some("git_operations")),
1422 rule("artifacts", "events_", "git", Some("git_operations")),
1423 ]
1424 }
1425
1426 fn revenue_map() -> BTreeMap<String, String> {
1427 [("sandbox", "sandboxes"), ("git", "git"), ("plan", "platform")].iter().map(|(k, v)| (k.to_string(), v.to_string())).collect()
1428 }
1429
1430 fn line(day: &str, source: &str, product: &str, meter: &str, quantity: f64, cost: f64) -> LineRow {
1431 LineRow { day: day.into(), source: source.into(), product: product.into(), meter: meter.into(), quantity, cost_usd: cost }
1432 }
1433
1434 fn usage(day: &str, workspace: &str, key: &str, value: i64, cash: i64, cost: i64) -> UsageRow {
1435 UsageRow { day: day.into(), workspace: workspace.into(), key: key.into(), value, cash, cost }
1436 }
1437
1438 #[test]
1439 fn a_day_puts_the_bill_g1ts_counts_and_charges_side_by_side() {
1440 let lines = vec![
1441 line("2026-10-15", SOURCE_BILLABLE, "containers", "container_memory", 1000.0, 2.00),
1442 line("2026-10-15", SOURCE_BILLABLE, "artifacts", "artifacts_operations", 30_000.0, 3.00),
1443 // Artifacts' own events: not used while the bill has a count.
1444 line("2026-10-15", SOURCE_ARTIFACTS, "artifacts", "events_pull", 29_000.0, 0.0),
1445 line("2026-10-15", SOURCE_BILLABLE, "workers", "workers_cpu_ms", 1.0, 0.50),
1446 line("2026-10-15", SOURCE_BILLABLE, "browser_rendering", "browser_hours", 2.0, 0.25),
1447 ];
1448 let own = vec![
1449 OwnRow { day: "2026-10-15".into(), meter: "git_operations".into(), workspace: "acme".into(), quantity: 7_500.0 },
1450 OwnRow { day: "2026-10-15".into(), meter: "git_operations".into(), workspace: "beta".into(), quantity: 2_500.0 },
1451 ];
1452 let usage = vec![
1453 usage("2026-10-15", "acme", "sandbox", 2_400_000, 1_000_000, 2_000_000),
1454 usage("2026-10-15", "beta", "sandbox", 1_200_000, 1_200_000, 1_000_000),
1455 usage("2026-10-15", "acme", "git", 600_000, 600_000, 500_000),
1456 usage("2026-10-15", "acme", "implement", 120_000, 120_000, 100_000),
1457 usage("2026-10-15", "beta", "plan", 20_000_000, 20_000_000, 0),
1458 ];
1459 let (days, workspaces) = fold(&rules(), &revenue_map(), &lines, &own, &usage);
1460 let get = |bucket: &str| days.iter().find(|d| d.bucket == bucket).unwrap();
1461 let sandboxes = get("sandboxes");
1462 assert_eq!((sandboxes.cf_cost_micros, sandboxes.own_cost_micros, sandboxes.value_micros, sandboxes.cash_micros), (2_000_000, 3_000_000, 3_600_000, 2_200_000));
1463 let git = get("git");
1464 assert_eq!(git.cf_cost_micros, 3_000_000);
1465 assert_eq!((git.cf_quantity, git.own_quantity), (30_000.0, 10_000.0));
1466 assert_eq!(get("platform").value_micros, 20_000_000);
1467 // Not mapped: a leak until someone maps it.
1468 assert_eq!(get(UNMAPPED).cf_cost_micros, 250_000);
1469 // Models: no Cloudflare line, their cost is g1t's own.
1470 assert_eq!(get("models").cost(), 100_000);
1471 // Git's cost shared by g1t's own counts (Cloudflare gave none per
1472 // workspace here): three quarters to acme.
1473 let share = |ws: &str, bucket: &str| workspaces.iter().find(|w| w.workspace == ws && w.bucket == bucket).map(|w| (w.cost, w.revenue));
1474 assert_eq!(share("acme", "git"), Some((2_250_000, 600_000)));
1475 assert_eq!(share("beta", "git"), Some((750_000, 0)));
1476 // Every bucket's cost is shared out exactly.
1477 for d in &days {
1478 let shared: i64 = workspaces.iter().filter(|w| w.bucket == d.bucket).map(|w| w.cost).sum();
1479 assert_eq!(shared, d.cost(), "{}", d.bucket);
1480 }
1481 }
1482
1483 #[test]
1484 fn artifacts_events_count_when_the_bill_does_not() {
1485 let lines = vec![
1486 line("2026-10-05", SOURCE_ARTIFACTS, "artifacts", "events_pull", 120.0, 0.0),
1487 line("2026-10-05", SOURCE_ARTIFACTS, "artifacts", "events_push", 30.0, 0.0),
1488 line("2026-10-05", SOURCE_ARTIFACTS, "artifacts", "events_ratelimited", 9.0, 0.0),
1489 ];
1490 let (days, _) = fold(&rules(), &revenue_map(), &lines, &[], &[]);
1491 assert_eq!(days[0].cf_quantity, 150.0);
1492 assert_eq!(days[0].cf_cost_micros, 0);
1493 }
1494
1495 #[test]
1496 fn month_end_meters_are_told_by_the_day_from_snapshots() {
1497 let snap = |day: &str, cost: i64, charge: i64| (day.to_string(), "acme".to_string(), "git".to_string(), cost, charge);
1498 let rows = pending_deltas(&[snap("2026-10-30", 100, 120), snap("2026-10-31", 250, 300), snap("2026-11-01", 40, 48), snap("2026-11-02", 40, 48)]);
1499 assert_eq!(
1500 rows.iter().map(|r| (r.day.as_str(), r.cost, r.value)).collect::<Vec<_>>(),
1501 vec![("2026-10-30", 100, 120), ("2026-10-31", 150, 180), ("2026-11-01", 40, 48)]
1502 );
1503 }
1504
1505 #[test]
1506 fn a_plan_payment_is_spread_over_the_month_it_pays_for() {
1507 let days = spread("2026-10-01T00:00:00.000Z", 20_000_000, 30);
1508 assert_eq!(days.len(), 30);
1509 assert_eq!(days[0], ("2026-10-01".to_string(), 666_667));
1510 assert_eq!(days[29], ("2026-10-30".to_string(), 666_666));
1511 assert_eq!(days.iter().map(|d| d.1).sum::<i64>(), 20_000_000);
1512 assert!(spread("2026-10-01", 0, 30).is_empty());
1513 assert_eq!(dollars(17_024_000), "$17.02");
1514 assert_eq!(dollars(-27_668_620), "-$27.67");
1515 assert_eq!(dollars(63_000), "$0.063");
1516 }
1517
1518 #[test]
1519 fn margins_and_deltas() {
1520 assert_eq!(margin_percent(1_200_000, 1_000_000).map(|m| (m * 100.0).round() / 100.0), Some(16.67));
1521 assert_eq!(margin_percent(0, 5), None);
1522 assert_eq!(delta_percent(110.0, 100.0), Some(10.0));
1523 assert_eq!(delta_percent(1.0, 0.0), None);
1524 }
1525
1526 fn day(bucket: &str, cf: i64, own: i64, value: i64, cfq: f64, ownq: f64) -> ProductDay {
1527 ProductDay { day: "2026-10-15".into(), bucket: bucket.into(), cf_cost_micros: cf, own_cost_micros: own, value_micros: value, cash_micros: value, cf_quantity: cfq, own_quantity: ownq }
1528 }
1529
1530 #[test]
1531 fn counts_more_than_the_threshold_apart_are_drift() {
1532 // Cloudflare counted 30,000 operations where g1t counted 10,000:
1533 // binding reads, perhaps. -66.7%.
1534 let drift = drifts("git", &[day("git", 3_000_000, 1_500_000, 1_800_000, 30_000.0, 10_000.0)], 10.0, true, 100_000);
1535 assert_eq!(drift.iter().map(|d| d.kind).collect::<Vec<_>>(), vec![DriftKind::Count, DriftKind::Cost]);
1536 assert!((drift[0].delta_percent.unwrap() + 66.666).abs() < 0.01);
1537 // 9% apart: within 10%.
1538 assert!(drifts("git", &[day("git", 1_000_000, 1_000_000, 1_200_000, 10_000.0, 10_900.0)], 10.0, true, 100_000).is_empty());
1539 // Uncounted products have no count drift.
1540 assert!(drifts("sandboxes", &[day("sandboxes", 1_000_000, 1_050_000, 1_200_000, 5.0, 0.0)], 10.0, false, 100_000).is_empty());
1541 }
1542
1543 #[test]
1544 fn cost_with_no_revenue_is_a_leak_but_not_for_running_g1t() {
1545 let leak = drifts("actions_cache", &[day("actions_cache", 400_000, 0, 0, 0.0, 0.0)], 10.0, false, 100_000);
1546 assert_eq!(leak.len(), 1);
1547 assert_eq!(leak[0].kind, DriftKind::Leak);
1548 assert!(drifts("platform", &[day("platform", 5_000_000, 0, 0, 0.0, 0.0)], 10.0, false, 100_000).is_empty());
1549 // Pennies say nothing.
1550 assert!(drifts("actions_cache", &[day("actions_cache", 50_000, 0, 0, 0.0, 0.0)], 10.0, false, 100_000).is_empty());
1551 assert!(drifts(UNMAPPED, &[day(UNMAPPED, 250_000, 0, 0, 0.0, 0.0)], 10.0, false, 100_000)[0].kind == DriftKind::Leak);
1552 }
1553
1554 #[test]
1555 fn a_margin_alert_needs_n_days_in_a_row_under_the_floor() {
1556 let s = |d: &str, revenue: i64, cost: i64| (d.to_string(), revenue, cost);
1557 // 5%, 0%, -20%: three days under 10%.
1558 let series = vec![s("10-13", 1_200_000, 1_000_000), s("10-14", 1_050_000, 1_000_000), s("10-15", 1_000_000, 1_000_000), s("10-16", 1_000_000, 1_200_000)];
1559 let (from, worst) = breach(&series, 10.0, 3, 100_000).unwrap();
1560 assert_eq!(from, "10-14");
1561 assert!((worst + 20.0).abs() < 1e-9);
1562 // A good day in the window clears it.
1563 let mended = vec![s("10-14", 1_050_000, 1_000_000), s("10-15", 1_300_000, 1_000_000), s("10-16", 1_000_000, 1_200_000)];
1564 assert!(breach(&mended, 10.0, 3, 100_000).is_none());
1565 // Cost with no revenue at all is the worst margin there is.
1566 assert_eq!(breach(&[s("10-16", 0, 500_000)], 10.0, 1, 100_000).unwrap().1, -100.0);
1567 // Too little cost to judge.
1568 assert!(breach(&[s("10-16", 0, 5_000)], 10.0, 1, 100_000).is_none());
1569 assert!(breach(&series, 10.0, 9, 100_000).is_none());
1570 }
1571
1572 #[test]
1573 fn shared_costs_add_up_to_the_bill() {
1574 let w = |k: &str, v: f64| (k.to_string(), v);
1575 assert_eq!(attribute(100, &[w("a", 1.0), w("b", 1.0), w("c", 1.0)]), vec![("a".into(), 34), ("b".into(), 33), ("c".into(), 33)]);
1576 assert_eq!(attribute(10, &[w("a", 3.0), w("b", 1.0), w("a", 0.0)]), vec![("a".into(), 8), ("b".into(), 2)]);
1577 assert!(attribute(10, &[w("a", 0.0)]).is_empty());
1578 assert!(attribute(0, &[w("a", 1.0)]).is_empty());
1579 }
1580
1581 #[test]
1582 fn a_workspace_that_costs_more_than_it_pays_is_flagged() {
1583 let rows = vec![("acme".to_string(), 5_000_000, 1_000_000), ("beta".to_string(), 900_000, 0), ("gamma".to_string(), 2_000_000, 3_000_000)];
1584 let found = anomalies(&rows, 1.0, 1_000_000);
1585 assert_eq!(found, vec![("acme".to_string(), 5_000_000, 1_000_000)]);
1586 // At twice its revenue as the threshold, $5 against $3 is fine.
1587 assert!(anomalies(&[("acme".to_string(), 5_000_000, 3_000_000)], 2.0, 1_000_000).is_empty());
1588 }
1589
1590 #[test]
1591 fn a_git_operation_costs_what_cloudflare_counts_for_it() {
1592 // $0.15 per 1,000 of Cloudflare's operations, on the charged days.
1593 let rate = billed_rate(&[(10_000.0, 0.0), (20_000.0, 3.0), (30_000.0, 4.5), (5_000.0, 0.75)]).unwrap();
1594 assert!((rate - 0.000_15).abs() < 1e-12);
1595 // Cloudflare counted 3 for every 1 g1t did: binding reads count.
1596 let per_op = derived_unit_cost(rate, 300_000.0, 100_000.0).unwrap();
1597 let per_thousand_micros = per_op * unit_size("1,000 operations") * 1e6;
1598 assert!((per_thousand_micros - 450_000.0).abs() < 1e-6, "{per_thousand_micros}");
1599 // Too few of g1t's units to say.
1600 assert!(derived_unit_cost(rate, 3_000.0, 500.0).is_none());
1601 assert!(billed_rate(&[(10_000.0, 0.0)]).is_none());
1602 assert_eq!(unit_size("million requests"), 1e6);
1603 assert_eq!(unit_size("second"), 1.0);
1604 }
1605}