"backend/cmd/vscode:/vscode.git/clone" did not exist on "a43da6225449c68f486b796139a4144c9fbe24fc"
Commit 6ac8ccde authored by erio's avatar erio
Browse files

fix: merge 30 general improvements from release branch

Bug fixes:
- Detached context for GetAccountConcurrencyBatch (prevent all-zero on request cancel)
- Filter soft-deleted users in GetByGroupID
- Stripe CSP policy (allow Stripe.js in script-src and frame-src)
- WebSearch API key validation on save
- RECHARGING status in payment result success check
- Windows test fixes (logger Sync deadlock, config path escaping)

Feature enhancements:
- Webhook multi-instance dispatch (extractOutTradeNo + GetWebhookProvider)
- EasyPay mobile H5 payment (device param + PayURL2)
- SSE error propagation in WebSearch emulation
- AccountStatsCost DTO field for admin usage logs
- Plans sort by sort_order instead of created_at
- UsageMapHook for streaming response usage data
- apicompat Instructions field passthrough
- EffectiveLoadFactor for ops concurrency/metrics
- Usage billing RETURNING balance for notify system
- BulkUpdate mixed channel warning with details
- println to slog migration in auth cache
- Wire ProviderSet cleanup
- CI cache-dependency-path optimization

Frontend:
- Refund eligibility check per provider (canRequestRefund)
- Plan sort_order editing
- Dead code cleanup (simulate_claude_max, client_affinity)
- GroupsView platform switch guard
- channels features_config API type
- UsageView account_stats_cost export
parent f1297a36
...@@ -17,6 +17,7 @@ jobs: ...@@ -17,6 +17,7 @@ jobs:
go-version-file: backend/go.mod go-version-file: backend/go.mod
check-latest: false check-latest: false
cache: true cache: true
cache-dependency-path: backend/go.sum
- name: Verify Go version - name: Verify Go version
run: | run: |
go version | grep -q 'go1.26.2' go version | grep -q 'go1.26.2'
...@@ -36,6 +37,7 @@ jobs: ...@@ -36,6 +37,7 @@ jobs:
go-version-file: backend/go.mod go-version-file: backend/go.mod
check-latest: false check-latest: false
cache: true cache: true
cache-dependency-path: backend/go.sum
- name: Verify Go version - name: Verify Go version
run: | run: |
go version | grep -q 'go1.26.2' go version | grep -q 'go1.26.2'
......
...@@ -36,19 +36,13 @@ func initializeApplication(buildInfo handler.BuildInfo) (*Application, error) { ...@@ -36,19 +36,13 @@ func initializeApplication(buildInfo handler.BuildInfo) (*Application, error) {
// Business layer ProviderSets // Business layer ProviderSets
repository.ProviderSet, repository.ProviderSet,
service.ProviderSet, service.ProviderSet,
payment.ProviderSet,
middleware.ProviderSet, middleware.ProviderSet,
handler.ProviderSet, handler.ProviderSet,
// Server layer ProviderSet // Server layer ProviderSet
server.ProviderSet, server.ProviderSet,
// Payment providers
payment.ProvideRegistry,
payment.ProvideEncryptionKey,
payment.ProvideDefaultLoadBalancer,
service.ProvidePaymentConfigService,
service.ProvidePaymentOrderExpiryService,
// Privacy client factory for OpenAI training opt-out // Privacy client factory for OpenAI training opt-out
providePrivacyClientFactory, providePrivacyClientFactory,
......
...@@ -28,7 +28,7 @@ const ( ...@@ -28,7 +28,7 @@ const (
// DefaultCSPPolicy is the default Content-Security-Policy with nonce support // DefaultCSPPolicy is the default Content-Security-Policy with nonce support
// __CSP_NONCE__ will be replaced with actual nonce at request time by the SecurityHeaders middleware // __CSP_NONCE__ will be replaced with actual nonce at request time by the SecurityHeaders middleware
const DefaultCSPPolicy = "default-src 'self'; script-src 'self' __CSP_NONCE__ https://challenges.cloudflare.com https://static.cloudflareinsights.com; style-src 'self' 'unsafe-inline' https://fonts.googleapis.com; img-src 'self' data: https:; font-src 'self' data: https://fonts.gstatic.com; connect-src 'self' https:; frame-src https://challenges.cloudflare.com; frame-ancestors 'none'; base-uri 'self'; form-action 'self'" const DefaultCSPPolicy = "default-src 'self'; script-src 'self' __CSP_NONCE__ https://challenges.cloudflare.com https://static.cloudflareinsights.com https://*.stripe.com; style-src 'self' 'unsafe-inline' https://fonts.googleapis.com; img-src 'self' data: https:; font-src 'self' data: https://fonts.gstatic.com; connect-src 'self' https:; frame-src https://challenges.cloudflare.com https://*.stripe.com; frame-ancestors 'none'; base-uri 'self'; form-action 'self'"
// UMQ(用户消息队列)模式常量 // UMQ(用户消息队列)模式常量
const ( const (
......
...@@ -233,12 +233,13 @@ func TestLoadForcedCodexInstructionsTemplate(t *testing.T) { ...@@ -233,12 +233,13 @@ func TestLoadForcedCodexInstructionsTemplate(t *testing.T) {
configPath := filepath.Join(tempDir, "config.yaml") configPath := filepath.Join(tempDir, "config.yaml")
require.NoError(t, os.WriteFile(templatePath, []byte("server-prefix\n\n{{ .ExistingInstructions }}"), 0o644)) require.NoError(t, os.WriteFile(templatePath, []byte("server-prefix\n\n{{ .ExistingInstructions }}"), 0o644))
require.NoError(t, os.WriteFile(configPath, []byte("gateway:\n forced_codex_instructions_template_file: \""+templatePath+"\"\n"), 0o644)) yamlSafePath := filepath.ToSlash(templatePath)
require.NoError(t, os.WriteFile(configPath, []byte("gateway:\n forced_codex_instructions_template_file: \""+yamlSafePath+"\"\n"), 0o644))
t.Setenv("DATA_DIR", tempDir) t.Setenv("DATA_DIR", tempDir)
cfg, err := Load() cfg, err := Load()
require.NoError(t, err) require.NoError(t, err)
require.Equal(t, templatePath, cfg.Gateway.ForcedCodexInstructionsTemplateFile) require.Equal(t, yamlSafePath, cfg.Gateway.ForcedCodexInstructionsTemplateFile)
require.Equal(t, "server-prefix\n\n{{ .ExistingInstructions }}", cfg.Gateway.ForcedCodexInstructionsTemplate) require.Equal(t, "server-prefix\n\n{{ .ExistingInstructions }}", cfg.Gateway.ForcedCodexInstructionsTemplate)
} }
......
...@@ -1412,6 +1412,12 @@ func (h *AccountHandler) BulkUpdate(c *gin.Context) { ...@@ -1412,6 +1412,12 @@ func (h *AccountHandler) BulkUpdate(c *gin.Context) {
c.JSON(409, gin.H{ c.JSON(409, gin.H{
"error": "mixed_channel_warning", "error": "mixed_channel_warning",
"message": mixedErr.Error(), "message": mixedErr.Error(),
"details": gin.H{
"group_id": mixedErr.GroupID,
"group_name": mixedErr.GroupName,
"current_platform": mixedErr.CurrentPlatform,
"other_platform": mixedErr.OtherPlatform,
},
}) })
return return
} }
......
...@@ -628,6 +628,7 @@ func UsageLogFromServiceAdmin(l *service.UsageLog) *AdminUsageLog { ...@@ -628,6 +628,7 @@ func UsageLogFromServiceAdmin(l *service.UsageLog) *AdminUsageLog {
ModelMappingChain: l.ModelMappingChain, ModelMappingChain: l.ModelMappingChain,
BillingTier: l.BillingTier, BillingTier: l.BillingTier,
AccountRateMultiplier: l.AccountRateMultiplier, AccountRateMultiplier: l.AccountRateMultiplier,
AccountStatsCost: l.AccountStatsCost,
IPAddress: l.IPAddress, IPAddress: l.IPAddress,
Account: AccountSummaryFromService(l.Account), Account: AccountSummaryFromService(l.Account),
} }
......
...@@ -427,6 +427,8 @@ type AdminUsageLog struct { ...@@ -427,6 +427,8 @@ type AdminUsageLog struct {
// AccountRateMultiplier 账号计费倍率快照(nil 表示按 1.0 处理) // AccountRateMultiplier 账号计费倍率快照(nil 表示按 1.0 处理)
AccountRateMultiplier *float64 `json:"account_rate_multiplier"` AccountRateMultiplier *float64 `json:"account_rate_multiplier"`
// AccountStatsCost 自定义定价规则计算的账号统计费用(nil 表示使用默认公式)
AccountStatsCost *float64 `json:"account_stats_cost,omitempty"`
// IPAddress 用户请求 IP(仅管理员可见) // IPAddress 用户请求 IP(仅管理员可见)
IPAddress *string `json:"ip_address,omitempty"` IPAddress *string `json:"ip_address,omitempty"`
......
...@@ -4,6 +4,7 @@ import ( ...@@ -4,6 +4,7 @@ import (
"io" "io"
"log/slog" "log/slog"
"net/http" "net/http"
"net/url"
"strings" "strings"
"github.com/Wei-Shaw/sub2api/internal/payment" "github.com/Wei-Shaw/sub2api/internal/payment"
...@@ -72,9 +73,13 @@ func (h *PaymentWebhookHandler) handleNotify(c *gin.Context, providerKey string) ...@@ -72,9 +73,13 @@ func (h *PaymentWebhookHandler) handleNotify(c *gin.Context, providerKey string)
rawBody = string(body) rawBody = string(body)
} }
provider, err := h.registry.GetProviderByKey(providerKey) // Extract out_trade_no to look up the order's specific provider instance.
// This is needed when multiple instances of the same provider exist (e.g. multiple EasyPay accounts).
outTradeNo := extractOutTradeNo(rawBody, providerKey)
provider, err := h.paymentService.GetWebhookProvider(c.Request.Context(), providerKey, outTradeNo)
if err != nil { if err != nil {
slog.Warn("[Payment Webhook] provider not registered", "provider", providerKey, "error", err) slog.Warn("[Payment Webhook] provider not found", "provider", providerKey, "outTradeNo", outTradeNo, "error", err)
writeSuccessResponse(c, providerKey) writeSuccessResponse(c, providerKey)
return return
} }
...@@ -111,19 +116,40 @@ func (h *PaymentWebhookHandler) handleNotify(c *gin.Context, providerKey string) ...@@ -111,19 +116,40 @@ func (h *PaymentWebhookHandler) handleNotify(c *gin.Context, providerKey string)
writeSuccessResponse(c, providerKey) writeSuccessResponse(c, providerKey)
} }
// extractOutTradeNo parses the webhook body to find the out_trade_no.
// This allows looking up the correct provider instance before verification.
func extractOutTradeNo(rawBody, providerKey string) string {
switch providerKey {
case payment.TypeEasyPay:
values, err := url.ParseQuery(rawBody)
if err == nil {
return values.Get("out_trade_no")
}
}
// For other providers (Stripe, Alipay direct, WxPay direct), the registry
// typically has only one instance, so no instance lookup is needed.
return ""
}
// wxpaySuccessResponse is the JSON response expected by WeChat Pay webhook. // wxpaySuccessResponse is the JSON response expected by WeChat Pay webhook.
type wxpaySuccessResponse struct { type wxpaySuccessResponse struct {
Code string `json:"code"` Code string `json:"code"`
Message string `json:"message"` Message string `json:"message"`
} }
// WeChat Pay webhook success response constants.
const (
wxpaySuccessCode = "SUCCESS"
wxpaySuccessMessage = "成功"
)
// writeSuccessResponse sends the provider-specific success response. // writeSuccessResponse sends the provider-specific success response.
// WeChat Pay requires JSON {"code":"SUCCESS","message":"成功"}; // WeChat Pay requires JSON {"code":"SUCCESS","message":"成功"};
// Stripe expects an empty 200; others accept plain text "success". // Stripe expects an empty 200; others accept plain text "success".
func writeSuccessResponse(c *gin.Context, providerKey string) { func writeSuccessResponse(c *gin.Context, providerKey string) {
switch providerKey { switch providerKey {
case payment.TypeWxpay: case payment.TypeWxpay:
c.JSON(http.StatusOK, wxpaySuccessResponse{Code: "SUCCESS", Message: "成功"}) c.JSON(http.StatusOK, wxpaySuccessResponse{Code: wxpaySuccessCode, Message: wxpaySuccessMessage})
case payment.TypeStripe: case payment.TypeStripe:
c.String(http.StatusOK, "") c.String(http.StatusOK, "")
default: default:
......
...@@ -27,6 +27,8 @@ const ( ...@@ -27,6 +27,8 @@ const (
maxEasypayResponseSize = 1 << 20 // 1MB maxEasypayResponseSize = 1 << 20 // 1MB
tradeStatusSuccess = "TRADE_SUCCESS" tradeStatusSuccess = "TRADE_SUCCESS"
signTypeMD5 = "MD5" signTypeMD5 = "MD5"
paymentModePopup = "popup"
deviceMobile = "mobile"
) )
// EasyPay implements payment.Provider for the EasyPay aggregation platform. // EasyPay implements payment.Provider for the EasyPay aggregation platform.
...@@ -61,7 +63,7 @@ func (e *EasyPay) CreatePayment(ctx context.Context, req payment.CreatePaymentRe ...@@ -61,7 +63,7 @@ func (e *EasyPay) CreatePayment(ctx context.Context, req payment.CreatePaymentRe
// Payment mode determined by instance config, not payment type. // Payment mode determined by instance config, not payment type.
// "popup" → hosted page (submit.php); "qrcode"/default → API call (mapi.php). // "popup" → hosted page (submit.php); "qrcode"/default → API call (mapi.php).
mode := e.config["paymentMode"] mode := e.config["paymentMode"]
if mode == "popup" { if mode == paymentModePopup {
return e.createRedirectPayment(req) return e.createRedirectPayment(req)
} }
return e.createAPIPayment(ctx, req) return e.createAPIPayment(ctx, req)
...@@ -81,6 +83,9 @@ func (e *EasyPay) createRedirectPayment(req payment.CreatePaymentRequest) (*paym ...@@ -81,6 +83,9 @@ func (e *EasyPay) createRedirectPayment(req payment.CreatePaymentRequest) (*paym
if cid := e.resolveCID(req.PaymentType); cid != "" { if cid := e.resolveCID(req.PaymentType); cid != "" {
params["cid"] = cid params["cid"] = cid
} }
if req.IsMobile {
params["device"] = deviceMobile
}
params["sign"] = easyPaySign(params, e.config["pkey"]) params["sign"] = easyPaySign(params, e.config["pkey"])
params["sign_type"] = signTypeMD5 params["sign_type"] = signTypeMD5
...@@ -106,7 +111,7 @@ func (e *EasyPay) createAPIPayment(ctx context.Context, req payment.CreatePaymen ...@@ -106,7 +111,7 @@ func (e *EasyPay) createAPIPayment(ctx context.Context, req payment.CreatePaymen
params["cid"] = cid params["cid"] = cid
} }
if req.IsMobile { if req.IsMobile {
params["device"] = "mobile" params["device"] = deviceMobile
} }
params["sign"] = easyPaySign(params, e.config["pkey"]) params["sign"] = easyPaySign(params, e.config["pkey"])
params["sign_type"] = signTypeMD5 params["sign_type"] = signTypeMD5
...@@ -120,6 +125,7 @@ func (e *EasyPay) createAPIPayment(ctx context.Context, req payment.CreatePaymen ...@@ -120,6 +125,7 @@ func (e *EasyPay) createAPIPayment(ctx context.Context, req payment.CreatePaymen
Msg string `json:"msg"` Msg string `json:"msg"`
TradeNo string `json:"trade_no"` TradeNo string `json:"trade_no"`
PayURL string `json:"payurl"` PayURL string `json:"payurl"`
PayURL2 string `json:"payurl2"` // H5 mobile payment URL
QRCode string `json:"qrcode"` QRCode string `json:"qrcode"`
} }
if err := json.Unmarshal(body, &resp); err != nil { if err := json.Unmarshal(body, &resp); err != nil {
...@@ -128,7 +134,11 @@ func (e *EasyPay) createAPIPayment(ctx context.Context, req payment.CreatePaymen ...@@ -128,7 +134,11 @@ func (e *EasyPay) createAPIPayment(ctx context.Context, req payment.CreatePaymen
if resp.Code != easypayCodeSuccess { if resp.Code != easypayCodeSuccess {
return nil, fmt.Errorf("easypay error: %s", resp.Msg) return nil, fmt.Errorf("easypay error: %s", resp.Msg)
} }
return &payment.CreatePaymentResponse{TradeNo: resp.TradeNo, PayURL: resp.PayURL, QRCode: resp.QRCode}, nil payURL := resp.PayURL
if req.IsMobile && resp.PayURL2 != "" {
payURL = resp.PayURL2
}
return &payment.CreatePaymentResponse{TradeNo: resp.TradeNo, PayURL: payURL, QRCode: resp.QRCode}, nil
} }
// resolveURLs returns (notifyURL, returnURL) preferring request values, // resolveURLs returns (notifyURL, returnURL) preferring request values,
......
...@@ -18,6 +18,9 @@ const ( ...@@ -18,6 +18,9 @@ const (
BlockTypeFunction BlockTypeFunction
) )
// UsageMapHook is a callback that can modify usage data before it's emitted in SSE events.
type UsageMapHook func(usageMap map[string]any)
// StreamingProcessor 流式响应处理器 // StreamingProcessor 流式响应处理器
type StreamingProcessor struct { type StreamingProcessor struct {
blockType BlockType blockType BlockType
...@@ -30,6 +33,7 @@ type StreamingProcessor struct { ...@@ -30,6 +33,7 @@ type StreamingProcessor struct {
originalModel string originalModel string
webSearchQueries []string webSearchQueries []string
groundingChunks []GeminiGroundingChunk groundingChunks []GeminiGroundingChunk
usageMapHook UsageMapHook
// 累计 usage // 累计 usage
inputTokens int inputTokens int
...@@ -46,6 +50,28 @@ func NewStreamingProcessor(originalModel string) *StreamingProcessor { ...@@ -46,6 +50,28 @@ func NewStreamingProcessor(originalModel string) *StreamingProcessor {
} }
} }
// SetUsageMapHook sets an optional hook that modifies usage maps before they are emitted.
func (p *StreamingProcessor) SetUsageMapHook(fn UsageMapHook) {
p.usageMapHook = fn
}
func usageToMap(u ClaudeUsage) map[string]any {
m := map[string]any{
"input_tokens": u.InputTokens,
"output_tokens": u.OutputTokens,
}
if u.CacheCreationInputTokens > 0 {
m["cache_creation_input_tokens"] = u.CacheCreationInputTokens
}
if u.CacheReadInputTokens > 0 {
m["cache_read_input_tokens"] = u.CacheReadInputTokens
}
if u.ImageOutputTokens > 0 {
m["image_output_tokens"] = u.ImageOutputTokens
}
return m
}
// ProcessLine 处理 SSE 行,返回 Claude SSE 事件 // ProcessLine 处理 SSE 行,返回 Claude SSE 事件
func (p *StreamingProcessor) ProcessLine(line string) []byte { func (p *StreamingProcessor) ProcessLine(line string) []byte {
line = strings.TrimSpace(line) line = strings.TrimSpace(line)
...@@ -172,6 +198,13 @@ func (p *StreamingProcessor) emitMessageStart(v1Resp *V1InternalResponse) []byte ...@@ -172,6 +198,13 @@ func (p *StreamingProcessor) emitMessageStart(v1Resp *V1InternalResponse) []byte
responseID = "msg_" + generateRandomID() responseID = "msg_" + generateRandomID()
} }
var usageValue any = usage
if p.usageMapHook != nil {
usageMap := usageToMap(usage)
p.usageMapHook(usageMap)
usageValue = usageMap
}
message := map[string]any{ message := map[string]any{
"id": responseID, "id": responseID,
"type": "message", "type": "message",
...@@ -180,7 +213,7 @@ func (p *StreamingProcessor) emitMessageStart(v1Resp *V1InternalResponse) []byte ...@@ -180,7 +213,7 @@ func (p *StreamingProcessor) emitMessageStart(v1Resp *V1InternalResponse) []byte
"model": p.originalModel, "model": p.originalModel,
"stop_reason": nil, "stop_reason": nil,
"stop_sequence": nil, "stop_sequence": nil,
"usage": usage, "usage": usageValue,
} }
event := map[string]any{ event := map[string]any{
...@@ -492,13 +525,20 @@ func (p *StreamingProcessor) emitFinish(finishReason string) []byte { ...@@ -492,13 +525,20 @@ func (p *StreamingProcessor) emitFinish(finishReason string) []byte {
ImageOutputTokens: p.imageOutputTokens, ImageOutputTokens: p.imageOutputTokens,
} }
var usageValue any = usage
if p.usageMapHook != nil {
usageMap := usageToMap(usage)
p.usageMapHook(usageMap)
usageValue = usageMap
}
deltaEvent := map[string]any{ deltaEvent := map[string]any{
"type": "message_delta", "type": "message_delta",
"delta": map[string]any{ "delta": map[string]any{
"stop_reason": stopReason, "stop_reason": stopReason,
"stop_sequence": nil, "stop_sequence": nil,
}, },
"usage": usage, "usage": usageValue,
} }
_, _ = result.Write(p.formatSSE("message_delta", deltaEvent)) _, _ = result.Write(p.formatSSE("message_delta", deltaEvent))
......
...@@ -28,6 +28,7 @@ func ChatCompletionsToResponses(req *ChatCompletionsRequest) (*ResponsesRequest, ...@@ -28,6 +28,7 @@ func ChatCompletionsToResponses(req *ChatCompletionsRequest) (*ResponsesRequest,
out := &ResponsesRequest{ out := &ResponsesRequest{
Model: req.Model, Model: req.Model,
Instructions: req.Instructions,
Input: inputJSON, Input: inputJSON,
Temperature: req.Temperature, Temperature: req.Temperature,
TopP: req.TopP, TopP: req.TopP,
......
...@@ -152,6 +152,7 @@ type AnthropicDelta struct { ...@@ -152,6 +152,7 @@ type AnthropicDelta struct {
// ResponsesRequest is the request body for POST /v1/responses. // ResponsesRequest is the request body for POST /v1/responses.
type ResponsesRequest struct { type ResponsesRequest struct {
Model string `json:"model"` Model string `json:"model"`
Instructions string `json:"instructions,omitempty"`
Input json.RawMessage `json:"input"` // string or []ResponsesInputItem Input json.RawMessage `json:"input"` // string or []ResponsesInputItem
MaxOutputTokens *int `json:"max_output_tokens,omitempty"` MaxOutputTokens *int `json:"max_output_tokens,omitempty"`
Temperature *float64 `json:"temperature,omitempty"` Temperature *float64 `json:"temperature,omitempty"`
...@@ -337,6 +338,7 @@ type ResponsesStreamEvent struct { ...@@ -337,6 +338,7 @@ type ResponsesStreamEvent struct {
type ChatCompletionsRequest struct { type ChatCompletionsRequest struct {
Model string `json:"model"` Model string `json:"model"`
Messages []ChatMessage `json:"messages"` Messages []ChatMessage `json:"messages"`
Instructions string `json:"instructions,omitempty"` // OpenAI Responses API compat
MaxTokens *int `json:"max_tokens,omitempty"` MaxTokens *int `json:"max_tokens,omitempty"`
MaxCompletionTokens *int `json:"max_completion_tokens,omitempty"` MaxCompletionTokens *int `json:"max_completion_tokens,omitempty"`
Temperature *float64 `json:"temperature,omitempty"` Temperature *float64 `json:"temperature,omitempty"`
......
...@@ -10,7 +10,13 @@ import ( ...@@ -10,7 +10,13 @@ import (
) )
func TestInit_DualOutput(t *testing.T) { func TestInit_DualOutput(t *testing.T) {
tmpDir := t.TempDir() // Use os.MkdirTemp instead of t.TempDir to avoid cleanup failures
// when lumberjack holds file handles on Windows.
tmpDir, err := os.MkdirTemp("", "logger-test-*")
if err != nil {
t.Fatalf("create temp dir: %v", err)
}
t.Cleanup(func() { _ = os.RemoveAll(tmpDir) })
logPath := filepath.Join(tmpDir, "logs", "sub2api.log") logPath := filepath.Join(tmpDir, "logs", "sub2api.log")
origStdout := os.Stdout origStdout := os.Stdout
...@@ -57,7 +63,9 @@ func TestInit_DualOutput(t *testing.T) { ...@@ -57,7 +63,9 @@ func TestInit_DualOutput(t *testing.T) {
L().Info("dual-output-info") L().Info("dual-output-info")
L().Warn("dual-output-warn") L().Warn("dual-output-warn")
Sync()
// Skip Sync() — on Windows, fsync on pipes deadlocks (FlushFileBuffers).
// The log data is already in the pipe buffer; closing writers is sufficient.
_ = stdoutW.Close() _ = stdoutW.Close()
_ = stderrW.Close() _ = stderrW.Close()
...@@ -166,7 +174,9 @@ func TestInit_CallerShouldPointToCallsite(t *testing.T) { ...@@ -166,7 +174,9 @@ func TestInit_CallerShouldPointToCallsite(t *testing.T) {
} }
L().Info("caller-check") L().Info("caller-check")
Sync() // Skip Sync() — on Windows, fsync on pipes deadlocks (FlushFileBuffers).
os.Stdout = origStdout
os.Stderr = origStderr
_ = stdoutW.Close() _ = stdoutW.Close()
logBytes, _ := io.ReadAll(stdoutR) logBytes, _ := io.ReadAll(stdoutR)
......
...@@ -77,7 +77,7 @@ func TestStdLogBridgeRoutesLevels(t *testing.T) { ...@@ -77,7 +77,7 @@ func TestStdLogBridgeRoutesLevels(t *testing.T) {
log.Printf("service started") log.Printf("service started")
log.Printf("Warning: queue full") log.Printf("Warning: queue full")
log.Printf("Forward request failed: timeout") log.Printf("Forward request failed: timeout")
Sync() // Skip Sync() — on Windows, fsync on pipes deadlocks (FlushFileBuffers).
_ = stdoutW.Close() _ = stdoutW.Close()
_ = stderrW.Close() _ = stderrW.Close()
...@@ -139,7 +139,7 @@ func TestLegacyPrintfRoutesLevels(t *testing.T) { ...@@ -139,7 +139,7 @@ func TestLegacyPrintfRoutesLevels(t *testing.T) {
LegacyPrintf("service.test", "request started") LegacyPrintf("service.test", "request started")
LegacyPrintf("service.test", "Warning: queue full") LegacyPrintf("service.test", "Warning: queue full")
LegacyPrintf("service.test", "forward failed: timeout") LegacyPrintf("service.test", "forward failed: timeout")
Sync() // Skip Sync() — on Windows, fsync on pipes deadlocks (FlushFileBuffers).
_ = stdoutW.Close() _ = stdoutW.Close()
_ = stderrW.Close() _ = stderrW.Close()
......
...@@ -113,9 +113,11 @@ func (r *usageBillingRepository) applyUsageBillingEffects(ctx context.Context, t ...@@ -113,9 +113,11 @@ func (r *usageBillingRepository) applyUsageBillingEffects(ctx context.Context, t
} }
if cmd.BalanceCost > 0 { if cmd.BalanceCost > 0 {
if err := deductUsageBillingBalance(ctx, tx, cmd.UserID, cmd.BalanceCost); err != nil { newBalance, err := deductUsageBillingBalance(ctx, tx, cmd.UserID, cmd.BalanceCost)
if err != nil {
return err return err
} }
result.NewBalance = &newBalance
} }
if cmd.APIKeyQuotaCost > 0 { if cmd.APIKeyQuotaCost > 0 {
...@@ -133,9 +135,11 @@ func (r *usageBillingRepository) applyUsageBillingEffects(ctx context.Context, t ...@@ -133,9 +135,11 @@ func (r *usageBillingRepository) applyUsageBillingEffects(ctx context.Context, t
} }
if cmd.AccountQuotaCost > 0 && (strings.EqualFold(cmd.AccountType, service.AccountTypeAPIKey) || strings.EqualFold(cmd.AccountType, service.AccountTypeBedrock)) { if cmd.AccountQuotaCost > 0 && (strings.EqualFold(cmd.AccountType, service.AccountTypeAPIKey) || strings.EqualFold(cmd.AccountType, service.AccountTypeBedrock)) {
if err := incrementUsageBillingAccountQuota(ctx, tx, cmd.AccountID, cmd.AccountQuotaCost); err != nil { quotaState, err := incrementUsageBillingAccountQuota(ctx, tx, cmd.AccountID, cmd.AccountQuotaCost)
if err != nil {
return err return err
} }
result.QuotaState = quotaState
} }
return nil return nil
...@@ -169,24 +173,22 @@ func incrementUsageBillingSubscription(ctx context.Context, tx *sql.Tx, subscrip ...@@ -169,24 +173,22 @@ func incrementUsageBillingSubscription(ctx context.Context, tx *sql.Tx, subscrip
return service.ErrSubscriptionNotFound return service.ErrSubscriptionNotFound
} }
func deductUsageBillingBalance(ctx context.Context, tx *sql.Tx, userID int64, amount float64) error { func deductUsageBillingBalance(ctx context.Context, tx *sql.Tx, userID int64, amount float64) (float64, error) {
res, err := tx.ExecContext(ctx, ` var newBalance float64
err := tx.QueryRowContext(ctx, `
UPDATE users UPDATE users
SET balance = balance - $1, SET balance = balance - $1,
updated_at = NOW() updated_at = NOW()
WHERE id = $2 AND deleted_at IS NULL WHERE id = $2 AND deleted_at IS NULL
`, amount, userID) RETURNING balance
if err != nil { `, amount, userID).Scan(&newBalance)
return err if errors.Is(err, sql.ErrNoRows) {
return 0, service.ErrUserNotFound
} }
affected, err := res.RowsAffected()
if err != nil { if err != nil {
return err return 0, err
}
if affected > 0 {
return nil
} }
return service.ErrUserNotFound return newBalance, nil
} }
func incrementUsageBillingAPIKeyQuota(ctx context.Context, tx *sql.Tx, apiKeyID int64, amount float64) (bool, error) { func incrementUsageBillingAPIKeyQuota(ctx context.Context, tx *sql.Tx, apiKeyID int64, amount float64) (bool, error) {
...@@ -240,7 +242,7 @@ func incrementUsageBillingAPIKeyRateLimit(ctx context.Context, tx *sql.Tx, apiKe ...@@ -240,7 +242,7 @@ func incrementUsageBillingAPIKeyRateLimit(ctx context.Context, tx *sql.Tx, apiKe
return nil return nil
} }
func incrementUsageBillingAccountQuota(ctx context.Context, tx *sql.Tx, accountID int64, amount float64) error { func incrementUsageBillingAccountQuota(ctx context.Context, tx *sql.Tx, accountID int64, amount float64) (*service.AccountQuotaState, error) {
rows, err := tx.QueryContext(ctx, rows, err := tx.QueryContext(ctx,
`UPDATE accounts SET extra = ( `UPDATE accounts SET extra = (
COALESCE(extra, '{}'::jsonb) COALESCE(extra, '{}'::jsonb)
...@@ -279,32 +281,40 @@ func incrementUsageBillingAccountQuota(ctx context.Context, tx *sql.Tx, accountI ...@@ -279,32 +281,40 @@ func incrementUsageBillingAccountQuota(ctx context.Context, tx *sql.Tx, accountI
WHERE id = $2 AND deleted_at IS NULL WHERE id = $2 AND deleted_at IS NULL
RETURNING RETURNING
COALESCE((extra->>'quota_used')::numeric, 0), COALESCE((extra->>'quota_used')::numeric, 0),
COALESCE((extra->>'quota_limit')::numeric, 0)`, COALESCE((extra->>'quota_limit')::numeric, 0),
COALESCE((extra->>'quota_daily_used')::numeric, 0),
COALESCE((extra->>'quota_daily_limit')::numeric, 0),
COALESCE((extra->>'quota_weekly_used')::numeric, 0),
COALESCE((extra->>'quota_weekly_limit')::numeric, 0)`,
amount, accountID) amount, accountID)
if err != nil { if err != nil {
return err return nil, err
} }
defer func() { _ = rows.Close() }() defer func() { _ = rows.Close() }()
var newUsed, limit float64 var state service.AccountQuotaState
if rows.Next() { if rows.Next() {
if err := rows.Scan(&newUsed, &limit); err != nil { if err := rows.Scan(
return err &state.TotalUsed, &state.TotalLimit,
&state.DailyUsed, &state.DailyLimit,
&state.WeeklyUsed, &state.WeeklyLimit,
); err != nil {
return nil, err
} }
} else { } else {
if err := rows.Err(); err != nil { if err := rows.Err(); err != nil {
return err return nil, err
} }
return service.ErrAccountNotFound return nil, service.ErrAccountNotFound
} }
if err := rows.Err(); err != nil { if err := rows.Err(); err != nil {
return err return nil, err
} }
if limit > 0 && newUsed >= limit && (newUsed-amount) < limit { if state.TotalLimit > 0 && state.TotalUsed >= state.TotalLimit && (state.TotalUsed-amount) < state.TotalLimit {
if err := enqueueSchedulerOutbox(ctx, tx, service.SchedulerOutboxEventAccountChanged, &accountID, nil, nil); err != nil { if err := enqueueSchedulerOutbox(ctx, tx, service.SchedulerOutboxEventAccountChanged, &accountID, nil, nil); err != nil {
logger.LegacyPrintf("repository.usage_billing", "[SchedulerOutbox] enqueue quota exceeded failed: account=%d err=%v", accountID, err) logger.LegacyPrintf("repository.usage_billing", "[SchedulerOutbox] enqueue quota exceeded failed: account=%d err=%v", accountID, err)
return err return nil, err
} }
} }
return nil return &state, nil
} }
...@@ -100,7 +100,7 @@ func (r *userGroupRateRepository) GetByGroupID(ctx context.Context, groupID int6 ...@@ -100,7 +100,7 @@ func (r *userGroupRateRepository) GetByGroupID(ctx context.Context, groupID int6
query := ` query := `
SELECT ugr.user_id, u.username, u.email, COALESCE(u.notes, ''), u.status, ugr.rate_multiplier SELECT ugr.user_id, u.username, u.email, COALESCE(u.notes, ''), u.status, ugr.rate_multiplier
FROM user_group_rate_multipliers ugr FROM user_group_rate_multipliers ugr
JOIN users u ON u.id = ugr.user_id JOIN users u ON u.id = ugr.user_id AND u.deleted_at IS NULL
WHERE ugr.group_id = $1 WHERE ugr.group_id = $1
ORDER BY ugr.user_id ORDER BY ugr.user_id
` `
......
...@@ -26,6 +26,7 @@ func RegisterPaymentRoutes( ...@@ -26,6 +26,7 @@ func RegisterPaymentRoutes(
authenticated.Use(middleware.BackendModeUserGuard(settingService)) authenticated.Use(middleware.BackendModeUserGuard(settingService))
{ {
authenticated.GET("/config", paymentHandler.GetPaymentConfig) authenticated.GET("/config", paymentHandler.GetPaymentConfig)
authenticated.GET("/checkout-info", paymentHandler.GetCheckoutInfo)
authenticated.GET("/plans", paymentHandler.GetPlans) authenticated.GET("/plans", paymentHandler.GetPlans)
authenticated.GET("/channels", paymentHandler.GetChannels) authenticated.GET("/channels", paymentHandler.GetChannels)
authenticated.GET("/limits", paymentHandler.GetLimits) authenticated.GET("/limits", paymentHandler.GetLimits)
...@@ -33,6 +34,7 @@ func RegisterPaymentRoutes( ...@@ -33,6 +34,7 @@ func RegisterPaymentRoutes(
orders := authenticated.Group("/orders") orders := authenticated.Group("/orders")
{ {
orders.POST("", paymentHandler.CreateOrder) orders.POST("", paymentHandler.CreateOrder)
orders.POST("/verify", paymentHandler.VerifyOrder)
orders.GET("/my", paymentHandler.GetMyOrders) orders.GET("/my", paymentHandler.GetMyOrders)
orders.GET("/:id", paymentHandler.GetOrder) orders.GET("/:id", paymentHandler.GetOrder)
orders.POST("/:id/cancel", paymentHandler.CancelOrder) orders.POST("/:id/cancel", paymentHandler.CancelOrder)
...@@ -52,6 +54,8 @@ func RegisterPaymentRoutes( ...@@ -52,6 +54,8 @@ func RegisterPaymentRoutes(
// --- Webhook endpoints (no auth) --- // --- Webhook endpoints (no auth) ---
webhook := v1.Group("/payment/webhook") webhook := v1.Group("/payment/webhook")
{ {
// EasyPay sends GET callbacks with query params
webhook.GET("/easypay", webhookHandler.EasyPayNotify)
webhook.POST("/easypay", webhookHandler.EasyPayNotify) webhook.POST("/easypay", webhookHandler.EasyPayNotify)
webhook.POST("/alipay", webhookHandler.AlipayNotify) webhook.POST("/alipay", webhookHandler.AlipayNotify)
webhook.POST("/wxpay", webhookHandler.WxpayNotify) webhook.POST("/wxpay", webhookHandler.WxpayNotify)
......
...@@ -6,6 +6,7 @@ import ( ...@@ -6,6 +6,7 @@ import (
"encoding/hex" "encoding/hex"
"errors" "errors"
"fmt" "fmt"
"log/slog"
"math/rand/v2" "math/rand/v2"
"time" "time"
...@@ -99,7 +100,7 @@ func (s *APIKeyService) StartAuthCacheInvalidationSubscriber(ctx context.Context ...@@ -99,7 +100,7 @@ func (s *APIKeyService) StartAuthCacheInvalidationSubscriber(ctx context.Context
s.authCacheL1.Del(cacheKey) s.authCacheL1.Del(cacheKey)
}); err != nil { }); err != nil {
// Log but don't fail - L1 cache will still work, just without cross-instance invalidation // Log but don't fail - L1 cache will still work, just without cross-instance invalidation
println("[Service] Warning: failed to start auth cache invalidation subscriber:", err.Error()) slog.Warn("failed to start auth cache invalidation subscriber", "error", err)
} }
} }
......
...@@ -81,9 +81,9 @@ type wildcardMappingEntry struct { ...@@ -81,9 +81,9 @@ type wildcardMappingEntry struct {
type channelCache struct { type channelCache struct {
// 热路径查找 // 热路径查找
pricingByGroupModel map[channelModelKey]*ChannelModelPricing // (groupID, platform, model) → 定价 pricingByGroupModel map[channelModelKey]*ChannelModelPricing // (groupID, platform, model) → 定价
wildcardByGroupPlatform map[channelGroupPlatformKey][]*wildcardPricingEntry // (groupID, platform) → 通配符定价(前缀长度降序 wildcardByGroupPlatform map[channelGroupPlatformKey][]*wildcardPricingEntry // (groupID, platform) → 通配符定价(按配置顺序,先匹配先使用
mappingByGroupModel map[channelModelKey]string // (groupID, platform, model) → 映射目标 mappingByGroupModel map[channelModelKey]string // (groupID, platform, model) → 映射目标
wildcardMappingByGP map[channelGroupPlatformKey][]*wildcardMappingEntry // (groupID, platform) → 通配符映射(前缀长度降序 wildcardMappingByGP map[channelGroupPlatformKey][]*wildcardMappingEntry // (groupID, platform) → 通配符映射(按配置顺序,先匹配先使用
channelByGroupID map[int64]*Channel // groupID → 渠道 channelByGroupID map[int64]*Channel // groupID → 渠道
groupPlatform map[int64]string // groupID → platform groupPlatform map[int64]string // groupID → platform
...@@ -680,6 +680,7 @@ func (s *ChannelService) Create(ctx context.Context, input *CreateChannelInput) ...@@ -680,6 +680,7 @@ func (s *ChannelService) Create(ctx context.Context, input *CreateChannelInput)
ModelPricing: input.ModelPricing, ModelPricing: input.ModelPricing,
ModelMapping: input.ModelMapping, ModelMapping: input.ModelMapping,
Features: input.Features, Features: input.Features,
FeaturesConfig: input.FeaturesConfig,
ApplyPricingToAccountStats: input.ApplyPricingToAccountStats, ApplyPricingToAccountStats: input.ApplyPricingToAccountStats,
AccountStatsPricingRules: input.AccountStatsPricingRules, AccountStatsPricingRules: input.AccountStatsPricingRules,
} }
...@@ -780,6 +781,9 @@ func (s *ChannelService) applyUpdateInput(ctx context.Context, channel *Channel, ...@@ -780,6 +781,9 @@ func (s *ChannelService) applyUpdateInput(ctx context.Context, channel *Channel,
if input.BillingModelSource != "" { if input.BillingModelSource != "" {
channel.BillingModelSource = input.BillingModelSource channel.BillingModelSource = input.BillingModelSource
} }
if input.FeaturesConfig != nil {
channel.FeaturesConfig = input.FeaturesConfig
}
if input.ApplyPricingToAccountStats != nil { if input.ApplyPricingToAccountStats != nil {
channel.ApplyPricingToAccountStats = *input.ApplyPricingToAccountStats channel.ApplyPricingToAccountStats = *input.ApplyPricingToAccountStats
} }
...@@ -959,6 +963,7 @@ type CreateChannelInput struct { ...@@ -959,6 +963,7 @@ type CreateChannelInput struct {
BillingModelSource string BillingModelSource string
RestrictModels bool RestrictModels bool
Features string Features string
FeaturesConfig map[string]any
ApplyPricingToAccountStats bool ApplyPricingToAccountStats bool
AccountStatsPricingRules []AccountStatsPricingRule AccountStatsPricingRules []AccountStatsPricingRule
} }
...@@ -974,6 +979,7 @@ type UpdateChannelInput struct { ...@@ -974,6 +979,7 @@ type UpdateChannelInput struct {
BillingModelSource string BillingModelSource string
RestrictModels *bool RestrictModels *bool
Features *string Features *string
FeaturesConfig map[string]any
ApplyPricingToAccountStats *bool ApplyPricingToAccountStats *bool
AccountStatsPricingRules *[]AccountStatsPricingRule AccountStatsPricingRules *[]AccountStatsPricingRule
} }
...@@ -343,8 +343,9 @@ func (s *ConcurrencyService) StartSlotCleanupWorker(accountRepo AccountRepositor ...@@ -343,8 +343,9 @@ func (s *ConcurrencyService) StartSlotCleanupWorker(accountRepo AccountRepositor
}() }()
} }
// GetAccountConcurrencyBatch gets current concurrency counts for multiple accounts // GetAccountConcurrencyBatch gets current concurrency counts for multiple accounts.
// Returns a map of accountID -> current concurrency count // Uses a detached context with timeout to prevent HTTP request cancellation from
// causing the entire batch to fail (which would show all concurrency as 0).
func (s *ConcurrencyService) GetAccountConcurrencyBatch(ctx context.Context, accountIDs []int64) (map[int64]int, error) { func (s *ConcurrencyService) GetAccountConcurrencyBatch(ctx context.Context, accountIDs []int64) (map[int64]int, error) {
if len(accountIDs) == 0 { if len(accountIDs) == 0 {
return map[int64]int{}, nil return map[int64]int{}, nil
...@@ -356,5 +357,11 @@ func (s *ConcurrencyService) GetAccountConcurrencyBatch(ctx context.Context, acc ...@@ -356,5 +357,11 @@ func (s *ConcurrencyService) GetAccountConcurrencyBatch(ctx context.Context, acc
} }
return result, nil return result, nil
} }
return s.cache.GetAccountConcurrencyBatch(ctx, accountIDs)
// Use a detached context so that a cancelled HTTP request doesn't cause
// the Redis pipeline to fail and return all-zero concurrency counts.
redisCtx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
defer cancel()
return s.cache.GetAccountConcurrencyBatch(redisCtx, accountIDs)
} }
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment