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