auths_id/storage/registry/

backend.rs

//! Registry backend trait and error types.
//!
//! # Architecture: Cache/Index, Not Source of Truth
//!
//! The registry is an **indexed materialized view** for performance:
//!
//! - **Identity KEL (Key Event Log)** = source of truth
//!   - Events in `v1/identities/<shard>/<prefix>/events/` are canonical
//!   - All other state is derived from the KEL
//!
//! - **Registry** = cached index for O(1) lookups
//!   - `state.json`, `tip.json` are performance caches
//!   - Can be rebuilt from KEL at any time
//!   - NOT authoritative for trust or authorization decisions
//!
//! - **Backend validates structure, not trust**
//!   - Backends may validate JSON structure, SAID integrity
//!   - Backends may NOT decide trust, authorization, or expiry
//!   - Policy engine (separate layer) handles authorization
//!
//! This separation ensures backends remain testable, replaceable, and
//! free from policy coupling.
//!
//! # Overwrite Semantics
//!
//! Different data types have different overwrite semantics:
//!
//! | Data Type | Semantic | Behavior |
//! |-----------|----------|----------|
//! | KEL Events | **Append-only** | Once written, immutable. `append_event()` refuses overwrites. |
//! | Device Attestations | **Latest-view** | `store_attestation()` overwrites existing. History in `attestation_history/`. |
//! | Org Member Attestations | **Latest-view** | `store_org_member()` overwrites existing for same member DID. |
//!
//! The "latest-view" pattern means the current file represents only the latest state.
//! Historical data is preserved separately for audit purposes.

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};

/// An operation to be committed atomically.
///
/// Collected into an [`AtomicWriteBatch`] and committed via
/// [`RegistryBackend::commit_batch`]. Git backends write all operations
/// in a single commit; non-Git backends fall back to sequential writes.
#[derive(Debug, Clone)]
pub enum AtomicWriteOp {
    /// Store a device attestation (latest-view overwrite).
    StoreAttestation(Attestation),
    /// Store an org member attestation (latest-view overwrite).
    StoreOrgMember { org: String, member: Attestation },
    /// Append a KEL event with optional CESR signature attachment.
    AppendEvent {
        prefix: Prefix,
        event: Event,
        attachment: Vec<u8>,
    },
    /// Append a TEL (credential-status) event under an issuer's registry.
    ///
    /// Staged alongside the issuer's anchoring `ixn` (an [`AtomicWriteOp::AppendEvent`])
    /// so the TEL event and its KEL anchor land in one commit. See the
    /// `RegistryBackend` TEL doc-block for the atomicity justification.
    AppendTelEvent {
        /// Issuing AID controlling the registry.
        issuer: Prefix,
        /// Registry SAID (`vcp.d`).
        registry_said: Said,
        /// Credential SAID (`iss.i`/`rev.i`; equals the registry SAID for a `vcp`).
        credential_said: Said,
        /// TEL event sequence number (`0` for `vcp`/`iss`, `1` for `rev`).
        sn: u128,
        /// The TEL event's canonical insertion-order JSON bytes.
        event_bytes: Vec<u8>,
    },
    /// Store an ACDC credential blob under an issuer's namespace.
    ///
    /// Staged alongside the `iss` TEL event and the issuer's anchoring `ixn` so the
    /// blob, the TEL event, and the KEL anchor land in one commit.
    StoreCredential {
        /// Issuing AID.
        issuer: Prefix,
        /// Credential SAID (`acdc.d`).
        credential_said: Said,
        /// The ACDC credential's canonical insertion-order JSON bytes.
        credential_bytes: Vec<u8>,
    },
}

/// Batch of write operations to be committed atomically.
///
/// Build one via [`AtomicWriteBatch::new`], stage operations, then pass
/// to [`RegistryBackend::commit_batch`].
#[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
    }

    /// Stage a TEL event (`vcp`/`iss`/`rev`) under an issuer's registry.
    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
    }

    /// Stage an ACDC credential blob under an issuer's namespace.
    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()
    }
}

/// Specific reasons a tenant ID is rejected.
///
/// `PathTraversal` is intentionally absent: the strict allowlist
/// (`[a-z0-9_-]`) makes traversal components unexpressible.
/// If the allowed charset expands (e.g. to include `.`), reintroduce
/// an explicit traversal check before adding it to the allowlist.
#[derive(Debug, Error, PartialEq)]
#[non_exhaustive]
pub enum TenantIdError {
    /// ID is empty or exceeds 64 characters.
    #[error("must be 1–64 characters (got {0})")]
    InvalidLength(usize),

    /// ID contains a character outside `[a-z0-9_-]`.
    /// Carries the first offending character for precise error messages.
    #[error("contains disallowed character {0:?} (only [a-z0-9_-] allowed)")]
    InvalidCharacter(char),

    /// ID matches a reserved route segment or system name.
    #[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"),
        }
    }
}

/// A tenant identifier that has been normalized (lowercased) and validated.
///
/// Construct via [`ValidatedTenantId::new`] which enforces:
/// - Length 1–64 characters
/// - Only `[a-z0-9_-]` characters (input is lowercased before checking)
/// - Not a reserved name (`admin`, `health`, `metrics`)
///
/// Passing this type through boundaries guarantees the ID is safe for
/// filesystem paths without further checking.
///
/// Usage:
/// ```ignore
/// let tid = ValidatedTenantId::new("Acme")?;
/// assert_eq!(tid.as_str(), "acme");
/// ```
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub struct ValidatedTenantId(String);

/// Reserved names that collide with current or planned API route segments.
const RESERVED_TENANT_IDS: &[&str] = &["admin", "health", "metrics"];

impl ValidatedTenantId {
    /// Normalize and validate a raw tenant identifier.
    ///
    /// Input is lowercased, then checked against the validation rules.
    pub fn new(raw: impl Into<String>) -> Result<Self, RegistryError> {
        let normalized = raw.into().to_lowercase();
        validate_tenant_id_inner(&normalized)?;
        Ok(Self(normalized))
    }

    /// Returns the canonical (lowercase) tenant ID string.
    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(())
}

/// Errors that can occur during registry operations.
#[derive(Debug, Error)]
#[non_exhaustive]
pub enum RegistryError {
    /// Storage backend operation failed
    #[error("Storage error: {0}")]
    Storage(#[source] Box<dyn std::error::Error + Send + Sync>),

    /// Invalid KERI prefix format
    #[error("Invalid prefix '{prefix}': {reason}")]
    InvalidPrefix { prefix: String, reason: String },

    /// Invalid device DID format
    #[error("Invalid device DID '{did}': {reason}")]
    InvalidDeviceDid { did: String, reason: String },

    /// Event already exists at this sequence number
    #[error("Event already exists: {prefix} seq {seq}")]
    EventExists { prefix: String, seq: u128 },

    /// Sequence number gap detected
    #[error("Sequence gap for {prefix}: expected {expected}, got {got}")]
    SequenceGap {
        prefix: String,
        expected: u128,
        got: u128,
    },

    /// Entity not found
    #[error("Not found: {entity_type} '{id}'")]
    NotFound { entity_type: String, id: String },

    /// JSON serialization/deserialization error
    #[error("Serialization error: {0}")]
    Serialization(#[from] serde_json::Error),

    /// Concurrent modification detected (CAS failure)
    #[error("Concurrent modification: {0}")]
    ConcurrentModification(String),

    /// SAID (Self-Addressing Identifier) mismatch (computed != stored)
    #[error("SAID mismatch: expected {expected}, got {actual}")]
    SaidMismatch { expected: String, actual: String },

    /// Invalid event structure (cannot compute SAID)
    #[error("Invalid event: {reason}")]
    InvalidEvent { reason: String },

    /// I/O error
    #[error("I/O error: {0}")]
    Io(#[from] std::io::Error),

    /// Internal error
    #[error("Internal error: {0}")]
    Internal(String),

    /// Invalid tenant ID
    #[error("invalid tenant ID '{tenant_id}': {kind}")]
    InvalidTenantId {
        tenant_id: String,
        kind: TenantIdError,
    },

    /// Attestation validation error
    #[error("Attestation error: {0}")]
    Attestation(String),

    /// Stale or replayed attestation rejected
    #[error("Stale attestation: {0}")]
    StaleAttestation(String),

    /// Method is not yet implemented by this backend.
    #[error("Not implemented: {method}")]
    NotImplemented {
        /// Name of the unimplemented method.
        method: &'static str,
    },

    /// A batch operation failed validation at a specific event index.
    #[error("Batch validation failed at index {index}: {source}")]
    BatchValidationFailed {
        /// Zero-based index of the failing event in the batch.
        index: usize,
        /// The underlying validation error.
        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 {
    /// Create a storage error from any error type.
    ///
    /// Use this to wrap backend-specific errors (e.g., git2::Error) in the
    /// generic Storage variant.
    pub fn storage<E: std::error::Error + Send + Sync + 'static>(err: E) -> Self {
        Self::Storage(Box::new(err))
    }

    /// Create a not-found error for an identity.
    pub fn identity_not_found(prefix: &Prefix) -> Self {
        Self::NotFound {
            entity_type: "identity".into(),
            id: prefix.as_str().into(),
        }
    }

    /// Create a not-found error for an event.
    pub fn event_not_found(prefix: &Prefix, seq: u128) -> Self {
        Self::NotFound {
            entity_type: "event".into(),
            id: format!("{} seq {}", prefix.as_str(), seq),
        }
    }

    /// Create a not-found error for a device.
    pub fn device_not_found(did: &str) -> Self {
        Self::NotFound {
            entity_type: "device".into(),
            id: did.into(),
        }
    }

    /// Create a not-found error for an org.
    pub fn org_not_found(org: &str) -> Self {
        Self::NotFound {
            entity_type: "org".into(),
            id: org.into(),
        }
    }
}

/// Visitor over stored ACDC credential blobs: `(issuer, credential_said, bytes)`.
///
/// Aliased to keep the [`RegistryBackend::visit_credentials`] signature readable.
pub type CredentialVisitor<'a> = dyn FnMut(&Prefix, &Said, &[u8]) -> ControlFlow<()> + 'a;

/// Visitor over persisted TEL coordinates: `(issuer, registry_said, credential_said)`.
///
/// Aliased to keep the [`RegistryBackend::visit_tel_registries`] signature readable.
pub type TelRegistryVisitor<'a> = dyn FnMut(&Prefix, &Said, &Said) -> ControlFlow<()> + 'a;

/// Trait for registry storage backends.
///
/// This trait defines operations for storing and retrieving:
/// - KERI events and key states
/// - Device attestations
/// - Organization memberships
///
/// Implementations must be thread-safe (`Send + Sync`).
///
/// # FROZEN SURFACE
///
/// This trait's method set is frozen. Do not add new methods without:
/// 1. Documented justification in a doc comment
/// 2. Review confirming the method is required for registry operations
///
/// The registry is a cache/index, not a source of truth. New methods should
/// only be added if they are essential for indexing or retrieval operations.
pub trait RegistryBackend: Send + Sync {
    // (visitor closure aliases for the credential/TEL enumeration surface live
    // at module scope below — see `CredentialVisitor` / `TelRegistryVisitor`.)
    // =========================================================================
    // KEL Operations
    // =========================================================================

    /// Append an event to an identity's KEL.
    ///
    /// # Semantics: Append-Only
    ///
    /// Events are **append-only** - once written, they are immutable.
    /// This function **refuses to overwrite** existing events.
    ///
    /// # Constraints (enforced by implementation)
    ///
    /// - Refuses if event file already exists (append-only, no rewrites)
    /// - Refuses if seq != tip.sequence + 1 (except seq 0 for inception)
    /// - Validates event.said() matches content hash expectations
    ///
    /// Also writes state.json with updated KeyState after each append.
    ///
    /// # Arguments
    ///
    /// * `prefix` - The KERI identifier prefix
    /// * `event` - The event to append
    fn append_event(&self, prefix: &Prefix, event: &Event) -> Result<(), RegistryError>;

    /// Append an event together with its CESR-attachment bytes (externalized
    /// signatures and other CESR groups).
    ///
    /// REQUIRED — no default (RT-002). A default that delegated to `append_event`
    /// silently DROPPED the attachment, so backends lost event signatures and the
    /// verify path could only replay KELs structurally — the exact gap RT-002
    /// exploits. Every backend MUST persist the attachment so a KEL can be
    /// AUTHENTICATED, not merely structurally replayed. (This tightens two
    /// existing methods; it does not expand the frozen surface.)
    fn append_signed_event(
        &self,
        prefix: &Prefix,
        event: &Event,
        attachment: &[u8],
    ) -> Result<(), RegistryError>;

    /// Read the CESR-attachment bytes for an event at the given sequence, or
    /// `None` if the event genuinely carries no attachment.
    ///
    /// REQUIRED — no default (RT-002). A default returning `None` made stored
    /// signatures unreadable — the read-side half of the same gap.
    fn get_attachment(&self, prefix: &Prefix, seq: u128) -> Result<Option<Vec<u8>>, RegistryError>;

    /// Get a single event by sequence number (random access).
    ///
    /// # Arguments
    ///
    /// * `prefix` - The KERI identifier prefix
    /// * `seq` - The sequence number of the event
    fn get_event(&self, prefix: &Prefix, seq: u128) -> Result<Event, RegistryError>;

    /// Visit events starting from a sequence (streaming via visitor pattern).
    ///
    /// Calls `visitor` for each event. Return `ControlFlow::Break(())` to stop early.
    ///
    /// # Arguments
    ///
    /// * `prefix` - The KERI identifier prefix
    /// * `from_seq` - Start reading from this sequence number
    /// * `visitor` - Callback invoked for each event
    fn visit_events(
        &self,
        prefix: &Prefix,
        from_seq: u128,
        visitor: &mut dyn FnMut(&Event) -> ControlFlow<()>,
    ) -> Result<(), RegistryError>;

    /// Get tip info without reading events.
    ///
    /// # Arguments
    ///
    /// * `prefix` - The KERI identifier prefix
    fn get_tip(&self, prefix: &Prefix) -> Result<TipInfo, RegistryError>;

    /// Get key state.
    ///
    /// First tries state.json (verified against tip.said).
    /// Falls back to full KEL replay if state.json missing or stale.
    ///
    /// # Arguments
    ///
    /// * `prefix` - The KERI identifier prefix
    fn get_key_state(&self, prefix: &Prefix) -> Result<KeyState, RegistryError>;

    /// Write a pre-computed `KeyState` directly into the registry tree.
    ///
    /// # When to Use
    ///
    /// This is for the archival worker's write-through path: the worker
    /// receives a `KeyState` from the mpsc channel (which was already
    /// validated and written to Redis Tier 0) and needs to persist it
    /// to Git Tier 1 without re-running KEL validation.
    ///
    /// This is **not** the append path for new KERI events. Use `append_event`
    /// for that. This method only updates `state.json` — it does not append
    /// any event blob or update `tip.json`. This is valid because the caller
    /// received the `KeyState` from a successful `append_event` call.
    ///
    /// # Arguments
    ///
    /// * `prefix` - The KERI identifier prefix
    /// * `state` - The pre-validated `KeyState` to persist
    fn write_key_state(&self, prefix: &Prefix, state: &KeyState) -> Result<(), RegistryError>;

    /// Visit all identity prefixes (streaming for tooling/export).
    ///
    /// Calls `visitor` for each identity prefix. Return `ControlFlow::Break(())` to stop early.
    fn visit_identities(
        &self,
        visitor: &mut dyn FnMut(&str) -> ControlFlow<()>,
    ) -> Result<(), RegistryError>;

    // =========================================================================
    // Attestation Operations
    // =========================================================================

    /// Store a device attestation.
    ///
    /// # Semantics: Latest-View (overwrites)
    ///
    /// Attestations use **latest-view** semantics - the current file represents
    /// the latest state only. This function **overwrites any existing attestation**
    /// for the same device DID.
    ///
    /// Historical attestations are preserved in `attestation_history/` via
    /// `visit_attestation_history`.
    ///
    /// # Arguments
    ///
    /// * `attestation` - The attestation to store
    fn store_attestation(&self, attestation: &Attestation) -> Result<(), RegistryError>;

    /// Load an attestation for a device.
    ///
    /// Returns `None` if no attestation exists for the device.
    ///
    /// # Arguments
    ///
    /// * `did` - The device DID
    fn load_attestation(&self, did: &CanonicalDid) -> Result<Option<Attestation>, RegistryError>;

    /// Visit attestation history for a device (append-only audit trail).
    ///
    /// Iterates historical attestations in chronological order (oldest first).
    /// Each entry represents a point-in-time snapshot of the device's attestation.
    ///
    /// # Arguments
    ///
    /// * `did` - The device DID
    /// * `visitor` - Callback invoked for each historical attestation
    fn visit_attestation_history(
        &self,
        did: &CanonicalDid,
        visitor: &mut dyn FnMut(&Attestation) -> ControlFlow<()>,
    ) -> Result<(), RegistryError>;

    /// Visit all device DIDs in the registry.
    ///
    /// Calls `visitor` for each device DID. Return `ControlFlow::Break(())` to stop early.
    fn visit_devices(
        &self,
        visitor: &mut dyn FnMut(&CanonicalDid) -> ControlFlow<()>,
    ) -> Result<(), RegistryError>;

    // =========================================================================
    // Org Operations
    // =========================================================================

    /// Store an org member attestation.
    ///
    /// # Semantics: Latest-View (overwrites)
    ///
    /// Member attestations use **latest-view** semantics - the current file represents
    /// the latest state only. This function **overwrites any existing attestation**
    /// for the same member DID within the org.
    ///
    /// # Arguments
    ///
    /// * `org` - The org DID prefix
    /// * `member` - The member's attestation
    fn store_org_member(&self, org: &str, member: &Attestation) -> Result<(), RegistryError>;

    /// Visit all members of an org (low-level, validation-aware).
    ///
    /// Iterates all member entries with their validation status.
    /// Even invalid entries call visitor (for audit/tooling).
    ///
    /// # Arguments
    ///
    /// * `org` - The org DID prefix
    /// * `visitor` - Callback invoked for each member entry
    fn visit_org_member_attestations(
        &self,
        org: &str,
        visitor: &mut dyn FnMut(&OrgMemberEntry) -> ControlFlow<()>,
    ) -> Result<(), RegistryError>;

    /// List org members with filtering (high-level API).
    ///
    /// Returns validated, filtered member views sorted deterministically by DID.
    ///
    /// # No Policy Decisions
    ///
    /// This method does NOT compute revoked/expired status. All valid attestations
    /// are returned with `status: Active`. Use the `revoked_at` and `expires_at` fields
    /// to compute actual status in your policy layer:
    ///
    /// ```ignore
    /// let actual_status = compute_status_from_view(&view, now);
    /// ```
    ///
    /// The `include_statuses` filter is ignored (status filtering is policy). The
    /// `roles_any`/`capabilities_*` filters are also inert: role/caps authority is
    /// KEL-native (delegator-anchored scope seal), not the attestation, so this method
    /// no longer reads attestation role/caps to filter or populate views.
    ///
    /// # Arguments
    ///
    /// * `org` - The org DID prefix
    /// * `_filter` - Filter criteria (retained for API stability; role/caps filters are inert)
    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,
                ),
            };

            // For valid attestations, surface identity facts only.
            //
            // Role/capabilities authority is KEL-native (the delegator-anchored scope
            // seal — see `crate::storage::registry::org_member` and the SDK org
            // delegation path), NOT the attestation. The scope-seal-derived values are
            // not available at this storage seam, so `MemberView.role`/`.capabilities`
            // are left empty here (display-only, fail-closed): a consumer that needs
            // authority must resolve it from the KEL, never from these fields.
            // The `roles_any`/`capabilities_*` filters are no longer applied for the
            // same reason — filtering on attestation-borne role/caps would re-introduce
            // the retired authority reader.
            if let Some(att) = att_opt {
                // Fail closed: an attestation whose issuer is not a `did:keri:`
                // identity is dropped from the view rather than surfaced with an
                // unvalidated issuer.
                let Ok(issuer) = IdentityDID::parse(att.issuer.as_str()) else {
                    return ControlFlow::Continue(());
                };
                members.push(MemberView {
                    did: entry.did.clone(),
                    status,
                    role: None,
                    capabilities: vec![],
                    issuer,
                    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(())
        })?;

        // Sort deterministically by DID only (status not computed here)
        members.sort_by(|a, b| a.did.to_string().cmp(&b.did.to_string()));

        Ok(members)
    }

    // =========================================================================
    // Lifecycle
    // =========================================================================

    /// Initialize the registry if it has not been set up yet.
    ///
    /// Creates the initial registry structure (e.g., first Git commit, database
    /// schema). Returns `true` if initialization was performed, `false` if the
    /// registry was already initialized.
    ///
    /// Usage:
    /// ```ignore
    /// let was_initialized = backend.init_if_needed()?;
    /// ```
    fn init_if_needed(&self) -> Result<bool, RegistryError>;

    // =========================================================================
    // Metadata
    // =========================================================================

    /// Get registry metadata.
    fn metadata(&self) -> Result<RegistryMetadata, RegistryError>;

    // =========================================================================
    // Fast-path operations (index-accelerated)
    // =========================================================================

    /// List org members using the SQLite index when available.
    ///
    /// # Justification for addition to frozen surface
    ///
    /// This is a performance optimization of the existing `list_org_members` method,
    /// not a new capability. The default implementation delegates to `list_org_members`
    /// (O(n) Git scan). Backends with an index override this for O(1) SQLite lookup.
    ///
    /// # Filter limitations
    ///
    /// The index stores only DID, rid, issuer, revoked_at, and expires_at.
    /// If the filter includes `roles_any`, `capabilities_any`, or `capabilities_all`,
    /// the implementation falls back to the Git scan (those fields require the full
    /// attestation blob). Status filtering is always deferred to the policy layer.
    ///
    /// # Arguments
    ///
    /// * `org` - The org DID prefix
    /// * `filter` - Filter criteria for members
    fn list_org_members_fast(
        &self,
        org: &str,
        filter: &MemberFilter,
    ) -> Result<Vec<MemberView>, RegistryError> {
        self.list_org_members(org, filter)
    }

    // =========================================================================
    // TEL (Transaction Event Log) Operations (FROZEN-TRAIT EXCEPTION)
    //
    // Justification (atomicity — same exception that justified the write-batch
    // surface above): a credential issuance writes three artifacts that MUST be
    // consistent — the ACDC blob, the `iss` TEL event, and the issuer KEL `ixn`
    // carrying the TEL anchor seal. A crash between any two leaves a dangling
    // state: an anchored-but-absent TEL event, or a credential blob with no
    // status log, or a TEL event the KEL never anchored (forgeable). These three
    // writes therefore commit together via `commit_batch`, exactly as the
    // attestation-blob + KEL-ixn pair does. The TEL is a derived, KEL-anchored
    // index (the registry remains a cache, not a source of truth); these methods
    // add only persistence + retrieval of that index, never trust decisions.
    //
    // Reviewed and approved as a frozen-surface extension for Epic F (credential
    // registry). Backends that do not persist a TEL may leave the default
    // `NotImplemented` stubs.
    // =========================================================================

    /// Append a TEL event (`vcp`/`iss`/`rev`) under an issuer's registry.
    ///
    /// # Semantics: Append-Only
    ///
    /// TEL events are immutable once written. Implementations refuse to overwrite
    /// an existing event at the same `(registry, credential, sn)` coordinate.
    ///
    /// Prefer staging this in an [`AtomicWriteBatch`] alongside the anchoring `ixn`
    /// (and, for an `iss`, the ACDC blob) so all three land in one commit.
    ///
    /// # Arguments
    ///
    /// * `issuer` - Issuing AID controlling the registry.
    /// * `registry_said` - Registry SAID (`vcp.d`).
    /// * `credential_said` - Credential SAID (equals the registry SAID for a `vcp`).
    /// * `sn` - TEL event sequence number.
    /// * `event_bytes` - The TEL event's canonical insertion-order JSON bytes.
    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",
        })
    }

    /// Visit the persisted TEL events for one credential, oldest first.
    ///
    /// Calls `visitor` with each event's raw JSON bytes in ascending `sn` order.
    /// Return `ControlFlow::Break(())` to stop early. A credential with no TEL
    /// events visits nothing and returns `Ok(())`.
    ///
    /// # Arguments
    ///
    /// * `issuer` - Issuing AID.
    /// * `registry_said` - Registry SAID.
    /// * `credential_said` - Credential SAID.
    /// * `visitor` - Callback invoked for each TEL event's JSON bytes.
    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",
        })
    }

    /// Store an ACDC credential blob under an issuer's namespace.
    ///
    /// Latest-view by credential SAID; since a credential SAID is content-addressed,
    /// re-storing the same SAID is idempotent.
    ///
    /// # Arguments
    ///
    /// * `issuer` - Issuing AID.
    /// * `credential_said` - Credential SAID (`acdc.d`).
    /// * `credential_bytes` - The ACDC's canonical insertion-order JSON bytes.
    fn store_credential(
        &self,
        _issuer: &Prefix,
        _credential_said: &Said,
        _credential_bytes: &[u8],
    ) -> Result<(), RegistryError> {
        Err(RegistryError::NotImplemented {
            method: "store_credential",
        })
    }

    /// Load an ACDC credential blob, or `None` if absent.
    ///
    /// # Arguments
    ///
    /// * `issuer` - Issuing AID.
    /// * `credential_said` - Credential SAID.
    fn load_credential(
        &self,
        _issuer: &Prefix,
        _credential_said: &Said,
    ) -> Result<Option<Vec<u8>>, RegistryError> {
        Err(RegistryError::NotImplemented {
            method: "load_credential",
        })
    }

    /// Visit every stored ACDC credential blob, in `(issuer, credential)` order.
    ///
    /// Calls `visitor` with each credential's `(issuer, credential_said,
    /// credential_bytes)`. Return `ControlFlow::Break(())` to stop early. A
    /// registry holding no credentials visits nothing and returns `Ok(())`.
    ///
    /// This is the enumeration half of the credential surface (the keyed reader
    /// is [`load_credential`]): tooling that must reconstruct a whole fleet on a
    /// cold machine (`registry pull`) has no way to know the credential SAIDs in
    /// advance, so it walks them here.
    ///
    /// # Arguments
    ///
    /// * `visitor` - Callback invoked for each credential blob.
    fn visit_credentials(&self, _visitor: &mut CredentialVisitor<'_>) -> Result<(), RegistryError> {
        Err(RegistryError::NotImplemented {
            method: "visit_credentials",
        })
    }

    /// Visit every persisted TEL registry coordinate, in tree order.
    ///
    /// Calls `visitor` with each `(issuer, registry_said, credential_said)`
    /// directory that holds TEL events — including the registry's own `vcp`
    /// slot, where `credential_said == registry_said`. Return
    /// `ControlFlow::Break(())` to stop early. A registry holding no TEL visits
    /// nothing and returns `Ok(())`.
    ///
    /// Pairs with [`visit_tel_events`] (which streams one coordinate's events):
    /// enumeration tooling (`registry pull`) discovers the coordinates here, then
    /// reads each one's events to copy them onto the cold machine.
    ///
    /// # Arguments
    ///
    /// * `visitor` - Callback invoked for each TEL coordinate.
    fn visit_tel_registries(
        &self,
        _visitor: &mut TelRegistryVisitor<'_>,
    ) -> Result<(), RegistryError> {
        Err(RegistryError::NotImplemented {
            method: "visit_tel_registries",
        })
    }

    // =========================================================================
    // Atomic Batch Writes (FROZEN-TRAIT EXCEPTION)
    //
    // Justification: Attestation blob + KEL ixn event must land in a single
    // Git commit. Without this, a crash between two writes leaves orphaned
    // blobs or dangling seals — a security gap where forged attestations
    // could exist without KEL evidence. See verification_bindings.md Gap 1.
    // =========================================================================

    /// Commit a batch of write operations atomically.
    ///
    /// Git backends override this to stage all operations in a single
    /// `TreeMutator` and produce one commit. The default implementation
    /// falls back to sequential individual writes (non-atomic, acceptable
    /// for in-memory test backends).
    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(())
    }
}

/// Blanket impl so `Arc<dyn RegistryBackend + Send + Sync>` can be used directly
/// as a `RegistryBackend` at call sites (e.g., tests, extractors).
impl<T: RegistryBackend + ?Sized> RegistryBackend for Arc<T> {
    fn append_event(&self, prefix: &Prefix, event: &Event) -> Result<(), RegistryError> {
        (**self).append_event(prefix, event)
    }

    // RT-002 root cause: this blanket impl forwards every method EXCEPT these two,
    // so before the trait defaults were removed, `Arc<dyn RegistryBackend>` (how
    // the SDK holds `ctx.registry`) silently dropped attachments on write and
    // returned None on read — the inception's signature was lost, and the bundle
    // producer saw `get_attachment(prefix, 0) == None`. Forward both.
    fn append_signed_event(
        &self,
        prefix: &Prefix,
        event: &Event,
        attachment: &[u8],
    ) -> Result<(), RegistryError> {
        (**self).append_signed_event(prefix, event, attachment)
    }

    fn get_attachment(&self, prefix: &Prefix, seq: u128) -> Result<Option<Vec<u8>>, RegistryError> {
        (**self).get_attachment(prefix, seq)
    }

    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 visit_credentials(&self, visitor: &mut CredentialVisitor<'_>) -> Result<(), RegistryError> {
        (**self).visit_credentials(visitor)
    }

    fn visit_tel_registries(
        &self,
        visitor: &mut TelRegistryVisitor<'_>,
    ) -> Result<(), RegistryError> {
        (**self).visit_tel_registries(visitor)
    }

    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");
    }
}