systemprompt_users/repository/user/
list.rs

1//! User listing queries.
2//!
3//! Copyright (c) systemprompt.io — Business Source License 1.1.
4//! See <https://systemprompt.io> for licensing details.
5
6use systemprompt_identifiers::UserId;
7
8use crate::error::{Result, UserError};
9use crate::models::{User, UserActivity, UserRole, UserStatus, UserWithSessions};
10use crate::repository::{MAX_PAGE_SIZE, UserRepository};
11
12impl UserRepository {
13    pub async fn find_with_sessions(&self, user_id: &UserId) -> Result<Option<UserWithSessions>> {
14        let deleted_status = UserStatus::Deleted.as_str();
15        let row = sqlx::query_as!(
16            UserWithSessions,
17            r#"
18            SELECT
19                u.id, u.name, u.email, u.full_name, u.status, u.roles, u.created_at,
20                COUNT(s.session_id) FILTER (WHERE s.ended_at IS NULL) as "active_sessions!",
21                MAX(s.last_activity_at) as last_session_at
22            FROM users u
23            LEFT JOIN user_sessions s ON s.user_id = u.id
24            WHERE u.id = $1 AND u.status != $2
25            GROUP BY u.id
26            "#,
27            user_id.as_str(),
28            deleted_status
29        )
30        .fetch_optional(&*self.pool)
31        .await?;
32
33        Ok(row)
34    }
35
36    pub async fn get_activity(&self, user_id: &UserId) -> Result<UserActivity> {
37        let row = sqlx::query_as!(
38            UserActivity,
39            r#"
40            SELECT
41                u.id as user_id,
42                MAX(s.last_activity_at) as last_active,
43                COUNT(DISTINCT s.session_id) as "session_count!",
44                COUNT(DISTINCT t.task_id) as "task_count!",
45                0::bigint as "message_count!"
46            FROM users u
47            LEFT JOIN user_sessions s ON s.user_id = u.id
48            LEFT JOIN agent_tasks t ON t.user_id = u.id
49            WHERE u.id = $1
50            GROUP BY u.id
51            "#,
52            user_id.as_str()
53        )
54        .fetch_one(&*self.pool)
55        .await?;
56
57        Ok(row)
58    }
59
60    pub async fn list(&self, limit: i64, offset: i64) -> Result<Vec<User>> {
61        let safe_limit = limit.min(MAX_PAGE_SIZE);
62        let deleted_status = UserStatus::Deleted.as_str();
63        let rows = sqlx::query_as!(
64            User,
65            r#"
66            SELECT id, name, email, full_name, display_name, status, email_verified,
67                   roles, avatar_url, is_bot, is_scanner, created_at, updated_at
68            FROM users
69            WHERE status != $1
70            ORDER BY created_at DESC
71            LIMIT $2 OFFSET $3
72            "#,
73            deleted_status,
74            safe_limit,
75            offset
76        )
77        .fetch_all(&*self.pool)
78        .await?;
79
80        Ok(rows)
81    }
82
83    pub async fn list_all(&self) -> Result<Vec<User>> {
84        let deleted_status = UserStatus::Deleted.as_str();
85        let rows = sqlx::query_as!(
86            User,
87            r#"
88            SELECT id, name, email, full_name, display_name, status, email_verified,
89                   roles, avatar_url, is_bot, is_scanner, created_at, updated_at
90            FROM users
91            WHERE status != $1
92            ORDER BY created_at DESC
93            "#,
94            deleted_status
95        )
96        .fetch_all(&*self.pool)
97        .await?;
98
99        Ok(rows)
100    }
101
102    pub async fn search(&self, query: &str, limit: i64) -> Result<Vec<User>> {
103        let safe_limit = limit.min(MAX_PAGE_SIZE);
104        let pattern = format!("%{query}%");
105        let deleted_status = UserStatus::Deleted.as_str();
106        let rows = sqlx::query_as!(
107            User,
108            r#"
109            SELECT id, name, email, full_name, display_name, status, email_verified,
110                   roles, avatar_url, is_bot, is_scanner, created_at, updated_at
111            FROM users
112            WHERE status != $1
113              AND (name ILIKE $2 OR email ILIKE $2 OR full_name ILIKE $2)
114            ORDER BY
115                CASE WHEN name ILIKE $2 THEN 0 ELSE 1 END,
116                created_at DESC
117            LIMIT $3
118            "#,
119            deleted_status,
120            pattern,
121            safe_limit
122        )
123        .fetch_all(&*self.pool)
124        .await?;
125
126        Ok(rows)
127    }
128
129    pub async fn count(&self) -> Result<i64> {
130        let deleted_status = UserStatus::Deleted.as_str();
131        let result = sqlx::query_scalar!(
132            r#"SELECT COUNT(*) as "count!" FROM users WHERE status != $1"#,
133            deleted_status
134        )
135        .fetch_one(&*self.pool)
136        .await?;
137
138        Ok(result)
139    }
140
141    pub async fn bulk_update_status(&self, user_ids: &[UserId], new_status: &str) -> Result<u64> {
142        let ids: Vec<String> = user_ids.iter().map(ToString::to_string).collect();
143        let result = sqlx::query!(
144            r#"
145            UPDATE users
146            SET status = $1, updated_at = NOW()
147            WHERE id = ANY($2)
148            "#,
149            new_status,
150            &ids[..]
151        )
152        .execute(&*self.write_pool)
153        .await?;
154
155        Ok(result.rows_affected())
156    }
157
158    pub async fn bulk_delete(&self, user_ids: &[UserId]) -> Result<u64> {
159        let deleted_status = UserStatus::Deleted.as_str();
160        let ids: Vec<String> = user_ids.iter().map(ToString::to_string).collect();
161        let result = sqlx::query!(
162            r#"
163            UPDATE users
164            SET status = $1, updated_at = NOW()
165            WHERE id = ANY($2)
166            "#,
167            deleted_status,
168            &ids[..]
169        )
170        .execute(&*self.write_pool)
171        .await?;
172
173        Ok(result.rows_affected())
174    }
175
176    pub async fn list_by_filter(
177        &self,
178        status: Option<&str>,
179        role: Option<&str>,
180        older_than_days: Option<i64>,
181        limit: i64,
182    ) -> Result<Vec<User>> {
183        let safe_limit = limit.min(MAX_PAGE_SIZE);
184        let deleted_status = UserStatus::Deleted.as_str();
185
186        let rows = sqlx::query_as!(
187            User,
188            r#"
189            SELECT id, name, email, full_name, display_name, status, email_verified,
190                   roles, avatar_url, is_bot, is_scanner, created_at, updated_at
191            FROM users
192            WHERE status != $1
193              AND ($2::text IS NULL OR status = $2)
194              AND ($3::text IS NULL OR $3 = ANY(roles))
195              AND ($4::bigint IS NULL OR created_at < NOW() - make_interval(days => $4::int))
196            ORDER BY created_at DESC
197            LIMIT $5
198            "#,
199            deleted_status,
200            status,
201            role,
202            older_than_days,
203            safe_limit
204        )
205        .fetch_all(&*self.pool)
206        .await?;
207
208        Ok(rows)
209    }
210
211    pub async fn is_temporary_anonymous(&self, id: &UserId) -> Result<bool> {
212        let anonymous_role = UserRole::Anonymous.as_str();
213        let result = sqlx::query_scalar!(
214            r#"
215            SELECT $1 = ANY(roles) as "is_anonymous!"
216            FROM users
217            WHERE id = $2
218            "#,
219            anonymous_role,
220            id.as_str()
221        )
222        .fetch_optional(&*self.pool)
223        .await?;
224
225        result.ok_or(UserError::NotFound(id.clone()))
226    }
227
228    pub async fn list_non_anonymous_with_sessions(
229        &self,
230        limit: i64,
231    ) -> Result<Vec<UserWithSessions>> {
232        let safe_limit = limit.min(MAX_PAGE_SIZE);
233        let deleted_status = UserStatus::Deleted.as_str();
234        let anonymous_role = UserRole::Anonymous.as_str();
235        let rows = sqlx::query_as!(
236            UserWithSessions,
237            r#"
238            SELECT
239                u.id, u.name, u.email, u.full_name, u.status, u.roles, u.created_at,
240                COUNT(s.session_id) FILTER (WHERE s.ended_at IS NULL) as "active_sessions!",
241                MAX(s.last_activity_at) as last_session_at
242            FROM users u
243            LEFT JOIN user_sessions s ON s.user_id = u.id
244            WHERE u.status != $1
245              AND NOT ($2 = ANY(u.roles))
246            GROUP BY u.id
247            ORDER BY last_session_at DESC NULLS LAST
248            LIMIT $3
249            "#,
250            deleted_status,
251            anonymous_role,
252            safe_limit
253        )
254        .fetch_all(&*self.pool)
255        .await?;
256
257        Ok(rows)
258    }
259}
systemprompt_users/repository/user/list.rs

systemprompt_users/repository/user/
list.rs
