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::{
10    User, UserActivity, UserActivityRow, UserRole, UserRow, UserStatus, UserWithSessions,
11    UserWithSessionsRow,
12};
13use crate::repository::{MAX_PAGE_SIZE, UserRepository};
14
15impl UserRepository {
16    pub async fn find_with_sessions(&self, user_id: &UserId) -> Result<Option<UserWithSessions>> {
17        let deleted_status = UserStatus::Deleted.as_str();
18        let row = sqlx::query_as!(
19            UserWithSessionsRow,
20            r#"
21            SELECT
22                u.id, u.name, u.email, u.full_name, u.status, u.roles, u.created_at,
23                COUNT(s.session_id) FILTER (WHERE s.ended_at IS NULL) as "active_sessions!",
24                MAX(s.last_activity_at) as last_session_at
25            FROM users u
26            LEFT JOIN user_sessions s ON s.user_id = u.id
27            WHERE u.id = $1 AND u.status != $2
28            GROUP BY u.id
29            "#,
30            user_id.as_str(),
31            deleted_status
32        )
33        .fetch_optional(&*self.pool)
34        .await?
35        .map(UserWithSessions::try_from)
36        .transpose()?;
37
38        Ok(row)
39    }
40
41    pub async fn get_activity(&self, user_id: &UserId) -> Result<UserActivity> {
42        let row = sqlx::query_as!(
43            UserActivityRow,
44            r#"
45            SELECT
46                u.id as user_id,
47                MAX(s.last_activity_at) as last_active,
48                COUNT(s.session_id) as "session_count!",
49                COALESCE(SUM(s.task_count), 0)::bigint as "task_count!",
50                COALESCE(SUM(s.message_count), 0)::bigint as "message_count!"
51            FROM users u
52            LEFT JOIN user_sessions s ON s.user_id = u.id
53            WHERE u.id = $1
54            GROUP BY u.id
55            "#,
56            user_id.as_str()
57        )
58        .fetch_one(&*self.pool)
59        .await
60        .map(UserActivity::from)?;
61
62        Ok(row)
63    }
64
65    pub async fn list(&self, limit: i64, offset: i64) -> Result<Vec<User>> {
66        self.list_filtered(limit, offset, false).await
67    }
68
69    pub async fn list_including_anonymous(&self, limit: i64, offset: i64) -> Result<Vec<User>> {
70        self.list_filtered(limit, offset, true).await
71    }
72
73    async fn list_filtered(
74        &self,
75        limit: i64,
76        offset: i64,
77        include_anonymous: bool,
78    ) -> Result<Vec<User>> {
79        let safe_limit = limit.min(MAX_PAGE_SIZE);
80        let deleted_status = UserStatus::Deleted.as_str();
81        let anonymous_role = UserRole::Anonymous.as_str();
82        let rows = sqlx::query_as!(
83            UserRow,
84            r#"
85            SELECT id, name, email, full_name, display_name, status, email_verified,
86                   roles, avatar_url, is_bot, is_scanner, created_at, updated_at
87            FROM users
88            WHERE status != $1
89              AND ($4 OR NOT ($5 = ANY(roles)))
90            ORDER BY created_at DESC
91            LIMIT $2 OFFSET $3
92            "#,
93            deleted_status,
94            safe_limit,
95            offset,
96            include_anonymous,
97            anonymous_role
98        )
99        .fetch_all(&*self.pool)
100        .await?
101        .into_iter()
102        .map(User::try_from)
103        .collect::<Result<Vec<_>>>()?;
104
105        Ok(rows)
106    }
107
108    pub async fn list_all(&self) -> Result<Vec<User>> {
109        let deleted_status = UserStatus::Deleted.as_str();
110        let anonymous_role = UserRole::Anonymous.as_str();
111        let rows = sqlx::query_as!(
112            UserRow,
113            r#"
114            SELECT id, name, email, full_name, display_name, status, email_verified,
115                   roles, avatar_url, is_bot, is_scanner, created_at, updated_at
116            FROM users
117            WHERE status != $1
118              AND NOT ($2 = ANY(roles))
119            ORDER BY created_at DESC
120            "#,
121            deleted_status,
122            anonymous_role
123        )
124        .fetch_all(&*self.pool)
125        .await?
126        .into_iter()
127        .map(User::try_from)
128        .collect::<Result<Vec<_>>>()?;
129
130        Ok(rows)
131    }
132
133    pub async fn search(&self, query: &str, limit: i64) -> Result<Vec<User>> {
134        self.search_filtered(query, limit, false).await
135    }
136
137    pub async fn search_including_anonymous(&self, query: &str, limit: i64) -> Result<Vec<User>> {
138        self.search_filtered(query, limit, true).await
139    }
140
141    async fn search_filtered(
142        &self,
143        query: &str,
144        limit: i64,
145        include_anonymous: bool,
146    ) -> Result<Vec<User>> {
147        let safe_limit = limit.min(MAX_PAGE_SIZE);
148        let pattern = format!("%{query}%");
149        let deleted_status = UserStatus::Deleted.as_str();
150        let anonymous_role = UserRole::Anonymous.as_str();
151        let rows = sqlx::query_as!(
152            UserRow,
153            r#"
154            SELECT id, name, email, full_name, display_name, status, email_verified,
155                   roles, avatar_url, is_bot, is_scanner, created_at, updated_at
156            FROM users
157            WHERE status != $1
158              AND ($4 OR NOT ($5 = ANY(roles)))
159              AND (name ILIKE $2 OR email ILIKE $2 OR full_name ILIKE $2)
160            ORDER BY
161                CASE WHEN name ILIKE $2 THEN 0 ELSE 1 END,
162                created_at DESC
163            LIMIT $3
164            "#,
165            deleted_status,
166            pattern,
167            safe_limit,
168            include_anonymous,
169            anonymous_role
170        )
171        .fetch_all(&*self.pool)
172        .await?
173        .into_iter()
174        .map(User::try_from)
175        .collect::<Result<Vec<_>>>()?;
176
177        Ok(rows)
178    }
179
180    pub async fn count(&self) -> Result<i64> {
181        self.count_filtered(false).await
182    }
183
184    pub async fn count_including_anonymous(&self) -> Result<i64> {
185        self.count_filtered(true).await
186    }
187
188    async fn count_filtered(&self, include_anonymous: bool) -> Result<i64> {
189        let deleted_status = UserStatus::Deleted.as_str();
190        let anonymous_role = UserRole::Anonymous.as_str();
191        let result = sqlx::query_scalar!(
192            r#"SELECT COUNT(*) as "count!" FROM users
193               WHERE status != $1
194                 AND ($2 OR NOT ($3 = ANY(roles)))"#,
195            deleted_status,
196            include_anonymous,
197            anonymous_role
198        )
199        .fetch_one(&*self.pool)
200        .await?;
201
202        Ok(result)
203    }
204
205    pub async fn list_by_filter(
206        &self,
207        status: Option<UserStatus>,
208        role: Option<&str>,
209        older_than_days: Option<i64>,
210        limit: i64,
211    ) -> Result<Vec<User>> {
212        let safe_limit = limit.min(MAX_PAGE_SIZE);
213        let deleted_status = UserStatus::Deleted.as_str();
214        let status = status.as_ref().map(UserStatus::as_str);
215
216        let rows = sqlx::query_as!(
217            UserRow,
218            r#"
219            SELECT id, name, email, full_name, display_name, status, email_verified,
220                   roles, avatar_url, is_bot, is_scanner, created_at, updated_at
221            FROM users
222            WHERE status != $1
223              AND ($2::text IS NULL OR status = $2)
224              AND ($3::text IS NULL OR $3 = ANY(roles))
225              AND ($4::bigint IS NULL OR created_at < NOW() - make_interval(days => $4::int))
226            ORDER BY created_at DESC
227            LIMIT $5
228            "#,
229            deleted_status,
230            status,
231            role,
232            older_than_days,
233            safe_limit
234        )
235        .fetch_all(&*self.pool)
236        .await?
237        .into_iter()
238        .map(User::try_from)
239        .collect::<Result<Vec<_>>>()?;
240
241        Ok(rows)
242    }
243
244    pub async fn is_temporary_anonymous(&self, id: &UserId) -> Result<bool> {
245        let anonymous_role = UserRole::Anonymous.as_str();
246        let result = sqlx::query_scalar!(
247            r#"
248            SELECT $1 = ANY(roles) as "is_anonymous!"
249            FROM users
250            WHERE id = $2
251            "#,
252            anonymous_role,
253            id.as_str()
254        )
255        .fetch_optional(&*self.pool)
256        .await?;
257
258        result.ok_or(UserError::NotFound(id.clone()))
259    }
260
261    pub async fn list_non_anonymous_with_sessions(
262        &self,
263        limit: i64,
264    ) -> Result<Vec<UserWithSessions>> {
265        let safe_limit = limit.min(MAX_PAGE_SIZE);
266        let deleted_status = UserStatus::Deleted.as_str();
267        let anonymous_role = UserRole::Anonymous.as_str();
268        let rows = sqlx::query_as!(
269            UserWithSessionsRow,
270            r#"
271            SELECT
272                u.id, u.name, u.email, u.full_name, u.status, u.roles, u.created_at,
273                COUNT(s.session_id) FILTER (WHERE s.ended_at IS NULL) as "active_sessions!",
274                MAX(s.last_activity_at) as last_session_at
275            FROM users u
276            LEFT JOIN user_sessions s ON s.user_id = u.id
277            WHERE u.status != $1
278              AND NOT ($2 = ANY(u.roles))
279            GROUP BY u.id
280            ORDER BY last_session_at DESC NULLS LAST
281            LIMIT $3
282            "#,
283            deleted_status,
284            anonymous_role,
285            safe_limit
286        )
287        .fetch_all(&*self.pool)
288        .await?
289        .into_iter()
290        .map(UserWithSessions::try_from)
291        .collect::<Result<Vec<_>>>()?;
292
293        Ok(rows)
294    }
295}
systemprompt_users/repository/user/list.rs

systemprompt_users/repository/user/
list.rs
