opsxapi/apps/server/internal/proxy/stream.go

package proxy

import (
	"bufio"
	"bytes"
	"encoding/base64"
	"encoding/json"
	"io"
	"net/http"
	"time"

	"github.com/gin-gonic/gin"
)

// ProxyHTTPStream 以 NDJSON 流式转发响应（SSE / chunked / 大响应体）。
// 事件序列：meta → chunk* → end | error
type streamEvent struct {
	Type    string            `json:"type"`
	Status  int               `json:"status,omitempty"`
	Headers map[string]string `json:"headers,omitempty"`
	Data    string            `json:"data,omitempty"`
	Message string            `json:"message,omitempty"`
	TimeMS  int64             `json:"timeMs,omitempty"`
	Size    int               `json:"size,omitempty"`
}

func (h *Handler) ProxyHTTPStream(c *gin.Context) {
	var spec HttpSpec
	if err := c.ShouldBindJSON(&spec); err != nil {
		c.JSON(http.StatusBadRequest, gin.H{"error": "invalid HttpSpec", "detail": err.Error()})
		return
	}

	method := normalizeMethod(spec.Method)
	u, err := ValidateURL(spec.URL)
	if err != nil {
		c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
		return
	}
	query := u.Query()
	for key, value := range spec.Query {
		query.Set(key, value)
	}
	u.RawQuery = query.Encode()

	var body io.Reader
	if spec.Body != nil && spec.Body.Mode != "none" && method != http.MethodGet && method != http.MethodHead {
		body = bytes.NewBufferString(spec.Body.Raw)
	}

	req, err := http.NewRequestWithContext(c.Request.Context(), method, u.String(), body)
	if err != nil {
		c.JSON(http.StatusBadRequest, gin.H{"error": "cannot create request", "detail": err.Error()})
		return
	}
	for key, value := range spec.Headers {
		if !isHopByHop(key) {
			req.Header.Set(key, value)
		}
	}
	if spec.Body != nil && spec.Body.Mode == "json" && req.Header.Get("Content-Type") == "" {
		req.Header.Set("Content-Type", "application/json")
	}

	client := h.BaseClient
	if spec.TimeoutMS > 0 {
		client = &http.Client{Timeout: time.Duration(spec.TimeoutMS) * time.Millisecond}
	} else {
		// 流式场景默认不设整体超时（SSE 长连接），交给客户端断开
		client = &http.Client{}
	}

	started := time.Now()
	res, err := client.Do(req)
	if err != nil {
		c.JSON(http.StatusBadGateway, gin.H{"error": "upstream request failed", "detail": err.Error()})
		return
	}
	defer res.Body.Close()

	headers := make(map[string]string, len(res.Header))
	for key, values := range res.Header {
		headers[key] = joinNonEmpty(values)
	}

	flusher, ok := c.Writer.(http.Flusher)
	if !ok {
		c.JSON(http.StatusInternalServerError, gin.H{"error": "streaming unsupported"})
		return
	}
	c.Header("Content-Type", "application/x-ndjson")
	c.Header("Cache-Control", "no-cache")
	c.Header("X-Accel-Buffering", "no")
	c.Status(http.StatusOK)

	enc := json.NewEncoder(c.Writer)
	writeEvent := func(ev streamEvent) bool {
		if err := enc.Encode(ev); err != nil {
			return false
		}
		flusher.Flush()
		return true
	}

	headers["content-type"] = res.Header.Get("Content-Type")
	if !writeEvent(streamEvent{Type: "meta", Status: res.StatusCode, Headers: headers}) {
		return
	}

	reader := bufio.NewReaderSize(res.Body, 32*1024)
	buf := make([]byte, 16*1024)
	total := 0
	for {
		n, readErr := reader.Read(buf)
		if n > 0 {
			total += n
			ok := writeEvent(streamEvent{
				Type: "chunk",
				Data: base64.StdEncoding.EncodeToString(buf[:n]),
				Size: n,
			})
			if !ok {
				return
			}
		}
		if readErr != nil {
			break
		}
		if c.Request.Context().Err() != nil {
			return
		}
	}
	writeEvent(streamEvent{Type: "end", TimeMS: time.Since(started).Milliseconds(), Size: total})
}

func joinNonEmpty(values []string) string {
	out := ""
	for i, v := range values {
		if i > 0 {
			out += ", "
		}
		out += v
	}
	return out
}