feat(gateway): 实现API网关响应模板和请求映射增强功能

- 添加getLastHttpStepId方法获取最后HTTP步骤ID用于响应体设置判断
- 实现响应模板功能支持自定义JSON响应格式通过responseTemplate配置
- 增强请求映射逻辑支持查询参数、请求头和请求体的灵活映射与合并
- 优化HTTP步骤处理器中的请求载荷转换和响应映射逻辑
- 在START节点中支持公共参数自动继承到后续HTTP步骤
- 添加查询参数环境变量绑定便于JSONata表达式访问
- 实现向后兼容的默认响应格式保持现有功能不变

zt-module-databus/zt-module-databus-server/src/main/java/com/zt/plat/module/databus/framework/integration/gateway/core/ApiGatewayExecutionService.java

@@ -50,6 +50,7 @@ public class ApiGatewayExecutionService {
     private static final String HEADER_REMOTE_ADDRESS = org.springframework.integration.http.HttpHeaders.PREFIX + "remoteAddress";
     private static final String LOCAL_DEBUG_REMOTE_ADDRESS = "127.0.0.1";
     private static final String ATTR_DEBUG_INVOKE = "gatewayDebugInvoke";
+    private static final String ATTR_API_AGGREGATE = "apiDefinitionAggregate";
 
     private final ApiGatewayRequestMapper requestMapper;
     private final ApiFlowDispatcher apiFlowDispatcher;
@@ -78,14 +79,22 @@ public class ApiGatewayExecutionService {
         ApiInvocationContext context = message.getPayload();
         accessLogger.onRequest(context);
         ApiInvocationContext responseContext;
-        ApiDefinitionAggregate debugAggregate = null;
+        ApiDefinitionAggregate aggregate = null;
         try {
             enforceCredentialAuthorization(context);
             if (Boolean.TRUE.equals(context.getAttributes().get(ATTR_DEBUG_INVOKE))) {
-                debugAggregate = resolveDebugAggregate(context);
+                aggregate = resolveDebugAggregate(context);
+            } else {
+                // 对于非调试调用，也需要获取 aggregate 以便后续应用响应模板
+                aggregate = apiDefinitionService.findByCodeAndVersion(context.getApiCode(), context.getApiVersion())
+                        .orElse(null);
             }
-            if (debugAggregate != null) {
+            // 将 aggregate 存储到上下文中，供响应构建时使用
-                responseContext = apiFlowDispatcher.dispatchWithAggregate(debugAggregate, context);
+            if (aggregate != null) {
+                context.getAttributes().put(ATTR_API_AGGREGATE, aggregate);
+            }
+            if (aggregate != null && Boolean.TRUE.equals(context.getAttributes().get(ATTR_DEBUG_INVOKE))) {
+                responseContext = apiFlowDispatcher.dispatchWithAggregate(aggregate, context);
             } else {
                 responseContext = apiFlowDispatcher.dispatch(context.getApiCode(), context.getApiVersion(), context);
             }
@@ -239,12 +248,64 @@ public class ApiGatewayExecutionService {
         String message = StringUtils.hasText(context.getResponseMessage())
                 ? context.getResponseMessage()
                 : HttpStatus.valueOf(status).getReasonPhrase();
-    return ApiGatewayResponse.builder()
+
-        .code(status)
+        // 尝试从上下文中获取 API 定义聚合对象
-        .message(message)
+        ApiDefinitionAggregate aggregate = (ApiDefinitionAggregate) context.getAttributes().get(ATTR_API_AGGREGATE);
-        .response(context.getResponseBody())
+        String responseTemplate = aggregate != null ? aggregate.getDefinition().getResponseTemplate() : null;
-        .traceId(TracerUtils.getTraceId())
+
-        .build();
+        // 如果配置了响应模板，则应用模板；否则使用默认格式（向后兼容）
+        if (StringUtils.hasText(responseTemplate)) {
+            return buildResponseWithTemplate(responseTemplate, context, status, message);
+        }
+
+        // 默认响应格式（向后兼容）
+        return ApiGatewayResponse.builder()
+            .code(status)
+            .message(message)
+            .response(context.getResponseBody())
+            .traceId(TracerUtils.getTraceId())
+            .build();
+    }
+
+    /**
+     * 根据响应模板构建响应对象
+     * 模板格式示例：{"code": 0, "message": "success", "data": {...}}
+     * 模板中的 data 字段会被实际响应数据替换，其他字段保持用户配置的原始值
+     */
+    private ApiGatewayResponse buildResponseWithTemplate(String templateJson, ApiInvocationContext context, int status, String message) {
+        try {
+            // 解析模板 JSON
+            Map<String, Object> template = objectMapper.readValue(templateJson, Map.class);
+
+            // 构建最终响应数据，保留模板中的所有字段
+            Map<String, Object> responseData = new LinkedHashMap<>(template);
+
+            // 只替换 data 字段为实际的响应体数据
+            // 其他字段（如 code、message）保持用户配置的原始值
+            if (responseData.containsKey("data")) {
+                responseData.put("data", context.getResponseBody());
+            }
+
+            // 返回使用模板结构的响应
+            // 注意：ApiGatewayResponse 的 code 和 message 是 HTTP 层面的状态
+            // responseData 中的 code 和 message 是业务层面的状态
+            return ApiGatewayResponse.builder()
+                .code(status)
+                .message(message)
+                .response(responseData)
+                .traceId(TracerUtils.getTraceId())
+                .build();
+
+        } catch (JsonProcessingException ex) {
+            log.warn("[API-PORTAL] 解析响应模板失败，使用默认格式: {}", ex.getMessage());
+            // 解析失败时回退到默认格式
+            return ApiGatewayResponse.builder()
+                .code(status)
+                .message(message)
+                .response(context.getResponseBody())
+                .traceId(TracerUtils.getTraceId())
+                .build();
+        }
     }
 
     /**

zt-module-databus/zt-module-databus-server/src/main/java/com/zt/plat/module/databus/framework/integration/gateway/domain/ApiDefinitionAggregate.java

@@ -40,4 +40,23 @@ public class ApiDefinitionAggregate {
         return credentialBindings == null ? Collections.emptyList() : credentialBindings;
     }
 
+    /**
+     * 获取最后一个 HTTP 步骤的 ID（order 最大的 HTTP 步骤）
+     * 用于判断是否需要设置 responseBody
+     */
+    public Long getLastHttpStepId() {
+        return getSteps().stream()
+                .filter(stepDef -> "HTTP".equalsIgnoreCase(stepDef.getStep().getType()))
+                .max((s1, s2) -> {
+                    Integer order1 = s1.getStep().getStepOrder();
+                    Integer order2 = s2.getStep().getStepOrder();
+                    if (order1 == null && order2 == null) return 0;
+                    if (order1 == null) return -1;
+                    if (order2 == null) return 1;
+                    return order1.compareTo(order2);
+                })
+                .map(stepDef -> stepDef.getStep().getId())
+                .orElse(null);
+    }
+
 }

zt-module-databus/zt-module-databus-server/src/main/java/com/zt/plat/module/databus/framework/integration/gateway/expression/GatewayExpressionHelper.java

@@ -96,30 +96,72 @@ public final class GatewayExpressionHelper {
     }
 
     private static void applyRequestMutations(ApiInvocationContext context, Map<?, ?> map) {
-        Object body = firstNonNull(map.get("requestBody"), map.get("body"));
+        // 处理查询参数
-        if (body != null) {
-            context.setRequestBody(body);
-        }
-        Object headers = map.get("requestHeaders");
-        if (headers instanceof Map<?, ?> headerMap) {
-            context.getRequestHeaders().putAll(toStringMap(headerMap));
-        }
         Object query = firstNonNull(map.get("requestQuery"), map.get("requestQueryParams"), map.get("query"));
         if (query instanceof Map<?, ?> queryMap) {
             context.getRequestQueryParams().putAll(toObjectMap(queryMap));
         }
+
+        // 处理请求头
+        Object headers = map.get("requestHeaders");
+        if (headers instanceof Map<?, ?> headerMap) {
+            context.getRequestHeaders().putAll(toStringMap(headerMap));
+        }
+
+        // 处理 body - 使用和 HTTP 步骤一致的逻辑
+        // 检查是否显式指定了 body 字段
+        boolean explicitBody = containsKeyIgnoreCase(map, "body", "payload", "requestBody");
+
+        if (explicitBody) {
+            // 如果显式指定了 body，提取并与原有 body 合并
+            Object body = firstNonNull(map.get("requestBody"), map.get("body"), map.get("payload"));
+            Object existingBody = context.getRequestBody();
+
+            if (body instanceof Map && existingBody instanceof Map) {
+                // 两者都是 Map，合并以保留未映射的字段
+                Map<String, Object> mergedBody = new LinkedHashMap<>(toObjectMap((Map<?, ?>) existingBody));
+                mergedBody.putAll(toObjectMap((Map<?, ?>) body));
+                context.setRequestBody(mergedBody);
+            } else if (body != null) {
+                // 直接替换
+                context.setRequestBody(body);
+            }
+        } else if (query == null) {
+            // 如果没有显式 body 字段，也没有 query 字段
+            // 说明整个 map 是新的 body，与原有 body 合并
+            Object existingBody = context.getRequestBody();
+            if (existingBody instanceof Map) {
+                Map<String, Object> mergedBody = new LinkedHashMap<>(toObjectMap((Map<?, ?>) existingBody));
+                mergedBody.putAll(toObjectMap(map));
+                context.setRequestBody(mergedBody);
+            } else {
+                // 原有 body 不是 Map，直接替换
+                context.setRequestBody(toObjectMap(map));
+            }
+        }
+        // 如果有 query 字段但没有显式 body 字段，保留原有 body（不做任何操作）
     }
 
     private static void applyResponseMutations(ApiInvocationContext context, Map<?, ?> map) {
         Object body = firstNonNull(map.get("responseBody"), map.get("body"));
+        Object status = map.get("responseStatus");
+        Object message = map.get("responseMessage");
+
+        // 如果有 body 字段，则提取 body
         if (body != null) {
             context.setResponseBody(body);
         }
-        Object status = map.get("responseStatus");
+        // 如果没有 body 字段，但也没有 responseStatus 或 responseMessage 字段
+        // 说明整个 map 就是响应体，直接设置为 responseBody
+        else if (status == null && message == null) {
+            context.setResponseBody(toObjectMap(map));
+        }
+        // 如果没有 body 字段，但有 responseStatus 或 responseMessage 字段
+        // 则只设置状态和消息，保留原有的 responseBody
+
         if (status != null) {
             context.setResponseStatus(asInteger(status));
         }
-        Object message = map.get("responseMessage");
         if (message != null) {
             context.setResponseMessage(String.valueOf(message));
         }
@@ -165,4 +207,18 @@ public final class GatewayExpressionHelper {
         }
         return null;
     }
+
+    private static boolean containsKeyIgnoreCase(Map<?, ?> source, String... keys) {
+        if (source == null || source.isEmpty()) {
+            return false;
+        }
+        for (String key : keys) {
+            for (Object entryKey : source.keySet()) {
+                if (key.equalsIgnoreCase(String.valueOf(entryKey))) {
+                    return true;
+                }
+            }
+        }
+        return false;
+    }
 }

zt-module-databus/zt-module-databus-server/src/main/java/com/zt/plat/module/databus/framework/integration/gateway/expression/JsonataExpressionEvaluator.java

@@ -43,6 +43,13 @@ public class JsonataExpressionEvaluator implements ExpressionEvaluator {
             environment.setVariable("vars", objectMapper.valueToTree(context.getVariables() == null ? java.util.Collections.emptyMap() : context.getVariables()));
             environment.setVariable("headers", objectMapper.valueToTree(context.getHeaders() == null ? java.util.Collections.emptyMap() : context.getHeaders()));
             environment.setVariable("ctx", objectMapper.valueToTree(context.getInvocation()));
+
+            // 直接绑定查询参数，方便访问
+            if (context.getInvocation() != null && context.getInvocation().getRequestQueryParams() != null) {
+                environment.setVariable("query", objectMapper.valueToTree(context.getInvocation().getRequestQueryParams()));
+            } else {
+                environment.setVariable("query", objectMapper.valueToTree(java.util.Collections.emptyMap()));
+            }
         } catch (EvaluateRuntimeException e) {
             throw ServiceExceptionUtil.exception(API_JSONATA_BIND_FAILED);
         }

zt-module-databus/zt-module-databus-server/src/main/java/com/zt/plat/module/databus/framework/integration/gateway/step/impl/HttpStepHandler.java

@@ -68,6 +68,7 @@ public class HttpStepHandler implements ApiStepHandler {
 
     private HttpRequestPayload coerceRequestPayload(Object evaluated, Object fallbackBody, Map<String, Object> fallbackQuery) {
         Map<String, Object> querySnapshot = new LinkedHashMap<>(fallbackQuery);
+
         if (evaluated == null) {
             return HttpRequestPayload.of(fallbackBody, querySnapshot);
         }
@@ -80,23 +81,47 @@ public class HttpStepHandler implements ApiStepHandler {
             mergeQueryParams(querySnapshot, multiValueMap);
             return HttpRequestPayload.of(fallbackBody, querySnapshot);
         }
-        if (evaluated instanceof Map<?, ?> map) {
+        if (evaluated instanceof Map<?, ?> evaluatedMap) {
-            Object queryPart = extractCaseInsensitive(map, "query", "queryParams", "params");
+            // 处理查询参数
+            Object queryPart = extractCaseInsensitive(evaluatedMap, "query", "queryParams", "params");
             if (queryPart != null) {
                 mergeQueryParams(querySnapshot, queryPart);
             }
-            boolean explicitBody = containsKeyIgnoreCase(map, "body", "payload");
+
+            // 检查是否显式指定了body字段
+            boolean explicitBody = containsKeyIgnoreCase(evaluatedMap, "body", "payload");
             Object body = explicitBody
-                    ? Optional.ofNullable(extractCaseInsensitive(map, "body", "payload")).orElse(fallbackBody)
+                    ? Optional.ofNullable(extractCaseInsensitive(evaluatedMap, "body", "payload")).orElse(fallbackBody)
                     : (queryPart != null ? fallbackBody : evaluated);
-            if (!explicitBody && queryPart == null) {
+
+            if (explicitBody) {
+                // 如果显式指定了body，将原始body和映射body合并
+                if (body instanceof Map && fallbackBody instanceof Map) {
+                    Map<String, Object> mergedBody = new LinkedHashMap<>((Map<String, Object>) fallbackBody);
+                    mergedBody.putAll((Map<String, Object>) body);
+                    return HttpRequestPayload.of(mergedBody, querySnapshot);
+                }
+                return HttpRequestPayload.of(body, querySnapshot);
+            } else if (queryPart != null) {
+                // 如果有查询参数部分但没有显式body，保留原始body
+                return HttpRequestPayload.of(fallbackBody, querySnapshot);
+            } else {
+                // 如果没有查询参数也没有显式body，说明整个evaluatedMap是新的body
+                if (fallbackBody instanceof Map) {
+                    // 将原始body和映射结果合并，保留原始字段
+                    Map<String, Object> mergedBody = new LinkedHashMap<>((Map<String, Object>) fallbackBody);
+                    ((Map<String, Object>) evaluatedMap).forEach((key, value) -> {
+                        mergedBody.put((String) key, value);
+                    });
+                    return HttpRequestPayload.of(mergedBody, querySnapshot);
+                }
                 return HttpRequestPayload.of(evaluated, querySnapshot);
             }
-            return HttpRequestPayload.of(body, querySnapshot);
         }
         return HttpRequestPayload.of(evaluated, querySnapshot);
     }
 
+
     @Override
     public GenericHandler<ApiInvocationContext> build(ApiDefinitionAggregate aggregate, ApiStepDefinition stepDefinition) {
         return (payload, headers) -> {
@@ -134,7 +159,7 @@ public class HttpStepHandler implements ApiStepHandler {
                         .success(true)
                         .elapsed(Duration.between(start, Instant.now()))
                         .build());
-                applyResponseMapping(stepDefinition, payload, headers, response);
+                applyResponseMapping(aggregate, stepDefinition, payload, headers, response);
             } catch (Exception ex) {
                 payload.addStepResult(ApiStepResult.builder()
                         .stepId(stepDefinition.getStep().getId())
@@ -157,22 +182,66 @@ public class HttpStepHandler implements ApiStepHandler {
         ExpressionSpec spec = ExpressionSpecParser.parse(stepDefinition.getStep().getRequestMappingExpr(), ExpressionTypeEnum.JSON);
         Map<String, Object> baseQuery = new LinkedHashMap<>(context.getRequestQueryParams());
         Object fallbackBody = context.getRequestBody();
+
         if (spec == null) {
             return HttpRequestPayload.of(fallbackBody, baseQuery);
         }
-        Object evaluated = expressionExecutor.evaluate(spec, context, fallbackBody, headers);
+
-        return coerceRequestPayload(evaluated, fallbackBody, baseQuery);
+        // 将查询参数合并到 payload 中，使表达式可以通过 $.query.k 访问
+        // 这样可以避免 JSONata 环境变量访问的限制
+        Object enrichedPayload = enrichPayloadWithQuery(fallbackBody, baseQuery);
+
+        Object evaluated = expressionExecutor.evaluate(spec, context, enrichedPayload, headers);
+
+        // 如果映射结果是Map且原始body也是Map，合并两个Map以保留未映射的字段
+        // 但如果映射结果包含显式的 body/query 结构，则不进行合并，直接传递
+        HttpRequestPayload result;
+        if (evaluated instanceof Map && fallbackBody instanceof Map) {
+            Map<String, Object> evaluatedMap = (Map<String, Object>) evaluated;
+
+            // 检查是否包含显式的 body 或 query 键（表示结构化映射）
+            boolean hasStructuredMapping = containsKeyIgnoreCase(evaluatedMap, "body", "payload")
+                    || containsKeyIgnoreCase(evaluatedMap, "query", "queryParams", "params");
+
+            if (hasStructuredMapping) {
+                // 有结构化映射，直接传递给 coerceRequestPayload 处理
+                result = coerceRequestPayload(evaluated, fallbackBody, baseQuery);
+            } else {
+                // 没有结构化映射，进行字段级合并以保留未映射的字段
+                Map<String, Object> originalBodyMap = (Map<String, Object>) fallbackBody;
+                Map<String, Object> mergedBody = new LinkedHashMap<>(originalBodyMap);
+                mergedBody.putAll(evaluatedMap);
+                result = coerceRequestPayload(mergedBody, fallbackBody, baseQuery);
+            }
+        } else {
+            result = coerceRequestPayload(evaluated, fallbackBody, baseQuery);
+        }
+
+        return result;
     }
 
-    private void applyResponseMapping(ApiStepDefinition stepDefinition, ApiInvocationContext context, Map<String, Object> headers, Object response) throws Exception {
+
+    private void applyResponseMapping(ApiDefinitionAggregate aggregate, ApiStepDefinition stepDefinition, ApiInvocationContext context, Map<String, Object> headers, Object response) throws Exception {
         ExpressionSpec spec = ExpressionSpecParser.parse(stepDefinition.getStep().getResponseMappingExpr(), ExpressionTypeEnum.JSON);
         if (spec == null) {
             context.setResponseBody(response);
             return;
         }
+
         Object mapped = expressionExecutor.evaluate(spec, context, response, headers);
+
+        // 判断当前步骤是否是最后一个 HTTP 步骤
+        Long lastHttpStepId = aggregate.getLastHttpStepId();
+        boolean isLastHttpStep = lastHttpStepId != null && lastHttpStepId.equals(stepDefinition.getStep().getId());
+
         if (mapped instanceof Map<?, ?> map) {
+            // 将映射结果放入 variables
             map.forEach((key, value) -> context.getVariables().put(String.valueOf(key), value));
+
+            // 如果是最后一个 HTTP 步骤，也要设置 responseBody，供 END 步骤使用
+            if (isLastHttpStep) {
+                context.setResponseBody(mapped);
+            }
         } else {
             context.setResponseBody(mapped);
         }
@@ -330,7 +399,8 @@ public class HttpStepHandler implements ApiStepHandler {
             mergeQueryParams(queryParams, requestPayload.body());
         }
         applyQueryParams(builder, queryParams);
-        return builder.build(true).toUri();
+        URI uri = builder.build(true).toUri();
+        return uri;
     }
 
     private Object extractCaseInsensitive(Map<?, ?> source, String... keys) {
@@ -435,4 +505,26 @@ public class HttpStepHandler implements ApiStepHandler {
         }
         return false;
     }
+
+    /**
+     * 将查询参数合并到 payload 中，使表达式可以通过 $.query.k 访问
+     * 这样可以避免 JSONata 环境变量访问的限制
+     */
+    private Object enrichPayloadWithQuery(Object originalPayload, Map<String, Object> queryParams) {
+        if (queryParams == null || queryParams.isEmpty()) {
+            return originalPayload;
+        }
+
+        Map<String, Object> enriched = new LinkedHashMap<>();
+
+        // 如果原始 payload 是 Map，复制所有字段
+        if (originalPayload instanceof Map) {
+            enriched.putAll((Map<String, Object>) originalPayload);
+        }
+
+        // 添加查询参数到 query 字段
+        enriched.put("query", new LinkedHashMap<>(queryParams));
+
+        return enriched;
+    }
 }

zt-module-databus/zt-module-databus-server/src/main/java/com/zt/plat/module/databus/framework/integration/gateway/step/impl/StartStepHandler.java

@@ -18,10 +18,19 @@ import org.springframework.stereotype.Component;
 
 import java.time.Duration;
 import java.time.Instant;
+import java.util.LinkedHashMap;
+import java.util.Map;
+
 import static com.zt.plat.module.databus.service.gateway.impl.GatewayServiceErrorCodeConstants.API_STEP_START_EXECUTION_FAILED;
 
 /**
  * Handler for START orchestration nodes.
+ *
+ * <p>增强功能：START 节点的映射结果会存入 context.requestBody，
+ * 使得后续 HTTP 步骤可以自动继承这些公共参数，避免重复配置。
+ *
+ * <p>注意：不将参数存入 context.variables，避免与后续步骤的响应结果冲突。
+ * variables 应该保留给步骤执行结果使用。
  */
 @Component
 @RequiredArgsConstructor
@@ -40,11 +49,22 @@ public class StartStepHandler implements ApiStepHandler {
             Instant start = Instant.now();
             Object snapshotBefore = GatewayExpressionHelper.snapshotRequest(payload);
             try {
-                ExpressionSpec spec = ExpressionSpecParser.parse(stepDefinition.getStep().getRequestMappingExpr(), ExpressionTypeEnum.JSON);
+                ExpressionSpec spec = ExpressionSpecParser.parse(
+                    stepDefinition.getStep().getRequestMappingExpr(),
+                    ExpressionTypeEnum.JSON
+                );
                 if (spec != null) {
-                    Object evaluated = expressionExecutor.evaluate(spec, payload, payload.getRequestBody(), headers);
+                    // 将查询参数合并到 payload 中，使表达式可以通过 $.query.k 访问
+                    // 这样和 HTTP 步骤的映射逻辑保持一致
+                    Object enrichedPayload = enrichPayloadWithQuery(payload.getRequestBody(), payload.getRequestQueryParams());
+
+                    Object evaluated = expressionExecutor.evaluate(spec, payload, enrichedPayload, headers);
+
+                    // 应用上下文变更（包括 requestBody、requestQuery 等）
+                    // 支持字段级映射，将外部字段名转换为内部字段名
                     GatewayExpressionHelper.applyContextMutations(payload, evaluated, true, false);
                 }
+
                 payload.addStepResult(ApiStepResult.builder()
                         .stepId(stepDefinition.getStep().getId())
                         .stepType(stepDefinition.getStep().getType())
@@ -70,4 +90,26 @@ public class StartStepHandler implements ApiStepHandler {
             return payload;
         };
     }
+
+    /**
+     * 将查询参数合并到 payload 中，使表达式可以通过 $.query.k 访问
+     * 这样可以避免 JSONata 环境变量访问的限制
+     */
+    private Object enrichPayloadWithQuery(Object originalPayload, Map<String, Object> queryParams) {
+        if (queryParams == null || queryParams.isEmpty()) {
+            return originalPayload;
+        }
+
+        Map<String, Object> enriched = new LinkedHashMap<>();
+
+        // 如果原始 payload 是 Map，复制所有字段
+        if (originalPayload instanceof Map) {
+            enriched.putAll((Map<String, Object>) originalPayload);
+        }
+
+        // 添加查询参数到 query 字段
+        enriched.put("query", new LinkedHashMap<>(queryParams));
+
+        return enriched;
+    }
 }