systemprompt_users/repository/user/
operations.rs

1use chrono::{Duration, Utc};
2use systemprompt_identifiers::UserId;
3
4use crate::error::{Result, UserError};
5use crate::models::{User, UserRole, UserStatus};
6use crate::repository::UserRepository;
7
8#[derive(Debug)]
9pub struct UpdateUserParams<'a> {
10    pub email: &'a str,
11    pub full_name: Option<&'a str>,
12    pub display_name: Option<&'a str>,
13    pub status: UserStatus,
14}
15
16impl UserRepository {
17    pub async fn create(
18        &self,
19        name: &str,
20        email: &str,
21        full_name: Option<&str>,
22        display_name: Option<&str>,
23    ) -> Result<User> {
24        let now = Utc::now();
25        let id = UserId::new(uuid::Uuid::new_v4().to_string());
26        let display_name_val = display_name.or(full_name);
27        let status = UserStatus::Active.as_str();
28        let role = UserRole::User.as_str();
29
30        let row = sqlx::query_as!(
31            User,
32            r#"
33            INSERT INTO users (
34                id, name, email, full_name, display_name,
35                status, email_verified, roles, is_bot,
36                created_at, updated_at
37            )
38            VALUES ($1, $2, $3, $4, $5, $6, false, ARRAY[$7]::TEXT[], false, $8, $8)
39            RETURNING id, name, email, full_name, display_name, status, email_verified,
40                      roles, avatar_url, is_bot, is_scanner, created_at, updated_at
41            "#,
42            id.as_str(),
43            name,
44            email,
45            full_name,
46            display_name_val,
47            status,
48            role,
49            now
50        )
51        .fetch_one(&*self.write_pool)
52        .await?;
53
54        Ok(row)
55    }
56
57    pub async fn create_anonymous(&self, fingerprint: &str) -> Result<User> {
58        let email = format!("{}@anonymous.local", fingerprint);
59
60        // Read first: repeat bot traffic hits an existing row, so avoid the
61        // upsert's per-request write lock and commit on the hot path.
62        if let Some(existing) = sqlx::query_as!(
63            User,
64            r#"
65            SELECT id, name, email, full_name, display_name, status, email_verified,
66                   roles, avatar_url, is_bot, is_scanner, created_at, updated_at
67            FROM users
68            WHERE email = $1
69            "#,
70            email
71        )
72        .fetch_optional(&*self.pool)
73        .await?
74        {
75            return Ok(existing);
76        }
77
78        // Miss: insert. ON CONFLICT covers a concurrent first request.
79        let user_id = uuid::Uuid::new_v4();
80        let id = UserId::new(user_id.to_string());
81        let name = format!("anonymous_{}", &user_id.to_string()[..8]);
82        let now = Utc::now();
83        let status = UserStatus::Active.as_str();
84        let role = UserRole::Anonymous.as_str();
85
86        let row = sqlx::query_as!(
87            User,
88            r#"
89            INSERT INTO users (
90                id, name, email, status, email_verified, roles,
91                is_bot, created_at, updated_at
92            )
93            VALUES ($1, $2, $3, $4, false, ARRAY[$5]::TEXT[], false, $6, $6)
94            ON CONFLICT (email) DO UPDATE SET updated_at = $6
95            RETURNING id, name, email, full_name, display_name, status, email_verified,
96                      roles, avatar_url, is_bot, is_scanner, created_at, updated_at
97            "#,
98            id.as_str(),
99            name,
100            email,
101            status,
102            role,
103            now
104        )
105        .fetch_one(&*self.write_pool)
106        .await?;
107
108        Ok(row)
109    }
110
111    pub async fn update_email(&self, id: &UserId, email: &str) -> Result<User> {
112        let row = sqlx::query_as!(
113            User,
114            r#"
115            UPDATE users
116            SET email = $1, email_verified = false, updated_at = $2
117            WHERE id = $3
118            RETURNING id, name, email, full_name, display_name, status, email_verified,
119                      roles, avatar_url, is_bot, is_scanner, created_at, updated_at
120            "#,
121            email,
122            Utc::now(),
123            id.as_str()
124        )
125        .fetch_optional(&*self.write_pool)
126        .await?
127        .ok_or_else(|| UserError::NotFound(id.clone()))?;
128
129        Ok(row)
130    }
131
132    pub async fn update_full_name(&self, id: &UserId, full_name: &str) -> Result<User> {
133        let row = sqlx::query_as!(
134            User,
135            r#"
136            UPDATE users
137            SET full_name = $1, updated_at = $2
138            WHERE id = $3
139            RETURNING id, name, email, full_name, display_name, status, email_verified,
140                      roles, avatar_url, is_bot, is_scanner, created_at, updated_at
141            "#,
142            full_name,
143            Utc::now(),
144            id.as_str()
145        )
146        .fetch_optional(&*self.write_pool)
147        .await?
148        .ok_or_else(|| UserError::NotFound(id.clone()))?;
149
150        Ok(row)
151    }
152
153    pub async fn update_status(&self, id: &UserId, status: UserStatus) -> Result<User> {
154        let row = sqlx::query_as!(
155            User,
156            r#"
157            UPDATE users
158            SET status = $1, updated_at = $2
159            WHERE id = $3
160            RETURNING id, name, email, full_name, display_name, status, email_verified,
161                      roles, avatar_url, is_bot, is_scanner, created_at, updated_at
162            "#,
163            status.as_str(),
164            Utc::now(),
165            id.as_str()
166        )
167        .fetch_optional(&*self.write_pool)
168        .await?
169        .ok_or_else(|| UserError::NotFound(id.clone()))?;
170
171        Ok(row)
172    }
173
174    pub async fn update_email_verified(&self, id: &UserId, verified: bool) -> Result<User> {
175        let row = sqlx::query_as!(
176            User,
177            r#"
178            UPDATE users
179            SET email_verified = $1, updated_at = $2
180            WHERE id = $3
181            RETURNING id, name, email, full_name, display_name, status, email_verified,
182                      roles, avatar_url, is_bot, is_scanner, created_at, updated_at
183            "#,
184            verified,
185            Utc::now(),
186            id.as_str()
187        )
188        .fetch_optional(&*self.write_pool)
189        .await?
190        .ok_or_else(|| UserError::NotFound(id.clone()))?;
191
192        Ok(row)
193    }
194
195    pub async fn update_display_name(&self, id: &UserId, display_name: &str) -> Result<User> {
196        let row = sqlx::query_as!(
197            User,
198            r#"
199            UPDATE users
200            SET display_name = $1, updated_at = $2
201            WHERE id = $3
202            RETURNING id, name, email, full_name, display_name, status, email_verified,
203                      roles, avatar_url, is_bot, is_scanner, created_at, updated_at
204            "#,
205            display_name,
206            Utc::now(),
207            id.as_str()
208        )
209        .fetch_optional(&*self.write_pool)
210        .await?
211        .ok_or_else(|| UserError::NotFound(id.clone()))?;
212
213        Ok(row)
214    }
215
216    pub async fn update_all_fields(
217        &self,
218        id: &UserId,
219        params: UpdateUserParams<'_>,
220    ) -> Result<User> {
221        let row = sqlx::query_as!(
222            User,
223            r#"
224            UPDATE users
225            SET email = $1, full_name = $2, display_name = $3, status = $4, updated_at = $5
226            WHERE id = $6
227            RETURNING id, name, email, full_name, display_name, status, email_verified,
228                      roles, avatar_url, is_bot, is_scanner, created_at, updated_at
229            "#,
230            params.email,
231            params.full_name,
232            params.display_name,
233            params.status.as_str(),
234            Utc::now(),
235            id.as_str()
236        )
237        .fetch_optional(&*self.write_pool)
238        .await?
239        .ok_or_else(|| UserError::NotFound(id.clone()))?;
240
241        Ok(row)
242    }
243
244    pub async fn assign_roles(&self, id: &UserId, roles: &[String]) -> Result<User> {
245        let row = sqlx::query_as!(
246            User,
247            r#"
248            UPDATE users
249            SET roles = $1, updated_at = $2
250            WHERE id = $3
251            RETURNING id, name, email, full_name, display_name, status, email_verified,
252                      roles, avatar_url, is_bot, is_scanner, created_at, updated_at
253            "#,
254            roles,
255            Utc::now(),
256            id.as_str()
257        )
258        .fetch_optional(&*self.write_pool)
259        .await?
260        .ok_or_else(|| UserError::NotFound(id.clone()))?;
261
262        Ok(row)
263    }
264
265    pub async fn delete(&self, id: &UserId) -> Result<()> {
266        let result = sqlx::query!(r#"DELETE FROM users WHERE id = $1"#, id.as_str())
267            .execute(&*self.write_pool)
268            .await?;
269
270        if result.rows_affected() == 0 {
271            return Err(UserError::NotFound(id.clone()));
272        }
273
274        Ok(())
275    }
276
277    pub async fn cleanup_old_anonymous(&self, days: i32) -> Result<u64> {
278        let cutoff = Utc::now() - Duration::days(i64::from(days));
279        let anonymous_role = UserRole::Anonymous.as_str();
280        let result = sqlx::query!(
281            r#"
282            DELETE FROM users u
283            WHERE $1 = ANY(u.roles)
284              AND u.created_at < $2
285              AND NOT EXISTS (
286                  SELECT 1
287                  FROM user_sessions s
288                  WHERE s.user_id = u.id
289                    AND s.ended_at IS NULL
290              )
291            "#,
292            anonymous_role,
293            cutoff
294        )
295        .execute(&*self.write_pool)
296        .await?;
297
298        Ok(result.rows_affected())
299    }
300}
systemprompt_users/repository/user/operations.rs

systemprompt_users/repository/user/
operations.rs
