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 row = sqlx::query_as!(
59            User,
60            r#"
61            UPDATE users
62            SET status = $1, updated_at = $2
63            WHERE id = $3
64            RETURNING id, name, email, full_name, display_name, status, email_verified,
65                      roles, avatar_url, is_bot, is_scanner, created_at, updated_at
66            "#,
67            status.as_str(),
68            Utc::now(),
69            id.as_str()
70        )
71        .fetch_optional(&*self.write_pool)
72        .await?
73        .ok_or_else(|| UserError::NotFound(id.clone()))?;
74        Ok(row)
75    }
76
77    pub async fn update_email_verified(&self, id: &UserId, verified: bool) -> Result<User> {
78        let row = sqlx::query_as!(
79            User,
80            r#"
81            UPDATE users
82            SET email_verified = $1, updated_at = $2
83            WHERE id = $3
84            RETURNING id, name, email, full_name, display_name, status, email_verified,
85                      roles, avatar_url, is_bot, is_scanner, created_at, updated_at
86            "#,
87            verified,
88            Utc::now(),
89            id.as_str()
90        )
91        .fetch_optional(&*self.write_pool)
92        .await?
93        .ok_or_else(|| UserError::NotFound(id.clone()))?;
94
95        Ok(row)
96    }
97
98    pub async fn update_display_name(&self, id: &UserId, display_name: &str) -> Result<User> {
99        let row = sqlx::query_as!(
100            User,
101            r#"
102            UPDATE users
103            SET display_name = $1, updated_at = $2
104            WHERE id = $3
105            RETURNING id, name, email, full_name, display_name, status, email_verified,
106                      roles, avatar_url, is_bot, is_scanner, created_at, updated_at
107            "#,
108            display_name,
109            Utc::now(),
110            id.as_str()
111        )
112        .fetch_optional(&*self.write_pool)
113        .await?
114        .ok_or_else(|| UserError::NotFound(id.clone()))?;
115
116        Ok(row)
117    }
118
119    pub async fn update_all_fields(
120        &self,
121        id: &UserId,
122        params: UpdateUserParams<'_>,
123    ) -> Result<User> {
124        let row = sqlx::query_as!(
125            User,
126            r#"
127            UPDATE users
128            SET email = $1, full_name = $2, display_name = $3, status = $4, updated_at = $5
129            WHERE id = $6
130            RETURNING id, name, email, full_name, display_name, status, email_verified,
131                      roles, avatar_url, is_bot, is_scanner, created_at, updated_at
132            "#,
133            params.email,
134            params.full_name,
135            params.display_name,
136            params.status.as_str(),
137            Utc::now(),
138            id.as_str()
139        )
140        .fetch_optional(&*self.write_pool)
141        .await?
142        .ok_or_else(|| UserError::NotFound(id.clone()))?;
143        Ok(row)
144    }
145
146    pub async fn assign_roles(&self, id: &UserId, roles: &[String]) -> Result<User> {
147        let row = sqlx::query_as!(
148            User,
149            r#"
150            UPDATE users
151            SET roles = $1, updated_at = $2
152            WHERE id = $3
153            RETURNING id, name, email, full_name, display_name, status, email_verified,
154                      roles, avatar_url, is_bot, is_scanner, created_at, updated_at
155            "#,
156            roles,
157            Utc::now(),
158            id.as_str()
159        )
160        .fetch_optional(&*self.write_pool)
161        .await?
162        .ok_or_else(|| UserError::NotFound(id.clone()))?;
163
164        Ok(row)
165    }
166
167    pub async fn delete(&self, id: &UserId) -> Result<Vec<super::PurgeCount>> {
168        let mut tx = self.write_pool.begin().await?;
169        let sessions = sqlx::query!("DELETE FROM user_sessions WHERE user_id = $1", id.as_str())
170            .execute(&mut *tx)
171            .await?;
172        let mut removed = Self::purge_user_rows(&mut tx, id).await?;
173        removed.insert(
174            0,
175            super::PurgeCount {
176                owner: "systemprompt-core",
177                table: "user_sessions",
178                rows: i64::try_from(sessions.rows_affected()).unwrap_or(i64::MAX),
179            },
180        );
181        let result = sqlx::query!(r#"DELETE FROM users WHERE id = $1"#, id.as_str())
182            .execute(&mut *tx)
183            .await?;
184
185        if result.rows_affected() == 0 {
186            return Err(UserError::NotFound(id.clone()));
187        }
188
189        tx.commit().await?;
190
191        Ok(removed)
192    }
193
194    pub async fn cleanup_old_anonymous(&self, days: i32) -> Result<u64> {
195        let mut tx = self.write_pool.begin().await?;
196        let cutoff = Utc::now() - Duration::days(i64::from(days));
197        let anonymous_role = UserRole::Anonymous.as_str();
198        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)
199            .execute(&mut *tx)
200            .await?;
201        let result = sqlx::query!(
202            r#"
203            DELETE FROM users u
204            WHERE $1 = ANY(u.roles)
205              AND u.created_at < $2
206              AND NOT EXISTS (
207                  SELECT 1
208                  FROM user_sessions s
209                  WHERE s.user_id = u.id
210                    AND s.ended_at IS NULL
211              )
212            "#,
213            anonymous_role,
214            cutoff
215        )
216        .execute(&mut *tx)
217        .await?;
218
219        tx.commit().await?;
220
221        Ok(result.rows_affected())
222    }
223
224    pub async fn count_old_anonymous(&self, days: i32) -> Result<i64> {
225        let cutoff = Utc::now() - Duration::days(i64::from(days));
226        let anonymous_role = UserRole::Anonymous.as_str();
227        let count = sqlx::query_scalar!(
228            r#"
229            SELECT COUNT(*) as "count!"
230            FROM users u
231            WHERE $1 = ANY(u.roles)
232              AND u.created_at < $2
233              AND NOT EXISTS (
234                  SELECT 1
235                  FROM user_sessions s
236                  WHERE s.user_id = u.id
237                    AND s.ended_at IS NULL
238              )
239            "#,
240            anonymous_role,
241            cutoff
242        )
243        .fetch_one(&*self.write_pool)
244        .await?;
245
246        Ok(count)
247    }
248}
systemprompt_users/repository/user/updates.rs

systemprompt_users/repository/user/
updates.rs
