lxq/mydreamjob-server/src/test/java/com/mydreamjob/onboarding/OnboardingConcurrencyIntegrationTest.java

package com.mydreamjob.onboarding;

import com.mydreamjob.auth.User;
import com.mydreamjob.auth.UserRepository;
import com.mydreamjob.common.exception.BusinessException;
import com.mydreamjob.common.exception.ErrorCode;
import com.mydreamjob.company.Company;
import com.mydreamjob.company.CompanyRepository;
import com.mydreamjob.onboarding.dto.CompanyChecklistItemCreateRequest;
import com.mydreamjob.onboarding.dto.CompanyChecklistItemView;
import com.mydreamjob.onboarding.dto.CompanyKnowledgeUpdateRequest;
import com.mydreamjob.onboarding.dto.CompanyKnowledgeView;
import com.mydreamjob.onboarding.dto.ProgressUpdateRequest;
import com.mydreamjob.onboarding.dto.ProgressUpdateResponse;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.boot.test.mock.mockito.MockBean;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.test.context.ActiveProfiles;
import org.springframework.test.context.TestPropertySource;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

import static org.assertj.core.api.Assertions.assertThat;

@SpringBootTest
@ActiveProfiles("integration-test")
@TestPropertySource(properties =
        "spring.datasource.url=jdbc:h2:mem:onboarding_concurrency;MODE=MySQL;DB_CLOSE_DELAY=-1;DB_CLOSE_ON_EXIT=FALSE")
class OnboardingConcurrencyIntegrationTest {

    private static final AtomicLong PHONE_SEQ = new AtomicLong(972_000_000L);
    private static final AtomicLong ITEM_SEQ = new AtomicLong(1L);

    @Autowired private OnboardingChecklistService checklistService;
    @Autowired private CompanyKnowledgeService knowledgeService;
    @Autowired private UserRepository userRepository;
    @Autowired private CompanyRepository companyRepository;
    @Autowired private OnboardingChecklistItemRepository checklistItemRepository;
    @Autowired private StudentOnboardingProgressRepository progressRepository;
    @Autowired private CompanyKnowledgeRepository knowledgeRepository;

    @MockBean private RedisTemplate<String, Object> redisTemplate;

    @BeforeEach
    void clearData() {
        progressRepository.deleteAll();
        knowledgeRepository.deleteAll();
        checklistItemRepository.deleteAll();
        companyRepository.deleteAll();
        userRepository.deleteAll();
    }

    @Test
    void identicalIdempotencyKeyConcurrentCreatesReplaySameRow() throws Exception {
        CompanyOwner owner = createCompanyOwner("幂等企业");
        CompanyChecklistItemCreateRequest request = createRequest("重复请求", "same-key");

        List<Attempt<CompanyChecklistItemView>> attempts = runConcurrently(
                () -> checklistService.createCompanyChecklistItem(owner.userId(), request),
                () -> checklistService.createCompanyChecklistItem(owner.userId(), request));

        assertThat(attempts).allSatisfy(attempt -> assertThat(attempt.error()).isNull());
        assertThat(attempts).extracting(attempt -> attempt.value().getItemId())
                .containsOnly(attempts.get(0).value().getItemId());
        assertThat(companyRows(owner.companyId()).stream()
                .filter(item -> "same-key".equals(item.getIdempotencyKey())))
                .hasSize(1);
    }

    @Test
    void idempotencyKeyRaceWithDifferentBodiesReturns9704() throws Exception {
        CompanyOwner owner = createCompanyOwner("幂等冲突企业");

        List<Attempt<CompanyChecklistItemView>> attempts = runConcurrently(
                () -> checklistService.createCompanyChecklistItem(
                        owner.userId(), createRequest("请求A", "conflict-key")),
                () -> checklistService.createCompanyChecklistItem(
                        owner.userId(), createRequest("请求B", "conflict-key")));

        assertThat(attempts).filteredOn(attempt -> attempt.error() == null).hasSize(1);
        assertThat(attempts).filteredOn(Attempt::failedWithIdempotencyConflict).hasSize(1);
        assertThat(companyRows(owner.companyId()).stream()
                .filter(item -> "conflict-key".equals(item.getIdempotencyKey())))
                .hasSize(1);
    }

    @Test
    void concurrentCreatesAtLimitBoundaryNeverWriteFiftyFirstItem() throws Exception {
        CompanyOwner owner = createCompanyOwner("上限企业");
        seedCompanyItems(owner.companyId(), 49);

        List<Attempt<CompanyChecklistItemView>> attempts = runConcurrently(
                () -> checklistService.createCompanyChecklistItem(
                        owner.userId(), createRequest("边界A", "limit-a")),
                () -> checklistService.createCompanyChecklistItem(
                        owner.userId(), createRequest("边界B", "limit-b")));

        assertThat(attempts).filteredOn(attempt -> attempt.error() == null).hasSize(1);
        assertThat(attempts).filteredOn(Attempt::failedWithItemLimit).hasSize(1);
        assertThat(companyRows(owner.companyId()))
                .filteredOn(item -> Boolean.TRUE.equals(item.getActive()))
                .hasSize(50);
    }

    @Test
    void inactiveItemsDoNotConsumeLimitAndExistingKeyStillReplays() throws Exception {
        CompanyOwner owner = createCompanyOwner("停用项企业");
        seedCompanyItems(owner.companyId(), 49);
        CompanyChecklistItemCreateRequest fiftieth = createRequest("第50条", "replay-key");
        CompanyChecklistItemView created = checklistService.createCompanyChecklistItem(
                owner.userId(), fiftieth);
        seedInactiveCompanyItems(owner.companyId(), 3);

        CompanyChecklistItemCreateRequest newRequest = createRequest("第51条", "new-over-limit");
        assertThat(failedWith(() -> checklistService.createCompanyChecklistItem(
                owner.userId(), newRequest))).isEqualTo(
                ErrorCode.ONBOARDING_ITEM_LIMIT_EXCEEDED.getCode());

        CompanyChecklistItemView replay = checklistService.createCompanyChecklistItem(
                owner.userId(), fiftieth);
        assertThat(replay.getItemId()).isEqualTo(created.getItemId());
        assertThat(companyRows(owner.companyId()).stream()
                .filter(item -> Boolean.TRUE.equals(item.getActive())))
                .hasSize(50);
    }

    @Test
    void sameIdempotencyKeyIsIsolatedByCompany() throws Exception {
        CompanyOwner first = createCompanyOwner("企业甲");
        CompanyOwner second = createCompanyOwner("企业乙");
        CompanyChecklistItemCreateRequest request = createRequest("独立企业请求", "shared-key");

        List<Attempt<CompanyChecklistItemView>> attempts = runConcurrently(
                () -> checklistService.createCompanyChecklistItem(first.userId(), request),
                () -> checklistService.createCompanyChecklistItem(second.userId(), request));

        assertThat(attempts).allSatisfy(attempt -> assertThat(attempt.error()).isNull());
        assertThat(companyRows(first.companyId())).hasSize(1);
        assertThat(companyRows(second.companyId())).hasSize(1);
    }

    @Test
    void firstProgressCreateWithNonZeroVersionReturns9705() {
        CompanyOwner owner = createCompanyOwner("进度版本企业");
        Long student = createUser(User.Role.STUDENT);
        OnboardingChecklistItem item = createStandardItem("进度版本项");

        ProgressUpdateRequest request = progressRequest(true, 1L);

        assertThat(failedWith(() -> checklistService.updateCompletion(
                student, owner.companyId(), item.getId(), request)))
                .isEqualTo(ErrorCode.ONBOARDING_PROGRESS_CONCURRENT_UPDATE.getCode());
        assertThat(progressRepository.findByStudentUserIdAndItemId(student, item.getId()))
                .isEmpty();
    }

    @Test
    void concurrentFirstProgressCreatesProduceSingleRow() throws Exception {
        CompanyOwner owner = createCompanyOwner("首次进度企业");
        Long student = createUser(User.Role.STUDENT);
        OnboardingChecklistItem item = createStandardItem("首次进度项");

        List<Callable<ProgressUpdateResponse>> calls = new ArrayList<>();
        for (int i = 0; i < 4; i++) {
            calls.add(() -> checklistService.updateCompletion(
                    student, owner.companyId(), item.getId(), progressRequest(true, 0L)));
        }

        List<Attempt<ProgressUpdateResponse>> attempts = runConcurrently(calls);

        assertThat(attempts).allSatisfy(attempt -> assertThat(attempt.error()).isNull());
        assertThat(progressRepository.findByStudentUserIdAndItemId(student, item.getId()))
                .isPresent();
        assertThat(progressRepository.findAll().stream()
                .filter(row -> row.getStudentUserId().equals(student)
                        && row.getItemId().equals(item.getId())))
                .hasSize(1);
    }

    @Test
    void simultaneousFirstKnowledgeUpsertsAreDeterministic() throws Exception {
        CompanyOwner owner = createCompanyOwner("知识并发企业");
        CompanyKnowledgeUpdateRequest request = knowledgeRequest("共同文化", null, null, null);

        List<Attempt<CompanyKnowledgeView>> attempts = runConcurrently(
                () -> knowledgeService.upsertCompanyKnowledge(owner.userId(), request),
                () -> knowledgeService.upsertCompanyKnowledge(owner.userId(), request));

        assertThat(attempts).allSatisfy(attempt -> assertThat(attempt.error()).isNull());
        assertThat(knowledgeRepository.findByCompanyId(owner.companyId())).isPresent();
        assertThat(knowledgeRepository.findAll().stream()
                .filter(row -> row.getCompanyId().equals(owner.companyId())))
                .hasSize(1);
    }

    @Test
    void identicalKnowledgeCreateReplayWithoutVersionIsIdempotent() {
        CompanyOwner owner = createCompanyOwner("知识重放企业");
        CompanyKnowledgeUpdateRequest request = knowledgeRequest("重放文化", "制度", null, null);

        knowledgeService.upsertCompanyKnowledge(owner.userId(), request);
        CompanyKnowledgeView replay = knowledgeService.upsertCompanyKnowledge(owner.userId(), request);

        assertThat(replay.getConfigured()).isTrue();
        assertThat(replay.getVersion()).isZero();
        assertThat(knowledgeRepository.findAll().stream()
                .filter(row -> row.getCompanyId().equals(owner.companyId())))
                .hasSize(1);
    }

    @Test
    void concurrentKnowledgeUpdatesWithSameVersionReturn9706() throws Exception {
        CompanyOwner owner = createCompanyOwner("知识版本企业");
        knowledgeRepository.save(CompanyKnowledge.builder()
                .companyId(owner.companyId())
                .culture("初始文化")
                .build());

        List<Attempt<CompanyKnowledgeView>> attempts = runConcurrently(
                () -> knowledgeService.upsertCompanyKnowledge(
                        owner.userId(), knowledgeRequest("更新A", null, null, 0L)),
                () -> knowledgeService.upsertCompanyKnowledge(
                        owner.userId(), knowledgeRequest("更新B", null, null, 0L)));

        assertThat(attempts).filteredOn(attempt -> attempt.error() == null).hasSize(1);
        assertThat(attempts).filteredOn(Attempt::failedWithKnowledgeConflict).hasSize(1);
        assertThat(knowledgeRepository.findByCompanyId(owner.companyId()).orElseThrow().getVersion())
                .isEqualTo(1L);
    }

    private int failedWith(ThrowingSupplier<?> supplier) {
        try {
            supplier.get();
            return -1;
        } catch (BusinessException ex) {
            return ex.getErrorCodeInt();
        }
    }

    private CompanyChecklistItemCreateRequest createRequest(String title, String idempotencyKey) {
        CompanyChecklistItemCreateRequest request = new CompanyChecklistItemCreateRequest();
        request.setTitle(title);
        request.setDescription(title + "说明");
        request.setCategory("ADMIN");
        request.setSortOrder(200);
        request.setRequired(false);
        request.setIdempotencyKey(idempotencyKey);
        return request;
    }

    private ProgressUpdateRequest progressRequest(boolean completed, Long version) {
        ProgressUpdateRequest request = new ProgressUpdateRequest();
        request.setCompleted(completed);
        request.setVersion(version);
        return request;
    }

    private CompanyKnowledgeUpdateRequest knowledgeRequest(
            String culture, String workPolicies, String onboardingGuide, Long version) {
        CompanyKnowledgeUpdateRequest request = new CompanyKnowledgeUpdateRequest();
        request.setCulture(culture);
        request.setWorkPolicies(workPolicies);
        request.setOnboardingGuide(onboardingGuide);
        request.setVersion(version);
        return request;
    }

    private List<OnboardingChecklistItem> seedCompanyItems(Long companyId, int count) {
        List<OnboardingChecklistItem> items = new ArrayList<>();
        for (int i = 0; i < count; i++) {
            items.add(OnboardingChecklistItem.builder()
                    .scopeType("COMPANY")
                    .scopeId(companyId)
                    .companyId(companyId)
                    .itemKey("seed-" + companyId + "-" + ITEM_SEQ.getAndIncrement())
                    .title("种子项-" + i)
                    .description("种子说明")
                    .category("OTHER")
                    .sortOrder(1000 + i)
                    .required(false)
                    .active(true)
                    .idempotencyKey("seed-key-" + companyId + "-" + i)
                    .requestHash("a".repeat(64))
                    .build());
        }
        return checklistItemRepository.saveAllAndFlush(items);
    }

    private void seedInactiveCompanyItems(Long companyId, int count) {
        List<OnboardingChecklistItem> items = new ArrayList<>();
        for (int i = 0; i < count; i++) {
            items.add(OnboardingChecklistItem.builder()
                    .scopeType("COMPANY")
                    .scopeId(companyId)
                    .companyId(companyId)
                    .itemKey("inactive-" + companyId + "-" + ITEM_SEQ.getAndIncrement())
                    .title("停用项-" + i)
                    .category("OTHER")
                    .sortOrder(2000 + i)
                    .required(false)
                    .active(false)
                    .build());
        }
        checklistItemRepository.saveAllAndFlush(items);
    }

    private OnboardingChecklistItem createStandardItem(String title) {
        return checklistItemRepository.saveAndFlush(OnboardingChecklistItem.builder()
                .scopeType("STANDARD")
                .scopeId(0L)
                .companyId(null)
                .itemKey("standard-" + ITEM_SEQ.getAndIncrement())
                .title(title)
                .description("通用说明")
                .category("DOCUMENT")
                .sortOrder(10)
                .required(true)
                .active(true)
                .build());
    }

    private List<OnboardingChecklistItem> companyRows(Long companyId) {
        return checklistItemRepository.findAll().stream()
                .filter(item -> companyId.equals(item.getCompanyId()))
                .toList();
    }

    private CompanyOwner createCompanyOwner(String name) {
        Long userId = createUser(User.Role.COMPANY);
        Long companyId = companyRepository.save(Company.builder()
                .ownerUserId(userId)
                .name(name)
                .verified(true)
                .build()).getId();
        return new CompanyOwner(userId, companyId);
    }

    private Long createUser(User.Role role) {
        return userRepository.save(User.builder()
                .phone(String.valueOf(PHONE_SEQ.getAndIncrement()))
                .nickname(role.name())
                .role(role)
                .roles(Set.of(role))
                .active(true)
                .verified(true)
                .build()).getId();
    }

    @SafeVarargs
    private final <T> List<Attempt<T>> runConcurrently(Callable<T>... actions) throws Exception {
        return runConcurrently(List.of(actions));
    }

    private <T> List<Attempt<T>> runConcurrently(List<Callable<T>> actions) throws Exception {
        ExecutorService pool = Executors.newFixedThreadPool(actions.size());
        CountDownLatch ready = new CountDownLatch(actions.size());
        CountDownLatch start = new CountDownLatch(1);
        List<Future<Attempt<T>>> futures = new ArrayList<>();
        try {
            for (Callable<T> action : actions) {
                futures.add(pool.submit(() -> {
                    ready.countDown();
                    if (!start.await(10, TimeUnit.SECONDS)) {
                        throw new IllegalStateException("concurrent test start timed out");
                    }
                    try {
                        return new Attempt<>(action.call(), null);
                    } catch (Throwable ex) {
                        return new Attempt<>(null, ex);
                    }
                }));
            }
            assertThat(ready.await(10, TimeUnit.SECONDS)).isTrue();
            start.countDown();

            List<Attempt<T>> results = new ArrayList<>();
            for (Future<Attempt<T>> future : futures) {
                results.add(future.get(20, TimeUnit.SECONDS));
            }
            return results;
        } finally {
            pool.shutdownNow();
        }
    }

    private record CompanyOwner(Long userId, Long companyId) {
    }

    private record Attempt<T>(T value, Throwable error) {
        boolean failedWithIdempotencyConflict() {
            return error instanceof BusinessException ex
                    && ex.getErrorCodeInt() == ErrorCode.ONBOARDING_IDEMPOTENCY_CONFLICT.getCode();
        }

        boolean failedWithItemLimit() {
            return error instanceof BusinessException ex
                    && ex.getErrorCodeInt() == ErrorCode.ONBOARDING_ITEM_LIMIT_EXCEEDED.getCode();
        }

        boolean failedWithKnowledgeConflict() {
            return error instanceof BusinessException ex
                    && ex.getErrorCodeInt()
                    == ErrorCode.ONBOARDING_KNOWLEDGE_CONCURRENT_UPDATE.getCode();
        }
    }

    @FunctionalInterface
    private interface ThrowingSupplier<T> {
        T get();
    }
}