增强Debug EML SSE诊断与心跳

2026-07-12 17:43:20 +08:00
parent 5d82b29c53
commit 937aa569ff
10 changed files with 237 additions and 6 deletions
--- a/server/src/main/java/cn/nianxx/thhotel/platform/debug/service/impl/DebugEmlSuperAgentProperties.java
+++ b/server/src/main/java/cn/nianxx/thhotel/platform/debug/service/impl/DebugEmlSuperAgentProperties.java
@@ -1,5 +1,6 @@
 package cn.nianxx.thhotel.platform.debug.service.impl;

+import java.time.Duration;
 import org.springframework.boot.context.properties.ConfigurationProperties;
 import org.springframework.stereotype.Component;

@@ -16,6 +17,8 @@ public class DebugEmlSuperAgentProperties {
    private String accessKey = "";
    /** 允许上传的最大文件字节数。 */
    private long maxFileBytes = 10 * 1024 * 1024L;
+    /** Debug SSE 心跳间隔，用于避免长时间调用 SuperAgent 时中间链路空闲断开。 */
+    private Duration sseHeartbeatInterval = Duration.ofSeconds(15);

    public boolean isEnabled() {
        return enabled;
@@ -40,4 +43,12 @@ public class DebugEmlSuperAgentProperties {
    public void setMaxFileBytes(long maxFileBytes) {
        this.maxFileBytes = maxFileBytes;
    }
+
+    public Duration getSseHeartbeatInterval() {
+        return sseHeartbeatInterval;
+    }
+
+    public void setSseHeartbeatInterval(Duration sseHeartbeatInterval) {
+        this.sseHeartbeatInterval = sseHeartbeatInterval;
+    }
 }
--- a/server/src/main/java/cn/nianxx/thhotel/platform/debug/service/impl/DebugEmlSuperAgentRunServiceImpl.java
+++ b/server/src/main/java/cn/nianxx/thhotel/platform/debug/service/impl/DebugEmlSuperAgentRunServiceImpl.java
@@ -42,6 +42,7 @@ import java.net.URLDecoder;
 import java.nio.charset.StandardCharsets;
 import java.security.MessageDigest;
 import java.security.NoSuchAlgorithmException;
+import java.time.Duration;
 import java.time.LocalDateTime;
 import java.time.ZoneOffset;
 import java.time.format.DateTimeFormatter;
@@ -51,9 +52,16 @@ import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.TimeUnit;
 import java.util.function.Consumer;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.springframework.http.HttpStatus;
 import org.springframework.stereotype.Service;
 import org.springframework.web.multipart.MultipartFile;
@@ -66,12 +74,23 @@ import org.springframework.web.client.RestClientResponseException;
@Service
 public class DebugEmlSuperAgentRunServiceImpl implements DebugEmlSuperAgentRunService {

+    private static final Logger log = LoggerFactory.getLogger(DebugEmlSuperAgentRunServiceImpl.class);
    private static final String SOURCE_PROVIDER = "DEBUG_EML_UPLOAD";
    private static final String SOURCE_CHANNEL = "EMAIL";
    private static final String SCHEMA_VERSION = "debug-eml-upload-v1";
    private static final DateTimeFormatter DATE_FOLDER_FORMATTER = DateTimeFormatter.BASIC_ISO_DATE;
    private static final Pattern CID_REFERENCE_PATTERN = Pattern.compile(
            "(?i)cid:(?:<[^>]+>|%3c[^\\s\"'>]+%3e|[^\\s\"'>]+)");
+    private static final Pattern JSON_SECRET_PATTERN = Pattern.compile(
+            "(?i)(\"(?:api[_-]?key|token|secret|password|cookie|authorization)\"\\s*:\\s*\")[^\"]*(\")");
+    private static final Pattern AUTHORIZATION_BEARER_SECRET_PATTERN = Pattern.compile(
+            "(?i)(authorization\\s*[:=]\\s*bearer\\s+)[^\\s,;]+");
+    private static final Pattern AUTHORIZATION_SECRET_PATTERN = Pattern.compile(
+            "(?i)(authorization\\s*[:=]\\s*)(?!bearer\\s)[^\\s,;]+");
+    private static final Pattern COOKIE_SECRET_PATTERN = Pattern.compile(
+            "(?i)(cookie\\s*[:=]\\s*)[^\\s,;]+");
+    private static final Pattern TEXT_SECRET_PATTERN = Pattern.compile(
+            "(?i)((?:api[_-]?key|token|secret|password)\\s*[:=]\\s*)[^\\s,;}]+");

    private final DebugEmlSuperAgentProperties properties;
    private final AliyunOssProperties ossProperties;
@@ -162,7 +181,9 @@ public class DebugEmlSuperAgentRunServiceImpl implements DebugEmlSuperAgentRunSe
                    "OSS 上传失败。",
                    exception);
        } catch (SuperAgentOpenApiException exception) {
-            markFailed(runId, superAgentFailureSummary(exception), DebugEmlSuperAgentRunStatus.SUPERAGENT_FAILED, nowUtc());
+            String safeSummary = superAgentFailureSummary(exception);
+            logSuperAgentOpenApiFailure(runId, safeSummary, exception);
+            markFailed(runId, safeSummary, DebugEmlSuperAgentRunStatus.SUPERAGENT_FAILED, nowUtc());
            throw new DebugEmlSuperAgentException(
                    HttpStatus.BAD_GATEWAY,
                    "SUPERAGENT_OPEN_API_FAILED",
@@ -233,7 +254,9 @@ public class DebugEmlSuperAgentRunServiceImpl implements DebugEmlSuperAgentRunSe
            streamSink.error(runId, "OSS_UPLOAD_FAILED", "OSS 上传失败。", DebugEmlSuperAgentRunStatus.FAILED.name());
        } catch (SuperAgentOpenApiException exception) {
            if (runId != null) {
-                markFailed(runId, superAgentFailureSummary(exception), DebugEmlSuperAgentRunStatus.SUPERAGENT_FAILED, nowUtc());
+                String safeSummary = superAgentFailureSummary(exception);
+                logSuperAgentOpenApiFailure(runId, safeSummary, exception);
+                markFailed(runId, safeSummary, DebugEmlSuperAgentRunStatus.SUPERAGENT_FAILED, nowUtc());
            }
            streamSink.error(
                    runId,
@@ -291,6 +314,52 @@ public class DebugEmlSuperAgentRunServiceImpl implements DebugEmlSuperAgentRunSe
        return "SuperAgent Open API 调用失败。";
    }

+    /**
+     * 记录 SuperAgent Debug 调用失败的安全诊断日志。只写异常类型和脱敏消息，不记录 API Key、Cookie、邮件正文或响应体。
+     */
+    private void logSuperAgentOpenApiFailure(
+            Long runId,
+            String safeSummary,
+            SuperAgentOpenApiException exception) {
+        log.warn(
+                "Debug EML SuperAgent Open API failed. debug_run_id={}, safe_error_summary={}, exception_chain={}",
+                runId,
+                safeSummary,
+                safeExceptionChain(exception));
+    }
+
+    /**
+     * 生成异常链摘要，帮助区分 timeout、connection reset、EOF 等网络/流读取问题。
+     */
+    private String safeExceptionChain(Throwable throwable) {
+        List<String> chain = new ArrayList<>();
+        Throwable current = throwable;
+        int depth = 0;
+        while (current != null && depth < 8) {
+            String message = trimToNull(current.getMessage());
+            String item = current.getClass().getSimpleName();
+            if (message != null) {
+                item += ": " + sanitizeLogMessage(message);
+            }
+            chain.add(item);
+            current = current.getCause();
+            depth++;
+        }
+        return String.join(" <- ", chain);
+    }
+
+    /**
+     * 日志消息脱敏并截断，避免外部异常把 Secret 或过长内容写入日志。
+     */
+    private String sanitizeLogMessage(String message) {
+        String sanitized = JSON_SECRET_PATTERN.matcher(message).replaceAll("$1[REDACTED]$2");
+        sanitized = AUTHORIZATION_BEARER_SECRET_PATTERN.matcher(sanitized).replaceAll("$1[REDACTED]");
+        sanitized = AUTHORIZATION_SECRET_PATTERN.matcher(sanitized).replaceAll("$1[REDACTED]");
+        sanitized = COOKIE_SECRET_PATTERN.matcher(sanitized).replaceAll("$1[REDACTED]");
+        sanitized = TEXT_SECRET_PATTERN.matcher(sanitized).replaceAll("$1[REDACTED]");
+        return safeErrorSummary(sanitized);
+    }
+
    /**
     * 取最底层异常，便于判断 HTTP、网络和解析失败类型。
     */
@@ -448,9 +517,11 @@ public class DebugEmlSuperAgentRunServiceImpl implements DebugEmlSuperAgentRunSe
                streamSink.trace(traceEvent);
            }
        };
-        SuperAgentOpenApiResult superAgentResult = streamSink == null
-                ? superAgentOpenApiClient.invokeMailDebug(superAgentRequest)
-                : superAgentOpenApiClient.invokeMailDebug(superAgentRequest, traceConsumer);
+        SuperAgentOpenApiResult superAgentResult = invokeSuperAgentWithHeartbeat(
+                runId,
+                superAgentRequest,
+                traceConsumer,
+                streamSink);
        List<SuperAgentOpenApiTraceEvent> traceEvents = streamedTraceEvents.isEmpty()
                ? superAgentResult.traceEvents()
                : List.copyOf(streamedTraceEvents);
@@ -503,6 +574,52 @@ public class DebugEmlSuperAgentRunServiceImpl implements DebugEmlSuperAgentRunSe
                status.name());
    }

+    /**
+     * 调用 SuperAgent 期间维持 Debug SSE 心跳，避免无 trace 输出时浏览器或反向代理认为连接空闲。
+     */
+    private SuperAgentOpenApiResult invokeSuperAgentWithHeartbeat(
+            Long runId,
+            SuperAgentMailDebugRequest request,
+            Consumer<SuperAgentOpenApiTraceEvent> traceConsumer,
+            DebugEmlSuperAgentStreamSink streamSink) {
+        if (streamSink == null) {
+            return superAgentOpenApiClient.invokeMailDebug(request);
+        }
+        ScheduledExecutorService heartbeatExecutor = null;
+        ScheduledFuture<?> heartbeatTask = null;
+        try {
+            Duration interval = properties.getSseHeartbeatInterval();
+            if (interval != null && !interval.isZero() && !interval.isNegative()) {
+                heartbeatExecutor = Executors.newSingleThreadScheduledExecutor(heartbeatThreadFactory(runId));
+                long intervalMillis = Math.max(1L, interval.toMillis());
+                heartbeatTask = heartbeatExecutor.scheduleAtFixedRate(
+                        () -> streamSink.heartbeat(runId),
+                        intervalMillis,
+                        intervalMillis,
+                        TimeUnit.MILLISECONDS);
+            }
+            return superAgentOpenApiClient.invokeMailDebug(request, traceConsumer);
+        } finally {
+            if (heartbeatTask != null) {
+                heartbeatTask.cancel(false);
+            }
+            if (heartbeatExecutor != null) {
+                heartbeatExecutor.shutdownNow();
+            }
+        }
+    }
+
+    /**
+     * 创建 Debug SSE 心跳线程；使用 daemon 线程，避免调试请求异常退出时阻塞 JVM 关闭。
+     */
+    private ThreadFactory heartbeatThreadFactory(Long runId) {
+        return runnable -> {
+            Thread thread = new Thread(runnable, "debug-eml-sse-heartbeat-" + runId);
+            thread.setDaemon(true);
+            return thread;
+        };
+    }
+
    /**
     * SourceMessage 已写入后先回填 Debug run，保证后续 SuperAgent 失败时仍可追踪入库消息和 OSS 原文。
     */
@@ -1050,6 +1167,11 @@ public class DebugEmlSuperAgentRunServiceImpl implements DebugEmlSuperAgentRunSe
         */
        void trace(SuperAgentOpenApiTraceEvent traceEvent);

+        /**
+         * 输出安全心跳事件，维持 Debug SSE 长连接。
+         */
+        void heartbeat(Long runId);
+
        /**
         * 输出最终 Debug 结果。
         */
@@ -1094,6 +1216,14 @@ public class DebugEmlSuperAgentRunServiceImpl implements DebugEmlSuperAgentRunSe
            send("superagent_trace", traceEvent);
        }

+        @Override
+        public void heartbeat(Long runId) {
+            Map<String, Object> data = new LinkedHashMap<>();
+            data.put("debug_run_id", runId == null ? null : runId.toString());
+            data.put("status", DebugEmlSuperAgentRunStatus.CALLING_SUPERAGENT.name());
+            send("debug_heartbeat", data);
+        }
+
        @Override
        public void result(DebugEmlSuperAgentRunResult result) {
            send("superagent_result", result);
@@ -1118,7 +1248,7 @@ public class DebugEmlSuperAgentRunServiceImpl implements DebugEmlSuperAgentRunSe
        /**
         * 写出一个 SSE 事件块并立即 flush。写失败通常表示客户端断开，只关闭流输出，不影响后续业务落库。
         */
-        private void send(String eventName, Object data) {
+        private synchronized void send(String eventName, Object data) {
            if (!active) {
                return;
            }