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

liu

20 小时以前 43aab58c9ecb1203770ba382efbdbae299ca87ec

 yudao-module-system/src/main/java/cn/iocoder/yudao/module/system/service/agent/imports/SystemAgentGenericImportService.java

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

import cn.hutool.core.util.StrUtil;
import cn.iocoder.yudao.framework.common.exception.ServiceException;
import cn.iocoder.yudao.framework.security.core.service.SecurityFrameworkService;
import cn.iocoder.yudao.module.system.api.agent.imports.*;
import cn.iocoder.yudao.module.system.dal.dataobject.agent.AgentImportCommandDO;
import cn.iocoder.yudao.module.system.dal.dataobject.agent.AgentImportCommandRowDO;
import cn.iocoder.yudao.module.system.dal.mysql.agent.AgentImportCommandMapper;
import cn.iocoder.yudao.module.system.dal.mysql.agent.AgentImportCommandRowMapper;
import cn.iocoder.yudao.module.system.framework.agent.AgentBridgeException;
import cn.iocoder.yudao.module.system.framework.agent.AgentPrincipal;
import cn.iocoder.yudao.module.system.service.storage.SystemAgentImportBridgeService;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import jakarta.annotation.Resource;
import org.springframework.beans.factory.ObjectProvider;
import org.springframework.beans.factory.annotation.Value;
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.nio.charset.StandardCharsets;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.time.Duration;
import java.time.LocalDateTime;
import java.util.*;

/**
 * é€šç”¨ä¸šåŠ¡å¯¼å…¥ç¼–排：适配器只负责单行校验/写入，本类统一处理权限、幂等、逐行事务与回执。
 */
@Service
@Validated
public class SystemAgentGenericImportService {

    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 ObjectProvider<AgentImportModuleAdapter> adapterProvider;
    @Resource
    private SecurityFrameworkService securityFrameworkService;
    @Resource
    private SystemAgentImportBridgeService attachmentService;
    @Resource
    private AgentImportCommandMapper commandMapper;
    @Resource
    private AgentImportCommandRowMapper commandRowMapper;
    @Resource
    private PlatformTransactionManager transactionManager;

    @Value("${yudao.agent.bridge.write-enabled:false}")
    private boolean writeEnabled;

    @Value("${yudao.agent.bridge.max-rows:5000}")
    private int maxRows;

    public List<Map<String, String>> modules() {
        return adapters().values().stream()
                .map(adapter -> Map.of("moduleId", adapter.moduleId(), "permission", adapter.permission()))
                .toList();
    }

    public AgentImportPreflightRespVO preflight(String moduleId, AgentImportPreflightReqVO request,
                                                AgentPrincipal principal) {
        AgentImportModuleAdapter adapter = requireAdapter(moduleId);
        requirePermission(adapter);
        validateAttachment(request.getAttachmentId(), request.getSourceSha256());
        validateRows(request);
        AgentImportPreflightRespVO result = adapter.preflight(principal, request);
        validatePlan(result, request);
        return result;
    }

    public AgentImportReceiptRespVO commit(String moduleId, AgentImportCommitReqVO request,
                                           AgentPrincipal principal) {
        if (!writeEnabled) {
            throw new AgentBridgeException(HttpStatus.SERVICE_UNAVAILABLE, "WRITE_DISABLED",
                    "智能体写入开关未打开，提交已拒绝");
        }
        AgentImportModuleAdapter adapter = requireAdapter(moduleId);
        requirePermission(adapter);
        validateAttachment(request.getAttachmentId(), request.getSourceSha256());
        validateRows(request);

        Long ownerUserId = principal.userId();
        String digest = requestDigest(moduleId, request);

        // 1. å¹‚等命中：命令 ID ä¼˜å…ˆï¼Œå…¶æ¬¡å¹‚等键
        AgentImportCommandDO existing = commandMapper.selectByCommandId(
                ownerUserId, moduleId, request.getCommandId());
        if (existing != null) {
            requireSameDigest(existing, digest);
            return toReceipt(existing);
        }
        existing = commandMapper.selectByIdempotencyKey(
                ownerUserId, moduleId, request.getIdempotencyKey());
        if (existing != null) {
            requireSameDigest(existing, digest);
            return toReceipt(existing);
        }

        // 2. é‡æ–°é¢„检并比对计划摘要，防止确认后主数据变化
        AgentImportPreflightRespVO plan = adapter.preflight(principal, request);
        validatePlan(plan, request);
        if (!StrUtil.equalsIgnoreCase(request.getPreflightHash(), plan.getPreflightHash())) {
            throw new AgentBridgeException(HttpStatus.CONFLICT, "PLAN_STALE",
                    "预检计划已变化，请重新预检并确认", false, Map.of("preflight", plan));
        }
        if (!Boolean.TRUE.equals(plan.getCanCommit())) {
            throw new AgentBridgeException(HttpStatus.UNPROCESSABLE_ENTITY, "PREFLIGHT_BLOCKED",
                    "预检存在阻断项，不能提交", false, Map.of("preflight", plan));
        }

        // 3. ç™»è®°å‘½ä»¤ï¼ˆå”¯ä¸€é”®å…œä½å¹¶å‘重复提交）
        AgentImportCommandDO command = AgentImportCommandDO.builder()
                .commandId(request.getCommandId())
                .idempotencyKey(request.getIdempotencyKey())
                .requestDigest(digest)
                .moduleId(moduleId)
                .ownerUserId(ownerUserId)
                .sessionId(request.getSessionId())
                .attachmentId(request.getAttachmentId())
                .sourceSha256(request.getSourceSha256().toLowerCase(Locale.ROOT))
                .preflightHash(request.getPreflightHash().toLowerCase(Locale.ROOT))
                .proposalHash(request.getProposalHash())
                .status(STATUS_RUNNING)
                .totalCount(request.getRows().size())
                .successCount(0).failedCount(0).skippedCount(0)
                .leaseExpiresAt(LocalDateTime.now().plus(LEASE_DURATION))
                .build();
        try {
            commandMapper.insert(command);
        } catch (DuplicateKeyException ex) {
            AgentImportCommandDO raced = commandMapper.selectByCommandId(
                    ownerUserId, moduleId, request.getCommandId());
            if (raced == null) {
                raced = commandMapper.selectByIdempotencyKey(
                        ownerUserId, moduleId, request.getIdempotencyKey());
            }
            if (raced == null) {
                throw ex;
            }
            requireSameDigest(raced, digest);
            return toReceipt(raced);
        }

        // 4. é€è¡Œç‹¬ç«‹äº‹åŠ¡æ‰§è¡Œï¼›å•è¡Œå¤±è´¥ä¸å½±å“å…¶ä»–è¡Œ
        int success = 0;
        int failed = 0;
        int skipped = 0;
        TransactionTemplate rowTx = new TransactionTemplate(transactionManager);
        try {
            for (int index = 0; index < request.getRows().size(); index++) {
                AgentImportPreflightReqVO.Row row = request.getRows().get(index);
                AgentImportPreflightRespVO.Row planned = plan.getRows().get(index);
                if ("SKIP".equals(planned.getAction())) {
                    commandRowMapper.insert(baseRow(moduleId, request.getCommandId(), row)
                            .action("SKIP").status("SKIPPED")
                            .targetItemId(planned.getTargetId()).retryable(false).build());
                    skipped++;
                    continue;
                }
                if (!"CREATE".equals(planned.getAction())) {
                    commandRowMapper.insert(baseRow(moduleId, request.getCommandId(), row)
                            .action("CREATE").status("FAILED")
                            .errorCode("PREFLIGHT_BLOCKED").errorMessage("该行预检未通过").retryable(false).build());
                    failed++;
                    continue;
                }
                try {
                    AgentImportRowReceipt receipt = rowTx.execute(status ->
                            adapter.commitRow(principal, request, planned, row));
                    if (receipt == null) {
                        throw new IllegalStateException("适配器未返回回执");
                    }
                    commandRowMapper.insert(baseRow(moduleId, request.getCommandId(), row)
                            .action(StrUtil.blankToDefault(receipt.action(), "CREATE"))
                            .status("SUCCESS")
                            .targetItemId(receipt.targetId())
                            .targetVersion(receipt.targetVersion())
                            .code(StrUtil.blankToDefault(receipt.code(), row.getValues().get("code")))
                            .name(StrUtil.blankToDefault(receipt.name(), row.getValues().get("name")))
                            .retryable(false).build());
                    success++;
                } catch (Exception ex) {
                    recordFailure(moduleId, request.getCommandId(), row, ex);
                    failed++;
                }
            }
        } catch (Exception ex) {
            command.setStatus(STATUS_UNKNOWN);
            command.setErrorCode("COMMIT_INTERRUPTED");
            command.setErrorMessage(trim(ex.getMessage(), 512));
            command.setSuccessCount(success);
            command.setFailedCount(failed);
            command.setSkippedCount(skipped);
            command.setUpdateTime(LocalDateTime.now());
            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());
        command.setUpdateTime(LocalDateTime.now());
        commandMapper.updateById(command);
        return toReceipt(command);
    }

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

    private void recordFailure(String moduleId, String commandId, AgentImportPreflightReqVO.Row row,
                               Exception ex) {
        String errorCode = "ROW_WRITE_FAILED";
        boolean retryable = true;
        String message = ex.getMessage();
        if (ex instanceof AgentBridgeException bridge) {
            errorCode = bridge.getErrorCode();
            retryable = bridge.isRetryable();
            message = bridge.getReason();
        } else if (ex instanceof ServiceException service) {
            errorCode = "ROW_WRITE_REJECTED";
            retryable = false;
            message = service.getMessage();
        }
        commandRowMapper.insert(baseRow(moduleId, commandId, row)
                .action("CREATE").status("FAILED")
                .errorCode(errorCode).errorMessage(trim(message, 512)).retryable(retryable).build());
    }

    private AgentImportPreflightRespVO.Row requirePlanRow(AgentImportPreflightRespVO plan, int index) {
        if (plan.getRows() == null || index >= plan.getRows().size()) {
            throw new AgentBridgeException(HttpStatus.BAD_GATEWAY, "ADAPTER_CONTRACT_INVALID",
                    "适配器返回的预检行数与请求不一致");
        }
        return plan.getRows().get(index);
    }

    private void validatePlan(AgentImportPreflightRespVO plan, AgentImportPreflightReqVO request) {
        if (plan == null || StrUtil.isBlank(plan.getPreflightHash())) {
            throw new AgentBridgeException(HttpStatus.BAD_GATEWAY, "ADAPTER_CONTRACT_INVALID",
                    "适配器未返回预检摘要");
        }
        if (plan.getRows() == null || plan.getRows().size() != request.getRows().size()) {
            throw new AgentBridgeException(HttpStatus.BAD_GATEWAY, "ADAPTER_CONTRACT_INVALID",
                    "适配器返回的预检行数与请求不一致");
        }
        for (int index = 0; index < plan.getRows().size(); index++) {
            if (!request.getRows().get(index).getCandidateId()
                    .equals(plan.getRows().get(index).getCandidateId())) {
                throw new AgentBridgeException(HttpStatus.BAD_GATEWAY, "ADAPTER_CONTRACT_INVALID",
                        "适配器返回的预检行顺序与请求不一致");
            }
        }
    }

    private void validateAttachment(Long attachmentId, String sourceSha256) {
        String actual = attachmentService.sha256OfAttachment(attachmentId);
        if (!StrUtil.equalsIgnoreCase(actual, sourceSha256)) {
            throw new AgentBridgeException(HttpStatus.CONFLICT, "SOURCE_CHANGED",
                    "附件内容已变化，请重新上传后再试", true, null);
        }
    }

    private void validateRows(AgentImportPreflightReqVO request) {
        if (request.getRows() == null || request.getRows().isEmpty()) {
            throw new AgentBridgeException(HttpStatus.UNPROCESSABLE_ENTITY, "ROWS_EMPTY", "没有可导入的数据行");
        }
        if (request.getRows().size() > maxRows) {
            throw new AgentBridgeException(HttpStatus.UNPROCESSABLE_ENTITY, "ROWS_TOO_MANY",
                    "单次导入超过 " + maxRows + " è¡Œï¼Œè¯·æ‹†åˆ†æ–‡ä»¶");
        }
    }

    private void requirePermission(AgentImportModuleAdapter adapter) {
        for (String permission : adapter.permissions()) {
            if (!securityFrameworkService.hasPermission(permission)) {
                throw new AgentBridgeException(HttpStatus.FORBIDDEN, "FORBIDDEN",
                        "当前用户缺少该模块的导入权限（" + permission + "）");
            }
        }
    }

    private AgentImportModuleAdapter requireAdapter(String moduleId) {
        AgentImportModuleAdapter adapter = adapters().get(moduleId);
        if (adapter == null) {
            throw new AgentBridgeException(HttpStatus.NOT_FOUND, "MODULE_NOT_SUPPORTED",
                    "不支持的导入模块：" + moduleId);
        }
        return adapter;
    }

    private Map<String, AgentImportModuleAdapter> adapters() {
        Map<String, AgentImportModuleAdapter> result = new LinkedHashMap<>();
        for (AgentImportModuleAdapter adapter : adapterProvider) {
            result.put(adapter.moduleId(), adapter);
        }
        return result;
    }

    private static AgentImportCommandRowDO.AgentImportCommandRowDOBuilder baseRow(
            String moduleId, String commandId, AgentImportPreflightReqVO.Row row) {
        return AgentImportCommandRowDO.builder()
                .commandId(commandId)
                .moduleId(moduleId)
                .candidateId(row.getCandidateId())
                .sourceRow(row.getRowNo())
                .code(row.getValues().get("code"))
                .name(row.getValues().get("name"));
    }

    private static void requireSameDigest(AgentImportCommandDO command, String digest) {
        if (!command.getRequestDigest().equals(digest)) {
            throw new AgentBridgeException(HttpStatus.CONFLICT, "IDEMPOTENCY_CONFLICT",
                    "同一命令或幂等键携带了不同的请求内容，请检查任务状态后重试");
        }
    }

    private static String requestDigest(String moduleId, AgentImportCommitReqVO request) {
        Map<String, Object> canonical = new LinkedHashMap<>();
        canonical.put("module_id", moduleId);
        canonical.put("attachment_id", request.getAttachmentId());
        canonical.put("session_id", request.getSessionId());
        canonical.put("source_sha256", StrUtil.nullToEmpty(request.getSourceSha256()).toLowerCase(Locale.ROOT));
        canonical.put("preflight_hash", StrUtil.nullToEmpty(request.getPreflightHash()).toLowerCase(Locale.ROOT));
        canonical.put("proposal_hash", StrUtil.nullToEmpty(request.getProposalHash()));
        canonical.put("proposal_revision", StrUtil.nullToEmpty(request.getProposalRevision()));
        canonical.put("options", request.getOptions() == null ? Map.of() : new TreeMap<>(request.getOptions()));
        List<Map<String, Object>> rows = new ArrayList<>();
        for (AgentImportPreflightReqVO.Row row : request.getRows()) {
            Map<String, Object> item = new LinkedHashMap<>();
            item.put("candidate_id", row.getCandidateId());
            item.put("row_no", row.getRowNo());
            item.put("values", row.getValues() == null ? Map.of() : new TreeMap<>(row.getValues()));
            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);
        }
    }

    /**
     * é€‚配器计算预检摘要的统一入口，避免各模块摘要口径漂移。
     */
    public static String preflightHash(String moduleId, AgentImportPreflightReqVO request,
                                       Object resolvedPayload) {
        Map<String, Object> canonical = new LinkedHashMap<>();
        canonical.put("module_id", moduleId);
        canonical.put("attachment_id", request.getAttachmentId());
        canonical.put("source_sha256", StrUtil.nullToEmpty(request.getSourceSha256()).toLowerCase(Locale.ROOT));
        canonical.put("proposal_revision", StrUtil.nullToEmpty(request.getProposalRevision()));
        canonical.put("options", request.getOptions() == null ? Map.of() : new TreeMap<>(request.getOptions()));
        List<Map<String, Object>> rows = new ArrayList<>();
        for (AgentImportPreflightReqVO.Row row : request.getRows()) {
            Map<String, Object> item = new LinkedHashMap<>();
            item.put("candidate_id", row.getCandidateId());
            item.put("row_no", row.getRowNo());
            item.put("values", row.getValues() == null ? Map.of() : new TreeMap<>(row.getValues()));
            rows.add(item);
        }
        canonical.put("rows", rows);
        canonical.put("resolved", resolvedPayload);
        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 AgentImportReceiptRespVO toReceipt(AgentImportCommandDO command) {
        String status = command.getStatus();
        if (STATUS_RUNNING.equals(status) && command.getLeaseExpiresAt() != null
                && command.getLeaseExpiresAt().isBefore(LocalDateTime.now())) {
            status = STATUS_UNKNOWN;
        }
        List<AgentImportCommandRowDO> rowDOs = commandRowMapper.selectListByCommandId(command.getCommandId());
        List<AgentImportReceiptRespVO.Row> rows = rowDOs.stream()
                .map(row -> AgentImportReceiptRespVO.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 AgentImportReceiptRespVO.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(AgentImportReceiptRespVO.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 String trim(String value, int maxLength) {
        if (value == null) {
            return null;
        }
        return value.length() <= maxLength ? value : value.substring(0, maxLength);
    }
}