From d81c29738120b052ddd52ed9ec1c50fbd68c5a91 Mon Sep 17 00:00:00 2001 From: wangziqi Date: Fri, 31 Jul 2026 16:26:39 +0800 Subject: [PATCH] fix(education): consolidate question import persistence --- .../QuestionImportJobController.java | 3 +- .../vo/QuestionImportPreviewReqVO.java | 5 +- .../importjob/QuestionImportJobDO.java | 32 ---- .../importjob/ContentImportAssetMapper.java | 7 + .../importjob/ContentImportJobMapper.java | 46 +++++ .../importjob/QuestionImportJobMapper.java | 54 ------ .../PreviewQuestionImportCommand.java | 3 +- .../QuestionImportJobServiceImpl.java | 169 +++++++++++------- .../QuestionImportJobControllerTest.java | 4 +- .../QuestionImportJobServiceImplTest.java | 123 +++++++------ 10 files changed, 218 insertions(+), 228 deletions(-) delete mode 100644 yudao-module-education/src/main/java/cn/iocoder/yudao/module/education/dal/dataobject/importjob/QuestionImportJobDO.java delete mode 100644 yudao-module-education/src/main/java/cn/iocoder/yudao/module/education/dal/mysql/importjob/QuestionImportJobMapper.java diff --git a/yudao-module-education/src/main/java/cn/iocoder/yudao/module/education/controller/admin/importjob/QuestionImportJobController.java b/yudao-module-education/src/main/java/cn/iocoder/yudao/module/education/controller/admin/importjob/QuestionImportJobController.java index 580007ec..896e9a58 100644 --- a/yudao-module-education/src/main/java/cn/iocoder/yudao/module/education/controller/admin/importjob/QuestionImportJobController.java +++ b/yudao-module-education/src/main/java/cn/iocoder/yudao/module/education/controller/admin/importjob/QuestionImportJobController.java @@ -28,8 +28,7 @@ public class QuestionImportJobController { @PreAuthorize("@ss.hasPermission('education:question-import:create')") public CommonResult preview(@Valid @RequestBody QuestionImportPreviewReqVO request) { return success(QuestionImportJobRespVO.from(service.requestPreview(new PreviewQuestionImportCommand( - request.getCommandKey(), request.getObjectKey(), request.getFileName(), request.getMediaType(), - request.getFileSize()), getLoginUserId()))); + request.getCommandKey(), request.getAssetId()), getLoginUserId()))); } @PostMapping("/{id}/execute") diff --git a/yudao-module-education/src/main/java/cn/iocoder/yudao/module/education/controller/admin/importjob/vo/QuestionImportPreviewReqVO.java b/yudao-module-education/src/main/java/cn/iocoder/yudao/module/education/controller/admin/importjob/vo/QuestionImportPreviewReqVO.java index b1f07f9a..57ce9679 100644 --- a/yudao-module-education/src/main/java/cn/iocoder/yudao/module/education/controller/admin/importjob/vo/QuestionImportPreviewReqVO.java +++ b/yudao-module-education/src/main/java/cn/iocoder/yudao/module/education/controller/admin/importjob/vo/QuestionImportPreviewReqVO.java @@ -8,8 +8,5 @@ import lombok.Data; @Data public class QuestionImportPreviewReqVO { @NotBlank private String commandKey; - @NotBlank private String objectKey; - @NotBlank private String fileName; - @NotBlank private String mediaType; - @NotNull @Positive private Long fileSize; + @NotNull @Positive private Long assetId; } diff --git a/yudao-module-education/src/main/java/cn/iocoder/yudao/module/education/dal/dataobject/importjob/QuestionImportJobDO.java b/yudao-module-education/src/main/java/cn/iocoder/yudao/module/education/dal/dataobject/importjob/QuestionImportJobDO.java deleted file mode 100644 index 020af382..00000000 --- a/yudao-module-education/src/main/java/cn/iocoder/yudao/module/education/dal/dataobject/importjob/QuestionImportJobDO.java +++ /dev/null @@ -1,32 +0,0 @@ -package cn.iocoder.yudao.module.education.dal.dataobject.importjob; - -import cn.iocoder.yudao.framework.tenant.core.db.TenantBaseDO; -import com.baomidou.mybatisplus.annotation.KeySequence; -import com.baomidou.mybatisplus.annotation.TableId; -import com.baomidou.mybatisplus.annotation.TableName; -import lombok.Data; -import lombok.EqualsAndHashCode; - -@TableName("education_question_import_job") -@KeySequence("education_question_import_job_seq") -@Data -@EqualsAndHashCode(callSuper = true) -public class QuestionImportJobDO extends TenantBaseDO { - @TableId private Long id; - private Long actorId; - private String objectKey; - private String fileName; - private String mediaType; - private Long fileSize; - private String previewCommandKey; - private String previewRequestHash; - private String executeCommandKey; - private String executeRequestHash; - private String status; - private String scanStatus; - private String parserStatus; - private String parsedPayload; - private Integer previewQuestionCount; - private Integer importedQuestionCount; - private String failureCode; -} diff --git a/yudao-module-education/src/main/java/cn/iocoder/yudao/module/education/dal/mysql/importjob/ContentImportAssetMapper.java b/yudao-module-education/src/main/java/cn/iocoder/yudao/module/education/dal/mysql/importjob/ContentImportAssetMapper.java index a3b99db3..5e7901f5 100644 --- a/yudao-module-education/src/main/java/cn/iocoder/yudao/module/education/dal/mysql/importjob/ContentImportAssetMapper.java +++ b/yudao-module-education/src/main/java/cn/iocoder/yudao/module/education/dal/mysql/importjob/ContentImportAssetMapper.java @@ -1,9 +1,16 @@ package cn.iocoder.yudao.module.education.dal.mysql.importjob; import cn.iocoder.yudao.framework.mybatis.core.mapper.BaseMapperX; +import cn.iocoder.yudao.framework.mybatis.core.query.LambdaQueryWrapperX; import cn.iocoder.yudao.module.education.dal.dataobject.importjob.ContentImportAssetDO; import org.apache.ibatis.annotations.Mapper; @Mapper public interface ContentImportAssetMapper extends BaseMapperX { + + default ContentImportAssetDO selectTenantAsset(Long tenantId, Long id) { + return selectOne(new LambdaQueryWrapperX() + .eq(ContentImportAssetDO::getTenantId, tenantId) + .eq(ContentImportAssetDO::getId, id)); + } } diff --git a/yudao-module-education/src/main/java/cn/iocoder/yudao/module/education/dal/mysql/importjob/ContentImportJobMapper.java b/yudao-module-education/src/main/java/cn/iocoder/yudao/module/education/dal/mysql/importjob/ContentImportJobMapper.java index 22172b7c..6d208db2 100644 --- a/yudao-module-education/src/main/java/cn/iocoder/yudao/module/education/dal/mysql/importjob/ContentImportJobMapper.java +++ b/yudao-module-education/src/main/java/cn/iocoder/yudao/module/education/dal/mysql/importjob/ContentImportJobMapper.java @@ -1,6 +1,7 @@ package cn.iocoder.yudao.module.education.dal.mysql.importjob; import cn.iocoder.yudao.framework.mybatis.core.mapper.BaseMapperX; +import cn.iocoder.yudao.framework.mybatis.core.query.LambdaQueryWrapperX; import cn.iocoder.yudao.module.education.dal.dataobject.importjob.ContentImportJobDO; import org.apache.ibatis.annotations.*; @@ -9,6 +10,24 @@ import java.util.List; @Mapper public interface ContentImportJobMapper extends BaseMapperX { + default ContentImportJobDO selectByPreviewKey(Long tenantId, String previewKey) { + return selectOne(new LambdaQueryWrapperX() + .eq(ContentImportJobDO::getTenantId, tenantId) + .eq(ContentImportJobDO::getPreviewKey, previewKey)); + } + + default ContentImportJobDO selectByExecuteKey(Long tenantId, String executeKey) { + return selectOne(new LambdaQueryWrapperX() + .eq(ContentImportJobDO::getTenantId, tenantId) + .eq(ContentImportJobDO::getExecuteKey, executeKey)); + } + + default ContentImportJobDO selectTenantJob(Long tenantId, Long id) { + return selectOne(new LambdaQueryWrapperX() + .eq(ContentImportJobDO::getTenantId, tenantId) + .eq(ContentImportJobDO::getId, id)); + } + @Insert(""" INSERT INTO education_content_import_job (tenant_id, asset_id, actor_id, preview_key, preview_request_hash, import_type, status, @@ -33,6 +52,21 @@ public interface ContentImportJobMapper extends BaseMapperX int requestExecute(@Param("tenantId") Long tenantId, @Param("id") Long id, @Param("executeKey") String executeKey, @Param("executeRequestHash") String executeRequestHash); + @Select(""" + UPDATE education_content_import_job + SET locked_by=#{workerId}, lease_token=#{leaseToken}, + lease_expires_at=CURRENT_TIMESTAMP + CAST(#{leaseSeconds} || ' seconds' AS INTERVAL), + last_heartbeat_at=CURRENT_TIMESTAMP, attempt_count=attempt_count + 1, + next_attempt_at=NULL, started_at=COALESCE(started_at,CURRENT_TIMESTAMP), update_time=CURRENT_TIMESTAMP + WHERE id=#{id} AND deleted=false AND status=#{expectedStatus} AND attempt_count < max_attempts + AND (next_attempt_at IS NULL OR next_attempt_at <= CURRENT_TIMESTAMP) + AND (lease_expires_at IS NULL OR lease_expires_at <= CURRENT_TIMESTAMP) + RETURNING * + """) + ContentImportJobDO claimById(@Param("id") Long id, @Param("expectedStatus") String expectedStatus, + @Param("workerId") String workerId, @Param("leaseToken") String leaseToken, + @Param("leaseSeconds") long leaseSeconds); + @Select(""" WITH candidate AS ( SELECT id FROM education_content_import_job @@ -86,6 +120,18 @@ public interface ContentImportJobMapper extends BaseMapperX @Param("resultSummary") String resultSummary, @Param("importedQuestionCount") Integer importedQuestionCount, @Param("failureCode") String failureCode, @Param("failureMessage") String failureMessage); + @Update(""" + UPDATE education_content_import_job + SET status='FAILED', failure_code='ATTEMPTS_EXHAUSTED', + failure_message='Maximum import attempts exhausted', finished_at=CURRENT_TIMESTAMP, + locked_by=NULL, lease_token=NULL, lease_expires_at=NULL, last_heartbeat_at=NULL, + update_time=CURRENT_TIMESTAMP + WHERE deleted=false AND status IN ('PREVIEW_PENDING','EXECUTE_PENDING') + AND attempt_count >= max_attempts + AND (lease_expires_at IS NULL OR lease_expires_at <= CURRENT_TIMESTAMP) + """) + int failExhausted(); + @Update("UPDATE education_content_import_job SET lease_expires_at=CURRENT_TIMESTAMP-INTERVAL '1 second' WHERE id=#{id}") int expireLeaseForTest(@Param("id") Long id); } diff --git a/yudao-module-education/src/main/java/cn/iocoder/yudao/module/education/dal/mysql/importjob/QuestionImportJobMapper.java b/yudao-module-education/src/main/java/cn/iocoder/yudao/module/education/dal/mysql/importjob/QuestionImportJobMapper.java deleted file mode 100644 index 3e45a6d0..00000000 --- a/yudao-module-education/src/main/java/cn/iocoder/yudao/module/education/dal/mysql/importjob/QuestionImportJobMapper.java +++ /dev/null @@ -1,54 +0,0 @@ -package cn.iocoder.yudao.module.education.dal.mysql.importjob; - -import cn.iocoder.yudao.framework.mybatis.core.mapper.BaseMapperX; -import cn.iocoder.yudao.framework.mybatis.core.query.LambdaQueryWrapperX; -import cn.iocoder.yudao.module.education.dal.dataobject.importjob.QuestionImportJobDO; -import org.apache.ibatis.annotations.Mapper; - -@Mapper -public interface QuestionImportJobMapper extends BaseMapperX { - - default QuestionImportJobDO selectByPreviewCommand(Long tenantId, String commandKey) { - return selectOne(new LambdaQueryWrapperX() - .eq(QuestionImportJobDO::getTenantId, tenantId) - .eq(QuestionImportJobDO::getPreviewCommandKey, commandKey)); - } - - default QuestionImportJobDO selectByExecuteCommand(Long tenantId, String commandKey) { - return selectOne(new LambdaQueryWrapperX() - .eq(QuestionImportJobDO::getTenantId, tenantId) - .eq(QuestionImportJobDO::getExecuteCommandKey, commandKey)); - } - - default QuestionImportJobDO selectTenantJob(Long tenantId, Long id) { - return selectOne(new LambdaQueryWrapperX() - .eq(QuestionImportJobDO::getTenantId, tenantId).eq(QuestionImportJobDO::getId, id)); - } - - default QuestionImportJobDO claimPreview(Long id) { - QuestionImportJobDO job = selectById(id); - return job != null && "PREVIEW_PENDING".equals(job.getStatus()) ? job : null; - } - - default QuestionImportJobDO claimExecute(Long id) { - QuestionImportJobDO job = selectById(id); - return job != null && "EXECUTE_PENDING".equals(job.getStatus()) ? job : null; - } - - default int completePreview(QuestionImportJobDO job) { return updateById(job); } - default int failPreview(Long id, String failureCode) { - QuestionImportJobDO job = new QuestionImportJobDO(); - job.setId(id); job.setStatus("FAILED"); job.setFailureCode(failureCode); - return updateById(job); - } - default int completeExecute(Long id, int count) { - QuestionImportJobDO job = new QuestionImportJobDO(); - job.setId(id); job.setStatus("COMPLETED"); job.setImportedQuestionCount(count); - return updateById(job); - } - default int failExecute(Long id, String failureCode) { - QuestionImportJobDO job = new QuestionImportJobDO(); - job.setId(id); job.setStatus("FAILED"); job.setFailureCode(failureCode); - return updateById(job); - } -} diff --git a/yudao-module-education/src/main/java/cn/iocoder/yudao/module/education/service/importjob/PreviewQuestionImportCommand.java b/yudao-module-education/src/main/java/cn/iocoder/yudao/module/education/service/importjob/PreviewQuestionImportCommand.java index 7626cf84..5fe854be 100644 --- a/yudao-module-education/src/main/java/cn/iocoder/yudao/module/education/service/importjob/PreviewQuestionImportCommand.java +++ b/yudao-module-education/src/main/java/cn/iocoder/yudao/module/education/service/importjob/PreviewQuestionImportCommand.java @@ -1,5 +1,4 @@ package cn.iocoder.yudao.module.education.service.importjob; -public record PreviewQuestionImportCommand(String commandKey, String objectKey, String fileName, - String mediaType, Long fileSize) { +public record PreviewQuestionImportCommand(String commandKey, Long assetId) { } diff --git a/yudao-module-education/src/main/java/cn/iocoder/yudao/module/education/service/importjob/QuestionImportJobServiceImpl.java b/yudao-module-education/src/main/java/cn/iocoder/yudao/module/education/service/importjob/QuestionImportJobServiceImpl.java index fd06ecb5..8a217122 100644 --- a/yudao-module-education/src/main/java/cn/iocoder/yudao/module/education/service/importjob/QuestionImportJobServiceImpl.java +++ b/yudao-module-education/src/main/java/cn/iocoder/yudao/module/education/service/importjob/QuestionImportJobServiceImpl.java @@ -4,8 +4,10 @@ import cn.iocoder.yudao.framework.common.util.json.JsonUtils; import cn.iocoder.yudao.framework.tenant.core.context.TenantContextHolder; import cn.iocoder.yudao.framework.tenant.core.util.TenantUtils; import cn.iocoder.yudao.module.education.config.EducationProperties; -import cn.iocoder.yudao.module.education.dal.dataobject.importjob.QuestionImportJobDO; -import cn.iocoder.yudao.module.education.dal.mysql.importjob.QuestionImportJobMapper; +import cn.iocoder.yudao.module.education.dal.dataobject.importjob.ContentImportAssetDO; +import cn.iocoder.yudao.module.education.dal.dataobject.importjob.ContentImportJobDO; +import cn.iocoder.yudao.module.education.dal.mysql.importjob.ContentImportAssetMapper; +import cn.iocoder.yudao.module.education.dal.mysql.importjob.ContentImportJobMapper; import cn.iocoder.yudao.module.education.enums.CatalogProviderMode; import cn.iocoder.yudao.module.education.service.question.authoring.QuestionDraftCommand; import cn.iocoder.yudao.module.education.service.question.authoring.TenantQuestionLifecycleService; @@ -17,6 +19,7 @@ import java.security.MessageDigest; import java.util.HexFormat; import java.util.List; import java.util.Optional; +import java.util.UUID; import static cn.iocoder.yudao.framework.common.exception.util.ServiceExceptionUtil.exception; import static cn.iocoder.yudao.module.education.enums.ErrorCodeConstants.*; @@ -24,17 +27,23 @@ import static cn.iocoder.yudao.module.education.enums.ErrorCodeConstants.*; @Service public class QuestionImportJobServiceImpl implements QuestionImportJobService { + private static final int MAX_ATTEMPTS = 5; + private static final long LEASE_SECONDS = 60; + private final EducationProperties properties; - private final QuestionImportJobMapper jobMapper; + private final ContentImportAssetMapper assetMapper; + private final ContentImportJobMapper jobMapper; private final Optional scanGateway; private final Optional parser; private final TenantQuestionLifecycleService lifecycleService; - public QuestionImportJobServiceImpl(EducationProperties properties, QuestionImportJobMapper jobMapper, + public QuestionImportJobServiceImpl(EducationProperties properties, ContentImportAssetMapper assetMapper, + ContentImportJobMapper jobMapper, Optional scanGateway, Optional parser, TenantQuestionLifecycleService lifecycleService) { this.properties = properties; + this.assetMapper = assetMapper; this.jobMapper = jobMapper; this.scanGateway = scanGateway; this.parser = parser; @@ -45,26 +54,27 @@ public class QuestionImportJobServiceImpl implements QuestionImportJobService { @Transactional(rollbackFor = Exception.class) public QuestionImportJobProjection requestPreview(PreviewQuestionImportCommand command, Long actorId) { Long tenantId = TenantContextHolder.getRequiredTenantId(); + ContentImportAssetDO asset = assetMapper.selectTenantAsset(tenantId, command.assetId()); + if (asset == null) throw exception(QUESTION_IMPORT_NOT_FOUND); String requestHash = hash(command); - QuestionImportJobDO existing = jobMapper.selectByPreviewCommand(tenantId, command.commandKey()); + ContentImportJobDO existing = jobMapper.selectByPreviewKey(tenantId, command.commandKey()); if (existing != null) { requireHash(existing.getPreviewRequestHash(), requestHash); - return project(existing); + return project(existing, asset); } - QuestionImportJobDO job = new QuestionImportJobDO(); + ContentImportJobDO job = new ContentImportJobDO(); job.setTenantId(tenantId); + job.setAssetId(asset.getId()); job.setActorId(actorId); - job.setObjectKey(command.objectKey()); - job.setFileName(command.fileName()); - job.setMediaType(command.mediaType()); - job.setFileSize(command.fileSize()); - job.setPreviewCommandKey(command.commandKey()); + job.setPreviewKey(command.commandKey()); job.setPreviewRequestHash(requestHash); - job.setStatus("PREVIEW_PENDING"); - job.setScanStatus("PENDING"); - job.setParserStatus("PENDING"); - jobMapper.insert(job); - return project(job); + job.setImportType("QUESTION_DRAFT"); + job.setMaxAttempts(MAX_ATTEMPTS); + job.setPreviewPayload("{}"); + jobMapper.insertPreviewIgnore(job); + ContentImportJobDO stored = jobMapper.selectByPreviewKey(tenantId, command.commandKey()); + requireHash(stored.getPreviewRequestHash(), requestHash); + return project(stored, asset); } @Override @@ -72,87 +82,108 @@ public class QuestionImportJobServiceImpl implements QuestionImportJobService { public QuestionImportJobProjection requestExecute(ExecuteQuestionImportCommand command, Long actorId) { assertJavaRead(); Long tenantId = TenantContextHolder.getRequiredTenantId(); - QuestionImportJobDO replay = jobMapper.selectByExecuteCommand(tenantId, command.commandKey()); String requestHash = hash(command); + ContentImportJobDO replay = jobMapper.selectByExecuteKey(tenantId, command.commandKey()); if (replay != null) { requireHash(replay.getExecuteRequestHash(), requestHash); - return project(replay); + return project(replay, requireAsset(tenantId, replay.getAssetId())); } - QuestionImportJobDO job = requireJob(tenantId, command.jobId()); - if (!"PREVIEW_READY".equals(job.getStatus()) || !"CLEAN".equals(job.getScanStatus()) - || !"PARSED".equals(job.getParserStatus()) || job.getParsedPayload() == null) { + ContentImportJobDO job = requireJob(tenantId, command.jobId()); + if (jobMapper.requestExecute(tenantId, job.getId(), command.commandKey(), requestHash) != 1) { throw exception(QUESTION_IMPORT_STATE_CONFLICT); } - job.setActorId(actorId); - job.setExecuteCommandKey(command.commandKey()); - job.setExecuteRequestHash(requestHash); - job.setStatus("EXECUTE_PENDING"); - jobMapper.updateById(job); - return project(job); + job = requireJob(tenantId, job.getId()); + return project(job, requireAsset(tenantId, job.getAssetId())); } @Override public QuestionImportJobProjection get(Long jobId) { - return project(requireJob(TenantContextHolder.getRequiredTenantId(), jobId)); + Long tenantId = TenantContextHolder.getRequiredTenantId(); + ContentImportJobDO job = requireJob(tenantId, jobId); + return project(job, requireAsset(tenantId, job.getAssetId())); } @Override public void preview(Long jobId) { - QuestionImportJobDO claimed = jobMapper.claimPreview(jobId); - if (claimed == null) return; - TenantUtils.execute(claimed.getTenantId(), () -> processPreview(claimed)); - } - - private void processPreview(QuestionImportJobDO job) { - ImportObjectScanGateway gateway = scanGateway.orElse(null); - if (gateway == null) { - jobMapper.failPreview(job.getId(), "SCAN_UNAVAILABLE"); - throw exception(QUESTION_IMPORT_SCAN_NOT_CLEAN); - } - ImportObjectScanGateway.ScanResult scanResult = gateway.scan(job.getObjectKey()); - if (scanResult != ImportObjectScanGateway.ScanResult.CLEAN) { - jobMapper.failPreview(job.getId(), "SCAN_" + scanResult.name()); - throw exception(QUESTION_IMPORT_SCAN_NOT_CLEAN); - } - job.setScanStatus("CLEAN"); - QuestionImportParser availableParser = parser.orElse(null); - if (availableParser == null) { - job.setParserStatus("UNAVAILABLE"); - job.setParsedPayload(null); - job.setPreviewQuestionCount(null); - } else { - QuestionImportParser.ParsedImport parsed = availableParser.parse(job.getObjectKey()); - job.setParserStatus("PARSED"); - job.setParsedPayload(parsed.payload()); - job.setPreviewQuestionCount(parsed.questions().size()); - } - job.setStatus("PREVIEW_READY"); - jobMapper.completePreview(job); + processClaimed(jobId, "PREVIEW_PENDING"); } @Override - @Transactional(rollbackFor = Exception.class) public void execute(Long jobId) { - QuestionImportJobDO claimed = jobMapper.claimExecute(jobId); - if (claimed == null) return; assertJavaRead(); - TenantUtils.execute(claimed.getTenantId(), () -> processExecute(claimed)); + processClaimed(jobId, "EXECUTE_PENDING"); } - private void processExecute(QuestionImportJobDO job) { + private void processClaimed(Long jobId, String expectedStatus) { + jobMapper.failExhausted(); + String token = UUID.randomUUID().toString(); + ContentImportJobDO job = jobMapper.claimById(jobId, expectedStatus, "education-import", token, LEASE_SECONDS); + if (job != null) TenantUtils.execute(job.getTenantId(), () -> process(job, token)); + } + + private void process(ContentImportJobDO job, String token) { + if ("PREVIEW_PENDING".equals(job.getStatus())) processPreview(job, token); + else processExecute(job, token); + } + + private void processPreview(ContentImportJobDO job, String token) { + ContentImportAssetDO asset = requireAsset(job.getTenantId(), job.getAssetId()); + ImportObjectScanGateway gateway = scanGateway.orElse(null); + if (gateway == null) { + finishFailed(job, token, "SCAN_UNAVAILABLE"); + throw exception(QUESTION_IMPORT_SCAN_NOT_CLEAN); + } + ImportObjectScanGateway.ScanResult scanResult = gateway.scan(asset.getObjectKey()); + if (scanResult != ImportObjectScanGateway.ScanResult.CLEAN) { + finishFailed(job, token, "SCAN_" + scanResult.name()); + throw exception(QUESTION_IMPORT_SCAN_NOT_CLEAN); + } + QuestionImportParser availableParser = parser.orElse(null); + String parserStatus = "UNAVAILABLE"; + String parsedPayload = null; + Integer questionCount = null; + String previewPayload = "{\"mode\":\"METADATA_ONLY\",\"executable\":false}"; + if (availableParser != null) { + QuestionImportParser.ParsedImport parsed = availableParser.parse(asset.getObjectKey()); + parserStatus = "PARSED"; + parsedPayload = parsed.payload(); + questionCount = parsed.questions().size(); + previewPayload = "{\"mode\":\"PARSED\",\"executable\":true}"; + } + int updated = jobMapper.finishClaim(job.getTenantId(), job.getId(), "PREVIEW_PENDING", "PREVIEW_READY", + token, "CLEAN", parserStatus, previewPayload, parsedPayload, questionCount, null, null, null, null); + if (updated != 1) throw exception(CONTENT_IMPORT_LEASE_LOST); + } + + private void processExecute(ContentImportJobDO job, String token) { if (!"CLEAN".equals(job.getScanStatus())) throw exception(QUESTION_IMPORT_SCAN_NOT_CLEAN); QuestionImportParser availableParser = parser.orElseThrow(() -> exception(QUESTION_IMPORT_PARSER_UNAVAILABLE)); List questions = availableParser.restore(job.getParsedPayload()); for (QuestionDraftCommand question : questions) lifecycleService.createDraft(question); - jobMapper.completeExecute(job.getId(), questions.size()); + int updated = jobMapper.finishClaim(job.getTenantId(), job.getId(), "EXECUTE_PENDING", "COMPLETED", + token, null, null, null, job.getParsedPayload(), job.getPreviewQuestionCount(), + "{\"result\":\"COMPLETED\"}", questions.size(), null, null); + if (updated != 1) throw exception(CONTENT_IMPORT_LEASE_LOST); } - private QuestionImportJobDO requireJob(Long tenantId, Long id) { - QuestionImportJobDO job = jobMapper.selectTenantJob(tenantId, id); + private void finishFailed(ContentImportJobDO job, String token, String failureCode) { + jobMapper.finishClaim(job.getTenantId(), job.getId(), job.getStatus(), "FAILED", token, + null, null, null, job.getParsedPayload(), job.getPreviewQuestionCount(), null, + job.getImportedQuestionCount(), failureCode, failureCode); + } + + private ContentImportJobDO requireJob(Long tenantId, Long id) { + ContentImportJobDO job = jobMapper.selectTenantJob(tenantId, id); if (job == null) throw exception(QUESTION_IMPORT_NOT_FOUND); return job; } + private ContentImportAssetDO requireAsset(Long tenantId, Long id) { + ContentImportAssetDO asset = assetMapper.selectTenantAsset(tenantId, id); + if (asset == null) throw exception(QUESTION_IMPORT_NOT_FOUND); + return asset; + } + private void assertJavaRead() { if (properties.getCatalogMode() != CatalogProviderMode.JAVA_READ) { throw exception(QUESTION_IMPORT_PROVIDER_UNSUPPORTED, properties.getCatalogMode()); @@ -173,9 +204,9 @@ public class QuestionImportJobServiceImpl implements QuestionImportJobService { } } - private QuestionImportJobProjection project(QuestionImportJobDO job) { + private QuestionImportJobProjection project(ContentImportJobDO job, ContentImportAssetDO asset) { return new QuestionImportJobProjection(job.getId(), job.getStatus(), job.getScanStatus(), - job.getParserStatus(), job.getFileName(), job.getFileSize(), job.getPreviewQuestionCount(), + job.getParserStatus(), asset.getFileName(), asset.getFileSizeBytes(), job.getPreviewQuestionCount(), job.getImportedQuestionCount(), job.getFailureCode()); } } diff --git a/yudao-module-education/src/test/java/cn/iocoder/yudao/module/education/controller/admin/importjob/QuestionImportJobControllerTest.java b/yudao-module-education/src/test/java/cn/iocoder/yudao/module/education/controller/admin/importjob/QuestionImportJobControllerTest.java index 7590232b..b10279d8 100644 --- a/yudao-module-education/src/test/java/cn/iocoder/yudao/module/education/controller/admin/importjob/QuestionImportJobControllerTest.java +++ b/yudao-module-education/src/test/java/cn/iocoder/yudao/module/education/controller/admin/importjob/QuestionImportJobControllerTest.java @@ -33,11 +33,11 @@ class QuestionImportJobControllerTest { @Test void controllerMapsPreviewRequestAndActor() { QuestionImportPreviewReqVO request = new QuestionImportPreviewReqVO(); - request.setCommandKey("preview-1"); request.setObjectKey("object/10/q.csv"); request.setFileName("q.csv"); - request.setMediaType("text/csv"); request.setFileSize(12L); + request.setCommandKey("preview-1"); request.setAssetId(88L); assertEquals(101L, controller.preview(request).getData().getId()); assertEquals("preview-1", service.preview.commandKey()); + assertEquals(88L, service.preview.assetId()); assertEquals(7L, service.actorId); } diff --git a/yudao-module-education/src/test/java/cn/iocoder/yudao/module/education/service/importjob/QuestionImportJobServiceImplTest.java b/yudao-module-education/src/test/java/cn/iocoder/yudao/module/education/service/importjob/QuestionImportJobServiceImplTest.java index 9c712c1a..a59d4576 100644 --- a/yudao-module-education/src/test/java/cn/iocoder/yudao/module/education/service/importjob/QuestionImportJobServiceImplTest.java +++ b/yudao-module-education/src/test/java/cn/iocoder/yudao/module/education/service/importjob/QuestionImportJobServiceImplTest.java @@ -3,8 +3,10 @@ package cn.iocoder.yudao.module.education.service.importjob; import cn.iocoder.yudao.framework.common.exception.ServiceException; import cn.iocoder.yudao.framework.tenant.core.context.TenantContextHolder; import cn.iocoder.yudao.module.education.config.EducationProperties; -import cn.iocoder.yudao.module.education.dal.dataobject.importjob.QuestionImportJobDO; -import cn.iocoder.yudao.module.education.dal.mysql.importjob.QuestionImportJobMapper; +import cn.iocoder.yudao.module.education.dal.dataobject.importjob.ContentImportAssetDO; +import cn.iocoder.yudao.module.education.dal.dataobject.importjob.ContentImportJobDO; +import cn.iocoder.yudao.module.education.dal.mysql.importjob.ContentImportAssetMapper; +import cn.iocoder.yudao.module.education.dal.mysql.importjob.ContentImportJobMapper; import cn.iocoder.yudao.module.education.enums.CatalogProviderMode; import cn.iocoder.yudao.module.education.service.question.authoring.QuestionDraftCommand; import cn.iocoder.yudao.module.education.service.question.authoring.TenantQuestionLifecycleService; @@ -12,7 +14,6 @@ import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.ExtendWith; -import org.mockito.ArgumentCaptor; import org.mockito.Mock; import org.mockito.junit.jupiter.MockitoExtension; @@ -21,13 +22,14 @@ import java.util.Optional; import static cn.iocoder.yudao.module.education.enums.ErrorCodeConstants.QUESTION_IMPORT_SCAN_NOT_CLEAN; import static org.junit.jupiter.api.Assertions.*; -import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.*; import static org.mockito.Mockito.*; @ExtendWith(MockitoExtension.class) class QuestionImportJobServiceImplTest { - @Mock private QuestionImportJobMapper jobMapper; + @Mock private ContentImportAssetMapper assetMapper; + @Mock private ContentImportJobMapper jobMapper; @Mock private ImportObjectScanGateway scanGateway; @Mock private QuestionImportParser parser; @Mock private TenantQuestionLifecycleService lifecycleService; @@ -39,108 +41,103 @@ class QuestionImportJobServiceImplTest { void setUp() { properties = new EducationProperties(); properties.setCatalogMode(CatalogProviderMode.JAVA_READ); - service = new QuestionImportJobServiceImpl(properties, jobMapper, Optional.of(scanGateway), + service = new QuestionImportJobServiceImpl(properties, assetMapper, jobMapper, Optional.of(scanGateway), Optional.of(parser), lifecycleService); TenantContextHolder.setTenantId(10L); } - @AfterEach - void tearDown() { - TenantContextHolder.clear(); - } + @AfterEach void tearDown() { TenantContextHolder.clear(); } @Test - void duplicatePreviewCommandReturnsExistingJobWithoutRescanning() { - QuestionImportJobDO existing = job(101L, "PREVIEW_READY"); + void duplicatePreviewUsesTenantBoundAdmittedAsset() { + ContentImportAssetDO asset = asset(88L); + ContentImportJobDO existing = job(101L, "PREVIEW_READY"); existing.setPreviewRequestHash(null); - when(jobMapper.selectByPreviewCommand(10L, "preview-1")).thenReturn(existing); + when(assetMapper.selectTenantAsset(10L, 88L)).thenReturn(asset); + when(jobMapper.selectByPreviewKey(10L, "preview-1")).thenReturn(existing); - QuestionImportJobProjection result = service.requestPreview(new PreviewQuestionImportCommand( - "preview-1", "object/10/questions.csv", "questions.csv", "text/csv", 128L), 7L); + QuestionImportJobProjection result = service.requestPreview(new PreviewQuestionImportCommand("preview-1", 88L), 7L); assertEquals(101L, result.id()); - verify(jobMapper, never()).insert((QuestionImportJobDO) any(QuestionImportJobDO.class)); - verifyNoInteractions(scanGateway, parser, lifecycleService); + assertEquals("questions.csv", result.fileName()); + verify(jobMapper, never()).insertPreviewIgnore(any()); } @Test - void cleanScanWithoutParserProducesMetadataOnlyPreviewReady() { - service = new QuestionImportJobServiceImpl(properties, jobMapper, Optional.of(scanGateway), + void crossTenantOrMissingAssetFailsBeforeJobInsert() { + when(assetMapper.selectTenantAsset(10L, 88L)).thenReturn(null); + assertThrows(ServiceException.class, + () -> service.requestPreview(new PreviewQuestionImportCommand("preview-1", 88L), 7L)); + verifyNoInteractions(jobMapper); + } + + @Test + void cleanScanWithoutParserProducesMetadataOnlyPreview() { + service = new QuestionImportJobServiceImpl(properties, assetMapper, jobMapper, Optional.of(scanGateway), Optional.empty(), lifecycleService); - QuestionImportJobDO claimed = job(101L, "PREVIEW_PENDING"); - claimed.setTenantId(10L); - when(jobMapper.claimPreview(101L)).thenReturn(claimed); + ContentImportJobDO claimed = job(101L, "PREVIEW_PENDING"); + when(jobMapper.claimById(eq(101L), eq("PREVIEW_PENDING"), anyString(), anyString(), anyLong())) + .thenReturn(claimed); + when(assetMapper.selectTenantAsset(10L, 88L)).thenReturn(asset(88L)); when(scanGateway.scan("object/10/questions.csv")).thenReturn(ImportObjectScanGateway.ScanResult.CLEAN); + when(jobMapper.finishClaim(eq(10L), eq(101L), eq("PREVIEW_PENDING"), eq("PREVIEW_READY"), + anyString(), eq("CLEAN"), eq("UNAVAILABLE"), contains("METADATA_ONLY"), isNull(), isNull(), + isNull(), isNull(), isNull(), isNull())).thenReturn(1); service.preview(101L); - ArgumentCaptor captor = ArgumentCaptor.forClass(QuestionImportJobDO.class); - verify(jobMapper).completePreview(captor.capture()); - assertEquals("PREVIEW_READY", captor.getValue().getStatus()); - assertEquals("CLEAN", captor.getValue().getScanStatus()); - assertEquals("UNAVAILABLE", captor.getValue().getParserStatus()); - assertNull(captor.getValue().getParsedPayload()); - verifyNoInteractions(lifecycleService); + verifyNoInteractions(parser, lifecycleService); } @Test - void unavailableScannerFailsClosedAndNeverParses() { - service = new QuestionImportJobServiceImpl(properties, jobMapper, Optional.empty(), + void unavailableScannerFailsClosedAndFencesTerminalWrite() { + service = new QuestionImportJobServiceImpl(properties, assetMapper, jobMapper, Optional.empty(), Optional.of(parser), lifecycleService); - when(jobMapper.claimPreview(101L)).thenReturn(job(101L, "PREVIEW_PENDING")); + when(jobMapper.claimById(eq(101L), eq("PREVIEW_PENDING"), anyString(), anyString(), anyLong())) + .thenReturn(job(101L, "PREVIEW_PENDING")); + when(assetMapper.selectTenantAsset(10L, 88L)).thenReturn(asset(88L)); ServiceException ex = assertThrows(ServiceException.class, () -> service.preview(101L)); assertEquals(QUESTION_IMPORT_SCAN_NOT_CLEAN.getCode(), ex.getCode()); - verify(jobMapper).failPreview(eq(101L), eq("SCAN_UNAVAILABLE")); + verify(jobMapper).finishClaim(eq(10L), eq(101L), eq("PREVIEW_PENDING"), eq("FAILED"), + anyString(), isNull(), isNull(), isNull(), isNull(), isNull(), isNull(), isNull(), + eq("SCAN_UNAVAILABLE"), eq("SCAN_UNAVAILABLE")); verifyNoInteractions(parser, lifecycleService); } @Test - void executePropagatesClaimedTenantAndCreatesEveryQuestionAsDraft() { + void executePropagatesClaimedTenantAndCreatesDrafts() { TenantContextHolder.clear(); - QuestionImportJobDO claimed = job(101L, "EXECUTE_PENDING"); - claimed.setTenantId(42L); - claimed.setScanStatus("CLEAN"); - claimed.setParserStatus("PARSED"); + ContentImportJobDO claimed = job(101L, "EXECUTE_PENDING"); + claimed.setTenantId(42L); claimed.setScanStatus("CLEAN"); claimed.setParserStatus("PARSED"); claimed.setParsedPayload("stored-preview"); - when(jobMapper.claimExecute(101L)).thenReturn(claimed); + when(jobMapper.claimById(eq(101L), eq("EXECUTE_PENDING"), anyString(), anyString(), anyLong())) + .thenReturn(claimed); when(parser.restore("stored-preview")).thenReturn(List.of(question("One"), question("Two"))); when(lifecycleService.createDraft(any())).thenAnswer(invocation -> { - assertEquals(42L, TenantContextHolder.getRequiredTenantId()); - return 100L; + assertEquals(42L, TenantContextHolder.getRequiredTenantId()); return 100L; }); + when(jobMapper.finishClaim(eq(42L), eq(101L), eq("EXECUTE_PENDING"), eq("COMPLETED"), + anyString(), isNull(), isNull(), isNull(), eq("stored-preview"), isNull(), anyString(), eq(2), + isNull(), isNull())).thenReturn(1); service.execute(101L); verify(lifecycleService, times(2)).createDraft(any()); - verify(jobMapper).completeExecute(101L, 2); assertNull(TenantContextHolder.getTenantId()); } - @Test - void executeFailsBeforeImportOutsideJavaRead() { - properties.setCatalogMode(CatalogProviderMode.SCALAR_READ); - when(jobMapper.claimExecute(101L)).thenReturn(job(101L, "EXECUTE_PENDING")); - - assertThrows(ServiceException.class, () -> service.execute(101L)); - - verifyNoInteractions(parser, lifecycleService); + private ContentImportAssetDO asset(Long id) { + ContentImportAssetDO asset = new ContentImportAssetDO(); asset.setId(id); asset.setTenantId(10L); + asset.setObjectKey("object/10/questions.csv"); asset.setFileName("questions.csv"); + asset.setMimeType("text/csv"); asset.setFileSizeBytes(128L); return asset; } - private QuestionImportJobDO job(Long id, String status) { - QuestionImportJobDO job = new QuestionImportJobDO(); - job.setId(id); - job.setTenantId(10L); - job.setActorId(7L); - job.setObjectKey("object/10/questions.csv"); - job.setFileName("questions.csv"); - job.setMediaType("text/csv"); - job.setFileSize(128L); - job.setPreviewCommandKey("preview-1"); - job.setPreviewRequestHash("hash"); - job.setStatus(status); - return job; + private ContentImportJobDO job(Long id, String status) { + ContentImportJobDO job = new ContentImportJobDO(); job.setId(id); job.setTenantId(10L); job.setAssetId(88L); + job.setActorId(7L); job.setPreviewKey("preview-1"); job.setPreviewRequestHash("hash"); + job.setStatus(status); return job; } private QuestionDraftCommand question(String stem) {