From a92eb168aa187cf419e54f89d48a7ff94edd6433 Mon Sep 17 00:00:00 2001 From: sagitchu Date: Sat, 28 Feb 2026 20:09:25 +0800 Subject: [PATCH] feat(diagnosis): add streaming progress support and tunnel-grouped forward list - Add SSE streaming endpoints for tunnel/forward diagnosis with real-time progress - Increase diagnosis timeout to 2 minutes with context propagation - Group forwards by tunnel within user groups in UI - Add nginx SSE proxy configuration for streaming endpoints --- go-backend/internal/app/app.go | 2 +- .../internal/http/handler/control_plane.go | 304 +++++++-- .../internal/http/handler/diagnosis_stream.go | 208 +++++++ .../internal/http/handler/federation.go | 12 +- go-backend/internal/http/handler/handler.go | 2 + go-backend/internal/http/handler/mutations.go | 9 +- vite-frontend/nginx.conf | 30 + vite-frontend/src/api/diagnosis-stream.ts | 343 +++++++++++ vite-frontend/src/api/index.ts | 4 +- vite-frontend/src/pages/forward.tsx | 579 ++++++++++++++---- vite-frontend/src/pages/tunnel.tsx | 246 +++++++- 11 files changed, 1511 insertions(+), 228 deletions(-) create mode 100644 go-backend/internal/http/handler/diagnosis_stream.go create mode 100644 vite-frontend/src/api/diagnosis-stream.ts diff --git a/go-backend/internal/app/app.go b/go-backend/internal/app/app.go index 5c013a8..17b231a 100644 --- a/go-backend/internal/app/app.go +++ b/go-backend/internal/app/app.go @@ -49,7 +49,7 @@ func New(cfg config.Config) (*App, error) { Handler: router, ReadTimeout: 30 * time.Second, ReadHeaderTimeout: 5 * time.Second, - WriteTimeout: 30 * time.Second, + WriteTimeout: 2 * time.Minute, IdleTimeout: 60 * time.Second, } diff --git a/go-backend/internal/http/handler/control_plane.go b/go-backend/internal/http/handler/control_plane.go index 86b691d..3ed4f0d 100644 --- a/go-backend/internal/http/handler/control_plane.go +++ b/go-backend/internal/http/handler/control_plane.go @@ -1,6 +1,7 @@ package handler import ( + "context" "errors" "fmt" "net" @@ -42,8 +43,29 @@ type diagnosisWorkItem struct { ipPreference string } +type diagnosisExecOptions struct { + commandTimeout time.Duration + pingTimeoutMS int + timeoutMessage string +} + +type diagnosisProgress struct { + Total int `json:"total"` + Completed int `json:"completed"` + Success int `json:"success"` + Failed int `json:"failed"` +} + +type diagnosisItemEmitter func(index int, item map[string]interface{}, progress diagnosisProgress) + const diagnosisMaxConcurrency = 8 +const ( + defaultNodeCommandTimeout = 6 * time.Second + diagnosisCommandTimeout = 2 * time.Minute + diagnosisRequestTimeout = 2 * time.Minute +) + func (h *Handler) resolveForwardAccess(r *http.Request, forwardID int64) (*forwardRecord, int64, int, error) { userID, roleID, err := userRoleFromRequest(r) if err != nil { @@ -309,16 +331,23 @@ func (h *Handler) applyNodeProtocolChange(nodeID int64, httpVal, tlsVal, socksVa } func (h *Handler) sendNodeCommand(nodeID int64, commandType string, data interface{}, tolerateExists bool, tolerateNotFound bool) (ws.CommandResult, error) { + return h.sendNodeCommandWithTimeout(nodeID, commandType, data, defaultNodeCommandTimeout, tolerateExists, tolerateNotFound) +} + +func (h *Handler) sendNodeCommandWithTimeout(nodeID int64, commandType string, data interface{}, timeout time.Duration, tolerateExists bool, tolerateNotFound bool) (ws.CommandResult, error) { var ( result ws.CommandResult err error ) + if timeout <= 0 { + timeout = defaultNodeCommandTimeout + } node, nodeErr := h.getNodeRecord(nodeID) if nodeErr == nil && node != nil && node.IsRemote == 1 { - result, err = h.sendRemoteNodeCommand(node, commandType, data) + result, err = h.sendRemoteNodeCommandWithTimeout(node, commandType, data, timeout) } else { - result, err = h.wsServer.SendCommand(nodeID, commandType, data, 6*time.Second) + result, err = h.wsServer.SendCommand(nodeID, commandType, data, timeout) } if err == nil { return result, nil @@ -338,6 +367,10 @@ func (h *Handler) sendNodeCommand(nodeID int64, commandType string, data interfa } func (h *Handler) sendRemoteNodeCommand(node *nodeRecord, commandType string, data interface{}) (ws.CommandResult, error) { + return h.sendRemoteNodeCommandWithTimeout(node, commandType, data, 0) +} + +func (h *Handler) sendRemoteNodeCommandWithTimeout(node *nodeRecord, commandType string, data interface{}, timeout time.Duration) (ws.CommandResult, error) { if node == nil { return ws.CommandResult{}, errors.New("节点不存在") } @@ -348,6 +381,9 @@ func (h *Handler) sendRemoteNodeCommand(node *nodeRecord, commandType string, da } fc := client.NewFederationClient() + if timeout > 0 { + fc = client.NewFederationClientWithTimeout(timeout) + } res, err := fc.Command(remoteURL, remoteToken, h.federationLocalDomain(), client.RuntimeNodeCommandRequest{ CommandType: commandType, Data: data, @@ -375,26 +411,45 @@ func (h *Handler) sendRemoteNodeCommand(node *nodeRecord, commandType string, da return result, nil } -func (h *Handler) diagnoseForwardRuntime(forward *forwardRecord) (map[string]interface{}, error) { +func (h *Handler) diagnoseForwardRuntime(ctx context.Context, forward *forwardRecord) (map[string]interface{}, error) { + if ctx == nil { + ctx = context.Background() + } + forwardName, workItems, err := h.prepareForwardDiagnosis(forward) + if err != nil { + return nil, err + } + + results := h.runDiagnosisWorkItems(ctx, workItems, nil) + + payload := map[string]interface{}{ + "forwardName": forwardName, + "timestamp": time.Now().UnixMilli(), + "results": results, + } + return payload, nil +} + +func (h *Handler) prepareForwardDiagnosis(forward *forwardRecord) (string, []diagnosisWorkItem, error) { if forward == nil { - return nil, errForwardNotFound + return "", nil, errForwardNotFound } targets, err := resolveDiagnosisTargets(forward.RemoteAddr) if err != nil { - return nil, err + return "", nil, err } tunnel, err := h.getTunnelRecord(forward.TunnelID) if err != nil { - return nil, err + return "", nil, err } chainRows, err := h.listChainNodesForTunnel(forward.TunnelID) if err != nil { - return nil, err + return "", nil, err } if len(chainRows) == 0 { - return nil, errors.New("隧道配置不完整") + return "", nil, errors.New("隧道配置不完整") } ipPreference := h.repo.GetTunnelIPPreference(forward.TunnelID) @@ -524,36 +579,49 @@ func (h *Handler) diagnoseForwardRuntime(forward *forwardRecord) (map[string]int } } - results := h.runDiagnosisWorkItems(workItems) + return forward.Name, workItems, nil +} + +func (h *Handler) diagnoseTunnelRuntime(ctx context.Context, tunnelID int64) (map[string]interface{}, error) { + if ctx == nil { + ctx = context.Background() + } + tunnelName, tunnelType, workItems, err := h.prepareTunnelDiagnosis(tunnelID) + if err != nil { + return nil, err + } + + results := h.runDiagnosisWorkItems(ctx, workItems, nil) payload := map[string]interface{}{ - "forwardName": forward.Name, - "timestamp": time.Now().UnixMilli(), - "results": results, + "tunnelName": tunnelName, + "tunnelType": tunnelType, + "timestamp": time.Now().UnixMilli(), + "results": results, } return payload, nil } -func (h *Handler) diagnoseTunnelRuntime(tunnelID int64) (map[string]interface{}, error) { +func (h *Handler) prepareTunnelDiagnosis(tunnelID int64) (string, string, []diagnosisWorkItem, error) { tunnel, err := h.getTunnelRecord(tunnelID) if err != nil { - return nil, err + return "", "", nil, err } tunnelName, err := h.repo.GetTunnelName(tunnelID) if err != nil { - return nil, err + return "", "", nil, err } if tunnelName == "" { - return nil, errors.New("隧道不存在") + return "", "", nil, errors.New("隧道不存在") } chainRows, err := h.listChainNodesForTunnel(tunnelID) if err != nil { - return nil, err + return "", "", nil, err } if len(chainRows) == 0 { - return nil, errors.New("隧道配置不完整") + return "", "", nil, errors.New("隧道配置不完整") } ipPreference := h.repo.GetTunnelIPPreference(tunnelID) @@ -676,15 +744,8 @@ func (h *Handler) diagnoseTunnelRuntime(tunnelID int64) (map[string]interface{}, } } - results := h.runDiagnosisWorkItems(workItems) - - payload := map[string]interface{}{ - "tunnelName": tunnelName, - "tunnelType": map[bool]string{true: "端口转发", false: "隧道转发"}[tunnel.Type == 1], - "timestamp": time.Now().UnixMilli(), - "results": results, - } - return payload, nil + tunnelType := map[bool]string{true: "端口转发", false: "隧道转发"}[tunnel.Type == 1] + return tunnelName, tunnelType, workItems, nil } func splitChainNodeGroups(rows []chainNodeRecord) ([]chainNodeRecord, [][]chainNodeRecord, []chainNodeRecord) { @@ -736,11 +797,85 @@ func resolveDiagnosisTargets(remoteAddr string) ([]diagnosisTarget, error) { return targets, nil } -func (h *Handler) runDiagnosisWorkItems(workItems []diagnosisWorkItem) []map[string]interface{} { +func diagnosisContextMessage(ctx context.Context) string { + if ctx == nil { + return "诊断超时（2分钟）" + } + switch ctx.Err() { + case context.DeadlineExceeded: + return "诊断超时（2分钟）" + case context.Canceled: + return "诊断已取消" + default: + return "诊断超时（2分钟）" + } +} + +func diagnosisExecOptionsFromContext(ctx context.Context) diagnosisExecOptions { + timeout := diagnosisCommandTimeout + if ctx != nil { + if deadline, ok := ctx.Deadline(); ok { + remaining := time.Until(deadline) + if remaining <= 0 { + remaining = 100 * time.Millisecond + } + if remaining < timeout { + timeout = remaining + } + } + } + if timeout <= 0 { + timeout = 100 * time.Millisecond + } + pingTimeoutMS := int(timeout / time.Millisecond) + if pingTimeoutMS <= 0 { + pingTimeoutMS = 100 + } + return diagnosisExecOptions{ + commandTimeout: timeout, + pingTimeoutMS: pingTimeoutMS, + timeoutMessage: diagnosisContextMessage(ctx), + } +} + +func newDiagnosisTimeoutItem(workItem diagnosisWorkItem, message string) map[string]interface{} { + targetPort := workItem.targetPort + if targetPort <= 0 { + targetPort = workItem.toNode.Port + } + item := newDiagnosisResultItem(workItem.fromNodeID, workItem.targetIP, targetPort, workItem.description, workItem.metadata) + item["success"] = false + if strings.TrimSpace(message) == "" { + message = "诊断超时（2分钟）" + } + item["message"] = message + return item +} + +func (h *Handler) executeDiagnosisWorkItem(workItem diagnosisWorkItem, options diagnosisExecOptions) map[string]interface{} { + single := make([]map[string]interface{}, 0, 1) + nodeCache := map[int64]*nodeRecord{} + if workItem.hasChainHop { + h.appendChainHopDiagnosis(&single, nodeCache, workItem.fromNodeID, workItem.toNode, workItem.description, workItem.metadata, workItem.ipPreference, options) + } else { + h.appendPathDiagnosis(&single, nodeCache, workItem.fromNodeID, workItem.targetIP, workItem.targetPort, workItem.description, workItem.metadata, options) + } + + if len(single) == 0 { + return newDiagnosisTimeoutItem(workItem, "诊断任务未返回结果") + } + return single[0] +} + +func (h *Handler) runDiagnosisWorkItems(ctx context.Context, workItems []diagnosisWorkItem, emitter diagnosisItemEmitter) []map[string]interface{} { + if ctx == nil { + ctx = context.Background() + } results := make([]map[string]interface{}, len(workItems)) if len(workItems) == 0 { return results } + workerLimit := diagnosisMaxConcurrency if workerLimit < 1 { workerLimit = 1 @@ -748,35 +883,68 @@ func (h *Handler) runDiagnosisWorkItems(workItems []diagnosisWorkItem) []map[str if workerLimit > len(workItems) { workerLimit = len(workItems) } - semaphore := make(chan struct{}, workerLimit) + + type diagnosisWorkResult struct { + index int + item map[string]interface{} + } + + jobs := make(chan int) + resultCh := make(chan diagnosisWorkResult, len(workItems)) var wg sync.WaitGroup - for i := range workItems { + for i := 0; i < workerLimit; i++ { wg.Add(1) - semaphore <- struct{}{} - go func(index int) { + go func() { defer wg.Done() - defer func() { <-semaphore }() - - workItem := workItems[index] - single := make([]map[string]interface{}, 0, 1) - nodeCache := map[int64]*nodeRecord{} - if workItem.hasChainHop { - h.appendChainHopDiagnosis(&single, nodeCache, workItem.fromNodeID, workItem.toNode, workItem.description, workItem.metadata, workItem.ipPreference) - } else { - h.appendPathDiagnosis(&single, nodeCache, workItem.fromNodeID, workItem.targetIP, workItem.targetPort, workItem.description, workItem.metadata) + for index := range jobs { + select { + case <-ctx.Done(): + resultCh <- diagnosisWorkResult{index: index, item: newDiagnosisTimeoutItem(workItems[index], diagnosisContextMessage(ctx))} + continue + default: + } + options := diagnosisExecOptionsFromContext(ctx) + resultCh <- diagnosisWorkResult{index: index, item: h.executeDiagnosisWorkItem(workItems[index], options)} } - - if len(single) == 0 { - results[index] = newDiagnosisResultItem(workItem.fromNodeID, workItem.targetIP, workItem.targetPort, workItem.description, workItem.metadata) - results[index]["success"] = false - results[index]["message"] = "诊断任务未返回结果" - return - } - results[index] = single[0] - }(i) + }() } + +enqueueLoop: + for i := 0; i < len(workItems); i++ { + select { + case <-ctx.Done(): + message := diagnosisContextMessage(ctx) + for j := i; j < len(workItems); j++ { + resultCh <- diagnosisWorkResult{index: j, item: newDiagnosisTimeoutItem(workItems[j], message)} + } + break enqueueLoop + case jobs <- i: + } + } + close(jobs) wg.Wait() + close(resultCh) + + progress := diagnosisProgress{Total: len(workItems)} + for result := range resultCh { + results[result.index] = result.item + progress.Completed++ + if asBool(result.item["success"], false) { + progress.Success++ + } else { + progress.Failed++ + } + if emitter != nil { + emitter(result.index, result.item, progress) + } + } + + for i := range results { + if results[i] == nil { + results[i] = newDiagnosisTimeoutItem(workItems[i], "诊断超时（2分钟）") + } + } return results } @@ -821,7 +989,7 @@ func (h *Handler) appendFailedDiagnosis(results *[]map[string]interface{}, nodeC *results = append(*results, item) } -func (h *Handler) appendPathDiagnosis(results *[]map[string]interface{}, nodeCache map[int64]*nodeRecord, fromNodeID int64, targetIP string, targetPort int, description string, metadata map[string]interface{}) { +func (h *Handler) appendPathDiagnosis(results *[]map[string]interface{}, nodeCache map[int64]*nodeRecord, fromNodeID int64, targetIP string, targetPort int, description string, metadata map[string]interface{}, options diagnosisExecOptions) { item := newDiagnosisResultItem(fromNodeID, targetIP, targetPort, description, metadata) fromNode, err := h.cachedNode(nodeCache, fromNodeID) @@ -838,9 +1006,9 @@ func (h *Handler) appendPathDiagnosis(results *[]map[string]interface{}, nodeCac pingErr error ) if fromNode.IsRemote == 1 { - pingData, pingErr = h.tcpPingViaRemoteNode(fromNode, targetIP, targetPort) + pingData, pingErr = h.tcpPingViaRemoteNode(fromNode, targetIP, targetPort, options) } else { - pingData, pingErr = h.tcpPingViaNode(fromNodeID, targetIP, targetPort) + pingData, pingErr = h.tcpPingViaNode(fromNodeID, targetIP, targetPort, options) } if pingErr != nil { item["success"] = false @@ -871,7 +1039,7 @@ func (h *Handler) appendPathDiagnosis(results *[]map[string]interface{}, nodeCac *results = append(*results, item) } -func (h *Handler) appendChainHopDiagnosis(results *[]map[string]interface{}, nodeCache map[int64]*nodeRecord, fromNodeID int64, toNode chainNodeRecord, description string, metadata map[string]interface{}, ipPreference string) { +func (h *Handler) appendChainHopDiagnosis(results *[]map[string]interface{}, nodeCache map[int64]*nodeRecord, fromNodeID int64, toNode chainNodeRecord, description string, metadata map[string]interface{}, ipPreference string, options diagnosisExecOptions) { fromNode, _ := h.cachedNode(nodeCache, fromNodeID) targetNode, err := h.cachedNode(nodeCache, toNode.NodeID) if err != nil { @@ -883,7 +1051,7 @@ func (h *Handler) appendChainHopDiagnosis(results *[]map[string]interface{}, nod h.appendFailedDiagnosis(results, nodeCache, fromNodeID, strings.Trim(strings.TrimSpace(targetNode.ServerIP), "[]"), toNode.Port, description, metadata, err.Error()) return } - h.appendPathDiagnosis(results, nodeCache, fromNodeID, targetIP, targetPort, description, metadata) + h.appendPathDiagnosis(results, nodeCache, fromNodeID, targetIP, targetPort, description, metadata, options) } func resolveChainProbeTarget(fromNode, targetNode *nodeRecord, preferredPort int, ipPreference string) (string, int, error) { @@ -936,13 +1104,19 @@ func (h *Handler) listChainNodesForTunnel(tunnelID int64) ([]chainNodeRecord, er return h.repo.ListChainNodesForTunnel(tunnelID) } -func (h *Handler) tcpPingViaNode(nodeID int64, ip string, port int) (map[string]interface{}, error) { - res, err := h.sendNodeCommand(nodeID, "TcpPing", map[string]interface{}{ +func (h *Handler) tcpPingViaNode(nodeID int64, ip string, port int, options diagnosisExecOptions) (map[string]interface{}, error) { + if options.commandTimeout <= 0 { + options.commandTimeout = diagnosisCommandTimeout + } + if options.pingTimeoutMS <= 0 { + options.pingTimeoutMS = int(diagnosisCommandTimeout / time.Millisecond) + } + res, err := h.sendNodeCommandWithTimeout(nodeID, "TcpPing", map[string]interface{}{ "ip": ip, "port": port, "count": 4, - "timeout": 5000, - }, false, false) + "timeout": options.pingTimeoutMS, + }, options.commandTimeout, false, false) if err != nil { return nil, err } @@ -952,7 +1126,7 @@ func (h *Handler) tcpPingViaNode(nodeID int64, ip string, port int) (map[string] return res.Data, nil } -func (h *Handler) tcpPingViaRemoteNode(node *nodeRecord, ip string, port int) (map[string]interface{}, error) { +func (h *Handler) tcpPingViaRemoteNode(node *nodeRecord, ip string, port int, options diagnosisExecOptions) (map[string]interface{}, error) { if node == nil { return nil, errors.New("节点不存在") } @@ -961,13 +1135,19 @@ func (h *Handler) tcpPingViaRemoteNode(node *nodeRecord, ip string, port int) (m if remoteURL == "" || remoteToken == "" { return nil, errors.New("远程节点缺少共享配置") } + if options.commandTimeout <= 0 { + options.commandTimeout = diagnosisCommandTimeout + } + if options.pingTimeoutMS <= 0 { + options.pingTimeoutMS = int(diagnosisCommandTimeout / time.Millisecond) + } - fc := client.NewFederationClient() + fc := client.NewFederationClientWithTimeout(options.commandTimeout) return fc.Diagnose(remoteURL, remoteToken, h.federationLocalDomain(), client.RuntimeDiagnoseRequest{ IP: strings.TrimSpace(ip), Port: port, Count: 4, - Timeout: 5000, + Timeout: options.pingTimeoutMS, }) } diff --git a/go-backend/internal/http/handler/diagnosis_stream.go b/go-backend/internal/http/handler/diagnosis_stream.go new file mode 100644 index 0000000..c75011b --- /dev/null +++ b/go-backend/internal/http/handler/diagnosis_stream.go @@ -0,0 +1,208 @@ +package handler + +import ( + "context" + "encoding/json" + "errors" + "net/http" + "strings" + "time" + + "go-backend/internal/http/response" +) + +type diagnosisStreamEvent struct { + Type string `json:"type"` + Data interface{} `json:"data,omitempty"` + TS int64 `json:"ts"` +} + +func prepareDiagnosisStreamResponse(w http.ResponseWriter) (http.Flusher, error) { + flusher, ok := w.(http.Flusher) + if !ok { + return nil, errors.New("当前服务不支持流式响应") + } + w.Header().Set("Content-Type", "application/x-ndjson; charset=utf-8") + w.Header().Set("Cache-Control", "no-cache") + w.Header().Set("Connection", "keep-alive") + w.Header().Set("X-Accel-Buffering", "no") + return flusher, nil +} + +func writeDiagnosisStreamEvent(encoder *json.Encoder, flusher http.Flusher, eventType string, data interface{}) error { + if encoder == nil || flusher == nil { + return errors.New("流式响应写入器未初始化") + } + event := diagnosisStreamEvent{Type: eventType, Data: data, TS: time.Now().UnixMilli()} + if err := encoder.Encode(event); err != nil { + return err + } + flusher.Flush() + return nil +} + +func summarizeDiagnosisProgress(results []map[string]interface{}) diagnosisProgress { + progress := diagnosisProgress{Total: len(results)} + for _, item := range results { + progress.Completed++ + if asBool(item["success"], false) { + progress.Success++ + } else { + progress.Failed++ + } + } + return progress +} + +func shouldIgnoreDiagnosisStreamError(err error) bool { + if err == nil { + return false + } + if errors.Is(err, context.Canceled) { + return true + } + msg := strings.ToLower(strings.TrimSpace(err.Error())) + if strings.Contains(msg, "broken pipe") || strings.Contains(msg, "connection reset by peer") { + return true + } + if strings.Contains(msg, "stream already closed") { + return true + } + return false +} + +func (h *Handler) streamDiagnosisRuntime(ctx context.Context, cancel context.CancelFunc, w http.ResponseWriter, startPayload map[string]interface{}, workItems []diagnosisWorkItem) error { + flusher, err := prepareDiagnosisStreamResponse(w) + if err != nil { + return err + } + encoder := json.NewEncoder(w) + + payload := map[string]interface{}{ + "total": len(workItems), + "timestamp": time.Now().UnixMilli(), + } + for key, value := range startPayload { + payload[key] = value + } + if err := writeDiagnosisStreamEvent(encoder, flusher, "start", payload); err != nil { + return err + } + + streamBroken := false + emitter := func(index int, item map[string]interface{}, progress diagnosisProgress) { + if streamBroken { + return + } + itemPayload := map[string]interface{}{ + "index": index, + "result": item, + "progress": progress, + } + if err := writeDiagnosisStreamEvent(encoder, flusher, "item", itemPayload); err != nil { + streamBroken = true + if cancel != nil { + cancel() + } + } + } + + results := h.runDiagnosisWorkItems(ctx, workItems, emitter) + if streamBroken { + return context.Canceled + } + + progress := summarizeDiagnosisProgress(results) + donePayload := map[string]interface{}{ + "progress": progress, + "timedOut": errors.Is(ctx.Err(), context.DeadlineExceeded), + } + return writeDiagnosisStreamEvent(encoder, flusher, "done", donePayload) +} + +func (h *Handler) tunnelDiagnoseStream(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + response.WriteJSON(w, response.ErrDefault("请求失败")) + return + } + id := asInt64FromBodyKey(r, w, "tunnelId") + if id <= 0 { + return + } + + tunnelName, tunnelType, workItems, err := h.prepareTunnelDiagnosis(id) + if err != nil { + if strings.Contains(err.Error(), "不存在") || strings.Contains(err.Error(), "不完整") { + response.WriteJSON(w, response.ErrDefault(err.Error())) + return + } + response.WriteJSON(w, response.Err(-2, err.Error())) + return + } + + ctx, cancel := context.WithTimeout(r.Context(), diagnosisRequestTimeout) + defer cancel() + + startPayload := map[string]interface{}{ + "tunnelName": tunnelName, + "tunnelType": tunnelType, + } + if err := h.streamDiagnosisRuntime(ctx, cancel, w, startPayload, workItems); err != nil { + if shouldIgnoreDiagnosisStreamError(err) { + return + } + if strings.Contains(err.Error(), "不支持流式响应") { + response.WriteJSON(w, response.ErrDefault(err.Error())) + return + } + return + } +} + +func (h *Handler) forwardDiagnoseStream(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + response.WriteJSON(w, response.ErrDefault("请求失败")) + return + } + id := asInt64FromBodyKey(r, w, "forwardId") + if id <= 0 { + return + } + + forward, _, _, err := h.resolveForwardAccess(r, id) + if err != nil { + if errors.Is(err, errForwardNotFound) { + response.WriteJSON(w, response.ErrDefault("转发不存在")) + return + } + response.WriteJSON(w, response.Err(-2, err.Error())) + return + } + + forwardName, workItems, err := h.prepareForwardDiagnosis(forward) + if err != nil { + if strings.Contains(err.Error(), "不存在") || strings.Contains(err.Error(), "不能为空") || strings.Contains(err.Error(), "错误") { + response.WriteJSON(w, response.ErrDefault(err.Error())) + return + } + response.WriteJSON(w, response.Err(-2, err.Error())) + return + } + + ctx, cancel := context.WithTimeout(r.Context(), diagnosisRequestTimeout) + defer cancel() + + startPayload := map[string]interface{}{ + "forwardName": forwardName, + } + if err := h.streamDiagnosisRuntime(ctx, cancel, w, startPayload, workItems); err != nil { + if shouldIgnoreDiagnosisStreamError(err) { + return + } + if strings.Contains(err.Error(), "不支持流式响应") { + response.WriteJSON(w, response.ErrDefault(err.Error())) + return + } + return + } +} diff --git a/go-backend/internal/http/handler/federation.go b/go-backend/internal/http/handler/federation.go index ee33d3f..5c40002 100644 --- a/go-backend/internal/http/handler/federation.go +++ b/go-backend/internal/http/handler/federation.go @@ -1232,16 +1232,20 @@ func (h *Handler) federationRuntimeDiagnose(w http.ResponseWriter, r *http.Reque if req.Count <= 0 { req.Count = 4 } - if req.Timeout <= 0 { - req.Timeout = 5000 + if req.Timeout <= 0 || req.Timeout > int(diagnosisCommandTimeout/time.Millisecond) { + req.Timeout = int(diagnosisCommandTimeout / time.Millisecond) + } + commandTimeout := time.Duration(req.Timeout) * time.Millisecond + if commandTimeout <= 0 || commandTimeout > diagnosisCommandTimeout { + commandTimeout = diagnosisCommandTimeout } - res, err := h.sendNodeCommand(share.NodeID, "TcpPing", map[string]interface{}{ + res, err := h.sendNodeCommandWithTimeout(share.NodeID, "TcpPing", map[string]interface{}{ "ip": req.IP, "port": req.Port, "count": req.Count, "timeout": req.Timeout, - }, false, false) + }, commandTimeout, false, false) if err != nil { response.WriteJSON(w, response.ErrDefault(err.Error())) return diff --git a/go-backend/internal/http/handler/handler.go b/go-backend/internal/http/handler/handler.go index 38aaac3..b872e76 100644 --- a/go-backend/internal/http/handler/handler.go +++ b/go-backend/internal/http/handler/handler.go @@ -134,6 +134,7 @@ func (h *Handler) Register(mux *http.ServeMux) { mux.HandleFunc("/api/v1/tunnel/update", h.tunnelUpdate) mux.HandleFunc("/api/v1/tunnel/delete", h.tunnelDelete) mux.HandleFunc("/api/v1/tunnel/diagnose", h.tunnelDiagnose) + mux.HandleFunc("/api/v1/tunnel/diagnose/stream", h.tunnelDiagnoseStream) mux.HandleFunc("/api/v1/tunnel/update-order", h.tunnelUpdateOrder) mux.HandleFunc("/api/v1/tunnel/batch-delete", h.tunnelBatchDelete) mux.HandleFunc("/api/v1/tunnel/batch-redeploy", h.tunnelBatchRedeploy) @@ -149,6 +150,7 @@ func (h *Handler) Register(mux *http.ServeMux) { mux.HandleFunc("/api/v1/forward/pause", h.forwardPause) mux.HandleFunc("/api/v1/forward/resume", h.forwardResume) mux.HandleFunc("/api/v1/forward/diagnose", h.forwardDiagnose) + mux.HandleFunc("/api/v1/forward/diagnose/stream", h.forwardDiagnoseStream) mux.HandleFunc("/api/v1/forward/update-order", h.forwardUpdateOrder) mux.HandleFunc("/api/v1/forward/batch-delete", h.forwardBatchDelete) mux.HandleFunc("/api/v1/forward/batch-pause", h.forwardBatchPause) diff --git a/go-backend/internal/http/handler/mutations.go b/go-backend/internal/http/handler/mutations.go index a17e9a5..aa44704 100644 --- a/go-backend/internal/http/handler/mutations.go +++ b/go-backend/internal/http/handler/mutations.go @@ -1,6 +1,7 @@ package handler import ( + "context" "crypto/rand" "database/sql" "encoding/hex" @@ -787,7 +788,9 @@ func (h *Handler) tunnelDiagnose(w http.ResponseWriter, r *http.Request) { if id <= 0 { return } - result, err := h.diagnoseTunnelRuntime(id) + ctx, cancel := context.WithTimeout(r.Context(), diagnosisRequestTimeout) + defer cancel() + result, err := h.diagnoseTunnelRuntime(ctx, id) if err != nil { if strings.Contains(err.Error(), "不存在") || strings.Contains(err.Error(), "不完整") { response.WriteJSON(w, response.ErrDefault(err.Error())) @@ -1399,7 +1402,9 @@ func (h *Handler) forwardDiagnose(w http.ResponseWriter, r *http.Request) { response.WriteJSON(w, response.Err(-2, err.Error())) return } - payload, err := h.diagnoseForwardRuntime(forward) + ctx, cancel := context.WithTimeout(r.Context(), diagnosisRequestTimeout) + defer cancel() + payload, err := h.diagnoseForwardRuntime(ctx, forward) if err != nil { if strings.Contains(err.Error(), "不存在") || strings.Contains(err.Error(), "不能为空") || strings.Contains(err.Error(), "错误") { response.WriteJSON(w, response.ErrDefault(err.Error())) diff --git a/vite-frontend/nginx.conf b/vite-frontend/nginx.conf index 207eb04..76e1883 100644 --- a/vite-frontend/nginx.conf +++ b/vite-frontend/nginx.conf @@ -55,11 +55,41 @@ http { } # API代理到后端 + location = /api/v1/tunnel/diagnose/stream { + proxy_set_header Host $host; + proxy_set_header X-Real-IP $remote_addr; + proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; + proxy_set_header X-Forwarded-Proto $scheme; + proxy_http_version 1.1; + proxy_buffering off; + proxy_cache off; + chunked_transfer_encoding on; + proxy_read_timeout 120s; + proxy_send_timeout 120s; + proxy_pass http://backend:6365/api/v1/tunnel/diagnose/stream; + } + + location = /api/v1/forward/diagnose/stream { + proxy_set_header Host $host; + proxy_set_header X-Real-IP $remote_addr; + proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; + proxy_set_header X-Forwarded-Proto $scheme; + proxy_http_version 1.1; + proxy_buffering off; + proxy_cache off; + chunked_transfer_encoding on; + proxy_read_timeout 120s; + proxy_send_timeout 120s; + proxy_pass http://backend:6365/api/v1/forward/diagnose/stream; + } + location ^~ /api/v1/ { proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; proxy_set_header X-Forwarded-Proto $scheme; + proxy_read_timeout 120s; + proxy_send_timeout 120s; proxy_pass http://backend:6365/api/v1/; } diff --git a/vite-frontend/src/api/diagnosis-stream.ts b/vite-frontend/src/api/diagnosis-stream.ts new file mode 100644 index 0000000..8b8dec7 --- /dev/null +++ b/vite-frontend/src/api/diagnosis-stream.ts @@ -0,0 +1,343 @@ +import axios from "axios"; + +import type { TunnelDiagnosisApiItem } from "@/api/types"; +import { clearSession, getToken } from "@/utils/session"; + +const DIAGNOSIS_STREAM_TIMEOUT_MS = 2 * 60 * 1000; + +type RawObject = Record; + +interface DiagnosisStreamRawEvent { + type?: string; + data?: unknown; + ts?: number; +} + +export interface DiagnosisStreamProgress { + total: number; + completed: number; + success: number; + failed: number; + timedOut?: boolean; +} + +export interface DiagnosisStreamItemPayload { + index: number; + result: TunnelDiagnosisApiItem; + progress: DiagnosisStreamProgress; +} + +export interface DiagnosisStreamRunResult { + fallback: boolean; + completed: boolean; + timedOut: boolean; + receivedItems: number; +} + +interface DiagnosisStreamCallbacks { + onStart?: (payload: RawObject) => void; + onItem: (payload: DiagnosisStreamItemPayload) => void; + onDone?: (payload: DiagnosisStreamProgress) => void; + onError?: (message: string) => void; +} + +interface RunDiagnosisStreamOptions extends DiagnosisStreamCallbacks { + path: string; + body: RawObject; + signal?: AbortSignal; +} + +const normalizeProgress = ( + payload: unknown, + fallback: DiagnosisStreamProgress, +): DiagnosisStreamProgress => { + if (!payload || typeof payload !== "object") { + return fallback; + } + const candidate = payload as RawObject; + const total = Number(candidate.total); + const completed = Number(candidate.completed); + const success = Number(candidate.success); + const failed = Number(candidate.failed); + + return { + total: Number.isFinite(total) && total >= 0 ? total : fallback.total, + completed: + Number.isFinite(completed) && completed >= 0 + ? completed + : fallback.completed, + success: + Number.isFinite(success) && success >= 0 ? success : fallback.success, + failed: Number.isFinite(failed) && failed >= 0 ? failed : fallback.failed, + timedOut: + typeof candidate.timedOut === "boolean" + ? candidate.timedOut + : fallback.timedOut, + }; +}; + +const resolveApiPath = (path: string): string => { + const normalizedPath = path.replace(/^\//, ""); + const baseURL = axios.defaults.baseURL || "/api/v1/"; + const normalizedBase = baseURL.endsWith("/") ? baseURL : `${baseURL}/`; + + return `${normalizedBase}${normalizedPath}`; +}; + +const isStreamSupported = (): boolean => { + return ( + typeof window !== "undefined" && + typeof fetch === "function" && + typeof TextDecoder !== "undefined" + ); +}; + +const handleTokenExpired = () => { + clearSession(); + if (window.location.pathname !== "/") { + window.location.href = "/"; + } +}; + +const combineAbortSignals = (signals: AbortSignal[]): AbortSignal => { + const controller = new AbortController(); + const onAbort = () => { + if (!controller.signal.aborted) { + controller.abort(); + } + }; + signals.forEach((signal) => { + if (signal.aborted) { + onAbort(); + return; + } + signal.addEventListener("abort", onAbort, { once: true }); + }); + return controller.signal; +}; + +const parseMessage = (err: unknown, fallback: string): string => { + if (err instanceof Error && err.message) { + return err.message; + } + return fallback; +}; + +const runDiagnosisStream = async ({ + path, + body, + signal, + onStart, + onItem, + onDone, + onError, +}: RunDiagnosisStreamOptions): Promise => { + if (!isStreamSupported()) { + return { fallback: true, completed: false, timedOut: false, receivedItems: 0 }; + } + + let receivedItems = 0; + let completed = false; + let timedOut = false; + let currentProgress: DiagnosisStreamProgress = { + total: 0, + completed: 0, + success: 0, + failed: 0, + }; + + const timeoutController = new AbortController(); + const timeoutId = window.setTimeout(() => { + timedOut = true; + timeoutController.abort(); + }, DIAGNOSIS_STREAM_TIMEOUT_MS); + + const mergedSignal = signal + ? combineAbortSignals([timeoutController.signal, signal]) + : timeoutController.signal; + + try { + const response = await fetch(resolveApiPath(path), { + method: "POST", + headers: { + "Content-Type": "application/json", + Accept: "application/x-ndjson, application/json", + Authorization: getToken() || "", + }, + body: JSON.stringify(body), + signal: mergedSignal, + }); + + if (response.status === 401) { + handleTokenExpired(); + return { fallback: false, completed: false, timedOut: false, receivedItems }; + } + + if (response.status === 404) { + return { fallback: true, completed: false, timedOut: false, receivedItems }; + } + + if (!response.ok || !response.body) { + const fallbackMessage = `请求失败(${response.status})`; + let message = fallbackMessage; + try { + const data = (await response.json()) as RawObject; + if (typeof data.msg === "string" && data.msg.trim()) { + message = data.msg; + } + } catch {} + if (receivedItems === 0) { + return { fallback: true, completed: false, timedOut: false, receivedItems }; + } + onError?.(message); + return { fallback: false, completed: false, timedOut: false, receivedItems }; + } + + const reader = response.body.getReader(); + const decoder = new TextDecoder(); + let buffer = ""; + + const processLine = (line: string) => { + if (!line) { + return; + } + let parsed: DiagnosisStreamRawEvent; + try { + parsed = JSON.parse(line) as DiagnosisStreamRawEvent; + } catch { + return; + } + + const eventType = (parsed.type || "").toLowerCase(); + if (eventType === "start") { + if (parsed.data && typeof parsed.data === "object") { + const startData = parsed.data as RawObject; + const startTotal = Number(startData.total); + if (Number.isFinite(startTotal) && startTotal >= 0) { + currentProgress = { ...currentProgress, total: startTotal }; + } + onStart?.(startData); + } + return; + } + + if (eventType === "item") { + if (!parsed.data || typeof parsed.data !== "object") { + return; + } + const itemData = parsed.data as RawObject; + const index = Number(itemData.index); + const result = itemData.result as TunnelDiagnosisApiItem | undefined; + if (!Number.isFinite(index) || !result || typeof result !== "object") { + return; + } + const progress = normalizeProgress(itemData.progress, currentProgress); + currentProgress = progress; + receivedItems += 1; + onItem({ + index, + result, + progress, + }); + return; + } + + if (eventType === "done") { + completed = true; + const donePayload = + parsed.data && typeof parsed.data === "object" + ? (parsed.data as RawObject) + : {}; + const doneProgress = normalizeProgress( + donePayload.progress ?? donePayload, + currentProgress, + ); + if (typeof donePayload.timedOut === "boolean") { + doneProgress.timedOut = donePayload.timedOut; + timedOut = donePayload.timedOut; + } + currentProgress = doneProgress; + onDone?.(doneProgress); + } + }; + + while (true) { + const { value, done } = await reader.read(); + if (done) { + break; + } + buffer += decoder.decode(value, { stream: true }); + const lines = buffer.split("\n"); + buffer = lines.pop() ?? ""; + lines.forEach((line) => processLine(line.trim())); + } + + const tail = buffer.trim(); + if (tail) { + processLine(tail); + } + + if (!completed && timedOut) { + const timeoutProgress = { + ...currentProgress, + timedOut: true, + }; + onDone?.(timeoutProgress); + } + + return { + fallback: false, + completed, + timedOut, + receivedItems, + }; + } catch (error) { + if (timedOut) { + const timeoutProgress = { + ...currentProgress, + timedOut: true, + }; + onDone?.(timeoutProgress); + return { fallback: false, completed: false, timedOut: true, receivedItems }; + } + + if (signal?.aborted) { + return { fallback: false, completed: false, timedOut: false, receivedItems }; + } + + if (receivedItems === 0) { + return { fallback: true, completed: false, timedOut: false, receivedItems }; + } + + onError?.(parseMessage(error, "流式诊断中断")); + return { fallback: false, completed: false, timedOut: false, receivedItems }; + } finally { + clearTimeout(timeoutId); + } +}; + +export const diagnoseTunnelStream = ( + tunnelId: number, + callbacks: DiagnosisStreamCallbacks, + signal?: AbortSignal, +) => { + return runDiagnosisStream({ + path: "/tunnel/diagnose/stream", + body: { tunnelId }, + signal, + ...callbacks, + }); +}; + +export const diagnoseForwardStream = ( + forwardId: number, + callbacks: DiagnosisStreamCallbacks, + signal?: AbortSignal, +) => { + return runDiagnosisStream({ + path: "/forward/diagnose/stream", + body: { forwardId }, + signal, + ...callbacks, + }); +}; diff --git a/vite-frontend/src/api/index.ts b/vite-frontend/src/api/index.ts index 3e84015..305d549 100644 --- a/vite-frontend/src/api/index.ts +++ b/vite-frontend/src/api/index.ts @@ -121,7 +121,7 @@ export const diagnoseTunnel = (tunnelId: number) => Network.post( "/tunnel/diagnose", { tunnelId }, - { timeout: 60 * 1000 }, + { timeout: 120 * 1000 }, ); export const updateTunnelOrder = (data: { tunnels: Array<{ id: number; inx: number }>; @@ -166,7 +166,7 @@ export const diagnoseForward = (forwardId: number) => Network.post( "/forward/diagnose", { forwardId }, - { timeout: 60 * 1000 }, + { timeout: 120 * 1000 }, ); // 转发排序操作 diff --git a/vite-frontend/src/pages/forward.tsx b/vite-frontend/src/pages/forward.tsx index 4f6c92d..dacaa5a 100644 --- a/vite-frontend/src/pages/forward.tsx +++ b/vite-frontend/src/pages/forward.tsx @@ -1,6 +1,6 @@ import type { SpeedLimitApiItem } from "@/api/types"; -import { useState, useEffect, useMemo } from "react"; +import { useState, useEffect, useMemo, useRef } from "react"; import toast from "react-hot-toast"; import { DndContext, @@ -76,6 +76,7 @@ import { getForwardDiagnosisQualityDisplay, type ForwardDiagnosisResult, } from "@/pages/forward/diagnosis"; +import { diagnoseForwardStream } from "@/api/diagnosis-stream"; import { executeForwardBatchChangeTunnel, executeForwardBatchDelete, @@ -133,10 +134,17 @@ interface ForwardForm { interface ForwardUserGroup { userId: number; userName: string; + tunnels: ForwardTunnelGroup[]; +} + +interface ForwardTunnelGroup { + tunnelKey: string; + tunnelName: string; items: Forward[]; } const UNKNOWN_FORWARD_USER_NAME = "未知用户"; +const UNCATEGORIZED_FORWARD_TUNNEL_NAME = "未分类"; const normalizeForwardUserName = (userName?: string): string => { const normalized = (userName || UNKNOWN_FORWARD_USER_NAME).trim(); @@ -151,6 +159,29 @@ const compareForwardUserNameAsc = (a: string, b: string): number => { }); }; +const normalizeForwardTunnelName = (tunnelName?: string): string => { + const normalized = (tunnelName || "").trim(); + + return normalized || UNCATEGORIZED_FORWARD_TUNNEL_NAME; +}; + +const buildForwardTunnelGroupKey = (tunnelName?: string): string => { + const normalized = normalizeForwardTunnelName(tunnelName); + + if (normalized === UNCATEGORIZED_FORWARD_TUNNEL_NAME) { + return "__uncategorized__"; + } + + return normalized.toLocaleLowerCase(); +}; + +const compareForwardTunnelNameAsc = (a: string, b: string): number => { + return a.localeCompare(b, "en", { + sensitivity: "base", + numeric: true, + }); +}; + export default function ForwardPage() { const [loading, setLoading] = useState(true); const [forwards, setForwards] = useState([]); @@ -198,6 +229,14 @@ export default function ForwardPage() { useState(null); const [diagnosisResult, setDiagnosisResult] = useState(null); + const [diagnosisProgress, setDiagnosisProgress] = useState({ + total: 0, + completed: 0, + success: 0, + failed: 0, + timedOut: false, + }); + const diagnosisAbortRef = useRef(null); const [addressModalTitle, setAddressModalTitle] = useState(""); const [addressList, setAddressList] = useState([]); @@ -250,6 +289,13 @@ export default function ForwardPage() { ); const [batchLoading, setBatchLoading] = useState(false); + useEffect(() => { + return () => { + diagnosisAbortRef.current?.abort(); + diagnosisAbortRef.current = null; + }; + }, []); + const parseShareIdFromTunnelName = (tunnelName: string): number | null => { const normalized = (tunnelName || "").trim(); @@ -800,28 +846,148 @@ export default function ForwardPage() { // 诊断转发 const handleDiagnose = async (forward: Forward) => { + diagnosisAbortRef.current?.abort(); + const abortController = new AbortController(); + diagnosisAbortRef.current = abortController; + setCurrentDiagnosisForward(forward); setDiagnosisModalOpen(true); setDiagnosisLoading(true); - setDiagnosisResult(null); + setDiagnosisProgress({ + total: 0, + completed: 0, + success: 0, + failed: 0, + timedOut: false, + }); + setDiagnosisResult({ + forwardName: forward.name, + timestamp: Date.now(), + results: [], + }); try { - const response = await diagnoseForward(forward.id); + let streamErrorMessage = ""; + const streamResult = await diagnoseForwardStream( + forward.id, + { + onStart: (payload) => { + const startForwardName = + typeof payload.forwardName === "string" && + payload.forwardName.trim() !== "" + ? payload.forwardName + : forward.name; + const startTotal = Number(payload.total); + setDiagnosisResult((prev) => ({ + forwardName: startForwardName, + timestamp: Date.now(), + results: prev?.results || [], + })); + if (Number.isFinite(startTotal) && startTotal >= 0) { + setDiagnosisProgress((prev) => ({ + ...prev, + total: startTotal, + })); + } + }, + onItem: ({ result, progress }) => { + setDiagnosisResult((prev) => { + const base: ForwardDiagnosisResult = prev || { + forwardName: forward.name, + timestamp: Date.now(), + results: [], + }; + const nextResults = [...base.results]; + const existingIndex = nextResults.findIndex( + (item) => + item.description === result.description && + item.nodeId === result.nodeId && + item.targetIp === result.targetIp && + item.targetPort === result.targetPort, + ); - if (response.code === 0) { - setDiagnosisResult(response.data as ForwardDiagnosisResult); - } else { - toast.error(response.msg || "诊断失败"); - setDiagnosisResult( - buildForwardDiagnosisFallbackResult({ - forwardName: forward.name, - remoteAddr: forward.remoteAddr, - description: "诊断失败", - message: response.msg || "诊断过程中发生错误", - }), - ); + if (existingIndex >= 0) { + nextResults[existingIndex] = result; + } else { + nextResults.push(result); + } + return { + ...base, + timestamp: Date.now(), + results: nextResults, + }; + }); + setDiagnosisProgress({ + total: progress.total, + completed: progress.completed, + success: progress.success, + failed: progress.failed, + timedOut: Boolean(progress.timedOut), + }); + }, + onDone: (progress) => { + setDiagnosisProgress({ + total: progress.total, + completed: progress.completed, + success: progress.success, + failed: progress.failed, + timedOut: Boolean(progress.timedOut), + }); + }, + onError: (message) => { + streamErrorMessage = message; + }, + }, + abortController.signal, + ); + + if (streamResult.fallback) { + const response = await diagnoseForward(forward.id); + + if (response.code === 0) { + const resultData = response.data as ForwardDiagnosisResult; + const successCount = resultData.results.filter((r) => r.success).length; + const failedCount = resultData.results.length - successCount; + setDiagnosisResult(resultData); + setDiagnosisProgress({ + total: resultData.results.length, + completed: resultData.results.length, + success: successCount, + failed: failedCount, + timedOut: false, + }); + } else { + toast.error(response.msg || "诊断失败"); + setDiagnosisResult( + buildForwardDiagnosisFallbackResult({ + forwardName: forward.name, + remoteAddr: forward.remoteAddr, + description: "诊断失败", + message: response.msg || "诊断过程中发生错误", + }), + ); + setDiagnosisProgress({ + total: 1, + completed: 1, + success: 0, + failed: 1, + timedOut: false, + }); + } + + return; + } + + if (streamErrorMessage) { + toast.error(streamErrorMessage); + } + if (streamResult.timedOut) { + toast.error("诊断达到2分钟超时，已返回当前结果"); } } catch { + if (abortController.signal.aborted) { + return; + } toast.error("网络错误，请重试"); setDiagnosisResult( buildForwardDiagnosisFallbackResult({ @@ -831,7 +997,17 @@ export default function ForwardPage() { message: "无法连接到服务器", }), ); + setDiagnosisProgress({ + total: 1, + completed: 1, + success: 0, + failed: 1, + timedOut: false, + }); } finally { + if (diagnosisAbortRef.current === abortController) { + diagnosisAbortRef.current = null; + } setDiagnosisLoading(false); } }; @@ -1170,9 +1346,13 @@ export default function ForwardPage() { const overForward = forwards.find((forward) => forward.id === overId); const activeUserId = activeForward?.userId ?? 0; const overUserId = overForward?.userId ?? 0; + const activeTunnelGroupKey = buildForwardTunnelGroupKey( + activeForward?.tunnelName, + ); + const overTunnelGroupKey = buildForwardTunnelGroupKey(overForward?.tunnelName); - // 仅允许在同一用户分组内拖拽，避免不同用户数据混排 - if (activeUserId !== overUserId) { + // 仅允许在同一用户+隧道分组内拖拽，避免混排 + if (activeUserId !== overUserId || activeTunnelGroupKey !== overTunnelGroupKey) { return; } @@ -1469,34 +1649,86 @@ export default function ForwardPage() { return []; } - const userGroupMap = new Map(); + type MutableForwardUserGroup = { + userId: number; + userName: string; + tunnelMap: Map; + }; + + const userGroupMap = new Map(); orderedForwards.forEach((forward) => { const userId = forward.userId ?? 0; const userName = normalizeForwardUserName(forward.userName); - const existingGroup = userGroupMap.get(userId); + const tunnelName = normalizeForwardTunnelName(forward.tunnelName); + const tunnelKey = buildForwardTunnelGroupKey(forward.tunnelName); + + let existingGroup = userGroupMap.get(userId); if (!existingGroup) { - userGroupMap.set(userId, { + existingGroup = { userId, userName, + tunnelMap: new Map(), + }; + userGroupMap.set(userId, existingGroup); + } else if ( + existingGroup.userName === UNKNOWN_FORWARD_USER_NAME && + userName !== UNKNOWN_FORWARD_USER_NAME + ) { + existingGroup.userName = userName; + } + + const existingTunnelGroup = existingGroup.tunnelMap.get(tunnelKey); + + if (!existingTunnelGroup) { + existingGroup.tunnelMap.set(tunnelKey, { + tunnelKey, + tunnelName, items: [forward], }); return; } - existingGroup.items.push(forward); + existingTunnelGroup.items.push(forward); if ( - existingGroup.userName === UNKNOWN_FORWARD_USER_NAME && - userName !== UNKNOWN_FORWARD_USER_NAME + existingTunnelGroup.tunnelName === UNCATEGORIZED_FORWARD_TUNNEL_NAME && + tunnelName !== UNCATEGORIZED_FORWARD_TUNNEL_NAME ) { - existingGroup.userName = userName; + existingTunnelGroup.tunnelName = tunnelName; } }); - const groups = Array.from(userGroupMap.values()); + const groups = Array.from(userGroupMap.values()).map((group) => { + const tunnels = Array.from(group.tunnelMap.values()); + + tunnels.sort((a, b) => { + const aIsUncategorized = + a.tunnelName === UNCATEGORIZED_FORWARD_TUNNEL_NAME; + const bIsUncategorized = + b.tunnelName === UNCATEGORIZED_FORWARD_TUNNEL_NAME; + + if (aIsUncategorized !== bIsUncategorized) { + return aIsUncategorized ? 1 : -1; + } + + const nameCompare = compareForwardTunnelNameAsc(a.tunnelName, b.tunnelName); + + if (nameCompare !== 0) { + return nameCompare; + } + + return compareForwardTunnelNameAsc(a.tunnelKey, b.tunnelKey); + }); + + return { + userId: group.userId, + userName: group.userName, + tunnels, + }; + }); groups.sort((a, b) => { if (isAdmin && tokenUserId !== null) { @@ -1521,7 +1753,10 @@ export default function ForwardPage() { }, [orderedForwards, isAdmin, tokenUserId]); const sortedForwards = useMemo( - () => groupedForwards.flatMap((group) => group.items), + () => + groupedForwards.flatMap((group) => + group.tunnels.flatMap((tunnel) => tunnel.items), + ), [groupedForwards], ); @@ -1673,17 +1908,6 @@ export default function ForwardPage() { {forward.name} - - - - {forward.tunnelName} - - -