Implement Geyser external deposit backup worker (#194)

Author: jeffyanta

pkg/code/async/geyser/backup.go

@@ -7,6 +7,8 @@ import (
 
 	"github.com/newrelic/go-agent/v3/newrelic"
 
+	"github.com/code-payments/code-server/pkg/code/common"
+	"github.com/code-payments/code-server/pkg/code/data/account"
 	"github.com/code-payments/code-server/pkg/code/data/timelock"
 	"github.com/code-payments/code-server/pkg/database/query"
 	"github.com/code-payments/code-server/pkg/metrics"
@@ -123,9 +125,39 @@ func (p *service) backupExternalDepositWorker(serviceCtx context.Context, interv
 				nr := serviceCtx.Value(metrics.NewRelicContextKey).(*newrelic.Application)
 				m := nr.StartTransaction("async__geyser_consumer_service__backup_external_deposit_worker")
 				defer m.End()
-				// tracedCtx := newrelic.NewContext(serviceCtx, m)
+				tracedCtx := newrelic.NewContext(serviceCtx, m)
+
+				accountInfoRecords, err := p.data.GetPrioritizedAccountInfosRequiringDepositSync(tracedCtx, 256)
+				if err == account.ErrAccountInfoNotFound {
+					return
+				} else if err != nil {
+					log.WithError(err).Warn("failed to get account info records")
+					return
+				}
+
+				var wg sync.WaitGroup
+				for _, accountInfoRecord := range accountInfoRecords {
+					wg.Add(1)
+
+					go func(accountInfoRecord *account.Record) {
+						defer wg.Done()
+
+						authorityAccount, err := common.NewAccountFromPublicKeyString(accountInfoRecord.AuthorityAccount)
+						if err != nil {
+							log.WithError(err).Warn("invalid authority account")
+							return
+						}
+
+						log := log.WithField("authority", authorityAccount.PublicKey().ToBase58())
 
-				// todo: implement me
+						err = fixMissingExternalDeposits(tracedCtx, p.data, p.vmIndexerClient, authorityAccount)
+						if err != nil {
+							log.WithError(err).Warn("failed to fix missing external deposits")
+						}
+					}(accountInfoRecord)
+				}
+
+				wg.Wait()
 			}()
 		case <-serviceCtx.Done():
 			return serviceCtx.Err()

pkg/code/async/geyser/config.go

@@ -22,20 +22,11 @@ const (
 	BackupTimelockWorkerIntervalConfigEnvName = envConfigPrefix + "BACKUP_TIMELOCK_WORKER_INTERVAL"
 	defaultBackupTimelockWorkerInterval       = 1 * time.Minute
 
-	BackupExternalDepositWorkerCountConfigEnvName = envConfigPrefix + "BACKUP_EXTERNAL_DEPOSIT_WORKER_COUNT"
-	defaultBackupExternalDepositWorkerCount       = 32
-
 	BackupExternalDepositWorkerIntervalConfigEnvName = envConfigPrefix + "BACKUP_EXTERNAL_DEPOSIT_WORKER_INTERVAL"
 	defaultBackupExternalDepositWorkerInterval       = 15 * time.Second
 
-	MessagingFeeCollectorPublicKeyConfigEnvName = envConfigPrefix + "MESSAGING_FEE_COLLECTOR_PUBLIC_KEY"
-	defaultMessagingFeeCollectorPublicKey       = "invalid" // ensure something valid is set
-
 	SwapSubsidizerPublicKeyConfigEnvName = envConfigPrefix + "SWAP_SUBSIDIZER_PUBLIC_KEY"
 	defaultSwapSubsidizerPublicKey       = "invalid" // ensure something valid is set
-
-	BackupMessagingWorkerIntervalConfigEnvName = envConfigPrefix + "BACKUP_MESSAGING_WORKER_INTERVAL"
-	defaultBackupMessagingWorkerInterval       = 15 * time.Minute // Decrease significantly once feature is live
 )
 
 type conf struct {
@@ -44,14 +35,10 @@ type conf struct {
 	programUpdateWorkerCount config.Uint64
 	programUpdateQueueSize   config.Uint64
 
-	backupExternalDepositWorkerCount    config.Uint64
 	backupExternalDepositWorkerInterval config.Duration
 
 	backupTimelockWorkerInterval config.Duration
 
-	messagingFeeCollectorPublicKey config.String
-	backupMessagingWorkerInterval  config.Duration
-
 	swapSubsidizerPublicKey config.String
 }
 
@@ -67,14 +54,10 @@ func WithEnvConfigs() ConfigProvider {
 			programUpdateWorkerCount: env.NewUint64Config(ProgramUpdateWorkerCountConfigEnvName, defaultProgramUpdateWorkerCount),
 			programUpdateQueueSize:   env.NewUint64Config(ProgramUpdateQueueSizeConfigEnvName, defaultProgramUpdateQueueSize),
 
-			backupExternalDepositWorkerCount:    env.NewUint64Config(BackupExternalDepositWorkerCountConfigEnvName, defaultBackupExternalDepositWorkerCount),
 			backupExternalDepositWorkerInterval: env.NewDurationConfig(BackupExternalDepositWorkerIntervalConfigEnvName, defaultBackupExternalDepositWorkerInterval),
 
 			backupTimelockWorkerInterval: env.NewDurationConfig(BackupTimelockWorkerIntervalConfigEnvName, defaultBackupTimelockWorkerInterval),
 
-			messagingFeeCollectorPublicKey: env.NewStringConfig(MessagingFeeCollectorPublicKeyConfigEnvName, defaultMessagingFeeCollectorPublicKey),
-			backupMessagingWorkerInterval:  env.NewDurationConfig(BackupMessagingWorkerIntervalConfigEnvName, defaultBackupMessagingWorkerInterval),
-
 			swapSubsidizerPublicKey: env.NewStringConfig(SwapSubsidizerPublicKeyConfigEnvName, defaultSwapSubsidizerPublicKey),
 		}
 	}