iotlcm/upstream-chan/internal/service/mingqian/client.go

package mingqian

import (
	"bytes"
	"context"
	"crypto/md5"
	"crypto/rand"
	"encoding/hex"
	"encoding/json"
	"io"
	"net/http"
	"net/url"
	"sort"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/gogf/gf/v2/errors/gerror"
)

const DefaultBaseURL = "http://iotapi.bjmqtx.com/openapi/wlw/v3/cool"

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

var accountGates sync.Map

func (limiter *gate) wait(ctx context.Context) error {
	for {
		if err := ctx.Err(); err != nil {
			return err
		}
		limiter.mu.Lock()
		delay := time.Until(limiter.next)
		if delay <= 0 {
			limiter.next = time.Now().Add(60 * time.Millisecond)
			limiter.mu.Unlock()
			return nil
		}
		limiter.mu.Unlock()
		timer := time.NewTimer(delay)
		select {
		case <-ctx.Done():
			timer.Stop()
			return ctx.Err()
		case <-timer.C:
		}
	}
}

type Client struct {
	baseURL string
	appID   string
	secret  string
	http    *http.Client
	gate    *gate
}

func NewClient(baseURL, appID, secret string, timeout time.Duration) (*Client, error) {
	baseURL = strings.TrimRight(strings.TrimSpace(baseURL), "/")
	parsed, err := url.Parse(baseURL)
	if err != nil || parsed.Host == "" || (parsed.Scheme != "http" && parsed.Scheme != "https") || parsed.User != nil || parsed.RawQuery != "" || parsed.Fragment != "" {
		return nil, gerror.New("mingqian: invalid api.baseUrl")
	}
	appID, secret = strings.TrimSpace(appID), strings.TrimSpace(secret)
	if appID == "" || secret == "" {
		return nil, gerror.New("mingqian: channel AppKey (appId) and AppSecret (APPKEY) are required")
	}
	if timeout <= 0 {
		return nil, gerror.New("mingqian: timeout must be positive")
	}
	limiter, _ := accountGates.LoadOrStore(baseURL+"|"+appID, &gate{})
	return &Client{baseURL: baseURL, appID: appID, secret: secret, gate: limiter.(*gate), http: &http.Client{Timeout: timeout, CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }}}, nil
}

func Sign(params map[string]string, secret string) string {
	keys := make([]string, 0, len(params))
	for key := range params {
		if key != "sign" {
			keys = append(keys, key)
		}
	}
	sort.Strings(keys)
	var source strings.Builder
	source.WriteString(secret)
	for _, key := range keys {
		source.WriteString(key)
		source.WriteString("_")
		source.WriteString(params[key])
	}
	source.WriteString(secret)
	sum := md5.Sum([]byte(source.String()))
	return hex.EncodeToString(sum[:])
}

func (client *Client) post(ctx context.Context, path string, data any, output any) error {
	if err := client.gate.wait(ctx); err != nil {
		return err
	}
	now := time.Now().In(time.FixedZone("Asia/Shanghai", 8*60*60))
	random := make([]byte, 3)
	if _, err := rand.Read(random); err != nil {
		return gerror.New("mingqian: cannot generate request ID")
	}
	common := map[string]string{
		"appId": client.appID, "version": "3.0", "reqtime": now.Format("2006-01-02"),
		"timestamp": strconv.FormatInt(now.Add(time.Minute).UnixMilli(), 10),
		"transid":   now.Format("20060102150405") + strconv.Itoa(1000 + now.Nanosecond()/1000000)[1:] + hex.EncodeToString(random),
	}
	common["sign"] = Sign(common, client.secret)
	payload := struct {
		Body map[string]string `json:"body"`
		Data string            `json:"data,omitempty"`
	}{Body: common}
	if data != nil {
		encoded, err := json.Marshal(data)
		if err != nil {
			return gerror.New("mingqian: invalid request data")
		}
		payload.Data = string(encoded)
	}
	encoded, err := json.Marshal(payload)
	if err != nil {
		return gerror.New("mingqian: invalid request envelope")
	}
	request, err := http.NewRequestWithContext(ctx, http.MethodPost, client.baseURL+path, bytes.NewReader(encoded))
	if err != nil {
		return gerror.New("mingqian: invalid request URL")
	}
	request.Header.Set("Content-Type", "application/json")
	response, err := client.http.Do(request)
	if err != nil {
		if ctx.Err() != nil {
			return ctx.Err()
		}
		return gerror.New("mingqian: upstream request failed or timed out; no automatic retry")
	}
	defer response.Body.Close()
	if response.StatusCode != http.StatusOK {
		return gerror.Newf("mingqian: HTTP %d", response.StatusCode)
	}
	body, err := io.ReadAll(io.LimitReader(response.Body, 4*1024*1024+1))
	if err != nil || len(body) > 4*1024*1024 {
		return gerror.New("mingqian: invalid or oversized response")
	}
	var envelope struct {
		Code *int            `json:"code"`
		Msg  string          `json:"msg"`
		Data json.RawMessage `json:"data"`
	}
	if err := json.Unmarshal(body, &envelope); err != nil {
		return gerror.New("mingqian: malformed response JSON")
	}
	if envelope.Code == nil {
		return gerror.New("mingqian: response code missing")
	}
	if *envelope.Code != 200 {
		return gerror.Newf("mingqian: business code %d", *envelope.Code)
	}
	if output != nil {
		if len(envelope.Data) == 0 || string(envelope.Data) == "null" {
			return gerror.New("mingqian: response data missing")
		}
		if err := json.Unmarshal(envelope.Data, output); err != nil {
			return gerror.New("mingqian: invalid response data")
		}
	}
	return nil
}