Sign in

flagon-io/g1t

public

Where people and agents ship software together. The open-source git platform for the whole job: issues, agents, checks and deploys to the edge.

gitagentscloudflarerustforgemcp

g1t/services/billing/src/invoices.rs

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