Merge remote-tracking branch 'origin/dev_business' into dev_business

liu

昨天 43aab58c9ecb1203770ba382efbdbae299ca87ec

 yudao-module-mdm/src/main/java/cn/iocoder/yudao/module/mdm/service/agent/MdmAgentImportCommandService.java

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,372 @@
package cn.iocoder.yudao.module.mdm.service.agent;

import cn.hutool.core.util.StrUtil;
import cn.iocoder.yudao.framework.common.exception.ServiceException;
import cn.iocoder.yudao.module.mdm.controller.admin.agent.vo.MdmAgentImportCommitReqVO;
import cn.iocoder.yudao.module.mdm.controller.admin.agent.vo.MdmAgentImportPreflightReqVO;
import cn.iocoder.yudao.module.mdm.controller.admin.agent.vo.MdmAgentImportPreflightRespVO;
import cn.iocoder.yudao.module.mdm.controller.admin.agent.vo.MdmAgentImportReceiptRespVO;
import cn.iocoder.yudao.module.mdm.controller.admin.item.vo.MdmItemImportExcelVO;
import cn.iocoder.yudao.module.mdm.controller.admin.item.vo.MdmItemSaveReqVO;
import cn.iocoder.yudao.module.mdm.dal.dataobject.agent.MdmAgentImportCommandDO;
import cn.iocoder.yudao.module.mdm.dal.dataobject.agent.MdmAgentImportCommandRowDO;
import cn.iocoder.yudao.module.mdm.dal.dataobject.item.MdmItemDO;
import cn.iocoder.yudao.module.mdm.dal.mysql.agent.MdmAgentImportCommandMapper;
import cn.iocoder.yudao.module.mdm.dal.mysql.agent.MdmAgentImportCommandRowMapper;
import cn.iocoder.yudao.module.mdm.service.item.MdmItemImportContext;
import cn.iocoder.yudao.module.mdm.service.item.MdmItemRowResolution;
import cn.iocoder.yudao.module.mdm.service.item.MdmItemService;
import cn.iocoder.yudao.module.system.framework.agent.AgentBridgeException;
import cn.iocoder.yudao.module.system.framework.agent.AgentPrincipal;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import jakarta.annotation.Resource;
import org.springframework.dao.DuplicateKeyException;
import org.springframework.http.HttpStatus;
import org.springframework.stereotype.Service;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.support.TransactionTemplate;
import org.springframework.validation.annotation.Validated;

import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.time.Duration;
import java.time.LocalDateTime;
import java.util.*;
import java.util.concurrent.atomic.AtomicReference;

/**
 * ç‰©æ–™å¯¼å…¥å¹‚等提交：同一 commandId/idempotencyKey é‡å¤æäº¤è¿”回原回执，绝不重复建物料。
 */
@Service
@Validated
public class MdmAgentImportCommandService {

    public static final String STATUS_RUNNING = "RUNNING";
    public static final String STATUS_SUCCEEDED = "SUCCEEDED";
    public static final String STATUS_PARTIAL_FAILED = "PARTIAL_FAILED";
    public static final String STATUS_FAILED = "FAILED";
    public static final String STATUS_UNKNOWN = "UNKNOWN";

    private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
    private static final Duration LEASE_DURATION = Duration.ofMinutes(5);

    @Resource
    private MdmAgentImportPreflightService preflightService;
    @Resource
    private MdmItemService itemService;
    @Resource
    private MdmAgentImportPolicy policy;
    @Resource
    private MdmAgentImportCommandMapper commandMapper;
    @Resource
    private MdmAgentImportCommandRowMapper commandRowMapper;
    @Resource
    private PlatformTransactionManager transactionManager;

    public MdmAgentImportReceiptRespVO commit(MdmAgentImportCommitReqVO reqVO, AgentPrincipal principal) {
        if (!policy.isWriteEnabled()) {
            throw new AgentBridgeException(HttpStatus.SERVICE_UNAVAILABLE, "WRITE_DISABLED",
                    "智能体写入开关未打开，提交已拒绝");
        }
        if (Boolean.TRUE.equals(reqVO.getSyncMes()) && !policy.isAllowMesSync()) {
            throw new AgentBridgeException(HttpStatus.UNPROCESSABLE_ENTITY, "MES_SYNC_DISABLED",
                    "服务未开放同步 MES，请关闭 syncMes åŽé‡è¯•");
        }
        String requestDigest = buildRequestDigest(reqVO);
        Long ownerUserId = principal.userId();

        // 1. å¹‚等命中：命令 ID ä¼˜å…ˆï¼Œå…¶æ¬¡å¹‚等键
        MdmAgentImportCommandDO existing = commandMapper.selectByCommandId(ownerUserId, reqVO.getCommandId());
        if (existing != null) {
            if (!existing.getRequestDigest().equals(requestDigest)) {
                throw new AgentBridgeException(HttpStatus.CONFLICT, "COMMAND_ID_CONFLICT",
                        "同一命令 ID æºå¸¦äº†ä¸åŒçš„请求内容，请检查任务状态后重试");
            }
            return toReceipt(existing);
        }
        existing = commandMapper.selectByIdempotencyKey(ownerUserId, reqVO.getIdempotencyKey());
        if (existing != null) {
            if (!existing.getRequestDigest().equals(requestDigest)) {
                throw new AgentBridgeException(HttpStatus.CONFLICT, "IDEMPOTENCY_CONFLICT",
                        "同一幂等键携带了不同的请求内容，请检查任务状态后重试");
            }
            return toReceipt(existing);
        }

        // 2. é‡æ–°é¢„检并比对计划摘要，防止确认后数据已变化
        MdmAgentImportPreflightRespVO preflight = preflightService.preflight(toPreflightRequest(reqVO), principal);
        if (!reqVO.getPreflightHash().equalsIgnoreCase(preflight.getPreflightHash())) {
            throw new AgentBridgeException(HttpStatus.CONFLICT, "PLAN_STALE",
                    "预检计划已变化（编码状态或主数据被改动），请重新预检并确认", false,
                    Map.of("preflight", preflight));
        }
        if (!Boolean.TRUE.equals(preflight.getCanCommit())) {
            throw new AgentBridgeException(HttpStatus.UNPROCESSABLE_ENTITY, "PREFLIGHT_BLOCKED",
                    "预检存在阻断项，不能提交", false, Map.of("preflight", preflight));
        }

        // 3. ç™»è®°å‘½ä»¤ï¼ˆå”¯ä¸€é”®å…œä½å¹¶å‘重复提交）
        MdmAgentImportCommandDO command = MdmAgentImportCommandDO.builder()
                .commandId(reqVO.getCommandId())
                .idempotencyKey(reqVO.getIdempotencyKey())
                .requestDigest(requestDigest)
                .ownerUserId(ownerUserId)
                .sessionId(reqVO.getSessionId())
                .attachmentId(reqVO.getAttachmentId())
                .sourceSha256(reqVO.getSourceSha256().toLowerCase(Locale.ROOT))
                .preflightHash(reqVO.getPreflightHash().toLowerCase(Locale.ROOT))
                .proposalHash(reqVO.getProposalHash())
                .syncMes(Boolean.TRUE.equals(reqVO.getSyncMes()))
                .status(STATUS_RUNNING)
                .totalCount(reqVO.getRows().size())
                .successCount(0).failedCount(0).skippedCount(0)
                .leaseExpiresAt(LocalDateTime.now().plus(LEASE_DURATION))
                .build();
        try {
            commandMapper.insert(command);
        } catch (DuplicateKeyException ex) {
            MdmAgentImportCommandDO raced = commandMapper.selectByCommandId(ownerUserId, reqVO.getCommandId());
            if (raced == null) {
                raced = commandMapper.selectByIdempotencyKey(ownerUserId, reqVO.getIdempotencyKey());
            }
            if (raced == null) {
                throw ex;
            }
            if (!raced.getRequestDigest().equals(requestDigest)) {
                throw new AgentBridgeException(HttpStatus.CONFLICT, "IDEMPOTENCY_CONFLICT",
                        "同一命令/幂等键携带了不同的请求内容，请检查任务状态后重试");
            }
            return toReceipt(raced);
        }

        // 4. é€è¡Œç‹¬ç«‹äº‹åŠ¡æ‰§è¡Œï¼›ä»»ä¸€è¡Œå¤±è´¥ä¸å½±å“å…¶ä»–è¡Œ
        int success = 0;
        int failed = 0;
        int skipped = 0;
        TransactionTemplate rowTxTemplate = new TransactionTemplate(transactionManager);
        List<MdmItemImportExcelVO> excelRows = reqVO.getRows().stream()
                .map(MdmAgentImportCommandService::toExcelVO).toList();
        MdmItemImportContext context = itemService.buildImportContext(excelRows);
        try {
            for (int index = 0; index < reqVO.getRows().size(); index++) {
                MdmAgentImportPreflightReqVO.Row row = reqVO.getRows().get(index);
                MdmAgentImportPreflightRespVO.Row planned = preflight.getRows().get(index);
                if ("SKIP".equals(planned.getAction())) {
                    commandRowMapper.insert(baseRow(command.getCommandId(), row)
                            .action("SKIP").status("SKIPPED")
                            .targetItemId(planned.getTargetItemId())
                            .retryable(false).build());
                    skipped++;
                    continue;
                }
                if (!"CREATE".equals(planned.getAction())) {
                    commandRowMapper.insert(baseRow(command.getCommandId(), row)
                            .action("CREATE").status("FAILED")
                            .errorCode("PREFLIGHT_BLOCKED").errorMessage("该行预检未通过").retryable(false).build());
                    failed++;
                    continue;
                }
                MdmItemRowResolution resolution = itemService.resolveImportRow(excelRows.get(index), context);
                if (!resolution.isSuccess()) {
                    commandRowMapper.insert(baseRow(command.getCommandId(), row)
                            .action("CREATE").status("FAILED")
                            .errorCode("ROW_INVALID").errorMessage(trim(resolution.failureMessage(), 512))
                            .retryable(false).build());
                    failed++;
                    continue;
                }
                AtomicReference<Throwable> rowError = new AtomicReference<>();
                boolean[] concurrentSkip = {false};
                try {
                    rowTxTemplate.executeWithoutResult(status -> {
                        try {
                            // æäº¤å‰é‡æŸ¥ï¼Œç¼©å°å¹¶å‘重复建物料的窗口
                            List<MdmItemDO> current = itemService.getItemListByCode(row.getCode().trim());
                            if (!current.isEmpty()) {
                                concurrentSkip[0] = true;
                                commandRowMapper.insert(baseRow(command.getCommandId(), row)
                                        .action("SKIP").status("SKIPPED")
                                        .targetItemId(current.get(0).getId()).retryable(false).build());
                                return;
                            }
                            MdmItemSaveReqVO saveReqVO = resolution.saveReqVO();
                            saveReqVO.setSyncMes(Boolean.TRUE.equals(reqVO.getSyncMes()));
                            Long itemId = itemService.createItem(saveReqVO);
                            commandRowMapper.insert(baseRow(command.getCommandId(), row)
                                    .action("CREATE").status("SUCCESS").targetItemId(itemId)
                                    .retryable(false).build());
                        } catch (Throwable t) {
                            rowError.set(t);
                            throw t;
                        }
                    });
                    if (concurrentSkip[0]) {
                        skipped++;
                    } else {
                        success++;
                    }
                } catch (Exception ex) {
                    Throwable cause = rowError.get() != null ? rowError.get() : ex;
                    String message = cause instanceof ServiceException
                            ? ((ServiceException) cause).getMessage() : cause.getMessage();
                    String errorCode = cause instanceof ServiceException ? "ITEM_CREATE_REJECTED" : "ITEM_CREATE_FAILED";
                    commandRowMapper.insert(baseRow(command.getCommandId(), row)
                            .action("CREATE").status("FAILED")
                            .errorCode(errorCode).errorMessage(trim(message, 512))
                            .retryable(!(cause instanceof ServiceException)).build());
                    failed++;
                }
            }
        } catch (Exception ex) {
            // åŸºç¡€è®¾æ–½å¼‚常中断：保留已完成行回执，命令标记 UNKNOWN ä¾›å¯¹è´¦
            command.setStatus(STATUS_UNKNOWN);
            command.setErrorCode("COMMIT_INTERRUPTED");
            command.setErrorMessage(trim(ex.getMessage(), 512));
            command.setSuccessCount(success);
            command.setFailedCount(failed);
            command.setSkippedCount(skipped);
            commandMapper.updateById(command);
            return toReceipt(command);
        }

        command.setSuccessCount(success);
        command.setFailedCount(failed);
        command.setSkippedCount(skipped);
        command.setStatus(failed == 0 ? STATUS_SUCCEEDED
                : (success + skipped == 0 ? STATUS_FAILED : STATUS_PARTIAL_FAILED));
        command.setLeaseExpiresAt(LocalDateTime.now());
        commandMapper.updateById(command);
        return toReceipt(command);
    }

    public MdmAgentImportReceiptRespVO getCommand(String commandId, AgentPrincipal principal) {
        MdmAgentImportCommandDO command = commandMapper.selectByCommandId(
                principal.userId(), commandId);
        if (command == null) {
            throw new AgentBridgeException(HttpStatus.NOT_FOUND, "COMMAND_NOT_FOUND", "导入命令不存在");
        }
        return toReceipt(command);
    }

    private MdmAgentImportReceiptRespVO toReceipt(MdmAgentImportCommandDO command) {
        String status = command.getStatus();
        if (STATUS_RUNNING.equals(status) && command.getLeaseExpiresAt() != null
                && command.getLeaseExpiresAt().isBefore(LocalDateTime.now())) {
            // ç§Ÿçº¦è¿‡æœŸè¯´æ˜Žæ‰§è¡Œè¿›ç¨‹å¯èƒ½å·²ä¸­æ–­ï¼›ä¸æ”¹åº“，仅如实告知调用方结果未知
            status = STATUS_UNKNOWN;
        }
        List<MdmAgentImportCommandRowDO> rowDOs = commandRowMapper.selectListByCommandId(command.getCommandId());
        List<MdmAgentImportReceiptRespVO.Row> rows = rowDOs.stream()
                .map(row -> MdmAgentImportReceiptRespVO.Row.builder()
                        .candidateId(row.getCandidateId())
                        .rowNo(row.getSourceRow())
                        .code(row.getCode())
                        .name(row.getName())
                        .action(row.getAction())
                        .status(row.getStatus())
                        .targetItemId(row.getTargetItemId())
                        .targetVersion(row.getTargetVersion())
                        .errorCode(row.getErrorCode())
                        .errorMessage(row.getErrorMessage())
                        .retryable(row.getRetryable())
                        .build())
                .toList();
        int success = (int) rowDOs.stream().filter(row -> "SUCCESS".equals(row.getStatus())).count();
        int failed = (int) rowDOs.stream().filter(row -> "FAILED".equals(row.getStatus())).count();
        int skipped = (int) rowDOs.stream().filter(row -> "SKIPPED".equals(row.getStatus())).count();
        int unknown = (int) rowDOs.stream().filter(row -> "UNKNOWN".equals(row.getStatus())).count();
        return MdmAgentImportReceiptRespVO.builder()
                .commandId(command.getCommandId())
                .idempotencyKey(command.getIdempotencyKey())
                .status(status)
                .attachmentId(command.getAttachmentId())
                .sessionId(command.getSessionId())
                .sourceSha256(command.getSourceSha256())
                .preflightHash(command.getPreflightHash())
                .proposalHash(command.getProposalHash())
                .summary(MdmAgentImportReceiptRespVO.Summary.builder()
                        .total(command.getTotalCount())
                        .success(success).failed(failed).skipped(skipped).unknown(unknown)
                        .build())
                .errorCode(command.getErrorCode())
                .errorMessage(command.getErrorMessage())
                .rows(rows)
                .createTime(command.getCreateTime())
                .completedTime(STATUS_RUNNING.equals(command.getStatus()) ? null : command.getUpdateTime())
                .build();
    }

    private static MdmAgentImportCommandRowDO.MdmAgentImportCommandRowDOBuilder baseRow(
            String commandId, MdmAgentImportPreflightReqVO.Row row) {
        return MdmAgentImportCommandRowDO.builder()
                .commandId(commandId)
                .candidateId(row.getCandidateId())
                .sourceRow(row.getRowNo())
                .code(row.getCode())
                .name(row.getName());
    }

    private static MdmItemImportExcelVO toExcelVO(MdmAgentImportPreflightReqVO.Row row) {
        return MdmItemImportExcelVO.builder()
                .code(row.getCode())
                .name(row.getName())
                .specification(row.getSpecification())
                .categoryCode(row.getCategoryCode())
                .unitMeasureCode(row.getUnitMeasureCode())
                .build();
    }

    private static MdmAgentImportPreflightReqVO toPreflightRequest(MdmAgentImportCommitReqVO reqVO) {
        MdmAgentImportPreflightReqVO preflightReqVO = new MdmAgentImportPreflightReqVO();
        preflightReqVO.setAttachmentId(reqVO.getAttachmentId());
        preflightReqVO.setSessionId(reqVO.getSessionId());
        preflightReqVO.setSourceSha256(reqVO.getSourceSha256());
        preflightReqVO.setProposalRevision(reqVO.getProposalRevision());
        preflightReqVO.setSyncMes(reqVO.getSyncMes());
        preflightReqVO.setRows(reqVO.getRows());
        return preflightReqVO;
    }

    /**
     * è¯·æ±‚摘要只包含业务内容，不含 commandId/idempotencyKey，便于识别“同键不同内容”。
     */
    private static String buildRequestDigest(MdmAgentImportCommitReqVO reqVO) {
        Map<String, Object> canonical = new LinkedHashMap<>();
        canonical.put("attachment_id", reqVO.getAttachmentId());
        canonical.put("session_id", reqVO.getSessionId());
        canonical.put("source_sha256", StrUtil.nullToEmpty(reqVO.getSourceSha256()).toLowerCase(Locale.ROOT));
        canonical.put("preflight_hash", StrUtil.nullToEmpty(reqVO.getPreflightHash()).toLowerCase(Locale.ROOT));
        canonical.put("proposal_hash", StrUtil.nullToEmpty(reqVO.getProposalHash()));
        canonical.put("proposal_revision", StrUtil.nullToEmpty(reqVO.getProposalRevision()));
        canonical.put("sync_mes", Boolean.TRUE.equals(reqVO.getSyncMes()));
        List<Map<String, Object>> rows = new ArrayList<>();
        for (MdmAgentImportPreflightReqVO.Row row : reqVO.getRows()) {
            Map<String, Object> item = new LinkedHashMap<>();
            item.put("candidate_id", row.getCandidateId());
            item.put("row_no", row.getRowNo());
            item.put("code", StrUtil.nullToEmpty(row.getCode()));
            item.put("name", StrUtil.nullToEmpty(row.getName()));
            item.put("specification", StrUtil.nullToEmpty(row.getSpecification()));
            item.put("category_code", StrUtil.nullToEmpty(row.getCategoryCode()));
            item.put("unit_measure_code", StrUtil.nullToEmpty(row.getUnitMeasureCode()));
            rows.add(item);
        }
        canonical.put("rows", rows);
        try {
            byte[] payload = OBJECT_MAPPER.writeValueAsBytes(canonical);
            return HexFormat.of().formatHex(MessageDigest.getInstance("SHA-256").digest(payload));
        } catch (NoSuchAlgorithmException ex) {
            throw new IllegalStateException("SHA-256 ä¸å¯ç”¨", ex);
        } catch (JsonProcessingException ex) {
            throw new IllegalStateException("请求摘要生成失败", ex);
        }
    }

    private static String trim(String value, int maxLength) {
        if (value == null) {
            return null;
        }
        return value.length() <= maxLength ? value : value.substring(0, maxLength);
    }
}