use crate::models::{AuthSession, AuthSessionRecord, AuthUser, AuthUserId};
use crate::resolver::{SessionCache, session_token_hash};
use chrono::{DateTime, Utc};
use platform_core::{AppContext, AppError, AppResult, DbPool, ErrorCode};
use std::sync::Arc;
#[async_trait::async_trait]
pub trait AuthUserRepository: std::fmt::Debug + Send + Sync {
async fn insert(&self, user: &AuthUser) -> AppResult<()>;
async fn find_by_id(&self, user_id: &AuthUserId) -> AppResult<Option<AuthUser>>;
async fn list(&self, limit: i64, cursor: Option<&str>) -> AppResult<Vec<AuthUser>>;
async fn find_session_by_id(&self, session_id: &str) -> AppResult<Option<AuthSessionRecord>>;
async fn list_sessions(
&self,
limit: i64,
cursor: Option<&str>,
) -> AppResult<Vec<AuthSessionRecord>>;
async fn revoke_session_by_id(
&self,
session_id: &str,
revoked_at: DateTime<Utc>,
) -> AppResult<bool>;
async fn set_user_disabled_at(
&self,
user_id: &AuthUserId,
disabled_at: Option<DateTime<Utc>>,
disabled_reason: Option<&str>,
disabled_until: Option<DateTime<Utc>>,
) -> AppResult<bool>;
}
#[derive(Debug, Clone)]
pub struct PostgresAuthUserRepository {
pool: DbPool,
session_cache: Option<Arc<dyn SessionCache>>,
}
impl PostgresAuthUserRepository {
#[must_use]
pub fn new(pool: DbPool) -> Self {
Self {
pool,
session_cache: None,
}
}
#[must_use]
pub fn new_with_session_cache(
pool: DbPool,
session_cache: Option<Arc<dyn SessionCache>>,
) -> Self {
Self {
pool,
session_cache,
}
}
#[must_use]
pub fn from_context(ctx: &AppContext) -> Self {
#[cfg(feature = "redis")]
{
return Self::new_with_session_cache(
ctx.db.clone(),
crate::redis_cache::session_cache_from_context(ctx),
);
}
#[cfg(not(feature = "redis"))]
Self::new(ctx.db.clone())
}
pub async fn create_dev_session(
&self,
user_id: AuthUserId,
session_id: String,
token: String,
created_at: DateTime<Utc>,
expires_at: DateTime<Utc>,
) -> AppResult<AuthSession> {
let mut tx = self.pool.begin().await.map_err(map_sql_error)?;
sqlx::query(
r#"
insert into auth.users (
id,
is_anonymous,
created_at,
disabled_at,
disabled_reason,
disabled_until
)
values ($1, false, $2, null, null, null)
on conflict (id) do nothing
"#,
)
.bind(&user_id.0)
.bind(created_at)
.execute(&mut *tx)
.await
.map_err(map_sql_error)?;
let active_user_exists = sqlx::query_scalar::<_, bool>(
r#"
select exists(
select 1
from auth.users
where id = $1
and (disabled_at is null or disabled_until <= now())
)
"#,
)
.bind(&user_id.0)
.fetch_one(&mut *tx)
.await
.map_err(map_sql_error)?;
if !active_user_exists {
return Err(AppError::new(ErrorCode::Forbidden, "Auth user is disabled"));
}
sqlx::query(
r#"
insert into auth.sessions (
id,
user_id,
token_hash,
device_id,
client_ip,
user_agent,
created_at,
expires_at,
revoked_at
)
values ($1, $2, $3, null, null, null, $4, $5, null)
"#,
)
.bind(&session_id)
.bind(&user_id.0)
.bind(session_token_hash(&token))
.bind(created_at)
.bind(expires_at)
.execute(&mut *tx)
.await
.map_err(map_sql_error)?;
tx.commit().await.map_err(map_sql_error)?;
Ok(AuthSession {
id: session_id,
user_id,
token,
device_id: None,
expires_at,
})
}
pub async fn revoke_session_token(
&self,
token: &str,
revoked_at: DateTime<Utc>,
) -> AppResult<bool> {
let token_hash = session_token_hash(token);
let revoked_token_hash = sqlx::query_scalar::<_, String>(
r#"
update auth.sessions
set revoked_at = $2
where token_hash = $1
and revoked_at is null
returning token_hash
"#,
)
.bind(&token_hash)
.bind(revoked_at)
.fetch_optional(&self.pool)
.await
.map_err(map_sql_error)?;
if revoked_token_hash.is_some() {
self.delete_cached_token_hash(&token_hash).await;
return Ok(true);
}
Ok(false)
}
async fn delete_cached_token_hash(&self, token_hash: &str) {
if let Some(cache) = &self.session_cache {
if let Err(error) = cache.delete(token_hash).await {
tracing::warn!(error = ?error, "failed to delete auth session cache");
}
}
}
}
#[async_trait::async_trait]
impl AuthUserRepository for PostgresAuthUserRepository {
async fn insert(&self, user: &AuthUser) -> AppResult<()> {
sqlx::query(
r#"
insert into auth.users (
id,
is_anonymous,
created_at,
disabled_at,
disabled_reason,
disabled_until
)
values ($1, $2, $3, $4, $5, $6)
"#,
)
.bind(&user.id.0)
.bind(user.is_anonymous)
.bind(user.created_at)
.bind(user.disabled_at)
.bind(user.disabled_reason.as_deref())
.bind(user.disabled_until)
.execute(&self.pool)
.await
.map(|_| ())
.map_err(map_sql_error)
}
async fn find_by_id(&self, user_id: &AuthUserId) -> AppResult<Option<AuthUser>> {
sqlx::query_as::<_, UserRow>(
r#"
select
id,
is_anonymous,
created_at,
case when disabled_until <= now() then null else disabled_at end,
case when disabled_until <= now() then null else disabled_reason end,
case when disabled_until <= now() then null else disabled_until end
from auth.users
where id = $1
"#,
)
.bind(&user_id.0)
.fetch_optional(&self.pool)
.await
.map(|row| row.map(user_from_row))
.map_err(map_sql_error)
}
async fn list(&self, limit: i64, cursor: Option<&str>) -> AppResult<Vec<AuthUser>> {
let rows = match cursor {
Some(after) => {
sqlx::query_as::<_, UserRow>(
r#"
select
id,
is_anonymous,
created_at,
case when disabled_until <= now() then null else disabled_at end,
case when disabled_until <= now() then null else disabled_reason end,
case when disabled_until <= now() then null else disabled_until end
from auth.users
where id > $1
order by id asc
limit $2
"#,
)
.bind(after)
.bind(limit)
.fetch_all(&self.pool)
.await
}
None => {
sqlx::query_as::<_, UserRow>(
r#"
select
id,
is_anonymous,
created_at,
case when disabled_until <= now() then null else disabled_at end,
case when disabled_until <= now() then null else disabled_reason end,
case when disabled_until <= now() then null else disabled_until end
from auth.users
order by id asc
limit $1
"#,
)
.bind(limit)
.fetch_all(&self.pool)
.await
}
}
.map_err(map_sql_error)?;
Ok(rows.into_iter().map(user_from_row).collect())
}
async fn find_session_by_id(&self, session_id: &str) -> AppResult<Option<AuthSessionRecord>> {
sqlx::query_as::<_, SessionRow>(
r#"
select id, user_id, device_id, client_ip, user_agent, created_at, expires_at, revoked_at
from auth.sessions
where id = $1
"#,
)
.bind(session_id)
.fetch_optional(&self.pool)
.await
.map(|row| row.map(session_from_row))
.map_err(map_sql_error)
}
async fn list_sessions(
&self,
limit: i64,
cursor: Option<&str>,
) -> AppResult<Vec<AuthSessionRecord>> {
let rows = match cursor {
Some(after) => {
sqlx::query_as::<_, SessionRow>(
r#"
select id, user_id, device_id, client_ip, user_agent, created_at, expires_at, revoked_at
from auth.sessions
where id > $1
order by id asc
limit $2
"#,
)
.bind(after)
.bind(limit)
.fetch_all(&self.pool)
.await
}
None => {
sqlx::query_as::<_, SessionRow>(
r#"
select id, user_id, device_id, client_ip, user_agent, created_at, expires_at, revoked_at
from auth.sessions
order by id asc
limit $1
"#,
)
.bind(limit)
.fetch_all(&self.pool)
.await
}
}
.map_err(map_sql_error)?;
Ok(rows.into_iter().map(session_from_row).collect())
}
async fn revoke_session_by_id(
&self,
session_id: &str,
revoked_at: DateTime<Utc>,
) -> AppResult<bool> {
let revoked_token_hash = sqlx::query_scalar::<_, String>(
r#"
update auth.sessions
set revoked_at = $2
where id = $1
and revoked_at is null
returning token_hash
"#,
)
.bind(session_id)
.bind(revoked_at)
.fetch_optional(&self.pool)
.await
.map_err(map_sql_error)?;
if let Some(token_hash) = revoked_token_hash {
self.delete_cached_token_hash(&token_hash).await;
return Ok(true);
}
Ok(false)
}
async fn set_user_disabled_at(
&self,
user_id: &AuthUserId,
disabled_at: Option<DateTime<Utc>>,
disabled_reason: Option<&str>,
disabled_until: Option<DateTime<Utc>>,
) -> AppResult<bool> {
let result = sqlx::query(
r#"
update auth.users
set disabled_at = $2,
disabled_reason = $3,
disabled_until = $4
where id = $1
"#,
)
.bind(&user_id.0)
.bind(disabled_at)
.bind(disabled_reason)
.bind(disabled_until)
.execute(&self.pool)
.await
.map_err(map_sql_error)?;
let changed = result.rows_affected() > 0;
if changed && disabled_at.is_some() {
let token_hashes = sqlx::query_scalar::<_, String>(
r#"
select token_hash
from auth.sessions
where user_id = $1
and revoked_at is null
and expires_at > now()
"#,
)
.bind(&user_id.0)
.fetch_all(&self.pool)
.await
.map_err(map_sql_error)?;
for token_hash in token_hashes {
self.delete_cached_token_hash(&token_hash).await;
}
}
Ok(changed)
}
}
type UserRow = (
String,
bool,
DateTime<Utc>,
Option<DateTime<Utc>>,
Option<String>,
Option<DateTime<Utc>>,
);
type SessionRow = (
String,
String,
Option<String>,
Option<String>,
Option<String>,
DateTime<Utc>,
DateTime<Utc>,
Option<DateTime<Utc>>,
);
fn user_from_row(row: UserRow) -> AuthUser {
let (id, is_anonymous, created_at, disabled_at, disabled_reason, disabled_until) = row;
AuthUser {
id: AuthUserId(id),
is_anonymous,
created_at,
disabled_at,
disabled_reason,
disabled_until,
}
}
fn session_from_row(row: SessionRow) -> AuthSessionRecord {
let (id, user_id, device_id, client_ip, user_agent, created_at, expires_at, revoked_at) = row;
AuthSessionRecord {
id,
user_id: AuthUserId(user_id),
device_id,
client_ip,
user_agent,
created_at,
expires_at,
revoked_at,
}
}
fn map_sql_error(source: sqlx::Error) -> AppError {
AppError::new(ErrorCode::Internal, "Internal server error").with_source(source)
}