use std::sync::Arc;
use std::time::Duration;
use issuerd_cluster::cache_keys;
use issuerd_core::{
Client, ClientId, ClientIdentifier, ClientScope, ClientScopeId, DistributedCache, Group,
Pagination, RealmId, Role, RoleId, Storage, User, UserId,
};
use tokio::sync::OnceCell;
use tracing::{debug, warn};
use crate::state::ServerState;
#[derive(Debug, Clone)]
pub struct UserClaims {
pub user: User,
pub groups: Vec<Group>,
pub realm_role_ids: Vec<RoleId>,
pub client_role_ids: Vec<RoleId>,
}
#[derive(Debug, Default, Clone)]
pub struct RealmCatalog {
pub roles: Vec<Role>,
pub scopes: Vec<ClientScope>,
}
#[derive(Debug, serde::Serialize, serde::Deserialize)]
struct UserClaimsEntry {
user: User,
groups: Vec<Group>,
realm_role_ids: Vec<RoleId>,
client_role_ids: Vec<RoleId>,
epoch: u64,
}
#[derive(Debug, serde::Serialize, serde::Deserialize)]
struct RealmCatalogEntry {
roles: Vec<Role>,
scopes: Vec<ClientScope>,
epoch: u64,
}
#[derive(Debug, serde::Serialize, serde::Deserialize)]
struct ClientEntry {
client: Client,
epoch: u64,
}
#[derive(Debug, serde::Serialize, serde::Deserialize)]
struct ClientScopesEntry {
default_scope_ids: Vec<ClientScopeId>,
epoch: u64,
}
#[derive(Debug, serde::Serialize, serde::Deserialize)]
struct ClientUuidEntry {
identifier: String,
epoch: u64,
}
pub struct ClaimsReader<'a> {
state: &'a Arc<ServerState>,
realm_id: RealmId,
epoch: OnceCell<Option<u64>>,
}
impl<'a> ClaimsReader<'a> {
pub fn new(state: &'a Arc<ServerState>, realm_id: &RealmId) -> Self {
Self {
state,
realm_id: realm_id.clone(),
epoch: OnceCell::new(),
}
}
fn ttl(&self) -> u64 {
self.state.config.cache.read_cache_ttl_secs
}
fn storage(&self) -> &dyn Storage {
self.state.storage.as_ref()
}
fn cache(&self) -> &Arc<dyn DistributedCache> {
&self.state.cache
}
async fn epoch(&self) -> Option<u64> {
*self
.epoch
.get_or_init(|| async {
let key = cache_keys::claims_epoch(self.realm_id.as_ref());
match self.state.cache.get(&key).await {
Ok(Some(bytes)) => match std::str::from_utf8(&bytes).ok()?.parse::<u64>() {
Ok(v) => Some(v),
Err(_) => {
warn!(realm = %self.realm_id, "claims epoch unparsable; treating as 0");
Some(0)
}
},
Ok(None) => Some(0),
Err(e) => {
warn!(realm = %self.realm_id, error = %e, "claims epoch read failed; falling back to storage");
None
}
}
})
.await
}
async fn read_entry<T: serde::de::DeserializeOwned>(
&self,
key: &str,
epoch: u64,
entry_epoch: fn(&T) -> u64,
) -> Option<T> {
match self.cache().get(key).await {
Ok(Some(bytes)) => match serde_json::from_slice::<T>(&bytes) {
Ok(entry) if entry_epoch(&entry) == epoch => {
debug!(realm = %self.realm_id, "claims cache: hit");
Some(entry)
}
Ok(_) => {
debug!(realm = %self.realm_id, "claims cache: stale epoch; re-reading");
None
}
Err(_) => {
debug!(realm = %self.realm_id, "claims cache: malformed entry; re-reading");
None
}
},
Ok(None) => None,
Err(e) => {
warn!(realm = %self.realm_id, error = %e, "claims cache read failed; falling back to storage");
None
}
}
}
async fn write_entry<T: serde::Serialize>(&self, key: &str, entry: &T) {
match serde_json::to_vec(entry) {
Ok(bytes) => {
if let Err(e) =
self.cache().set(key, bytes, Some(Duration::from_secs(self.ttl()))).await
{
warn!(realm = %self.realm_id, error = %e, "claims cache write failed");
}
}
Err(e) => {
warn!(realm = %self.realm_id, error = %e, "claims cache serialization failed");
}
}
}
pub async fn user_claims(&self, user_id: &UserId) -> Option<UserClaims> {
if self.ttl() == 0 {
return load_user_claims(self.storage(), &self.realm_id, user_id).await.0;
}
let Some(epoch) = self.epoch().await else {
return load_user_claims(self.storage(), &self.realm_id, user_id).await.0;
};
let key = cache_keys::user_claims(self.realm_id.as_ref(), user_id.as_ref());
if let Some(entry) = self.read_entry::<UserClaimsEntry>(&key, epoch, |e| e.epoch).await {
return Some(UserClaims {
user: entry.user,
groups: entry.groups,
realm_role_ids: entry.realm_role_ids,
client_role_ids: entry.client_role_ids,
});
}
let (bundle, degraded) = load_user_claims(self.storage(), &self.realm_id, user_id).await;
if let Some(bundle) = &bundle {
if !degraded {
let entry = UserClaimsEntry {
user: bundle.user.clone(),
groups: bundle.groups.clone(),
realm_role_ids: bundle.realm_role_ids.clone(),
client_role_ids: bundle.client_role_ids.clone(),
epoch,
};
self.write_entry(&key, &entry).await;
}
}
bundle
}
pub async fn realm_catalog(&self) -> RealmCatalog {
if self.ttl() == 0 {
return load_realm_catalog(self.storage(), &self.realm_id).await.0;
}
let Some(epoch) = self.epoch().await else {
return load_realm_catalog(self.storage(), &self.realm_id).await.0;
};
let key = cache_keys::realm_catalog(self.realm_id.as_ref());
if let Some(entry) = self.read_entry::<RealmCatalogEntry>(&key, epoch, |e| e.epoch).await {
return RealmCatalog {
roles: entry.roles,
scopes: entry.scopes,
};
}
let (catalog, degraded) = load_realm_catalog(self.storage(), &self.realm_id).await;
if !degraded {
let entry = RealmCatalogEntry {
roles: catalog.roles.clone(),
scopes: catalog.scopes.clone(),
epoch,
};
self.write_entry(&key, &entry).await;
}
catalog
}
pub async fn client(&self, client_id: &ClientIdentifier) -> Option<Client> {
if self.ttl() == 0 {
return load_client(self.storage(), &self.realm_id, client_id).await.0;
}
let Some(epoch) = self.epoch().await else {
return load_client(self.storage(), &self.realm_id, client_id).await.0;
};
let key = cache_keys::client(self.realm_id.as_ref(), client_id.as_ref());
if let Some(entry) = self.read_entry::<ClientEntry>(&key, epoch, |e| e.epoch).await {
return Some(entry.client);
}
let (client, degraded) = load_client(self.storage(), &self.realm_id, client_id).await;
if let Some(client) = &client {
if !degraded {
self.write_entry(
&key,
&ClientEntry {
client: client.clone(),
epoch,
},
)
.await;
}
}
client
}
pub async fn client_default_scope_ids(&self, client_uuid: &ClientId) -> Vec<ClientScopeId> {
if self.ttl() == 0 {
return load_default_scope_ids(self.storage(), &self.realm_id, client_uuid).await.0;
}
let Some(epoch) = self.epoch().await else {
return load_default_scope_ids(self.storage(), &self.realm_id, client_uuid).await.0;
};
let key = cache_keys::client_scopes(self.realm_id.as_ref(), client_uuid.as_ref());
if let Some(entry) = self.read_entry::<ClientScopesEntry>(&key, epoch, |e| e.epoch).await {
return entry.default_scope_ids;
}
let (ids, degraded) =
load_default_scope_ids(self.storage(), &self.realm_id, client_uuid).await;
if !degraded {
self.write_entry(
&key,
&ClientScopesEntry {
default_scope_ids: ids.clone(),
epoch,
},
)
.await;
}
ids
}
pub async fn client_identifier_by_uuid(&self, client_uuid: &ClientId) -> Option<String> {
if self.ttl() == 0 {
return load_client_identifier(self.storage(), &self.realm_id, client_uuid).await.0;
}
let Some(epoch) = self.epoch().await else {
return load_client_identifier(self.storage(), &self.realm_id, client_uuid).await.0;
};
let key = cache_keys::client_uuid(self.realm_id.as_ref(), client_uuid.as_ref());
if let Some(entry) = self.read_entry::<ClientUuidEntry>(&key, epoch, |e| e.epoch).await {
return Some(entry.identifier);
}
let (identifier, degraded) =
load_client_identifier(self.storage(), &self.realm_id, client_uuid).await;
if let Some(identifier) = &identifier {
if !degraded {
self.write_entry(
&key,
&ClientUuidEntry {
identifier: identifier.clone(),
epoch,
},
)
.await;
}
}
identifier
}
}
async fn load_user_claims(
storage: &dyn Storage,
realm_id: &RealmId,
user_id: &UserId,
) -> (Option<UserClaims>, bool) {
let user = match storage.get_user(realm_id, user_id).await {
Ok(Some(user)) => user,
Ok(None) => return (None, false),
Err(e) => {
debug!(realm = %realm_id, error = %e, "user read failed");
return (None, true);
}
};
let mut degraded = false;
let groups = match storage.list_user_groups(realm_id, user_id).await {
Ok(ids) => match storage.get_groups_batch(realm_id, &ids).await {
Ok(groups) => groups,
Err(e) => {
debug!(realm = %realm_id, error = %e, "group batch read failed");
degraded = true;
Vec::new()
}
},
Err(e) => {
debug!(realm = %realm_id, error = %e, "group membership read failed");
degraded = true;
Vec::new()
}
};
let realm_role_ids = match storage.list_user_realm_roles(realm_id, user_id).await {
Ok(ids) => ids,
Err(e) => {
debug!(realm = %realm_id, error = %e, "realm-role mapping read failed");
degraded = true;
Vec::new()
}
};
let client_role_ids = match storage.list_user_client_roles(realm_id, user_id).await {
Ok(ids) => ids,
Err(e) => {
debug!(realm = %realm_id, error = %e, "client-role mapping read failed");
degraded = true;
Vec::new()
}
};
(
Some(UserClaims {
user,
groups,
realm_role_ids,
client_role_ids,
}),
degraded,
)
}
async fn load_realm_catalog(storage: &dyn Storage, realm_id: &RealmId) -> (RealmCatalog, bool) {
let mut degraded = false;
let roles = match storage
.list_roles(
realm_id,
&Pagination {
first: 0,
max: i32::MAX,
},
)
.await
{
Ok(roles) => roles,
Err(e) => {
debug!(realm = %realm_id, error = %e, "role catalog read failed");
degraded = true;
Vec::new()
}
};
let scopes = match storage
.list_client_scopes(
realm_id,
&Pagination {
first: 0,
max: i32::MAX,
},
)
.await
{
Ok(scopes) => scopes,
Err(e) => {
debug!(realm = %realm_id, error = %e, "scope catalog read failed");
degraded = true;
Vec::new()
}
};
(RealmCatalog { roles, scopes }, degraded)
}
async fn load_client(
storage: &dyn Storage,
realm_id: &RealmId,
client_id: &ClientIdentifier,
) -> (Option<Client>, bool) {
match storage.get_client_by_client_id(realm_id, client_id).await {
Ok(Some(client)) => (Some(client), false),
Ok(None) => (None, false),
Err(e) => {
debug!(realm = %realm_id, error = %e, "client read failed");
(None, true)
}
}
}
async fn load_default_scope_ids(
storage: &dyn Storage,
realm_id: &RealmId,
client_uuid: &ClientId,
) -> (Vec<ClientScopeId>, bool) {
match storage.list_client_scope_assignments(realm_id, client_uuid).await {
Ok(assignments) => (
assignments
.into_iter()
.filter(|(_, is_default)| *is_default)
.map(|(scope_id, _)| scope_id)
.collect(),
false,
),
Err(e) => {
debug!(realm = %realm_id, error = %e, "scope assignment read failed");
(Vec::new(), true)
}
}
}
async fn load_client_identifier(
storage: &dyn Storage,
realm_id: &RealmId,
client_uuid: &ClientId,
) -> (Option<String>, bool) {
match storage.get_client(realm_id, client_uuid).await {
Ok(Some(client)) => (Some(client.client_id.to_string()), false),
Ok(None) => (None, false),
Err(e) => {
debug!(realm = %realm_id, error = %e, "client uuid read failed");
(None, true)
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use issuerd_core::{ClientAuthenticatorType, ClientProtocol, IssuerdError, Scope, Username};
use std::collections::HashMap;
async fn setup() -> (Arc<ServerState>, RealmId) {
let state = Arc::new(
ServerState::from_config(&crate::config::ServerConfig::default()).await.unwrap(),
);
(state, RealmId::new("master").unwrap())
}
fn reader<'a>(state: &'a Arc<ServerState>, realm_id: &RealmId) -> ClaimsReader<'a> {
ClaimsReader::new(state, realm_id)
}
fn user(user_id: &UserId, realm_id: &RealmId) -> User {
User {
id: user_id.clone(),
realm_id: realm_id.clone(),
username: Username::new("bob").unwrap(),
email: None,
email_verified: false,
first_name: None,
last_name: None,
enabled: true,
federation_link: None,
attributes: HashMap::new(),
required_actions: vec![],
created_at: chrono::Utc::now(),
updated_at: chrono::Utc::now(),
}
}
fn client(uuid: &str, realm_id: &RealmId, identifier: &str) -> Client {
Client {
id: ClientId::new(uuid).unwrap(),
realm_id: realm_id.clone(),
client_id: ClientIdentifier::new(identifier).unwrap(),
name: None,
description: None,
enabled: true,
protocol: ClientProtocol::OpenIdConnect,
public_client: true,
bearer_only: false,
client_authenticator_type: ClientAuthenticatorType::ClientSecret,
secret: None,
redirect_uris: vec![],
web_origins: vec![],
default_scopes: Scope::parse("openid"),
optional_scopes: Scope::empty(),
consent_required: false,
full_scope_allowed: true,
service_accounts_enabled: false,
protocol_mappers: Vec::new(),
scope_mappings: Default::default(),
attributes: HashMap::new(),
}
}
#[tokio::test]
async fn user_claims_caches_and_invalidation_reflects() {
let (state, realm_id) = setup().await;
let user_id = UserId::new("bob").unwrap();
state.storage.create_user(&realm_id, &user(&user_id, &realm_id)).await.unwrap();
let bundle = reader(&state, &realm_id).user_claims(&user_id).await.expect("bundle");
assert_eq!(bundle.user.username.as_str(), "bob");
let key = cache_keys::user_claims("master", "bob");
assert!(state.cache.get(&key).await.unwrap().is_some(), "bundle cached");
let mut renamed = user(&user_id, &realm_id);
renamed.username = Username::new("robert").unwrap();
state.storage.update_user(&realm_id, &renamed).await.unwrap();
let bundle = reader(&state, &realm_id).user_claims(&user_id).await.unwrap();
assert_eq!(bundle.user.username.as_str(), "bob", "stale entry served within TTL");
issuerd_cluster::invalidate::invalidate_user_claims(
state.cache.as_ref(),
&realm_id,
&user_id,
)
.await;
let bundle = reader(&state, &realm_id).user_claims(&user_id).await.unwrap();
assert_eq!(bundle.user.username.as_str(), "robert");
}
#[tokio::test]
async fn epoch_bump_invalidates_every_entry_kind() {
let (state, realm_id) = setup().await;
let user_id = UserId::new("bob").unwrap();
state.storage.create_user(&realm_id, &user(&user_id, &realm_id)).await.unwrap();
let uuid = ClientId::new("uuid-1").unwrap();
state
.storage
.create_client(&realm_id, &client("uuid-1", &realm_id, "my-app"))
.await
.unwrap();
let r = reader(&state, &realm_id);
assert!(r.user_claims(&user_id).await.is_some());
let _ = r.realm_catalog().await;
assert!(r.client(&ClientIdentifier::new("my-app").unwrap()).await.is_some());
let _ = r.client_default_scope_ids(&uuid).await;
assert_eq!(r.client_identifier_by_uuid(&uuid).await.as_deref(), Some("my-app"));
issuerd_cluster::invalidate::bump_claims_epoch(state.cache.as_ref(), &realm_id).await;
let r = reader(&state, &realm_id);
assert!(r.user_claims(&user_id).await.is_some());
let key = cache_keys::user_claims("master", "bob");
let raw = state.cache.get(&key).await.unwrap().unwrap();
let entry: serde_json::Value = serde_json::from_slice(&raw).unwrap();
assert_eq!(entry["epoch"], 1, "entry rewritten at the new epoch");
}
#[tokio::test]
async fn disabled_cache_is_pure_db() {
let mut cfg = crate::config::ServerConfig::default();
cfg.cache.read_cache_ttl_secs = 0;
let state = Arc::new(ServerState::from_config(&cfg).await.unwrap());
let realm_id = RealmId::new("master").unwrap();
let user_id = UserId::new("bob").unwrap();
state.storage.create_user(&realm_id, &user(&user_id, &realm_id)).await.unwrap();
assert!(reader(&state, &realm_id).user_claims(&user_id).await.is_some());
let key = cache_keys::user_claims("master", "bob");
assert!(state.cache.get(&key).await.unwrap().is_none(), "nothing cached");
let mut renamed = user(&user_id, &realm_id);
renamed.username = Username::new("robert").unwrap();
state.storage.update_user(&realm_id, &renamed).await.unwrap();
let bundle = reader(&state, &realm_id).user_claims(&user_id).await.unwrap();
assert_eq!(bundle.user.username.as_str(), "robert");
}
#[tokio::test]
async fn cache_outage_falls_back_to_storage() {
#[derive(Debug)]
struct DeadCache;
#[async_trait::async_trait]
impl DistributedCache for DeadCache {
async fn get(&self, _: &str) -> Result<Option<Vec<u8>>, IssuerdError> {
Err(IssuerdError::ServerError("cache down".into()))
}
async fn set(
&self,
_: &str,
_: Vec<u8>,
_: Option<Duration>,
) -> Result<(), IssuerdError> {
Err(IssuerdError::ServerError("cache down".into()))
}
async fn delete(&self, _: &str) -> Result<(), IssuerdError> {
Err(IssuerdError::ServerError("cache down".into()))
}
async fn compare_and_swap(
&self,
_: &str,
_: Option<Vec<u8>>,
_: Vec<u8>,
) -> Result<bool, IssuerdError> {
Err(IssuerdError::ServerError("cache down".into()))
}
async fn publish(&self, _: &str, _: Vec<u8>) -> Result<(), IssuerdError> {
Err(IssuerdError::ServerError("cache down".into()))
}
async fn subscribe(
&self,
_: &str,
_: Box<dyn Fn(Vec<u8>) + Send + Sync>,
) -> Result<(), IssuerdError> {
Err(IssuerdError::ServerError("cache down".into()))
}
}
let cfg = crate::config::ServerConfig::default();
let storage: Arc<dyn Storage> = Arc::new(issuerd_storage::InMemoryStorage::new());
let state = Arc::new(
ServerState::from_components(&cfg, storage, Arc::new(DeadCache)).await.unwrap(),
);
let realm_id = RealmId::new("master").unwrap();
let user_id = UserId::new("admin").unwrap();
state.storage.create_user(&realm_id, &user(&user_id, &realm_id)).await.unwrap();
let role = Role {
id: RoleId::new("role-1").unwrap(),
name: issuerd_core::RoleName::new("admin").unwrap(),
description: None,
realm_id: realm_id.clone(),
client_role: false,
client_id: None,
composite: false,
composites: vec![],
attributes: HashMap::new(),
};
state.storage.create_role(&realm_id, &role).await.unwrap();
let bundle = reader(&state, &realm_id).user_claims(&user_id).await;
assert!(bundle.is_some(), "stored user resolves despite cache outage");
let catalog = reader(&state, &realm_id).realm_catalog().await;
assert!(!catalog.roles.is_empty(), "stored roles resolve despite cache outage");
}
}