Sign in

flagon-io/g1t

public

Git for AI scale: a forge for thousands of agents working on the same code at once.

gitagentscloudflarerustforgemcp

g1t/services/billing/src/sales.rs

552 lines22,927 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.

Two limits, real invoices, trust that grows by itself, sales signals1//! What g1t's team needs to sell and support: month-by-month figures, the
2//! signals that say a workspace is worth a call, and what was done about
3//! it. Staff only, through sudo.g1t.sh.
4
5use g1t_contracts::billing::{
6 AdminAddNoteArgs, AdminOverviewArgs, AdminSalesArgs, AdminSetSalesArgs, AdminSignalsArgs, KindFigures,
7 LimitState, MonthFigures, Overview, SalesNote, SalesRecord, Signal, SignalKind, TermsKind, Trust,
8};
9use g1t_contracts::time::rfc3339;
10use g1t_contracts::{FailureCode, Outcome, new_id};
11use g1t_kit::now_ms;
12use serde::Deserialize;
13use worker::Result;
14use worker::wasm_bindgen::JsValue;
15
16use crate::Billing;
17use crate::features::dollars;
18use crate::limits::previous_month;
19
20pub(crate) const STAGES: &[&str] = &["none", "lead", "contacted", "negotiating", "won", "lost", "churn_risk"];
21
22/// A workspace this much ahead of last month's pace is growing.
23const GROWING_FACTOR: f64 = 1.5;
24/// Below this last month, growth is noise.
25const GROWING_FROM_MICROS: i64 = 10_000_000;
26/// Spending this much a month may suit custom terms or an enterprise.
27const HIGH_SPEND_MICROS: i64 = 500_000_000;
28
29/// The six months ending with `month`, oldest first.
30pub(crate) fn last_months(month: &str, count: usize) -> Vec<String> {
31 let mut months = vec![month.to_owned()];
32 while months.len() < count {
33 let earlier = previous_month(months.last().unwrap());
34 months.push(earlier);
35 }
36 months.reverse();
37 months
38}
39
40/// How urgent a kind of signal is: lower first.
41fn urgency(kind: SignalKind) -> u8 {
42 match kind {
43 SignalKind::AtLimit => 0,
44 SignalKind::Declined => 1,
45 SignalKind::NearCeiling => 2,
46 SignalKind::HighSpend => 3,
47 SignalKind::Growing => 4,
48 SignalKind::Established => 5,
49 SignalKind::FirstPayment => 6,
50 }
51}
52
53/// Whether this month, at its pace so far, is well ahead of last month.
54pub(crate) fn growing(this_month: i64, last_month: i64, day: u32, days_in_month: u32) -> bool {
55 if last_month < GROWING_FROM_MICROS || day == 0 {
56 return false;
57 }
58 let pace = this_month as f64 * f64::from(days_in_month) / f64::from(day);
59 pace >= last_month as f64 * GROWING_FACTOR
60}
61
62fn days_in(month: &str) -> u32 {
63 let year: i32 = month[..4].parse().unwrap_or(1970);
64 match month[5..7].parse::<u32>().unwrap_or(1) {
65 2 if (year % 4 == 0 && year % 100 != 0) || year % 400 == 0 => 29,
66 2 => 28,
67 4 | 6 | 9 | 11 => 30,
68 _ => 31,
69 }
70}
71
72#[derive(Deserialize)]
73struct MonthRow {
74 month: String,
75 charged: Option<i64>,
76 cost: Option<i64>,
77 paid: Option<i64>,
78}
79
80#[derive(Deserialize)]
81struct RecordRow {
82 stage: String,
83 owner: Option<String>,
84 next_step: Option<String>,
85 next_at: Option<String>,
86 updated_at: String,
87}
88
89#[derive(Deserialize)]
90struct NoteRow {
91 id: String,
92 text: String,
93 by: String,
94 created_at: String,
95}
96
97impl Billing {
98 /// Month-by-month figures for some workspaces (all, when empty).
99 pub(crate) async fn months_for(&self, workspaces: &[String], count: usize) -> Result<Vec<MonthFigures>> {
100 let months = last_months(&rfc3339(now_ms())[..7], count);
101 let since = format!("{}-01", months[0]);
102 let (filter, values): (String, Vec<JsValue>) = if workspaces.is_empty() {
103 (String::new(), vec![])
104 } else {
105 let marks = vec!["?"; workspaces.len()].join(", ");
106 (format!("AND workspace IN ({marks})"), workspaces.iter().map(|w| JsValue::from(w.as_str())).collect())
107 };
108 let rows = self
109 .db
110 .prepare(format!(
111 "SELECT substr(created_at, 1, 7) AS month,
112 -SUM(CASE WHEN kind = 'usage' THEN amount_micros END) AS charged,
113 SUM(CASE WHEN kind = 'usage' THEN cost_micros END) AS cost,
114 SUM(CASE WHEN kind = 'top_up' AND reference NOT LIKE 'crd%' THEN amount_micros END) AS paid
115 FROM ledger WHERE created_at >= '{since}' {filter} GROUP BY 1"
116 ))
117 .bind(&values)?
118 .all()
119 .await?
120 .results::<MonthRow>()?;
121 Ok(months
122 .into_iter()
123 .map(|month| {
124 let row = rows.iter().find(|r| r.month == month);
125 MonthFigures {
126 charged_micros: row.and_then(|r| r.charged).unwrap_or(0),
127 cost_micros: row.and_then(|r| r.cost).unwrap_or(0),
128 paid_micros: row.and_then(|r| r.paid).unwrap_or(0),
129 month,
130 }
131 })
132 .collect())
133 }
134
135 async fn record_row(&self, workspace: &str) -> Result<Option<RecordRow>> {
136 self.db
137 .prepare("SELECT * FROM sales_records WHERE workspace = ?")
138 .bind(&[workspace.into()])?
139 .first::<RecordRow>(None)
140 .await
141 }
142
143 pub(crate) async fn admin_sales(&self, a: AdminSalesArgs) -> Result<SalesRecord> {
144 let workspace = a.workspace.trim().to_lowercase();
145 let row = self.record_row(&workspace).await?;
146 let notes = self
147 .db
148 .prepare("SELECT id, text, by, created_at FROM sales_notes WHERE workspace = ? ORDER BY created_at DESC LIMIT 100")
149 .bind(&[workspace.as_str().into()])?
150 .all()
151 .await?
152 .results::<NoteRow>()?
153 .into_iter()
154 .map(|n| SalesNote { id: n.id, text: n.text, by: n.by, created_at: n.created_at })
155 .collect();
156 Ok(match row {
157 Some(row) => SalesRecord {
158 workspace,
159 stage: row.stage,
160 owner: row.owner,
161 next_step: row.next_step,
162 next_at: row.next_at,
163 notes,
164 updated_at: Some(row.updated_at),
165 },
166 None => SalesRecord {
167 workspace,
168 stage: "none".to_owned(),
169 owner: None,
170 next_step: None,
171 next_at: None,
172 notes,
173 updated_at: None,
174 },
175 })
176 }
177
178 pub(crate) async fn admin_set_sales(&self, a: AdminSetSalesArgs) -> Result<Outcome<SalesRecord>> {
179 let workspace = a.workspace.trim().to_lowercase();
180 if workspace.is_empty() || a.by.trim().is_empty() {
181 return Ok(Outcome::fail(FailureCode::Invalid, "Name the workspace, and who is making the change."));
182 }
183 if !STAGES.contains(&a.stage.as_str()) {
184 return Ok(Outcome::fail(FailureCode::Invalid, format!("A stage is one of: {}.", STAGES.join(", "))));
185 }
186 let clean = |value: &Option<String>| value.as_deref().map(str::trim).filter(|v| !v.is_empty()).map(str::to_owned);
187 let (owner, next_step, next_at) = (clean(&a.owner), clean(&a.next_step), clean(&a.next_at));
188 let before = self.record_row(&workspace).await?;
189 self.db
190 .prepare(
191 "INSERT INTO sales_records (workspace, stage, owner, next_step, next_at, updated_by, updated_at)
192 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)
193 ON CONFLICT (workspace) DO UPDATE SET stage = ?2, owner = ?3, next_step = ?4, next_at = ?5,
194 updated_by = ?6, updated_at = ?7",
195 )
196 .bind(&[
197 workspace.as_str().into(),
198 a.stage.as_str().into(),
199 crate::optional(owner.as_deref()),
200 crate::optional(next_step.as_deref()),
201 crate::optional(next_at.as_deref()),
202 a.by.as_str().into(),
203 rfc3339(now_ms()).into(),
204 ])?
205 .run()
206 .await?;
207 let account = self.account_of(&workspace).await?;
208 let was = before.map_or_else(|| "none".to_owned(), |b| b.stage);
209 let detail = format!(
210 "{workspace}: stage {was} → {}{}{}",
211 a.stage,
212 owner.as_deref().map(|o| format!(", owner {o}")).unwrap_or_default(),
213 next_step.as_deref().map(|s| format!(", next: {s}")).unwrap_or_default(),
214 );
215 self.audit(&account.id, "sales", &detail, &a.by).await?;
216 Ok(Outcome::Ok(self.admin_sales(AdminSalesArgs { workspace }).await?))
217 }
218
219 pub(crate) async fn admin_add_note(&self, a: AdminAddNoteArgs) -> Result<Outcome<SalesRecord>> {
220 let workspace = a.workspace.trim().to_lowercase();
221 let text = a.text.trim();
222 if workspace.is_empty() || text.is_empty() || a.by.trim().is_empty() {
223 return Ok(Outcome::fail(FailureCode::Invalid, "A note needs a workspace, some words, and who wrote it."));
224 }
225 if text.chars().count() > 4000 {
226 return Ok(Outcome::fail(FailureCode::Invalid, "Keep a note under 4,000 characters."));
227 }
228 let now = now_ms();
229 self.db
230 .prepare("INSERT INTO sales_notes (id, workspace, text, by, created_at) VALUES (?, ?, ?, ?, ?)")
231 .bind(&[new_id("note", now).into(), workspace.as_str().into(), text.into(), a.by.as_str().into(), rfc3339(now).into()])?
232 .run()
233 .await?;
234 Ok(Outcome::Ok(self.admin_sales(AdminSalesArgs { workspace }).await?))
235 }
236
237 /// Every workspace worth reaching out to, most urgent first.
238 pub(crate) async fn admin_signals(&self, _: AdminSignalsArgs) -> Result<Vec<Signal>> {
239 let now = rfc3339(now_ms());
240 let month = &now[..7];
241 let last = previous_month(month);
242 #[derive(Deserialize)]
243 struct Active {
244 workspace: String,
245 this_month: Option<i64>,
246 last_month: Option<i64>,
247 first_paid: Option<String>,
248 }
249 let active = self
250 .db
251 .prepare(
252 "SELECT workspace,
253 -SUM(CASE WHEN kind = 'usage' AND created_at >= ?1 THEN amount_micros END) AS this_month,
254 -SUM(CASE WHEN kind = 'usage' AND created_at >= ?2 AND created_at < ?1 THEN amount_micros END) AS last_month,
255 MIN(CASE WHEN kind = 'top_up' AND amount_micros > 0 AND reference NOT LIKE 'crd%' THEN created_at END) AS first_paid
256 FROM ledger GROUP BY workspace
257 HAVING MAX(created_at) >= ?2
258 LIMIT 500",
259 )
260 .bind(&[format!("{month}-01").into(), format!("{last}-01").into()])?
261 .all()
262 .await?
263 .results::<Active>()?;
264 let day: u32 = now[8..10].parse().unwrap_or(1);
265 let fortnight_ago = rfc3339(now_ms() - 14 * 24 * 60 * 60 * 1000);
266 let mut signals = vec![];
267 for row in active {
268 let limit = self.limit_of(&row.workspace).await?;
269 if limit.trust == Trust::Internal {
270 continue;
271 }
272 let this_month = row.this_month.unwrap_or(0).max(0);
273 let last_month = row.last_month.unwrap_or(0).max(0);
274 let record = self.record_row(&row.workspace).await?;
Billing lists every invoice and every staff change; signals carry follow-ups275 let (stage, owner, next_step, next_at) =
276 record.map_or((None, None, None, None), |r| (Some(r.stage), r.owner, r.next_step, r.next_at));
Two limits, real invoices, trust that grows by itself, sales signals277 let mut push = |kind: SignalKind, detail: String, value: i64| {
278 signals.push(Signal {
279 workspace: row.workspace.clone(),
280 kind,
281 detail,
282 value_micros: value,
283 stage: stage.clone(),
284 owner: owner.clone(),
Billing lists every invoice and every staff change; signals carry follow-ups285 next_step: next_step.clone(),
286 next_at: next_at.clone(),
Two limits, real invoices, trust that grows by itself, sales signals287 });
288 };
289 let declined = limit.message.as_deref().is_some_and(|m| m.contains("could not be charged"));
290 if declined {
291 push(SignalKind::Declined, limit.message.clone().unwrap_or_default(), limit.exposure_micros);
292 } else if limit.state == LimitState::Stopped {
293 push(SignalKind::AtLimit, limit.message.clone().unwrap_or_default(), limit.exposure_micros.max(limit.spent_micros));
294 } else if let Some(available) = limit.available_micros.filter(|a| *a > 0) {
295 if limit.exposure_micros * 5 >= available * 4 {
296 push(
297 SignalKind::NearCeiling,
298 format!(
299 "{} unpaid of the {} g1t allows it ({:?}); a call could raise it before it stops.",
300 dollars(limit.exposure_micros),
301 dollars(available),
302 limit.trust
303 ),
304 limit.exposure_micros,
305 );
306 }
307 }
308 if last_month >= HIGH_SPEND_MICROS {
309 let terms = self.terms_of(&row.workspace).await?;
310 if terms.kind == TermsKind::Standard && !limit.account.starts_with("ent_") {
311 push(
312 SignalKind::HighSpend,
313 format!("Spent {} last month on standard terms: worth offering custom terms or an enterprise.", dollars(last_month)),
314 last_month,
315 );
316 }
317 }
318 if growing(this_month, last_month, day, days_in(month)) {
319 push(
320 SignalKind::Growing,
321 format!(
322 "{} so far this month, on pace for about {}, against {} last month.",
323 dollars(this_month),
324 dollars((this_month as f64 * f64::from(days_in(month)) / f64::from(day.max(1))) as i64),
325 dollars(last_month)
326 ),
327 this_month,
328 );
329 }
330 if limit.trust == Trust::Established {
331 push(
332 SignalKind::Established,
333 format!(
334 "A steady customer: its limit now follows its spend ({} available).",
335 limit.available_micros.map(dollars).unwrap_or_default()
336 ),
337 last_month,
338 );
339 }
340 if row.first_paid.as_deref().is_some_and(|at| at >= fortnight_ago.as_str()) {
341 push(SignalKind::FirstPayment, "Paid g1t for the first time in the last two weeks: say hello.".to_owned(), this_month);
342 }
343 }
344 signals.sort_by(|a, b| urgency(a.kind).cmp(&urgency(b.kind)).then(b.value_micros.cmp(&a.value_micros)));
345 Ok(signals)
346 }
347
Billing lists every invoice and every staff change; signals carry follow-ups348 /// Every invoice g1t has sent, workspaces' and enterprises'.
349 pub(crate) async fn admin_invoices(&self, a: g1t_contracts::billing::AdminInvoicesArgs) -> Result<Vec<g1t_contracts::billing::InvoiceSummary>> {
350 #[derive(Deserialize)]
351 struct Row {
352 invoice_id: String,
353 kind: String,
354 account: String,
355 name: String,
356 reason: String,
357 period: String,
358 amount_micros: i64,
359 status: String,
360 hosted_url: Option<String>,
361 created_at: String,
362 paid_at: Option<String>,
363 }
364 let status = a.status.filter(|s| ["paid", "open", "failed", "overdue", "void"].contains(&s.as_str()));
365 let month = a.month.filter(|m| m.len() == 7 && m.chars().all(|c| c.is_ascii_digit() || c == '-'));
366 let rows = self
367 .db
368 .prepare(
369 "SELECT * FROM (
370 SELECT invoice_id, 'workspace' AS kind, workspace AS account, workspace AS name, reason, period,
371 amount_micros, status, hosted_url, created_at, paid_at
372 FROM workspace_invoices
373 UNION ALL
374 SELECT i.invoice_id, 'enterprise', i.account_id, COALESCE(b.name, i.account_id), 'enterprise', i.period,
375 i.amount_micros, i.status, i.hosted_url, i.created_at, i.paid_at
376 FROM enterprise_invoices i LEFT JOIN billing_accounts b ON b.id = i.account_id
377 )
378 WHERE (?1 IS NULL OR status = ?1) AND (?2 IS NULL OR substr(created_at, 1, 7) = ?2)
379 ORDER BY created_at DESC LIMIT 200",
380 )
381 .bind(&[crate::optional(status.as_deref()), crate::optional(month.as_deref())])?
382 .all()
383 .await?
384 .results::<Row>()?;
385 Ok(rows
386 .into_iter()
387 .map(|r| g1t_contracts::billing::InvoiceSummary {
388 invoice_id: r.invoice_id,
389 kind: r.kind,
390 account: r.account,
391 name: r.name,
392 reason: r.reason,
393 period: r.period,
394 amount_micros: r.amount_micros,
395 status: r.status,
396 hosted_url: r.hosted_url,
397 created_at: r.created_at,
398 paid_at: r.paid_at,
399 })
400 .collect())
401 }
402
403 /// Every change made in sudo and by Stripe, newest first.
404 pub(crate) async fn admin_audit(&self, a: g1t_contracts::billing::AdminAuditArgs) -> Result<Vec<g1t_contracts::billing::AdminAction>> {
405 #[derive(Deserialize)]
406 struct Row {
407 id: String,
408 account: String,
409 action: String,
410 detail: String,
411 by: String,
412 created_at: String,
413 }
414 let rows = self
415 .db
416 .prepare(
417 "SELECT id, account, action, detail, by, created_at FROM admin_actions
418 WHERE (?1 IS NULL OR by = ?1) AND (?2 IS NULL OR action = ?2) AND (?3 IS NULL OR created_at < ?3)
419 ORDER BY created_at DESC LIMIT 100",
420 )
421 .bind(&[
422 crate::optional(a.by.as_deref().map(str::trim).filter(|s| !s.is_empty())),
423 crate::optional(a.action.as_deref().map(str::trim).filter(|s| !s.is_empty())),
424 crate::optional(a.before.as_deref()),
425 ])?
426 .all()
427 .await?
428 .results::<Row>()?;
429 Ok(rows
430 .into_iter()
431 .map(|r| g1t_contracts::billing::AdminAction {
432 id: r.id,
433 account: r.account,
434 action: r.action,
435 detail: r.detail,
436 by: r.by,
437 created_at: r.created_at,
438 })
439 .collect())
440 }
441
Two limits, real invoices, trust that grows by itself, sales signals442 /// The business at a glance.
443 pub(crate) async fn admin_overview(&self, _: AdminOverviewArgs) -> Result<Overview> {
444 let now = rfc3339(now_ms());
445 let month = now[..7].to_owned();
446 let months = self.months_for(&[], 6).await?;
447 #[derive(Deserialize)]
448 struct KindRow {
449 kind: Option<String>,
450 charged: Option<i64>,
451 cost: Option<i64>,
452 }
453 let by_kind = self
454 .db
455 .prepare(
456 "SELECT CASE
457 WHEN task = 'sandbox' THEN 'Sandbox time'
458 WHEN task = 'deployments' THEN 'Deployments'
Team plan, an open-source pool, monthly trials and honest metering; the sidebar for everyone; a workspace that stays put459 WHEN task = 'security' THEN 'Security scans'
460 WHEN task = 'context' THEN 'Search embeddings'
461 WHEN task = 'storage' THEN 'Private storage'
Two limits, real invoices, trust that grows by itself, sales signals462 WHEN billed_to = 'workspace' THEN 'Own-provider runs'
463 ELSE 'Models' END AS kind,
464 -SUM(amount_micros) AS charged, SUM(cost_micros) AS cost
465 FROM ledger WHERE kind = 'usage' AND created_at >= ? GROUP BY 1 ORDER BY charged DESC",
466 )
467 .bind(&[format!("{month}-01").into()])?
468 .all()
469 .await?
470 .results::<KindRow>()?
471 .into_iter()
472 .map(|r| KindFigures {
473 kind: r.kind.unwrap_or_else(|| "Other".to_owned()),
474 charged_micros: r.charged.unwrap_or(0),
475 cost_micros: r.cost.unwrap_or(0),
476 })
477 .collect();
478 #[derive(Deserialize)]
479 struct Count {
480 n: Option<i64>,
481 }
482 let count = |sql: &'static str, args: Vec<JsValue>| {
483 let db = &self.db;
484 async move {
485 Ok::<i64, worker::Error>(db.prepare(sql).bind(&args)?.first::<Count>(None).await?.and_then(|c| c.n).unwrap_or(0))
486 }
487 };
488 let paying = count(
489 "SELECT COUNT(DISTINCT workspace) AS n FROM ledger
490 WHERE kind = 'top_up' AND amount_micros > 0 AND reference NOT LIKE 'crd%' AND created_at >= ?",
491 vec![rfc3339(now_ms() - 60 * 24 * 60 * 60 * 1000).into()],
492 )
493 .await?;
494 let open_invoices = count(
495 "SELECT (SELECT COALESCE(SUM(amount_micros), 0) FROM workspace_invoices WHERE status IN ('open', 'failed'))
496 + (SELECT COALESCE(SUM(amount_micros), 0) FROM enterprise_invoices WHERE status IN ('open', 'overdue')) AS n",
497 vec![],
498 )
499 .await?;
500 let follow_ups = count(
501 "SELECT COUNT(*) AS n FROM sales_records WHERE next_at IS NOT NULL AND next_at <= ? AND stage NOT IN ('won', 'lost')",
502 vec![now[..10].into()],
503 )
504 .await?;
505 let signals = self.admin_signals(AdminSignalsArgs {}).await?;
506 let tally = |kind: SignalKind| signals.iter().filter(|s| s.kind == kind).count() as u32;
507 Ok(Overview {
508 month,
509 months,
510 by_kind,
511 paying_workspaces: paying as u32,
512 stopped: tally(SignalKind::AtLimit),
513 near_ceiling: tally(SignalKind::NearCeiling),
514 declined: tally(SignalKind::Declined),
515 open_invoices_micros: open_invoices,
516 follow_ups_due: follow_ups as u32,
Team plan, an open-source pool, monthly trials and honest metering; the sidebar for everyone; a workspace that stays put517 pools: Some(self.pools().await?),
Two limits, real invoices, trust that grows by itself, sales signals518 })
519 }
520}
521
522#[cfg(test)]
523mod tests {
524 use super::*;
525
526 #[test]
527 fn six_months_end_with_this_one() {
528 assert_eq!(last_months("2026-02", 3), vec!["2025-12", "2026-01", "2026-02"]);
529 }
530
531 #[test]
532 fn growth_is_judged_on_pace_not_on_the_month_so_far() {
533 // Ten days in, $20 on a 30-day month is a $60 pace against $30.
534 assert!(growing(20_000_000, 30_000_000, 10, 30));
535 assert!(!growing(10_000_000, 30_000_000, 10, 30));
536 // Too small last month to say.
537 assert!(!growing(9_000_000, 1_000_000, 10, 30));
538 }
539
540 #[test]
541 fn the_most_urgent_comes_first() {
542 assert!(urgency(SignalKind::AtLimit) < urgency(SignalKind::NearCeiling));
543 assert!(urgency(SignalKind::Declined) < urgency(SignalKind::Growing));
544 }
545
546 #[test]
547 fn february_knows_its_leap_years() {
548 assert_eq!(days_in("2028-02"), 29);
549 assert_eq!(days_in("2026-02"), 28);
550 assert_eq!(days_in("2026-10"), 31);
551 }
552}