systemprompt_content/repository/link/
analytics.rs

1//! Link click and analytics repository.
2//!
3//! [`LinkAnalyticsRepository`] records click events and serves the aggregate
4//! click/conversion views over `link_clicks` and `campaign_links`, maintaining
5//! the denormalised counters on the link row as clicks arrive.
6//!
7//! Copyright (c) systemprompt.io — Business Source License 1.1.
8//! See <https://systemprompt.io> for licensing details.
9
10use crate::error::ContentError;
11use crate::models::{
12    CampaignPerformance, ContentJourneyNode, LinkClick, LinkPerformance, RecordClickParams,
13};
14use sqlx::PgPool;
15use std::sync::Arc;
16use systemprompt_database::DbPool;
17use systemprompt_identifiers::{
18    CampaignId, ContentId, ContextId, LinkClickId, LinkId, SessionId, TaskId, UserId,
19};
20
21#[derive(Debug, Clone)]
22pub struct LinkAnalyticsRepository {
23    pool: Arc<PgPool>,
24    write_pool: Arc<PgPool>,
25}
26
27impl LinkAnalyticsRepository {
28    pub fn new(db: &DbPool) -> Result<Self, ContentError> {
29        let pool = db.pool_arc().map_err(ContentError::Repository)?;
30        let write_pool = db.write_pool_arc().map_err(ContentError::Repository)?;
31        Ok(Self { pool, write_pool })
32    }
33
34    pub async fn get_link_performance(
35        &self,
36        link_id: &LinkId,
37    ) -> Result<Option<LinkPerformance>, sqlx::Error> {
38        sqlx::query_as!(
39            LinkPerformance,
40            r#"
41            SELECT
42                l.id as "link_id: LinkId",
43                COALESCE(l.click_count, 0)::bigint as "click_count!",
44                COALESCE(l.unique_click_count, 0)::bigint as "unique_click_count!",
45                COALESCE(l.conversion_count, 0)::bigint as "conversion_count!",
46                CASE
47                    WHEN COALESCE(l.click_count, 0) > 0 THEN
48                        COALESCE(l.conversion_count, 0)::float / l.click_count
49                    ELSE 0.0
50                END as conversion_rate
51            FROM campaign_links l
52            WHERE l.id = $1
53            "#,
54            link_id.as_str()
55        )
56        .fetch_optional(&*self.pool)
57        .await
58    }
59
60    pub async fn check_session_clicked_link(
61        &self,
62        link_id: &LinkId,
63        session_id: &SessionId,
64    ) -> Result<bool, sqlx::Error> {
65        let result = sqlx::query!(
66            r#"SELECT COALESCE(COUNT(*), 0)::bigint as "count!" FROM link_clicks WHERE link_id = $1 AND session_id = $2"#,
67            link_id.as_str(),
68            session_id.as_str()
69        )
70        .fetch_one(&*self.pool)
71        .await?;
72
73        Ok(result.count > 0)
74    }
75
76    pub async fn increment_link_clicks(
77        &self,
78        link_id: &LinkId,
79        is_first_click: bool,
80    ) -> Result<(), sqlx::Error> {
81        if is_first_click {
82            sqlx::query!(
83                "UPDATE campaign_links SET click_count = click_count + 1, unique_click_count = \
84                 unique_click_count + 1 WHERE id = $1",
85                link_id.as_str()
86            )
87            .execute(&*self.write_pool)
88            .await?;
89        } else {
90            sqlx::query!(
91                "UPDATE campaign_links SET click_count = click_count + 1 WHERE id = $1",
92                link_id.as_str()
93            )
94            .execute(&*self.write_pool)
95            .await?;
96        }
97        Ok(())
98    }
99
100    pub async fn get_clicks_by_link(
101        &self,
102        link_id: &LinkId,
103        limit: i64,
104        offset: i64,
105    ) -> Result<Vec<LinkClick>, sqlx::Error> {
106        sqlx::query_as!(
107            LinkClick,
108            r#"
109            SELECT id as "id: LinkClickId", link_id as "link_id: LinkId",
110                   session_id as "session_id: SessionId", user_id as "user_id: UserId",
111                   context_id as "context_id: ContextId", task_id as "task_id: TaskId",
112                   referrer_page, referrer_url, clicked_at, user_agent, ip_address,
113                   device_type, country, is_first_click, is_conversion, conversion_at,
114                   time_on_page_seconds, scroll_depth_percent
115            FROM link_clicks
116            WHERE link_id = $1
117            ORDER BY clicked_at DESC
118            LIMIT $2 OFFSET $3
119            "#,
120            link_id.as_str(),
121            limit,
122            offset
123        )
124        .fetch_all(&*self.pool)
125        .await
126    }
127
128    pub async fn get_content_journey_map(
129        &self,
130        limit: i64,
131        offset: i64,
132    ) -> Result<Vec<ContentJourneyNode>, sqlx::Error> {
133        let rows = sqlx::query!(
134            r#"
135            SELECT source_content_id, target_url, COALESCE(click_count, 0) as "click_count!"
136            FROM campaign_links
137            WHERE source_content_id IS NOT NULL AND click_count > 0
138            ORDER BY click_count DESC
139            LIMIT $1 OFFSET $2
140            "#,
141            limit,
142            offset
143        )
144        .fetch_all(&*self.pool)
145        .await?;
146
147        Ok(rows
148            .into_iter()
149            .filter_map(|r| {
150                Some(ContentJourneyNode {
151                    source_content_id: ContentId::new(r.source_content_id?),
152                    target_url: r.target_url,
153                    click_count: r.click_count,
154                })
155            })
156            .collect())
157    }
158
159    pub async fn get_campaign_performance(
160        &self,
161        campaign_id: &CampaignId,
162    ) -> Result<Option<CampaignPerformance>, sqlx::Error> {
163        sqlx::query_as!(
164            CampaignPerformance,
165            r#"
166            SELECT
167                campaign_id as "campaign_id!: CampaignId",
168                COALESCE(SUM(click_count), 0)::bigint as "total_clicks!",
169                COUNT(*)::bigint as "link_count!",
170                COUNT(DISTINCT source_content_id) as unique_visitors,
171                COALESCE(SUM(conversion_count), 0)::bigint as conversion_count
172            FROM campaign_links
173            WHERE campaign_id = $1
174            GROUP BY campaign_id
175            "#,
176            campaign_id.as_str()
177        )
178        .fetch_optional(&*self.pool)
179        .await
180    }
181
182    pub async fn record_click(&self, params: &RecordClickParams) -> Result<(), sqlx::Error> {
183        sqlx::query!(
184            r#"
185            INSERT INTO link_clicks (
186                id, link_id, session_id, user_id, context_id, task_id,
187                referrer_page, referrer_url, clicked_at, user_agent, ip_address,
188                device_type, country, is_first_click, is_conversion
189            )
190            VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15)
191            "#,
192            params.click_id.as_str(),
193            params.link_id.as_str(),
194            params.session_id.as_str(),
195            params.user_id.as_ref().map(UserId::as_str),
196            params.context_id.as_ref().map(ContextId::as_str),
197            params.task_id.as_ref().map(TaskId::as_str),
198            params.referrer_page,
199            params.referrer_url,
200            params.clicked_at,
201            params.user_agent,
202            params.ip_address,
203            params.device_type,
204            params.country,
205            params.is_first_click,
206            params.is_conversion
207        )
208        .execute(&*self.write_pool)
209        .await?;
210        Ok(())
211    }
212}
systemprompt_content/repository/link/analytics.rs

systemprompt_content/repository/link/
analytics.rs
