Add integration into swap worker for push notifications

Author: jeffyanta

pkg/code/async/geyser/external_deposit.go

@@ -377,7 +377,7 @@ func processPotentialExternalDepositIntoVm(ctx context.Context, data code_data.P
 		}
 
 		// Best-effort processing for notification back to the user
-		integration.OnDepositReceived(ctx, ownerAccount, mint, currencyName, uint64(deltaQuarksIntoOmnibus), usdMarketValue)
+		integration.OnDepositReceived(ctx, ownerAccount, mint, currencyName, usdMarketValue)
 
 		return nil
 	default:

pkg/code/async/geyser/integration.go

@@ -8,5 +8,5 @@ import (
 
 // Integration allows for notifications based on events processed by Geyser
 type Integration interface {
-	OnDepositReceived(ctx context.Context, owner, mint *common.Account, currencyName string, quarksReceived uint64, usdMarketValue float64) error
+	OnDepositReceived(ctx context.Context, owner, mint *common.Account, currencyName string, usdMarketValue float64) error
 }

pkg/code/async/swap/integration.go

@@ -0,0 +1,13 @@
+package async_swap
+
+import (
+	"context"
+
+	"github.com/code-payments/code-server/pkg/code/common"
+	"github.com/code-payments/code-server/pkg/currency"
+)
+
+// Integration allows for notifications based on events processed by the swap worker
+type Integration interface {
+	OnSwapFinalized(ctx context.Context, owner, mint *common.Account, currencyName string, region currency.Code, nativeAmount float64) error
+}

pkg/code/async/swap/service.go

@@ -12,23 +12,26 @@ import (
 )
 
 type service struct {
-	log  *logrus.Entry
-	conf *conf
-	data code_data.Provider
+	log         *logrus.Entry
+	conf        *conf
+	data        code_data.Provider
+	integration Integration
 }
 
-func New(data code_data.Provider, configProvider ConfigProvider) async.Service {
+func New(data code_data.Provider, integration Integration, configProvider ConfigProvider) async.Service {
 	return &service{
-		log:  logrus.StandardLogger().WithField("service", "swap"),
-		conf: configProvider(),
-		data: data,
+		log:         logrus.StandardLogger().WithField("service", "swap"),
+		conf:        configProvider(),
+		data:        data,
+		integration: integration,
 	}
 
 }
 
 func (p *service) Start(ctx context.Context, interval time.Duration) error {
 
 	for _, state := range []swap.State{
+		swap.StateCreated,
 		swap.StateFunding,
 		swap.StateSubmitting,
 	} {

pkg/code/async/swap/util.go

@@ -152,6 +152,34 @@ func (p *service) updateBalances(ctx context.Context, record *swap.Record) error
 	})
 }
 
+func (p *service) notifySwapFinalized(ctx context.Context, swapRecord *swap.Record) error {
+	owner, err := common.NewAccountFromPublicKeyString(swapRecord.Owner)
+	if err != nil {
+		return err
+	}
+
+	toMint, err := common.NewAccountFromPublicKeyString(swapRecord.ToMint)
+	if err != nil {
+		return err
+	}
+
+	currencyName := common.CoreMintName
+	if !common.IsCoreMint(toMint) {
+		currencyMetadataRecord, err := p.data.GetCurrencyMetadata(ctx, toMint.PublicKey().ToBase58())
+		if err != nil {
+			return nil
+		}
+		currencyName = currencyMetadataRecord.Name
+	}
+
+	fundingIntentRecord, err := p.data.GetIntent(ctx, swapRecord.FundingId)
+	if err != nil {
+		return err
+	}
+
+	return p.integration.OnSwapFinalized(ctx, owner, toMint, currencyName, fundingIntentRecord.SendPublicPaymentMetadata.ExchangeCurrency, fundingIntentRecord.SendPublicPaymentMetadata.NativeAmount)
+}
+
 func (p *service) markNonceReleasedDueToSubmittedTransaction(ctx context.Context, record *swap.Record) error {
 	err := p.validateSwapState(record, swap.StateSubmitting)
 	if err != nil {

pkg/code/async/swap/worker.go

@@ -73,6 +73,8 @@ func (p *service) worker(serviceCtx context.Context, state swap.State, interval
 
 func (p *service) handle(ctx context.Context, record *swap.Record) error {
 	switch record.State {
+	case swap.StateCreated:
+		return p.handleStateCreated(ctx, record)
 	case swap.StateFunding:
 		return p.handleStateFunding(ctx, record)
 	case swap.StateSubmitting:
@@ -81,6 +83,14 @@ func (p *service) handle(ctx context.Context, record *swap.Record) error {
 	return nil
 }
 
+func (p *service) handleStateCreated(ctx context.Context, record *swap.Record) error {
+	if err := p.validateSwapState(record, swap.StateCreated); err != nil {
+		return err
+	}
+
+	return nil
+}
+
 func (p *service) handleStateFunding(ctx context.Context, record *swap.Record) error {
 	if err := p.validateSwapState(record, swap.StateFunding); err != nil {
 		return err
@@ -118,7 +128,12 @@ func (p *service) handleStateSubmitting(ctx context.Context, record *swap.Record
 			if err != nil {
 				return errors.Wrap(err, "error updating balances")
 			}
-			return p.markSwapFinalized(ctx, record)
+			err = p.markSwapFinalized(ctx, record)
+			if err != nil {
+				return errors.Wrap(err, "error marking swap as finalized")
+			}
+
+			go p.notifySwapFinalized(ctx, record)
 		}
 	}