Skip to content
Sign up

flagon-io/g1t

public
0
887 linesCodeBlameRaw
1//! The price book as versions, and how it changes.
2//!
3//! A price is its cost plus the markup, and costs move: Cloudflare's
4//! rates, how much CPU sandboxes use, what Artifacts turns out to count as
5//! an operation. So prices must move too, without surprising anyone and
6//! without g1t selling at a loss.
7//!
8//! - **Versions.** Every price is a row in `price_versions`, never changed
9//! once written. `prices` holds the version in force; a version waiting
10//! for its date is applied by the daily run, and recorded in
11//! `price_changes` (the public record on the pricing page). A charge
12//! records the version it was made at (`ledger.price_version`), so what a
13//! past statement says is always explained by the prices of then.
14//! - **Proposals.** The keeper and the reconciler measure costs; what they
15//! find is proposed (`price_proposals`). A move under 2% is noise. One
16//! within the guardrail (`auto_apply_percent`, 25%) is applied on its own
17//! when `auto_apply` is on. Anything larger, or more than four times off
18//! (suspect), waits for staff to approve or reject in sudo.
19//! - **Notice.** A fall applies at once: customers only gain. A rise
20//! applies `notice_days` (14) after it is decided, and a monthly meter's
21//! at the start of the month after that, so no month is charged at two
22//! prices. Owners of workspaces on the plan are emailed once per rise.
23//! Cost-plus means the rise does come: margin protection is for new usage
24//! after the notice, never retroactive.
25
26use g1t_contracts::billing::*;
27use g1t_contracts::time::{parse_rfc3339, rfc3339};
28use g1t_contracts::{FailureCode, Outcome, new_id};
29use g1t_kit::now_ms;
30use serde::Deserialize;
31use serde_json::Value;
32use worker::Result;
33
34use crate::Billing;
35
36/// Smaller moves are noise.
37pub(crate) const MIN_CHANGE: f64 = 0.02;
38/// A measurement outside this factor of the current cost is suspect: it
39/// is proposed for a person to look at, never applied on its own.
40pub(crate) const MAX_FACTOR: f64 = 4.0;
41
42/// Meters from when g1t ran agents and sold model access (the agent rate,
43/// its weights, models at the provider's price, and the AI Gateway), and
44/// the card fee from before it was folded into g1t's prices. They stay in
45/// the tables for the statements that name them, but are never shown as
46/// prices, proposed or charged again.
47pub(crate) const RETIRED_METERS: [&str; 10] = [
48 "agent_models",
49 "agent_tokens",
50 "agent_tokens_own",
51 "agent_token_weight_input",
52 "agent_token_weight_output",
53 "agent_token_weight_cache_read",
54 "agent_token_weight_cache_write",
55 "gateway_models",
56 "card_fee_percent",
57 "card_fee_fixed",
58];
59
60/// Whether `meter` is one of the [`RETIRED_METERS`].
61pub(crate) fn retired(meter: &str) -> bool {
62 RETIRED_METERS.contains(&meter)
63}
64
65/// Meters charged once a month from the month's total (`storage`). A rise
66/// in one takes effect at the start of a month.
67pub(crate) const MONTHLY: [&str; 9] = [
68 "git_operations",
69 "private_storage",
70 "actions_cache",
71 "custom_domain_month",
72 "embedding_tokens",
73 "scan_cpu",
74 "scan_rows",
75 "package_storage",
76 "telemetry_gb",
77];
78
79/// The price meters a month-end source is charged at, for the ledger's
80/// `price_version`.
81pub(crate) fn meters_of(source: &str) -> &'static [&'static str] {
82 match source {
83 "git" => &["git_operations"],
84 "storage" => &["private_storage"],
85 "context" => &["embedding_tokens"],
86 "security" => &["scan_cpu", "scan_rows"],
87 "domains" => &["custom_domain_month"],
88 "cache" => &["actions_cache"],
89 "package_storage" => &["package_storage"],
90 "telemetry" => &["telemetry_gb"],
91 _ => &[],
92 }
93}
94
95/// Rises smaller than this, in percent, are listed on the pricing page
96/// with their date but not emailed: the keeper moves sandbox seconds by a
97/// percent or two at a time, and an email for each would be noise.
98const EMAIL_RISE_PERCENT: f64 = 5.0;
99
100/// What a meter's `cost_micros` is (`PriceVersion::basis`): a weight of the
101/// agent rate's tokens is a multiplier; the agent rate and security
102/// activation are prices g1t sets, with no per-unit cost behind them (the
103/// models a run uses are charged apart, at cost); everything else is a
104/// cost g1t pays.
105pub(crate) fn basis_of(meter: &str) -> &'static str {
106 if meter.starts_with("agent_token_weight_") {
107 "weight"
108 } else if matches!(meter, "agent_tokens" | "agent_tokens_own") || meter == crate::features::SECURITY_METER {
109 "rate"
110 } else {
111 "cost"
112 }
113}
114
115/// Marks the proposals behind a meter's versions still waiting for their
116/// date as superseded, before a newer version replaces them; who decided
117/// it, and when, stay. Parameter: the meter.
118pub(crate) const SUPERSEDE_WAITING_SQL: &str = "UPDATE price_proposals SET status = 'superseded'
119 WHERE version_id IN (SELECT id FROM price_versions WHERE meter = ?1 AND applied_at IS NULL)";
120
121/// Owners told of rises per run, at most; the rest on the next run.
122const NOTICES_PER_RUN: usize = 40;
123
124/// What may change prices without a person.
125#[derive(Clone, Copy, Debug, PartialEq)]
126pub(crate) struct Guard {
127 pub auto_apply: bool,
128 /// The largest move applied on its own, in percent either way.
129 pub auto_percent: f64,
130 /// Days between deciding a rise and charging it.
131 pub notice_days: u32,
132}
133
134impl From<&CostSettings> for Guard {
135 fn from(s: &CostSettings) -> Self {
136 Guard { auto_apply: s.auto_apply, auto_percent: s.auto_apply_percent, notice_days: s.notice_days }
137 }
138}
139
140#[derive(Clone, Debug, PartialEq)]
141pub(crate) enum Decision {
142 /// Too small to matter.
143 Nothing,
144 /// Applied without a person, from `effective_ms`.
145 Auto { effective_ms: u64 },
146 /// Waits for staff. `suspect` when the measurement is wildly off.
147 Approve { suspect: bool },
148}
149
150/// When a new cost takes effect: a fall at once; a rise after the notice
151/// period, and for a monthly meter at the start of the first month after
152/// it.
153pub(crate) fn effective_ms(current: f64, new: f64, now_ms: u64, notice_days: u32, monthly: bool) -> u64 {
154 if new <= current {
155 return now_ms;
156 }
157 let after = now_ms + u64::from(notice_days) * crate::costs::DAY_MS;
158 if !monthly {
159 return after;
160 }
161 let stamp = rfc3339(after);
162 if &stamp[8..] == "01T00:00:00.000Z" {
163 return after;
164 }
165 parse_rfc3339(&crate::credits::next_month_start(&stamp[..7])).unwrap_or(after)
166}
167
168/// What to do with a measured cost against the one in force (or the one
169/// already scheduled).
170pub(crate) fn decide(current: f64, measured: f64, guard: Guard, now_ms: u64, monthly: bool) -> Decision {
171 if !measured.is_finite() || measured <= 0.0 || !current.is_finite() || current <= 0.0 {
172 return Decision::Nothing;
173 }
174 let ratio = measured / current;
175 if (ratio - 1.0).abs() < MIN_CHANGE {
176 return Decision::Nothing;
177 }
178 if !(1.0 / MAX_FACTOR..=MAX_FACTOR).contains(&ratio) {
179 return Decision::Approve { suspect: true };
180 }
181 if guard.auto_apply && (ratio - 1.0).abs() * 100.0 <= guard.auto_percent {
182 return Decision::Auto { effective_ms: effective_ms(current, measured, now_ms, guard.notice_days, monthly) };
183 }
184 Decision::Approve { suspect: false }
185}
186
187/// A version of one meter's price.
188#[derive(Clone, Debug, PartialEq, Deserialize)]
189pub(crate) struct Version {
190 pub id: String,
191 pub meter: String,
192 pub version: u32,
193 pub cost_micros: f64,
194 pub markup_percent: u32,
195 pub effective_at: String,
196 pub reason: String,
197 pub created_by: String,
198 pub applied_at: Option<String>,
199}
200
201/// The version in force for `meter` at `at`: the newest whose date has
202/// come. Old versions never change, so a past month's charges are always
203/// explained by the version of then.
204pub(crate) fn in_force<'a>(versions: &'a [Version], meter: &str, at: &str) -> Option<&'a Version> {
205 versions
206 .iter()
207 .filter(|v| v.meter == meter && v.effective_at.as_str() <= at)
208 .max_by(|a, b| a.effective_at.cmp(&b.effective_at).then(a.version.cmp(&b.version)))
209}
210
211/// Settings from `cost_settings` rows, defaults for anything missing or
212/// unreadable.
213pub(crate) fn settings_from(rows: &[(String, String)]) -> CostSettings {
214 let mut s = CostSettings::default();
215 for (key, value) in rows {
216 let value = value.trim();
217 match key.as_str() {
218 "auto_apply" => s.auto_apply = value == "true",
219 "auto_apply_percent" => s.auto_apply_percent = value.parse().unwrap_or(s.auto_apply_percent),
220 "notice_days" => s.notice_days = value.parse().unwrap_or(s.notice_days),
221 "margin_floor_percent" => s.margin_floor_percent = value.parse().unwrap_or(s.margin_floor_percent),
222 "alert_days" => s.alert_days = value.parse().unwrap_or(s.alert_days),
223 "min_daily_cost_micros" => s.min_daily_cost_micros = value.parse().unwrap_or(s.min_daily_cost_micros),
224 "anomaly_factor" => s.anomaly_factor = value.parse().unwrap_or(s.anomaly_factor),
225 "anomaly_floor_micros" => s.anomaly_floor_micros = value.parse().unwrap_or(s.anomaly_floor_micros),
226 _ => {}
227 }
228 }
229 s
230}
231
232/// Why settings cannot be saved, if they cannot.
233pub(crate) fn check_settings(s: &CostSettings) -> std::result::Result<(), String> {
234 if !(0.0..=100.0).contains(&s.auto_apply_percent) {
235 return Err("The guardrail is a percentage from 0 to 100.".into());
236 }
237 if s.notice_days > 90 {
238 return Err("Notice is at most 90 days.".into());
239 }
240 if !(-100.0..=100.0).contains(&s.margin_floor_percent) {
241 return Err("The margin floor is a percentage.".into());
242 }
243 if s.alert_days == 0 || s.alert_days > 30 {
244 return Err("Alert after 1 to 30 days.".into());
245 }
246 if s.min_daily_cost_micros < 0 || s.anomaly_floor_micros < 0 || !(s.anomaly_factor > 0.0 && s.anomaly_factor.is_finite()) {
247 return Err("Amounts and the factor must be positive.".into());
248 }
249 Ok(())
250}
251
252#[derive(Deserialize)]
253struct ProposalRow {
254 id: String,
255 meter: String,
256 current_cost_micros: f64,
257 proposed_cost_micros: f64,
258 reason: String,
259 source: String,
260 suspect: i64,
261 status: String,
262 created_at: String,
263 decided_at: Option<String>,
264 decided_by: Option<String>,
265 note: Option<String>,
266 version_id: Option<String>,
267 title: Option<String>,
268 unit: Option<String>,
269 markup_percent: Option<u32>,
270 effective_at: Option<String>,
271}
272
273impl From<ProposalRow> for PriceProposal {
274 fn from(r: ProposalRow) -> Self {
275 let change = if r.current_cost_micros > 0.0 { (r.proposed_cost_micros / r.current_cost_micros - 1.0) * 100.0 } else { 0.0 };
276 PriceProposal {
277 title: r.title.unwrap_or_else(|| r.meter.clone()),
278 unit: r.unit.unwrap_or_default(),
279 markup_percent: r.markup_percent.unwrap_or(20),
280 id: r.id,
281 meter: r.meter,
282 current_cost_micros: r.current_cost_micros,
283 proposed_cost_micros: r.proposed_cost_micros,
284 change_percent: change,
285 reason: r.reason,
286 source: r.source,
287 suspect: r.suspect != 0,
288 status: r.status,
289 created_at: r.created_at,
290 decided_at: r.decided_at,
291 decided_by: r.decided_by,
292 note: r.note,
293 effective_at: r.version_id.and(r.effective_at),
294 }
295 }
296}
297
298const PROPOSAL_SQL: &str = "SELECT p.*, pr.title, pr.unit, pr.markup_percent, v.effective_at
299 FROM price_proposals p
300 LEFT JOIN prices pr ON pr.meter = p.meter
301 LEFT JOIN price_versions v ON v.id = p.version_id";
302
303impl Billing {
304 pub(crate) async fn cost_settings(&self) -> Result<CostSettings> {
305 #[derive(Deserialize)]
306 struct Row {
307 key: String,
308 value: String,
309 }
310 let rows = self.db.prepare("SELECT key, value FROM cost_settings").all().await?.results::<Row>()?;
311 Ok(settings_from(&rows.into_iter().map(|r| (r.key, r.value)).collect::<Vec<_>>()))
312 }
313
314 pub(crate) async fn admin_set_cost_settings(&self, a: AdminSetCostSettingsArgs) -> Result<Outcome<CostSettings>> {
315 if let Err(why) = check_settings(&a.settings) {
316 return Ok(Outcome::fail(FailureCode::Invalid, why));
317 }
318 let s = &a.settings;
319 let now = rfc3339(now_ms());
320 let pairs = [
321 ("auto_apply", s.auto_apply.to_string()),
322 ("auto_apply_percent", s.auto_apply_percent.to_string()),
323 ("notice_days", s.notice_days.to_string()),
324 ("margin_floor_percent", s.margin_floor_percent.to_string()),
325 ("alert_days", s.alert_days.to_string()),
326 ("min_daily_cost_micros", s.min_daily_cost_micros.to_string()),
327 ("anomaly_factor", s.anomaly_factor.to_string()),
328 ("anomaly_floor_micros", s.anomaly_floor_micros.to_string()),
329 ];
330 let mut statements = Vec::new();
331 for (key, value) in &pairs {
332 statements.push(
333 self.db
334 .prepare(
335 "INSERT INTO cost_settings (key, value, updated_at, updated_by) VALUES (?1, ?2, ?3, ?4)
336 ON CONFLICT (key) DO UPDATE SET value = ?2, updated_at = ?3, updated_by = ?4",
337 )
338 .bind(&[(*key).into(), value.as_str().into(), now.as_str().into(), a.by.as_str().into()])?,
339 );
340 }
341 self.db.batch(statements).await?;
342 let detail = pairs.iter().map(|(k, v)| format!("{k}={v}")).collect::<Vec<_>>().join(", ");
343 self.audit("costs", "cost_settings", &detail, &a.by).await?;
344 Ok(Outcome::Ok(self.cost_settings().await?))
345 }
346
347 /// The cost a new measurement is judged against: the newest version
348 /// scheduled, else the one in force.
349 async fn latest_cost(&self, meter: &str) -> Result<Option<(f64, u32)>> {
350 #[derive(Deserialize)]
351 struct Row {
352 cost_micros: f64,
353 markup_percent: u32,
354 }
355 let scheduled = self
356 .db
357 .prepare("SELECT cost_micros, markup_percent FROM price_versions WHERE meter = ? AND applied_at IS NULL ORDER BY version DESC LIMIT 1")
358 .bind(&[meter.into()])?
359 .first::<Row>(None)
360 .await?;
361 if let Some(row) = scheduled {
362 return Ok(Some((row.cost_micros, row.markup_percent)));
363 }
364 Ok(self
365 .db
366 .prepare("SELECT cost_micros, markup_percent FROM prices WHERE meter = ?")
367 .bind(&[meter.into()])?
368 .first::<Row>(None)
369 .await?
370 .map(|r| (r.cost_micros, r.markup_percent)))
371 }
372
373 /// Proposes a measured cost for a meter. Returns what happened, in
374 /// words, or None when there was nothing to do.
375 pub(crate) async fn propose(&self, meter: &str, measured: f64, reason: &str, source: &str) -> Result<Option<String>> {
376 let Some((current, markup)) = self.latest_cost(meter).await? else {
377 return Ok(None);
378 };
379 // Ten-thousandths of a micro are plenty; floating-point tails are not.
380 let measured = (measured * 10_000.0).round() / 10_000.0;
381 let settings = self.cost_settings().await?;
382 let now = now_ms();
383 let decision = decide(current, measured, Guard::from(&settings), now, MONTHLY.contains(&meter));
384 if decision == Decision::Nothing {
385 return Ok(None);
386 }
387 let stamp = rfc3339(now);
388 let id = new_id("ppr", now);
389 // A newer measurement replaces an open one.
390 self.db
391 .prepare("UPDATE price_proposals SET status = 'superseded', decided_at = ? WHERE meter = ? AND status = 'open'")
392 .bind(&[stamp.as_str().into(), meter.into()])?
393 .run()
394 .await?;
395 let (status, suspect) = match decision {
396 Decision::Auto { .. } => ("applied", false),
397 Decision::Approve { suspect } => ("open", suspect),
398 Decision::Nothing => unreachable!(),
399 };
400 self.db
401 .prepare(
402 "INSERT INTO price_proposals (id, meter, current_cost_micros, proposed_cost_micros, reason, source, suspect, status, created_at, decided_at, decided_by)
403 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
404 )
405 .bind(&[
406 id.as_str().into(),
407 meter.into(),
408 current.into(),
409 measured.into(),
410 reason.into(),
411 source.into(),
412 i32::from(suspect).into(),
413 status.into(),
414 stamp.as_str().into(),
415 crate::optional((status == "applied").then_some(stamp.as_str())),
416 crate::optional((status == "applied").then_some("guardrail")),
417 ])?
418 .run()
419 .await?;
420 if status == "open" {
421 return Ok(Some(format!(
422 "proposed {measured:.4} against {current:.4}{}, waiting for staff",
423 if suspect { " (far off: look before approving)" } else { "" }
424 )));
425 }
426 let Decision::Auto { effective_ms } = decision else { unreachable!() };
427 let version = self.schedule_version(meter, measured, markup, effective_ms, reason, source, Some(&id)).await?;
428 self.apply_due_versions().await?;
429 Ok(Some(format!("applied {measured:.4} (was {current:.4}) from {} as {version}", rfc3339(effective_ms))))
430 }
431
432 /// Writes the next version of a meter's price, to take effect at
433 /// `effective_ms`. Returns its id.
434 #[allow(clippy::too_many_arguments)]
435 pub(crate) async fn schedule_version(
436 &self,
437 meter: &str,
438 cost: f64,
439 markup: u32,
440 effective_ms: u64,
441 reason: &str,
442 by: &str,
443 proposal: Option<&str>,
444 ) -> Result<String> {
445 #[derive(Deserialize)]
446 struct Top {
447 version: Option<u32>,
448 }
449 let next = self
450 .db
451 .prepare("SELECT MAX(version) AS version FROM price_versions WHERE meter = ?")
452 .bind(&[meter.into()])?
453 .first::<Top>(None)
454 .await?
455 .and_then(|t| t.version)
456 .unwrap_or(0)
457 + 1;
458 let id = format!("pv_{meter}_{next}");
459 let now = rfc3339(now_ms());
460 // A newer decision replaces a rise still waiting for its date. The
461 // proposal behind it never took effect: superseded, not "applied"
462 // with no date it is in force from.
463 self.db
464 .batch(vec![
465 self.db.prepare(SUPERSEDE_WAITING_SQL).bind(&[meter.into()])?,
466 self.db.prepare("DELETE FROM price_versions WHERE meter = ? AND applied_at IS NULL").bind(&[meter.into()])?,
467 ])
468 .await?;
469 self.db
470 .prepare(
471 "INSERT INTO price_versions (id, meter, version, cost_micros, markup_percent, effective_at, reason, created_by, proposal_id, created_at)
472 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
473 )
474 .bind(&[
475 id.as_str().into(),
476 meter.into(),
477 next.into(),
478 cost.into(),
479 markup.into(),
480 rfc3339(effective_ms).into(),
481 reason.into(),
482 by.into(),
483 crate::optional(proposal),
484 now.as_str().into(),
485 ])?
486 .run()
487 .await?;
488 if let Some(proposal) = proposal {
489 self.db
490 .prepare("UPDATE price_proposals SET version_id = ? WHERE id = ?")
491 .bind(&[id.as_str().into(), proposal.into()])?
492 .run()
493 .await?;
494 }
495 Ok(id)
496 }
497
498 /// Puts every version whose date has come into the price book, oldest
499 /// first, each once, with a public record of the change.
500 pub(crate) async fn apply_due_versions(&self) -> Result<u32> {
501 let now = rfc3339(now_ms());
502 let due = self
503 .db
504 .prepare("SELECT * FROM price_versions WHERE applied_at IS NULL AND effective_at <= ? ORDER BY effective_at, version")
505 .bind(&[now.as_str().into()])?
506 .all()
507 .await?
508 .results::<Version>()?;
509 let mut applied = 0;
510 for v in due {
511 let claimed = self
512 .db
513 .prepare("UPDATE price_versions SET applied_at = ? WHERE id = ? AND applied_at IS NULL RETURNING id")
514 .bind(&[now.as_str().into(), v.id.as_str().into()])?
515 .first::<Value>(None)
516 .await?;
517 if claimed.is_none() {
518 continue;
519 }
520 let reason = if v.created_by.contains('@') { format!("{} (approved by g1t staff)", v.reason) } else { v.reason.clone() };
521 self.db
522 .batch(vec![
523 self.db
524 .prepare(
525 "INSERT INTO price_changes (id, meter, old_cost_micros, new_cost_micros, markup_percent, old_markup_percent, reason, created_at)
526 SELECT ?, meter, cost_micros, ?, ?, CASE WHEN markup_percent <> ? THEN markup_percent END, ?, ? FROM prices WHERE meter = ?",
527 )
528 .bind(&[
529 new_id("prc", now_ms()).into(),
530 v.cost_micros.into(),
531 v.markup_percent.into(),
532 v.markup_percent.into(),
533 reason.as_str().into(),
534 now.as_str().into(),
535 v.meter.as_str().into(),
536 ])?,
537 self.db
538 .prepare("UPDATE prices SET cost_micros = ?, markup_percent = ?, source = 'cloudflare', updated_at = ? WHERE meter = ?")
539 .bind(&[v.cost_micros.into(), v.markup_percent.into(), now.as_str().into(), v.meter.as_str().into()])?,
540 ])
541 .await?;
542 applied += 1;
543 }
544 Ok(applied)
545 }
546
547 /// The id of the price version in force for `meter` now, for the
548 /// ledger.
549 pub(crate) async fn version_now(&self, meter: &str) -> Result<Option<String>> {
550 let versions = self
551 .db
552 .prepare("SELECT * FROM price_versions WHERE meter = ? AND applied_at IS NOT NULL")
553 .bind(&[meter.into()])?
554 .all()
555 .await?
556 .results::<Version>()?;
557 Ok(in_force(&versions, meter, &rfc3339(now_ms())).map(|v| v.id.clone()))
558 }
559
560 pub(crate) async fn proposals(&self) -> Result<Vec<PriceProposal>> {
561 Ok(self
562 .db
563 .prepare(format!(
564 "{PROPOSAL_SQL} ORDER BY CASE WHEN p.status = 'open' THEN 0 ELSE 1 END, p.created_at DESC LIMIT 40"
565 ))
566 .all()
567 .await?
568 .results::<ProposalRow>()?
569 .into_iter()
570 .map(PriceProposal::from)
571 .collect())
572 }
573
574 pub(crate) async fn versions(&self) -> Result<Vec<PriceVersion>> {
575 Ok(self
576 .db
577 .prepare("SELECT * FROM price_versions ORDER BY CASE WHEN applied_at IS NULL THEN 0 ELSE 1 END, effective_at DESC, version DESC LIMIT 60")
578 .all()
579 .await?
580 .results::<Version>()?
581 .into_iter()
582 .map(|v| PriceVersion {
583 price_micros: Price::price_for(v.cost_micros, v.markup_percent),
584 basis: basis_of(&v.meter).to_owned(),
585 id: v.id,
586 meter: v.meter,
587 version: v.version,
588 cost_micros: v.cost_micros,
589 markup_percent: v.markup_percent,
590 effective_at: v.effective_at,
591 reason: v.reason,
592 created_by: v.created_by,
593 applied_at: v.applied_at,
594 })
595 .collect())
596 }
597
598 /// Prices not in force yet, for the pricing page's "coming" changes.
599 pub(crate) async fn coming_changes(&self) -> Result<Vec<PriceChange>> {
600 #[derive(Deserialize)]
601 struct Row {
602 meter: String,
603 old_cost_micros: Option<f64>,
604 cost_micros: f64,
605 markup_percent: u32,
606 reason: String,
607 created_at: String,
608 effective_at: String,
609 }
610 Ok(self
611 .db
612 .prepare(
613 "SELECT v.meter, p.cost_micros AS old_cost_micros, v.cost_micros, v.markup_percent, v.reason, v.created_at, v.effective_at
614 FROM price_versions v LEFT JOIN prices p ON p.meter = v.meter
615 WHERE v.applied_at IS NULL ORDER BY v.effective_at",
616 )
617 .all()
618 .await?
619 .results::<Row>()?
620 .into_iter()
621 .map(|r| PriceChange {
622 old_cost_micros: r.old_cost_micros.unwrap_or(r.cost_micros),
623 meter: r.meter,
624 new_cost_micros: r.cost_micros,
625 markup_percent: r.markup_percent,
626 old_markup_percent: None,
627 reason: r.reason,
628 created_at: r.created_at,
629 effective_at: Some(r.effective_at),
630 })
631 .collect())
632 }
633
634 /// `admin_decide_proposal`: an approved fall applies now; an approved
635 /// rise after the notice period.
636 pub(crate) async fn admin_decide_proposal(&self, a: AdminDecideProposalArgs) -> Result<Outcome<PriceProposal>> {
637 let found = self
638 .db
639 .prepare(format!("{PROPOSAL_SQL} WHERE p.id = ?"))
640 .bind(&[a.id.as_str().into()])?
641 .first::<ProposalRow>(None)
642 .await?;
643 let Some(found) = found else {
644 return Ok(Outcome::fail(FailureCode::NotFound, "No such proposal."));
645 };
646 if found.status != "open" {
647 return Ok(Outcome::fail(FailureCode::Conflict, format!("This proposal is already {}.", found.status)));
648 }
649 let approve = match a.decision.as_str() {
650 "approve" => true,
651 "reject" => false,
652 _ => return Ok(Outcome::fail(FailureCode::Invalid, "Approve or reject.")),
653 };
654 if !approve && a.note.trim().is_empty() {
655 return Ok(Outcome::fail(FailureCode::Invalid, "Say why it is rejected, for whoever measures it next."));
656 }
657 let now = now_ms();
658 let stamp = rfc3339(now);
659 self.db
660 .prepare("UPDATE price_proposals SET status = ?, decided_at = ?, decided_by = ?, note = ? WHERE id = ? AND status = 'open'")
661 .bind(&[
662 if approve { "approved" } else { "rejected" }.into(),
663 stamp.as_str().into(),
664 a.by.as_str().into(),
665 a.note.trim().into(),
666 a.id.as_str().into(),
667 ])?
668 .run()
669 .await?;
670 if approve {
671 let settings = self.cost_settings().await?;
672 let (current, markup) = self.latest_cost(&found.meter).await?.unwrap_or((found.current_cost_micros, 20));
673 let effective = effective_ms(current, found.proposed_cost_micros, now, settings.notice_days, MONTHLY.contains(&found.meter.as_str()));
674 self.schedule_version(&found.meter, found.proposed_cost_micros, markup, effective, &found.reason, &a.by, Some(&found.id))
675 .await?;
676 self.apply_due_versions().await?;
677 }
678 self.audit(
679 "costs",
680 if approve { "price_approved" } else { "price_rejected" },
681 &format!("{}: {:.4} to {:.4}. {}", found.meter, found.current_cost_micros, found.proposed_cost_micros, a.note.trim()),
682 &a.by,
683 )
684 .await?;
685 let row = self
686 .db
687 .prepare(format!("{PROPOSAL_SQL} WHERE p.id = ?"))
688 .bind(&[a.id.as_str().into()])?
689 .first::<ProposalRow>(None)
690 .await?;
691 Ok(match row {
692 Some(row) => Outcome::Ok(row.into()),
693 None => Outcome::fail(FailureCode::NotFound, "No such proposal."),
694 })
695 }
696
697 /// Emails owners of workspaces on the plan about each rise still to
698 /// come, once per rise per workspace.
699 pub(crate) async fn tell_owners_of_rises(&self, identity: &worker::Fetcher) -> Result<()> {
700 #[derive(Deserialize)]
701 struct Rise {
702 id: String,
703 title: Option<String>,
704 old_cost: f64,
705 cost_micros: f64,
706 markup_percent: u32,
707 unit: Option<String>,
708 effective_at: String,
709 reason: String,
710 }
711 let rises = self
712 .db
713 .prepare(
714 "SELECT v.id, p.title, p.cost_micros AS old_cost, v.cost_micros, v.markup_percent, p.unit, v.effective_at, v.reason
715 FROM price_versions v JOIN prices p ON p.meter = v.meter
716 WHERE v.applied_at IS NULL AND v.cost_micros > p.cost_micros * ?",
717 )
718 .bind(&[(1.0 + EMAIL_RISE_PERCENT / 100.0).into()])?
719 .all()
720 .await?
721 .results::<Rise>()?;
722 let mut sent = 0;
723 for rise in rises {
724 #[derive(Deserialize)]
725 struct Workspace {
726 workspace: String,
727 }
728 let workspaces = self
729 .db
730 .prepare(
731 "SELECT DISTINCT workspace FROM subscriptions WHERE feature = 'plan' AND status IN ('active', 'canceling')
732 AND workspace NOT IN (SELECT workspace FROM price_notices WHERE version_id = ?) LIMIT ?",
733 )
734 .bind(&[rise.id.as_str().into(), ((NOTICES_PER_RUN - sent) as u32).into()])?
735 .all()
736 .await?
737 .results::<Workspace>()?;
738 let title = rise.title.clone().unwrap_or_default();
739 let unit = rise.unit.clone().unwrap_or_default();
740 let price = |cost: f64| crate::features::dollars(Price::price_for(cost, rise.markup_percent).round() as i64);
741 let intro = format!(
742 "From {}, {title} on g1t goes from {} to {} per {unit}. It is what g1t pays Cloudflare plus 20%, and the cost moved: {} Nothing already charged changes.",
743 &rise.effective_at[..10],
744 price(rise.old_cost),
745 price(rise.cost_micros),
746 rise.reason
747 );
748 for Workspace { workspace } in workspaces {
749 let args = g1t_contracts::identity::NotifyOwnersArgs {
750 workspace: workspace.clone(),
751 subject: format!("g1t: {title} costs more from {}", &rise.effective_at[..10]),
752 intro: intro.clone(),
753 action: "See prices".to_owned(),
754 link: "https://g1t.sh/pricing".to_owned(),
755 footer: "You get this because you own a workspace on the g1t plan. How prices follow costs: https://docs.g1t.sh/guides/usage-and-billing/#how-prices-are-set".to_owned(),
756 };
757 // Recorded whether or not anyone was there to tell.
758 if let Err(error) = g1t_kit::call::<_, u32>(identity, "notify_owners", &args).await {
759 worker::console_error!("could not tell {workspace} of a price rise: {error}");
760 continue;
761 }
762 self.db
763 .prepare("INSERT OR IGNORE INTO price_notices (version_id, workspace, sent_at) VALUES (?, ?, ?)")
764 .bind(&[rise.id.as_str().into(), workspace.as_str().into(), rfc3339(now_ms()).into()])?
765 .run()
766 .await?;
767 sent += 1;
768 if sent >= NOTICES_PER_RUN {
769 return Ok(());
770 }
771 }
772 }
773 Ok(())
774 }
775}
776
777#[cfg(test)]
778mod tests {
779 use super::*;
780
781 const NOW: &str = "2026-10-06T04:17:00.000Z";
782
783 fn now() -> u64 {
784 parse_rfc3339(NOW).unwrap()
785 }
786
787 fn guard() -> Guard {
788 Guard { auto_apply: true, auto_percent: 25.0, notice_days: 14 }
789 }
790
791 #[test]
792 fn small_moves_are_noise_and_wild_ones_wait_for_a_person() {
793 assert_eq!(decide(21.0, 21.2, guard(), now(), false), Decision::Nothing);
794 assert_eq!(decide(21.0, 0.0, guard(), now(), false), Decision::Nothing);
795 assert_eq!(decide(21.0, 200.0, guard(), now(), false), Decision::Approve { suspect: true });
796 assert_eq!(decide(21.0, 2.0, guard(), now(), false), Decision::Approve { suspect: true });
797 }
798
799 #[test]
800 fn moves_inside_the_guardrail_apply_themselves_falls_at_once_rises_after_notice() {
801 // Down 19%: at once.
802 assert_eq!(decide(21.0, 17.0, guard(), now(), false), Decision::Auto { effective_ms: now() });
803 // Up 19%: after 14 days.
804 let Decision::Auto { effective_ms } = decide(21.0, 25.0, guard(), now(), false) else { panic!() };
805 assert_eq!(rfc3339(effective_ms), "2026-10-20T04:17:00.000Z");
806 // A monthly meter's rise waits for the month after the notice.
807 let Decision::Auto { effective_ms } = decide(150_000.0, 180_000.0, guard(), now(), true) else { panic!() };
808 assert_eq!(rfc3339(effective_ms), "2026-11-01T00:00:00.000Z");
809 let late = parse_rfc3339("2026-10-25T00:00:00Z").unwrap();
810 assert_eq!(rfc3339(effective_ms_for(late)), "2026-12-01T00:00:00.000Z");
811 }
812
813 fn effective_ms_for(now: u64) -> u64 {
814 effective_ms(1.0, 2.0, now, 14, true)
815 }
816
817 #[test]
818 fn a_versions_cost_says_what_it_is() {
819 // The agent rate's $0.25 a million tokens is g1t's price, not what
820 // tokens cost g1t; the weights are multipliers.
821 assert_eq!(basis_of("agent_tokens"), "rate");
822 assert_eq!(basis_of("agent_tokens_own"), "rate");
823 assert_eq!(basis_of("security_activation"), "rate");
824 assert_eq!(basis_of("agent_token_weight_cache_read"), "weight");
825 // A provider's dollar passed on at cost, and Cloudflare's units, are costs.
826 for meter in ["agent_models", "gateway_models", "sandbox_second", "git_operations", "card_fee_percent"] {
827 assert_eq!(basis_of(meter), "cost", "{meter}");
828 }
829 }
830
831 #[test]
832 fn a_rise_replaced_before_its_date_is_superseded_not_applied() {
833 // 2026-10-07's sandbox rise (v2, due 10-21) was replaced by
834 // 10-08's (v3, due 10-22): v2's proposal never took effect.
835 let sql = SUPERSEDE_WAITING_SQL;
836 assert_eq!(crate::rename::parameters(sql), 1);
837 assert!(sql.contains("status = 'superseded'") && sql.contains("applied_at IS NULL") && !sql.contains("decided_"), "{sql}");
838 // A rise still waits its notice: the replacement is no shortcut.
839 let Decision::Auto { effective_ms } = decide(18.1817, 19.5181, guard(), now(), false) else { panic!() };
840 assert_eq!(rfc3339(effective_ms), "2026-10-20T04:17:00.000Z");
841 }
842
843 #[test]
844 fn past_the_guardrail_or_with_auto_off_staff_decide() {
845 // Up 50%: staff.
846 assert_eq!(decide(150_000.0, 225_000.0, guard(), now(), true), Decision::Approve { suspect: false });
847 // Down 30%: staff too; the guardrail is either way.
848 assert_eq!(decide(100.0, 70.0, guard(), now(), false), Decision::Approve { suspect: false });
849 let off = Guard { auto_apply: false, ..guard() };
850 assert_eq!(decide(21.0, 22.0, off, now(), false), Decision::Approve { suspect: false });
851 }
852
853 #[test]
854 fn a_past_statement_keeps_the_price_of_its_time() {
855 let v = |version: u32, cost: f64, at: &str| Version {
856 id: format!("pv_git_operations_{version}"),
857 meter: "git_operations".into(),
858 version,
859 cost_micros: cost,
860 markup_percent: 20,
861 effective_at: at.into(),
862 reason: String::new(),
863 created_by: "keeper".into(),
864 applied_at: None,
865 };
866 let versions = vec![v(1, 150_000.0, "2026-10-05T00:00:00Z"), v(2, 450_000.0, "2026-11-01T00:00:00.000Z"), v(3, 400_000.0, "2026-12-01T00:00:00.000Z")];
867 // October's charges were at version 1, whatever came later.
868 assert_eq!(in_force(&versions, "git_operations", "2026-10-31T23:59:59Z").unwrap().version, 1);
869 assert_eq!(in_force(&versions, "git_operations", "2026-11-15T00:00:00Z").unwrap().version, 2);
870 assert_eq!(in_force(&versions, "git_operations", "2027-01-01T00:00:00Z").unwrap().cost_micros, 400_000.0);
871 assert!(in_force(&versions, "git_operations", "2026-01-01T00:00:00Z").is_none());
872 assert!(in_force(&versions, "sandbox_second", "2027-01-01T00:00:00Z").is_none());
873 }
874
875 #[test]
876 fn settings_read_with_defaults_and_are_checked() {
877 let s = settings_from(&[("auto_apply".into(), "false".into()), ("notice_days".into(), "30".into()), ("anomaly_factor".into(), "x".into())]);
878 assert!(!s.auto_apply);
879 assert_eq!(s.notice_days, 30);
880 assert_eq!(s.anomaly_factor, 1.0);
881 assert_eq!(s.auto_apply_percent, 25.0);
882 assert!(check_settings(&s).is_ok());
883 assert!(check_settings(&CostSettings { notice_days: 120, ..s.clone() }).is_err());
884 assert!(check_settings(&CostSettings { alert_days: 0, ..s.clone() }).is_err());
885 assert!(check_settings(&CostSettings { auto_apply_percent: 150.0, ..s }).is_err());
886 }
887}