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

systemprompt_analytics/repository/fingerprint/
mutations.rs
