feat(workflow): 添加任务结果轮询和手动查询功能

- 新增 GET /v2/ai/order/v2/queryResult 接口用于手动查询任务结果
- 实现 WorkflowTaskClient 调用外部工作流服务查询任务状态
- 创建 WorkflowTaskResultHandler 统一处理任务结果逻辑
- 添加定时任务每5分钟轮询超时的执行中任务防止通知遗漏
- 将分布式锁从 JDBC 迁移到 Redis 提升性能
- 更新配置文件添加 workflow_host 配置项
- 完善 API 文档详细说明查询接口的使用方法和响应格式

doc/AI_ORDER_V2_API.md

@@ -196,3 +196,114 @@ GET /v2/ai/order/v2/get/ORD_202603090001
   "timestamp": 1773033000000
 }
 ```
+
+---
+
+## 4. 手动查询任务结果
+
+*   **URL**: `/v2/ai/order/v2/queryResult`
+*   **Method**: `GET`
+*   **描述**: 根据 taskId 主动向外部 Workflow 服务查询任务结果，并自动处理（更新订单、相册、推送通知）。用于 Redis 通知遗漏时的手动补偿。
+
+### 请求参数
+
+| 字段名 | 类型 | 必填 | 说明 |
+| :--- | :--- | :--- | :--- |
+| `taskId` | string | 是 | 外部任务ID，例如 `2b80cfe9-eb90-44c2-9201-55ecff33a85d` |
+
+### 响应参数（`data` 字段）
+
+返回外部 Workflow 服务原始响应 `WorkflowTaskMessage` 对象：
+
+| 字段名 | 类型 | 说明 |
+| :--- | :--- | :--- |
+| `taskId` | string | 任务ID |
+| `status` | string | 任务状态（`successes`=成功, `failed`=失败, `running`=执行中） |
+| `message` | string | 消息/错误信息 |
+| `type` | string | 任务类型 |
+| `data` | object | 任务结果数据 |
+| `data.images` | string[] | 生成的图片URL列表 |
+| `data.videos` | string[] | 生成的视频URL列表 |
+| `request` | object | 任务请求参数 |
+| `request.template` | string | 模版名称 |
+| `request.inputs` | object | 输入参数 |
+| `createTime` | long | 创建时间（时间戳） |
+| `updateTime` | long | 更新时间（时间戳） |
+
+### 处理逻辑
+
+1. 调用外部接口 `GET {workflow_host}/result?taskId=xxx` 获取任务结果
+2. 如果 `status` 为 `successes`：更新订单状态为完成、更新相册、推送用户通知
+3. 如果 `status` 为失败状态：执行退款并标记订单/相册失败
+4. 如果 `status` 为 `running`/`pending`/`processing` 等执行中状态：跳过处理，仅返回当前状态
+
+### 示例请求
+
+```http
+GET /v2/ai/order/v2/queryResult?taskId=2b80cfe9-eb90-44c2-9201-55ecff33a85d
+```
+
+### 示例响应（任务已完成）
+
+```json
+{
+  "code": 200,
+  "message": "success",
+  "success": true,
+  "data": {
+    "taskId": "2b80cfe9-eb90-44c2-9201-55ecff33a85d",
+    "status": "successes",
+    "message": null,
+    "type": "workflow",
+    "data": {
+      "images": [
+        "https://cdn.example.com/result1.jpg",
+        "https://cdn.example.com/result2.jpg"
+      ],
+      "videos": null
+    },
+    "request": {
+      "template": "spring_photo",
+      "inputs": {
+        "style": "natural"
+      }
+    },
+    "createTime": 1773033000,
+    "updateTime": 1773033300
+  },
+  "timestamp": 1773033600000
+}
+```
+
+### 示例响应（任务仍在执行中）
+
+```json
+{
+  "code": 200,
+  "message": "success",
+  "success": true,
+  "data": {
+    "taskId": "2b80cfe9-eb90-44c2-9201-55ecff33a85d",
+    "status": "running",
+    "message": null,
+    "type": "workflow",
+    "data": null,
+    "request": null,
+    "createTime": 1773033000,
+    "updateTime": null
+  },
+  "timestamp": 1773033600000
+}
+```
+
+### 示例响应（查询异常）
+
+```json
+{
+  "code": 500,
+  "message": "查询异常: Connection refused",
+  "success": false,
+  "data": null,
+  "timestamp": 1773033600000
+}
+```

pom.xml

@@ -155,10 +155,9 @@
             <version>${shedlock.version}</version>
         </dependency>
 
-        <!-- JDBC + MySQL -->
         <dependency>
             <groupId>net.javacrumbs.shedlock</groupId>
-            <artifactId>shedlock-provider-jdbc-template</artifactId>
+            <artifactId>shedlock-provider-redis-spring</artifactId>
             <version>${shedlock.version}</version>
         </dependency>
 

src/main/java/com/aitori/client/WorkflowTaskClient.java

@@ -0,0 +1,13 @@
+package com.aitori.client;
+
+import com.aitori.entity.message.WorkflowTaskMessage;
+import com.dtflys.forest.annotation.BaseRequest;
+import com.dtflys.forest.annotation.Get;
+import com.dtflys.forest.annotation.Query;
+
+@BaseRequest(baseURL = "${workflow_host}")
+public interface WorkflowTaskClient {
+
+    @Get(url = "/result")
+    WorkflowTaskMessage getTaskResult(@Query("taskId") String taskId);
+}

src/main/java/com/aitori/config/SchedulerConfig.java

@@ -1,27 +1,20 @@
 package com.aitori.config;
 
 import net.javacrumbs.shedlock.core.LockProvider;
-import net.javacrumbs.shedlock.provider.jdbctemplate.JdbcTemplateLockProvider;
+import net.javacrumbs.shedlock.provider.redis.spring.RedisLockProvider;
 import net.javacrumbs.shedlock.spring.annotation.EnableSchedulerLock;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
-import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.data.redis.connection.RedisConnectionFactory;
 import org.springframework.scheduling.annotation.EnableScheduling;
 
-import javax.sql.DataSource;
-
 @Configuration
 @EnableScheduling
 @EnableSchedulerLock(defaultLockAtMostFor = "PT30M")
 public class SchedulerConfig {
 
     @Bean
-    public LockProvider lockProvider(DataSource dataSource) {
-        return new JdbcTemplateLockProvider(
-                JdbcTemplateLockProvider.Configuration.builder()
-                        .withJdbcTemplate(new JdbcTemplate(dataSource))
-                        .usingDbTime()
-                        .build()
-        );
+    public LockProvider lockProvider(RedisConnectionFactory connectionFactory) {
+        return new RedisLockProvider(connectionFactory);
     }
-}
+}

src/main/java/com/aitori/controller/TAiOrderV2Controller.java

@@ -1,16 +1,23 @@
 package com.aitori.controller;
 
+import com.aitori.client.WorkflowTaskClient;
+import com.aitori.dto.R;
 import com.aitori.entity.TAiOrderV2;
+import com.aitori.entity.message.WorkflowTaskMessage;
 import com.aitori.entity.request.AiOrderV2QueryRequest;
 import com.aitori.entity.vo.AiOrderV2VO;
 import com.aitori.service.TAiOrderV2Service;
+import com.aitori.service.WorkflowTaskResultHandler;
 import com.baomidou.mybatisplus.core.metadata.IPage;
 import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
 import io.swagger.v3.oas.annotations.Operation;
+import io.swagger.v3.oas.annotations.Parameter;
 import io.swagger.v3.oas.annotations.tags.Tag;
 import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
 import org.springframework.web.bind.annotation.*;
 
+@Slf4j
 @RestController
 @RequestMapping("/v2/ai/order/v2")
 @Tag(name = "订单V2管理接口")
@@ -18,6 +25,8 @@ import org.springframework.web.bind.annotation.*;
 public class TAiOrderV2Controller {
 
     private final TAiOrderV2Service aiOrderV2Service;
+    private final WorkflowTaskClient workflowTaskClient;
+    private final WorkflowTaskResultHandler resultHandler;
 
     @GetMapping("/list")
     @Operation(summary = "订单V2列表", description = "分页查询订单V2列表")
@@ -31,4 +40,21 @@ public class TAiOrderV2Controller {
     public TAiOrderV2 get(@PathVariable String id) {
         return aiOrderV2Service.getById(id);
     }
+
+    @GetMapping("/queryResult")
+    @Operation(summary = "手动查询任务结果", description = "根据taskId查询外部任务结果并处理，用于通知遗漏时手动补偿")
+    public R<WorkflowTaskMessage> queryResult(@Parameter(description = "外部任务ID", required = true, example = "2b80cfe9-eb90-44c2-9201-55ecff33a85d") @RequestParam String taskId) {
+        log.info("手动查询任务结果, taskId={}", taskId);
+        try {
+            WorkflowTaskMessage result = workflowTaskClient.getTaskResult(taskId);
+            if (result == null) {
+                return R.fail("外部接口未返回结果, taskId=" + taskId);
+            }
+            resultHandler.handlePolledResult(result);
+            return R.success(result);
+        } catch (Exception e) {
+            log.error("手动查询任务结果异常, taskId={}", taskId, e);
+            return R.fail("查询异常: " + e.getMessage());
+        }
+    }
 }

src/main/java/com/aitori/entity/message/WorkflowTaskMessage.java

@@ -1,42 +1,60 @@
 package com.aitori.entity.message;
 
 import com.fasterxml.jackson.annotation.JsonAlias;
+import io.swagger.v3.oas.annotations.media.Schema;
 import lombok.Data;
 
 import java.util.List;
 import java.util.Map;
 
 @Data
+@Schema(description = "Workflow任务消息")
 public class WorkflowTaskMessage {
 
     @JsonAlias("task_id")
+    @Schema(description = "任务ID")
     private String taskId;
 
+    @Schema(description = "任务状态 (successes=成功, failed=失败, running=执行中)")
     private String status;
 
+    @Schema(description = "消息/错误信息")
     private String message;
 
+    @Schema(description = "任务类型")
     private String type;
 
+    @Schema(description = "任务结果数据")
     private TaskData data;
 
+    @Schema(description = "任务请求参数")
     private TaskRequest request;
 
     @JsonAlias("create_time")
+    @Schema(description = "创建时间(时间戳)")
     private Long createTime;
 
     @JsonAlias("update_time")
+    @Schema(description = "更新时间(时间戳)")
     private Long updateTime;
 
     @Data
+    @Schema(description = "任务结果数据")
     public static class TaskData {
+        @Schema(description = "生成的图片URL列表")
         private List<String> images;
+
+        @Schema(description = "生成的视频URL列表")
         private List<String> videos;
     }
 
     @Data
+    @Schema(description = "任务请求参数")
     public static class TaskRequest {
+        @Schema(description = "模版名称")
         private String template;
+
+        @Schema(description = "输入参数")
         private Map<String, Object> inputs;
     }
 }

src/main/java/com/aitori/listener/WorkflowTaskMessageListener.java

@@ -1,16 +1,7 @@
 package com.aitori.listener;
 
-import com.aitori.entity.TAiAlbum;
-import com.aitori.entity.TAiOrderV2;
-import com.aitori.entity.TAiStyle;
 import com.aitori.entity.message.WorkflowTaskMessage;
-import com.aitori.service.TAiAlbumService;
-import com.aitori.service.TAiOrderV2Service;
-import com.aitori.service.TAiStyleService;
-import com.aitori.service.TUserMessageService;
-import com.aitori.utils.VideoThumbnailUtil;
-import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
-import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
+import com.aitori.service.WorkflowTaskResultHandler;
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import lombok.RequiredArgsConstructor;
@@ -20,24 +11,12 @@ import org.springframework.data.redis.connection.Message;
 import org.springframework.data.redis.connection.MessageListener;
 import org.springframework.stereotype.Component;
 
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.LinkedHashMap;
-import java.util.List;
-import java.util.Map;
-
 @Slf4j
 @Component
 @RequiredArgsConstructor
 public class WorkflowTaskMessageListener implements MessageListener {
 
-    private static final String STATUS_SUCCESS = "successes";
-    private static final int MSG_TYPE_ORDER_V2_FINISH = 11;
-
-    private final TAiOrderV2Service aiOrderV2Service;
-    private final TAiAlbumService aiAlbumService;
-    private final TAiStyleService aiStyleService;
-    private final TUserMessageService userMessageService;
+    private final WorkflowTaskResultHandler resultHandler;
     private final ObjectMapper objectMapper;
     private final AsyncTaskExecutor asyncTaskExecutor;
 
@@ -57,214 +36,10 @@ public class WorkflowTaskMessageListener implements MessageListener {
 
         asyncTaskExecutor.execute(() -> {
             try {
-                handleTaskMessage(taskMessage);
+                resultHandler.handleNotification(taskMessage);
             } catch (Exception e) {
                 log.error("处理 WorkflowTask 消息异常, taskId: {}", taskMessage.getTaskId(), e);
             }
         });
     }
-
-    private void handleTaskMessage(WorkflowTaskMessage taskMessage) {
-        String taskId = taskMessage.getTaskId();
-        if (taskId == null || taskId.isBlank()) {
-            log.warn("WorkflowTask 消息缺少 taskId, 跳过处理");
-            return;
-        }
-
-        TAiOrderV2 order = aiOrderV2Service.getOne(
-                new LambdaQueryWrapper<TAiOrderV2>().eq(TAiOrderV2::getTaskId, taskId)
-        );
-        if (order == null) {
-            log.warn("未找到 taskId={} 对应的订单, 跳过处理", taskId);
-            return;
-        }
-
-        if (order.getTaskStatus() != null && (order.getTaskStatus() == 2 || order.getTaskStatus() == 3)) {
-            log.info("订单已处理, taskId={}, orderId={}, taskStatus={}, 跳过重复处理",
-                    taskId, order.getId(), order.getTaskStatus());
-            return;
-        }
-
-        if (STATUS_SUCCESS.equals(taskMessage.getStatus())) {
-            handleSuccess(taskId, order, taskMessage.getData());
-        } else {
-            aiOrderV2Service.refundOrder(order, taskMessage.getMessage());
-            failAlbumIfExists(order.getId());
-            log.info("任务失败已退款, taskId={}, orderId={}, status={}, message={}",
-                    taskId, order.getId(), taskMessage.getStatus(), taskMessage.getMessage());
-        }
-    }
-
-    private void handleSuccess(String taskId, TAiOrderV2 order, WorkflowTaskMessage.TaskData data) {
-        String imagesArrayJson = buildImagesArrayJson(data != null ? data.getImages() : null);
-        String videosArrayJson = buildVideosArrayJson(data != null ? data.getVideos() : null);
-
-        String resultJson = assembleResultJson(imagesArrayJson, videosArrayJson);
-        LambdaUpdateWrapper<TAiOrderV2> updateWrapper = new LambdaUpdateWrapper<>();
-        updateWrapper.eq(TAiOrderV2::getTaskId, taskId)
-                .set(TAiOrderV2::getTaskStatus, 2)
-                .set(TAiOrderV2::getResult, resultJson)
-                .set(TAiOrderV2::getErrorMessage, null);
-        aiOrderV2Service.update(updateWrapper);
-        log.info("任务完成, taskId={}, orderId={}", taskId, order.getId());
-
-        TAiAlbum album = updateAlbumIfExists(order.getId(), imagesArrayJson, videosArrayJson);
-        notifyUser(order, album);
-    }
-
-    /**
-     * 组装 result JSON，格式:
-     * images → {"images":"[\"url1\",\"url2\"]"}
-     * video  → {"videos":"[{\"url\":\"...\",\"thumbnailUrl\":\"...\"}]"}
-     */
-    private String assembleResultJson(String imagesArrayJson, String videosArrayJson) {
-        Map<String, String> result = new LinkedHashMap<>();
-        if (imagesArrayJson != null) {
-            result.put("images", imagesArrayJson);
-        }
-        if (videosArrayJson != null) {
-            result.put("videos", videosArrayJson);
-        }
-        if (result.isEmpty()) {
-            return null;
-        }
-        try {
-            return objectMapper.writeValueAsString(result);
-        } catch (JsonProcessingException e) {
-            log.error("序列化 result 失败", e);
-            return null;
-        }
-    }
-
-    private String buildImagesArrayJson(List<String> images) {
-        if (images == null || images.isEmpty()) {
-            return null;
-        }
-        try {
-            return objectMapper.writeValueAsString(images);
-        } catch (JsonProcessingException e) {
-            log.error("序列化 images 失败", e);
-            return null;
-        }
-    }
-
-    /**
-     * 对每个视频 URL 用 ffmpeg 截取第1秒缩略图，上传后构建带 thumbnailUrl 的 JSON 数组字符串
-     */
-    private String buildVideosArrayJson(List<String> videoUrls) {
-        if (videoUrls == null || videoUrls.isEmpty()) {
-            return null;
-        }
-
-        List<Map<String, String>> videoItems = new ArrayList<>();
-        for (String videoUrl : videoUrls) {
-            Map<String, String> item = new HashMap<>();
-            item.put("url", videoUrl);
-
-            String thumbnailUrl = VideoThumbnailUtil.extractAndUpload(videoUrl);
-            if (thumbnailUrl != null) {
-                item.put("thumbnailUrl", thumbnailUrl);
-            }
-            videoItems.add(item);
-        }
-
-        try {
-            return objectMapper.writeValueAsString(videoItems);
-        } catch (JsonProcessingException e) {
-            log.error("序列化 videos 失败", e);
-            return null;
-        }
-    }
-
-    /**
-     * 根据 orderId 查找相册，若存在则将预计算好的 images/videos JSON 更新到相册的 pictures/videos 字段
-     */
-    private TAiAlbum updateAlbumIfExists(String orderId, String imagesArrayJson, String videosArrayJson) {
-        if (orderId == null) {
-            return null;
-        }
-
-        TAiAlbum album = aiAlbumService.getOne(
-                new LambdaQueryWrapper<TAiAlbum>().eq(TAiAlbum::getOrderId, orderId)
-        );
-        if (album == null) {
-            log.info("orderId={} 无关联相册, 跳过相册更新", orderId);
-            return null;
-        }
-
-        LambdaUpdateWrapper<TAiAlbum> albumUpdate = new LambdaUpdateWrapper<>();
-        albumUpdate.eq(TAiAlbum::getId, album.getId());
-
-        boolean needUpdate = false;
-
-        if (imagesArrayJson != null) {
-            albumUpdate.set(TAiAlbum::getPictures, imagesArrayJson);
-            needUpdate = true;
-        }
-        if (videosArrayJson != null) {
-            albumUpdate.set(TAiAlbum::getVideos, videosArrayJson);
-            needUpdate = true;
-        }
-
-        if (needUpdate) {
-            albumUpdate.set(TAiAlbum::getStatus, 3);
-            aiAlbumService.update(albumUpdate);
-            log.info("相册更新成功, albumId={}, orderId={}", album.getId(), orderId);
-        }
-        return album;
-    }
-
-    private void failAlbumIfExists(String orderId) {
-        if (orderId == null) {
-            return;
-        }
-        TAiAlbum album = aiAlbumService.getOne(
-                new LambdaQueryWrapper<TAiAlbum>().eq(TAiAlbum::getOrderId, orderId)
-        );
-        if (album == null) {
-            return;
-        }
-        LambdaUpdateWrapper<TAiAlbum> albumUpdate = new LambdaUpdateWrapper<>();
-        albumUpdate.eq(TAiAlbum::getId, album.getId())
-                .set(TAiAlbum::getStatus, 0);
-        aiAlbumService.update(albumUpdate);
-        log.info("相册标记失败, albumId={}, orderId={}", album.getId(), orderId);
-    }
-
-    private void notifyUser(TAiOrderV2 order, TAiAlbum album) {
-        if (order.getUserId() == null || order.getStyleId() == null) {
-            log.warn("订单缺少 userId 或 styleId, 跳过消息通知, orderId={}", order.getId());
-            return;
-        }
-
-        TAiStyle style = aiStyleService.getById(order.getStyleId());
-        if (style == null) {
-            log.warn("未找到 styleId={} 对应的风格, 跳过消息通知", order.getStyleId());
-            return;
-        }
-
-        String styleName = style.getName();
-        log.info("styleName:{}", styleName);
-
-        String title;
-        String content = switch (order.getOrderLang() != null ? order.getOrderLang() : "") {
-            case "zh_CN" -> {
-                title = "图像生成已完成";
-                yield String.format("「%s」中所述修改相应参数的值。现在可以从生成的相册中查看和下载图像。", styleName);
-            }
-            case "en_US" -> {
-                title = "Image creation completed";
-                yield String.format("「%s」 An image was generated. You can view and download images from generated albums.", styleName);
-            }
-            default -> {
-                title = "画像生成は完了しました";
-                yield String.format("「%s」の画像が生成されました。生成されたアルバムから画像を表示およびダウンロードできるようになりました。", styleName);
-            }
-        };
-
-        Integer albumId = album != null ? album.getId().intValue() : null;
-        log.info("发送消息给用户:{}, orderLang:{}", order.getUserId(), order.getOrderLang());
-        userMessageService.sendMessage(order.getUserId(), title, content, order.getId(),
-                MSG_TYPE_ORDER_V2_FINISH, albumId, style.getCoverImg());
-    }
 }

src/main/java/com/aitori/scheduler/WorkflowTaskScheduler.java

@@ -0,0 +1,75 @@
+package com.aitori.scheduler;
+
+import com.aitori.client.WorkflowTaskClient;
+import com.aitori.entity.TAiOrderV2;
+import com.aitori.entity.message.WorkflowTaskMessage;
+import com.aitori.service.TAiOrderV2Service;
+import com.aitori.service.WorkflowTaskResultHandler;
+import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import net.javacrumbs.shedlock.spring.annotation.SchedulerLock;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
+
+import java.time.LocalDateTime;
+import java.util.List;
+
+@Slf4j
+@Component
+@RequiredArgsConstructor
+public class WorkflowTaskScheduler {
+
+    private final TAiOrderV2Service aiOrderV2Service;
+    private final WorkflowTaskClient workflowTaskClient;
+    private final WorkflowTaskResultHandler resultHandler;
+
+    /**
+     * 每5分钟轮询一次：查询创建时间超过5分钟且仍处于"执行中"的订单，
+     * 主动向外部接口查询结果，防止 Redis 通知遗漏导致订单卡住。
+     */
+    @Scheduled(fixedRate = 5 * 60 * 1000, initialDelay = 60 * 1000)
+    @SchedulerLock(name = "pollPendingTaskResults", lockAtMostFor = "PT10M", lockAtLeastFor = "PT1M")
+    public void pollPendingTaskResults() {
+        LocalDateTime threshold = LocalDateTime.now().minusMinutes(5);
+        List<TAiOrderV2> pendingOrders = aiOrderV2Service.list(
+                new LambdaQueryWrapper<TAiOrderV2>()
+                        .eq(TAiOrderV2::getTaskStatus, 1)
+                        .isNotNull(TAiOrderV2::getTaskId)
+                        .le(TAiOrderV2::getCreateTime, threshold)
+        );
+
+        if (pendingOrders.isEmpty()) {
+            log.debug("无超时执行中的任务需要轮询");
+            return;
+        }
+
+        log.info("开始轮询超时任务, 共{}条", pendingOrders.size());
+        int processed = 0;
+        int stillRunning = 0;
+        int failed = 0;
+
+        for (TAiOrderV2 order : pendingOrders) {
+            try {
+                WorkflowTaskMessage result = workflowTaskClient.getTaskResult(order.getTaskId());
+                if (result == null) {
+                    log.warn("外部接口未返回结果, taskId={}, orderId={}", order.getTaskId(), order.getId());
+                    failed++;
+                    continue;
+                }
+                boolean done = resultHandler.handlePolledResult(result);
+                if (done) {
+                    processed++;
+                } else {
+                    stillRunning++;
+                }
+            } catch (Exception e) {
+                failed++;
+                log.error("轮询任务结果异常, taskId={}, orderId={}", order.getTaskId(), order.getId(), e);
+            }
+        }
+
+        log.info("轮询超时任务完成: 总数={}, 已处理={}, 仍执行中={}, 异常={}",
+                pendingOrders.size(), processed, stillRunning, failed);
+    }
+}

src/main/java/com/aitori/service/WorkflowTaskResultHandler.java

@@ -0,0 +1,257 @@
+package com.aitori.service;
+
+import com.aitori.entity.TAiAlbum;
+import com.aitori.entity.TAiOrderV2;
+import com.aitori.entity.TAiStyle;
+import com.aitori.entity.message.WorkflowTaskMessage;
+import com.aitori.utils.VideoThumbnailUtil;
+import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
+import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Service;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+@Slf4j
+@Service
+@RequiredArgsConstructor
+public class WorkflowTaskResultHandler {
+
+    private static final String STATUS_SUCCESS = "successes";
+    private static final int MSG_TYPE_ORDER_V2_FINISH = 11;
+    private static final Set<String> RUNNING_STATUSES = Set.of("submitted", "pending", "processing", "queued");
+
+    private final TAiOrderV2Service aiOrderV2Service;
+    private final TAiAlbumService aiAlbumService;
+    private final TAiStyleService aiStyleService;
+    private final TUserMessageService userMessageService;
+    private final ObjectMapper objectMapper;
+
+    /**
+     * 处理 Redis 通知的任务消息（非终态一律当作失败）
+     */
+    public void handleNotification(WorkflowTaskMessage taskMessage) {
+        handleTaskMessage(taskMessage, false);
+    }
+
+    /**
+     * 轮询/手动查询的任务结果处理，遇到非终态会跳过
+     *
+     * @return true 已处理（成功或失败），false 任务仍在执行中
+     */
+    public boolean handlePolledResult(WorkflowTaskMessage taskMessage) {
+        return handleTaskMessage(taskMessage, true);
+    }
+
+    /**
+     * @param skipIfRunning true 时遇到非终态状态跳过而不是当失败处理
+     * @return true 已处理
+     */
+    private boolean handleTaskMessage(WorkflowTaskMessage taskMessage, boolean skipIfRunning) {
+        String taskId = taskMessage.getTaskId();
+        if (taskId == null || taskId.isBlank()) {
+            log.warn("WorkflowTask 消息缺少 taskId, 跳过处理");
+            return false;
+        }
+
+        String status = taskMessage.getStatus();
+        if (skipIfRunning && (status == null || RUNNING_STATUSES.contains(status.toLowerCase()))) {
+            log.info("任务仍在执行中, taskId={}, status={}", taskId, status);
+            return false;
+        }
+
+        TAiOrderV2 order = aiOrderV2Service.getOne(
+                new LambdaQueryWrapper<TAiOrderV2>().eq(TAiOrderV2::getTaskId, taskId)
+        );
+        if (order == null) {
+            log.warn("未找到 taskId={} 对应的订单, 跳过处理", taskId);
+            return false;
+        }
+
+        if (order.getTaskStatus() != null && (order.getTaskStatus() == 2 || order.getTaskStatus() == 3)) {
+            log.info("订单已处理, taskId={}, orderId={}, taskStatus={}, 跳过重复处理",
+                    taskId, order.getId(), order.getTaskStatus());
+            return true;
+        }
+
+        if (STATUS_SUCCESS.equals(status)) {
+            handleSuccess(taskId, order, taskMessage.getData());
+        } else {
+            aiOrderV2Service.refundOrder(order, taskMessage.getMessage());
+            failAlbumIfExists(order.getId());
+            log.info("任务失败已退款, taskId={}, orderId={}, status={}, message={}",
+                    taskId, order.getId(), status, taskMessage.getMessage());
+        }
+        return true;
+    }
+
+    private void handleSuccess(String taskId, TAiOrderV2 order, WorkflowTaskMessage.TaskData data) {
+        String imagesArrayJson = buildImagesArrayJson(data != null ? data.getImages() : null);
+        String videosArrayJson = buildVideosArrayJson(data != null ? data.getVideos() : null);
+
+        String resultJson = assembleResultJson(imagesArrayJson, videosArrayJson);
+        LambdaUpdateWrapper<TAiOrderV2> updateWrapper = new LambdaUpdateWrapper<>();
+        updateWrapper.eq(TAiOrderV2::getTaskId, taskId)
+                .set(TAiOrderV2::getTaskStatus, 2)
+                .set(TAiOrderV2::getResult, resultJson)
+                .set(TAiOrderV2::getErrorMessage, null);
+        aiOrderV2Service.update(updateWrapper);
+        log.info("任务完成, taskId={}, orderId={}", taskId, order.getId());
+
+        TAiAlbum album = updateAlbumIfExists(order.getId(), imagesArrayJson, videosArrayJson);
+        notifyUser(order, album);
+    }
+
+    private String assembleResultJson(String imagesArrayJson, String videosArrayJson) {
+        Map<String, String> result = new LinkedHashMap<>();
+        if (imagesArrayJson != null) {
+            result.put("images", imagesArrayJson);
+        }
+        if (videosArrayJson != null) {
+            result.put("videos", videosArrayJson);
+        }
+        if (result.isEmpty()) {
+            return null;
+        }
+        try {
+            return objectMapper.writeValueAsString(result);
+        } catch (JsonProcessingException e) {
+            log.error("序列化 result 失败", e);
+            return null;
+        }
+    }
+
+    private String buildImagesArrayJson(List<String> images) {
+        if (images == null || images.isEmpty()) {
+            return null;
+        }
+        try {
+            return objectMapper.writeValueAsString(images);
+        } catch (JsonProcessingException e) {
+            log.error("序列化 images 失败", e);
+            return null;
+        }
+    }
+
+    private String buildVideosArrayJson(List<String> videoUrls) {
+        if (videoUrls == null || videoUrls.isEmpty()) {
+            return null;
+        }
+
+        List<Map<String, String>> videoItems = new ArrayList<>();
+        for (String videoUrl : videoUrls) {
+            Map<String, String> item = new HashMap<>();
+            item.put("url", videoUrl);
+
+            String thumbnailUrl = VideoThumbnailUtil.extractAndUpload(videoUrl);
+            if (thumbnailUrl != null) {
+                item.put("thumbnailUrl", thumbnailUrl);
+            }
+            videoItems.add(item);
+        }
+
+        try {
+            return objectMapper.writeValueAsString(videoItems);
+        } catch (JsonProcessingException e) {
+            log.error("序列化 videos 失败", e);
+            return null;
+        }
+    }
+
+    private TAiAlbum updateAlbumIfExists(String orderId, String imagesArrayJson, String videosArrayJson) {
+        if (orderId == null) {
+            return null;
+        }
+
+        TAiAlbum album = aiAlbumService.getOne(
+                new LambdaQueryWrapper<TAiAlbum>().eq(TAiAlbum::getOrderId, orderId)
+        );
+        if (album == null) {
+            log.info("orderId={} 无关联相册, 跳过相册更新", orderId);
+            return null;
+        }
+
+        LambdaUpdateWrapper<TAiAlbum> albumUpdate = new LambdaUpdateWrapper<>();
+        albumUpdate.eq(TAiAlbum::getId, album.getId());
+
+        boolean needUpdate = false;
+
+        if (imagesArrayJson != null) {
+            albumUpdate.set(TAiAlbum::getPictures, imagesArrayJson);
+            needUpdate = true;
+        }
+        if (videosArrayJson != null) {
+            albumUpdate.set(TAiAlbum::getVideos, videosArrayJson);
+            needUpdate = true;
+        }
+
+        if (needUpdate) {
+            albumUpdate.set(TAiAlbum::getStatus, 3);
+            aiAlbumService.update(albumUpdate);
+            log.info("相册更新成功, albumId={}, orderId={}", album.getId(), orderId);
+        }
+        return album;
+    }
+
+    private void failAlbumIfExists(String orderId) {
+        if (orderId == null) {
+            return;
+        }
+        TAiAlbum album = aiAlbumService.getOne(
+                new LambdaQueryWrapper<TAiAlbum>().eq(TAiAlbum::getOrderId, orderId)
+        );
+        if (album == null) {
+            return;
+        }
+        LambdaUpdateWrapper<TAiAlbum> albumUpdate = new LambdaUpdateWrapper<>();
+        albumUpdate.eq(TAiAlbum::getId, album.getId())
+                .set(TAiAlbum::getStatus, 0);
+        aiAlbumService.update(albumUpdate);
+        log.info("相册标记失败, albumId={}, orderId={}", album.getId(), orderId);
+    }
+
+    private void notifyUser(TAiOrderV2 order, TAiAlbum album) {
+        if (order.getUserId() == null || order.getStyleId() == null) {
+            log.warn("订单缺少 userId 或 styleId, 跳过消息通知, orderId={}", order.getId());
+            return;
+        }
+
+        TAiStyle style = aiStyleService.getById(order.getStyleId());
+        if (style == null) {
+            log.warn("未找到 styleId={} 对应的风格, 跳过消息通知", order.getStyleId());
+            return;
+        }
+
+        String styleName = style.getName();
+        log.info("styleName:{}", styleName);
+
+        String title;
+        String content = switch (order.getOrderLang() != null ? order.getOrderLang() : "") {
+            case "zh_CN" -> {
+                title = "图像生成已完成";
+                yield String.format("「%s」中所述修改相应参数的值。现在可以从生成的相册中查看和下载图像。", styleName);
+            }
+            case "en_US" -> {
+                title = "Image creation completed";
+                yield String.format("「%s」 An image was generated. You can view and download images from generated albums.", styleName);
+            }
+            default -> {
+                title = "画像生成は完了しました";
+                yield String.format("「%s」の画像が生成されました。生成されたアルバムから画像を表示およびダウンロードできるようになりました。", styleName);
+            }
+        };
+
+        Integer albumId = album != null ? album.getId().intValue() : null;
+        log.info("发送消息给用户:{}, orderLang:{}", order.getUserId(), order.getOrderLang());
+        userMessageService.sendMessage(order.getUserId(), title, content, order.getId(),
+                MSG_TYPE_ORDER_V2_FINISH, albumId, style.getCoverImg());
+    }
+}

src/main/resources/application.yaml

@@ -80,6 +80,7 @@ forest:
     model_health_host: https://api.sensenova.cn
     auth_host: https://sit-platform.senseauto.com
     product_name: sensechat
+    workflow_host: ${WORKFLOW_HOST:http://localhost:8080}
 
 
 # ---  actuator  ---