Add Geyser app integration for notifications

Author: Jeff Yanta

pkg/code/async/geyser/backup.go

@@ -150,7 +150,7 @@ func (p *service) backupExternalDepositWorker(serviceCtx context.Context, interv
 
 						log := log.WithField("authority", authorityAccount.PublicKey().ToBase58())
 
-						err = fixMissingExternalDeposits(tracedCtx, p.data, p.vmIndexerClient, authorityAccount)
+						err = fixMissingExternalDeposits(tracedCtx, p.data, p.vmIndexerClient, p.integration, authorityAccount)
 						if err != nil {
 							log.WithError(err).Warn("failed to fix missing external deposits")
 						}

pkg/code/async/geyser/external_deposit.go

@@ -38,7 +38,7 @@ var (
 	syncedDepositCache = cache.NewCache(1_000_000)
 )
 
-func fixMissingExternalDeposits(ctx context.Context, data code_data.Provider, vmIndexerClient indexerpb.IndexerClient, userAuthority *common.Account) error {
+func fixMissingExternalDeposits(ctx context.Context, data code_data.Provider, vmIndexerClient indexerpb.IndexerClient, integration Integration, userAuthority *common.Account) error {
 	err := maybeInitiateExternalDepositIntoVm(ctx, data, vmIndexerClient, userAuthority)
 	if err != nil {
 		return errors.Wrap(err, "error depositing into the vm")
@@ -51,7 +51,7 @@ func fixMissingExternalDeposits(ctx context.Context, data code_data.Provider, vm
 
 	var anyError error
 	for _, signature := range signatures {
-		err := processPotentialExternalDepositIntoVm(ctx, data, signature, userAuthority)
+		err := processPotentialExternalDepositIntoVm(ctx, data, integration, signature, userAuthority)
 		if err != nil {
 			anyError = errors.Wrap(err, "error processing signature for external deposit into vm")
 		}
@@ -202,7 +202,7 @@ func findPotentialExternalDepositsIntoVm(ctx context.Context, data code_data.Pro
 	}
 }
 
-func processPotentialExternalDepositIntoVm(ctx context.Context, data code_data.Provider, signature string, userAuthority *common.Account) error {
+func processPotentialExternalDepositIntoVm(ctx context.Context, data code_data.Provider, integration Integration, signature string, userAuthority *common.Account) error {
 	vmDepositAta, err := userAuthority.ToVmDepositAssociatedTokenAccount(common.CodeVmAccount, common.CoreMintAccount)
 	if err != nil {
 		return errors.Wrap(err, "error getting vm deposit ata")
@@ -276,6 +276,11 @@ func processPotentialExternalDepositIntoVm(ctx context.Context, data code_data.P
 			return nil
 		}
 
+		ownerAccount, err := common.NewAccountFromPublicKeyString(accountInfoRecord.OwnerAccount)
+		if err != nil {
+			return errors.Wrap(err, "invalid owner account")
+		}
+
 		usdExchangeRecord, err := data.GetExchangeRate(ctx, currency_lib.USD, time.Now())
 		if err != nil {
 			return errors.Wrap(err, "error getting usd rate")
@@ -287,7 +292,7 @@ func processPotentialExternalDepositIntoVm(ctx context.Context, data code_data.P
 			IntentId:   getExternalDepositIntentID(signature, userVirtualTimelockVaultAccount),
 			IntentType: intent.ExternalDeposit,
 
-			InitiatorOwnerAccount: accountInfoRecord.OwnerAccount,
+			InitiatorOwnerAccount: ownerAccount.PublicKey().ToBase58(),
 
 			ExternalDepositMetadata: &intent.ExternalDepositMetadata{
 				DestinationTokenAccount: userVirtualTimelockVaultAccount.PublicKey().ToBase58(),
@@ -322,6 +327,9 @@ func processPotentialExternalDepositIntoVm(ctx context.Context, data code_data.P
 
 		syncedDepositCache.Insert(cacheKey, true, 1)
 
+		// Best-effort processing for notification back to the user
+		integration.OnDepositReceived(ctx, ownerAccount, uint64(deltaQuarksIntoOmnibus))
+
 		return nil
 	default:
 		syncedDepositCache.Insert(cacheKey, true, 1)

pkg/code/async/geyser/handler.go

@@ -31,13 +31,15 @@ type TokenProgramAccountHandler struct {
 	conf            *conf
 	data            code_data.Provider
 	vmIndexerClient indexerpb.IndexerClient
+	integration     Integration
 }
 
-func NewTokenProgramAccountHandler(conf *conf, data code_data.Provider, vmIndexerClient indexerpb.IndexerClient) ProgramAccountUpdateHandler {
+func NewTokenProgramAccountHandler(conf *conf, data code_data.Provider, vmIndexerClient indexerpb.IndexerClient, integration Integration) ProgramAccountUpdateHandler {
 	return &TokenProgramAccountHandler{
 		conf:            conf,
 		data:            data,
 		vmIndexerClient: vmIndexerClient,
+		integration:     integration,
 	}
 }
 
@@ -97,7 +99,7 @@ func (h *TokenProgramAccountHandler) Handle(ctx context.Context, update *geyserp
 			return nil
 		}
 
-		err = processPotentialExternalDepositIntoVm(ctx, h.data, signature, userAuthorityAccount)
+		err = processPotentialExternalDepositIntoVm(ctx, h.data, h.integration, signature, userAuthorityAccount)
 		if err != nil {
 			return errors.Wrap(err, "error processing signature for external deposit into vm")
 		}
@@ -116,8 +118,8 @@ func (h *TokenProgramAccountHandler) Handle(ctx context.Context, update *geyserp
 	}
 }
 
-func initializeProgramAccountUpdateHandlers(conf *conf, data code_data.Provider, vmIndexerClient indexerpb.IndexerClient) map[string]ProgramAccountUpdateHandler {
+func initializeProgramAccountUpdateHandlers(conf *conf, data code_data.Provider, vmIndexerClient indexerpb.IndexerClient, integration Integration) map[string]ProgramAccountUpdateHandler {
 	return map[string]ProgramAccountUpdateHandler{
-		base58.Encode(token.ProgramKey): NewTokenProgramAccountHandler(conf, data, vmIndexerClient),
+		base58.Encode(token.ProgramKey): NewTokenProgramAccountHandler(conf, data, vmIndexerClient, integration),
 	}
 }

pkg/code/async/geyser/integration.go

@@ -0,0 +1,12 @@
+package async_geyser
+
+import (
+	"context"
+
+	"github.com/code-payments/code-server/pkg/code/common"
+)
+
+// Integration allows for notifications based on events processed by Geyser
+type Integration interface {
+	OnDepositReceived(ctx context.Context, owner *common.Account, quarksReceived uint64) error
+}

pkg/code/async/geyser/service.go

@@ -27,6 +27,8 @@ type service struct {
 	vmIndexerClient indexerpb.IndexerClient
 	conf            *conf
 
+	integration Integration
+
 	programUpdatesChan    chan *geyserpb.SubscribeUpdateAccount
 	programUpdateHandlers map[string]ProgramAccountUpdateHandler
 
@@ -44,15 +46,16 @@ type service struct {
 	backupExternalDepositWorkerStatus bool
 }
 
-func New(data code_data.Provider, vmIndexerClient indexerpb.IndexerClient, configProvider ConfigProvider) async.Service {
+func New(data code_data.Provider, vmIndexerClient indexerpb.IndexerClient, integration Integration, configProvider ConfigProvider) async.Service {
 	conf := configProvider()
 	return &service{
 		log:                        logrus.StandardLogger().WithField("service", "geyser_consumer"),
 		data:                       data,
 		vmIndexerClient:            vmIndexerClient,
 		conf:                       configProvider(),
+		integration:                integration,
 		programUpdatesChan:         make(chan *geyserpb.SubscribeUpdateAccount, conf.programUpdateQueueSize.Get(context.Background())),
-		programUpdateHandlers:      initializeProgramAccountUpdateHandlers(conf, data, vmIndexerClient),
+		programUpdateHandlers:      initializeProgramAccountUpdateHandlers(conf, data, vmIndexerClient, integration),
 		programUpdateWorkerMetrics: make(map[int]*eventWorkerMetrics),
 	}
 }