feat: display unapproved comments created by the current user (halo-d…

…ev#3102)

#### What type of PR is this?
/kind improvement
/area core
/milestone 2.2.x
#### What this PR does / why we need it:
主题端显示当前用户创建的未审核通过的评论/回复

在开启了评论审核功能的前提下：
- 如果非匿名用户在主题端创建的评论或回复必须审核通过才能显示。
- 如果用户登录后评论或回复会被立即展示在列表，但仅限创建者可见，审核通过后对所有人可见。
#### Which issue(s) this PR fixes:

Fixes halo-dev#2731

#### Special notes for your reviewer:
/cc @halo-dev/sig-halo 
#### Does this PR introduce a user-facing change?
```release-note
主题端支持显示当前用户创建的未审核的评论
```

src/main/java/run/halo/app/theme/finders/impl/CommentFinderImpl.java

@@ -1,12 +1,17 @@
 package run.halo.app.theme.finders.impl;
 
+import java.security.Principal;
 import java.time.Instant;
 import java.util.Comparator;
 import java.util.List;
 import java.util.Objects;
 import java.util.function.Function;
 import java.util.function.Predicate;
+import org.apache.commons.lang3.BooleanUtils;
 import org.apache.commons.lang3.ObjectUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.springframework.security.core.context.ReactiveSecurityContextHolder;
+import org.springframework.security.core.context.SecurityContext;
 import org.springframework.util.Assert;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
@@ -17,6 +22,7 @@
 import run.halo.app.extension.ListResult;
 import run.halo.app.extension.ReactiveExtensionClient;
 import run.halo.app.extension.Ref;
+import run.halo.app.infra.AnonymousUserConst;
 import run.halo.app.theme.finders.CommentFinder;
 import run.halo.app.theme.finders.Finder;
 import run.halo.app.theme.finders.vo.CommentVo;
@@ -45,35 +51,40 @@ public Mono<CommentVo> getByName(String name) {
 
     @Override
     public Mono<ListResult<CommentVo>> list(Ref ref, Integer page, Integer size) {
-        return client.list(Comment.class, fixedPredicate(ref),
-                defaultComparator(),
-                pageNullSafe(page), sizeNullSafe(size))
-            .flatMap(list -> Flux.fromStream(list.get().map(this::toCommentVo))
-                .concatMap(Function.identity())
-                .collectList()
-                .map(commentVos -> new ListResult<>(list.getPage(), list.getSize(), list.getTotal(),
-                    commentVos)
-                )
-            )
-            .defaultIfEmpty(new ListResult<>(page, size, 0L, List.of()));
+        return fixedCommentPredicate(ref)
+            .flatMap(fixedPredicate ->
+                client.list(Comment.class, fixedPredicate,
+                        defaultComparator(),
+                        pageNullSafe(page), sizeNullSafe(size))
+                    .flatMap(list -> Flux.fromStream(list.get().map(this::toCommentVo))
+                        .concatMap(Function.identity())
+                        .collectList()
+                        .map(commentVos -> new ListResult<>(list.getPage(), list.getSize(),
+                            list.getTotal(),
+                            commentVos)
+                        )
+                    )
+                    .defaultIfEmpty(new ListResult<>(page, size, 0L, List.of()))
+            );
     }
 
     @Override
     public Mono<ListResult<ReplyVo>> listReply(String commentName, Integer page, Integer size) {
         Comparator<Reply> comparator =
             Comparator.comparing(reply -> reply.getMetadata().getCreationTimestamp());
-        return client.list(Reply.class,
-                reply -> reply.getSpec().getCommentName().equals(commentName)
-                    && Objects.equals(false, reply.getSpec().getHidden())
-                    && Objects.equals(true, reply.getSpec().getApproved()),
-                comparator.reversed(), pageNullSafe(page), sizeNullSafe(size))
-            .flatMap(list -> Flux.fromStream(list.get().map(this::toReplyVo))
-                .concatMap(Function.identity())
-                .collectList()
-                .map(replyVos -> new ListResult<>(list.getPage(), list.getSize(), list.getTotal(),
-                    replyVos))
-            )
-            .defaultIfEmpty(new ListResult<>(page, size, 0L, List.of()));
+        return fixedReplyPredicate(commentName)
+            .flatMap(fixedPredicate ->
+                client.list(Reply.class, fixedPredicate,
+                        comparator.reversed(), pageNullSafe(page), sizeNullSafe(size))
+                    .flatMap(list -> Flux.fromStream(list.get().map(this::toReplyVo))
+                        .concatMap(Function.identity())
+                        .collectList()
+                        .map(replyVos -> new ListResult<>(list.getPage(), list.getSize(),
+                            list.getTotal(),
+                            replyVos))
+                    )
+                    .defaultIfEmpty(new ListResult<>(page, size, 0L, List.of()))
+            );
     }
 
     private Mono<CommentVo> toCommentVo(Comment comment) {
@@ -102,11 +113,56 @@ private Mono<OwnerInfo> getOwnerInfo(Comment.CommentOwner owner) {
             .defaultIfEmpty(OwnerInfo.ghostUser());
     }
 
-    private Predicate<Comment> fixedPredicate(Ref ref) {
+    private Mono<Predicate<Comment>> fixedCommentPredicate(Ref ref) {
         Assert.notNull(ref, "Comment subject reference must not be null");
-        return comment -> comment.getSpec().getSubjectRef().equals(ref)
-            && Objects.equals(false, comment.getSpec().getHidden())
-            && Objects.equals(true, comment.getSpec().getApproved());
+        // Ref must be equal to the comment subject
+        Predicate<Comment> refPredicate = comment -> comment.getSpec().getSubjectRef().equals(ref)
+            && comment.getMetadata().getDeletionTimestamp() == null;
+
+        // is approved and not hidden
+        Predicate<Comment> approvedPredicate =
+            comment -> BooleanUtils.isFalse(comment.getSpec().getHidden())
+                && BooleanUtils.isTrue(comment.getSpec().getApproved());
+        return getCurrentUserWithoutAnonymous()
+            .map(username -> {
+                Predicate<Comment> isOwner = comment -> {
+                    Comment.CommentOwner owner = comment.getSpec().getOwner();
+                    return owner != null && StringUtils.equals(username, owner.getName());
+                };
+                return approvedPredicate.or(isOwner);
+            })
+            .defaultIfEmpty(approvedPredicate)
+            .map(refPredicate::and);
+    }
+
+    private Mono<Predicate<Reply>> fixedReplyPredicate(String commentName) {
+        Assert.notNull(commentName, "The commentName must not be null");
+        // The comment name must be equal to the comment name of the reply
+        Predicate<Reply> commentNamePredicate =
+            reply -> reply.getSpec().getCommentName().equals(commentName)
+                && reply.getMetadata().getDeletionTimestamp() == null;
+
+        // is approved and not hidden
+        Predicate<Reply> approvedPredicate =
+            reply -> BooleanUtils.isFalse(reply.getSpec().getHidden())
+                && BooleanUtils.isTrue(reply.getSpec().getApproved());
+        return getCurrentUserWithoutAnonymous()
+            .map(username -> {
+                Predicate<Reply> isOwner = reply -> {
+                    Comment.CommentOwner owner = reply.getSpec().getOwner();
+                    return owner != null && StringUtils.equals(username, owner.getName());
+                };
+                return approvedPredicate.or(isOwner);
+            })
+            .defaultIfEmpty(approvedPredicate)
+            .map(commentNamePredicate::and);
+    }
+
+    Mono<String> getCurrentUserWithoutAnonymous() {
+        return ReactiveSecurityContextHolder.getContext()
+            .mapNotNull(SecurityContext::getAuthentication)
+            .map(Principal::getName)
+            .filter(username -> !AnonymousUserConst.PRINCIPAL.equals(username));
     }
 
     static Comparator<Comment> defaultComparator() {