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
This commit is contained in:
Yaroslav Vorobiov
parent
51d4e5c275
commit
6d95e34e39
@ -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
|
||||
}
|
||||
|
||||
|
||||
@ -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{
|
||||
|
||||
@ -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)
|
||||
|
||||
@ -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"`
|
||||
}
|
||||
|
||||
@ -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,
|
||||
|
||||
@ -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
|
||||
}
|
||||
|
||||
125
satellite/payments/storjscan/service_test.go
Normal file
125
satellite/payments/storjscan/service_test.go
Normal file
@ -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)
|
||||
})
|
||||
})
|
||||
}
|
||||
@ -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"`
|
||||
}
|
||||
|
||||
@ -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,
|
||||
|
||||
Loading…
Reference in New Issue
Block a user