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