headless_lms_models/
users.rs

1use crate::library::oauth::Digest;
2use crate::prelude::*;
3use utoipa::ToSchema;
4
5#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, ToSchema)]
6
7pub struct User {
8    pub id: Uuid,
9    pub created_at: DateTime<Utc>,
10    pub updated_at: DateTime<Utc>,
11    pub deleted_at: Option<DateTime<Utc>>,
12    pub upstream_id: Option<i32>,
13    pub email_domain: Option<String>,
14}
15
16/// The domain part of an address, as stored in `users.email_domain`.
17///
18/// Every writer of `user_details.email` must use this, or the derived column drifts silently.
19pub fn email_domain_from_email(email: &str) -> Option<&str> {
20    email.trim().split('@').next_back()
21}
22
23pub async fn insert(
24    conn: &mut PgConnection,
25    pkey_policy: PKeyPolicy<Uuid>,
26    email: &str,
27    first_name: Option<&str>,
28    last_name: Option<&str>,
29) -> ModelResult<Uuid> {
30    let mut tx = conn.begin().await?;
31    let email_domain = email_domain_from_email(email);
32    let res = sqlx::query!(
33        "
34INSERT INTO users (id, email_domain)
35VALUES ($1, $2)
36RETURNING *
37",
38        pkey_policy.into_uuid(),
39        email_domain
40    )
41    .fetch_one(&mut *tx)
42    .await?;
43
44    let _res2 = sqlx::query!(
45        "
46INSERT INTO user_details (user_id, email, first_name, last_name)
47VALUES ($1, $2, $3, $4)
48",
49        res.id,
50        email,
51        first_name,
52        last_name
53    )
54    .execute(&mut *tx)
55    .await?;
56    tx.commit().await?;
57    Ok(res.id)
58}
59
60pub async fn insert_with_upstream_id_and_moocfi_id(
61    conn: &mut PgConnection,
62    email: &str,
63    first_name: Option<&str>,
64    last_name: Option<&str>,
65    upstream_id: i32,
66    moocfi_id: Uuid,
67) -> ModelResult<User> {
68    info!("The user is not in the database yet, inserting");
69    let email_domain = email_domain_from_email(email);
70    let mut tx = conn.begin().await?;
71    let user = sqlx::query_as!(
72        User,
73        r#"
74INSERT INTO
75  users (id, upstream_id, email_domain)
76VALUES ($1, $2, $3)
77RETURNING *;
78          "#,
79        moocfi_id,
80        upstream_id,
81        email_domain
82    )
83    .fetch_one(&mut *tx)
84    .await?;
85
86    let _res2 = sqlx::query!(
87        "
88INSERT INTO user_details (user_id, email, first_name, last_name)
89VALUES ($1, $2, $3, $4)
90",
91        user.id,
92        email,
93        first_name,
94        last_name
95    )
96    .execute(&mut *tx)
97    .await?;
98    tx.commit().await?;
99    Ok(user)
100}
101
102/// Looks up a user by email (case-insensitive) using the `lower(email)` index on `user_details`.
103pub async fn get_by_email(conn: &mut PgConnection, email: &str) -> ModelResult<User> {
104    let user = sqlx::query_as!(
105        User,
106        "
107SELECT users.*
108FROM user_details
109JOIN users ON (user_details.user_id = users.id)
110WHERE lower(user_details.email) = lower($1)
111        ",
112        email
113    )
114    .fetch_one(conn)
115    .await?;
116    Ok(user)
117}
118
119pub async fn get_by_id(conn: &mut PgConnection, id: Uuid) -> ModelResult<User> {
120    let user = sqlx::query_as!(
121        User,
122        "
123SELECT *
124FROM users
125WHERE id = $1
126        ",
127        id
128    )
129    .fetch_one(conn)
130    .await?;
131    Ok(user)
132}
133
134pub async fn get_by_ids(conn: &mut PgConnection, ids: &[Uuid]) -> ModelResult<Vec<User>> {
135    let users = sqlx::query_as!(
136        User,
137        "
138SELECT *
139FROM users
140WHERE id = ANY($1)
141  AND deleted_at IS NULL
142        ",
143        ids
144    )
145    .fetch_all(conn)
146    .await?;
147    Ok(users)
148}
149
150/// Like [`get_by_id`], but only returns a user that is not soft-deleted.
151///
152/// A soft-deleted (banned/removed) user yields `RecordNotFound`, the same as a nonexistent id,
153/// so callers that must reject deleted accounts can treat both cases identically. [`get_by_id`]
154/// deliberately does not filter, since some callers fetch a user's own deleted row.
155pub async fn get_active_by_id(conn: &mut PgConnection, id: Uuid) -> ModelResult<User> {
156    let user = sqlx::query_as!(
157        User,
158        "
159SELECT *
160FROM users
161WHERE id = $1
162  AND deleted_at IS NULL
163        ",
164        id
165    )
166    .fetch_one(conn)
167    .await?;
168    Ok(user)
169}
170
171pub async fn find_by_upstream_id(
172    conn: &mut PgConnection,
173    upstream_id: i32,
174) -> ModelResult<Option<User>> {
175    let user = sqlx::query_as!(
176        User,
177        "SELECT * FROM users WHERE upstream_id = $1 AND deleted_at IS NULL",
178        upstream_id
179    )
180    .fetch_optional(conn)
181    .await?;
182    Ok(user)
183}
184
185/// Includes all users who have returned an exercise on a course course instance
186pub async fn get_all_user_ids_with_user_exercise_states_on_course(
187    conn: &mut PgConnection,
188    course_id: Uuid,
189) -> ModelResult<Vec<Uuid>> {
190    let res = sqlx::query!(
191        "
192SELECT DISTINCT user_id
193FROM user_exercise_states
194WHERE course_id = $1
195  AND deleted_at IS NULL
196        ",
197        course_id
198    )
199    .map(|x| x.user_id)
200    .fetch_all(conn)
201    .await?;
202    Ok(res)
203}
204
205pub async fn get_users_by_course_instance_enrollment(
206    conn: &mut PgConnection,
207    course_instance_id: Uuid,
208) -> ModelResult<Vec<User>> {
209    let res = sqlx::query_as!(
210        User,
211        "
212SELECT *
213FROM users
214WHERE id IN (
215    SELECT user_id
216    FROM course_instance_enrollments
217    WHERE course_instance_id = $1
218      AND deleted_at IS NULL
219  )
220",
221        course_instance_id,
222    )
223    .fetch_all(&mut *conn)
224    .await?;
225    Ok(res)
226}
227
228pub async fn get_users_ids_in_db_from_upstream_ids(
229    conn: &mut PgConnection,
230    upstream_ids: &[i32],
231) -> ModelResult<Vec<Uuid>> {
232    let res = sqlx::query!(
233        "
234SELECT *
235FROM users
236WHERE upstream_id IN (
237    SELECT UNNEST($1::integer [])
238  )
239AND deleted_at IS NULL
240",
241        upstream_ids,
242    )
243    .fetch_all(&mut *conn)
244    .await?;
245    Ok(res.iter().map(|x| x.id).collect::<Vec<_>>())
246}
247
248/// Writes the new email onto `user_details` and keeps `users.email_domain` in step, within an
249/// already-open transaction. The `clear_email_verification` trigger drops proof of the old
250/// address as part of the `user_details` update.
251async fn apply_email_update(
252    tx: &mut PgConnection,
253    user_id: Uuid,
254    new_email: &str,
255) -> ModelResult<()> {
256    sqlx::query!(
257        "UPDATE user_details SET email = $1 WHERE user_id = $2",
258        new_email,
259        user_id,
260    )
261    .execute(&mut *tx)
262    .await?;
263
264    let email_domain = email_domain_from_email(new_email);
265    sqlx::query!(
266        "UPDATE users SET email_domain = $1 WHERE id = $2",
267        email_domain,
268        user_id,
269    )
270    .execute(&mut *tx)
271    .await?;
272
273    Ok(())
274}
275
276/// Points the account with this upstream id at a new address. The returned local user id lets
277/// the caller mail a fresh link.
278pub async fn update_email_for_user(
279    conn: &mut PgConnection,
280    upstream_id: &i32,
281    new_email: String,
282) -> ModelResult<Uuid> {
283    info!("Updating user (Upstream id: {upstream_id})");
284    let mut tx = conn.begin().await?;
285
286    let user = sqlx::query_as!(
287        User,
288        "SELECT * FROM users WHERE upstream_id = $1 AND deleted_at IS NULL",
289        upstream_id
290    )
291    .fetch_one(&mut *tx)
292    .await?;
293
294    apply_email_update(&mut tx, user.id, &new_email).await?;
295
296    tx.commit().await?;
297
298    info!("Email change succeeded");
299    Ok(user.id)
300}
301
302/// Points the account at a new address by user id rather than upstream id, for accounts (e.g.
303/// local-only ones) that have no upstream id. Same effect as [update_email_for_user] otherwise.
304pub async fn update_email_for_user_by_id(
305    conn: &mut PgConnection,
306    user_id: Uuid,
307    new_email: &str,
308) -> ModelResult<()> {
309    let mut tx = conn.begin().await?;
310    apply_email_update(&mut tx, user_id, new_email).await?;
311    tx.commit().await?;
312    Ok(())
313}
314
315/// Soft-deletes the user and takes their OAuth credentials down with the account.
316///
317/// Returns the digests of the deleted access tokens so a caller holding the exercise-services
318/// token cache can evict them; otherwise a deleted account keeps authenticating that API from a
319/// cache hit until the entry ages out (see `domain::exercise_services::token`).
320pub async fn delete_user(conn: &mut PgConnection, id: Uuid) -> ModelResult<Vec<Digest>> {
321    info!("Deleting user {id}");
322    let mut tx = conn.begin().await?;
323    crate::email_deliveries::soft_delete_unsent_retryable_deliveries_for_user(&mut tx, id).await?;
324    sqlx::query!("DELETE FROM user_details WHERE user_id = $1", id,)
325        .execute(&mut *tx)
326        .await?;
327    sqlx::query!("DELETE FROM user_passwords WHERE user_id = $1", id,)
328        .execute(&mut *tx)
329        .await?;
330    sqlx::query!(
331        "UPDATE users set deleted_at = now() WHERE id = $1 AND deleted_at IS NULL",
332        id,
333    )
334    .execute(&mut *tx)
335    .await?;
336    sqlx::query!(
337        "UPDATE roles set deleted_at = now() WHERE user_id = $1 AND deleted_at IS NULL",
338        id,
339    )
340    .execute(&mut *tx)
341    .await?;
342    let revoked_access_digests =
343        crate::oauth_refresh_tokens::OAuthRefreshTokens::revoke_all_grants_of_user_in_transaction(
344            &mut tx, id,
345        )
346        .await?;
347    tx.commit().await?;
348    info!("Deletion succeeded");
349    Ok(revoked_access_digests)
350}
headless_lms_models/users.rs

headless_lms_models/
users.rs
