systemprompt_analytics/repository/fingerprint/
mutations.rs

1//! Fingerprint-reputation state changes for `FingerprintRepository`.
2//!
3//! Upserts a fingerprint on each session, updates velocity and session-count
4//! metrics, raises and clears abuse flags, and adjusts the reputation score.
5//! All writes go to the write pool.
6//!
7//! Copyright (c) systemprompt.io — Business Source License 1.1.
8//! See <https://systemprompt.io> for licensing details.
9
10use crate::Result;
11use systemprompt_identifiers::UserId;
12
13use super::FingerprintRepository;
14use crate::models::{FingerprintReputation, FlagReason};
15
16impl FingerprintRepository {
17    pub async fn upsert_fingerprint(
18        &self,
19        fingerprint_hash: &str,
20        ip_address: Option<&str>,
21        user_agent: Option<&str>,
22        user_id: Option<&UserId>,
23    ) -> Result<FingerprintReputation> {
24        let user_ids = user_id.map_or_else(Vec::new, |u| vec![u.as_str().to_owned()]);
25
26        let row = sqlx::query_as!(
27            FingerprintReputation,
28            r#"
29            INSERT INTO fingerprint_reputation (
30                fingerprint_hash, last_ip_address, last_user_agent,
31                associated_user_ids, total_session_count
32            )
33            VALUES ($1, $2, $3, $4, 1)
34            ON CONFLICT (fingerprint_hash) DO UPDATE SET
35                last_seen_at = CURRENT_TIMESTAMP,
36                last_ip_address = COALESCE($2, fingerprint_reputation.last_ip_address),
37                last_user_agent = COALESCE($3, fingerprint_reputation.last_user_agent),
38                total_session_count = fingerprint_reputation.total_session_count + 1,
39                associated_user_ids = CASE
40                    WHEN array_length($4, 1) > 0 AND NOT ($4[1] = ANY(fingerprint_reputation.associated_user_ids))
41                    THEN array_cat(fingerprint_reputation.associated_user_ids, $4)
42                    ELSE fingerprint_reputation.associated_user_ids
43                END,
44                updated_at = CURRENT_TIMESTAMP
45            RETURNING
46                fingerprint_hash,
47                first_seen_at,
48                last_seen_at,
49                total_session_count,
50                active_session_count,
51                total_request_count,
52                requests_last_hour,
53                peak_requests_per_minute,
54                sustained_high_velocity_minutes,
55                is_flagged,
56                flag_reason,
57                flagged_at,
58                reputation_score,
59                abuse_incidents,
60                last_abuse_at,
61                last_ip_address,
62                last_user_agent,
63                associated_user_ids,
64                updated_at
65            "#,
66            fingerprint_hash,
67            ip_address,
68            user_agent,
69            &user_ids[..],
70        )
71        .fetch_one(&*self.write_pool)
72        .await?;
73
74        Ok(row)
75    }
76
77    pub async fn flag_fingerprint(
78        &self,
79        fingerprint_hash: &str,
80        reason: FlagReason,
81        new_score: i32,
82    ) -> Result<()> {
83        sqlx::query!(
84            r#"
85            UPDATE fingerprint_reputation
86            SET is_flagged = TRUE,
87                flag_reason = $2,
88                flagged_at = CURRENT_TIMESTAMP,
89                reputation_score = $3,
90                abuse_incidents = abuse_incidents + 1,
91                last_abuse_at = CURRENT_TIMESTAMP,
92                updated_at = CURRENT_TIMESTAMP
93            WHERE fingerprint_hash = $1
94            "#,
95            fingerprint_hash,
96            reason.as_str(),
97            new_score,
98        )
99        .execute(&*self.write_pool)
100        .await?;
101
102        Ok(())
103    }
104
105    pub async fn update_velocity_metrics(
106        &self,
107        fingerprint_hash: &str,
108        requests_last_hour: i32,
109        peak_requests_per_minute: f32,
110        sustained_high_velocity_minutes: i32,
111    ) -> Result<()> {
112        sqlx::query!(
113            r#"
114            UPDATE fingerprint_reputation
115            SET requests_last_hour = $2,
116                peak_requests_per_minute = $3,
117                sustained_high_velocity_minutes = $4,
118                total_request_count = total_request_count + 1,
119                updated_at = CURRENT_TIMESTAMP
120            WHERE fingerprint_hash = $1
121            "#,
122            fingerprint_hash,
123            requests_last_hour,
124            peak_requests_per_minute,
125            sustained_high_velocity_minutes,
126        )
127        .execute(&*self.write_pool)
128        .await?;
129
130        Ok(())
131    }
132
133    pub async fn update_active_session_count(
134        &self,
135        fingerprint_hash: &str,
136        active_count: i32,
137    ) -> Result<()> {
138        sqlx::query!(
139            r#"
140            UPDATE fingerprint_reputation
141            SET active_session_count = $2,
142                updated_at = CURRENT_TIMESTAMP
143            WHERE fingerprint_hash = $1
144            "#,
145            fingerprint_hash,
146            active_count,
147        )
148        .execute(&*self.write_pool)
149        .await?;
150
151        Ok(())
152    }
153
154    pub async fn increment_request_count(&self, fingerprint_hash: &str) -> Result<()> {
155        sqlx::query!(
156            r#"
157            UPDATE fingerprint_reputation
158            SET total_request_count = total_request_count + 1,
159                updated_at = CURRENT_TIMESTAMP
160            WHERE fingerprint_hash = $1
161            "#,
162            fingerprint_hash,
163        )
164        .execute(&*self.write_pool)
165        .await?;
166
167        Ok(())
168    }
169
170    pub async fn clear_flag(&self, fingerprint_hash: &str) -> Result<()> {
171        sqlx::query!(
172            r#"
173            UPDATE fingerprint_reputation
174            SET is_flagged = FALSE,
175                flag_reason = NULL,
176                flagged_at = NULL,
177                updated_at = CURRENT_TIMESTAMP
178            WHERE fingerprint_hash = $1
179            "#,
180            fingerprint_hash,
181        )
182        .execute(&*self.write_pool)
183        .await?;
184
185        Ok(())
186    }
187
188    pub async fn adjust_reputation_score(&self, fingerprint_hash: &str, delta: i32) -> Result<i32> {
189        let row = sqlx::query_scalar!(
190            r#"
191            UPDATE fingerprint_reputation
192            SET reputation_score = GREATEST(0, LEAST(100, reputation_score + $2)),
193                updated_at = CURRENT_TIMESTAMP
194            WHERE fingerprint_hash = $1
195            RETURNING reputation_score as "reputation_score!"
196            "#,
197            fingerprint_hash,
198            delta,
199        )
200        .fetch_one(&*self.write_pool)
201        .await?;
202
203        Ok(row)
204    }
205}
systemprompt_analytics/repository/fingerprint/mutations.rs

systemprompt_analytics/repository/fingerprint/
mutations.rs
