fix: a verified email can be reused (#6064)

#### What type of PR is this?
/kind bug
/area core
/milestone 2.17.x

#### What this PR does / why we need it:
修复已验证邮箱可以重复的问题

如果出现多个重复的已验证邮箱，则只保留一个其他的设置为未验证

#### Does this PR introduce a user-facing change?
```release-note
修复已验证邮箱可以重复的问题
```

application/src/main/java/run/halo/app/core/extension/reconciler/UserReconciler.java

@@ -20,6 +20,7 @@ import run.halo.app.core.extension.UserConnection;
 import run.halo.app.core.extension.attachment.Attachment;
 import run.halo.app.core.extension.service.AttachmentService;
 import run.halo.app.core.extension.service.RoleService;
+import run.halo.app.core.extension.service.UserService;
 import run.halo.app.extension.ExtensionClient;
 import run.halo.app.extension.GroupKind;
 import run.halo.app.extension.MetadataUtil;
@@ -47,6 +48,7 @@ public class UserReconciler implements Reconciler<Request> {
         .fixedBackoff(300)
         .retryOn(IllegalStateException.class)
         .build();
+    private final UserService userService;
 
     @Override
     public Result reconcile(Request request) {
@@ -60,10 +62,36 @@ public class UserReconciler implements Reconciler<Request> {
             ensureRoleNamesAnno(request.name());
             updatePermalink(request.name());
             handleAvatar(request.name());
+
+            checkVerifiedEmail(user);
+            client.update(user);
         });
         return new Result(false, null);
     }
 
+    private void checkVerifiedEmail(User user) {
+        var username = user.getMetadata().getName();
+        if (!user.getSpec().isEmailVerified()) {
+            return;
+        }
+        var email = user.getSpec().getEmail();
+        if (StringUtils.isBlank(email)) {
+            return;
+        }
+        if (checkEmailInUse(username, email)) {
+            user.getSpec().setEmailVerified(false);
+        }
+    }
+
+    private Boolean checkEmailInUse(String username, String email) {
+        return userService.listByEmail(email)
+            .filter(existUser -> existUser.getSpec().isEmailVerified())
+            .filter(existUser -> !existUser.getMetadata().getName().equals(username))
+            .hasElements()
+            .blockOptional()
+            .orElse(false);
+    }
+
     private void handleAvatar(String name) {
         client.fetch(User.class, name).ifPresent(user -> {
             Map<String, String> annotations = MetadataUtil.nullSafeAnnotations(user);

application/src/main/java/run/halo/app/core/extension/service/UserService.java

@@ -25,4 +25,6 @@ public interface UserService {
     Mono<User> createUser(User user, Set<String> roles);
 
     Mono<Boolean> confirmPassword(String username, String rawPassword);
+
+    Flux<User> listByEmail(String email);
 }

application/src/main/java/run/halo/app/core/extension/service/UserServiceImpl.java

@@ -1,6 +1,9 @@
 package run.halo.app.core.extension.service;
 
+import static org.springframework.data.domain.Sort.Order.asc;
+import static org.springframework.data.domain.Sort.Order.desc;
 import static run.halo.app.core.extension.RoleBinding.containsUser;
+import static run.halo.app.extension.index.query.QueryFactory.equal;
 
 import java.util.HashSet;
 import java.util.Objects;
@@ -8,6 +11,7 @@ import java.util.Set;
 import lombok.RequiredArgsConstructor;
 import org.apache.commons.lang3.BooleanUtils;
 import org.springframework.context.ApplicationEventPublisher;
+import org.springframework.data.domain.Sort;
 import org.springframework.security.crypto.password.PasswordEncoder;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
@@ -21,8 +25,10 @@ import run.halo.app.core.extension.Role;
 import run.halo.app.core.extension.RoleBinding;
 import run.halo.app.core.extension.User;
 import run.halo.app.event.user.PasswordChangedEvent;
+import run.halo.app.extension.ListOptions;
 import run.halo.app.extension.ReactiveExtensionClient;
 import run.halo.app.extension.exception.ExtensionNotFoundException;
+import run.halo.app.extension.router.selector.FieldSelector;
 import run.halo.app.infra.SystemConfigurableEnvironmentFetcher;
 import run.halo.app.infra.SystemSetting;
 import run.halo.app.infra.exception.AccessDeniedException;
@@ -198,6 +204,15 @@ public class UserServiceImpl implements UserService {
             .hasElement();
     }
 
+    @Override
+    public Flux<User> listByEmail(String email) {
+        var listOptions = new ListOptions();
+        listOptions.setFieldSelector(FieldSelector.of(equal("spec.email", email)));
+        return client.listAll(User.class, listOptions, Sort.by(desc("metadata.creationTimestamp"),
+            asc("metadata.name"))
+        );
+    }
+
     void publishPasswordChangedEvent(String username) {
         eventPublisher.publishEvent(new PasswordChangedEvent(this, username));
     }

application/src/main/java/run/halo/app/core/extension/service/impl/EmailVerificationServiceImpl.java

@@ -20,6 +20,7 @@ import run.halo.app.core.extension.User;
 import run.halo.app.core.extension.notification.Reason;
 import run.halo.app.core.extension.notification.Subscription;
 import run.halo.app.core.extension.service.EmailVerificationService;
+import run.halo.app.core.extension.service.UserService;
 import run.halo.app.extension.GroupVersion;
 import run.halo.app.extension.MetadataUtil;
 import run.halo.app.extension.ReactiveExtensionClient;
@@ -46,6 +47,7 @@ public class EmailVerificationServiceImpl implements EmailVerificationService {
     private final ReactiveExtensionClient client;
     private final NotificationReasonEmitter reasonEmitter;
     private final NotificationCenter notificationCenter;
+    private final UserService userService;
 
     @Override
     public Mono<Void> sendVerificationCode(String username, String email) {
@@ -80,28 +82,51 @@ public class EmailVerificationServiceImpl implements EmailVerificationService {
         Assert.state(StringUtils.isNotBlank(username), "Username must not be blank");
         Assert.state(StringUtils.isNotBlank(code), "Code must not be blank");
         return Mono.defer(() -> client.get(User.class, username)
-                .flatMap(user -> {
+                .flatMap(user -> verifyUserEmail(user, code))
-                    var annotations = MetadataUtil.nullSafeAnnotations(user);
-                    var emailToVerify = annotations.get(User.EMAIL_TO_VERIFY);
-                    if (StringUtils.isBlank(emailToVerify)) {
-                        return Mono.error(EmailVerificationFailed::new);
-                    }
-                    var verified =
-                        emailVerificationManager.verifyCode(username, emailToVerify, code);
-                    if (!verified) {
-                        return Mono.error(EmailVerificationFailed::new);
-                    }
-                    user.getSpec().setEmailVerified(true);
-                    user.getSpec().setEmail(emailToVerify);
-                    emailVerificationManager.removeCode(username, emailToVerify);
-                    return client.update(user);
-                })
             )
             .retryWhen(Retry.backoff(8, Duration.ofMillis(100))
-                .filter(OptimisticLockingFailureException.class::isInstance))
+                .filter(OptimisticLockingFailureException.class::isInstance));
+    }
+
+    private Mono<Void> verifyUserEmail(User user, String code) {
+        var username = user.getMetadata().getName();
+        var annotations = MetadataUtil.nullSafeAnnotations(user);
+        var emailToVerify = annotations.get(User.EMAIL_TO_VERIFY);
+
+        if (StringUtils.isBlank(emailToVerify)) {
+            return Mono.error(EmailVerificationFailed::new);
+        }
+
+        var verified = emailVerificationManager.verifyCode(username, emailToVerify, code);
+        if (!verified) {
+            return Mono.error(EmailVerificationFailed::new);
+        }
+
+        return isEmailInUse(username, emailToVerify)
+            .flatMap(inUse -> {
+                if (inUse) {
+                    return Mono.error(new EmailVerificationFailed("Email already in use.",
+                        null,
+                        "problemDetail.user.email.verify.emailInUse",
+                        null)
+                    );
+                }
+                // remove code when verified
+                emailVerificationManager.removeCode(username, emailToVerify);
+                user.getSpec().setEmailVerified(true);
+                user.getSpec().setEmail(emailToVerify);
+                return client.update(user);
+            })
             .then();
     }
 
+    Mono<Boolean> isEmailInUse(String username, String emailToVerify) {
+        return userService.listByEmail(emailToVerify)
+            .filter(user -> user.getSpec().isEmailVerified())
+            .filter(user -> !user.getMetadata().getName().equals(username))
+            .hasElements();
+    }
+
     @Override
     public Mono<Void> sendRegisterVerificationCode(String email) {
         Assert.state(StringUtils.isNotBlank(email), "Email must not be blank");

application/src/main/java/run/halo/app/infra/SchemeInitializer.java

@@ -87,6 +87,12 @@ public class SchemeInitializer implements ApplicationListener<ApplicationContext
                 .setName("spec.displayName")
                 .setIndexFunc(
                     simpleAttribute(User.class, user -> user.getSpec().getDisplayName())));
+            indexSpecs.add(new IndexSpec()
+                .setName("spec.email")
+                .setIndexFunc(simpleAttribute(User.class, user -> {
+                    var email = user.getSpec().getEmail();
+                    return StringUtils.isBlank(email) ? null : email;
+                })));
             indexSpecs.add(new IndexSpec()
                 .setName(User.USER_RELATED_ROLES_INDEX)
                 .setIndexFunc(multiValueAttribute(User.class, user -> {

application/src/main/resources/config/i18n/messages.properties

@@ -54,6 +54,7 @@ problemDetail.run.halo.app.infra.exception.PluginDependenciesNotEnabledException
 
 problemDetail.index.duplicateKey=The value of {0} already exists for unique index {1}, please rename it and retry.
 problemDetail.user.email.verify.maxAttempts=Too many verification attempts, please try again later.
+problemDetail.user.email.verify.emailInUse=The email has been used, please change the email and retry.
 problemDetail.user.password.unsatisfied=The password does not meet the specifications.
 problemDetail.user.username.unsatisfied=The username does not meet the specifications.
 problemDetail.user.oldPassword.notMatch=The old password does not match.

application/src/main/resources/config/i18n/messages_zh.properties

@@ -31,6 +31,7 @@ problemDetail.run.halo.app.infra.exception.PluginDependenciesNotEnabledException
 
 problemDetail.index.duplicateKey=唯一索引 {1} 中的值 {0} 已存在，请更名后重试。
 problemDetail.user.email.verify.maxAttempts=尝试次数过多，请稍候再试。
+problemDetail.user.email.verify.emailInUse=邮箱已被使用, 请更换邮箱后重试。
 problemDetail.user.password.unsatisfied=密码不符合规范。
 problemDetail.user.username.unsatisfied=用户名不符合规范。
 problemDetail.user.oldPassword.notMatch=旧密码不匹配。

application/src/test/java/run/halo/app/core/extension/reconciler/UserReconcilerTest.java

@@ -70,10 +70,10 @@ class UserReconcilerTest {
         when(client.fetch(eq(User.class), eq("fake-user")))
             .thenReturn(Optional.of(user("fake-user")));
         userReconciler.reconcile(new Reconciler.Request("fake-user"));
-        verify(client, times(3)).update(any(User.class));
+        verify(client, times(4)).update(any(User.class));
 
         ArgumentCaptor<User> captor = ArgumentCaptor.forClass(User.class);
-        verify(client, times(3)).update(captor.capture());
+        verify(client, times(4)).update(captor.capture());
         assertThat(captor.getValue().getStatus().getPermalink())
             .isEqualTo("http://localhost:8090/authors/fake-user");
     }
@@ -83,7 +83,7 @@ class UserReconcilerTest {
         when(client.fetch(eq(User.class), eq(AnonymousUserConst.PRINCIPAL)))
             .thenReturn(Optional.of(user(AnonymousUserConst.PRINCIPAL)));
         userReconciler.reconcile(new Reconciler.Request(AnonymousUserConst.PRINCIPAL));
-        verify(client, times(2)).update(any(User.class));
+        verify(client, times(3)).update(any(User.class));
     }
 
     @Test
@@ -108,7 +108,7 @@ class UserReconcilerTest {
 
         userReconciler.reconcile(new Reconciler.Request("fake-user"));
         ArgumentCaptor<User> captor = ArgumentCaptor.forClass(User.class);
-        verify(client, times(3)).update(captor.capture());
+        verify(client, times(4)).update(captor.capture());
         User user = captor.getAllValues().get(1);
         assertThat(user.getMetadata().getAnnotations().get(User.ROLE_NAMES_ANNO))
             .isEqualTo("[\"fake-role\"]");