systemprompt_analytics/feedback/
totals.rs

1//! Raw-reference totals deduplicate shared requests and conversation
2//! assessments.
3//!
4//! Copyright (c) systemprompt.io — Business Source License 1.1.
5//! See <https://systemprompt.io> for licensing details.
6
7use super::{FactsTotals, FeedbackFactsRepository, validation};
8use crate::Result;
9use chrono::{DateTime, Utc};
10use systemprompt_identifiers::{ManagedResourceId, UserId};
11
12impl FeedbackFactsRepository {
13    pub async fn reference_totals(
14        &self,
15        owner: &UserId,
16        from: DateTime<Utc>,
17        to: DateTime<Utc>,
18        resource: Option<&ManagedResourceId>,
19    ) -> Result<FactsTotals> {
20        if from >= to {
21            return Err(validation::invalid());
22        }
23        let resource = resource.map(ManagedResourceId::as_str);
24        let mut tx = self.pool.begin().await?;
25        sqlx::query!("SET TRANSACTION ISOLATION LEVEL REPEATABLE READ READ ONLY")
26            .execute(&mut *tx)
27            .await?;
28        let invocations = sqlx::query!(r#"SELECT COUNT(*) AS "total!",COUNT(*) FILTER(WHERE resource_id IS NOT NULL) AS "verified!" FROM analytics_normalized_facts WHERE owner_id=$1 AND fact_kind='invocation' AND NOT deleted AND occurred_at>=$2 AND occurred_at<$3 AND ($4::text IS NULL OR resource_id=$4)"#, owner.as_str(), from, to, resource).fetch_one(&mut *tx).await?;
29        let requests = sqlx::query!(r#"SELECT COUNT(*) AS "total!",COUNT(*) FILTER(WHERE NOT succeeded) AS "failed!",COUNT(*) FILTER(WHERE amount_micros IS NOT NULL) AS "priced!",COUNT(*) FILTER(WHERE latency_micros IS NOT NULL) AS "measured!",COUNT(*) FILTER(WHERE input_tokens IS NOT NULL AND output_tokens IS NOT NULL) AS "tokens_measured!" FROM analytics_normalized_facts r WHERE owner_id=$1 AND fact_kind='request' AND NOT deleted AND occurred_at>=$2 AND occurred_at<$3 AND ($4::text IS NULL OR EXISTS(SELECT 1 FROM analytics_normalized_facts a WHERE a.owner_id=r.owner_id AND a.fact_kind='resource_association' AND NOT a.deleted AND a.resource_id=$4 AND a.request_source=r.source AND a.request_id=r.fact_id))"#, owner.as_str(), from, to, resource).fetch_one(&mut *tx).await?;
30        let spend = sqlx::query!(r#"SELECT currency AS "currency!",SUM(amount_micros)::text AS "amount!" FROM analytics_normalized_facts r WHERE owner_id=$1 AND fact_kind='request' AND NOT deleted AND occurred_at>=$2 AND occurred_at<$3 AND currency IS NOT NULL AND ($4::text IS NULL OR EXISTS(SELECT 1 FROM analytics_normalized_facts a WHERE a.owner_id=r.owner_id AND a.fact_kind='resource_association' AND NOT a.deleted AND a.resource_id=$4 AND a.request_source=r.source AND a.request_id=r.fact_id)) GROUP BY currency"#, owner.as_str(), from, to, resource).fetch_all(&mut *tx).await?;
31        let assessments = sqlx::query!(r#"SELECT COUNT(DISTINCT (conversation_source,conversation_id)) AS "total!",COUNT(DISTINCT (conversation_source,conversation_id)) FILTER(WHERE assessment_status='scored') AS "scored!",COUNT(DISTINCT (conversation_source,conversation_id)) FILTER(WHERE assessment_status='failed') AS "failed!" FROM (SELECT DISTINCT ON (conversation_source,conversation_id) * FROM analytics_normalized_facts WHERE owner_id=$1 AND fact_kind='assessment' AND NOT deleted ORDER BY conversation_source,conversation_id,occurred_at DESC,source,fact_id) a WHERE occurred_at>=$2 AND occurred_at<$3 AND ($4::text IS NULL OR EXISTS(SELECT 1 FROM analytics_normalized_facts i WHERE i.owner_id=a.owner_id AND i.fact_kind='invocation' AND NOT i.deleted AND i.resource_id=$4 AND i.source=a.invocation_source AND i.fact_id=a.invocation_id))"#, owner.as_str(), from, to, resource).fetch_one(&mut *tx).await?;
32        let mut totals = FactsTotals {
33            invocations: invocations.total,
34            verified_invocations: invocations.verified,
35            requests: requests.total,
36            failed_requests: requests.failed,
37            priced_requests: requests.priced,
38            latency_measured_requests: requests.measured,
39            token_measured_requests: requests.tokens_measured,
40            assessed_conversations: assessments.scored,
41            assessment_conversations: assessments.total,
42            failed_assessments: assessments.failed,
43            related_spend_non_additive: resource.is_some(),
44            ..FactsTotals::default()
45        };
46        for row in spend {
47            totals.spend_by_currency.insert(
48                row.currency,
49                row.amount.parse().map_err(|_error| validation::invalid())?,
50            );
51        }
52        tx.commit().await?;
53        Ok(totals)
54    }
55}
systemprompt_analytics/feedback/totals.rs

systemprompt_analytics/feedback/
totals.rs
