1use crate::models::{AuthSession, AuthSessionRecord, AuthUser, AuthUserId};
2use crate::resolver::{SessionCache, session_token_hash};
3use chrono::{DateTime, Utc};
4use platform_core::{AppError, AppResult, DbPool, ErrorCode};
5use std::sync::Arc;
6
7#[async_trait::async_trait]
8pub trait AuthUserRepository: std::fmt::Debug + Send + Sync {
9 async fn insert(&self, user: &AuthUser) -> AppResult<()>;
10 async fn find_by_id(&self, user_id: &AuthUserId) -> AppResult<Option<AuthUser>>;
11 async fn list(&self, limit: i64, cursor: Option<&str>) -> AppResult<Vec<AuthUser>>;
12 async fn find_session_by_id(&self, session_id: &str) -> AppResult<Option<AuthSessionRecord>>;
13 async fn list_sessions(
14 &self,
15 limit: i64,
16 cursor: Option<&str>,
17 ) -> AppResult<Vec<AuthSessionRecord>>;
18 async fn revoke_session_by_id(
19 &self,
20 session_id: &str,
21 revoked_at: DateTime<Utc>,
22 ) -> AppResult<bool>;
23 async fn set_user_disabled_at(
24 &self,
25 user_id: &AuthUserId,
26 disabled_at: Option<DateTime<Utc>>,
27 disabled_reason: Option<&str>,
28 disabled_until: Option<DateTime<Utc>>,
29 ) -> AppResult<bool>;
30}
31
32#[derive(Debug, Clone)]
33pub struct PostgresAuthUserRepository {
34 pool: DbPool,
35 session_cache: Option<Arc<dyn SessionCache>>,
36}
37
38impl PostgresAuthUserRepository {
39 #[must_use]
40 pub fn new(pool: DbPool) -> Self {
41 Self {
42 pool,
43 session_cache: None,
44 }
45 }
46
47 #[must_use]
48 pub fn new_with_session_cache(
49 pool: DbPool,
50 session_cache: Option<Arc<dyn SessionCache>>,
51 ) -> Self {
52 Self {
53 pool,
54 session_cache,
55 }
56 }
57
58 pub async fn create_dev_session(
59 &self,
60 user_id: AuthUserId,
61 session_id: String,
62 token: String,
63 created_at: DateTime<Utc>,
64 expires_at: DateTime<Utc>,
65 ) -> AppResult<AuthSession> {
66 let mut tx = self.pool.begin().await.map_err(map_sql_error)?;
67
68 sqlx::query(
69 r#"
70 insert into auth.users (id, created_at, disabled_at, disabled_reason, disabled_until)
71 values ($1, $2, null, null, null)
72 on conflict (id) do nothing
73 "#,
74 )
75 .bind(&user_id.0)
76 .bind(created_at)
77 .execute(&mut *tx)
78 .await
79 .map_err(map_sql_error)?;
80
81 let active_user_exists = sqlx::query_scalar::<_, bool>(
82 r#"
83 select exists(
84 select 1
85 from auth.users
86 where id = $1
87 and (disabled_at is null or disabled_until <= now())
88 )
89 "#,
90 )
91 .bind(&user_id.0)
92 .fetch_one(&mut *tx)
93 .await
94 .map_err(map_sql_error)?;
95
96 if !active_user_exists {
97 return Err(AppError::new(ErrorCode::Forbidden, "Auth user is disabled"));
98 }
99
100 sqlx::query(
101 r#"
102 insert into auth.sessions (id, user_id, token_hash, device_id, created_at, expires_at, revoked_at)
103 values ($1, $2, $3, null, $4, $5, null)
104 "#,
105 )
106 .bind(&session_id)
107 .bind(&user_id.0)
108 .bind(session_token_hash(&token))
109 .bind(created_at)
110 .bind(expires_at)
111 .execute(&mut *tx)
112 .await
113 .map_err(map_sql_error)?;
114
115 tx.commit().await.map_err(map_sql_error)?;
116
117 Ok(AuthSession {
118 id: session_id,
119 user_id,
120 token,
121 device_id: None,
122 expires_at,
123 })
124 }
125
126 pub async fn revoke_session_token(
127 &self,
128 token: &str,
129 revoked_at: DateTime<Utc>,
130 ) -> AppResult<bool> {
131 let token_hash = session_token_hash(token);
132 let revoked_token_hash = sqlx::query_scalar::<_, String>(
133 r#"
134 update auth.sessions
135 set revoked_at = $2
136 where token_hash = $1
137 and revoked_at is null
138 returning token_hash
139 "#,
140 )
141 .bind(&token_hash)
142 .bind(revoked_at)
143 .fetch_optional(&self.pool)
144 .await
145 .map_err(map_sql_error)?;
146
147 if revoked_token_hash.is_some() {
148 self.delete_cached_token_hash(&token_hash).await;
149 return Ok(true);
150 }
151
152 Ok(false)
153 }
154
155 async fn delete_cached_token_hash(&self, token_hash: &str) {
156 if let Some(cache) = &self.session_cache {
157 if let Err(error) = cache.delete(token_hash).await {
158 tracing::warn!(error = ?error, "failed to delete auth session cache");
159 }
160 }
161 }
162}
163
164#[async_trait::async_trait]
165impl AuthUserRepository for PostgresAuthUserRepository {
166 async fn insert(&self, user: &AuthUser) -> AppResult<()> {
167 sqlx::query(
168 r#"
169 insert into auth.users (
170 id,
171 created_at,
172 disabled_at,
173 disabled_reason,
174 disabled_until
175 )
176 values ($1, $2, $3, $4, $5)
177 "#,
178 )
179 .bind(&user.id.0)
180 .bind(user.created_at)
181 .bind(user.disabled_at)
182 .bind(user.disabled_reason.as_deref())
183 .bind(user.disabled_until)
184 .execute(&self.pool)
185 .await
186 .map(|_| ())
187 .map_err(map_sql_error)
188 }
189
190 async fn find_by_id(&self, user_id: &AuthUserId) -> AppResult<Option<AuthUser>> {
191 sqlx::query_as::<_, UserRow>(
192 r#"
193 select
194 id,
195 created_at,
196 case when disabled_until <= now() then null else disabled_at end,
197 case when disabled_until <= now() then null else disabled_reason end,
198 case when disabled_until <= now() then null else disabled_until end
199 from auth.users
200 where id = $1
201 "#,
202 )
203 .bind(&user_id.0)
204 .fetch_optional(&self.pool)
205 .await
206 .map(|row| row.map(user_from_row))
207 .map_err(map_sql_error)
208 }
209
210 async fn list(&self, limit: i64, cursor: Option<&str>) -> AppResult<Vec<AuthUser>> {
211 let rows = match cursor {
212 Some(after) => {
213 sqlx::query_as::<_, UserRow>(
214 r#"
215 select
216 id,
217 created_at,
218 case when disabled_until <= now() then null else disabled_at end,
219 case when disabled_until <= now() then null else disabled_reason end,
220 case when disabled_until <= now() then null else disabled_until end
221 from auth.users
222 where id > $1
223 order by id asc
224 limit $2
225 "#,
226 )
227 .bind(after)
228 .bind(limit)
229 .fetch_all(&self.pool)
230 .await
231 }
232 None => {
233 sqlx::query_as::<_, UserRow>(
234 r#"
235 select
236 id,
237 created_at,
238 case when disabled_until <= now() then null else disabled_at end,
239 case when disabled_until <= now() then null else disabled_reason end,
240 case when disabled_until <= now() then null else disabled_until end
241 from auth.users
242 order by id asc
243 limit $1
244 "#,
245 )
246 .bind(limit)
247 .fetch_all(&self.pool)
248 .await
249 }
250 }
251 .map_err(map_sql_error)?;
252
253 Ok(rows.into_iter().map(user_from_row).collect())
254 }
255
256 async fn find_session_by_id(&self, session_id: &str) -> AppResult<Option<AuthSessionRecord>> {
257 sqlx::query_as::<_, SessionRow>(
258 r#"
259 select id, user_id, device_id, created_at, expires_at, revoked_at
260 from auth.sessions
261 where id = $1
262 "#,
263 )
264 .bind(session_id)
265 .fetch_optional(&self.pool)
266 .await
267 .map(|row| row.map(session_from_row))
268 .map_err(map_sql_error)
269 }
270
271 async fn list_sessions(
272 &self,
273 limit: i64,
274 cursor: Option<&str>,
275 ) -> AppResult<Vec<AuthSessionRecord>> {
276 let rows = match cursor {
277 Some(after) => {
278 sqlx::query_as::<_, SessionRow>(
279 r#"
280 select id, user_id, device_id, created_at, expires_at, revoked_at
281 from auth.sessions
282 where id > $1
283 order by id asc
284 limit $2
285 "#,
286 )
287 .bind(after)
288 .bind(limit)
289 .fetch_all(&self.pool)
290 .await
291 }
292 None => {
293 sqlx::query_as::<_, SessionRow>(
294 r#"
295 select id, user_id, device_id, created_at, expires_at, revoked_at
296 from auth.sessions
297 order by id asc
298 limit $1
299 "#,
300 )
301 .bind(limit)
302 .fetch_all(&self.pool)
303 .await
304 }
305 }
306 .map_err(map_sql_error)?;
307
308 Ok(rows.into_iter().map(session_from_row).collect())
309 }
310
311 async fn revoke_session_by_id(
312 &self,
313 session_id: &str,
314 revoked_at: DateTime<Utc>,
315 ) -> AppResult<bool> {
316 let revoked_token_hash = sqlx::query_scalar::<_, String>(
317 r#"
318 update auth.sessions
319 set revoked_at = $2
320 where id = $1
321 and revoked_at is null
322 returning token_hash
323 "#,
324 )
325 .bind(session_id)
326 .bind(revoked_at)
327 .fetch_optional(&self.pool)
328 .await
329 .map_err(map_sql_error)?;
330
331 if let Some(token_hash) = revoked_token_hash {
332 self.delete_cached_token_hash(&token_hash).await;
333 return Ok(true);
334 }
335
336 Ok(false)
337 }
338
339 async fn set_user_disabled_at(
340 &self,
341 user_id: &AuthUserId,
342 disabled_at: Option<DateTime<Utc>>,
343 disabled_reason: Option<&str>,
344 disabled_until: Option<DateTime<Utc>>,
345 ) -> AppResult<bool> {
346 let result = sqlx::query(
347 r#"
348 update auth.users
349 set disabled_at = $2,
350 disabled_reason = $3,
351 disabled_until = $4
352 where id = $1
353 "#,
354 )
355 .bind(&user_id.0)
356 .bind(disabled_at)
357 .bind(disabled_reason)
358 .bind(disabled_until)
359 .execute(&self.pool)
360 .await
361 .map_err(map_sql_error)?;
362
363 let changed = result.rows_affected() > 0;
364 if changed && disabled_at.is_some() {
365 let token_hashes = sqlx::query_scalar::<_, String>(
366 r#"
367 select token_hash
368 from auth.sessions
369 where user_id = $1
370 and revoked_at is null
371 and expires_at > now()
372 "#,
373 )
374 .bind(&user_id.0)
375 .fetch_all(&self.pool)
376 .await
377 .map_err(map_sql_error)?;
378
379 for token_hash in token_hashes {
380 self.delete_cached_token_hash(&token_hash).await;
381 }
382 }
383
384 Ok(changed)
385 }
386}
387
388type UserRow = (
389 String,
390 DateTime<Utc>,
391 Option<DateTime<Utc>>,
392 Option<String>,
393 Option<DateTime<Utc>>,
394);
395type SessionRow = (
396 String,
397 String,
398 Option<String>,
399 DateTime<Utc>,
400 DateTime<Utc>,
401 Option<DateTime<Utc>>,
402);
403
404fn user_from_row(row: UserRow) -> AuthUser {
405 let (id, created_at, disabled_at, disabled_reason, disabled_until) = row;
406 AuthUser {
407 id: AuthUserId(id),
408 created_at,
409 disabled_at,
410 disabled_reason,
411 disabled_until,
412 }
413}
414
415fn session_from_row(row: SessionRow) -> AuthSessionRecord {
416 let (id, user_id, device_id, created_at, expires_at, revoked_at) = row;
417 AuthSessionRecord {
418 id,
419 user_id: AuthUserId(user_id),
420 device_id,
421 created_at,
422 expires_at,
423 revoked_at,
424 }
425}
426
427fn map_sql_error(source: sqlx::Error) -> AppError {
428 AppError::new(ErrorCode::Internal, "Internal server error").with_source(source)
429}