systemprompt_users/repository/user/
operations.rs

1//! User row creation and update operations.
2//!
3//! Copyright (c) systemprompt.io — Business Source License 1.1.
4//! See <https://systemprompt.io> for licensing details.
5
6use chrono::{Duration, Utc};
7use systemprompt_identifiers::UserId;
8
9use crate::error::{Result, UserError};
10use crate::models::{User, UserRole, UserStatus, normalise_email};
11use crate::repository::UserRepository;
12
13#[derive(Debug)]
14pub struct UpdateUserParams<'a> {
15    pub email: &'a str,
16    pub full_name: Option<&'a str>,
17    pub display_name: Option<&'a str>,
18    pub status: UserStatus,
19}
20
21impl UserRepository {
22    pub async fn create(
23        &self,
24        name: &str,
25        email: &str,
26        full_name: Option<&str>,
27        display_name: Option<&str>,
28    ) -> Result<User> {
29        let now = Utc::now();
30        let id = UserId::new(uuid::Uuid::new_v4().to_string());
31        let display_name_val = display_name.or(full_name);
32        let status = UserStatus::Active.as_str();
33        let role = UserRole::User.as_str();
34        let email = normalise_email(email);
35
36        let row = sqlx::query_as!(
37            User,
38            r#"
39            INSERT INTO users (
40                id, name, email, full_name, display_name,
41                status, email_verified, roles, is_bot,
42                created_at, updated_at
43            )
44            VALUES ($1, $2, $3, $4, $5, $6, false, ARRAY[$7]::TEXT[], false, $8, $8)
45            RETURNING id, name, email, full_name, display_name, status, email_verified,
46                      roles, avatar_url, is_bot, is_scanner, created_at, updated_at
47            "#,
48            id.as_str(),
49            name,
50            email,
51            full_name,
52            display_name_val,
53            status,
54            role,
55            now
56        )
57        .fetch_one(&*self.write_pool)
58        .await?;
59
60        Ok(row)
61    }
62
63    /// Inserts the user, yielding `None` when one already holds that name or
64    /// email.
65    ///
66    /// `create` is the right call when the caller owns the identity it is
67    /// minting. This one is for the auto-provisioning paths, where several
68    /// processes resolve the *same* well-known identity — the local-trial
69    /// `admin` — and would otherwise each read "absent" and then race to
70    /// insert it, leaving every loser with a unique-violation the caller can
71    /// only tell apart by matching on the driver's error text.
72    pub async fn create_if_absent(
73        &self,
74        name: &str,
75        email: &str,
76        full_name: Option<&str>,
77        display_name: Option<&str>,
78    ) -> Result<Option<User>> {
79        let now = Utc::now();
80        let id = UserId::new(uuid::Uuid::new_v4().to_string());
81        let display_name_val = display_name.or(full_name);
82        let status = UserStatus::Active.as_str();
83        let role = UserRole::User.as_str();
84        let email = normalise_email(email);
85
86        let row = sqlx::query_as!(
87            User,
88            r#"
89            INSERT INTO users (
90                id, name, email, full_name, display_name,
91                status, email_verified, roles, is_bot,
92                created_at, updated_at
93            )
94            VALUES ($1, $2, $3, $4, $5, $6, false, ARRAY[$7]::TEXT[], false, $8, $8)
95            ON CONFLICT DO NOTHING
96            RETURNING id, name, email, full_name, display_name, status, email_verified,
97                      roles, avatar_url, is_bot, is_scanner, created_at, updated_at
98            "#,
99            id.as_str(),
100            name,
101            email,
102            full_name,
103            display_name_val,
104            status,
105            role,
106            now
107        )
108        .fetch_optional(&*self.write_pool)
109        .await?;
110
111        Ok(row)
112    }
113
114    pub async fn create_anonymous(&self, fingerprint: &str) -> Result<User> {
115        let email = normalise_email(&format!("{}@anonymous.local", fingerprint));
116
117        // Why: Read first: repeat bot traffic hits an existing row, so avoid the
118        // upsert's per-request write lock and commit on the hot path.
119        if let Some(existing) = sqlx::query_as!(
120            User,
121            r#"
122            SELECT id, name, email, full_name, display_name, status, email_verified,
123                   roles, avatar_url, is_bot, is_scanner, created_at, updated_at
124            FROM users
125            WHERE email = $1
126            "#,
127            email
128        )
129        .fetch_optional(&*self.pool)
130        .await?
131        {
132            return Ok(existing);
133        }
134
135        // Why: ON CONFLICT covers a concurrent first request racing this insert.
136        let user_id = uuid::Uuid::new_v4();
137        let id = UserId::new(user_id.to_string());
138        let name = format!("anonymous_{}", &user_id.to_string()[..8]);
139        let now = Utc::now();
140        let status = UserStatus::Active.as_str();
141        let role = UserRole::Anonymous.as_str();
142
143        let row = sqlx::query_as!(
144            User,
145            r#"
146            INSERT INTO users (
147                id, name, email, status, email_verified, roles,
148                is_bot, created_at, updated_at
149            )
150            VALUES ($1, $2, $3, $4, false, ARRAY[$5]::TEXT[], false, $6, $6)
151            ON CONFLICT (email) DO UPDATE SET updated_at = $6
152            RETURNING id, name, email, full_name, display_name, status, email_verified,
153                      roles, avatar_url, is_bot, is_scanner, created_at, updated_at
154            "#,
155            id.as_str(),
156            name,
157            email,
158            status,
159            role,
160            now
161        )
162        .fetch_one(&*self.write_pool)
163        .await?;
164
165        Ok(row)
166    }
167
168    pub async fn update_email(&self, id: &UserId, email: &str) -> Result<User> {
169        let row = sqlx::query_as!(
170            User,
171            r#"
172            UPDATE users
173            SET email = $1, email_verified = false, updated_at = $2
174            WHERE id = $3
175            RETURNING id, name, email, full_name, display_name, status, email_verified,
176                      roles, avatar_url, is_bot, is_scanner, created_at, updated_at
177            "#,
178            email,
179            Utc::now(),
180            id.as_str()
181        )
182        .fetch_optional(&*self.write_pool)
183        .await?
184        .ok_or_else(|| UserError::NotFound(id.clone()))?;
185
186        Ok(row)
187    }
188
189    pub async fn update_full_name(&self, id: &UserId, full_name: &str) -> Result<User> {
190        let row = sqlx::query_as!(
191            User,
192            r#"
193            UPDATE users
194            SET full_name = $1, updated_at = $2
195            WHERE id = $3
196            RETURNING id, name, email, full_name, display_name, status, email_verified,
197                      roles, avatar_url, is_bot, is_scanner, created_at, updated_at
198            "#,
199            full_name,
200            Utc::now(),
201            id.as_str()
202        )
203        .fetch_optional(&*self.write_pool)
204        .await?
205        .ok_or_else(|| UserError::NotFound(id.clone()))?;
206
207        Ok(row)
208    }
209
210    pub async fn update_status(&self, id: &UserId, status: UserStatus) -> Result<User> {
211        let row = sqlx::query_as!(
212            User,
213            r#"
214            UPDATE users
215            SET status = $1, updated_at = $2
216            WHERE id = $3
217            RETURNING id, name, email, full_name, display_name, status, email_verified,
218                      roles, avatar_url, is_bot, is_scanner, created_at, updated_at
219            "#,
220            status.as_str(),
221            Utc::now(),
222            id.as_str()
223        )
224        .fetch_optional(&*self.write_pool)
225        .await?
226        .ok_or_else(|| UserError::NotFound(id.clone()))?;
227
228        Ok(row)
229    }
230
231    pub async fn update_email_verified(&self, id: &UserId, verified: bool) -> Result<User> {
232        let row = sqlx::query_as!(
233            User,
234            r#"
235            UPDATE users
236            SET email_verified = $1, updated_at = $2
237            WHERE id = $3
238            RETURNING id, name, email, full_name, display_name, status, email_verified,
239                      roles, avatar_url, is_bot, is_scanner, created_at, updated_at
240            "#,
241            verified,
242            Utc::now(),
243            id.as_str()
244        )
245        .fetch_optional(&*self.write_pool)
246        .await?
247        .ok_or_else(|| UserError::NotFound(id.clone()))?;
248
249        Ok(row)
250    }
251
252    pub async fn update_display_name(&self, id: &UserId, display_name: &str) -> Result<User> {
253        let row = sqlx::query_as!(
254            User,
255            r#"
256            UPDATE users
257            SET display_name = $1, updated_at = $2
258            WHERE id = $3
259            RETURNING id, name, email, full_name, display_name, status, email_verified,
260                      roles, avatar_url, is_bot, is_scanner, created_at, updated_at
261            "#,
262            display_name,
263            Utc::now(),
264            id.as_str()
265        )
266        .fetch_optional(&*self.write_pool)
267        .await?
268        .ok_or_else(|| UserError::NotFound(id.clone()))?;
269
270        Ok(row)
271    }
272
273    pub async fn update_all_fields(
274        &self,
275        id: &UserId,
276        params: UpdateUserParams<'_>,
277    ) -> Result<User> {
278        let row = sqlx::query_as!(
279            User,
280            r#"
281            UPDATE users
282            SET email = $1, full_name = $2, display_name = $3, status = $4, updated_at = $5
283            WHERE id = $6
284            RETURNING id, name, email, full_name, display_name, status, email_verified,
285                      roles, avatar_url, is_bot, is_scanner, created_at, updated_at
286            "#,
287            params.email,
288            params.full_name,
289            params.display_name,
290            params.status.as_str(),
291            Utc::now(),
292            id.as_str()
293        )
294        .fetch_optional(&*self.write_pool)
295        .await?
296        .ok_or_else(|| UserError::NotFound(id.clone()))?;
297
298        Ok(row)
299    }
300
301    pub async fn assign_roles(&self, id: &UserId, roles: &[String]) -> Result<User> {
302        let row = sqlx::query_as!(
303            User,
304            r#"
305            UPDATE users
306            SET roles = $1, updated_at = $2
307            WHERE id = $3
308            RETURNING id, name, email, full_name, display_name, status, email_verified,
309                      roles, avatar_url, is_bot, is_scanner, created_at, updated_at
310            "#,
311            roles,
312            Utc::now(),
313            id.as_str()
314        )
315        .fetch_optional(&*self.write_pool)
316        .await?
317        .ok_or_else(|| UserError::NotFound(id.clone()))?;
318
319        Ok(row)
320    }
321
322    pub async fn delete(&self, id: &UserId) -> Result<()> {
323        let result = sqlx::query!(r#"DELETE FROM users WHERE id = $1"#, id.as_str())
324            .execute(&*self.write_pool)
325            .await?;
326
327        if result.rows_affected() == 0 {
328            return Err(UserError::NotFound(id.clone()));
329        }
330
331        Ok(())
332    }
333
334    pub async fn cleanup_old_anonymous(&self, days: i32) -> Result<u64> {
335        let cutoff = Utc::now() - Duration::days(i64::from(days));
336        let anonymous_role = UserRole::Anonymous.as_str();
337        let result = sqlx::query!(
338            r#"
339            DELETE FROM users u
340            WHERE $1 = ANY(u.roles)
341              AND u.created_at < $2
342              AND NOT EXISTS (
343                  SELECT 1
344                  FROM user_sessions s
345                  WHERE s.user_id = u.id
346                    AND s.ended_at IS NULL
347              )
348            "#,
349            anonymous_role,
350            cutoff
351        )
352        .execute(&*self.write_pool)
353        .await?;
354
355        Ok(result.rows_affected())
356    }
357
358    pub async fn count_old_anonymous(&self, days: i32) -> Result<i64> {
359        let cutoff = Utc::now() - Duration::days(i64::from(days));
360        let anonymous_role = UserRole::Anonymous.as_str();
361        let count = sqlx::query_scalar!(
362            r#"
363            SELECT COUNT(*) as "count!"
364            FROM users u
365            WHERE $1 = ANY(u.roles)
366              AND u.created_at < $2
367              AND NOT EXISTS (
368                  SELECT 1
369                  FROM user_sessions s
370                  WHERE s.user_id = u.id
371                    AND s.ended_at IS NULL
372              )
373            "#,
374            anonymous_role,
375            cutoff
376        )
377        .fetch_one(&*self.write_pool)
378        .await?;
379
380        Ok(count)
381    }
382}
systemprompt_users/repository/user/operations.rs

systemprompt_users/repository/user/
operations.rs
