Merge pull request #905 from rsksmart/feat/fly2158/transfer_to_cold_wallet_watcher

feat/fly2158/cold to hot wallet transfer 2
Merging this PR (part 2 of the feature) because part 1 (#901) was initially created to give room to colleagues to review it in parallel with the development of this part 2. Since nobody reviewed part 1, this second part is being merged so the full context of the feature is in one PR with everything together.

Author: AndresQuijano

internal/adapters/entrypoints/watcher/common.go

@@ -21,6 +21,7 @@ const (
 	rskEclipseCheckInterval          = 15 * time.Second
 	btcReleaseCheckInterval          = 3 * time.Minute
 	assetMetricsUpdateInterval       = 1 * time.Minute
+	transferColdWalletInterval       = 5 * time.Minute
 )
 
 type Watcher interface {

internal/adapters/entrypoints/watcher/monitoring/cold_wallet_metrics_watcher.go

@@ -0,0 +1,74 @@
+package monitoring
+
+import (
+	"context"
+
+	"github.com/rsksmart/liquidity-provider-server/internal/entities"
+	"github.com/rsksmart/liquidity-provider-server/internal/entities/cold_wallet"
+	log "github.com/sirupsen/logrus"
+)
+
+type ColdWalletMetricsWatcher struct {
+	appMetrics   *Metrics
+	eventBus     entities.EventBus
+	closeChannel chan struct{}
+}
+
+func NewColdWalletMetricsWatcher(
+	appMetrics *Metrics,
+	eventBus entities.EventBus,
+) *ColdWalletMetricsWatcher {
+	closeChannel := make(chan struct{}, 1)
+	return &ColdWalletMetricsWatcher{
+		appMetrics:   appMetrics,
+		eventBus:     eventBus,
+		closeChannel: closeChannel,
+	}
+}
+
+func (watcher *ColdWalletMetricsWatcher) Prepare(ctx context.Context) error {
+	return nil
+}
+
+// nolint: cyclop
+func (watcher *ColdWalletMetricsWatcher) Start() {
+	rbtcThresholdChannel := watcher.eventBus.Subscribe(cold_wallet.RbtcTransferredDueToThresholdEventId)
+	btcThresholdChannel := watcher.eventBus.Subscribe(cold_wallet.BtcTransferredDueToThresholdEventId)
+	rbtcTimeForcingChannel := watcher.eventBus.Subscribe(cold_wallet.RbtcTransferredDueToTimeForcingEventId)
+	btcTimeForcingChannel := watcher.eventBus.Subscribe(cold_wallet.BtcTransferredDueToTimeForcingEventId)
+
+metricLoop:
+	for {
+		select {
+		case event := <-rbtcThresholdChannel:
+			if typedEvent, ok := event.(cold_wallet.RbtcTransferredDueToThresholdEvent); ok {
+				watcher.appMetrics.ColdWalletTransfersMetric.WithLabelValues("rbtc", "threshold").Inc()
+				watcher.appMetrics.ColdWalletLastAmountMetric.WithLabelValues("rbtc").Set(weiToBtcFloat64(typedEvent.Amount))
+			}
+		case event := <-btcThresholdChannel:
+			if typedEvent, ok := event.(cold_wallet.BtcTransferredDueToThresholdEvent); ok {
+				watcher.appMetrics.ColdWalletTransfersMetric.WithLabelValues("btc", "threshold").Inc()
+				watcher.appMetrics.ColdWalletLastAmountMetric.WithLabelValues("btc").Set(weiToBtcFloat64(typedEvent.Amount))
+			}
+		case event := <-rbtcTimeForcingChannel:
+			if typedEvent, ok := event.(cold_wallet.RbtcTransferredDueToTimeForcingEvent); ok {
+				watcher.appMetrics.ColdWalletTransfersMetric.WithLabelValues("rbtc", "time_forcing").Inc()
+				watcher.appMetrics.ColdWalletLastAmountMetric.WithLabelValues("rbtc").Set(weiToBtcFloat64(typedEvent.Amount))
+			}
+		case event := <-btcTimeForcingChannel:
+			if typedEvent, ok := event.(cold_wallet.BtcTransferredDueToTimeForcingEvent); ok {
+				watcher.appMetrics.ColdWalletTransfersMetric.WithLabelValues("btc", "time_forcing").Inc()
+				watcher.appMetrics.ColdWalletLastAmountMetric.WithLabelValues("btc").Set(weiToBtcFloat64(typedEvent.Amount))
+			}
+		case <-watcher.closeChannel:
+			close(watcher.closeChannel)
+			break metricLoop
+		}
+	}
+}
+
+func (watcher *ColdWalletMetricsWatcher) Shutdown(closeChannel chan<- bool) {
+	watcher.closeChannel <- struct{}{}
+	closeChannel <- true
+	log.Debug("Cold wallet metrics watcher shutdown completed")
+}