use async_trait::async_trait;
use chrono::{DateTime, Utc};
use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use uuid::Uuid;

use crate::email::{auth_email_event_catalog, default_event_type_for_template_key};
use crate::error::{AuthError, AuthResult};
use crate::types::{
    Account, AdminAuditLog, ApiKey, AthenaClient, CreateAccount, CreateAdminAuditLog, CreateApiKey,
    CreateAthenaClient, CreateEmailSendFailure, CreateEmailTemplate, CreateInvitation,
    CreateMember, CreateOAuthClient, CreateOrganization, CreatePasskey, CreateSentEmail,
    CreateSession, CreateSessionIpProfileRecord, CreateTwoFactor, CreateUser, CreateVerification,
    EmailEventType, EmailSendFailure, EmailTemplate, Invitation, InvitationStatus, IpProfileRecord,
    ListAdminAuditLogsParams, ListEmailEventTypesParams, ListEmailSendFailuresParams,
    ListEmailTemplatesParams, ListSentEmailsParams, ListUsersParams, Member, OAuthClient,
    Organization, Passkey, SentEmail, Session, SessionIpProfileRecord, TwoFactor, UpdateAccount,
    UpdateApiKey, UpdateEmailSendFailure, UpdateEmailTemplate, UpdateOrganization, UpdateSentEmail,
    UpdateUser, UpsertIpProfileRecord, User, Verification,
};

pub use super::memory_traits::{
    MemoryAccount, MemoryApiKey, MemoryInvitation, MemoryMember, MemoryOrganization, MemoryPasskey,
    MemorySession, MemoryTwoFactor, MemoryUser, MemoryVerification,
};

use super::traits::{
    AccountOps, ApiKeyOps, InvitationOps, MemberOps, OrganizationOps, PasskeyOps, SessionOps,
    TwoFactorOps, UserOps, VerificationOps,
};

/// In-memory database adapter for testing and development.
///
/// Generic over entity types — use default type parameters for the built-in
/// types, or supply your own custom structs that implement the `Memory*`
/// traits.
///
/// ```rust,ignore
/// // Using built-in types (no turbofish needed):
/// let adapter = MemoryDatabaseAdapter::new();
///
/// // Using custom types:
/// let adapter = MemoryDatabaseAdapter::<MyUser, MySession, MyAccount,
///     MyOrg, MyMember, MyInvitation, MyVerification>::new();
/// ```
pub struct MemoryDatabaseAdapter<
    U = User,
    S = Session,
    A = Account,
    O = Organization,
    M = Member,
    I = Invitation,
    V = Verification,
    P = Passkey,
> {
    users: Arc<Mutex<HashMap<String, U>>>,
    sessions: Arc<Mutex<HashMap<String, S>>>,
    accounts: Arc<Mutex<HashMap<String, A>>>,
    verifications: Arc<Mutex<HashMap<String, V>>>,
    email_index: Arc<Mutex<HashMap<String, String>>>,
    username_index: Arc<Mutex<HashMap<String, String>>>,
    organizations: Arc<Mutex<HashMap<String, O>>>,
    members: Arc<Mutex<HashMap<String, M>>>,
    invitations: Arc<Mutex<HashMap<String, I>>>,
    slug_index: Arc<Mutex<HashMap<String, String>>>,
    two_factors: Arc<Mutex<HashMap<String, TwoFactor>>>,
    api_keys: Arc<Mutex<HashMap<String, ApiKey>>>,
    passkeys: Arc<Mutex<HashMap<String, P>>>,
    passkey_credential_index: Arc<Mutex<HashMap<String, String>>>,
    athena_clients: Arc<Mutex<HashMap<String, AthenaClient>>>,
    oauth_clients: Arc<Mutex<HashMap<String, OAuthClient>>>,
    sent_emails: Arc<Mutex<HashMap<String, SentEmail>>>,
    email_send_failures: Arc<Mutex<HashMap<String, EmailSendFailure>>>,
    email_templates: Arc<Mutex<HashMap<String, EmailTemplate>>>,
    email_event_types: Arc<Mutex<HashMap<String, EmailEventType>>>,
    admin_audit_logs: Arc<Mutex<Vec<AdminAuditLog>>>,
    ip_profiles_by_ip: Arc<Mutex<HashMap<String, IpProfileRecord>>>,
    session_ip_profiles_by_session_id: Arc<Mutex<HashMap<String, SessionIpProfileRecord>>>,
}

/// Constructor for the default (built-in) entity types.
/// Use `Default::default()` for custom type parameterizations.
impl MemoryDatabaseAdapter {
    pub fn new() -> Self {
        Self::default()
    }
}

impl<U, S, A, O, M, I, V, P> Default for MemoryDatabaseAdapter<U, S, A, O, M, I, V, P> {
    fn default() -> Self {
        Self {
            users: Arc::new(Mutex::new(HashMap::new())),
            sessions: Arc::new(Mutex::new(HashMap::new())),
            accounts: Arc::new(Mutex::new(HashMap::new())),
            verifications: Arc::new(Mutex::new(HashMap::new())),
            email_index: Arc::new(Mutex::new(HashMap::new())),
            username_index: Arc::new(Mutex::new(HashMap::new())),
            organizations: Arc::new(Mutex::new(HashMap::new())),
            members: Arc::new(Mutex::new(HashMap::new())),
            invitations: Arc::new(Mutex::new(HashMap::new())),
            slug_index: Arc::new(Mutex::new(HashMap::new())),
            two_factors: Arc::new(Mutex::new(HashMap::new())),
            api_keys: Arc::new(Mutex::new(HashMap::new())),
            passkeys: Arc::new(Mutex::new(HashMap::new())),
            passkey_credential_index: Arc::new(Mutex::new(HashMap::new())),
            athena_clients: Arc::new(Mutex::new(HashMap::new())),
            oauth_clients: Arc::new(Mutex::new(HashMap::new())),
            sent_emails: Arc::new(Mutex::new(HashMap::new())),
            email_send_failures: Arc::new(Mutex::new(HashMap::new())),
            email_templates: Arc::new(Mutex::new(HashMap::new())),
            email_event_types: Arc::new(Mutex::new(default_email_event_types())),
            admin_audit_logs: Arc::new(Mutex::new(Vec::new())),
            ip_profiles_by_ip: Arc::new(Mutex::new(HashMap::new())),
            session_ip_profiles_by_session_id: Arc::new(Mutex::new(HashMap::new())),
        }
    }
}

fn default_email_event_types() -> HashMap<String, EmailEventType> {
    let now = Utc::now();
    auth_email_event_catalog()
        .iter()
        .map(|entry| {
            let row = EmailEventType {
                event_type: entry.event_type.to_string(),
                category: entry.category.to_string(),
                description: entry.description.to_string(),
                default_template_key: entry.default_template_key.map(str::to_string),
                required_variables: serde_json::json!(entry.required_variables),
                optional_variables: serde_json::json!(entry.optional_variables),
                is_active: true,
                is_system: true,
                metadata: serde_json::json!({ "seed": "core_catalog" }),
                created_at: now,
                updated_at: now,
            };
            (row.event_type.clone(), row)
        })
        .collect()
}

// -- UserOps --

#[async_trait]
impl<U, S, A, O, M, I, V, P> UserOps for MemoryDatabaseAdapter<U, S, A, O, M, I, V, P>
where
    U: MemoryUser,
    S: MemorySession,
    A: MemoryAccount,
    O: MemoryOrganization,
    M: MemoryMember,
    I: MemoryInvitation,
    V: MemoryVerification,
    P: MemoryPasskey,
{
    type User = U;

    async fn create_user(&self, create_user: CreateUser) -> AuthResult<U> {
        let mut users = self
            .users
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        let mut email_index = self
            .email_index
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        let mut username_index = self
            .username_index
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());

        let id = create_user
            .id
            .clone()
            .unwrap_or_else(|| Uuid::new_v4().to_string());

        if let Some(email) = &create_user.email
            && email_index.contains_key(email)
        {
            return Err(AuthError::config("Email already exists"));
        }

        if let Some(username) = &create_user.username
            && username_index.contains_key(username)
        {
            return Err(AuthError::conflict(
                "A user with this username already exists",
            ));
        }

        let now = Utc::now();
        let user = U::from_create(id.clone(), &create_user, now);

        users.insert(id.clone(), user.clone());

        if let Some(email) = &create_user.email {
            email_index.insert(email.clone(), id.clone());
        }
        if let Some(username) = &create_user.username {
            username_index.insert(username.clone(), id);
        }

        Ok(user)
    }

    async fn get_user_by_id(&self, id: &str) -> AuthResult<Option<U>> {
        let users = self
            .users
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        Ok(users.get(id).cloned())
    }

    async fn get_user_by_email(&self, email: &str) -> AuthResult<Option<U>> {
        let email_index = self
            .email_index
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        let users = self
            .users
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());

        if let Some(user_id) = email_index.get(email) {
            Ok(users.get(user_id).cloned())
        } else {
            Ok(None)
        }
    }

    async fn get_user_by_username(&self, username: &str) -> AuthResult<Option<U>> {
        let username_index = self
            .username_index
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        let users = self
            .users
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());

        if let Some(user_id) = username_index.get(username) {
            Ok(users.get(user_id).cloned())
        } else {
            Ok(None)
        }
    }

    async fn update_user(&self, id: &str, update: UpdateUser) -> AuthResult<U> {
        let mut users = self
            .users
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        let mut email_index = self
            .email_index
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        let mut username_index = self
            .username_index
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());

        let user = users.get_mut(id).ok_or(AuthError::UserNotFound)?;

        // Update indices BEFORE mutation (read old values via trait getters)
        if let Some(new_email) = &update.email {
            if let Some(old_email) = user.email() {
                email_index.remove(old_email);
            }
            email_index.insert(new_email.clone(), id.to_string());
        }

        if let Some(ref new_username) = update.username {
            if let Some(old_username) = user.username() {
                username_index.remove(old_username);
            }
            username_index.insert(new_username.clone(), id.to_string());
        }

        user.apply_update(&update);
        Ok(user.clone())
    }

    async fn delete_user(&self, id: &str) -> AuthResult<()> {
        let mut users = self
            .users
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        let mut email_index = self
            .email_index
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        let mut username_index = self
            .username_index
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());

        if let Some(user) = users.remove(id) {
            if let Some(email) = user.email() {
                email_index.remove(email);
            }
            if let Some(username) = user.username() {
                username_index.remove(username);
            }
        }

        Ok(())
    }

    async fn list_users(&self, params: ListUsersParams) -> AuthResult<(Vec<U>, usize)> {
        let users = self
            .users
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        let mut result: Vec<U> = users.values().cloned().collect();

        // Apply search filter
        if let Some(search_value) = &params.search_value {
            let field = params.search_field.as_deref().unwrap_or("email");
            let op = params.search_operator.as_deref().unwrap_or("contains");
            let sv = search_value.to_lowercase();
            result.retain(|u| {
                let field_val = match field {
                    "name" => u.name().unwrap_or("").to_lowercase(),
                    _ => u.email().unwrap_or("").to_lowercase(),
                };
                match op {
                    "starts_with" => field_val.starts_with(&sv),
                    "ends_with" => field_val.ends_with(&sv),
                    _ => field_val.contains(&sv),
                }
            });
        }

        // Apply filter
        if let Some(filter_value) = &params.filter_value {
            let field = params.filter_field.as_deref().unwrap_or("email");
            let op = params.filter_operator.as_deref().unwrap_or("eq");
            let fv = filter_value.to_lowercase();
            result.retain(|u| {
                let field_val = match field {
                    "name" => u.name().unwrap_or("").to_lowercase(),
                    "role" => u.role().unwrap_or("").to_lowercase(),
                    _ => u.email().unwrap_or("").to_lowercase(),
                };
                match op {
                    "contains" => field_val.contains(&fv),
                    "starts_with" => field_val.starts_with(&fv),
                    "ends_with" => field_val.ends_with(&fv),
                    "ne" => field_val != fv,
                    _ => field_val == fv,
                }
            });
        }

        // Apply sort
        if let Some(sort_by) = &params.sort_by {
            let desc = params.sort_direction.as_deref() == Some("desc");
            result.sort_by(|a, b| {
                let av = match sort_by.as_str() {
                    "name" => a.name().unwrap_or("").to_string(),
                    "createdAt" => a.created_at().to_rfc3339(),
                    _ => a.email().unwrap_or("").to_string(),
                };
                let bv = match sort_by.as_str() {
                    "name" => b.name().unwrap_or("").to_string(),
                    "createdAt" => b.created_at().to_rfc3339(),
                    _ => b.email().unwrap_or("").to_string(),
                };
                if desc { bv.cmp(&av) } else { av.cmp(&bv) }
            });
        }

        let total = result.len();
        let offset = params.offset.unwrap_or(0);
        let limit = params.limit.unwrap_or(100);
        let paged: Vec<U> = result.into_iter().skip(offset).take(limit).collect();

        Ok((paged, total))
    }
}

// -- SessionOps --

#[async_trait]
impl<U, S, A, O, M, I, V, P> SessionOps for MemoryDatabaseAdapter<U, S, A, O, M, I, V, P>
where
    U: MemoryUser,
    S: MemorySession,
    A: MemoryAccount,
    O: MemoryOrganization,
    M: MemoryMember,
    I: MemoryInvitation,
    V: MemoryVerification,
    P: MemoryPasskey,
{
    type Session = S;

    async fn create_session(&self, create_session: CreateSession) -> AuthResult<S> {
        let mut sessions = self
            .sessions
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());

        let id = Uuid::new_v4().to_string();
        let token = format!("session_{}", Uuid::new_v4());
        let now = Utc::now();
        let session = S::from_create(id, token.clone(), &create_session, now);

        sessions.insert(token, session.clone());
        Ok(session)
    }

    async fn get_session(&self, token: &str) -> AuthResult<Option<S>> {
        let mut sessions = self
            .sessions
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        let now = Utc::now();

        let Some(session) = sessions.get_mut(token) else {
            return Ok(None);
        };

        if session.active() && session.expires_at() <= now {
            session.set_active(false);
            session.set_updated_at(now);
            return Ok(None);
        }

        if !session.active() {
            return Ok(None);
        }

        Ok(Some(session.clone()))
    }

    async fn get_user_sessions(&self, user_id: &str) -> AuthResult<Vec<S>> {
        let mut sessions = self
            .sessions
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        let now = Utc::now();

        for session in sessions.values_mut() {
            if session.user_id() == user_id && session.active() && session.expires_at() <= now {
                session.set_active(false);
                session.set_updated_at(now);
            }
        }

        Ok(sessions
            .values()
            .filter(|s| s.user_id() == user_id && s.is_active_now())
            .cloned()
            .collect())
    }

    async fn update_session_expiry(
        &self,
        token: &str,
        expires_at: DateTime<Utc>,
    ) -> AuthResult<()> {
        let mut sessions = self
            .sessions
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        if let Some(session) = sessions.get_mut(token) {
            session.set_expires_at(expires_at);
            session.set_updated_at(Utc::now());
        }
        Ok(())
    }

    async fn delete_session(&self, token: &str) -> AuthResult<()> {
        let mut sessions = self
            .sessions
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        sessions.remove(token);
        Ok(())
    }

    async fn delete_user_sessions(&self, user_id: &str) -> AuthResult<()> {
        let mut sessions = self
            .sessions
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        sessions.retain(|_, s| s.user_id() != user_id);
        Ok(())
    }

    async fn delete_expired_sessions(&self) -> AuthResult<usize> {
        let mut sessions = self
            .sessions
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        let now = Utc::now();
        let initial_count = sessions.len();
        sessions.retain(|_, s| s.expires_at() > now && s.active());
        Ok(initial_count - sessions.len())
    }

    async fn update_session_active_organization(
        &self,
        token: &str,
        organization_id: Option<&str>,
    ) -> AuthResult<S> {
        let mut sessions = self
            .sessions
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        let session = sessions.get_mut(token).ok_or(AuthError::SessionNotFound)?;
        session.set_active_organization_id(organization_id.map(|s| s.to_string()));
        session.set_updated_at(Utc::now());
        Ok(session.clone())
    }

    async fn get_ip_profile_by_ip(&self, ip_address: &str) -> AuthResult<Option<IpProfileRecord>> {
        let profiles = self
            .ip_profiles_by_ip
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        Ok(profiles.get(ip_address).cloned())
    }

    async fn upsert_ip_profile(
        &self,
        input: UpsertIpProfileRecord,
    ) -> AuthResult<Option<IpProfileRecord>> {
        let mut profiles = self
            .ip_profiles_by_ip
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        let now = Utc::now();

        if let Some(existing) = profiles.get_mut(&input.ip_address) {
            existing.provider = input.provider;
            existing.country_code = input.country_code;
            existing.country_name = input.country_name;
            existing.region_name = input.region_name;
            existing.city = input.city;
            existing.latitude = input.latitude;
            existing.longitude = input.longitude;
            existing.timezone = input.timezone;
            existing.isp = input.isp;
            existing.organization = input.organization;
            existing.asn = input.asn;
            existing.is_vpn = input.is_vpn;
            existing.is_proxy = input.is_proxy;
            existing.is_tor = input.is_tor;
            existing.is_hosting = input.is_hosting;
            existing.raw_payload = input.raw_payload;
            existing.lookup_expires_at = input.lookup_expires_at;
            existing.last_seen_at = now;
            existing.updated_at = now;
            return Ok(Some(existing.clone()));
        }

        let record = IpProfileRecord {
            id: Uuid::new_v4().to_string(),
            ip_address: input.ip_address.clone(),
            provider: input.provider,
            country_code: input.country_code,
            country_name: input.country_name,
            region_name: input.region_name,
            city: input.city,
            latitude: input.latitude,
            longitude: input.longitude,
            timezone: input.timezone,
            isp: input.isp,
            organization: input.organization,
            asn: input.asn,
            is_vpn: input.is_vpn,
            is_proxy: input.is_proxy,
            is_tor: input.is_tor,
            is_hosting: input.is_hosting,
            raw_payload: input.raw_payload,
            lookup_expires_at: input.lookup_expires_at,
            first_seen_at: now,
            last_seen_at: now,
            created_at: now,
            updated_at: now,
        };
        profiles.insert(input.ip_address, record.clone());
        Ok(Some(record))
    }

    async fn create_session_ip_profile(
        &self,
        input: CreateSessionIpProfileRecord,
    ) -> AuthResult<Option<SessionIpProfileRecord>> {
        let mut session_profiles = self
            .session_ip_profiles_by_session_id
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        if let Some(existing) = session_profiles.get(&input.session_id) {
            return Ok(Some(existing.clone()));
        }

        let record = SessionIpProfileRecord {
            id: Uuid::new_v4().to_string(),
            session_id: input.session_id.clone(),
            session_token: input.session_token,
            user_id: input.user_id,
            ip_address: input.ip_address,
            ip_profile_id: input.ip_profile_id,
            country_code: input.country_code,
            country_name: input.country_name,
            is_vpn: input.is_vpn,
            is_proxy: input.is_proxy,
            is_tor: input.is_tor,
            is_hosting: input.is_hosting,
            country_changed: input.country_changed,
            previous_country_code: input.previous_country_code,
            provider: input.provider,
            created_at: Utc::now(),
        };

        session_profiles.insert(input.session_id, record.clone());
        Ok(Some(record))
    }

    async fn get_latest_session_country_code(
        &self,
        user_id: &str,
        exclude_session_id: Option<&str>,
    ) -> AuthResult<Option<String>> {
        let session_profiles = self
            .session_ip_profiles_by_session_id
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        let latest = session_profiles
            .values()
            .filter(|profile| profile.user_id == user_id)
            .filter(|profile| {
                if let Some(exclude_id) = exclude_session_id {
                    profile.session_id != exclude_id
                } else {
                    true
                }
            })
            .max_by_key(|profile| profile.created_at)
            .and_then(|profile| profile.country_code.clone());
        Ok(latest)
    }

    async fn get_session_ip_profile_by_session_id(
        &self,
        session_id: &str,
    ) -> AuthResult<Option<SessionIpProfileRecord>> {
        let session_profiles = self
            .session_ip_profiles_by_session_id
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        Ok(session_profiles.get(session_id).cloned())
    }
}

// -- AccountOps --

#[async_trait]
impl<U, S, A, O, M, I, V, P> AccountOps for MemoryDatabaseAdapter<U, S, A, O, M, I, V, P>
where
    U: MemoryUser,
    S: MemorySession,
    A: MemoryAccount,
    O: MemoryOrganization,
    M: MemoryMember,
    I: MemoryInvitation,
    V: MemoryVerification,
    P: MemoryPasskey,
{
    type Account = A;

    async fn create_account(&self, create_account: CreateAccount) -> AuthResult<A> {
        let mut accounts = self
            .accounts
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());

        let id = Uuid::new_v4().to_string();
        let now = Utc::now();
        let account = A::from_create(id.clone(), &create_account, now);

        accounts.insert(id, account.clone());
        Ok(account)
    }

    async fn get_account(
        &self,
        provider: &str,
        provider_account_id: &str,
    ) -> AuthResult<Option<A>> {
        let accounts = self
            .accounts
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        Ok(accounts
            .values()
            .find(|acc| acc.provider_id() == provider && acc.account_id() == provider_account_id)
            .cloned())
    }

    async fn get_user_accounts(&self, user_id: &str) -> AuthResult<Vec<A>> {
        let accounts = self
            .accounts
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        Ok(accounts
            .values()
            .filter(|acc| acc.user_id() == user_id)
            .cloned()
            .collect())
    }

    async fn update_account(&self, id: &str, update: UpdateAccount) -> AuthResult<A> {
        let mut accounts = self
            .accounts
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        let account = accounts
            .get_mut(id)
            .ok_or_else(|| AuthError::not_found("Account not found"))?;
        account.apply_update(&update);
        Ok(account.clone())
    }

    async fn delete_account(&self, id: &str) -> AuthResult<()> {
        let mut accounts = self
            .accounts
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        accounts.remove(id);
        Ok(())
    }
}

// -- VerificationOps --

#[async_trait]
impl<U, S, A, O, M, I, V, P> VerificationOps for MemoryDatabaseAdapter<U, S, A, O, M, I, V, P>
where
    U: MemoryUser,
    S: MemorySession,
    A: MemoryAccount,
    O: MemoryOrganization,
    M: MemoryMember,
    I: MemoryInvitation,
    V: MemoryVerification,
    P: MemoryPasskey,
{
    type Verification = V;

    async fn create_verification(&self, create_verification: CreateVerification) -> AuthResult<V> {
        let mut verifications = self
            .verifications
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());

        let id = Uuid::new_v4().to_string();
        let now = Utc::now();
        let verification = V::from_create(id.clone(), &create_verification, now);

        verifications.insert(id, verification.clone());
        Ok(verification)
    }

    async fn get_verification(&self, identifier: &str, value: &str) -> AuthResult<Option<V>> {
        let verifications = self
            .verifications
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        let now = Utc::now();
        Ok(verifications
            .values()
            .find(|v| v.identifier() == identifier && v.value() == value && v.expires_at() > now)
            .cloned())
    }

    async fn get_verification_by_value(&self, value: &str) -> AuthResult<Option<V>> {
        let verifications = self
            .verifications
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        let now = Utc::now();
        Ok(verifications
            .values()
            .find(|v| v.value() == value && v.expires_at() > now)
            .cloned())
    }

    async fn get_verification_by_identifier(&self, identifier: &str) -> AuthResult<Option<V>> {
        let verifications = self
            .verifications
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        let now = Utc::now();
        Ok(verifications
            .values()
            .find(|v| v.identifier() == identifier && v.expires_at() > now)
            .cloned())
    }

    async fn consume_verification(&self, identifier: &str, value: &str) -> AuthResult<Option<V>> {
        let mut verifications = self
            .verifications
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        let now = Utc::now();

        let matched_id = verifications
            .iter()
            .filter_map(|(id, verification)| {
                if verification.identifier() == identifier
                    && verification.value() == value
                    && verification.expires_at() > now
                {
                    Some((id, verification.created_at()))
                } else {
                    None
                }
            })
            .max_by_key(|(_, created_at)| *created_at)
            .map(|(id, _)| id.clone());

        if let Some(id) = matched_id {
            Ok(verifications.remove(&id))
        } else {
            Ok(None)
        }
    }

    async fn delete_verification(&self, id: &str) -> AuthResult<()> {
        let mut verifications = self
            .verifications
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        verifications.remove(id);
        Ok(())
    }

    async fn delete_expired_verifications(&self) -> AuthResult<usize> {
        let mut verifications = self
            .verifications
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        let now = Utc::now();
        let initial_count = verifications.len();
        verifications.retain(|_, v| v.expires_at() > now);
        Ok(initial_count - verifications.len())
    }

    async fn create_email_send_failure(&self, failure: CreateEmailSendFailure) -> AuthResult<()> {
        self.create_email_send_failure_record(failure).await?;
        Ok(())
    }

    async fn create_email_send_failure_record(
        &self,
        failure: CreateEmailSendFailure,
    ) -> AuthResult<EmailSendFailure> {
        let mut rows = self
            .email_send_failures
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        let now = Utc::now();
        let row = EmailSendFailure {
            id: Uuid::new_v4().to_string(),
            user_id: failure.user_id,
            recipient_email: failure.recipient_email,
            flow: failure.flow,
            provider: failure.provider,
            error_message: failure.error_message,
            metadata: failure.metadata.unwrap_or_else(|| serde_json::json!({})),
            resolved: false,
            resolution_note: None,
            created_at: now,
            updated_at: now,
        };
        rows.insert(row.id.clone(), row.clone());
        Ok(row)
    }

    async fn create_sent_email(&self, sent: CreateSentEmail) -> AuthResult<()> {
        self.create_sent_email_record(sent).await?;
        Ok(())
    }

    async fn create_sent_email_record(&self, sent: CreateSentEmail) -> AuthResult<SentEmail> {
        let mut rows = self
            .sent_emails
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        let now = Utc::now();
        let row = SentEmail {
            id: Uuid::new_v4().to_string(),
            recipient_email: sent.recipient_email,
            subject: sent.subject,
            from_address: sent.from_address,
            from_name: sent.from_name,
            text_body: sent.text_body,
            html_body: sent.html_body,
            provider: sent.provider,
            flow: sent.flow,
            metadata: sent.metadata.unwrap_or_else(|| serde_json::json!({})),
            created_at: now,
            updated_at: now,
        };
        rows.insert(row.id.clone(), row.clone());
        Ok(row)
    }

    async fn list_sent_emails(&self, limit: usize, offset: usize) -> AuthResult<Vec<SentEmail>> {
        self.list_sent_emails_with_params(ListSentEmailsParams {
            limit: Some(limit),
            offset: Some(offset),
            ..Default::default()
        })
        .await
    }

    async fn list_sent_emails_with_params(
        &self,
        params: ListSentEmailsParams,
    ) -> AuthResult<Vec<SentEmail>> {
        let rows = self
            .sent_emails
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        let limit = params.limit.unwrap_or(100).min(1000);
        let offset = params.offset.unwrap_or(0);

        let recipient_filter = params
            .recipient_email
            .map(|value| value.trim().to_ascii_lowercase())
            .filter(|value| !value.is_empty());
        let provider_filter = params
            .provider
            .map(|value| value.trim().to_ascii_lowercase())
            .filter(|value| !value.is_empty());
        let flow_filter = params
            .flow
            .map(|value| value.trim().to_ascii_lowercase())
            .filter(|value| !value.is_empty());
        let subject_contains_filter = params
            .subject_contains
            .map(|value| value.trim().to_ascii_lowercase())
            .filter(|value| !value.is_empty());
        let created_after_ts = params
            .created_after
            .map(|created_after| created_after.timestamp_micros());
        let created_before_ts = params
            .created_before
            .map(|created_before| created_before.timestamp_micros());

        let mut out: Vec<SentEmail> = rows
            .values()
            .filter(|row| {
                if let Some(ref recipient) = recipient_filter
                    && row.recipient_email.to_ascii_lowercase() != *recipient
                {
                    return false;
                }
                if let Some(ref provider) = provider_filter
                    && row.provider.to_ascii_lowercase() != *provider
                {
                    return false;
                }
                if let Some(ref flow) = flow_filter {
                    let row_flow = row.flow.as_deref().unwrap_or("").to_ascii_lowercase();
                    if row_flow != *flow {
                        return false;
                    }
                }
                if let Some(ref subject_contains) = subject_contains_filter
                    && !row
                        .subject
                        .to_ascii_lowercase()
                        .contains(subject_contains.as_str())
                {
                    return false;
                }
                if let Some(created_after_ts) = created_after_ts
                    && row.created_at.timestamp_micros() < created_after_ts
                {
                    return false;
                }
                if let Some(created_before_ts) = created_before_ts
                    && row.created_at.timestamp_micros() > created_before_ts
                {
                    return false;
                }
                true
            })
            .cloned()
            .collect();

        out.sort_by(|a, b| b.created_at.cmp(&a.created_at));
        Ok(out.into_iter().skip(offset).take(limit).collect())
    }

    async fn count_sent_emails(&self, params: ListSentEmailsParams) -> AuthResult<usize> {
        let rows = self
            .sent_emails
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        let recipient_filter = params
            .recipient_email
            .as_deref()
            .map(str::trim)
            .map(str::to_ascii_lowercase)
            .filter(|value| !value.is_empty());
        let provider_filter = params
            .provider
            .as_deref()
            .map(str::trim)
            .map(str::to_ascii_lowercase)
            .filter(|value| !value.is_empty());
        let flow_filter = params
            .flow
            .as_deref()
            .map(str::trim)
            .map(str::to_ascii_lowercase)
            .filter(|value| !value.is_empty());
        let subject_contains_filter = params
            .subject_contains
            .as_deref()
            .map(str::trim)
            .map(str::to_ascii_lowercase)
            .filter(|value| !value.is_empty());
        let created_after_ts = params
            .created_after
            .as_ref()
            .map(|created_after| created_after.timestamp_micros());
        let created_before_ts = params
            .created_before
            .as_ref()
            .map(|created_before| created_before.timestamp_micros());

        let total = rows
            .values()
            .filter(|row| {
                if let Some(ref recipient) = recipient_filter
                    && row.recipient_email.to_ascii_lowercase() != *recipient
                {
                    return false;
                }
                if let Some(ref provider) = provider_filter
                    && row.provider.to_ascii_lowercase() != *provider
                {
                    return false;
                }
                if let Some(ref flow) = flow_filter {
                    let row_flow = row.flow.as_deref().unwrap_or("").to_ascii_lowercase();
                    if row_flow != *flow {
                        return false;
                    }
                }
                if let Some(ref subject_contains) = subject_contains_filter
                    && !row
                        .subject
                        .to_ascii_lowercase()
                        .contains(subject_contains.as_str())
                {
                    return false;
                }
                if let Some(created_after_ts) = created_after_ts
                    && row.created_at.timestamp_micros() < created_after_ts
                {
                    return false;
                }
                if let Some(created_before_ts) = created_before_ts
                    && row.created_at.timestamp_micros() > created_before_ts
                {
                    return false;
                }
                true
            })
            .count();

        Ok(total)
    }

    async fn list_sent_emails_for_recipient(
        &self,
        recipient_email: &str,
        limit: usize,
        offset: usize,
    ) -> AuthResult<Vec<SentEmail>> {
        self.list_sent_emails_with_params(ListSentEmailsParams {
            limit: Some(limit),
            offset: Some(offset),
            recipient_email: Some(recipient_email.to_string()),
            ..Default::default()
        })
        .await
    }

    async fn get_sent_email_by_id(&self, id: &str) -> AuthResult<Option<SentEmail>> {
        let rows = self
            .sent_emails
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        Ok(rows.get(id).cloned())
    }

    async fn update_sent_email(&self, id: &str, update: UpdateSentEmail) -> AuthResult<SentEmail> {
        let mut rows = self
            .sent_emails
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        let row = rows
            .get_mut(id)
            .ok_or_else(|| AuthError::not_found("Sent email not found"))?;

        if let Some(recipient_email) = update.recipient_email {
            row.recipient_email = recipient_email;
        }
        if let Some(subject) = update.subject {
            row.subject = subject;
        }
        if let Some(from_address) = update.from_address {
            row.from_address = from_address;
        }
        if let Some(from_name) = update.from_name {
            row.from_name = from_name;
        }
        if let Some(text_body) = update.text_body {
            row.text_body = text_body;
        }
        if let Some(html_body) = update.html_body {
            row.html_body = html_body;
        }
        if let Some(provider) = update.provider {
            row.provider = provider;
        }
        if let Some(flow) = update.flow {
            row.flow = flow;
        }
        if let Some(metadata) = update.metadata {
            row.metadata = metadata;
        }
        row.updated_at = Utc::now();

        Ok(row.clone())
    }

    async fn delete_sent_email(&self, id: &str) -> AuthResult<()> {
        let mut rows = self
            .sent_emails
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        if rows.remove(id).is_none() {
            return Err(AuthError::not_found("Sent email not found"));
        }
        Ok(())
    }

    async fn list_email_send_failures(
        &self,
        params: ListEmailSendFailuresParams,
    ) -> AuthResult<Vec<EmailSendFailure>> {
        let rows = self
            .email_send_failures
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        let limit = params.limit.unwrap_or(100).min(1000);
        let offset = params.offset.unwrap_or(0);
        let recipient_filter = params
            .recipient_email
            .map(|value| value.trim().to_ascii_lowercase())
            .filter(|value| !value.is_empty());
        let provider_filter = params
            .provider
            .map(|value| value.trim().to_ascii_lowercase())
            .filter(|value| !value.is_empty());
        let flow_filter = params
            .flow
            .map(|value| value.trim().to_ascii_lowercase())
            .filter(|value| !value.is_empty());
        let created_after_ts = params
            .created_after
            .map(|created_after| created_after.timestamp_micros());
        let created_before_ts = params
            .created_before
            .map(|created_before| created_before.timestamp_micros());

        let mut out: Vec<EmailSendFailure> = rows
            .values()
            .filter(|row| {
                if let Some(ref recipient) = recipient_filter
                    && row.recipient_email.to_ascii_lowercase() != *recipient
                {
                    return false;
                }
                if let Some(ref provider) = provider_filter {
                    let row_provider = row.provider.as_deref().unwrap_or("").to_ascii_lowercase();
                    if row_provider != *provider {
                        return false;
                    }
                }
                if let Some(ref flow) = flow_filter
                    && row.flow.to_ascii_lowercase() != *flow
                {
                    return false;
                }
                if let Some(resolved) = params.resolved
                    && row.resolved != resolved
                {
                    return false;
                }
                if let Some(created_after_ts) = created_after_ts
                    && row.created_at.timestamp_micros() < created_after_ts
                {
                    return false;
                }
                if let Some(created_before_ts) = created_before_ts
                    && row.created_at.timestamp_micros() > created_before_ts
                {
                    return false;
                }
                true
            })
            .cloned()
            .collect();

        out.sort_by(|a, b| b.created_at.cmp(&a.created_at));
        Ok(out.into_iter().skip(offset).take(limit).collect())
    }

    async fn count_email_send_failures(
        &self,
        params: ListEmailSendFailuresParams,
    ) -> AuthResult<usize> {
        let rows = self
            .email_send_failures
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        let recipient_filter = params
            .recipient_email
            .as_deref()
            .map(str::trim)
            .map(str::to_ascii_lowercase)
            .filter(|value| !value.is_empty());
        let provider_filter = params
            .provider
            .as_deref()
            .map(str::trim)
            .map(str::to_ascii_lowercase)
            .filter(|value| !value.is_empty());
        let flow_filter = params
            .flow
            .as_deref()
            .map(str::trim)
            .map(str::to_ascii_lowercase)
            .filter(|value| !value.is_empty());
        let created_after_ts = params
            .created_after
            .as_ref()
            .map(|created_after| created_after.timestamp_micros());
        let created_before_ts = params
            .created_before
            .as_ref()
            .map(|created_before| created_before.timestamp_micros());

        let total = rows
            .values()
            .filter(|row| {
                if let Some(ref recipient) = recipient_filter
                    && row.recipient_email.to_ascii_lowercase() != *recipient
                {
                    return false;
                }
                if let Some(ref provider) = provider_filter {
                    let row_provider = row.provider.as_deref().unwrap_or("").to_ascii_lowercase();
                    if row_provider != *provider {
                        return false;
                    }
                }
                if let Some(ref flow) = flow_filter
                    && row.flow.to_ascii_lowercase() != *flow
                {
                    return false;
                }
                if let Some(resolved) = params.resolved
                    && row.resolved != resolved
                {
                    return false;
                }
                if let Some(created_after_ts) = created_after_ts
                    && row.created_at.timestamp_micros() < created_after_ts
                {
                    return false;
                }
                if let Some(created_before_ts) = created_before_ts
                    && row.created_at.timestamp_micros() > created_before_ts
                {
                    return false;
                }
                true
            })
            .count();

        Ok(total)
    }

    async fn get_email_send_failure_by_id(&self, id: &str) -> AuthResult<Option<EmailSendFailure>> {
        let rows = self
            .email_send_failures
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        Ok(rows.get(id).cloned())
    }

    async fn update_email_send_failure(
        &self,
        id: &str,
        update: UpdateEmailSendFailure,
    ) -> AuthResult<EmailSendFailure> {
        let mut rows = self
            .email_send_failures
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        let row = rows
            .get_mut(id)
            .ok_or_else(|| AuthError::not_found("Email send failure not found"))?;

        if let Some(resolved) = update.resolved {
            row.resolved = resolved;
        }
        if let Some(resolution_note) = update.resolution_note {
            row.resolution_note = resolution_note;
        }
        if let Some(metadata) = update.metadata {
            row.metadata = metadata;
        }
        row.updated_at = Utc::now();
        Ok(row.clone())
    }

    async fn delete_email_send_failure(&self, id: &str) -> AuthResult<()> {
        let mut rows = self
            .email_send_failures
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        if rows.remove(id).is_none() {
            return Err(AuthError::not_found("Email send failure not found"));
        }
        Ok(())
    }

    async fn create_admin_audit_log(&self, log: CreateAdminAuditLog) -> AuthResult<()> {
        let mut rows = self
            .admin_audit_logs
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        rows.push(AdminAuditLog {
            id: Uuid::new_v4().to_string(),
            actor_user_id: log.actor_user_id,
            actor_session_token: log.actor_session_token,
            action: log.action,
            target_type: log.target_type,
            target_id: log.target_id,
            success: log.success,
            error_message: log.error_message,
            metadata: log.metadata.unwrap_or_else(|| serde_json::json!({})),
            created_at: Utc::now(),
        });
        Ok(())
    }

    async fn list_admin_audit_logs(
        &self,
        limit: usize,
        offset: usize,
    ) -> AuthResult<Vec<AdminAuditLog>> {
        self.list_admin_audit_logs_with_params(ListAdminAuditLogsParams {
            limit: Some(limit),
            offset: Some(offset),
            ..Default::default()
        })
        .await
    }

    async fn list_admin_audit_logs_with_params(
        &self,
        params: ListAdminAuditLogsParams,
    ) -> AuthResult<Vec<AdminAuditLog>> {
        let rows = self
            .admin_audit_logs
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        let limit = params.limit.unwrap_or(100).min(1000);
        let offset = params.offset.unwrap_or(0);
        let actor_user_id_filter = params
            .actor_user_id
            .as_deref()
            .map(str::trim)
            .filter(|value| !value.is_empty());
        let action_filter = params
            .action
            .as_deref()
            .map(str::trim)
            .map(str::to_ascii_lowercase)
            .filter(|value| !value.is_empty());
        let target_type_filter = params
            .target_type
            .as_deref()
            .map(str::trim)
            .map(str::to_ascii_lowercase)
            .filter(|value| !value.is_empty());
        let target_id_filter = params
            .target_id
            .as_deref()
            .map(str::trim)
            .filter(|value| !value.is_empty());
        let created_after_ts = params
            .created_after
            .as_ref()
            .map(|created_after| created_after.timestamp_micros());
        let created_before_ts = params
            .created_before
            .as_ref()
            .map(|created_before| created_before.timestamp_micros());

        let mut out: Vec<AdminAuditLog> = rows
            .iter()
            .filter(|row| {
                if let Some(actor_user_id) = actor_user_id_filter
                    && row.actor_user_id.as_deref() != Some(actor_user_id)
                {
                    return false;
                }
                if let Some(ref action) = action_filter
                    && row.action.to_ascii_lowercase() != *action
                {
                    return false;
                }
                if let Some(ref target_type) = target_type_filter {
                    let row_target_type = row
                        .target_type
                        .as_deref()
                        .unwrap_or("")
                        .to_ascii_lowercase();
                    if row_target_type != *target_type {
                        return false;
                    }
                }
                if let Some(target_id) = target_id_filter
                    && row.target_id.as_deref() != Some(target_id)
                {
                    return false;
                }
                if let Some(success) = params.success
                    && row.success != success
                {
                    return false;
                }
                if let Some(created_after_ts) = created_after_ts
                    && row.created_at.timestamp_micros() < created_after_ts
                {
                    return false;
                }
                if let Some(created_before_ts) = created_before_ts
                    && row.created_at.timestamp_micros() > created_before_ts
                {
                    return false;
                }
                true
            })
            .cloned()
            .collect();

        out.sort_by(|a, b| b.created_at.cmp(&a.created_at));
        Ok(out.into_iter().skip(offset).take(limit).collect())
    }

    async fn count_admin_audit_logs(&self, params: ListAdminAuditLogsParams) -> AuthResult<usize> {
        let rows = self
            .admin_audit_logs
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        let actor_user_id_filter = params
            .actor_user_id
            .as_deref()
            .map(str::trim)
            .filter(|value| !value.is_empty());
        let action_filter = params
            .action
            .as_deref()
            .map(str::trim)
            .map(str::to_ascii_lowercase)
            .filter(|value| !value.is_empty());
        let target_type_filter = params
            .target_type
            .as_deref()
            .map(str::trim)
            .map(str::to_ascii_lowercase)
            .filter(|value| !value.is_empty());
        let target_id_filter = params
            .target_id
            .as_deref()
            .map(str::trim)
            .filter(|value| !value.is_empty());
        let created_after_ts = params
            .created_after
            .as_ref()
            .map(|created_after| created_after.timestamp_micros());
        let created_before_ts = params
            .created_before
            .as_ref()
            .map(|created_before| created_before.timestamp_micros());

        let total = rows
            .iter()
            .filter(|row| {
                if let Some(actor_user_id) = actor_user_id_filter
                    && row.actor_user_id.as_deref() != Some(actor_user_id)
                {
                    return false;
                }
                if let Some(ref action) = action_filter
                    && row.action.to_ascii_lowercase() != *action
                {
                    return false;
                }
                if let Some(ref target_type) = target_type_filter {
                    let row_target_type = row
                        .target_type
                        .as_deref()
                        .unwrap_or("")
                        .to_ascii_lowercase();
                    if row_target_type != *target_type {
                        return false;
                    }
                }
                if let Some(target_id) = target_id_filter
                    && row.target_id.as_deref() != Some(target_id)
                {
                    return false;
                }
                if let Some(success) = params.success
                    && row.success != success
                {
                    return false;
                }
                if let Some(created_after_ts) = created_after_ts
                    && row.created_at.timestamp_micros() < created_after_ts
                {
                    return false;
                }
                if let Some(created_before_ts) = created_before_ts
                    && row.created_at.timestamp_micros() > created_before_ts
                {
                    return false;
                }
                true
            })
            .count();
        Ok(total)
    }

    async fn create_athena_client(&self, input: CreateAthenaClient) -> AuthResult<AthenaClient> {
        let mut clients = self
            .athena_clients
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        if clients
            .values()
            .any(|c| c.client_name.eq_ignore_ascii_case(&input.client_name))
        {
            return Err(AuthError::conflict("Athena client already exists"));
        }

        let now = Utc::now();
        let row = AthenaClient {
            id: Uuid::new_v4().to_string(),
            client_name: input.client_name,
            description: input.description,
            is_active: input.is_active.unwrap_or(true),
            pg_uri_env_var: input.pg_uri_env_var,
            metadata: input.metadata.unwrap_or_else(|| serde_json::json!({})),
            created_at: now,
            updated_at: now,
        };
        clients.insert(row.id.clone(), row.clone());
        Ok(row)
    }

    async fn list_athena_clients(&self) -> AuthResult<Vec<AthenaClient>> {
        let clients = self
            .athena_clients
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        let mut out: Vec<AthenaClient> = clients.values().cloned().collect();
        out.sort_by(|a, b| b.created_at.cmp(&a.created_at));
        Ok(out)
    }

    async fn create_oauth_client(&self, input: CreateOAuthClient) -> AuthResult<OAuthClient> {
        let mut clients = self
            .oauth_clients
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        if clients
            .values()
            .any(|c| c.client_name.eq_ignore_ascii_case(&input.client_name))
        {
            return Err(AuthError::conflict("OAuth client already exists"));
        }

        let now = Utc::now();
        let row = OAuthClient {
            id: Uuid::new_v4().to_string(),
            client_name: input.client_name,
            provider: input.provider,
            description: input.description,
            response_type: input.response_type.unwrap_or_else(|| "code".to_string()),
            grant_type: input
                .grant_type
                .unwrap_or_else(|| "authorization_code".to_string()),
            token_authentication_method: input
                .token_authentication_method
                .unwrap_or_else(|| "none".to_string()),
            redirect_uris: input.redirect_uris,
            client_url: input.client_url,
            scopes: input.scopes.unwrap_or_default(),
            is_active: input.is_active.unwrap_or(true),
            metadata: input.metadata.unwrap_or_else(|| serde_json::json!({})),
            created_at: now,
            updated_at: now,
        };
        clients.insert(row.id.clone(), row.clone());
        Ok(row)
    }

    async fn list_oauth_clients(&self) -> AuthResult<Vec<OAuthClient>> {
        let clients = self
            .oauth_clients
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        let mut out: Vec<OAuthClient> = clients.values().cloned().collect();
        out.sort_by(|a, b| b.created_at.cmp(&a.created_at));
        Ok(out)
    }

    async fn find_active_oauth_client_by_provider_and_redirect_uri(
        &self,
        provider: &str,
        redirect_uri: &str,
    ) -> AuthResult<Option<OAuthClient>> {
        let clients = self
            .oauth_clients
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        Ok(clients
            .values()
            .filter(|client| client.is_active)
            .find(|client| {
                client.provider.eq_ignore_ascii_case(provider)
                    && client.redirect_uris.iter().any(|uri| uri == redirect_uri)
            })
            .cloned())
    }

    async fn get_active_email_template(
        &self,
        template_key: &str,
        locale: Option<&str>,
    ) -> AuthResult<Option<EmailTemplate>> {
        let templates = self
            .email_templates
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        let preferred_locale = locale
            .map(str::trim)
            .filter(|value| !value.is_empty())
            .unwrap_or("en")
            .to_ascii_lowercase();
        let fallback_locale = preferred_locale
            .split('-')
            .next()
            .map(str::trim)
            .filter(|value| !value.is_empty())
            .unwrap_or("en")
            .to_string();

        let mut candidates: Vec<EmailTemplate> = templates
            .values()
            .filter(|template| {
                template.is_active
                    && template.template_key == template_key
                    && (template.locale.eq_ignore_ascii_case(&preferred_locale)
                        || template.locale.eq_ignore_ascii_case(&fallback_locale)
                        || template.locale.eq_ignore_ascii_case("en"))
            })
            .cloned()
            .collect();

        candidates.sort_by(|a, b| {
            let rank = |value: &str| -> usize {
                if value.eq_ignore_ascii_case(&preferred_locale) {
                    0
                } else if value.eq_ignore_ascii_case(&fallback_locale) {
                    1
                } else if value.eq_ignore_ascii_case("en") {
                    2
                } else {
                    3
                }
            };

            rank(&a.locale)
                .cmp(&rank(&b.locale))
                .then_with(|| b.updated_at.cmp(&a.updated_at))
        });

        Ok(candidates.into_iter().next())
    }

    async fn get_active_email_template_for_event(
        &self,
        event_type: &str,
        template_key: &str,
        locale: Option<&str>,
    ) -> AuthResult<Option<EmailTemplate>> {
        let templates = self
            .email_templates
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        let preferred_locale = locale
            .map(str::trim)
            .filter(|value| !value.is_empty())
            .unwrap_or("en")
            .to_ascii_lowercase();
        let fallback_locale = preferred_locale
            .split('-')
            .next()
            .map(str::trim)
            .filter(|value| !value.is_empty())
            .unwrap_or("en")
            .to_string();

        let mut candidates: Vec<EmailTemplate> = templates
            .values()
            .filter(|template| {
                template.is_active
                    && (template.event_type.as_deref() == Some(event_type)
                        || template.template_key == template_key)
                    && (template.locale.eq_ignore_ascii_case(&preferred_locale)
                        || template.locale.eq_ignore_ascii_case(&fallback_locale)
                        || template.locale.eq_ignore_ascii_case("en"))
            })
            .cloned()
            .collect();

        candidates.sort_by(|a, b| {
            let source_rank = |template: &EmailTemplate| -> usize {
                if template.event_type.as_deref() == Some(event_type) {
                    0
                } else if template.template_key == template_key {
                    1
                } else {
                    2
                }
            };
            let locale_rank = |value: &str| -> usize {
                if value.eq_ignore_ascii_case(&preferred_locale) {
                    0
                } else if value.eq_ignore_ascii_case(&fallback_locale) {
                    1
                } else if value.eq_ignore_ascii_case("en") {
                    2
                } else {
                    3
                }
            };

            source_rank(a)
                .cmp(&source_rank(b))
                .then_with(|| locale_rank(&a.locale).cmp(&locale_rank(&b.locale)))
                .then_with(|| b.updated_at.cmp(&a.updated_at))
        });

        Ok(candidates.into_iter().next())
    }

    async fn create_email_template(&self, input: CreateEmailTemplate) -> AuthResult<EmailTemplate> {
        let mut templates = self
            .email_templates
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        let event_types = self
            .email_event_types
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        let normalized_key = input.template_key.trim().to_string();
        let normalized_event_type = input
            .event_type
            .as_deref()
            .map(str::trim)
            .filter(|value| !value.is_empty())
            .map(str::to_string)
            .or_else(|| default_event_type_for_template_key(&normalized_key).map(str::to_string));
        let normalized_locale = input
            .locale
            .as_deref()
            .map(str::trim)
            .filter(|value| !value.is_empty())
            .unwrap_or("en")
            .to_ascii_lowercase();

        if templates.values().any(|template| {
            template.template_key == normalized_key && template.locale == normalized_locale
        }) {
            return Err(AuthError::conflict(
                "Email template with this templateKey and locale already exists",
            ));
        }

        if let Some(event_type) = normalized_event_type.as_deref()
            && !event_types.contains_key(event_type)
        {
            return Err(AuthError::bad_request(format!(
                "Unknown eventType '{event_type}'"
            )));
        }

        let now = Utc::now();
        let row = EmailTemplate {
            id: Uuid::new_v4().to_string(),
            template_key: normalized_key,
            event_type: normalized_event_type,
            locale: normalized_locale,
            subject_template: input.subject_template,
            text_template: input.text_template,
            html_template: input.html_template,
            variables: input.variables.unwrap_or_else(|| serde_json::json!([])),
            variable_bindings: input
                .variable_bindings
                .unwrap_or_else(|| serde_json::json!([])),
            is_active: input.is_active.unwrap_or(true),
            metadata: input.metadata.unwrap_or_else(|| serde_json::json!({})),
            created_at: now,
            updated_at: now,
        };
        templates.insert(row.id.clone(), row.clone());
        Ok(row)
    }

    async fn list_email_templates(
        &self,
        params: ListEmailTemplatesParams,
    ) -> AuthResult<Vec<EmailTemplate>> {
        let templates = self
            .email_templates
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        let limit = params.limit.unwrap_or(100).min(1000);
        let offset = params.offset.unwrap_or(0);
        let template_key_filter = params
            .template_key
            .map(|value| value.trim().to_string())
            .filter(|value| !value.is_empty());
        let event_type_filter = params
            .event_type
            .map(|value| value.trim().to_string())
            .filter(|value| !value.is_empty());
        let locale_filter = params
            .locale
            .map(|value| value.trim().to_ascii_lowercase())
            .filter(|value| !value.is_empty());

        let mut rows: Vec<EmailTemplate> = templates
            .values()
            .filter(|template| {
                if let Some(ref template_key) = template_key_filter
                    && &template.template_key != template_key
                {
                    return false;
                }

                if let Some(ref event_type) = event_type_filter
                    && template.event_type.as_deref() != Some(event_type.as_str())
                {
                    return false;
                }

                if let Some(ref locale) = locale_filter
                    && &template.locale != locale
                {
                    return false;
                }

                if let Some(is_active) = params.is_active
                    && template.is_active != is_active
                {
                    return false;
                }

                true
            })
            .cloned()
            .collect();

        rows.sort_by(|a, b| b.updated_at.cmp(&a.updated_at));
        Ok(rows.into_iter().skip(offset).take(limit).collect())
    }

    async fn get_email_template_by_id(&self, id: &str) -> AuthResult<Option<EmailTemplate>> {
        let templates = self
            .email_templates
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        Ok(templates.get(id).cloned())
    }

    async fn count_email_templates(&self, params: ListEmailTemplatesParams) -> AuthResult<usize> {
        let templates = self
            .email_templates
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        let template_key_filter = params
            .template_key
            .as_deref()
            .map(str::trim)
            .map(str::to_string)
            .filter(|value| !value.is_empty());
        let event_type_filter = params
            .event_type
            .as_deref()
            .map(str::trim)
            .map(str::to_string)
            .filter(|value| !value.is_empty());
        let locale_filter = params
            .locale
            .as_deref()
            .map(str::trim)
            .map(str::to_ascii_lowercase)
            .filter(|value| !value.is_empty());

        let total = templates
            .values()
            .filter(|template| {
                if let Some(ref template_key) = template_key_filter
                    && &template.template_key != template_key
                {
                    return false;
                }

                if let Some(ref event_type) = event_type_filter
                    && template.event_type.as_deref() != Some(event_type.as_str())
                {
                    return false;
                }

                if let Some(ref locale) = locale_filter
                    && &template.locale != locale
                {
                    return false;
                }

                if let Some(is_active) = params.is_active
                    && template.is_active != is_active
                {
                    return false;
                }

                true
            })
            .count();

        Ok(total)
    }

    async fn update_email_template(
        &self,
        id: &str,
        update: UpdateEmailTemplate,
    ) -> AuthResult<EmailTemplate> {
        let mut templates = self
            .email_templates
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        let event_types = self
            .email_event_types
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        let current = templates
            .get(id)
            .cloned()
            .ok_or_else(|| AuthError::not_found("Email template not found"))?;

        let updated_template_key = update
            .template_key
            .map(|value| value.trim().to_string())
            .filter(|value| !value.is_empty())
            .unwrap_or(current.template_key.clone());
        let updated_event_type = match update.event_type {
            Some(Some(value)) => {
                let normalized = value.trim().to_string();
                if normalized.is_empty() {
                    None
                } else {
                    Some(normalized)
                }
            }
            Some(None) => None,
            None => current.event_type.clone(),
        };
        let updated_locale = update
            .locale
            .map(|value| value.trim().to_ascii_lowercase())
            .filter(|value| !value.is_empty())
            .unwrap_or(current.locale.clone());

        let duplicate_exists = templates.values().any(|template| {
            template.id != id
                && template.template_key == updated_template_key
                && template.locale == updated_locale
        });
        if duplicate_exists {
            return Err(AuthError::conflict(
                "Email template with this templateKey and locale already exists",
            ));
        }

        if let Some(event_type) = updated_event_type.as_deref()
            && !event_types.contains_key(event_type)
        {
            return Err(AuthError::bad_request(format!(
                "Unknown eventType '{event_type}'"
            )));
        }

        let template = templates
            .get_mut(id)
            .ok_or_else(|| AuthError::not_found("Email template not found"))?;

        template.template_key = updated_template_key;
        template.event_type = updated_event_type;
        template.locale = updated_locale;
        if let Some(subject_template) = update.subject_template {
            template.subject_template = subject_template;
        }
        if let Some(text_template) = update.text_template {
            template.text_template = text_template;
        }
        if let Some(html_template) = update.html_template {
            template.html_template = html_template;
        }
        if let Some(variables) = update.variables {
            template.variables = variables;
        }
        if let Some(variable_bindings) = update.variable_bindings {
            template.variable_bindings = variable_bindings;
        }
        if let Some(is_active) = update.is_active {
            template.is_active = is_active;
        }
        if let Some(metadata) = update.metadata {
            template.metadata = metadata;
        }
        template.updated_at = Utc::now();

        Ok(template.clone())
    }

    async fn delete_email_template(&self, id: &str) -> AuthResult<()> {
        let mut templates = self
            .email_templates
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        if templates.remove(id).is_none() {
            return Err(AuthError::not_found("Email template not found"));
        }
        Ok(())
    }

    async fn list_email_event_types(
        &self,
        params: ListEmailEventTypesParams,
    ) -> AuthResult<Vec<EmailEventType>> {
        let event_types = self
            .email_event_types
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        let limit = params.limit.unwrap_or(100).min(1000);
        let offset = params.offset.unwrap_or(0);
        let event_type_filter = params
            .event_type
            .map(|value| value.trim().to_string())
            .filter(|value| !value.is_empty());
        let category_filter = params
            .category
            .map(|value| value.trim().to_string())
            .filter(|value| !value.is_empty());

        let mut rows: Vec<EmailEventType> = event_types
            .values()
            .filter(|row| {
                if let Some(ref event_type) = event_type_filter
                    && &row.event_type != event_type
                {
                    return false;
                }

                if let Some(ref category) = category_filter
                    && &row.category != category
                {
                    return false;
                }

                if let Some(is_active) = params.is_active
                    && row.is_active != is_active
                {
                    return false;
                }

                if let Some(is_system) = params.is_system
                    && row.is_system != is_system
                {
                    return false;
                }

                true
            })
            .cloned()
            .collect();

        rows.sort_by(|a, b| a.event_type.cmp(&b.event_type));
        Ok(rows.into_iter().skip(offset).take(limit).collect())
    }

    async fn count_email_event_types(
        &self,
        params: ListEmailEventTypesParams,
    ) -> AuthResult<usize> {
        let event_types = self
            .email_event_types
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        let event_type_filter = params
            .event_type
            .as_deref()
            .map(str::trim)
            .map(str::to_string)
            .filter(|value| !value.is_empty());
        let category_filter = params
            .category
            .as_deref()
            .map(str::trim)
            .map(str::to_string)
            .filter(|value| !value.is_empty());

        let total = event_types
            .values()
            .filter(|row| {
                if let Some(ref event_type) = event_type_filter
                    && &row.event_type != event_type
                {
                    return false;
                }

                if let Some(ref category) = category_filter
                    && &row.category != category
                {
                    return false;
                }

                if let Some(is_active) = params.is_active
                    && row.is_active != is_active
                {
                    return false;
                }

                if let Some(is_system) = params.is_system
                    && row.is_system != is_system
                {
                    return false;
                }

                true
            })
            .count();

        Ok(total)
    }

    async fn get_email_event_type(&self, event_type: &str) -> AuthResult<Option<EmailEventType>> {
        let event_types = self
            .email_event_types
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        Ok(event_types.get(event_type).cloned())
    }
}

// -- OrganizationOps --

#[async_trait]
impl<U, S, A, O, M, I, V, P> OrganizationOps for MemoryDatabaseAdapter<U, S, A, O, M, I, V, P>
where
    U: MemoryUser,
    S: MemorySession,
    A: MemoryAccount,
    O: MemoryOrganization,
    M: MemoryMember,
    I: MemoryInvitation,
    V: MemoryVerification,
    P: MemoryPasskey,
{
    type Organization = O;

    async fn create_organization(&self, create_org: CreateOrganization) -> AuthResult<O> {
        let mut organizations = self
            .organizations
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        let mut slug_index = self
            .slug_index
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());

        if slug_index.contains_key(&create_org.slug) {
            return Err(AuthError::conflict("Organization slug already exists"));
        }

        let id = create_org
            .id
            .clone()
            .unwrap_or_else(|| Uuid::new_v4().to_string());
        let now = Utc::now();
        let organization = O::from_create(id.clone(), &create_org, now);

        organizations.insert(id.clone(), organization.clone());
        slug_index.insert(create_org.slug.clone(), id);

        Ok(organization)
    }

    async fn get_organization_by_id(&self, id: &str) -> AuthResult<Option<O>> {
        let organizations = self
            .organizations
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        Ok(organizations.get(id).cloned())
    }

    async fn get_organization_by_slug(&self, slug: &str) -> AuthResult<Option<O>> {
        let slug_index = self
            .slug_index
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        let organizations = self
            .organizations
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());

        if let Some(org_id) = slug_index.get(slug) {
            Ok(organizations.get(org_id).cloned())
        } else {
            Ok(None)
        }
    }

    async fn update_organization(&self, id: &str, update: UpdateOrganization) -> AuthResult<O> {
        let mut organizations = self
            .organizations
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        let mut slug_index = self
            .slug_index
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());

        let org = organizations
            .get_mut(id)
            .ok_or_else(|| AuthError::not_found("Organization not found"))?;

        // Update slug index BEFORE mutation
        if let Some(new_slug) = &update.slug {
            let current_slug = org.slug().to_string();
            if *new_slug != current_slug {
                if slug_index.contains_key(new_slug.as_str()) {
                    return Err(AuthError::conflict("Organization slug already exists"));
                }
                slug_index.remove(&current_slug);
                slug_index.insert(new_slug.clone(), id.to_string());
            }
        }

        org.apply_update(&update);
        Ok(org.clone())
    }

    async fn delete_organization(&self, id: &str) -> AuthResult<()> {
        let mut organizations = self
            .organizations
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        let mut slug_index = self
            .slug_index
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        let mut members = self
            .members
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        let mut invitations = self
            .invitations
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());

        if let Some(org) = organizations.remove(id) {
            slug_index.remove(org.slug());
        }

        members.retain(|_, m| m.organization_id() != id);
        invitations.retain(|_, i| i.organization_id() != id);

        Ok(())
    }

    async fn list_user_organizations(&self, user_id: &str) -> AuthResult<Vec<O>> {
        let members = self
            .members
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        let organizations = self
            .organizations
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());

        let org_ids: Vec<String> = members
            .values()
            .filter(|m| m.user_id() == user_id)
            .map(|m| m.organization_id().to_string())
            .collect();

        let orgs = org_ids
            .iter()
            .filter_map(|id| organizations.get(id).cloned())
            .collect();

        Ok(orgs)
    }
}

// -- MemberOps --

#[async_trait]
impl<U, S, A, O, M, I, V, P> MemberOps for MemoryDatabaseAdapter<U, S, A, O, M, I, V, P>
where
    U: MemoryUser,
    S: MemorySession,
    A: MemoryAccount,
    O: MemoryOrganization,
    M: MemoryMember,
    I: MemoryInvitation,
    V: MemoryVerification,
    P: MemoryPasskey,
{
    type Member = M;

    async fn create_member(&self, create_member: CreateMember) -> AuthResult<M> {
        let mut members = self
            .members
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());

        let exists = members.values().any(|m| {
            m.organization_id() == create_member.organization_id
                && m.user_id() == create_member.user_id
        });

        if exists {
            return Err(AuthError::conflict(
                "User is already a member of this organization",
            ));
        }

        let id = Uuid::new_v4().to_string();
        let now = Utc::now();
        let member = M::from_create(id.clone(), &create_member, now);

        members.insert(id, member.clone());
        Ok(member)
    }

    async fn get_member(&self, organization_id: &str, user_id: &str) -> AuthResult<Option<M>> {
        let members = self
            .members
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        Ok(members
            .values()
            .find(|m| m.organization_id() == organization_id && m.user_id() == user_id)
            .cloned())
    }

    async fn get_member_by_id(&self, id: &str) -> AuthResult<Option<M>> {
        let members = self
            .members
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        Ok(members.get(id).cloned())
    }

    async fn update_member_role(&self, member_id: &str, role: &str) -> AuthResult<M> {
        let mut members = self
            .members
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        let member = members
            .get_mut(member_id)
            .ok_or_else(|| AuthError::not_found("Member not found"))?;
        member.set_role(role.to_string());
        Ok(member.clone())
    }

    async fn delete_member(&self, member_id: &str) -> AuthResult<()> {
        let mut members = self
            .members
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        members.remove(member_id);
        Ok(())
    }

    async fn list_organization_members(&self, organization_id: &str) -> AuthResult<Vec<M>> {
        let members = self
            .members
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        Ok(members
            .values()
            .filter(|m| m.organization_id() == organization_id)
            .cloned()
            .collect())
    }

    async fn count_organization_members(&self, organization_id: &str) -> AuthResult<usize> {
        let members = self
            .members
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        Ok(members
            .values()
            .filter(|m| m.organization_id() == organization_id)
            .count())
    }

    async fn count_organization_owners(&self, organization_id: &str) -> AuthResult<usize> {
        let members = self
            .members
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        Ok(members
            .values()
            .filter(|m| m.organization_id() == organization_id && m.role() == "owner")
            .count())
    }
}

// -- InvitationOps --

#[async_trait]
impl<U, S, A, O, M, I, V, P> InvitationOps for MemoryDatabaseAdapter<U, S, A, O, M, I, V, P>
where
    U: MemoryUser,
    S: MemorySession,
    A: MemoryAccount,
    O: MemoryOrganization,
    M: MemoryMember,
    I: MemoryInvitation,
    V: MemoryVerification,
    P: MemoryPasskey,
{
    type Invitation = I;

    async fn create_invitation(&self, create_inv: CreateInvitation) -> AuthResult<I> {
        let mut invitations = self
            .invitations
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());

        let id = Uuid::new_v4().to_string();
        let now = Utc::now();
        let invitation = I::from_create(id.clone(), &create_inv, now);

        invitations.insert(id, invitation.clone());
        Ok(invitation)
    }

    async fn get_invitation_by_id(&self, id: &str) -> AuthResult<Option<I>> {
        let invitations = self
            .invitations
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        Ok(invitations.get(id).cloned())
    }

    async fn get_pending_invitation(
        &self,
        organization_id: &str,
        email: &str,
    ) -> AuthResult<Option<I>> {
        let invitations = self
            .invitations
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        Ok(invitations
            .values()
            .find(|i| {
                i.organization_id() == organization_id
                    && i.email().to_lowercase() == email.to_lowercase()
                    && *i.status() == InvitationStatus::Pending
            })
            .cloned())
    }

    async fn update_invitation_status(&self, id: &str, status: InvitationStatus) -> AuthResult<I> {
        let mut invitations = self
            .invitations
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        let invitation = invitations
            .get_mut(id)
            .ok_or_else(|| AuthError::not_found("Invitation not found"))?;
        invitation.set_status(status);
        Ok(invitation.clone())
    }

    async fn list_organization_invitations(&self, organization_id: &str) -> AuthResult<Vec<I>> {
        let invitations = self
            .invitations
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        Ok(invitations
            .values()
            .filter(|i| i.organization_id() == organization_id)
            .cloned()
            .collect())
    }

    async fn list_user_invitations(&self, email: &str) -> AuthResult<Vec<I>> {
        let invitations = self
            .invitations
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        let now = Utc::now();
        Ok(invitations
            .values()
            .filter(|i| {
                i.email().to_lowercase() == email.to_lowercase()
                    && *i.status() == InvitationStatus::Pending
                    && i.expires_at() > now
            })
            .cloned()
            .collect())
    }
}

// -- TwoFactorOps --

#[async_trait]
impl<U, S, A, O, M, I, V, P> TwoFactorOps for MemoryDatabaseAdapter<U, S, A, O, M, I, V, P>
where
    U: MemoryUser,
    S: MemorySession,
    A: MemoryAccount,
    O: MemoryOrganization,
    M: MemoryMember,
    I: MemoryInvitation,
    V: MemoryVerification,
    P: MemoryPasskey,
{
    type TwoFactor = TwoFactor;

    async fn create_two_factor(&self, create: CreateTwoFactor) -> AuthResult<TwoFactor> {
        let mut two_factors = self
            .two_factors
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());

        // Check if user already has 2FA
        if two_factors.values().any(|tf| tf.user_id == create.user_id) {
            return Err(AuthError::conflict(
                "Two-factor authentication already enabled for this user",
            ));
        }

        let id = Uuid::new_v4().to_string();
        let now = Utc::now();
        let two_factor: TwoFactor = MemoryTwoFactor::from_create(id.clone(), &create, now);

        two_factors.insert(id, two_factor.clone());
        Ok(two_factor)
    }

    async fn get_two_factor_by_user_id(&self, user_id: &str) -> AuthResult<Option<TwoFactor>> {
        let two_factors = self
            .two_factors
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        Ok(two_factors
            .values()
            .find(|tf| tf.user_id == user_id)
            .cloned())
    }

    async fn update_two_factor_backup_codes(
        &self,
        user_id: &str,
        backup_codes: &str,
    ) -> AuthResult<TwoFactor> {
        let mut two_factors = self
            .two_factors
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        let two_factor = two_factors
            .values_mut()
            .find(|tf| tf.user_id == user_id)
            .ok_or_else(|| AuthError::not_found("Two-factor record not found"))?;
        two_factor.set_backup_codes(backup_codes.to_string());
        Ok(two_factor.clone())
    }

    async fn delete_two_factor(&self, user_id: &str) -> AuthResult<()> {
        let mut two_factors = self
            .two_factors
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        two_factors.retain(|_, tf| tf.user_id != user_id);
        Ok(())
    }
}

// -- ApiKeyOps --

#[async_trait]
impl<U, S, A, O, M, I, V, P> ApiKeyOps for MemoryDatabaseAdapter<U, S, A, O, M, I, V, P>
where
    U: MemoryUser,
    S: MemorySession,
    A: MemoryAccount,
    O: MemoryOrganization,
    M: MemoryMember,
    I: MemoryInvitation,
    V: MemoryVerification,
    P: MemoryPasskey,
{
    type ApiKey = ApiKey;

    async fn create_api_key(&self, input: CreateApiKey) -> AuthResult<ApiKey> {
        let mut api_keys = self
            .api_keys
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());

        if api_keys.values().any(|k| k.key_hash == input.key_hash) {
            return Err(AuthError::conflict("API key already exists"));
        }

        let id = Uuid::new_v4().to_string();
        let now = Utc::now();
        let api_key: ApiKey = MemoryApiKey::from_create(id.clone(), &input, now);

        api_keys.insert(id, api_key.clone());
        Ok(api_key)
    }

    async fn get_api_key_by_id(&self, id: &str) -> AuthResult<Option<ApiKey>> {
        let api_keys = self
            .api_keys
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        Ok(api_keys.get(id).cloned())
    }

    async fn get_api_key_by_hash(&self, hash: &str) -> AuthResult<Option<ApiKey>> {
        let api_keys = self
            .api_keys
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        Ok(api_keys.values().find(|k| k.key_hash == hash).cloned())
    }

    async fn list_api_keys_by_user(&self, user_id: &str) -> AuthResult<Vec<ApiKey>> {
        let api_keys = self
            .api_keys
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        let mut keys: Vec<ApiKey> = api_keys
            .values()
            .filter(|k| k.user_id == user_id)
            .cloned()
            .collect();
        keys.sort_by(|a, b| b.created_at.cmp(&a.created_at));
        Ok(keys)
    }

    async fn update_api_key(&self, id: &str, update: UpdateApiKey) -> AuthResult<ApiKey> {
        let mut api_keys = self
            .api_keys
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        let api_key = api_keys
            .get_mut(id)
            .ok_or_else(|| AuthError::not_found("API key not found"))?;
        api_key.apply_update(&update);
        Ok(api_key.clone())
    }

    async fn delete_api_key(&self, id: &str) -> AuthResult<()> {
        let mut api_keys = self
            .api_keys
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        api_keys.remove(id);
        Ok(())
    }

    async fn delete_expired_api_keys(&self) -> AuthResult<usize> {
        let mut api_keys = self
            .api_keys
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        let now = Utc::now();
        let initial_count = api_keys.len();
        api_keys.retain(|_, k| {
            if let Some(expires_at) = &k.expires_at
                && let Ok(exp) = chrono::DateTime::parse_from_rfc3339(expires_at)
            {
                return exp > now;
            }
            true // keep keys without expiration
        });
        Ok(initial_count - api_keys.len())
    }
}

// -- PasskeyOps --

#[async_trait]
impl<U, S, A, O, M, I, V, P> PasskeyOps for MemoryDatabaseAdapter<U, S, A, O, M, I, V, P>
where
    U: MemoryUser,
    S: MemorySession,
    A: MemoryAccount,
    O: MemoryOrganization,
    M: MemoryMember,
    I: MemoryInvitation,
    V: MemoryVerification,
    P: MemoryPasskey,
{
    type Passkey = P;

    async fn create_passkey(&self, input: CreatePasskey) -> AuthResult<P> {
        let mut credential_index = self
            .passkey_credential_index
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        let mut passkeys = self
            .passkeys
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());

        if credential_index.contains_key(&input.credential_id) {
            return Err(AuthError::conflict(
                "A passkey with this credential ID already exists",
            ));
        }

        let id = Uuid::new_v4().to_string();
        let now = Utc::now();
        let passkey = P::from_create(id.clone(), &input, now);

        credential_index.insert(input.credential_id.clone(), id.clone());
        passkeys.insert(id, passkey.clone());
        Ok(passkey)
    }

    async fn get_passkey_by_id(&self, id: &str) -> AuthResult<Option<P>> {
        let passkeys = self
            .passkeys
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        Ok(passkeys.get(id).cloned())
    }

    async fn get_passkey_by_credential_id(&self, credential_id: &str) -> AuthResult<Option<P>> {
        let passkey_id = {
            let credential_index = self
                .passkey_credential_index
                .lock()
                .unwrap_or_else(|poisoned| poisoned.into_inner());
            credential_index.get(credential_id).cloned()
        };

        let passkeys = self
            .passkeys
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());

        if let Some(id) = passkey_id {
            Ok(passkeys.get(&id).cloned())
        } else {
            Ok(None)
        }
    }

    async fn list_passkeys_by_user(&self, user_id: &str) -> AuthResult<Vec<P>> {
        let passkeys = self
            .passkeys
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        let mut matched: Vec<P> = passkeys
            .values()
            .filter(|p| p.user_id() == user_id)
            .cloned()
            .collect();
        matched.sort_by_key(|p| std::cmp::Reverse(p.created_at()));
        Ok(matched)
    }

    async fn update_passkey_counter(&self, id: &str, counter: u64) -> AuthResult<P> {
        let mut passkeys = self
            .passkeys
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        let passkey = passkeys
            .get_mut(id)
            .ok_or_else(|| AuthError::not_found("Passkey not found"))?;
        passkey.set_counter(counter);
        Ok(passkey.clone())
    }

    async fn update_passkey_name(&self, id: &str, name: &str) -> AuthResult<P> {
        let mut passkeys = self
            .passkeys
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        let passkey = passkeys
            .get_mut(id)
            .ok_or_else(|| AuthError::not_found("Passkey not found"))?;
        passkey.set_name(name.to_string());
        Ok(passkey.clone())
    }

    async fn delete_passkey(&self, id: &str) -> AuthResult<()> {
        let mut credential_index = self
            .passkey_credential_index
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        let mut passkeys = self
            .passkeys
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());

        if let Some(passkey) = passkeys.remove(id) {
            credential_index.remove(passkey.credential_id());
        }
        Ok(())
    }
}
