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/invoices.rs

359 lines15,513 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//! A workspace's invoices from g1t.
2//!
3//! Every time g1t charges a workspace's card, it is a real Stripe invoice:
4//! when each month closes, and when the workspace nears its limit mid-month
5//! (a threshold invoice, as Cloudflare and Fly do). Each is itemised by
6//! what was used since the last one, with any credit paid in advance taken
7//! off and anything left unpaid from before added, so its total is exactly
8//! what is owed. Stripe charges the card, emails the receipt, and keeps
9//! the invoice and its PDF in the workspace's billing page.
10
11use g1t_contracts::billing::{EntryKind, InvoiceItem, InvoicesArgs, WorkspaceInvoice};
12use g1t_contracts::time::rfc3339;
13use g1t_contracts::{FailureCode, Outcome};
14use g1t_kit::now_ms;
15use serde::Deserialize;
16use serde_json::Value;
17use worker::Result;
18
19use crate::Billing;
20
21/// The invoice's lines: what was used since the last one, by kind, then
22/// whatever makes the total what is owed.
23pub(crate) fn invoice_lines(used: &[(String, i64)], owed: i64) -> Vec<InvoiceItem> {
24 let mut lines: Vec<InvoiceItem> = used
25 .iter()
26 .filter(|(_, amount)| *amount > 0)
27 .map(|(kind, amount)| InvoiceItem { description: kind.clone(), amount_micros: *amount })
28 .collect();
29 let difference = owed - lines.iter().map(|l| l.amount_micros).sum::<i64>();
30 if difference < 0 {
31 lines.push(InvoiceItem { description: "Paid in advance".to_owned(), amount_micros: difference });
32 } else if difference > 0 {
33 lines.push(InvoiceItem { description: "Unpaid from earlier".to_owned(), amount_micros: difference });
34 }
35 lines
36}
37
38#[derive(Deserialize)]
39struct InvoiceRow {
40 invoice_id: String,
41 workspace: String,
42 reason: String,
43 period: String,
44 amount_micros: i64,
45 status: String,
46 hosted_url: Option<String>,
47 pdf_url: Option<String>,
48 created_at: String,
49}
50
51#[derive(Deserialize)]
52struct LineRow {
53 description: String,
54 amount_micros: i64,
55}
56
57/// A Stripe invoice, as far as billing reads it.
58#[derive(Deserialize)]
59struct StripeInvoice {
60 id: String,
61 #[serde(default)]
62 status: Option<String>,
63 #[serde(default)]
64 hosted_invoice_url: Option<String>,
65 #[serde(default)]
66 invoice_pdf: Option<String>,
67 #[serde(default)]
68 amount_paid: i64,
69 #[serde(default)]
70 charge: Option<String>,
71}
72
73impl Billing {
74 /// Invoices the workspace for what it owes, charging its card. `Ok(Err)`
75 /// says why not, when there was nothing to do or no card.
76 pub(crate) async fn invoice_workspace(
77 &self,
78 workspace: &str,
79 reason: &str,
80 period: &str,
81 ) -> Result<std::result::Result<WorkspaceInvoice, String>> {
82 let Some(stripe) = &self.stripe else { return Ok(Err("Payments are not set up.".into())) };
83 let Some(account) = self.row(workspace).await? else { return Ok(Err("Nothing billed yet.".into())) };
84 let Some(customer) = account.customer_id else { return Ok(Err("No card on file.".into())) };
85 let owed = (-account.balance_micros).max(0);
Team plan, an open-source pool, monthly trials and honest metering; the sidebar for everyone; a workspace that stays put86 // A month's close charges no less than the minimum
87 // (`MIN_CHARGE_MICROS`), so a payment's fee is never most of it;
88 // less carries over. A charge because a limit was reached always
89 // goes through, so a new workspace's small limit never strands it.
90 if reason == "month" && !crate::limits::worth_charging(owed, self.plans.min_charge_micros) {
91 return Ok(Err(format!(
92 "{} is owed, under the {} minimum charge; it carries over to the next invoice.",
93 crate::features::dollars(owed),
94 crate::features::dollars(self.plans.min_charge_micros)
95 )));
Two limits, real invoices, trust that grows by itself, sales signals96 }
97 // What was used since the last invoice, by kind.
98 #[derive(Deserialize)]
99 struct Last {
100 through_at: Option<String>,
101 }
102 let since = self
103 .db
104 .prepare("SELECT MAX(through_at) AS through_at FROM workspace_invoices WHERE workspace = ? AND status <> 'void'")
105 .bind(&[workspace.into()])?
106 .first::<Last>(None)
107 .await?
108 .and_then(|l| l.through_at)
109 .unwrap_or_default();
110 #[derive(Deserialize)]
111 struct Used {
112 kind: String,
113 charged: Option<i64>,
114 }
115 let now = rfc3339(now_ms());
116 let used: Vec<(String, i64)> = self
117 .db
118 .prepare(
119 "SELECT CASE
120 WHEN task = 'sandbox' THEN 'Sandbox time'
121 WHEN task = 'deployments' THEN 'Deployments: builds and usage past the plan'
Team plan, an open-source pool, monthly trials and honest metering; the sidebar for everyone; a workspace that stays put122 WHEN task = 'security' THEN 'Security scans'
123 WHEN task = 'context' THEN 'Search embeddings'
124 WHEN task = 'storage' THEN 'Private repository storage'
Two limits, real invoices, trust that grows by itself, sales signals125 WHEN billed_to = 'workspace' THEN 'Runs on your own model provider'
126 ELSE 'Agents on g1t''s models' END AS kind,
127 -SUM(amount_micros) AS charged
128 FROM ledger WHERE workspace = ? AND kind = 'usage' AND created_at > ? AND created_at <= ?
129 GROUP BY 1 ORDER BY charged DESC",
130 )
131 .bind(&[workspace.into(), since.as_str().into(), now.as_str().into()])?
132 .all()
133 .await?
134 .results::<Used>()?
135 .into_iter()
136 .map(|u| (u.kind, u.charged.unwrap_or(0)))
137 .collect();
138 let lines = invoice_lines(&used, owed);
139 let key = format!("ws-invoice/{workspace}/{reason}/{period}/{}", owed / 10_000);
140 for (position, line) in lines.iter().enumerate() {
141 let fields = [
142 ("customer", customer.clone()),
143 ("amount", (line.amount_micros / 10_000).to_string()),
144 ("currency", "usd".to_owned()),
145 ("description", line.description.clone()),
146 ("metadata[workspace]", workspace.to_owned()),
147 ];
148 let _: Value = stripe.post_idempotent("/invoiceitems", &fields, &format!("{key}/item/{position}")).await?;
149 }
150 let description = match reason {
151 "month" => format!("g1t usage for {workspace}, {period}"),
152 _ => format!("g1t usage for {workspace}, charged as it neared its limit"),
153 };
154 let fields = [
155 ("customer", customer.clone()),
156 ("collection_method", "charge_automatically".to_owned()),
157 ("auto_advance", "false".to_owned()),
158 ("pending_invoice_items_behavior", "include".to_owned()),
159 ("description", description),
160 ("metadata[g1t_workspace]", workspace.to_owned()),
161 ("metadata[reason]", reason.to_owned()),
162 ("metadata[period]", period.to_owned()),
163 ];
164 let draft: StripeInvoice = stripe.post_idempotent("/invoices", &fields, &key).await?;
165 // A retry finds it finalized already; that is fine.
166 let _ = stripe.post::<Value>(&format!("/invoices/{}/finalize", draft.id), &[]).await;
167 // Charge the card now; a decline comes back as an error.
168 let paid = stripe.post::<StripeInvoice>(&format!("/invoices/{}/pay", draft.id), &[("off_session", "true".to_owned())]).await;
169 let invoice: StripeInvoice = stripe.get(&format!("/invoices/{}", draft.id)).await?;
170 let total = lines.iter().map(|l| l.amount_micros).sum::<i64>();
171 let status = if invoice.status.as_deref() == Some("paid") { "paid" } else { "failed" };
172 let mut writes = vec![self
173 .db
174 .prepare(
175 "INSERT OR REPLACE INTO workspace_invoices
176 (invoice_id, workspace, reason, period, amount_micros, status, hosted_url, pdf_url, through_at, created_at, paid_at)
177 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
178 )
179 .bind(&[
180 invoice.id.as_str().into(),
181 workspace.into(),
182 reason.into(),
183 period.into(),
184 (total as f64).into(),
185 status.into(),
186 crate::optional(invoice.hosted_invoice_url.as_deref()),
187 crate::optional(invoice.invoice_pdf.as_deref()),
188 now.as_str().into(),
189 now.as_str().into(),
190 crate::optional((status == "paid").then_some(now.as_str())),
191 ])?];
192 for (position, line) in lines.iter().enumerate() {
193 writes.push(
194 self.db
195 .prepare("INSERT OR REPLACE INTO workspace_invoice_lines (invoice_id, position, description, amount_micros) VALUES (?, ?, ?, ?)")
196 .bind(&[invoice.id.as_str().into(), (position as u32).into(), line.description.as_str().into(), (line.amount_micros as f64).into()])?,
197 );
198 }
199 self.db.batch(writes).await?;
200 if status == "paid" {
201 self.credit_invoice(workspace, &invoice).await?;
202 } else {
203 let error = paid.err().map_or_else(|| "the card was declined".to_owned(), |e| e.to_string().chars().take(200).collect());
204 self.mark_declined(workspace, &error).await?;
205 }
206 Ok(Ok(WorkspaceInvoice {
207 invoice_id: invoice.id,
208 workspace: workspace.to_owned(),
209 reason: reason.to_owned(),
210 period: period.to_owned(),
211 amount_micros: total,
212 status: status.to_owned(),
213 hosted_url: invoice.hosted_invoice_url,
214 pdf_url: invoice.invoice_pdf,
215 lines,
216 created_at: now,
217 }))
218 }
219
220 /// Enters an invoice's payment once, with the kind of card that paid.
221 async fn credit_invoice(&self, workspace: &str, invoice: &StripeInvoice) -> Result<bool> {
222 let seen = self
223 .db
224 .prepare("SELECT id FROM ledger WHERE reference = ?")
225 .bind(&[invoice.id.as_str().into()])?
226 .first::<Value>(None)
227 .await?;
228 if seen.is_some() {
229 return Ok(false);
230 }
231 let amount = invoice.amount_paid * 10_000;
232 if amount <= 0 {
233 return Ok(false);
234 }
235 self.enter(workspace, EntryKind::TopUp, amount, &format!("Paid invoice {}", invoice.id), &invoice.id, None, None, None, None)
236 .await?;
237 // Prepaid cards pay, but never raise the limit.
238 if let (Some(stripe), Some(charge)) = (&self.stripe, &invoice.charge) {
239 if let Ok(charge) = stripe.get::<Value>(&format!("/charges/{charge}")).await {
240 if let Some(funding) = charge["payment_method_details"]["card"]["funding"].as_str() {
241 self.db
242 .prepare("UPDATE ledger SET funding = ? WHERE reference = ?")
243 .bind(&[funding.into(), invoice.id.as_str().into()])?
244 .run()
245 .await?;
246 }
247 }
248 }
249 Ok(true)
250 }
251
252 pub(crate) async fn mark_declined(&self, workspace: &str, error: &str) -> Result<()> {
253 let now = rfc3339(now_ms());
254 self.db
255 .prepare(
256 "INSERT INTO limits (workspace, autopay_failed_at, autopay_error, updated_at) VALUES (?1, ?2, ?3, ?2)
257 ON CONFLICT (workspace) DO UPDATE SET autopay_failed_at = ?2, autopay_error = ?3, updated_at = ?2",
258 )
259 .bind(&[workspace.into(), now.as_str().into(), error.into()])?
260 .run()
261 .await?;
262 Ok(())
263 }
264
265 /// A workspace invoice paid later, on Stripe's page or by a retry.
266 pub(crate) async fn workspace_invoice_paid(&self, invoice_id: &str) -> Result<Option<String>> {
267 #[derive(Deserialize)]
268 struct Row {
269 workspace: String,
270 }
271 let Some(row) = self
272 .db
273 .prepare("SELECT workspace FROM workspace_invoices WHERE invoice_id = ?")
274 .bind(&[invoice_id.into()])?
275 .first::<Row>(None)
276 .await?
277 else {
278 return Ok(None);
279 };
280 let Some(stripe) = &self.stripe else { return Ok(None) };
281 let invoice: StripeInvoice = stripe.get(&format!("/invoices/{invoice_id}")).await?;
282 self.db
283 .prepare("UPDATE workspace_invoices SET status = 'paid', paid_at = ? WHERE invoice_id = ?")
284 .bind(&[rfc3339(now_ms()).into(), invoice_id.into()])?
285 .run()
286 .await?;
287 let credited = self.credit_invoice(&row.workspace, &invoice).await?;
288 Ok(Some(format!(
289 "invoice {invoice_id} for {} paid{}",
290 row.workspace,
291 if credited { "" } else { " (already credited)" }
292 )))
293 }
294
295 pub(crate) async fn workspace_invoices(&self, workspace: &str) -> Result<Vec<WorkspaceInvoice>> {
296 let rows = self
297 .db
298 .prepare("SELECT * FROM workspace_invoices WHERE workspace = ? ORDER BY created_at DESC LIMIT 36")
299 .bind(&[workspace.into()])?
300 .all()
301 .await?
302 .results::<InvoiceRow>()?;
303 let mut invoices = vec![];
304 for row in rows {
305 let lines = self
306 .db
307 .prepare("SELECT description, amount_micros FROM workspace_invoice_lines WHERE invoice_id = ? ORDER BY position")
308 .bind(&[row.invoice_id.as_str().into()])?
309 .all()
310 .await?
311 .results::<LineRow>()?
312 .into_iter()
313 .map(|l| InvoiceItem { description: l.description, amount_micros: l.amount_micros })
314 .collect();
315 invoices.push(WorkspaceInvoice {
316 invoice_id: row.invoice_id,
317 workspace: row.workspace,
318 reason: row.reason,
319 period: row.period,
320 amount_micros: row.amount_micros,
321 status: row.status,
322 hosted_url: row.hosted_url,
323 pdf_url: row.pdf_url,
324 lines,
325 created_at: row.created_at,
326 });
327 }
328 Ok(invoices)
329 }
330
331 /// `invoices`: for the workspace's members.
332 pub(crate) async fn invoices(&self, a: InvoicesArgs) -> Result<Outcome<Vec<WorkspaceInvoice>>> {
333 let workspace = a.workspace.to_lowercase();
334 if !a.viewer.is_some_and(|viewer| viewer.is_member(&workspace)) {
335 return Ok(Outcome::fail(FailureCode::Forbidden, "Only members can see a workspace's invoices."));
336 }
337 Ok(Outcome::Ok(self.workspace_invoices(&workspace).await?))
338 }
339}
340
341#[cfg(test)]
342mod tests {
343 use super::*;
344
345 #[test]
346 fn an_invoice_adds_up_to_what_is_owed() {
347 let used = vec![("Agents on g1t's models".to_owned(), 40_000_000), ("Sandbox time".to_owned(), 10_000_000)];
348 // $10 of credit was paid in advance.
349 let lines = invoice_lines(&used, 40_000_000);
350 assert_eq!(lines.last().unwrap().description, "Paid in advance");
351 assert_eq!(lines.iter().map(|l| l.amount_micros).sum::<i64>(), 40_000_000);
352 // $5 was left unpaid from before.
353 let lines = invoice_lines(&used, 55_000_000);
354 assert_eq!(lines.last().unwrap().description, "Unpaid from earlier");
355 assert_eq!(lines.iter().map(|l| l.amount_micros).sum::<i64>(), 55_000_000);
356 // Exactly what was used.
357 assert_eq!(invoice_lines(&used, 50_000_000).len(), 2);
358 }
359}