iotlcm/upstream-chan/internal/logic/upstreamimpl/recharge/client.go

package recharge

import (
	"bytes"
	"context"
	"crypto/md5"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"strconv"
	"strings"
	"sync"
	"time"
)

const DefaultBaseURL = "http://www.sdtywl.cn/"
const requestInterval = time.Second

var requestGates sync.Map

// 上游业务码（综合充值平台API文档-通用反馈状态码）
const (
	CodeSuccess      = 200 // 成功
	CodeMissingParam = 400 // 缺少参数
	CodeSignError    = 401 // sign错误
	CodeForbidden    = 403 // 无权访问
	CodePwdKeyError  = 404 // pwdkey错误
	CodeNotWhiteList = 407 // 非白名单IP
	CodeRateLimited  = 429 // 访问超频（单接口60次/分）
	CodeServerError  = 500 // 其他错误
)

// BusinessError 上游业务错误（HTTP 200 但 code != 200）
type BusinessError struct {
	Code int
	Msg  string
}

func (e *BusinessError) Error() string {
	return fmt.Sprintf("recharge upstream business error, code %d, msg %s", e.Code, e.Msg)
}

// Response 上游统一响应信封
type Response struct {
	Code int             `json:"code"`
	Msg  string          `json:"msg"`
	Data json.RawMessage `json:"data"`
}

type Config struct {
	BaseURL string
	PwdKey  string
	Pwdiv   string
	Timeout time.Duration
}

type Client struct {
	baseURL string
	pwdKey  string
	pwdiv   string
	http    *http.Client
}

type requestGate struct {
	mu   sync.Mutex
	next time.Time
}

func (gate *requestGate) wait(ctx context.Context) error {
	for {
		if err := ctx.Err(); err != nil {
			return err
		}
		gate.mu.Lock()
		delay := time.Until(gate.next)
		if delay <= 0 {
			gate.next = time.Now().Add(requestInterval)
			gate.mu.Unlock()
			return nil
		}
		gate.mu.Unlock()

		timer := time.NewTimer(delay)
		select {
		case <-ctx.Done():
			timer.Stop()
			return ctx.Err()
		case <-timer.C:
		}
	}
}

func NewClient(cfg Config) *Client {
	baseURL := cfg.BaseURL
	if baseURL == "" {
		baseURL = DefaultBaseURL
	}
	timeout := cfg.Timeout
	if timeout <= 0 {
		timeout = 10 * time.Second
	}
	return &Client{
		baseURL: strings.TrimSuffix(baseURL, "/"),
		pwdKey:  cfg.PwdKey,
		pwdiv:   cfg.Pwdiv,
		http:    &http.Client{Timeout: timeout},
	}
}

// DoPost 调用上游 POST 接口，返回业务信封；HTTP 层错误直接返回 error，
// 业务码（code != 200）由调用方判断
func (c *Client) DoPost(ctx context.Context, endpoint string, data any) (*Response, error) {
	if err := c.requestGate(endpoint).wait(ctx); err != nil {
		return nil, err
	}
	timestamp := strconv.FormatInt(time.Now().UnixMilli(), 10)
	payload := map[string]any{
		"pwdkey":    c.pwdKey,
		"timestamp": timestamp,
		"sign":      c.sign(timestamp),
	}
	if data != nil {
		payload["data"] = data
	}
	body, err := json.Marshal(payload)
	if err != nil {
		return nil, err
	}
	return c.do(ctx, http.MethodPost, c.baseURL+"/prod-api/open/api"+endpoint, body)
}

func (c *Client) requestGate(endpoint string) *requestGate {
	key := c.baseURL + "|" + c.pwdKey + "|" + strings.TrimSpace(endpoint)
	gate, _ := requestGates.LoadOrStore(key, &requestGate{})
	return gate.(*requestGate)
}

func (c *Client) sign(timestamp string) string {
	sum := md5.Sum([]byte(c.pwdKey + c.pwdiv + timestamp))
	return strings.ToLower(hex.EncodeToString(sum[:]))
}

func (c *Client) do(ctx context.Context, method, endpoint string, body []byte) (*Response, error) {
	var reader io.Reader
	if body != nil {
		reader = bytes.NewReader(body)
	}
	req, err := http.NewRequestWithContext(ctx, method, endpoint, reader)
	if err != nil {
		return nil, err
	}
	req.Header.Set("Accept", "application/json")
	if body != nil {
		req.Header.Set("Content-Type", "application/json; charset=utf-8")
	}

	resp, err := c.http.Do(req)
	if err != nil {
		return nil, err
	}
	defer resp.Body.Close()

	respBody, err := io.ReadAll(resp.Body)
	if err != nil {
		return nil, err
	}
	if resp.StatusCode < http.StatusOK || resp.StatusCode >= http.StatusMultipleChoices {
		return nil, fmt.Errorf("recharge upstream http status %d: %s", resp.StatusCode, string(respBody))
	}
	var output Response
	if err := json.Unmarshal(respBody, &output); err != nil {
		return nil, fmt.Errorf("decode recharge upstream response: %w, body=%s", err, string(respBody))
	}
	return &output, nil
}