diff --git a/services/identity/src/client_service.rs b/services/identity/src/client_service.rs index 30aecbd24..4363cd221 100644 --- a/services/identity/src/client_service.rs +++ b/services/identity/src/client_service.rs @@ -1,1063 +1,1074 @@ // Standard library imports use std::str::FromStr; // External crate imports use comm_lib::aws::DynamoDBError; use comm_lib::shared::reserved_users::RESERVED_USERNAME_SET; use comm_opaque2::grpc::protocol_error_to_grpc_status; use rand::rngs::OsRng; use serde::{Deserialize, Serialize}; use siwe::eip55; use tonic::Response; use tracing::{debug, error, info, warn}; // Workspace crate imports use crate::config::CONFIG; use crate::constants::request_metadata; use crate::database::{ DBDeviceTypeInt, DatabaseClient, DeviceType, KeyPayload, }; use crate::error::{DeviceListError, Error as DBError}; use crate::grpc_services::protos::unauth::{ find_user_id_request, AddReservedUsernamesRequest, AuthResponse, Empty, FindUserIdRequest, FindUserIdResponse, GenerateNonceResponse, OpaqueLoginFinishRequest, OpaqueLoginStartRequest, OpaqueLoginStartResponse, RegistrationFinishRequest, RegistrationStartRequest, RegistrationStartResponse, RemoveReservedUsernameRequest, ReservedRegistrationStartRequest, ReservedWalletRegistrationRequest, SecondaryDeviceKeysUploadRequest, VerifyUserAccessTokenRequest, VerifyUserAccessTokenResponse, WalletAuthRequest, GetFarcasterUsersRequest, GetFarcasterUsersResponse }; use crate::grpc_services::shared::get_value; use crate::grpc_utils::{ ChallengeResponse, DeviceKeyUploadActions, NonceChallenge, }; use crate::nonce::generate_nonce_data; use crate::reserved_users::{ validate_account_ownership_message_and_get_user_id, validate_add_reserved_usernames_message, validate_remove_reserved_username_message, }; use crate::siwe::{ is_valid_ethereum_address, parse_and_verify_siwe_message, SocialProof, }; use crate::token::{AccessTokenData, AuthType}; pub use crate::grpc_services::protos::unauth::identity_client_service_server::{ IdentityClientService, IdentityClientServiceServer, }; use crate::regex::is_valid_username; #[derive(Clone, Serialize, Deserialize)] pub enum WorkflowInProgress { Registration(Box), Login(Box), Update(UpdateState), } #[derive(Clone, Serialize, Deserialize)] pub struct UserRegistrationInfo { pub username: String, pub flattened_device_key_upload: FlattenedDeviceKeyUpload, pub user_id: Option, pub farcaster_id: Option, } #[derive(Clone, Serialize, Deserialize)] pub struct UserLoginInfo { pub user_id: String, pub flattened_device_key_upload: FlattenedDeviceKeyUpload, pub opaque_server_login: comm_opaque2::server::Login, pub device_to_remove: Option, } #[derive(Clone, Serialize, Deserialize)] pub struct UpdateState { pub user_id: String, } #[derive(Clone, Serialize, Deserialize)] pub struct FlattenedDeviceKeyUpload { pub device_id_key: String, pub key_payload: String, pub key_payload_signature: String, pub content_prekey: String, pub content_prekey_signature: String, pub content_one_time_keys: Vec, pub notif_prekey: String, pub notif_prekey_signature: String, pub notif_one_time_keys: Vec, pub device_type: DeviceType, } #[derive(derive_more::Constructor)] pub struct ClientService { client: DatabaseClient, } #[tonic::async_trait] impl IdentityClientService for ClientService { async fn register_password_user_start( &self, request: tonic::Request, ) -> Result, tonic::Status> { let message = request.into_inner(); debug!("Received registration request for: {}", message.username); if !is_valid_username(&message.username) || is_valid_ethereum_address(&message.username) { return Err(tonic::Status::invalid_argument("invalid username")); } self.check_username_taken(&message.username).await?; let username_in_reserved_usernames_table = self .client .username_in_reserved_usernames_table(&message.username) .await .map_err(handle_db_error)?; if username_in_reserved_usernames_table { return Err(tonic::Status::already_exists("username already exists")); } if RESERVED_USERNAME_SET.contains(&message.username) { return Err(tonic::Status::invalid_argument("username reserved")); } let registration_state = construct_user_registration_info( &message, None, message.username.clone(), message.farcaster_id.clone(), )?; let server_registration = comm_opaque2::server::Registration::new(); let server_message = server_registration .start( &CONFIG.server_setup, &message.opaque_registration_request, message.username.as_bytes(), ) .map_err(protocol_error_to_grpc_status)?; let session_id = self .client .insert_workflow(WorkflowInProgress::Registration(Box::new( registration_state, ))) .await .map_err(handle_db_error)?; let response = RegistrationStartResponse { session_id, opaque_registration_response: server_message, }; Ok(Response::new(response)) } async fn register_reserved_password_user_start( &self, request: tonic::Request, ) -> Result, tonic::Status> { let message = request.into_inner(); self.check_username_taken(&message.username).await?; if RESERVED_USERNAME_SET.contains(&message.username) { return Err(tonic::Status::invalid_argument("username reserved")); } let username_in_reserved_usernames_table = self .client .username_in_reserved_usernames_table(&message.username) .await .map_err(handle_db_error)?; if !username_in_reserved_usernames_table { return Err(tonic::Status::permission_denied("username not reserved")); } let user_id = validate_account_ownership_message_and_get_user_id( &message.username, &message.keyserver_message, &message.keyserver_signature, )?; let registration_state = construct_user_registration_info( &message, Some(user_id), message.username.clone(), None, )?; let server_registration = comm_opaque2::server::Registration::new(); let server_message = server_registration .start( &CONFIG.server_setup, &message.opaque_registration_request, message.username.as_bytes(), ) .map_err(protocol_error_to_grpc_status)?; let session_id = self .client .insert_workflow(WorkflowInProgress::Registration(Box::new( registration_state, ))) .await .map_err(handle_db_error)?; let response = RegistrationStartResponse { session_id, opaque_registration_response: server_message, }; Ok(Response::new(response)) } async fn register_password_user_finish( &self, request: tonic::Request, ) -> Result, tonic::Status> { let code_version = get_code_version(&request); let message = request.into_inner(); if let Some(WorkflowInProgress::Registration(state)) = self .client .get_workflow(message.session_id) .await .map_err(handle_db_error)? { let server_registration = comm_opaque2::server::Registration::new(); let password_file = server_registration .finish(&message.opaque_registration_upload) .map_err(protocol_error_to_grpc_status)?; let login_time = chrono::Utc::now(); let device_id = state.flattened_device_key_upload.device_id_key.clone(); let user_id = self .client .add_password_user_to_users_table( *state, password_file, code_version, login_time, ) .await .map_err(handle_db_error)?; // Create access token let token = AccessTokenData::with_created_time( user_id.clone(), device_id, login_time, crate::token::AuthType::Password, &mut OsRng, ); let access_token = token.access_token.clone(); self .client .put_access_token_data(token) .await .map_err(handle_db_error)?; let response = AuthResponse { user_id, access_token, }; Ok(Response::new(response)) } else { Err(tonic::Status::not_found("session not found")) } } async fn log_in_password_user_start( &self, request: tonic::Request, ) -> Result, tonic::Status> { let message = request.into_inner(); debug!("Attempting to log in user: {:?}", &message.username); let user_id_and_password_file = self .client .get_user_id_and_password_file_from_username(&message.username) .await .map_err(handle_db_error)?; let (user_id, password_file_bytes) = if let Some(data) = user_id_and_password_file { data } else { // It's possible that the user attempting login is already registered // on Ashoat's keyserver. If they are, we should send back a gRPC status // code instructing them to get a signed message from Ashoat's keyserver // in order to claim their username and register with the Identity // service. let username_in_reserved_usernames_table = self .client .username_in_reserved_usernames_table(&message.username) .await .map_err(handle_db_error)?; if username_in_reserved_usernames_table { return Err(tonic::Status::failed_precondition( "need keyserver message to claim username", )); } return Err(tonic::Status::not_found("user not found")); }; let flattened_device_key_upload = construct_flattened_device_key_upload(&message)?; let maybe_device_to_remove = self .get_keyserver_device_to_remove( &user_id, &flattened_device_key_upload.device_id_key, message.force.unwrap_or(false), &flattened_device_key_upload.device_type, ) .await?; let mut server_login = comm_opaque2::server::Login::new(); let server_response = server_login .start( &CONFIG.server_setup, &password_file_bytes, &message.opaque_login_request, message.username.as_bytes(), ) .map_err(protocol_error_to_grpc_status)?; let login_state = construct_user_login_info( user_id, server_login, flattened_device_key_upload, maybe_device_to_remove, )?; let session_id = self .client .insert_workflow(WorkflowInProgress::Login(Box::new(login_state))) .await .map_err(handle_db_error)?; let response = Response::new(OpaqueLoginStartResponse { session_id, opaque_login_response: server_response, }); Ok(response) } async fn log_in_password_user_finish( &self, request: tonic::Request, ) -> Result, tonic::Status> { let code_version = get_code_version(&request); let message = request.into_inner(); if let Some(WorkflowInProgress::Login(state)) = self .client .get_workflow(message.session_id) .await .map_err(handle_db_error)? { let mut server_login = state.opaque_server_login.clone(); server_login .finish(&message.opaque_login_upload) .map_err(protocol_error_to_grpc_status)?; if let Some(device_to_remove) = state.device_to_remove { self .client .remove_device(state.user_id.clone(), device_to_remove) .await .map_err(handle_db_error)?; } let login_time = chrono::Utc::now(); self .client .add_user_device( state.user_id.clone(), state.flattened_device_key_upload.clone(), code_version, login_time, ) .await .map_err(handle_db_error)?; // Create access token let token = AccessTokenData::with_created_time( state.user_id.clone(), state.flattened_device_key_upload.device_id_key, login_time, crate::token::AuthType::Password, &mut OsRng, ); let access_token = token.access_token.clone(); self .client .put_access_token_data(token) .await .map_err(handle_db_error)?; let response = AuthResponse { user_id: state.user_id, access_token, }; Ok(Response::new(response)) } else { Err(tonic::Status::not_found("session not found")) } } async fn log_in_wallet_user( &self, request: tonic::Request, ) -> Result, tonic::Status> { let code_version = get_code_version(&request); let message = request.into_inner(); let parsed_message = parse_and_verify_siwe_message( &message.siwe_message, &message.siwe_signature, )?; self.verify_and_remove_nonce(&parsed_message.nonce).await?; let wallet_address = eip55(&parsed_message.address); let flattened_device_key_upload = construct_flattened_device_key_upload(&message)?; let login_time = chrono::Utc::now(); let Some(user_id) = self .client .get_user_id_from_user_info(wallet_address.clone(), &AuthType::Wallet) .await .map_err(handle_db_error)? else { // It's possible that the user attempting login is already registered // on Ashoat's keyserver. If they are, we should send back a gRPC status // code instructing them to get a signed message from Ashoat's keyserver // in order to claim their wallet address and register with the Identity // service. let username_in_reserved_usernames_table = self .client .username_in_reserved_usernames_table(&wallet_address) .await .map_err(handle_db_error)?; if username_in_reserved_usernames_table { return Err(tonic::Status::failed_precondition( "need keyserver message to claim username", )); } return Err(tonic::Status::not_found("user not found")); }; self .client .add_user_device( user_id.clone(), flattened_device_key_upload.clone(), code_version, chrono::Utc::now(), ) .await .map_err(handle_db_error)?; // Create access token let token = AccessTokenData::with_created_time( user_id.clone(), flattened_device_key_upload.device_id_key, login_time, crate::token::AuthType::Wallet, &mut OsRng, ); let access_token = token.access_token.clone(); self .client .put_access_token_data(token) .await .map_err(handle_db_error)?; let response = AuthResponse { user_id, access_token, }; Ok(Response::new(response)) } async fn register_wallet_user( &self, request: tonic::Request, ) -> Result, tonic::Status> { let code_version = get_code_version(&request); let message = request.into_inner(); let parsed_message = parse_and_verify_siwe_message( &message.siwe_message, &message.siwe_signature, )?; match self .client .get_nonce_from_nonces_table(&parsed_message.nonce) .await .map_err(handle_db_error)? { None => return Err(tonic::Status::invalid_argument("invalid nonce")), Some(nonce) if nonce.is_expired() => { // we don't need to remove the nonce from the table here // because the DynamoDB TTL will take care of it return Err(tonic::Status::aborted("nonce expired")); } Some(_) => self .client .remove_nonce_from_nonces_table(&parsed_message.nonce) .await .map_err(handle_db_error)?, }; let wallet_address = eip55(&parsed_message.address); self.check_wallet_address_taken(&wallet_address).await?; let username_in_reserved_usernames_table = self .client .username_in_reserved_usernames_table(&wallet_address) .await .map_err(handle_db_error)?; if username_in_reserved_usernames_table { return Err(tonic::Status::already_exists( "wallet address already exists", )); } let flattened_device_key_upload = construct_flattened_device_key_upload(&message)?; let login_time = chrono::Utc::now(); let social_proof = SocialProof::new(message.siwe_message, message.siwe_signature); let serialized_social_proof = serde_json::to_string(&social_proof) .map_err(|_| tonic::Status::invalid_argument("invalid_social_proof"))?; let user_id = self .client .add_wallet_user_to_users_table( flattened_device_key_upload.clone(), wallet_address, serialized_social_proof, None, code_version, login_time, message.farcaster_id, ) .await .map_err(handle_db_error)?; // Create access token let token = AccessTokenData::with_created_time( user_id.clone(), flattened_device_key_upload.device_id_key, login_time, crate::token::AuthType::Wallet, &mut OsRng, ); let access_token = token.access_token.clone(); self .client .put_access_token_data(token) .await .map_err(handle_db_error)?; let response = AuthResponse { user_id, access_token, }; Ok(Response::new(response)) } async fn register_reserved_wallet_user( &self, request: tonic::Request, ) -> Result, tonic::Status> { let code_version = get_code_version(&request); let message = request.into_inner(); let parsed_message = parse_and_verify_siwe_message( &message.siwe_message, &message.siwe_signature, )?; self.verify_and_remove_nonce(&parsed_message.nonce).await?; let wallet_address = eip55(&parsed_message.address); self.check_wallet_address_taken(&wallet_address).await?; let wallet_address_in_reserved_usernames_table = self .client .username_in_reserved_usernames_table(&wallet_address) .await .map_err(handle_db_error)?; if !wallet_address_in_reserved_usernames_table { return Err(tonic::Status::permission_denied( "wallet address not reserved", )); } let user_id = validate_account_ownership_message_and_get_user_id( &wallet_address, &message.keyserver_message, &message.keyserver_signature, )?; let flattened_device_key_upload = construct_flattened_device_key_upload(&message)?; let social_proof = SocialProof::new(message.siwe_message, message.siwe_signature); let serialized_social_proof = serde_json::to_string(&social_proof) .map_err(|_| tonic::Status::invalid_argument("invalid_social_proof"))?; let login_time = chrono::Utc::now(); self .client .add_wallet_user_to_users_table( flattened_device_key_upload.clone(), wallet_address, serialized_social_proof, Some(user_id.clone()), code_version, login_time, None, ) .await .map_err(handle_db_error)?; let token = AccessTokenData::with_created_time( user_id.clone(), flattened_device_key_upload.device_id_key, login_time, crate::token::AuthType::Wallet, &mut OsRng, ); let access_token = token.access_token.clone(); self .client .put_access_token_data(token) .await .map_err(handle_db_error)?; let response = AuthResponse { user_id, access_token, }; Ok(Response::new(response)) } async fn upload_keys_for_registered_device_and_log_in( &self, request: tonic::Request, ) -> Result, tonic::Status> { let code_version = get_code_version(&request); let message = request.into_inner(); let challenge_response = ChallengeResponse::try_from(&message)?; let flattened_device_key_upload = construct_flattened_device_key_upload(&message)?; let user_id = message.user_id; let device_id = flattened_device_key_upload.device_id_key.clone(); let NonceChallenge { nonce } = challenge_response.verify_and_get_message(&device_id)?; self.verify_and_remove_nonce(&nonce).await?; let Some(device_list) = self .client .get_current_device_list(&user_id) .await .map_err(handle_db_error)? else { warn!("User {} does not have valid device list. Secondary device auth impossible.", user_id); return Err(tonic::Status::aborted("device list error")); }; if !device_list.device_ids.contains(&device_id) { return Err(tonic::Status::permission_denied( "device not in device list", )); } let login_time = chrono::Utc::now(); let user_identifier = self .client .get_user_identifier(&user_id) .await .map_err(handle_db_error)?; let token = AccessTokenData::with_created_time( user_id.clone(), device_id, login_time, user_identifier.into(), &mut OsRng, ); let access_token = token.access_token.clone(); self .client .put_access_token_data(token) .await .map_err(handle_db_error)?; self .client .put_device_data( &user_id, flattened_device_key_upload, code_version, login_time, ) .await .map_err(handle_db_error)?; let response = AuthResponse { user_id, access_token, }; Ok(Response::new(response)) } async fn generate_nonce( &self, _request: tonic::Request, ) -> Result, tonic::Status> { let nonce_data = generate_nonce_data(&mut OsRng); match self .client .add_nonce_to_nonces_table(nonce_data.clone()) .await { Ok(_) => Ok(Response::new(GenerateNonceResponse { nonce: nonce_data.nonce, })), Err(e) => Err(handle_db_error(e)), } } async fn verify_user_access_token( &self, request: tonic::Request, ) -> Result, tonic::Status> { let message = request.into_inner(); debug!("Verifying device: {}", &message.device_id); let token_valid = self .client .verify_access_token( message.user_id, message.device_id.clone(), message.access_token, ) .await .map_err(handle_db_error)?; let response = Response::new(VerifyUserAccessTokenResponse { token_valid }); debug!( "device {} was verified: {}", &message.device_id, token_valid ); Ok(response) } async fn add_reserved_usernames( &self, request: tonic::Request, ) -> Result, tonic::Status> { let message = request.into_inner(); let user_details = validate_add_reserved_usernames_message( &message.message, &message.signature, )?; let filtered_user_details = self .client .filter_out_taken_usernames(user_details) .await .map_err(handle_db_error)?; self .client .add_usernames_to_reserved_usernames_table(filtered_user_details) .await .map_err(handle_db_error)?; let response = Response::new(Empty {}); Ok(response) } async fn remove_reserved_username( &self, request: tonic::Request, ) -> Result, tonic::Status> { let message = request.into_inner(); let username = validate_remove_reserved_username_message( &message.message, &message.signature, )?; self .client .delete_username_from_reserved_usernames_table(username) .await .map_err(handle_db_error)?; let response = Response::new(Empty {}); Ok(response) } async fn ping( &self, _request: tonic::Request, ) -> Result, tonic::Status> { let response = Response::new(Empty {}); Ok(response) } async fn find_user_id( &self, request: tonic::Request, ) -> Result, tonic::Status> { let message = request.into_inner(); use find_user_id_request::Identifier; let (user_ident, auth_type) = match message.identifier { None => { return Err(tonic::Status::invalid_argument("no identifier provided")) } Some(Identifier::Username(username)) => (username, AuthType::Password), Some(Identifier::WalletAddress(address)) => (address, AuthType::Wallet), }; let (is_reserved_result, user_id_result) = tokio::join!( self .client .username_in_reserved_usernames_table(&user_ident), self .client .get_user_id_from_user_info(user_ident.clone(), &auth_type), ); let is_reserved = is_reserved_result.map_err(handle_db_error)?; let user_id = user_id_result.map_err(handle_db_error)?; Ok(Response::new(FindUserIdResponse { user_id, is_reserved, })) } async fn get_farcaster_users( &self, - _request: tonic::Request, + request: tonic::Request, ) -> Result, tonic::Status> { - unimplemented!(); + let message = request.into_inner(); + + let farcaster_users = self + .client + .get_farcaster_users(message.farcaster_ids) + .await + .map_err(handle_db_error)? + .into_iter() + .map(|d| d.0) + .collect(); + + Ok(Response::new(GetFarcasterUsersResponse { farcaster_users })) } } impl ClientService { async fn check_username_taken( &self, username: &str, ) -> Result<(), tonic::Status> { let username_taken = self .client .username_taken(username.to_string()) .await .map_err(handle_db_error)?; if username_taken { return Err(tonic::Status::already_exists("username already exists")); } Ok(()) } async fn check_wallet_address_taken( &self, wallet_address: &str, ) -> Result<(), tonic::Status> { let wallet_address_taken = self .client .wallet_address_taken(wallet_address.to_string()) .await .map_err(handle_db_error)?; if wallet_address_taken { return Err(tonic::Status::already_exists( "wallet address already exists", )); } Ok(()) } async fn verify_and_remove_nonce( &self, nonce: &str, ) -> Result<(), tonic::Status> { match self .client .get_nonce_from_nonces_table(nonce) .await .map_err(handle_db_error)? { None => return Err(tonic::Status::invalid_argument("invalid nonce")), Some(nonce) if nonce.is_expired() => { // we don't need to remove the nonce from the table here // because the DynamoDB TTL will take care of it return Err(tonic::Status::aborted("nonce expired")); } Some(nonce_data) => self .client .remove_nonce_from_nonces_table(&nonce_data.nonce) .await .map_err(handle_db_error)?, }; Ok(()) } async fn get_keyserver_device_to_remove( &self, user_id: &str, new_keyserver_device_id: &str, force: bool, device_type: &DeviceType, ) -> Result, tonic::Status> { if device_type != &DeviceType::Keyserver { return Ok(None); } let maybe_keyserver_device_id = self .client .get_keyserver_device_id_for_user(user_id) .await .map_err(handle_db_error)?; let Some(existing_keyserver_device_id) = maybe_keyserver_device_id else { return Ok(None); }; if new_keyserver_device_id == existing_keyserver_device_id { return Ok(None); } if force { info!( "keyserver {} will be removed from the device list", existing_keyserver_device_id ); Ok(Some(existing_keyserver_device_id)) } else { Err(tonic::Status::already_exists( "user already has a keyserver", )) } } } pub fn handle_db_error(db_error: DBError) -> tonic::Status { match db_error { DBError::AwsSdk(DynamoDBError::InternalServerError(_)) | DBError::AwsSdk(DynamoDBError::ProvisionedThroughputExceededException( _, )) | DBError::AwsSdk(DynamoDBError::RequestLimitExceeded(_)) => { tonic::Status::unavailable("please retry") } DBError::DeviceList(DeviceListError::InvalidDeviceListUpdate) => { tonic::Status::invalid_argument("invalid device list update") } e => { error!("Encountered an unexpected error: {}", e); tonic::Status::failed_precondition("unexpected error") } } } fn construct_user_registration_info( message: &impl DeviceKeyUploadActions, user_id: Option, username: String, farcaster_id: Option, ) -> Result { Ok(UserRegistrationInfo { username, flattened_device_key_upload: construct_flattened_device_key_upload( message, )?, user_id, farcaster_id, }) } fn construct_user_login_info( user_id: String, opaque_server_login: comm_opaque2::server::Login, flattened_device_key_upload: FlattenedDeviceKeyUpload, device_to_remove: Option, ) -> Result { Ok(UserLoginInfo { user_id, flattened_device_key_upload, opaque_server_login, device_to_remove, }) } fn construct_flattened_device_key_upload( message: &impl DeviceKeyUploadActions, ) -> Result { let key_info = KeyPayload::from_str(&message.payload()?) .map_err(|_| tonic::Status::invalid_argument("malformed payload"))?; let flattened_device_key_upload = FlattenedDeviceKeyUpload { device_id_key: key_info.primary_identity_public_keys.ed25519, key_payload: message.payload()?, key_payload_signature: message.payload_signature()?, content_prekey: message.content_prekey()?, content_prekey_signature: message.content_prekey_signature()?, content_one_time_keys: message.one_time_content_prekeys()?, notif_prekey: message.notif_prekey()?, notif_prekey_signature: message.notif_prekey_signature()?, notif_one_time_keys: message.one_time_notif_prekeys()?, device_type: DeviceType::try_from(DBDeviceTypeInt(message.device_type()?)) .map_err(handle_db_error)?, }; Ok(flattened_device_key_upload) } fn get_code_version(req: &tonic::Request) -> u64 { get_value(req, request_metadata::CODE_VERSION) .and_then(|version| version.parse().ok()) .unwrap_or_else(|| { warn!( "Could not retrieve code version from request: {:?}. Defaulting to 0", req ); Default::default() }) } diff --git a/services/identity/src/database/farcaster.rs b/services/identity/src/database/farcaster.rs index 32c71321b..6ec30381d 100644 --- a/services/identity/src/database/farcaster.rs +++ b/services/identity/src/database/farcaster.rs @@ -1,88 +1,88 @@ use comm_lib::aws::ddb::types::AttributeValue; use comm_lib::database::AttributeExtractor; use comm_lib::database::AttributeMap; use comm_lib::database::DBItemAttributeError; use comm_lib::database::DBItemError; use comm_lib::database::Value; -use grpc_clients::identity::protos::unauth::FarcasterUser; use tracing::error; use crate::constants::USERS_TABLE; use crate::constants::USERS_TABLE_FARCASTER_ID_ATTRIBUTE_NAME; use crate::constants::USERS_TABLE_FARCASTER_ID_INDEX; use crate::constants::USERS_TABLE_PARTITION_KEY; use crate::constants::USERS_TABLE_USERNAME_ATTRIBUTE; use crate::constants::USERS_TABLE_WALLET_ADDRESS_ATTRIBUTE; +use crate::grpc_services::protos::unauth::FarcasterUser; use super::DatabaseClient; use super::Error; -struct FarcasterUserData(FarcasterUser); +pub struct FarcasterUserData(pub FarcasterUser); impl DatabaseClient { - async fn get_farcaster_users( + pub async fn get_farcaster_users( &self, farcaster_ids: Vec, ) -> Result, Error> { let mut users: Vec = Vec::new(); for id in farcaster_ids { let query_response = self .client .query() .table_name(USERS_TABLE) .index_name(USERS_TABLE_FARCASTER_ID_INDEX) .key_condition_expression(format!( "{} = :val", USERS_TABLE_FARCASTER_ID_ATTRIBUTE_NAME )) .expression_attribute_values(":val", AttributeValue::S(id)) .send() .await .map_err(|e| { error!("Failed to query users by farcasterID: {:?}", e); Error::AwsSdk(e.into()) })? .items .and_then(|mut items| items.pop()) .map(FarcasterUserData::try_from) .transpose() .map_err(Error::from)?; if let Some(data) = query_response { users.push(data); } } Ok(users) } } impl TryFrom for FarcasterUserData { type Error = DBItemError; fn try_from(mut attrs: AttributeMap) -> Result { let user_id = attrs.take_attr(USERS_TABLE_PARTITION_KEY)?; let maybe_username = attrs.take_attr(USERS_TABLE_USERNAME_ATTRIBUTE)?; let maybe_wallet_address = attrs.take_attr(USERS_TABLE_WALLET_ADDRESS_ATTRIBUTE)?; let username = match (maybe_username, maybe_wallet_address) { (Some(u), _) => u, (_, Some(w)) => w, (_, _) => { return Err(DBItemError { attribute_name: USERS_TABLE_USERNAME_ATTRIBUTE.to_string(), attribute_value: Value::AttributeValue(None), attribute_error: DBItemAttributeError::Missing, }); } }; let farcaster_id = attrs.take_attr(USERS_TABLE_FARCASTER_ID_ATTRIBUTE_NAME)?; Ok(Self(FarcasterUser { user_id, username, farcaster_id, })) } }