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        if let Some(existing) = sqlx::query_as!(
118            User,
119            r#"
120            SELECT id, name, email, full_name, display_name, status, email_verified,
121                   roles, avatar_url, is_bot, is_scanner, created_at, updated_at
122            FROM users
123            WHERE email = $1
124            "#,
125            email
126        )
127        .fetch_optional(&*self.pool)
128        .await?
129        {
130            return Ok(existing);
131        }
132
133        let user_id = uuid::Uuid::new_v4();
134        let id = UserId::new(user_id.to_string());
135        let name = format!("anonymous_{}", &user_id.to_string()[..8]);
136        let now = Utc::now();
137        let status = UserStatus::Active.as_str();
138        let role = UserRole::Anonymous.as_str();
139
140        let row = sqlx::query_as!(
141            User,
142            r#"
143            INSERT INTO users (
144                id, name, email, status, email_verified, roles,
145                is_bot, created_at, updated_at
146            )
147            VALUES ($1, $2, $3, $4, false, ARRAY[$5]::TEXT[], false, $6, $6)
148            ON CONFLICT (email) DO UPDATE SET updated_at = $6
149            RETURNING id, name, email, full_name, display_name, status, email_verified,
150                      roles, avatar_url, is_bot, is_scanner, created_at, updated_at
151            "#,
152            id.as_str(),
153            name,
154            email,
155            status,
156            role,
157            now
158        )
159        .fetch_one(&*self.write_pool)
160        .await?;
161
162        Ok(row)
163    }
164
165    pub async fn update_email(&self, id: &UserId, email: &str) -> Result<User> {
166        let row = sqlx::query_as!(
167            User,
168            r#"
169            UPDATE users
170            SET email = $1, email_verified = false, updated_at = $2
171            WHERE id = $3
172            RETURNING id, name, email, full_name, display_name, status, email_verified,
173                      roles, avatar_url, is_bot, is_scanner, created_at, updated_at
174            "#,
175            email,
176            Utc::now(),
177            id.as_str()
178        )
179        .fetch_optional(&*self.write_pool)
180        .await?
181        .ok_or_else(|| UserError::NotFound(id.clone()))?;
182
183        Ok(row)
184    }
185
186    pub async fn update_full_name(&self, id: &UserId, full_name: &str) -> Result<User> {
187        let row = sqlx::query_as!(
188            User,
189            r#"
190            UPDATE users
191            SET full_name = $1, updated_at = $2
192            WHERE id = $3
193            RETURNING id, name, email, full_name, display_name, status, email_verified,
194                      roles, avatar_url, is_bot, is_scanner, created_at, updated_at
195            "#,
196            full_name,
197            Utc::now(),
198            id.as_str()
199        )
200        .fetch_optional(&*self.write_pool)
201        .await?
202        .ok_or_else(|| UserError::NotFound(id.clone()))?;
203
204        Ok(row)
205    }
206
207    pub async fn update_status(&self, id: &UserId, status: UserStatus) -> Result<User> {
208        let row = sqlx::query_as!(
209            User,
210            r#"
211            UPDATE users
212            SET status = $1, updated_at = $2
213            WHERE id = $3
214            RETURNING id, name, email, full_name, display_name, status, email_verified,
215                      roles, avatar_url, is_bot, is_scanner, created_at, updated_at
216            "#,
217            status.as_str(),
218            Utc::now(),
219            id.as_str()
220        )
221        .fetch_optional(&*self.write_pool)
222        .await?
223        .ok_or_else(|| UserError::NotFound(id.clone()))?;
224
225        Ok(row)
226    }
227
228    pub async fn update_email_verified(&self, id: &UserId, verified: bool) -> Result<User> {
229        let row = sqlx::query_as!(
230            User,
231            r#"
232            UPDATE users
233            SET email_verified = $1, updated_at = $2
234            WHERE id = $3
235            RETURNING id, name, email, full_name, display_name, status, email_verified,
236                      roles, avatar_url, is_bot, is_scanner, created_at, updated_at
237            "#,
238            verified,
239            Utc::now(),
240            id.as_str()
241        )
242        .fetch_optional(&*self.write_pool)
243        .await?
244        .ok_or_else(|| UserError::NotFound(id.clone()))?;
245
246        Ok(row)
247    }
248
249    pub async fn update_display_name(&self, id: &UserId, display_name: &str) -> Result<User> {
250        let row = sqlx::query_as!(
251            User,
252            r#"
253            UPDATE users
254            SET display_name = $1, updated_at = $2
255            WHERE id = $3
256            RETURNING id, name, email, full_name, display_name, status, email_verified,
257                      roles, avatar_url, is_bot, is_scanner, created_at, updated_at
258            "#,
259            display_name,
260            Utc::now(),
261            id.as_str()
262        )
263        .fetch_optional(&*self.write_pool)
264        .await?
265        .ok_or_else(|| UserError::NotFound(id.clone()))?;
266
267        Ok(row)
268    }
269
270    pub async fn update_all_fields(
271        &self,
272        id: &UserId,
273        params: UpdateUserParams<'_>,
274    ) -> Result<User> {
275        let row = sqlx::query_as!(
276            User,
277            r#"
278            UPDATE users
279            SET email = $1, full_name = $2, display_name = $3, status = $4, updated_at = $5
280            WHERE id = $6
281            RETURNING id, name, email, full_name, display_name, status, email_verified,
282                      roles, avatar_url, is_bot, is_scanner, created_at, updated_at
283            "#,
284            params.email,
285            params.full_name,
286            params.display_name,
287            params.status.as_str(),
288            Utc::now(),
289            id.as_str()
290        )
291        .fetch_optional(&*self.write_pool)
292        .await?
293        .ok_or_else(|| UserError::NotFound(id.clone()))?;
294
295        Ok(row)
296    }
297
298    pub async fn assign_roles(&self, id: &UserId, roles: &[String]) -> Result<User> {
299        let row = sqlx::query_as!(
300            User,
301            r#"
302            UPDATE users
303            SET roles = $1, updated_at = $2
304            WHERE id = $3
305            RETURNING id, name, email, full_name, display_name, status, email_verified,
306                      roles, avatar_url, is_bot, is_scanner, created_at, updated_at
307            "#,
308            roles,
309            Utc::now(),
310            id.as_str()
311        )
312        .fetch_optional(&*self.write_pool)
313        .await?
314        .ok_or_else(|| UserError::NotFound(id.clone()))?;
315
316        Ok(row)
317    }
318
319    pub async fn delete(&self, id: &UserId) -> Result<()> {
320        let result = sqlx::query!(r#"DELETE FROM users WHERE id = $1"#, id.as_str())
321            .execute(&*self.write_pool)
322            .await?;
323
324        if result.rows_affected() == 0 {
325            return Err(UserError::NotFound(id.clone()));
326        }
327
328        Ok(())
329    }
330
331    pub async fn cleanup_old_anonymous(&self, days: i32) -> Result<u64> {
332        let cutoff = Utc::now() - Duration::days(i64::from(days));
333        let anonymous_role = UserRole::Anonymous.as_str();
334        let result = sqlx::query!(
335            r#"
336            DELETE FROM users u
337            WHERE $1 = ANY(u.roles)
338              AND u.created_at < $2
339              AND NOT EXISTS (
340                  SELECT 1
341                  FROM user_sessions s
342                  WHERE s.user_id = u.id
343                    AND s.ended_at IS NULL
344              )
345            "#,
346            anonymous_role,
347            cutoff
348        )
349        .execute(&*self.write_pool)
350        .await?;
351
352        Ok(result.rows_affected())
353    }
354
355    pub async fn count_old_anonymous(&self, days: i32) -> Result<i64> {
356        let cutoff = Utc::now() - Duration::days(i64::from(days));
357        let anonymous_role = UserRole::Anonymous.as_str();
358        let count = sqlx::query_scalar!(
359            r#"
360            SELECT COUNT(*) as "count!"
361            FROM users u
362            WHERE $1 = ANY(u.roles)
363              AND u.created_at < $2
364              AND NOT EXISTS (
365                  SELECT 1
366                  FROM user_sessions s
367                  WHERE s.user_id = u.id
368                    AND s.ended_at IS NULL
369              )
370            "#,
371            anonymous_role,
372            cutoff
373        )
374        .fetch_one(&*self.write_pool)
375        .await?;
376
377        Ok(count)
378    }
379}
systemprompt_users/repository/user/operations.rs

systemprompt_users/repository/user/
operations.rs
