feat: 接入 CoinGlass 清算热力图隔离链路

internal/worker/coinglass_collector.go

@@ -0,0 +1,132 @@
+package worker
+
+import (
+	"context"
+	"errors"
+	"log/slog"
+	"strings"
+	"sync/atomic"
+	"time"
+
+	"cryptoHermes/internal/entity"
+	cgapi "cryptoHermes/internal/repo/webapi/coinglass"
+	"cryptoHermes/internal/usecase"
+)
+
+type CoinglassCollectorDeps struct {
+	Provider       usecase.CoinglassProvider
+	LiqHeatMapRepo usecase.CoinglassLiqHeatMapRepository
+
+	Symbols            []string
+	LiqHeatMapInterval string
+	LiqHeatMapLimit    int
+	BackoffInitial     time.Duration
+	BackoffMax         time.Duration
+	Logger             *slog.Logger
+}
+
+type CoinglassCollector struct {
+	d CoinglassCollectorDeps
+
+	running atomic.Bool
+
+	fakeSuccessStreak int
+	disabledUntil     time.Time
+	backoff           time.Duration
+	nextAllowed       time.Time
+}
+
+func NewCoinglassCollector(d CoinglassCollectorDeps) *CoinglassCollector {
+	if d.LiqHeatMapInterval == "" {
+		d.LiqHeatMapInterval = "5"
+	}
+	if d.LiqHeatMapLimit <= 0 {
+		d.LiqHeatMapLimit = 288
+	}
+	if d.BackoffInitial <= 0 {
+		d.BackoffInitial = 30 * time.Second
+	}
+	if d.BackoffMax <= 0 {
+		d.BackoffMax = 10 * time.Minute
+	}
+	if d.Logger == nil {
+		d.Logger = slog.Default()
+	}
+	return &CoinglassCollector{d: d}
+}
+
+func (c *CoinglassCollector) CollectLiqHeatMap(ctx context.Context) error {
+	if !c.running.CompareAndSwap(false, true) {
+		c.d.Logger.Info("coinglass_liq_heatmap_skip_running")
+		return nil
+	}
+	defer c.running.Store(false)
+
+	now := time.Now()
+	if now.Before(c.disabledUntil) {
+		c.d.Logger.Warn("coinglass_liq_heatmap_skip_disabled", "until", c.disabledUntil.Format(time.RFC3339))
+		return nil
+	}
+	if now.Before(c.nextAllowed) {
+		c.d.Logger.Warn("coinglass_liq_heatmap_skip_backoff", "until", c.nextAllowed.Format(time.RFC3339))
+		return nil
+	}
+
+	hadError := false
+	for _, symbol := range c.d.Symbols {
+		symbol = strings.ToUpper(strings.TrimSpace(symbol))
+		if symbol == "" {
+			continue
+		}
+
+		item, err := c.d.Provider.GetLiqHeatMap(ctx, symbol, c.d.LiqHeatMapInterval, c.d.LiqHeatMapLimit)
+		if err != nil {
+			hadError = true
+			c.recordLiqHeatMapError(symbol, err)
+			continue
+		}
+
+		if err := c.d.LiqHeatMapRepo.UpsertMany(ctx, []entity.CGLiqHeatMap{*item}); err != nil {
+			hadError = true
+			c.d.Logger.Error("coinglass_liq_heatmap_upsert_failed", "symbol", symbol, "err", err)
+			continue
+		}
+
+		c.fakeSuccessStreak = 0
+		c.d.Logger.Info("coinglass_liq_heatmap_collect_ok", "symbol", symbol, "interval", c.d.LiqHeatMapInterval)
+	}
+
+	if hadError {
+		c.bumpBackoff()
+	} else {
+		c.backoff = 0
+		c.nextAllowed = time.Time{}
+	}
+	return nil
+}
+
+func (c *CoinglassCollector) recordLiqHeatMapError(symbol string, err error) {
+	if errors.Is(err, cgapi.ErrFakeSuccess) {
+		c.fakeSuccessStreak++
+		c.d.Logger.Error("coinglass_liq_heatmap_fake_success", "symbol", symbol, "streak", c.fakeSuccessStreak, "err", err)
+		if c.fakeSuccessStreak >= 3 {
+			c.disabledUntil = time.Now().Add(time.Hour)
+			c.d.Logger.Error("coinglass_liq_heatmap_disabled", "until", c.disabledUntil.Format(time.RFC3339))
+		}
+		return
+	}
+
+	c.d.Logger.Error("coinglass_liq_heatmap_collect_failed", "symbol", symbol, "err", err)
+}
+
+func (c *CoinglassCollector) bumpBackoff() {
+	if c.backoff == 0 {
+		c.backoff = c.d.BackoffInitial
+	} else {
+		c.backoff *= 2
+	}
+	if c.backoff > c.d.BackoffMax {
+		c.backoff = c.d.BackoffMax
+	}
+	c.nextAllowed = time.Now().Add(c.backoff)
+}