optimize: subject update sync logic. (#632)

* optimize: subject update sync logic.

* optimize: upgrade version to 0.14.2 and update CHANGELOG.MD

CHANGELOG.MD

@@ -2,6 +2,12 @@
 
 更新日志文档，版本顺序从新到旧，最新版本在最前(上)面。
 
+# 0.14.2
+
+## 问题修复
+
+- 条目更新乐观锁问题 #631
+
 # 0.14.1
 
 ## 优化

gradle.properties

@@ -1 +1 @@
-version=0.14.1
+version=0.14.2

server/src/main/java/run/ikaros/server/core/subject/service/impl/SubjectServiceImpl.java

@@ -315,20 +315,31 @@ public Mono<Void> update(Subject subject) {
             .flatMap(episodeRepository::save)
             .checkpoint("UpdateEpisodeEntities")
 
-            // 条目同步的更新逻辑是: 不存在则新增，存在则更新
+            // 条目同步的更新逻辑是: 不存在则新增，存在则忽略
             .then(Mono.justOrEmpty(subject.getSyncs()))
             .flatMapMany(subjectSyncs -> Flux.fromStream(subjectSyncs.stream()))
             .flatMap(subjectSync -> copyProperties(subjectSync, new SubjectSyncEntity()))
             .map(entity -> entity.setSubjectId(subjectId.get()))
-            .flatMap(entity -> subjectSyncRepository.findBySubjectIdAndPlatformAndPlatformId(
-                entity.getSubjectId(), entity.getPlatform(), entity.getPlatformId()
-            ).switchIfEmpty(Mono.just(entity)
-                .doOnSuccess(e2 -> log.debug("create new subject sync record: [{}].", e2))))
-            .flatMap(subjectSyncRepository::save)
+            .flatMap(this::updateSubjectSyncEntity)
             .checkpoint("UpdateSubjectSyncEntities")
             .then();
     }
 
+    /**
+     * 不存在则新增，存在则忽略.
+     */
+    private Mono<SubjectSyncEntity> updateSubjectSyncEntity(SubjectSyncEntity entity) {
+        Assert.notNull(entity, "'subjectSyncEntity' must not null.");
+        return subjectSyncRepository.existsBySubjectIdAndPlatformAndPlatformId(
+                entity.getSubjectId(), entity.getPlatform(), entity.getPlatformId())
+            .filter(exists -> exists)
+            .flatMap(exists -> subjectSyncRepository.findBySubjectIdAndPlatformAndPlatformId(
+                entity.getSubjectId(), entity.getPlatform(), entity.getPlatformId()))
+            .switchIfEmpty(Mono.just(entity)
+                .doOnSuccess(e2 -> log.debug("create new subject sync record: [{}].", e2))
+                .flatMap(subjectSyncRepository::save));
+    }
+
     @Override
     public Mono<Void> deleteById(Long id) {
         Assert.isTrue(id > 0, "'id' must gt 0.");

server/src/main/java/run/ikaros/server/store/repository/SubjectSyncRepository.java

@@ -17,6 +17,10 @@ Mono<SubjectSyncEntity> findBySubjectIdAndPlatformAndPlatformId(Long subjectId,
                                                                     SubjectSyncPlatform platform,
                                                                     String platformId);
 
+    Mono<Boolean> existsBySubjectIdAndPlatformAndPlatformId(Long subjectId,
+                                                            SubjectSyncPlatform platform,
+                                                            String platformId);
+
     Mono<Boolean> existsByPlatformAndPlatformId(SubjectSyncPlatform platform, String platformId);
 
     Flux<SubjectSyncEntity> findAllBySubjectId(Long subjectId);

server/src/test/java/run/ikaros/server/core/subject/SubjectServiceTest.java

@@ -7,6 +7,7 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Objects;
+import java.util.Random;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReference;
 import org.junit.jupiter.api.AfterEach;
@@ -207,5 +208,83 @@ void update() {
             .verifyComplete();
 
 
+    }
+
+    @Test
+    void updateLarge() {
+        Subject subject = createSubjectInstance();
+        final Random random = new Random();
+
+        StepVerifier.create(subjectService.create(subject))
+            .expectNext(subject)
+            .verifyComplete();
+
+        assertThat(subject.getId()).isNotNull();
+
+        // update
+        final String newName = "subject-name-unit-test";
+        subject.setName(newName);
+
+
+        List<Episode> episodes = subject.getEpisodes();
+        episodes.clear();
+        final int epCount = 100;
+        for (int i = 0; i < epCount; i++) {
+            Episode episode = Episode.builder()
+                .airTime(LocalDateTime.now())
+                .sequence(i)
+                .group(EpisodeGroup.MAIN)
+                .name("ep-" + i)
+                .nameCn("第二集").build();
+            episodes.add(episode);
+        }
+
+        final List<SubjectSync> syncs = subject.getSyncs();
+
+        SubjectSync aniDbSync = new SubjectSync();
+        aniDbSync.setSyncTime(LocalDateTime.now());
+        aniDbSync.setPlatform(SubjectSyncPlatform.AniDB);
+        aniDbSync.setSubjectId(subject.getId());
+        aniDbSync.setPlatformId(String.valueOf(random.nextInt(0, 30000)));
+        syncs.add(aniDbSync);
+        SubjectSync bgmtvSync = new SubjectSync();
+        bgmtvSync.setSyncTime(LocalDateTime.now());
+        bgmtvSync.setPlatform(SubjectSyncPlatform.BGM_TV);
+        bgmtvSync.setSubjectId(subject.getId());
+        bgmtvSync.setPlatformId(String.valueOf(random.nextInt(0, 30000)));
+        syncs.add(bgmtvSync);
+        SubjectSync tmDbSync = new SubjectSync();
+        tmDbSync.setSyncTime(LocalDateTime.now());
+        tmDbSync.setPlatform(SubjectSyncPlatform.TMDB);
+        tmDbSync.setSubjectId(subject.getId());
+        tmDbSync.setPlatformId(String.valueOf(random.nextInt(0, 30000)));
+        syncs.add(tmDbSync);
+
+        subject.setSyncs(syncs);
+
+        StepVerifier.create(subjectService.update(subject))
+            .verifyComplete();
+
+        StepVerifier.create(subjectService.findById(subject.getId()))
+            .expectNextMatches(newSubject -> {
+                    boolean result = true;
+                    result = newName.equals(newSubject.getName());
+                    result = newSubject.getTotalEpisodes() == epCount;
+                    if (!result) {
+                        return false;
+                    }
+
+                    List<Episode> newEps = newSubject.getEpisodes();
+                    result = newEps.size() == epCount;
+
+                    List<SubjectSync> newSyncs = newSubject.getSyncs();
+
+                    result = newSyncs.size() == syncs.size();
+                    return result;
+                }
+            )
+            .verifyComplete();
+
+
     }
 }