ws proxy support (#2308)

* feat: ws proxy

* fix: remove unused type

node/pkg/websocketfetcher/app.go

@@ -115,7 +115,7 @@ func New() *App {
 }
 
 func (a *App) Init(ctx context.Context, opts ...AppOption) error {
-	// TODO: Proxy support
+
 	cexFactories := map[string]func(context.Context, ...common.FetcherOption) (common.FetcherInterface, error){
 		"binance":  binance.New,
 		"coinbase": coinbase.New,
@@ -198,6 +198,8 @@ func (a *App) initializeCex(ctx context.Context, appConfig AppConfig) error {
 	a.buffer = make(chan *common.FeedData, appConfig.BufferSize)
 	a.storeInterval = appConfig.StoreInterval
 
+	wsProxy := os.Getenv("WS_PROXY")
+
 	for name, factory := range appConfig.CexFactories {
 		if _, ok := feedMap[name]; !ok {
 			log.Warn().Msgf("no feeds for %s", name)
@@ -207,6 +209,7 @@ func (a *App) initializeCex(ctx context.Context, appConfig AppConfig) error {
 			ctx,
 			common.WithFeedDataBuffer(a.buffer),
 			common.WithFeedMaps(feedMap[name]),
+			common.WithProxy(wsProxy),
 		)
 		if err != nil {
 			log.Error().Err(err).Msgf("error in creating %s fetcher", name)

node/pkg/websocketfetcher/common/type.go

@@ -23,7 +23,6 @@ const (
 	VolumeFetchTimeout  = 6 * time.Second
 )
 
-type Proxy = types.Proxy
 type Feed = types.Feed
 type FeedData = types.FeedData
 

node/pkg/wss/utils.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"net/http"
 	"net/url"
+	"strings"
 	"time"
 
 	"bisonai.com/miko/node/pkg/utils/retrier"
@@ -16,6 +17,10 @@ import (
 func (ws *WebsocketHelper) Dial(ctx context.Context) error {
 	dialOption := &websocket.DialOptions{}
 	if ws.Proxy != "" {
+		if strings.HasPrefix(ws.Endpoint, "wss") {
+			ws.Endpoint = strings.Replace(ws.Endpoint, "wss", "ws", 1)
+		}
+
 		proxyURL, err := url.Parse(ws.Proxy)
 		if err != nil {
 			return err