feat: 完善工作流 Public API 调用能力
- 支持 JSON 文件 URL 简写与 Multipart 单请求文件上传 - 完善执行拓扑、枚举状态、节点名称、恢复校验和安全错误响应 - 增加临时上传生命周期清理并升级 MinIO SDK - 重构工作流接口调用说明弹窗的扁平响应式布局
This commit is contained in:
@@ -1,5 +1,6 @@
|
||||
package tech.easyflow.ai.easyagentsflow.config;
|
||||
|
||||
import com.easyagents.flow.core.chain.event.ChainStatusChangeEvent;
|
||||
import com.easyagents.flow.core.chain.repository.ChainDefinitionRepository;
|
||||
import com.easyagents.flow.core.chain.repository.ChainDefinitionSnapshotRepository;
|
||||
import com.easyagents.flow.core.chain.repository.ChainStateRepository;
|
||||
@@ -13,6 +14,7 @@ import org.springframework.context.annotation.Configuration;
|
||||
import tech.easyflow.ai.easyagentsflow.listener.ChainErrorListenerForSave;
|
||||
import tech.easyflow.ai.easyagentsflow.listener.ChainEventListenerForSave;
|
||||
import tech.easyflow.ai.easyagentsflow.listener.NodeErrorListenerForSave;
|
||||
import tech.easyflow.ai.easyagentsflow.upload.WorkflowApiUploadCleanupListener;
|
||||
|
||||
import javax.annotation.Resource;
|
||||
import java.time.Duration;
|
||||
@@ -36,6 +38,8 @@ public class ChainExecutorConfig {
|
||||
@Resource
|
||||
private ChainEventListenerForSave chainEventListenerForSave;
|
||||
@Resource
|
||||
private WorkflowApiUploadCleanupListener workflowApiUploadCleanupListener;
|
||||
@Resource
|
||||
private WorkflowExecutionBudgetProperties workflowExecutionBudgetProperties;
|
||||
@Resource
|
||||
private WorkflowRuntimeProperties workflowRuntimeProperties;
|
||||
@@ -84,6 +88,9 @@ public class ChainExecutorConfig {
|
||||
*/
|
||||
private void saveStepsListeners(ChainExecutor chainExecutor) {
|
||||
chainExecutor.addEventListener(chainEventListenerForSave);
|
||||
chainExecutor.addEventListener(
|
||||
ChainStatusChangeEvent.class,
|
||||
workflowApiUploadCleanupListener);
|
||||
chainExecutor.addErrorListener(new ChainErrorListenerForSave());
|
||||
chainExecutor.addNodeErrorListener(new NodeErrorListenerForSave());
|
||||
}
|
||||
|
||||
@@ -51,8 +51,17 @@ public class TinyFlowService {
|
||||
}
|
||||
ChainInfo res = getChainInfo(executeId, chainState);
|
||||
|
||||
if (nodes != null) {
|
||||
if (nodes != null && !nodes.isEmpty()) {
|
||||
Map<String, String> resolvedNodeNames =
|
||||
chainExecutor.getInstanceNodeNames(chainState);
|
||||
Map<String, String> nodeNames = resolvedNodeNames == null
|
||||
? Map.of()
|
||||
: resolvedNodeNames;
|
||||
for (NodeInfo node : nodes) {
|
||||
if (node != null
|
||||
&& StringUtil.noText(node.getNodeName())) {
|
||||
node.setNodeName(nodeNames.get(node.getNodeId()));
|
||||
}
|
||||
processNodeState(executeId, node, chainState, nodeStateRepository);
|
||||
res.getNodes().put(node.getNodeId(), node);
|
||||
}
|
||||
|
||||
@@ -12,6 +12,9 @@ import tech.easyflow.common.web.exceptions.BusinessException;
|
||||
import tech.easyflow.ai.entity.Workflow;
|
||||
|
||||
import javax.annotation.Resource;
|
||||
import java.net.URI;
|
||||
import java.net.URLDecoder;
|
||||
import java.nio.charset.StandardCharsets;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Collection;
|
||||
import java.util.LinkedHashMap;
|
||||
@@ -88,6 +91,53 @@ public class WorkflowRunningParameterResolver {
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 解析开始节点中可通过 multipart 上传的文件参数名。
|
||||
*
|
||||
* @param content 工作流内容
|
||||
* @return 保持开始节点定义顺序的文件参数名集合
|
||||
*/
|
||||
public Set<String> resolveFileParameterNames(String content) {
|
||||
List<Parameter> startParameters = resolveStartParameters(content);
|
||||
Set<String> names = new LinkedHashSet<>();
|
||||
if (startParameters == null || startParameters.isEmpty()) {
|
||||
return names;
|
||||
}
|
||||
for (Parameter parameter : startParameters) {
|
||||
String name = trimToNull(
|
||||
parameter == null ? null : parameter.getName());
|
||||
if (StringUtils.hasText(name) && isFileParameter(parameter)) {
|
||||
names.add(name);
|
||||
}
|
||||
}
|
||||
return names;
|
||||
}
|
||||
|
||||
/**
|
||||
* 解析开始节点中必须提供值的文件参数名。
|
||||
*
|
||||
* @param content 工作流内容
|
||||
* @return 保持开始节点定义顺序的必填文件参数名集合
|
||||
*/
|
||||
public Set<String> resolveRequiredFileParameterNames(
|
||||
String content) {
|
||||
List<Parameter> startParameters = resolveStartParameters(content);
|
||||
Set<String> names = new LinkedHashSet<>();
|
||||
if (startParameters == null || startParameters.isEmpty()) {
|
||||
return names;
|
||||
}
|
||||
for (Parameter parameter : startParameters) {
|
||||
String name = trimToNull(
|
||||
parameter == null ? null : parameter.getName());
|
||||
if (StringUtils.hasText(name)
|
||||
&& isFileParameter(parameter)
|
||||
&& parameter.isRequired()) {
|
||||
names.add(name);
|
||||
}
|
||||
}
|
||||
return names;
|
||||
}
|
||||
|
||||
/**
|
||||
* 归一化工作流运行时变量，确保文件参数统一为文件对象数组。
|
||||
*
|
||||
@@ -508,9 +558,9 @@ public class WorkflowRunningParameterResolver {
|
||||
Set<String> seenFilePaths = new LinkedHashSet<>();
|
||||
long totalSize = 0L;
|
||||
for (Object candidate : candidates) {
|
||||
if (!(candidate instanceof Map<?, ?> fileMap)) {
|
||||
throw new BusinessException("文件参数 " + parameterName + " 的输入格式不正确，必须为文件对象或文件对象数组");
|
||||
}
|
||||
Map<?, ?> fileMap = normalizeFileCandidate(
|
||||
candidate,
|
||||
parameterName);
|
||||
String fileName = trimObjectToNull(fileMap.get("fileName"));
|
||||
String filePath = trimObjectToNull(fileMap.get("filePath"));
|
||||
if (!StringUtils.hasText(fileName)) {
|
||||
@@ -541,6 +591,94 @@ public class WorkflowRunningParameterResolver {
|
||||
return normalized;
|
||||
}
|
||||
|
||||
/**
|
||||
* 将文件对象或远程 URL 字符串转换为统一文件描述。
|
||||
*
|
||||
* @param candidate 原始文件值
|
||||
* @param parameterName 工作流文件参数名
|
||||
* @return 可继续执行通用校验的文件描述
|
||||
* @throws BusinessException URL 无效或无法识别文件名时抛出
|
||||
*/
|
||||
private Map<?, ?> normalizeFileCandidate(
|
||||
Object candidate,
|
||||
String parameterName) {
|
||||
if (candidate instanceof Map<?, ?> fileMap) {
|
||||
return fileMap;
|
||||
}
|
||||
if (!(candidate instanceof String stringValue)) {
|
||||
throw new BusinessException(
|
||||
"文件参数 " + parameterName
|
||||
+ " 的输入格式不正确，必须为文件 URL、文件对象或对应数组");
|
||||
}
|
||||
String fileUrl = trimToNull(stringValue);
|
||||
if (!isHttpUrl(fileUrl)) {
|
||||
throw new BusinessException(
|
||||
"文件参数 " + parameterName
|
||||
+ " 仅支持 HTTP/HTTPS 文件 URL");
|
||||
}
|
||||
|
||||
Map<String, Object> normalized = new LinkedHashMap<>();
|
||||
normalized.put("fileName", resolveRemoteFileName(
|
||||
fileUrl,
|
||||
parameterName));
|
||||
normalized.put("filePath", fileUrl);
|
||||
return normalized;
|
||||
}
|
||||
|
||||
/**
|
||||
* 从远程 URL 路径中提取并解码文件名。
|
||||
*
|
||||
* @param fileUrl 远程文件 URL
|
||||
* @param parameterName 工作流文件参数名
|
||||
* @return 带扩展名的文件名
|
||||
* @throws BusinessException URL 无效或路径中没有可识别文件名时抛出
|
||||
*/
|
||||
private String resolveRemoteFileName(
|
||||
String fileUrl,
|
||||
String parameterName) {
|
||||
try {
|
||||
URI uri = URI.create(fileUrl);
|
||||
String rawPath = uri.getRawPath();
|
||||
int lastSlash = rawPath == null ? -1 : rawPath.lastIndexOf('/');
|
||||
String rawFileName = lastSlash < 0
|
||||
? rawPath
|
||||
: rawPath.substring(lastSlash + 1);
|
||||
String fileName = StringUtils.hasText(rawFileName)
|
||||
? URLDecoder.decode(
|
||||
rawFileName.replace("+", "%2B"),
|
||||
StandardCharsets.UTF_8)
|
||||
: null;
|
||||
int lastDot = fileName == null ? -1 : fileName.lastIndexOf('.');
|
||||
String extension = lastDot < 0
|
||||
? null
|
||||
: fileName.substring(lastDot + 1);
|
||||
if (!StringUtils.hasText(uri.getRawAuthority())
|
||||
|| !StringUtils.hasText(fileName)
|
||||
|| lastDot <= 0
|
||||
|| !StringUtils.hasText(extension)
|
||||
|| !extension.matches("[A-Za-z0-9]{1,16}")
|
||||
|| fileName.indexOf('/') >= 0
|
||||
|| fileName.indexOf('\\') >= 0) {
|
||||
throw invalidRemoteFileName(parameterName);
|
||||
}
|
||||
return fileName;
|
||||
} catch (IllegalArgumentException exception) {
|
||||
throw invalidRemoteFileName(parameterName);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 构建无法从 URL 识别文件名时的统一业务异常。
|
||||
*
|
||||
* @param parameterName 工作流文件参数名
|
||||
* @return 统一业务异常
|
||||
*/
|
||||
private BusinessException invalidRemoteFileName(String parameterName) {
|
||||
return new BusinessException(
|
||||
"文件参数 " + parameterName
|
||||
+ " 的 URL 路径无法识别带扩展名的文件名，请改用包含 fileName 和 filePath 的文件对象");
|
||||
}
|
||||
|
||||
private void collectFileValues(Object value, List<Object> result) {
|
||||
if (value == null) {
|
||||
return;
|
||||
|
||||
@@ -0,0 +1,135 @@
|
||||
package tech.easyflow.ai.easyagentsflow.upload;
|
||||
|
||||
import org.springframework.stereotype.Component;
|
||||
import org.springframework.web.multipart.MultipartFile;
|
||||
import tech.easyflow.common.web.multipart.MultipartFileMetadataNormalizer;
|
||||
|
||||
import java.io.File;
|
||||
import java.io.IOException;
|
||||
import java.io.InputStream;
|
||||
import java.util.Objects;
|
||||
|
||||
/**
|
||||
* 为工作流 Public API 创建文件名和内容类型一致的 Multipart 文件视图。
|
||||
*/
|
||||
@Component
|
||||
public class WorkflowApiMultipartFileNormalizer {
|
||||
|
||||
/**
|
||||
* 归一化单个文件 Part，不读取或复制文件内容。
|
||||
*
|
||||
* @param file 原始文件 Part
|
||||
* @return 归一化文件视图；输入为空时返回 {@code null}
|
||||
*/
|
||||
public MultipartFile normalize(MultipartFile file) {
|
||||
if (file == null) {
|
||||
return null;
|
||||
}
|
||||
String filename = MultipartFileMetadataNormalizer.sanitizeFilename(
|
||||
file.getOriginalFilename());
|
||||
String contentType =
|
||||
MultipartFileMetadataNormalizer.normalizeContentType(
|
||||
filename,
|
||||
file.getContentType());
|
||||
if (Objects.equals(filename, file.getOriginalFilename())
|
||||
&& Objects.equals(contentType, file.getContentType())) {
|
||||
return file;
|
||||
}
|
||||
return new NormalizedMultipartFile(
|
||||
file,
|
||||
filename,
|
||||
contentType);
|
||||
}
|
||||
|
||||
/**
|
||||
* 仅覆盖安全元数据并委托文件内容访问的 Multipart 视图。
|
||||
*/
|
||||
private static final class NormalizedMultipartFile
|
||||
implements MultipartFile {
|
||||
|
||||
private final MultipartFile delegate;
|
||||
private final String originalFilename;
|
||||
private final String contentType;
|
||||
|
||||
/**
|
||||
* 创建归一化文件视图。
|
||||
*
|
||||
* @param delegate 原始文件
|
||||
* @param originalFilename 安全文件名
|
||||
* @param contentType 标准内容类型
|
||||
*/
|
||||
private NormalizedMultipartFile(
|
||||
MultipartFile delegate,
|
||||
String originalFilename,
|
||||
String contentType) {
|
||||
this.delegate = delegate;
|
||||
this.originalFilename = originalFilename;
|
||||
this.contentType = contentType;
|
||||
}
|
||||
|
||||
/**
|
||||
* {@inheritDoc}
|
||||
*/
|
||||
@Override
|
||||
public String getName() {
|
||||
return delegate.getName();
|
||||
}
|
||||
|
||||
/**
|
||||
* {@inheritDoc}
|
||||
*/
|
||||
@Override
|
||||
public String getOriginalFilename() {
|
||||
return originalFilename;
|
||||
}
|
||||
|
||||
/**
|
||||
* {@inheritDoc}
|
||||
*/
|
||||
@Override
|
||||
public String getContentType() {
|
||||
return contentType;
|
||||
}
|
||||
|
||||
/**
|
||||
* {@inheritDoc}
|
||||
*/
|
||||
@Override
|
||||
public boolean isEmpty() {
|
||||
return delegate.isEmpty();
|
||||
}
|
||||
|
||||
/**
|
||||
* {@inheritDoc}
|
||||
*/
|
||||
@Override
|
||||
public long getSize() {
|
||||
return delegate.getSize();
|
||||
}
|
||||
|
||||
/**
|
||||
* {@inheritDoc}
|
||||
*/
|
||||
@Override
|
||||
public byte[] getBytes() throws IOException {
|
||||
return delegate.getBytes();
|
||||
}
|
||||
|
||||
/**
|
||||
* {@inheritDoc}
|
||||
*/
|
||||
@Override
|
||||
public InputStream getInputStream() throws IOException {
|
||||
return delegate.getInputStream();
|
||||
}
|
||||
|
||||
/**
|
||||
* {@inheritDoc}
|
||||
*/
|
||||
@Override
|
||||
public void transferTo(File destination)
|
||||
throws IOException, IllegalStateException {
|
||||
delegate.transferTo(destination);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,43 @@
|
||||
package tech.easyflow.ai.easyagentsflow.upload;
|
||||
|
||||
import java.util.Map;
|
||||
|
||||
/**
|
||||
* Public Workflow API multipart 文件准备结果。
|
||||
*/
|
||||
public class WorkflowApiPreparedUpload {
|
||||
|
||||
private final String requestId;
|
||||
private final Map<String, Object> variables;
|
||||
|
||||
/**
|
||||
* 创建文件准备结果。
|
||||
*
|
||||
* @param requestId 临时上传请求 ID
|
||||
* @param variables 已注入文件描述的工作流变量
|
||||
*/
|
||||
public WorkflowApiPreparedUpload(
|
||||
String requestId,
|
||||
Map<String, Object> variables) {
|
||||
this.requestId = requestId;
|
||||
this.variables = variables;
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取临时上传请求 ID。
|
||||
*
|
||||
* @return 临时上传请求 ID
|
||||
*/
|
||||
public String getRequestId() {
|
||||
return requestId;
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取已注入文件描述的工作流变量。
|
||||
*
|
||||
* @return 工作流变量
|
||||
*/
|
||||
public Map<String, Object> getVariables() {
|
||||
return variables;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,41 @@
|
||||
package tech.easyflow.ai.easyagentsflow.upload;
|
||||
|
||||
import org.springframework.util.StringUtils;
|
||||
|
||||
import java.io.Serializable;
|
||||
|
||||
/**
|
||||
* Public Workflow API 已准备或已写入的临时文件。
|
||||
*
|
||||
* @param filePath 工作流运行时读取 URL；物理写入完成前为空
|
||||
* @param storageLocator 可恢复文件存储定位符
|
||||
*/
|
||||
public record WorkflowApiStoredFile(
|
||||
String filePath,
|
||||
String storageLocator) implements Serializable {
|
||||
|
||||
/**
|
||||
* 创建临时文件记录。
|
||||
*
|
||||
* @throws IllegalArgumentException 恢复定位符为空时抛出
|
||||
*/
|
||||
public WorkflowApiStoredFile {
|
||||
if (!StringUtils.hasText(storageLocator)) {
|
||||
throw new IllegalArgumentException(
|
||||
"工作流临时文件恢复定位符不能为空");
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 返回写入完成后的临时文件记录。
|
||||
*
|
||||
* @param resolvedFilePath 文件读取 URL
|
||||
* @return 包含原恢复定位符的新记录
|
||||
*/
|
||||
public WorkflowApiStoredFile withFilePath(
|
||||
String resolvedFilePath) {
|
||||
return new WorkflowApiStoredFile(
|
||||
resolvedFilePath,
|
||||
storageLocator);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,42 @@
|
||||
package tech.easyflow.ai.easyagentsflow.upload;
|
||||
|
||||
import com.easyagents.flow.core.chain.Chain;
|
||||
import com.easyagents.flow.core.chain.Event;
|
||||
import com.easyagents.flow.core.chain.event.ChainStatusChangeEvent;
|
||||
import com.easyagents.flow.core.chain.listener.ChainEventListener;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
/**
|
||||
* 工作流终态事件到临时上传清理队列的桥接监听器。
|
||||
*/
|
||||
@Component
|
||||
public class WorkflowApiUploadCleanupListener implements ChainEventListener {
|
||||
|
||||
private final WorkflowApiUploadLifecycleService lifecycleService;
|
||||
|
||||
/**
|
||||
* 创建临时上传清理监听器。
|
||||
*
|
||||
* @param lifecycleService 临时上传生命周期服务
|
||||
*/
|
||||
public WorkflowApiUploadCleanupListener(
|
||||
WorkflowApiUploadLifecycleService lifecycleService) {
|
||||
this.lifecycleService = lifecycleService;
|
||||
}
|
||||
|
||||
/**
|
||||
* 在工作流进入终态后触发异步清理登记。
|
||||
*
|
||||
* @param event 工作流事件
|
||||
* @param chain 工作流实例
|
||||
*/
|
||||
@Override
|
||||
public void onEvent(Event event, Chain chain) {
|
||||
if (event instanceof ChainStatusChangeEvent statusChangeEvent
|
||||
&& statusChangeEvent.getStatus() != null
|
||||
&& statusChangeEvent.getStatus().isTerminal()) {
|
||||
lifecycleService.markExecutionTerminal(
|
||||
chain.getStateInstanceId());
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,67 @@
|
||||
package tech.easyflow.ai.easyagentsflow.upload;
|
||||
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import org.springframework.scheduling.annotation.Scheduled;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
import java.time.Duration;
|
||||
|
||||
/**
|
||||
* Public Workflow API 临时上传兜底清理任务。
|
||||
*/
|
||||
@Component
|
||||
public class WorkflowApiUploadCleanupScheduler {
|
||||
|
||||
private static final Logger LOG = LoggerFactory.getLogger(
|
||||
WorkflowApiUploadCleanupScheduler.class);
|
||||
private static final int CLEANUP_BATCH_SIZE = 100;
|
||||
private static final int CLEANUP_MAX_RECORDS_PER_RUN = 2_000;
|
||||
private static final Duration CLEANUP_TIME_BUDGET =
|
||||
Duration.ofSeconds(30);
|
||||
|
||||
private final WorkflowApiUploadLifecycleService lifecycleService;
|
||||
|
||||
/**
|
||||
* 创建临时上传清理任务。
|
||||
*
|
||||
* @param lifecycleService 临时上传生命周期服务
|
||||
*/
|
||||
public WorkflowApiUploadCleanupScheduler(
|
||||
WorkflowApiUploadLifecycleService lifecycleService) {
|
||||
this.lifecycleService = lifecycleService;
|
||||
}
|
||||
|
||||
/**
|
||||
* 定期清理终态、启动失败或状态已丢失的临时上传。
|
||||
*/
|
||||
@Scheduled(
|
||||
fixedDelayString =
|
||||
"${easyflow.workflow.api-upload.cleanup-interval:1m}")
|
||||
public void cleanup() {
|
||||
try {
|
||||
int processed = 0;
|
||||
long deadline = System.nanoTime()
|
||||
+ CLEANUP_TIME_BUDGET.toNanos();
|
||||
while (processed < CLEANUP_MAX_RECORDS_PER_RUN
|
||||
&& System.nanoTime() < deadline) {
|
||||
int batchSize = Math.min(
|
||||
CLEANUP_BATCH_SIZE,
|
||||
CLEANUP_MAX_RECORDS_PER_RUN - processed);
|
||||
int batchProcessed =
|
||||
lifecycleService.cleanupExpired(batchSize);
|
||||
processed += batchProcessed;
|
||||
if (batchProcessed < batchSize) {
|
||||
break;
|
||||
}
|
||||
}
|
||||
if (processed > 0) {
|
||||
LOG.info(
|
||||
"已处理 {} 条工作流 API 临时上传清理记录",
|
||||
processed);
|
||||
}
|
||||
} catch (RuntimeException error) {
|
||||
LOG.error("工作流 API 临时上传定时清理失败", error);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,688 @@
|
||||
package tech.easyflow.ai.easyagentsflow.upload;
|
||||
|
||||
import com.easyagents.flow.core.chain.ChainState;
|
||||
import com.easyagents.flow.core.chain.repository.ChainStateRepository;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import org.springframework.beans.factory.annotation.Qualifier;
|
||||
import org.springframework.stereotype.Service;
|
||||
import org.springframework.util.StringUtils;
|
||||
import org.springframework.web.multipart.MultipartFile;
|
||||
import tech.easyflow.ai.easyagentsflow.service.WorkflowRunningParameterResolver;
|
||||
import tech.easyflow.common.cache.RedisLockExecutor;
|
||||
import tech.easyflow.common.filestorage.FileStorageService;
|
||||
import tech.easyflow.common.filestorage.FileStorageWriteHandle;
|
||||
import tech.easyflow.common.filestorage.FileStorageWriteResult;
|
||||
import tech.easyflow.common.web.exceptions.BusinessException;
|
||||
|
||||
import java.net.ConnectException;
|
||||
import java.net.SocketTimeoutException;
|
||||
import java.net.UnknownHostException;
|
||||
import java.net.http.HttpTimeoutException;
|
||||
import java.time.Duration;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Collection;
|
||||
import java.util.LinkedHashMap;
|
||||
import java.util.List;
|
||||
import java.util.Locale;
|
||||
import java.util.Map;
|
||||
import java.util.Optional;
|
||||
import java.util.Set;
|
||||
import java.util.UUID;
|
||||
import java.util.concurrent.TimeoutException;
|
||||
|
||||
/**
|
||||
* Public Workflow API multipart 临时文件生命周期服务。
|
||||
*
|
||||
* <p>上传文件在请求线程中写入统一存储，工作流运行或挂起期间保留，
|
||||
* 工作流终态、启动失败或状态丢失后由幂等清理流程删除。</p>
|
||||
*/
|
||||
@Service
|
||||
public class WorkflowApiUploadLifecycleService {
|
||||
|
||||
private static final Logger LOG = LoggerFactory.getLogger(
|
||||
WorkflowApiUploadLifecycleService.class);
|
||||
private static final Duration STAGED_RETENTION = Duration.ofHours(1);
|
||||
private static final Duration TERMINAL_RETENTION = Duration.ofHours(1);
|
||||
private static final Duration ACTIVE_RECHECK = Duration.ofHours(24);
|
||||
private static final Duration CLEANUP_LOCK_WAIT = Duration.ZERO;
|
||||
private static final Duration CLEANUP_LOCK_LEASE = Duration.ofMinutes(5);
|
||||
private static final Duration CLEANUP_RETRY_DELAY =
|
||||
Duration.ofMinutes(5);
|
||||
private static final String STORAGE_PATH_PREFIX =
|
||||
"workflow-api-upload/";
|
||||
private static final String CLEANUP_LOCK_PREFIX =
|
||||
"easyflow:workflow:api-upload:cleanup-lock:";
|
||||
|
||||
private final WorkflowRunningParameterResolver parameterResolver;
|
||||
private final FileStorageService fileStorageService;
|
||||
private final WorkflowApiMultipartFileNormalizer fileNormalizer;
|
||||
private final WorkflowApiUploadStore uploadStore;
|
||||
private final ChainStateRepository chainStateRepository;
|
||||
private final RedisLockExecutor redisLockExecutor;
|
||||
|
||||
/**
|
||||
* 创建工作流 API 临时文件生命周期服务。
|
||||
*
|
||||
* @param parameterResolver 工作流运行参数解析器
|
||||
* @param fileStorageService 文件存储服务
|
||||
* @param fileNormalizer Multipart 文件元数据归一化器
|
||||
* @param uploadStore 临时上传记录存储
|
||||
* @param chainStateRepository 工作流状态仓储
|
||||
* @param redisLockExecutor Redis 分布式锁执行器
|
||||
*/
|
||||
public WorkflowApiUploadLifecycleService(
|
||||
WorkflowRunningParameterResolver parameterResolver,
|
||||
@Qualifier("default") FileStorageService fileStorageService,
|
||||
WorkflowApiMultipartFileNormalizer fileNormalizer,
|
||||
WorkflowApiUploadStore uploadStore,
|
||||
ChainStateRepository chainStateRepository,
|
||||
RedisLockExecutor redisLockExecutor) {
|
||||
this.parameterResolver = parameterResolver;
|
||||
this.fileStorageService = fileStorageService;
|
||||
this.fileNormalizer = fileNormalizer;
|
||||
this.uploadStore = uploadStore;
|
||||
this.chainStateRepository = chainStateRepository;
|
||||
this.redisLockExecutor = redisLockExecutor;
|
||||
}
|
||||
|
||||
/**
|
||||
* 校验、存储 multipart 文件并注入工作流变量。
|
||||
*
|
||||
* @param workflowContent 已发布工作流内容
|
||||
* @param variables 普通运行变量
|
||||
* @param fileParts 以工作流文件参数名分组的 multipart 文件
|
||||
* @return 临时上传准备结果
|
||||
*/
|
||||
public WorkflowApiPreparedUpload prepare(
|
||||
String workflowContent,
|
||||
Map<String, Object> variables,
|
||||
Map<String, List<MultipartFile>> fileParts) {
|
||||
if (fileParts == null || fileParts.isEmpty()) {
|
||||
throw new BusinessException(
|
||||
400,
|
||||
40016,
|
||||
"multipart 请求至少需要上传一个 files.<开始节点参数名> 文件 Part");
|
||||
}
|
||||
|
||||
Map<String, List<MultipartFile>> normalizedFileParts =
|
||||
normalizeFileParts(fileParts);
|
||||
|
||||
Set<String> fileParameterNames =
|
||||
parameterResolver.resolveFileParameterNames(workflowContent);
|
||||
Set<String> requiredFileParameterNames =
|
||||
parameterResolver.resolveRequiredFileParameterNames(
|
||||
workflowContent);
|
||||
Map<String, Object> baseVariables = new LinkedHashMap<>();
|
||||
if (variables != null) {
|
||||
baseVariables.putAll(variables);
|
||||
}
|
||||
validateFileParts(
|
||||
workflowContent,
|
||||
baseVariables,
|
||||
normalizedFileParts,
|
||||
fileParameterNames,
|
||||
requiredFileParameterNames);
|
||||
|
||||
WorkflowApiUploadRecord record = new WorkflowApiUploadRecord();
|
||||
long now = System.currentTimeMillis();
|
||||
record.setRequestId(UUID.randomUUID().toString().replace("-", ""));
|
||||
record.setCreatedAt(now);
|
||||
record.setCleanupAt(now + STAGED_RETENTION.toMillis());
|
||||
|
||||
try {
|
||||
uploadStore.create(record);
|
||||
Map<String, Object> resolvedVariables =
|
||||
storeFiles(
|
||||
baseVariables,
|
||||
normalizedFileParts,
|
||||
record);
|
||||
Map<String, Object> normalized =
|
||||
parameterResolver.normalizeRuntimeVariables(
|
||||
workflowContent,
|
||||
resolvedVariables);
|
||||
return new WorkflowApiPreparedUpload(
|
||||
record.getRequestId(),
|
||||
normalized);
|
||||
} catch (RuntimeException | Error error) {
|
||||
try {
|
||||
cleanupPreparationFailure(record);
|
||||
} catch (RuntimeException cleanupError) {
|
||||
error.addSuppressed(cleanupError);
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 在工作流首个节点启动前绑定执行 ID。
|
||||
*
|
||||
* @param requestId 临时上传请求 ID
|
||||
* @param executeId 工作流执行 ID
|
||||
*/
|
||||
public void bindExecution(String requestId, String executeId) {
|
||||
uploadStore.bindExecution(requestId, executeId);
|
||||
WorkflowApiUploadRecord record = uploadStore.find(requestId)
|
||||
.orElseThrow(() -> new IllegalStateException(
|
||||
"工作流临时上传记录不存在: " + requestId));
|
||||
uploadStore.schedule(
|
||||
record,
|
||||
System.currentTimeMillis() + ACTIVE_RECHECK.toMillis());
|
||||
}
|
||||
|
||||
/**
|
||||
* 标记工作流执行已进入终态，交由清理任务删除文件。
|
||||
*
|
||||
* @param executeId 工作流执行 ID
|
||||
*/
|
||||
public void markExecutionTerminal(String executeId) {
|
||||
uploadStore.findByExecutionId(executeId).ifPresent(record ->
|
||||
uploadStore.schedule(
|
||||
record,
|
||||
System.currentTimeMillis()
|
||||
+ TERMINAL_RETENTION.toMillis()));
|
||||
}
|
||||
|
||||
/**
|
||||
* 终止尚未成功启动的临时上传并立即清理。
|
||||
*
|
||||
* @param requestId 临时上传请求 ID
|
||||
*/
|
||||
public void abort(String requestId) {
|
||||
cleanupRequest(requestId, true);
|
||||
}
|
||||
|
||||
/**
|
||||
* 清理一批到期上传记录。
|
||||
*
|
||||
* @param limit 单次最大处理数量
|
||||
* @return 已领取并完成一次处理的上传记录数量
|
||||
*/
|
||||
public int cleanupExpired(int limit) {
|
||||
int processed = 0;
|
||||
int safeLimit = Math.max(1, limit);
|
||||
for (int index = 0; index < safeLimit; index++) {
|
||||
long now = System.currentTimeMillis();
|
||||
Optional<String> claimed = uploadStore.claimExpired(
|
||||
now,
|
||||
now + CLEANUP_RETRY_DELAY.toMillis());
|
||||
if (claimed.isEmpty()) {
|
||||
break;
|
||||
}
|
||||
String requestId = claimed.get();
|
||||
processed++;
|
||||
try {
|
||||
cleanupRequest(requestId, false);
|
||||
} catch (RuntimeException error) {
|
||||
LOG.error(
|
||||
"清理工作流 API 临时上传失败，requestId={}",
|
||||
requestId,
|
||||
error);
|
||||
}
|
||||
}
|
||||
return processed;
|
||||
}
|
||||
|
||||
/**
|
||||
* 校验 multipart 文件字段和既有变量冲突，并复用运行参数校验限制。
|
||||
*
|
||||
* @param workflowContent 工作流内容
|
||||
* @param variables 普通变量
|
||||
* @param fileParts 文件 Part
|
||||
* @param fileParameterNames 文件参数名
|
||||
* @param requiredFileParameterNames 必填文件参数名
|
||||
*/
|
||||
private void validateFileParts(
|
||||
String workflowContent,
|
||||
Map<String, Object> variables,
|
||||
Map<String, List<MultipartFile>> fileParts,
|
||||
Set<String> fileParameterNames,
|
||||
Set<String> requiredFileParameterNames) {
|
||||
for (String requiredParameterName
|
||||
: requiredFileParameterNames) {
|
||||
List<MultipartFile> uploaded =
|
||||
fileParts.get(requiredParameterName);
|
||||
if (!hasValue(variables.get(requiredParameterName))
|
||||
&& (uploaded == null || uploaded.isEmpty())) {
|
||||
throw new BusinessException(
|
||||
400,
|
||||
40016,
|
||||
"缺少必填文件参数 " + requiredParameterName
|
||||
+ "，请使用 files."
|
||||
+ requiredParameterName);
|
||||
}
|
||||
}
|
||||
Map<String, Object> candidates = new LinkedHashMap<>(variables);
|
||||
for (Map.Entry<String, List<MultipartFile>> entry :
|
||||
fileParts.entrySet()) {
|
||||
String parameterName = entry.getKey();
|
||||
if (!StringUtils.hasText(parameterName)
|
||||
|| !fileParameterNames.contains(parameterName)) {
|
||||
throw new BusinessException(
|
||||
400,
|
||||
40016,
|
||||
"文件 Part " + parameterName
|
||||
+ " 不是开始节点的文件参数");
|
||||
}
|
||||
if (hasValue(variables.get(parameterName))) {
|
||||
throw new BusinessException(
|
||||
400,
|
||||
40016,
|
||||
"文件参数 " + parameterName
|
||||
+ " 不能同时通过 metadata 和文件 Part 传值");
|
||||
}
|
||||
List<MultipartFile> files = entry.getValue();
|
||||
if (files == null || files.isEmpty()) {
|
||||
throw new BusinessException(
|
||||
400,
|
||||
40016,
|
||||
"文件参数 " + parameterName + " 不能为空");
|
||||
}
|
||||
List<Map<String, Object>> descriptors =
|
||||
new ArrayList<>(files.size());
|
||||
for (int index = 0; index < files.size(); index++) {
|
||||
MultipartFile file = files.get(index);
|
||||
validateMultipartFile(file, parameterName);
|
||||
descriptors.add(fileDescriptor(
|
||||
file,
|
||||
"multipart://" + parameterName + "/" + index));
|
||||
}
|
||||
candidates.put(parameterName, descriptors);
|
||||
}
|
||||
try {
|
||||
parameterResolver.normalizeRuntimeVariables(
|
||||
workflowContent,
|
||||
candidates);
|
||||
} catch (BusinessException error) {
|
||||
throw translateFileValidationFailure(error);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 将运行参数解析器中的文件校验错误转换为稳定公共错误码。
|
||||
*
|
||||
* @param error 原文件参数校验异常
|
||||
* @return 原异常或带稳定错误码的异常
|
||||
*/
|
||||
private BusinessException translateFileValidationFailure(
|
||||
BusinessException error) {
|
||||
String message = error.getMessage();
|
||||
if (message == null || !message.startsWith("文件参数 ")) {
|
||||
return error;
|
||||
}
|
||||
boolean limitExceeded = message.contains("超过")
|
||||
|| message.contains("最多上传");
|
||||
return new BusinessException(
|
||||
limitExceeded ? 413 : 400,
|
||||
limitExceeded ? 41301 : 40016,
|
||||
message,
|
||||
error);
|
||||
}
|
||||
|
||||
/**
|
||||
* 将通过校验的文件写入统一存储。
|
||||
*
|
||||
* @param variables 普通变量
|
||||
* @param fileParts 文件 Part
|
||||
* @param record 上传记录
|
||||
* @return 已注入真实存储路径的变量
|
||||
*/
|
||||
private Map<String, Object> storeFiles(
|
||||
Map<String, Object> variables,
|
||||
Map<String, List<MultipartFile>> fileParts,
|
||||
WorkflowApiUploadRecord record) {
|
||||
Map<String, Object> resolved = new LinkedHashMap<>(variables);
|
||||
for (Map.Entry<String, List<MultipartFile>> entry :
|
||||
fileParts.entrySet()) {
|
||||
List<Map<String, Object>> descriptors =
|
||||
new ArrayList<>(entry.getValue().size());
|
||||
for (MultipartFile file : entry.getValue()) {
|
||||
FileStorageWriteHandle writeHandle;
|
||||
try {
|
||||
writeHandle = fileStorageService.prepareRecoverableWrite(
|
||||
STORAGE_PATH_PREFIX
|
||||
+ record.getRequestId(),
|
||||
buildStorageFilename(
|
||||
file,
|
||||
record.getStoredFiles().size()));
|
||||
} catch (RuntimeException error) {
|
||||
throw translateStorageFailure(error);
|
||||
}
|
||||
String locator = writeHandle.encodeLocator();
|
||||
int storedFileIndex = record.getStoredFiles().size();
|
||||
record.getStoredFiles().add(
|
||||
new WorkflowApiStoredFile(null, locator));
|
||||
// 先持久化精确 locator，物理写入中途退出后仍可由清理任务定位。
|
||||
uploadStore.save(record);
|
||||
|
||||
FileStorageWriteResult writeResult;
|
||||
try {
|
||||
writeResult = fileStorageService.saveRecoverable(
|
||||
file,
|
||||
writeHandle);
|
||||
} catch (RuntimeException error) {
|
||||
throw translateStorageFailure(error);
|
||||
}
|
||||
if (!locator.equals(writeResult.getLocator())) {
|
||||
throw new IllegalStateException(
|
||||
"文件存储返回了不一致的恢复定位符");
|
||||
}
|
||||
if (!StringUtils.hasText(writeResult.getUrl())) {
|
||||
throw new IllegalStateException(
|
||||
"文件存储未返回有效路径: "
|
||||
+ file.getOriginalFilename());
|
||||
}
|
||||
record.getStoredFiles().set(
|
||||
storedFileIndex,
|
||||
record.getStoredFiles()
|
||||
.get(storedFileIndex)
|
||||
.withFilePath(writeResult.getUrl()));
|
||||
uploadStore.save(record);
|
||||
descriptors.add(fileDescriptor(
|
||||
file,
|
||||
writeResult.getUrl()));
|
||||
}
|
||||
resolved.put(entry.getKey(), descriptors);
|
||||
}
|
||||
return resolved;
|
||||
}
|
||||
|
||||
/**
|
||||
* 构建不包含用户目录片段的稳定存储文件名。
|
||||
*
|
||||
* @param file 上传文件
|
||||
* @param index 当前请求内文件序号
|
||||
* @return 安全存储文件名
|
||||
*/
|
||||
private String buildStorageFilename(
|
||||
MultipartFile file,
|
||||
int index) {
|
||||
String original = file.getOriginalFilename();
|
||||
String extension = "";
|
||||
int separator = original == null
|
||||
? -1
|
||||
: original.lastIndexOf('.');
|
||||
if (separator >= 0 && separator < original.length() - 1) {
|
||||
String candidate = original.substring(separator + 1);
|
||||
if (candidate.length() <= 16
|
||||
&& candidate.matches("[A-Za-z0-9]+")) {
|
||||
extension = "." + candidate.toLowerCase(Locale.ROOT);
|
||||
}
|
||||
}
|
||||
return String.format(
|
||||
Locale.ROOT,
|
||||
"%03d-%s%s",
|
||||
index,
|
||||
UUID.randomUUID().toString().replace("-", ""),
|
||||
extension);
|
||||
}
|
||||
|
||||
/**
|
||||
* 校验单个 multipart 文件的基础元数据。
|
||||
*
|
||||
* @param file 文件
|
||||
* @param parameterName 工作流文件参数名
|
||||
*/
|
||||
private void validateMultipartFile(
|
||||
MultipartFile file,
|
||||
String parameterName) {
|
||||
if (file == null || file.isEmpty()) {
|
||||
throw new BusinessException(
|
||||
400,
|
||||
40016,
|
||||
"文件参数 " + parameterName + " 包含空文件");
|
||||
}
|
||||
if (!StringUtils.hasText(file.getOriginalFilename())) {
|
||||
throw new BusinessException(
|
||||
400,
|
||||
40016,
|
||||
"文件参数 " + parameterName + " 缺少文件名");
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 归一化全部文件 Part 的文件名和内容类型。
|
||||
*
|
||||
* @param fileParts 原始文件 Part
|
||||
* @return 保持参数和文件顺序的归一化视图
|
||||
*/
|
||||
private Map<String, List<MultipartFile>> normalizeFileParts(
|
||||
Map<String, List<MultipartFile>> fileParts) {
|
||||
Map<String, List<MultipartFile>> normalized =
|
||||
new LinkedHashMap<>();
|
||||
for (Map.Entry<String, List<MultipartFile>> entry
|
||||
: fileParts.entrySet()) {
|
||||
List<MultipartFile> source = entry.getValue();
|
||||
if (source == null) {
|
||||
normalized.put(entry.getKey(), List.of());
|
||||
continue;
|
||||
}
|
||||
normalized.put(
|
||||
entry.getKey(),
|
||||
source.stream()
|
||||
.map(fileNormalizer::normalize)
|
||||
.toList());
|
||||
}
|
||||
return normalized;
|
||||
}
|
||||
|
||||
/**
|
||||
* 将对象存储异常转换为不泄露底层配置的公共错误。
|
||||
*
|
||||
* @param error 原始存储异常
|
||||
* @return 安全业务异常
|
||||
*/
|
||||
private BusinessException translateStorageFailure(
|
||||
RuntimeException error) {
|
||||
if (isTransientStorageFailure(error)) {
|
||||
return new BusinessException(
|
||||
503,
|
||||
50301,
|
||||
"文件存储暂时不可用，请稍后重试",
|
||||
error);
|
||||
}
|
||||
return new BusinessException(
|
||||
500,
|
||||
50001,
|
||||
"文件存储处理失败，请联系管理员并提供 requestId",
|
||||
error);
|
||||
}
|
||||
|
||||
/**
|
||||
* 保守识别可直接重试的网络和超时故障。
|
||||
*
|
||||
* @param error 原始异常
|
||||
* @return 是否为暂时性依赖故障
|
||||
*/
|
||||
private boolean isTransientStorageFailure(Throwable error) {
|
||||
Throwable current = error;
|
||||
while (current != null) {
|
||||
if (current instanceof ConnectException
|
||||
|| current instanceof SocketTimeoutException
|
||||
|| current instanceof UnknownHostException
|
||||
|| current instanceof HttpTimeoutException
|
||||
|| current instanceof TimeoutException) {
|
||||
return true;
|
||||
}
|
||||
String className = current.getClass().getSimpleName();
|
||||
if ("InsufficientDataException".equals(className)
|
||||
|| "ServerException".equals(className)) {
|
||||
return true;
|
||||
}
|
||||
String message = current.getMessage();
|
||||
if (message != null) {
|
||||
String normalized = message.toLowerCase(Locale.ROOT);
|
||||
if (normalized.contains("timeout")
|
||||
|| normalized.contains("timed out")
|
||||
|| normalized.contains("connection refused")
|
||||
|| normalized.contains("temporarily unavailable")
|
||||
|| normalized.contains("service unavailable")) {
|
||||
return true;
|
||||
}
|
||||
}
|
||||
current = current.getCause();
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
/**
|
||||
* 构建工作流运行态文件描述。
|
||||
*
|
||||
* @param file multipart 文件
|
||||
* @param filePath 存储路径或校验占位路径
|
||||
* @return 文件描述
|
||||
*/
|
||||
private Map<String, Object> fileDescriptor(
|
||||
MultipartFile file,
|
||||
String filePath) {
|
||||
Map<String, Object> descriptor = new LinkedHashMap<>();
|
||||
descriptor.put("fileName", file.getOriginalFilename());
|
||||
descriptor.put("filePath", filePath);
|
||||
if (StringUtils.hasText(file.getContentType())) {
|
||||
descriptor.put("contentType", file.getContentType());
|
||||
}
|
||||
descriptor.put("size", file.getSize());
|
||||
return descriptor;
|
||||
}
|
||||
|
||||
/**
|
||||
* 判断 metadata 中是否已经提供有效值。
|
||||
*
|
||||
* @param value metadata 变量值
|
||||
* @return 是否存在有效值
|
||||
*/
|
||||
private boolean hasValue(Object value) {
|
||||
if (value == null) {
|
||||
return false;
|
||||
}
|
||||
if (value instanceof String text) {
|
||||
return StringUtils.hasText(text);
|
||||
}
|
||||
if (value instanceof Collection<?> collection) {
|
||||
return !collection.isEmpty();
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
/**
|
||||
* 使用请求线程持有的最新恢复定位符清理准备阶段失败的文件。
|
||||
*
|
||||
* <p>当 Redis 更新恰好失败时，重新读取的记录可能缺少最后一次状态，
|
||||
* 因此必须优先使用内存中的记录执行补偿。</p>
|
||||
*
|
||||
* @param record 请求线程持有的最新上传记录
|
||||
*/
|
||||
private void cleanupPreparationFailure(
|
||||
WorkflowApiUploadRecord record) {
|
||||
try {
|
||||
deleteFiles(record);
|
||||
uploadStore.remove(record);
|
||||
} catch (RuntimeException cleanupError) {
|
||||
try {
|
||||
uploadStore.save(record);
|
||||
} catch (RuntimeException persistenceError) {
|
||||
cleanupError.addSuppressed(persistenceError);
|
||||
}
|
||||
throw cleanupError;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 在分布式锁下清理一条上传记录。
|
||||
*
|
||||
* @param requestId 上传请求 ID
|
||||
* @param force 是否忽略工作流运行状态立即清理
|
||||
* @return 是否成功删除记录
|
||||
*/
|
||||
private boolean cleanupRequest(String requestId, boolean force) {
|
||||
RedisLockExecutor.LockHandle handle =
|
||||
redisLockExecutor.tryAcquire(
|
||||
CLEANUP_LOCK_PREFIX + requestId,
|
||||
CLEANUP_LOCK_WAIT,
|
||||
CLEANUP_LOCK_LEASE);
|
||||
if (handle == null) {
|
||||
return false;
|
||||
}
|
||||
try (handle) {
|
||||
WorkflowApiUploadRecord record =
|
||||
uploadStore.find(requestId).orElse(null);
|
||||
if (record == null) {
|
||||
uploadStore.removeMissingIndex(requestId);
|
||||
return false;
|
||||
}
|
||||
if (!force && shouldRetain(record)) {
|
||||
uploadStore.schedule(
|
||||
record,
|
||||
System.currentTimeMillis()
|
||||
+ ACTIVE_RECHECK.toMillis());
|
||||
return false;
|
||||
}
|
||||
deleteFiles(record);
|
||||
uploadStore.remove(record);
|
||||
return true;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 判断上传记录是否仍被运行中或挂起的工作流使用。
|
||||
*
|
||||
* @param record 上传记录
|
||||
* @return 是否需要继续保留
|
||||
*/
|
||||
private boolean shouldRetain(WorkflowApiUploadRecord record) {
|
||||
if (!StringUtils.hasText(record.getExecuteId())) {
|
||||
return false;
|
||||
}
|
||||
ChainState state = chainStateRepository.load(record.getExecuteId());
|
||||
return state != null
|
||||
&& state.getStatus() != null
|
||||
&& !state.getStatus().isTerminal();
|
||||
}
|
||||
|
||||
/**
|
||||
* 幂等删除记录中的全部临时文件。
|
||||
*
|
||||
* @param record 上传记录
|
||||
*/
|
||||
private void deleteFiles(WorkflowApiUploadRecord record) {
|
||||
RuntimeException firstFailure = null;
|
||||
for (WorkflowApiStoredFile storedFile :
|
||||
record.getStoredFiles()) {
|
||||
try {
|
||||
fileStorageService.deleteRecoverable(
|
||||
FileStorageWriteHandle.decodeLocator(
|
||||
storedFile.storageLocator()));
|
||||
} catch (RuntimeException error) {
|
||||
firstFailure = appendFailure(firstFailure, error);
|
||||
}
|
||||
}
|
||||
// 兼容开发阶段已经写入 Redis 的旧版 URL 记录。
|
||||
for (String filePath : record.getFilePaths()) {
|
||||
try {
|
||||
fileStorageService.delete(filePath);
|
||||
} catch (RuntimeException error) {
|
||||
firstFailure = appendFailure(firstFailure, error);
|
||||
}
|
||||
}
|
||||
if (firstFailure != null) {
|
||||
throw firstFailure;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 聚合文件清理异常并保留全部失败原因。
|
||||
*
|
||||
* @param firstFailure 首个异常
|
||||
* @param currentFailure 当前异常
|
||||
* @return 聚合后的首个异常
|
||||
*/
|
||||
private RuntimeException appendFailure(
|
||||
RuntimeException firstFailure,
|
||||
RuntimeException currentFailure) {
|
||||
if (firstFailure == null) {
|
||||
return currentFailure;
|
||||
}
|
||||
firstFailure.addSuppressed(currentFailure);
|
||||
return firstFailure;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,133 @@
|
||||
package tech.easyflow.ai.easyagentsflow.upload;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* Public Workflow API 临时上传记录。
|
||||
*/
|
||||
public class WorkflowApiUploadRecord {
|
||||
|
||||
private String requestId;
|
||||
private String executeId;
|
||||
private List<WorkflowApiStoredFile> storedFiles = new ArrayList<>();
|
||||
/**
|
||||
* 兼容早期临时上传记录的旧版路径；新记录使用 {@link #storedFiles}。
|
||||
*/
|
||||
private List<String> filePaths = new ArrayList<>();
|
||||
private long createdAt;
|
||||
private long cleanupAt;
|
||||
|
||||
/**
|
||||
* 获取上传请求 ID。
|
||||
*
|
||||
* @return 上传请求 ID
|
||||
*/
|
||||
public String getRequestId() {
|
||||
return requestId;
|
||||
}
|
||||
|
||||
/**
|
||||
* 设置上传请求 ID。
|
||||
*
|
||||
* @param requestId 上传请求 ID
|
||||
*/
|
||||
public void setRequestId(String requestId) {
|
||||
this.requestId = requestId;
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取工作流执行 ID。
|
||||
*
|
||||
* @return 工作流执行 ID
|
||||
*/
|
||||
public String getExecuteId() {
|
||||
return executeId;
|
||||
}
|
||||
|
||||
/**
|
||||
* 设置工作流执行 ID。
|
||||
*
|
||||
* @param executeId 工作流执行 ID
|
||||
*/
|
||||
public void setExecuteId(String executeId) {
|
||||
this.executeId = executeId;
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取带恢复定位符的临时文件。
|
||||
*
|
||||
* @return 临时文件记录
|
||||
*/
|
||||
public List<WorkflowApiStoredFile> getStoredFiles() {
|
||||
return storedFiles;
|
||||
}
|
||||
|
||||
/**
|
||||
* 设置带恢复定位符的临时文件。
|
||||
*
|
||||
* @param storedFiles 临时文件记录
|
||||
*/
|
||||
public void setStoredFiles(
|
||||
List<WorkflowApiStoredFile> storedFiles) {
|
||||
this.storedFiles = storedFiles == null
|
||||
? new ArrayList<>()
|
||||
: new ArrayList<>(storedFiles);
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取旧版临时文件路径。
|
||||
*
|
||||
* @return 临时文件路径
|
||||
*/
|
||||
public List<String> getFilePaths() {
|
||||
return filePaths;
|
||||
}
|
||||
|
||||
/**
|
||||
* 设置旧版临时文件路径。
|
||||
*
|
||||
* @param filePaths 临时文件路径
|
||||
*/
|
||||
public void setFilePaths(List<String> filePaths) {
|
||||
this.filePaths = filePaths == null
|
||||
? new ArrayList<>()
|
||||
: new ArrayList<>(filePaths);
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取创建时间。
|
||||
*
|
||||
* @return Unix 毫秒时间戳
|
||||
*/
|
||||
public long getCreatedAt() {
|
||||
return createdAt;
|
||||
}
|
||||
|
||||
/**
|
||||
* 设置创建时间。
|
||||
*
|
||||
* @param createdAt Unix 毫秒时间戳
|
||||
*/
|
||||
public void setCreatedAt(long createdAt) {
|
||||
this.createdAt = createdAt;
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取下次清理检查时间。
|
||||
*
|
||||
* @return Unix 毫秒时间戳
|
||||
*/
|
||||
public long getCleanupAt() {
|
||||
return cleanupAt;
|
||||
}
|
||||
|
||||
/**
|
||||
* 设置下次清理检查时间。
|
||||
*
|
||||
* @param cleanupAt Unix 毫秒时间戳
|
||||
*/
|
||||
public void setCleanupAt(long cleanupAt) {
|
||||
this.cleanupAt = cleanupAt;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,345 @@
|
||||
package tech.easyflow.ai.easyagentsflow.upload;
|
||||
|
||||
import com.fasterxml.jackson.core.JsonProcessingException;
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import org.springframework.data.redis.core.StringRedisTemplate;
|
||||
import org.springframework.data.redis.core.script.DefaultRedisScript;
|
||||
import org.springframework.stereotype.Component;
|
||||
import org.springframework.util.StringUtils;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.util.Arrays;
|
||||
import java.util.List;
|
||||
import java.util.Optional;
|
||||
|
||||
/**
|
||||
* Public Workflow API 临时上传记录的 Redis 存储。
|
||||
*/
|
||||
@Component
|
||||
public class WorkflowApiUploadStore {
|
||||
|
||||
private static final String RECORD_KEY_PREFIX =
|
||||
"easyflow:workflow:{api-upload}:record:";
|
||||
private static final String EXECUTION_KEY_PREFIX =
|
||||
"easyflow:workflow:{api-upload}:execution:";
|
||||
private static final String CLEANUP_INDEX =
|
||||
"easyflow:workflow:{api-upload}:cleanup";
|
||||
private static final Duration EXECUTION_INDEX_TTL =
|
||||
Duration.ofDays(30);
|
||||
private static final DefaultRedisScript<Long> SAVE_SCHEDULED_SCRIPT =
|
||||
longScript(
|
||||
"redis.call('set', KEYS[1], ARGV[1]); "
|
||||
+ "redis.call('zadd', KEYS[2], ARGV[2], ARGV[3]); "
|
||||
+ "if ARGV[4] == '1' "
|
||||
+ "and redis.call('get', KEYS[3]) == ARGV[3] then "
|
||||
+ "redis.call('pexpire', KEYS[3], ARGV[5]); end; "
|
||||
+ "return 1");
|
||||
private static final DefaultRedisScript<Long> BIND_EXECUTION_SCRIPT =
|
||||
longScript(
|
||||
"redis.call('set', KEYS[1], ARGV[1]); "
|
||||
+ "redis.call('psetex', KEYS[2], ARGV[3], ARGV[2]); "
|
||||
+ "if ARGV[4] == '1' and KEYS[3] ~= KEYS[2] "
|
||||
+ "and redis.call('get', KEYS[3]) == ARGV[2] then "
|
||||
+ "redis.call('del', KEYS[3]); end; "
|
||||
+ "return 1");
|
||||
private static final DefaultRedisScript<Long>
|
||||
REMOVE_EXECUTION_INDEX_SCRIPT =
|
||||
longScript(
|
||||
"if redis.call('get', KEYS[1]) == ARGV[1] then "
|
||||
+ "return redis.call('del', KEYS[1]); end; "
|
||||
+ "return 0");
|
||||
private static final DefaultRedisScript<String> CLAIM_EXPIRED_SCRIPT =
|
||||
stringScript(
|
||||
"local values = redis.call("
|
||||
+ "'zrangebyscore', KEYS[1], '-inf', ARGV[1], "
|
||||
+ "'LIMIT', 0, 1); "
|
||||
+ "if #values == 0 then return nil; end; "
|
||||
+ "redis.call('zadd', KEYS[1], ARGV[2], values[1]); "
|
||||
+ "return values[1]");
|
||||
private static final DefaultRedisScript<Long> REMOVE_SCRIPT =
|
||||
longScript(
|
||||
"redis.call('del', KEYS[1]); "
|
||||
+ "if ARGV[2] == '1' "
|
||||
+ "and redis.call('get', KEYS[2]) == ARGV[1] then "
|
||||
+ "redis.call('del', KEYS[2]); end; "
|
||||
+ "redis.call('zrem', KEYS[3], ARGV[1]); "
|
||||
+ "return 1");
|
||||
|
||||
private final StringRedisTemplate redisTemplate;
|
||||
private final ObjectMapper objectMapper;
|
||||
|
||||
/**
|
||||
* 创建临时上传记录存储。
|
||||
*
|
||||
* @param redisTemplate Redis 模板
|
||||
* @param objectMapper JSON 映射器
|
||||
*/
|
||||
public WorkflowApiUploadStore(StringRedisTemplate redisTemplate,
|
||||
ObjectMapper objectMapper) {
|
||||
this.redisTemplate = redisTemplate;
|
||||
this.objectMapper = objectMapper;
|
||||
}
|
||||
|
||||
/**
|
||||
* 新建临时上传记录并登记清理时间。
|
||||
*
|
||||
* @param record 上传记录
|
||||
*/
|
||||
public void create(WorkflowApiUploadRecord record) {
|
||||
requireRecord(record);
|
||||
saveScheduled(record);
|
||||
}
|
||||
|
||||
/**
|
||||
* 保存上传记录的最新内容。
|
||||
*
|
||||
* @param record 上传记录
|
||||
*/
|
||||
public void save(WorkflowApiUploadRecord record) {
|
||||
requireRecord(record);
|
||||
redisTemplate.opsForValue().set(
|
||||
recordKey(record.getRequestId()),
|
||||
serialize(record));
|
||||
}
|
||||
|
||||
/**
|
||||
* 将上传请求绑定到工作流执行实例。
|
||||
*
|
||||
* @param requestId 上传请求 ID
|
||||
* @param executeId 工作流执行 ID
|
||||
*/
|
||||
public void bindExecution(String requestId, String executeId) {
|
||||
if (!StringUtils.hasText(executeId)) {
|
||||
throw new IllegalArgumentException("工作流执行 ID 不能为空");
|
||||
}
|
||||
WorkflowApiUploadRecord record = find(requestId)
|
||||
.orElseThrow(() -> new IllegalStateException(
|
||||
"工作流临时上传记录不存在: " + requestId));
|
||||
String previousExecuteId = record.getExecuteId();
|
||||
record.setExecuteId(executeId);
|
||||
redisTemplate.execute(
|
||||
BIND_EXECUTION_SCRIPT,
|
||||
Arrays.asList(
|
||||
recordKey(requestId),
|
||||
executionKey(executeId),
|
||||
StringUtils.hasText(previousExecuteId)
|
||||
? executionKey(previousExecuteId)
|
||||
: recordKey(requestId)),
|
||||
serialize(record),
|
||||
requestId,
|
||||
String.valueOf(EXECUTION_INDEX_TTL.toMillis()),
|
||||
StringUtils.hasText(previousExecuteId) ? "1" : "0");
|
||||
}
|
||||
|
||||
/**
|
||||
* 按上传请求 ID 查找记录。
|
||||
*
|
||||
* @param requestId 上传请求 ID
|
||||
* @return 上传记录
|
||||
*/
|
||||
public Optional<WorkflowApiUploadRecord> find(String requestId) {
|
||||
if (!StringUtils.hasText(requestId)) {
|
||||
return Optional.empty();
|
||||
}
|
||||
String value = redisTemplate.opsForValue().get(recordKey(requestId));
|
||||
if (!StringUtils.hasText(value)) {
|
||||
return Optional.empty();
|
||||
}
|
||||
try {
|
||||
return Optional.of(objectMapper.readValue(
|
||||
value,
|
||||
WorkflowApiUploadRecord.class));
|
||||
} catch (JsonProcessingException error) {
|
||||
throw new IllegalStateException(
|
||||
"读取工作流临时上传记录失败: " + requestId,
|
||||
error);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 按工作流执行 ID 查找上传记录。
|
||||
*
|
||||
* @param executeId 工作流执行 ID
|
||||
* @return 上传记录
|
||||
*/
|
||||
public Optional<WorkflowApiUploadRecord> findByExecutionId(
|
||||
String executeId) {
|
||||
if (!StringUtils.hasText(executeId)) {
|
||||
return Optional.empty();
|
||||
}
|
||||
String requestId = redisTemplate.opsForValue().get(
|
||||
executionKey(executeId));
|
||||
Optional<WorkflowApiUploadRecord> record = find(requestId);
|
||||
if (StringUtils.hasText(requestId)
|
||||
&& (record.isEmpty()
|
||||
|| !executeId.equals(record.get().getExecuteId()))) {
|
||||
redisTemplate.execute(
|
||||
REMOVE_EXECUTION_INDEX_SCRIPT,
|
||||
List.of(executionKey(executeId)),
|
||||
requestId);
|
||||
return Optional.empty();
|
||||
}
|
||||
return record;
|
||||
}
|
||||
|
||||
/**
|
||||
* 更新记录的下次清理检查时间。
|
||||
*
|
||||
* @param record 上传记录
|
||||
* @param cleanupAt Unix 毫秒时间戳
|
||||
*/
|
||||
public void schedule(WorkflowApiUploadRecord record, long cleanupAt) {
|
||||
requireRecord(record);
|
||||
record.setCleanupAt(cleanupAt);
|
||||
saveScheduled(record);
|
||||
}
|
||||
|
||||
/**
|
||||
* 原子领取一条到期上传请求，并提前设置失败重试时间。
|
||||
*
|
||||
* @param now 当前 Unix 毫秒时间戳
|
||||
* @param retryAt 领取后默认重试时间
|
||||
* @return 领取到的上传请求 ID
|
||||
*/
|
||||
public Optional<String> claimExpired(
|
||||
long now,
|
||||
long retryAt) {
|
||||
String requestId = redisTemplate.execute(
|
||||
CLAIM_EXPIRED_SCRIPT,
|
||||
List.of(CLEANUP_INDEX),
|
||||
String.valueOf(now),
|
||||
String.valueOf(retryAt));
|
||||
return Optional.ofNullable(requestId);
|
||||
}
|
||||
|
||||
/**
|
||||
* 删除上传记录、执行索引和清理索引。
|
||||
*
|
||||
* @param record 上传记录
|
||||
*/
|
||||
public void remove(WorkflowApiUploadRecord record) {
|
||||
requireRecord(record);
|
||||
boolean hasExecuteId =
|
||||
StringUtils.hasText(record.getExecuteId());
|
||||
redisTemplate.execute(
|
||||
REMOVE_SCRIPT,
|
||||
Arrays.asList(
|
||||
recordKey(record.getRequestId()),
|
||||
hasExecuteId
|
||||
? executionKey(record.getExecuteId())
|
||||
: recordKey(record.getRequestId()),
|
||||
CLEANUP_INDEX),
|
||||
record.getRequestId(),
|
||||
hasExecuteId ? "1" : "0");
|
||||
}
|
||||
|
||||
/**
|
||||
* 删除已经缺少详情记录的残留清理索引。
|
||||
*
|
||||
* @param requestId 上传请求 ID
|
||||
*/
|
||||
public void removeMissingIndex(String requestId) {
|
||||
if (StringUtils.hasText(requestId)) {
|
||||
redisTemplate.opsForZSet().remove(CLEANUP_INDEX, requestId);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 序列化上传记录。
|
||||
*
|
||||
* @param record 上传记录
|
||||
* @return JSON 文本
|
||||
*/
|
||||
private String serialize(WorkflowApiUploadRecord record) {
|
||||
try {
|
||||
return objectMapper.writeValueAsString(record);
|
||||
} catch (JsonProcessingException error) {
|
||||
throw new IllegalStateException(
|
||||
"写入工作流临时上传记录失败: "
|
||||
+ record.getRequestId(),
|
||||
error);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 原子保存记录与清理索引，并续期执行索引。
|
||||
*
|
||||
* @param record 上传记录
|
||||
*/
|
||||
private void saveScheduled(WorkflowApiUploadRecord record) {
|
||||
boolean hasExecuteId =
|
||||
StringUtils.hasText(record.getExecuteId());
|
||||
redisTemplate.execute(
|
||||
SAVE_SCHEDULED_SCRIPT,
|
||||
Arrays.asList(
|
||||
recordKey(record.getRequestId()),
|
||||
CLEANUP_INDEX,
|
||||
hasExecuteId
|
||||
? executionKey(record.getExecuteId())
|
||||
: recordKey(record.getRequestId())),
|
||||
serialize(record),
|
||||
String.valueOf(record.getCleanupAt()),
|
||||
record.getRequestId(),
|
||||
hasExecuteId ? "1" : "0",
|
||||
String.valueOf(EXECUTION_INDEX_TTL.toMillis()));
|
||||
}
|
||||
|
||||
/**
|
||||
* 创建返回 Long 的 Redis Lua 脚本。
|
||||
*
|
||||
* @param text Lua 文本
|
||||
* @return Redis 脚本
|
||||
*/
|
||||
private static DefaultRedisScript<Long> longScript(
|
||||
String text) {
|
||||
DefaultRedisScript<Long> script = new DefaultRedisScript<>();
|
||||
script.setScriptText(text);
|
||||
script.setResultType(Long.class);
|
||||
return script;
|
||||
}
|
||||
|
||||
/**
|
||||
* 创建返回字符串的 Redis Lua 脚本。
|
||||
*
|
||||
* @param text Lua 文本
|
||||
* @return Redis 脚本
|
||||
*/
|
||||
private static DefaultRedisScript<String> stringScript(
|
||||
String text) {
|
||||
DefaultRedisScript<String> script = new DefaultRedisScript<>();
|
||||
script.setScriptText(text);
|
||||
script.setResultType(String.class);
|
||||
return script;
|
||||
}
|
||||
|
||||
/**
|
||||
* 校验记录主键。
|
||||
*
|
||||
* @param record 上传记录
|
||||
*/
|
||||
private void requireRecord(WorkflowApiUploadRecord record) {
|
||||
if (record == null || !StringUtils.hasText(record.getRequestId())) {
|
||||
throw new IllegalArgumentException("工作流临时上传请求 ID 不能为空");
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 构建记录 Redis Key。
|
||||
*
|
||||
* @param requestId 上传请求 ID
|
||||
* @return Redis Key
|
||||
*/
|
||||
private String recordKey(String requestId) {
|
||||
return RECORD_KEY_PREFIX + requestId;
|
||||
}
|
||||
|
||||
/**
|
||||
* 构建执行实例 Redis Key。
|
||||
*
|
||||
* @param executeId 工作流执行 ID
|
||||
* @return Redis Key
|
||||
*/
|
||||
private String executionKey(String executeId) {
|
||||
return EXECUTION_KEY_PREFIX + executeId;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,129 @@
|
||||
package tech.easyflow.ai.easyagentsflow.upload;
|
||||
|
||||
import org.springframework.beans.factory.annotation.Qualifier;
|
||||
import org.springframework.stereotype.Component;
|
||||
import org.springframework.util.StringUtils;
|
||||
import tech.easyflow.common.filestorage.FileStorageService;
|
||||
import tech.easyflow.common.filestorage.FileStorageWriteHandle;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.io.InputStream;
|
||||
import java.net.URI;
|
||||
import java.util.Optional;
|
||||
import java.util.regex.Matcher;
|
||||
import java.util.regex.Pattern;
|
||||
|
||||
/**
|
||||
* 读取经过 Public Workflow API 上传记录验证的临时文件。
|
||||
*
|
||||
* <p>外部文件描述只提供公开读取路径。仅当路径中的随机请求 ID、Redis 上传记录、
|
||||
* 完整文件 URL 和可恢复存储句柄全部匹配时，才允许绕过公网 URL 限制并直接读取物理对象。</p>
|
||||
*/
|
||||
@Component
|
||||
public class WorkflowApiUploadedFileReader {
|
||||
|
||||
private static final String STORAGE_PATH_PREFIX = "workflow-api-upload/";
|
||||
private static final Pattern MANAGED_PATH_PATTERN = Pattern.compile(
|
||||
"(?:^|/)workflow-api-upload/([0-9a-f]{32})/([^/]+)$");
|
||||
|
||||
private final WorkflowApiUploadStore uploadStore;
|
||||
private final FileStorageService fileStorageService;
|
||||
|
||||
/**
|
||||
* 创建工作流 API 上传文件读取器。
|
||||
*
|
||||
* @param uploadStore 临时上传记录存储
|
||||
* @param fileStorageService 默认文件存储路由
|
||||
*/
|
||||
public WorkflowApiUploadedFileReader(
|
||||
WorkflowApiUploadStore uploadStore,
|
||||
@Qualifier("default") FileStorageService fileStorageService) {
|
||||
this.uploadStore = uploadStore;
|
||||
this.fileStorageService = fileStorageService;
|
||||
}
|
||||
|
||||
/**
|
||||
* 在路径属于受管工作流上传文件时校验记录并打开物理对象。
|
||||
*
|
||||
* @param filePath 工作流文件描述中的完整读取路径
|
||||
* @return 受管文件流；普通文件路径或普通远端 URL 返回空
|
||||
* @throws IOException 上传记录已失效、引用不匹配或物理对象无法读取时抛出
|
||||
*/
|
||||
public Optional<InputStream> openVerified(String filePath) throws IOException {
|
||||
ManagedPath managedPath = parseManagedPath(filePath).orElse(null);
|
||||
if (managedPath == null) {
|
||||
return Optional.empty();
|
||||
}
|
||||
WorkflowApiUploadRecord record = uploadStore.find(managedPath.requestId())
|
||||
.orElseThrow(() -> new IOException("工作流上传文件已失效，请重新上传"));
|
||||
WorkflowApiStoredFile storedFile = record.getStoredFiles().stream()
|
||||
.filter(file -> file != null && filePath.equals(file.filePath()))
|
||||
.findFirst()
|
||||
.orElseThrow(() -> new IOException("工作流上传文件引用与上传记录不匹配"));
|
||||
|
||||
final FileStorageWriteHandle handle;
|
||||
try {
|
||||
handle = FileStorageWriteHandle.decodeLocator(storedFile.storageLocator());
|
||||
} catch (IllegalArgumentException exception) {
|
||||
throw new IOException("工作流上传文件存储定位符无效", exception);
|
||||
}
|
||||
String expectedStoragePath = STORAGE_PATH_PREFIX + managedPath.requestId() + "/";
|
||||
if (!expectedStoragePath.equals(handle.getPath())
|
||||
|| !managedPath.filename().equals(handle.getFilename())) {
|
||||
throw new IOException("工作流上传文件存储定位与上传请求不匹配");
|
||||
}
|
||||
try {
|
||||
return Optional.of(fileStorageService.readRecoverable(handle));
|
||||
} catch (RuntimeException exception) {
|
||||
throw new IOException("读取工作流上传文件失败", exception);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 判断路径结构是否属于系统生成的工作流 API 上传目录。
|
||||
*
|
||||
* <p>该判断只用于选择 I/O 隔离通道，不能替代 {@link #openVerified(String)} 的授权校验。</p>
|
||||
*
|
||||
* @param filePath 文件读取路径
|
||||
* @return 路径结构匹配时返回 true
|
||||
*/
|
||||
public boolean isManagedPathCandidate(String filePath) {
|
||||
return parseManagedPath(filePath).isPresent();
|
||||
}
|
||||
|
||||
/**
|
||||
* 从 URL 或相对路径中解析受管请求 ID 与固定文件名。
|
||||
*
|
||||
* @param filePath 原始文件路径
|
||||
* @return 受管路径信息
|
||||
*/
|
||||
private Optional<ManagedPath> parseManagedPath(String filePath) {
|
||||
if (!StringUtils.hasText(filePath)) {
|
||||
return Optional.empty();
|
||||
}
|
||||
final String path;
|
||||
try {
|
||||
URI uri = URI.create(filePath);
|
||||
path = uri.getPath();
|
||||
} catch (IllegalArgumentException exception) {
|
||||
return Optional.empty();
|
||||
}
|
||||
if (!StringUtils.hasText(path)) {
|
||||
return Optional.empty();
|
||||
}
|
||||
Matcher matcher = MANAGED_PATH_PATTERN.matcher(path);
|
||||
if (!matcher.find()) {
|
||||
return Optional.empty();
|
||||
}
|
||||
return Optional.of(new ManagedPath(matcher.group(1), matcher.group(2)));
|
||||
}
|
||||
|
||||
/**
|
||||
* 系统受管上传路径中的可信定位片段。
|
||||
*
|
||||
* @param requestId 随机上传请求 ID
|
||||
* @param filename 系统生成的存储文件名
|
||||
*/
|
||||
private record ManagedPath(String requestId, String filename) {
|
||||
}
|
||||
}
|
||||
@@ -1,6 +1,7 @@
|
||||
package tech.easyflow.ai.node;
|
||||
|
||||
import com.easyagents.flow.core.util.IoBulkhead;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.beans.factory.annotation.Qualifier;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
@@ -12,6 +13,7 @@ import tech.easyflow.ai.document.exception.DocumentParseBridgeException;
|
||||
import tech.easyflow.ai.document.service.DocumentParseBridgeService;
|
||||
import tech.easyflow.ai.document.support.DocumentInputStreamSupport;
|
||||
import tech.easyflow.ai.document.support.DocumentParseSourceType;
|
||||
import tech.easyflow.ai.easyagentsflow.upload.WorkflowApiUploadedFileReader;
|
||||
import tech.easyflow.common.filestorage.FileStorageService;
|
||||
import tech.easyflow.common.util.StringUtil;
|
||||
import tech.easyflow.common.web.exceptions.BusinessException;
|
||||
@@ -27,6 +29,7 @@ import java.util.LinkedHashMap;
|
||||
import java.util.LinkedHashSet;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Optional;
|
||||
import java.util.Set;
|
||||
|
||||
/**
|
||||
@@ -49,6 +52,7 @@ public class DocNodeFileContentExtractor {
|
||||
private final DocumentParseBridgeService documentParseBridgeService;
|
||||
private final FileStorageService fileStorageService;
|
||||
private final ReaderManager readerManager;
|
||||
private final WorkflowApiUploadedFileReader uploadedFileReader;
|
||||
|
||||
/**
|
||||
* 创建文件内容提取器。
|
||||
@@ -56,13 +60,31 @@ public class DocNodeFileContentExtractor {
|
||||
* @param documentParseBridgeService 统一文档解析桥接服务
|
||||
* @param fileStorageService 文件存储服务
|
||||
* @param readerManager 默认读取器管理器
|
||||
* @param uploadedFileReader 已验证的工作流 API 上传文件读取器
|
||||
*/
|
||||
@Autowired
|
||||
public DocNodeFileContentExtractor(DocumentParseBridgeService documentParseBridgeService,
|
||||
@Qualifier("default") FileStorageService fileStorageService,
|
||||
ReaderManager readerManager) {
|
||||
ReaderManager readerManager,
|
||||
WorkflowApiUploadedFileReader uploadedFileReader) {
|
||||
this.documentParseBridgeService = documentParseBridgeService;
|
||||
this.fileStorageService = fileStorageService;
|
||||
this.readerManager = readerManager;
|
||||
this.uploadedFileReader = uploadedFileReader;
|
||||
}
|
||||
|
||||
/**
|
||||
* 创建不启用工作流 API 上传识别的提取器，供同包隔离测试使用。
|
||||
*
|
||||
* @param documentParseBridgeService 统一文档解析桥接服务
|
||||
* @param fileStorageService 文件存储服务
|
||||
* @param readerManager 默认读取器管理器
|
||||
*/
|
||||
DocNodeFileContentExtractor(
|
||||
DocumentParseBridgeService documentParseBridgeService,
|
||||
FileStorageService fileStorageService,
|
||||
ReaderManager readerManager) {
|
||||
this(documentParseBridgeService, fileStorageService, readerManager, null);
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -305,7 +327,9 @@ public class DocNodeFileContentExtractor {
|
||||
DocumentSourceRef sourceRef, Path target) throws IOException {
|
||||
String filePath = sourceRef.getFilePath();
|
||||
boolean localStorage = StringUtil.hasText(filePath)
|
||||
&& !isRemoteUrl(filePath);
|
||||
&& (!isRemoteUrl(filePath)
|
||||
|| (uploadedFileReader != null
|
||||
&& uploadedFileReader.isManagedPathCandidate(filePath)));
|
||||
if (localStorage) {
|
||||
try (IoBulkhead.Permit ignored =
|
||||
IoBulkhead.storage().acquire("storage:document-read");
|
||||
@@ -340,6 +364,14 @@ public class DocNodeFileContentExtractor {
|
||||
|
||||
private InputStream openInputStream(DocumentSourceRef sourceRef) throws IOException {
|
||||
String filePath = sourceRef.getFilePath();
|
||||
if (uploadedFileReader != null && StringUtil.hasText(filePath)) {
|
||||
Optional<InputStream> managed = uploadedFileReader.openVerified(filePath);
|
||||
if (managed.isPresent()) {
|
||||
return DocumentInputStreamSupport.limit(
|
||||
managed.get(),
|
||||
FILE_MAX_SINGLE_SIZE);
|
||||
}
|
||||
}
|
||||
if (StringUtil.hasText(filePath) && isRemoteUrl(filePath)) {
|
||||
return DocumentInputStreamSupport.openRemote(filePath, FILE_MAX_SINGLE_SIZE);
|
||||
}
|
||||
|
||||
@@ -17,6 +17,7 @@ import tech.easyflow.common.web.exceptions.BusinessException;
|
||||
|
||||
import java.lang.reflect.Field;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
import static org.mockito.Mockito.mock;
|
||||
import static org.mockito.Mockito.times;
|
||||
@@ -82,6 +83,8 @@ public class TinyFlowServiceTest {
|
||||
.thenReturn(nodeStateRepository);
|
||||
when(chainStateRepository.load(EXECUTE_ID))
|
||||
.thenReturn(chainState);
|
||||
when(chainExecutor.getInstanceNodeNames(chainState))
|
||||
.thenReturn(Map.of(NODE_ID, "文档解析"));
|
||||
when(nodeStateRepository.load(EXECUTE_ID, NODE_ID))
|
||||
.thenReturn(null);
|
||||
TinyFlowService service = service(chainExecutor);
|
||||
@@ -96,7 +99,12 @@ public class TinyFlowServiceTest {
|
||||
Assert.assertEquals(
|
||||
Integer.valueOf(NodeStatus.READY.getValue()),
|
||||
result.getNodes().get(NODE_ID).getStatus());
|
||||
Assert.assertEquals(
|
||||
"文档解析",
|
||||
result.getNodes().get(NODE_ID).getNodeName());
|
||||
verify(chainStateRepository, times(1)).load(EXECUTE_ID);
|
||||
verify(chainExecutor, times(1))
|
||||
.getInstanceNodeNames(chainState);
|
||||
verify(nodeStateRepository, times(1))
|
||||
.load(EXECUTE_ID, NODE_ID);
|
||||
}
|
||||
|
||||
@@ -159,6 +159,47 @@ public class WorkflowRunningParameterResolverTest {
|
||||
Assert.assertEquals("file", fields.get(0).get("type"));
|
||||
}
|
||||
|
||||
/**
|
||||
* multipart 文件字段名应从解析后的开始节点参数中按定义顺序返回。
|
||||
*
|
||||
* @throws Exception 反射注入失败
|
||||
*/
|
||||
@Test
|
||||
public void testResolveFileParameterNamesShouldReturnStartFileFields()
|
||||
throws Exception {
|
||||
WorkflowRunningParameterResolver resolver = newResolver();
|
||||
|
||||
Assert.assertEquals(
|
||||
List.of("attachments"),
|
||||
List.copyOf(resolver.resolveFileParameterNames(
|
||||
workflowContentWithStartParameters())));
|
||||
}
|
||||
|
||||
/**
|
||||
* 必填文件字段名应从开始节点参数定义中单独解析。
|
||||
*
|
||||
* @throws Exception 反射注入失败
|
||||
*/
|
||||
@Test
|
||||
public void testResolveRequiredFileParameterNamesShouldKeepOrder()
|
||||
throws Exception {
|
||||
WorkflowRunningParameterResolver resolver = newResolver();
|
||||
JSONObject startData = data("开始");
|
||||
JSONArray parameters = startParameters();
|
||||
parameters.getJSONObject(1).put("required", true);
|
||||
startData.put("parameters", parameters);
|
||||
String content = workflowJson(
|
||||
array(
|
||||
node("s1", "startNode", null, startData),
|
||||
node("e1", "endNode", null, data("结束"))),
|
||||
array(edge("e1", "s1", "e1")));
|
||||
|
||||
Assert.assertEquals(
|
||||
List.of("attachments"),
|
||||
List.copyOf(resolver
|
||||
.resolveRequiredFileParameterNames(content)));
|
||||
}
|
||||
|
||||
/**
|
||||
* 文件参数运行值应统一归一化为数组并按 filePath 去重。
|
||||
*
|
||||
@@ -178,6 +219,89 @@ public class WorkflowRunningParameterResolverTest {
|
||||
Assert.assertTrue(((List<?>) attachments).get(0) instanceof Map<?, ?>);
|
||||
}
|
||||
|
||||
/**
|
||||
* 文件参数应接受远程 URL 字符串数组并自动提取文件名。
|
||||
*
|
||||
* @throws Exception 反射注入失败
|
||||
*/
|
||||
@Test
|
||||
public void testNormalizeRuntimeVariablesShouldAcceptRemoteFileUrls()
|
||||
throws Exception {
|
||||
WorkflowRunningParameterResolver resolver = newResolver();
|
||||
String firstUrl = "https://files.example.com/contracts/"
|
||||
+ "%E5%90%88%E5%90%8C%20v1.docx?signature=test";
|
||||
String secondUrl = "https://files.example.com/contracts/report.pdf";
|
||||
Map<String, Object> variables = new LinkedHashMap<>();
|
||||
variables.put("attachments", List.of(firstUrl, secondUrl));
|
||||
|
||||
Map<String, Object> normalized = resolver.normalizeRuntimeVariables(
|
||||
workflowContentWithStartParameters(),
|
||||
variables);
|
||||
|
||||
List<?> attachments = (List<?>) normalized.get("attachments");
|
||||
Assert.assertEquals(2, attachments.size());
|
||||
Assert.assertEquals(
|
||||
"合同 v1.docx",
|
||||
((Map<?, ?>) attachments.get(0)).get("fileName"));
|
||||
Assert.assertEquals(
|
||||
firstUrl,
|
||||
((Map<?, ?>) attachments.get(0)).get("filePath"));
|
||||
Assert.assertEquals(
|
||||
"report.pdf",
|
||||
((Map<?, ?>) attachments.get(1)).get("fileName"));
|
||||
}
|
||||
|
||||
/**
|
||||
* 单个远程文件 URL 也应归一化为文件对象数组。
|
||||
*
|
||||
* @throws Exception 反射注入失败
|
||||
*/
|
||||
@Test
|
||||
public void testNormalizeRuntimeVariablesShouldAcceptSingleRemoteFileUrl()
|
||||
throws Exception {
|
||||
WorkflowRunningParameterResolver resolver = newResolver();
|
||||
Map<String, Object> variables = new LinkedHashMap<>();
|
||||
variables.put(
|
||||
"attachments",
|
||||
"https://files.example.com/contracts/contract.docx");
|
||||
|
||||
Map<String, Object> normalized = resolver.normalizeRuntimeVariables(
|
||||
workflowContentWithStartParameters(),
|
||||
variables);
|
||||
|
||||
List<?> attachments = (List<?>) normalized.get("attachments");
|
||||
Assert.assertEquals(1, attachments.size());
|
||||
Assert.assertEquals(
|
||||
"contract.docx",
|
||||
((Map<?, ?>) attachments.get(0)).get("fileName"));
|
||||
}
|
||||
|
||||
/**
|
||||
* 无法从 URL 路径识别文件扩展名时应给出可恢复的格式提示。
|
||||
*
|
||||
* @throws Exception 反射注入失败
|
||||
*/
|
||||
@Test
|
||||
public void testNormalizeRuntimeVariablesShouldRejectAmbiguousRemoteUrl()
|
||||
throws Exception {
|
||||
WorkflowRunningParameterResolver resolver = newResolver();
|
||||
Map<String, Object> variables = new LinkedHashMap<>();
|
||||
variables.put(
|
||||
"attachments",
|
||||
"https://files.example.com/download?id=contract");
|
||||
|
||||
BusinessException exception = Assert.assertThrows(
|
||||
BusinessException.class,
|
||||
() -> resolver.normalizeRuntimeVariables(
|
||||
workflowContentWithStartParameters(),
|
||||
variables));
|
||||
|
||||
Assert.assertEquals(
|
||||
"文件参数 attachments 的 URL 路径无法识别带扩展名的文件名，"
|
||||
+ "请改用包含 fileName 和 filePath 的文件对象",
|
||||
exception.getMessage());
|
||||
}
|
||||
|
||||
/**
|
||||
* 多文件参数应按 filePath 去重并保留已有非文件变量。
|
||||
*
|
||||
@@ -270,6 +394,36 @@ public class WorkflowRunningParameterResolverTest {
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 文件参数应拒绝超过十个文件的输入。
|
||||
*
|
||||
* @throws Exception 反射注入失败
|
||||
*/
|
||||
@Test
|
||||
public void testNormalizeRuntimeVariablesShouldEnforceFileCountLimit()
|
||||
throws Exception {
|
||||
WorkflowRunningParameterResolver resolver = newResolver();
|
||||
List<Map<String, Object>> files = new java.util.ArrayList<>();
|
||||
for (int index = 0; index < 11; index++) {
|
||||
files.add(fileValue(
|
||||
"file-" + index + ".pdf",
|
||||
"/files/file-" + index + ".pdf",
|
||||
1L));
|
||||
}
|
||||
Map<String, Object> variables = new LinkedHashMap<>();
|
||||
variables.put("attachments", files);
|
||||
|
||||
BusinessException exception = Assert.assertThrows(
|
||||
BusinessException.class,
|
||||
() -> resolver.normalizeRuntimeVariables(
|
||||
workflowContentWithStartParameters(),
|
||||
variables));
|
||||
|
||||
Assert.assertEquals(
|
||||
"文件参数 attachments 最多上传 10 个文件",
|
||||
exception.getMessage());
|
||||
}
|
||||
|
||||
/**
|
||||
* 旧版图片 URL 应归一化为 URL 图片描述。
|
||||
*
|
||||
|
||||
@@ -0,0 +1,28 @@
|
||||
package tech.easyflow.ai.easyagentsflow.upload;
|
||||
|
||||
import org.junit.Test;
|
||||
import org.mockito.Mockito;
|
||||
|
||||
/**
|
||||
* {@link WorkflowApiUploadCleanupScheduler} 批量排空测试。
|
||||
*/
|
||||
public class WorkflowApiUploadCleanupSchedulerTest {
|
||||
|
||||
/**
|
||||
* 验证一次调度会连续处理多批到期记录。
|
||||
*/
|
||||
@Test
|
||||
public void cleanupShouldDrainMultipleBatches() {
|
||||
WorkflowApiUploadLifecycleService lifecycleService =
|
||||
Mockito.mock(WorkflowApiUploadLifecycleService.class);
|
||||
Mockito.when(lifecycleService.cleanupExpired(100))
|
||||
.thenReturn(100, 100, 20);
|
||||
WorkflowApiUploadCleanupScheduler scheduler =
|
||||
new WorkflowApiUploadCleanupScheduler(lifecycleService);
|
||||
|
||||
scheduler.cleanup();
|
||||
|
||||
Mockito.verify(lifecycleService, Mockito.times(3))
|
||||
.cleanupExpired(100);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,552 @@
|
||||
package tech.easyflow.ai.easyagentsflow.upload;
|
||||
|
||||
import com.easyagents.flow.core.chain.repository.ChainStateRepository;
|
||||
import org.junit.Assert;
|
||||
import org.junit.Test;
|
||||
import org.mockito.ArgumentCaptor;
|
||||
import org.mockito.InOrder;
|
||||
import org.mockito.Mockito;
|
||||
import org.springframework.web.multipart.MultipartFile;
|
||||
import tech.easyflow.ai.easyagentsflow.service.WorkflowRunningParameterResolver;
|
||||
import tech.easyflow.common.cache.RedisLockExecutor;
|
||||
import tech.easyflow.common.filestorage.FileStorageService;
|
||||
import tech.easyflow.common.filestorage.FileStorageWriteHandle;
|
||||
import tech.easyflow.common.filestorage.FileStorageWriteResult;
|
||||
import tech.easyflow.common.web.exceptions.BusinessException;
|
||||
|
||||
import java.util.LinkedHashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Optional;
|
||||
import java.util.Set;
|
||||
|
||||
/**
|
||||
* {@link WorkflowApiUploadLifecycleService} multipart 文件生命周期测试。
|
||||
*/
|
||||
public class WorkflowApiUploadLifecycleServiceTest {
|
||||
|
||||
/**
|
||||
* 验证同名文件 Part 会按顺序保存并注入文件对象数组。
|
||||
*/
|
||||
@Test
|
||||
public void prepareShouldStoreRepeatedFilePartsInOrder() {
|
||||
Fixture fixture = fixture();
|
||||
MultipartFile first = file("first.pdf", "application/pdf", 10L);
|
||||
MultipartFile second = file(
|
||||
"second.docx",
|
||||
"application/vnd.openxmlformats-officedocument.wordprocessingml.document",
|
||||
20L);
|
||||
Mockito.when(fixture.parameterResolver.resolveFileParameterNames("flow"))
|
||||
.thenReturn(Set.of("documents"));
|
||||
Mockito.when(fixture.parameterResolver.normalizeRuntimeVariables(
|
||||
Mockito.eq("flow"),
|
||||
Mockito.anyMap()))
|
||||
.thenAnswer(invocation -> new LinkedHashMap<>(
|
||||
invocation.getArgument(1)));
|
||||
FileStorageWriteHandle firstHandle = handle("first.pdf");
|
||||
FileStorageWriteHandle secondHandle = handle("second.docx");
|
||||
Mockito.when(fixture.fileStorageService.prepareRecoverableWrite(
|
||||
Mockito.anyString(),
|
||||
Mockito.anyString()))
|
||||
.thenReturn(firstHandle, secondHandle);
|
||||
Mockito.when(fixture.fileStorageService.saveRecoverable(
|
||||
first,
|
||||
firstHandle))
|
||||
.thenReturn(new FileStorageWriteResult(
|
||||
"/files/first.pdf",
|
||||
firstHandle.encodeLocator()));
|
||||
Mockito.when(fixture.fileStorageService.saveRecoverable(
|
||||
second,
|
||||
secondHandle))
|
||||
.thenReturn(new FileStorageWriteResult(
|
||||
"/files/second.docx",
|
||||
secondHandle.encodeLocator()));
|
||||
|
||||
WorkflowApiPreparedUpload prepared = fixture.service.prepare(
|
||||
"flow",
|
||||
Map.of("user_input", "解析"),
|
||||
Map.of("documents", List.of(first, second)));
|
||||
|
||||
Assert.assertNotNull(prepared.getRequestId());
|
||||
Assert.assertEquals("解析", prepared.getVariables().get("user_input"));
|
||||
@SuppressWarnings("unchecked")
|
||||
List<Map<String, Object>> documents =
|
||||
(List<Map<String, Object>>) prepared.getVariables()
|
||||
.get("documents");
|
||||
Assert.assertEquals(2, documents.size());
|
||||
Assert.assertEquals(
|
||||
"/files/first.pdf",
|
||||
documents.get(0).get("filePath"));
|
||||
Assert.assertEquals(
|
||||
"/files/second.docx",
|
||||
documents.get(1).get("filePath"));
|
||||
|
||||
ArgumentCaptor<WorkflowApiUploadRecord> recordCaptor =
|
||||
ArgumentCaptor.forClass(WorkflowApiUploadRecord.class);
|
||||
Mockito.verify(fixture.uploadStore).create(
|
||||
recordCaptor.capture());
|
||||
Assert.assertEquals(
|
||||
List.of("/files/first.pdf", "/files/second.docx"),
|
||||
recordCaptor.getValue().getStoredFiles().stream()
|
||||
.map(WorkflowApiStoredFile::filePath)
|
||||
.toList());
|
||||
InOrder writeOrder = Mockito.inOrder(
|
||||
fixture.uploadStore,
|
||||
fixture.fileStorageService);
|
||||
writeOrder.verify(fixture.uploadStore)
|
||||
.save(Mockito.any(WorkflowApiUploadRecord.class));
|
||||
writeOrder.verify(fixture.fileStorageService)
|
||||
.saveRecoverable(first, firstHandle);
|
||||
}
|
||||
|
||||
/**
|
||||
* 验证非法客户端 MIME 会按扩展名归一化，并同时用于存储和文件描述。
|
||||
*/
|
||||
@Test
|
||||
public void prepareShouldNormalizeInvalidContentType() {
|
||||
Fixture fixture = fixture();
|
||||
MultipartFile file = file(
|
||||
"C:\\fakepath\\report.docx",
|
||||
"Other",
|
||||
10L);
|
||||
Mockito.when(fixture.parameterResolver
|
||||
.resolveFileParameterNames("flow"))
|
||||
.thenReturn(Set.of("documents"));
|
||||
Mockito.when(fixture.parameterResolver.normalizeRuntimeVariables(
|
||||
Mockito.eq("flow"),
|
||||
Mockito.anyMap()))
|
||||
.thenAnswer(invocation -> new LinkedHashMap<>(
|
||||
invocation.getArgument(1)));
|
||||
FileStorageWriteHandle handle = handle("report.docx");
|
||||
Mockito.when(fixture.fileStorageService.prepareRecoverableWrite(
|
||||
Mockito.anyString(),
|
||||
Mockito.anyString()))
|
||||
.thenReturn(handle);
|
||||
Mockito.when(fixture.fileStorageService.saveRecoverable(
|
||||
Mockito.any(MultipartFile.class),
|
||||
Mockito.eq(handle)))
|
||||
.thenReturn(new FileStorageWriteResult(
|
||||
"/files/report.docx",
|
||||
handle.encodeLocator()));
|
||||
|
||||
WorkflowApiPreparedUpload prepared = fixture.service.prepare(
|
||||
"flow",
|
||||
Map.of(),
|
||||
Map.of("documents", List.of(file)));
|
||||
|
||||
ArgumentCaptor<MultipartFile> storedFile =
|
||||
ArgumentCaptor.forClass(MultipartFile.class);
|
||||
Mockito.verify(fixture.fileStorageService).saveRecoverable(
|
||||
storedFile.capture(),
|
||||
Mockito.eq(handle));
|
||||
Assert.assertEquals(
|
||||
"report.docx",
|
||||
storedFile.getValue().getOriginalFilename());
|
||||
Assert.assertEquals(
|
||||
"application/vnd.openxmlformats-officedocument.wordprocessingml.document",
|
||||
storedFile.getValue().getContentType());
|
||||
@SuppressWarnings("unchecked")
|
||||
List<Map<String, Object>> documents =
|
||||
(List<Map<String, Object>>) prepared.getVariables()
|
||||
.get("documents");
|
||||
Assert.assertEquals(
|
||||
storedFile.getValue().getContentType(),
|
||||
documents.get(0).get("contentType"));
|
||||
}
|
||||
|
||||
/**
|
||||
* 验证对象存储超时返回可重试的 50301，并补偿临时文件。
|
||||
*/
|
||||
@Test
|
||||
public void prepareShouldTranslateStorageTimeoutAndCleanup() {
|
||||
Fixture fixture = fixture();
|
||||
MultipartFile file = file(
|
||||
"report.pdf",
|
||||
"application/pdf",
|
||||
10L);
|
||||
Mockito.when(fixture.parameterResolver
|
||||
.resolveFileParameterNames("flow"))
|
||||
.thenReturn(Set.of("documents"));
|
||||
Mockito.when(fixture.parameterResolver.normalizeRuntimeVariables(
|
||||
Mockito.eq("flow"),
|
||||
Mockito.anyMap()))
|
||||
.thenAnswer(invocation -> new LinkedHashMap<>(
|
||||
invocation.getArgument(1)));
|
||||
FileStorageWriteHandle handle = handle("report.pdf");
|
||||
Mockito.when(fixture.fileStorageService.prepareRecoverableWrite(
|
||||
Mockito.anyString(),
|
||||
Mockito.anyString()))
|
||||
.thenReturn(handle);
|
||||
Mockito.when(fixture.fileStorageService.saveRecoverable(
|
||||
file,
|
||||
handle))
|
||||
.thenThrow(new IllegalStateException(
|
||||
"storage timeout",
|
||||
new java.net.SocketTimeoutException("timeout")));
|
||||
|
||||
BusinessException exception = Assert.assertThrows(
|
||||
BusinessException.class,
|
||||
() -> fixture.service.prepare(
|
||||
"flow",
|
||||
Map.of(),
|
||||
Map.of("documents", List.of(file))));
|
||||
|
||||
Assert.assertEquals(503, exception.getHttpStatus());
|
||||
Assert.assertEquals(50301, exception.getErrorCode());
|
||||
Assert.assertFalse(exception.getMessage().contains("Socket"));
|
||||
Mockito.verify(fixture.fileStorageService)
|
||||
.deleteRecoverable(handle);
|
||||
Mockito.verify(fixture.uploadStore)
|
||||
.remove(Mockito.any(WorkflowApiUploadRecord.class));
|
||||
}
|
||||
|
||||
/**
|
||||
* 验证存储鉴权等非暂时性错误返回安全的 50001，并执行补偿。
|
||||
*/
|
||||
@Test
|
||||
public void prepareShouldHidePermanentStorageFailureAndCleanup() {
|
||||
Fixture fixture = fixture();
|
||||
MultipartFile file = file(
|
||||
"report.pdf",
|
||||
"application/pdf",
|
||||
10L);
|
||||
Mockito.when(fixture.parameterResolver
|
||||
.resolveFileParameterNames("flow"))
|
||||
.thenReturn(Set.of("documents"));
|
||||
Mockito.when(fixture.parameterResolver.normalizeRuntimeVariables(
|
||||
Mockito.eq("flow"),
|
||||
Mockito.anyMap()))
|
||||
.thenAnswer(invocation -> new LinkedHashMap<>(
|
||||
invocation.getArgument(1)));
|
||||
FileStorageWriteHandle handle = handle("report.pdf");
|
||||
Mockito.when(fixture.fileStorageService.prepareRecoverableWrite(
|
||||
Mockito.anyString(),
|
||||
Mockito.anyString()))
|
||||
.thenReturn(handle);
|
||||
Mockito.when(fixture.fileStorageService.saveRecoverable(
|
||||
file,
|
||||
handle))
|
||||
.thenThrow(new IllegalStateException(
|
||||
"AccessKey=secret, endpoint=http://internal:9000"));
|
||||
|
||||
BusinessException exception = Assert.assertThrows(
|
||||
BusinessException.class,
|
||||
() -> fixture.service.prepare(
|
||||
"flow",
|
||||
Map.of(),
|
||||
Map.of("documents", List.of(file))));
|
||||
|
||||
Assert.assertEquals(500, exception.getHttpStatus());
|
||||
Assert.assertEquals(50001, exception.getErrorCode());
|
||||
Assert.assertFalse(exception.getMessage().contains("secret"));
|
||||
Mockito.verify(fixture.fileStorageService)
|
||||
.deleteRecoverable(handle);
|
||||
Mockito.verify(fixture.uploadStore)
|
||||
.remove(Mockito.any(WorkflowApiUploadRecord.class));
|
||||
}
|
||||
|
||||
/**
|
||||
* 验证未知文件 Part 在写入存储前被拒绝。
|
||||
*/
|
||||
@Test
|
||||
public void prepareShouldRejectUnknownFilePartBeforeStorage() {
|
||||
Fixture fixture = fixture();
|
||||
MultipartFile file = file("data.pdf", "application/pdf", 10L);
|
||||
Mockito.when(fixture.parameterResolver.resolveFileParameterNames("flow"))
|
||||
.thenReturn(Set.of("documents"));
|
||||
|
||||
BusinessException exception = Assert.assertThrows(
|
||||
BusinessException.class,
|
||||
() -> fixture.service.prepare(
|
||||
"flow",
|
||||
Map.of(),
|
||||
Map.of("unknown", List.of(file))));
|
||||
|
||||
Assert.assertTrue(exception.getMessage().contains("unknown"));
|
||||
Mockito.verifyNoInteractions(fixture.fileStorageService);
|
||||
Mockito.verifyNoInteractions(fixture.uploadStore);
|
||||
}
|
||||
|
||||
/**
|
||||
* 验证缺少开始节点必填文件字段时在存储前返回 40016。
|
||||
*/
|
||||
@Test
|
||||
public void prepareShouldRejectMissingRequiredFileBeforeStorage() {
|
||||
Fixture fixture = fixture();
|
||||
MultipartFile file = file("data.pdf", "application/pdf", 10L);
|
||||
Mockito.when(fixture.parameterResolver
|
||||
.resolveFileParameterNames("flow"))
|
||||
.thenReturn(Set.of("documents", "appendix"));
|
||||
Mockito.when(fixture.parameterResolver
|
||||
.resolveRequiredFileParameterNames("flow"))
|
||||
.thenReturn(Set.of("documents", "appendix"));
|
||||
|
||||
BusinessException exception = Assert.assertThrows(
|
||||
BusinessException.class,
|
||||
() -> fixture.service.prepare(
|
||||
"flow",
|
||||
Map.of(),
|
||||
Map.of("documents", List.of(file))));
|
||||
|
||||
Assert.assertEquals(40016, exception.getErrorCode());
|
||||
Assert.assertTrue(exception.getMessage().contains("appendix"));
|
||||
Mockito.verifyNoInteractions(fixture.fileStorageService);
|
||||
Mockito.verifyNoInteractions(fixture.uploadStore);
|
||||
}
|
||||
|
||||
/**
|
||||
* 验证文件数量和大小限制统一转换为 41301。
|
||||
*/
|
||||
@Test
|
||||
public void prepareShouldTranslateFileLimitToPayloadTooLarge() {
|
||||
Fixture fixture = fixture();
|
||||
MultipartFile file = file("data.pdf", "application/pdf", 10L);
|
||||
Mockito.when(fixture.parameterResolver
|
||||
.resolveFileParameterNames("flow"))
|
||||
.thenReturn(Set.of("documents"));
|
||||
Mockito.when(fixture.parameterResolver.normalizeRuntimeVariables(
|
||||
Mockito.eq("flow"),
|
||||
Mockito.anyMap()))
|
||||
.thenThrow(new BusinessException(
|
||||
"文件参数 documents 最多上传 10 个文件"));
|
||||
|
||||
BusinessException exception = Assert.assertThrows(
|
||||
BusinessException.class,
|
||||
() -> fixture.service.prepare(
|
||||
"flow",
|
||||
Map.of(),
|
||||
Map.of("documents", List.of(file))));
|
||||
|
||||
Assert.assertEquals(413, exception.getHttpStatus());
|
||||
Assert.assertEquals(41301, exception.getErrorCode());
|
||||
Mockito.verifyNoInteractions(fixture.fileStorageService);
|
||||
Mockito.verifyNoInteractions(fixture.uploadStore);
|
||||
}
|
||||
|
||||
/**
|
||||
* 验证文件写入后 Redis 更新失败时仍使用内存路径执行补偿删除。
|
||||
*/
|
||||
@Test
|
||||
public void prepareShouldDeleteSavedFileWhenRecordUpdateFails() {
|
||||
Fixture fixture = fixture();
|
||||
MultipartFile file = file("data.pdf", "application/pdf", 10L);
|
||||
Mockito.when(fixture.parameterResolver.resolveFileParameterNames("flow"))
|
||||
.thenReturn(Set.of("documents"));
|
||||
Mockito.when(fixture.parameterResolver.normalizeRuntimeVariables(
|
||||
Mockito.eq("flow"),
|
||||
Mockito.anyMap()))
|
||||
.thenAnswer(invocation -> new LinkedHashMap<>(
|
||||
invocation.getArgument(1)));
|
||||
FileStorageWriteHandle handle = handle("data.pdf");
|
||||
Mockito.when(fixture.fileStorageService.prepareRecoverableWrite(
|
||||
Mockito.anyString(),
|
||||
Mockito.anyString()))
|
||||
.thenReturn(handle);
|
||||
Mockito.when(fixture.fileStorageService.saveRecoverable(
|
||||
file,
|
||||
handle))
|
||||
.thenReturn(new FileStorageWriteResult(
|
||||
"/files/data.pdf",
|
||||
handle.encodeLocator()));
|
||||
Mockito.doNothing()
|
||||
.doThrow(new IllegalStateException("Redis 写入失败"))
|
||||
.when(fixture.uploadStore)
|
||||
.save(Mockito.any(WorkflowApiUploadRecord.class));
|
||||
|
||||
IllegalStateException exception = Assert.assertThrows(
|
||||
IllegalStateException.class,
|
||||
() -> fixture.service.prepare(
|
||||
"flow",
|
||||
Map.of(),
|
||||
Map.of("documents", List.of(file))));
|
||||
|
||||
Assert.assertTrue(exception.getMessage().contains("Redis"));
|
||||
Mockito.verify(fixture.fileStorageService)
|
||||
.deleteRecoverable(handle);
|
||||
Mockito.verify(fixture.uploadStore)
|
||||
.remove(Mockito.any(WorkflowApiUploadRecord.class));
|
||||
}
|
||||
|
||||
/**
|
||||
* 验证启动失败后的 abort 会幂等删除已保存文件和上传记录。
|
||||
*/
|
||||
@Test
|
||||
public void abortShouldDeleteStoredFilesAndRecord() {
|
||||
Fixture fixture = fixture();
|
||||
WorkflowApiUploadRecord record = new WorkflowApiUploadRecord();
|
||||
record.setRequestId("request-1");
|
||||
FileStorageWriteHandle firstHandle = handle("a.pdf");
|
||||
FileStorageWriteHandle secondHandle = handle("b.pdf");
|
||||
record.setStoredFiles(List.of(
|
||||
new WorkflowApiStoredFile(
|
||||
"/files/a.pdf",
|
||||
firstHandle.encodeLocator()),
|
||||
new WorkflowApiStoredFile(
|
||||
"/files/b.pdf",
|
||||
secondHandle.encodeLocator())));
|
||||
RedisLockExecutor.LockHandle handle =
|
||||
Mockito.mock(RedisLockExecutor.LockHandle.class);
|
||||
Mockito.when(fixture.redisLockExecutor.tryAcquire(
|
||||
Mockito.anyString(),
|
||||
Mockito.any(),
|
||||
Mockito.any()))
|
||||
.thenReturn(handle);
|
||||
Mockito.when(fixture.uploadStore.find("request-1"))
|
||||
.thenReturn(Optional.of(record));
|
||||
|
||||
fixture.service.abort("request-1");
|
||||
|
||||
Mockito.verify(fixture.fileStorageService)
|
||||
.deleteRecoverable(firstHandle);
|
||||
Mockito.verify(fixture.fileStorageService)
|
||||
.deleteRecoverable(secondHandle);
|
||||
Mockito.verify(fixture.uploadStore).remove(record);
|
||||
Mockito.verify(handle).close();
|
||||
}
|
||||
|
||||
/**
|
||||
* 验证单条清理失败不会阻塞后续到期记录。
|
||||
*/
|
||||
@Test
|
||||
public void cleanupExpiredShouldContinueAfterFailedRecord() {
|
||||
Fixture fixture = fixture();
|
||||
FileStorageWriteHandle failedFile = handle("failed.pdf");
|
||||
FileStorageWriteHandle goodFile = handle("good.pdf");
|
||||
WorkflowApiUploadRecord failedRecord =
|
||||
storedRecord("failed", failedFile);
|
||||
WorkflowApiUploadRecord goodRecord =
|
||||
storedRecord("good", goodFile);
|
||||
RedisLockExecutor.LockHandle failedLock =
|
||||
Mockito.mock(RedisLockExecutor.LockHandle.class);
|
||||
RedisLockExecutor.LockHandle goodLock =
|
||||
Mockito.mock(RedisLockExecutor.LockHandle.class);
|
||||
Mockito.when(fixture.uploadStore.claimExpired(
|
||||
Mockito.anyLong(),
|
||||
Mockito.anyLong()))
|
||||
.thenReturn(
|
||||
Optional.of("failed"),
|
||||
Optional.of("good"),
|
||||
Optional.empty());
|
||||
Mockito.when(fixture.redisLockExecutor.tryAcquire(
|
||||
Mockito.anyString(),
|
||||
Mockito.any(),
|
||||
Mockito.any()))
|
||||
.thenReturn(failedLock, goodLock);
|
||||
Mockito.when(fixture.uploadStore.find("failed"))
|
||||
.thenReturn(Optional.of(failedRecord));
|
||||
Mockito.when(fixture.uploadStore.find("good"))
|
||||
.thenReturn(Optional.of(goodRecord));
|
||||
Mockito.doThrow(new IllegalStateException("对象存储不可用"))
|
||||
.when(fixture.fileStorageService)
|
||||
.deleteRecoverable(failedFile);
|
||||
|
||||
int processed = fixture.service.cleanupExpired(10);
|
||||
|
||||
Assert.assertEquals(2, processed);
|
||||
Mockito.verify(fixture.fileStorageService)
|
||||
.deleteRecoverable(goodFile);
|
||||
Mockito.verify(fixture.uploadStore).remove(goodRecord);
|
||||
}
|
||||
|
||||
/**
|
||||
* 创建 multipart 文件桩。
|
||||
*
|
||||
* @param name 文件名
|
||||
* @param contentType MIME 类型
|
||||
* @param size 文件大小
|
||||
* @return multipart 文件桩
|
||||
*/
|
||||
private MultipartFile file(
|
||||
String name,
|
||||
String contentType,
|
||||
long size) {
|
||||
MultipartFile file = Mockito.mock(MultipartFile.class);
|
||||
Mockito.when(file.isEmpty()).thenReturn(false);
|
||||
Mockito.when(file.getOriginalFilename()).thenReturn(name);
|
||||
Mockito.when(file.getContentType()).thenReturn(contentType);
|
||||
Mockito.when(file.getSize()).thenReturn(size);
|
||||
return file;
|
||||
}
|
||||
|
||||
/**
|
||||
* 创建可恢复文件存储句柄。
|
||||
*
|
||||
* @param filename 固定文件名
|
||||
* @return 测试句柄
|
||||
*/
|
||||
private FileStorageWriteHandle handle(String filename) {
|
||||
return new FileStorageWriteHandle(
|
||||
"localFileStorage",
|
||||
"",
|
||||
"/tmp/easyflow-test",
|
||||
"workflow-api-upload/test",
|
||||
filename);
|
||||
}
|
||||
|
||||
/**
|
||||
* 创建包含单个临时文件的上传记录。
|
||||
*
|
||||
* @param requestId 请求 ID
|
||||
* @param handle 文件句柄
|
||||
* @return 上传记录
|
||||
*/
|
||||
private WorkflowApiUploadRecord storedRecord(
|
||||
String requestId,
|
||||
FileStorageWriteHandle handle) {
|
||||
WorkflowApiUploadRecord record =
|
||||
new WorkflowApiUploadRecord();
|
||||
record.setRequestId(requestId);
|
||||
record.setStoredFiles(List.of(
|
||||
new WorkflowApiStoredFile(
|
||||
"/files/" + handle.getFilename(),
|
||||
handle.encodeLocator())));
|
||||
return record;
|
||||
}
|
||||
|
||||
/**
|
||||
* 创建生命周期服务测试夹具。
|
||||
*
|
||||
* @return 测试夹具
|
||||
*/
|
||||
private Fixture fixture() {
|
||||
WorkflowRunningParameterResolver parameterResolver =
|
||||
Mockito.mock(WorkflowRunningParameterResolver.class);
|
||||
FileStorageService fileStorageService =
|
||||
Mockito.mock(FileStorageService.class);
|
||||
WorkflowApiUploadStore uploadStore =
|
||||
Mockito.mock(WorkflowApiUploadStore.class);
|
||||
ChainStateRepository chainStateRepository =
|
||||
Mockito.mock(ChainStateRepository.class);
|
||||
RedisLockExecutor redisLockExecutor =
|
||||
Mockito.mock(RedisLockExecutor.class);
|
||||
return new Fixture(
|
||||
new WorkflowApiUploadLifecycleService(
|
||||
parameterResolver,
|
||||
fileStorageService,
|
||||
new WorkflowApiMultipartFileNormalizer(),
|
||||
uploadStore,
|
||||
chainStateRepository,
|
||||
redisLockExecutor),
|
||||
parameterResolver,
|
||||
fileStorageService,
|
||||
uploadStore,
|
||||
redisLockExecutor);
|
||||
}
|
||||
|
||||
/**
|
||||
* 生命周期服务测试夹具。
|
||||
*
|
||||
* @param service 被测服务
|
||||
* @param parameterResolver 参数解析器
|
||||
* @param fileStorageService 文件存储
|
||||
* @param uploadStore 上传记录存储
|
||||
* @param redisLockExecutor 分布式锁执行器
|
||||
*/
|
||||
private record Fixture(
|
||||
WorkflowApiUploadLifecycleService service,
|
||||
WorkflowRunningParameterResolver parameterResolver,
|
||||
FileStorageService fileStorageService,
|
||||
WorkflowApiUploadStore uploadStore,
|
||||
RedisLockExecutor redisLockExecutor) {
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,135 @@
|
||||
package tech.easyflow.ai.easyagentsflow.upload;
|
||||
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import org.junit.Assert;
|
||||
import org.junit.Test;
|
||||
import org.mockito.ArgumentCaptor;
|
||||
import org.mockito.ArgumentMatchers;
|
||||
import org.mockito.Mockito;
|
||||
import org.springframework.data.redis.core.StringRedisTemplate;
|
||||
import org.springframework.data.redis.core.ValueOperations;
|
||||
import org.springframework.data.redis.core.script.RedisScript;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* {@link WorkflowApiUploadStore} Redis 原子性契约测试。
|
||||
*/
|
||||
public class WorkflowApiUploadStoreTest {
|
||||
|
||||
/**
|
||||
* 验证记录与清理索引通过同槽 Lua 脚本原子创建。
|
||||
*/
|
||||
@Test
|
||||
@SuppressWarnings({"unchecked", "rawtypes"})
|
||||
public void createShouldUseSameSlotAtomicScript() {
|
||||
StringRedisTemplate redisTemplate =
|
||||
Mockito.mock(StringRedisTemplate.class);
|
||||
Mockito.doReturn(1L).when(redisTemplate).execute(
|
||||
ArgumentMatchers.<RedisScript<Long>>any(),
|
||||
ArgumentMatchers.<List<String>>any(),
|
||||
ArgumentMatchers.anyString(),
|
||||
ArgumentMatchers.anyString(),
|
||||
ArgumentMatchers.anyString(),
|
||||
ArgumentMatchers.anyString(),
|
||||
ArgumentMatchers.anyString());
|
||||
WorkflowApiUploadStore store = new WorkflowApiUploadStore(
|
||||
redisTemplate,
|
||||
new ObjectMapper());
|
||||
WorkflowApiUploadRecord record = record("request-1", null);
|
||||
|
||||
store.create(record);
|
||||
|
||||
ArgumentCaptor<RedisScript<Long>> scriptCaptor =
|
||||
ArgumentCaptor.forClass((Class) RedisScript.class);
|
||||
ArgumentCaptor<List<String>> keysCaptor =
|
||||
ArgumentCaptor.forClass((Class) List.class);
|
||||
Mockito.verify(redisTemplate).execute(
|
||||
scriptCaptor.capture(),
|
||||
keysCaptor.capture(),
|
||||
ArgumentMatchers.anyString(),
|
||||
ArgumentMatchers.anyString(),
|
||||
ArgumentMatchers.anyString(),
|
||||
ArgumentMatchers.anyString(),
|
||||
ArgumentMatchers.anyString());
|
||||
Assert.assertTrue(
|
||||
scriptCaptor.getValue().getScriptAsString()
|
||||
.contains("redis.call('zadd'"));
|
||||
Assert.assertEquals(3, keysCaptor.getValue().size());
|
||||
Assert.assertTrue(keysCaptor.getValue().stream()
|
||||
.allMatch(key -> key.contains("{api-upload}")));
|
||||
}
|
||||
|
||||
/**
|
||||
* 验证重新绑定执行 ID 时会在同一脚本中清除旧索引。
|
||||
*
|
||||
* @throws Exception 上传记录序列化失败
|
||||
*/
|
||||
@Test
|
||||
@SuppressWarnings({"unchecked", "rawtypes"})
|
||||
public void bindExecutionShouldReplacePreviousIndexAtomically()
|
||||
throws Exception {
|
||||
StringRedisTemplate redisTemplate =
|
||||
Mockito.mock(StringRedisTemplate.class);
|
||||
ValueOperations<String, String> valueOperations =
|
||||
Mockito.mock(ValueOperations.class);
|
||||
Mockito.when(redisTemplate.opsForValue())
|
||||
.thenReturn(valueOperations);
|
||||
ObjectMapper objectMapper = new ObjectMapper();
|
||||
WorkflowApiUploadRecord record =
|
||||
record("request-1", "execution-old");
|
||||
Mockito.when(valueOperations.get(
|
||||
"easyflow:workflow:{api-upload}:record:request-1"))
|
||||
.thenReturn(objectMapper.writeValueAsString(record));
|
||||
Mockito.doReturn(1L).when(redisTemplate).execute(
|
||||
ArgumentMatchers.<RedisScript<Long>>any(),
|
||||
ArgumentMatchers.<List<String>>any(),
|
||||
ArgumentMatchers.anyString(),
|
||||
ArgumentMatchers.anyString(),
|
||||
ArgumentMatchers.anyString(),
|
||||
ArgumentMatchers.anyString());
|
||||
WorkflowApiUploadStore store = new WorkflowApiUploadStore(
|
||||
redisTemplate,
|
||||
objectMapper);
|
||||
|
||||
store.bindExecution("request-1", "execution-new");
|
||||
|
||||
ArgumentCaptor<RedisScript<Long>> scriptCaptor =
|
||||
ArgumentCaptor.forClass((Class) RedisScript.class);
|
||||
ArgumentCaptor<List<String>> keysCaptor =
|
||||
ArgumentCaptor.forClass((Class) List.class);
|
||||
Mockito.verify(redisTemplate).execute(
|
||||
scriptCaptor.capture(),
|
||||
keysCaptor.capture(),
|
||||
ArgumentMatchers.anyString(),
|
||||
ArgumentMatchers.anyString(),
|
||||
ArgumentMatchers.anyString(),
|
||||
ArgumentMatchers.anyString());
|
||||
Assert.assertEquals(
|
||||
"easyflow:workflow:{api-upload}:execution:execution-new",
|
||||
keysCaptor.getValue().get(1));
|
||||
Assert.assertEquals(
|
||||
"easyflow:workflow:{api-upload}:execution:execution-old",
|
||||
keysCaptor.getValue().get(2));
|
||||
Assert.assertTrue(
|
||||
scriptCaptor.getValue().getScriptAsString()
|
||||
.contains("redis.call('del', KEYS[3])"));
|
||||
}
|
||||
|
||||
/**
|
||||
* 创建测试上传记录。
|
||||
*
|
||||
* @param requestId 上传请求 ID
|
||||
* @param executeId 执行 ID
|
||||
* @return 上传记录
|
||||
*/
|
||||
private WorkflowApiUploadRecord record(
|
||||
String requestId,
|
||||
String executeId) {
|
||||
WorkflowApiUploadRecord record = new WorkflowApiUploadRecord();
|
||||
record.setRequestId(requestId);
|
||||
record.setExecuteId(executeId);
|
||||
record.setCleanupAt(1_000L);
|
||||
return record;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,145 @@
|
||||
package tech.easyflow.ai.easyagentsflow.upload;
|
||||
|
||||
import org.junit.Assert;
|
||||
import org.junit.Test;
|
||||
import org.mockito.Mockito;
|
||||
import tech.easyflow.common.filestorage.FileStorageService;
|
||||
import tech.easyflow.common.filestorage.FileStorageWriteHandle;
|
||||
|
||||
import java.io.ByteArrayInputStream;
|
||||
import java.io.IOException;
|
||||
import java.io.InputStream;
|
||||
import java.nio.charset.StandardCharsets;
|
||||
import java.util.List;
|
||||
import java.util.Optional;
|
||||
|
||||
/**
|
||||
* {@link WorkflowApiUploadedFileReader} 上传记录授权边界测试。
|
||||
*/
|
||||
public class WorkflowApiUploadedFileReaderTest {
|
||||
|
||||
private static final String REQUEST_ID =
|
||||
"0123456789abcdef0123456789abcdef";
|
||||
private static final String FILENAME =
|
||||
"000-abcdefabcdefabcdefabcdefabcdefab.docx";
|
||||
private static final String FILE_URL =
|
||||
"http://127.0.0.1:39000/easyflow/attachment/"
|
||||
+ "workflow-api-upload/" + REQUEST_ID + "/" + FILENAME;
|
||||
|
||||
/**
|
||||
* 验证 URL、上传记录与恢复句柄完全匹配后按固定后端读取。
|
||||
*
|
||||
* @throws Exception 测试流读取失败时抛出
|
||||
*/
|
||||
@Test
|
||||
public void shouldReadExactRecordedUploadByRecoverableHandle() throws Exception {
|
||||
WorkflowApiUploadStore uploadStore = Mockito.mock(WorkflowApiUploadStore.class);
|
||||
FileStorageService fileStorageService = Mockito.mock(FileStorageService.class);
|
||||
WorkflowApiUploadedFileReader reader =
|
||||
new WorkflowApiUploadedFileReader(uploadStore, fileStorageService);
|
||||
FileStorageWriteHandle handle = handle();
|
||||
WorkflowApiUploadRecord record = record(FILE_URL, handle);
|
||||
byte[] content = "document-content".getBytes(StandardCharsets.UTF_8);
|
||||
Mockito.when(uploadStore.find(REQUEST_ID)).thenReturn(Optional.of(record));
|
||||
Mockito.when(fileStorageService.readRecoverable(handle))
|
||||
.thenReturn(new ByteArrayInputStream(content));
|
||||
|
||||
Optional<InputStream> opened = reader.openVerified(FILE_URL);
|
||||
|
||||
Assert.assertTrue(opened.isPresent());
|
||||
try (InputStream inputStream = opened.orElseThrow()) {
|
||||
Assert.assertArrayEquals(content, inputStream.readAllBytes());
|
||||
}
|
||||
Mockito.verify(fileStorageService).readRecoverable(handle);
|
||||
}
|
||||
|
||||
/**
|
||||
* 验证普通远端 URL 不访问上传记录，也不获得内部读取权限。
|
||||
*
|
||||
* @throws IOException 路径解析失败时抛出
|
||||
*/
|
||||
@Test
|
||||
public void shouldIgnoreOrdinaryRemoteUrl() throws IOException {
|
||||
WorkflowApiUploadStore uploadStore = Mockito.mock(WorkflowApiUploadStore.class);
|
||||
FileStorageService fileStorageService = Mockito.mock(FileStorageService.class);
|
||||
WorkflowApiUploadedFileReader reader =
|
||||
new WorkflowApiUploadedFileReader(uploadStore, fileStorageService);
|
||||
|
||||
Optional<InputStream> opened = reader.openVerified(
|
||||
"http://127.0.0.1:39000/easyflow/attachment/ordinary.docx");
|
||||
|
||||
Assert.assertTrue(opened.isEmpty());
|
||||
Mockito.verifyNoInteractions(uploadStore, fileStorageService);
|
||||
}
|
||||
|
||||
/**
|
||||
* 验证看似系统目录的 URL 在 Redis 记录不存在时明确判定为失效。
|
||||
*/
|
||||
@Test
|
||||
public void shouldRejectManagedPathWhenUploadRecordExpired() {
|
||||
WorkflowApiUploadStore uploadStore = Mockito.mock(WorkflowApiUploadStore.class);
|
||||
FileStorageService fileStorageService = Mockito.mock(FileStorageService.class);
|
||||
WorkflowApiUploadedFileReader reader =
|
||||
new WorkflowApiUploadedFileReader(uploadStore, fileStorageService);
|
||||
Mockito.when(uploadStore.find(REQUEST_ID)).thenReturn(Optional.empty());
|
||||
|
||||
IOException exception = Assert.assertThrows(
|
||||
IOException.class,
|
||||
() -> reader.openVerified(FILE_URL));
|
||||
|
||||
Assert.assertTrue(exception.getMessage().contains("已失效"));
|
||||
Mockito.verifyNoInteractions(fileStorageService);
|
||||
}
|
||||
|
||||
/**
|
||||
* 验证同一请求 ID 下未记录的 URL 不能复用其他文件的存储 locator。
|
||||
*/
|
||||
@Test
|
||||
public void shouldRejectUrlThatDoesNotExactlyMatchStoredFile() {
|
||||
WorkflowApiUploadStore uploadStore = Mockito.mock(WorkflowApiUploadStore.class);
|
||||
FileStorageService fileStorageService = Mockito.mock(FileStorageService.class);
|
||||
WorkflowApiUploadedFileReader reader =
|
||||
new WorkflowApiUploadedFileReader(uploadStore, fileStorageService);
|
||||
Mockito.when(uploadStore.find(REQUEST_ID)).thenReturn(Optional.of(
|
||||
record(FILE_URL + "?different=true", handle())));
|
||||
|
||||
IOException exception = Assert.assertThrows(
|
||||
IOException.class,
|
||||
() -> reader.openVerified(FILE_URL));
|
||||
|
||||
Assert.assertTrue(exception.getMessage().contains("不匹配"));
|
||||
Mockito.verifyNoInteractions(fileStorageService);
|
||||
}
|
||||
|
||||
/**
|
||||
* 创建与系统上传目录一致的恢复句柄。
|
||||
*
|
||||
* @return 测试句柄
|
||||
*/
|
||||
private FileStorageWriteHandle handle() {
|
||||
return new FileStorageWriteHandle(
|
||||
"local",
|
||||
"",
|
||||
"/tmp/easyflow-test",
|
||||
"workflow-api-upload/" + REQUEST_ID,
|
||||
FILENAME);
|
||||
}
|
||||
|
||||
/**
|
||||
* 创建包含单个受管文件的上传记录。
|
||||
*
|
||||
* @param fileUrl 记录中的完整文件 URL
|
||||
* @param handle 文件存储句柄
|
||||
* @return 上传记录
|
||||
*/
|
||||
private WorkflowApiUploadRecord record(
|
||||
String fileUrl,
|
||||
FileStorageWriteHandle handle) {
|
||||
WorkflowApiUploadRecord record = new WorkflowApiUploadRecord();
|
||||
record.setRequestId(REQUEST_ID);
|
||||
record.setStoredFiles(List.of(new WorkflowApiStoredFile(
|
||||
fileUrl,
|
||||
handle.encodeLocator())));
|
||||
return record;
|
||||
}
|
||||
}
|
||||
@@ -2,6 +2,8 @@ package tech.easyflow.ai.node;
|
||||
|
||||
import org.junit.Assert;
|
||||
import org.junit.Test;
|
||||
import org.mockito.Mockito;
|
||||
import tech.easyflow.ai.easyagentsflow.upload.WorkflowApiUploadedFileReader;
|
||||
import tech.easyflow.ai.document.model.DocumentParseTaskInfo;
|
||||
import tech.easyflow.ai.document.model.DocumentParseTaskStatus;
|
||||
import tech.easyflow.ai.document.model.DocumentParsedResult;
|
||||
@@ -14,13 +16,12 @@ import java.io.ByteArrayInputStream;
|
||||
import java.io.File;
|
||||
import java.io.IOException;
|
||||
import java.io.InputStream;
|
||||
import java.net.InetSocketAddress;
|
||||
import java.nio.charset.StandardCharsets;
|
||||
import java.util.Arrays;
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import com.sun.net.httpserver.HttpServer;
|
||||
import java.util.Optional;
|
||||
|
||||
/**
|
||||
* {@link DocNodeFileContentExtractor} 单元测试。
|
||||
@@ -175,42 +176,63 @@ public class DocNodeFileContentExtractorTest {
|
||||
}
|
||||
|
||||
/**
|
||||
* 验证远端素材 URL 的非桥接文件不会误走本地存储读取。
|
||||
* 验证普通远端素材 URL 的非桥接文件仍拒绝访问回环地址。
|
||||
*/
|
||||
@Test
|
||||
public void shouldReadRemoteUrlForUnsupportedType() {
|
||||
public void shouldRejectLoopbackRemoteUrlForUnsupportedType() {
|
||||
RecordingDocumentParseBridgeService bridgeService = new RecordingDocumentParseBridgeService();
|
||||
HttpServer server;
|
||||
try {
|
||||
server = HttpServer.create(new InetSocketAddress(0), 0);
|
||||
} catch (IOException e) {
|
||||
throw new RuntimeException(e);
|
||||
DocNodeFileContentExtractor extractor = new DocNodeFileContentExtractor(
|
||||
bridgeService,
|
||||
new FailingFileStorageService(),
|
||||
new ReadingReaderManager()
|
||||
);
|
||||
|
||||
RuntimeException exception = Assert.assertThrows(
|
||||
RuntimeException.class,
|
||||
() -> extractor.extract(buildFileValue(
|
||||
"note.txt",
|
||||
"http://127.0.0.1:39000/note.txt",
|
||||
"text/plain")));
|
||||
|
||||
Throwable cause = exception;
|
||||
while (cause != null
|
||||
&& !(cause instanceof java.net.UnknownHostException)) {
|
||||
cause = cause.getCause();
|
||||
}
|
||||
byte[] body = "remote text".getBytes(StandardCharsets.UTF_8);
|
||||
server.createContext("/note.txt", exchange -> {
|
||||
exchange.sendResponseHeaders(200, body.length);
|
||||
exchange.getResponseBody().write(body);
|
||||
exchange.close();
|
||||
});
|
||||
server.start();
|
||||
try {
|
||||
DocNodeFileContentExtractor extractor = new DocNodeFileContentExtractor(
|
||||
Assert.assertNotNull(cause);
|
||||
Assert.assertNull(bridgeService.lastSource);
|
||||
}
|
||||
|
||||
/**
|
||||
* 验证受管上传 URL 的非桥接文件通过记录校验后走内部存储读取。
|
||||
*
|
||||
* @throws IOException 测试流配置失败时抛出
|
||||
*/
|
||||
@Test
|
||||
public void shouldReadVerifiedManagedUploadForUnsupportedType() throws IOException {
|
||||
RecordingDocumentParseBridgeService bridgeService = new RecordingDocumentParseBridgeService();
|
||||
WorkflowApiUploadedFileReader uploadedFileReader =
|
||||
Mockito.mock(WorkflowApiUploadedFileReader.class);
|
||||
String fileUrl = "http://127.0.0.1:39000/easyflow/attachment/"
|
||||
+ "workflow-api-upload/0123456789abcdef0123456789abcdef/note.txt";
|
||||
byte[] body = "managed text".getBytes(StandardCharsets.UTF_8);
|
||||
Mockito.when(uploadedFileReader.isManagedPathCandidate(fileUrl))
|
||||
.thenReturn(true);
|
||||
Mockito.when(uploadedFileReader.openVerified(fileUrl))
|
||||
.thenReturn(Optional.of(new ByteArrayInputStream(body)));
|
||||
DocNodeFileContentExtractor extractor = new DocNodeFileContentExtractor(
|
||||
bridgeService,
|
||||
new FailingFileStorageService(),
|
||||
new ReadingReaderManager()
|
||||
);
|
||||
new ReadingReaderManager(),
|
||||
uploadedFileReader);
|
||||
|
||||
String content = extractor.extract(buildFileValue(
|
||||
String content = extractor.extract(buildFileValue(
|
||||
"note.txt",
|
||||
"http://127.0.0.1:" + server.getAddress().getPort() + "/note.txt",
|
||||
"text/plain"
|
||||
));
|
||||
fileUrl,
|
||||
"text/plain"));
|
||||
|
||||
Assert.assertEquals("remote text", content);
|
||||
Assert.assertNull(bridgeService.lastSource);
|
||||
} finally {
|
||||
server.stop(0);
|
||||
}
|
||||
Assert.assertEquals("managed text", content);
|
||||
Assert.assertNull(bridgeService.lastSource);
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
Reference in New Issue
Block a user