mirror of
https://github.com/Wei-Shaw/sub2api.git
synced 2026-09-24 16:05:44 +08:00
Merge pull request #3807 from Wei-Shaw/refactor/gateway-cc-pipeline-and-file-split
refactor(gateway): 抽取 CC forwarder 公共管线并拆分两大 service 文件
This commit is contained in:
Wesley Liddick
@@ -0,0 +1,810 @@
|
||||
package service
|
||||
|
||||
// 本文件由 gateway_service.go 纯移动拆分而来：Anthropic APIKey 直通
|
||||
// （passthrough）转发路径及其流式/非流式响应与 usage 解析。仅做代码搬迁，
|
||||
// 无任何行为变更。
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"strings"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
"github.com/Wei-Shaw/sub2api/internal/pkg/logger"
|
||||
"github.com/Wei-Shaw/sub2api/internal/util/responseheaders"
|
||||
"github.com/tidwall/gjson"
|
||||
|
||||
"github.com/gin-gonic/gin"
|
||||
)
|
||||
|
||||
type anthropicPassthroughForwardInput struct {
|
||||
Body []byte
|
||||
Parsed *ParsedRequest
|
||||
RequestModel string
|
||||
OriginalModel string
|
||||
RequestStream bool
|
||||
StartTime time.Time
|
||||
}
|
||||
|
||||
func (s *GatewayService) forwardAnthropicAPIKeyPassthrough(
|
||||
ctx context.Context,
|
||||
c *gin.Context,
|
||||
account *Account,
|
||||
body []byte,
|
||||
reqModel string,
|
||||
originalModel string,
|
||||
reqStream bool,
|
||||
startTime time.Time,
|
||||
) (*ForwardResult, error) {
|
||||
return s.forwardAnthropicAPIKeyPassthroughWithInput(ctx, c, account, anthropicPassthroughForwardInput{
|
||||
Body: body,
|
||||
RequestModel: reqModel,
|
||||
OriginalModel: originalModel,
|
||||
RequestStream: reqStream,
|
||||
StartTime: startTime,
|
||||
})
|
||||
}
|
||||
|
||||
func (s *GatewayService) forwardAnthropicAPIKeyPassthroughWithInput(
|
||||
ctx context.Context,
|
||||
c *gin.Context,
|
||||
account *Account,
|
||||
input anthropicPassthroughForwardInput,
|
||||
) (*ForwardResult, error) {
|
||||
token, tokenType, err := s.GetAccessToken(ctx, account)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if tokenType != "apikey" {
|
||||
return nil, fmt.Errorf("anthropic api key passthrough requires apikey token, got: %s", tokenType)
|
||||
}
|
||||
|
||||
proxyURL := ""
|
||||
if account.ProxyID != nil && account.Proxy != nil {
|
||||
proxyURL = account.Proxy.URL()
|
||||
}
|
||||
|
||||
logger.LegacyPrintf("service.gateway", "[Anthropic 自动透传] 命中 API Key 透传分支: account=%d name=%s model=%s stream=%v",
|
||||
account.ID, account.Name, input.RequestModel, input.RequestStream)
|
||||
|
||||
if c != nil {
|
||||
c.Set("anthropic_passthrough", true)
|
||||
}
|
||||
// Pre-filter: strip empty text blocks (including nested in tool_result) to prevent upstream 400.
|
||||
input.Body = StripEmptyTextBlocks(input.Body)
|
||||
// Pre-filter: strip web-search history blocks the upstream cannot accept
|
||||
// (emulation-synthesized ones always; genuine ones additionally for
|
||||
// passback-required third-party upstreams such as GLM/Kimi/DeepSeek,
|
||||
// which reject server_tool_use with 400). input.RequestModel 已是映射后的模型 ID。
|
||||
input.Body = FilterWebSearchHistoryBlocks(input.Body, input.RequestModel)
|
||||
if input.Parsed != nil {
|
||||
// 透传分支也会改写实际 wire body，成功 usage hash 依赖这里同步当前 body。
|
||||
if err := input.Parsed.ReplaceBody(input.Body); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
|
||||
var resp *http.Response
|
||||
retryStart := time.Now()
|
||||
for attempt := 1; attempt <= maxRetryAttempts; attempt++ {
|
||||
upstreamCtx, releaseUpstreamCtx := detachStreamUpstreamContext(ctx, input.RequestStream)
|
||||
upstreamReq, wireBody, err := s.buildUpstreamRequestAnthropicAPIKeyPassthrough(upstreamCtx, c, account, input.Body, token)
|
||||
releaseUpstreamCtx()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if input.Parsed != nil && !bytes.Equal(wireBody, input.Body) {
|
||||
// build 阶段会按 beta 能力清理 body，发送前同步到 ParsedRequest 当前视图。
|
||||
if err := input.Parsed.ReplaceBody(wireBody); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
input.Body = input.Parsed.Body.Bytes()
|
||||
}
|
||||
|
||||
resp, err = s.httpUpstream.DoWithTLS(upstreamReq, proxyURL, account.ID, account.Concurrency, s.tlsFPProfileService.ResolveTLSProfile(account))
|
||||
if err != nil {
|
||||
if resp != nil && resp.Body != nil {
|
||||
_ = resp.Body.Close()
|
||||
}
|
||||
safeErr := sanitizeUpstreamErrorMessage(err.Error())
|
||||
setOpsUpstreamError(c, 0, safeErr, "")
|
||||
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
|
||||
Platform: account.Platform,
|
||||
AccountID: account.ID,
|
||||
AccountName: account.Name,
|
||||
UpstreamStatusCode: 0,
|
||||
UpstreamURL: safeUpstreamURL(upstreamReq.URL.String()),
|
||||
Passthrough: true,
|
||||
Kind: "request_error",
|
||||
Message: safeErr,
|
||||
})
|
||||
c.JSON(http.StatusBadGateway, gin.H{
|
||||
"type": "error",
|
||||
"error": gin.H{
|
||||
"type": "upstream_error",
|
||||
"message": "Upstream request failed",
|
||||
},
|
||||
})
|
||||
return nil, fmt.Errorf("upstream request failed: %s", safeErr)
|
||||
}
|
||||
|
||||
// 透传分支禁止 400 请求体降级重试（该重试会改写请求体）
|
||||
if resp.StatusCode >= 400 && resp.StatusCode != 400 && s.shouldRetryUpstreamError(account, resp.StatusCode) {
|
||||
if attempt < maxRetryAttempts {
|
||||
elapsed := time.Since(retryStart)
|
||||
if elapsed >= maxRetryElapsed {
|
||||
break
|
||||
}
|
||||
|
||||
delay := retryBackoffDelay(attempt)
|
||||
remaining := maxRetryElapsed - elapsed
|
||||
if delay > remaining {
|
||||
delay = remaining
|
||||
}
|
||||
if delay <= 0 {
|
||||
break
|
||||
}
|
||||
|
||||
respBody, _ := s.readUpstreamErrorBody(resp)
|
||||
_ = resp.Body.Close()
|
||||
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
|
||||
Platform: account.Platform,
|
||||
AccountID: account.ID,
|
||||
AccountName: account.Name,
|
||||
UpstreamStatusCode: resp.StatusCode,
|
||||
UpstreamRequestID: resp.Header.Get("x-request-id"),
|
||||
UpstreamURL: safeUpstreamURL(upstreamReq.URL.String()),
|
||||
Passthrough: true,
|
||||
Kind: "retry",
|
||||
Message: extractUpstreamErrorMessage(respBody),
|
||||
Detail: func() string {
|
||||
if s.cfg != nil && s.cfg.Gateway.LogUpstreamErrorBody {
|
||||
return truncateString(string(respBody), s.cfg.Gateway.LogUpstreamErrorBodyMaxBytes)
|
||||
}
|
||||
return ""
|
||||
}(),
|
||||
})
|
||||
logger.LegacyPrintf("service.gateway", "Anthropic passthrough account %d: upstream error %d, retry %d/%d after %v (elapsed=%v/%v)",
|
||||
account.ID, resp.StatusCode, attempt, maxRetryAttempts, delay, elapsed, maxRetryElapsed)
|
||||
if err := sleepWithContext(ctx, delay); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
continue
|
||||
}
|
||||
break
|
||||
}
|
||||
|
||||
break
|
||||
}
|
||||
if resp == nil || resp.Body == nil {
|
||||
return nil, errors.New("upstream request failed: empty response")
|
||||
}
|
||||
defer func() { _ = resp.Body.Close() }()
|
||||
|
||||
if resp.StatusCode >= 400 && s.shouldRetryUpstreamError(account, resp.StatusCode) {
|
||||
if s.shouldFailoverUpstreamError(resp.StatusCode) {
|
||||
respBody, _ := s.readUpstreamErrorBody(resp)
|
||||
_ = resp.Body.Close()
|
||||
resp.Body = io.NopCloser(bytes.NewReader(respBody))
|
||||
|
||||
logger.LegacyPrintf("service.gateway", "[Anthropic Passthrough] Upstream error (retry exhausted, failover): Account=%d(%s) Status=%d RequestID=%s Body=%s",
|
||||
account.ID, account.Name, resp.StatusCode, resp.Header.Get("x-request-id"), truncateString(string(respBody), 1000))
|
||||
|
||||
s.handleRetryExhaustedSideEffects(ctx, resp, account)
|
||||
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
|
||||
Platform: account.Platform,
|
||||
AccountID: account.ID,
|
||||
AccountName: account.Name,
|
||||
UpstreamStatusCode: resp.StatusCode,
|
||||
UpstreamRequestID: resp.Header.Get("x-request-id"),
|
||||
Passthrough: true,
|
||||
Kind: "retry_exhausted_failover",
|
||||
Message: extractUpstreamErrorMessage(respBody),
|
||||
Detail: func() string {
|
||||
if s.cfg != nil && s.cfg.Gateway.LogUpstreamErrorBody {
|
||||
return truncateString(string(respBody), s.cfg.Gateway.LogUpstreamErrorBodyMaxBytes)
|
||||
}
|
||||
return ""
|
||||
}(),
|
||||
})
|
||||
return nil, &UpstreamFailoverError{
|
||||
StatusCode: resp.StatusCode,
|
||||
ResponseBody: respBody,
|
||||
RetryableOnSameAccount: account.IsPoolMode() && account.IsPoolModeRetryableStatus(resp.StatusCode),
|
||||
}
|
||||
}
|
||||
return s.handleRetryExhaustedError(ctx, resp, c, account)
|
||||
}
|
||||
|
||||
if resp.StatusCode >= 400 && s.shouldFailoverUpstreamError(resp.StatusCode) {
|
||||
respBody, _ := s.readUpstreamErrorBody(resp)
|
||||
_ = resp.Body.Close()
|
||||
resp.Body = io.NopCloser(bytes.NewReader(respBody))
|
||||
|
||||
logger.LegacyPrintf("service.gateway", "[Anthropic Passthrough] Upstream error (failover): Account=%d(%s) Status=%d RequestID=%s Body=%s",
|
||||
account.ID, account.Name, resp.StatusCode, resp.Header.Get("x-request-id"), truncateString(string(respBody), 1000))
|
||||
|
||||
s.handleFailoverSideEffects(ctx, resp, account, input.RequestModel)
|
||||
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
|
||||
Platform: account.Platform,
|
||||
AccountID: account.ID,
|
||||
AccountName: account.Name,
|
||||
UpstreamStatusCode: resp.StatusCode,
|
||||
UpstreamRequestID: resp.Header.Get("x-request-id"),
|
||||
Passthrough: true,
|
||||
Kind: "failover",
|
||||
Message: extractUpstreamErrorMessage(respBody),
|
||||
Detail: func() string {
|
||||
if s.cfg != nil && s.cfg.Gateway.LogUpstreamErrorBody {
|
||||
return truncateString(string(respBody), s.cfg.Gateway.LogUpstreamErrorBodyMaxBytes)
|
||||
}
|
||||
return ""
|
||||
}(),
|
||||
})
|
||||
return nil, &UpstreamFailoverError{
|
||||
StatusCode: resp.StatusCode,
|
||||
ResponseBody: respBody,
|
||||
RetryableOnSameAccount: account.IsPoolMode() && account.IsPoolModeRetryableStatus(resp.StatusCode),
|
||||
}
|
||||
}
|
||||
|
||||
if resp.StatusCode >= 400 {
|
||||
return s.handleErrorResponse(ctx, resp, c, account, input.RequestModel)
|
||||
}
|
||||
|
||||
var usage *ClaudeUsage
|
||||
var firstTokenMs *int
|
||||
var clientDisconnect bool
|
||||
if input.RequestStream {
|
||||
streamResult, err := s.handleStreamingResponseAnthropicAPIKeyPassthrough(ctx, resp, c, account, input.StartTime, input.RequestModel)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
usage = streamResult.usage
|
||||
firstTokenMs = streamResult.firstTokenMs
|
||||
clientDisconnect = streamResult.clientDisconnect
|
||||
} else {
|
||||
usage, err = s.handleNonStreamingResponseAnthropicAPIKeyPassthrough(ctx, resp, c, account)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
if usage == nil {
|
||||
usage = &ClaudeUsage{}
|
||||
}
|
||||
|
||||
return &ForwardResult{
|
||||
RequestID: resp.Header.Get("x-request-id"),
|
||||
Usage: *usage,
|
||||
Model: input.OriginalModel,
|
||||
UpstreamModel: input.RequestModel,
|
||||
Stream: input.RequestStream,
|
||||
Duration: time.Since(input.StartTime),
|
||||
FirstTokenMs: firstTokenMs,
|
||||
ClientDisconnect: clientDisconnect,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (s *GatewayService) buildUpstreamRequestAnthropicAPIKeyPassthrough(
|
||||
ctx context.Context,
|
||||
c *gin.Context,
|
||||
account *Account,
|
||||
body []byte,
|
||||
token string,
|
||||
) (*http.Request, []byte, error) {
|
||||
targetURL := claudeAPIURL
|
||||
baseURL := account.GetBaseURL()
|
||||
if baseURL != "" {
|
||||
validatedURL, err := s.validateUpstreamBaseURL(baseURL)
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
targetURL = validatedURL + "/v1/messages?beta=true"
|
||||
}
|
||||
|
||||
// 能力维度 body sanitize：透传路径上 anthropic-beta header 原样透传客户端值，
|
||||
// 依此决定是否保留 body 中的 context_management。避免“客户端 body 带字段但
|
||||
// header 忘记带 beta token”的客户端 bug 在透传场景下让上游 400。
|
||||
clientBeta := ""
|
||||
if c != nil && c.Request != nil {
|
||||
clientBeta = getHeaderRaw(c.Request.Header, "anthropic-beta")
|
||||
}
|
||||
// 账号覆写了 anthropic-beta 时，覆写值即最终上游值：净化以覆写值为准
|
||||
if beta, ok := account.HeaderOverrideValue("anthropic-beta"); ok {
|
||||
clientBeta = beta
|
||||
}
|
||||
if sanitized, changed := sanitizeAnthropicBodyForBetaTokens(body, clientBeta); changed {
|
||||
body = sanitized
|
||||
}
|
||||
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodPost, targetURL, bytes.NewReader(body))
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
|
||||
if c != nil && c.Request != nil {
|
||||
for key, values := range c.Request.Header {
|
||||
lowerKey := strings.ToLower(strings.TrimSpace(key))
|
||||
if !allowedHeaders[lowerKey] {
|
||||
continue
|
||||
}
|
||||
wireKey := resolveWireCasing(key)
|
||||
for _, v := range values {
|
||||
addHeaderRaw(req.Header, wireKey, v)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// 覆盖入站鉴权残留，并注入上游认证
|
||||
req.Header.Del("authorization")
|
||||
req.Header.Del("x-api-key")
|
||||
req.Header.Del("x-goog-api-key")
|
||||
req.Header.Del("cookie")
|
||||
setAnthropicAPIKeyAuthHeader(req.Header, account, token)
|
||||
|
||||
if getHeaderRaw(req.Header, "content-type") == "" {
|
||||
setHeaderRaw(req.Header, "content-type", "application/json")
|
||||
}
|
||||
if getHeaderRaw(req.Header, "anthropic-version") == "" {
|
||||
setHeaderRaw(req.Header, "anthropic-version", "2023-06-01")
|
||||
}
|
||||
|
||||
// 账号级请求头覆写（最终生效，覆盖上面所有来源的同名头）
|
||||
account.ApplyHeaderOverrides(req.Header)
|
||||
|
||||
return req, body, nil
|
||||
}
|
||||
|
||||
func (s *GatewayService) handleStreamingResponseAnthropicAPIKeyPassthrough(
|
||||
ctx context.Context,
|
||||
resp *http.Response,
|
||||
c *gin.Context,
|
||||
account *Account,
|
||||
startTime time.Time,
|
||||
model string,
|
||||
) (*streamingResult, error) {
|
||||
if s.rateLimitService != nil {
|
||||
s.rateLimitService.UpdateSessionWindow(ctx, account, resp.Header)
|
||||
}
|
||||
|
||||
writeAnthropicPassthroughResponseHeaders(c.Writer.Header(), resp.Header, s.responseHeaderFilter)
|
||||
|
||||
contentType := strings.TrimSpace(resp.Header.Get("Content-Type"))
|
||||
if contentType == "" {
|
||||
contentType = "text/event-stream"
|
||||
}
|
||||
c.Header("Content-Type", contentType)
|
||||
if c.Writer.Header().Get("Cache-Control") == "" {
|
||||
c.Header("Cache-Control", "no-cache")
|
||||
}
|
||||
if c.Writer.Header().Get("Connection") == "" {
|
||||
c.Header("Connection", "keep-alive")
|
||||
}
|
||||
c.Header("X-Accel-Buffering", "no")
|
||||
if v := resp.Header.Get("x-request-id"); v != "" {
|
||||
c.Header("x-request-id", v)
|
||||
}
|
||||
|
||||
w := c.Writer
|
||||
flusher, ok := w.(http.Flusher)
|
||||
if !ok {
|
||||
return nil, errors.New("streaming not supported")
|
||||
}
|
||||
|
||||
usage := &ClaudeUsage{}
|
||||
var firstTokenMs *int
|
||||
clientDisconnected := false
|
||||
sawTerminalEvent := false
|
||||
|
||||
scanner := bufio.NewScanner(resp.Body)
|
||||
maxLineSize := defaultMaxLineSize
|
||||
if s.cfg != nil && s.cfg.Gateway.MaxLineSize > 0 {
|
||||
maxLineSize = s.cfg.Gateway.MaxLineSize
|
||||
}
|
||||
scanBuf := getSSEScannerBuf64K()
|
||||
scanner.Buffer(scanBuf[:0], maxLineSize)
|
||||
|
||||
type scanEvent struct {
|
||||
line string
|
||||
err error
|
||||
}
|
||||
events := make(chan scanEvent, 16)
|
||||
done := make(chan struct{})
|
||||
sendEvent := func(ev scanEvent) bool {
|
||||
select {
|
||||
case events <- ev:
|
||||
return true
|
||||
case <-done:
|
||||
return false
|
||||
}
|
||||
}
|
||||
var lastReadAt int64
|
||||
atomic.StoreInt64(&lastReadAt, time.Now().UnixNano())
|
||||
go func(scanBuf *sseScannerBuf64K) {
|
||||
defer putSSEScannerBuf64K(scanBuf)
|
||||
defer close(events)
|
||||
for scanner.Scan() {
|
||||
atomic.StoreInt64(&lastReadAt, time.Now().UnixNano())
|
||||
if !sendEvent(scanEvent{line: scanner.Text()}) {
|
||||
return
|
||||
}
|
||||
}
|
||||
if err := scanner.Err(); err != nil {
|
||||
_ = sendEvent(scanEvent{err: err})
|
||||
}
|
||||
}(scanBuf)
|
||||
defer close(done)
|
||||
|
||||
streamInterval := time.Duration(0)
|
||||
if s.cfg != nil && s.cfg.Gateway.StreamDataIntervalTimeout > 0 {
|
||||
streamInterval = time.Duration(s.cfg.Gateway.StreamDataIntervalTimeout) * time.Second
|
||||
}
|
||||
var intervalTicker *time.Ticker
|
||||
if streamInterval > 0 {
|
||||
intervalTicker = time.NewTicker(streamInterval)
|
||||
defer intervalTicker.Stop()
|
||||
}
|
||||
var intervalCh <-chan time.Time
|
||||
if intervalTicker != nil {
|
||||
intervalCh = intervalTicker.C
|
||||
}
|
||||
|
||||
keepaliveInterval := time.Duration(0)
|
||||
if s.cfg != nil && s.cfg.Gateway.StreamKeepaliveInterval > 0 {
|
||||
keepaliveInterval = time.Duration(s.cfg.Gateway.StreamKeepaliveInterval) * time.Second
|
||||
}
|
||||
var keepaliveTimer *time.Timer
|
||||
if keepaliveInterval > 0 {
|
||||
keepaliveTimer = time.NewTimer(keepaliveInterval)
|
||||
defer keepaliveTimer.Stop()
|
||||
}
|
||||
var keepaliveCh <-chan time.Time
|
||||
if keepaliveTimer != nil {
|
||||
keepaliveCh = keepaliveTimer.C
|
||||
}
|
||||
lastDataAt := time.Now()
|
||||
resetKeepaliveTimer := func() {
|
||||
if keepaliveTimer == nil {
|
||||
return
|
||||
}
|
||||
if !keepaliveTimer.Stop() {
|
||||
select {
|
||||
case <-keepaliveTimer.C:
|
||||
default:
|
||||
}
|
||||
}
|
||||
keepaliveTimer.Reset(keepaliveInterval)
|
||||
}
|
||||
inPartialEvent := false
|
||||
|
||||
for {
|
||||
select {
|
||||
case ev, ok := <-events:
|
||||
if !ok {
|
||||
if !clientDisconnected {
|
||||
// 兜底补刷，确保最后一个未以空行结尾的事件也能及时送达客户端。
|
||||
flusher.Flush()
|
||||
}
|
||||
if !sawTerminalEvent {
|
||||
if clientDisconnected && streamInterval > 0 {
|
||||
lastRead := time.Unix(0, atomic.LoadInt64(&lastReadAt))
|
||||
if time.Since(lastRead) >= streamInterval {
|
||||
return &streamingResult{usage: usage, firstTokenMs: firstTokenMs, clientDisconnect: true}, fmt.Errorf("stream usage incomplete after timeout")
|
||||
}
|
||||
}
|
||||
return &streamingResult{usage: usage, firstTokenMs: firstTokenMs, clientDisconnect: clientDisconnected}, fmt.Errorf("stream usage incomplete: missing terminal event")
|
||||
}
|
||||
return &streamingResult{usage: usage, firstTokenMs: firstTokenMs, clientDisconnect: clientDisconnected}, nil
|
||||
}
|
||||
if ev.err != nil {
|
||||
if sawTerminalEvent {
|
||||
return &streamingResult{usage: usage, firstTokenMs: firstTokenMs, clientDisconnect: clientDisconnected}, nil
|
||||
}
|
||||
if clientDisconnected {
|
||||
return &streamingResult{usage: usage, firstTokenMs: firstTokenMs, clientDisconnect: true}, fmt.Errorf("stream usage incomplete after disconnect: %w", ev.err)
|
||||
}
|
||||
if errors.Is(ev.err, context.Canceled) || errors.Is(ev.err, context.DeadlineExceeded) {
|
||||
return &streamingResult{usage: usage, firstTokenMs: firstTokenMs, clientDisconnect: true}, fmt.Errorf("stream usage incomplete: %w", ev.err)
|
||||
}
|
||||
if errors.Is(ev.err, bufio.ErrTooLong) {
|
||||
logger.LegacyPrintf("service.gateway", "[Anthropic passthrough] SSE line too long: account=%d max_size=%d error=%v", account.ID, maxLineSize, ev.err)
|
||||
return &streamingResult{usage: usage, firstTokenMs: firstTokenMs}, ev.err
|
||||
}
|
||||
return &streamingResult{usage: usage, firstTokenMs: firstTokenMs}, fmt.Errorf("stream read error: %w", ev.err)
|
||||
}
|
||||
|
||||
line := ev.line
|
||||
if data, ok := extractAnthropicSSEDataLine(line); ok {
|
||||
trimmed := strings.TrimSpace(data)
|
||||
if anthropicStreamEventIsTerminal("", trimmed) {
|
||||
sawTerminalEvent = true
|
||||
}
|
||||
if firstTokenMs == nil && trimmed != "" && trimmed != "[DONE]" {
|
||||
ms := int(time.Since(startTime).Milliseconds())
|
||||
firstTokenMs = &ms
|
||||
}
|
||||
s.parseSSEUsagePassthrough(data, usage)
|
||||
} else {
|
||||
trimmed := strings.TrimSpace(line)
|
||||
if strings.HasPrefix(trimmed, "event:") && anthropicStreamEventIsTerminal(strings.TrimSpace(strings.TrimPrefix(trimmed, "event:")), "") {
|
||||
sawTerminalEvent = true
|
||||
}
|
||||
}
|
||||
|
||||
if !clientDisconnected {
|
||||
restored := string(reverseToolNamesIfPresent(c, []byte(line)))
|
||||
if _, err := io.WriteString(w, restored); err != nil {
|
||||
clientDisconnected = true
|
||||
logger.LegacyPrintf("service.gateway", "[Anthropic passthrough] Client disconnected during streaming, continue draining upstream for usage: account=%d", account.ID)
|
||||
} else if _, err := io.WriteString(w, "\n"); err != nil {
|
||||
clientDisconnected = true
|
||||
logger.LegacyPrintf("service.gateway", "[Anthropic passthrough] Client disconnected during streaming, continue draining upstream for usage: account=%d", account.ID)
|
||||
} else if line == "" {
|
||||
// 按 SSE 事件边界刷出，减少每行 flush 带来的 syscall 开销。
|
||||
flusher.Flush()
|
||||
lastDataAt = time.Now()
|
||||
resetKeepaliveTimer()
|
||||
inPartialEvent = false
|
||||
} else {
|
||||
inPartialEvent = true
|
||||
}
|
||||
}
|
||||
|
||||
case <-intervalCh:
|
||||
lastRead := time.Unix(0, atomic.LoadInt64(&lastReadAt))
|
||||
if time.Since(lastRead) < streamInterval {
|
||||
continue
|
||||
}
|
||||
if clientDisconnected {
|
||||
return &streamingResult{usage: usage, firstTokenMs: firstTokenMs, clientDisconnect: true}, fmt.Errorf("stream usage incomplete after timeout")
|
||||
}
|
||||
logger.LegacyPrintf("service.gateway", "[Anthropic passthrough] Stream data interval timeout: account=%d model=%s interval=%s", account.ID, model, streamInterval)
|
||||
if s.rateLimitService != nil {
|
||||
s.rateLimitService.HandleStreamTimeout(ctx, account, model)
|
||||
}
|
||||
return &streamingResult{usage: usage, firstTokenMs: firstTokenMs}, fmt.Errorf("stream data interval timeout")
|
||||
|
||||
case <-keepaliveCh:
|
||||
if clientDisconnected {
|
||||
continue
|
||||
}
|
||||
if inPartialEvent {
|
||||
resetKeepaliveTimer()
|
||||
continue
|
||||
}
|
||||
if time.Since(lastDataAt) < keepaliveInterval {
|
||||
resetKeepaliveTimer()
|
||||
continue
|
||||
}
|
||||
if _, err := fmt.Fprint(w, "event: ping\ndata: {\"type\": \"ping\"}\n\n"); err != nil {
|
||||
clientDisconnected = true
|
||||
logger.LegacyPrintf("service.gateway", "[Anthropic passthrough] Client disconnected during keepalive ping, continue draining upstream for usage: account=%d", account.ID)
|
||||
continue
|
||||
}
|
||||
flusher.Flush()
|
||||
lastDataAt = time.Now()
|
||||
resetKeepaliveTimer()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func extractAnthropicSSEDataLine(line string) (string, bool) {
|
||||
if !strings.HasPrefix(line, "data:") {
|
||||
return "", false
|
||||
}
|
||||
start := len("data:")
|
||||
for start < len(line) {
|
||||
if line[start] != ' ' && line[start] != '\t' {
|
||||
break
|
||||
}
|
||||
start++
|
||||
}
|
||||
return line[start:], true
|
||||
}
|
||||
|
||||
func (s *GatewayService) parseSSEUsagePassthrough(data string, usage *ClaudeUsage) {
|
||||
if usage == nil || data == "" || data == "[DONE]" {
|
||||
return
|
||||
}
|
||||
|
||||
parsed := gjson.Parse(data)
|
||||
switch parsed.Get("type").String() {
|
||||
case "message_start":
|
||||
msgUsage := parsed.Get("message.usage")
|
||||
if msgUsage.Exists() {
|
||||
usage.InputTokens = int(msgUsage.Get("input_tokens").Int())
|
||||
usage.CacheCreationInputTokens = int(msgUsage.Get("cache_creation_input_tokens").Int())
|
||||
usage.CacheReadInputTokens = int(msgUsage.Get("cache_read_input_tokens").Int())
|
||||
|
||||
// 保持与通用解析一致：message_start 允许覆盖 5m/1h 明细（包括 0）。
|
||||
cc5m := msgUsage.Get("cache_creation.ephemeral_5m_input_tokens")
|
||||
cc1h := msgUsage.Get("cache_creation.ephemeral_1h_input_tokens")
|
||||
if cc5m.Exists() || cc1h.Exists() {
|
||||
usage.CacheCreation5mTokens = int(cc5m.Int())
|
||||
usage.CacheCreation1hTokens = int(cc1h.Int())
|
||||
}
|
||||
}
|
||||
case "message_delta":
|
||||
deltaUsage := parsed.Get("usage")
|
||||
if deltaUsage.Exists() {
|
||||
if v := deltaUsage.Get("input_tokens").Int(); v > 0 {
|
||||
usage.InputTokens = int(v)
|
||||
}
|
||||
if v := deltaUsage.Get("output_tokens").Int(); v > 0 {
|
||||
usage.OutputTokens = int(v)
|
||||
}
|
||||
if v := deltaUsage.Get("cache_creation_input_tokens").Int(); v > 0 {
|
||||
usage.CacheCreationInputTokens = int(v)
|
||||
}
|
||||
if v := deltaUsage.Get("cache_read_input_tokens").Int(); v > 0 {
|
||||
usage.CacheReadInputTokens = int(v)
|
||||
}
|
||||
|
||||
cc5m := deltaUsage.Get("cache_creation.ephemeral_5m_input_tokens")
|
||||
cc1h := deltaUsage.Get("cache_creation.ephemeral_1h_input_tokens")
|
||||
if cc5m.Exists() && cc5m.Int() > 0 {
|
||||
usage.CacheCreation5mTokens = int(cc5m.Int())
|
||||
}
|
||||
if cc1h.Exists() && cc1h.Int() > 0 {
|
||||
usage.CacheCreation1hTokens = int(cc1h.Int())
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if usage.CacheReadInputTokens == 0 {
|
||||
if cached := parsed.Get("message.usage.cached_tokens").Int(); cached > 0 {
|
||||
usage.CacheReadInputTokens = int(cached)
|
||||
}
|
||||
if cached := parsed.Get("usage.cached_tokens").Int(); usage.CacheReadInputTokens == 0 && cached > 0 {
|
||||
usage.CacheReadInputTokens = int(cached)
|
||||
}
|
||||
}
|
||||
if usage.CacheCreationInputTokens == 0 {
|
||||
cc5m := parsed.Get("message.usage.cache_creation.ephemeral_5m_input_tokens").Int()
|
||||
cc1h := parsed.Get("message.usage.cache_creation.ephemeral_1h_input_tokens").Int()
|
||||
if cc5m == 0 && cc1h == 0 {
|
||||
cc5m = parsed.Get("usage.cache_creation.ephemeral_5m_input_tokens").Int()
|
||||
cc1h = parsed.Get("usage.cache_creation.ephemeral_1h_input_tokens").Int()
|
||||
}
|
||||
total := cc5m + cc1h
|
||||
if total > 0 {
|
||||
usage.CacheCreationInputTokens = int(total)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func parseClaudeUsageFromResponseBody(body []byte) *ClaudeUsage {
|
||||
usage := &ClaudeUsage{}
|
||||
if len(body) == 0 {
|
||||
return usage
|
||||
}
|
||||
|
||||
parsed := gjson.ParseBytes(body)
|
||||
usageNode := parsed.Get("usage")
|
||||
if !usageNode.Exists() {
|
||||
return usage
|
||||
}
|
||||
|
||||
usage.InputTokens = int(usageNode.Get("input_tokens").Int())
|
||||
usage.OutputTokens = int(usageNode.Get("output_tokens").Int())
|
||||
usage.CacheCreationInputTokens = int(usageNode.Get("cache_creation_input_tokens").Int())
|
||||
usage.CacheReadInputTokens = int(usageNode.Get("cache_read_input_tokens").Int())
|
||||
|
||||
cc5m := usageNode.Get("cache_creation.ephemeral_5m_input_tokens").Int()
|
||||
cc1h := usageNode.Get("cache_creation.ephemeral_1h_input_tokens").Int()
|
||||
if cc5m > 0 || cc1h > 0 {
|
||||
usage.CacheCreation5mTokens = int(cc5m)
|
||||
usage.CacheCreation1hTokens = int(cc1h)
|
||||
}
|
||||
if usage.CacheCreationInputTokens == 0 && (cc5m > 0 || cc1h > 0) {
|
||||
usage.CacheCreationInputTokens = int(cc5m + cc1h)
|
||||
}
|
||||
if usage.CacheReadInputTokens == 0 {
|
||||
if cached := usageNode.Get("cached_tokens").Int(); cached > 0 {
|
||||
usage.CacheReadInputTokens = int(cached)
|
||||
}
|
||||
}
|
||||
return usage
|
||||
}
|
||||
|
||||
func (s *GatewayService) invalidNonStreamingJSONFailoverError(
|
||||
ctx context.Context,
|
||||
resp *http.Response,
|
||||
account *Account,
|
||||
body []byte,
|
||||
parseErr error,
|
||||
requestedModel ...string,
|
||||
) error {
|
||||
const statusCode = http.StatusBadGateway
|
||||
|
||||
accountID := int64(0)
|
||||
accountName := ""
|
||||
retryableOnSameAccount := false
|
||||
if account != nil {
|
||||
accountID = account.ID
|
||||
accountName = account.Name
|
||||
retryableOnSameAccount = account.IsPoolMode() && account.IsPoolModeRetryableStatus(statusCode)
|
||||
}
|
||||
|
||||
logger.LegacyPrintf(
|
||||
"service.gateway",
|
||||
"Account %d(%s): upstream returned non-JSON 2xx response, attempting failover: status=%d request_id=%s error=%v",
|
||||
accountID,
|
||||
accountName,
|
||||
resp.StatusCode,
|
||||
resp.Header.Get("x-request-id"),
|
||||
parseErr,
|
||||
)
|
||||
|
||||
if s.rateLimitService != nil && account != nil {
|
||||
if len(requestedModel) > 0 {
|
||||
s.rateLimitService.HandleUpstreamError(ctx, account, statusCode, resp.Header, body, requestedModel[0])
|
||||
} else {
|
||||
s.rateLimitService.HandleUpstreamError(ctx, account, statusCode, resp.Header, body)
|
||||
}
|
||||
}
|
||||
|
||||
return &UpstreamFailoverError{
|
||||
StatusCode: statusCode,
|
||||
ResponseBody: body,
|
||||
ResponseHeaders: resp.Header,
|
||||
RetryableOnSameAccount: retryableOnSameAccount,
|
||||
}
|
||||
}
|
||||
|
||||
func (s *GatewayService) handleNonStreamingResponseAnthropicAPIKeyPassthrough(
|
||||
ctx context.Context,
|
||||
resp *http.Response,
|
||||
c *gin.Context,
|
||||
account *Account,
|
||||
) (*ClaudeUsage, error) {
|
||||
if s.rateLimitService != nil {
|
||||
s.rateLimitService.UpdateSessionWindow(ctx, account, resp.Header)
|
||||
}
|
||||
|
||||
body, err := ReadUpstreamResponseBody(resp.Body, s.cfg, c, anthropicTooLargeError)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if resp.StatusCode >= http.StatusOK && resp.StatusCode < http.StatusMultipleChoices {
|
||||
var raw json.RawMessage
|
||||
if err := json.Unmarshal(body, &raw); err != nil {
|
||||
return nil, s.invalidNonStreamingJSONFailoverError(ctx, resp, account, body, err)
|
||||
}
|
||||
}
|
||||
|
||||
usage := parseClaudeUsageFromResponseBody(body)
|
||||
|
||||
writeAnthropicPassthroughResponseHeaders(c.Writer.Header(), resp.Header, s.responseHeaderFilter)
|
||||
contentType := strings.TrimSpace(resp.Header.Get("Content-Type"))
|
||||
if contentType == "" {
|
||||
contentType = "application/json"
|
||||
}
|
||||
body = reverseToolNamesIfPresent(c, body)
|
||||
c.Data(resp.StatusCode, contentType, body)
|
||||
return usage, nil
|
||||
}
|
||||
|
||||
func writeAnthropicPassthroughResponseHeaders(dst http.Header, src http.Header, filter *responseheaders.CompiledHeaderFilter) {
|
||||
if dst == nil || src == nil {
|
||||
return
|
||||
}
|
||||
if filter != nil {
|
||||
responseheaders.WriteFilteredHeaders(dst, src, filter)
|
||||
return
|
||||
}
|
||||
if v := strings.TrimSpace(src.Get("Content-Type")); v != "" {
|
||||
dst.Set("Content-Type", v)
|
||||
}
|
||||
if v := strings.TrimSpace(src.Get("x-request-id")); v != "" {
|
||||
dst.Set("x-request-id", v)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,412 @@
|
||||
package service
|
||||
|
||||
// 本文件由 gateway_service.go 纯移动拆分而来：Bedrock 上游转发（CC 兼容转换、
|
||||
// 请求构建、错误处理与非流式响应）。仅做代码搬迁，无任何行为变更。
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/Wei-Shaw/sub2api/internal/pkg/logger"
|
||||
|
||||
"github.com/gin-gonic/gin"
|
||||
)
|
||||
|
||||
// ApplyBedrockCCCompat 应用 Bedrock CC 兼容转换（渠道级模型映射后调用）
|
||||
// 清理 body 中 Anthropic API 专有字段、修复 thinking/tool_use ID、过滤 beta token，
|
||||
// 同时过滤 HTTP header 中的 anthropic-beta（防止 Passthrough 路径透传不支持的 token）。
|
||||
func (s *GatewayService) ApplyBedrockCCCompat(c *gin.Context, body []byte, model string, account *Account, groupID *int64) []byte {
|
||||
if !s.isBedrockCCCompatEnabled(c.Request.Context(), account, groupID) {
|
||||
return body
|
||||
}
|
||||
body = sanitizeBedrockCCFields(body)
|
||||
body = sanitizeBedrockThinking(body, model)
|
||||
body = sanitizeBedrockToolUseIDs(body)
|
||||
body = sanitizeBedrockCCBetaTokens(body, model)
|
||||
// 过滤 HTTP header 中的 anthropic-beta，只保留 Bedrock 支持的 token
|
||||
if betaHeader := c.GetHeader("anthropic-beta"); betaHeader != "" {
|
||||
if filtered := ResolveBedrockBetaTokens(betaHeader, body, model); len(filtered) > 0 {
|
||||
c.Request.Header.Set("anthropic-beta", strings.Join(filtered, ", "))
|
||||
} else {
|
||||
c.Request.Header.Del("anthropic-beta")
|
||||
}
|
||||
}
|
||||
return body
|
||||
}
|
||||
|
||||
// isBedrockCCCompatEnabled 检查渠道是否启用了 Bedrock CC 兼容模式
|
||||
func (s *GatewayService) isBedrockCCCompatEnabled(ctx context.Context, account *Account, groupID *int64) bool {
|
||||
if groupID == nil || s.channelService == nil {
|
||||
return false
|
||||
}
|
||||
ch, err := s.channelService.GetChannelForGroup(ctx, *groupID)
|
||||
if err != nil || ch == nil {
|
||||
return false
|
||||
}
|
||||
return ch.IsBedrockCCCompatEnabled(account.Platform)
|
||||
}
|
||||
|
||||
// forwardBedrock 转发请求到 AWS Bedrock
|
||||
func (s *GatewayService) forwardBedrock(
|
||||
ctx context.Context,
|
||||
c *gin.Context,
|
||||
account *Account,
|
||||
parsed *ParsedRequest,
|
||||
startTime time.Time,
|
||||
) (*ForwardResult, error) {
|
||||
reqModel := parsed.Model
|
||||
reqStream := parsed.Stream
|
||||
body := parsed.Body.Bytes()
|
||||
|
||||
region := bedrockRuntimeRegion(account)
|
||||
mappedModel, ok := ResolveBedrockModelID(account, reqModel)
|
||||
if !ok {
|
||||
return nil, fmt.Errorf("unsupported bedrock model: %s", reqModel)
|
||||
}
|
||||
if mappedModel != reqModel {
|
||||
logger.LegacyPrintf("service.gateway", "[Bedrock] Model mapping: %s -> %s (account: %s)", reqModel, mappedModel, account.Name)
|
||||
}
|
||||
|
||||
betaHeader := ""
|
||||
if c != nil && c.Request != nil {
|
||||
betaHeader = c.GetHeader("anthropic-beta")
|
||||
}
|
||||
|
||||
// 准备请求体（注入 anthropic_version/anthropic_beta，移除 Bedrock 不支持的字段，清理 cache_control）
|
||||
betaTokens, err := s.resolveBedrockBetaTokensForRequest(ctx, account, betaHeader, body, mappedModel)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
bedrockBody, err := PrepareBedrockRequestBodyWithTokens(body, mappedModel, betaTokens, false)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("prepare bedrock request body: %w", err)
|
||||
}
|
||||
|
||||
proxyURL := ""
|
||||
if account.ProxyID != nil && account.Proxy != nil {
|
||||
proxyURL = account.Proxy.URL()
|
||||
}
|
||||
|
||||
logger.LegacyPrintf("service.gateway", "[Bedrock] 命中 Bedrock 分支: account=%d name=%s model=%s->%s stream=%v",
|
||||
account.ID, account.Name, reqModel, mappedModel, reqStream)
|
||||
|
||||
// 根据账号类型选择认证方式
|
||||
var signer *BedrockSigner
|
||||
var bedrockAPIKey string
|
||||
if account.IsBedrockAPIKey() {
|
||||
bedrockAPIKey = account.GetCredential("api_key")
|
||||
if bedrockAPIKey == "" {
|
||||
return nil, fmt.Errorf("api_key not found in bedrock credentials")
|
||||
}
|
||||
} else {
|
||||
signer, err = NewBedrockSignerFromAccount(account)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("create bedrock signer: %w", err)
|
||||
}
|
||||
}
|
||||
|
||||
// 执行上游请求（含重试）
|
||||
resp, err := s.executeBedrockUpstream(ctx, c, account, bedrockBody, mappedModel, region, reqStream, signer, bedrockAPIKey, proxyURL)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer func() { _ = resp.Body.Close() }()
|
||||
|
||||
// 将 Bedrock 的 x-amzn-requestid 映射到 x-request-id，
|
||||
// 使通用错误处理函数（handleErrorResponse、handleRetryExhaustedError）能正确提取 AWS request ID。
|
||||
if awsReqID := resp.Header.Get("x-amzn-requestid"); awsReqID != "" && resp.Header.Get("x-request-id") == "" {
|
||||
resp.Header.Set("x-request-id", awsReqID)
|
||||
}
|
||||
|
||||
// 错误/failover 处理
|
||||
if resp.StatusCode >= 400 {
|
||||
return s.handleBedrockUpstreamErrors(ctx, resp, c, account)
|
||||
}
|
||||
|
||||
// Bedrock 分支绕过通用 Forward 成功路径，这里保持上游接受回调语义一致。
|
||||
if parsed.OnUpstreamAccepted != nil {
|
||||
parsed.OnUpstreamAccepted()
|
||||
}
|
||||
|
||||
// 响应处理
|
||||
var usage *ClaudeUsage
|
||||
var firstTokenMs *int
|
||||
var clientDisconnect bool
|
||||
if reqStream {
|
||||
streamResult, err := s.handleBedrockStreamingResponse(ctx, resp, c, account, startTime, reqModel)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
usage = streamResult.usage
|
||||
firstTokenMs = streamResult.firstTokenMs
|
||||
clientDisconnect = streamResult.clientDisconnect
|
||||
} else {
|
||||
usage, err = s.handleBedrockNonStreamingResponse(ctx, resp, c, account)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
if usage == nil {
|
||||
usage = &ClaudeUsage{}
|
||||
}
|
||||
|
||||
return &ForwardResult{
|
||||
RequestID: resp.Header.Get("x-amzn-requestid"),
|
||||
Usage: *usage,
|
||||
Model: reqModel,
|
||||
UpstreamModel: mappedModel,
|
||||
Stream: reqStream,
|
||||
Duration: time.Since(startTime),
|
||||
FirstTokenMs: firstTokenMs,
|
||||
ClientDisconnect: clientDisconnect,
|
||||
}, nil
|
||||
}
|
||||
|
||||
// executeBedrockUpstream 执行 Bedrock 上游请求（含重试逻辑）
|
||||
func (s *GatewayService) executeBedrockUpstream(
|
||||
ctx context.Context,
|
||||
c *gin.Context,
|
||||
account *Account,
|
||||
body []byte,
|
||||
modelID string,
|
||||
region string,
|
||||
stream bool,
|
||||
signer *BedrockSigner,
|
||||
apiKey string,
|
||||
proxyURL string,
|
||||
) (*http.Response, error) {
|
||||
var resp *http.Response
|
||||
var err error
|
||||
retryStart := time.Now()
|
||||
for attempt := 1; attempt <= maxRetryAttempts; attempt++ {
|
||||
var upstreamReq *http.Request
|
||||
if account.IsBedrockAPIKey() {
|
||||
upstreamReq, err = s.buildUpstreamRequestBedrockAPIKey(ctx, body, modelID, region, stream, apiKey)
|
||||
} else {
|
||||
upstreamReq, err = s.buildUpstreamRequestBedrock(ctx, body, modelID, region, stream, signer)
|
||||
}
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
resp, err = s.httpUpstream.DoWithTLS(upstreamReq, proxyURL, account.ID, account.Concurrency, nil)
|
||||
if err != nil {
|
||||
if resp != nil && resp.Body != nil {
|
||||
_ = resp.Body.Close()
|
||||
}
|
||||
safeErr := sanitizeUpstreamErrorMessage(err.Error())
|
||||
setOpsUpstreamError(c, 0, safeErr, "")
|
||||
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
|
||||
Platform: account.Platform,
|
||||
AccountID: account.ID,
|
||||
AccountName: account.Name,
|
||||
UpstreamStatusCode: 0,
|
||||
UpstreamURL: safeUpstreamURL(upstreamReq.URL.String()),
|
||||
Kind: "request_error",
|
||||
Message: safeErr,
|
||||
})
|
||||
c.JSON(http.StatusBadGateway, gin.H{
|
||||
"type": "error",
|
||||
"error": gin.H{
|
||||
"type": "upstream_error",
|
||||
"message": "Upstream request failed",
|
||||
},
|
||||
})
|
||||
return nil, fmt.Errorf("upstream request failed: %s", safeErr)
|
||||
}
|
||||
|
||||
if resp.StatusCode >= 400 && resp.StatusCode != 400 && s.shouldRetryUpstreamError(account, resp.StatusCode) {
|
||||
if attempt < maxRetryAttempts {
|
||||
elapsed := time.Since(retryStart)
|
||||
if elapsed >= maxRetryElapsed {
|
||||
break
|
||||
}
|
||||
|
||||
delay := retryBackoffDelay(attempt)
|
||||
remaining := maxRetryElapsed - elapsed
|
||||
if delay > remaining {
|
||||
delay = remaining
|
||||
}
|
||||
if delay <= 0 {
|
||||
break
|
||||
}
|
||||
|
||||
respBody, _ := s.readUpstreamErrorBody(resp)
|
||||
_ = resp.Body.Close()
|
||||
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
|
||||
Platform: account.Platform,
|
||||
AccountID: account.ID,
|
||||
AccountName: account.Name,
|
||||
UpstreamStatusCode: resp.StatusCode,
|
||||
UpstreamURL: safeUpstreamURL(upstreamReq.URL.String()),
|
||||
Kind: "retry",
|
||||
Message: extractUpstreamErrorMessage(respBody),
|
||||
Detail: func() string {
|
||||
if s.cfg != nil && s.cfg.Gateway.LogUpstreamErrorBody {
|
||||
return truncateString(string(respBody), s.cfg.Gateway.LogUpstreamErrorBodyMaxBytes)
|
||||
}
|
||||
return ""
|
||||
}(),
|
||||
})
|
||||
logger.LegacyPrintf("service.gateway", "[Bedrock] account %d: upstream error %d, retry %d/%d after %v",
|
||||
account.ID, resp.StatusCode, attempt, maxRetryAttempts, delay)
|
||||
if err := sleepWithContext(ctx, delay); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
continue
|
||||
}
|
||||
break
|
||||
}
|
||||
|
||||
break
|
||||
}
|
||||
if resp == nil || resp.Body == nil {
|
||||
return nil, errors.New("upstream request failed: empty response")
|
||||
}
|
||||
return resp, nil
|
||||
}
|
||||
|
||||
// handleBedrockUpstreamErrors 处理 Bedrock 上游 4xx/5xx 错误（failover + 错误响应）
|
||||
func (s *GatewayService) handleBedrockUpstreamErrors(
|
||||
ctx context.Context,
|
||||
resp *http.Response,
|
||||
c *gin.Context,
|
||||
account *Account,
|
||||
) (*ForwardResult, error) {
|
||||
// retry exhausted + failover
|
||||
if s.shouldRetryUpstreamError(account, resp.StatusCode) {
|
||||
if s.shouldFailoverUpstreamError(resp.StatusCode) {
|
||||
respBody, _ := s.readUpstreamErrorBody(resp)
|
||||
_ = resp.Body.Close()
|
||||
resp.Body = io.NopCloser(bytes.NewReader(respBody))
|
||||
|
||||
logger.LegacyPrintf("service.gateway", "[Bedrock] Upstream error (retry exhausted, failover): Account=%d(%s) Status=%d Body=%s",
|
||||
account.ID, account.Name, resp.StatusCode, truncateString(string(respBody), 1000))
|
||||
|
||||
s.handleRetryExhaustedSideEffects(ctx, resp, account)
|
||||
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
|
||||
Platform: account.Platform,
|
||||
AccountID: account.ID,
|
||||
AccountName: account.Name,
|
||||
UpstreamStatusCode: resp.StatusCode,
|
||||
Kind: "retry_exhausted_failover",
|
||||
Message: extractUpstreamErrorMessage(respBody),
|
||||
})
|
||||
return nil, &UpstreamFailoverError{
|
||||
StatusCode: resp.StatusCode,
|
||||
ResponseBody: respBody,
|
||||
RetryableOnSameAccount: account.IsPoolMode() && account.IsPoolModeRetryableStatus(resp.StatusCode),
|
||||
}
|
||||
}
|
||||
return s.handleRetryExhaustedError(ctx, resp, c, account)
|
||||
}
|
||||
|
||||
// non-retryable failover
|
||||
if s.shouldFailoverUpstreamError(resp.StatusCode) {
|
||||
respBody, _ := s.readUpstreamErrorBody(resp)
|
||||
_ = resp.Body.Close()
|
||||
resp.Body = io.NopCloser(bytes.NewReader(respBody))
|
||||
|
||||
s.handleFailoverSideEffects(ctx, resp, account)
|
||||
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
|
||||
Platform: account.Platform,
|
||||
AccountID: account.ID,
|
||||
AccountName: account.Name,
|
||||
UpstreamStatusCode: resp.StatusCode,
|
||||
Kind: "failover",
|
||||
Message: extractUpstreamErrorMessage(respBody),
|
||||
})
|
||||
return nil, &UpstreamFailoverError{
|
||||
StatusCode: resp.StatusCode,
|
||||
ResponseBody: respBody,
|
||||
RetryableOnSameAccount: account.IsPoolMode() && account.IsPoolModeRetryableStatus(resp.StatusCode),
|
||||
}
|
||||
}
|
||||
|
||||
// other errors
|
||||
return s.handleErrorResponse(ctx, resp, c, account)
|
||||
}
|
||||
|
||||
// buildUpstreamRequestBedrock 构建 Bedrock 上游请求
|
||||
func (s *GatewayService) buildUpstreamRequestBedrock(
|
||||
ctx context.Context,
|
||||
body []byte,
|
||||
modelID string,
|
||||
region string,
|
||||
stream bool,
|
||||
signer *BedrockSigner,
|
||||
) (*http.Request, error) {
|
||||
targetURL := BuildBedrockURL(region, modelID, stream)
|
||||
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodPost, targetURL, bytes.NewReader(body))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
req.Header.Set("Accept", "application/json")
|
||||
|
||||
// SigV4 签名
|
||||
if err := signer.SignRequest(ctx, req, body); err != nil {
|
||||
return nil, fmt.Errorf("sign bedrock request: %w", err)
|
||||
}
|
||||
|
||||
return req, nil
|
||||
}
|
||||
|
||||
// buildUpstreamRequestBedrockAPIKey 构建 Bedrock API Key (Bearer Token) 上游请求
|
||||
func (s *GatewayService) buildUpstreamRequestBedrockAPIKey(
|
||||
ctx context.Context,
|
||||
body []byte,
|
||||
modelID string,
|
||||
region string,
|
||||
stream bool,
|
||||
apiKey string,
|
||||
) (*http.Request, error) {
|
||||
targetURL := BuildBedrockURL(region, modelID, stream)
|
||||
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodPost, targetURL, bytes.NewReader(body))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
req.Header.Set("Accept", "application/json")
|
||||
req.Header.Set("Authorization", "Bearer "+apiKey)
|
||||
|
||||
return req, nil
|
||||
}
|
||||
|
||||
// handleBedrockNonStreamingResponse 处理 Bedrock 非流式响应
|
||||
// Bedrock InvokeModel 非流式响应的 body 格式与 Claude API 兼容
|
||||
func (s *GatewayService) handleBedrockNonStreamingResponse(
|
||||
ctx context.Context,
|
||||
resp *http.Response,
|
||||
c *gin.Context,
|
||||
account *Account,
|
||||
) (*ClaudeUsage, error) {
|
||||
body, err := ReadUpstreamResponseBody(resp.Body, s.cfg, c, anthropicTooLargeError)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// 转换 Bedrock 特有的 amazon-bedrock-invocationMetrics 为标准 Anthropic usage 格式
|
||||
// 并移除该字段避免透传给客户端
|
||||
body = transformBedrockInvocationMetrics(body)
|
||||
|
||||
usage := parseClaudeUsageFromResponseBody(body)
|
||||
|
||||
c.Header("Content-Type", "application/json")
|
||||
if v := resp.Header.Get("x-amzn-requestid"); v != "" {
|
||||
c.Header("x-request-id", v)
|
||||
}
|
||||
c.Data(resp.StatusCode, "application/json", body)
|
||||
return usage, nil
|
||||
}
|
||||
File diff suppressed because it is too large
Load Diff
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,324 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/Wei-Shaw/sub2api/internal/pkg/apicompat"
|
||||
"github.com/Wei-Shaw/sub2api/internal/pkg/logger"
|
||||
"github.com/Wei-Shaw/sub2api/internal/util/responseheaders"
|
||||
"github.com/gin-gonic/gin"
|
||||
"go.uber.org/zap"
|
||||
)
|
||||
|
||||
// 本文件收敛三个 CC（Chat Completions）forwarder 之间重复的 HTTP 管线与 SSE
|
||||
// 循环骨架（PR #3802 遗留项）：
|
||||
//
|
||||
// - forwardAsRawChatCompletions （原生 CC 直转）
|
||||
// - forwardResponsesViaRawChatCompletions（/v1/responses → CC 回退）
|
||||
// - forwardAnthropicViaRawChatCompletions（/v1/messages → CC 回退）
|
||||
//
|
||||
// 以及 messages / chat_completions 两条 Responses 主路径中逐字相同的错误处理块。
|
||||
// 所有 helper 都是对既有内联代码的等价提取，不改变任何行为；各路径的差异
|
||||
// （GLM effort 归一化、fast policy、Grok 分支、ClientDisconnect 语义等）仍留在
|
||||
// 调用方，属于有意保留的行为差异，不在此强行统一。
|
||||
|
||||
// newUpstreamSSEScanner 构造读取上游 SSE 流的行扫描器，按配置放大单行上限。
|
||||
func (s *OpenAIGatewayService) newUpstreamSSEScanner(r io.Reader) *bufio.Scanner {
|
||||
scanner := bufio.NewScanner(r)
|
||||
maxLineSize := defaultMaxLineSize
|
||||
if s.cfg != nil && s.cfg.Gateway.MaxLineSize > 0 {
|
||||
maxLineSize = s.cfg.Gateway.MaxLineSize
|
||||
}
|
||||
scanner.Buffer(make([]byte, 0, 64*1024), maxLineSize)
|
||||
return scanner
|
||||
}
|
||||
|
||||
// newStreamHeaderWriter 返回幂等的 SSE 响应头写入闭包：首次调用时透传过滤后的
|
||||
// 上游响应头并写入标准 SSE 头 + 200 状态码，后续调用为 no-op。延迟到首个事件
|
||||
// 写出前才提交响应头，使上游早期失败仍可改走 failover 或非流式错误响应。
|
||||
func (s *OpenAIGatewayService) newStreamHeaderWriter(c *gin.Context, upstream http.Header) func() {
|
||||
headersWritten := false
|
||||
return func() {
|
||||
if headersWritten {
|
||||
return
|
||||
}
|
||||
headersWritten = true
|
||||
if s.responseHeaderFilter != nil {
|
||||
responseheaders.WriteFilteredHeaders(c.Writer.Header(), upstream, s.responseHeaderFilter)
|
||||
}
|
||||
c.Writer.Header().Set("Content-Type", "text/event-stream")
|
||||
c.Writer.Header().Set("Cache-Control", "no-cache")
|
||||
c.Writer.Header().Set("Connection", "keep-alive")
|
||||
c.Writer.Header().Set("X-Accel-Buffering", "no")
|
||||
c.Writer.WriteHeader(http.StatusOK)
|
||||
}
|
||||
}
|
||||
|
||||
// readOpenAIUpstreamError 读取上游错误体并把 resp.Body 回卷为可重读的副本
|
||||
// （下游 handleXxxErrorResponse 需要再次读取），返回原始错误体与脱敏后的
|
||||
// 上游错误消息。
|
||||
func (s *OpenAIGatewayService) readOpenAIUpstreamError(resp *http.Response) ([]byte, string) {
|
||||
respBody := s.readUpstreamErrorBody(resp)
|
||||
_ = resp.Body.Close()
|
||||
resp.Body = io.NopCloser(bytes.NewReader(respBody))
|
||||
|
||||
upstreamMsg := strings.TrimSpace(extractUpstreamErrorMessage(respBody))
|
||||
upstreamMsg = sanitizeUpstreamErrorMessage(upstreamMsg)
|
||||
return respBody, upstreamMsg
|
||||
}
|
||||
|
||||
// failoverOpenAIUpstreamHTTPError 对 >=400 的上游响应做 failover 判定：命中时
|
||||
// 记录 ops 事件、执行账号级错误处置并返回 *UpstreamFailoverError；未命中返回
|
||||
// nil，调用方继续走各自端点格式的非 failover 错误处理链。
|
||||
func (s *OpenAIGatewayService) failoverOpenAIUpstreamHTTPError(
|
||||
ctx context.Context,
|
||||
c *gin.Context,
|
||||
account *Account,
|
||||
resp *http.Response,
|
||||
respBody []byte,
|
||||
upstreamMsg string,
|
||||
upstreamModel string,
|
||||
) *UpstreamFailoverError {
|
||||
if !s.shouldFailoverOpenAIUpstreamResponse(resp.StatusCode, upstreamMsg, respBody) {
|
||||
return nil
|
||||
}
|
||||
upstreamDetail := ""
|
||||
if s.cfg != nil && s.cfg.Gateway.LogUpstreamErrorBody {
|
||||
maxBytes := s.cfg.Gateway.LogUpstreamErrorBodyMaxBytes
|
||||
if maxBytes <= 0 {
|
||||
maxBytes = 2048
|
||||
}
|
||||
upstreamDetail = truncateString(string(respBody), maxBytes)
|
||||
}
|
||||
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
|
||||
Platform: account.Platform,
|
||||
AccountID: account.ID,
|
||||
AccountName: account.Name,
|
||||
UpstreamStatusCode: resp.StatusCode,
|
||||
UpstreamRequestID: resp.Header.Get("x-request-id"),
|
||||
Kind: "failover",
|
||||
Message: upstreamMsg,
|
||||
Detail: upstreamDetail,
|
||||
})
|
||||
s.handleOpenAIAccountUpstreamError(ctx, account, resp.StatusCode, resp.Header, respBody, upstreamModel)
|
||||
return &UpstreamFailoverError{
|
||||
StatusCode: resp.StatusCode,
|
||||
ResponseBody: respBody,
|
||||
RetryableOnSameAccount: account.IsPoolMode() && (account.IsPoolModeRetryableStatus(resp.StatusCode) || isOpenAITransientProcessingError(resp.StatusCode, upstreamMsg, respBody)),
|
||||
}
|
||||
}
|
||||
|
||||
// openAIChatCompletionsTargetURL 解析账号的（非 Grok）Chat Completions 上游端点。
|
||||
func (s *OpenAIGatewayService) openAIChatCompletionsTargetURL(account *Account) (string, error) {
|
||||
baseURL := account.GetOpenAIBaseURL()
|
||||
if baseURL == "" {
|
||||
baseURL = "https://api.openai.com"
|
||||
}
|
||||
validatedURL, err := s.validateUpstreamBaseURL(baseURL)
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("invalid base_url: %w", err)
|
||||
}
|
||||
return buildOpenAIChatCompletionsURL(validatedURL), nil
|
||||
}
|
||||
|
||||
// resolveCCFallbackTarget 解析两条 CC 回退路径共用的账号凭证与上游端点
|
||||
// （回退路径仅面向 APIKey 账号，凭证恒为 openai api_key）。
|
||||
func (s *OpenAIGatewayService) resolveCCFallbackTarget(account *Account) (apiKey string, targetURL string, err error) {
|
||||
apiKey = account.GetOpenAIApiKey()
|
||||
if apiKey == "" {
|
||||
return "", "", fmt.Errorf("account %d missing api_key", account.ID)
|
||||
}
|
||||
targetURL, err = s.openAIChatCompletionsTargetURL(account)
|
||||
if err != nil {
|
||||
return "", "", err
|
||||
}
|
||||
return apiKey, targetURL, nil
|
||||
}
|
||||
|
||||
// sendCCUpstreamRequest 构建并发送 CC 上游请求：分离的上游 context、OpenAI HTTP
|
||||
// profile、标准头（含流式 Accept 切换）、客户端 header 白名单透传、自定义 UA 与
|
||||
// 账号级 header 覆写，最后经代理发出。传输层失败（DNS/TCP/TLS，无 HTTP 响应）
|
||||
// 统一由 handleOpenAIUpstreamTransportError 归一为 failover。
|
||||
//
|
||||
// userAgent 为空时保留默认 UA；Grok 的默认 UA 兜底由调用方解析后传入。
|
||||
func (s *OpenAIGatewayService) sendCCUpstreamRequest(
|
||||
ctx context.Context,
|
||||
c *gin.Context,
|
||||
account *Account,
|
||||
targetURL string,
|
||||
body []byte,
|
||||
stream bool,
|
||||
bearerToken string,
|
||||
userAgent string,
|
||||
) (*http.Response, error) {
|
||||
upstreamCtx, releaseUpstreamCtx := detachUpstreamContext(ctx)
|
||||
upstreamReq, err := http.NewRequestWithContext(upstreamCtx, http.MethodPost, targetURL, bytes.NewReader(body))
|
||||
releaseUpstreamCtx()
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("build upstream request: %w", err)
|
||||
}
|
||||
upstreamReq = upstreamReq.WithContext(WithHTTPUpstreamProfile(upstreamReq.Context(), HTTPUpstreamProfileOpenAI))
|
||||
upstreamReq.Header.Set("Content-Type", "application/json")
|
||||
upstreamReq.Header.Set("Authorization", "Bearer "+bearerToken)
|
||||
if stream {
|
||||
upstreamReq.Header.Set("Accept", "text/event-stream")
|
||||
} else {
|
||||
upstreamReq.Header.Set("Accept", "application/json")
|
||||
}
|
||||
|
||||
// 透传白名单中的客户端 header。详见 openaiCCRawAllowedHeaders 的设计说明。
|
||||
for key, values := range c.Request.Header {
|
||||
lowerKey := strings.ToLower(key)
|
||||
if openaiCCRawAllowedHeaders[lowerKey] {
|
||||
for _, v := range values {
|
||||
upstreamReq.Header.Add(key, v)
|
||||
}
|
||||
}
|
||||
}
|
||||
if userAgent != "" {
|
||||
upstreamReq.Header.Set("user-agent", userAgent)
|
||||
}
|
||||
|
||||
// 账号级请求头覆写（仅 openai api_key 账号启用时生效）
|
||||
account.ApplyHeaderOverrides(upstreamReq.Header)
|
||||
|
||||
proxyURL := ""
|
||||
if account.Proxy != nil {
|
||||
proxyURL = account.Proxy.URL()
|
||||
}
|
||||
resp, err := s.httpUpstream.Do(upstreamReq, proxyURL, account.ID, account.Concurrency)
|
||||
if err != nil {
|
||||
return nil, s.handleOpenAIUpstreamTransportError(ctx, c, account, err, false)
|
||||
}
|
||||
return resp, nil
|
||||
}
|
||||
|
||||
// ccStreamScanState 是 scanCCStream 返回的读取状态快照。
|
||||
type ccStreamScanState struct {
|
||||
// Usage 为 include_usage chunk 中最近一次出现的用量（上游可能重复发送，
|
||||
// 总是保留最新值）；终态事件中的用量由调用方在 finalize 阶段自行覆盖。
|
||||
Usage OpenAIUsage
|
||||
// FirstTokenMs 为首个实际输出 chunk（排除 usage-only chunk）的到达时延。
|
||||
FirstTokenMs *int
|
||||
// SawDone 表示上游发出了 [DONE] 哨兵。
|
||||
SawDone bool
|
||||
// Err 为 scanner 读错误（客户端 context 取消不属于此类，会原样带出）。
|
||||
// 非 nil 时调用方必须跳过 finalize 并返回 usage-incomplete 错误，避免
|
||||
// 把上游截断伪装成正常收尾。
|
||||
Err error
|
||||
}
|
||||
|
||||
// scanCCStream 驱动两条 CC 回退路径共享的 SSE 读循环：提取 data 行、在 [DONE]
|
||||
// 哨兵处停止、保留最新 usage、记录首 token 时延，并把每个解析成功的 chunk 交给
|
||||
// emit 回调做各自的协议转换与写出。读错误按既有约定过滤 context 取消类噪声后
|
||||
// 记入 Warn 日志。
|
||||
func (s *OpenAIGatewayService) scanCCStream(
|
||||
resp *http.Response,
|
||||
logPrefix string,
|
||||
requestID string,
|
||||
startTime time.Time,
|
||||
emit func(*apicompat.ChatCompletionsChunk),
|
||||
) ccStreamScanState {
|
||||
var st ccStreamScanState
|
||||
|
||||
scanner := s.newUpstreamSSEScanner(resp.Body)
|
||||
for scanner.Scan() {
|
||||
line := scanner.Text()
|
||||
payload, ok := extractOpenAISSEDataLine(line)
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
payload = strings.TrimSpace(payload)
|
||||
if payload == "" {
|
||||
continue
|
||||
}
|
||||
if payload == "[DONE]" {
|
||||
st.SawDone = true
|
||||
break
|
||||
}
|
||||
|
||||
if u := extractCCStreamUsage(payload); u != nil {
|
||||
st.Usage = *u
|
||||
}
|
||||
|
||||
var chunk apicompat.ChatCompletionsChunk
|
||||
if err := json.Unmarshal([]byte(payload), &chunk); err != nil {
|
||||
logger.L().Warn(logPrefix+": failed to parse chat stream chunk",
|
||||
zap.Error(err),
|
||||
zap.String("request_id", requestID),
|
||||
)
|
||||
continue
|
||||
}
|
||||
if st.FirstTokenMs == nil && !isOpenAIChatUsageOnlyStreamChunk(payload) && chatChunkStartsResponsesOutput(&chunk) {
|
||||
ms := int(time.Since(startTime).Milliseconds())
|
||||
st.FirstTokenMs = &ms
|
||||
}
|
||||
emit(&chunk)
|
||||
}
|
||||
|
||||
if err := scanner.Err(); err != nil {
|
||||
if !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded) {
|
||||
logger.L().Warn(logPrefix+": stream read error",
|
||||
zap.Error(err),
|
||||
zap.String("request_id", requestID),
|
||||
)
|
||||
}
|
||||
st.Err = err
|
||||
}
|
||||
return st
|
||||
}
|
||||
|
||||
// logCCStreamMissingDoneSentinel 记录"上游未发 [DONE] 哨兵即结束"的 debug 日志。
|
||||
func logCCStreamMissingDoneSentinel(logPrefix, requestID string) {
|
||||
logger.L().Debug(logPrefix+": upstream stream ended without done sentinel",
|
||||
zap.String("request_id", requestID),
|
||||
)
|
||||
}
|
||||
|
||||
// readCCUpstreamJSONResponse 读取并解析 CC 非流式 JSON 响应，失败时以调用方
|
||||
// 端点格式回写错误；成功时顺带提取 usage。
|
||||
func (s *OpenAIGatewayService) readCCUpstreamJSONResponse(
|
||||
c *gin.Context,
|
||||
resp *http.Response,
|
||||
writeError compatErrorWriter,
|
||||
) (*apicompat.ChatCompletionsResponse, OpenAIUsage, error) {
|
||||
respBody, err := ReadUpstreamResponseBody(resp.Body, s.cfg, c, openAITooLargeError)
|
||||
if err != nil {
|
||||
if !errors.Is(err, ErrUpstreamResponseBodyTooLarge) {
|
||||
writeError(c, http.StatusBadGateway, "api_error", "Failed to read upstream response")
|
||||
}
|
||||
return nil, OpenAIUsage{}, fmt.Errorf("read upstream body: %w", err)
|
||||
}
|
||||
|
||||
var ccResp apicompat.ChatCompletionsResponse
|
||||
if err := json.Unmarshal(respBody, &ccResp); err != nil {
|
||||
writeError(c, http.StatusBadGateway, "api_error", "Failed to parse upstream response")
|
||||
return nil, OpenAIUsage{}, fmt.Errorf("parse chat completions response: %w", err)
|
||||
}
|
||||
|
||||
usage := OpenAIUsage{}
|
||||
if parsed, ok := extractOpenAIUsageFromJSONBytes(respBody); ok {
|
||||
usage = parsed
|
||||
}
|
||||
return &ccResp, usage, nil
|
||||
}
|
||||
|
||||
// writeOpenAIResponsesFallbackError 以 /v1/responses 回退路径的既有错误格式回写
|
||||
// （裸 error 对象；不调用 MarkResponseCommitted，与原内联写法保持一致）。
|
||||
func writeOpenAIResponsesFallbackError(c *gin.Context, statusCode int, errType, message string) {
|
||||
c.JSON(statusCode, gin.H{
|
||||
"error": gin.H{
|
||||
"type": errType,
|
||||
"message": message,
|
||||
},
|
||||
})
|
||||
}
|
||||
@@ -1,13 +1,10 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"strings"
|
||||
"sync/atomic"
|
||||
@@ -269,12 +266,7 @@ func (s *OpenAIGatewayService) ForwardAsChatCompletions(
|
||||
|
||||
// 8. Handle error response with failover
|
||||
if resp.StatusCode >= 400 {
|
||||
respBody := s.readUpstreamErrorBody(resp)
|
||||
_ = resp.Body.Close()
|
||||
resp.Body = io.NopCloser(bytes.NewReader(respBody))
|
||||
|
||||
upstreamMsg := strings.TrimSpace(extractUpstreamErrorMessage(respBody))
|
||||
upstreamMsg = sanitizeUpstreamErrorMessage(upstreamMsg)
|
||||
respBody, upstreamMsg := s.readOpenAIUpstreamError(resp)
|
||||
if account.Type == AccountTypeAPIKey &&
|
||||
openai_compat.ResolveResponsesSupport(account.Extra) == openai_compat.ResponsesSupportUnknown &&
|
||||
!isResponsesEndpointSupportedByStatus(resp.StatusCode) {
|
||||
@@ -285,31 +277,8 @@ func (s *OpenAIGatewayService) ForwardAsChatCompletions(
|
||||
)
|
||||
return s.forwardAsRawChatCompletions(ctx, c, account, body, defaultMappedModel)
|
||||
}
|
||||
if s.shouldFailoverOpenAIUpstreamResponse(resp.StatusCode, upstreamMsg, respBody) {
|
||||
upstreamDetail := ""
|
||||
if s.cfg != nil && s.cfg.Gateway.LogUpstreamErrorBody {
|
||||
maxBytes := s.cfg.Gateway.LogUpstreamErrorBodyMaxBytes
|
||||
if maxBytes <= 0 {
|
||||
maxBytes = 2048
|
||||
}
|
||||
upstreamDetail = truncateString(string(respBody), maxBytes)
|
||||
}
|
||||
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
|
||||
Platform: account.Platform,
|
||||
AccountID: account.ID,
|
||||
AccountName: account.Name,
|
||||
UpstreamStatusCode: resp.StatusCode,
|
||||
UpstreamRequestID: resp.Header.Get("x-request-id"),
|
||||
Kind: "failover",
|
||||
Message: upstreamMsg,
|
||||
Detail: upstreamDetail,
|
||||
})
|
||||
s.handleOpenAIAccountUpstreamError(ctx, account, resp.StatusCode, resp.Header, respBody, upstreamModel)
|
||||
return nil, &UpstreamFailoverError{
|
||||
StatusCode: resp.StatusCode,
|
||||
ResponseBody: respBody,
|
||||
RetryableOnSameAccount: account.IsPoolMode() && (account.IsPoolModeRetryableStatus(resp.StatusCode) || isOpenAITransientProcessingError(resp.StatusCode, upstreamMsg, respBody)),
|
||||
}
|
||||
if foErr := s.failoverOpenAIUpstreamHTTPError(ctx, c, account, resp, respBody, upstreamMsg, upstreamModel); foErr != nil {
|
||||
return nil, foErr
|
||||
}
|
||||
return s.handleChatCompletionsErrorResponse(resp, c, account, billingModel)
|
||||
}
|
||||
@@ -500,22 +469,7 @@ func (s *OpenAIGatewayService) handleChatStreamingResponse(
|
||||
requestBodyLen int,
|
||||
) (*OpenAIForwardResult, error) {
|
||||
requestID := resp.Header.Get("x-request-id")
|
||||
|
||||
headersWritten := false
|
||||
writeStreamHeaders := func() {
|
||||
if headersWritten {
|
||||
return
|
||||
}
|
||||
headersWritten = true
|
||||
if s.responseHeaderFilter != nil {
|
||||
responseheaders.WriteFilteredHeaders(c.Writer.Header(), resp.Header, s.responseHeaderFilter)
|
||||
}
|
||||
c.Writer.Header().Set("Content-Type", "text/event-stream")
|
||||
c.Writer.Header().Set("Cache-Control", "no-cache")
|
||||
c.Writer.Header().Set("Connection", "keep-alive")
|
||||
c.Writer.Header().Set("X-Accel-Buffering", "no")
|
||||
c.Writer.WriteHeader(http.StatusOK)
|
||||
}
|
||||
writeStreamHeaders := s.newStreamHeaderWriter(c, resp.Header)
|
||||
|
||||
state := apicompat.NewResponsesEventToChatState()
|
||||
state.Model = originalModel
|
||||
@@ -533,12 +487,7 @@ func (s *OpenAIGatewayService) handleChatStreamingResponse(
|
||||
var streamFailoverErr *UpstreamFailoverError
|
||||
var streamNonFailoverErr error
|
||||
|
||||
scanner := bufio.NewScanner(resp.Body)
|
||||
maxLineSize := defaultMaxLineSize
|
||||
if s.cfg != nil && s.cfg.Gateway.MaxLineSize > 0 {
|
||||
maxLineSize = s.cfg.Gateway.MaxLineSize
|
||||
}
|
||||
scanner.Buffer(make([]byte, 0, 64*1024), maxLineSize)
|
||||
scanner := s.newUpstreamSSEScanner(resp.Body)
|
||||
|
||||
streamInterval := time.Duration(0)
|
||||
if s.cfg != nil && s.cfg.Gateway.StreamDataIntervalTimeout > 0 {
|
||||
|
||||
@@ -1,13 +1,10 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"strings"
|
||||
"time"
|
||||
@@ -148,69 +145,26 @@ func (s *OpenAIGatewayService) forwardAsRawChatCompletions(
|
||||
zap.Bool("stream", clientStream),
|
||||
)
|
||||
|
||||
// 5. Build upstream request
|
||||
// 5. Build and send upstream request via the shared CC pipeline
|
||||
targetURL, err := s.rawChatCompletionsURL(account)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
upstreamCtx, releaseUpstreamCtx := detachUpstreamContext(ctx)
|
||||
upstreamReq, err := http.NewRequestWithContext(upstreamCtx, http.MethodPost, targetURL, bytes.NewReader(upstreamBody))
|
||||
releaseUpstreamCtx()
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("build upstream request: %w", err)
|
||||
}
|
||||
upstreamReq = upstreamReq.WithContext(WithHTTPUpstreamProfile(upstreamReq.Context(), HTTPUpstreamProfileOpenAI))
|
||||
upstreamReq.Header.Set("Content-Type", "application/json")
|
||||
upstreamReq.Header.Set("Authorization", "Bearer "+token)
|
||||
if clientStream {
|
||||
upstreamReq.Header.Set("Accept", "text/event-stream")
|
||||
} else {
|
||||
upstreamReq.Header.Set("Accept", "application/json")
|
||||
}
|
||||
|
||||
// 透传白名单中的客户端 header。详见 openaiCCRawAllowedHeaders 的设计说明。
|
||||
for key, values := range c.Request.Header {
|
||||
lowerKey := strings.ToLower(key)
|
||||
if openaiCCRawAllowedHeaders[lowerKey] {
|
||||
for _, v := range values {
|
||||
upstreamReq.Header.Add(key, v)
|
||||
}
|
||||
}
|
||||
}
|
||||
customUA := account.GetOpenAIUserAgent()
|
||||
if customUA != "" {
|
||||
upstreamReq.Header.Set("user-agent", customUA)
|
||||
} else if account.Platform == PlatformGrok {
|
||||
upstreamReq.Header.Set("user-agent", "sub2api-grok/1.0")
|
||||
if customUA == "" && account.Platform == PlatformGrok {
|
||||
customUA = "sub2api-grok/1.0"
|
||||
}
|
||||
|
||||
// 账号级请求头覆写（仅 openai api_key 账号启用时生效）
|
||||
account.ApplyHeaderOverrides(upstreamReq.Header)
|
||||
|
||||
// 6. Send request
|
||||
proxyURL := ""
|
||||
if account.Proxy != nil {
|
||||
proxyURL = account.Proxy.URL()
|
||||
}
|
||||
resp, err := s.httpUpstream.Do(upstreamReq, proxyURL, account.ID, account.Concurrency)
|
||||
resp, err := s.sendCCUpstreamRequest(ctx, c, account, targetURL, upstreamBody, clientStream, token, customUA)
|
||||
if err != nil {
|
||||
return nil, s.handleOpenAIUpstreamTransportError(ctx, c, account, err, false)
|
||||
return nil, err
|
||||
}
|
||||
defer func() { _ = resp.Body.Close() }()
|
||||
|
||||
// 7. Handle error response with failover
|
||||
if resp.StatusCode >= 400 {
|
||||
respBody := s.readUpstreamErrorBody(resp)
|
||||
_ = resp.Body.Close()
|
||||
resp.Body = io.NopCloser(bytes.NewReader(respBody))
|
||||
respBody, upstreamMsg := s.readOpenAIUpstreamError(resp)
|
||||
if account.Platform == PlatformGrok {
|
||||
s.updateGrokUsageSnapshot(ctx, account.ID, xai.ParseQuotaHeaders(resp.Header, resp.StatusCode))
|
||||
}
|
||||
|
||||
upstreamMsg := strings.TrimSpace(extractUpstreamErrorMessage(respBody))
|
||||
upstreamMsg = sanitizeUpstreamErrorMessage(upstreamMsg)
|
||||
if account.Platform == PlatformGrok {
|
||||
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
|
||||
Platform: account.Platform,
|
||||
AccountID: account.ID,
|
||||
@@ -230,31 +184,8 @@ func (s *OpenAIGatewayService) forwardAsRawChatCompletions(
|
||||
}
|
||||
return s.handleChatCompletionsErrorResponse(resp, c, account, billingModel)
|
||||
}
|
||||
if s.shouldFailoverOpenAIUpstreamResponse(resp.StatusCode, upstreamMsg, respBody) {
|
||||
upstreamDetail := ""
|
||||
if s.cfg != nil && s.cfg.Gateway.LogUpstreamErrorBody {
|
||||
maxBytes := s.cfg.Gateway.LogUpstreamErrorBodyMaxBytes
|
||||
if maxBytes <= 0 {
|
||||
maxBytes = 2048
|
||||
}
|
||||
upstreamDetail = truncateString(string(respBody), maxBytes)
|
||||
}
|
||||
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
|
||||
Platform: account.Platform,
|
||||
AccountID: account.ID,
|
||||
AccountName: account.Name,
|
||||
UpstreamStatusCode: resp.StatusCode,
|
||||
UpstreamRequestID: resp.Header.Get("x-request-id"),
|
||||
Kind: "failover",
|
||||
Message: upstreamMsg,
|
||||
Detail: upstreamDetail,
|
||||
})
|
||||
s.handleOpenAIAccountUpstreamError(ctx, account, resp.StatusCode, resp.Header, respBody, upstreamModel)
|
||||
return nil, &UpstreamFailoverError{
|
||||
StatusCode: resp.StatusCode,
|
||||
ResponseBody: respBody,
|
||||
RetryableOnSameAccount: account.IsPoolMode() && (account.IsPoolModeRetryableStatus(resp.StatusCode) || isOpenAITransientProcessingError(resp.StatusCode, upstreamMsg, respBody)),
|
||||
}
|
||||
if foErr := s.failoverOpenAIUpstreamHTTPError(ctx, c, account, resp, respBody, upstreamMsg, upstreamModel); foErr != nil {
|
||||
return nil, foErr
|
||||
}
|
||||
return s.handleChatCompletionsErrorResponse(resp, c, account, billingModel)
|
||||
}
|
||||
@@ -286,15 +217,7 @@ func (s *OpenAIGatewayService) rawChatCompletionsURL(account *Account) (string,
|
||||
return targetURL, nil
|
||||
}
|
||||
|
||||
baseURL := account.GetOpenAIBaseURL()
|
||||
if baseURL == "" {
|
||||
baseURL = "https://api.openai.com"
|
||||
}
|
||||
validatedURL, err := s.validateUpstreamBaseURL(baseURL)
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("invalid base_url: %w", err)
|
||||
}
|
||||
return buildOpenAIChatCompletionsURL(validatedURL), nil
|
||||
return s.openAIChatCompletionsTargetURL(account)
|
||||
}
|
||||
|
||||
// streamRawChatCompletions 透传上游 CC SSE 流到客户端，并提取 usage（包括
|
||||
@@ -316,29 +239,8 @@ func (s *OpenAIGatewayService) streamRawChatCompletions(
|
||||
requestBodyLen int,
|
||||
) (*OpenAIForwardResult, error) {
|
||||
requestID := resp.Header.Get("x-request-id")
|
||||
|
||||
headersWritten := false
|
||||
writeStreamHeaders := func() {
|
||||
if headersWritten {
|
||||
return
|
||||
}
|
||||
headersWritten = true
|
||||
if s.responseHeaderFilter != nil {
|
||||
responseheaders.WriteFilteredHeaders(c.Writer.Header(), resp.Header, s.responseHeaderFilter)
|
||||
}
|
||||
c.Writer.Header().Set("Content-Type", "text/event-stream")
|
||||
c.Writer.Header().Set("Cache-Control", "no-cache")
|
||||
c.Writer.Header().Set("Connection", "keep-alive")
|
||||
c.Writer.Header().Set("X-Accel-Buffering", "no")
|
||||
c.Writer.WriteHeader(http.StatusOK)
|
||||
}
|
||||
|
||||
scanner := bufio.NewScanner(resp.Body)
|
||||
maxLineSize := defaultMaxLineSize
|
||||
if s.cfg != nil && s.cfg.Gateway.MaxLineSize > 0 {
|
||||
maxLineSize = s.cfg.Gateway.MaxLineSize
|
||||
}
|
||||
scanner.Buffer(make([]byte, 0, 64*1024), maxLineSize)
|
||||
writeStreamHeaders := s.newStreamHeaderWriter(c, resp.Header)
|
||||
scanner := s.newUpstreamSSEScanner(resp.Body)
|
||||
|
||||
var usage OpenAIUsage
|
||||
var firstTokenMs *int
|
||||
|
||||
@@ -1,13 +1,10 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"strings"
|
||||
"sync/atomic"
|
||||
@@ -322,16 +319,12 @@ func (s *OpenAIGatewayService) ForwardAsAnthropic(
|
||||
|
||||
// 8. Handle error response with failover
|
||||
if resp.StatusCode >= 400 {
|
||||
respBody := s.readUpstreamErrorBody(resp)
|
||||
_ = resp.Body.Close()
|
||||
resp.Body = io.NopCloser(bytes.NewReader(respBody))
|
||||
respBody, upstreamMsg := s.readOpenAIUpstreamError(resp)
|
||||
if account.Platform == PlatformGrok {
|
||||
s.updateGrokUsageSnapshot(ctx, account.ID, xai.ParseQuotaHeaders(resp.Header, resp.StatusCode))
|
||||
s.handleGrokAccountUpstreamError(ctx, account, resp.StatusCode, resp.Header, respBody)
|
||||
}
|
||||
|
||||
upstreamMsg := strings.TrimSpace(extractUpstreamErrorMessage(respBody))
|
||||
upstreamMsg = sanitizeUpstreamErrorMessage(upstreamMsg)
|
||||
if previousResponseID != "" && (isOpenAICompatPreviousResponseNotFound(resp.StatusCode, upstreamMsg, respBody) || isOpenAICompatPreviousResponseUnsupported(resp.StatusCode, upstreamMsg, respBody)) {
|
||||
if isOpenAICompatPreviousResponseUnsupported(resp.StatusCode, upstreamMsg, respBody) {
|
||||
s.disableOpenAICompatSessionContinuation(ctx, c, account, promptCacheKey)
|
||||
@@ -345,31 +338,8 @@ func (s *OpenAIGatewayService) ForwardAsAnthropic(
|
||||
)
|
||||
return s.ForwardAsAnthropic(ctx, c, account, body, promptCacheKey, defaultMappedModel)
|
||||
}
|
||||
if s.shouldFailoverOpenAIUpstreamResponse(resp.StatusCode, upstreamMsg, respBody) {
|
||||
upstreamDetail := ""
|
||||
if s.cfg != nil && s.cfg.Gateway.LogUpstreamErrorBody {
|
||||
maxBytes := s.cfg.Gateway.LogUpstreamErrorBodyMaxBytes
|
||||
if maxBytes <= 0 {
|
||||
maxBytes = 2048
|
||||
}
|
||||
upstreamDetail = truncateString(string(respBody), maxBytes)
|
||||
}
|
||||
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
|
||||
Platform: account.Platform,
|
||||
AccountID: account.ID,
|
||||
AccountName: account.Name,
|
||||
UpstreamStatusCode: resp.StatusCode,
|
||||
UpstreamRequestID: resp.Header.Get("x-request-id"),
|
||||
Kind: "failover",
|
||||
Message: upstreamMsg,
|
||||
Detail: upstreamDetail,
|
||||
})
|
||||
s.handleOpenAIAccountUpstreamError(ctx, account, resp.StatusCode, resp.Header, respBody, upstreamModel)
|
||||
return nil, &UpstreamFailoverError{
|
||||
StatusCode: resp.StatusCode,
|
||||
ResponseBody: respBody,
|
||||
RetryableOnSameAccount: account.IsPoolMode() && (account.IsPoolModeRetryableStatus(resp.StatusCode) || isOpenAITransientProcessingError(resp.StatusCode, upstreamMsg, respBody)),
|
||||
}
|
||||
if foErr := s.failoverOpenAIUpstreamHTTPError(ctx, c, account, resp, respBody, upstreamMsg, upstreamModel); foErr != nil {
|
||||
return nil, foErr
|
||||
}
|
||||
// Non-failover error: return Anthropic-formatted error to client
|
||||
return s.handleAnthropicErrorResponse(resp, c, account, billingModel)
|
||||
@@ -573,12 +543,7 @@ func (s *OpenAIGatewayService) readOpenAICompatBufferedTerminal(
|
||||
return nil, usage, acc, errors.New("upstream response body is nil")
|
||||
}
|
||||
|
||||
scanner := bufio.NewScanner(resp.Body)
|
||||
maxLineSize := defaultMaxLineSize
|
||||
if s.cfg != nil && s.cfg.Gateway.MaxLineSize > 0 {
|
||||
maxLineSize = s.cfg.Gateway.MaxLineSize
|
||||
}
|
||||
scanner.Buffer(make([]byte, 0, 64*1024), maxLineSize)
|
||||
scanner := s.newUpstreamSSEScanner(resp.Body)
|
||||
|
||||
streamInterval := time.Duration(0)
|
||||
if s.cfg != nil && s.cfg.Gateway.StreamDataIntervalTimeout > 0 {
|
||||
@@ -737,22 +702,7 @@ func (s *OpenAIGatewayService) handleAnthropicStreamingResponse(
|
||||
startTime time.Time,
|
||||
) (*OpenAIForwardResult, error) {
|
||||
requestID := resp.Header.Get("x-request-id")
|
||||
|
||||
headersWritten := false
|
||||
writeStreamHeaders := func() {
|
||||
if headersWritten {
|
||||
return
|
||||
}
|
||||
headersWritten = true
|
||||
if s.responseHeaderFilter != nil {
|
||||
responseheaders.WriteFilteredHeaders(c.Writer.Header(), resp.Header, s.responseHeaderFilter)
|
||||
}
|
||||
c.Writer.Header().Set("Content-Type", "text/event-stream")
|
||||
c.Writer.Header().Set("Cache-Control", "no-cache")
|
||||
c.Writer.Header().Set("Connection", "keep-alive")
|
||||
c.Writer.Header().Set("X-Accel-Buffering", "no")
|
||||
c.Writer.WriteHeader(http.StatusOK)
|
||||
}
|
||||
writeStreamHeaders := s.newStreamHeaderWriter(c, resp.Header)
|
||||
|
||||
state := apicompat.NewResponsesEventToAnthropicState()
|
||||
state.Model = originalModel
|
||||
@@ -763,12 +713,7 @@ func (s *OpenAIGatewayService) handleAnthropicStreamingResponse(
|
||||
clientDisconnected := false
|
||||
clientOutputStarted := false
|
||||
|
||||
scanner := bufio.NewScanner(resp.Body)
|
||||
maxLineSize := defaultMaxLineSize
|
||||
if s.cfg != nil && s.cfg.Gateway.MaxLineSize > 0 {
|
||||
maxLineSize = s.cfg.Gateway.MaxLineSize
|
||||
}
|
||||
scanner.Buffer(make([]byte, 0, 64*1024), maxLineSize)
|
||||
scanner := s.newUpstreamSSEScanner(resp.Body)
|
||||
|
||||
streamInterval := time.Duration(0)
|
||||
if s.cfg != nil && s.cfg.Gateway.StreamDataIntervalTimeout > 0 {
|
||||
|
||||
@@ -1,13 +1,9 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"strings"
|
||||
"time"
|
||||
@@ -100,91 +96,22 @@ func (s *OpenAIGatewayService) forwardAnthropicViaRawChatCompletions(
|
||||
zap.Bool("stream", clientStream),
|
||||
)
|
||||
|
||||
// 3. Build upstream request
|
||||
apiKey := account.GetOpenAIApiKey()
|
||||
if apiKey == "" {
|
||||
return nil, fmt.Errorf("account %d missing api_key", account.ID)
|
||||
}
|
||||
baseURL := account.GetOpenAIBaseURL()
|
||||
if baseURL == "" {
|
||||
baseURL = "https://api.openai.com"
|
||||
}
|
||||
validatedURL, err := s.validateUpstreamBaseURL(baseURL)
|
||||
// 3. Build and send upstream request via the shared CC pipeline
|
||||
apiKey, targetURL, err := s.resolveCCFallbackTarget(account)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("invalid base_url: %w", err)
|
||||
return nil, err
|
||||
}
|
||||
targetURL := buildOpenAIChatCompletionsURL(validatedURL)
|
||||
|
||||
upstreamCtx, releaseUpstreamCtx := detachUpstreamContext(ctx)
|
||||
upstreamReq, err := http.NewRequestWithContext(upstreamCtx, http.MethodPost, targetURL, bytes.NewReader(chatBody))
|
||||
releaseUpstreamCtx()
|
||||
resp, err := s.sendCCUpstreamRequest(ctx, c, account, targetURL, chatBody, clientStream, apiKey, account.GetOpenAIUserAgent())
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("build upstream request: %w", err)
|
||||
}
|
||||
upstreamReq = upstreamReq.WithContext(WithHTTPUpstreamProfile(upstreamReq.Context(), HTTPUpstreamProfileOpenAI))
|
||||
upstreamReq.Header.Set("Content-Type", "application/json")
|
||||
upstreamReq.Header.Set("Authorization", "Bearer "+apiKey)
|
||||
if clientStream {
|
||||
upstreamReq.Header.Set("Accept", "text/event-stream")
|
||||
} else {
|
||||
upstreamReq.Header.Set("Accept", "application/json")
|
||||
}
|
||||
for key, values := range c.Request.Header {
|
||||
lowerKey := strings.ToLower(key)
|
||||
if openaiCCRawAllowedHeaders[lowerKey] {
|
||||
for _, v := range values {
|
||||
upstreamReq.Header.Add(key, v)
|
||||
}
|
||||
}
|
||||
}
|
||||
if customUA := account.GetOpenAIUserAgent(); customUA != "" {
|
||||
upstreamReq.Header.Set("user-agent", customUA)
|
||||
}
|
||||
account.ApplyHeaderOverrides(upstreamReq.Header)
|
||||
|
||||
proxyURL := ""
|
||||
if account.Proxy != nil {
|
||||
proxyURL = account.Proxy.URL()
|
||||
}
|
||||
resp, err := s.httpUpstream.Do(upstreamReq, proxyURL, account.ID, account.Concurrency)
|
||||
if err != nil {
|
||||
return nil, s.handleOpenAIUpstreamTransportError(ctx, c, account, err, false)
|
||||
return nil, err
|
||||
}
|
||||
defer func() { _ = resp.Body.Close() }()
|
||||
|
||||
// 4. Handle error responses
|
||||
if resp.StatusCode >= 400 {
|
||||
respBody := s.readUpstreamErrorBody(resp)
|
||||
_ = resp.Body.Close()
|
||||
resp.Body = io.NopCloser(bytes.NewReader(respBody))
|
||||
|
||||
upstreamMsg := strings.TrimSpace(extractUpstreamErrorMessage(respBody))
|
||||
upstreamMsg = sanitizeUpstreamErrorMessage(upstreamMsg)
|
||||
if s.shouldFailoverOpenAIUpstreamResponse(resp.StatusCode, upstreamMsg, respBody) {
|
||||
upstreamDetail := ""
|
||||
if s.cfg != nil && s.cfg.Gateway.LogUpstreamErrorBody {
|
||||
maxBytes := s.cfg.Gateway.LogUpstreamErrorBodyMaxBytes
|
||||
if maxBytes <= 0 {
|
||||
maxBytes = 2048
|
||||
}
|
||||
upstreamDetail = truncateString(string(respBody), maxBytes)
|
||||
}
|
||||
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
|
||||
Platform: account.Platform,
|
||||
AccountID: account.ID,
|
||||
AccountName: account.Name,
|
||||
UpstreamStatusCode: resp.StatusCode,
|
||||
UpstreamRequestID: resp.Header.Get("x-request-id"),
|
||||
Kind: "failover",
|
||||
Message: upstreamMsg,
|
||||
Detail: upstreamDetail,
|
||||
})
|
||||
s.handleOpenAIAccountUpstreamError(ctx, account, resp.StatusCode, resp.Header, respBody, upstreamModel)
|
||||
return nil, &UpstreamFailoverError{
|
||||
StatusCode: resp.StatusCode,
|
||||
ResponseBody: respBody,
|
||||
RetryableOnSameAccount: account.IsPoolMode() && (account.IsPoolModeRetryableStatus(resp.StatusCode) || isOpenAITransientProcessingError(resp.StatusCode, upstreamMsg, respBody)),
|
||||
}
|
||||
respBody, upstreamMsg := s.readOpenAIUpstreamError(resp)
|
||||
if foErr := s.failoverOpenAIUpstreamHTTPError(ctx, c, account, resp, respBody, upstreamMsg, upstreamModel); foErr != nil {
|
||||
return nil, foErr
|
||||
}
|
||||
// Non-failover error: return Anthropic-formatted error to client via the
|
||||
// shared compat handler (passthrough rules, ops recording, cyber_policy).
|
||||
@@ -209,28 +136,14 @@ func (s *OpenAIGatewayService) bufferChatCompletionsAsAnthropic(
|
||||
startTime time.Time,
|
||||
) (*OpenAIForwardResult, error) {
|
||||
requestID := resp.Header.Get("x-request-id")
|
||||
respBody, err := ReadUpstreamResponseBody(resp.Body, s.cfg, c, openAITooLargeError)
|
||||
ccResp, usage, err := s.readCCUpstreamJSONResponse(c, resp, writeAnthropicError)
|
||||
if err != nil {
|
||||
if !errors.Is(err, ErrUpstreamResponseBodyTooLarge) {
|
||||
writeAnthropicError(c, http.StatusBadGateway, "api_error", "Failed to read upstream response")
|
||||
}
|
||||
return nil, fmt.Errorf("read upstream body: %w", err)
|
||||
return nil, err
|
||||
}
|
||||
|
||||
var ccResp apicompat.ChatCompletionsResponse
|
||||
if err := json.Unmarshal(respBody, &ccResp); err != nil {
|
||||
writeAnthropicError(c, http.StatusBadGateway, "api_error", "Failed to parse upstream response")
|
||||
return nil, fmt.Errorf("parse chat completions response: %w", err)
|
||||
}
|
||||
responsesResp := apicompat.ChatCompletionsResponseToResponses(&ccResp, originalModel)
|
||||
responsesResp := apicompat.ChatCompletionsResponseToResponses(ccResp, originalModel)
|
||||
|
||||
anthropicResp := apicompat.ResponsesToAnthropic(responsesResp, originalModel)
|
||||
|
||||
usage := OpenAIUsage{}
|
||||
if parsed, ok := extractOpenAIUsageFromJSONBytes(respBody); ok {
|
||||
usage = parsed
|
||||
}
|
||||
|
||||
if s.responseHeaderFilter != nil {
|
||||
responseheaders.WriteFilteredHeaders(c.Writer.Header(), resp.Header, s.responseHeaderFilter)
|
||||
}
|
||||
@@ -260,71 +173,18 @@ func (s *OpenAIGatewayService) streamChatCompletionsAsAnthropic(
|
||||
startTime time.Time,
|
||||
) (*OpenAIForwardResult, error) {
|
||||
requestID := resp.Header.Get("x-request-id")
|
||||
headersWritten := false
|
||||
writeStreamHeaders := func() {
|
||||
if headersWritten {
|
||||
return
|
||||
}
|
||||
headersWritten = true
|
||||
if s.responseHeaderFilter != nil {
|
||||
responseheaders.WriteFilteredHeaders(c.Writer.Header(), resp.Header, s.responseHeaderFilter)
|
||||
}
|
||||
c.Writer.Header().Set("Content-Type", "text/event-stream")
|
||||
c.Writer.Header().Set("Cache-Control", "no-cache")
|
||||
c.Writer.Header().Set("Connection", "keep-alive")
|
||||
c.Writer.Header().Set("X-Accel-Buffering", "no")
|
||||
c.Writer.WriteHeader(http.StatusOK)
|
||||
}
|
||||
writeStreamHeaders := s.newStreamHeaderWriter(c, resp.Header)
|
||||
|
||||
ccState := apicompat.NewChatCompletionsToResponsesStreamState(originalModel)
|
||||
anthropicState := apicompat.NewResponsesEventToAnthropicState()
|
||||
anthropicState.Model = originalModel
|
||||
var usage OpenAIUsage
|
||||
var firstTokenMs *int
|
||||
clientDisconnected := false
|
||||
sawDone := false
|
||||
|
||||
scanner := bufio.NewScanner(resp.Body)
|
||||
maxLineSize := defaultMaxLineSize
|
||||
if s.cfg != nil && s.cfg.Gateway.MaxLineSize > 0 {
|
||||
maxLineSize = s.cfg.Gateway.MaxLineSize
|
||||
}
|
||||
scanner.Buffer(make([]byte, 0, 64*1024), maxLineSize)
|
||||
|
||||
for scanner.Scan() {
|
||||
line := scanner.Text()
|
||||
payload, ok := extractOpenAISSEDataLine(line)
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
payload = strings.TrimSpace(payload)
|
||||
if payload == "" {
|
||||
continue
|
||||
}
|
||||
if payload == "[DONE]" {
|
||||
sawDone = true
|
||||
break
|
||||
}
|
||||
|
||||
if u := extractCCStreamUsage(payload); u != nil {
|
||||
usage = *u
|
||||
}
|
||||
|
||||
var chunk apicompat.ChatCompletionsChunk
|
||||
if err := json.Unmarshal([]byte(payload), &chunk); err != nil {
|
||||
logger.L().Warn("openai messages chat fallback: failed to parse chat stream chunk",
|
||||
zap.Error(err),
|
||||
zap.String("request_id", requestID),
|
||||
)
|
||||
continue
|
||||
}
|
||||
if firstTokenMs == nil && !isOpenAIChatUsageOnlyStreamChunk(payload) && chatChunkStartsResponsesOutput(&chunk) {
|
||||
ms := int(time.Since(startTime).Milliseconds())
|
||||
firstTokenMs = &ms
|
||||
}
|
||||
|
||||
// 与 responses 兄弟不同：客户端断开后仍继续做事件转换（喂 anthropicState），
|
||||
// 仅跳过写出，保证 finalize 阶段的 usage 汇总不受断开影响。
|
||||
emitChunk := func(chunk *apicompat.ChatCompletionsChunk) {
|
||||
// CC chunk → Responses events → Anthropic events
|
||||
responsesEvents := apicompat.ChatCompletionsChunkToResponsesEvents(&chunk, ccState)
|
||||
responsesEvents := apicompat.ChatCompletionsChunkToResponsesEvents(chunk, ccState)
|
||||
for _, rEvent := range responsesEvents {
|
||||
anthropicEvents := apicompat.ResponsesEventToAnthropicEvents(&rEvent, anthropicState)
|
||||
if clientDisconnected {
|
||||
@@ -347,13 +207,10 @@ func (s *OpenAIGatewayService) streamChatCompletionsAsAnthropic(
|
||||
}
|
||||
}
|
||||
|
||||
if err := scanner.Err(); err != nil {
|
||||
if !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded) {
|
||||
logger.L().Warn("openai messages chat fallback: stream read error",
|
||||
zap.Error(err),
|
||||
zap.String("request_id", requestID),
|
||||
)
|
||||
}
|
||||
scan := s.scanCCStream(resp, "openai messages chat fallback", requestID, startTime, emitChunk)
|
||||
usage := scan.Usage
|
||||
|
||||
if scan.Err != nil {
|
||||
// Broken upstream read: skip finalization so no synthetic message_stop
|
||||
// masks the truncation, and surface the error to flag usage incomplete
|
||||
// (mirrors forwardResponsesViaRawChatCompletions).
|
||||
@@ -367,9 +224,9 @@ func (s *OpenAIGatewayService) streamChatCompletionsAsAnthropic(
|
||||
ServiceTier: serviceTier,
|
||||
Stream: true,
|
||||
Duration: time.Since(startTime),
|
||||
FirstTokenMs: firstTokenMs,
|
||||
FirstTokenMs: scan.FirstTokenMs,
|
||||
ClientDisconnect: clientDisconnected,
|
||||
}, fmt.Errorf("stream usage incomplete: %w", err)
|
||||
}, fmt.Errorf("stream usage incomplete: %w", scan.Err)
|
||||
}
|
||||
|
||||
// Finalize CC→Responses stream (emit response.completed)
|
||||
@@ -397,10 +254,8 @@ func (s *OpenAIGatewayService) streamChatCompletionsAsAnthropic(
|
||||
if !clientDisconnected {
|
||||
c.Writer.Flush()
|
||||
}
|
||||
if !sawDone {
|
||||
logger.L().Debug("openai messages chat fallback: upstream stream ended without done sentinel",
|
||||
zap.String("request_id", requestID),
|
||||
)
|
||||
if !scan.SawDone {
|
||||
logCCStreamMissingDoneSentinel("openai messages chat fallback", requestID)
|
||||
}
|
||||
|
||||
return &OpenAIForwardResult{
|
||||
@@ -413,7 +268,7 @@ func (s *OpenAIGatewayService) streamChatCompletionsAsAnthropic(
|
||||
ServiceTier: serviceTier,
|
||||
Stream: true,
|
||||
Duration: time.Since(startTime),
|
||||
FirstTokenMs: firstTokenMs,
|
||||
FirstTokenMs: scan.FirstTokenMs,
|
||||
ClientDisconnect: clientDisconnected,
|
||||
}, nil
|
||||
}
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -1,13 +1,10 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"strings"
|
||||
"time"
|
||||
@@ -31,22 +28,12 @@ func (s *OpenAIGatewayService) forwardResponsesViaRawChatCompletions(
|
||||
|
||||
var responsesReq apicompat.ResponsesRequest
|
||||
if err := json.Unmarshal(body, &responsesReq); err != nil {
|
||||
c.JSON(http.StatusBadRequest, gin.H{
|
||||
"error": gin.H{
|
||||
"type": "invalid_request_error",
|
||||
"message": "Failed to parse request body",
|
||||
},
|
||||
})
|
||||
writeOpenAIResponsesFallbackError(c, http.StatusBadRequest, "invalid_request_error", "Failed to parse request body")
|
||||
return nil, fmt.Errorf("parse responses request: %w", err)
|
||||
}
|
||||
originalModel := strings.TrimSpace(responsesReq.Model)
|
||||
if originalModel == "" {
|
||||
c.JSON(http.StatusBadRequest, gin.H{
|
||||
"error": gin.H{
|
||||
"type": "invalid_request_error",
|
||||
"message": "model is required",
|
||||
},
|
||||
})
|
||||
writeOpenAIResponsesFallbackError(c, http.StatusBadRequest, "invalid_request_error", "model is required")
|
||||
return nil, fmt.Errorf("missing model in request")
|
||||
}
|
||||
|
||||
@@ -56,12 +43,7 @@ func (s *OpenAIGatewayService) forwardResponsesViaRawChatCompletions(
|
||||
|
||||
chatReq, err := apicompat.ResponsesToChatCompletionsRequest(&responsesReq)
|
||||
if err != nil {
|
||||
c.JSON(http.StatusBadRequest, gin.H{
|
||||
"error": gin.H{
|
||||
"type": "invalid_request_error",
|
||||
"message": err.Error(),
|
||||
},
|
||||
})
|
||||
writeOpenAIResponsesFallbackError(c, http.StatusBadRequest, "invalid_request_error", err.Error())
|
||||
return nil, fmt.Errorf("convert responses to chat completions: %w", err)
|
||||
}
|
||||
|
||||
@@ -98,94 +80,21 @@ func (s *OpenAIGatewayService) forwardResponsesViaRawChatCompletions(
|
||||
zap.Bool("stream", clientStream),
|
||||
)
|
||||
|
||||
apiKey := account.GetOpenAIApiKey()
|
||||
if apiKey == "" {
|
||||
return nil, fmt.Errorf("account %d missing api_key", account.ID)
|
||||
}
|
||||
baseURL := account.GetOpenAIBaseURL()
|
||||
if baseURL == "" {
|
||||
baseURL = "https://api.openai.com"
|
||||
}
|
||||
validatedURL, err := s.validateUpstreamBaseURL(baseURL)
|
||||
// Build and send upstream request via the shared CC pipeline
|
||||
apiKey, targetURL, err := s.resolveCCFallbackTarget(account)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("invalid base_url: %w", err)
|
||||
return nil, err
|
||||
}
|
||||
targetURL := buildOpenAIChatCompletionsURL(validatedURL)
|
||||
|
||||
upstreamCtx, releaseUpstreamCtx := detachUpstreamContext(ctx)
|
||||
upstreamReq, err := http.NewRequestWithContext(upstreamCtx, http.MethodPost, targetURL, bytes.NewReader(chatBody))
|
||||
releaseUpstreamCtx()
|
||||
resp, err := s.sendCCUpstreamRequest(ctx, c, account, targetURL, chatBody, clientStream, apiKey, account.GetOpenAIUserAgent())
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("build upstream request: %w", err)
|
||||
}
|
||||
upstreamReq = upstreamReq.WithContext(WithHTTPUpstreamProfile(upstreamReq.Context(), HTTPUpstreamProfileOpenAI))
|
||||
upstreamReq.Header.Set("Content-Type", "application/json")
|
||||
upstreamReq.Header.Set("Authorization", "Bearer "+apiKey)
|
||||
if clientStream {
|
||||
upstreamReq.Header.Set("Accept", "text/event-stream")
|
||||
} else {
|
||||
upstreamReq.Header.Set("Accept", "application/json")
|
||||
}
|
||||
for key, values := range c.Request.Header {
|
||||
lowerKey := strings.ToLower(key)
|
||||
if openaiCCRawAllowedHeaders[lowerKey] {
|
||||
for _, v := range values {
|
||||
upstreamReq.Header.Add(key, v)
|
||||
}
|
||||
}
|
||||
}
|
||||
if customUA := account.GetOpenAIUserAgent(); customUA != "" {
|
||||
upstreamReq.Header.Set("user-agent", customUA)
|
||||
}
|
||||
|
||||
// 账号级请求头覆写（仅 openai api_key 账号启用时生效）
|
||||
account.ApplyHeaderOverrides(upstreamReq.Header)
|
||||
|
||||
proxyURL := ""
|
||||
if account.Proxy != nil {
|
||||
proxyURL = account.Proxy.URL()
|
||||
}
|
||||
resp, err := s.httpUpstream.Do(upstreamReq, proxyURL, account.ID, account.Concurrency)
|
||||
if err != nil {
|
||||
// Transport-level failure (proxy/DNS/TCP/TLS — no HTTP response). Convert to
|
||||
// a failover so the handler switches to a healthy account, and temporarily
|
||||
// unschedule the account on durable faults (e.g. rejected proxy credentials).
|
||||
return nil, s.handleOpenAIUpstreamTransportError(ctx, c, account, err, false)
|
||||
return nil, err
|
||||
}
|
||||
defer func() { _ = resp.Body.Close() }()
|
||||
|
||||
if resp.StatusCode >= 400 {
|
||||
respBody := s.readUpstreamErrorBody(resp)
|
||||
_ = resp.Body.Close()
|
||||
resp.Body = io.NopCloser(bytes.NewReader(respBody))
|
||||
|
||||
upstreamMsg := strings.TrimSpace(extractUpstreamErrorMessage(respBody))
|
||||
upstreamMsg = sanitizeUpstreamErrorMessage(upstreamMsg)
|
||||
if s.shouldFailoverOpenAIUpstreamResponse(resp.StatusCode, upstreamMsg, respBody) {
|
||||
upstreamDetail := ""
|
||||
if s.cfg != nil && s.cfg.Gateway.LogUpstreamErrorBody {
|
||||
maxBytes := s.cfg.Gateway.LogUpstreamErrorBodyMaxBytes
|
||||
if maxBytes <= 0 {
|
||||
maxBytes = 2048
|
||||
}
|
||||
upstreamDetail = truncateString(string(respBody), maxBytes)
|
||||
}
|
||||
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
|
||||
Platform: account.Platform,
|
||||
AccountID: account.ID,
|
||||
AccountName: account.Name,
|
||||
UpstreamStatusCode: resp.StatusCode,
|
||||
UpstreamRequestID: resp.Header.Get("x-request-id"),
|
||||
Kind: "failover",
|
||||
Message: upstreamMsg,
|
||||
Detail: upstreamDetail,
|
||||
})
|
||||
s.handleOpenAIAccountUpstreamError(ctx, account, resp.StatusCode, resp.Header, respBody, upstreamModel)
|
||||
return nil, &UpstreamFailoverError{
|
||||
StatusCode: resp.StatusCode,
|
||||
ResponseBody: respBody,
|
||||
RetryableOnSameAccount: account.IsPoolMode() && (account.IsPoolModeRetryableStatus(resp.StatusCode) || isOpenAITransientProcessingError(resp.StatusCode, upstreamMsg, respBody)),
|
||||
}
|
||||
respBody, upstreamMsg := s.readOpenAIUpstreamError(resp)
|
||||
if foErr := s.failoverOpenAIUpstreamHTTPError(ctx, c, account, resp, respBody, upstreamMsg, upstreamModel); foErr != nil {
|
||||
return nil, foErr
|
||||
}
|
||||
return s.handleErrorResponse(ctx, resp, c, account, chatBody, billingModel)
|
||||
}
|
||||
@@ -207,35 +116,11 @@ func (s *OpenAIGatewayService) bufferChatCompletionsAsResponses(
|
||||
startTime time.Time,
|
||||
) (*OpenAIForwardResult, error) {
|
||||
requestID := resp.Header.Get("x-request-id")
|
||||
respBody, err := ReadUpstreamResponseBody(resp.Body, s.cfg, c, openAITooLargeError)
|
||||
ccResp, usage, err := s.readCCUpstreamJSONResponse(c, resp, writeOpenAIResponsesFallbackError)
|
||||
if err != nil {
|
||||
if !errors.Is(err, ErrUpstreamResponseBodyTooLarge) {
|
||||
c.JSON(http.StatusBadGateway, gin.H{
|
||||
"error": gin.H{
|
||||
"type": "api_error",
|
||||
"message": "Failed to read upstream response",
|
||||
},
|
||||
})
|
||||
}
|
||||
return nil, fmt.Errorf("read upstream body: %w", err)
|
||||
}
|
||||
|
||||
var ccResp apicompat.ChatCompletionsResponse
|
||||
if err := json.Unmarshal(respBody, &ccResp); err != nil {
|
||||
c.JSON(http.StatusBadGateway, gin.H{
|
||||
"error": gin.H{
|
||||
"type": "api_error",
|
||||
"message": "Failed to parse upstream response",
|
||||
},
|
||||
})
|
||||
return nil, fmt.Errorf("parse chat completions response: %w", err)
|
||||
}
|
||||
responsesResp := apicompat.ChatCompletionsResponseToResponses(&ccResp, originalModel)
|
||||
|
||||
usage := OpenAIUsage{}
|
||||
if parsed, ok := extractOpenAIUsageFromJSONBytes(respBody); ok {
|
||||
usage = parsed
|
||||
return nil, err
|
||||
}
|
||||
responsesResp := apicompat.ChatCompletionsResponseToResponses(ccResp, originalModel)
|
||||
|
||||
if s.responseHeaderFilter != nil {
|
||||
responseheaders.WriteFilteredHeaders(c.Writer.Header(), resp.Header, s.responseHeaderFilter)
|
||||
@@ -266,27 +151,10 @@ func (s *OpenAIGatewayService) streamChatCompletionsAsResponses(
|
||||
startTime time.Time,
|
||||
) (*OpenAIForwardResult, error) {
|
||||
requestID := resp.Header.Get("x-request-id")
|
||||
headersWritten := false
|
||||
writeStreamHeaders := func() {
|
||||
if headersWritten {
|
||||
return
|
||||
}
|
||||
headersWritten = true
|
||||
if s.responseHeaderFilter != nil {
|
||||
responseheaders.WriteFilteredHeaders(c.Writer.Header(), resp.Header, s.responseHeaderFilter)
|
||||
}
|
||||
c.Writer.Header().Set("Content-Type", "text/event-stream")
|
||||
c.Writer.Header().Set("Cache-Control", "no-cache")
|
||||
c.Writer.Header().Set("Connection", "keep-alive")
|
||||
c.Writer.Header().Set("X-Accel-Buffering", "no")
|
||||
c.Writer.WriteHeader(http.StatusOK)
|
||||
}
|
||||
writeStreamHeaders := s.newStreamHeaderWriter(c, resp.Header)
|
||||
|
||||
state := apicompat.NewChatCompletionsToResponsesStreamState(originalModel)
|
||||
var usage OpenAIUsage
|
||||
var firstTokenMs *int
|
||||
clientDisconnected := false
|
||||
sawDone := false
|
||||
|
||||
writeEvents := func(events []apicompat.ResponsesStreamEvent) {
|
||||
if clientDisconnected || len(events) == 0 {
|
||||
@@ -314,57 +182,14 @@ func (s *OpenAIGatewayService) streamChatCompletionsAsResponses(
|
||||
c.Writer.Flush()
|
||||
}
|
||||
|
||||
scanner := bufio.NewScanner(resp.Body)
|
||||
maxLineSize := defaultMaxLineSize
|
||||
if s.cfg != nil && s.cfg.Gateway.MaxLineSize > 0 {
|
||||
maxLineSize = s.cfg.Gateway.MaxLineSize
|
||||
}
|
||||
scanner.Buffer(make([]byte, 0, 64*1024), maxLineSize)
|
||||
scan := s.scanCCStream(resp, "openai responses chat fallback", requestID, startTime, func(chunk *apicompat.ChatCompletionsChunk) {
|
||||
writeEvents(apicompat.ChatCompletionsChunkToResponsesEvents(chunk, state))
|
||||
})
|
||||
|
||||
for scanner.Scan() {
|
||||
line := scanner.Text()
|
||||
payload, ok := extractOpenAISSEDataLine(line)
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
payload = strings.TrimSpace(payload)
|
||||
if payload == "" {
|
||||
continue
|
||||
}
|
||||
if payload == "[DONE]" {
|
||||
sawDone = true
|
||||
break
|
||||
}
|
||||
|
||||
if u := extractCCStreamUsage(payload); u != nil {
|
||||
usage = *u
|
||||
}
|
||||
|
||||
var chunk apicompat.ChatCompletionsChunk
|
||||
if err := json.Unmarshal([]byte(payload), &chunk); err != nil {
|
||||
logger.L().Warn("openai responses chat fallback: failed to parse chat stream chunk",
|
||||
zap.Error(err),
|
||||
zap.String("request_id", requestID),
|
||||
)
|
||||
continue
|
||||
}
|
||||
if firstTokenMs == nil && !isOpenAIChatUsageOnlyStreamChunk(payload) && chatChunkStartsResponsesOutput(&chunk) {
|
||||
ms := int(time.Since(startTime).Milliseconds())
|
||||
firstTokenMs = &ms
|
||||
}
|
||||
writeEvents(apicompat.ChatCompletionsChunkToResponsesEvents(&chunk, state))
|
||||
}
|
||||
|
||||
if err := scanner.Err(); err != nil {
|
||||
if !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded) {
|
||||
logger.L().Warn("openai responses chat fallback: stream read error",
|
||||
zap.Error(err),
|
||||
zap.String("request_id", requestID),
|
||||
)
|
||||
}
|
||||
if scan.Err != nil {
|
||||
return &OpenAIForwardResult{
|
||||
RequestID: requestID,
|
||||
Usage: usage,
|
||||
Usage: scan.Usage,
|
||||
Model: originalModel,
|
||||
BillingModel: billingModel,
|
||||
UpstreamModel: upstreamModel,
|
||||
@@ -372,8 +197,8 @@ func (s *OpenAIGatewayService) streamChatCompletionsAsResponses(
|
||||
ServiceTier: serviceTier,
|
||||
Stream: true,
|
||||
Duration: time.Since(startTime),
|
||||
FirstTokenMs: firstTokenMs,
|
||||
}, fmt.Errorf("stream usage incomplete: %w", err)
|
||||
FirstTokenMs: scan.FirstTokenMs,
|
||||
}, fmt.Errorf("stream usage incomplete: %w", scan.Err)
|
||||
}
|
||||
|
||||
writeEvents(apicompat.FinalizeChatCompletionsResponsesStream(state))
|
||||
@@ -386,15 +211,13 @@ func (s *OpenAIGatewayService) streamChatCompletionsAsResponses(
|
||||
c.Writer.Flush()
|
||||
}
|
||||
}
|
||||
if !sawDone {
|
||||
logger.L().Debug("openai responses chat fallback: upstream stream ended without done sentinel",
|
||||
zap.String("request_id", requestID),
|
||||
)
|
||||
if !scan.SawDone {
|
||||
logCCStreamMissingDoneSentinel("openai responses chat fallback", requestID)
|
||||
}
|
||||
|
||||
return &OpenAIForwardResult{
|
||||
RequestID: requestID,
|
||||
Usage: usage,
|
||||
Usage: scan.Usage,
|
||||
Model: originalModel,
|
||||
BillingModel: billingModel,
|
||||
UpstreamModel: upstreamModel,
|
||||
@@ -402,7 +225,7 @@ func (s *OpenAIGatewayService) streamChatCompletionsAsResponses(
|
||||
ServiceTier: serviceTier,
|
||||
Stream: true,
|
||||
Duration: time.Since(startTime),
|
||||
FirstTokenMs: firstTokenMs,
|
||||
FirstTokenMs: scan.FirstTokenMs,
|
||||
}, nil
|
||||
}
|
||||
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,658 @@
|
||||
package service
|
||||
|
||||
// 本文件由 openai_gateway_service.go 纯移动拆分而来：用量记录、计费成本计算与
|
||||
// Codex 用量快照。仅做代码搬迁，无任何行为变更。
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/Wei-Shaw/sub2api/internal/config"
|
||||
"github.com/Wei-Shaw/sub2api/internal/pkg/logger"
|
||||
"github.com/Wei-Shaw/sub2api/internal/pkg/timezone"
|
||||
"go.uber.org/zap"
|
||||
)
|
||||
|
||||
// OpenAIRecordUsageInput input for recording usage
|
||||
type OpenAIRecordUsageInput struct {
|
||||
Result *OpenAIForwardResult
|
||||
APIKey *APIKey
|
||||
User *User
|
||||
Account *Account
|
||||
Subscription *UserSubscription
|
||||
InboundEndpoint string
|
||||
UpstreamEndpoint string
|
||||
UserAgent string // 请求的 User-Agent
|
||||
IPAddress string // 请求的客户端 IP 地址
|
||||
RequestPayloadHash string
|
||||
APIKeyService APIKeyQuotaUpdater
|
||||
QuotaPlatform string // user×platform quota platform resolved by the handler before async billing.
|
||||
// CyberBlocked 为 true 时把该用量行标记为 cyber（request_type=cyber），计费逻辑不变。
|
||||
CyberBlocked bool
|
||||
ChannelUsageFields
|
||||
}
|
||||
|
||||
// CyberPolicyUsageInput 是 cyber 拒绝、未走正常 RecordUsage 的请求记录用量的入参。
|
||||
// 用量按上游真实 token 计费，与 WS cyber 及正常请求口径一致（InputTokens/OutputTokens
|
||||
// 取自上游 response.failed 报告的 usage，即 mark.UpstreamInTok/OutTok）。
|
||||
type CyberPolicyUsageInput struct {
|
||||
APIKey *APIKey
|
||||
Account *Account
|
||||
Subscription *UserSubscription
|
||||
RequestID string
|
||||
Model string
|
||||
Stream bool
|
||||
InputTokens int
|
||||
OutputTokens int
|
||||
// 渠道归因与请求级 meta，使 cyber 计费行与正常 RecordUsage 行口径一致
|
||||
// （否则 cyber 行 channel_id 等为空，渠道维度统计会遗漏 cyber 命中）。
|
||||
InboundEndpoint string
|
||||
UpstreamEndpoint string
|
||||
UserAgent string
|
||||
IPAddress string
|
||||
RequestPayloadHash string
|
||||
APIKeyService APIKeyQuotaUpdater
|
||||
ChannelUsageFields
|
||||
}
|
||||
|
||||
// RecordCyberPolicyUsageLog 为被上游 cyber_policy 拒绝、未走正常 RecordUsage 的请求
|
||||
// （HTTP forward 返回错误路径）记录用量并按上游真实 token 计费，使其与 WS cyber 路径、
|
||||
// 与正常请求的计费口径统一（不再是 tokens=0 免费行）。token 取自上游 response.failed
|
||||
// 报告的 usage（非流式直接拒通常为 0，cost 随之为 0）。复用 RecordUsage 完成成本计算、
|
||||
// 扣费与用量行写入（request_type=cyber 由 CyberBlocked 置位）。仅 forward 返回错误的
|
||||
// 路径由 handler 调用，避免与成功路径的正常 RecordUsage 重复。
|
||||
func (s *OpenAIGatewayService) RecordCyberPolicyUsageLog(ctx context.Context, in CyberPolicyUsageInput) {
|
||||
if s == nil || in.APIKey == nil || in.APIKey.User == nil || in.Account == nil || strings.TrimSpace(in.Model) == "" {
|
||||
return
|
||||
}
|
||||
result := &OpenAIForwardResult{
|
||||
RequestID: in.RequestID,
|
||||
Model: in.Model,
|
||||
Stream: in.Stream,
|
||||
Usage: OpenAIUsage{
|
||||
InputTokens: in.InputTokens,
|
||||
OutputTokens: in.OutputTokens,
|
||||
},
|
||||
}
|
||||
if err := s.RecordUsage(ctx, &OpenAIRecordUsageInput{
|
||||
Result: result,
|
||||
APIKey: in.APIKey,
|
||||
User: in.APIKey.User,
|
||||
Account: in.Account,
|
||||
Subscription: in.Subscription,
|
||||
InboundEndpoint: in.InboundEndpoint,
|
||||
UpstreamEndpoint: in.UpstreamEndpoint,
|
||||
UserAgent: in.UserAgent,
|
||||
IPAddress: in.IPAddress,
|
||||
RequestPayloadHash: in.RequestPayloadHash,
|
||||
APIKeyService: in.APIKeyService,
|
||||
ChannelUsageFields: in.ChannelUsageFields,
|
||||
CyberBlocked: true,
|
||||
}); err != nil {
|
||||
logger.LegacyPrintf("service.openai_gateway", "cyber usage record failed: request_id=%s err=%v", in.RequestID, err)
|
||||
}
|
||||
}
|
||||
|
||||
// RecordUsage records usage and deducts balance
|
||||
func (s *OpenAIGatewayService) RecordUsage(ctx context.Context, input *OpenAIRecordUsageInput) error {
|
||||
if input == nil {
|
||||
return errors.New("openai usage input is nil")
|
||||
}
|
||||
result := input.Result
|
||||
if result == nil {
|
||||
return errors.New("openai usage result is nil")
|
||||
}
|
||||
if s.rateLimitService != nil && input.Account != nil && input.Account.Platform == PlatformOpenAI {
|
||||
s.rateLimitService.ResetOpenAI403Counter(ctx, input.Account.ID)
|
||||
}
|
||||
|
||||
apiKey := input.APIKey
|
||||
user := input.User
|
||||
account := input.Account
|
||||
subscription := input.Subscription
|
||||
ApplyOpenAIImageBillingResolution(result)
|
||||
|
||||
// 计算实际的新输入token（减去缓存读取的token）
|
||||
// 因为 input_tokens 包含了 cache_read_tokens，而缓存读取的token不应按输入价格计费
|
||||
actualInputTokens := result.Usage.InputTokens - result.Usage.CacheReadInputTokens
|
||||
if actualInputTokens < 0 {
|
||||
actualInputTokens = 0
|
||||
}
|
||||
|
||||
// Calculate cost
|
||||
tokens := UsageTokens{
|
||||
InputTokens: actualInputTokens,
|
||||
ImageInputTokens: result.Usage.ImageInputTokens,
|
||||
OutputTokens: result.Usage.OutputTokens,
|
||||
CacheCreationTokens: result.Usage.CacheCreationInputTokens,
|
||||
CacheReadTokens: result.Usage.CacheReadInputTokens,
|
||||
ImageOutputTokens: result.Usage.ImageOutputTokens,
|
||||
}
|
||||
|
||||
// Get rate multiplier
|
||||
multiplier := 1.0
|
||||
if s.cfg != nil {
|
||||
multiplier = s.cfg.Default.RateMultiplier
|
||||
}
|
||||
if apiKey.GroupID != nil && apiKey.Group != nil {
|
||||
resolver := s.userGroupRateResolver
|
||||
if resolver == nil {
|
||||
resolver = newUserGroupRateResolver(nil, nil, resolveUserGroupRateCacheTTL(s.cfg), nil, "service.openai_gateway")
|
||||
}
|
||||
multiplier = resolver.Resolve(ctx, user.ID, *apiKey.GroupID, apiKey.Group.RateMultiplier)
|
||||
}
|
||||
// token 倍率叠加高峰因子（token 计费含图片 token，图片按次倍率不受影响）。高峰因子按请求时刻现算，
|
||||
// 不并入上面的 Resolve，以免污染 user:group 倍率缓存。
|
||||
multiplier, imageMultiplier := computePeakAwareMultipliers(apiKey, multiplier, timezone.Now())
|
||||
|
||||
var cost *CostBreakdown
|
||||
var err error
|
||||
billingModel := forwardResultBillingModel(result.Model, result.UpstreamModel)
|
||||
if result.BillingModel != "" {
|
||||
billingModel = strings.TrimSpace(result.BillingModel)
|
||||
}
|
||||
if input.BillingModelSource == BillingModelSourceChannelMapped && input.ChannelMappedModel != "" && input.ChannelMappedModel != input.OriginalModel {
|
||||
billingModel = input.ChannelMappedModel
|
||||
}
|
||||
if input.BillingModelSource == BillingModelSourceRequested && input.OriginalModel != "" {
|
||||
billingModel = input.OriginalModel
|
||||
}
|
||||
billingModels := usageBillingModelCandidates(
|
||||
billingModel,
|
||||
result.BillingModel,
|
||||
input.ChannelMappedModel,
|
||||
input.OriginalModel,
|
||||
result.UpstreamModel,
|
||||
result.Model,
|
||||
)
|
||||
serviceTier := ""
|
||||
if result.ServiceTier != nil {
|
||||
serviceTier = strings.TrimSpace(*result.ServiceTier)
|
||||
}
|
||||
cost, err = s.calculateOpenAIRecordUsageCost(ctx, result, apiKey, billingModels, multiplier, imageMultiplier, tokens, serviceTier)
|
||||
if err != nil {
|
||||
if !isUsagePricingUnavailableError(err) {
|
||||
return err
|
||||
}
|
||||
logger.L().With(
|
||||
zap.String("component", "service.openai_gateway"),
|
||||
zap.Strings("billing_models", billingModels),
|
||||
zap.String("requested_model", input.OriginalModel),
|
||||
zap.String("mapped_model", input.ChannelMappedModel),
|
||||
zap.String("upstream_model", result.UpstreamModel),
|
||||
zap.Int64("api_key_id", apiKey.ID),
|
||||
zap.Int64("account_id", account.ID),
|
||||
).Warn("openai_usage.pricing_missing_record_zero_cost", zap.Error(err))
|
||||
cost = &CostBreakdown{BillingMode: string(BillingModeToken)}
|
||||
}
|
||||
|
||||
// Determine billing type
|
||||
isSubscriptionBilling := subscription != nil && apiKey.Group != nil && apiKey.Group.IsSubscriptionType()
|
||||
billingType := BillingTypeBalance
|
||||
if isSubscriptionBilling {
|
||||
billingType = BillingTypeSubscription
|
||||
}
|
||||
|
||||
// Create usage log
|
||||
durationMs := int(result.Duration.Milliseconds())
|
||||
accountRateMultiplier := account.BillingRateMultiplier()
|
||||
requestID := resolveUsageBillingRequestID(ctx, result.RequestID)
|
||||
if result.OpenAIWSMode {
|
||||
if upstreamRequestID := strings.TrimSpace(result.RequestID); upstreamRequestID != "" {
|
||||
requestID = upstreamRequestID
|
||||
}
|
||||
}
|
||||
|
||||
// 确定 RequestedModel（渠道映射前的原始模型）
|
||||
requestedModel := result.Model
|
||||
if input.OriginalModel != "" {
|
||||
requestedModel = input.OriginalModel
|
||||
}
|
||||
|
||||
usageLog := &UsageLog{
|
||||
UserID: user.ID,
|
||||
APIKeyID: apiKey.ID,
|
||||
AccountID: account.ID,
|
||||
RequestID: requestID,
|
||||
Model: result.Model,
|
||||
RequestedModel: requestedModel,
|
||||
UpstreamModel: optionalNonEqualStringPtr(result.UpstreamModel, result.Model),
|
||||
ServiceTier: result.ServiceTier,
|
||||
ReasoningEffort: result.ReasoningEffort,
|
||||
InboundEndpoint: optionalTrimmedStringPtr(input.InboundEndpoint),
|
||||
UpstreamEndpoint: optionalTrimmedStringPtr(input.UpstreamEndpoint),
|
||||
InputTokens: actualInputTokens,
|
||||
OutputTokens: result.Usage.OutputTokens,
|
||||
CacheCreationTokens: result.Usage.CacheCreationInputTokens,
|
||||
CacheReadTokens: result.Usage.CacheReadInputTokens,
|
||||
ImageOutputTokens: result.Usage.ImageOutputTokens,
|
||||
ImageCount: result.ImageCount,
|
||||
ImageSize: optionalTrimmedStringPtr(result.ImageSize),
|
||||
ImageInputSize: optionalTrimmedStringPtr(result.ImageInputSize),
|
||||
ImageOutputSize: optionalTrimmedStringPtr(result.ImageOutputSize),
|
||||
ImageSizeSource: optionalTrimmedStringPtr(result.ImageSizeSource),
|
||||
ImageSizeBreakdown: result.ImageSizeBreakdown,
|
||||
}
|
||||
if cost != nil {
|
||||
usageLog.InputCost = cost.InputCost
|
||||
usageLog.OutputCost = cost.OutputCost
|
||||
usageLog.ImageOutputCost = cost.ImageOutputCost
|
||||
usageLog.CacheCreationCost = cost.CacheCreationCost
|
||||
usageLog.CacheReadCost = cost.CacheReadCost
|
||||
usageLog.TotalCost = cost.TotalCost
|
||||
usageLog.ActualCost = cost.ActualCost
|
||||
}
|
||||
if result.ImageCount > 0 && (cost == nil || cost.BillingMode != string(BillingModeToken)) {
|
||||
usageLog.RateMultiplier = imageMultiplier
|
||||
} else {
|
||||
usageLog.RateMultiplier = multiplier
|
||||
}
|
||||
usageLog.AccountRateMultiplier = &accountRateMultiplier
|
||||
usageLog.BillingType = billingType
|
||||
usageLog.Stream = result.Stream
|
||||
if input.CyberBlocked {
|
||||
usageLog.RequestType = RequestTypeCyberBlocked
|
||||
}
|
||||
usageLog.OpenAIWSMode = result.OpenAIWSMode
|
||||
usageLog.DurationMs = &durationMs
|
||||
usageLog.FirstTokenMs = result.FirstTokenMs
|
||||
usageLog.CreatedAt = time.Now()
|
||||
// 设置渠道信息
|
||||
usageLog.ChannelID = optionalInt64Ptr(input.ChannelID)
|
||||
usageLog.ModelMappingChain = optionalTrimmedStringPtr(input.ModelMappingChain)
|
||||
// 设置计费模式
|
||||
if cost != nil && cost.BillingMode != "" {
|
||||
billingMode := cost.BillingMode
|
||||
usageLog.BillingMode = &billingMode
|
||||
} else if result.ImageCount > 0 {
|
||||
billingMode := string(BillingModeImage)
|
||||
usageLog.BillingMode = &billingMode
|
||||
} else {
|
||||
billingMode := string(BillingModeToken)
|
||||
usageLog.BillingMode = &billingMode
|
||||
}
|
||||
// 添加 UserAgent
|
||||
if input.UserAgent != "" {
|
||||
usageLog.UserAgent = &input.UserAgent
|
||||
}
|
||||
|
||||
// 添加 IPAddress
|
||||
if input.IPAddress != "" {
|
||||
usageLog.IPAddress = &input.IPAddress
|
||||
}
|
||||
|
||||
if apiKey.GroupID != nil {
|
||||
usageLog.GroupID = apiKey.GroupID
|
||||
}
|
||||
if subscription != nil {
|
||||
usageLog.SubscriptionID = &subscription.ID
|
||||
}
|
||||
|
||||
// 计算账号统计定价费用（使用最终上游模型匹配自定义规则）
|
||||
if apiKey.GroupID != nil {
|
||||
applyAccountStatsCost(ctx, usageLog, s.channelService, s.billingService,
|
||||
account.ID, *apiKey.GroupID, result.UpstreamModel, result.Model,
|
||||
tokens, cost.TotalCost,
|
||||
)
|
||||
}
|
||||
|
||||
if s.cfg != nil && s.cfg.RunMode == config.RunModeSimple {
|
||||
writeUsageLogBestEffort(ctx, s.usageLogRepo, usageLog, "service.openai_gateway")
|
||||
logger.LegacyPrintf("service.openai_gateway", "[SIMPLE MODE] Usage recorded (not billed): user=%d, tokens=%d", usageLog.UserID, usageLog.TotalTokens())
|
||||
s.deferredService.ScheduleLastUsedUpdate(account.ID)
|
||||
return nil
|
||||
}
|
||||
|
||||
// Async usage billing runs outside the original request context, so it
|
||||
// cannot recover ForcePlatform there. Fall back for internal/test callers.
|
||||
quotaPlatform := input.QuotaPlatform
|
||||
if quotaPlatform == "" {
|
||||
quotaPlatform = PlatformFromAPIKey(apiKey)
|
||||
}
|
||||
|
||||
billingErr := func() error {
|
||||
_, err := applyUsageBilling(ctx, requestID, usageLog, &postUsageBillingParams{
|
||||
Cost: cost,
|
||||
User: user,
|
||||
APIKey: apiKey,
|
||||
Account: account,
|
||||
Subscription: subscription,
|
||||
RequestPayloadHash: resolveUsageBillingPayloadFingerprint(ctx, input.RequestPayloadHash),
|
||||
IsSubscriptionBill: isSubscriptionBilling,
|
||||
AccountRateMultiplier: accountRateMultiplier,
|
||||
APIKeyService: input.APIKeyService,
|
||||
Platform: quotaPlatform,
|
||||
}, s.billingDeps(), s.usageBillingRepo)
|
||||
return err
|
||||
}()
|
||||
|
||||
if billingErr != nil {
|
||||
return billingErr
|
||||
}
|
||||
writeUsageLogBestEffort(ctx, s.usageLogRepo, usageLog, "service.openai_gateway")
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *OpenAIGatewayService) calculateOpenAIRecordUsageCost(
|
||||
ctx context.Context,
|
||||
result *OpenAIForwardResult,
|
||||
apiKey *APIKey,
|
||||
billingModels []string,
|
||||
multiplier float64,
|
||||
imageMultiplier float64,
|
||||
tokens UsageTokens,
|
||||
serviceTier string,
|
||||
) (*CostBreakdown, error) {
|
||||
billingModel := firstUsageBillingModel(billingModels)
|
||||
if result != nil && result.ImageCount > 0 {
|
||||
// 渠道定价为 token 计费时走 token 路径，否则走图片计费
|
||||
if resolved := s.resolveOpenAIChannelPricing(ctx, billingModel, apiKey); resolved == nil || resolved.Mode != BillingModeToken {
|
||||
return s.calculateOpenAIImageCost(ctx, billingModel, apiKey, result, imageMultiplier), nil
|
||||
}
|
||||
}
|
||||
if len(billingModels) == 0 || billingModel == "" {
|
||||
return nil, errors.New("openai usage billing model is empty")
|
||||
}
|
||||
var lastErr error
|
||||
for _, candidate := range billingModels {
|
||||
candidate = strings.TrimSpace(candidate)
|
||||
if candidate == "" {
|
||||
continue
|
||||
}
|
||||
cost, err := s.calculateOpenAIRecordUsageTokenCost(ctx, apiKey, candidate, multiplier, tokens, serviceTier)
|
||||
if err == nil {
|
||||
return cost, nil
|
||||
}
|
||||
lastErr = err
|
||||
}
|
||||
if lastErr == nil {
|
||||
lastErr = errors.New("no non-empty billing model candidates")
|
||||
}
|
||||
return nil, fmt.Errorf("calculate OpenAI usage cost failed for billing models %s: %w", strings.Join(billingModels, ","), lastErr)
|
||||
}
|
||||
|
||||
func isUsagePricingUnavailableError(err error) bool {
|
||||
if err == nil {
|
||||
return false
|
||||
}
|
||||
if errors.Is(err, ErrModelPricingUnavailable) {
|
||||
return true
|
||||
}
|
||||
msg := strings.ToLower(err.Error())
|
||||
return strings.Contains(msg, "no pricing available") || strings.Contains(msg, "pricing not found")
|
||||
}
|
||||
|
||||
func (s *OpenAIGatewayService) calculateOpenAIRecordUsageTokenCost(
|
||||
ctx context.Context,
|
||||
apiKey *APIKey,
|
||||
billingModel string,
|
||||
multiplier float64,
|
||||
tokens UsageTokens,
|
||||
serviceTier string,
|
||||
) (*CostBreakdown, error) {
|
||||
if s.resolver != nil && apiKey.Group != nil {
|
||||
gid := apiKey.Group.ID
|
||||
return s.billingService.CalculateCostUnified(CostInput{
|
||||
Ctx: ctx,
|
||||
Model: billingModel,
|
||||
GroupID: &gid,
|
||||
Tokens: tokens,
|
||||
RequestCount: 1,
|
||||
RateMultiplier: multiplier,
|
||||
ServiceTier: serviceTier,
|
||||
Resolver: s.resolver,
|
||||
})
|
||||
}
|
||||
return s.billingService.CalculateCostWithServiceTier(billingModel, tokens, multiplier, serviceTier)
|
||||
}
|
||||
|
||||
func (s *OpenAIGatewayService) calculateOpenAIImageCost(
|
||||
ctx context.Context,
|
||||
billingModel string,
|
||||
apiKey *APIKey,
|
||||
result *OpenAIForwardResult,
|
||||
multiplier float64,
|
||||
) *CostBreakdown {
|
||||
sizeTier := NormalizeImageBillingTierOrDefault(result.ImageSize)
|
||||
if resolved := s.resolveOpenAIChannelPricing(ctx, billingModel, apiKey); resolved != nil &&
|
||||
(resolved.Mode == BillingModePerRequest || resolved.Mode == BillingModeImage) {
|
||||
gid := apiKey.Group.ID
|
||||
cost, err := s.billingService.CalculateCostUnified(CostInput{
|
||||
Ctx: ctx,
|
||||
Model: billingModel,
|
||||
GroupID: &gid,
|
||||
RequestCount: result.ImageCount,
|
||||
SizeTier: sizeTier,
|
||||
RateMultiplier: multiplier,
|
||||
Resolver: s.resolver,
|
||||
Resolved: resolved,
|
||||
})
|
||||
if err == nil {
|
||||
return cost
|
||||
}
|
||||
logger.LegacyPrintf("service.openai_gateway", "Calculate image channel cost failed: %v", err)
|
||||
}
|
||||
|
||||
var groupConfig *ImagePriceConfig
|
||||
if apiKey != nil && apiKey.Group != nil {
|
||||
groupConfig = &ImagePriceConfig{
|
||||
Price1K: apiKey.Group.ImagePrice1K,
|
||||
Price2K: apiKey.Group.ImagePrice2K,
|
||||
Price4K: apiKey.Group.ImagePrice4K,
|
||||
}
|
||||
}
|
||||
return s.billingService.CalculateImageCost(billingModel, sizeTier, result.ImageCount, groupConfig, multiplier)
|
||||
}
|
||||
|
||||
func (s *OpenAIGatewayService) resolveOpenAIChannelPricing(ctx context.Context, billingModel string, apiKey *APIKey) *ResolvedPricing {
|
||||
if s.resolver == nil || apiKey == nil || apiKey.Group == nil {
|
||||
return nil
|
||||
}
|
||||
gid := apiKey.Group.ID
|
||||
resolved := s.resolver.Resolve(ctx, PricingInput{Model: billingModel, GroupID: &gid})
|
||||
if resolved.Source == PricingSourceChannel {
|
||||
return resolved
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// ParseCodexRateLimitHeaders extracts Codex usage limits from response headers.
|
||||
// Exported for use in ratelimit_service when handling OpenAI 429 responses.
|
||||
func ParseCodexRateLimitHeaders(headers http.Header) *OpenAICodexUsageSnapshot {
|
||||
snapshot := &OpenAICodexUsageSnapshot{}
|
||||
hasData := false
|
||||
|
||||
// Helper to parse float64 from header
|
||||
parseFloat := func(key string) *float64 {
|
||||
if v := headers.Get(key); v != "" {
|
||||
if f, err := strconv.ParseFloat(v, 64); err == nil {
|
||||
return &f
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// Helper to parse int from header
|
||||
parseInt := func(key string) *int {
|
||||
if v := headers.Get(key); v != "" {
|
||||
if i, err := strconv.Atoi(v); err == nil {
|
||||
return &i
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// Primary (weekly) limits
|
||||
if v := parseFloat("x-codex-primary-used-percent"); v != nil {
|
||||
snapshot.PrimaryUsedPercent = v
|
||||
hasData = true
|
||||
}
|
||||
if v := parseInt("x-codex-primary-reset-after-seconds"); v != nil {
|
||||
snapshot.PrimaryResetAfterSeconds = v
|
||||
hasData = true
|
||||
}
|
||||
if v := parseInt("x-codex-primary-window-minutes"); v != nil {
|
||||
snapshot.PrimaryWindowMinutes = v
|
||||
hasData = true
|
||||
}
|
||||
|
||||
// Secondary (5h) limits
|
||||
if v := parseFloat("x-codex-secondary-used-percent"); v != nil {
|
||||
snapshot.SecondaryUsedPercent = v
|
||||
hasData = true
|
||||
}
|
||||
if v := parseInt("x-codex-secondary-reset-after-seconds"); v != nil {
|
||||
snapshot.SecondaryResetAfterSeconds = v
|
||||
hasData = true
|
||||
}
|
||||
if v := parseInt("x-codex-secondary-window-minutes"); v != nil {
|
||||
snapshot.SecondaryWindowMinutes = v
|
||||
hasData = true
|
||||
}
|
||||
|
||||
// Overflow ratio
|
||||
if v := parseFloat("x-codex-primary-over-secondary-limit-percent"); v != nil {
|
||||
snapshot.PrimaryOverSecondaryPercent = v
|
||||
hasData = true
|
||||
}
|
||||
|
||||
if !hasData {
|
||||
return nil
|
||||
}
|
||||
|
||||
snapshot.UpdatedAt = time.Now().Format(time.RFC3339)
|
||||
return snapshot
|
||||
}
|
||||
|
||||
func codexSnapshotBaseTime(snapshot *OpenAICodexUsageSnapshot, fallback time.Time) time.Time {
|
||||
if snapshot == nil {
|
||||
return fallback
|
||||
}
|
||||
if snapshot.UpdatedAt == "" {
|
||||
return fallback
|
||||
}
|
||||
base, err := time.Parse(time.RFC3339, snapshot.UpdatedAt)
|
||||
if err != nil {
|
||||
return fallback
|
||||
}
|
||||
return base
|
||||
}
|
||||
|
||||
func codexResetAtRFC3339(base time.Time, resetAfterSeconds *int) *string {
|
||||
if resetAfterSeconds == nil {
|
||||
return nil
|
||||
}
|
||||
sec := *resetAfterSeconds
|
||||
if sec < 0 {
|
||||
sec = 0
|
||||
}
|
||||
resetAt := base.Add(time.Duration(sec) * time.Second).Format(time.RFC3339)
|
||||
return &resetAt
|
||||
}
|
||||
|
||||
func buildCodexUsageExtraUpdates(snapshot *OpenAICodexUsageSnapshot, fallbackNow time.Time) map[string]any {
|
||||
if snapshot == nil {
|
||||
return nil
|
||||
}
|
||||
|
||||
baseTime := codexSnapshotBaseTime(snapshot, fallbackNow)
|
||||
updates := make(map[string]any)
|
||||
|
||||
// 保存原始 primary/secondary 字段，便于排查问题
|
||||
if snapshot.PrimaryUsedPercent != nil {
|
||||
updates["codex_primary_used_percent"] = *snapshot.PrimaryUsedPercent
|
||||
}
|
||||
if snapshot.PrimaryResetAfterSeconds != nil {
|
||||
updates["codex_primary_reset_after_seconds"] = *snapshot.PrimaryResetAfterSeconds
|
||||
}
|
||||
if snapshot.PrimaryWindowMinutes != nil {
|
||||
updates["codex_primary_window_minutes"] = *snapshot.PrimaryWindowMinutes
|
||||
}
|
||||
if snapshot.SecondaryUsedPercent != nil {
|
||||
updates["codex_secondary_used_percent"] = *snapshot.SecondaryUsedPercent
|
||||
}
|
||||
if snapshot.SecondaryResetAfterSeconds != nil {
|
||||
updates["codex_secondary_reset_after_seconds"] = *snapshot.SecondaryResetAfterSeconds
|
||||
}
|
||||
if snapshot.SecondaryWindowMinutes != nil {
|
||||
updates["codex_secondary_window_minutes"] = *snapshot.SecondaryWindowMinutes
|
||||
}
|
||||
if snapshot.PrimaryOverSecondaryPercent != nil {
|
||||
updates["codex_primary_over_secondary_percent"] = *snapshot.PrimaryOverSecondaryPercent
|
||||
}
|
||||
updates["codex_usage_updated_at"] = baseTime.Format(time.RFC3339)
|
||||
|
||||
// 归一化到 5h/7d 规范字段
|
||||
if normalized := snapshot.Normalize(); normalized != nil {
|
||||
if normalized.Used5hPercent != nil {
|
||||
updates["codex_5h_used_percent"] = *normalized.Used5hPercent
|
||||
}
|
||||
if normalized.Reset5hSeconds != nil {
|
||||
updates["codex_5h_reset_after_seconds"] = *normalized.Reset5hSeconds
|
||||
}
|
||||
if normalized.Window5hMinutes != nil {
|
||||
updates["codex_5h_window_minutes"] = *normalized.Window5hMinutes
|
||||
}
|
||||
if normalized.Used7dPercent != nil {
|
||||
updates["codex_7d_used_percent"] = *normalized.Used7dPercent
|
||||
}
|
||||
if normalized.Reset7dSeconds != nil {
|
||||
updates["codex_7d_reset_after_seconds"] = *normalized.Reset7dSeconds
|
||||
}
|
||||
if normalized.Window7dMinutes != nil {
|
||||
updates["codex_7d_window_minutes"] = *normalized.Window7dMinutes
|
||||
}
|
||||
if reset5hAt := codexResetAtRFC3339(baseTime, normalized.Reset5hSeconds); reset5hAt != nil {
|
||||
updates["codex_5h_reset_at"] = *reset5hAt
|
||||
}
|
||||
if reset7dAt := codexResetAtRFC3339(baseTime, normalized.Reset7dSeconds); reset7dAt != nil {
|
||||
updates["codex_7d_reset_at"] = *reset7dAt
|
||||
}
|
||||
}
|
||||
|
||||
return updates
|
||||
}
|
||||
|
||||
// updateCodexUsageSnapshot saves the Codex usage snapshot to account's Extra field
|
||||
// updateCodexUsageSnapshot 把 /responses 的 x-codex-* 全局头快照写入账号 codex_* Extra。
|
||||
// ⚠️ 调用方必须排除 spark 影子账号(account.IsShadow()):影子的 codex_* 仅由 QueryUsage
|
||||
// (/wham/usage bengalfox 道)更新,不能被全局头口径污染(外审第7轮 P1)。本函数仅持 accountID,
|
||||
// 无法在此自检影子,故守卫前置到各调用点。
|
||||
func (s *OpenAIGatewayService) updateCodexUsageSnapshot(ctx context.Context, accountID int64, snapshot *OpenAICodexUsageSnapshot) {
|
||||
if snapshot == nil {
|
||||
return
|
||||
}
|
||||
if s == nil || s.accountRepo == nil {
|
||||
return
|
||||
}
|
||||
|
||||
now := time.Now()
|
||||
updates := buildCodexUsageExtraUpdates(snapshot, now)
|
||||
if len(updates) == 0 {
|
||||
return
|
||||
}
|
||||
if !s.getCodexSnapshotThrottle().Allow(accountID, now) {
|
||||
return
|
||||
}
|
||||
|
||||
go func() {
|
||||
updateCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
_ = s.accountRepo.UpdateExtra(updateCtx, accountID, updates)
|
||||
}()
|
||||
}
|
||||
|
||||
func (s *OpenAIGatewayService) UpdateCodexUsageSnapshotFromHeaders(ctx context.Context, accountID int64, headers http.Header) {
|
||||
if accountID <= 0 || headers == nil {
|
||||
return
|
||||
}
|
||||
if snapshot := ParseCodexRateLimitHeaders(headers); snapshot != nil {
|
||||
s.updateCodexUsageSnapshot(ctx, accountID, snapshot)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user