systemprompt_analytics/repository/funnel/
stats.rs

1use anyhow::Result;
2use chrono::{DateTime, Utc};
3use systemprompt_identifiers::FunnelId;
4
5use super::FunnelRepository;
6use crate::models::{FunnelStats, FunnelStep, FunnelStepStats};
7
8impl FunnelRepository {
9    pub async fn get_stats(
10        &self,
11        funnel_id: &FunnelId,
12        since: DateTime<Utc>,
13    ) -> Result<FunnelStats> {
14        let funnel = self
15            .find_by_id(funnel_id)
16            .await?
17            .ok_or_else(|| anyhow::anyhow!("Funnel not found: {}", funnel_id))?;
18
19        let total_entries = sqlx::query_scalar!(
20            r#"
21            SELECT COUNT(*) as "count!"
22            FROM funnel_progress
23            WHERE funnel_id = $1 AND created_at >= $2
24            "#,
25            funnel_id.as_str(),
26            since
27        )
28        .fetch_one(&*self.pool)
29        .await?;
30
31        let total_completions = sqlx::query_scalar!(
32            r#"
33            SELECT COUNT(*) as "count!"
34            FROM funnel_progress
35            WHERE funnel_id = $1 AND created_at >= $2 AND completed_at IS NOT NULL
36            "#,
37            funnel_id.as_str(),
38            since
39        )
40        .fetch_one(&*self.pool)
41        .await?;
42
43        let overall_conversion_rate = if total_entries > 0 {
44            (total_completions as f64 / total_entries as f64) * 100.0
45        } else {
46            0.0
47        };
48
49        let step_stats = self
50            .calculate_step_stats(funnel_id, &funnel.steps, since)
51            .await?;
52
53        Ok(FunnelStats {
54            funnel_id: funnel_id.clone(),
55            funnel_name: funnel.funnel.name,
56            total_entries,
57            total_completions,
58            overall_conversion_rate,
59            step_stats,
60        })
61    }
62
63    async fn calculate_step_stats(
64        &self,
65        funnel_id: &FunnelId,
66        steps: &[FunnelStep],
67        since: DateTime<Utc>,
68    ) -> Result<Vec<FunnelStepStats>> {
69        if steps.is_empty() {
70            return Ok(Vec::new());
71        }
72
73        let step_orders: Vec<i32> = steps.iter().map(|s| s.step_order).collect();
74
75        let rows = sqlx::query!(
76            r#"
77            SELECT
78                s.step_order,
79                COUNT(*) FILTER (WHERE fp.current_step >= s.step_order) as "entered_count!",
80                COUNT(*) FILTER (WHERE fp.current_step > s.step_order) as "exited_count!"
81            FROM UNNEST($3::int4[]) AS s(step_order)
82            LEFT JOIN funnel_progress fp
83                ON fp.funnel_id = $1 AND fp.created_at >= $2
84            GROUP BY s.step_order
85            ORDER BY s.step_order
86            "#,
87            funnel_id.as_str(),
88            since,
89            &step_orders
90        )
91        .fetch_all(&*self.pool)
92        .await?;
93
94        let mut stats = Vec::with_capacity(steps.len());
95        for row in rows {
96            let entered_count = row.entered_count;
97            let exited_count = row.exited_count;
98            let conversion_rate = if entered_count > 0 {
99                (exited_count as f64 / entered_count as f64) * 100.0
100            } else {
101                0.0
102            };
103
104            stats.push(FunnelStepStats {
105                step_order: row.step_order.unwrap_or(0),
106                entered_count,
107                exited_count,
108                conversion_rate,
109                avg_time_to_next_ms: None,
110            });
111        }
112
113        Ok(stats)
114    }
115}
systemprompt_analytics/repository/funnel/stats.rs

systemprompt_analytics/repository/funnel/
stats.rs
