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