exchange_go/services/excservice/binancesocketmanager.go

package excservice

import (
	"context"
	"crypto/tls"
	"errors"
	"fmt"
	"go-admin/common/global"
	"go-admin/common/helper"
	"go-admin/models/binancedto"
	"go-admin/models/commondto"
	"go-admin/pkg/jsonhelper"
	"go-admin/pkg/utility"
	"math/rand"
	"net"
	"net/http"
	"net/url"
	"os"
	"os/signal"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"github.com/bytedance/sonic"
	log "github.com/go-admin-team/go-admin-core/logger"
	"github.com/gorilla/websocket"
	"golang.org/x/net/proxy"
)

type BinanceWebSocketManager struct {
	ws          *websocket.Conn
	stopChannel chan struct{}
	url         string
	/* 0-现货 1-合约 */
	wsType       int
	apiKey       string
	apiSecret    string
	proxyType    string
	proxyAddress string
	reconnect    chan struct{}
	isStopped    bool       // 标记 WebSocket 是否已主动停止
	mu           sync.Mutex // 用于控制并发访问 isStopped
	cancelFunc   context.CancelFunc
	listenKey    string      // 新增字段
	reconnecting atomic.Bool // 防止重复重连
	ConnectTime  time.Time   // 当前连接建立时间
}

// 已有连接
var SpotSockets = map[string]*BinanceWebSocketManager{}
var FutureSockets = map[string]*BinanceWebSocketManager{}

func NewBinanceWebSocketManager(wsType int, apiKey, apiSecret, proxyType, proxyAddress string) *BinanceWebSocketManager {
	url := ""

	switch wsType {
	case 0:
		url = "wss://stream.binance.com:9443/ws"
	case 1:
		url = "wss://fstream.binance.com/ws"
	}

	return &BinanceWebSocketManager{
		stopChannel:  make(chan struct{}, 10),
		reconnect:    make(chan struct{}, 10),
		isStopped:    false,
		url:          url,
		wsType:       wsType,
		apiKey:       apiKey,
		apiSecret:    apiSecret,
		proxyType:    proxyType,
		proxyAddress: proxyAddress,
	}
}

func (wm *BinanceWebSocketManager) GetKey() string {
	return wm.apiKey
}

func (wm *BinanceWebSocketManager) Start() {
	utility.SafeGo(wm.run)
	// wm.run()
}

// 重启连接
func (wm *BinanceWebSocketManager) Restart(apiKey, apiSecret, proxyType, proxyAddress string) *BinanceWebSocketManager {
	wm.mu.Lock()
	defer wm.mu.Unlock()

	wm.apiKey = apiKey
	wm.apiSecret = apiSecret
	wm.proxyType = proxyType
	wm.proxyAddress = proxyAddress

	if wm.isStopped {
		wm.isStopped = false
		utility.SafeGo(wm.run)
	} else {
		log.Warnf("调用restart")
		wm.triggerReconnect(true)
	}

	return wm
}

// 触发重连
func (wm *BinanceWebSocketManager) triggerReconnect(force bool) {
	if force {
		wm.reconnecting.Store(false) // 强制重置标志位
	}

	if wm.reconnecting.CompareAndSwap(false, true) {
		log.Warnf("准备重连 key: %s wsType: %v", wm.apiKey, wm.wsType)
		// 发送信号触发重连协程
		select {
		case wm.reconnect <- struct{}{}:
		default:
			// 防止阻塞，如果通道满了就跳过
			log.Debugf("reconnect 信号已存在，跳过 key:%s", wm.apiKey)
		}
	}
}
func Restart(wm *BinanceWebSocketManager) {
	log.Warnf("调用restart")
	wm.triggerReconnect(true)
}
func (wm *BinanceWebSocketManager) run() {
	ctx, cancel := context.WithCancel(context.Background())
	wm.cancelFunc = cancel

	utility.SafeGo(wm.handleSignal)

	// 计算错误记录键
	errKey := fmt.Sprintf(global.API_WEBSOCKET_ERR, wm.apiKey)
	errMessage := commondto.WebSocketErr{Time: time.Now()}
	helper.DefaultRedis.SetString(errKey, jsonhelper.ToJsonString(errMessage))

	for {
		select {
		case <-ctx.Done():
			return
		default:
			if err := wm.connect(ctx); err != nil {
				wm.handleConnectionError(errKey, err)

				if wm.isErrorCountExceeded(errKey) {
					log.Error("连接 %s WebSocket 时出错次数过多，停止 WebSocket 管理器: %v", wm.wsType, wm.apiKey)
					wm.Stop()
					return
				}

				time.Sleep(5 * time.Second)
				continue
			}

			<-wm.stopChannel
			log.Info("停止 %s WebSocket 管理器...", getWsTypeName(wm.wsType))
			wm.Stop()
			return
		}
	}
}

// handleConnectionError 处理 WebSocket 连接错误
func (wm *BinanceWebSocketManager) handleConnectionError(errKey string, err error) {
	// 从 Redis 获取错误记录
	var errMessage commondto.WebSocketErr
	val, _ := helper.DefaultRedis.GetString(errKey)
	if val != "" {
		sonic.UnmarshalString(val, &errMessage)
	}

	// 更新错误记录
	errMessage.Count++
	errMessage.Time = time.Now()
	errMessage.ErrorMessage = err.Error()

	// 将错误记录保存到 Redis
	if data, err := sonic.MarshalString(errMessage); err == nil {
		helper.DefaultRedis.SetString(errKey, data)
	}

	// 记录错误日志
	log.Error("连接 %s WebSocket 时出错: %v, 错误: %v", wm.wsType, wm.apiKey, err)
}

// isErrorCountExceeded 检查错误次数是否超过阈值
func (wm *BinanceWebSocketManager) isErrorCountExceeded(errKey string) bool {
	val, _ := helper.DefaultRedis.GetString(errKey)
	if val == "" {
		return false
	}

	var errMessage commondto.WebSocketErr
	if err := sonic.UnmarshalString(val, &errMessage); err != nil {
		return false
	}

	return errMessage.Count >= 5
}

// 处理终止信号
func (wm *BinanceWebSocketManager) handleSignal() {
	ch := make(chan os.Signal)
	signal.Notify(ch, os.Interrupt)
	<-ch
	wm.Stop()
}

func (wm *BinanceWebSocketManager) connect(ctx context.Context) error {
	dialer, err := wm.getDialer()
	if err != nil {
		return err
	}

	listenKey, err := wm.getListenKey()
	if err != nil {
		return err
	}

	wm.listenKey = listenKey
	url := fmt.Sprintf("%s/%s", wm.url, listenKey)
	wm.ws, _, err = dialer.Dial(url, nil)
	if err != nil {
		return err
	}

	// 连接成功，更新连接时间
	wm.ConnectTime = time.Now()
	log.Info(fmt.Sprintf("已连接到 Binance %s WebSocket【%s】 key:%s", getWsTypeName(wm.wsType), wm.apiKey, listenKey))

	// Ping处理
	wm.ws.SetPingHandler(func(appData string) error {
		log.Info(fmt.Sprintf("收到 wstype: %v key:%s Ping 消息【%s】", wm.wsType, wm.apiKey, appData))

		for x := 0; x < 5; x++ {
			if err := wm.ws.WriteControl(websocket.PongMessage, []byte(appData), time.Now().Add(time.Second*10)); err != nil {
				log.Error("binance 回应pong失败 次数:", strconv.Itoa(x), " err:", err)

				time.Sleep(time.Second * 1)
				continue
			}

			break
		}

		setLastTime(wm)
		return nil
	})

	utility.SafeGo(func() { wm.startListenKeyRenewal2(ctx) })
	utility.SafeGo(func() { wm.readMessages(ctx) })
	utility.SafeGo(func() { wm.handleReconnect(ctx) })
	utility.SafeGo(func() { wm.startPingLoop(ctx) })
	// utility.SafeGo(func() { wm.startDeadCheck(ctx) })

	return nil
}

// ReplaceConnection 创建新连接并关闭旧连接，实现无缝连接替换
func (wm *BinanceWebSocketManager) ReplaceConnection() error {
	wm.mu.Lock()
	if wm.isStopped {
		wm.mu.Unlock()
		return errors.New("WebSocket 已停止")
	}
	oldCtxCancel := wm.cancelFunc
	oldConn := wm.ws
	wm.mu.Unlock()

	log.Infof("🔄 正在替换连接: %s", wm.apiKey)

	// 步骤 1：先获取新的 listenKey 和连接
	newListenKey, err := wm.getListenKey()
	if err != nil {
		return fmt.Errorf("获取新 listenKey 失败: %w", err)
	}

	dialer, err := wm.getDialer()
	if err != nil {
		return err
	}

	newURL := fmt.Sprintf("%s/%s", wm.url, newListenKey)
	newConn, _, err := dialer.Dial(newURL, nil)
	if err != nil {
		return fmt.Errorf("新连接 Dial 失败: %w", err)
	}

	// 步骤 2：创建新的上下文并启动协程
	newCtx, newCancel := context.WithCancel(context.Background())

	// 设置 ping handler
	newConn.SetPingHandler(func(appData string) error {
		log.Infof("收到 Ping（新连接） key:%s msg:%s", wm.apiKey, appData)
		for x := 0; x < 5; x++ {
			if err := newConn.WriteControl(websocket.PongMessage, []byte(appData), time.Now().Add(10*time.Second)); err != nil {
				log.Errorf("Pong 失败 %d 次 err:%v", x, err)
				time.Sleep(time.Second)
				continue
			}
			break
		}
		setLastTime(wm)
		return nil
	})

	// 步骤 3：安全切换连接
	wm.mu.Lock()
	wm.ws = newConn
	wm.listenKey = newListenKey
	wm.ConnectTime = time.Now()
	wm.cancelFunc = newCancel
	wm.mu.Unlock()

	log.Infof("✅ 替换连接成功: %s listenKey: %s", wm.apiKey, newListenKey)

	// 步骤 4：启动新连接协程
	go wm.startListenKeyRenewal2(newCtx)
	go wm.readMessages(newCtx)
	go wm.handleReconnect(newCtx)
	go wm.startPingLoop(newCtx)
	// go wm.startDeadCheck(newCtx)

	// 步骤 5：关闭旧连接、取消旧协程
	if oldCtxCancel != nil {
		oldCtxCancel()
	}
	if oldConn != nil {
		_ = oldConn.Close()
		log.Infof("🔒 旧连接已关闭: %s", wm.apiKey)
	}

	return nil
}

// 更新最后通信时间
func setLastTime(wm *BinanceWebSocketManager) {
	subKey := fmt.Sprintf(global.USER_SUBSCRIBE, wm.apiKey)
	val, _ := helper.DefaultRedis.GetString(subKey)
	now := time.Now()
	var data binancedto.UserSubscribeState
	if val != "" {
		sonic.Unmarshal([]byte(val), &data)
	}

	if wm.wsType == 0 {
		data.SpotLastTime = &now
	} else {
		data.FuturesLastTime = &now
	}

	val, _ = sonic.MarshalString(&data)

	if val != "" {
		helper.DefaultRedis.SetString(subKey, val)
	}

}

func (wm *BinanceWebSocketManager) getDialer() (*websocket.Dialer, error) {
	if wm.proxyAddress == "" {
		return &websocket.Dialer{}, nil
	}

	if !strings.HasPrefix(wm.proxyAddress, "http://") && !strings.HasPrefix(wm.proxyAddress, "https://") && !strings.HasPrefix(wm.proxyAddress, "socks5://") {
		wm.proxyAddress = wm.proxyType + "://" + wm.proxyAddress
	}

	proxyURL, err := url.Parse(wm.proxyAddress)
	if err != nil {
		return nil, fmt.Errorf("failed to parse proxy URL: %v", err)
	}

	transport := &http.Transport{
		TLSClientConfig: &tls.Config{InsecureSkipVerify: false},
	}

	switch proxyURL.Scheme {
	case "socks5":
		return wm.createSocks5Dialer(proxyURL)
	case "http", "https":
		transport.Proxy = http.ProxyURL(proxyURL)
		return &websocket.Dialer{Proxy: transport.Proxy, TLSClientConfig: transport.TLSClientConfig}, nil
	default:
		return nil, fmt.Errorf("unsupported proxy scheme: %s", proxyURL.Scheme)
	}
}

func (wm *BinanceWebSocketManager) createSocks5Dialer(proxyURL *url.URL) (*websocket.Dialer, error) {
	auth := &proxy.Auth{}
	if proxyURL.User != nil {
		auth.User = proxyURL.User.Username()
		auth.Password, _ = proxyURL.User.Password()
	}

	socksDialer, err := proxy.SOCKS5("tcp", proxyURL.Host, auth, proxy.Direct)
	if err != nil {
		return nil, fmt.Errorf("failed to create SOCKS5 proxy dialer: %v", err)
	}

	return &websocket.Dialer{
		NetDialContext: func(ctx context.Context, network, address string) (net.Conn, error) {
			return socksDialer.Dial(network, address)
		},
	}, nil
}

// 复用创建HTTP客户端的逻辑
func (wm *BinanceWebSocketManager) createBinanceClient() (*helper.BinanceClient, error) {
	return helper.NewBinanceClient(wm.apiKey, wm.apiSecret, wm.proxyType, wm.proxyAddress)
}

// 获取listenKey
func (wm *BinanceWebSocketManager) getListenKey() (string, error) {
	client, err := wm.createBinanceClient()
	if err != nil {
		return "", err
	}

	var resp []byte
	switch wm.wsType {
	case 0:
		resp, _, err = client.SendSpotRequestByKey("/api/v3/userDataStream", "POST", nil)
	case 1:
		resp, _, err = client.SendFuturesRequestByKey("/fapi/v1/listenKey", "POST", nil)

	default:
		log.Error("链接类型错误")
		return "", errors.New("链接类型错误")
	}

	if err != nil {
		return "", err
	}

	var dataMap map[string]interface{}
	if err := sonic.Unmarshal(resp, &dataMap); err != nil {
		return "", err
	}

	if listenKey, ok := dataMap["listenKey"]; ok {
		return listenKey.(string), nil
	}

	return "", errors.New("listenKey 不存在")
}

// 接收消息
func (wm *BinanceWebSocketManager) readMessages(ctx context.Context) {
	defer wm.ws.Close()

	for {
		select {
		case <-ctx.Done():
			return
		default:
			if wm.isStopped {
				return
			}

			_, msg, err := wm.ws.ReadMessage()
			if err != nil && strings.Contains(err.Error(), "websocket: close") {
				if !wm.isStopped {
					log.Error("收到关闭消息", err.Error())
					wm.triggerReconnect(false)
				}

				log.Error("websocket 关闭")
				return
			} else if err != nil {
				log.Error("读取消息时出错: %v", err)
				return
			}
			wm.handleOrderUpdate(msg)

		}
	}
}

func (wm *BinanceWebSocketManager) handleOrderUpdate(msg []byte) {
	setLastTime(wm)

	if reconnect, _ := ReceiveListen(msg, wm.wsType, wm.apiKey); reconnect {
		log.Errorf("收到重连请求")
		wm.triggerReconnect(false)
	}
}

// Stop 安全停止 WebSocket
func (wm *BinanceWebSocketManager) Stop() {
	wm.mu.Lock()
	defer wm.mu.Unlock()

	if wm.isStopped {
		return
	}
	wm.isStopped = true

	select {
	case <-wm.stopChannel:
	default:
		close(wm.stopChannel)
	}

	if wm.cancelFunc != nil {
		wm.cancelFunc()
		wm.cancelFunc = nil
	}

	if wm.ws != nil {
		if err := wm.ws.Close(); err != nil {
			log.Errorf("WebSocket Close 错误 key:%s err:%v", wm.apiKey, err)
		}
		wm.ws = nil
	}

	log.Infof("WebSocket 已完全停止 key:%s", wm.apiKey)
	wm.stopChannel = make(chan struct{}, 10)
}

// handleReconnect 使用指数退避并保持永不退出
func (wm *BinanceWebSocketManager) handleReconnect(ctx context.Context) {
	const maxRetries = 100
	baseDelay := time.Second * 2
	retryCount := 0

	for {
		select {
		case <-ctx.Done():
			log.Infof("handleReconnect context done: %s", wm.apiKey)
			return

		case <-wm.reconnect:
			wm.mu.Lock()
			if wm.isStopped {
				wm.mu.Unlock()
				return
			}
			wm.mu.Unlock()

			log.Warnf("WebSocket 连接断开，准备重连 key:%s", wm.apiKey)

			for {
				wm.mu.Lock()
				if wm.ws != nil {
					_ = wm.ws.Close()
					wm.ws = nil
				}
				if wm.cancelFunc != nil {
					wm.cancelFunc()
					wm.cancelFunc = nil
				}
				wm.mu.Unlock()

				newCtx, cancel := context.WithCancel(context.Background())
				wm.mu.Lock()
				wm.cancelFunc = cancel
				wm.mu.Unlock()

				if err := wm.connect(newCtx); err != nil {
					log.Errorf("🔌 重连失败（%d/%d）key:%s，err: %v", retryCount+1, maxRetries, wm.apiKey, err)
					cancel()
					retryCount++

					if retryCount >= maxRetries {
						log.Errorf("❌ 重连失败次数过多，停止重连逻辑 key:%s", wm.apiKey)
						wm.reconnecting.Store(false)
						return
					}

					delay := baseDelay * time.Duration(1<<retryCount)
					if delay > time.Minute*5 {
						delay = time.Minute * 5
					}
					log.Warnf("等待 %v 后重试...", delay)
					time.Sleep(delay)
					continue
				}

				log.Infof("✅ 重连成功 key:%s", wm.apiKey)
				retryCount = 0
				wm.reconnecting.Store(false)

				// ✅ 重连成功后开启假死检测
				utility.SafeGo(func() { wm.startDeadCheck(newCtx) })

				break
			}
		}
	}
}

// startDeadCheck 替代 Start 中的定时器，绑定连接生命周期
func (wm *BinanceWebSocketManager) startDeadCheck(ctx context.Context) {
	ticker := time.NewTicker(1 * time.Minute)
	defer ticker.Stop()

	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			if wm.isStopped {
				return
			}
			wm.DeadCheck()
		}
	}
}

// 假死检测
func (wm *BinanceWebSocketManager) DeadCheck() {
	subKey := fmt.Sprintf(global.USER_SUBSCRIBE, wm.apiKey)
	val, _ := helper.DefaultRedis.GetString(subKey)
	if val == "" {
		log.Warnf("没有订阅信息，无法进行假死检测")
		return
	}

	var data binancedto.UserSubscribeState
	_ = sonic.Unmarshal([]byte(val), &data)

	var lastTime *time.Time
	if wm.wsType == 0 {
		lastTime = data.SpotLastTime
	} else {
		lastTime = data.FuturesLastTime
	}

	// 定义最大静默时间（超出视为假死）
	var timeout time.Duration
	if wm.wsType == 0 {
		timeout = 40 * time.Second // Spot 每 20s ping，40s 足够
	} else {
		timeout = 6 * time.Minute // Futures 每 3 分钟 ping
	}

	if lastTime != nil && time.Since(*lastTime) > timeout {
		log.Warnf("检测到假死连接 key:%s type:%v, 距离上次通信: %v, 触发重连", wm.apiKey, wm.wsType, time.Since(*lastTime))
		wm.triggerReconnect(true)
	}
}

// 主动心跳发送机制
func (wm *BinanceWebSocketManager) startPingLoop(ctx context.Context) {
	ticker := time.NewTicker(1 * time.Minute)
	defer ticker.Stop()

	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			if wm.isStopped {
				return
			}
			err := wm.ws.WriteMessage(websocket.PingMessage, []byte("ping"))
			if err != nil {
				log.Error("主动 Ping Binance 失败:", err)
				wm.triggerReconnect(false)
			}
		}
	}
}

// 定期删除listenkey 并重启ws
// func (wm *BinanceWebSocketManager) startListenKeyRenewal(ctx context.Context, listenKey string) {
// 	time.Sleep(30 * time.Minute)

// 	select {
// 	case <-ctx.Done():
// 		return
// 	default:
// 		if err := wm.deleteListenKey(listenKey); err != nil {
// 			log.Error("Failed to renew listenKey: ,type:%v key: %s", wm.wsType, wm.apiKey, err)
// 		} else {
// 			log.Debug("Successfully delete listenKey")
// 			wm.triggerReconnect()
// 		}
// 	}

// }

// 定时续期
func (wm *BinanceWebSocketManager) startListenKeyRenewal2(ctx context.Context) {
	ticker := time.NewTicker(30 * time.Minute)
	defer func() {
		log.Debug("定时续期任务退出 key:", wm.apiKey)
		ticker.Stop()
	}()

	for {
		select {
		case <-ticker.C:
			if wm.isStopped {
				return
			}

			if err := wm.renewListenKey(wm.listenKey); err != nil {
				log.Error("Failed to renew listenKey: ,type:%v key: %s", wm.wsType, wm.apiKey, err)
			}
		case <-ctx.Done():
			return
		}
	}
}

/*
删除listenkey
*/
// func (wm *BinanceWebSocketManager) deleteListenKey(listenKey string) error {
// 	client, err := wm.createBinanceClient()
// 	if err != nil {
// 		return err
// 	}

// 	var resp []byte

// 	switch wm.wsType {
// 	case 0:
// 		path := fmt.Sprintf("/api/v3/userDataStream")
// 		params := map[string]interface{}{
// 			"listenKey": listenKey,
// 		}
// 		resp, _, err = client.SendSpotRequestByKey(path, "DELETE", params)

// 		log.Debug(fmt.Sprintf("deleteListenKey resp: %s", string(resp)))
// 	case 1:
// 		resp, _, err = client.SendFuturesRequestByKey("/fapi/v1/listenKey", "DELETE", nil)
// 		log.Debug(fmt.Sprintf("deleteListenKey resp: %s", string(resp)))
// 	default:
// 		return errors.New("unknown ws type")
// 	}

// 	return err
// }

func (wm *BinanceWebSocketManager) renewListenKey(listenKey string) error {
	client, err := wm.createBinanceClient()
	if err != nil {
		return err
	}

	var resp []byte

	switch wm.wsType {
	case 0:
		path := fmt.Sprintf("/api/v3/userDataStream?listenKey=%s", listenKey)
		resp, _, err = client.SendSpotRequestByKey(path, "PUT", nil)
		log.Debug(fmt.Sprintf("renewListenKey resp: %s", string(resp)))
	case 1:
		// path := fmt.Sprintf("/fapi/v1/listenKey", listenKey)
		resp, _, err = client.SendFuturesRequestByKey("/fapi/v1/listenKey", "PUT", nil)
		log.Debug(fmt.Sprintf("renewListenKey resp: %s", string(resp)))
	default:
		return errors.New("unknown ws type")
	}
	return nil
}

func getWsTypeName(wsType int) string {
	switch wsType {
	case 0:
		return "spot"
	case 1:
		return "futures"
	default:
		return "unknown"
	}
}
func getUUID() string {
	return fmt.Sprintf("%s-%s-%s-%s-%s", randomHex(8), randomHex(4), randomHex(4), randomHex(4), randomHex(12))
}

func randomHex(n int) string {
	rand.New(rand.NewSource(time.Now().UnixNano()))
	hexChars := "0123456789abcdef"
	bytes := make([]byte, n)
	for i := 0; i < n; i++ {
		bytes[i] = hexChars[rand.Intn(len(hexChars))]
	}
	return string(bytes)
}