use std::collections::HashMap;
use std::time::Duration;

use config::oauth::OAuthConfig;
use tonic::metadata::{Ascii, MetadataValue};
use tonic::service::Interceptor;
use tonic::service::interceptor::InterceptedService;
use tonic::transport::{Channel, Endpoint};
use tonic::{Request, Status};
use uuid::Uuid;

use crate::error::KenAuthError;
use crate::proto;

#[derive(Clone)]
pub struct RpcInterceptor {
    key: Option<MetadataValue<Ascii>>,
}

impl RpcInterceptor {
    pub fn new(rpc_key: &str) -> Result<Self, KenAuthError> {
        let key = if rpc_key.is_empty() {
            None
        } else {
            Some(
                MetadataValue::try_from(rpc_key)
                    .map_err(|_| KenAuthError::Config("rpc_key is not ASCII".into()))?,
            )
        };
        Ok(Self { key })
    }
}

impl Interceptor for RpcInterceptor {
    fn call(&mut self, mut request: Request<()>) -> Result<Request<()>, Status> {
        if let Some(key) = &self.key {
            request
                .metadata_mut()
                .insert(crate::RPC_KEY_HEADER, key.clone());
        }
        if let Ok(value) = MetadataValue::try_from(Uuid::now_v7().to_string()) {
            request
                .metadata_mut()
                .insert(crate::REQUEST_ID_HEADER, value);
        }
        Ok(request)
    }
}

type Intercepted = InterceptedService<Channel, RpcInterceptor>;

#[derive(Clone)]
pub struct KenAuthClient {
    channel: Channel,
    interceptor: RpcInterceptor,
    read_timeout: Duration,
    write_timeout: Duration,
}

impl KenAuthClient {
    pub fn connect(config: &OAuthConfig) -> Result<Self, KenAuthError> {
        if !config.enabled {
            return Err(KenAuthError::Config("oauth.enabled is false".into()));
        }
        Self::connect_rpc(
            &config.grpc_endpoint,
            &config.rpc_key,
            config.http_timeout_ms,
        )
    }

    pub fn connect_rpc(
        grpc_endpoint: &str,
        rpc_key: &str,
        http_timeout_ms: u64,
    ) -> Result<Self, KenAuthError> {
        if grpc_endpoint.is_empty() {
            return Err(KenAuthError::Config("grpc endpoint is empty".into()));
        }
        let timeout = Duration::from_millis(http_timeout_ms.max(200));
        let endpoint = Endpoint::from_shared(grpc_endpoint.to_string())
            .map_err(|err| KenAuthError::Config(err.to_string()))?
            .connect_timeout(timeout)
            .timeout(timeout);
        let channel = endpoint.connect_lazy();
        Ok(Self {
            channel,
            interceptor: RpcInterceptor::new(rpc_key)?,
            read_timeout: Duration::from_millis(300),
            write_timeout: timeout.max(Duration::from_secs(2)),
        })
    }

    pub fn channel(&self) -> Channel {
        self.channel.clone()
    }

    pub async fn check_health(&self) -> Result<(), KenAuthError> {
        let mut client = tonic_health::pb::health_client::HealthClient::with_interceptor(
            self.channel.clone(),
            self.interceptor.clone(),
        );
        let mut request = Request::new(tonic_health::pb::HealthCheckRequest {
            service: String::new(),
        });
        request.set_timeout(Duration::from_secs(2));
        client
            .check(request)
            .await
            .map_err(KenAuthError::from_status)?;
        Ok(())
    }

    pub fn identity(
        &self,
    ) -> proto::v1::identity_service_client::IdentityServiceClient<Intercepted> {
        proto::v1::identity_service_client::IdentityServiceClient::with_interceptor(
            self.channel.clone(),
            self.interceptor.clone(),
        )
    }

    pub fn authority(
        &self,
    ) -> proto::internal::v1::authority_service_client::AuthorityServiceClient<Intercepted> {
        proto::internal::v1::authority_service_client::AuthorityServiceClient::with_interceptor(
            self.channel.clone(),
            self.interceptor.clone(),
        )
    }

    pub fn email(
        &self,
    ) -> proto::internal::v1::email_service_client::EmailServiceClient<Intercepted> {
        proto::internal::v1::email_service_client::EmailServiceClient::with_interceptor(
            self.channel.clone(),
            self.interceptor.clone(),
        )
    }

    pub fn session(&self) -> proto::v1::session_service_client::SessionServiceClient<Intercepted> {
        proto::v1::session_service_client::SessionServiceClient::with_interceptor(
            self.channel.clone(),
            self.interceptor.clone(),
        )
    }

    pub fn organization(
        &self,
    ) -> proto::v1::organization_service_client::OrganizationServiceClient<Intercepted> {
        proto::v1::organization_service_client::OrganizationServiceClient::with_interceptor(
            self.channel.clone(),
            self.interceptor.clone(),
        )
    }

    pub fn token(&self) -> proto::v1::token_service_client::TokenServiceClient<Intercepted> {
        proto::v1::token_service_client::TokenServiceClient::with_interceptor(
            self.channel.clone(),
            self.interceptor.clone(),
        )
    }

    pub fn authorization(
        &self,
    ) -> proto::v1::authorization_service_client::AuthorizationServiceClient<Intercepted> {
        proto::v1::authorization_service_client::AuthorizationServiceClient::with_interceptor(
            self.channel.clone(),
            self.interceptor.clone(),
        )
    }

    pub fn access_key(
        &self,
    ) -> proto::v1::access_key_service_client::AccessKeyServiceClient<Intercepted> {
        proto::v1::access_key_service_client::AccessKeyServiceClient::with_interceptor(
            self.channel.clone(),
            self.interceptor.clone(),
        )
    }

    pub fn ssh_key(&self) -> proto::v1::ssh_key_service_client::SshKeyServiceClient<Intercepted> {
        proto::v1::ssh_key_service_client::SshKeyServiceClient::with_interceptor(
            self.channel.clone(),
            self.interceptor.clone(),
        )
    }

    pub fn mfa(&self) -> proto::v1::mfa_service_client::MfaServiceClient<Intercepted> {
        proto::v1::mfa_service_client::MfaServiceClient::with_interceptor(
            self.channel.clone(),
            self.interceptor.clone(),
        )
    }

    pub fn agreement(
        &self,
    ) -> proto::v1::agreement_service_client::AgreementServiceClient<Intercepted> {
        proto::v1::agreement_service_client::AgreementServiceClient::with_interceptor(
            self.channel.clone(),
            self.interceptor.clone(),
        )
    }

    #[allow(clippy::too_many_arguments)]
    pub async fn change_password(
        &self,
        user_id: i64,
        old_password_hash: &str,
        new_password_hash: &str,
        keep_session_id: &str,
        client_ip: &str,
    ) -> Result<proto::v1::ChangePasswordResponse, KenAuthError> {
        let mut client = self.identity();
        client
            .change_password(self.write_request(proto::v1::ChangePasswordRequest {
                user_id,
                old_password_hash: old_password_hash.to_string(),
                new_password_hash: new_password_hash.to_string(),
                keep_session_id: keep_session_id.to_string(),
                client_ip: client_ip.to_string(),
            }))
            .await
            .map_err(KenAuthError::from_status)
            .map(|response| response.into_inner())
    }

    pub async fn get_mfa_status(
        &self,
        user_id: i64,
    ) -> Result<proto::v1::GetMfaStatusResponse, KenAuthError> {
        let mut client = self.mfa();
        client
            .get_mfa_status(self.read_request(proto::v1::GetMfaStatusRequest { user_id }))
            .await
            .map_err(KenAuthError::from_status)
            .map(|response| response.into_inner())
    }

    pub async fn configure_totp(
        &self,
        user_id: i64,
        name: &str,
        client_ip: &str,
    ) -> Result<proto::v1::ConfigureTotpResponse, KenAuthError> {
        let mut client = self.mfa();
        client
            .configure_totp(self.write_request(proto::v1::ConfigureTotpRequest {
                user_id,
                name: name.to_string(),
                client_ip: client_ip.to_string(),
            }))
            .await
            .map_err(KenAuthError::from_status)
            .map(|response| response.into_inner())
    }

    pub async fn enable_totp(
        &self,
        user_id: i64,
        code: &str,
        password_hash: &str,
        client_ip: &str,
    ) -> Result<proto::v1::EnableTotpResponse, KenAuthError> {
        let mut client = self.mfa();
        client
            .enable_totp(self.write_request(proto::v1::EnableTotpRequest {
                user_id,
                code: code.to_string(),
                password_hash: password_hash.to_string(),
                client_ip: client_ip.to_string(),
            }))
            .await
            .map_err(KenAuthError::from_status)
            .map(|response| response.into_inner())
    }

    pub async fn disable_totp(
        &self,
        user_id: i64,
        code: &str,
        client_ip: &str,
    ) -> Result<proto::v1::DisableTotpResponse, KenAuthError> {
        let mut client = self.mfa();
        client
            .disable_totp(self.write_request(proto::v1::DisableTotpRequest {
                user_id,
                code: code.to_string(),
                client_ip: client_ip.to_string(),
            }))
            .await
            .map_err(KenAuthError::from_status)
            .map(|response| response.into_inner())
    }

    pub async fn regenerate_recovery_codes(
        &self,
        user_id: i64,
        password_hash: &str,
        client_ip: &str,
    ) -> Result<proto::v1::RegenerateRecoveryCodesResponse, KenAuthError> {
        let mut client = self.mfa();
        client
            .regenerate_recovery_codes(self.write_request(
                proto::v1::RegenerateRecoveryCodesRequest {
                    user_id,
                    password_hash: password_hash.to_string(),
                    client_ip: client_ip.to_string(),
                },
            ))
            .await
            .map_err(KenAuthError::from_status)
            .map(|response| response.into_inner())
    }

    pub async fn list_access_keys(
        &self,
        user_id: i64,
        include_revoked: bool,
        page: proto::common::v1::PageRequest,
    ) -> Result<proto::v1::ListAccessKeysResponse, KenAuthError> {
        let mut client = self.access_key();
        client
            .list_access_keys(self.read_request(proto::v1::ListAccessKeysRequest {
                user_id,
                include_revoked,
                page: Some(page),
            }))
            .await
            .map_err(KenAuthError::from_status)
            .map(|response| response.into_inner())
    }

    #[allow(clippy::too_many_arguments)]
    pub async fn create_access_key(
        &self,
        user_id: i64,
        name: &str,
        description: &str,
        scopes: Vec<i32>,
        expires_in_secs: i64,
        client_ip: &str,
    ) -> Result<proto::v1::CreateAccessKeyResponse, KenAuthError> {
        let mut client = self.access_key();
        client
            .create_access_key(self.write_request(proto::v1::CreateAccessKeyRequest {
                user_id,
                name: name.to_string(),
                description: description.to_string(),
                scopes,
                expires_in_secs,
                client_ip: client_ip.to_string(),
            }))
            .await
            .map_err(KenAuthError::from_status)
            .map(|response| response.into_inner())
    }

    pub async fn rotate_access_key(
        &self,
        user_id: i64,
        access_key_id: &str,
        client_ip: &str,
    ) -> Result<proto::v1::RotateAccessKeyResponse, KenAuthError> {
        let mut client = self.access_key();
        client
            .rotate_access_key(self.write_request(proto::v1::RotateAccessKeyRequest {
                user_id,
                access_key_id: access_key_id.to_string(),
                client_ip: client_ip.to_string(),
            }))
            .await
            .map_err(KenAuthError::from_status)
            .map(|response| response.into_inner())
    }

    pub async fn revoke_access_key(
        &self,
        user_id: i64,
        access_key_id: &str,
        reason: &str,
    ) -> Result<proto::v1::RevokeAccessKeyResponse, KenAuthError> {
        let mut client = self.access_key();
        client
            .revoke_access_key(self.write_request(proto::v1::RevokeAccessKeyRequest {
                user_id,
                access_key_id: access_key_id.to_string(),
                reason: reason.to_string(),
            }))
            .await
            .map_err(KenAuthError::from_status)
            .map(|response| response.into_inner())
    }

    pub async fn list_user_ssh_keys(
        &self,
        user_id: i64,
        include_revoked: bool,
        page: proto::common::v1::PageRequest,
    ) -> Result<proto::v1::ListUserSshKeysResponse, KenAuthError> {
        let mut client = self.ssh_key();
        client
            .list_user_ssh_keys(self.read_request(proto::v1::ListUserSshKeysRequest {
                user_id,
                include_revoked,
                page: Some(page),
            }))
            .await
            .map_err(KenAuthError::from_status)
            .map(|response| response.into_inner())
    }

    pub async fn add_ssh_key(
        &self,
        user_id: i64,
        name: &str,
        public_key: &str,
        description: &str,
        client_ip: &str,
    ) -> Result<proto::v1::AddSshKeyResponse, KenAuthError> {
        let mut client = self.ssh_key();
        client
            .add_ssh_key(self.write_request(proto::v1::AddSshKeyRequest {
                user_id,
                name: name.to_string(),
                public_key: public_key.to_string(),
                description: description.to_string(),
                client_ip: client_ip.to_string(),
            }))
            .await
            .map_err(KenAuthError::from_status)
            .map(|response| response.into_inner())
    }

    pub async fn revoke_ssh_key(
        &self,
        user_id: i64,
        key_id: &str,
        reason: &str,
        bump_security_version: bool,
    ) -> Result<proto::v1::RevokeSshKeyResponse, KenAuthError> {
        let mut client = self.ssh_key();
        client
            .revoke_ssh_key(self.write_request(proto::v1::RevokeSshKeyRequest {
                user_id,
                key_id: key_id.to_string(),
                reason: reason.to_string(),
                bump_security_version,
            }))
            .await
            .map_err(KenAuthError::from_status)
            .map(|response| response.into_inner())
    }

    pub async fn list_user_sessions(
        &self,
        user_id: i64,
        include_revoked: bool,
        include_expired: bool,
        page: proto::common::v1::PageRequest,
    ) -> Result<proto::v1::ListUserSessionsResponse, KenAuthError> {
        let mut client = self.session();
        client
            .list_user_sessions(self.read_request(proto::v1::ListUserSessionsRequest {
                user_id,
                include_revoked,
                include_expired,
                page: Some(page),
            }))
            .await
            .map_err(KenAuthError::from_status)
            .map(|response| response.into_inner())
    }

    pub async fn revoke_user_sessions(
        &self,
        user_id: i64,
        except_session_id: &str,
        reason: &str,
        bump_security_version: bool,
    ) -> Result<proto::v1::RevokeUserSessionsResponse, KenAuthError> {
        let mut client = self.session();
        client
            .revoke_user_sessions(self.write_request(proto::v1::RevokeUserSessionsRequest {
                user_id,
                except_session_id: except_session_id.to_string(),
                reason: reason.to_string(),
                bump_security_version,
            }))
            .await
            .map_err(KenAuthError::from_status)
            .map(|response| response.into_inner())
    }

    pub fn read_request<T>(&self, message: T) -> Request<T> {
        let mut request = Request::new(message);
        request.set_timeout(self.read_timeout);
        request
    }

    pub fn write_request<T>(&self, message: T) -> Request<T> {
        let mut request = Request::new(message);
        request.set_timeout(self.write_timeout);
        request
    }

    /// Enqueue one plain text / HTML mail on `kenoauth`.
    ///
    /// `kenoauth` owns SMTP, retries and backpressure; a `Ok` here means the
    /// message was accepted into its queue, while delivery outcome is visible
    /// through `GetQueueStatus` / the delivery events on that side.
    pub async fn send_mail(
        &self,
        to: &str,
        subject: &str,
        text: &str,
        html: &str,
    ) -> Result<(), KenAuthError> {
        let mut client = self.email();
        let request = proto::internal::v1::SendEmailRequest {
            request_id: Uuid::now_v7().to_string(),
            to: vec![proto::internal::v1::EmailAddress {
                address: to.to_string(),
                name: String::new(),
            }],
            cc: Vec::new(),
            bcc: Vec::new(),
            subject: subject.to_string(),
            text: text.to_string(),
            html: html.to_string(),
            reply_to: None,
            attachments: Vec::new(),
            metadata: HashMap::new(),
            trace_id: String::new(),
        };
        client
            .send_email(self.write_request(request))
            .await
            .map_err(KenAuthError::from_status)?;
        Ok(())
    }

    pub async fn get_session(
        &self,
        user_id: i64,
        version: i32,
        session_id: &str,
    ) -> Result<proto::internal::v1::AuthorityAccess, KenAuthError> {
        let mut client = self.authority();
        client
            .get_session(self.read_request(proto::internal::v1::GetSessionRequest {
                user_id,
                version,
                session_id: session_id.to_string(),
            }))
            .await
            .map_err(KenAuthError::from_status)
            .map(|response| response.into_inner().access.unwrap_or_default())
    }

    pub async fn batch_get_authority_users(
        &self,
        user_ids: Vec<i64>,
    ) -> Result<Vec<proto::internal::v1::AuthorityUser>, KenAuthError> {
        let mut client = self.authority();
        client
            .batch_get_user(
                self.read_request(proto::internal::v1::BatchGetUserRequest { user_ids }),
            )
            .await
            .map_err(KenAuthError::from_status)
            .map(|response| response.into_inner().users)
    }

    pub async fn get_user_status(
        &self,
        user_id: i64,
    ) -> Result<proto::common::v1::UserStatus, KenAuthError> {
        let mut client = self.identity();
        let user = Some(proto::common::v1::UserRef {
            key: Some(proto::common::v1::user_ref::Key::UserId(user_id)),
        });
        client
            .get_user_status(self.read_request(proto::v1::GetUserStatusRequest { user }))
            .await
            .map_err(KenAuthError::from_status)
            .map(|response| response.into_inner().status.unwrap_or_default())
    }

    pub async fn get_user_public_by_ref(
        &self,
        key: proto::common::v1::user_ref::Key,
    ) -> Result<proto::common::v1::UserPublic, KenAuthError> {
        let mut client = self.identity();
        let user = Some(proto::common::v1::UserRef { key: Some(key) });
        client
            .get_user_public(self.read_request(proto::v1::GetUserPublicRequest { user }))
            .await
            .map_err(KenAuthError::from_status)?
            .into_inner()
            .user
            .ok_or(KenAuthError::NotFound)
    }

    pub async fn get_authority_user(
        &self,
        user_id: i64,
    ) -> Result<proto::internal::v1::AuthorityUser, KenAuthError> {
        let mut client = self.authority();
        client
            .get_user(self.read_request(proto::internal::v1::GetUserRequest { user_id }))
            .await
            .map_err(KenAuthError::from_status)?
            .into_inner()
            .user
            .ok_or(KenAuthError::NotFound)
    }

    /// The only supported write path for `username` / `displayname`.
    /// `version` is the account's current optimistic-lock version.
    #[allow(clippy::too_many_arguments)]
    pub async fn update_authority_user(
        &self,
        user_id: i64,
        username: Option<String>,
        displayname: Option<String>,
        clear_displayname: bool,
        public_show_email: Option<bool>,
        version: i32,
    ) -> Result<proto::internal::v1::AuthorityUser, KenAuthError> {
        let mut client = self.authority();
        client
            .update_user(self.write_request(proto::internal::v1::UpdateUserRequest {
                user_id,
                patch: Some(proto::internal::v1::AuthorityUserPatch {
                    version,
                    username,
                    displayname,
                    clear_displayname,
                    public_show_email,
                }),
            }))
            .await
            .map_err(KenAuthError::from_status)?
            .into_inner()
            .user
            .ok_or(KenAuthError::NotFound)
    }

    pub async fn batch_get_user_public(
        &self,
        user_ids: Vec<i64>,
    ) -> Result<Vec<proto::common::v1::UserPublic>, KenAuthError> {
        let mut client = self.identity();
        client
            .batch_get_user_public(
                self.read_request(proto::v1::BatchGetUserPublicRequest { user_ids }),
            )
            .await
            .map_err(KenAuthError::from_status)
            .map(|response| response.into_inner().users)
    }

    pub async fn get_user_public(
        &self,
        user_id: i64,
    ) -> Result<proto::common::v1::UserPublic, KenAuthError> {
        let mut client = self.identity();
        let user = Some(proto::common::v1::UserRef {
            key: Some(proto::common::v1::user_ref::Key::UserId(user_id)),
        });
        client
            .get_user_public(self.read_request(proto::v1::GetUserPublicRequest { user }))
            .await
            .map_err(KenAuthError::from_status)?
            .into_inner()
            .user
            .ok_or(KenAuthError::NotFound)
    }

    pub async fn revoke_session(
        &self,
        user_id: i64,
        session_id: &str,
        reason: &str,
    ) -> Result<(), KenAuthError> {
        let mut client = self.session();
        client
            .revoke_session(self.write_request(proto::v1::RevokeSessionRequest {
                user_id,
                session_id: session_id.to_string(),
                reason: reason.to_string(),
            }))
            .await
            .map_err(KenAuthError::from_status)?;
        Ok(())
    }

    pub async fn verify_access_key(
        &self,
        token: &str,
        required_scopes: Vec<i32>,
    ) -> Result<Option<proto::common::v1::Principal>, KenAuthError> {
        let mut client = self.access_key();
        let response = client
            .verify_access_key(self.read_request(proto::v1::VerifyAccessKeyRequest {
                access_key: token.to_string(),
                required_scopes,
                client_ip: String::new(),
            }))
            .await
            .map_err(KenAuthError::from_status)?
            .into_inner();
        if !response.valid {
            return Ok(None);
        }
        Ok(response.principal)
    }

    pub async fn issue_service_token(
        &self,
        user_id: i64,
        audience: i32,
        scopes: Vec<i32>,
        ttl_seconds: u32,
        event_id: &str,
    ) -> Result<proto::v1::IssueServiceTokenResponse, KenAuthError> {
        let mut client = self.token();
        client
            .issue_service_token(self.write_request(proto::v1::IssueServiceTokenRequest {
                audience,
                subject_user_id: user_id,
                scopes,
                ttl_seconds,
                constraint: None,
                event_id: event_id.to_string(),
            }))
            .await
            .map_err(KenAuthError::from_status)
            .map(|response| response.into_inner())
    }

    pub async fn introspect_token(
        &self,
        token: &str,
    ) -> Result<proto::v1::IntrospectTokenResponse, KenAuthError> {
        let mut client = self.token();
        let credential = Some(proto::common::v1::Credential {
            kind: Some(proto::common::v1::credential::Kind::ServiceToken(
                token.to_string(),
            )),
            request_id: String::new(),
        });
        client
            .introspect_token(self.read_request(proto::v1::IntrospectTokenRequest {
                credential,
                required_scopes: Vec::new(),
            }))
            .await
            .map_err(KenAuthError::from_status)
            .map(|response| response.into_inner())
    }

    pub async fn lookup_ssh_key(
        &self,
        public_key: &str,
    ) -> Result<Option<proto::common::v1::Principal>, KenAuthError> {
        let mut client = self.ssh_key();
        let response = client
            .lookup_ssh_key(self.read_request(proto::v1::LookupSshKeyRequest {
                fingerprint: String::new(),
                public_key: public_key.to_string(),
                require_active: true,
            }))
            .await
            .map_err(KenAuthError::from_status)?
            .into_inner();
        if !response.found {
            return Ok(None);
        }
        Ok(response.principal)
    }

    pub async fn ensure_bot_user(
        &self,
        username: &str,
        displayname: &str,
        system: bool,
    ) -> Result<proto::common::v1::UserPublic, KenAuthError> {
        let mut client = self.identity();
        client
            .ensure_bot_user(self.write_request(proto::v1::EnsureBotUserRequest {
                username: username.to_string(),
                displayname: displayname.to_string(),
                system,
            }))
            .await
            .map_err(KenAuthError::from_status)?
            .into_inner()
            .user
            .ok_or(KenAuthError::NotFound)
    }

    pub async fn ensure_bot_member(
        &self,
        organization_id: &str,
        user_id: i64,
        role_keys: Vec<String>,
        role_ids: Vec<String>,
    ) -> Result<proto::v1::OrganizationMemberInfo, KenAuthError> {
        let mut client = self.organization();
        client
            .ensure_bot_member(self.write_request(proto::v1::EnsureBotMemberRequest {
                organization_id: organization_id.to_string(),
                user_id,
                role_keys,
                role_ids,
            }))
            .await
            .map_err(KenAuthError::from_status)?
            .into_inner()
            .member
            .ok_or(KenAuthError::NotFound)
    }

    pub async fn get_organization(
        &self,
        actor_user_id: i64,
        organization_id: &str,
        slug: &str,
    ) -> Result<proto::v1::GetOrganizationResponse, KenAuthError> {
        let mut client = self.organization();
        client
            .get_organization(self.read_request(proto::v1::GetOrganizationRequest {
                actor_user_id,
                organization_id: organization_id.to_string(),
                slug: slug.to_string(),
            }))
            .await
            .map_err(KenAuthError::from_status)
            .map(|response| response.into_inner())
    }

    pub async fn get_member(
        &self,
        actor_user_id: i64,
        organization_id: &str,
        user_id: i64,
    ) -> Result<Option<proto::v1::OrganizationMemberInfo>, KenAuthError> {
        let mut client = self.organization();
        match client
            .get_member(self.read_request(proto::v1::GetMemberRequest {
                actor_user_id,
                organization_id: organization_id.to_string(),
                member_id: String::new(),
                user_id,
            }))
            .await
        {
            Ok(response) => Ok(response.into_inner().member),
            Err(status) if status.code() == tonic::Code::NotFound => Ok(None),
            Err(status) => Err(KenAuthError::from_status(status)),
        }
    }

    pub async fn list_members(
        &self,
        actor_user_id: i64,
        organization_id: &str,
        include_inactive: bool,
        include_banned: bool,
        page: proto::common::v1::PageRequest,
    ) -> Result<proto::v1::ListMembersResponse, KenAuthError> {
        let mut client = self.organization();
        client
            .list_members(self.read_request(proto::v1::ListMembersRequest {
                actor_user_id,
                organization_id: organization_id.to_string(),
                include_inactive,
                include_banned,
                page: Some(page),
            }))
            .await
            .map_err(KenAuthError::from_status)
            .map(|response| response.into_inner())
    }

    pub async fn list_organizations(
        &self,
        actor_user_id: i64,
        user_id: i64,
        include_archived: bool,
        page: proto::common::v1::PageRequest,
    ) -> Result<proto::v1::ListOrganizationsResponse, KenAuthError> {
        let mut client = self.organization();
        client
            .list_organizations(self.read_request(proto::v1::ListOrganizationsRequest {
                actor_user_id,
                user_id,
                include_archived,
                page: Some(page),
            }))
            .await
            .map_err(KenAuthError::from_status)
            .map(|response| response.into_inner())
    }

    pub async fn list_public_organizations(
        &self,
        query: &str,
        page: proto::common::v1::PageRequest,
    ) -> Result<proto::v1::ListPublicOrganizationsResponse, KenAuthError> {
        let mut client = self.organization();
        client
            .list_public_organizations(self.read_request(
                proto::v1::ListPublicOrganizationsRequest {
                    query: query.to_string(),
                    page: Some(page),
                },
            ))
            .await
            .map_err(KenAuthError::from_status)
            .map(|response| response.into_inner())
    }

    pub async fn check_permissions(
        &self,
        actor_user_id: i64,
        organization_id: &str,
        user_id: i64,
        permissions: Vec<String>,
    ) -> Result<proto::v1::CheckPermissionsResponse, KenAuthError> {
        let mut client = self.organization();
        client
            .check_permissions(self.read_request(proto::v1::CheckPermissionsRequest {
                actor_user_id,
                organization_id: organization_id.to_string(),
                user_id,
                permissions,
            }))
            .await
            .map_err(KenAuthError::from_status)
            .map(|response| response.into_inner())
    }

    pub async fn create_organization(
        &self,
        actor_user_id: i64,
        slug: &str,
        name: &str,
        description: &str,
        avatar_url: &str,
        join_policy: Option<i32>,
        allow_join: Option<bool>,
        require_email_verified: Option<bool>,
        require_mfa: Option<bool>,
        welcome_message: Option<String>,
        member_must_enable_2fa: Option<bool>,
        allow_invite: Option<bool>,
        allow_join_level: Option<i32>,
    ) -> Result<proto::v1::CreateOrganizationResponse, KenAuthError> {
        let mut client = self.organization();
        client
            .create_organization(self.write_request(proto::v1::CreateOrganizationRequest {
                actor_user_id,
                slug: slug.to_string(),
                name: name.to_string(),
                description: description.to_string(),
                avatar_url: avatar_url.to_string(),
                join_policy,
                allow_join,
                require_email_verified,
                require_mfa,
                welcome_message,
                member_must_enable_2fa,
                allow_invite,
                allow_join_level,
            }))
            .await
            .map_err(KenAuthError::from_status)
            .map(|response| response.into_inner())
    }

    #[allow(clippy::too_many_arguments)]
    pub async fn update_organization(
        &self,
        actor_user_id: i64,
        organization_id: &str,
        name: Option<String>,
        description: Option<String>,
        avatar_url: Option<String>,
        expected_version: i32,
        join_policy: Option<i32>,
        allow_join: Option<bool>,
        require_email_verified: Option<bool>,
        require_mfa: Option<bool>,
        welcome_message: Option<String>,
        member_must_enable_2fa: Option<bool>,
        allow_invite: Option<bool>,
        allow_join_level: Option<i32>,
    ) -> Result<proto::v1::OrganizationInfo, KenAuthError> {
        let mut client = self.organization();
        client
            .update_organization(self.write_request(proto::v1::UpdateOrganizationRequest {
                actor_user_id,
                organization_id: organization_id.to_string(),
                name,
                description,
                avatar_url,
                expected_version,
                join_policy,
                allow_join,
                require_email_verified,
                require_mfa,
                welcome_message,
                member_must_enable_2fa,
                allow_invite,
                allow_join_level,
            }))
            .await
            .map_err(KenAuthError::from_status)?
            .into_inner()
            .organization
            .ok_or(KenAuthError::NotFound)
    }

    pub async fn archive_organization(
        &self,
        actor_user_id: i64,
        organization_id: &str,
        reason: &str,
    ) -> Result<(), KenAuthError> {
        let mut client = self.organization();
        client
            .archive_organization(self.write_request(proto::v1::ArchiveOrganizationRequest {
                actor_user_id,
                organization_id: organization_id.to_string(),
                reason: reason.to_string(),
            }))
            .await
            .map_err(KenAuthError::from_status)?;
        Ok(())
    }

    pub async fn restore_organization(
        &self,
        actor_user_id: i64,
        organization_id: &str,
    ) -> Result<(), KenAuthError> {
        let mut client = self.organization();
        client
            .restore_organization(self.write_request(proto::v1::RestoreOrganizationRequest {
                actor_user_id,
                organization_id: organization_id.to_string(),
            }))
            .await
            .map_err(KenAuthError::from_status)?;
        Ok(())
    }

    pub async fn delete_organization(
        &self,
        actor_user_id: i64,
        organization_id: &str,
        force: bool,
        reason: &str,
    ) -> Result<(), KenAuthError> {
        let mut client = self.organization();
        client
            .delete_organization(self.write_request(proto::v1::DeleteOrganizationRequest {
                actor_user_id,
                organization_id: organization_id.to_string(),
                force,
                reason: reason.to_string(),
            }))
            .await
            .map_err(KenAuthError::from_status)?;
        Ok(())
    }
    pub async fn add_member(
        &self,
        actor_user_id: i64,
        organization_id: &str,
        user_id: i64,
        role: i32,
        role_keys: Vec<String>,
    ) -> Result<(proto::v1::OrganizationMemberInfo, bool), KenAuthError> {
        let mut client = self.organization();
        let response = client
            .add_member(self.write_request(proto::v1::AddMemberRequest {
                actor_user_id,
                organization_id: organization_id.to_string(),
                user_id,
                role,
                role_keys,
            }))
            .await
            .map_err(KenAuthError::from_status)?
            .into_inner();
        let member = response.member.ok_or(KenAuthError::NotFound)?;
        Ok((member, response.created))
    }

    pub async fn remove_member(
        &self,
        actor_user_id: i64,
        organization_id: &str,
        user_id: i64,
        reason: &str,
    ) -> Result<bool, KenAuthError> {
        let mut client = self.organization();
        client
            .remove_member(self.write_request(proto::v1::RemoveMemberRequest {
                actor_user_id,
                organization_id: organization_id.to_string(),
                member_id: String::new(),
                user_id,
                reason: reason.to_string(),
            }))
            .await
            .map_err(KenAuthError::from_status)
            .map(|response| response.into_inner().removed)
    }

    pub async fn leave_organization(
        &self,
        actor_user_id: i64,
        organization_id: &str,
    ) -> Result<bool, KenAuthError> {
        let mut client = self.organization();
        client
            .leave_organization(self.write_request(proto::v1::LeaveOrganizationRequest {
                actor_user_id,
                organization_id: organization_id.to_string(),
            }))
            .await
            .map_err(KenAuthError::from_status)
            .map(|response| response.into_inner().left)
    }

    #[allow(clippy::too_many_arguments)]
    pub async fn update_member(
        &self,
        actor_user_id: i64,
        organization_id: &str,
        user_id: i64,
        alias_name: Option<Option<String>>,
        alias_avatar_url: Option<Option<String>>,
        message_notifications: Option<String>,
        muted: Option<bool>,
        mute_until: Option<Option<prost_types::Timestamp>>,
        mobile_push: Option<bool>,
        suppress_everyone: Option<bool>,
        suppress_roles: Option<bool>,
        hide_muted_channels: Option<bool>,
    ) -> Result<proto::v1::OrganizationMemberInfo, KenAuthError> {
        let (clear_mute_until, mute_until) = match mute_until {
            Some(None) => (Some(true), None),
            Some(Some(value)) => (None, Some(value)),
            None => (None, None),
        };
        let mut client = self.organization();
        client
            .update_member(self.write_request(proto::v1::UpdateMemberRequest {
                actor_user_id,
                organization_id: organization_id.to_string(),
                member_id: String::new(),
                alias_name: alias_name.map(|value| value.unwrap_or_default()),
                alias_avatar_url: alias_avatar_url.map(|value| value.unwrap_or_default()),
                message_notifications,
                muted,
                clear_mute_until,
                mute_until,
                mobile_push,
                suppress_everyone,
                suppress_roles,
                hide_muted_channels,
                expected_version: 0,
                user_id,
            }))
            .await
            .map_err(KenAuthError::from_status)?
            .into_inner()
            .member
            .ok_or(KenAuthError::NotFound)
    }

    pub async fn ban_member(
        &self,
        actor_user_id: i64,
        organization_id: &str,
        user_id: i64,
        reason: &str,
    ) -> Result<proto::v1::OrganizationMemberInfo, KenAuthError> {
        let mut client = self.organization();
        client
            .ban_member(self.write_request(proto::v1::BanMemberRequest {
                actor_user_id,
                organization_id: organization_id.to_string(),
                member_id: String::new(),
                reason: reason.to_string(),
                user_id,
            }))
            .await
            .map_err(KenAuthError::from_status)?
            .into_inner()
            .member
            .ok_or(KenAuthError::NotFound)
    }

    pub async fn unban_member(
        &self,
        actor_user_id: i64,
        organization_id: &str,
        user_id: i64,
    ) -> Result<proto::v1::OrganizationMemberInfo, KenAuthError> {
        let mut client = self.organization();
        client
            .unban_member(self.write_request(proto::v1::UnbanMemberRequest {
                actor_user_id,
                organization_id: organization_id.to_string(),
                member_id: String::new(),
                user_id,
            }))
            .await
            .map_err(KenAuthError::from_status)?
            .into_inner()
            .member
            .ok_or(KenAuthError::NotFound)
    }

    #[allow(clippy::too_many_arguments)]
    pub async fn update_ban(
        &self,
        actor_user_id: i64,
        organization_id: &str,
        user_id: i64,
        reason: Option<Option<String>>,
        expires_at: Option<Option<prost_types::Timestamp>>,
        allow_speak: Option<bool>,
        allow_push_repo: Option<bool>,
        allow_comment_any: Option<bool>,
    ) -> Result<proto::v1::OrganizationMemberInfo, KenAuthError> {
        let (clear_expires_at, expires_at) = match expires_at {
            Some(None) => (Some(true), None),
            Some(Some(value)) => (None, Some(value)),
            None => (None, None),
        };
        let mut client = self.organization();
        client
            .update_ban(self.write_request(proto::v1::UpdateBanRequest {
                actor_user_id,
                organization_id: organization_id.to_string(),
                user_id,
                reason: reason.map(|value| value.unwrap_or_default()),
                clear_expires_at,
                expires_at,
                allow_agent_join: None,
                allow_speak,
                allow_push_repo,
                allow_comment_any,
            }))
            .await
            .map_err(KenAuthError::from_status)?
            .into_inner()
            .ban
            .ok_or(KenAuthError::NotFound)
    }

    #[allow(clippy::too_many_arguments)]
    pub async fn create_role(
        &self,
        actor_user_id: i64,
        organization_id: &str,
        key: &str,
        name: &str,
        description: &str,
        rank: i32,
        self_assignable: bool,
        permissions: Vec<String>,
    ) -> Result<proto::v1::OrganizationRoleInfo, KenAuthError> {
        let mut client = self.organization();
        client
            .create_role(self.write_request(proto::v1::CreateRoleRequest {
                actor_user_id,
                organization_id: organization_id.to_string(),
                key: key.to_string(),
                name: name.to_string(),
                description: description.to_string(),
                rank,
                default_role: false,
                permissions: permissions.into_iter().map(allow_grant).collect(),
                self_assignable,
            }))
            .await
            .map_err(KenAuthError::from_status)?
            .into_inner()
            .role
            .ok_or(KenAuthError::NotFound)
    }

    #[allow(clippy::too_many_arguments)]
    pub async fn update_role(
        &self,
        actor_user_id: i64,
        organization_id: &str,
        role_id: &str,
        name: Option<String>,
        description: Option<String>,
        rank: Option<i32>,
        self_assignable: Option<bool>,
        expected_version: i32,
    ) -> Result<proto::v1::OrganizationRoleInfo, KenAuthError> {
        let mut client = self.organization();
        client
            .update_role(self.write_request(proto::v1::UpdateRoleRequest {
                actor_user_id,
                organization_id: organization_id.to_string(),
                role_id: role_id.to_string(),
                name,
                description,
                rank,
                default_role: None,
                expected_version,
                self_assignable,
            }))
            .await
            .map_err(KenAuthError::from_status)?
            .into_inner()
            .role
            .ok_or(KenAuthError::NotFound)
    }

    pub async fn set_role_permissions(
        &self,
        actor_user_id: i64,
        organization_id: &str,
        role_id: &str,
        permissions: Vec<String>,
    ) -> Result<proto::v1::OrganizationRoleInfo, KenAuthError> {
        let mut client = self.organization();
        client
            .set_role_permissions(self.write_request(proto::v1::SetRolePermissionsRequest {
                actor_user_id,
                organization_id: organization_id.to_string(),
                role_id: role_id.to_string(),
                permissions: permissions.into_iter().map(allow_grant).collect(),
            }))
            .await
            .map_err(KenAuthError::from_status)?
            .into_inner()
            .role
            .ok_or(KenAuthError::NotFound)
    }

    pub async fn delete_role(
        &self,
        actor_user_id: i64,
        organization_id: &str,
        role_id: &str,
    ) -> Result<bool, KenAuthError> {
        let mut client = self.organization();
        client
            .delete_role(self.write_request(proto::v1::DeleteRoleRequest {
                actor_user_id,
                organization_id: organization_id.to_string(),
                role_id: role_id.to_string(),
            }))
            .await
            .map_err(KenAuthError::from_status)
            .map(|response| response.into_inner().deleted)
    }

    pub async fn assign_role(
        &self,
        actor_user_id: i64,
        organization_id: &str,
        user_id: i64,
        role_id: &str,
    ) -> Result<proto::v1::OrganizationMemberInfo, KenAuthError> {
        let mut client = self.organization();
        client
            .assign_role(self.write_request(proto::v1::AssignRoleRequest {
                actor_user_id,
                organization_id: organization_id.to_string(),
                member_id: String::new(),
                role_id: role_id.to_string(),
                user_id,
            }))
            .await
            .map_err(KenAuthError::from_status)?
            .into_inner()
            .member
            .ok_or(KenAuthError::NotFound)
    }

    pub async fn unassign_role(
        &self,
        actor_user_id: i64,
        organization_id: &str,
        user_id: i64,
        role_id: &str,
    ) -> Result<proto::v1::OrganizationMemberInfo, KenAuthError> {
        let mut client = self.organization();
        client
            .unassign_role(self.write_request(proto::v1::UnassignRoleRequest {
                actor_user_id,
                organization_id: organization_id.to_string(),
                member_id: String::new(),
                role_id: role_id.to_string(),
                user_id,
            }))
            .await
            .map_err(KenAuthError::from_status)?
            .into_inner()
            .member
            .ok_or(KenAuthError::NotFound)
    }

    #[allow(clippy::too_many_arguments)]
    pub async fn create_invite(
        &self,
        actor_user_id: i64,
        organization_id: &str,
        method: i32,
        max_uses: i32,
        expires_at: Option<prost_types::Timestamp>,
        role_ids: Vec<String>,
        target_user_id: i64,
        email: &str,
    ) -> Result<(proto::v1::OrganizationInviteInfo, String), KenAuthError> {
        let mut client = self.organization();
        let response = client
            .create_invite(self.write_request(proto::v1::CreateInviteRequest {
                actor_user_id,
                organization_id: organization_id.to_string(),
                method,
                max_uses,
                expires_at,
                role_id: role_ids.first().cloned().unwrap_or_default(),
                target_user_id,
                email: email.to_string(),
                role_ids,
            }))
            .await
            .map_err(KenAuthError::from_status)?
            .into_inner();
        let invite = response.invite.ok_or(KenAuthError::NotFound)?;
        Ok((invite, response.token))
    }

    pub async fn revoke_invite(
        &self,
        actor_user_id: i64,
        organization_id: &str,
        invite_id: &str,
    ) -> Result<bool, KenAuthError> {
        let mut client = self.organization();
        client
            .revoke_invite(self.write_request(proto::v1::RevokeInviteRequest {
                actor_user_id,
                organization_id: organization_id.to_string(),
                invite_id: invite_id.to_string(),
            }))
            .await
            .map_err(KenAuthError::from_status)
            .map(|response| response.into_inner().revoked)
    }

    pub async fn redeem_invite(
        &self,
        actor_user_id: i64,
        token: &str,
    ) -> Result<proto::v1::RedeemInviteResponse, KenAuthError> {
        let mut client = self.organization();
        client
            .redeem_invite(self.write_request(proto::v1::RedeemInviteRequest {
                actor_user_id,
                token: token.to_string(),
            }))
            .await
            .map_err(KenAuthError::from_status)
            .map(|response| response.into_inner())
    }

    pub async fn create_join_rule(
        &self,
        actor_user_id: i64,
        organization_id: &str,
        role_id: &str,
        trigger: &str,
        matcher: &str,
        enabled: bool,
    ) -> Result<proto::v1::JoinRuleInfo, KenAuthError> {
        let mut client = self.organization();
        client
            .create_join_rule(self.write_request(proto::v1::CreateJoinRuleRequest {
                actor_user_id,
                organization_id: organization_id.to_string(),
                role_id: role_id.to_string(),
                trigger: trigger.to_string(),
                matcher: matcher.to_string(),
                enabled,
            }))
            .await
            .map_err(KenAuthError::from_status)?
            .into_inner()
            .rule
            .ok_or(KenAuthError::NotFound)
    }

    #[allow(clippy::too_many_arguments)]
    pub async fn update_join_rule(
        &self,
        actor_user_id: i64,
        organization_id: &str,
        rule_id: &str,
        role_id: Option<String>,
        trigger: Option<String>,
        matcher: Option<String>,
        enabled: Option<bool>,
    ) -> Result<proto::v1::JoinRuleInfo, KenAuthError> {
        let mut client = self.organization();
        client
            .update_join_rule(self.write_request(proto::v1::UpdateJoinRuleRequest {
                actor_user_id,
                organization_id: organization_id.to_string(),
                rule_id: rule_id.to_string(),
                role_id,
                trigger,
                matcher,
                enabled,
            }))
            .await
            .map_err(KenAuthError::from_status)?
            .into_inner()
            .rule
            .ok_or(KenAuthError::NotFound)
    }

    pub async fn delete_join_rule(
        &self,
        actor_user_id: i64,
        organization_id: &str,
        rule_id: &str,
    ) -> Result<bool, KenAuthError> {
        let mut client = self.organization();
        client
            .delete_join_rule(self.write_request(proto::v1::DeleteJoinRuleRequest {
                actor_user_id,
                organization_id: organization_id.to_string(),
                rule_id: rule_id.to_string(),
            }))
            .await
            .map_err(KenAuthError::from_status)
            .map(|response| response.into_inner().deleted)
    }

    pub async fn create_join_request(
        &self,
        actor_user_id: i64,
        organization_id: &str,
        note: &str,
    ) -> Result<proto::v1::JoinRequestInfo, KenAuthError> {
        let mut client = self.organization();
        client
            .create_join_request(self.write_request(proto::v1::CreateJoinRequestRequest {
                actor_user_id,
                organization_id: organization_id.to_string(),
                note: note.to_string(),
            }))
            .await
            .map_err(KenAuthError::from_status)?
            .into_inner()
            .request
            .ok_or(KenAuthError::NotFound)
    }

    pub async fn withdraw_join_request(
        &self,
        actor_user_id: i64,
        organization_id: &str,
        request_id: &str,
    ) -> Result<proto::v1::JoinRequestInfo, KenAuthError> {
        let mut client = self.organization();
        client
            .withdraw_join_request(self.write_request(proto::v1::WithdrawJoinRequestRequest {
                actor_user_id,
                organization_id: organization_id.to_string(),
                request_id: request_id.to_string(),
            }))
            .await
            .map_err(KenAuthError::from_status)?
            .into_inner()
            .request
            .ok_or(KenAuthError::NotFound)
    }

    pub async fn approve_join_request(
        &self,
        actor_user_id: i64,
        organization_id: &str,
        request_id: &str,
    ) -> Result<
        (
            proto::v1::JoinRequestInfo,
            Option<proto::v1::OrganizationMemberInfo>,
        ),
        KenAuthError,
    > {
        let mut client = self.organization();
        let response = client
            .approve_join_request(self.write_request(proto::v1::ApproveJoinRequestRequest {
                actor_user_id,
                organization_id: organization_id.to_string(),
                request_id: request_id.to_string(),
            }))
            .await
            .map_err(KenAuthError::from_status)?
            .into_inner();
        let request = response.request.ok_or(KenAuthError::NotFound)?;
        Ok((request, response.member))
    }

    pub async fn reject_join_request(
        &self,
        actor_user_id: i64,
        organization_id: &str,
        request_id: &str,
        note: &str,
    ) -> Result<proto::v1::JoinRequestInfo, KenAuthError> {
        let mut client = self.organization();
        client
            .reject_join_request(self.write_request(proto::v1::RejectJoinRequestRequest {
                actor_user_id,
                organization_id: organization_id.to_string(),
                request_id: request_id.to_string(),
                note: note.to_string(),
            }))
            .await
            .map_err(KenAuthError::from_status)?
            .into_inner()
            .request
            .ok_or(KenAuthError::NotFound)
    }
}

fn allow_grant(permission: String) -> proto::v1::OrganizationPermissionGrant {
    proto::v1::OrganizationPermissionGrant {
        permission,
        effect: proto::v1::PermissionEffect::Allow as i32,
    }
}