systemprompt-analytics 0.53.0

Analytics for systemprompt.io AI governance infrastructure. Session, agent, tool, and microdollar-precision cost attribution across the MCP governance pipeline.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52

//! Session geolocation enrichment through the session owner.
//!
//! Copyright (c) systemprompt.io — Business Source License 1.1.
//! See <https://systemprompt.io> for licensing details.

use super::SessionRepository;
use crate::{AnalyticsError, GeoIpReader, Result};
use systemprompt_identifiers::SessionId;

impl SessionRepository {
    pub(super) async fn backfill_geo(
        &self,
        reader: Option<&GeoIpReader>,
        batch_size: i64,
    ) -> Result<u64> {
        if batch_size <= 0 {
            return Err(AnalyticsError::invalid_argument(
                "batch size must be positive",
            ));
        }
        let mut updated = 0;
        let mut after: Option<SessionId> = None;
        loop {
            let rows = self
                .owner
                .sessions_missing_geo(after.as_ref(), batch_size)
                .await
                .map_err(AnalyticsError::from)?;
            let Some(last) = rows.last() else {
                break;
            };
            after = Some(last.0.clone());
            for (session_id, ip) in rows {
                if let Some((country, region, city)) =
                    crate::services::extractor::geoip::lookup_geoip(&ip, reader)
                {
                    updated += self
                        .owner
                        .set_session_geo(
                            &session_id,
                            country.as_deref(),
                            region.as_deref(),
                            city.as_deref(),
                        )
                        .await
                        .map_err(AnalyticsError::from)?;
                }
            }
        }
        Ok(updated)
    }
}