1use std::time::Duration;
10
11use chrono::{DateTime, Datelike, Utc};
12
13use crate::cache::{Cache, acquire_lock_async};
14use crate::error::{AppError, Result};
15use crate::usage::{AnthropicApiSnapshot, finite_amount};
16use crate::vendor::{MAX_BODY_BYTES, read_body_capped};
17
18use super::types::{CostReport, page_covered_through, page_dollars};
19
20pub const BASE_URL: &str = "https://api.anthropic.com";
21pub const ANTHROPIC_VERSION: &str = "2023-06-01";
22const HTTP_TIMEOUT: Duration = Duration::from_secs(15);
23const LOCK_TIMEOUT: Duration = Duration::from_secs(15);
24const MAX_PAGES: usize = 12;
27
28#[derive(Debug, Clone)]
29pub struct Endpoints {
30 pub cost_report: String,
31}
32
33impl Default for Endpoints {
34 fn default() -> Self {
35 Self {
36 cost_report: format!("{BASE_URL}/v1/organizations/cost_report"),
37 }
38 }
39}
40
41pub type FetchOutcome = crate::outcome::Outcome<AnthropicApiSnapshot>;
44
45fn month_start_rfc3339(now: DateTime<Utc>) -> String {
47 format!("{:04}-{:02}-01T00:00:00Z", now.year(), now.month())
48}
49
50fn target_key(admin_key: &str) -> String {
56 use std::hash::{Hash, Hasher};
57 let mut hasher = std::collections::hash_map::DefaultHasher::new();
58 admin_key.hash(&mut hasher);
59 format!("key:{:016x}", hasher.finish())
60}
61
62fn validate_limit(limit: Option<f64>) -> Result<Option<f64>> {
63 if let Some(value) = limit
64 && (!value.is_finite() || value <= 0.0)
65 {
66 return Err(AppError::Schema(
67 "anthropic-api monthly_limit must be finite and greater than zero; \
68 remove it to show spend without a limit"
69 .into(),
70 ));
71 }
72 Ok(limit)
73}
74
75pub async fn fetch_snapshot(
78 client: &reqwest::Client,
79 admin_key: &str,
80 cache: &Cache,
81 endpoints: &Endpoints,
82 cache_ttl: Duration,
83 limit: Option<f64>,
84) -> Result<FetchOutcome> {
85 fetch_snapshot_at(
86 client,
87 admin_key,
88 cache,
89 endpoints,
90 cache_ttl,
91 limit,
92 Utc::now(),
93 )
94 .await
95}
96
97pub async fn fetch_snapshot_at(
100 client: &reqwest::Client,
101 admin_key: &str,
102 cache: &Cache,
103 endpoints: &Endpoints,
104 cache_ttl: Duration,
105 limit: Option<f64>,
106 now: DateTime<Utc>,
107) -> Result<FetchOutcome> {
108 let limit = validate_limit(limit)?;
109 cache.ensure_dir()?;
110 let _lock = acquire_lock_async(&cache.lock_path(), LOCK_TIMEOUT).await?;
111
112 let month = month_start_rfc3339(now);
115 let target = target_key(admin_key);
116
117 if let Some(bytes) = cache.fresh_payload(cache_ttl)?
118 && let Ok(outcome) = reuse_cache(&bytes, cache, false, limit, &month, &target)
119 {
120 return Ok(outcome);
121 }
122
123 match fetch_live(client, endpoints, admin_key, now).await {
124 Ok((spent, covered_through)) => {
125 let snap = AnthropicApiSnapshot {
126 spent,
127 limit,
128 covered_through,
129 };
130 let bytes = serde_json::to_vec(&serde_json::json!({
131 "month": month,
132 "target": target,
133 "snapshot": {
134 "spent": snap.spent,
135 "limit": snap.limit,
136 "covered_through": snap.covered_through,
137 },
138 }))?;
139 cache.write_payload(&bytes)?;
140 Ok(crate::outcome::Outcome::fresh(snap))
141 }
142 Err(e) if e.is_transient() => fallback_silent(cache, limit, &month, &target, e),
143 Err(AppError::Http { status, body }) => {
144 cache.mark_stale();
145 let diag = cache.write_last_error(status, &body);
146 fallback_with_error(
147 cache,
148 Some(diag),
149 limit,
150 &month,
151 &target,
152 AppError::Http { status, body },
153 )
154 }
155 Err(e) => {
156 cache.mark_stale();
157 let diag = cache.write_last_error(0, &e.to_string());
158 fallback_with_error(cache, Some(diag), limit, &month, &target, e)
159 }
160 }
161}
162
163fn fallback_silent(
164 cache: &Cache,
165 limit: Option<f64>,
166 month: &str,
167 target: &str,
168 original: AppError,
169) -> Result<FetchOutcome> {
170 crate::outcome::fallback(cache, None, original, |bytes| {
171 cached_snapshot(bytes, limit, month, target)
172 })
173}
174
175fn fallback_with_error(
183 cache: &Cache,
184 last_error: Option<(u16, String)>,
185 limit: Option<f64>,
186 month: &str,
187 target: &str,
188 original: AppError,
189) -> Result<FetchOutcome> {
190 crate::outcome::fallback(cache, last_error, original, |bytes| {
191 cached_snapshot(bytes, limit, month, target)
192 })
193}
194
195fn reuse_cache(
196 bytes: &[u8],
197 cache: &Cache,
198 stale: bool,
199 limit: Option<f64>,
200 month: &str,
201 target: &str,
202) -> Result<FetchOutcome> {
203 let snapshot = cached_snapshot(bytes, limit, month, target)?;
204 Ok(crate::outcome::Outcome::cached(snapshot, cache, stale))
205}
206
207fn cached_snapshot(
210 bytes: &[u8],
211 limit: Option<f64>,
212 month: &str,
213 target: &str,
214) -> Result<AnthropicApiSnapshot> {
215 let (spent, covered_through) = parse_cached_spend(bytes, month, target)?;
216 Ok(AnthropicApiSnapshot {
217 spent,
218 limit,
219 covered_through,
220 })
221}
222
223fn parse_cached_spend(
224 bytes: &[u8],
225 month: &str,
226 target: &str,
227) -> Result<(f64, Option<chrono::NaiveDate>)> {
228 let v: serde_json::Value = serde_json::from_slice(bytes)?;
229 let cached_month = v.get("month").and_then(serde_json::Value::as_str);
232 if cached_month != Some(month) {
233 return Err(AppError::Schema(format!(
234 "anthropic-api cache is for a different month ({}); refetching",
235 cached_month.unwrap_or("unknown")
236 )));
237 }
238 let cached_target = v.get("target").and_then(serde_json::Value::as_str);
239 if cached_target != Some(target) {
240 return Err(AppError::Schema(
241 "anthropic-api cache belongs to a different Admin key; refetching".into(),
242 ));
243 }
244 let s = v
245 .get("snapshot")
246 .ok_or_else(|| AppError::Schema("anthropic-api cache missing 'snapshot'".into()))?;
247 let spent = s["spent"]
248 .as_f64()
249 .ok_or_else(|| AppError::Schema("anthropic-api cache missing 'spent'".into()))?;
250 let spent = crate::usage::finite_amount("anthropic-api cache", "spent", spent)?;
251 let covered_through = s
252 .get("covered_through")
253 .and_then(serde_json::Value::as_str)
254 .map(|raw| {
255 chrono::NaiveDate::parse_from_str(raw, "%Y-%m-%d").map_err(|e| {
256 AppError::Schema(format!(
257 "anthropic-api cache has invalid 'covered_through' {raw:?}: {e}"
258 ))
259 })
260 })
261 .transpose()?;
262 Ok((spent, covered_through))
263}
264
265async fn fetch_live(
266 client: &reqwest::Client,
267 endpoints: &Endpoints,
268 admin_key: &str,
269 now: DateTime<Utc>,
270) -> Result<(f64, Option<chrono::NaiveDate>)> {
271 let starting_at = month_start_rfc3339(now);
272 let mut total = 0.0;
273 let mut page: Option<String> = None;
274 let mut seen_pages: Vec<String> = Vec::new();
275 let mut covered_through = None;
276
277 for _ in 0..MAX_PAGES {
278 let mut req = client
279 .get(&endpoints.cost_report)
280 .header("x-api-key", admin_key)
281 .header("anthropic-version", ANTHROPIC_VERSION)
282 .query(&[
283 ("starting_at", starting_at.as_str()),
284 ("bucket_width", "1d"),
285 ]);
286 if let Some(p) = &page {
287 req = req.query(&[("page", p.as_str())]);
288 }
289
290 let resp = tokio::time::timeout(HTTP_TIMEOUT, req.send())
291 .await
292 .map_err(|_| {
293 AppError::Transport(format!("anthropic-api timeout: {}", endpoints.cost_report))
294 })??;
295
296 let status = resp.status();
297 let bytes = read_body_capped(resp, MAX_BODY_BYTES).await?;
298 if !status.is_success() {
299 let body = String::from_utf8_lossy(&bytes).chars().take(200).collect();
300 return Err(AppError::Http {
301 status: status.as_u16(),
302 body,
303 });
304 }
305
306 let report: CostReport = serde_json::from_slice(&bytes)
307 .map_err(|e| AppError::Schema(format!("anthropic-api cost_report: {e}")))?;
308 total = finite_amount(
309 "anthropic-api",
310 "cost_report running total",
311 total + page_dollars(&report)?,
312 )?;
313 covered_through = covered_through.max(page_covered_through(&report)?);
314
315 match (report.has_more, report.next_page) {
319 (false, _) => return Ok((total, covered_through)),
320 (true, None) => {
321 return Err(AppError::Schema(
322 "anthropic-api cost_report: has_more is true but next_page is missing; \
323 refusing to report a partial month"
324 .into(),
325 ));
326 }
327 (true, Some(p)) if p.trim().is_empty() => {
328 return Err(AppError::Schema(
329 "anthropic-api cost_report: has_more is true but next_page is empty; \
330 refusing to report a partial month"
331 .into(),
332 ));
333 }
334 (true, Some(p)) => {
335 if seen_pages.contains(&p) {
336 return Err(AppError::Schema(format!(
337 "anthropic-api cost_report: pagination repeated cursor {p:?}; \
338 refusing to report a partial month"
339 )));
340 }
341 seen_pages.push(p.clone());
342 page = Some(p);
343 }
344 }
345 }
346 Err(AppError::Schema(format!(
347 "anthropic-api cost_report: more than {MAX_PAGES} pages for one month; \
348 refusing to report a partial month"
349 )))
350}
351
352#[cfg(test)]
353mod tests {
354 use super::*;
355 use chrono::TimeZone;
356 use tempfile::TempDir;
357
358 fn cache_fixture() -> (TempDir, Cache) {
359 let td = TempDir::new().unwrap();
360 let cache = Cache::at(td.path().join("anthropic_api"));
361 cache.ensure_dir().unwrap();
362 (td, cache)
363 }
364
365 #[test]
366 fn month_start_is_first_of_month_utc() {
367 let now = Utc.with_ymd_and_hms(2026, 7, 19, 15, 8, 0).unwrap();
368 assert_eq!(month_start_rfc3339(now), "2026-07-01T00:00:00Z");
369 }
370
371 #[tokio::test]
372 async fn live_fetch_sums_month_to_date_and_divides_by_100() {
373 let mut server = mockito::Server::new_async().await;
374 server
375 .mock("GET", "/v1/organizations/cost_report")
376 .match_header("x-api-key", "sk-ant-admin01-test")
377 .match_query(mockito::Matcher::Any)
378 .with_status(200)
379 .with_body(
380 r#"{"data":[{"ending_at":"2026-07-19T00:00:00Z",
381 "results":[{"amount":"100.0","currency":"USD"},
382 {"amount":"34.0","currency":"USD"}]}],
383 "has_more":false,"next_page":null}"#,
384 )
385 .create_async()
386 .await;
387
388 let (_td, cache) = cache_fixture();
389 let client = reqwest::Client::new();
390 let endpoints = Endpoints {
391 cost_report: format!("{}/v1/organizations/cost_report", server.url()),
392 };
393 let out = fetch_snapshot(
394 &client,
395 "sk-ant-admin01-test",
396 &cache,
397 &endpoints,
398 Duration::from_secs(0),
399 Some(1000.0),
400 )
401 .await
402 .unwrap();
403 assert!((out.snapshot.spent - 1.34).abs() < 1e-9);
405 assert_eq!(out.snapshot.limit, Some(1000.0));
406 assert_eq!(
407 out.snapshot.covered_through,
408 Some(chrono::NaiveDate::from_ymd_opt(2026, 7, 18).unwrap())
409 );
410 assert!(!out.stale);
411 }
412
413 #[tokio::test]
414 async fn http_401_falls_back_to_cache_when_present() {
415 let mut server = mockito::Server::new_async().await;
416 server
417 .mock("GET", "/v1/organizations/cost_report")
418 .match_query(mockito::Matcher::Any)
419 .with_status(401)
420 .with_body(r#"{"error":{"message":"invalid x-api-key"}}"#)
421 .create_async()
422 .await;
423
424 let (_td, cache) = cache_fixture();
425 let now = at(2026, 7, 19);
426 cache
427 .write_payload(
428 serde_json::json!({
429 "month": month_start_rfc3339(now),
430 "target": target_key("k"),
431 "snapshot": { "spent": 2.5, "limit": null },
432 })
433 .to_string()
434 .as_bytes(),
435 )
436 .unwrap();
437
438 let client = reqwest::Client::new();
439 let endpoints = Endpoints {
440 cost_report: format!("{}/v1/organizations/cost_report", server.url()),
441 };
442 let out = fetch_snapshot_at(
443 &client,
444 "k",
445 &cache,
446 &endpoints,
447 Duration::from_secs(0),
448 Some(50.0),
449 now,
450 )
451 .await
452 .unwrap();
453 assert!(out.stale);
454 assert!((out.snapshot.spent - 2.5).abs() < 1e-9);
455 assert_eq!(out.snapshot.limit, Some(50.0));
457 assert_eq!(out.last_error.as_ref().map(|(c, _)| *c), Some(401));
458 }
459
460 fn at(y: i32, m: u32, d: u32) -> DateTime<Utc> {
462 chrono::NaiveDate::from_ymd_opt(y, m, d)
463 .unwrap()
464 .and_hms_opt(12, 0, 0)
465 .unwrap()
466 .and_utc()
467 }
468
469 fn ok_body(cents: &str) -> String {
470 format!(
471 r#"{{"data":[{{"results":[{{"amount":"{cents}","currency":"USD"}}]}}],"has_more":false}}"#
472 )
473 }
474
475 #[tokio::test]
476 async fn month_rollover_refetches_instead_of_showing_last_month() {
477 let mut server = mockito::Server::new_async().await;
480 server
481 .mock("GET", "/v1/organizations/cost_report")
482 .match_query(mockito::Matcher::Any)
483 .with_status(200)
484 .with_body(ok_body("250.0"))
485 .create_async()
486 .await;
487
488 let (_td, cache) = cache_fixture();
489 cache
490 .write_payload(
491 serde_json::json!({
492 "month": month_start_rfc3339(at(2026, 6, 30)),
493 "target": target_key("k"),
494 "snapshot": { "spent": 987.0, "limit": null },
495 })
496 .to_string()
497 .as_bytes(),
498 )
499 .unwrap();
500
501 let client = reqwest::Client::new();
502 let endpoints = Endpoints {
503 cost_report: format!("{}/v1/organizations/cost_report", server.url()),
504 };
505 let out = fetch_snapshot_at(
507 &client,
508 "k",
509 &cache,
510 &endpoints,
511 Duration::from_secs(3600),
512 None,
513 at(2026, 7, 1),
514 )
515 .await
516 .unwrap();
517 assert!((out.snapshot.spent - 2.5).abs() < 1e-9);
518 assert!(!out.stale);
519 }
520
521 #[tokio::test]
522 async fn last_months_cache_is_not_served_during_an_outage() {
523 let mut server = mockito::Server::new_async().await;
526 server
527 .mock("GET", "/v1/organizations/cost_report")
528 .match_query(mockito::Matcher::Any)
529 .with_status(500)
530 .with_body("upstream boom")
531 .create_async()
532 .await;
533
534 let (_td, cache) = cache_fixture();
535 cache
536 .write_payload(
537 serde_json::json!({
538 "month": month_start_rfc3339(at(2026, 6, 30)),
539 "target": target_key("k"),
540 "snapshot": { "spent": 987.0, "limit": null },
541 })
542 .to_string()
543 .as_bytes(),
544 )
545 .unwrap();
546
547 let client = reqwest::Client::new();
548 let endpoints = Endpoints {
549 cost_report: format!("{}/v1/organizations/cost_report", server.url()),
550 };
551 let out = fetch_snapshot_at(
552 &client,
553 "k",
554 &cache,
555 &endpoints,
556 Duration::from_secs(0),
557 None,
558 at(2026, 7, 1),
559 )
560 .await;
561 assert!(out.is_err(), "expected an error, got {out:?}");
562 }
563
564 #[tokio::test]
565 async fn first_run_auth_failure_preserves_the_original_error() {
566 let mut server = mockito::Server::new_async().await;
569 server
570 .mock("GET", "/v1/organizations/cost_report")
571 .match_query(mockito::Matcher::Any)
572 .with_status(401)
573 .with_body(r#"{"error":{"message":"invalid x-api-key"}}"#)
574 .create_async()
575 .await;
576
577 let (_td, cache) = cache_fixture();
578 let client = reqwest::Client::new();
579 let endpoints = Endpoints {
580 cost_report: format!("{}/v1/organizations/cost_report", server.url()),
581 };
582 let err = fetch_snapshot_at(
583 &client,
584 "k",
585 &cache,
586 &endpoints,
587 Duration::from_secs(0),
588 None,
589 at(2026, 7, 19),
590 )
591 .await
592 .unwrap_err();
593 assert!(
594 matches!(err, AppError::Http { status: 401, .. }),
595 "original error must survive, got {err:?}"
596 );
597 assert!(err.to_string().contains("invalid x-api-key"));
598 }
599
600 #[tokio::test]
601 async fn has_more_without_next_page_is_an_error_not_a_partial_month() {
602 let mut server = mockito::Server::new_async().await;
603 server
604 .mock("GET", "/v1/organizations/cost_report")
605 .match_query(mockito::Matcher::Any)
606 .with_status(200)
607 .with_body(
608 r#"{"data":[{"results":[{"amount":"100.0","currency":"USD"}]}],
609 "has_more":true}"#,
610 )
611 .create_async()
612 .await;
613
614 let (_td, cache) = cache_fixture();
615 let client = reqwest::Client::new();
616 let endpoints = Endpoints {
617 cost_report: format!("{}/v1/organizations/cost_report", server.url()),
618 };
619 let out = fetch_snapshot_at(
620 &client,
621 "k",
622 &cache,
623 &endpoints,
624 Duration::from_secs(0),
625 None,
626 at(2026, 7, 19),
627 )
628 .await;
629 assert!(out.is_err(), "partial month must not be reported: {out:?}");
630 }
631
632 #[tokio::test]
633 async fn repeated_pagination_cursor_is_an_error() {
634 let mut server = mockito::Server::new_async().await;
637 server
638 .mock("GET", "/v1/organizations/cost_report")
639 .match_query(mockito::Matcher::Any)
640 .with_status(200)
641 .with_body(
642 r#"{"data":[{"results":[{"amount":"100.0","currency":"USD"}]}],
643 "has_more":true,"next_page":"same"}"#,
644 )
645 .expect_at_least(1)
646 .create_async()
647 .await;
648
649 let (_td, cache) = cache_fixture();
650 let client = reqwest::Client::new();
651 let endpoints = Endpoints {
652 cost_report: format!("{}/v1/organizations/cost_report", server.url()),
653 };
654 let out = fetch_snapshot_at(
655 &client,
656 "k",
657 &cache,
658 &endpoints,
659 Duration::from_secs(0),
660 None,
661 at(2026, 7, 19),
662 )
663 .await;
664 assert!(out.is_err(), "cursor loop must not be reported: {out:?}");
665 }
666
667 #[tokio::test]
668 async fn malformed_200_is_not_cached_as_zero_spend() {
669 let mut server = mockito::Server::new_async().await;
670 server
671 .mock("GET", "/v1/organizations/cost_report")
672 .match_query(mockito::Matcher::Any)
673 .with_status(200)
674 .with_body(r#"{"error":{"message":"permission_error"}}"#)
675 .create_async()
676 .await;
677
678 let (_td, cache) = cache_fixture();
679 let client = reqwest::Client::new();
680 let endpoints = Endpoints {
681 cost_report: format!("{}/v1/organizations/cost_report", server.url()),
682 };
683 let out = fetch_snapshot_at(
684 &client,
685 "k",
686 &cache,
687 &endpoints,
688 Duration::from_secs(0),
689 None,
690 at(2026, 7, 19),
691 )
692 .await;
693 assert!(out.is_err(), "expected a schema error, got {out:?}");
694 assert!(cache.maybe_payload().unwrap().is_none());
696 }
697
698 #[tokio::test]
699 async fn a_month_with_no_spend_is_cached_as_a_real_zero() {
700 let mut server = mockito::Server::new_async().await;
702 server
703 .mock("GET", "/v1/organizations/cost_report")
704 .match_query(mockito::Matcher::Any)
705 .with_status(200)
706 .with_body(r#"{"data":[],"has_more":false,"next_page":null}"#)
707 .create_async()
708 .await;
709
710 let (_td, cache) = cache_fixture();
711 let client = reqwest::Client::new();
712 let endpoints = Endpoints {
713 cost_report: format!("{}/v1/organizations/cost_report", server.url()),
714 };
715 let out = fetch_snapshot_at(
716 &client,
717 "k",
718 &cache,
719 &endpoints,
720 Duration::from_secs(0),
721 None,
722 at(2026, 7, 19),
723 )
724 .await
725 .unwrap();
726 assert_eq!(out.snapshot.spent, 0.0);
727 assert!(!out.stale);
728 assert!(cache.maybe_payload().unwrap().is_some());
729 }
730
731 #[tokio::test]
732 async fn switching_admin_key_refetches_instead_of_reusing_another_organization() {
733 let mut server = mockito::Server::new_async().await;
734 let first = server
735 .mock("GET", "/v1/organizations/cost_report")
736 .match_header("x-api-key", "org-a-key")
737 .match_query(mockito::Matcher::Any)
738 .with_status(200)
739 .with_body(ok_body("100.0"))
740 .expect(1)
741 .create_async()
742 .await;
743 let second = server
744 .mock("GET", "/v1/organizations/cost_report")
745 .match_header("x-api-key", "org-b-key")
746 .match_query(mockito::Matcher::Any)
747 .with_status(200)
748 .with_body(ok_body("250.0"))
749 .expect(1)
750 .create_async()
751 .await;
752
753 let (_td, cache) = cache_fixture();
754 let client = reqwest::Client::new();
755 let endpoints = Endpoints {
756 cost_report: format!("{}/v1/organizations/cost_report", server.url()),
757 };
758 let now = at(2026, 7, 19);
759 let a = fetch_snapshot_at(
760 &client,
761 "org-a-key",
762 &cache,
763 &endpoints,
764 Duration::ZERO,
765 None,
766 now,
767 )
768 .await
769 .unwrap();
770 assert_eq!(a.snapshot.spent, 1.0);
771
772 let b = fetch_snapshot_at(
774 &client,
775 "org-b-key",
776 &cache,
777 &endpoints,
778 Duration::from_secs(3600),
779 None,
780 now,
781 )
782 .await
783 .unwrap();
784 assert_eq!(b.snapshot.spent, 2.5);
785 first.assert_async().await;
786 second.assert_async().await;
787 }
788
789 #[tokio::test]
790 async fn invalid_monthly_limits_fail_before_network_or_cache_access() {
791 let client = reqwest::Client::new();
792 let cache = Cache::at(std::path::PathBuf::from("unused-invalid-limit-cache"));
793 for limit in [0.0, -1.0, f64::INFINITY, f64::NAN] {
794 let err = fetch_snapshot_at(
795 &client,
796 "key",
797 &cache,
798 &Endpoints::default(),
799 Duration::ZERO,
800 Some(limit),
801 at(2026, 7, 19),
802 )
803 .await
804 .unwrap_err();
805 assert!(err.to_string().contains("monthly_limit"), "{err:?}");
806 }
807 assert!(!cache.dir().exists());
808 }
809}