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

## 输入引用

引用相关的 docs 文档链接：

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

## Context Maps

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

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

## 目标

实现 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 枚举

```java
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 接口

```java
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 实现类

```java
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 控制器

```java
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 | 代码结构清晰，符合项目规范 | 无 |