systemprompt_users/repository/user/
updates.rs

1//! User record field updates, role assignment, and anonymous-account cleanup.
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 super::operations::UpdateUserParams;
10use crate::error::{Result, UserError};
11use crate::models::{User, UserRole, UserStatus};
12use crate::repository::UserRepository;
13
14impl UserRepository {
15    pub async fn update_email(&self, id: &UserId, email: &str) -> Result<User> {
16        let row = sqlx::query_as!(
17            User,
18            r#"
19            UPDATE users
20            SET email = $1, email_verified = false, updated_at = $2
21            WHERE id = $3
22            RETURNING id, name, email, full_name, display_name, status, email_verified,
23                      roles, avatar_url, is_bot, is_scanner, created_at, updated_at
24            "#,
25            email,
26            Utc::now(),
27            id.as_str()
28        )
29        .fetch_optional(&*self.write_pool)
30        .await?
31        .ok_or_else(|| UserError::NotFound(id.clone()))?;
32
33        Ok(row)
34    }
35
36    pub async fn update_full_name(&self, id: &UserId, full_name: &str) -> Result<User> {
37        let row = sqlx::query_as!(
38            User,
39            r#"
40            UPDATE users
41            SET full_name = $1, updated_at = $2
42            WHERE id = $3
43            RETURNING id, name, email, full_name, display_name, status, email_verified,
44                      roles, avatar_url, is_bot, is_scanner, created_at, updated_at
45            "#,
46            full_name,
47            Utc::now(),
48            id.as_str()
49        )
50        .fetch_optional(&*self.write_pool)
51        .await?
52        .ok_or_else(|| UserError::NotFound(id.clone()))?;
53
54        Ok(row)
55    }
56
57    pub async fn update_status(&self, id: &UserId, status: UserStatus) -> Result<User> {
58        let mut tx = self.write_pool.begin().await?;
59        sqlx::query!("SELECT public.begin_user_privacy() AS prepared")
60            .fetch_one(&mut *tx)
61            .await?;
62        let row = sqlx::query_as!(
63            User,
64            r#"
65            UPDATE users
66            SET status = $1, updated_at = $2
67            WHERE id = $3
68            RETURNING id, name, email, full_name, display_name, status, email_verified,
69                      roles, avatar_url, is_bot, is_scanner, created_at, updated_at
70            "#,
71            status.as_str(),
72            Utc::now(),
73            id.as_str()
74        )
75        .fetch_optional(&mut *tx)
76        .await?
77        .ok_or_else(|| UserError::NotFound(id.clone()))?;
78
79        sqlx::query!("SELECT public.finish_user_privacy() AS finished")
80            .fetch_one(&mut *tx)
81            .await?;
82        tx.commit().await?;
83
84        Ok(row)
85    }
86
87    pub async fn update_email_verified(&self, id: &UserId, verified: bool) -> Result<User> {
88        let row = sqlx::query_as!(
89            User,
90            r#"
91            UPDATE users
92            SET email_verified = $1, updated_at = $2
93            WHERE id = $3
94            RETURNING id, name, email, full_name, display_name, status, email_verified,
95                      roles, avatar_url, is_bot, is_scanner, created_at, updated_at
96            "#,
97            verified,
98            Utc::now(),
99            id.as_str()
100        )
101        .fetch_optional(&*self.write_pool)
102        .await?
103        .ok_or_else(|| UserError::NotFound(id.clone()))?;
104
105        Ok(row)
106    }
107
108    pub async fn update_display_name(&self, id: &UserId, display_name: &str) -> Result<User> {
109        let row = sqlx::query_as!(
110            User,
111            r#"
112            UPDATE users
113            SET display_name = $1, updated_at = $2
114            WHERE id = $3
115            RETURNING id, name, email, full_name, display_name, status, email_verified,
116                      roles, avatar_url, is_bot, is_scanner, created_at, updated_at
117            "#,
118            display_name,
119            Utc::now(),
120            id.as_str()
121        )
122        .fetch_optional(&*self.write_pool)
123        .await?
124        .ok_or_else(|| UserError::NotFound(id.clone()))?;
125
126        Ok(row)
127    }
128
129    pub async fn update_all_fields(
130        &self,
131        id: &UserId,
132        params: UpdateUserParams<'_>,
133    ) -> Result<User> {
134        let mut tx = self.write_pool.begin().await?;
135        sqlx::query!("SELECT public.begin_user_privacy() AS prepared")
136            .fetch_one(&mut *tx)
137            .await?;
138        let row = sqlx::query_as!(
139            User,
140            r#"
141            UPDATE users
142            SET email = $1, full_name = $2, display_name = $3, status = $4, updated_at = $5
143            WHERE id = $6
144            RETURNING id, name, email, full_name, display_name, status, email_verified,
145                      roles, avatar_url, is_bot, is_scanner, created_at, updated_at
146            "#,
147            params.email,
148            params.full_name,
149            params.display_name,
150            params.status.as_str(),
151            Utc::now(),
152            id.as_str()
153        )
154        .fetch_optional(&mut *tx)
155        .await?
156        .ok_or_else(|| UserError::NotFound(id.clone()))?;
157
158        sqlx::query!("SELECT public.finish_user_privacy() AS finished")
159            .fetch_one(&mut *tx)
160            .await?;
161        tx.commit().await?;
162
163        Ok(row)
164    }
165
166    pub async fn assign_roles(&self, id: &UserId, roles: &[String]) -> Result<User> {
167        let row = sqlx::query_as!(
168            User,
169            r#"
170            UPDATE users
171            SET roles = $1, updated_at = $2
172            WHERE id = $3
173            RETURNING id, name, email, full_name, display_name, status, email_verified,
174                      roles, avatar_url, is_bot, is_scanner, created_at, updated_at
175            "#,
176            roles,
177            Utc::now(),
178            id.as_str()
179        )
180        .fetch_optional(&*self.write_pool)
181        .await?
182        .ok_or_else(|| UserError::NotFound(id.clone()))?;
183
184        Ok(row)
185    }
186
187    pub async fn delete(&self, id: &UserId) -> Result<Vec<super::PurgeCount>> {
188        let mut tx = self.write_pool.begin().await?;
189        sqlx::query!("SELECT public.begin_user_privacy() AS prepared")
190            .fetch_one(&mut *tx)
191            .await?;
192        let sessions = sqlx::query!("DELETE FROM user_sessions WHERE user_id = $1", id.as_str())
193            .execute(&mut *tx)
194            .await?;
195        let mut removed = Self::purge_user_rows(&mut tx, id).await?;
196        removed.insert(
197            0,
198            super::PurgeCount {
199                owner: "systemprompt-core",
200                table: "user_sessions",
201                rows: i64::try_from(sessions.rows_affected()).unwrap_or(i64::MAX),
202            },
203        );
204        let result = sqlx::query!(r#"DELETE FROM users WHERE id = $1"#, id.as_str())
205            .execute(&mut *tx)
206            .await?;
207
208        if result.rows_affected() == 0 {
209            return Err(UserError::NotFound(id.clone()));
210        }
211
212        sqlx::query!("SELECT public.finish_user_privacy() AS finished")
213            .fetch_one(&mut *tx)
214            .await?;
215        tx.commit().await?;
216
217        Ok(removed)
218    }
219
220    pub async fn cleanup_old_anonymous(&self, days: i32) -> Result<u64> {
221        let mut tx = self.write_pool.begin().await?;
222        sqlx::query!("SELECT public.begin_user_privacy() AS prepared")
223            .fetch_one(&mut *tx)
224            .await?;
225        let cutoff = Utc::now() - Duration::days(i64::from(days));
226        let anonymous_role = UserRole::Anonymous.as_str();
227        sqlx::query!("DELETE FROM user_sessions WHERE user_id IN (SELECT u.id FROM users u WHERE $1 = ANY(u.roles) AND u.created_at < $2 AND NOT EXISTS (SELECT 1 FROM user_sessions s WHERE s.user_id = u.id AND s.ended_at IS NULL))", anonymous_role, cutoff)
228            .execute(&mut *tx)
229            .await?;
230        let result = sqlx::query!(
231            r#"
232            DELETE FROM users u
233            WHERE $1 = ANY(u.roles)
234              AND u.created_at < $2
235              AND NOT EXISTS (
236                  SELECT 1
237                  FROM user_sessions s
238                  WHERE s.user_id = u.id
239                    AND s.ended_at IS NULL
240              )
241            "#,
242            anonymous_role,
243            cutoff
244        )
245        .execute(&mut *tx)
246        .await?;
247
248        sqlx::query!("SELECT public.finish_user_privacy() AS finished")
249            .fetch_one(&mut *tx)
250            .await?;
251        tx.commit().await?;
252
253        Ok(result.rows_affected())
254    }
255
256    pub async fn count_old_anonymous(&self, days: i32) -> Result<i64> {
257        let cutoff = Utc::now() - Duration::days(i64::from(days));
258        let anonymous_role = UserRole::Anonymous.as_str();
259        let count = sqlx::query_scalar!(
260            r#"
261            SELECT COUNT(*) as "count!"
262            FROM users u
263            WHERE $1 = ANY(u.roles)
264              AND u.created_at < $2
265              AND NOT EXISTS (
266                  SELECT 1
267                  FROM user_sessions s
268                  WHERE s.user_id = u.id
269                    AND s.ended_at IS NULL
270              )
271            "#,
272            anonymous_role,
273            cutoff
274        )
275        .fetch_one(&*self.write_pool)
276        .await?;
277
278        Ok(count)
279    }
280}
systemprompt_users/repository/user/updates.rs

systemprompt_users/repository/user/
updates.rs
