satellite/payments/billing: add bonus transactions

- only applies to storjscan transactions
- applies a 10% bonus by default
- bonus transactions have a distinct source "type" to allow for
  filtering on the frontend

Fixes: https://github.com/storj/storj/issues/5702

Change-Id: I32d65f776c58bcb41227ff5bc77a8e4cb62a9add

satellite/core.go

@@ -582,6 +582,7 @@ func New(log *zap.Logger, full *identity.FullIdentity, db DB,
 			peer.DB.Billing(),
 			config.Payments.BillingConfig.Interval,
 			config.Payments.BillingConfig.DisableLoop,
+			config.Payments.BonusRate,
 		)
 		peer.Services.Add(lifecycle.Item{
 			Name:  "billing:chore",

satellite/payments/billing/chore.go

@@ -26,16 +26,18 @@ type Chore struct {
 	TransactionCycle *sync2.Cycle
 
 	disableLoop bool
+	bonusRate   int64
 }
 
 // NewChore creates new chore.
-func NewChore(log *zap.Logger, paymentTypes []PaymentType, transactionsDB TransactionsDB, interval time.Duration, disableLoop bool) *Chore {
+func NewChore(log *zap.Logger, paymentTypes []PaymentType, transactionsDB TransactionsDB, interval time.Duration, disableLoop bool, bonusRate int64) *Chore {
 	return &Chore{
 		log:              log,
 		paymentTypes:     paymentTypes,
 		transactionsDB:   transactionsDB,
 		TransactionCycle: sync2.NewCycle(interval),
 		disableLoop:      disableLoop,
+		bonusRate:        bonusRate,
 	}
 }
 
@@ -61,7 +63,11 @@ func (chore *Chore) Run(ctx context.Context) (err error) {
 				continue
 			}
 			for _, transaction := range transactions {
-				_, err = chore.transactionsDB.Insert(ctx, transaction)
+				if bonus, ok := prepareBonusTransaction(chore.bonusRate, paymentType.Source(), transaction); ok {
+					_, err = chore.transactionsDB.Insert(ctx, transaction, bonus)
+				} else {
+					_, err = chore.transactionsDB.Insert(ctx, transaction)
+				}
 				if err != nil {
 					chore.log.Error("error storing transaction to db", zap.Error(ChoreErr.Wrap(err)))
 					// we need to halt storing transactions if one fails, so that it can be tried again on the next loop.

satellite/payments/billing/chore_test.go

@@ -4,151 +4,216 @@
 package billing_test
 
 import (
+	"bytes"
 	"context"
-	"encoding/json"
+	"fmt"
 	"testing"
 	"time"
 
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"github.com/zeebo/errs"
 	"go.uber.org/zap/zaptest"
 
 	"storj.io/common/currency"
 	"storj.io/common/testcontext"
 	"storj.io/common/testrand"
 	"storj.io/common/uuid"
-	"storj.io/storj/private/blockchain"
 	"storj.io/storj/satellite"
 	"storj.io/storj/satellite/payments/billing"
 	"storj.io/storj/satellite/satellitedb/satellitedbtest"
 )
 
 func TestChore(t *testing.T) {
-	satellitedbtest.Run(t, func(ctx *testcontext.Context, t *testing.T, db satellite.DB) {
+	ts := makeTimestamp()
-		logger := zaptest.NewLogger(t)
 
-		userID := testrand.UUID()
+	const otherSource = "NOT-STORJCAN"
-		billingDB := db.Billing()
 
-		var batch, runningBatch []billing.Transaction
+	var (
-		paymentType := struct{ mockPayment }{}
+		mike   = testrand.UUID()
-		paymentType.mockSource = func() string { return "mockPaymentService" }
+		joe    = testrand.UUID()
-		paymentType.mockType = func() billing.TransactionType { return billing.TransactionTypeCredit }
+		robert = testrand.UUID()
-		paymentType.mockGetNewTransactions = func(ctx context.Context,
+
-			lastTransactionTime time.Time, metadata []byte) ([]billing.Transaction, error) {
+		names = map[uuid.UUID]string{
-			return batch, nil
+			mike:   "mike",
+			joe:    "joe",
+			robert: "robert",
 		}
 
-		chore := billing.NewChore(logger, []billing.PaymentType{paymentType}, billingDB, time.Minute, false)
+		mike1   = makeFakeTransaction(mike, billing.StorjScanSource, billing.TransactionTypeCredit, 1000, ts, `{"fake": "mike1"}`)
+		mike2   = makeFakeTransaction(mike, billing.StorjScanSource, billing.TransactionTypeCredit, 2000, ts.Add(time.Second*2), `{"fake": "mike2"}`)
+		joe1    = makeFakeTransaction(joe, billing.StorjScanSource, billing.TransactionTypeCredit, 500, ts.Add(time.Second), `{"fake": "joe1"}`)
+		joe2    = makeFakeTransaction(joe, billing.StorjScanSource, billing.TransactionTypeDebit, -100, ts.Add(time.Second), `{"fake": "joe1"}`)
+		robert1 = makeFakeTransaction(robert, otherSource, billing.TransactionTypeCredit, 3000, ts.Add(time.Second), `{"fake": "robert1"}`)
+
+		mike1Bonus = makeBonusTransaction(mike, 100, mike1.Timestamp, mike1.Metadata)
+		mike2Bonus = makeBonusTransaction(mike, 200, mike2.Timestamp, mike2.Metadata)
+		joe1Bonus  = makeBonusTransaction(joe, 50, joe1.Timestamp, joe1.Metadata)
+	)
+
+	assertTXs := func(ctx *testcontext.Context, t *testing.T, db billing.TransactionsDB, userID uuid.UUID, expectedTXs []billing.Transaction) {
+		t.Helper()
+
+		actualTXs, err := db.List(ctx, userID)
+		require.NoError(t, err)
+		for i := 0; i < len(expectedTXs) && i < len(actualTXs); i++ {
+			assertTxEqual(t, expectedTXs[i], actualTXs[i], "unexpected transaction at index %d", i)
+		}
+		for i := len(expectedTXs); i < len(actualTXs); i++ {
+			assert.Fail(t, "extra unexpected transaction", "index=%d tx=%+v", i, actualTXs[i])
+		}
+		for i := len(actualTXs); i < len(expectedTXs); i++ {
+			assert.Fail(t, "missing expected transaction", "index=%d tx=%+v", i, expectedTXs[i])
+		}
+	}
+
+	assertBalance := func(ctx *testcontext.Context, t *testing.T, db billing.TransactionsDB, userID uuid.UUID, expected currency.Amount) {
+		t.Helper()
+		actual, err := db.GetBalance(ctx, userID)
+		require.NoError(t, err)
+		assert.Equal(t, expected, actual, "unexpected balance for user %s (%q)", userID, names[userID])
+	}
+
+	runTest := func(ctx *testcontext.Context, t *testing.T, db billing.TransactionsDB, bonusRate int64, mikeTXs, joeTXs, robertTXs []billing.Transaction, mikeBalance, joeBalance, robertBalance currency.Amount) {
+		paymentTypes := []billing.PaymentType{
+			newFakePaymentType(billing.StorjScanSource,
+				[]billing.Transaction{mike1, joe1, joe2},
+				[]billing.Transaction{mike2},
+			),
+			newFakePaymentType(otherSource,
+				[]billing.Transaction{robert1},
+			),
+		}
+
+		chore := billing.NewChore(zaptest.NewLogger(t), paymentTypes, db, time.Hour, false, bonusRate)
 		ctx.Go(func() error {
 			return chore.Run(ctx)
 		})
 		defer ctx.Check(chore.Close)
 
+		// Trigger (at least) two loops to process all batches.
 		chore.TransactionCycle.Pause()
-
+		chore.TransactionCycle.TriggerWait()
-		batch = createBatch(t, userID, 0, 0)
-		runningBatch = append(runningBatch, batch...)
-
 		chore.TransactionCycle.TriggerWait()
 		chore.TransactionCycle.Pause()
 
-		transactions, err := billingDB.List(ctx, userID)
+		assertTXs(ctx, t, db, mike, mikeTXs)
-		require.NoError(t, err)
+		assertTXs(ctx, t, db, joe, joeTXs)
-		require.Equal(t, len(runningBatch), len(transactions))
+		assertTXs(ctx, t, db, robert, robertTXs)
-		for _, act := range transactions {
+		assertBalance(ctx, t, db, mike, mikeBalance)
-			for _, exp := range runningBatch {
+		assertBalance(ctx, t, db, joe, joeBalance)
-				if act.ID == exp.ID {
+		assertBalance(ctx, t, db, robert, robertBalance)
-					compareTransactions(t, exp, act)
+	}
-					break
-				}
-			}
-		}
 
-		batch = createBatch(t, userID, 3, 4)
+	t.Run("without StorjScan bonus", func(t *testing.T) {
-		runningBatch = append(runningBatch, batch...)
+		satellitedbtest.Run(t, func(ctx *testcontext.Context, t *testing.T, db satellite.DB) {
+			runTest(ctx, t, db.Billing(), 0,
+				[]billing.Transaction{mike2, mike1},
+				[]billing.Transaction{joe1, joe2},
+				[]billing.Transaction{robert1},
+				currency.AmountFromBaseUnits(30000000, currency.USDollarsMicro),
+				currency.AmountFromBaseUnits(4000000, currency.USDollarsMicro),
+				currency.AmountFromBaseUnits(30000000, currency.USDollarsMicro),
+			)
+		})
+	})
 
-		chore.TransactionCycle.TriggerWait()
+	t.Run("with StorjScan bonus", func(t *testing.T) {
-		chore.TransactionCycle.Pause()
+		satellitedbtest.Run(t, func(ctx *testcontext.Context, t *testing.T, db satellite.DB) {
-
+			runTest(ctx, t, db.Billing(), 10,
-		transactions, err = billingDB.List(ctx, userID)
+				[]billing.Transaction{mike2, mike2Bonus, mike1, mike1Bonus},
-		require.NoError(t, err)
+				[]billing.Transaction{joe1, joe1Bonus, joe2},
-		require.Equal(t, len(runningBatch), len(transactions))
+				[]billing.Transaction{robert1},
-		for _, act := range transactions {
+				currency.AmountFromBaseUnits(33000000, currency.USDollarsMicro),
-			for _, exp := range runningBatch {
+				currency.AmountFromBaseUnits(4500000, currency.USDollarsMicro),
-				if act.ID == exp.ID {
+				currency.AmountFromBaseUnits(30000000, currency.USDollarsMicro),
-					compareTransactions(t, exp, act)
+			)
-					break
+		})
-				}
-			}
-		}
 	})
 }
 
-func createBatch(t *testing.T, userID uuid.UUID, blockNumber int64, logIndex int) []billing.Transaction {
+func makeFakeTransaction(userID uuid.UUID, source string, typ billing.TransactionType, amountUSD int64, timestamp time.Time, metadata string) billing.Transaction {
-	tenUSD := currency.AmountFromBaseUnits(1000, currency.USDollars)
+	return billing.Transaction{
-	twentyUSD := currency.AmountFromBaseUnits(2000, currency.USDollars)
-	thirtyUSD := currency.AmountFromBaseUnits(3000, currency.USDollars)
-
-	address, err := blockchain.BytesToAddress(testrand.Bytes(20))
-	require.NoError(t, err)
-
-	metadata, err := json.Marshal(map[string]interface{}{
-		"ReferenceID": "some invoice ID",
-		"Wallet":      address.Hex(),
-		"BlockNumber": blockNumber,
-		"LogIndex":    logIndex,
-	})
-	require.NoError(t, err)
-
-	credit10TX := billing.Transaction{
 		UserID:      userID,
-		Amount:      tenUSD,
+		Amount:      currency.AmountFromBaseUnits(amountUSD, currency.USDollars),
-		Description: "credit from mock payment",
+		Description: fmt.Sprintf("%s transaction", source),
-		Source:      "mockPaymentService",
+		Source:      source,
+		Status:      billing.TransactionStatusCompleted,
+		Type:        typ,
+		Metadata:    []byte(metadata),
+		Timestamp:   timestamp,
+	}
+}
+
+func makeBonusTransaction(userID uuid.UUID, amountUSD int64, timestamp time.Time, metadata []byte) billing.Transaction {
+	return billing.Transaction{
+		UserID:      userID,
+		Amount:      currency.AmountFromBaseUnits(amountUSD, currency.USDollars),
+		Description: "STORJ Token Bonus (10%)",
+		Source:      billing.StorjScanBonusSource,
 		Status:      billing.TransactionStatusCompleted,
 		Type:        billing.TransactionTypeCredit,
 		Metadata:    metadata,
-		Timestamp:   time.Now().Add(time.Second),
+		Timestamp:   timestamp,
-		CreatedAt:   time.Now(),
 	}
-
-	credit20TX := billing.Transaction{
-		UserID:      userID,
-		Amount:      twentyUSD,
-		Description: "credit from mock payment",
-		Source:      "mockPaymentService",
-		Status:      billing.TransactionStatusCompleted,
-		Type:        billing.TransactionTypeCredit,
-		Metadata:    metadata,
-		Timestamp:   time.Now().Add(time.Second * 2),
-		CreatedAt:   time.Now(),
-	}
-
-	credit30TX := billing.Transaction{
-		UserID:      userID,
-		Amount:      thirtyUSD,
-		Description: "credit from mock payment",
-		Source:      "mockPaymentService",
-		Status:      billing.TransactionStatusCompleted,
-		Type:        billing.TransactionTypeCredit,
-		Metadata:    metadata,
-		Timestamp:   time.Now().Add(time.Second * 4),
-		CreatedAt:   time.Now(),
-	}
-	return []billing.Transaction{credit10TX, credit20TX, credit30TX}
 }
 
-// setup mock payment type.
+type fakePaymentType struct {
-var _ billing.PaymentType = (*mockPayment)(nil)
+	source              string
-
+	txType              billing.TransactionType
-type mockPayment struct {
+	txBatches           [][]billing.Transaction
-	mockSource             func() string
+	lastTransactionTime time.Time
-	mockType               func() billing.TransactionType
+	lastMetadata        []byte
-	mockGetNewTransactions func(ctx context.Context, lastTransactionTime time.Time, metadata []byte) ([]billing.Transaction, error)
 }
 
-func (t mockPayment) Source() string                { return t.mockSource() }
+func newFakePaymentType(source string, txBatches ...[]billing.Transaction) *fakePaymentType {
-func (t mockPayment) Type() billing.TransactionType { return t.mockType() }
+	return &fakePaymentType{
-func (t mockPayment) GetNewTransactions(ctx context.Context, lastTransactionTime time.Time, metadata []byte) ([]billing.Transaction, error) {
+		source:    source,
-	return t.mockGetNewTransactions(ctx, lastTransactionTime, metadata)
+		txType:    billing.TransactionTypeCredit,
+		txBatches: txBatches,
+	}
+}
+
+func (pt *fakePaymentType) Source() string                { return pt.source }
+func (pt *fakePaymentType) Type() billing.TransactionType { return pt.txType }
+func (pt *fakePaymentType) GetNewTransactions(ctx context.Context, lastTransactionTime time.Time, metadata []byte) ([]billing.Transaction, error) {
+	// Ensure that the chore is passing up the expected fields
+	switch {
+	case !pt.lastTransactionTime.Equal(lastTransactionTime):
+		return nil, errs.New("expected last timestamp %q but got %q", pt.lastTransactionTime, lastTransactionTime)
+	case !bytes.Equal(pt.lastMetadata, metadata):
+		return nil, errs.New("expected metadata %q but got %q", string(pt.lastMetadata), string(metadata))
+	}
+
+	var txs []billing.Transaction
+	if len(pt.txBatches) > 0 {
+		txs = pt.txBatches[0]
+		pt.txBatches = pt.txBatches[1:]
+		if len(txs) > 0 {
+			// Set up the next expected fields
+			pt.lastTransactionTime = txs[len(txs)-1].Timestamp
+			pt.lastMetadata = txs[len(txs)-1].Metadata
+		}
+	}
+	return txs, nil
+}
+
+func assertTxEqual(t *testing.T, exp, act billing.Transaction, msgAndArgs ...interface{}) {
+	// Assert that the actual transaction has a database id and created at date
+	assert.NotZero(t, act.ID)
+	assert.NotEqual(t, time.Time{}, act.CreatedAt)
+
+	act.ID = 0
+	exp.ID = 0
+	act.CreatedAt = time.Time{}
+	exp.CreatedAt = time.Time{}
+
+	// Do a little hack to patch up the currency on the transactions since
+	// the amount loaded from the database is likely in micro dollars.
+	if exp.Amount.Currency() == currency.USDollars && act.Amount.Currency() == currency.USDollarsMicro {
+		exp.Amount = currency.AmountFromDecimal(
+			exp.Amount.AsDecimal().Truncate(act.Amount.Currency().DecimalPlaces()),
+			act.Amount.Currency())
+	}
+	assert.Equal(t, exp, act, msgAndArgs...)
 }

satellite/payments/billing/transactions.go

@@ -5,6 +5,7 @@ package billing
 
 import (
 	"context"
+	"fmt"
 	"time"
 
 	"github.com/zeebo/errs"
@@ -81,6 +82,13 @@ type PaymentType interface {
 	GetNewTransactions(ctx context.Context, lastTransactionTime time.Time, metadata []byte) ([]Transaction, error)
 }
 
+// Well-known PaymentType sources.
+const (
+	StripeSource         = "stripe"
+	StorjScanSource      = "storjscan"
+	StorjScanBonusSource = "storjscanbonus"
+)
+
 // Transaction defines billing related transaction info that is stored in the DB.
 type Transaction struct {
 	ID          int64
@@ -94,3 +102,33 @@ type Transaction struct {
 	Timestamp   time.Time
 	CreatedAt   time.Time
 }
+
+func prepareBonusTransaction(bonusRate int64, source string, transaction Transaction) (Transaction, bool) {
+	// Bonus transactions only apply when enabled (i.e. positive rate) and
+	// for StorjScan transactions.
+	switch {
+	case bonusRate <= 0:
+		return Transaction{}, false
+	case source != StorjScanSource:
+		return Transaction{}, false
+	case transaction.Type != TransactionTypeCredit:
+		// This is defensive. Storjscan shouldn't provide "debit" transactions.
+		return Transaction{}, false
+	}
+
+	return Transaction{
+		UserID:      transaction.UserID,
+		Amount:      calculateBonusAmount(transaction.Amount, bonusRate),
+		Description: fmt.Sprintf("STORJ Token Bonus (%d%%)", bonusRate),
+		Source:      StorjScanBonusSource,
+		Status:      TransactionStatusCompleted,
+		Type:        TransactionTypeCredit,
+		Timestamp:   transaction.Timestamp,
+		Metadata:    append([]byte(nil), transaction.Metadata...),
+	}, true
+}
+
+func calculateBonusAmount(amount currency.Amount, bonusRate int64) currency.Amount {
+	bonusUnits := amount.BaseUnits() * bonusRate / 100
+	return currency.AmountFromBaseUnits(bonusUnits, amount.Currency())
+}

satellite/payments/billing/transactions_test.go

@@ -60,8 +60,7 @@ func TestTransactionsDBList(t *testing.T) {
 			Status:      txStatus,
 			Type:        txType,
 			Metadata:    metadata,
-			Timestamp:   time.Now(),
+			Timestamp:   makeTimestamp(),
-			CreatedAt:   time.Now(),
 		}
 
 		txs = append(txs, createTX)
@@ -115,8 +114,7 @@ func TestTransactionsDBBalance(t *testing.T) {
 		Status:      billing.TransactionStatusCompleted,
 		Type:        billing.TransactionTypeCredit,
 		Metadata:    metadata,
-		Timestamp:   time.Now().Add(time.Second),
+		Timestamp:   makeTimestamp().Add(time.Second),
-		CreatedAt:   time.Now(),
 	}
 
 	credit30TX := billing.Transaction{
@@ -127,8 +125,7 @@ func TestTransactionsDBBalance(t *testing.T) {
 		Status:      billing.TransactionStatusCompleted,
 		Type:        billing.TransactionTypeCredit,
 		Metadata:    metadata,
-		Timestamp:   time.Now().Add(time.Second * 2),
+		Timestamp:   makeTimestamp().Add(time.Second * 2),
-		CreatedAt:   time.Now(),
 	}
 
 	charge20TX := billing.Transaction{
@@ -139,8 +136,7 @@ func TestTransactionsDBBalance(t *testing.T) {
 		Status:      billing.TransactionStatusCompleted,
 		Type:        billing.TransactionTypeDebit,
 		Metadata:    metadata,
-		Timestamp:   time.Now().Add(time.Second * 3),
+		Timestamp:   makeTimestamp().Add(time.Second * 3),
-		CreatedAt:   time.Now(),
 	}
 
 	t.Run("add 10 USD to account", func(t *testing.T) {
@@ -215,8 +211,7 @@ func TestUpdateTransactions(t *testing.T) {
 		Status:      billing.TransactionStatusCompleted,
 		Type:        billing.TransactionTypeCredit,
 		Metadata:    metadata,
-		Timestamp:   time.Now().Add(time.Second),
+		Timestamp:   makeTimestamp().Add(time.Second),
-		CreatedAt:   time.Now(),
 	}
 
 	t.Run("update metadata", func(t *testing.T) {
@@ -257,10 +252,6 @@ func TestUpdateTransactions(t *testing.T) {
 	})
 }
 
-func TestUpdateMetadata(t *testing.T) {
-
-}
-
 // compareTransactions is a helper method to compare tx used to create db entry,
 // with the tx returned from the db. Method doesn't compare created at field, but
 // ensures that is not empty.
@@ -271,14 +262,20 @@ func compareTransactions(t *testing.T, exp, act billing.Transaction) {
 	assert.Equal(t, exp.Status, act.Status)
 	assert.Equal(t, exp.Source, act.Source)
 	assert.Equal(t, exp.Type, act.Type)
-	var expUpdatedMetadata map[string]string
+	var expUpdatedMetadata map[string]interface{}
-	var actUpdatedMetadata map[string]string
+	var actUpdatedMetadata map[string]interface{}
 	err := json.Unmarshal(exp.Metadata, &expUpdatedMetadata)
 	require.NoError(t, err)
 	err = json.Unmarshal(act.Metadata, &actUpdatedMetadata)
 	require.NoError(t, err)
 	assert.Equal(t, expUpdatedMetadata["ReferenceID"], actUpdatedMetadata["ReferenceID"])
 	assert.Equal(t, expUpdatedMetadata["Wallet"], actUpdatedMetadata["Wallet"])
-	assert.WithinDuration(t, exp.Timestamp, act.Timestamp, time.Microsecond) // database timestamps use microsecond precision
+	assert.Equal(t, exp.Timestamp, act.Timestamp)
-	assert.False(t, act.CreatedAt.IsZero())
+	assert.NotEqual(t, time.Time{}, act.CreatedAt)
+}
+
+func makeTimestamp() time.Time {
+	// Truncate to microseconds to paper over a loss of nanosecond precision
+	// going in and out of the database due to timestamp column resolution.
+	return time.Now().Truncate(time.Microsecond)
 }

satellite/payments/storjscan/service.go

@@ -118,7 +118,7 @@ func (service *Service) Payments(ctx context.Context, wallet blockchain.Address,
 
 // Source defines the billing transaction source for storjscan payments.
 func (service *Service) Source() string {
-	return "storjscan"
+	return billing.StorjScanSource
 }
 
 // Type defines the billing transaction type for storjscan payments.

satellite/payments/stripe/service.go

@@ -425,7 +425,7 @@ func (service *Service) createTokenPaymentBillingTransaction(ctx context.Context
 		UserID:      userID,
 		Amount:      currency.AmountFromBaseUnits(amount, currency.USDollars),
 		Description: "Paid Stripe Invoice",
-		Source:      "stripe",
+		Source:      billing.StripeSource,
 		Status:      billing.TransactionStatusPending,
 		Type:        billing.TransactionTypeDebit,
 		Metadata:    metadata,

satellite/payments/stripe/service_test.go

@@ -360,7 +360,7 @@ func TestService_InvoiceItemsFromZeroTokenBalance(t *testing.T) {
 			UserID:      userID,
 			Amount:      currency.AmountFromBaseUnits(1000, currency.USDollars),
 			Description: "token payment credit",
-			Source:      "storjscan",
+			Source:      billing.StorjScanSource,
 			Status:      billing.TransactionStatusCompleted,
 			Type:        billing.TransactionTypeCredit,
 			Metadata:    nil,