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