From 6d95e34e3916a55dc0dc90e061cc6e220de1f5e8 Mon Sep 17 00:00:00 2001 From: Yaroslav Vorobiov Date: Wed, 1 Jun 2022 22:59:47 +0100 Subject: [PATCH] satellite/payments: add payments method for deposit wallets Add payments method to payments to DepositWallets interface. Exposes payments retrieval API for a particular wallet to other systems such as console billing API. Change-Id: Ifcb3a35514aab50be00f6360007954980b5d8b38 --- satellite/api.go | 6 +- satellite/payments/storjscan/chore.go | 17 +-- satellite/payments/storjscan/chore_test.go | 25 ++-- satellite/payments/storjscan/paymentsdb.go | 31 ++--- .../payments/storjscan/paymentsdb_test.go | 61 ++++----- satellite/payments/storjscan/service.go | 64 ++++++--- satellite/payments/storjscan/service_test.go | 125 ++++++++++++++++++ satellite/payments/tokens.go | 25 ++++ satellite/satellitedb/storjscanpayments.go | 7 +- 9 files changed, 269 insertions(+), 92 deletions(-) create mode 100644 satellite/payments/storjscan/service_test.go diff --git a/satellite/api.go b/satellite/api.go index 36736ba8f..73c9097d4 100644 --- a/satellite/api.go +++ b/satellite/api.go @@ -526,10 +526,14 @@ func NewAPI(log *zap.Logger, full *identity.FullIdentity, db DB, pc.Storjscan.Auth.Identifier, pc.Storjscan.Auth.Secret) - peer.Payments.StorjscanService, err = storjscan.NewService(peer.DB, peer.Payments.StorjscanClient) + peer.Payments.StorjscanService = storjscan.NewService(log.Named("storjscan-service"), + peer.DB.Wallets(), + peer.DB.StorjscanPayments(), + peer.Payments.StorjscanClient) if err != nil { return nil, errs.Combine(err, peer.Close()) } + peer.Payments.DepositWallets = peer.Payments.StorjscanService } diff --git a/satellite/payments/storjscan/chore.go b/satellite/payments/storjscan/chore.go index 71bc4972b..a54dbc349 100644 --- a/satellite/payments/storjscan/chore.go +++ b/satellite/payments/storjscan/chore.go @@ -11,6 +11,7 @@ import ( "go.uber.org/zap" "storj.io/common/sync2" + "storj.io/storj/satellite/payments" "storj.io/storj/satellite/payments/monetary" ) @@ -46,7 +47,7 @@ func (chore *Chore) Run(ctx context.Context) (err error) { return chore.TransactionCycle.Run(ctx, func(ctx context.Context) error { var from int64 - blockNumber, err := chore.paymentsDB.LastBlock(ctx, PaymentStatusConfirmed) + blockNumber, err := chore.paymentsDB.LastBlock(ctx, payments.PaymentStatusConfirmed) switch { case err == nil: from = blockNumber + 1 @@ -57,22 +58,22 @@ func (chore *Chore) Run(ctx context.Context) (err error) { return nil } - payments, err := chore.client.Payments(ctx, from) + latestPayments, err := chore.client.Payments(ctx, from) if err != nil { chore.log.Error("error retrieving payments", zap.Error(ChoreErr.Wrap(err))) return nil } - if len(payments.Payments) == 0 { + if len(latestPayments.Payments) == 0 { return nil } var cachedPayments []CachedPayment - for _, payment := range payments.Payments { - var status PaymentStatus - if payments.LatestBlock.Number-payment.BlockNumber >= int64(chore.confirmations) { - status = PaymentStatusConfirmed + for _, payment := range latestPayments.Payments { + var status payments.PaymentStatus + if latestPayments.LatestBlock.Number-payment.BlockNumber >= int64(chore.confirmations) { + status = payments.PaymentStatusConfirmed } else { - status = PaymentStatusPending + status = payments.PaymentStatusPending } cachedPayments = append(cachedPayments, CachedPayment{ diff --git a/satellite/payments/storjscan/chore_test.go b/satellite/payments/storjscan/chore_test.go index d249854a0..546a06641 100644 --- a/satellite/payments/storjscan/chore_test.go +++ b/satellite/payments/storjscan/chore_test.go @@ -17,6 +17,7 @@ import ( "storj.io/common/testcontext" "storj.io/storj/satellite" + "storj.io/storj/satellite/payments" "storj.io/storj/satellite/payments/monetary" "storj.io/storj/satellite/payments/storjscan" "storj.io/storj/satellite/payments/storjscan/blockchaintest" @@ -31,7 +32,7 @@ func TestChore(t *testing.T) { const confirmations = 12 - var payments []storjscan.Payment + var pmnts []storjscan.Payment var cachedPayments []storjscan.CachedPayment latestBlock := storjscan.Header{ @@ -41,7 +42,7 @@ func TestChore(t *testing.T) { } addPayments := func(count int) { - l := len(payments) + l := len(pmnts) for i := l; i < l+count; i++ { payment := storjscan.Payment{ From: blockchaintest.NewAddress(), @@ -53,13 +54,13 @@ func TestChore(t *testing.T) { LogIndex: i, Timestamp: now.Add(time.Duration(i) * time.Second), } - payments = append(payments, payment) + pmnts = append(pmnts, payment) cachedPayments = append(cachedPayments, storjscan.CachedPayment{ From: payment.From, To: payment.To, TokenValue: monetary.AmountFromBaseUnits(payment.TokenValue.Int64(), monetary.StorjToken), - Status: storjscan.PaymentStatusPending, + Status: payments.PaymentStatusPending, BlockHash: payment.BlockHash, BlockNumber: payment.BlockNumber, Transaction: payment.Transaction, @@ -69,15 +70,15 @@ func TestChore(t *testing.T) { } latestBlock = storjscan.Header{ - Hash: payments[len(payments)-1].BlockHash, - Number: payments[len(payments)-1].BlockNumber, - Timestamp: payments[len(payments)-1].Timestamp, + Hash: pmnts[len(pmnts)-1].BlockHash, + Number: pmnts[len(pmnts)-1].BlockNumber, + Timestamp: pmnts[len(pmnts)-1].Timestamp, } for i := 0; i < len(cachedPayments); i++ { if latestBlock.Number-cachedPayments[i].BlockNumber >= confirmations { - cachedPayments[i].Status = storjscan.PaymentStatusConfirmed + cachedPayments[i].Status = payments.PaymentStatusConfirmed } else { - cachedPayments[i].Status = storjscan.PaymentStatusPending + cachedPayments[i].Status = payments.PaymentStatusPending } } } @@ -106,7 +107,7 @@ func TestChore(t *testing.T) { } } - storjscantest.ServePayments(t, w, from, latestBlock, payments) + storjscantest.ServePayments(t, w, from, latestBlock, pmnts) })) defer server.Close() @@ -126,7 +127,7 @@ func TestChore(t *testing.T) { addPayments(100) chore.TransactionCycle.TriggerWait() - last, err := paymentsDB.LastBlock(ctx, storjscan.PaymentStatusPending) + last, err := paymentsDB.LastBlock(ctx, payments.PaymentStatusPending) require.NoError(t, err) require.EqualValues(t, 99, last) actual, err := paymentsDB.List(ctx) @@ -136,7 +137,7 @@ func TestChore(t *testing.T) { addPayments(100) chore.TransactionCycle.TriggerWait() - last, err = paymentsDB.LastBlock(ctx, storjscan.PaymentStatusPending) + last, err = paymentsDB.LastBlock(ctx, payments.PaymentStatusPending) require.NoError(t, err) require.EqualValues(t, 199, last) actual, err = paymentsDB.List(ctx) diff --git a/satellite/payments/storjscan/paymentsdb.go b/satellite/payments/storjscan/paymentsdb.go index 3bb22dea4..9a04cf25a 100644 --- a/satellite/payments/storjscan/paymentsdb.go +++ b/satellite/payments/storjscan/paymentsdb.go @@ -10,6 +10,7 @@ import ( "github.com/zeebo/errs" "storj.io/storj/private/blockchain" + "storj.io/storj/satellite/payments" "storj.io/storj/satellite/payments/monetary" ) @@ -27,30 +28,20 @@ type PaymentsDB interface { // ListWallet returns list of storjscan payments order by block number and log index desc. ListWallet(ctx context.Context, wallet blockchain.Address, limit int, offset int64) ([]CachedPayment, error) // LastBlock returns the highest block known to DB for specified payment status. - LastBlock(ctx context.Context, status PaymentStatus) (int64, error) + LastBlock(ctx context.Context, status payments.PaymentStatus) (int64, error) // DeletePending removes all pending transactions from the DB. DeletePending(ctx context.Context) error } -// PaymentStatus indicates payment status. -type PaymentStatus string - -const ( - // PaymentStatusConfirmed indicates that payment has required number of confirmations. - PaymentStatusConfirmed = "confirmed" - // PaymentStatusPending indicates that payment has not meet confirmation requirements. - PaymentStatusPending = "pending" -) - // CachedPayment holds cached data of storjscan payment. type CachedPayment struct { - From blockchain.Address - To blockchain.Address - TokenValue monetary.Amount - Status PaymentStatus - BlockHash blockchain.Hash - BlockNumber int64 - Transaction blockchain.Hash - LogIndex int - Timestamp time.Time + From blockchain.Address `json:"from"` + To blockchain.Address `json:"to"` + TokenValue monetary.Amount `json:"tokenValue"` + Status payments.PaymentStatus `json:"status"` + BlockHash blockchain.Hash `json:"blockHash"` + BlockNumber int64 `json:"blockNumber"` + Transaction blockchain.Hash `json:"transaction"` + LogIndex int `json:"logIndex"` + Timestamp time.Time `json:"timestamp"` } diff --git a/satellite/payments/storjscan/paymentsdb_test.go b/satellite/payments/storjscan/paymentsdb_test.go index 709403e60..3698cc4ed 100644 --- a/satellite/payments/storjscan/paymentsdb_test.go +++ b/satellite/payments/storjscan/paymentsdb_test.go @@ -15,6 +15,7 @@ import ( "storj.io/common/testrand" "storj.io/storj/private/blockchain" "storj.io/storj/satellite" + "storj.io/storj/satellite/payments" "storj.io/storj/satellite/payments/monetary" "storj.io/storj/satellite/payments/storjscan" "storj.io/storj/satellite/payments/storjscan/blockchaintest" @@ -26,22 +27,22 @@ func TestPaymentsDBInsertBatch(t *testing.T) { paymentsDB := db.StorjscanPayments() now := time.Now().Truncate(time.Second) - var payments []storjscan.CachedPayment + var cachedPayments []storjscan.CachedPayment for i := 0; i < 100; i++ { - payments = append(payments, storjscan.CachedPayment{ + cachedPayments = append(cachedPayments, storjscan.CachedPayment{ From: blockchaintest.NewAddress(), To: blockchaintest.NewAddress(), TokenValue: monetary.AmountFromBaseUnits(1000, monetary.StorjToken), BlockHash: blockchaintest.NewHash(), BlockNumber: int64(i), Transaction: blockchaintest.NewHash(), - Status: storjscan.PaymentStatusConfirmed, + Status: payments.PaymentStatusConfirmed, LogIndex: i, Timestamp: now.Add(time.Duration(i) * time.Second), }) } - err := paymentsDB.InsertBatch(ctx, payments) + err := paymentsDB.InsertBatch(ctx, cachedPayments) require.NoError(t, err) }) } @@ -71,70 +72,70 @@ func TestPaymentsDBList(t *testing.T) { TokenValue: new(big.Int).SetInt64(testrand.Int63n(1000)), Transaction: tx1, Index: 0, - }, storjscan.PaymentStatusConfirmed), + }, payments.PaymentStatusConfirmed), blocks[0].NewPayment(paymentLog{ From: blockchaintest.NewAddress(), To: wallet1, TokenValue: new(big.Int).SetInt64(testrand.Int63n(1000)), Transaction: tx1, Index: 1, - }, storjscan.PaymentStatusConfirmed), + }, payments.PaymentStatusConfirmed), blocks[0].NewPayment(paymentLog{ From: blockchaintest.NewAddress(), To: wallet1, TokenValue: new(big.Int).SetInt64(testrand.Int63n(1000)), Transaction: blockchaintest.NewHash(), Index: 2, - }, storjscan.PaymentStatusConfirmed), + }, payments.PaymentStatusConfirmed), blocks[1].NewPayment(paymentLog{ From: blockchaintest.NewAddress(), To: wallet1, TokenValue: new(big.Int).SetInt64(testrand.Int63n(1000)), Transaction: blockchaintest.NewHash(), Index: 0, - }, storjscan.PaymentStatusConfirmed), + }, payments.PaymentStatusConfirmed), blocks[1].NewPayment(paymentLog{ From: blockchaintest.NewAddress(), To: wallet2, TokenValue: new(big.Int).SetInt64(testrand.Int63n(1000)), Transaction: blockchaintest.NewHash(), Index: 1, - }, storjscan.PaymentStatusConfirmed), + }, payments.PaymentStatusConfirmed), blocks[2].NewPayment(paymentLog{ From: blockchaintest.NewAddress(), To: wallet2, TokenValue: new(big.Int).SetInt64(testrand.Int63n(1000)), Transaction: blockchaintest.NewHash(), Index: 0, - }, storjscan.PaymentStatusConfirmed), + }, payments.PaymentStatusConfirmed), blocks[3].NewPayment(paymentLog{ From: blockchaintest.NewAddress(), To: wallet1, TokenValue: new(big.Int).SetInt64(testrand.Int63n(1000)), Transaction: blockchaintest.NewHash(), Index: 0, - }, storjscan.PaymentStatusConfirmed), + }, payments.PaymentStatusConfirmed), blocks[4].NewPayment(paymentLog{ From: blockchaintest.NewAddress(), To: wallet1, TokenValue: new(big.Int).SetInt64(testrand.Int63n(1000)), Transaction: blockchaintest.NewHash(), Index: 0, - }, storjscan.PaymentStatusPending), + }, payments.PaymentStatusPending), blocks[4].NewPayment(paymentLog{ From: blockchaintest.NewAddress(), To: wallet2, TokenValue: new(big.Int).SetInt64(testrand.Int63n(1000)), Transaction: tx2, Index: 1, - }, storjscan.PaymentStatusPending), + }, payments.PaymentStatusPending), blocks[4].NewPayment(paymentLog{ From: blockchaintest.NewAddress(), To: wallet2, TokenValue: new(big.Int).SetInt64(testrand.Int63n(1000)), Transaction: tx2, Index: 2, - }, storjscan.PaymentStatusPending), + }, payments.PaymentStatusPending), } err := paymentsDB.InsertBatch(ctx, expected) @@ -162,13 +163,13 @@ func TestPaymentsDBLastBlock(t *testing.T) { paymentsDB := db.StorjscanPayments() now := time.Now().Truncate(time.Second) - var payments []storjscan.CachedPayment + var cachedPayments []storjscan.CachedPayment for i := 0; i < 10; i++ { - payments = append(payments, storjscan.CachedPayment{ + cachedPayments = append(cachedPayments, storjscan.CachedPayment{ From: blockchaintest.NewAddress(), To: blockchaintest.NewAddress(), TokenValue: monetary.AmountFromBaseUnits(1000, monetary.StorjToken), - Status: storjscan.PaymentStatusConfirmed, + Status: payments.PaymentStatusConfirmed, BlockHash: blockchaintest.NewHash(), BlockNumber: int64(i), Transaction: blockchaintest.NewHash(), @@ -176,11 +177,11 @@ func TestPaymentsDBLastBlock(t *testing.T) { Timestamp: now.Add(time.Duration(i) * time.Second), }) } - payments = append(payments, storjscan.CachedPayment{ + cachedPayments = append(cachedPayments, storjscan.CachedPayment{ From: blockchaintest.NewAddress(), To: blockchaintest.NewAddress(), TokenValue: monetary.AmountFromBaseUnits(1000, monetary.StorjToken), - Status: storjscan.PaymentStatusPending, + Status: payments.PaymentStatusPending, BlockHash: blockchaintest.NewHash(), BlockNumber: int64(10), Transaction: blockchaintest.NewHash(), @@ -188,16 +189,16 @@ func TestPaymentsDBLastBlock(t *testing.T) { Timestamp: now.Add(time.Duration(10) * time.Second), }) - err := paymentsDB.InsertBatch(ctx, payments) + err := paymentsDB.InsertBatch(ctx, cachedPayments) require.NoError(t, err) t.Run("payment status confirmed", func(t *testing.T) { - last, err := paymentsDB.LastBlock(ctx, storjscan.PaymentStatusConfirmed) + last, err := paymentsDB.LastBlock(ctx, payments.PaymentStatusConfirmed) require.NoError(t, err) require.EqualValues(t, 9, last) }) t.Run("payment status pending", func(t *testing.T) { - last, err := paymentsDB.LastBlock(ctx, storjscan.PaymentStatusPending) + last, err := paymentsDB.LastBlock(ctx, payments.PaymentStatusPending) require.NoError(t, err) require.EqualValues(t, 10, last) }) @@ -206,7 +207,7 @@ func TestPaymentsDBLastBlock(t *testing.T) { func TestPaymentsDBLastBlockNoPayments(t *testing.T) { satellitedbtest.Run(t, func(ctx *testcontext.Context, t *testing.T, db satellite.DB) { - _, err := db.StorjscanPayments().LastBlock(ctx, storjscan.PaymentStatusConfirmed) + _, err := db.StorjscanPayments().LastBlock(ctx, payments.PaymentStatusConfirmed) require.True(t, errs.Is(err, storjscan.ErrNoPayments)) }) } @@ -232,42 +233,42 @@ func TestPaymentsDBDeletePending(t *testing.T) { TokenValue: new(big.Int).SetInt64(testrand.Int63n(1000)), Transaction: blockchaintest.NewHash(), Index: 0, - }, storjscan.PaymentStatusConfirmed), + }, payments.PaymentStatusConfirmed), blocks[1].NewPayment(paymentLog{ From: blockchaintest.NewAddress(), To: blockchaintest.NewAddress(), TokenValue: new(big.Int).SetInt64(testrand.Int63n(1000)), Transaction: blockchaintest.NewHash(), Index: 0, - }, storjscan.PaymentStatusConfirmed), + }, payments.PaymentStatusConfirmed), blocks[2].NewPayment(paymentLog{ From: blockchaintest.NewAddress(), To: blockchaintest.NewAddress(), TokenValue: new(big.Int).SetInt64(testrand.Int63n(1000)), Transaction: blockchaintest.NewHash(), Index: 0, - }, storjscan.PaymentStatusConfirmed), + }, payments.PaymentStatusConfirmed), blocks[3].NewPayment(paymentLog{ From: blockchaintest.NewAddress(), To: blockchaintest.NewAddress(), TokenValue: new(big.Int).SetInt64(testrand.Int63n(1000)), Transaction: blockchaintest.NewHash(), Index: 0, - }, storjscan.PaymentStatusConfirmed), + }, payments.PaymentStatusConfirmed), blocks[4].NewPayment(paymentLog{ From: blockchaintest.NewAddress(), To: blockchaintest.NewAddress(), TokenValue: new(big.Int).SetInt64(testrand.Int63n(1000)), Transaction: blockchaintest.NewHash(), Index: 0, - }, storjscan.PaymentStatusPending), + }, payments.PaymentStatusPending), blocks[5].NewPayment(paymentLog{ From: blockchaintest.NewAddress(), To: blockchaintest.NewAddress(), TokenValue: new(big.Int).SetInt64(testrand.Int63n(1000)), Transaction: blockchaintest.NewHash(), Index: 0, - }, storjscan.PaymentStatusPending), + }, payments.PaymentStatusPending), } require.NoError(t, paymentsDB.InsertBatch(ctx, payments)) @@ -293,7 +294,7 @@ type blockHeader struct { Timestamp time.Time } -func (block blockHeader) NewPayment(log paymentLog, status storjscan.PaymentStatus) storjscan.CachedPayment { +func (block blockHeader) NewPayment(log paymentLog, status payments.PaymentStatus) storjscan.CachedPayment { return storjscan.CachedPayment{ From: log.From, To: log.To, diff --git a/satellite/payments/storjscan/service.go b/satellite/payments/storjscan/service.go index d736ffb51..14d579ea4 100644 --- a/satellite/payments/storjscan/service.go +++ b/satellite/payments/storjscan/service.go @@ -7,47 +7,48 @@ import ( "context" "github.com/zeebo/errs" + "go.uber.org/zap" "storj.io/common/uuid" "storj.io/storj/private/blockchain" "storj.io/storj/satellite/payments" ) -var ( - // Error defines storjscan service error. - Error = errs.Class("storjscan service") -) +// ErrService is storjscan service error class. +var ErrService = errs.Class("storjscan service") // ensures that Wallets implements payments.Wallets. var _ payments.DepositWallets = (*Service)(nil) -// Service is an implementation for payment service via Stripe and Coinpayments. -// -// architecture: Service +// Service exposes API to interact with storjscan payments provider. type Service struct { - walletsDB WalletsDB - storjscanClient *Client + log *zap.Logger + walletsDB WalletsDB + paymentsDB PaymentsDB + client *Client } -// NewService creates a Service instance. -func NewService(db DB, storjscanClient *Client) (*Service, error) { +// NewService creates new storjscan service instance. +func NewService(log *zap.Logger, walletsDB WalletsDB, paymentsDB PaymentsDB, client *Client) *Service { return &Service{ - walletsDB: db.Wallets(), - storjscanClient: storjscanClient, - }, nil + log: log, + walletsDB: walletsDB, + paymentsDB: paymentsDB, + client: client, + } } // Claim gets a new crypto wallet and associates it with a user. func (service *Service) Claim(ctx context.Context, userID uuid.UUID) (_ blockchain.Address, err error) { defer mon.Task()(&ctx)(&err) - address, err := service.storjscanClient.ClaimNewEthAddress(ctx) + address, err := service.client.ClaimNewEthAddress(ctx) if err != nil { - return blockchain.Address{}, Error.Wrap(err) + return blockchain.Address{}, ErrService.Wrap(err) } err = service.walletsDB.Add(ctx, userID, address) if err != nil { - return blockchain.Address{}, Error.Wrap(err) + return blockchain.Address{}, ErrService.Wrap(err) } return address, nil @@ -58,5 +59,32 @@ func (service *Service) Get(ctx context.Context, userID uuid.UUID) (_ blockchain defer mon.Task()(&ctx)(&err) address, err := service.walletsDB.Get(ctx, userID) - return address, Error.Wrap(err) + return address, ErrService.Wrap(err) +} + +// Payments retrieves payments for specific wallet. +func (service *Service) Payments(ctx context.Context, wallet blockchain.Address, limit int, offset int64) (_ []payments.WalletPayment, err error) { + defer mon.Task()(&ctx)(&err) + + cachedPayments, err := service.paymentsDB.ListWallet(ctx, wallet, limit, offset) + if err != nil { + return nil, ErrService.Wrap(err) + } + + var walletPayments []payments.WalletPayment + for _, pmnt := range cachedPayments { + walletPayments = append(walletPayments, payments.WalletPayment{ + From: pmnt.From, + To: pmnt.To, + TokenValue: pmnt.TokenValue, + Status: pmnt.Status, + BlockHash: pmnt.BlockHash, + BlockNumber: pmnt.BlockNumber, + Transaction: pmnt.Transaction, + LogIndex: pmnt.LogIndex, + Timestamp: pmnt.Timestamp, + }) + } + + return walletPayments, nil } diff --git a/satellite/payments/storjscan/service_test.go b/satellite/payments/storjscan/service_test.go new file mode 100644 index 000000000..fcfff2ab8 --- /dev/null +++ b/satellite/payments/storjscan/service_test.go @@ -0,0 +1,125 @@ +// Copyright (C) 2022 Storj Labs, Inc. +// See LICENSE for copying information. + +package storjscan_test + +import ( + "testing" + "time" + + "github.com/stretchr/testify/require" + "go.uber.org/zap/zaptest" + + "storj.io/common/testcontext" + "storj.io/storj/satellite" + "storj.io/storj/satellite/payments" + "storj.io/storj/satellite/payments/monetary" + "storj.io/storj/satellite/payments/storjscan" + "storj.io/storj/satellite/payments/storjscan/blockchaintest" + "storj.io/storj/satellite/satellitedb/satellitedbtest" +) + +func TestServicePayments(t *testing.T) { + satellitedbtest.Run(t, func(ctx *testcontext.Context, t *testing.T, db satellite.DB) { + paymentsDB := db.StorjscanPayments() + now := time.Now().Truncate(time.Second) + + wallet1 := blockchaintest.NewAddress() + wallet2 := blockchaintest.NewAddress() + + walletPayments := []payments.WalletPayment{ + { + From: blockchaintest.NewAddress(), + To: wallet1, + TokenValue: monetary.AmountFromBaseUnits(100, monetary.StorjToken), + Status: payments.PaymentStatusConfirmed, + BlockHash: blockchaintest.NewHash(), + BlockNumber: 0, + Transaction: blockchaintest.NewHash(), + LogIndex: 0, + Timestamp: now, + }, + { + From: blockchaintest.NewAddress(), + To: wallet1, + TokenValue: monetary.AmountFromBaseUnits(100, monetary.StorjToken), + Status: payments.PaymentStatusConfirmed, + BlockHash: blockchaintest.NewHash(), + BlockNumber: 0, + Transaction: blockchaintest.NewHash(), + LogIndex: 1, + Timestamp: now, + }, + { + From: blockchaintest.NewAddress(), + To: wallet2, + TokenValue: monetary.AmountFromBaseUnits(100, monetary.StorjToken), + Status: payments.PaymentStatusConfirmed, + BlockHash: blockchaintest.NewHash(), + BlockNumber: 0, + Transaction: blockchaintest.NewHash(), + LogIndex: 2, + Timestamp: now, + }, + { + From: blockchaintest.NewAddress(), + To: wallet1, + TokenValue: monetary.AmountFromBaseUnits(200, monetary.StorjToken), + Status: payments.PaymentStatusPending, + BlockHash: blockchaintest.NewHash(), + BlockNumber: 1, + Transaction: blockchaintest.NewHash(), + LogIndex: 0, + Timestamp: now.Add(15 * time.Second), + }, + } + + var cachedPayments []storjscan.CachedPayment + for _, pmnt := range walletPayments { + cachedPayments = append(cachedPayments, storjscan.CachedPayment{ + From: pmnt.From, + To: pmnt.To, + TokenValue: pmnt.TokenValue, + Status: pmnt.Status, + BlockHash: pmnt.BlockHash, + BlockNumber: pmnt.BlockNumber, + Transaction: pmnt.Transaction, + LogIndex: pmnt.LogIndex, + Timestamp: pmnt.Timestamp, + }) + } + err := paymentsDB.InsertBatch(ctx, cachedPayments) + require.NoError(t, err) + + service := storjscan.NewService(zaptest.NewLogger(t), db.Wallets(), paymentsDB, nil) + + t.Run("wallet 1", func(t *testing.T) { + expected := []payments.WalletPayment{walletPayments[0], walletPayments[1], walletPayments[3]} + + actual, err := service.Payments(ctx, wallet1, 5, 0) + require.NoError(t, err) + require.Equal(t, expected, actual) + }) + t.Run("wallet 1 from offset", func(t *testing.T) { + expected := []payments.WalletPayment{walletPayments[1], walletPayments[3]} + + actual, err := service.Payments(ctx, wallet1, 5, 1) + require.NoError(t, err) + require.Equal(t, expected, actual) + }) + t.Run("wallet 1 with limit", func(t *testing.T) { + expected := []payments.WalletPayment{walletPayments[0], walletPayments[1]} + + actual, err := service.Payments(ctx, wallet1, 2, 0) + require.NoError(t, err) + require.Equal(t, expected, actual) + }) + t.Run("wallet 2", func(t *testing.T) { + expected := []payments.WalletPayment{walletPayments[2]} + + actual, err := service.Payments(ctx, wallet2, 1, 0) + require.NoError(t, err) + require.Equal(t, expected, actual) + }) + }) +} diff --git a/satellite/payments/tokens.go b/satellite/payments/tokens.go index f3859b171..85ceb784d 100644 --- a/satellite/payments/tokens.go +++ b/satellite/payments/tokens.go @@ -34,6 +34,8 @@ type DepositWallets interface { Claim(ctx context.Context, userID uuid.UUID) (blockchain.Address, error) // Get returns the crypto wallet address associated with the given user. Get(ctx context.Context, userID uuid.UUID) (blockchain.Address, error) + // Payments returns payments for a particular wallet. + Payments(ctx context.Context, wallet blockchain.Address, limit int, offset int64) ([]WalletPayment, error) } // TransactionStatus defines allowed statuses @@ -97,3 +99,26 @@ type DepositBonus struct { Percentage int64 CreatedAt time.Time } + +// PaymentStatus indicates payment status. +type PaymentStatus string + +const ( + // PaymentStatusConfirmed indicates that payment has required number of confirmations. + PaymentStatusConfirmed = "confirmed" + // PaymentStatusPending indicates that payment has not meet confirmation requirements. + PaymentStatusPending = "pending" +) + +// WalletPayment holds storj token payment data. +type WalletPayment struct { + From blockchain.Address `json:"from"` + To blockchain.Address `json:"to"` + TokenValue monetary.Amount `json:"tokenValue"` + Status PaymentStatus `json:"status"` + BlockHash blockchain.Hash `json:"blockHash"` + BlockNumber int64 `json:"blockNumber"` + Transaction blockchain.Hash `json:"transaction"` + LogIndex int `json:"logIndex"` + Timestamp time.Time `json:"timestamp"` +} diff --git a/satellite/satellitedb/storjscanpayments.go b/satellite/satellitedb/storjscanpayments.go index aecff849c..f6c8b717e 100644 --- a/satellite/satellitedb/storjscanpayments.go +++ b/satellite/satellitedb/storjscanpayments.go @@ -9,6 +9,7 @@ import ( "storj.io/private/dbutil/pgutil" "storj.io/storj/private/blockchain" + "storj.io/storj/satellite/payments" "storj.io/storj/satellite/payments/monetary" "storj.io/storj/satellite/payments/storjscan" "storj.io/storj/satellite/satellitedb/dbx" @@ -126,7 +127,7 @@ func (storjscanPayments *storjscanPayments) ListWallet(ctx context.Context, wall } // LastBlock returns the highest block known to DB. -func (storjscanPayments *storjscanPayments) LastBlock(ctx context.Context, status storjscan.PaymentStatus) (_ int64, err error) { +func (storjscanPayments *storjscanPayments) LastBlock(ctx context.Context, status payments.PaymentStatus) (_ int64, err error) { defer mon.Task()(&ctx)(&err) blockNumber, err := storjscanPayments.db.First_StorjscanPayment_BlockNumber_By_Status_OrderBy_Desc_BlockNumber_Desc_LogIndex( @@ -144,7 +145,7 @@ func (storjscanPayments *storjscanPayments) LastBlock(ctx context.Context, statu // DeletePending removes all pending transactions from the DB. func (storjscanPayments storjscanPayments) DeletePending(ctx context.Context) error { _, err := storjscanPayments.db.Delete_StorjscanPayment_By_Status(ctx, - dbx.StorjscanPayment_Status(storjscan.PaymentStatusPending)) + dbx.StorjscanPayment_Status(payments.PaymentStatusPending)) return err } @@ -152,7 +153,7 @@ func (storjscanPayments storjscanPayments) DeletePending(ctx context.Context) er func fromDBXPayment(dbxPmnt *dbx.StorjscanPayment) storjscan.CachedPayment { payment := storjscan.CachedPayment{ TokenValue: monetary.AmountFromBaseUnits(dbxPmnt.TokenValue, monetary.StorjToken), - Status: storjscan.PaymentStatus(dbxPmnt.Status), + Status: payments.PaymentStatus(dbxPmnt.Status), BlockNumber: dbxPmnt.BlockNumber, LogIndex: dbxPmnt.LogIndex, Timestamp: dbxPmnt.Timestamp,