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

systemprompt_users/repository/user/
list.rs
