sbol_db_postgres/repo/
users.rs

1//! The account store over Postgres.
2//!
3//! Rows map to [`User`]. Usernames are unique. Production migration permits
4//! duplicate classic emails; lookup therefore prefers an exact username and
5//! rejects an ambiguous email instead of selecting an arbitrary account.
6
7use async_trait::async_trait;
8use chrono::Utc;
9use sbol_db_core::{DomainError, NewUser, User, UserId};
10use sbol_db_storage::{UserListQuery, UserPage, UserStore};
11use sqlx::Row;
12use uuid::Uuid;
13
14use crate::repo::db_err;
15use crate::PgPool;
16
17const USER_COLS: &str = "id, username, name, email, affiliation, password_hash, \
18    graph_uri, is_admin, is_curator, is_member, reset_password_link, created_at, updated_at";
19
20#[derive(Clone)]
21pub struct PgUserStore {
22    pool: PgPool,
23}
24
25impl PgUserStore {
26    pub fn new(pool: PgPool) -> Self {
27        Self { pool }
28    }
29}
30
31#[async_trait]
32impl UserStore for PgUserStore {
33    async fn create_user(&self, new_user: NewUser) -> Result<User, DomainError> {
34        let id = UserId::new();
35        let now = Utc::now();
36        let row = sqlx::query(&format!(
37            "INSERT INTO sbh_user \
38             (id, username, name, email, affiliation, password_hash, \
39              graph_uri, is_admin, is_curator, is_member, created_at, updated_at) \
40             VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12) \
41             RETURNING {USER_COLS}"
42        ))
43        .bind(id.as_uuid())
44        .bind(&new_user.username)
45        .bind(&new_user.name)
46        .bind(&new_user.email)
47        .bind(new_user.affiliation.as_deref())
48        .bind(&new_user.password_hash)
49        .bind(&new_user.graph_uri)
50        .bind(new_user.is_admin)
51        .bind(new_user.is_curator)
52        .bind(new_user.is_member)
53        .bind(now)
54        .bind(now)
55        .fetch_one(&self.pool)
56        .await
57        .map_err(db_err)?;
58        row_to_user(row)
59    }
60
61    async fn find_by_email_or_username(
62        &self,
63        identifier: &str,
64    ) -> Result<Option<User>, DomainError> {
65        let username = sqlx::query(&format!(
66            "SELECT {USER_COLS} FROM sbh_user WHERE username = $1"
67        ))
68        .bind(identifier)
69        .fetch_optional(&self.pool)
70        .await
71        .map_err(db_err)?;
72        if let Some(row) = username {
73            return row_to_user(row).map(Some);
74        }
75        let rows = sqlx::query(&format!(
76            "SELECT {USER_COLS} FROM sbh_user WHERE email = $1 ORDER BY username LIMIT 2"
77        ))
78        .bind(identifier)
79        .fetch_all(&self.pool)
80        .await
81        .map_err(db_err)?;
82        match rows.len() {
83            0 => Ok(None),
84            1 => row_to_user(rows.into_iter().next().expect("one row")).map(Some),
85            _ => Err(DomainError::Validation(
86                "multiple accounts use this email; log in with your username".to_owned(),
87            )),
88        }
89    }
90
91    async fn get_by_id(&self, id: UserId) -> Result<Option<User>, DomainError> {
92        let row = sqlx::query(&format!("SELECT {USER_COLS} FROM sbh_user WHERE id = $1"))
93            .bind(id.as_uuid())
94            .fetch_optional(&self.pool)
95            .await
96            .map_err(db_err)?;
97        row.map(row_to_user).transpose()
98    }
99
100    async fn list_users(&self) -> Result<Vec<User>, DomainError> {
101        let rows = sqlx::query(&format!(
102            "SELECT {USER_COLS} FROM sbh_user ORDER BY username"
103        ))
104        .fetch_all(&self.pool)
105        .await
106        .map_err(db_err)?;
107        rows.into_iter().map(row_to_user).collect()
108    }
109
110    async fn page_users(&self, query: &UserListQuery) -> Result<UserPage, DomainError> {
111        let needle = query
112            .text
113            .as_deref()
114            .map(str::trim)
115            .filter(|value| !value.is_empty())
116            .map(str::to_lowercase);
117        let predicate = "($1::text IS NULL OR position($1 in lower(username)) > 0 OR \
118            position($1 in lower(name)) > 0 OR position($1 in lower(email)) > 0 OR \
119            position($1 in lower(coalesce(affiliation, ''))) > 0)";
120        let total: i64 =
121            sqlx::query_scalar(&format!("SELECT count(*) FROM sbh_user WHERE {predicate}"))
122                .bind(needle.as_deref())
123                .fetch_one(&self.pool)
124                .await
125                .map_err(db_err)?;
126        let rows = sqlx::query(&format!(
127            "SELECT {USER_COLS} FROM sbh_user WHERE {predicate} \
128             ORDER BY lower(username), username LIMIT $2 OFFSET $3"
129        ))
130        .bind(needle.as_deref())
131        .bind(i64::from(query.limit))
132        .bind(i64::try_from(query.offset).unwrap_or(i64::MAX))
133        .fetch_all(&self.pool)
134        .await
135        .map_err(db_err)?;
136        Ok(UserPage {
137            total: total.max(0) as u64,
138            items: rows
139                .into_iter()
140                .map(row_to_user)
141                .collect::<Result<_, _>>()?,
142        })
143    }
144
145    async fn update_user(&self, user: &User) -> Result<User, DomainError> {
146        let row = sqlx::query(&format!(
147            "UPDATE sbh_user \
148                SET name = $2, email = $3, affiliation = $4, \
149                    is_admin = $5, is_curator = $6, is_member = $7, updated_at = $8 \
150              WHERE id = $1 \
151             RETURNING {USER_COLS}"
152        ))
153        .bind(user.id.as_uuid())
154        .bind(&user.name)
155        .bind(&user.email)
156        .bind(user.affiliation.as_deref())
157        .bind(user.is_admin)
158        .bind(user.is_curator)
159        .bind(user.is_member)
160        .bind(Utc::now())
161        .fetch_optional(&self.pool)
162        .await
163        .map_err(db_err)?
164        .ok_or_else(|| DomainError::NotFound(format!("user {}", user.id)))?;
165        row_to_user(row)
166    }
167
168    async fn set_sole_admin(&self, id: UserId) -> Result<(), DomainError> {
169        let mut tx = self.pool.begin().await.map_err(db_err)?;
170        let exists: bool =
171            sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM sbh_user WHERE id = $1)")
172                .bind(id.as_uuid())
173                .fetch_one(&mut *tx)
174                .await
175                .map_err(db_err)?;
176        if !exists {
177            return Err(DomainError::NotFound(format!("user {id}")));
178        }
179        let now = Utc::now();
180        sqlx::query(
181            "UPDATE sbh_user \
182                SET updated_at = CASE \
183                        WHEN is_admin IS DISTINCT FROM (id = $1) THEN $2 \
184                        ELSE updated_at \
185                    END, \
186                    is_admin = (id = $1)",
187        )
188        .bind(id.as_uuid())
189        .bind(now)
190        .execute(&mut *tx)
191        .await
192        .map_err(db_err)?;
193        tx.commit().await.map_err(db_err)?;
194        Ok(())
195    }
196
197    async fn set_password_hash(&self, id: UserId, password_hash: &str) -> Result<(), DomainError> {
198        sqlx::query("UPDATE sbh_user SET password_hash = $2, updated_at = $3 WHERE id = $1")
199            .bind(id.as_uuid())
200            .bind(password_hash)
201            .bind(Utc::now())
202            .execute(&self.pool)
203            .await
204            .map_err(db_err)?;
205        Ok(())
206    }
207
208    async fn set_reset_link(&self, id: UserId, link: Option<&str>) -> Result<(), DomainError> {
209        sqlx::query("UPDATE sbh_user SET reset_password_link = $2 WHERE id = $1")
210            .bind(id.as_uuid())
211            .bind(link)
212            .execute(&self.pool)
213            .await
214            .map_err(db_err)?;
215        Ok(())
216    }
217
218    async fn consume_reset_link(&self, link: &str) -> Result<Option<User>, DomainError> {
219        let row = sqlx::query(&format!(
220            "UPDATE sbh_user SET reset_password_link = NULL \
221              WHERE reset_password_link = $1 \
222             RETURNING {USER_COLS}"
223        ))
224        .bind(link)
225        .fetch_optional(&self.pool)
226        .await
227        .map_err(db_err)?;
228        row.map(row_to_user).transpose()
229    }
230
231    async fn delete_user(&self, id: UserId) -> Result<bool, DomainError> {
232        let result = sqlx::query("DELETE FROM sbh_user WHERE id = $1")
233            .bind(id.as_uuid())
234            .execute(&self.pool)
235            .await
236            .map_err(db_err)?;
237        Ok(result.rows_affected() > 0)
238    }
239
240    async fn any_admin(&self) -> Result<bool, DomainError> {
241        let exists: bool =
242            sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM sbh_user WHERE is_admin = true)")
243                .fetch_one(&self.pool)
244                .await
245                .map_err(db_err)?;
246        Ok(exists)
247    }
248}
249
250fn row_to_user(row: sqlx::postgres::PgRow) -> Result<User, DomainError> {
251    let id: Uuid = row.try_get("id").map_err(db_err)?;
252    Ok(User {
253        id: UserId(id),
254        username: row.try_get("username").map_err(db_err)?,
255        name: row.try_get("name").map_err(db_err)?,
256        email: row.try_get("email").map_err(db_err)?,
257        affiliation: row.try_get("affiliation").map_err(db_err)?,
258        password_hash: row.try_get("password_hash").map_err(db_err)?,
259        graph_uri: row.try_get("graph_uri").map_err(db_err)?,
260        is_admin: row.try_get("is_admin").map_err(db_err)?,
261        is_curator: row.try_get("is_curator").map_err(db_err)?,
262        is_member: row.try_get("is_member").map_err(db_err)?,
263        reset_password_link: row.try_get("reset_password_link").map_err(db_err)?,
264        created_at: row.try_get("created_at").map_err(db_err)?,
265        updated_at: row.try_get("updated_at").map_err(db_err)?,
266    })
267}
sbol_db_postgres/repo/users.rs

sbol_db_postgres/repo/
users.rs
