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<()> {
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        sqlx::query!("DELETE FROM user_sessions WHERE user_id = $1", id.as_str())
193            .execute(&mut *tx)
194            .await?;
195        let result = sqlx::query!(r#"DELETE FROM users WHERE id = $1"#, id.as_str())
196            .execute(&mut *tx)
197            .await?;
198
199        if result.rows_affected() == 0 {
200            return Err(UserError::NotFound(id.clone()));
201        }
202
203        sqlx::query!("SELECT public.finish_user_privacy() AS finished")
204            .fetch_one(&mut *tx)
205            .await?;
206        tx.commit().await?;
207
208        Ok(())
209    }
210
211    pub async fn cleanup_old_anonymous(&self, days: i32) -> Result<u64> {
212        let mut tx = self.write_pool.begin().await?;
213        sqlx::query!("SELECT public.begin_user_privacy() AS prepared")
214            .fetch_one(&mut *tx)
215            .await?;
216        let cutoff = Utc::now() - Duration::days(i64::from(days));
217        let anonymous_role = UserRole::Anonymous.as_str();
218        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)
219            .execute(&mut *tx)
220            .await?;
221        let result = sqlx::query!(
222            r#"
223            DELETE FROM users u
224            WHERE $1 = ANY(u.roles)
225              AND u.created_at < $2
226              AND NOT EXISTS (
227                  SELECT 1
228                  FROM user_sessions s
229                  WHERE s.user_id = u.id
230                    AND s.ended_at IS NULL
231              )
232            "#,
233            anonymous_role,
234            cutoff
235        )
236        .execute(&mut *tx)
237        .await?;
238
239        sqlx::query!("SELECT public.finish_user_privacy() AS finished")
240            .fetch_one(&mut *tx)
241            .await?;
242        tx.commit().await?;
243
244        Ok(result.rows_affected())
245    }
246
247    pub async fn count_old_anonymous(&self, days: i32) -> Result<i64> {
248        let cutoff = Utc::now() - Duration::days(i64::from(days));
249        let anonymous_role = UserRole::Anonymous.as_str();
250        let count = sqlx::query_scalar!(
251            r#"
252            SELECT COUNT(*) as "count!"
253            FROM users u
254            WHERE $1 = ANY(u.roles)
255              AND u.created_at < $2
256              AND NOT EXISTS (
257                  SELECT 1
258                  FROM user_sessions s
259                  WHERE s.user_id = u.id
260                    AND s.ended_at IS NULL
261              )
262            "#,
263            anonymous_role,
264            cutoff
265        )
266        .fetch_one(&*self.write_pool)
267        .await?;
268
269        Ok(count)
270    }
271}
systemprompt_users/repository/user/updates.rs

systemprompt_users/repository/user/
updates.rs
