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, UserRow, 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            UserRow,
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        .map(User::try_from)
32        .transpose()?
33        .ok_or_else(|| UserError::NotFound(id.clone()))?;
34
35        Ok(row)
36    }
37
38    pub async fn update_full_name(&self, id: &UserId, full_name: &str) -> Result<User> {
39        let row = sqlx::query_as!(
40            UserRow,
41            r#"
42            UPDATE users
43            SET full_name = $1, updated_at = $2
44            WHERE id = $3
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            full_name,
49            Utc::now(),
50            id.as_str()
51        )
52        .fetch_optional(&*self.write_pool)
53        .await?
54        .map(User::try_from)
55        .transpose()?
56        .ok_or_else(|| UserError::NotFound(id.clone()))?;
57
58        Ok(row)
59    }
60
61    pub async fn update_status(&self, id: &UserId, status: UserStatus) -> Result<User> {
62        let row = sqlx::query_as!(
63            UserRow,
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(&*self.write_pool)
76        .await?
77        .map(User::try_from)
78        .transpose()?
79        .ok_or_else(|| UserError::NotFound(id.clone()))?;
80        Ok(row)
81    }
82
83    pub async fn update_email_verified(&self, id: &UserId, verified: bool) -> Result<User> {
84        let row = sqlx::query_as!(
85            UserRow,
86            r#"
87            UPDATE users
88            SET email_verified = $1, updated_at = $2
89            WHERE id = $3
90            RETURNING id, name, email, full_name, display_name, status, email_verified,
91                      roles, avatar_url, is_bot, is_scanner, created_at, updated_at
92            "#,
93            verified,
94            Utc::now(),
95            id.as_str()
96        )
97        .fetch_optional(&*self.write_pool)
98        .await?
99        .map(User::try_from)
100        .transpose()?
101        .ok_or_else(|| UserError::NotFound(id.clone()))?;
102
103        Ok(row)
104    }
105
106    pub async fn update_display_name(&self, id: &UserId, display_name: &str) -> Result<User> {
107        let row = sqlx::query_as!(
108            UserRow,
109            r#"
110            UPDATE users
111            SET display_name = $1, updated_at = $2
112            WHERE id = $3
113            RETURNING id, name, email, full_name, display_name, status, email_verified,
114                      roles, avatar_url, is_bot, is_scanner, created_at, updated_at
115            "#,
116            display_name,
117            Utc::now(),
118            id.as_str()
119        )
120        .fetch_optional(&*self.write_pool)
121        .await?
122        .map(User::try_from)
123        .transpose()?
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 row = sqlx::query_as!(
135            UserRow,
136            r#"
137            UPDATE users
138            SET email = $1, full_name = $2, display_name = $3, status = $4, updated_at = $5
139            WHERE id = $6
140            RETURNING id, name, email, full_name, display_name, status, email_verified,
141                      roles, avatar_url, is_bot, is_scanner, created_at, updated_at
142            "#,
143            params.email,
144            params.full_name,
145            params.display_name,
146            params.status.as_str(),
147            Utc::now(),
148            id.as_str()
149        )
150        .fetch_optional(&*self.write_pool)
151        .await?
152        .map(User::try_from)
153        .transpose()?
154        .ok_or_else(|| UserError::NotFound(id.clone()))?;
155        Ok(row)
156    }
157
158    pub async fn assign_roles(&self, id: &UserId, roles: &[String]) -> Result<User> {
159        let row = sqlx::query_as!(
160            UserRow,
161            r#"
162            UPDATE users
163            SET roles = $1, updated_at = $2
164            WHERE id = $3
165            RETURNING id, name, email, full_name, display_name, status, email_verified,
166                      roles, avatar_url, is_bot, is_scanner, created_at, updated_at
167            "#,
168            roles,
169            Utc::now(),
170            id.as_str()
171        )
172        .fetch_optional(&*self.write_pool)
173        .await?
174        .map(User::try_from)
175        .transpose()?
176        .ok_or_else(|| UserError::NotFound(id.clone()))?;
177
178        Ok(row)
179    }
180
181    pub async fn delete(&self, id: &UserId) -> Result<Vec<super::PurgeCount>> {
182        let mut tx = self.write_pool.begin().await?;
183        let sessions = sqlx::query!("DELETE FROM user_sessions WHERE user_id = $1", id.as_str())
184            .execute(&mut *tx)
185            .await?;
186        let mut removed = Self::purge_user_rows(&mut tx, id).await?;
187        removed.insert(
188            0,
189            super::PurgeCount {
190                owner: "systemprompt-core",
191                table: "user_sessions",
192                rows: i64::try_from(sessions.rows_affected()).unwrap_or(i64::MAX),
193            },
194        );
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        tx.commit().await?;
204
205        Ok(removed)
206    }
207
208    pub async fn cleanup_old_anonymous(&self, days: i32) -> Result<u64> {
209        let mut tx = self.write_pool.begin().await?;
210        let cutoff = Utc::now() - Duration::days(i64::from(days));
211        let anonymous_role = UserRole::Anonymous.as_str();
212        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)
213            .execute(&mut *tx)
214            .await?;
215        let result = sqlx::query!(
216            r#"
217            DELETE FROM users u
218            WHERE $1 = ANY(u.roles)
219              AND u.created_at < $2
220              AND NOT EXISTS (
221                  SELECT 1
222                  FROM user_sessions s
223                  WHERE s.user_id = u.id
224                    AND s.ended_at IS NULL
225              )
226            "#,
227            anonymous_role,
228            cutoff
229        )
230        .execute(&mut *tx)
231        .await?;
232
233        tx.commit().await?;
234
235        Ok(result.rows_affected())
236    }
237
238    pub async fn count_old_anonymous(&self, days: i32) -> Result<i64> {
239        let cutoff = Utc::now() - Duration::days(i64::from(days));
240        let anonymous_role = UserRole::Anonymous.as_str();
241        let count = sqlx::query_scalar!(
242            r#"
243            SELECT COUNT(*) as "count!"
244            FROM users u
245            WHERE $1 = ANY(u.roles)
246              AND u.created_at < $2
247              AND NOT EXISTS (
248                  SELECT 1
249                  FROM user_sessions s
250                  WHERE s.user_id = u.id
251                    AND s.ended_at IS NULL
252              )
253            "#,
254            anonymous_role,
255            cutoff
256        )
257        .fetch_one(&*self.write_pool)
258        .await?;
259
260        Ok(count)
261    }
262}
systemprompt_users/repository/user/updates.rs

systemprompt_users/repository/user/
updates.rs
