| 1 | //! Provider cost reports: what an AI provider says the workspace's |
| 2 | //! organization spent, day by day, read from the provider itself. |
| 3 | //! |
| 4 | //! g1t meters only what runs through it. A team's other agents (CI |
| 5 | //! scripts, other tools) are on the provider's bill too, and the provider's |
| 6 | //! own report is the ground truth for all of it. A workspace owner connects |
| 7 | //! the organization's admin key; billing checks it with one small report |
| 8 | //! request, seals it with `BILLING_KEY` bound to the connection's id, and |
| 9 | //! never returns or logs it (only its last four characters are kept in |
| 10 | //! the clear, as its hint). |
| 11 | //! |
| 12 | //! - **Anthropic**: the Usage and Cost Admin API's |
| 13 | //! `GET /v1/organizations/cost_report`, with an Admin API key |
| 14 | //! (`x-api-key`, `anthropic-version: 2023-06-01`): daily buckets |
| 15 | //! (`bucket_width=1d`, at most 31 a page), grouped by `workspace_id` and |
| 16 | //! `description`, amounts as decimal strings of cents in USD, paged with |
| 17 | //! `page` = `next_page` while `has_more`. |
| 18 | //! - **OpenAI**: `GET /v1/organization/costs`, with an admin key as a |
| 19 | //! bearer token: daily buckets from `start_time` (Unix seconds, at most |
| 20 | //! 180 a page), grouped by `line_item` and `project_id`, amounts as |
| 21 | //! `amount.value` in dollars, paged the same way. |
| 22 | //! |
| 23 | //! Once a day (`keeper::DAILY`) every connection is read again: the last |
| 24 | //! [`REVISE_DAYS`] days, since providers revise recent days (further back |
| 25 | //! when reads have failed since), and [`BACKFILL_DAYS`] on its first read. |
| 26 | //! A read replaces the days it covers, so reading twice changes nothing. |
| 27 | //! A provider that fails records why on its connection and the others are |
| 28 | //! still read. |
| 29 | |
| 30 | use std::collections::{BTreeMap, BTreeSet}; |
| 31 | |
| 32 | use g1t_contracts::billing::{ |
| 33 | ConnectProviderArgs, CostProvider, DisconnectProviderArgs, ProviderConnection, ProviderConnectionsArgs, |
| 34 | ProviderCostDay, ProviderCosts, ProviderCostsArgs, |
| 35 | }; |
| 36 | use g1t_contracts::time::{parse_rfc3339, rfc3339}; |
| 37 | use g1t_contracts::{FailureCode, Outcome, PrincipalKind, Role, User, new_id}; |
| 38 | use g1t_kit::now_ms; |
| 39 | use serde::Deserialize; |
| 40 | use serde_json::Value; |
| 41 | use worker::{Fetch, Headers, Method, Request, RequestInit, Result}; |
| 42 | |
| 43 | use crate::Billing; |
| 44 | |
| 45 | const DAY_MS: u64 = 24 * 60 * 60 * 1000; |
| 46 | /// How far back the first read goes, today included. |
| 47 | pub(crate) const BACKFILL_DAYS: u64 = 90; |
| 48 | /// How many recent days every read asks for again, today included. |
| 49 | pub(crate) const REVISE_DAYS: u64 = 7; |
| 50 | /// The most pages read from one provider at once: 90 days are 3 of |
| 51 | /// Anthropic's and 1 of OpenAI's. |
| 52 | const MAX_PAGES: u32 = 12; |
| 53 | /// The most rows `provider_costs` returns; past it the answer is not complete. |
| 54 | const MAX_ROWS: usize = 20_000; |
| 55 | /// The longest key accepted. |
| 56 | const MAX_KEY: usize = 512; |
| 57 | /// Statements per D1 batch when writing a read. |
| 58 | const BATCH: usize = 50; |
| 59 | |
| 60 | const ANTHROPIC_COSTS: &str = "https://api.anthropic.com/v1/organizations/cost_report"; |
| 61 | const ANTHROPIC_VERSION: &str = "2023-06-01"; |
| 62 | const OPENAI_COSTS: &str = "https://api.openai.com/v1/organization/costs"; |
| 63 | /// As Anthropic asks of integrations. |
| 64 | const USER_AGENT: &str = "g1t/1.0 (https://g1t.sh)"; |
| 65 | |
| 66 | const NO_SEALER: &str = "Provider cost reports are not set up on this g1t: it has no key to keep secrets with."; |
| 67 | const MEMBERS_ONLY: &str = "Only members can see a workspace's provider costs."; |
| 68 | |
| 69 | /// What the list and every answer read: never the key. |
| 70 | const LIST_SQL: &str = "SELECT provider, hint, connected_by, connected_at, last_synced_at, last_error, synced_from |
| 71 | FROM provider_connections WHERE workspace = ? ORDER BY provider"; |
| 72 | |
| 73 | /// One read's days replaced: those the provider answered for, `?3` to `?4`. |
| 74 | pub(crate) const DELETE_DAYS: &str = |
| 75 | "DELETE FROM provider_costs WHERE workspace = ?1 AND provider = ?2 AND day >= ?3 AND day <= ?4"; |
| 76 | |
| 77 | /// One day's line item, written again if it is there already. |
| 78 | pub(crate) const UPSERT_COST: &str = |
| 79 | "INSERT INTO provider_costs (workspace, provider, day, item, scope, micros, currency, synced_at) |
| 80 | VALUES (?1, ?2, ?3, ?4, ?5, ?6, 'USD', ?7) |
| 81 | ON CONFLICT (workspace, provider, day, item, scope) DO UPDATE SET |
| 82 | micros = excluded.micros, currency = excluded.currency, synced_at = excluded.synced_at"; |
| 83 | |
| 84 | /// A read that worked: the first day asked for moves back, never forward. |
| 85 | pub(crate) const NOTE_SYNCED: &str = "UPDATE provider_connections SET last_synced_at = ?1, last_error = ?2, |
| 86 | synced_from = CASE WHEN synced_from IS NULL OR synced_from > ?3 THEN ?3 ELSE synced_from END |
| 87 | WHERE id = ?4"; |
| 88 | |
| 89 | /// A connection as the list reads it. |
| 90 | #[derive(Deserialize)] |
| 91 | struct ListedRow { |
| 92 | provider: String, |
| 93 | hint: String, |
| 94 | connected_by: String, |
| 95 | connected_at: String, |
| 96 | last_synced_at: Option<String>, |
| 97 | last_error: Option<String>, |
| 98 | synced_from: Option<String>, |
| 99 | } |
| 100 | |
| 101 | fn to_connection(row: ListedRow) -> Option<ProviderConnection> { |
| 102 | Some(ProviderConnection { |
| 103 | provider: CostProvider::parse(&row.provider)?, |
| 104 | hint: row.hint, |
| 105 | connected_by: row.connected_by, |
| 106 | connected_at: row.connected_at, |
| 107 | last_synced_at: row.last_synced_at, |
| 108 | last_error: row.last_error, |
| 109 | synced_from: row.synced_from, |
| 110 | }) |
| 111 | } |
| 112 | |
| 113 | /// A connection as a read needs it, sealed key and all. Never answered. |
| 114 | #[derive(Deserialize)] |
| 115 | struct SealedRow { |
| 116 | id: String, |
| 117 | workspace: String, |
| 118 | provider: String, |
| 119 | credential: String, |
| 120 | last_synced_at: Option<String>, |
| 121 | } |
| 122 | |
| 123 | /// Only an owner, as a person, connects or disconnects a provider: the |
| 124 | /// key reads the whole organization's spend. |
| 125 | pub(crate) fn owner_only<T>(actor: &User, workspace: &str) -> Option<Outcome<T>> { |
| 126 | (actor.role_in(workspace) != Some(Role::Owner) || actor.kind != PrincipalKind::User).then(|| { |
| 127 | Outcome::fail( |
| 128 | FailureCode::Forbidden, |
| 129 | "Only an owner of the workspace can connect or disconnect a provider.", |
| 130 | ) |
| 131 | }) |
| 132 | } |
| 133 | |
| 134 | /// `YYYY-MM-DD` of an instant, UTC. |
| 135 | pub(crate) fn day_of(ms: u64) -> String { |
| 136 | rfc3339(ms)[..10].to_owned() |
| 137 | } |
| 138 | |
| 139 | /// The first instant of a `YYYY-MM-DD` day, UTC. |
| 140 | pub(crate) fn day_start(day: &str) -> Option<u64> { |
| 141 | (day.len() == 10).then_some(())?; |
| 142 | parse_rfc3339(&format!("{day}T00:00:00Z")) |
| 143 | } |
| 144 | |
| 145 | /// What a read asks for: from the first day (`YYYY-MM-DD`) to the start |
| 146 | /// of tomorrow (exclusive, in ms). The last [`REVISE_DAYS`] days; back to |
| 147 | /// the day before the last good read when reads have failed since; all |
| 148 | /// [`BACKFILL_DAYS`] when there has been none. |
| 149 | pub(crate) fn window(now: u64, last_synced_at: Option<&str>) -> (String, u64) { |
| 150 | let today = now - now % DAY_MS; |
| 151 | let until = today + DAY_MS; |
| 152 | let earliest = today - (BACKFILL_DAYS - 1) * DAY_MS; |
| 153 | let from = match last_synced_at.and_then(parse_rfc3339) { |
| 154 | None => earliest, |
| 155 | Some(last) => { |
| 156 | let revise = today - (REVISE_DAYS - 1) * DAY_MS; |
| 157 | let missed = (last - last % DAY_MS).saturating_sub(DAY_MS); |
| 158 | revise.min(missed).max(earliest) |
| 159 | } |
| 160 | }; |
| 161 | (day_of(from), until) |
| 162 | } |
| 163 | |
| 164 | /// Percent-encodes a query value: a page token may hold `=`, `+` or `/`. |
| 165 | fn encode(value: &str) -> String { |
| 166 | value |
| 167 | .bytes() |
| 168 | .map(|b| match b { |
| 169 | b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'_' | b'.' | b'~' => (b as char).to_string(), |
| 170 | _ => format!("%{b:02X}"), |
| 171 | }) |
| 172 | .collect() |
| 173 | } |
| 174 | |
| 175 | /// Anthropic's cost report from `from` (`YYYY-MM-DD`) to `until`, by day, |
| 176 | /// grouped by its workspace and description. |
| 177 | pub(crate) fn anthropic_url(from: &str, until: u64, page: Option<&str>) -> String { |
| 178 | let mut url = format!( |
| 179 | "{ANTHROPIC_COSTS}?starting_at={from}T00:00:00Z&ending_at={}&bucket_width=1d&group_by%5B%5D=workspace_id&group_by%5B%5D=description&limit=31", |
| 180 | encode(&format!("{}Z", &rfc3339(until)[..19])) |
| 181 | ); |
| 182 | if let Some(page) = page { |
| 183 | url.push_str(&format!("&page={}", encode(page))); |
| 184 | } |
| 185 | url |
| 186 | } |
| 187 | |
| 188 | /// OpenAI's costs from `from` (`YYYY-MM-DD`) to `until`, by day, grouped by |
| 189 | /// line item and project. |
| 190 | pub(crate) fn openai_url(from: &str, until: u64, page: Option<&str>) -> String { |
| 191 | let start = day_start(from).unwrap_or(0) / 1000; |
| 192 | let mut url = format!( |
| 193 | "{OPENAI_COSTS}?start_time={start}&end_time={}&bucket_width=1d&group_by=line_item&group_by=project_id&limit=180", |
| 194 | until / 1000 |
| 195 | ); |
| 196 | if let Some(page) = page { |
| 197 | url.push_str(&format!("&page={}", encode(page))); |
| 198 | } |
| 199 | url |
| 200 | } |
| 201 | |
| 202 | /// The smallest request that proves a key can read the report: yesterday, |
| 203 | /// one bucket, ungrouped. |
| 204 | fn check_url(provider: CostProvider, now: u64) -> String { |
| 205 | let today = now - now % DAY_MS; |
| 206 | let yesterday = today - DAY_MS; |
| 207 | match provider { |
| 208 | CostProvider::Anthropic => format!( |
| 209 | "{ANTHROPIC_COSTS}?starting_at={}&ending_at={}&bucket_width=1d&limit=1", |
| 210 | encode(&format!("{}Z", &rfc3339(yesterday)[..19])), |
| 211 | encode(&format!("{}Z", &rfc3339(today)[..19])) |
| 212 | ), |
| 213 | CostProvider::Openai => format!( |
| 214 | "{OPENAI_COSTS}?start_time={}&end_time={}&bucket_width=1d&limit=1", |
| 215 | yesterday / 1000, |
| 216 | today / 1000 |
| 217 | ), |
| 218 | } |
| 219 | } |
| 220 | |
| 221 | /// One cost as the provider reported it. |
| 222 | #[derive(Clone, Debug, PartialEq)] |
| 223 | pub(crate) struct CostRow { |
| 224 | pub day: String, |
| 225 | /// Anthropic's description, OpenAI's line item; empty for none. |
| 226 | pub item: String, |
| 227 | /// The provider's own workspace or project; empty for its default. |
| 228 | pub scope: String, |
| 229 | pub micros: i64, |
| 230 | pub currency: String, |
| 231 | } |
| 232 | |
| 233 | /// One page of a report: the days it answered for (every bucket, costs or |
| 234 | /// not), its costs, and the next page's token. |
| 235 | #[derive(Debug, Default)] |
| 236 | pub(crate) struct Page { |
| 237 | pub days: Vec<String>, |
| 238 | pub rows: Vec<CostRow>, |
| 239 | pub next: Option<String>, |
| 240 | } |
| 241 | |
| 242 | /// A decimal string of cents (`"123.78912"`) in millionths of a dollar, |
| 243 | /// exactly, rounded half away from zero at the last place. |
| 244 | pub(crate) fn cents_to_micros(cents: &str) -> Option<i64> { |
| 245 | let text = cents.trim(); |
| 246 | let (negative, digits) = match text.strip_prefix('-') { |
| 247 | Some(rest) => (true, rest), |
| 248 | None => (false, text), |
| 249 | }; |
| 250 | let (whole, fraction) = digits.split_once('.').unwrap_or((digits, "")); |
| 251 | if whole.is_empty() && fraction.is_empty() { |
| 252 | return None; |
| 253 | } |
| 254 | if !whole.bytes().all(|b| b.is_ascii_digit()) || !fraction.bytes().all(|b| b.is_ascii_digit()) { |
| 255 | return None; |
| 256 | } |
| 257 | // A cent is 10,000 millionths: four places of the fraction, and the |
| 258 | // fifth to round on. |
| 259 | let whole: i64 = if whole.is_empty() { 0 } else { whole.parse().ok()? }; |
| 260 | let mut places: Vec<u8> = fraction.bytes().map(|b| b - b'0').collect(); |
| 261 | places.resize(5, 0); |
| 262 | let kept = places[..4].iter().fold(0i64, |n, d| n * 10 + i64::from(*d)); |
| 263 | let up = i64::from(places[4] >= 5); |
| 264 | let micros = whole.checked_mul(10_000)?.checked_add(kept + up)?; |
| 265 | Some(if negative { -micros } else { micros }) |
| 266 | } |
| 267 | |
| 268 | /// Dollars (`0.06`) in millionths of a dollar, rounded. |
| 269 | pub(crate) fn dollars_to_micros(dollars: f64) -> Option<i64> { |
| 270 | dollars.is_finite().then(|| (dollars * 1_000_000.0).round() as i64) |
| 271 | } |
| 272 | |
| 273 | fn text(value: &Value) -> String { |
| 274 | value.as_str().unwrap_or_default().to_owned() |
| 275 | } |
| 276 | |
| 277 | /// A page of Anthropic's cost report, as documented: `data` holds buckets |
| 278 | /// with `starting_at` and `results`; each result has `amount` (a decimal |
| 279 | /// string in the currency's lowest unit), `currency`, and, when grouped, |
| 280 | /// `description` and `workspace_id` (null for the default workspace). |
| 281 | pub(crate) fn parse_anthropic(body: &Value) -> std::result::Result<Page, String> { |
| 282 | let buckets = body["data"].as_array().ok_or("Anthropic's cost report had no data.")?; |
| 283 | let mut page = Page::default(); |
| 284 | for bucket in buckets { |
| 285 | let start = bucket["starting_at"] |
| 286 | .as_str() |
| 287 | .filter(|s| s.len() >= 10) |
| 288 | .ok_or("A bucket in Anthropic's cost report had no start.")?; |
| 289 | let day = start[..10].to_owned(); |
| 290 | for result in bucket["results"].as_array().into_iter().flatten() { |
| 291 | let amount = result["amount"] |
| 292 | .as_str() |
| 293 | .ok_or("A cost in Anthropic's report had no amount.")?; |
| 294 | let micros = cents_to_micros(amount) |
| 295 | .ok_or_else(|| format!("Anthropic reported an amount g1t can't read: {amount}"))?; |
| 296 | let item = [&result["description"], &result["model"], &result["cost_type"]] |
| 297 | .into_iter() |
| 298 | .find_map(|v| v.as_str().filter(|s| !s.is_empty())) |
| 299 | .unwrap_or_default() |
| 300 | .to_owned(); |
| 301 | page.rows.push(CostRow { |
| 302 | day: day.clone(), |
| 303 | item, |
| 304 | scope: text(&result["workspace_id"]), |
| 305 | micros, |
| 306 | currency: result["currency"].as_str().unwrap_or("USD").to_owned(), |
| 307 | }); |
| 308 | } |
| 309 | page.days.push(day); |
| 310 | } |
| 311 | if body["has_more"].as_bool() == Some(true) { |
| 312 | page.next = body["next_page"].as_str().filter(|s| !s.is_empty()).map(str::to_owned); |
| 313 | } |
| 314 | Ok(page) |
| 315 | } |
| 316 | |
| 317 | /// A page of OpenAI's costs, as its reference has it: `data` holds buckets |
| 318 | /// with `start_time` (Unix seconds) and their costs (`results` in the |
| 319 | /// reference's example, `result` in its schema); each has |
| 320 | /// `amount.value` (dollars) and `amount.currency` (`usd`), and, when |
| 321 | /// grouped, `line_item` and `project_id`. |
| 322 | pub(crate) fn parse_openai(body: &Value) -> std::result::Result<Page, String> { |
| 323 | let buckets = body["data"].as_array().ok_or("OpenAI's costs had no data.")?; |
| 324 | let mut page = Page::default(); |
| 325 | for bucket in buckets { |
| 326 | let start = bucket["start_time"] |
| 327 | .as_u64() |
| 328 | .ok_or("A bucket in OpenAI's costs had no start.")?; |
| 329 | let day = day_of(start * 1000); |
| 330 | let results = bucket["results"].as_array().or_else(|| bucket["result"].as_array()); |
| 331 | for result in results.into_iter().flatten() { |
| 332 | let value = result["amount"]["value"] |
| 333 | .as_f64() |
| 334 | .ok_or("A cost in OpenAI's report had no amount.")?; |
| 335 | let micros = dollars_to_micros(value).ok_or("OpenAI reported an amount g1t can't read.")?; |
| 336 | page.rows.push(CostRow { |
| 337 | day: day.clone(), |
| 338 | item: text(&result["line_item"]), |
| 339 | scope: text(&result["project_id"]), |
| 340 | micros, |
| 341 | currency: result["amount"]["currency"].as_str().unwrap_or("usd").to_owned(), |
| 342 | }); |
| 343 | } |
| 344 | page.days.push(day); |
| 345 | } |
| 346 | if body["has_more"].as_bool() == Some(true) { |
| 347 | page.next = body["next_page"].as_str().filter(|s| !s.is_empty()).map(str::to_owned); |
| 348 | } |
| 349 | Ok(page) |
| 350 | } |
| 351 | |
| 352 | /// The rows summed by day, item and scope (a description can repeat across |
| 353 | /// service tiers), and the currencies other than US dollars that were left |
| 354 | /// out. |
| 355 | pub(crate) fn aggregate(rows: &[CostRow]) -> (BTreeMap<(String, String, String), i64>, BTreeSet<String>) { |
| 356 | let mut sums = BTreeMap::new(); |
| 357 | let mut other = BTreeSet::new(); |
| 358 | for row in rows { |
| 359 | if !row.currency.eq_ignore_ascii_case("usd") { |
| 360 | other.insert(row.currency.to_uppercase()); |
| 361 | continue; |
| 362 | } |
| 363 | *sums |
| 364 | .entry((row.day.clone(), row.item.clone(), row.scope.clone())) |
| 365 | .or_insert(0i64) += row.micros; |
| 366 | } |
| 367 | (sums, other) |
| 368 | } |
| 369 | |
| 370 | /// Replaces every occurrence of the key with its hint. |
| 371 | pub(crate) fn scrub(text: &str, credential: &str) -> String { |
| 372 | if credential.len() < 8 { |
| 373 | return text.to_owned(); |
| 374 | } |
| 375 | text.replace(credential, &g1t_secrets::hint(credential)) |
| 376 | } |
| 377 | |
| 378 | /// A key that is plainly the wrong kind, before anything is sent: what to |
| 379 | /// use instead. |
| 380 | pub(crate) fn precheck(provider: CostProvider, credential: &str) -> std::result::Result<(), String> { |
| 381 | if credential.is_empty() { |
| 382 | return Err(format!("Paste {}'s admin key.", provider.label())); |
| 383 | } |
| 384 | if credential.len() > MAX_KEY || !credential.bytes().all(|b| b.is_ascii_graphic()) { |
| 385 | return Err("That isn't a key: a key is one line of letters, digits and dashes, with no spaces.".to_owned()); |
| 386 | } |
| 387 | match provider { |
| 388 | CostProvider::Anthropic if credential.starts_with("sk-ant-api") => Err( |
| 389 | "This is an Anthropic API key. The cost report needs an Admin API key (it starts sk-ant-admin), made in the Claude Console under Settings, Admin keys." |
| 390 | .to_owned(), |
| 391 | ), |
| 392 | CostProvider::Anthropic if credential.starts_with("sk-admin-") || credential.starts_with("sk-proj-") => { |
| 393 | Err("This looks like an OpenAI key. Connect it as OpenAI, or paste an Anthropic Admin API key (sk-ant-admin…).".to_owned()) |
| 394 | } |
| 395 | CostProvider::Openai if credential.starts_with("sk-proj-") || credential.starts_with("sk-svcacct-") => Err( |
| 396 | "This is a project API key. OpenAI's Costs API needs an admin key (it starts sk-admin-), made in the OpenAI platform under Organization settings, Admin keys." |
| 397 | .to_owned(), |
| 398 | ), |
| 399 | CostProvider::Openai if credential.starts_with("sk-ant-") => { |
| 400 | Err("This looks like an Anthropic key. Connect it as Anthropic, or paste an OpenAI admin key (sk-admin-…).".to_owned()) |
| 401 | } |
| 402 | _ => Ok(()), |
| 403 | } |
| 404 | } |
| 405 | |
| 406 | /// Why a provider refused a report request, in a sentence that says what |
| 407 | /// to do. The provider's own words are added where they help, with the key |
| 408 | /// scrubbed out of them. |
| 409 | pub(crate) fn explain(provider: CostProvider, status: u16, body: &Value, credential: &str) -> String { |
| 410 | let label = provider.label(); |
| 411 | let said = body["error"]["message"] |
| 412 | .as_str() |
| 413 | .map(|m| scrub(m, credential)) |
| 414 | .map(|m| m.chars().take(200).collect::<String>()); |
| 415 | let base = match (provider, status) { |
| 416 | (CostProvider::Anthropic, 401) => { |
| 417 | "Anthropic didn't accept this key. Use an Admin API key (sk-ant-admin…) from the Claude Console, under Settings, Admin keys.".to_owned() |
| 418 | } |
| 419 | (CostProvider::Anthropic, 403) => { |
| 420 | "This key can't read the organization's cost report. Only an Admin API key works: workspace API keys don't.".to_owned() |
| 421 | } |
| 422 | (CostProvider::Anthropic, 404) => { |
| 423 | "Anthropic has no cost report for this key. The Admin API needs an organization: individual accounts don't have one.".to_owned() |
| 424 | } |
| 425 | (CostProvider::Openai, 401) => { |
| 426 | "OpenAI didn't accept this key. Use an admin key from the OpenAI platform, under Organization settings, Admin keys.".to_owned() |
| 427 | } |
| 428 | (CostProvider::Openai, 403) => { |
| 429 | "This key can't read the organization's costs. Use an organization admin key, not a project or user API key.".to_owned() |
| 430 | } |
| 431 | (CostProvider::Openai, 404) => "OpenAI has no costs for this key. Use an organization admin key.".to_owned(), |
| 432 | (_, 429) => format!("{label} is limiting requests just now. g1t tries again at the next daily read."), |
| 433 | (_, 500..=599) => format!("{label} didn't answer ({status}). g1t tries again at the next daily read."), |
| 434 | _ => format!("{label} refused the request ({status})."), |
| 435 | }; |
| 436 | match said { |
| 437 | Some(said) if !said.is_empty() && status != 429 && status < 500 => format!("{base} {label} said: {said}"), |
| 438 | _ => base, |
| 439 | } |
| 440 | } |
| 441 | |
| 442 | /// Whether what the providers said covers every day from `since`. |
| 443 | pub(crate) fn complete(connections: &[ProviderConnection], since: &str) -> bool { |
| 444 | !connections.is_empty() |
| 445 | && connections.iter().all(|c| { |
| 446 | c.last_error.is_none() |
| 447 | && c.last_synced_at.is_some() |
| 448 | && c.synced_from.as_deref().is_some_and(|from| from <= since) |
| 449 | }) |
| 450 | } |
| 451 | |
| 452 | /// Asks a provider for one page of its report: the answer, or why not. |
| 453 | async fn get(provider: CostProvider, credential: &str, url: &str) -> std::result::Result<Value, String> { |
| 454 | let sent = async { |
| 455 | let headers = Headers::new(); |
| 456 | match provider { |
| 457 | CostProvider::Anthropic => { |
| 458 | headers.set("x-api-key", credential)?; |
| 459 | headers.set("anthropic-version", ANTHROPIC_VERSION)?; |
| 460 | } |
| 461 | CostProvider::Openai => headers.set("authorization", &format!("Bearer {credential}"))?, |
| 462 | } |
| 463 | headers.set("user-agent", USER_AGENT)?; |
| 464 | headers.set("accept", "application/json")?; |
| 465 | let mut init = RequestInit::new(); |
| 466 | init.with_method(Method::Get).with_headers(headers); |
| 467 | let mut response = Fetch::Request(Request::new_with_init(url, &init)?).send().await?; |
| 468 | let status = response.status_code(); |
| 469 | let body: Value = response.json().await.unwrap_or(Value::Null); |
| 470 | Ok::<_, worker::Error>((status, body)) |
| 471 | }; |
| 472 | match sent.await { |
| 473 | Ok((200, body)) => Ok(body), |
| 474 | Ok((status, body)) => Err(explain(provider, status, &body, credential)), |
| 475 | // Never the error itself: it could carry the request. |
| 476 | Err(_) => Err(format!( |
| 477 | "g1t couldn't reach {}. It tries again at the next daily read.", |
| 478 | provider.label() |
| 479 | )), |
| 480 | } |
| 481 | } |
| 482 | |
| 483 | /// Every page of a provider's report from `from` to `until`. |
| 484 | async fn read_report( |
| 485 | provider: CostProvider, |
| 486 | credential: &str, |
| 487 | from: &str, |
| 488 | until: u64, |
| 489 | ) -> std::result::Result<Page, String> { |
| 490 | let mut all = Page::default(); |
| 491 | let mut token: Option<String> = None; |
| 492 | for _ in 0..MAX_PAGES { |
| 493 | let url = match provider { |
| 494 | CostProvider::Anthropic => anthropic_url(from, until, token.as_deref()), |
| 495 | CostProvider::Openai => openai_url(from, until, token.as_deref()), |
| 496 | }; |
| 497 | let body = get(provider, credential, &url).await?; |
| 498 | let page = match provider { |
| 499 | CostProvider::Anthropic => parse_anthropic(&body)?, |
| 500 | CostProvider::Openai => parse_openai(&body)?, |
| 501 | }; |
| 502 | all.days.extend(page.days); |
| 503 | all.rows.extend(page.rows); |
| 504 | match page.next { |
| 505 | Some(next) => token = Some(next), |
| 506 | None => return Ok(all), |
| 507 | } |
| 508 | } |
| 509 | all.next = token; |
| 510 | Ok(all) |
| 511 | } |
| 512 | |
| 513 | impl Billing { |
| 514 | async fn listed(&self, workspace: &str) -> Result<Vec<ProviderConnection>> { |
| 515 | let rows = self |
| 516 | .db |
| 517 | .prepare(LIST_SQL) |
| 518 | .bind(&[workspace.into()])? |
| 519 | .all() |
| 520 | .await? |
| 521 | .results::<ListedRow>()?; |
| 522 | Ok(rows.into_iter().filter_map(to_connection).collect()) |
| 523 | } |
| 524 | |
| 525 | pub(crate) async fn provider_connections( |
| 526 | &self, |
| 527 | a: ProviderConnectionsArgs, |
| 528 | ) -> Result<Outcome<Vec<ProviderConnection>>> { |
| 529 | let workspace = a.workspace.to_lowercase(); |
| 530 | if !a.viewer.is_some_and(|viewer| viewer.is_member(&workspace)) { |
| 531 | return Ok(Outcome::fail(FailureCode::Forbidden, MEMBERS_ONLY)); |
| 532 | } |
| 533 | Ok(Outcome::Ok(self.listed(&workspace).await?)) |
| 534 | } |
| 535 | |
| 536 | pub(crate) async fn connect_provider(&self, a: ConnectProviderArgs) -> Result<Outcome<ProviderConnection>> { |
| 537 | let workspace = a.workspace.to_lowercase(); |
| 538 | if let Some(refused) = owner_only(&a.actor, &workspace) { |
| 539 | return Ok(refused); |
| 540 | } |
| 541 | let Some(sealer) = &self.sealer else { |
| 542 | return Ok(Outcome::fail(FailureCode::Conflict, NO_SEALER)); |
| 543 | }; |
| 544 | let provider = a.provider; |
| 545 | let credential = a.credential.trim(); |
| 546 | if let Err(problem) = precheck(provider, credential) { |
| 547 | return Ok(Outcome::fail(FailureCode::Invalid, problem)); |
| 548 | } |
| 549 | let now = now_ms(); |
| 550 | if let Err(problem) = get(provider, credential, &check_url(provider, now)).await { |
| 551 | return Ok(Outcome::fail(FailureCode::Invalid, problem)); |
| 552 | } |
| 553 | // A new id every time, and the key sealed to it: connecting again |
| 554 | // replaces the row's id with the key. The costs read with the old |
| 555 | // key may be another organization's, so they go. |
| 556 | let id = new_id("pcn", now); |
| 557 | let sealed = sealer.seal(credential, &id); |
| 558 | self.db |
| 559 | .batch(vec![ |
| 560 | self.db |
| 561 | .prepare("DELETE FROM provider_costs WHERE workspace = ?1 AND provider = ?2") |
| 562 | .bind(&[workspace.as_str().into(), provider.name().into()])?, |
| 563 | self.db |
| 564 | .prepare( |
| 565 | "INSERT INTO provider_connections (id, workspace, provider, credential, hint, connected_by, connected_at) |
| 566 | VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7) |
| 567 | ON CONFLICT (workspace, provider) DO UPDATE SET |
| 568 | id = excluded.id, credential = excluded.credential, hint = excluded.hint, |
| 569 | connected_by = excluded.connected_by, connected_at = excluded.connected_at, |
| 570 | last_synced_at = NULL, last_error = NULL, synced_from = NULL", |
| 571 | ) |
| 572 | .bind(&[ |
| 573 | id.as_str().into(), |
| 574 | workspace.as_str().into(), |
| 575 | provider.name().into(), |
| 576 | sealed.as_str().into(), |
| 577 | g1t_secrets::hint(credential).into(), |
| 578 | a.actor.username.as_str().into(), |
| 579 | rfc3339(now).into(), |
| 580 | ])?, |
| 581 | ]) |
| 582 | .await?; |
| 583 | // The first 90 days at once, so the page has something to show; |
| 584 | // a failure is kept on the connection, which stays. |
| 585 | let row = SealedRow { |
| 586 | id, |
| 587 | workspace: workspace.clone(), |
| 588 | provider: provider.name().to_owned(), |
| 589 | credential: sealed, |
| 590 | last_synced_at: None, |
| 591 | }; |
| 592 | if let Err(error) = self.sync_connection(&row, now).await { |
| 593 | worker::console_error!( |
| 594 | "provider costs: {workspace} {}: writing the first read failed: {error}", |
| 595 | provider.name() |
| 596 | ); |
| 597 | } |
| 598 | let listed = self.listed(&workspace).await?; |
| 599 | Ok(match listed.into_iter().find(|c| c.provider == provider) { |
| 600 | Some(connection) => Outcome::Ok(connection), |
| 601 | None => Outcome::fail(FailureCode::NotFound, "The connection was not saved."), |
| 602 | }) |
| 603 | } |
| 604 | |
| 605 | pub(crate) async fn disconnect_provider(&self, a: DisconnectProviderArgs) -> Result<Outcome<bool>> { |
| 606 | let workspace = a.workspace.to_lowercase(); |
| 607 | if let Some(refused) = owner_only(&a.actor, &workspace) { |
| 608 | return Ok(refused); |
| 609 | } |
| 610 | let was = self.listed(&workspace).await?.iter().any(|c| c.provider == a.provider); |
| 611 | let bind = |
| 612 | || -> Vec<worker::wasm_bindgen::JsValue> { vec![workspace.as_str().into(), a.provider.name().into()] }; |
| 613 | self.db |
| 614 | .batch(vec![ |
| 615 | self.db |
| 616 | .prepare("DELETE FROM provider_connections WHERE workspace = ?1 AND provider = ?2") |
| 617 | .bind(&bind())?, |
| 618 | self.db |
| 619 | .prepare("DELETE FROM provider_costs WHERE workspace = ?1 AND provider = ?2") |
| 620 | .bind(&bind())?, |
| 621 | ]) |
| 622 | .await?; |
| 623 | Ok(Outcome::Ok(was)) |
| 624 | } |
| 625 | |
| 626 | pub(crate) async fn provider_costs(&self, a: ProviderCostsArgs) -> Result<Outcome<ProviderCosts>> { |
| 627 | let workspace = a.workspace.to_lowercase(); |
| 628 | if !a.viewer.is_some_and(|viewer| viewer.is_member(&workspace)) { |
| 629 | return Ok(Outcome::fail(FailureCode::Forbidden, MEMBERS_ONLY)); |
| 630 | } |
| 631 | let since = a.since.get(..10).unwrap_or_default().to_owned(); |
| 632 | if day_start(&since).is_none() { |
| 633 | return Ok(Outcome::fail(FailureCode::Invalid, "Give since as a day, YYYY-MM-DD.")); |
| 634 | } |
| 635 | #[derive(Deserialize)] |
| 636 | struct DayRow { |
| 637 | day: String, |
| 638 | provider: String, |
| 639 | item: String, |
| 640 | micros: Option<f64>, |
| 641 | } |
| 642 | let connections = self.listed(&workspace).await?; |
| 643 | let rows = self |
| 644 | .db |
| 645 | .prepare( |
| 646 | "SELECT day, provider, item, SUM(micros) AS micros FROM provider_costs |
| 647 | WHERE workspace = ?1 AND day >= ?2 |
| 648 | GROUP BY day, provider, item ORDER BY day, provider, item LIMIT ?3", |
| 649 | ) |
| 650 | .bind(&[ |
| 651 | workspace.as_str().into(), |
| 652 | since.as_str().into(), |
| 653 | ((MAX_ROWS + 1) as f64).into(), |
| 654 | ])? |
| 655 | .all() |
| 656 | .await? |
| 657 | .results::<DayRow>()?; |
| 658 | let cut = rows.len() > MAX_ROWS; |
| 659 | let days = rows |
| 660 | .into_iter() |
| 661 | .take(MAX_ROWS) |
| 662 | .filter_map(|row| { |
| 663 | Some(ProviderCostDay { |
| 664 | day: row.day, |
| 665 | provider: CostProvider::parse(&row.provider)?, |
| 666 | item: row.item, |
| 667 | micros: row.micros.unwrap_or(0.0) as i64, |
| 668 | }) |
| 669 | }) |
| 670 | .collect(); |
| 671 | let complete = !cut && complete(&connections, &since); |
| 672 | Ok(Outcome::Ok(ProviderCosts { |
| 673 | connections, |
| 674 | days, |
| 675 | complete, |
| 676 | })) |
| 677 | } |
| 678 | |
| 679 | /// Reads one connection's report and keeps it. A provider's refusal is |
| 680 | /// kept on the connection; only a failure to write is an error. |
| 681 | async fn sync_connection(&self, row: &SealedRow, now: u64) -> Result<std::result::Result<usize, String>> { |
| 682 | let Some(provider) = CostProvider::parse(&row.provider) else { |
| 683 | return Ok(Err("Unknown provider.".to_owned())); |
| 684 | }; |
| 685 | let problem = |message: String| async move { |
| 686 | self.db |
| 687 | .prepare("UPDATE provider_connections SET last_error = ?1 WHERE id = ?2") |
| 688 | .bind(&[message.as_str().into(), row.id.as_str().into()])? |
| 689 | .run() |
| 690 | .await?; |
| 691 | Ok::<_, worker::Error>(Err(message)) |
| 692 | }; |
| 693 | let Some(sealer) = &self.sealer else { |
| 694 | return problem(NO_SEALER.to_owned()).await; |
| 695 | }; |
| 696 | let Some(credential) = sealer.open(&row.credential, &row.id) else { |
| 697 | return problem("g1t can't open the saved key. Connect the provider again.".to_owned()).await; |
| 698 | }; |
| 699 | let (from, until) = window(now, row.last_synced_at.as_deref()); |
| 700 | let page = match read_report(provider, &credential, &from, until).await { |
| 701 | Ok(page) => page, |
| 702 | Err(message) => return problem(message).await, |
| 703 | }; |
| 704 | let (sums, other) = aggregate(&page.rows); |
| 705 | let at = rfc3339(now); |
| 706 | let days: BTreeSet<&String> = page.days.iter().collect(); |
| 707 | let mut statements = vec![]; |
| 708 | if let (Some(first), Some(last)) = (days.first(), days.last()) { |
| 709 | statements.push(self.db.prepare(DELETE_DAYS).bind(&[ |
| 710 | row.workspace.as_str().into(), |
| 711 | provider.name().into(), |
| 712 | first.as_str().into(), |
| 713 | last.as_str().into(), |
| 714 | ])?); |
| 715 | } |
| 716 | for ((day, item, scope), micros) in &sums { |
| 717 | statements.push(self.db.prepare(UPSERT_COST).bind(&[ |
| 718 | row.workspace.as_str().into(), |
| 719 | provider.name().into(), |
| 720 | day.as_str().into(), |
| 721 | item.as_str().into(), |
| 722 | scope.as_str().into(), |
| 723 | (*micros as f64).into(), |
| 724 | at.as_str().into(), |
| 725 | ])?); |
| 726 | } |
| 727 | let written = sums.len(); |
| 728 | let mut statements = statements.into_iter().peekable(); |
| 729 | while statements.peek().is_some() { |
| 730 | self.db.batch(statements.by_ref().take(BATCH).collect()).await?; |
| 731 | } |
| 732 | let note = if page.next.is_some() { |
| 733 | Some(format!( |
| 734 | "{} had more pages than g1t reads at once; the earliest days may be missing.", |
| 735 | provider.label() |
| 736 | )) |
| 737 | } else if !other.is_empty() { |
| 738 | let currencies: Vec<String> = other.into_iter().collect(); |
| 739 | Some(format!( |
| 740 | "Some costs were reported in {}, not US dollars, and are left out.", |
| 741 | currencies.join(", ") |
| 742 | )) |
| 743 | } else { |
| 744 | None |
| 745 | }; |
| 746 | // Reaching back further than before only once the provider said so. |
| 747 | let from = days |
| 748 | .first() |
| 749 | .map_or(from.clone(), |first| (*first).clone().min(from.clone())); |
| 750 | self.db |
| 751 | .prepare(NOTE_SYNCED) |
| 752 | .bind(&[ |
| 753 | at.as_str().into(), |
| 754 | crate::optional(note.as_deref()), |
| 755 | from.as_str().into(), |
| 756 | row.id.as_str().into(), |
| 757 | ])? |
| 758 | .run() |
| 759 | .await?; |
| 760 | Ok(match note { |
| 761 | Some(note) => Err(note), |
| 762 | None => Ok(written), |
| 763 | }) |
| 764 | } |
| 765 | |
| 766 | /// Every connection's report, read again: once a day. Each provider on |
| 767 | /// its own, one at a time; one failing never stops the rest. Returns |
| 768 | /// how many were read and how many failed. |
| 769 | pub(crate) async fn sync_provider_costs(&self) -> Result<(u32, u32)> { |
| 770 | let rows = self |
| 771 | .db |
| 772 | .prepare( |
| 773 | "SELECT id, workspace, provider, credential, last_synced_at FROM provider_connections |
| 774 | ORDER BY last_synced_at", |
| 775 | ) |
| 776 | .all() |
| 777 | .await? |
| 778 | .results::<SealedRow>()?; |
| 779 | let (mut read, mut failed) = (0, 0); |
| 780 | let now = now_ms(); |
| 781 | for row in &rows { |
| 782 | match self.sync_connection(row, now).await { |
| 783 | Ok(Ok(_)) => read += 1, |
| 784 | Ok(Err(message)) => { |
| 785 | failed += 1; |
| 786 | worker::console_error!("provider costs: {} {}: {message}", row.workspace, row.provider); |
| 787 | } |
| 788 | Err(error) => { |
| 789 | failed += 1; |
| 790 | worker::console_error!( |
| 791 | "provider costs: {} {}: writing failed: {error}", |
| 792 | row.workspace, |
| 793 | row.provider |
| 794 | ); |
| 795 | } |
| 796 | } |
| 797 | } |
| 798 | Ok((read, failed)) |
| 799 | } |
| 800 | } |
| 801 | |
| 802 | #[cfg(test)] |
| 803 | mod tests { |
| 804 | use super::*; |
| 805 | use serde_json::json; |
| 806 | |
| 807 | /// The response example in Anthropic's Get Cost Report reference |
| 808 | /// (platform.claude.com/docs/en/api/beta/organization/cost_report/retrieve). |
| 809 | const ANTHROPIC_EXAMPLE: &str = r#"{ |
| 810 | "data": [ |
| 811 | { |
| 812 | "ending_at": "2025-08-02T00:00:00Z", |
| 813 | "results": [ |
| 814 | { |
| 815 | "amount": "123.78912", |
| 816 | "context_window": "0-200k", |
| 817 | "cost_type": "tokens", |
| 818 | "currency": "USD", |
| 819 | "description": "Claude Opus 5.5 Usage - Input Tokens", |
| 820 | "inference_geo": "global", |
| 821 | "model": "claude-opus-5-5", |
| 822 | "service_tier": "standard", |
| 823 | "speed": "standard", |
| 824 | "token_type": "cache_creation.ephemeral_1h_input_tokens", |
| 825 | "workspace_id": "wrkspc_01JwQvzr7rXLA5AGx3HKfFUJ" |
| 826 | } |
| 827 | ], |
| 828 | "starting_at": "2025-08-01T00:00:00Z" |
| 829 | } |
| 830 | ], |
| 831 | "has_more": true, |
| 832 | "next_page": "page_MjAyNS0wNS0xNFQwMDowMDowMFo=" |
| 833 | }"#; |
| 834 | |
| 835 | /// The response example for GET /v1/organization/costs in OpenAI's |
| 836 | /// OpenAPI description (github.com/openai/openai-openapi). |
| 837 | const OPENAI_EXAMPLE: &str = r#"{ |
| 838 | "object": "page", |
| 839 | "data": [ |
| 840 | { |
| 841 | "object": "bucket", |
| 842 | "start_time": 1730419200, |
| 843 | "end_time": 1730505600, |
| 844 | "results": [ |
| 845 | { |
| 846 | "object": "organization.costs.result", |
| 847 | "amount": { |
| 848 | "value": 0.06, |
| 849 | "currency": "usd" |
| 850 | }, |
| 851 | "line_item": null, |
| 852 | "project_id": null |
| 853 | } |
| 854 | ] |
| 855 | } |
| 856 | ], |
| 857 | "has_more": false, |
| 858 | "next_page": null |
| 859 | }"#; |
| 860 | |
| 861 | fn user(role: &str, kind: &str) -> User { |
| 862 | serde_json::from_value(json!({ |
| 863 | "id": "usr_1", "username": "ada", "kind": kind, "verified": true, |
| 864 | "workspaces": [{ "slug": "acme", "role": role }], |
| 865 | })) |
| 866 | .unwrap() |
| 867 | } |
| 868 | |
| 869 | #[test] |
| 870 | fn anthropics_documented_report_reads_as_cents_by_day() { |
| 871 | let page = parse_anthropic(&serde_json::from_str(ANTHROPIC_EXAMPLE).unwrap()).unwrap(); |
| 872 | assert_eq!(page.days, ["2025-08-01"]); |
| 873 | assert_eq!( |
| 874 | page.rows, |
| 875 | [CostRow { |
| 876 | day: "2025-08-01".into(), |
| 877 | item: "Claude Opus 5.5 Usage - Input Tokens".into(), |
| 878 | scope: "wrkspc_01JwQvzr7rXLA5AGx3HKfFUJ".into(), |
| 879 | // 123.78912 cents is $1.2378912. |
| 880 | micros: 1_237_891, |
| 881 | currency: "USD".into(), |
| 882 | }] |
| 883 | ); |
| 884 | assert_eq!(page.next.as_deref(), Some("page_MjAyNS0wNS0xNFQwMDowMDowMFo=")); |
| 885 | // The default workspace is null, and the last page has no token. |
| 886 | let last = parse_anthropic(&json!({ |
| 887 | "data": [{ "starting_at": "2025-08-02T00:00:00Z", "ending_at": "2025-08-03T00:00:00Z", "results": [] }, |
| 888 | { "starting_at": "2025-08-03T00:00:00Z", "ending_at": "2025-08-04T00:00:00Z", |
| 889 | "results": [{ "amount": "50", "currency": "USD", "description": null, "workspace_id": null, "cost_type": "web_search" }] }], |
| 890 | "has_more": false, "next_page": null, |
| 891 | })) |
| 892 | .unwrap(); |
| 893 | assert_eq!(last.days, ["2025-08-02", "2025-08-03"]); |
| 894 | assert_eq!(last.rows[0].scope, ""); |
| 895 | assert_eq!(last.rows[0].item, "web_search"); |
| 896 | assert_eq!(last.rows[0].micros, 500_000); |
| 897 | assert_eq!(last.next, None); |
| 898 | assert!(parse_anthropic(&json!({ "error": { "type": "authentication_error" } })).is_err()); |
| 899 | } |
| 900 | |
| 901 | #[test] |
| 902 | fn openais_documented_costs_read_as_dollars_by_day() { |
| 903 | let page = parse_openai(&serde_json::from_str(OPENAI_EXAMPLE).unwrap()).unwrap(); |
| 904 | // 1730419200 is 2024-11-01 00:00 UTC. |
| 905 | assert_eq!(page.days, ["2024-11-01"]); |
| 906 | assert_eq!( |
| 907 | page.rows, |
| 908 | [CostRow { |
| 909 | day: "2024-11-01".into(), |
| 910 | item: "".into(), |
| 911 | scope: "".into(), |
| 912 | micros: 60_000, |
| 913 | currency: "usd".into() |
| 914 | }] |
| 915 | ); |
| 916 | assert_eq!(page.next, None); |
| 917 | // The schema names the list `result`; grouped, a line item and a project. |
| 918 | let grouped = parse_openai(&json!({ |
| 919 | "object": "page", |
| 920 | "data": [{ "object": "bucket", "start_time": 1730419200, "end_time": 1730505600, |
| 921 | "result": [{ "object": "organization.costs.result", "amount": { "value": 1.5, "currency": "usd" }, |
| 922 | "line_item": "Image models", "project_id": "proj_abc" }] }], |
| 923 | "has_more": true, "next_page": "page_2", |
| 924 | })) |
| 925 | .unwrap(); |
| 926 | assert_eq!(grouped.rows[0].item, "Image models"); |
| 927 | assert_eq!(grouped.rows[0].scope, "proj_abc"); |
| 928 | assert_eq!(grouped.rows[0].micros, 1_500_000); |
| 929 | assert_eq!(grouped.next.as_deref(), Some("page_2")); |
| 930 | } |
| 931 | |
| 932 | #[test] |
| 933 | fn cents_become_millionths_exactly() { |
| 934 | assert_eq!(cents_to_micros("123.45"), Some(1_234_500)); |
| 935 | assert_eq!(cents_to_micros("0.00005"), Some(1)); |
| 936 | assert_eq!(cents_to_micros("0.00004"), Some(0)); |
| 937 | assert_eq!(cents_to_micros("7"), Some(70_000)); |
| 938 | assert_eq!(cents_to_micros(".5"), Some(5_000)); |
| 939 | assert_eq!(cents_to_micros("-12.5"), Some(-125_000)); |
| 940 | assert_eq!(cents_to_micros("1e3"), None); |
| 941 | assert_eq!(cents_to_micros(""), None); |
| 942 | assert_eq!(dollars_to_micros(0.06), Some(60_000)); |
| 943 | assert_eq!(dollars_to_micros(f64::NAN), None); |
| 944 | } |
| 945 | |
| 946 | #[test] |
| 947 | fn costs_are_summed_by_day_item_and_scope_in_dollars_only() { |
| 948 | let row = |item: &str, micros: i64, currency: &str| CostRow { |
| 949 | day: "2026-10-01".into(), |
| 950 | item: item.into(), |
| 951 | scope: "".into(), |
| 952 | micros, |
| 953 | currency: currency.into(), |
| 954 | }; |
| 955 | let (sums, other) = aggregate(&[ |
| 956 | row("Opus input", 10, "USD"), |
| 957 | row("Opus input", 5, "USD"), |
| 958 | row("Opus output", 7, "usd"), |
| 959 | row("x", 9, "eur"), |
| 960 | ]); |
| 961 | assert_eq!(sums.len(), 2); |
| 962 | assert_eq!(sums[&("2026-10-01".into(), "Opus input".into(), "".into())], 15); |
| 963 | assert_eq!(other.into_iter().collect::<Vec<_>>(), ["EUR"]); |
| 964 | } |
| 965 | |
| 966 | #[test] |
| 967 | fn a_read_asks_for_recent_days_again_and_ninety_the_first_time() { |
| 968 | let now = parse_rfc3339("2026-10-18T09:30:00.000Z").unwrap(); |
| 969 | let (from, until) = window(now, None); |
| 970 | assert_eq!(from, "2026-07-21"); |
| 971 | assert_eq!(rfc3339(until), "2026-10-19T00:00:00.000Z"); |
| 972 | assert_eq!(window(now, Some("2026-10-17T04:17:00.000Z")).0, "2026-10-12"); |
| 973 | // Reads failing since the 1st: from the day before it. |
| 974 | assert_eq!(window(now, Some("2026-10-01T04:17:00.000Z")).0, "2026-09-30"); |
| 975 | // Never more than 90 days. |
| 976 | assert_eq!(window(now, Some("2025-01-01T00:00:00.000Z")).0, "2026-07-21"); |
| 977 | } |
| 978 | |
| 979 | #[test] |
| 980 | fn report_requests_follow_the_documented_parameters() { |
| 981 | let until = parse_rfc3339("2026-10-19T00:00:00.000Z").unwrap(); |
| 982 | assert_eq!( |
| 983 | anthropic_url("2026-10-12", until, None), |
| 984 | "https://api.anthropic.com/v1/organizations/cost_report?starting_at=2026-10-12T00:00:00Z&ending_at=2026-10-19T00%3A00%3A00Z&bucket_width=1d&group_by%5B%5D=workspace_id&group_by%5B%5D=description&limit=31" |
| 985 | ); |
| 986 | assert!(anthropic_url("2026-10-12", until, Some("page_a=")).ends_with("&page=page_a%3D")); |
| 987 | assert_eq!( |
| 988 | openai_url("2024-11-01", until, None), |
| 989 | "https://api.openai.com/v1/organization/costs?start_time=1730419200&end_time=1792368000&bucket_width=1d&group_by=line_item&group_by=project_id&limit=180" |
| 990 | ); |
| 991 | let now = parse_rfc3339("2026-10-18T09:30:00.000Z").unwrap(); |
| 992 | assert!( |
| 993 | check_url(CostProvider::Anthropic, now) |
| 994 | .contains("starting_at=2026-10-17T00%3A00%3A00Z&ending_at=2026-10-18T00%3A00%3A00Z") |
| 995 | ); |
| 996 | assert!(check_url(CostProvider::Openai, now).ends_with("&limit=1")); |
| 997 | } |
| 998 | |
| 999 | #[test] |
| 1000 | fn the_wrong_kind_of_key_is_named_before_anything_is_sent() { |
| 1001 | assert!( |
| 1002 | precheck(CostProvider::Anthropic, "sk-ant-api03-abcd") |
| 1003 | .unwrap_err() |
| 1004 | .contains("Admin API key") |
| 1005 | ); |
| 1006 | assert!(precheck(CostProvider::Anthropic, "sk-ant-admin01-abcd").is_ok()); |
| 1007 | assert!( |
| 1008 | precheck(CostProvider::Openai, "sk-proj-abcd") |
| 1009 | .unwrap_err() |
| 1010 | .contains("admin key") |
| 1011 | ); |
| 1012 | assert!( |
| 1013 | precheck(CostProvider::Openai, "sk-ant-admin01-abcd") |
| 1014 | .unwrap_err() |
| 1015 | .contains("Anthropic") |
| 1016 | ); |
| 1017 | assert!(precheck(CostProvider::Openai, "sk-admin-abcd").is_ok()); |
| 1018 | assert!(precheck(CostProvider::Openai, "").is_err()); |
| 1019 | assert!(precheck(CostProvider::Openai, "sk-admin abcd").is_err()); |
| 1020 | } |
| 1021 | |
| 1022 | #[test] |
| 1023 | fn a_refusal_says_why_and_never_repeats_the_key() { |
| 1024 | let key = "sk-ant-admin01-SECRETSECRETwxyz"; |
| 1025 | let body = |
| 1026 | json!({ "error": { "type": "authentication_error", "message": format!("invalid x-api-key {key}") } }); |
| 1027 | let said = explain(CostProvider::Anthropic, 401, &body, key); |
| 1028 | assert!( |
| 1029 | said.starts_with("Anthropic didn't accept this key. Use an Admin API key"), |
| 1030 | "{said}" |
| 1031 | ); |
| 1032 | assert!(!said.contains("SECRET"), "{said}"); |
| 1033 | assert!(said.contains("…wxyz"), "{said}"); |
| 1034 | assert!(explain(CostProvider::Openai, 403, &Value::Null, key).contains("not a project or user API key")); |
| 1035 | assert!(explain(CostProvider::Openai, 429, &body, key).contains("tries again")); |
| 1036 | assert!(!explain(CostProvider::Openai, 503, &body, key).contains("SECRET")); |
| 1037 | } |
| 1038 | |
| 1039 | #[test] |
| 1040 | fn only_an_owner_as_a_person_connects_a_provider() { |
| 1041 | assert!(owner_only::<()>(&user("owner", "user"), "acme").is_none()); |
| 1042 | assert!( |
| 1043 | matches!(owner_only::<()>(&user("member", "user"), "acme"), Some(Outcome::Fail(failure)) if failure.code == FailureCode::Forbidden) |
| 1044 | ); |
| 1045 | assert!(owner_only::<()>(&user("owner", "workspace"), "acme").is_some()); |
| 1046 | assert!(owner_only::<()>(&user("owner", "user"), "other").is_some()); |
| 1047 | } |
| 1048 | |
| 1049 | #[test] |
| 1050 | fn no_answer_carries_the_key() { |
| 1051 | let key = "sk-admin-SECRETSECRETwxyz"; |
| 1052 | // The list never selects it. |
| 1053 | assert!(!LIST_SQL.contains("credential")); |
| 1054 | // What is stored is sealed. |
| 1055 | let sealer = g1t_secrets::Sealer::new(&"ab".repeat(32)).unwrap(); |
| 1056 | let sealed = sealer.seal(key, "pcn_1"); |
| 1057 | assert!(!sealed.contains("SECRET")); |
| 1058 | // A connection, a list and the costs, as sent. |
| 1059 | let connection = to_connection(ListedRow { |
| 1060 | provider: "openai".into(), |
| 1061 | hint: g1t_secrets::hint(key), |
| 1062 | connected_by: "ada".into(), |
| 1063 | connected_at: "2026-10-18T00:00:00.000Z".into(), |
| 1064 | last_synced_at: None, |
| 1065 | last_error: Some(explain( |
| 1066 | CostProvider::Openai, |
| 1067 | 401, |
| 1068 | &json!({ "error": { "message": key } }), |
| 1069 | key, |
| 1070 | )), |
| 1071 | synced_from: None, |
| 1072 | }) |
| 1073 | .unwrap(); |
| 1074 | let costs = ProviderCosts { |
| 1075 | connections: vec![connection.clone()], |
| 1076 | days: vec![], |
| 1077 | complete: false, |
| 1078 | }; |
| 1079 | for json in [ |
| 1080 | serde_json::to_string(&connection).unwrap(), |
| 1081 | serde_json::to_string(&costs).unwrap(), |
| 1082 | ] { |
| 1083 | assert!(!json.contains("SECRET"), "{json}"); |
| 1084 | assert!(!json.contains("credential"), "{json}"); |
| 1085 | } |
| 1086 | assert_eq!(connection.hint, "…wxyz"); |
| 1087 | // And the arguments, logged, name it only as redacted. |
| 1088 | let args = ConnectProviderArgs { |
| 1089 | actor: user("owner", "user"), |
| 1090 | workspace: "acme".into(), |
| 1091 | provider: CostProvider::Openai, |
| 1092 | credential: key.into(), |
| 1093 | }; |
| 1094 | assert!(!format!("{args:?}").contains("SECRET")); |
| 1095 | } |
| 1096 | |
| 1097 | #[test] |
| 1098 | fn costs_are_complete_only_when_every_provider_has_read_back_far_enough() { |
| 1099 | let connection = |from: Option<&str>, error: Option<&str>| ProviderConnection { |
| 1100 | provider: CostProvider::Anthropic, |
| 1101 | hint: "…abcd".into(), |
| 1102 | connected_by: "ada".into(), |
| 1103 | connected_at: "2026-10-01T00:00:00.000Z".into(), |
| 1104 | last_synced_at: Some("2026-10-18T04:17:00.000Z".into()), |
| 1105 | last_error: error.map(str::to_owned), |
| 1106 | synced_from: from.map(str::to_owned), |
| 1107 | }; |
| 1108 | assert!(!complete(&[], "2026-10-01")); |
| 1109 | assert!(complete(&[connection(Some("2026-07-21"), None)], "2026-10-01")); |
| 1110 | assert!(!complete(&[connection(Some("2026-10-05"), None)], "2026-10-01")); |
| 1111 | assert!(!complete( |
| 1112 | &[connection(Some("2026-07-21"), Some("Anthropic didn't answer"))], |
| 1113 | "2026-10-01" |
| 1114 | )); |
| 1115 | } |
| 1116 | |
| 1117 | #[test] |
| 1118 | fn a_read_replaces_its_days_and_reading_twice_changes_nothing() { |
| 1119 | let db = rusqlite::Connection::open_in_memory().unwrap(); |
| 1120 | db.execute_batch(include_str!("../migrations/0067_provider_costs.sql")) |
| 1121 | .unwrap(); |
| 1122 | let write = |day: &str, item: &str, micros: i64, at: &str| { |
| 1123 | db.execute( |
| 1124 | UPSERT_COST, |
| 1125 | rusqlite::params!["acme", "anthropic", day, item, "", micros, at], |
| 1126 | ) |
| 1127 | .unwrap(); |
| 1128 | }; |
| 1129 | let read = || -> Vec<(String, String, i64)> { |
| 1130 | let mut statement = db |
| 1131 | .prepare("SELECT day, item, micros FROM provider_costs ORDER BY day, item") |
| 1132 | .unwrap(); |
| 1133 | statement |
| 1134 | .query_map([], |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?))) |
| 1135 | .unwrap() |
| 1136 | .map(Result::unwrap) |
| 1137 | .collect() |
| 1138 | }; |
| 1139 | write("2026-10-16", "Opus input", 100, "t1"); |
| 1140 | write("2026-10-17", "Opus input", 200, "t1"); |
| 1141 | write("2026-10-17", "Opus output", 50, "t1"); |
| 1142 | // The same read again: the same rows. |
| 1143 | write("2026-10-17", "Opus input", 200, "t1"); |
| 1144 | assert_eq!(read().len(), 3); |
| 1145 | // The provider revised the 17th: the output line went away and the |
| 1146 | // input grew. The 16th is outside this read and stays. |
| 1147 | db.execute( |
| 1148 | DELETE_DAYS, |
| 1149 | rusqlite::params!["acme", "anthropic", "2026-10-17", "2026-10-18"], |
| 1150 | ) |
| 1151 | .unwrap(); |
| 1152 | write("2026-10-17", "Opus input", 250, "t2"); |
| 1153 | assert_eq!( |
| 1154 | read(), |
| 1155 | [ |
| 1156 | ("2026-10-16".into(), "Opus input".into(), 100), |
| 1157 | ("2026-10-17".into(), "Opus input".into(), 250) |
| 1158 | ] |
| 1159 | ); |
| 1160 | // A successful read moves the first day back, never forward. |
| 1161 | db.execute( |
| 1162 | "INSERT INTO provider_connections (id, workspace, provider, credential, hint, connected_by, connected_at) VALUES ('pcn_1', 'acme', 'anthropic', 'v1:x', '…abcd', 'ada', 't0')", |
| 1163 | [], |
| 1164 | ) |
| 1165 | .unwrap(); |
| 1166 | let note = |from: &str| { |
| 1167 | db.execute( |
| 1168 | NOTE_SYNCED, |
| 1169 | rusqlite::params!["t3", Option::<String>::None, from, "pcn_1"], |
| 1170 | ) |
| 1171 | .unwrap(); |
| 1172 | db.query_row("SELECT synced_from FROM provider_connections", [], |r| { |
| 1173 | r.get::<_, String>(0) |
| 1174 | }) |
| 1175 | .unwrap() |
| 1176 | }; |
| 1177 | assert_eq!(note("2026-07-21"), "2026-07-21"); |
| 1178 | assert_eq!(note("2026-10-12"), "2026-07-21"); |
| 1179 | // One connection per workspace and provider. |
| 1180 | assert!(db |
| 1181 | .execute( |
| 1182 | "INSERT INTO provider_connections (id, workspace, provider, credential, hint, connected_by, connected_at) VALUES ('pcn_2', 'acme', 'anthropic', 'v1:y', '…efgh', 'ada', 't4')", |
| 1183 | [], |
| 1184 | ) |
| 1185 | .is_err()); |
| 1186 | } |
| 1187 | } |