kris/datai

Fork 0

Kris 3d290a8dd1 feat：归档当前所有文件

2026-01-28 18:23:25 +08:00

22 KiB

Raw Permalink Blame History

Prompt - 元数据拉取核心功能实现

输入引用

引用相关的 docs 文档链接：

REQ-010-6.md - 元数据拉取核心功能需求
ADR-0015.md - 元数据拉取核心功能架构决策
metadata-module.md - Salesforce Metadata API 模块说明（唯一真源）
index.md - Salesforce SOAP API Java 客户端参考文档（唯一真源）
REQ-010-1.md - 数据库表结构设计和创建
REQ-010-2.md - 基础实体类和Mapper创建
REQ-010-3.md - Salesforce组织配置管理
REQ-010-4.md - 元数据任务定义管理
REQ-010-5.md - Metadata API客户端封装

Context Maps

强制列出本次 Prompt 依赖的 Canvas 文件：

Authentication.canvas - 项目架构视觉化展示
相关节点: 集成核心 - 提供与Salesforce的各种连接方式
相关节点: SessionManager - 会话管理，提供登录服务

目标

实现 Salesforce 元数据拉取的核心功能，包括手动触发拉取、异步拉取执行、状态监控、拉取历史记录、拉取进度查询、拉取取消功能。

输出格式

1. 代码结构

datai-salesforce-metadata/
├── src/main/java/com/datai/metadata/
│   ├── controller/
│   │   └── MetadataRetrieveController.java - 元数据拉取控制器
│   ├── service/
│   │   ├── IMetadataRetrieveService.java - 元数据拉取服务接口
│   │   └── impl/
│   │       └── MetadataRetrieveServiceImpl.java - 元数据拉取服务实现
│   ├── enums/
│   │   └── RetrieveStatus.java - 拉取状态枚举
│   └── dto/
│       ├── RetrieveTriggerRequest.java - 拉取触发请求DTO
│       ├── RetrieveTriggerResponse.java - 拉取触发响应DTO
│       ├── RetrieveProgressResponse.java - 拉取进度响应DTO
│       └── RetrieveHistoryResponse.java - 拉取历史响应DTO
└── src/main/resources/
    └── mapper/
        └── metadata/
            └── DataiMetaJobExecutionMapper.xml - 作业执行Mapper XML

2. 代码示例

2.1 RetrieveStatus 枚举

package com.datai.metadata.enums;

import com.baomidou.mybatisplus.annotation.EnumValue;
import com.fasterxml.jackson.annotation.JsonValue;

/**
 * 拉取状态枚举
 */
public enum RetrieveStatus {
    
    PENDING("Pending", "待处理"),
    PROCESSING("Processing", "处理中"),
    SUCCESS("Success", "成功"),
    FAILED("Failed", "失败"),
    PARTIAL_SUCCESS("Partial_Success", "部分成功"),
    CANCELLED("Cancelled", "已取消");
    
    @EnumValue
    @JsonValue
    private final String code;
    
    private final String description;
    
    RetrieveStatus(String code, String description) {
        this.code = code;
        this.description = description;
    }
    
    public String getCode() {
        return code;
    }
    
    public String getDescription() {
        return description;
    }
}

2.2 IMetadataRetrieveService 接口

package com.datai.metadata.service;

import com.baomidou.mybatisplus.core.metadata.IPage;
import com.datai.metadata.dto.*;
import com.datai.metadata.entity.DataiMetaJobExecution;

/**
 * 元数据拉取服务接口
 */
public interface IMetadataRetrieveService {

    /**
     * 触发拉取
     * 
     * @param request 拉取触发请求
     * @return 拉取触发响应
     */
    RetrieveTriggerResponse triggerRetrieve(RetrieveTriggerRequest request);

    /**
     * 查询拉取进度
     * 
     * @param jobId 作业ID
     * @return 拉取进度响应
     */
    RetrieveProgressResponse getRetrieveProgress(String jobId);

    /**
     * 取消拉取
     * 
     * @param jobId 作业ID
     */
    void cancelRetrieve(String jobId);

    /**
     * 查询拉取历史
     * 
     * @param pageNum 页码
     * @param pageSize 页大小
     * @param taskId 任务ID
     * @param orgConfigId 组织配置ID
     * @param status 状态
     * @return 拉取历史响应
     */
    IPage<RetrieveHistoryResponse> getRetrieveHistory(int pageNum, int pageSize, 
            Long taskId, Long orgConfigId, String status);
}

2.3 MetadataRetrieveServiceImpl 实现类

package com.datai.metadata.service.impl;

import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.datai.common.core.domain.AjaxResult;
import com.datai.metadata.client.*;
import com.datai.metadata.dto.*;
import com.datai.metadata.entity.DataiMetaJobExecution;
import com.datai.metadata.entity.DataiMetaTask;
import com.datai.metadata.enums.RetrieveStatus;
import com.datai.metadata.mapper.DataiMetaJobExecutionMapper;
import com.datai.metadata.mapper.DataiMetaTaskMapper;
import com.datai.metadata.service.IMetadataApiService;
import com.datai.metadata.service.IMetadataRetrieveService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.concurrent.ConcurrentHashMap;

/**
 * 元数据拉取服务实现类
 */
@Slf4j
@Service
public class MetadataRetrieveServiceImpl implements IMetadataRetrieveService {

    @Autowired
    private IMetadataApiService metadataApiService;

    @Autowired
    private DataiMetaJobExecutionMapper jobExecutionMapper;

    @Autowired
    private DataiMetaTaskMapper taskMapper;

    private final ConcurrentHashMap<String, RetrieveProgress> progressMap = new ConcurrentHashMap<>();

    @Override
    @Transactional
    public RetrieveTriggerResponse triggerRetrieve(RetrieveTriggerRequest request) {
        // 查询任务
        DataiMetaTask task = taskMapper.selectById(request.getTaskId());
        if (task == null) {
            throw new BusinessException("任务不存在");
        }

        // 创建作业执行记录
        DataiMetaJobExecution jobExecution = new DataiMetaJobExecution();
        jobExecution.setTaskId(request.getTaskId());
        jobExecution.setOrgConfigId(request.getOrgConfigId());
        jobExecution.setJobType(task.getTaskType());
        jobExecution.setStatus(RetrieveStatus.PENDING.getCode());
        jobExecution.setCreatedBy(request.getCreatedBy());
        jobExecutionMapper.insert(jobExecution);

        // 异步执行拉取
        executeRetrieveAsync(jobExecution.getId(), task);

        // 返回响应
        RetrieveTriggerResponse response = new RetrieveTriggerResponse();
        response.setJobId(jobExecution.getId().toString());
        response.setStatus(RetrieveStatus.PENDING.getCode());
        response.setMessage("拉取任务已创建");

        return response;
    }

    @Override
    public RetrieveProgressResponse getRetrieveProgress(String jobId) {
        // 查询作业执行记录
        DataiMetaJobExecution jobExecution = jobExecutionMapper.selectById(Long.parseLong(jobId));
        if (jobExecution == null) {
            throw new BusinessException("作业不存在");
        }

        // 查询进度
        RetrieveProgress progress = progressMap.get(jobId);
        if (progress == null) {
            progress = new RetrieveProgress();
            progress.setStatus(jobExecution.getStatus());
            progress.setProgress(jobExecution.getStatus().equals(RetrieveStatus.SUCCESS.getCode()) ? 100 : 0);
        }

        // 返回响应
        RetrieveProgressResponse response = new RetrieveProgressResponse();
        response.setJobId(jobId);
        response.setStatus(progress.getStatus());
        response.setProgress(progress.getProgress());
        response.setMessage(progress.getMessage());

        return response;
    }

    @Override
    @Transactional
    public void cancelRetrieve(String jobId) {
        // 查询作业执行记录
        DataiMetaJobExecution jobExecution = jobExecutionMapper.selectById(Long.parseLong(jobId));
        if (jobExecution == null) {
            throw new BusinessException("作业不存在");
        }

        // 检查状态
        if (!RetrieveStatus.PROCESSING.getCode().equals(jobExecution.getStatus())) {
            throw new BusinessException("只能取消处理中的作业");
        }

        // 取消作业
        metadataApiService.cancelJob(jobId);

        // 更新状态
        jobExecution.setStatus(RetrieveStatus.CANCELLED.getCode());
        jobExecutionMapper.updateById(jobExecution);

        // 清除进度
        progressMap.remove(jobId);

        log.info("拉取任务已取消: {}", jobId);
    }

    @Override
    public IPage<RetrieveHistoryResponse> getRetrieveHistory(int pageNum, int pageSize, 
            Long taskId, Long orgConfigId, String status) {
        // 构建查询条件
        QueryWrapper<DataiMetaJobExecution> queryWrapper = new QueryWrapper<>();
        queryWrapper.eq("job_type", "retrieve");
        
        if (taskId != null) {
            queryWrapper.eq("task_id", taskId);
        }
        
        if (orgConfigId != null) {
            queryWrapper.eq("org_config_id", orgConfigId);
        }
        
        if (status != null && !status.isEmpty()) {
            queryWrapper.eq("status", status);
        }
        
        queryWrapper.orderByDesc("created_time");

        // 分页查询
        Page<DataiMetaJobExecution> page = new Page<>(pageNum, pageSize);
        IPage<DataiMetaJobExecution> jobExecutionPage = jobExecutionMapper.selectPage(page, queryWrapper);

        // 转换为响应
        IPage<RetrieveHistoryResponse> responsePage = new Page<>(pageNum, pageSize, jobExecutionPage.getTotal());
        for (DataiMetaJobExecution jobExecution : jobExecutionPage.getRecords()) {
            RetrieveHistoryResponse response = new RetrieveHistoryResponse();
            BeanUtils.copyProperties(jobExecution, response);
            responsePage.getRecords().add(response);
        }

        return responsePage;
    }

    /**
     * 异步执行拉取
     * 
     * @param jobExecutionId 作业执行ID
     * @param task 任务
     */
    @Async("metadataTaskExecutor")
    public void executeRetrieveAsync(Long jobExecutionId, DataiMetaTask task) {
        String jobId = jobExecutionId.toString();
        
        try {
            // 更新状态为处理中
            updateJobStatus(jobExecutionId, RetrieveStatus.PROCESSING);
            
            // 更新进度
            updateProgress(jobId, RetrieveStatus.PROCESSING.getCode(), 0, "正在拉取元数据");

            // 调用 Metadata API 拉取
            CompletableFuture<RetrieveResult> future = metadataApiService.retrieveAsync(
                    task.getOrgConfigId(), 
                    task.getPackageXml(), 
                    task.getApiVersion()
            );

            // 等待拉取完成
            RetrieveResult result = future.get();

            // 更新进度
            updateProgress(jobId, RetrieveStatus.PROCESSING.getCode(), 80, "正在下载 Zip 文件");

            // 保存 Zip 文件
            saveZipFile(jobExecutionId, result.getZipFile());

            // 更新状态为成功
            updateJobStatus(jobExecutionId, RetrieveStatus.SUCCESS);

            // 更新进度
            updateProgress(jobId, RetrieveStatus.SUCCESS.getCode(), 100, "拉取成功");

            // 清除进度
            progressMap.remove(jobId);

        } catch (Exception e) {
            log.error("拉取元数据失败: {}", jobId, e);
            
            // 更新状态为失败
            updateJobStatus(jobExecutionId, RetrieveStatus.FAILED);
            
            // 更新进度
            updateProgress(jobId, RetrieveStatus.FAILED.getCode(), 0, "拉取失败: " + e.getMessage());
            
            // 清除进度
            progressMap.remove(jobId);
        }
    }

    /**
     * 更新作业状态
     * 
     * @param jobExecutionId 作业执行ID
     * @param status 状态
     */
    private void updateJobStatus(Long jobExecutionId, RetrieveStatus status) {
        DataiMetaJobExecution jobExecution = new DataiMetaJobExecution();
        jobExecution.setId(jobExecutionId);
        jobExecution.setStatus(status.getCode());
        jobExecutionMapper.updateById(jobExecution);
    }

    /**
     * 更新进度
     * 
     * @param jobId 作业ID
     * @param status 状态
     * @param progress 进度
     * @param message 消息
     */
    private void updateProgress(String jobId, String status, int progress, String message) {
        RetrieveProgress retrieveProgress = new RetrieveProgress();
        retrieveProgress.setStatus(status);
        retrieveProgress.setProgress(progress);
        retrieveProgress.setMessage(message);
        progressMap.put(jobId, retrieveProgress);
    }

    /**
     * 保存 Zip 文件
     * 
     * @param jobExecutionId 作业执行ID
     * @param zipFile Zip 文件字节数组
     */
    private void saveZipFile(Long jobExecutionId, byte[] zipFile) {
        // 实现略
    }

    /**
     * 拉取进度
     */
    private static class RetrieveProgress {
        private String status;
        private int progress;
        private String message;

        public String getStatus() {
            return status;
        }

        public void setStatus(String status) {
            this.status = status;
        }

        public int getProgress() {
            return progress;
        }

        public void setProgress(int progress) {
            this.progress = progress;
        }

        public String getMessage() {
            return message;
        }

        public void setMessage(String message) {
            this.message = message;
        }
    }
}

2.4 MetadataRetrieveController 控制器

package com.datai.metadata.controller;

import com.baomidou.mybatisplus.core.metadata.IPage;
import com.datai.common.core.controller.BaseController;
import com.datai.common.core.domain.AjaxResult;
import com.datai.metadata.dto.*;
import com.datai.metadata.service.IMetadataRetrieveService;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.validation.annotation.Validated;
import org.springframework.web.bind.annotation.*;

import javax.validation.Valid;

/**
 * 元数据拉取控制器
 */
@Slf4j
@Api(tags = "元数据拉取管理")
@RestController
@RequestMapping("/metadata/retrieve")
public class MetadataRetrieveController extends BaseController {

    @Autowired
    private IMetadataRetrieveService metadataRetrieveService;

    /**
     * 触发拉取
     * 
     * @param request 拉取触发请求
     * @return 拉取触发响应
     */
    @ApiOperation("触发拉取")
    @PostMapping("/trigger")
    public AjaxResult triggerRetrieve(@Valid @RequestBody RetrieveTriggerRequest request) {
        RetrieveTriggerResponse response = metadataRetrieveService.triggerRetrieve(request);
        return AjaxResult.success(response);
    }

    /**
     * 查询拉取进度
     * 
     * @param jobId 作业ID
     * @return 拉取进度响应
     */
    @ApiOperation("查询拉取进度")
    @GetMapping("/progress/{jobId}")
    public AjaxResult getRetrieveProgress(@PathVariable String jobId) {
        RetrieveProgressResponse response = metadataRetrieveService.getRetrieveProgress(jobId);
        return AjaxResult.success(response);
    }

    /**
     * 取消拉取
     * 
     * @param jobId 作业ID
     * @return 操作结果
     */
    @ApiOperation("取消拉取")
    @DeleteMapping("/{jobId}")
    public AjaxResult cancelRetrieve(@PathVariable String jobId) {
        metadataRetrieveService.cancelRetrieve(jobId);
        return AjaxResult.success();
    }

    /**
     * 查询拉取历史
     * 
     * @param pageNum 页码
     * @param pageSize 页大小
     * @param taskId 任务ID
     * @param orgConfigId 组织配置ID
     * @param status 状态
     * @return 拉取历史响应
     */
    @ApiOperation("查询拉取历史")
    @GetMapping("/history")
    public AjaxResult getRetrieveHistory(
            @RequestParam(defaultValue = "1") int pageNum,
            @RequestParam(defaultValue = "10") int pageSize,
            @RequestParam(required = false) Long taskId,
            @RequestParam(required = false) Long orgConfigId,
            @RequestParam(required = false) String status) {
        IPage<RetrieveHistoryResponse> response = metadataRetrieveService.getRetrieveHistory(
                pageNum, pageSize, taskId, orgConfigId, status);
        return AjaxResult.success(response);
    }
}

3. 单元测试

必须为以下类编写单元测试：

MetadataRetrieveServiceImpl
MetadataRetrieveController

4. 集成测试

必须编写以下集成测试：

手动触发拉取测试
异步拉取执行测试
状态监控测试
拉取历史记录测试
拉取进度查询测试
拉取取消测试

5. API 文档

为以下接口编写 API 文档：

POST /metadata/retrieve/trigger - 触发拉取
GET /metadata/retrieve/progress/{jobId} - 查询拉取进度
DELETE /metadata/retrieve/{jobId} - 取消拉取
GET /metadata/retrieve/history - 查询拉取历史

约束

技术栈限制

后端: Spring Boot 3, MyBatis Plus, MySQL
前端: Vue 3, Element Plus
Salesforce API: Metadata API, Partner API
异步执行: 必须使用 Spring 的 @Async 注解和线程池
状态轮询: 必须使用定时任务实现状态轮询

架构约束

模块约束: 必须在 datai-salesforce-metadata 模块下实现
认证约束: 必须使用 SessionManager 进行会话管理和自动重新登录
API 约束: 必须遵循 Salesforce Metadata API 调用规范

性能要求

异步执行: 异步执行不影响系统响应
状态轮询: 状态轮询频率合理（如每 5 秒轮询一次）
历史记录查询: 历史记录查询性能良好，支持分页和条件查询

安全性要求

敏感信息: 不记录敏感信息（如密码、令牌）
错误信息: 不暴露系统内部信息

兼容性要求

API 版本: 支持 API 版本 58.0、57.0、56.0
Java 版本: 支持 Java 17

Rule Set

"请严格参考 @Authentication.canvas 中的状态机转移逻辑，不要自行发挥。"

具体规则：

必须使用 Canvas 中定义的类名和方法名
必须遵循 Canvas 中定义的调用关系
必须使用 SessionManager 进行会话管理和自动重新登录
必须使用现有的认证模块进行 OAuth 认证
必须使用现有的集成核心功能进行 API 调用
必须遵循现有的异常处理机制
必须遵循现有的日志记录规范

验收标准

功能完整性

手动触发拉取功能正常工作：能够成功手动触发拉取，支持选择任务ID和组织配置，触发成功返回 Job ID
异步拉取执行正常工作：异步拉取执行正常工作，使用线程池管理异步任务，支持并发拉取，拉取任务不阻塞系统响应
状态监控正常工作：状态监控正常工作，使用状态机管理拉取状态，支持状态查询，状态更新及时
拉取历史记录正常工作：拉取历史记录成功，支持分页查询，支持条件查询，历史记录完整
拉取进度查询正常工作：拉取进度查询成功，支持实时进度查询，进度信息准确，支持进度百分比显示
拉取取消功能正常工作：拉取取消功能正常工作，支持取消正在进行的拉取任务，取消后资源正确释放，取消状态更新及时

代码正确性

代码符合项目编码规范，有清晰的注释
代码结构清晰，易于扩展和维护
代码易于单元测试和集成测试

性能指标

拉取操作不影响系统响应，系统响应时间在可接受范围内
状态轮询频率合理，不会导致 API 限流
历史记录查询性能良好，支持分页和条件查询

风险

输出质量风险

代码质量: AI 生成的代码可能存在 bug 或性能问题
- 缓解措施: 编写详细的单元测试和集成测试，进行代码审查

技术实现风险

异步执行: 异步执行机制复杂可能导致状态管理困难
- 缓解措施: 使用状态机管理拉取状态，使用 ConcurrentHashMap 存储异步任务状态
状态轮询: 状态轮询频率不当可能导致 API 限流
- 缓解措施: 设置合理的轮询频率（如每 5 秒轮询一次），使用超时机制防止无限轮询
拉取取消: 拉取取消功能复杂可能导致资源泄漏
- 缓解措施: 使用 Future.cancel() 取消异步任务，使用状态机管理取消状态
历史记录: 拉取历史记录过多可能影响查询性能
- 缓解措施: 使用分页查询，定期清理历史记录
进度查询: 进度查询不准确可能导致用户体验差
- 缓解措施: 使用轮询机制获取进度，使用缓存提高查询性能

时间成本风险

开发周期: 开发复杂度高，可能延长开发周期
- 缓解措施: 分阶段实施，先实现核心功能，再实现辅助功能

使用记录

日期	使用场景	输入参数	输出结果	反馈	改进措施
2026-01-18	元数据拉取核心功能实现	REQ-010-6, ADR-0015	MetadataRetrieveController, MetadataRetrieveService, RetrieveStatus	代码结构清晰，符合项目规范	无

22 KiB Raw Permalink Blame History Unescape Escape