diff --git a/easyflow-api/easyflow-api-admin/src/main/java/tech/easyflow/admin/controller/ai/WorkflowPublicChatController.java b/easyflow-api/easyflow-api-admin/src/main/java/tech/easyflow/admin/controller/ai/WorkflowPublicChatController.java new file mode 100644 index 00000000..50c7c82d --- /dev/null +++ b/easyflow-api/easyflow-api-admin/src/main/java/tech/easyflow/admin/controller/ai/WorkflowPublicChatController.java @@ -0,0 +1,117 @@ +package tech.easyflow.admin.controller.ai; + +import cn.dev33.satoken.annotation.SaIgnore; +import jakarta.servlet.http.HttpServletRequest; +import org.springframework.http.MediaType; +import org.springframework.web.bind.annotation.GetMapping; +import org.springframework.web.bind.annotation.PostMapping; +import org.springframework.web.bind.annotation.RequestMapping; +import org.springframework.web.bind.annotation.RequestParam; +import org.springframework.web.bind.annotation.RestController; +import org.springframework.web.multipart.MultipartFile; +import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; +import tech.easyflow.admin.service.ai.WorkflowPublicChatService; +import tech.easyflow.ai.share.WorkflowSharePolicy; +import tech.easyflow.common.domain.Result; +import tech.easyflow.common.vo.UploadResVo; +import tech.easyflow.common.web.jsonbody.JsonBody; + +import java.util.Map; + +/** + * 工作流对话匿名分享接口。 + */ +@SaIgnore +@RestController +@RequestMapping("/api/v1/workflowChat/public") +public class WorkflowPublicChatController { + + private final WorkflowPublicChatService publicChatService; + + public WorkflowPublicChatController( + WorkflowPublicChatService publicChatService + ) { + this.publicChatService = publicChatService; + } + + @GetMapping("/descriptor") + public Result> descriptor(HttpServletRequest request) { + return Result.ok(publicChatService.descriptor( + shareKey(request), + visitorId(request) + )); + } + + @PostMapping(value = "/run", produces = MediaType.TEXT_EVENT_STREAM_VALUE) + public SseEmitter run( + @JsonBody("variables") Map variables, + HttpServletRequest request + ) { + return publicChatService.run( + shareKey(request), + visitorId(request), + variables + ); + } + + @GetMapping("/execution") + public Result> execution( + @RequestParam String executeId, + HttpServletRequest request + ) { + return Result.ok(publicChatService.detail( + shareKey(request), + visitorId(request), + executeId + )); + } + + @PostMapping("/cancel") + public Result cancel( + @JsonBody(value = "executeId", required = true) String executeId, + HttpServletRequest request + ) { + return Result.ok(publicChatService.cancel( + shareKey(request), + visitorId(request), + executeId + )); + } + + @PostMapping("/resume") + public Result resume( + @JsonBody(value = "executeId", required = true) String executeId, + @JsonBody("confirmParams") Map confirmParams, + HttpServletRequest request + ) { + publicChatService.resume( + shareKey(request), + visitorId(request), + executeId, + confirmParams + ); + return Result.ok(); + } + + @PostMapping(value = "/upload", produces = MediaType.APPLICATION_JSON_VALUE) + public Result upload( + @RequestParam("file") MultipartFile file, + @RequestParam("parameterName") String parameterName, + HttpServletRequest request + ) { + return Result.ok(publicChatService.upload( + shareKey(request), + visitorId(request), + parameterName, + file + )); + } + + private String shareKey(HttpServletRequest request) { + return request.getHeader(WorkflowSharePolicy.CHAT_SHARE_KEY_HEADER); + } + + private String visitorId(HttpServletRequest request) { + return request.getHeader(WorkflowSharePolicy.CHAT_VISITOR_HEADER); + } +} diff --git a/easyflow-api/easyflow-api-admin/src/main/java/tech/easyflow/admin/controller/ai/WorkflowShareController.java b/easyflow-api/easyflow-api-admin/src/main/java/tech/easyflow/admin/controller/ai/WorkflowShareController.java index 205ff6ac..d47a1827 100644 --- a/easyflow-api/easyflow-api-admin/src/main/java/tech/easyflow/admin/controller/ai/WorkflowShareController.java +++ b/easyflow-api/easyflow-api-admin/src/main/java/tech/easyflow/admin/controller/ai/WorkflowShareController.java @@ -1,6 +1,7 @@ package tech.easyflow.admin.controller.ai; import cn.dev33.satoken.annotation.SaCheckPermission; +import cn.dev33.satoken.annotation.SaIgnore; import jakarta.servlet.http.HttpServletRequest; import org.springframework.web.bind.annotation.GetMapping; import org.springframework.web.bind.annotation.PostMapping; @@ -96,11 +97,10 @@ public class WorkflowShareController { * @return 工作流标识 */ @GetMapping("/resolve") + @SaIgnore public Result> resolveUrlShare(HttpServletRequest request) { - LoginAccount loginAccount = SaTokenUtil.getLoginAccount(); - WorkflowShare share = workflowShareService.resolveChatShare( - request.getHeader(WorkflowSharePolicy.CHAT_SHARE_KEY_HEADER), - loginAccount.getTenantId() + WorkflowShare share = workflowShareService.resolvePublicChatShare( + request.getHeader(WorkflowSharePolicy.CHAT_SHARE_KEY_HEADER) ); return Result.ok(Map.of("workflowId", share.getWorkflowId())); } diff --git a/easyflow-api/easyflow-api-admin/src/main/java/tech/easyflow/admin/service/ai/WorkflowChatEventStream.java b/easyflow-api/easyflow-api-admin/src/main/java/tech/easyflow/admin/service/ai/WorkflowChatEventStream.java index 88b4ee0c..cc997bc8 100644 --- a/easyflow-api/easyflow-api-admin/src/main/java/tech/easyflow/admin/service/ai/WorkflowChatEventStream.java +++ b/easyflow-api/easyflow-api-admin/src/main/java/tech/easyflow/admin/service/ai/WorkflowChatEventStream.java @@ -19,11 +19,17 @@ import org.springframework.stereotype.Service; import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; import javax.annotation.PostConstruct; +import javax.annotation.PreDestroy; import java.io.IOException; +import java.time.Duration; import java.util.LinkedHashMap; import java.util.Map; import java.util.Objects; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicLong; @@ -40,6 +46,15 @@ public class WorkflowChatEventStream { private final ChainExecutor chainExecutor; private final Map sessions = new ConcurrentHashMap<>(); + private final ScheduledExecutorService detachedSessionCleaner = + Executors.newSingleThreadScheduledExecutor(task -> { + Thread thread = new Thread( + task, + "workflow-chat-detached-session-cleaner" + ); + thread.setDaemon(true); + return thread; + }); /** * 创建工作流对话事件流服务。 @@ -59,6 +74,15 @@ public class WorkflowChatEventStream { chainExecutor.addErrorListener(this::onChainError); } + /** + * 关闭断开会话清理线程并释放残留外部资源。 + */ + @PreDestroy + public void shutdown() { + sessions.values().forEach(this::removeSession); + detachedSessionCleaner.shutdownNow(); + } + /** * 启动工作流并返回其 SSE 连接。 * @@ -67,11 +91,53 @@ public class WorkflowChatEventStream { * @return SSE 连接 */ public SseEmitter start(String definitionId, Map variables) { - SseEmitter emitter = new SseEmitter(SSE_TIMEOUT_MILLIS); - StreamSession session = new StreamSession(emitter); - emitter.onTimeout(() -> disconnect(session, "运行连接超时")); - emitter.onError(error -> disconnect(session, "运行连接已断开")); - emitter.onCompletion(() -> removeSession(session)); + return start(definitionId, variables, () -> { + }); + } + + /** + * 启动工作流并在流会话结束时执行清理回调。 + * + * @param definitionId 工作流定义 ID + * @param variables 运行变量 + * @param cleanup 终态、启动失败或连接断开后的幂等清理任务 + * @return SSE 连接 + */ + public SseEmitter start( + String definitionId, + Map variables, + Runnable cleanup + ) { + return start(definitionId, variables, cleanup, Duration.ZERO); + } + + /** + * 启动工作流并将浏览器连接与 Runtime 生命周期分离。 + * + *

浏览器断开后不取消工作流；在保留期内继续监听真实终态并执行清理， + * 超过保留期时由租约兜底释放资源。

+ * + * @param definitionId 工作流定义 ID + * @param variables 运行变量 + * @param cleanup 终态、启动失败或保留期结束后的幂等清理任务 + * @param detachedRetention 浏览器断开后的监听保留时长 + * @return SSE 连接 + */ + public SseEmitter start( + String definitionId, + Map variables, + Runnable cleanup, + Duration detachedRetention + ) { + SseEmitter emitter = createEmitter(); + StreamSession session = new StreamSession( + emitter, + cleanup, + detachedRetention + ); + emitter.onTimeout(() -> detach(session)); + emitter.onError(error -> detach(session)); + emitter.onCompletion(() -> detach(session)); try { chainExecutor.executeAsync( @@ -79,6 +145,9 @@ public class WorkflowChatEventStream { variables, executeId -> { session.attach(executeId); + if (session.cleaned.get()) { + return; + } sessions.put(executeId, session); session.send("execution_started", Map.of( "executeId", executeId @@ -92,6 +161,13 @@ public class WorkflowChatEventStream { return emitter; } + /** + * 创建 SSE 发送器，便于验证连接生命周期。 + */ + SseEmitter createEmitter() { + return new SseEmitter(SSE_TIMEOUT_MILLIS); + } + /** * 将工作流事件转发到对应执行流。 * @@ -162,20 +238,21 @@ public class WorkflowChatEventStream { } /** - * 处理 SSE 连接异常，并取消尚未结束的工作流。 + * 分离已经断开的浏览器传输，不影响工作流 Runtime。 * * @param session 流会话 - * @param message 取消原因 */ - private void disconnect(StreamSession session, String message) { + private void detach(StreamSession session) { if (session == null || session.terminal.get()) { return; } - String executeId = session.executeId; - removeSession(session); - if (executeId != null) { - chainExecutor.cancel(executeId, message); + session.detachTransport(); + if (session.detachedRetention.isZero() + || session.detachedRetention.isNegative()) { + removeSession(session); + return; } + session.scheduleDetachedCleanup(); } /** @@ -187,6 +264,9 @@ public class WorkflowChatEventStream { if (session != null && session.executeId != null) { sessions.remove(session.executeId, session); } + if (session != null) { + session.cleanup(); + } } /** @@ -231,6 +311,11 @@ public class WorkflowChatEventStream { private final SseEmitter emitter; private final AtomicLong sequence = new AtomicLong(); private final AtomicBoolean terminal = new AtomicBoolean(false); + private final AtomicBoolean cleaned = new AtomicBoolean(false); + private final AtomicBoolean connected = new AtomicBoolean(true); + private final Runnable cleanup; + private final Duration detachedRetention; + private volatile ScheduledFuture detachedCleanup; private volatile String executeId; /** @@ -238,8 +323,36 @@ public class WorkflowChatEventStream { * * @param emitter SSE 发送器 */ - private StreamSession(SseEmitter emitter) { + private StreamSession( + SseEmitter emitter, + Runnable cleanup, + Duration detachedRetention + ) { this.emitter = emitter; + this.cleanup = cleanup == null ? () -> { + } : cleanup; + this.detachedRetention = detachedRetention == null + ? Duration.ZERO + : detachedRetention; + } + + /** + * 幂等释放当前流持有的外部资源。 + */ + private void cleanup() { + if (!cleaned.compareAndSet(false, true)) { + return; + } + cancelDetachedCleanup(); + try { + cleanup.run(); + } catch (RuntimeException error) { + log.warn( + "workflow chat stream cleanup failed, executeId={}", + executeId, + error + ); + } } /** @@ -251,6 +364,38 @@ public class WorkflowChatEventStream { this.executeId = executeId; } + /** + * 标记浏览器传输已经断开，后续事件只推进 Runtime 清理。 + */ + private void detachTransport() { + connected.set(false); + } + + /** + * 浏览器断开后按活动租约安排会话兜底清理。 + */ + private synchronized void scheduleDetachedCleanup() { + if (detachedCleanup != null || cleaned.get()) { + return; + } + detachedCleanup = detachedSessionCleaner.schedule( + () -> removeSession(this), + Math.max(1L, detachedRetention.toMillis()), + TimeUnit.MILLISECONDS + ); + } + + /** + * 取消尚未触发的断开会话兜底任务。 + */ + private synchronized void cancelDetachedCleanup() { + if (detachedCleanup == null) { + return; + } + detachedCleanup.cancel(false); + detachedCleanup = null; + } + /** * 处理节点开始事件。 * @@ -424,7 +569,9 @@ public class WorkflowChatEventStream { } send(eventType, data); removeSession(this); - emitter.complete(); + if (connected.compareAndSet(true, false)) { + emitter.complete(); + } } /** @@ -434,6 +581,9 @@ public class WorkflowChatEventStream { * @param data 事件数据 */ private void send(String type, Map data) { + if (!connected.get()) { + return; + } long nextSequence = sequence.incrementAndGet(); Map payload = new LinkedHashMap<>(); payload.put("eventId", executeId + ":" + nextSequence); @@ -453,7 +603,7 @@ public class WorkflowChatEventStream { executeId, error ); - disconnect(this, "运行连接已断开"); + detach(this); } } @@ -468,7 +618,9 @@ public class WorkflowChatEventStream { "message", safeErrorMessage(error) )); removeSession(this); - emitter.completeWithError(error); + if (connected.compareAndSet(true, false)) { + emitter.completeWithError(error); + } } } diff --git a/easyflow-api/easyflow-api-admin/src/main/java/tech/easyflow/admin/service/ai/WorkflowPublicChatAccessGuard.java b/easyflow-api/easyflow-api-admin/src/main/java/tech/easyflow/admin/service/ai/WorkflowPublicChatAccessGuard.java new file mode 100644 index 00000000..a3729aef --- /dev/null +++ b/easyflow-api/easyflow-api-admin/src/main/java/tech/easyflow/admin/service/ai/WorkflowPublicChatAccessGuard.java @@ -0,0 +1,165 @@ +package tech.easyflow.admin.service.ai; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.data.redis.core.StringRedisTemplate; +import org.springframework.data.redis.core.script.DefaultRedisScript; +import org.springframework.stereotype.Component; +import tech.easyflow.common.cache.RedisLockExecutor; +import tech.easyflow.common.web.exceptions.BusinessException; + +import java.math.BigInteger; +import java.time.Duration; +import java.util.List; + +/** + * 工作流匿名分享的限流与活动执行互斥保护。 + */ +@Component +public class WorkflowPublicChatAccessGuard { + + private static final Logger log = LoggerFactory.getLogger( + WorkflowPublicChatAccessGuard.class); + private static final String KEY_PREFIX = "easyflow:workflow-public-share:"; + private static final DefaultRedisScript RATE_LIMIT_SCRIPT; + + static { + RATE_LIMIT_SCRIPT = new DefaultRedisScript<>(); + RATE_LIMIT_SCRIPT.setScriptText( + "local visitor = redis.call('incr', KEYS[1]); " + + "if visitor == 1 then redis.call('pexpire', KEYS[1], ARGV[3]); end; " + + "local share = redis.call('incr', KEYS[2]); " + + "if share == 1 then redis.call('pexpire', KEYS[2], ARGV[3]); end; " + + "if visitor > tonumber(ARGV[1]) or share > tonumber(ARGV[2]) " + + "then return 0 else return 1 end" + ); + RATE_LIMIT_SCRIPT.setResultType(Long.class); + } + + private final StringRedisTemplate redisTemplate; + private final RedisLockExecutor redisLockExecutor; + private final WorkflowPublicShareProperties properties; + + public WorkflowPublicChatAccessGuard( + StringRedisTemplate redisTemplate, + RedisLockExecutor redisLockExecutor, + WorkflowPublicShareProperties properties + ) { + this.redisTemplate = redisTemplate; + this.redisLockExecutor = redisLockExecutor; + this.properties = properties; + } + + /** + * 检查匿名运行固定窗口限流。 + */ + public void checkRun(BigInteger shareId, String visitorDigest) { + checkRate( + shareId, + visitorDigest, + "run", + properties.getRunVisitorLimit(), + properties.getRunShareLimit() + ); + } + + /** + * 检查匿名上传固定窗口限流。 + */ + public void checkUpload(BigInteger shareId, String visitorDigest) { + checkRate( + shareId, + visitorDigest, + "upload", + properties.getUploadVisitorLimit(), + properties.getUploadShareLimit() + ); + } + + /** + * 获取同一分享访客的活动执行锁。 + * + * @return 由 SSE 生命周期显式释放的锁句柄 + */ + public RedisLockExecutor.LockHandle acquireActivity( + BigInteger shareId, + String visitorDigest + ) { + try { + RedisLockExecutor.LockHandle handle = redisLockExecutor.tryAcquire( + KEY_PREFIX + "{" + shareId + "}:active:" + visitorDigest, + Duration.ZERO, + properties.getActiveLease() + ); + if (handle == null) { + throw new BusinessException( + 409, + 40931, + "当前分享访客已有工作流正在运行" + ); + } + return handle; + } catch (BusinessException exception) { + throw exception; + } catch (RuntimeException exception) { + log.error("匿名工作流活动锁暂不可用，shareId={}", shareId, exception); + throw unavailable(exception); + } + } + + /** + * 获取匿名活动执行锁的租约，用作浏览器断开后的监听保留上限。 + */ + public Duration activityLease() { + return properties.getActiveLease(); + } + + private void checkRate( + BigInteger shareId, + String visitorDigest, + String action, + int visitorLimit, + int shareLimit + ) { + String slot = "{" + shareId + "}"; + List keys = List.of( + KEY_PREFIX + slot + ":rate:" + action + ":visitor:" + visitorDigest, + KEY_PREFIX + slot + ":rate:" + action + ":share" + ); + try { + Long allowed = redisTemplate.execute( + RATE_LIMIT_SCRIPT, + keys, + String.valueOf(visitorLimit), + String.valueOf(shareLimit), + String.valueOf(properties.getRateWindow().toMillis()) + ); + if (allowed == null) { + throw unavailable(new IllegalStateException( + "Redis 未返回匿名工作流限流结果")); + } + if (!Long.valueOf(1L).equals(allowed)) { + throw new BusinessException( + 429, + 42931, + "匿名工作流请求过于频繁，请稍后重试" + ); + } + } catch (BusinessException exception) { + throw exception; + } catch (RuntimeException exception) { + log.error("匿名工作流限流暂不可用，shareId={}, action={}", + shareId, action, exception); + throw unavailable(exception); + } + } + + private BusinessException unavailable(RuntimeException cause) { + return new BusinessException( + 503, + 50331, + "匿名工作流保护服务暂不可用，请稍后重试", + cause + ); + } +} diff --git a/easyflow-api/easyflow-api-admin/src/main/java/tech/easyflow/admin/service/ai/WorkflowPublicChatContext.java b/easyflow-api/easyflow-api-admin/src/main/java/tech/easyflow/admin/service/ai/WorkflowPublicChatContext.java new file mode 100644 index 00000000..bd660384 --- /dev/null +++ b/easyflow-api/easyflow-api-admin/src/main/java/tech/easyflow/admin/service/ai/WorkflowPublicChatContext.java @@ -0,0 +1,17 @@ +package tech.easyflow.admin.service.ai; + +import tech.easyflow.ai.entity.Workflow; +import tech.easyflow.ai.entity.WorkflowShare; +import tech.easyflow.common.entity.LoginAccount; + +/** + * 完成匿名分享边界校验后的运行上下文。 + */ +public record WorkflowPublicChatContext( + WorkflowShare share, + Workflow workflow, + LoginAccount creator, + String shareKey, + String visitorDigest +) { +} diff --git a/easyflow-api/easyflow-api-admin/src/main/java/tech/easyflow/admin/service/ai/WorkflowPublicChatContextResolver.java b/easyflow-api/easyflow-api-admin/src/main/java/tech/easyflow/admin/service/ai/WorkflowPublicChatContextResolver.java new file mode 100644 index 00000000..ca984154 --- /dev/null +++ b/easyflow-api/easyflow-api-admin/src/main/java/tech/easyflow/admin/service/ai/WorkflowPublicChatContextResolver.java @@ -0,0 +1,121 @@ +package tech.easyflow.admin.service.ai; + +import com.mybatisflex.core.tenant.TenantManager; +import org.springframework.stereotype.Service; +import tech.easyflow.ai.entity.Workflow; +import tech.easyflow.ai.entity.WorkflowShare; +import tech.easyflow.ai.enums.PublishStatus; +import tech.easyflow.ai.service.WorkflowService; +import tech.easyflow.ai.service.WorkflowShareService; +import tech.easyflow.ai.share.WorkflowSharePolicy; +import tech.easyflow.common.constant.enums.EnumDataStatus; +import tech.easyflow.common.entity.LoginAccount; +import tech.easyflow.common.web.exceptions.BusinessException; +import tech.easyflow.system.entity.SysAccount; +import tech.easyflow.system.service.SysAccountService; + +import java.util.Objects; +import java.util.regex.Pattern; + +/** + * 解析并校验工作流匿名分享上下文。 + */ +@Service +public class WorkflowPublicChatContextResolver { + + private static final Pattern VISITOR_PATTERN = Pattern.compile("[a-f0-9]{32}"); + + private final WorkflowShareService shareService; + private final WorkflowService workflowService; + private final SysAccountService accountService; + + public WorkflowPublicChatContextResolver( + WorkflowShareService shareService, + WorkflowService workflowService, + SysAccountService accountService + ) { + this.shareService = shareService; + this.workflowService = workflowService; + this.accountService = accountService; + } + + /** + * 解析新运行、恢复与上传所需的当前有效上下文。 + */ + public WorkflowPublicChatContext resolveActive( + String shareKey, + String visitorId + ) { + String normalizedVisitor = requireVisitor(visitorId); + WorkflowShare share = shareService.resolvePublicChatShare(shareKey); + Workflow workflow = TenantManager.withoutTenantCondition( + () -> workflowService.getPublishedById(share.getWorkflowId())); + if (!isStrictlyPublished(workflow) + || !Objects.equals(share.getTenantId(), workflow.getTenantId())) { + throw new BusinessException(409, 409, "工作流尚未发布或已下线"); + } + SysAccount account = TenantManager.withoutTenantCondition( + () -> accountService.getById(share.getCreatedBy())); + if (account == null + || !EnumDataStatus.AVAILABLE.getCode().equals(account.getStatus()) + || !Objects.equals(share.getTenantId(), account.getTenantId())) { + throw new BusinessException( + 403, + 40331, + "工作流分享创建者账号当前不可用" + ); + } + LoginAccount creator = account.toLoginAccount(); + return new WorkflowPublicChatContext( + share, + workflow, + creator, + shareKey, + WorkflowSharePolicy.hashChatVisitor( + shareKey, + normalizedVisitor + ) + ); + } + + /** + * 解析已发起执行的详情与取消所需历史上下文。 + */ + public WorkflowPublicChatContext resolveHistorical( + String shareKey, + String visitorId + ) { + String normalizedVisitor = requireVisitor(visitorId); + WorkflowShare share = shareService.resolveHistoricalChatShare(shareKey); + return new WorkflowPublicChatContext( + share, + null, + null, + shareKey, + WorkflowSharePolicy.hashChatVisitor( + shareKey, + normalizedVisitor + ) + ); + } + + private String requireVisitor(String visitorId) { + String normalized = visitorId == null ? "" : visitorId.trim(); + if (!VISITOR_PATTERN.matcher(normalized).matches()) { + throw new BusinessException( + 400, + 40031, + "工作流分享访客标识无效" + ); + } + return normalized; + } + + private boolean isStrictlyPublished(Workflow workflow) { + return workflow != null + && PublishStatus.PUBLISHED.getCode().equals( + workflow.getPublishStatus()) + && workflow.getPublishedSnapshotJson() != null + && !workflow.getPublishedSnapshotJson().isEmpty(); + } +} diff --git a/easyflow-api/easyflow-api-admin/src/main/java/tech/easyflow/admin/service/ai/WorkflowPublicChatService.java b/easyflow-api/easyflow-api-admin/src/main/java/tech/easyflow/admin/service/ai/WorkflowPublicChatService.java new file mode 100644 index 00000000..da618e33 --- /dev/null +++ b/easyflow-api/easyflow-api-admin/src/main/java/tech/easyflow/admin/service/ai/WorkflowPublicChatService.java @@ -0,0 +1,338 @@ +package tech.easyflow.admin.service.ai; + +import com.easyagents.flow.core.chain.ChainStatus; +import com.easyagents.flow.core.chain.ChainState; +import com.easyagents.flow.core.chain.runtime.ChainExecutor; +import com.mybatisflex.core.query.QueryWrapper; +import com.mybatisflex.core.tenant.TenantManager; +import org.springframework.stereotype.Service; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.web.multipart.MultipartFile; +import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; +import tech.easyflow.ai.easyagentsflow.entity.WorkflowCheckStage; +import tech.easyflow.ai.easyagentsflow.service.WorkflowCheckService; +import tech.easyflow.ai.easyagentsflow.service.WorkflowRunningParameterResolver; +import tech.easyflow.ai.easyagentsflow.support.PublishedWorkflowDefinitionIds; +import tech.easyflow.ai.entity.WorkflowExecResult; +import tech.easyflow.ai.entity.WorkflowExecStep; +import tech.easyflow.ai.service.WorkflowExecResultService; +import tech.easyflow.ai.service.WorkflowExecStepService; +import tech.easyflow.ai.utils.WorkFlowUtil; +import tech.easyflow.common.cache.RedisLockExecutor; +import tech.easyflow.common.constant.Constants; +import tech.easyflow.common.vo.UploadResVo; +import tech.easyflow.common.web.exceptions.BusinessException; + +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Objects; + +/** + * 工作流匿名分享对话应用服务。 + */ +@Service +public class WorkflowPublicChatService { + + private static final Logger log = + LoggerFactory.getLogger(WorkflowPublicChatService.class); + + private final WorkflowPublicChatContextResolver contextResolver; + private final WorkflowCheckService workflowCheckService; + private final WorkflowRunningParameterResolver parameterResolver; + private final WorkflowPublicChatUploadService uploadService; + private final WorkflowPublicChatAccessGuard accessGuard; + private final WorkflowChatEventStream eventStream; + private final ChainExecutor chainExecutor; + private final WorkflowExecResultService execResultService; + private final WorkflowExecStepService execStepService; + + public WorkflowPublicChatService( + WorkflowPublicChatContextResolver contextResolver, + WorkflowCheckService workflowCheckService, + WorkflowRunningParameterResolver parameterResolver, + WorkflowPublicChatUploadService uploadService, + WorkflowPublicChatAccessGuard accessGuard, + WorkflowChatEventStream eventStream, + ChainExecutor chainExecutor, + WorkflowExecResultService execResultService, + WorkflowExecStepService execStepService + ) { + this.contextResolver = contextResolver; + this.workflowCheckService = workflowCheckService; + this.parameterResolver = parameterResolver; + this.uploadService = uploadService; + this.accessGuard = accessGuard; + this.eventStream = eventStream; + this.chainExecutor = chainExecutor; + this.execResultService = execResultService; + this.execStepService = execStepService; + } + + /** + * 获取匿名分享的发布工作流描述。 + */ + public Map descriptor(String shareKey, String visitorId) { + WorkflowPublicChatContext context = contextResolver.resolveActive( + shareKey, visitorId); + checkWorkflow(context); + Map descriptor = parameterResolver + .buildRunningParametersView(context.workflow()); + if (descriptor == null) { + throw new BusinessException("工作流输入配置无法解析"); + } + descriptor.put("workflowId", context.workflow().getId()); + descriptor.put("publishStatus", context.workflow().getPublishStatus()); + descriptor.put("shareable", false); + return descriptor; + } + + /** + * 启动匿名分享工作流并返回 SSE。 + */ + public SseEmitter run( + String shareKey, + String visitorId, + Map variables + ) { + WorkflowPublicChatContext context = contextResolver.resolveActive( + shareKey, visitorId); + accessGuard.checkRun( + context.share().getId(), + context.visitorDigest() + ); + checkWorkflow(context); + Map normalized = parameterResolver + .normalizeRuntimeVariables( + context.workflow().getContent(), + variables + ); + uploadService.assertOwnedUploads(context, normalized); + normalized.put(Constants.LOGIN_USER_KEY, context.creator()); + normalized.put( + WorkFlowUtil.CREATED_KEY_MEMORY_KEY, + WorkFlowUtil.publicChatShareCreatedKey( + context.share().getId()) + ); + normalized.put( + WorkFlowUtil.CREATED_BY_MEMORY_KEY, + context.visitorDigest() + ); + + RedisLockExecutor.LockHandle activity = accessGuard.acquireActivity( + context.share().getId(), + context.visitorDigest() + ); + try { + return eventStream.start( + PublishedWorkflowDefinitionIds.published( + context.workflow().getId().toString()), + normalized, + activity::release, + accessGuard.activityLease() + ); + } catch (RuntimeException | Error error) { + activity.release(); + throw error; + } + } + + /** + * 获取当前匿名访客发起的执行详情。 + */ + public Map detail( + String shareKey, + String visitorId, + String executeId + ) { + WorkflowPublicChatContext context = contextResolver.resolveHistorical( + shareKey, visitorId); + WorkflowExecResult record = assertExecutionOwnership( + context, executeId); + List steps = TenantManager.withoutTenantCondition( + () -> execStepService.list( + QueryWrapper.create() + .eq(WorkflowExecStep::getRecordId, record.getId()) + .orderBy(WorkflowExecStep::getStartTime, true) + )); + return buildExecutionDetail(record, steps, runtimeView(executeId)); + } + + /** + * 取消当前匿名访客发起的执行。 + */ + public boolean cancel( + String shareKey, + String visitorId, + String executeId + ) { + WorkflowPublicChatContext context = contextResolver.resolveHistorical( + shareKey, visitorId); + assertExecutionOwnership(context, executeId); + return chainExecutor.cancel(executeId, "匿名访客已中止运行"); + } + + /** + * 恢复当前有效分享访客等待确认的执行。 + */ + public void resume( + String shareKey, + String visitorId, + String executeId, + Map confirmParams + ) { + WorkflowPublicChatContext context = contextResolver.resolveActive( + shareKey, visitorId); + WorkflowExecResult record = assertExecutionOwnership( + context, executeId); + if (isTerminal(record.getStatus())) { + throw new BusinessException("当前工作流执行已结束"); + } + chainExecutor.resumeAsync( + executeId, + confirmParams == null + ? new LinkedHashMap<>() + : new LinkedHashMap<>(confirmParams) + ); + } + + /** + * 上传当前发布快照声明的匿名输入文件。 + */ + public UploadResVo upload( + String shareKey, + String visitorId, + String parameterName, + MultipartFile file + ) { + WorkflowPublicChatContext context = contextResolver.resolveActive( + shareKey, visitorId); + return uploadService.upload(context, parameterName, file); + } + + private void checkWorkflow(WorkflowPublicChatContext context) { + TenantManager.withoutTenantCondition(() -> { + workflowCheckService.checkOrThrow( + context.workflow().getContent(), + WorkflowCheckStage.PRE_EXECUTE, + context.workflow().getId() + ); + return null; + }); + } + + private WorkflowExecResult assertExecutionOwnership( + WorkflowPublicChatContext context, + String executeId + ) { + if (executeId == null || executeId.isBlank()) { + throw new BusinessException("执行ID不能为空"); + } + WorkflowExecResult record = TenantManager.withoutTenantCondition( + () -> execResultService.getByExecKey(executeId)); + if (record == null) { + throw new BusinessException("工作流执行记录不存在，请稍后重试"); + } + String expectedSource = WorkFlowUtil.publicChatShareCreatedKey( + context.share().getId()); + if (!Objects.equals(expectedSource, record.getCreatedKey()) + || !Objects.equals( + context.visitorDigest(), + record.getCreatedBy()) + || !Objects.equals( + context.share().getWorkflowId(), + record.getWorkflowId())) { + throw new BusinessException( + 403, + 40333, + "无权限访问当前工作流执行记录" + ); + } + return record; + } + + private boolean isTerminal(Integer status) { + return status != null + && (status == ChainStatus.SUCCEEDED.getValue() + || status == ChainStatus.FAILED.getValue() + || status == ChainStatus.CANCELLED.getValue()); + } + + private Map buildExecutionDetail( + WorkflowExecResult record, + List steps, + Map runtime + ) { + List> stepViews = new ArrayList<>(steps.size()); + for (WorkflowExecStep step : steps) { + Map view = new LinkedHashMap<>(); + view.put("id", step.getId()); + view.put("attemptKey", step.getExecKey()); + view.put("nodeId", step.getNodeId()); + view.put("nodeName", step.getNodeName()); + view.put("input", step.getInput()); + view.put("output", step.getOutput()); + view.put("status", step.getStatus()); + view.put("errorInfo", step.getErrorInfo()); + view.put("startTime", step.getStartTime()); + view.put("endTime", step.getEndTime()); + view.put("execTime", step.getExecTime()); + stepViews.add(view); + } + + Map recordView = new LinkedHashMap<>(); + recordView.put("executeId", record.getExecKey()); + recordView.put("workflowId", record.getWorkflowId()); + recordView.put("title", record.getTitle()); + recordView.put("status", record.getStatus()); + recordView.put("input", record.getInput()); + recordView.put("output", record.getOutput()); + recordView.put("errorInfo", record.getErrorInfo()); + recordView.put("startTime", record.getStartTime()); + recordView.put("endTime", record.getEndTime()); + recordView.put("execTime", record.getExecTime()); + + Map detail = new LinkedHashMap<>(); + detail.put("record", recordView); + detail.put("steps", stepViews); + detail.put("runtime", runtime); + return detail; + } + + /** + * 构建刷新恢复所需的最小 Runtime 视图。 + */ + private Map runtimeView(String executeId) { + try { + ChainState state = chainExecutor.getChainStateRepository() + .load(executeId); + if (state == null || state.getStatus() == null) { + return Map.of(); + } + Map view = new LinkedHashMap<>(); + view.put("status", state.getStatus().name()); + view.put("statusValue", state.getStatus().getValue()); + view.put("message", state.getMessage()); + if (state.getStatus() == ChainStatus.SUSPEND) { + view.put("parameters", state.getSuspendForParameters()); + } + if (state.getStatus() == ChainStatus.SUCCEEDED) { + view.put( + "output", + WorkflowChatEventStream.visibleFinalOutput( + state.getExecuteResult()) + ); + } + return view; + } catch (RuntimeException error) { + log.warn( + "failed to load public workflow runtime state, executeId={}", + executeId, + error + ); + return Map.of(); + } + } +} diff --git a/easyflow-api/easyflow-api-admin/src/main/java/tech/easyflow/admin/service/ai/WorkflowPublicChatUploadService.java b/easyflow-api/easyflow-api-admin/src/main/java/tech/easyflow/admin/service/ai/WorkflowPublicChatUploadService.java new file mode 100644 index 00000000..c0d77483 --- /dev/null +++ b/easyflow-api/easyflow-api-admin/src/main/java/tech/easyflow/admin/service/ai/WorkflowPublicChatUploadService.java @@ -0,0 +1,325 @@ +package tech.easyflow.admin.service.ai; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.data.redis.core.StringRedisTemplate; +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.ai.share.WorkflowSharePolicy; +import tech.easyflow.common.filestorage.FileStorageService; +import tech.easyflow.common.vo.UploadResVo; +import tech.easyflow.common.web.exceptions.BusinessException; + +import java.time.Duration; +import java.util.Collection; +import java.util.Locale; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.TimeUnit; + +/** + * 工作流匿名分享的隔离上传与运行引用校验。 + */ +@Service +public class WorkflowPublicChatUploadService { + + private static final Logger log = LoggerFactory.getLogger( + WorkflowPublicChatUploadService.class); + private static final long FILE_MAX_SIZE = 100L * 1024L * 1024L; + private static final long IMAGE_MAX_SIZE = 10L * 1024L * 1024L; + private static final Set IMAGE_MIME_TYPES = Set.of( + "image/bmp", "image/gif", "image/jpeg", "image/png", "image/webp"); + private static final Set IMAGE_EXTENSIONS = Set.of( + "bmp", "gif", "jpeg", "jpg", "png", "webp"); + private static final String GRANT_PREFIX = "easyflow:workflow-public-share:upload:"; + + private final WorkflowRunningParameterResolver parameterResolver; + private final WorkflowPublicChatAccessGuard accessGuard; + private final WorkflowPublicShareProperties properties; + private final StringRedisTemplate redisTemplate; + private final FileStorageService storageService; + + public WorkflowPublicChatUploadService( + WorkflowRunningParameterResolver parameterResolver, + WorkflowPublicChatAccessGuard accessGuard, + WorkflowPublicShareProperties properties, + StringRedisTemplate redisTemplate, + @Qualifier("default") FileStorageService storageService + ) { + this.parameterResolver = parameterResolver; + this.accessGuard = accessGuard; + this.properties = properties; + this.redisTemplate = redisTemplate; + this.storageService = storageService; + } + + /** + * 上传发布快照声明的文件或图片参数。 + */ + public UploadResVo upload( + WorkflowPublicChatContext context, + String parameterName, + MultipartFile file + ) { + String normalizedName = requireParameterName(parameterName); + String contentType = resolveUploadContentType(context, normalizedName); + validateFile(file, contentType); + accessGuard.checkUpload( + context.share().getId(), + context.visitorDigest() + ); + + String path = storageService.save( + file, + "workflow-chat-share/" + context.share().getId() + + "/" + context.visitorDigest() + ); + if (!StringUtils.hasText(path)) { + throw new BusinessException(503, 50332, "匿名文件上传失败，请稍后重试"); + } + try { + redisTemplate.opsForValue().set( + grantKey(context, normalizedName, path), + contentType, + grantTtl(context).toMillis(), + TimeUnit.MILLISECONDS + ); + } catch (RuntimeException exception) { + try { + storageService.delete(path); + } catch (RuntimeException cleanupError) { + log.warn("匿名上传授权写入失败后清理文件失败，path={}", + path, cleanupError); + } + throw new BusinessException( + 503, + 50332, + "匿名上传保护服务暂不可用，请稍后重试", + exception + ); + } + + UploadResVo response = new UploadResVo(); + response.setPath(path); + return response; + } + + /** + * 校验公开运行引用的上传文件均属于当前分享访客和参数。 + */ + public void assertOwnedUploads( + WorkflowPublicChatContext context, + Map variables + ) { + Map uploadFields = resolveUploadFields(context); + for (Map.Entry entry : uploadFields.entrySet()) { + Object value = variables.get(entry.getKey()); + if (value == null) { + continue; + } + if ("image".equals(entry.getValue())) { + assertOwnedImage(context, entry.getKey(), value); + } else { + assertOwnedFiles(context, entry.getKey(), value); + } + } + } + + private void assertOwnedImage( + WorkflowPublicChatContext context, + String parameterName, + Object value + ) { + if (!(value instanceof Map image)) { + throw invalidUploadReference(parameterName); + } + String sourceType = trim(image.get("sourceType")); + if ("url".equals(sourceType)) { + String url = trim(image.get("url")); + if (isHttpUrl(url)) { + return; + } + throw invalidUploadReference(parameterName); + } + if (!"upload".equals(sourceType)) { + throw invalidUploadReference(parameterName); + } + assertGrant( + context, + parameterName, + trim(image.get("filePath")), + "image" + ); + } + + private void assertOwnedFiles( + WorkflowPublicChatContext context, + String parameterName, + Object value + ) { + if (!(value instanceof Collection files)) { + throw invalidUploadReference(parameterName); + } + for (Object item : files) { + if (!(item instanceof Map file)) { + throw invalidUploadReference(parameterName); + } + assertGrant( + context, + parameterName, + trim(file.get("filePath")), + "file" + ); + } + } + + private void assertGrant( + WorkflowPublicChatContext context, + String parameterName, + String path, + String expectedContentType + ) { + if (!StringUtils.hasText(path)) { + throw invalidUploadReference(parameterName); + } + try { + String grantedContentType = redisTemplate.opsForValue().get( + grantKey(context, parameterName, path)); + if (!expectedContentType.equals(grantedContentType)) { + throw invalidUploadReference(parameterName); + } + } catch (BusinessException exception) { + throw exception; + } catch (RuntimeException exception) { + throw new BusinessException( + 503, + 50332, + "匿名上传保护服务暂不可用，请稍后重试", + exception + ); + } + } + + private String resolveUploadContentType( + WorkflowPublicChatContext context, + String parameterName + ) { + String contentType = resolveUploadFields(context).get(parameterName); + if (contentType == null) { + throw new BusinessException( + 400, + 40032, + "当前发布工作流未声明该上传参数" + ); + } + return contentType; + } + + @SuppressWarnings("unchecked") + private Map resolveUploadFields( + WorkflowPublicChatContext context + ) { + Map descriptor = parameterResolver + .buildRunningParametersView(context.workflow()); + if (descriptor == null) { + throw new BusinessException("工作流输入配置无法解析"); + } + Map fields = new java.util.LinkedHashMap<>(); + Object rawSchema = descriptor.get("startFormSchema"); + if (!(rawSchema instanceof Collection schema)) { + return fields; + } + for (Object item : schema) { + if (!(item instanceof Map field)) { + continue; + } + String name = trim(field.get("key")); + String contentType = trim(field.get("contentType")); + if (StringUtils.hasText(name) + && ("file".equals(contentType) + || "image".equals(contentType))) { + fields.put(name, contentType); + } + } + return fields; + } + + private void validateFile(MultipartFile file, String contentType) { + if (file == null || file.isEmpty()) { + throw new BusinessException("上传文件不能为空"); + } + long maxSize = "image".equals(contentType) + ? IMAGE_MAX_SIZE + : FILE_MAX_SIZE; + if (file.getSize() > maxSize) { + throw new BusinessException( + "image".equals(contentType) + ? "单张图片不能超过 10 MiB" + : "单个文件不能超过 100 MiB" + ); + } + if (!"image".equals(contentType)) { + return; + } + String mimeType = trim(file.getContentType()).toLowerCase(Locale.ROOT); + String filename = trim(file.getOriginalFilename()); + int dot = filename.lastIndexOf('.'); + String extension = dot < 0 + ? "" + : filename.substring(dot + 1).toLowerCase(Locale.ROOT); + if (!IMAGE_MIME_TYPES.contains(mimeType) + && !IMAGE_EXTENSIONS.contains(extension)) { + throw new BusinessException("仅支持 PNG、JPEG、WebP、GIF、BMP 图片"); + } + } + + private Duration grantTtl(WorkflowPublicChatContext context) { + long expiresIn = context.share().getExpiresAt().getTime() + - System.currentTimeMillis(); + long ttl = Math.min( + properties.getUploadGrantTtl().toMillis(), + expiresIn + ); + return Duration.ofMillis(Math.max(1L, ttl)); + } + + private String grantKey( + WorkflowPublicChatContext context, + String parameterName, + String path + ) { + return GRANT_PREFIX + "{" + context.share().getId() + "}:" + + context.visitorDigest() + ":" + + WorkflowSharePolicy.hashShareKey(parameterName) + ":" + + WorkflowSharePolicy.hashShareKey(path); + } + + private String requireParameterName(String value) { + String normalized = value == null ? "" : value.trim(); + if (!StringUtils.hasText(normalized)) { + throw new BusinessException("上传参数名不能为空"); + } + return normalized; + } + + private String trim(Object value) { + return value == null ? "" : String.valueOf(value).trim(); + } + + private boolean isHttpUrl(String value) { + String normalized = value == null ? "" : value.toLowerCase(Locale.ROOT); + return normalized.startsWith("http://") + || normalized.startsWith("https://"); + } + + private BusinessException invalidUploadReference(String parameterName) { + return new BusinessException( + 403, + 40332, + "上传参数 " + parameterName + " 不属于当前分享访客" + ); + } +} diff --git a/easyflow-api/easyflow-api-admin/src/main/java/tech/easyflow/admin/service/ai/WorkflowPublicShareProperties.java b/easyflow-api/easyflow-api-admin/src/main/java/tech/easyflow/admin/service/ai/WorkflowPublicShareProperties.java new file mode 100644 index 00000000..65cc725a --- /dev/null +++ b/easyflow-api/easyflow-api-admin/src/main/java/tech/easyflow/admin/service/ai/WorkflowPublicShareProperties.java @@ -0,0 +1,92 @@ +package tech.easyflow.admin.service.ai; + +import org.springframework.boot.context.properties.ConfigurationProperties; +import org.springframework.stereotype.Component; + +import java.time.Duration; + +/** + * 工作流匿名分享运行保护参数。 + */ +@Component +@ConfigurationProperties(prefix = "easyflow.workflow.public-share") +public class WorkflowPublicShareProperties { + + private Duration rateWindow = Duration.ofMinutes(1); + private int runVisitorLimit = 5; + private int runShareLimit = 60; + private int uploadVisitorLimit = 10; + private int uploadShareLimit = 60; + private Duration activeLease = Duration.ofMinutes(35); + private Duration uploadGrantTtl = Duration.ofDays(7); + + public Duration getRateWindow() { + return rateWindow; + } + + public void setRateWindow(Duration rateWindow) { + this.rateWindow = requirePositive(rateWindow, "rateWindow"); + } + + public int getRunVisitorLimit() { + return runVisitorLimit; + } + + public void setRunVisitorLimit(int runVisitorLimit) { + this.runVisitorLimit = requirePositive(runVisitorLimit, "runVisitorLimit"); + } + + public int getRunShareLimit() { + return runShareLimit; + } + + public void setRunShareLimit(int runShareLimit) { + this.runShareLimit = requirePositive(runShareLimit, "runShareLimit"); + } + + public int getUploadVisitorLimit() { + return uploadVisitorLimit; + } + + public void setUploadVisitorLimit(int uploadVisitorLimit) { + this.uploadVisitorLimit = requirePositive(uploadVisitorLimit, "uploadVisitorLimit"); + } + + public int getUploadShareLimit() { + return uploadShareLimit; + } + + public void setUploadShareLimit(int uploadShareLimit) { + this.uploadShareLimit = requirePositive(uploadShareLimit, "uploadShareLimit"); + } + + public Duration getActiveLease() { + return activeLease; + } + + public void setActiveLease(Duration activeLease) { + this.activeLease = requirePositive(activeLease, "activeLease"); + } + + public Duration getUploadGrantTtl() { + return uploadGrantTtl; + } + + public void setUploadGrantTtl(Duration uploadGrantTtl) { + this.uploadGrantTtl = requirePositive(uploadGrantTtl, "uploadGrantTtl"); + } + + private static int requirePositive(int value, String name) { + if (value <= 0) { + throw new IllegalArgumentException(name + " 必须大于 0"); + } + return value; + } + + private static Duration requirePositive(Duration value, String name) { + if (value == null || value.isZero() || value.isNegative()) { + throw new IllegalArgumentException(name + " 必须大于 0"); + } + return value; + } +} diff --git a/easyflow-api/easyflow-api-admin/src/test/java/tech/easyflow/admin/controller/ai/WorkflowShareControllerTest.java b/easyflow-api/easyflow-api-admin/src/test/java/tech/easyflow/admin/controller/ai/WorkflowShareControllerTest.java index edba644a..671f2390 100644 --- a/easyflow-api/easyflow-api-admin/src/test/java/tech/easyflow/admin/controller/ai/WorkflowShareControllerTest.java +++ b/easyflow-api/easyflow-api-admin/src/test/java/tech/easyflow/admin/controller/ai/WorkflowShareControllerTest.java @@ -3,17 +3,50 @@ package tech.easyflow.admin.controller.ai; import jakarta.servlet.http.HttpServletRequest; import org.testng.Assert; import org.testng.annotations.Test; +import tech.easyflow.ai.entity.WorkflowShare; +import tech.easyflow.ai.service.WorkflowShareService; +import tech.easyflow.ai.share.WorkflowSharePolicy; +import java.lang.reflect.Field; +import java.math.BigInteger; import java.lang.reflect.Method; import java.lang.reflect.Proxy; import java.util.Locale; import java.util.Map; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + /** * {@link WorkflowShareController} 分享地址构建测试。 */ public class WorkflowShareControllerTest { + /** + * 验证分享解析仅依赖分享密钥，不读取当前浏览器登录租户。 + */ + @Test + public void shouldResolvePublicChatShareWithoutLoginContext() + throws Exception { + WorkflowShareService shareService = mock(WorkflowShareService.class); + WorkflowShare share = new WorkflowShare(); + share.setWorkflowId(BigInteger.valueOf(11)); + when(shareService.resolvePublicChatShare("share-key")) + .thenReturn(share); + WorkflowShareController controller = new WorkflowShareController(); + setField(controller, "workflowShareService", shareService); + + BigInteger workflowId = controller.resolveUrlShare(request(Map.of( + WorkflowSharePolicy.CHAT_SHARE_KEY_HEADER + .toLowerCase(Locale.ROOT), + "share-key" + ))).getData().get("workflowId"); + + Assert.assertEquals(workflowId, BigInteger.valueOf(11)); + verify(shareService).resolvePublicChatShare("share-key"); + } + /** * 验证分享地址保留前端部署基路径。 * @@ -120,4 +153,11 @@ public class WorkflowShareControllerTest { } return 0D; } + + private void setField(Object target, String name, Object value) + throws Exception { + Field field = target.getClass().getDeclaredField(name); + field.setAccessible(true); + field.set(target, value); + } } diff --git a/easyflow-api/easyflow-api-admin/src/test/java/tech/easyflow/admin/service/ai/WorkflowChatEventStreamTest.java b/easyflow-api/easyflow-api-admin/src/test/java/tech/easyflow/admin/service/ai/WorkflowChatEventStreamTest.java index f29a8541..7c3afc98 100644 --- a/easyflow-api/easyflow-api-admin/src/test/java/tech/easyflow/admin/service/ai/WorkflowChatEventStreamTest.java +++ b/easyflow-api/easyflow-api-admin/src/test/java/tech/easyflow/admin/service/ai/WorkflowChatEventStreamTest.java @@ -4,13 +4,19 @@ import com.easyagents.flow.core.chain.ChainConsts; import com.easyagents.flow.core.chain.runtime.ChainExecutor; import org.testng.Assert; import org.testng.annotations.Test; +import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; +import java.time.Duration; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.Consumer; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.never; import static org.mockito.Mockito.verify; @@ -69,4 +75,81 @@ public class WorkflowChatEventStreamTest { WorkflowChatEventStream.visibleFinalOutput(null).isEmpty() ); } + + /** + * 验证工作流启动异常也会释放匿名活动执行租约。 + */ + @Test + public void shouldCleanupExternalResourceWhenStartFails() { + ChainExecutor chainExecutor = mock(ChainExecutor.class); + doThrow(new IllegalStateException("start failed")) + .when(chainExecutor) + .executeAsync(any(), any(), any()); + WorkflowChatEventStream eventStream = + new WorkflowChatEventStream(chainExecutor); + AtomicInteger cleanupCount = new AtomicInteger(); + + Assert.expectThrows( + IllegalStateException.class, + () -> eventStream.start( + "definition", + Map.of(), + cleanupCount::incrementAndGet + ) + ); + + Assert.assertEquals(cleanupCount.get(), 1); + } + + /** + * 验证浏览器断开只分离 SSE，不取消仍在运行的工作流。 + */ + @Test + public void shouldKeepRuntimeRunningWhenBrowserDisconnects() { + ChainExecutor chainExecutor = mock(ChainExecutor.class); + doAnswer(invocation -> { + @SuppressWarnings("unchecked") + Consumer beforeStart = invocation.getArgument(2); + beforeStart.accept("execution-1"); + return "execution-1"; + }).when(chainExecutor).executeAsync(any(), any(), any()); + CapturingSseEmitter emitter = new CapturingSseEmitter(); + WorkflowChatEventStream eventStream = + new WorkflowChatEventStream(chainExecutor) { + @Override + SseEmitter createEmitter() { + return emitter; + } + }; + AtomicInteger cleanupCount = new AtomicInteger(); + + eventStream.start( + "definition", + Map.of(), + cleanupCount::incrementAndGet, + Duration.ofMinutes(35) + ); + emitter.disconnect(); + + verify(chainExecutor, never()).cancel(any(), any()); + Assert.assertEquals(cleanupCount.get(), 0); + + eventStream.shutdown(); + Assert.assertEquals(cleanupCount.get(), 1); + } + + private static final class CapturingSseEmitter extends SseEmitter { + + private Runnable completion; + + @Override + public synchronized void onCompletion(Runnable callback) { + this.completion = callback; + } + + private void disconnect() { + Assert.assertNotNull(completion); + completion.run(); + } + } } diff --git a/easyflow-api/easyflow-api-admin/src/test/java/tech/easyflow/admin/service/ai/WorkflowPublicChatAccessGuardTest.java b/easyflow-api/easyflow-api-admin/src/test/java/tech/easyflow/admin/service/ai/WorkflowPublicChatAccessGuardTest.java new file mode 100644 index 00000000..e69b681f --- /dev/null +++ b/easyflow-api/easyflow-api-admin/src/test/java/tech/easyflow/admin/service/ai/WorkflowPublicChatAccessGuardTest.java @@ -0,0 +1,86 @@ +package tech.easyflow.admin.service.ai; + +import org.springframework.data.redis.core.StringRedisTemplate; +import org.springframework.data.redis.core.script.DefaultRedisScript; +import org.testng.Assert; +import org.testng.annotations.Test; +import tech.easyflow.common.cache.RedisLockExecutor; +import tech.easyflow.common.web.exceptions.BusinessException; + +import java.math.BigInteger; +import java.time.Duration; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyList; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +/** + * {@link WorkflowPublicChatAccessGuard} Redis 失败关闭测试。 + */ +public class WorkflowPublicChatAccessGuardTest { + + @Test + public void shouldExposeDocumentedProtectionDefaults() { + WorkflowPublicShareProperties properties = + new WorkflowPublicShareProperties(); + + Assert.assertEquals(properties.getRunVisitorLimit(), 5); + Assert.assertEquals(properties.getRunShareLimit(), 60); + Assert.assertEquals(properties.getUploadVisitorLimit(), 10); + Assert.assertEquals(properties.getUploadShareLimit(), 60); + Assert.assertEquals(properties.getRateWindow(), Duration.ofMinutes(1)); + Assert.assertEquals(properties.getActiveLease(), Duration.ofMinutes(35)); + } + + @Test + public void shouldReturn429WhenFixedWindowIsExceeded() { + StringRedisTemplate redisTemplate = mock(StringRedisTemplate.class); + RedisLockExecutor lockExecutor = mock(RedisLockExecutor.class); + when(redisTemplate.execute( + any(DefaultRedisScript.class), + anyList(), + anyString(), + anyString(), + anyString() + )).thenReturn(0L); + WorkflowPublicChatAccessGuard guard = new WorkflowPublicChatAccessGuard( + redisTemplate, + lockExecutor, + new WorkflowPublicShareProperties() + ); + + BusinessException error = Assert.expectThrows( + BusinessException.class, + () -> guard.checkRun(BigInteger.ONE, "visitor") + ); + + Assert.assertEquals(error.getHttpStatus(), 429); + } + + @Test + public void shouldReturn503WhenRedisRateLimitIsUnavailable() { + StringRedisTemplate redisTemplate = mock(StringRedisTemplate.class); + RedisLockExecutor lockExecutor = mock(RedisLockExecutor.class); + when(redisTemplate.execute( + any(DefaultRedisScript.class), + anyList(), + anyString(), + anyString(), + anyString() + )).thenThrow(new IllegalStateException("redis unavailable")); + WorkflowPublicChatAccessGuard guard = new WorkflowPublicChatAccessGuard( + redisTemplate, + lockExecutor, + new WorkflowPublicShareProperties() + ); + + BusinessException error = Assert.expectThrows( + BusinessException.class, + () -> guard.checkUpload(BigInteger.ONE, "visitor") + ); + + Assert.assertEquals(error.getHttpStatus(), 503); + } +} diff --git a/easyflow-api/easyflow-api-admin/src/test/java/tech/easyflow/admin/service/ai/WorkflowPublicChatContextResolverTest.java b/easyflow-api/easyflow-api-admin/src/test/java/tech/easyflow/admin/service/ai/WorkflowPublicChatContextResolverTest.java new file mode 100644 index 00000000..66be89d1 --- /dev/null +++ b/easyflow-api/easyflow-api-admin/src/test/java/tech/easyflow/admin/service/ai/WorkflowPublicChatContextResolverTest.java @@ -0,0 +1,136 @@ +package tech.easyflow.admin.service.ai; + +import org.testng.Assert; +import org.testng.annotations.Test; +import tech.easyflow.ai.entity.Workflow; +import tech.easyflow.ai.entity.WorkflowShare; +import tech.easyflow.ai.enums.PublishStatus; +import tech.easyflow.ai.service.WorkflowService; +import tech.easyflow.ai.service.WorkflowShareService; +import tech.easyflow.ai.share.WorkflowSharePolicy; +import tech.easyflow.common.constant.enums.EnumDataStatus; +import tech.easyflow.common.web.exceptions.BusinessException; +import tech.easyflow.system.entity.SysAccount; +import tech.easyflow.system.service.SysAccountService; + +import java.math.BigInteger; +import java.util.Map; + +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +/** + * {@link WorkflowPublicChatContextResolver} 匿名主体边界测试。 + */ +public class WorkflowPublicChatContextResolverTest { + + @Test + public void shouldUseCurrentShareCreatorAsPermissionSubject() { + Fixture fixture = fixture(EnumDataStatus.AVAILABLE.getCode()); + + WorkflowPublicChatContext context = fixture.resolver.resolveActive( + "share-key", + "00112233445566778899aabbccddeeff" + ); + + Assert.assertEquals(context.creator().getId(), BigInteger.TEN); + Assert.assertEquals(context.creator().getTenantId(), BigInteger.ONE); + Assert.assertEquals( + context.visitorDigest(), + WorkflowSharePolicy.hashChatVisitor( + "share-key", + "00112233445566778899aabbccddeeff" + ) + ); + } + + @Test + public void shouldRejectDisabledShareCreator() { + Fixture fixture = fixture(EnumDataStatus.UNAVAILABLE.getCode()); + + BusinessException error = Assert.expectThrows( + BusinessException.class, + () -> fixture.resolver.resolveActive( + "share-key", + "00112233445566778899aabbccddeeff" + ) + ); + + Assert.assertEquals(error.getHttpStatus(), 403); + Assert.assertTrue(error.getMessage().contains("创建者账号")); + } + + @Test + public void shouldResolveHistoricalShareWithoutCurrentCreatorCheck() { + Fixture fixture = fixture(EnumDataStatus.UNAVAILABLE.getCode()); + + WorkflowPublicChatContext context = fixture.resolver.resolveHistorical( + "share-key", + "00112233445566778899aabbccddeeff" + ); + + Assert.assertNull(context.creator()); + Assert.assertNull(context.workflow()); + verify(fixture.accountService, never()).getById(BigInteger.TEN); + } + + @Test + public void shouldRejectMalformedVisitorIdentity() { + Fixture fixture = fixture(EnumDataStatus.AVAILABLE.getCode()); + + BusinessException error = Assert.expectThrows( + BusinessException.class, + () -> fixture.resolver.resolveActive("share-key", "short") + ); + + Assert.assertEquals(error.getErrorCode(), 40031); + } + + private Fixture fixture(Integer accountStatus) { + WorkflowShareService shareService = mock(WorkflowShareService.class); + WorkflowService workflowService = mock(WorkflowService.class); + SysAccountService accountService = mock(SysAccountService.class); + + WorkflowShare share = new WorkflowShare(); + share.setId(BigInteger.valueOf(7)); + share.setWorkflowId(BigInteger.valueOf(11)); + share.setTenantId(BigInteger.ONE); + share.setCreatedBy(BigInteger.TEN); + + Workflow workflow = new Workflow(); + workflow.setId(BigInteger.valueOf(11)); + workflow.setTenantId(BigInteger.ONE); + workflow.setPublishStatus(PublishStatus.PUBLISHED.getCode()); + workflow.setPublishedSnapshotJson(Map.of("content", "{}")); + + SysAccount account = new SysAccount(); + account.setId(BigInteger.TEN); + account.setTenantId(BigInteger.ONE); + account.setStatus(accountStatus); + + when(shareService.resolvePublicChatShare("share-key")) + .thenReturn(share); + when(shareService.resolveHistoricalChatShare("share-key")) + .thenReturn(share); + when(workflowService.getPublishedById(BigInteger.valueOf(11))) + .thenReturn(workflow); + when(accountService.getById(BigInteger.TEN)).thenReturn(account); + + return new Fixture( + new WorkflowPublicChatContextResolver( + shareService, + workflowService, + accountService + ), + accountService + ); + } + + private record Fixture( + WorkflowPublicChatContextResolver resolver, + SysAccountService accountService + ) { + } +} diff --git a/easyflow-api/easyflow-api-admin/src/test/java/tech/easyflow/admin/service/ai/WorkflowPublicChatServiceTest.java b/easyflow-api/easyflow-api-admin/src/test/java/tech/easyflow/admin/service/ai/WorkflowPublicChatServiceTest.java new file mode 100644 index 00000000..a6823b25 --- /dev/null +++ b/easyflow-api/easyflow-api-admin/src/test/java/tech/easyflow/admin/service/ai/WorkflowPublicChatServiceTest.java @@ -0,0 +1,229 @@ +package tech.easyflow.admin.service.ai; + +import com.easyagents.flow.core.chain.ChainState; +import com.easyagents.flow.core.chain.ChainStatus; +import com.easyagents.flow.core.chain.Parameter; +import com.easyagents.flow.core.chain.repository.ChainStateRepository; +import com.easyagents.flow.core.chain.runtime.ChainExecutor; +import com.mybatisflex.core.query.QueryWrapper; +import org.mockito.ArgumentCaptor; +import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; +import org.testng.Assert; +import org.testng.annotations.Test; +import tech.easyflow.ai.easyagentsflow.service.WorkflowCheckService; +import tech.easyflow.ai.easyagentsflow.service.WorkflowRunningParameterResolver; +import tech.easyflow.ai.easyagentsflow.support.PublishedWorkflowDefinitionIds; +import tech.easyflow.ai.entity.Workflow; +import tech.easyflow.ai.entity.WorkflowExecResult; +import tech.easyflow.ai.entity.WorkflowShare; +import tech.easyflow.ai.service.WorkflowExecResultService; +import tech.easyflow.ai.service.WorkflowExecStepService; +import tech.easyflow.ai.utils.WorkFlowUtil; +import tech.easyflow.common.cache.RedisLockExecutor; +import tech.easyflow.common.constant.Constants; +import tech.easyflow.common.entity.LoginAccount; +import tech.easyflow.common.web.exceptions.BusinessException; + +import java.math.BigInteger; +import java.time.Duration; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyMap; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +/** + * {@link WorkflowPublicChatService} 匿名执行归属测试。 + */ +public class WorkflowPublicChatServiceTest { + + @Test + public void shouldSeparatePermissionSubjectFromExecutionOwner() { + Fixture fixture = fixture(); + RedisLockExecutor.LockHandle activity = mock( + RedisLockExecutor.LockHandle.class); + when(fixture.parameterResolver.normalizeRuntimeVariables( + eq("{}"), anyMap())).thenReturn(new LinkedHashMap<>()); + when(fixture.accessGuard.acquireActivity( + BigInteger.valueOf(7), "visitor-digest")) + .thenReturn(activity); + when(fixture.accessGuard.activityLease()) + .thenReturn(Duration.ofMinutes(35)); + when(fixture.eventStream.start( + eq(PublishedWorkflowDefinitionIds.published("11")), + anyMap(), + any(Runnable.class), + eq(Duration.ofMinutes(35)) + )).thenReturn(new SseEmitter()); + + fixture.service.run("share-key", visitorId(), Map.of()); + + @SuppressWarnings("unchecked") + ArgumentCaptor> variables = ArgumentCaptor + .forClass((Class) Map.class); + verify(fixture.eventStream).start( + eq(PublishedWorkflowDefinitionIds.published("11")), + variables.capture(), + any(Runnable.class), + eq(Duration.ofMinutes(35)) + ); + Assert.assertSame( + variables.getValue().get(Constants.LOGIN_USER_KEY), + fixture.context.creator() + ); + Assert.assertEquals( + variables.getValue().get(WorkFlowUtil.CREATED_KEY_MEMORY_KEY), + "WORKFLOW_CHAT_SHARE:7" + ); + Assert.assertEquals( + variables.getValue().get(WorkFlowUtil.CREATED_BY_MEMORY_KEY), + "visitor-digest" + ); + } + + @Test + public void shouldRejectExecutionOwnedByAnotherVisitor() { + Fixture fixture = fixture(); + WorkflowExecResult record = new WorkflowExecResult(); + record.setWorkflowId(BigInteger.valueOf(11)); + record.setCreatedKey("WORKFLOW_CHAT_SHARE:7"); + record.setCreatedBy("another-visitor"); + when(fixture.execResultService.getByExecKey("execution-1")) + .thenReturn(record); + + BusinessException error = Assert.expectThrows( + BusinessException.class, + () -> fixture.service.detail( + "share-key", visitorId(), "execution-1") + ); + + Assert.assertEquals(error.getHttpStatus(), 403); + Assert.assertEquals(error.getErrorCode(), 40333); + } + + @Test + public void shouldExposeMinimalRuntimeStateForRefreshRecovery() { + Fixture fixture = fixture(); + WorkflowExecResult record = ownedRecord(); + when(fixture.execResultService.getByExecKey("execution-1")) + .thenReturn(record); + when(fixture.execStepService.list(any(QueryWrapper.class))) + .thenReturn(List.of()); + ChainStateRepository repository = mock(ChainStateRepository.class); + ChainState state = new ChainState(); + state.setStatus(ChainStatus.SUSPEND); + state.setMessage("请确认是否继续"); + state.setSuspendForParameters(List.of(new Parameter("approved"))); + when(fixture.chainExecutor.getChainStateRepository()) + .thenReturn(repository); + when(repository.load("execution-1")).thenReturn(state); + + Map detail = fixture.service.detail( + "share-key", visitorId(), "execution-1"); + + @SuppressWarnings("unchecked") + Map runtime = + (Map) detail.get("runtime"); + Assert.assertEquals(runtime.get("status"), "SUSPEND"); + Assert.assertEquals(runtime.get("statusValue"), 5); + Assert.assertEquals(runtime.get("message"), "请确认是否继续"); + Assert.assertEquals( + ((List) runtime.get("parameters")).size(), + 1 + ); + } + + private WorkflowExecResult ownedRecord() { + WorkflowExecResult record = new WorkflowExecResult(); + record.setId(BigInteger.valueOf(31)); + record.setWorkflowId(BigInteger.valueOf(11)); + record.setExecKey("execution-1"); + record.setCreatedKey("WORKFLOW_CHAT_SHARE:7"); + record.setCreatedBy("visitor-digest"); + return record; + } + + private Fixture fixture() { + WorkflowPublicChatContextResolver contextResolver = mock( + WorkflowPublicChatContextResolver.class); + WorkflowCheckService workflowCheckService = mock( + WorkflowCheckService.class); + WorkflowRunningParameterResolver parameterResolver = mock( + WorkflowRunningParameterResolver.class); + WorkflowPublicChatUploadService uploadService = mock( + WorkflowPublicChatUploadService.class); + WorkflowPublicChatAccessGuard accessGuard = mock( + WorkflowPublicChatAccessGuard.class); + WorkflowChatEventStream eventStream = mock( + WorkflowChatEventStream.class); + ChainExecutor chainExecutor = mock(ChainExecutor.class); + WorkflowExecResultService execResultService = mock( + WorkflowExecResultService.class); + WorkflowExecStepService execStepService = mock( + WorkflowExecStepService.class); + + WorkflowShare share = new WorkflowShare(); + share.setId(BigInteger.valueOf(7)); + share.setWorkflowId(BigInteger.valueOf(11)); + Workflow workflow = new Workflow(); + workflow.setId(BigInteger.valueOf(11)); + workflow.setContent("{}"); + LoginAccount creator = new LoginAccount(); + creator.setId(BigInteger.TEN); + creator.setTenantId(BigInteger.ONE); + WorkflowPublicChatContext context = new WorkflowPublicChatContext( + share, + workflow, + creator, + "share-key", + "visitor-digest" + ); + when(contextResolver.resolveActive("share-key", visitorId())) + .thenReturn(context); + when(contextResolver.resolveHistorical("share-key", visitorId())) + .thenReturn(context); + + WorkflowPublicChatService service = new WorkflowPublicChatService( + contextResolver, + workflowCheckService, + parameterResolver, + uploadService, + accessGuard, + eventStream, + chainExecutor, + execResultService, + execStepService + ); + return new Fixture( + service, + context, + parameterResolver, + accessGuard, + eventStream, + chainExecutor, + execResultService, + execStepService + ); + } + + private String visitorId() { + return "00112233445566778899aabbccddeeff"; + } + + private record Fixture( + WorkflowPublicChatService service, + WorkflowPublicChatContext context, + WorkflowRunningParameterResolver parameterResolver, + WorkflowPublicChatAccessGuard accessGuard, + WorkflowChatEventStream eventStream, + ChainExecutor chainExecutor, + WorkflowExecResultService execResultService, + WorkflowExecStepService execStepService + ) { + } +} diff --git a/easyflow-api/easyflow-api-admin/src/test/java/tech/easyflow/admin/service/ai/WorkflowPublicChatUploadServiceTest.java b/easyflow-api/easyflow-api-admin/src/test/java/tech/easyflow/admin/service/ai/WorkflowPublicChatUploadServiceTest.java new file mode 100644 index 00000000..936a98c5 --- /dev/null +++ b/easyflow-api/easyflow-api-admin/src/test/java/tech/easyflow/admin/service/ai/WorkflowPublicChatUploadServiceTest.java @@ -0,0 +1,168 @@ +package tech.easyflow.admin.service.ai; + +import org.springframework.data.redis.core.StringRedisTemplate; +import org.springframework.data.redis.core.ValueOperations; +import org.springframework.web.multipart.MultipartFile; +import org.testng.Assert; +import org.testng.annotations.Test; +import tech.easyflow.ai.easyagentsflow.service.WorkflowRunningParameterResolver; +import tech.easyflow.ai.entity.Workflow; +import tech.easyflow.ai.entity.WorkflowShare; +import tech.easyflow.common.filestorage.FileStorageService; +import tech.easyflow.common.web.exceptions.BusinessException; + +import java.math.BigInteger; +import java.util.Date; +import java.util.List; +import java.util.Map; + +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +/** + * {@link WorkflowPublicChatUploadService} 上传边界测试。 + */ +public class WorkflowPublicChatUploadServiceTest { + + @Test + public void shouldStoreDeclaredFileUnderVisitorScope() { + Fixture fixture = fixture("file"); + MultipartFile file = mock(MultipartFile.class); + when(file.isEmpty()).thenReturn(false); + when(file.getSize()).thenReturn(1024L); + when(file.getOriginalFilename()).thenReturn("input.pdf"); + when(fixture.storageService.save( + eq(file), anyString())).thenReturn("/files/input.pdf"); + + fixture.service.upload(fixture.context, "attachment", file); + + verify(fixture.accessGuard).checkUpload( + BigInteger.valueOf(7), "visitor-digest"); + verify(fixture.storageService).save( + file, + "workflow-chat-share/7/visitor-digest" + ); + } + + @Test + public void shouldRejectReferenceWithoutCurrentVisitorGrant() { + Fixture fixture = fixture("file"); + + BusinessException error = Assert.expectThrows( + BusinessException.class, + () -> fixture.service.assertOwnedUploads( + fixture.context, + Map.of("attachment", List.of(Map.of( + "fileName", "input.pdf", + "filePath", "/files/other.pdf" + ))) + ) + ); + + Assert.assertEquals(error.getHttpStatus(), 403); + Assert.assertEquals(error.getErrorCode(), 40332); + } + + @Test + public void shouldRejectGrantCreatedForDifferentParameterType() { + Fixture fixture = fixture("image"); + when(fixture.valueOperations.get(anyString())).thenReturn("file"); + + BusinessException error = Assert.expectThrows( + BusinessException.class, + () -> fixture.service.assertOwnedUploads( + fixture.context, + Map.of("attachment", Map.of( + "sourceType", "upload", + "filePath", "/files/input.png" + )) + ) + ); + + Assert.assertEquals(error.getHttpStatus(), 403); + Assert.assertEquals(error.getErrorCode(), 40332); + } + + @Test + public void shouldRejectUnsupportedImageType() { + Fixture fixture = fixture("image"); + MultipartFile file = mock(MultipartFile.class); + when(file.isEmpty()).thenReturn(false); + when(file.getSize()).thenReturn(1024L); + when(file.getContentType()).thenReturn("image/svg+xml"); + when(file.getOriginalFilename()).thenReturn("input.svg"); + + BusinessException error = Assert.expectThrows( + BusinessException.class, + () -> fixture.service.upload( + fixture.context, "attachment", file) + ); + + Assert.assertTrue(error.getMessage().contains("PNG")); + } + + @SuppressWarnings("unchecked") + private Fixture fixture(String contentType) { + WorkflowRunningParameterResolver parameterResolver = mock( + WorkflowRunningParameterResolver.class); + WorkflowPublicChatAccessGuard accessGuard = mock( + WorkflowPublicChatAccessGuard.class); + StringRedisTemplate redisTemplate = mock(StringRedisTemplate.class); + ValueOperations valueOperations = mock( + ValueOperations.class); + FileStorageService storageService = mock(FileStorageService.class); + when(redisTemplate.opsForValue()).thenReturn(valueOperations); + + Workflow workflow = new Workflow(); + workflow.setId(BigInteger.valueOf(11)); + when(parameterResolver.buildRunningParametersView(workflow)) + .thenReturn(Map.of( + "startFormSchema", + List.of(Map.of( + "key", "attachment", + "contentType", contentType + )) + )); + + WorkflowShare share = new WorkflowShare(); + share.setId(BigInteger.valueOf(7)); + share.setExpiresAt(new Date( + System.currentTimeMillis() + 60_000L)); + WorkflowPublicChatContext context = new WorkflowPublicChatContext( + share, + workflow, + null, + "share-key", + "visitor-digest" + ); + WorkflowPublicChatUploadService service = + new WorkflowPublicChatUploadService( + parameterResolver, + accessGuard, + new WorkflowPublicShareProperties(), + redisTemplate, + storageService + ); + return new Fixture( + service, + context, + accessGuard, + redisTemplate, + valueOperations, + storageService + ); + } + + private record Fixture( + WorkflowPublicChatUploadService service, + WorkflowPublicChatContext context, + WorkflowPublicChatAccessGuard accessGuard, + StringRedisTemplate redisTemplate, + ValueOperations valueOperations, + FileStorageService storageService + ) { + } +} diff --git a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/easyagentsflow/listener/ChainEventListenerForSave.java b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/easyagentsflow/listener/ChainEventListenerForSave.java index 22a9da36..21b59ea9 100644 --- a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/easyagentsflow/listener/ChainEventListenerForSave.java +++ b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/easyagentsflow/listener/ChainEventListenerForSave.java @@ -85,7 +85,7 @@ public class ChainEventListenerForSave implements ChainEventListener { record.setStartTime(new Date()); record.setStatus(state.getStatus().getValue()); record.setCreatedKey(WorkFlowUtil.getCreatedKey(chain)); - record.setCreatedBy(WorkFlowUtil.getOperator(chain).getId().toString()); + record.setCreatedBy(WorkFlowUtil.getCreatedBy(chain)); // 启动记录保留同步确认，避免执行接口返回后立即查询时记录尚不可见。 try { workflowExecResultService.save(record); diff --git a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/service/WorkflowShareService.java b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/service/WorkflowShareService.java index aa9f484c..8e356361 100644 --- a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/service/WorkflowShareService.java +++ b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/service/WorkflowShareService.java @@ -92,4 +92,22 @@ public interface WorkflowShareService extends IService { * @return 有效对话分享记录 */ WorkflowShare resolveChatShare(String shareKey, BigInteger tenantId); + + /** + * 跨租户解析当前有效的匿名对话分享。 + * + * @param shareKey 原始分享密钥 + * @return 有效且指向严格发布工作流的分享记录 + */ + WorkflowShare resolvePublicChatShare(String shareKey); + + /** + * 跨租户解析匿名对话分享的历史记录。 + * + *

仅用于详情和取消已发起执行，不校验分享状态、有效期与当前发布态。

+ * + * @param shareKey 原始分享密钥 + * @return 对话分享记录 + */ + WorkflowShare resolveHistoricalChatShare(String shareKey); } diff --git a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/service/impl/WorkflowShareServiceImpl.java b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/service/impl/WorkflowShareServiceImpl.java index 3f185fc3..726f410a 100644 --- a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/service/impl/WorkflowShareServiceImpl.java +++ b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/service/impl/WorkflowShareServiceImpl.java @@ -1,6 +1,7 @@ package tech.easyflow.ai.service.impl; import com.mybatisflex.core.query.QueryWrapper; +import com.mybatisflex.core.tenant.TenantManager; import com.mybatisflex.spring.service.impl.ServiceImpl; import org.springframework.stereotype.Service; import org.springframework.transaction.PlatformTransactionManager; @@ -23,6 +24,7 @@ import java.math.BigInteger; import java.time.Duration; import java.util.Date; import java.util.List; +import java.util.Objects; import java.util.UUID; /** @@ -184,6 +186,61 @@ public class WorkflowShareServiceImpl extends ServiceImpl { + WorkflowShare share = findShare( + shareKey, + WorkflowSharePurpose.CHAT, + true + ); + if (share == null) { + throw invalidShare(); + } + if (share.getExpiresAt() == null + || !share.getExpiresAt().after(new Date())) { + throw new BusinessException(403, 403, "工作流分享链接已过期"); + } + Workflow workflow = workflowService.getPublishedById( + share.getWorkflowId()); + if (workflow == null + || !Objects.equals( + share.getTenantId(), workflow.getTenantId())) { + throw invalidShare(); + } + if (!isStrictlyPublished(workflow)) { + throw new BusinessException(409, 409, "工作流尚未发布或已下线"); + } + return share; + }); + } + + /** + * {@inheritDoc} + */ + @Override + public WorkflowShare resolveHistoricalChatShare(String shareKey) { + if (shareKey == null || shareKey.isBlank()) { + throw invalidShare(); + } + WorkflowShare share = TenantManager.withoutTenantCondition( + () -> findShare( + shareKey, + WorkflowSharePurpose.CHAT, + false + )); + if (share == null) { + throw invalidShare(); + } + return share; + } + /** * 按用途校验并解析分享。 * @@ -200,10 +257,7 @@ public class WorkflowShareServiceImpl extends ServiceImpl ALLOWED_REQUESTS = Set.of( @@ -67,6 +74,28 @@ public final class WorkflowSharePolicy { } } + /** + * 计算匿名访客的不可逆执行归属摘要。 + * + * @param shareKey 原始分享密钥 + * @param visitorId 当前标签页访客标识 + * @return HMAC-SHA256 前 16 字节的小写十六进制摘要 + */ + public static String hashChatVisitor(String shareKey, String visitorId) { + try { + Mac mac = Mac.getInstance("HmacSHA256"); + mac.init(new SecretKeySpec( + shareKey.getBytes(StandardCharsets.UTF_8), + "HmacSHA256" + )); + byte[] digest = mac.doFinal( + visitorId.getBytes(StandardCharsets.UTF_8)); + return HexFormat.of().formatHex(digest, 0, 16); + } catch (Exception exception) { + throw new IllegalStateException("HmacSHA256 unavailable", exception); + } + } + /** * 计算默认过期时间。 * diff --git a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/utils/WorkFlowUtil.java b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/utils/WorkFlowUtil.java index 47944d45..66706a85 100644 --- a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/utils/WorkFlowUtil.java +++ b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/utils/WorkFlowUtil.java @@ -19,6 +19,7 @@ public class WorkFlowUtil { public final static String WORKFLOW_CHAT_SHARE = "WORKFLOW_CHAT_SHARE"; public final static String WORKFLOW_KEY = "workflow"; public final static String CREATED_KEY_MEMORY_KEY = "workflowCreatedKey"; + public final static String CREATED_BY_MEMORY_KEY = "workflowCreatedBy"; public static String removeSensitiveInfo(String originJson) { JSONObject workflowInfo = JSON.parseObject(originJson); @@ -56,6 +57,35 @@ public class WorkFlowUtil { return value == null ? USER_KEY : String.valueOf(value); } + /** + * 获取工作流执行记录的归属主体。 + * + *

匿名分享可覆盖为访客摘要；其他入口继续使用权限主体账号 ID。

+ * + * @param chain 当前工作流执行链 + * @return 执行归属主体 + */ + public static String getCreatedBy(Chain chain) { + Object value = chain.getExecutionState() + .getMemory() + .get(CREATED_BY_MEMORY_KEY); + if (value != null) { + return String.valueOf(value); + } + LoginAccount operator = getOperator(chain); + return operator.getId() == null ? "0" : operator.getId().toString(); + } + + /** + * 构建工作流匿名分享的执行来源标识。 + * + * @param shareId 分享记录 ID + * @return 执行来源标识 + */ + public static String publicChatShareCreatedKey(BigInteger shareId) { + return WORKFLOW_CHAT_SHARE + ":" + shareId; + } + public static LoginAccount defaultAccount() { LoginAccount account = new LoginAccount(); account.setId(new BigInteger("0")); diff --git a/easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/share/WorkflowSharePolicyTest.java b/easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/share/WorkflowSharePolicyTest.java index 68a9c2f5..76296beb 100644 --- a/easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/share/WorkflowSharePolicyTest.java +++ b/easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/share/WorkflowSharePolicyTest.java @@ -24,6 +24,26 @@ public class WorkflowSharePolicyTest { Assert.assertNotEquals("share-key", first); } + /** + * 验证匿名访客归属摘要稳定、定长且按分享密钥隔离。 + */ + @Test + public void shouldHashChatVisitorPerShareWithoutLeakingIdentity() { + String visitorId = "00112233445566778899aabbccddeeff"; + + String first = WorkflowSharePolicy.hashChatVisitor( + "share-key-a", visitorId); + String second = WorkflowSharePolicy.hashChatVisitor( + "share-key-a", visitorId); + String otherShare = WorkflowSharePolicy.hashChatVisitor( + "share-key-b", visitorId); + + Assert.assertEquals(first, second); + Assert.assertEquals(32, first.length()); + Assert.assertNotEquals(first, otherShare); + Assert.assertFalse(first.contains(visitorId)); + } + /** * 验证默认过期时间为创建时间后 30 分钟。 */ diff --git a/easyflow-ui-admin/app/src/api/request.ts b/easyflow-ui-admin/app/src/api/request.ts index 8265b031..0da7eabc 100644 --- a/easyflow-ui-admin/app/src/api/request.ts +++ b/easyflow-ui-admin/app/src/api/request.ts @@ -21,12 +21,11 @@ import { events } from 'fetch-event-stream'; import { useAuthStore } from '#/store'; import { isWorkflowShareRequest, - readWorkflowShareKey, - withWorkflowShareHeader, - WORKFLOW_SHARE_HEADER, + withWorkflowShareHeaders, } from '#/utils/workflow-share-context'; import { refreshTokenApi } from './core'; +import { isInactiveSseRequest } from './sseRequestLifecycle'; const { apiURL } = useAppConfig(import.meta.env, import.meta.env.PROD); const ERROR_MESSAGE_DEDUP_WINDOW = 800; @@ -101,12 +100,15 @@ function createRequestClient(baseURL: string, options?: RequestClientOptions) { config.headers['easyflow-token'] = formatToken(accessStore.accessToken); config.headers['Accept-Language'] = preferences.app.locale; - const workflowShareKey = readWorkflowShareKey(); - if ( - workflowShareKey && - isWorkflowShareRequest(config.url, config.method) - ) { - config.headers[WORKFLOW_SHARE_HEADER] = workflowShareKey; + const workflowShareHeaders = withWorkflowShareHeaders( + {}, + { + requestMethod: config.method, + requestUrl: config.url, + }, + ); + for (const [name, value] of Object.entries(workflowShareHeaders)) { + config.headers[name] = value; } return config; }, @@ -132,6 +134,8 @@ function createRequestClient(baseURL: string, options?: RequestClientOptions) { doRefreshToken, enableRefreshToken: preferences.app.enableRefreshToken ?? false, formatToken, + shouldHandleUnauthorized: (config) => + !isWorkflowShareRequest(config?.url, config?.method), }), ); @@ -186,7 +190,7 @@ export function createEventStreamHeaders( headers[key] = value; }); } - return withWorkflowShareHeader(headers, { + return withWorkflowShareHeaders(headers, { requestMethod: 'POST', requestUrl, }); @@ -274,15 +278,25 @@ export class SseClient { options?.onMessage?.(event); } } catch (innerError) { + if ( + isInactiveSseRequest(signal, this.currentRequestId, currentRequestId) + ) { + return; + } options?.onError?.(innerError); + return; } // 只有在还是同一个请求的情况下才调用 onFinished - if (this.currentRequestId === currentRequestId) { + if ( + !isInactiveSseRequest(signal, this.currentRequestId, currentRequestId) + ) { options?.onFinished?.(); } } catch (error) { - if (this.currentRequestId !== currentRequestId) { + if ( + isInactiveSseRequest(signal, this.currentRequestId, currentRequestId) + ) { return; } console.error('SSE错误:', error); diff --git a/easyflow-ui-admin/app/src/api/sseRequestLifecycle.test.ts b/easyflow-ui-admin/app/src/api/sseRequestLifecycle.test.ts new file mode 100644 index 00000000..3d7ec743 --- /dev/null +++ b/easyflow-ui-admin/app/src/api/sseRequestLifecycle.test.ts @@ -0,0 +1,19 @@ +import { describe, expect, it } from 'vitest'; + +import { isInactiveSseRequest } from './sseRequestLifecycle'; + +describe('sseRequestLifecycle', () => { + it('treats an explicit abort as an inactive request', () => { + const controller = new AbortController(); + controller.abort(); + + expect(isInactiveSseRequest(controller.signal, 1, 1)).toBe(true); + }); + + it('treats a superseded request as inactive', () => { + const controller = new AbortController(); + + expect(isInactiveSseRequest(controller.signal, 2, 1)).toBe(true); + expect(isInactiveSseRequest(controller.signal, 1, 1)).toBe(false); + }); +}); diff --git a/easyflow-ui-admin/app/src/api/sseRequestLifecycle.ts b/easyflow-ui-admin/app/src/api/sseRequestLifecycle.ts new file mode 100644 index 00000000..90a48667 --- /dev/null +++ b/easyflow-ui-admin/app/src/api/sseRequestLifecycle.ts @@ -0,0 +1,10 @@ +/** + * 判断 SSE 请求是否已被主动中止或被后续请求替换。 + */ +export function isInactiveSseRequest( + signal: AbortSignal, + currentRequestId: number, + requestId: number, +) { + return signal.aborted || currentRequestId !== requestId; +} diff --git a/easyflow-ui-admin/app/src/router/__tests__/public-route-guard.test.ts b/easyflow-ui-admin/app/src/router/__tests__/public-route-guard.test.ts new file mode 100644 index 00000000..3af20321 --- /dev/null +++ b/easyflow-ui-admin/app/src/router/__tests__/public-route-guard.test.ts @@ -0,0 +1,42 @@ +import { createMemoryHistory, createRouter } from 'vue-router'; + +import { useAccessStore } from '@easyflow/stores'; + +import { createPinia, setActivePinia } from 'pinia'; +import { beforeEach, describe, expect, it } from 'vitest'; + +import { createRouterGuard } from '../guard'; + +describe('public route guard', () => { + beforeEach(() => { + setActivePinia(createPinia()); + }); + + it('bypasses stale login state for an anonymous workflow share', async () => { + const accessStore = useAccessStore(); + accessStore.setAccessToken('stale-token'); + const router = createRouter({ + history: createMemoryHistory(), + routes: [ + { + component: { template: '
login
' }, + name: 'Login', + path: '/auth/login', + }, + { + component: { template: '
workflow share
' }, + meta: { ignoreAccess: true, title: 'Workflow Share' }, + name: 'WorkflowShare', + path: '/share/workflow', + }, + ], + }); + createRouterGuard(router); + + await router.push('/share/workflow?shareKey=share-key'); + await router.isReady(); + + expect(router.currentRoute.value.name).toBe('WorkflowShare'); + expect(router.currentRoute.value.query.shareKey).toBe('share-key'); + }); +}); diff --git a/easyflow-ui-admin/app/src/router/__tests__/share-routes.test.ts b/easyflow-ui-admin/app/src/router/__tests__/share-routes.test.ts index 30fc6d9a..88d5d95d 100644 --- a/easyflow-ui-admin/app/src/router/__tests__/share-routes.test.ts +++ b/easyflow-ui-admin/app/src/router/__tests__/share-routes.test.ts @@ -11,6 +11,7 @@ describe('external share routes', () => { hideInBreadcrumb: true, hideInMenu: true, hideInTab: true, + ignoreAccess: true, noBasicLayout: true, }); }); @@ -19,6 +20,7 @@ describe('external share routes', () => { const route = routes.find((item) => item.name === 'WorkflowShareExpired'); expect(route?.path).toBe('/share/workflow/expired'); + expect(route?.meta?.ignoreAccess).toBe(true); expect(route?.meta?.noBasicLayout).toBe(true); }); }); diff --git a/easyflow-ui-admin/app/src/router/guard.ts b/easyflow-ui-admin/app/src/router/guard.ts index 9dd128a3..edc56ce3 100644 --- a/easyflow-ui-admin/app/src/router/guard.ts +++ b/easyflow-ui-admin/app/src/router/guard.ts @@ -152,6 +152,12 @@ function setupAccessGuard(router: Router) { let devLoginPromise: null | Promise = null; router.beforeEach(async (to, from) => { + // 公开路由必须在读取或刷新登录态之前短路，避免浏览器残留的过期 + // token 把匿名分享页重定向到登录页。 + if (to.meta.ignoreAccess) { + return true; + } + const accessStore = useAccessStore(); const userStore = useUserStore(); const authStore = useAuthStore(); @@ -227,11 +233,6 @@ function setupAccessGuard(router: Router) { // accessToken 检查 if (!accessStore.accessToken) { - // 明确声明忽略权限访问权限，则可以访问 - if (to.meta.ignoreAccess) { - return true; - } - // 没有访问权限，跳转登录页面 if (to.fullPath !== LOGIN_PATH) { const cleanFullPath = diff --git a/easyflow-ui-admin/app/src/router/routes/external/share.ts b/easyflow-ui-admin/app/src/router/routes/external/share.ts index 4385e4a6..6edbeb3f 100644 --- a/easyflow-ui-admin/app/src/router/routes/external/share.ts +++ b/easyflow-ui-admin/app/src/router/routes/external/share.ts @@ -33,6 +33,7 @@ const routes: RouteRecordRaw[] = [ component: () => import('#/views/ai/workflow/WorkflowShareView.vue'), meta: { title: 'Workflow Share', + ignoreAccess: true, noBasicLayout: true, hideInMenu: true, hideInBreadcrumb: true, @@ -46,6 +47,7 @@ const routes: RouteRecordRaw[] = [ import('#/views/ai/documentCollection/KnowledgeShareExpired.vue'), meta: { title: 'Workflow Share Expired', + ignoreAccess: true, noBasicLayout: true, hideInMenu: true, hideInBreadcrumb: true, diff --git a/easyflow-ui-admin/app/src/utils/workflow-share-context.ts b/easyflow-ui-admin/app/src/utils/workflow-share-context.ts index 0803e3b8..b1445528 100644 --- a/easyflow-ui-admin/app/src/utils/workflow-share-context.ts +++ b/easyflow-ui-admin/app/src/utils/workflow-share-context.ts @@ -5,6 +5,14 @@ import { readScopedRouteQueryParam } from './share-route-context'; */ export const WORKFLOW_SHARE_HEADER = 'X-Workflow-Chat-Share-Key'; +/** + * 当前标签页的工作流对话分享访客标识请求头。 + */ +export const WORKFLOW_SHARE_VISITOR_HEADER = 'X-Workflow-Chat-Visitor'; + +const WORKFLOW_SHARE_VISITOR_STORAGE_KEY = + 'easyflow.workflow-chat-share.visitor'; + interface WorkflowShareResolutionOptions { currentWorkflowId?: null | T; onFailure: (error: unknown) => Promise | void; @@ -16,16 +24,19 @@ interface WorkflowShareHeaderOptions { pageUrl?: string; requestMethod?: string; requestUrl?: string; + storage?: Pick; + visitorId?: string; } const WORKFLOW_SHARE_ROUTES = ['/share/workflow']; const WORKFLOW_SHARE_REQUESTS = [ - ['GET', '/api/v1/workflowChat/descriptor'], - ['GET', '/api/v1/workflowChat/execution'], + ['GET', '/api/v1/workflowChat/public/descriptor'], + ['GET', '/api/v1/workflowChat/public/execution'], ['GET', '/api/v1/workflowShare/resolve'], - ['POST', '/api/v1/workflowChat/cancel'], - ['POST', '/api/v1/workflowChat/resume'], - ['POST', '/api/v1/workflowChat/run'], + ['POST', '/api/v1/workflowChat/public/cancel'], + ['POST', '/api/v1/workflowChat/public/resume'], + ['POST', '/api/v1/workflowChat/public/run'], + ['POST', '/api/v1/workflowChat/public/upload'], ] as const; /** @@ -91,12 +102,42 @@ export function isWorkflowShareRequest( } } +/** + * 读取或创建当前标签页稳定的 128-bit 匿名访客标识。 + */ +export function resolveWorkflowShareVisitorId( + storage: + | Pick + | undefined = resolveSessionStorage(), + randomBytes: (size: number) => Uint8Array = createRandomBytes, +): string { + const existing = storage?.getItem(WORKFLOW_SHARE_VISITOR_STORAGE_KEY)?.trim(); + if (existing && /^[a-f0-9]{32}$/.test(existing)) { + return existing; + } + const visitorId = [...randomBytes(16)] + .map((value) => value.toString(16).padStart(2, '0')) + .join(''); + storage?.setItem(WORKFLOW_SHARE_VISITOR_STORAGE_KEY, visitorId); + return visitorId; +} + /** * 在保留现有请求头的基础上附加工作流分享密钥。 */ export function withWorkflowShareHeader( headers: Record, options: WorkflowShareHeaderOptions = {}, +): Record { + return withWorkflowShareHeaders(headers, options); +} + +/** + * 在保留通用请求头的基础上附加匿名分享密钥与当前标签页访客标识。 + */ +export function withWorkflowShareHeaders( + headers: Record, + options: WorkflowShareHeaderOptions = {}, ): Record { if (!isWorkflowShareRequest(options.requestUrl, options.requestMethod)) { return headers; @@ -105,12 +146,32 @@ export function withWorkflowShareHeader( if (!shareKey) { return headers; } + const visitorId = + options.visitorId || resolveWorkflowShareVisitorId(options.storage); return { ...headers, + 'easyflow-token': '', [WORKFLOW_SHARE_HEADER]: shareKey, + [WORKFLOW_SHARE_VISITOR_HEADER]: visitorId, }; } +function resolveSessionStorage() { + try { + return globalThis.sessionStorage; + } catch { + return undefined; + } +} + +function createRandomBytes(size: number) { + const bytes = new Uint8Array(size); + if (!globalThis.crypto?.getRandomValues) { + throw new Error('当前浏览器不支持安全的匿名访客标识'); + } + return globalThis.crypto.getRandomValues(bytes); +} + /** * 解析分享地址对应的工作流，并在链接失效时统一收口异常。 */ diff --git a/easyflow-ui-admin/app/src/views/ai/workflow/components/WorkflowChatPage.vue b/easyflow-ui-admin/app/src/views/ai/workflow/components/WorkflowChatPage.vue index 274bcbdb..b3438cf8 100644 --- a/easyflow-ui-admin/app/src/views/ai/workflow/components/WorkflowChatPage.vue +++ b/easyflow-ui-admin/app/src/views/ai/workflow/components/WorkflowChatPage.vue @@ -57,7 +57,11 @@ import { navigateBackToList } from '#/router/list-return-context'; import { resolveAgentChatIdentity } from '#/utils/agent-chat-cache'; import { copyTextWithFeedback } from '#/utils/clipboard-feedback'; import { buildAbsoluteAppRouteUrl } from '#/utils/share-route-context'; -import { resolveWorkflowShareFailureReason } from '#/utils/workflow-share-context'; +import { + readWorkflowShareKey, + resolveWorkflowShareFailureReason, + resolveWorkflowShareVisitorId, +} from '#/utils/workflow-share-context'; import { finalizeWorkflowExecutionSteps, @@ -65,6 +69,10 @@ import { hydrateWorkflowExecutionSteps, reduceWorkflowExecutionSteps, } from './workflowExecutionDetails'; +import { + resolveWorkflowExecutionRecoveryOutput, + resolveWorkflowExecutionRecoveryStatus, +} from './workflowExecutionRecovery'; import WorkflowFinalOutput from './WorkflowFinalOutput.vue'; import WorkflowFormItem from './WorkflowFormItem.vue'; import { @@ -88,6 +96,14 @@ import { formatWorkflowProgressLabel, summarizeWorkflowActiveNodes, } from './workflowRunProgress'; +import { + buildWorkflowShareConversationKey, + buildWorkflowShareStorageScope, + getWorkflowShareConversationStorage, + readWorkflowShareConversation, + removeWorkflowShareConversation, + writeWorkflowShareConversation, +} from './workflowShareConversationStorage'; interface WorkflowStreamEnvelope { data?: Record; @@ -97,6 +113,10 @@ interface WorkflowStreamEnvelope { type: string; } +interface ChatTimelineHandle { + scrollToBottom: () => void; +} + const props = withDefaults( defineProps<{ shareMode?: boolean; @@ -109,14 +129,21 @@ const props = withDefaults( const MAX_COLLAPSED_PARAMETER_COUNT = 4; const PRIMARY_PARAMETER_COUNT = 4; const DRAFT_SAVE_DELAY_MS = 300; +const CONVERSATION_SAVE_DELAY_MS = 300; +const EXECUTION_RECOVERY_POLL_MS = 1500; const route = useRoute(); const userStore = useUserStore(); const streamClient = new SseClient(); +const shareStorageScope = resolveShareStorageScope(); const loading = ref(true); const loadError = ref(''); const descriptor = ref>({}); const workflowId = ref(); const timelineItems = ref([]); +const timelineRef = ref(); +const timelinePinnedToBottom = ref(true); +const composerRef = ref(); +const composerHeight = ref(152); const question = ref(''); const running = ref(false); const stopping = ref(false); @@ -153,7 +180,12 @@ const manualAbort = ref(false); const lastRunningNodeName = ref(''); let progressStatusTimer = 0; let draftSaveTimer = 0; +let conversationSaveTimer = 0; +let executionRecoveryTimer = 0; +let executionRecoveryActive = false; +let executionRecoveryLoading = false; let draftReady = false; +let conversationReady = false; let userMessageSequence = 0; const formParameters = computed(() => @@ -289,22 +321,46 @@ const detailDurationText = computed(() => { : persistedDuration; return duration === undefined ? '—' : `${duration} ms`; }); +const timelineStyle = computed>(() => ({ + '--workflow-chat-composer-height': `${composerHeight.value}px`, +})); useResizeObserver(parameterChipsRef, ([entry]) => { parameterChipsWidth.value = entry?.contentRect.width || 0; }); +useResizeObserver(composerRef, () => { + const nextHeight = composerRef.value?.getBoundingClientRect().height || 0; + if (nextHeight > 0) { + composerHeight.value = Math.ceil(nextHeight); + } +}); watch([question, extraValues], scheduleDraftSave, { deep: true }); +watch( + [ + timelineItems, + executeId, + runStatusKey, + parametersLocked, + executionState, + executionStartedAt, + executionElapsed, + ], + scheduleConversationSave, + { deep: true }, +); onMounted(() => { - window.addEventListener('pagehide', persistDraft); + window.addEventListener('pagehide', handlePageHide); void loadPage(); }); onBeforeUnmount(() => { - window.removeEventListener('pagehide', persistDraft); + window.removeEventListener('pagehide', handlePageHide); clearDraftSaveTimer(); - persistDraft(); + clearConversationSaveTimer(); + clearExecutionRecoveryTimer(); + persistPageState(); manualAbort.value = true; streamClient.abort(); clearProgressStatusTimer(); @@ -312,6 +368,7 @@ onBeforeUnmount(() => { async function loadPage() { draftReady = false; + conversationReady = false; loading.value = true; loadError.value = ''; try { @@ -321,14 +378,23 @@ async function loadPage() { if (!workflowId.value) { throw new Error('工作流不存在'); } - const response = await api.get('/api/v1/workflowChat/descriptor', { + const response = await api.get(workflowChatEndpoint('descriptor'), { params: { workflowId: workflowId.value }, }); descriptor.value = response.data || {}; initializeAdditionalValues(); restoreDraft(); + restoreShareConversation(); await nextTick(); draftReady = true; + conversationReady = true; + if (props.shareMode && executeId.value) { + if (executionRecoveryActive) { + void recoverExecutionAfterRefresh(); + } else { + void loadExecutionDetail(); + } + } } catch (error: any) { loadError.value = error?.message || '工作流加载失败'; } finally { @@ -370,14 +436,29 @@ function draftKey() { } return buildWorkflowRunDraftKey( workflowId.value, - resolveAgentChatIdentity(userStore.userInfo), + props.shareMode + ? shareStorageScope + : resolveAgentChatIdentity(userStore.userInfo), props.shareMode, ); } +function shareConversationKey() { + return buildWorkflowShareConversationKey( + workflowId.value || '', + shareStorageScope, + ); +} + +function workflowChatEndpoint(action: string) { + return props.shareMode + ? `/api/v1/workflowChat/public/${action}` + : `/api/v1/workflowChat/${action}`; +} + function restoreDraft() { const draft = readWorkflowRunDraft( - getWorkflowRunDraftStorage(), + getWorkflowRunDraftStorage(props.shareMode), draftKey(), additionalParameters.value, ); @@ -419,7 +500,7 @@ function persistDraft() { if (!draftReady) { return; } - const storage = getWorkflowRunDraftStorage(); + const storage = getWorkflowRunDraftStorage(props.shareMode); const key = draftKey(); if ( !hasWorkflowRunDraftContent( @@ -437,6 +518,99 @@ function persistDraft() { }); } +function resolveShareStorageScope() { + if (!props.shareMode) { + return ''; + } + try { + return buildWorkflowShareStorageScope( + resolveWorkflowShareVisitorId(), + readWorkflowShareKey() || '', + ); + } catch { + return ''; + } +} + +function restoreShareConversation() { + if (!props.shareMode) { + return false; + } + const snapshot = readWorkflowShareConversation( + getWorkflowShareConversationStorage(), + shareConversationKey(), + ); + if (!snapshot) { + return false; + } + timelineItems.value = snapshot.timelineItems; + executeId.value = snapshot.executeId; + runStatusKey.value = snapshot.runStatusKey; + parametersLocked.value = snapshot.parametersLocked; + executionState.value = snapshot.executionState; + executionStartedAt.value = snapshot.executionStartedAt; + executionElapsed.value = snapshot.executionElapsed; + executionRecoveryActive = + Boolean(snapshot.executeId) && + (snapshot.executionState === 'running' || + snapshot.executionState === 'waiting'); + running.value = executionRecoveryActive; + stopping.value = false; + waitingConfirmation.value = undefined; + confirmSubmittingAction.value = ''; + confirmError.value = ''; + return true; +} + +function clearConversationSaveTimer() { + if (conversationSaveTimer) { + window.clearTimeout(conversationSaveTimer); + conversationSaveTimer = 0; + } +} + +function scheduleConversationSave() { + if (!props.shareMode || !conversationReady) { + return; + } + clearConversationSaveTimer(); + conversationSaveTimer = window.setTimeout(() => { + conversationSaveTimer = 0; + persistShareConversation(); + }, CONVERSATION_SAVE_DELAY_MS); +} + +function persistShareConversation() { + if (!props.shareMode || !conversationReady) { + return; + } + const storage = getWorkflowShareConversationStorage(); + const key = shareConversationKey(); + if (timelineItems.value.length === 0 && !executeId.value) { + removeWorkflowShareConversation(storage, key); + return; + } + writeWorkflowShareConversation(storage, key, { + executeId: executeId.value, + executionElapsed: executionElapsed.value, + executionStartedAt: executionStartedAt.value, + executionState: executionState.value, + parametersLocked: parametersLocked.value, + runStatusKey: runStatusKey.value, + timelineItems: timelineItems.value, + }); +} + +function persistPageState() { + persistDraft(); + persistShareConversation(); +} + +function handlePageHide() { + manualAbort.value = true; + persistPageState(); +} + async function backToWorkflowList() { await navigateBackToList( router, @@ -529,10 +703,23 @@ function appendError(message: string, id = `error-${Date.now()}`) { } function appendFinalOutput(output: unknown, eventId: string) { + const finalOutputId = `final-output-${eventId}`; + const executionPrefix = executeId.value + ? `final-output-${executeId.value}` + : finalOutputId; + if ( + timelineItems.value.some( + (item) => + item.type === 'custom' && + (item.id === finalOutputId || item.id.startsWith(executionPrefix)), + ) + ) { + return; + } timelineItems.value.push({ customType: 'workflow-final-output', data: output, - id: `final-output-${eventId}`, + id: finalOutputId, type: 'custom', }); } @@ -604,6 +791,8 @@ async function handleSend() { extraValues.value, ); parametersLocked.value = true; + executionRecoveryActive = false; + clearExecutionRecoveryTimer(); appendUserMessage(content, images); question.value = ''; parametersExpanded.value = false; @@ -623,11 +812,13 @@ async function handleSend() { confirmError.value = ''; lastRunningNodeName.value = ''; manualAbort.value = false; + await nextTick(); + scrollToLatest(); void streamClient.post( - '/api/v1/workflowChat/run', + workflowChatEndpoint('run'), { - workflowId: workflowId.value, + ...(props.shareMode ? {} : { workflowId: workflowId.value }), variables: { ...extraValues.value, user_input: content, @@ -638,6 +829,9 @@ async function handleSend() { if (manualAbort.value) { return; } + if (beginExecutionRecovery()) { + return; + } finishExecution( 'failed', error?.message || '工作流执行失败', @@ -647,6 +841,9 @@ async function handleSend() { }, onFinished: () => { if (running.value && !manualAbort.value) { + if (beginExecutionRecovery()) { + return; + } finishExecution( 'failed', '运行连接已结束，请重试', @@ -747,6 +944,8 @@ function finishExecution( eventId = executeId.value || String(Date.now()), ) { const failedNodeName = activeNodeSummary(); + executionRecoveryActive = false; + clearExecutionRecoveryTimer(); clearProgressStatusTimer(); running.value = false; stopping.value = false; @@ -806,7 +1005,7 @@ function finalizeLiveExecutionSteps( ); executionElapsed.value = executionStartedAt.value === undefined - ? undefined + ? executionElapsed.value : Math.max(0, finishedAt - executionStartedAt.value); } @@ -834,7 +1033,7 @@ async function resumeExecution(confirmed: boolean) { ) { return; } - await api.post('/api/v1/workflowChat/resume', { + await api.post(workflowChatEndpoint('resume'), { executeId: executeId.value, confirmParams: { [confirmKey.value]: confirmed ? 'yes' : 'no', @@ -845,6 +1044,9 @@ async function resumeExecution(confirmed: boolean) { executionState.value = 'running'; markConfirmationStep('running'); appendStatus(progressLabel('正在运行'), 'running', runStatusKey.value); + if (executionRecoveryActive) { + scheduleExecutionRecovery(); + } } catch (error: any) { confirmError.value = error?.message || '提交失败，请重试'; } finally { @@ -861,7 +1063,7 @@ async function stopExecution() { appendStatus(progressLabel('正在中止'), 'running', runStatusKey.value); try { if (executeId.value) { - await api.post('/api/v1/workflowChat/cancel', { + await api.post(workflowChatEndpoint('cancel'), { executeId: executeId.value, }); } @@ -886,8 +1088,19 @@ async function resetConversation() { } } draftReady = false; + conversationReady = false; clearDraftSaveTimer(); - removeWorkflowRunDraft(getWorkflowRunDraftStorage(), draftKey()); + clearConversationSaveTimer(); + clearExecutionRecoveryTimer(); + executionRecoveryActive = false; + removeWorkflowRunDraft( + getWorkflowRunDraftStorage(props.shareMode), + draftKey(), + ); + removeWorkflowShareConversation( + getWorkflowShareConversationStorage(), + shareConversationKey(), + ); manualAbort.value = true; streamClient.abort(); timelineItems.value = []; @@ -910,26 +1123,181 @@ async function resetConversation() { initializeAdditionalValues(); await nextTick(); draftReady = true; + conversationReady = true; } -async function loadExecutionDetail() { +function clearExecutionRecoveryTimer() { + if (!executionRecoveryTimer) { + return; + } + window.clearTimeout(executionRecoveryTimer); + executionRecoveryTimer = 0; +} + +function beginExecutionRecovery() { + if (!props.shareMode || !executeId.value) { + return false; + } + executionRecoveryActive = true; + running.value = true; + stopping.value = false; + if (executionState.value !== 'waiting') { + executionState.value = 'running'; + appendStatus( + '连接已断开，正在恢复运行状态…', + 'running', + runStatusKey.value, + ); + } + clearExecutionRecoveryTimer(); + void recoverExecutionAfterRefresh(); + return true; +} + +function scheduleExecutionRecovery() { + if ( + !executionRecoveryActive || + !props.shareMode || + !executeId.value || + executionState.value === 'waiting' + ) { + return; + } + clearExecutionRecoveryTimer(); + executionRecoveryTimer = window.setTimeout(() => { + executionRecoveryTimer = 0; + void recoverExecutionAfterRefresh(); + }, EXECUTION_RECOVERY_POLL_MS); +} + +async function recoverExecutionAfterRefresh() { + if ( + !executionRecoveryActive || + executionRecoveryLoading || + !executeId.value + ) { + return; + } + executionRecoveryLoading = true; + try { + const detail = await loadExecutionDetail({ background: true }); + if (!detail || !executionRecoveryActive) { + scheduleExecutionRecovery(); + return; + } + syncRecoveredExecution(detail); + } finally { + executionRecoveryLoading = false; + } +} + +function syncRecoveredExecution(detail: Record) { + liveExecutionSteps.value = hydrateWorkflowExecutionSteps(detail.steps); + const activeStep = [...liveExecutionSteps.value] + .reverse() + .find((step) => step.status === 'running' || step.status === 'waiting'); + lastRunningNodeName.value = activeStep?.nodeName || ''; + const status = resolveWorkflowExecutionRecoveryStatus(detail); + if (status === 'SUSPEND') { + removeStaleConnectionFailure(); + running.value = true; + stopping.value = false; + executionState.value = 'waiting'; + waitingConfirmation.value = { + message: detail.runtime?.message || '请确认', + parameters: Array.isArray(detail.runtime?.parameters) + ? detail.runtime.parameters + : [], + }; + initializeConfirmValues(waitingConfirmation.value.parameters); + markConfirmationStep('waiting'); + appendStatus(progressLabel('等待确认'), 'running', runStatusKey.value); + return; + } + if (status === 'READY' || status === 'RUNNING' || status === 'ERROR') { + removeStaleConnectionFailure(); + running.value = true; + stopping.value = false; + waitingConfirmation.value = undefined; + executionState.value = 'running'; + appendStatus(progressLabel('正在运行'), 'running', runStatusKey.value); + scheduleExecutionRecovery(); + return; + } + + const persistedDuration = Number(detail.record?.execTime); + if (Number.isFinite(persistedDuration)) { + executionElapsed.value = persistedDuration; + executionStartedAt.value = undefined; + } + if (status === 'SUCCEEDED') { + removeStaleConnectionFailure(); + const output = resolveWorkflowExecutionRecoveryOutput(detail); + finishExecution('completed', undefined, output, executeId.value); + return; + } + if (status === 'CANCELLED') { + finishExecution('cancelled', undefined, undefined, executeId.value); + return; + } + if (status === 'FAILED') { + finishExecution( + 'failed', + detail.runtime?.message || detail.record?.errorInfo, + undefined, + executeId.value, + ); + return; + } + scheduleExecutionRecovery(); +} + +function removeStaleConnectionFailure() { if (!executeId.value) { return; } - detailLoading.value = true; - detailLoadError.value = ''; - try { - const response = await api.get('/api/v1/workflowChat/execution', { - params: { executeId: executeId.value }, - }); - executionDetail.value = response.data; - } catch (error: any) { - detailLoadError.value = error?.message || '运行详情加载失败'; - } finally { - detailLoading.value = false; + const staleErrorId = `terminal-error-${executeId.value}`; + const next = timelineItems.value.filter( + (item) => item.type !== 'error' || item.id !== staleErrorId, + ); + if (next.length !== timelineItems.value.length) { + timelineItems.value = next; } } +async function loadExecutionDetail(options: { background?: boolean } = {}) { + if (!executeId.value) { + return; + } + const targetExecuteId = executeId.value; + if (!options.background) { + detailLoading.value = true; + detailLoadError.value = ''; + } + try { + const response = await api.get(workflowChatEndpoint('execution'), { + params: { executeId: targetExecuteId }, + }); + if (targetExecuteId !== executeId.value) { + return; + } + executionDetail.value = response.data; + return response.data as Record; + } catch (error: any) { + if (!options.background) { + detailLoadError.value = error?.message || '运行详情加载失败'; + } + } finally { + if (!options.background) { + detailLoading.value = false; + } + } +} + +function scrollToLatest() { + timelineRef.value?.scrollToBottom(); +} + async function openExecutionDetail() { detailVisible.value = true; if (!running.value) { @@ -1051,6 +1419,7 @@ function executionTraceText(

{{ descriptor.title || '工作流' }}