@Override
  public Mono<SetProfileResponse> setProfile(final SetProfileRequest request) {
    validateRequest(request);
    return Mono.fromSupplier(AuthenticationUtil::requireAuthenticatedDevice)
        .flatMap(authenticatedDevice -> Mono.zip(
            Mono.fromFuture(() -> accountsManager.getByAccountIdentifierAsync(authenticatedDevice.accountIdentifier()))
                .map(maybeAccount -> maybeAccount.orElseThrow(Status.UNAUTHENTICATED::asRuntimeException)),
            Mono.fromFuture(() -> profilesManager.getAsync(authenticatedDevice.accountIdentifier(), request.getVersion()))
        ))
        .doOnNext(accountAndMaybeProfile -> {
          if (!request.getPaymentAddress().isEmpty()) {
            final boolean hasDisallowedPrefix =
                dynamicConfigurationManager.getConfiguration().getPaymentsConfiguration().getDisallowedPrefixes().stream()
                    .anyMatch(prefix -> accountAndMaybeProfile.getT1().getNumber().startsWith(prefix));
            if (hasDisallowedPrefix && accountAndMaybeProfile.getT2().map(VersionedProfile::paymentAddress).isEmpty()) {
              throw Status.PERMISSION_DENIED.asRuntimeException();
            }
          }
        })
        .flatMap(accountAndMaybeProfile -> {
          final Account account = accountAndMaybeProfile.getT1();
          final Optional<String> currentAvatar = accountAndMaybeProfile.getT2().map(VersionedProfile::avatar)
              .filter(avatar -> avatar.startsWith("profiles/"));
          final AvatarData avatarData = switch (AvatarChangeUtil.fromGrpcAvatarChange(request.getAvatarChange())) {
            case AVATAR_CHANGE_UNCHANGED -> new AvatarData(currentAvatar, currentAvatar, Optional.empty());
            case AVATAR_CHANGE_CLEAR -> new AvatarData(currentAvatar, Optional.empty(), Optional.empty());
            case AVATAR_CHANGE_UPDATE -> {
              final String updateAvatarObjectName = ProfileHelper.generateAvatarObjectName();
              yield new AvatarData(currentAvatar, Optional.of(updateAvatarObjectName),
                  Optional.of(generateAvatarUploadForm(updateAvatarObjectName)));
            }
          };

          final Mono<Void> profileSetMono = Mono.fromFuture(() -> profilesManager.setAsync(account.getUuid(),
              new VersionedProfile(
                  request.getVersion(),
                  request.getName().toByteArray(),
                  avatarData.finalAvatar().orElse(null),
                  request.getAboutEmoji().toByteArray(),
                  request.getAbout().toByteArray(),
                  request.getPaymentAddress().toByteArray(),
                  request.getPhoneNumberSharing().toByteArray(),
                  request.getCommitment().toByteArray())));

          final List<Mono<?>> updates = new ArrayList<>(2);
          final List<AccountBadge> updatedBadges = Optional.of(request.getBadgeIdsList())
              .map(badges -> ProfileHelper.mergeBadgeIdsWithExistingAccountBadges(clock, badgeConfigurationMap, badges, account.getBadges()))
              .orElseGet(account::getBadges);

          updates.add(Mono.fromFuture(() -> accountsManager.updateAsync(account, a -> {
            a.setBadges(clock, updatedBadges);
            a.setCurrentProfileVersion(request.getVersion());
          })));

          if (request.getAvatarChange() != AvatarChange.AVATAR_CHANGE_UNCHANGED && avatarData.currentAvatar().isPresent()) {
            updates.add(Mono.fromFuture(() -> asyncS3client.deleteObject(DeleteObjectRequest.builder()
                .bucket(bucket)
                .key(avatarData.currentAvatar().get())
                .build())));
          }
          return profileSetMono.thenMany(Flux.merge(updates)).then(Mono.just(avatarData));
        })
        .map(avatarData -> avatarData.uploadAttributes()
            .map(avatarUploadAttributes -> SetProfileResponse.newBuilder().setAttributes(avatarUploadAttributes).build())
            .orElse(SetProfileResponse.newBuilder().build())
        );
  }
@Test
  void setProfile() throws InvalidInputException {
    final byte[] commitment = new ProfileKey(new byte[32]).getCommitment(new ServiceId.Aci(AUTHENTICATED_ACI)).serialize();
    final byte[] validAboutEmoji = new byte[60];
    final byte[] validAbout = new byte[540];
    final byte[] validPaymentAddress = new byte[582];
    final byte[] validPhoneNumberSharing = new byte[29];

    final SetProfileRequest request = SetProfileRequest.newBuilder()
        .setVersion(VERSION)
        .setName(ByteString.copyFrom(VALID_NAME))
        .setAvatarChange(AvatarChange.AVATAR_CHANGE_UNCHANGED)
        .setAboutEmoji(ByteString.copyFrom(validAboutEmoji))
        .setAbout(ByteString.copyFrom(validAbout))
        .setPaymentAddress(ByteString.copyFrom(validPaymentAddress))
        .setPhoneNumberSharing(ByteString.copyFrom(validPhoneNumberSharing))
        .setCommitment(ByteString.copyFrom(commitment))
        .build();

    authenticatedServiceStub().setProfile(request);

    final ArgumentCaptor<VersionedProfile> profileArgumentCaptor = ArgumentCaptor.forClass(VersionedProfile.class);

    verify(profilesManager).setAsync(eq(account.getUuid()), profileArgumentCaptor.capture());

    final VersionedProfile profile = profileArgumentCaptor.getValue();

    assertThat(profile.commitment()).isEqualTo(commitment);
    assertThat(profile.avatar()).isNull();
    assertThat(profile.version()).isEqualTo(VERSION);
    assertThat(profile.name()).isEqualTo(VALID_NAME);
    assertThat(profile.aboutEmoji()).isEqualTo(validAboutEmoji);
    assertThat(profile.about()).isEqualTo(validAbout);
    assertThat(profile.paymentAddress()).isEqualTo(validPaymentAddress);
    assertThat(profile.phoneNumberSharing()).isEqualTo(validPhoneNumberSharing);
  }