use std::ops::ControlFlow;
use std::sync::Arc;
use auths_core::storage::keychain::IdentityDID;
use auths_verifier::core::{Attestation, ResourceId};
use auths_verifier::types::CanonicalDid;
use thiserror::Error;
use crate::keri::event::Event;
use crate::keri::state::KeyState;
use crate::keri::{Prefix, Said};
use super::org_member::{MemberFilter, MemberStatus, MemberView, OrgMemberEntry};
use super::schemas::{RegistryMetadata, TipInfo};
#[derive(Debug, Clone)]
pub enum AtomicWriteOp {
StoreAttestation(Attestation),
StoreOrgMember { org: String, member: Attestation },
AppendEvent {
prefix: Prefix,
event: Event,
attachment: Vec<u8>,
},
AppendTelEvent {
issuer: Prefix,
registry_said: Said,
credential_said: Said,
sn: u128,
event_bytes: Vec<u8>,
},
StoreCredential {
issuer: Prefix,
credential_said: Said,
credential_bytes: Vec<u8>,
},
}
#[derive(Debug, Clone, Default)]
pub struct AtomicWriteBatch {
ops: Vec<AtomicWriteOp>,
}
impl AtomicWriteBatch {
pub fn new() -> Self {
Self::default()
}
pub fn stage_attestation(&mut self, att: Attestation) -> &mut Self {
self.ops.push(AtomicWriteOp::StoreAttestation(att));
self
}
pub fn stage_org_member(&mut self, org: impl Into<String>, member: Attestation) -> &mut Self {
self.ops.push(AtomicWriteOp::StoreOrgMember {
org: org.into(),
member,
});
self
}
pub fn stage_event(&mut self, prefix: Prefix, event: Event, attachment: Vec<u8>) -> &mut Self {
self.ops.push(AtomicWriteOp::AppendEvent {
prefix,
event,
attachment,
});
self
}
pub fn stage_tel_event(
&mut self,
issuer: Prefix,
registry_said: Said,
credential_said: Said,
sn: u128,
event_bytes: Vec<u8>,
) -> &mut Self {
self.ops.push(AtomicWriteOp::AppendTelEvent {
issuer,
registry_said,
credential_said,
sn,
event_bytes,
});
self
}
pub fn stage_credential(
&mut self,
issuer: Prefix,
credential_said: Said,
credential_bytes: Vec<u8>,
) -> &mut Self {
self.ops.push(AtomicWriteOp::StoreCredential {
issuer,
credential_said,
credential_bytes,
});
self
}
pub fn ops(&self) -> &[AtomicWriteOp] {
&self.ops
}
pub fn is_empty(&self) -> bool {
self.ops.is_empty()
}
}
#[derive(Debug, Error, PartialEq)]
#[non_exhaustive]
pub enum TenantIdError {
#[error("must be 1–64 characters (got {0})")]
InvalidLength(usize),
#[error("contains disallowed character {0:?} (only [a-z0-9_-] allowed)")]
InvalidCharacter(char),
#[error("'{0}' is reserved")]
Reserved(String),
}
impl auths_core::error::AuthsErrorInfo for TenantIdError {
fn error_code(&self) -> &'static str {
match self {
Self::InvalidLength(_) => "AUTHS-E4851",
Self::InvalidCharacter(_) => "AUTHS-E4852",
Self::Reserved(_) => "AUTHS-E4853",
}
}
fn suggestion(&self) -> Option<&'static str> {
match self {
Self::InvalidLength(_) => Some("Tenant ID must be between 1 and 64 characters"),
Self::InvalidCharacter(_) => {
Some("Only lowercase letters, digits, hyphens, and underscores are allowed")
}
Self::Reserved(_) => Some("Choose a different tenant ID; this name is reserved"),
}
}
}
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub struct ValidatedTenantId(String);
const RESERVED_TENANT_IDS: &[&str] = &["admin", "health", "metrics"];
impl ValidatedTenantId {
pub fn new(raw: impl Into<String>) -> Result<Self, RegistryError> {
let normalized = raw.into().to_lowercase();
validate_tenant_id_inner(&normalized)?;
Ok(Self(normalized))
}
pub fn as_str(&self) -> &str {
&self.0
}
}
impl std::fmt::Display for ValidatedTenantId {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str(&self.0)
}
}
impl AsRef<str> for ValidatedTenantId {
fn as_ref(&self) -> &str {
&self.0
}
}
impl From<ValidatedTenantId> for String {
fn from(tid: ValidatedTenantId) -> Self {
tid.0
}
}
fn validate_tenant_id_inner(tenant_id: &str) -> Result<(), RegistryError> {
let len = tenant_id.len();
if len == 0 || len > 64 {
return Err(RegistryError::InvalidTenantId {
tenant_id: tenant_id.into(),
kind: TenantIdError::InvalidLength(len),
});
}
if let Some(bad_char) = tenant_id
.chars()
.find(|c| !matches!(c, 'a'..='z' | '0'..='9' | '-' | '_'))
{
return Err(RegistryError::InvalidTenantId {
tenant_id: tenant_id.into(),
kind: TenantIdError::InvalidCharacter(bad_char),
});
}
if RESERVED_TENANT_IDS.contains(&tenant_id) {
return Err(RegistryError::InvalidTenantId {
tenant_id: tenant_id.into(),
kind: TenantIdError::Reserved(tenant_id.into()),
});
}
Ok(())
}
#[derive(Debug, Error)]
#[non_exhaustive]
pub enum RegistryError {
#[error("Storage error: {0}")]
Storage(#[source] Box<dyn std::error::Error + Send + Sync>),
#[error("Invalid prefix '{prefix}': {reason}")]
InvalidPrefix { prefix: String, reason: String },
#[error("Invalid device DID '{did}': {reason}")]
InvalidDeviceDid { did: String, reason: String },
#[error("Event already exists: {prefix} seq {seq}")]
EventExists { prefix: String, seq: u128 },
#[error("Sequence gap for {prefix}: expected {expected}, got {got}")]
SequenceGap {
prefix: String,
expected: u128,
got: u128,
},
#[error("Not found: {entity_type} '{id}'")]
NotFound { entity_type: String, id: String },
#[error("Serialization error: {0}")]
Serialization(#[from] serde_json::Error),
#[error("Concurrent modification: {0}")]
ConcurrentModification(String),
#[error("SAID mismatch: expected {expected}, got {actual}")]
SaidMismatch { expected: String, actual: String },
#[error("Invalid event: {reason}")]
InvalidEvent { reason: String },
#[error("I/O error: {0}")]
Io(#[from] std::io::Error),
#[error("Internal error: {0}")]
Internal(String),
#[error("invalid tenant ID '{tenant_id}': {kind}")]
InvalidTenantId {
tenant_id: String,
kind: TenantIdError,
},
#[error("Attestation error: {0}")]
Attestation(String),
#[error("Stale attestation: {0}")]
StaleAttestation(String),
#[error("Not implemented: {method}")]
NotImplemented {
method: &'static str,
},
#[error("Batch validation failed at index {index}: {source}")]
BatchValidationFailed {
index: usize,
source: Box<RegistryError>,
},
}
impl auths_core::error::AuthsErrorInfo for RegistryError {
fn error_code(&self) -> &'static str {
match self {
Self::Storage(_) => "AUTHS-E4861",
Self::InvalidPrefix { .. } => "AUTHS-E4862",
Self::InvalidDeviceDid { .. } => "AUTHS-E4863",
Self::EventExists { .. } => "AUTHS-E4864",
Self::SequenceGap { .. } => "AUTHS-E4865",
Self::NotFound { .. } => "AUTHS-E4866",
Self::Serialization(_) => "AUTHS-E4867",
Self::ConcurrentModification(_) => "AUTHS-E4868",
Self::SaidMismatch { .. } => "AUTHS-E4869",
Self::InvalidEvent { .. } => "AUTHS-E4870",
Self::Io(_) => "AUTHS-E4871",
Self::Internal(_) => "AUTHS-E4872",
Self::InvalidTenantId { .. } => "AUTHS-E4873",
Self::Attestation(_) => "AUTHS-E4874",
Self::StaleAttestation(_) => "AUTHS-E4875",
Self::NotImplemented { .. } => "AUTHS-E4876",
Self::BatchValidationFailed { .. } => "AUTHS-E4877",
}
}
fn suggestion(&self) -> Option<&'static str> {
match self {
Self::Storage(_) => Some("Check storage backend connectivity"),
Self::InvalidPrefix { .. } => Some("KERI prefixes must start with 'E' (Blake3 SAID)"),
Self::InvalidDeviceDid { .. } => Some("Device DIDs must be in 'did:key:z...' format"),
Self::EventExists { .. } => Some("This event has already been appended to the KEL"),
Self::SequenceGap { .. } => Some("Events must be appended in strict sequence order"),
Self::NotFound { .. } => None,
Self::Serialization(_) => None,
Self::ConcurrentModification(_) => {
Some("Retry the operation; another process modified the registry")
}
Self::SaidMismatch { .. } => Some("The event content does not match its declared SAID"),
Self::InvalidEvent { .. } => None,
Self::Io(_) => Some("Check file permissions and disk space"),
Self::Internal(_) => None,
Self::InvalidTenantId { .. } => None,
Self::Attestation(_) => None,
Self::StaleAttestation(_) => {
Some("The attestation has been superseded by a newer version")
}
Self::NotImplemented { .. } => {
Some("This operation is not supported by the current backend")
}
Self::BatchValidationFailed { .. } => None,
}
}
}
impl RegistryError {
pub fn storage<E: std::error::Error + Send + Sync + 'static>(err: E) -> Self {
Self::Storage(Box::new(err))
}
pub fn identity_not_found(prefix: &Prefix) -> Self {
Self::NotFound {
entity_type: "identity".into(),
id: prefix.as_str().into(),
}
}
pub fn event_not_found(prefix: &Prefix, seq: u128) -> Self {
Self::NotFound {
entity_type: "event".into(),
id: format!("{} seq {}", prefix.as_str(), seq),
}
}
pub fn device_not_found(did: &str) -> Self {
Self::NotFound {
entity_type: "device".into(),
id: did.into(),
}
}
pub fn org_not_found(org: &str) -> Self {
Self::NotFound {
entity_type: "org".into(),
id: org.into(),
}
}
}
pub trait RegistryBackend: Send + Sync {
fn append_event(&self, prefix: &Prefix, event: &Event) -> Result<(), RegistryError>;
fn append_signed_event(
&self,
prefix: &Prefix,
event: &Event,
_attachment: &[u8],
) -> Result<(), RegistryError> {
self.append_event(prefix, event)
}
fn get_attachment(
&self,
_prefix: &Prefix,
_seq: u128,
) -> Result<Option<Vec<u8>>, RegistryError> {
Ok(None)
}
fn get_event(&self, prefix: &Prefix, seq: u128) -> Result<Event, RegistryError>;
fn visit_events(
&self,
prefix: &Prefix,
from_seq: u128,
visitor: &mut dyn FnMut(&Event) -> ControlFlow<()>,
) -> Result<(), RegistryError>;
fn get_tip(&self, prefix: &Prefix) -> Result<TipInfo, RegistryError>;
fn get_key_state(&self, prefix: &Prefix) -> Result<KeyState, RegistryError>;
fn write_key_state(&self, prefix: &Prefix, state: &KeyState) -> Result<(), RegistryError>;
fn visit_identities(
&self,
visitor: &mut dyn FnMut(&str) -> ControlFlow<()>,
) -> Result<(), RegistryError>;
fn store_attestation(&self, attestation: &Attestation) -> Result<(), RegistryError>;
fn load_attestation(&self, did: &CanonicalDid) -> Result<Option<Attestation>, RegistryError>;
fn visit_attestation_history(
&self,
did: &CanonicalDid,
visitor: &mut dyn FnMut(&Attestation) -> ControlFlow<()>,
) -> Result<(), RegistryError>;
fn visit_devices(
&self,
visitor: &mut dyn FnMut(&CanonicalDid) -> ControlFlow<()>,
) -> Result<(), RegistryError>;
fn store_org_member(&self, org: &str, member: &Attestation) -> Result<(), RegistryError>;
fn visit_org_member_attestations(
&self,
org: &str,
visitor: &mut dyn FnMut(&OrgMemberEntry) -> ControlFlow<()>,
) -> Result<(), RegistryError>;
fn list_org_members(
&self,
org: &str,
_filter: &MemberFilter,
) -> Result<Vec<MemberView>, RegistryError> {
let mut members = Vec::new();
self.visit_org_member_attestations(org, &mut |entry| {
let (status, att_opt, revoked_at) = match &entry.attestation {
Ok(att) => (MemberStatus::Active, Some(att), att.revoked_at),
Err(reason) => (
MemberStatus::Invalid {
reason: reason.clone(),
},
None,
None,
),
};
if let Some(att) = att_opt {
members.push(MemberView {
did: entry.did.clone(),
status,
role: None,
capabilities: vec![],
#[allow(clippy::disallowed_methods)] issuer: IdentityDID::new_unchecked(att.issuer.as_str()),
rid: att.rid.clone(),
revoked_at,
expires_at: att.expires_at,
timestamp: att.timestamp,
source_filename: entry.filename.clone(),
});
} else {
members.push(MemberView {
did: entry.did.clone(),
status,
role: None,
capabilities: vec![],
issuer: entry.org.clone(),
rid: ResourceId::new(""),
revoked_at: None,
expires_at: None,
timestamp: None,
source_filename: entry.filename.clone(),
});
}
ControlFlow::Continue(())
})?;
members.sort_by(|a, b| a.did.to_string().cmp(&b.did.to_string()));
Ok(members)
}
fn init_if_needed(&self) -> Result<bool, RegistryError>;
fn metadata(&self) -> Result<RegistryMetadata, RegistryError>;
fn list_org_members_fast(
&self,
org: &str,
filter: &MemberFilter,
) -> Result<Vec<MemberView>, RegistryError> {
self.list_org_members(org, filter)
}
fn append_tel_event(
&self,
_issuer: &Prefix,
_registry_said: &Said,
_credential_said: &Said,
_sn: u128,
_event_bytes: &[u8],
) -> Result<(), RegistryError> {
Err(RegistryError::NotImplemented {
method: "append_tel_event",
})
}
fn visit_tel_events(
&self,
_issuer: &Prefix,
_registry_said: &Said,
_credential_said: &Said,
_visitor: &mut dyn FnMut(&[u8]) -> ControlFlow<()>,
) -> Result<(), RegistryError> {
Err(RegistryError::NotImplemented {
method: "visit_tel_events",
})
}
fn store_credential(
&self,
_issuer: &Prefix,
_credential_said: &Said,
_credential_bytes: &[u8],
) -> Result<(), RegistryError> {
Err(RegistryError::NotImplemented {
method: "store_credential",
})
}
fn load_credential(
&self,
_issuer: &Prefix,
_credential_said: &Said,
) -> Result<Option<Vec<u8>>, RegistryError> {
Err(RegistryError::NotImplemented {
method: "load_credential",
})
}
fn commit_batch(&self, batch: &AtomicWriteBatch) -> Result<(), RegistryError> {
for op in batch.ops() {
match op {
AtomicWriteOp::StoreAttestation(att) => self.store_attestation(att)?,
AtomicWriteOp::StoreOrgMember { org, member } => {
self.store_org_member(org, member)?
}
AtomicWriteOp::AppendEvent {
prefix,
event,
attachment,
} => {
if attachment.is_empty() {
self.append_event(prefix, event)?;
} else {
self.append_signed_event(prefix, event, attachment)?;
}
}
AtomicWriteOp::AppendTelEvent {
issuer,
registry_said,
credential_said,
sn,
event_bytes,
} => {
self.append_tel_event(issuer, registry_said, credential_said, *sn, event_bytes)?
}
AtomicWriteOp::StoreCredential {
issuer,
credential_said,
credential_bytes,
} => self.store_credential(issuer, credential_said, credential_bytes)?,
}
}
Ok(())
}
}
impl<T: RegistryBackend + ?Sized> RegistryBackend for Arc<T> {
fn append_event(&self, prefix: &Prefix, event: &Event) -> Result<(), RegistryError> {
(**self).append_event(prefix, event)
}
fn get_event(&self, prefix: &Prefix, seq: u128) -> Result<Event, RegistryError> {
(**self).get_event(prefix, seq)
}
fn visit_events(
&self,
prefix: &Prefix,
from_seq: u128,
visitor: &mut dyn FnMut(&Event) -> ControlFlow<()>,
) -> Result<(), RegistryError> {
(**self).visit_events(prefix, from_seq, visitor)
}
fn get_tip(&self, prefix: &Prefix) -> Result<TipInfo, RegistryError> {
(**self).get_tip(prefix)
}
fn get_key_state(&self, prefix: &Prefix) -> Result<KeyState, RegistryError> {
(**self).get_key_state(prefix)
}
fn write_key_state(&self, prefix: &Prefix, state: &KeyState) -> Result<(), RegistryError> {
(**self).write_key_state(prefix, state)
}
fn visit_identities(
&self,
visitor: &mut dyn FnMut(&str) -> ControlFlow<()>,
) -> Result<(), RegistryError> {
(**self).visit_identities(visitor)
}
fn store_attestation(&self, attestation: &Attestation) -> Result<(), RegistryError> {
(**self).store_attestation(attestation)
}
fn load_attestation(&self, did: &CanonicalDid) -> Result<Option<Attestation>, RegistryError> {
(**self).load_attestation(did)
}
fn visit_attestation_history(
&self,
did: &CanonicalDid,
visitor: &mut dyn FnMut(&Attestation) -> ControlFlow<()>,
) -> Result<(), RegistryError> {
(**self).visit_attestation_history(did, visitor)
}
fn visit_devices(
&self,
visitor: &mut dyn FnMut(&CanonicalDid) -> ControlFlow<()>,
) -> Result<(), RegistryError> {
(**self).visit_devices(visitor)
}
fn store_org_member(&self, org: &str, member: &Attestation) -> Result<(), RegistryError> {
(**self).store_org_member(org, member)
}
fn visit_org_member_attestations(
&self,
org: &str,
visitor: &mut dyn FnMut(&OrgMemberEntry) -> ControlFlow<()>,
) -> Result<(), RegistryError> {
(**self).visit_org_member_attestations(org, visitor)
}
fn init_if_needed(&self) -> Result<bool, RegistryError> {
(**self).init_if_needed()
}
fn metadata(&self) -> Result<RegistryMetadata, RegistryError> {
(**self).metadata()
}
fn append_tel_event(
&self,
issuer: &Prefix,
registry_said: &Said,
credential_said: &Said,
sn: u128,
event_bytes: &[u8],
) -> Result<(), RegistryError> {
(**self).append_tel_event(issuer, registry_said, credential_said, sn, event_bytes)
}
fn visit_tel_events(
&self,
issuer: &Prefix,
registry_said: &Said,
credential_said: &Said,
visitor: &mut dyn FnMut(&[u8]) -> ControlFlow<()>,
) -> Result<(), RegistryError> {
(**self).visit_tel_events(issuer, registry_said, credential_said, visitor)
}
fn store_credential(
&self,
issuer: &Prefix,
credential_said: &Said,
credential_bytes: &[u8],
) -> Result<(), RegistryError> {
(**self).store_credential(issuer, credential_said, credential_bytes)
}
fn load_credential(
&self,
issuer: &Prefix,
credential_said: &Said,
) -> Result<Option<Vec<u8>>, RegistryError> {
(**self).load_credential(issuer, credential_said)
}
fn commit_batch(&self, batch: &AtomicWriteBatch) -> Result<(), RegistryError> {
(**self).commit_batch(batch)
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn error_identity_not_found() {
let prefix = Prefix::new_unchecked("ETest123".to_string());
let err = RegistryError::identity_not_found(&prefix);
assert!(err.to_string().contains("identity"));
assert!(err.to_string().contains("ETest123"));
}
#[test]
fn error_event_not_found() {
let prefix = Prefix::new_unchecked("ETest123".to_string());
let err = RegistryError::event_not_found(&prefix, 5);
assert!(err.to_string().contains("event"));
assert!(err.to_string().contains("seq 5"));
}
#[test]
fn error_device_not_found() {
let err = RegistryError::device_not_found("did:key:test");
assert!(err.to_string().contains("device"));
assert!(err.to_string().contains("did:key:test"));
}
#[test]
fn error_sequence_gap() {
let err = RegistryError::SequenceGap {
prefix: "ETest".into(),
expected: 5,
got: 7,
};
assert!(err.to_string().contains("expected 5"));
assert!(err.to_string().contains("got 7"));
}
#[test]
fn error_concurrent_modification() {
let err = RegistryError::ConcurrentModification("Registry was modified".into());
assert!(err.to_string().contains("Concurrent"));
assert!(err.to_string().contains("modified"));
}
#[test]
fn validated_tenant_id_normalizes_to_lowercase() {
let tid = ValidatedTenantId::new("ACME").unwrap();
assert_eq!(tid.as_str(), "acme");
}
#[test]
fn validated_tenant_id_rejects_empty() {
assert!(ValidatedTenantId::new("").is_err());
}
#[test]
fn validated_tenant_id_rejects_reserved() {
assert!(ValidatedTenantId::new("admin").is_err());
}
#[test]
fn validated_tenant_id_rejects_bad_chars() {
assert!(ValidatedTenantId::new("acme/sub").is_err());
assert!(ValidatedTenantId::new("../escape").is_err());
}
#[test]
fn validated_tenant_id_accepts_valid() {
let tid = ValidatedTenantId::new("my-tenant_123").unwrap();
assert_eq!(tid.as_str(), "my-tenant_123");
}
#[test]
fn validated_tenant_id_display() {
let tid = ValidatedTenantId::new("acme").unwrap();
assert_eq!(format!("{}", tid), "acme");
}
#[test]
fn validated_tenant_id_into_string() {
let tid = ValidatedTenantId::new("acme").unwrap();
let s: String = tid.into();
assert_eq!(s, "acme");
}
}