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

systemprompt_users/repository/user/
updates.rs
