1mod authorization;
2pub mod db;
3mod telemetry;
4mod token;
5
6use crate::{
7 api::CloudflareIpCountryHeader, build_clickhouse_client, db::UserId, executor::Executor, Cents,
8 Config, Error, Result,
9};
10use anyhow::{anyhow, Context as _};
11use authorization::authorize_access_to_language_model;
12use axum::routing::get;
13use axum::{
14 body::Body,
15 http::{self, HeaderName, HeaderValue, Request, StatusCode},
16 middleware::{self, Next},
17 response::{IntoResponse, Response},
18 routing::post,
19 Extension, Json, Router, TypedHeader,
20};
21use chrono::{DateTime, Duration, Utc};
22use collections::HashMap;
23use db::{usage_measure::UsageMeasure, ActiveUserCount, LlmDatabase};
24use futures::{Stream, StreamExt as _};
25use isahc_http_client::IsahcHttpClient;
26use rpc::ListModelsResponse;
27use rpc::{
28 proto::Plan, LanguageModelProvider, PerformCompletionParams, EXPIRED_LLM_TOKEN_HEADER_NAME,
29};
30use std::{
31 pin::Pin,
32 sync::Arc,
33 task::{Context, Poll},
34};
35use strum::IntoEnumIterator;
36use telemetry::{report_llm_rate_limit, report_llm_usage, LlmRateLimitEventRow, LlmUsageEventRow};
37use tokio::sync::RwLock;
38use util::ResultExt;
39
40pub use token::*;
41
42pub struct LlmState {
43 pub config: Config,
44 pub executor: Executor,
45 pub db: Arc<LlmDatabase>,
46 pub http_client: IsahcHttpClient,
47 pub clickhouse_client: Option<clickhouse::Client>,
48 active_user_count_by_model:
49 RwLock<HashMap<(LanguageModelProvider, String), (DateTime<Utc>, ActiveUserCount)>>,
50}
51
52const ACTIVE_USER_COUNT_CACHE_DURATION: Duration = Duration::seconds(30);
53
54impl LlmState {
55 pub async fn new(config: Config, executor: Executor) -> Result<Arc<Self>> {
56 let database_url = config
57 .llm_database_url
58 .as_ref()
59 .ok_or_else(|| anyhow!("missing LLM_DATABASE_URL"))?;
60 let max_connections = config
61 .llm_database_max_connections
62 .ok_or_else(|| anyhow!("missing LLM_DATABASE_MAX_CONNECTIONS"))?;
63
64 let mut db_options = db::ConnectOptions::new(database_url);
65 db_options.max_connections(max_connections);
66 let mut db = LlmDatabase::new(db_options, executor.clone()).await?;
67 db.initialize().await?;
68
69 let db = Arc::new(db);
70
71 let user_agent = format!("Zed Server/{}", env!("CARGO_PKG_VERSION"));
72 let http_client = IsahcHttpClient::builder()
73 .default_header("User-Agent", user_agent)
74 .build()
75 .map(IsahcHttpClient::from)
76 .context("failed to construct http client")?;
77
78 let this = Self {
79 executor,
80 db,
81 http_client,
82 clickhouse_client: config
83 .clickhouse_url
84 .as_ref()
85 .and_then(|_| build_clickhouse_client(&config).log_err()),
86 active_user_count_by_model: RwLock::new(HashMap::default()),
87 config,
88 };
89
90 Ok(Arc::new(this))
91 }
92
93 pub async fn get_active_user_count(
94 &self,
95 provider: LanguageModelProvider,
96 model: &str,
97 ) -> Result<ActiveUserCount> {
98 let now = Utc::now();
99
100 {
101 let active_user_count_by_model = self.active_user_count_by_model.read().await;
102 if let Some((last_updated, count)) =
103 active_user_count_by_model.get(&(provider, model.to_string()))
104 {
105 if now - *last_updated < ACTIVE_USER_COUNT_CACHE_DURATION {
106 return Ok(*count);
107 }
108 }
109 }
110
111 let mut cache = self.active_user_count_by_model.write().await;
112 let new_count = self.db.get_active_user_count(provider, model, now).await?;
113 cache.insert((provider, model.to_string()), (now, new_count));
114 Ok(new_count)
115 }
116}
117
118pub fn routes() -> Router<(), Body> {
119 Router::new()
120 .route("/models", get(list_models))
121 .route("/completion", post(perform_completion))
122 .layer(middleware::from_fn(validate_api_token))
123}
124
125async fn validate_api_token<B>(mut req: Request<B>, next: Next<B>) -> impl IntoResponse {
126 let token = req
127 .headers()
128 .get(http::header::AUTHORIZATION)
129 .and_then(|header| header.to_str().ok())
130 .ok_or_else(|| {
131 Error::http(
132 StatusCode::BAD_REQUEST,
133 "missing authorization header".to_string(),
134 )
135 })?
136 .strip_prefix("Bearer ")
137 .ok_or_else(|| {
138 Error::http(
139 StatusCode::BAD_REQUEST,
140 "invalid authorization header".to_string(),
141 )
142 })?;
143
144 let state = req.extensions().get::<Arc<LlmState>>().unwrap();
145 match LlmTokenClaims::validate(token, &state.config) {
146 Ok(claims) => {
147 if state.db.is_access_token_revoked(&claims.jti).await? {
148 return Err(Error::http(
149 StatusCode::UNAUTHORIZED,
150 "unauthorized".to_string(),
151 ));
152 }
153
154 tracing::Span::current()
155 .record("user_id", claims.user_id)
156 .record("login", claims.github_user_login.clone())
157 .record("authn.jti", &claims.jti)
158 .record("is_staff", claims.is_staff);
159
160 req.extensions_mut().insert(claims);
161 Ok::<_, Error>(next.run(req).await.into_response())
162 }
163 Err(ValidateLlmTokenError::Expired) => Err(Error::Http(
164 StatusCode::UNAUTHORIZED,
165 "unauthorized".to_string(),
166 [(
167 HeaderName::from_static(EXPIRED_LLM_TOKEN_HEADER_NAME),
168 HeaderValue::from_static("true"),
169 )]
170 .into_iter()
171 .collect(),
172 )),
173 Err(_err) => Err(Error::http(
174 StatusCode::UNAUTHORIZED,
175 "unauthorized".to_string(),
176 )),
177 }
178}
179
180async fn list_models(
181 Extension(state): Extension<Arc<LlmState>>,
182 Extension(claims): Extension<LlmTokenClaims>,
183 country_code_header: Option<TypedHeader<CloudflareIpCountryHeader>>,
184) -> Result<Json<ListModelsResponse>> {
185 let country_code = country_code_header.map(|header| header.to_string());
186
187 let mut accessible_models = Vec::new();
188
189 for (provider, model) in state.db.all_models() {
190 let authorize_result = authorize_access_to_language_model(
191 &state.config,
192 &claims,
193 country_code.as_deref(),
194 provider,
195 &model.name,
196 );
197
198 if authorize_result.is_ok() {
199 accessible_models.push(rpc::LanguageModel {
200 provider,
201 name: model.name,
202 });
203 }
204 }
205
206 Ok(Json(ListModelsResponse {
207 models: accessible_models,
208 }))
209}
210
211async fn perform_completion(
212 Extension(state): Extension<Arc<LlmState>>,
213 Extension(claims): Extension<LlmTokenClaims>,
214 country_code_header: Option<TypedHeader<CloudflareIpCountryHeader>>,
215 Json(params): Json<PerformCompletionParams>,
216) -> Result<impl IntoResponse> {
217 let model = normalize_model_name(
218 state.db.model_names_for_provider(params.provider),
219 params.model,
220 );
221
222 authorize_access_to_language_model(
223 &state.config,
224 &claims,
225 country_code_header
226 .map(|header| header.to_string())
227 .as_deref(),
228 params.provider,
229 &model,
230 )?;
231
232 check_usage_limit(&state, params.provider, &model, &claims).await?;
233
234 let stream = match params.provider {
235 LanguageModelProvider::Anthropic => {
236 let api_key = if claims.is_staff {
237 state
238 .config
239 .anthropic_staff_api_key
240 .as_ref()
241 .context("no Anthropic AI staff API key configured on the server")?
242 } else {
243 state
244 .config
245 .anthropic_api_key
246 .as_ref()
247 .context("no Anthropic AI API key configured on the server")?
248 };
249
250 let mut request: anthropic::Request =
251 serde_json::from_str(params.provider_request.get())?;
252
253 // Override the model on the request with the latest version of the model that is
254 // known to the server.
255 //
256 // Right now, we use the version that's defined in `model.id()`, but we will likely
257 // want to change this code once a new version of an Anthropic model is released,
258 // so that users can use the new version, without having to update Zed.
259 request.model = match model.as_str() {
260 "claude-3-5-sonnet" => anthropic::Model::Claude3_5Sonnet.id().to_string(),
261 "claude-3-opus" => anthropic::Model::Claude3Opus.id().to_string(),
262 "claude-3-haiku" => anthropic::Model::Claude3Haiku.id().to_string(),
263 "claude-3-sonnet" => anthropic::Model::Claude3Sonnet.id().to_string(),
264 _ => request.model,
265 };
266
267 let (chunks, rate_limit_info) = anthropic::stream_completion_with_rate_limit_info(
268 &state.http_client,
269 anthropic::ANTHROPIC_API_URL,
270 api_key,
271 request,
272 None,
273 )
274 .await
275 .map_err(|err| match err {
276 anthropic::AnthropicError::ApiError(ref api_error) => match api_error.code() {
277 Some(anthropic::ApiErrorCode::RateLimitError) => {
278 tracing::info!(
279 target: "upstream rate limit exceeded",
280 user_id = claims.user_id,
281 login = claims.github_user_login,
282 authn.jti = claims.jti,
283 is_staff = claims.is_staff,
284 provider = params.provider.to_string(),
285 model = model
286 );
287
288 Error::http(
289 StatusCode::TOO_MANY_REQUESTS,
290 "Upstream Anthropic rate limit exceeded.".to_string(),
291 )
292 }
293 Some(anthropic::ApiErrorCode::InvalidRequestError) => {
294 Error::http(StatusCode::BAD_REQUEST, api_error.message.clone())
295 }
296 Some(anthropic::ApiErrorCode::OverloadedError) => {
297 Error::http(StatusCode::SERVICE_UNAVAILABLE, api_error.message.clone())
298 }
299 Some(_) => {
300 Error::http(StatusCode::INTERNAL_SERVER_ERROR, api_error.message.clone())
301 }
302 None => Error::Internal(anyhow!(err)),
303 },
304 anthropic::AnthropicError::Other(err) => Error::Internal(err),
305 })?;
306
307 if let Some(rate_limit_info) = rate_limit_info {
308 tracing::info!(
309 target: "upstream rate limit",
310 is_staff = claims.is_staff,
311 provider = params.provider.to_string(),
312 model = model,
313 tokens_remaining = rate_limit_info.tokens_remaining,
314 requests_remaining = rate_limit_info.requests_remaining,
315 requests_reset = ?rate_limit_info.requests_reset,
316 tokens_reset = ?rate_limit_info.tokens_reset,
317 );
318 }
319
320 chunks
321 .map(move |event| {
322 let chunk = event?;
323 let (
324 input_tokens,
325 output_tokens,
326 cache_creation_input_tokens,
327 cache_read_input_tokens,
328 ) = match &chunk {
329 anthropic::Event::MessageStart {
330 message: anthropic::Response { usage, .. },
331 }
332 | anthropic::Event::MessageDelta { usage, .. } => (
333 usage.input_tokens.unwrap_or(0) as usize,
334 usage.output_tokens.unwrap_or(0) as usize,
335 usage.cache_creation_input_tokens.unwrap_or(0) as usize,
336 usage.cache_read_input_tokens.unwrap_or(0) as usize,
337 ),
338 _ => (0, 0, 0, 0),
339 };
340
341 anyhow::Ok(CompletionChunk {
342 bytes: serde_json::to_vec(&chunk).unwrap(),
343 input_tokens,
344 output_tokens,
345 cache_creation_input_tokens,
346 cache_read_input_tokens,
347 })
348 })
349 .boxed()
350 }
351 LanguageModelProvider::OpenAi => {
352 let api_key = state
353 .config
354 .openai_api_key
355 .as_ref()
356 .context("no OpenAI API key configured on the server")?;
357 let chunks = open_ai::stream_completion(
358 &state.http_client,
359 open_ai::OPEN_AI_API_URL,
360 api_key,
361 serde_json::from_str(params.provider_request.get())?,
362 None,
363 )
364 .await?;
365
366 chunks
367 .map(|event| {
368 event.map(|chunk| {
369 let input_tokens =
370 chunk.usage.as_ref().map_or(0, |u| u.prompt_tokens) as usize;
371 let output_tokens =
372 chunk.usage.as_ref().map_or(0, |u| u.completion_tokens) as usize;
373 CompletionChunk {
374 bytes: serde_json::to_vec(&chunk).unwrap(),
375 input_tokens,
376 output_tokens,
377 cache_creation_input_tokens: 0,
378 cache_read_input_tokens: 0,
379 }
380 })
381 })
382 .boxed()
383 }
384 LanguageModelProvider::Google => {
385 let api_key = state
386 .config
387 .google_ai_api_key
388 .as_ref()
389 .context("no Google AI API key configured on the server")?;
390 let chunks = google_ai::stream_generate_content(
391 &state.http_client,
392 google_ai::API_URL,
393 api_key,
394 serde_json::from_str(params.provider_request.get())?,
395 None,
396 )
397 .await?;
398
399 chunks
400 .map(|event| {
401 event.map(|chunk| {
402 // TODO - implement token counting for Google AI
403 CompletionChunk {
404 bytes: serde_json::to_vec(&chunk).unwrap(),
405 input_tokens: 0,
406 output_tokens: 0,
407 cache_creation_input_tokens: 0,
408 cache_read_input_tokens: 0,
409 }
410 })
411 })
412 .boxed()
413 }
414 };
415
416 Ok(Response::new(Body::wrap_stream(TokenCountingStream {
417 state,
418 claims,
419 provider: params.provider,
420 model,
421 input_tokens: 0,
422 output_tokens: 0,
423 cache_creation_input_tokens: 0,
424 cache_read_input_tokens: 0,
425 inner_stream: stream,
426 })))
427}
428
429fn normalize_model_name(known_models: Vec<String>, name: String) -> String {
430 if let Some(known_model_name) = known_models
431 .iter()
432 .filter(|known_model_name| name.starts_with(known_model_name.as_str()))
433 .max_by_key(|known_model_name| known_model_name.len())
434 {
435 known_model_name.to_string()
436 } else {
437 name
438 }
439}
440
441/// The maximum monthly spending an individual user can reach on the free tier
442/// before they have to pay.
443pub const FREE_TIER_MONTHLY_SPENDING_LIMIT: Cents = Cents::from_dollars(5);
444
445/// The default value to use for maximum spend per month if the user did not
446/// explicitly set a maximum spend.
447///
448/// Used to prevent surprise bills.
449pub const DEFAULT_MAX_MONTHLY_SPEND: Cents = Cents::from_dollars(10);
450
451/// The maximum lifetime spending an individual user can reach before being cut off.
452const LIFETIME_SPENDING_LIMIT: Cents = Cents::from_dollars(1_000);
453
454async fn check_usage_limit(
455 state: &Arc<LlmState>,
456 provider: LanguageModelProvider,
457 model_name: &str,
458 claims: &LlmTokenClaims,
459) -> Result<()> {
460 let model = state.db.model(provider, model_name)?;
461 let usage = state
462 .db
463 .get_usage(
464 UserId::from_proto(claims.user_id),
465 provider,
466 model_name,
467 Utc::now(),
468 )
469 .await?;
470
471 if state.config.is_llm_billing_enabled() {
472 if usage.spending_this_month >= FREE_TIER_MONTHLY_SPENDING_LIMIT {
473 if !claims.has_llm_subscription {
474 return Err(Error::http(
475 StatusCode::PAYMENT_REQUIRED,
476 "Maximum spending limit reached for this month.".to_string(),
477 ));
478 }
479 }
480 }
481
482 // TODO: Remove this once we've rolled out monthly spending limits.
483 if usage.lifetime_spending >= LIFETIME_SPENDING_LIMIT {
484 return Err(Error::http(
485 StatusCode::FORBIDDEN,
486 "Maximum spending limit reached.".to_string(),
487 ));
488 }
489
490 let active_users = state.get_active_user_count(provider, model_name).await?;
491
492 let users_in_recent_minutes = active_users.users_in_recent_minutes.max(1);
493 let users_in_recent_days = active_users.users_in_recent_days.max(1);
494
495 let per_user_max_requests_per_minute =
496 model.max_requests_per_minute as usize / users_in_recent_minutes;
497 let per_user_max_tokens_per_minute =
498 model.max_tokens_per_minute as usize / users_in_recent_minutes;
499 let per_user_max_tokens_per_day = model.max_tokens_per_day as usize / users_in_recent_days;
500
501 let checks = [
502 (
503 usage.requests_this_minute,
504 per_user_max_requests_per_minute,
505 UsageMeasure::RequestsPerMinute,
506 ),
507 (
508 usage.tokens_this_minute,
509 per_user_max_tokens_per_minute,
510 UsageMeasure::TokensPerMinute,
511 ),
512 (
513 usage.tokens_this_day,
514 per_user_max_tokens_per_day,
515 UsageMeasure::TokensPerDay,
516 ),
517 ];
518
519 for (used, limit, usage_measure) in checks {
520 // Temporarily bypass rate-limiting for staff members.
521 if claims.is_staff {
522 continue;
523 }
524
525 if used > limit {
526 let resource = match usage_measure {
527 UsageMeasure::RequestsPerMinute => "requests_per_minute",
528 UsageMeasure::TokensPerMinute => "tokens_per_minute",
529 UsageMeasure::TokensPerDay => "tokens_per_day",
530 };
531
532 if let Some(client) = state.clickhouse_client.as_ref() {
533 tracing::info!(
534 target: "user rate limit",
535 user_id = claims.user_id,
536 login = claims.github_user_login,
537 authn.jti = claims.jti,
538 is_staff = claims.is_staff,
539 provider = provider.to_string(),
540 model = model.name,
541 requests_this_minute = usage.requests_this_minute,
542 tokens_this_minute = usage.tokens_this_minute,
543 tokens_this_day = usage.tokens_this_day,
544 users_in_recent_minutes = users_in_recent_minutes,
545 users_in_recent_days = users_in_recent_days,
546 max_requests_per_minute = per_user_max_requests_per_minute,
547 max_tokens_per_minute = per_user_max_tokens_per_minute,
548 max_tokens_per_day = per_user_max_tokens_per_day,
549 );
550
551 report_llm_rate_limit(
552 client,
553 LlmRateLimitEventRow {
554 time: Utc::now().timestamp_millis(),
555 user_id: claims.user_id as i32,
556 is_staff: claims.is_staff,
557 plan: match claims.plan {
558 Plan::Free => "free".to_string(),
559 Plan::ZedPro => "zed_pro".to_string(),
560 },
561 model: model.name.clone(),
562 provider: provider.to_string(),
563 usage_measure: resource.to_string(),
564 requests_this_minute: usage.requests_this_minute as u64,
565 tokens_this_minute: usage.tokens_this_minute as u64,
566 tokens_this_day: usage.tokens_this_day as u64,
567 users_in_recent_minutes: users_in_recent_minutes as u64,
568 users_in_recent_days: users_in_recent_days as u64,
569 max_requests_per_minute: per_user_max_requests_per_minute as u64,
570 max_tokens_per_minute: per_user_max_tokens_per_minute as u64,
571 max_tokens_per_day: per_user_max_tokens_per_day as u64,
572 },
573 )
574 .await
575 .log_err();
576 }
577
578 return Err(Error::http(
579 StatusCode::TOO_MANY_REQUESTS,
580 format!("Rate limit exceeded. Maximum {} reached.", resource),
581 ));
582 }
583 }
584
585 Ok(())
586}
587
588struct CompletionChunk {
589 bytes: Vec<u8>,
590 input_tokens: usize,
591 output_tokens: usize,
592 cache_creation_input_tokens: usize,
593 cache_read_input_tokens: usize,
594}
595
596struct TokenCountingStream<S> {
597 state: Arc<LlmState>,
598 claims: LlmTokenClaims,
599 provider: LanguageModelProvider,
600 model: String,
601 input_tokens: usize,
602 output_tokens: usize,
603 cache_creation_input_tokens: usize,
604 cache_read_input_tokens: usize,
605 inner_stream: S,
606}
607
608impl<S> Stream for TokenCountingStream<S>
609where
610 S: Stream<Item = Result<CompletionChunk, anyhow::Error>> + Unpin,
611{
612 type Item = Result<Vec<u8>, anyhow::Error>;
613
614 fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
615 match Pin::new(&mut self.inner_stream).poll_next(cx) {
616 Poll::Ready(Some(Ok(mut chunk))) => {
617 chunk.bytes.push(b'\n');
618 self.input_tokens += chunk.input_tokens;
619 self.output_tokens += chunk.output_tokens;
620 self.cache_creation_input_tokens += chunk.cache_creation_input_tokens;
621 self.cache_read_input_tokens += chunk.cache_read_input_tokens;
622 Poll::Ready(Some(Ok(chunk.bytes)))
623 }
624 Poll::Ready(Some(Err(e))) => Poll::Ready(Some(Err(e))),
625 Poll::Ready(None) => Poll::Ready(None),
626 Poll::Pending => Poll::Pending,
627 }
628 }
629}
630
631impl<S> Drop for TokenCountingStream<S> {
632 fn drop(&mut self) {
633 let state = self.state.clone();
634 let claims = self.claims.clone();
635 let provider = self.provider;
636 let model = std::mem::take(&mut self.model);
637 let input_token_count = self.input_tokens;
638 let output_token_count = self.output_tokens;
639 let cache_creation_input_token_count = self.cache_creation_input_tokens;
640 let cache_read_input_token_count = self.cache_read_input_tokens;
641 self.state.executor.spawn_detached(async move {
642 let usage = state
643 .db
644 .record_usage(
645 UserId::from_proto(claims.user_id),
646 claims.is_staff,
647 provider,
648 &model,
649 input_token_count,
650 cache_creation_input_token_count,
651 cache_read_input_token_count,
652 output_token_count,
653 Utc::now(),
654 )
655 .await
656 .log_err();
657
658 if let Some(usage) = usage {
659 tracing::info!(
660 target: "user usage",
661 user_id = claims.user_id,
662 login = claims.github_user_login,
663 authn.jti = claims.jti,
664 is_staff = claims.is_staff,
665 requests_this_minute = usage.requests_this_minute,
666 tokens_this_minute = usage.tokens_this_minute,
667 );
668
669 if let Some(clickhouse_client) = state.clickhouse_client.as_ref() {
670 report_llm_usage(
671 clickhouse_client,
672 LlmUsageEventRow {
673 time: Utc::now().timestamp_millis(),
674 user_id: claims.user_id as i32,
675 is_staff: claims.is_staff,
676 plan: match claims.plan {
677 Plan::Free => "free".to_string(),
678 Plan::ZedPro => "zed_pro".to_string(),
679 },
680 model,
681 provider: provider.to_string(),
682 input_token_count: input_token_count as u64,
683 cache_creation_input_token_count: cache_creation_input_token_count
684 as u64,
685 cache_read_input_token_count: cache_read_input_token_count as u64,
686 output_token_count: output_token_count as u64,
687 requests_this_minute: usage.requests_this_minute as u64,
688 tokens_this_minute: usage.tokens_this_minute as u64,
689 tokens_this_day: usage.tokens_this_day as u64,
690 input_tokens_this_month: usage.input_tokens_this_month as u64,
691 cache_creation_input_tokens_this_month: usage
692 .cache_creation_input_tokens_this_month
693 as u64,
694 cache_read_input_tokens_this_month: usage
695 .cache_read_input_tokens_this_month
696 as u64,
697 output_tokens_this_month: usage.output_tokens_this_month as u64,
698 spending_this_month: usage.spending_this_month.0 as u64,
699 lifetime_spending: usage.lifetime_spending.0 as u64,
700 },
701 )
702 .await
703 .log_err();
704 }
705 }
706 })
707 }
708}
709
710pub fn log_usage_periodically(state: Arc<LlmState>) {
711 state.executor.clone().spawn_detached(async move {
712 loop {
713 state
714 .executor
715 .sleep(std::time::Duration::from_secs(30))
716 .await;
717
718 for provider in LanguageModelProvider::iter() {
719 for model in state.db.model_names_for_provider(provider) {
720 if let Some(active_user_count) = state
721 .get_active_user_count(provider, &model)
722 .await
723 .log_err()
724 {
725 tracing::info!(
726 target: "active user counts",
727 provider = provider.to_string(),
728 model = model,
729 users_in_recent_minutes = active_user_count.users_in_recent_minutes,
730 users_in_recent_days = active_user_count.users_in_recent_days,
731 );
732 }
733 }
734 }
735
736 if let Some(usages) = state
737 .db
738 .get_application_wide_usages_by_model(Utc::now())
739 .await
740 .log_err()
741 {
742 for usage in usages {
743 tracing::info!(
744 target: "computed usage",
745 provider = usage.provider.to_string(),
746 model = usage.model,
747 requests_this_minute = usage.requests_this_minute,
748 tokens_this_minute = usage.tokens_this_minute,
749 );
750 }
751 }
752 }
753 })
754}