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        self.list_filtered(limit, offset, false).await
62    }
63
64    pub async fn list_including_anonymous(&self, limit: i64, offset: i64) -> Result<Vec<User>> {
65        self.list_filtered(limit, offset, true).await
66    }
67
68    async fn list_filtered(
69        &self,
70        limit: i64,
71        offset: i64,
72        include_anonymous: bool,
73    ) -> Result<Vec<User>> {
74        let safe_limit = limit.min(MAX_PAGE_SIZE);
75        let deleted_status = UserStatus::Deleted.as_str();
76        let anonymous_role = UserRole::Anonymous.as_str();
77        let rows = sqlx::query_as!(
78            User,
79            r#"
80            SELECT id, name, email, full_name, display_name, status, email_verified,
81                   roles, avatar_url, is_bot, is_scanner, created_at, updated_at
82            FROM users
83            WHERE status != $1
84              AND ($4 OR NOT ($5 = ANY(roles)))
85            ORDER BY created_at DESC
86            LIMIT $2 OFFSET $3
87            "#,
88            deleted_status,
89            safe_limit,
90            offset,
91            include_anonymous,
92            anonymous_role
93        )
94        .fetch_all(&*self.pool)
95        .await?;
96
97        Ok(rows)
98    }
99
100    pub async fn list_all(&self) -> Result<Vec<User>> {
101        let deleted_status = UserStatus::Deleted.as_str();
102        let anonymous_role = UserRole::Anonymous.as_str();
103        let rows = sqlx::query_as!(
104            User,
105            r#"
106            SELECT id, name, email, full_name, display_name, status, email_verified,
107                   roles, avatar_url, is_bot, is_scanner, created_at, updated_at
108            FROM users
109            WHERE status != $1
110              AND NOT ($2 = ANY(roles))
111            ORDER BY created_at DESC
112            "#,
113            deleted_status,
114            anonymous_role
115        )
116        .fetch_all(&*self.pool)
117        .await?;
118
119        Ok(rows)
120    }
121
122    pub async fn search(&self, query: &str, limit: i64) -> Result<Vec<User>> {
123        self.search_filtered(query, limit, false).await
124    }
125
126    pub async fn search_including_anonymous(&self, query: &str, limit: i64) -> Result<Vec<User>> {
127        self.search_filtered(query, limit, true).await
128    }
129
130    async fn search_filtered(
131        &self,
132        query: &str,
133        limit: i64,
134        include_anonymous: bool,
135    ) -> Result<Vec<User>> {
136        let safe_limit = limit.min(MAX_PAGE_SIZE);
137        let pattern = format!("%{query}%");
138        let deleted_status = UserStatus::Deleted.as_str();
139        let anonymous_role = UserRole::Anonymous.as_str();
140        let rows = sqlx::query_as!(
141            User,
142            r#"
143            SELECT id, name, email, full_name, display_name, status, email_verified,
144                   roles, avatar_url, is_bot, is_scanner, created_at, updated_at
145            FROM users
146            WHERE status != $1
147              AND ($4 OR NOT ($5 = ANY(roles)))
148              AND (name ILIKE $2 OR email ILIKE $2 OR full_name ILIKE $2)
149            ORDER BY
150                CASE WHEN name ILIKE $2 THEN 0 ELSE 1 END,
151                created_at DESC
152            LIMIT $3
153            "#,
154            deleted_status,
155            pattern,
156            safe_limit,
157            include_anonymous,
158            anonymous_role
159        )
160        .fetch_all(&*self.pool)
161        .await?;
162
163        Ok(rows)
164    }
165
166    pub async fn count(&self) -> Result<i64> {
167        self.count_filtered(false).await
168    }
169
170    pub async fn count_including_anonymous(&self) -> Result<i64> {
171        self.count_filtered(true).await
172    }
173
174    async fn count_filtered(&self, include_anonymous: bool) -> Result<i64> {
175        let deleted_status = UserStatus::Deleted.as_str();
176        let anonymous_role = UserRole::Anonymous.as_str();
177        let result = sqlx::query_scalar!(
178            r#"SELECT COUNT(*) as "count!" FROM users
179               WHERE status != $1
180                 AND ($2 OR NOT ($3 = ANY(roles)))"#,
181            deleted_status,
182            include_anonymous,
183            anonymous_role
184        )
185        .fetch_one(&*self.pool)
186        .await?;
187
188        Ok(result)
189    }
190
191    pub async fn list_by_filter(
192        &self,
193        status: Option<&str>,
194        role: Option<&str>,
195        older_than_days: Option<i64>,
196        limit: i64,
197    ) -> Result<Vec<User>> {
198        let safe_limit = limit.min(MAX_PAGE_SIZE);
199        let deleted_status = UserStatus::Deleted.as_str();
200
201        let rows = sqlx::query_as!(
202            User,
203            r#"
204            SELECT id, name, email, full_name, display_name, status, email_verified,
205                   roles, avatar_url, is_bot, is_scanner, created_at, updated_at
206            FROM users
207            WHERE status != $1
208              AND ($2::text IS NULL OR status = $2)
209              AND ($3::text IS NULL OR $3 = ANY(roles))
210              AND ($4::bigint IS NULL OR created_at < NOW() - make_interval(days => $4::int))
211            ORDER BY created_at DESC
212            LIMIT $5
213            "#,
214            deleted_status,
215            status,
216            role,
217            older_than_days,
218            safe_limit
219        )
220        .fetch_all(&*self.pool)
221        .await?;
222
223        Ok(rows)
224    }
225
226    pub async fn is_temporary_anonymous(&self, id: &UserId) -> Result<bool> {
227        let anonymous_role = UserRole::Anonymous.as_str();
228        let result = sqlx::query_scalar!(
229            r#"
230            SELECT $1 = ANY(roles) as "is_anonymous!"
231            FROM users
232            WHERE id = $2
233            "#,
234            anonymous_role,
235            id.as_str()
236        )
237        .fetch_optional(&*self.pool)
238        .await?;
239
240        result.ok_or(UserError::NotFound(id.clone()))
241    }
242
243    pub async fn list_non_anonymous_with_sessions(
244        &self,
245        limit: i64,
246    ) -> Result<Vec<UserWithSessions>> {
247        let safe_limit = limit.min(MAX_PAGE_SIZE);
248        let deleted_status = UserStatus::Deleted.as_str();
249        let anonymous_role = UserRole::Anonymous.as_str();
250        let rows = sqlx::query_as!(
251            UserWithSessions,
252            r#"
253            SELECT
254                u.id, u.name, u.email, u.full_name, u.status, u.roles, u.created_at,
255                COUNT(s.session_id) FILTER (WHERE s.ended_at IS NULL) as "active_sessions!",
256                MAX(s.last_activity_at) as last_session_at
257            FROM users u
258            LEFT JOIN user_sessions s ON s.user_id = u.id
259            WHERE u.status != $1
260              AND NOT ($2 = ANY(u.roles))
261            GROUP BY u.id
262            ORDER BY last_session_at DESC NULLS LAST
263            LIMIT $3
264            "#,
265            deleted_status,
266            anonymous_role,
267            safe_limit
268        )
269        .fetch_all(&*self.pool)
270        .await?;
271
272        Ok(rows)
273    }
274}
systemprompt_users/repository/user/list.rs

systemprompt_users/repository/user/
list.rs
