use async_trait::async_trait;
use sqlx::{Row, postgres::PgPool};
use crate::{
error::{AuthError, Result},
session::{SessionData, SessionStore, TokenPair, generate_refresh_token, hash_token, unix_now},
};
pub const DEFAULT_TOKEN_ISSUER: &str = "fraiseql";
pub const DEFAULT_TOKEN_AUDIENCE: &str = "fraiseql-api";
pub const DEFAULT_TENANT_CLAIM: &str = "tenant_id";
pub struct PostgresSessionStore {
db: PgPool,
hs256_secret: Option<Vec<u8>>,
token_issuer: String,
token_audience: String,
tenant_claim: String,
}
pub const SESSIONS_SCHEMA_SQL: &str = r"
CREATE SCHEMA IF NOT EXISTS _system;
CREATE TABLE IF NOT EXISTS _system.sessions (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
user_id TEXT NOT NULL,
refresh_token_hash TEXT NOT NULL UNIQUE,
issued_at BIGINT NOT NULL,
expires_at BIGINT NOT NULL,
created_at TIMESTAMPTZ DEFAULT NOW(),
revoked_at TIMESTAMPTZ
);
-- The account's tenant when the session was minted (#1450), for revocation and
-- audit. NULL for a platform account or a principal with no account row.
ALTER TABLE _system.sessions ADD COLUMN IF NOT EXISTS tenant_id UUID;
CREATE INDEX IF NOT EXISTS idx_sessions_user_id ON _system.sessions(user_id);
CREATE INDEX IF NOT EXISTS idx_sessions_expires_at ON _system.sessions(expires_at);
CREATE INDEX IF NOT EXISTS idx_sessions_revoked_at ON _system.sessions(revoked_at);
";
impl PostgresSessionStore {
#[must_use]
pub fn new(db: PgPool) -> Self {
Self {
db,
hs256_secret: None,
token_issuer: DEFAULT_TOKEN_ISSUER.to_string(),
token_audience: DEFAULT_TOKEN_AUDIENCE.to_string(),
tenant_claim: DEFAULT_TENANT_CLAIM.to_string(),
}
}
#[must_use]
pub fn with_hs256_secret(db: PgPool, secret: Vec<u8>) -> Self {
Self {
hs256_secret: Some(secret),
..Self::new(db)
}
}
#[must_use]
pub fn with_tenant_claim(mut self, claim: impl Into<String>) -> Self {
self.tenant_claim = claim.into();
self
}
#[must_use]
pub fn with_token_claims(
mut self,
issuer: impl Into<String>,
audience: impl Into<String>,
) -> Self {
self.token_issuer = issuer.into();
self.token_audience = audience.into();
self
}
pub async fn init(&self) -> Result<()> {
sqlx::raw_sql(SESSIONS_SCHEMA_SQL).execute(&self.db).await.map_err(|e| {
AuthError::DatabaseError {
message: format!("Failed to initialize sessions table: {}", e),
}
})?;
Ok(())
}
fn generate_access_token(
&self,
user_id: &str,
tenant: Option<uuid::Uuid>,
expires_in: u64,
) -> Result<String> {
let now = unix_now()?;
let exp = now + expires_in;
let mut claims = crate::Claims {
sub: user_id.to_string(),
iat: now,
exp,
nbf: None,
iss: self.token_issuer.clone(),
aud: vec![self.token_audience.clone()],
extra: std::collections::HashMap::new(),
};
claims
.extra
.insert("jti".to_string(), serde_json::json!(uuid::Uuid::new_v4().to_string()));
if let Some(tenant) = tenant {
claims.extra.insert(
self.tenant_claim.clone(),
serde_json::json!(tenant.as_simple().to_string()),
);
}
match &self.hs256_secret {
Some(secret) => crate::jwt::generate_hs256_token(&claims, secret),
None => Err(AuthError::ConfigError {
message: "JWT signing is not configured for this PostgresSessionStore — \
construct it with with_hs256_secret. Refusing to issue an access \
token that no validator could verify."
.to_string(),
}),
}
}
}
async fn account_for_session(db: &PgPool, user_id: &str) -> Result<Option<uuid::Uuid>> {
let row = sqlx::query("SELECT active, tenant_id FROM core.tb_user WHERE user_id = $1")
.bind(user_id)
.fetch_optional(db)
.await;
let (active, tenant): (Option<bool>, Option<uuid::Uuid>) = match row {
Ok(Some(row)) => (Some(row.get("active")), row.get("tenant_id")),
Ok(None) => (None, None),
Err(sqlx::Error::Database(e)) if e.code().as_deref() == Some("42P01") => {
return Ok(None);
},
Err(e) => {
return Err(AuthError::DatabaseError {
message: format!("Failed to check account activation: {e}"),
});
},
};
if active == Some(false) {
tracing::warn!(
user_id = %user_id,
"refused a session for a deactivated account (SCIM active = false)"
);
return Err(AuthError::AccountDeactivated);
}
Ok(tenant)
}
#[async_trait]
impl SessionStore for PostgresSessionStore {
async fn create_session(&self, user_id: &str, expires_at: u64) -> Result<TokenPair> {
let tenant = account_for_session(&self.db, user_id).await?;
let refresh_token = generate_refresh_token();
let refresh_token_hash = hash_token(&refresh_token);
let now = unix_now()?;
let expires_in = expires_at.saturating_sub(now);
let access_token = self.generate_access_token(user_id, tenant, expires_in)?;
sqlx::query(
r"
INSERT INTO _system.sessions
(user_id, refresh_token_hash, issued_at, expires_at, tenant_id)
VALUES ($1, $2, $3, $4, $5)
",
)
.bind(user_id)
.bind(&refresh_token_hash)
.bind(now.cast_signed())
.bind(expires_at.cast_signed())
.bind(tenant)
.execute(&self.db)
.await
.map_err(|e| {
if e.to_string().contains("duplicate key") {
AuthError::SessionError {
message: "Refresh token already exists".to_string(),
}
} else {
AuthError::DatabaseError {
message: format!("Failed to create session: {}", e),
}
}
})?;
Ok(TokenPair {
access_token,
refresh_token,
expires_in,
})
}
async fn get_session(&self, refresh_token_hash: &str) -> Result<SessionData> {
let row = sqlx::query(
r"
SELECT user_id, issued_at, expires_at, refresh_token_hash
FROM _system.sessions
WHERE refresh_token_hash = $1 AND revoked_at IS NULL
",
)
.bind(refresh_token_hash)
.fetch_optional(&self.db)
.await
.map_err(|e| AuthError::DatabaseError {
message: format!("Failed to get session: {}", e),
})?
.ok_or(AuthError::TokenNotFound)?;
let user_id: String = row.get("user_id");
let issued_at: i64 = row.get("issued_at");
let expires_at: i64 = row.get("expires_at");
let refresh_token_hash: String = row.get("refresh_token_hash");
Ok(SessionData {
user_id,
issued_at: issued_at.cast_unsigned(),
expires_at: expires_at.cast_unsigned(),
refresh_token_hash,
})
}
async fn revoke_session(&self, refresh_token_hash: &str) -> Result<()> {
let result = sqlx::query(
r"
UPDATE _system.sessions
SET revoked_at = NOW()
WHERE refresh_token_hash = $1 AND revoked_at IS NULL
",
)
.bind(refresh_token_hash)
.execute(&self.db)
.await
.map_err(|e| AuthError::DatabaseError {
message: format!("Failed to revoke session: {}", e),
})?;
if result.rows_affected() == 0 {
return Err(AuthError::SessionError {
message: "Session not found or already revoked".to_string(),
});
}
Ok(())
}
async fn revoke_all_sessions(&self, user_id: &str) -> Result<()> {
sqlx::query(
r"
UPDATE _system.sessions
SET revoked_at = NOW()
WHERE user_id = $1 AND revoked_at IS NULL
",
)
.bind(user_id)
.execute(&self.db)
.await
.map_err(|e| AuthError::DatabaseError {
message: format!("Failed to revoke all sessions: {}", e),
})?;
Ok(())
}
}
#[cfg(test)]
mod tests;