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