use chrono::Utc;
use sqlx::Acquire;
use systemprompt_identifiers::UserId;
use systemprompt_traits::FederatedIdentityClaims;
use crate::error::Result;
use crate::models::{User, UserRole, UserStatus};
use crate::repository::UserRepository;
impl UserRepository {
pub async fn find_federated(&self, issuer: &str, external_sub: &str) -> Result<Option<UserId>> {
let row = sqlx::query!(
"SELECT user_id FROM federated_identities WHERE issuer = $1 AND external_sub = $2",
issuer,
external_sub
)
.fetch_optional(&*self.pool)
.await?;
Ok(row.map(|r| UserId::new(r.user_id)))
}
pub async fn find_or_create_federated(
&self,
issuer: &str,
external_sub: &str,
claims: &FederatedIdentityClaims,
) -> Result<User> {
let mut conn = self.write_pool.acquire().await?;
let mut tx = conn.begin().await?;
if let Some(existing) = sqlx::query!(
"UPDATE federated_identities SET last_seen_at = CURRENT_TIMESTAMP WHERE issuer = $1 \
AND external_sub = $2 RETURNING user_id",
issuer,
external_sub
)
.fetch_optional(&mut *tx)
.await?
{
let user = sqlx::query_as!(
User,
r#"
SELECT id, name, email, full_name, display_name, status,
email_verified, roles, avatar_url, is_bot, is_scanner,
created_at, updated_at
FROM users WHERE id = $1
"#,
existing.user_id
)
.fetch_one(&mut *tx)
.await?;
tx.commit().await?;
return Ok(user);
}
let now = Utc::now();
let id = UserId::new(uuid::Uuid::new_v4().to_string());
let name = claims
.preferred_username
.clone()
.or_else(|| claims.name.clone())
.unwrap_or_else(|| format!("fed_{}_{}", short_hash(issuer), short_hash(external_sub)));
let synthetic_email = || {
format!(
"{}@{}.federated.local",
short_hash(external_sub),
short_host(issuer)
)
};
let email = match (claims.email.as_deref(), claims.email_verified) {
(Some(addr), true) => addr.to_owned(),
(Some(addr), false) => {
tracing::warn!(
issuer,
external_sub,
upstream_email = addr,
"upstream IdP did not assert email_verified; using synthetic local email to \
prevent account-claim attacks"
);
synthetic_email()
},
(None, _) => synthetic_email(),
};
let display_name = claims.name.clone();
let status = UserStatus::Active.as_str();
let roles = normalised_roles(&claims.roles);
let user = sqlx::query_as!(
User,
r#"
INSERT INTO users (
id, name, email, full_name, display_name,
status, email_verified, roles, is_bot,
created_at, updated_at
)
VALUES ($1, $2, $3, $4, $5, $6, false, $7::TEXT[], false, $8, $8)
RETURNING id, name, email, full_name, display_name, status, email_verified,
roles, avatar_url, is_bot, is_scanner, created_at, updated_at
"#,
id.as_str(),
name,
email,
display_name.as_deref(),
display_name.as_deref(),
status,
&roles,
now,
)
.fetch_one(&mut *tx)
.await?;
sqlx::query!(
"INSERT INTO federated_identities (issuer, external_sub, user_id) VALUES ($1, $2, $3)",
issuer,
external_sub,
user.id.as_str()
)
.execute(&mut *tx)
.await?;
tx.commit().await?;
Ok(user)
}
}
fn normalised_roles(claim_roles: &[String]) -> Vec<String> {
if claim_roles.is_empty() {
vec![UserRole::User.as_str().to_owned()]
} else {
claim_roles.to_vec()
}
}
fn short_hash(s: &str) -> String {
use sha2::{Digest, Sha256};
let digest = Sha256::digest(s.as_bytes());
hex::encode(&digest[..6])
}
fn short_host(issuer: &str) -> String {
issuer
.trim_start_matches("https://")
.trim_start_matches("http://")
.split('/')
.next()
.unwrap_or("issuer")
.replace(['.', ':'], "-")
}