Merge pull request #30 from coinbase/patrick/historical-balance-lookup

[services] Add historical balance lookup

Author: patrick-ogrady

.dockerignore

@@ -1,2 +1,3 @@
 rosetta-bitcoin
 bitcoin-data
+cli-data

go.mod

@@ -5,7 +5,7 @@ go 1.13
 require (
 	github.com/btcsuite/btcd v0.21.0-beta
 	github.com/btcsuite/btcutil v1.0.2
-	github.com/coinbase/rosetta-sdk-go v0.5.7
+	github.com/coinbase/rosetta-sdk-go v0.5.8-0.20201027222031-dd9e29377d5f
 	github.com/dgraph-io/badger/v2 v2.2007.2
 	github.com/grpc-ecosystem/go-grpc-middleware v1.2.2
 	github.com/stretchr/testify v1.6.1

go.sum

@@ -61,8 +61,8 @@ github.com/client9/misspell v0.3.4 h1:ta993UF76GwbvJcIo3Y68y/M3WxlpEHPWIGDkJYwzJ
 github.com/client9/misspell v0.3.4/go.mod h1:qj6jICC3Q7zFZvVWo7KLAzC3yx5G7kyvSDkc90ppPyw=
 github.com/cloudflare/cloudflare-go v0.10.2-0.20190916151808-a80f83b9add9/go.mod h1:1MxXX1Ux4x6mqPmjkUgTP1CdXIBXKX7T+Jk9Gxrmx+U=
 github.com/cncf/udpa/go v0.0.0-20191209042840-269d4d468f6f/go.mod h1:M8M6+tZqaGXZJjfX53e64911xZQV5JYwmTeXPW+k8Sc=
-github.com/coinbase/rosetta-sdk-go v0.5.7 h1:BaR/+O3GzrsyunVNkVQHtjDCcId8G1Fh/RqEbeyExnk=
-github.com/coinbase/rosetta-sdk-go v0.5.7/go.mod h1:l5aNeyeZKBkmWbVdkdLpWdToQ6hTwI7cZ1OU9cMbljY=
+github.com/coinbase/rosetta-sdk-go v0.5.8-0.20201027222031-dd9e29377d5f h1:aWkN9dKMkMMpZKX5QycpePxH176Fj2fNNC7jESfLZw0=
+github.com/coinbase/rosetta-sdk-go v0.5.8-0.20201027222031-dd9e29377d5f/go.mod h1:l5aNeyeZKBkmWbVdkdLpWdToQ6hTwI7cZ1OU9cMbljY=
 github.com/coreos/etcd v3.3.10+incompatible/go.mod h1:uF7uidLiAD3TWHmW31ZFd/JWoc32PjwdhPthX9715RE=
 github.com/coreos/go-etcd v2.0.0+incompatible/go.mod h1:Jez6KQU2B/sWsbdaef3ED8NzMklzPG4d5KIOhIy30Tk=
 github.com/coreos/go-semver v0.2.0/go.mod h1:nnelYz7RCh+5ahJtPPxZlU+153eP4D4r3EedlOD2RNk=

indexer/balance_storage_handler.go

@@ -0,0 +1,46 @@
+// Copyright 2020 Coinbase, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package indexer
+
+import (
+	"context"
+
+	"github.com/coinbase/rosetta-sdk-go/parser"
+	"github.com/coinbase/rosetta-sdk-go/storage"
+	"github.com/coinbase/rosetta-sdk-go/types"
+)
+
+var _ storage.BalanceStorageHandler = (*BalanceStorageHandler)(nil)
+
+// BalanceStorageHandler implements storage.BalanceStorageHandler.
+type BalanceStorageHandler struct{}
+
+// BlockAdded is called whenever a block is committed to BlockStorage.
+func (h *BalanceStorageHandler) BlockAdded(
+	ctx context.Context,
+	block *types.Block,
+	changes []*parser.BalanceChange,
+) error {
+	return nil
+}
+
+// BlockRemoved is called whenever a block is removed from BlockStorage.
+func (h *BalanceStorageHandler) BlockRemoved(
+	ctx context.Context,
+	block *types.Block,
+	changes []*parser.BalanceChange,
+) error {
+	return nil
+}

indexer/balance_storage_helper.go

@@ -0,0 +1,62 @@
+// Copyright 2020 Coinbase, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package indexer
+
+import (
+	"context"
+
+	"github.com/coinbase/rosetta-sdk-go/asserter"
+	"github.com/coinbase/rosetta-sdk-go/parser"
+	"github.com/coinbase/rosetta-sdk-go/storage"
+	"github.com/coinbase/rosetta-sdk-go/types"
+)
+
+var _ storage.BalanceStorageHelper = (*BalanceStorageHelper)(nil)
+
+// BalanceStorageHelper implements storage.BalanceStorageHelper.
+type BalanceStorageHelper struct {
+	a *asserter.Asserter
+}
+
+// AccountBalance attempts to fetch the balance
+// for a missing account in storage.
+func (h *BalanceStorageHelper) AccountBalance(
+	ctx context.Context,
+	account *types.AccountIdentifier,
+	currency *types.Currency,
+	block *types.BlockIdentifier,
+) (*types.Amount, error) {
+	return &types.Amount{
+		Value:    zeroValue,
+		Currency: currency,
+	}, nil
+}
+
+// Asserter returns a *asserter.Asserter.
+func (h *BalanceStorageHelper) Asserter() *asserter.Asserter {
+	return h.a
+}
+
+// BalanceExemptions returns a list of *types.BalanceExemption.
+func (h *BalanceStorageHelper) BalanceExemptions() []*types.BalanceExemption {
+	return []*types.BalanceExemption{}
+}
+
+// ExemptFunc returns a parser.ExemptOperation.
+func (h *BalanceStorageHelper) ExemptFunc() parser.ExemptOperation {
+	return func(op *types.Operation) bool {
+		return false
+	}
+}

indexer/coin_storage_helper.go

@@ -0,0 +1,38 @@
+// Copyright 2020 Coinbase, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package indexer
+
+import (
+	"context"
+
+	"github.com/coinbase/rosetta-sdk-go/storage"
+	"github.com/coinbase/rosetta-sdk-go/types"
+)
+
+var _ storage.CoinStorageHelper = (*CoinStorageHelper)(nil)
+
+// CoinStorageHelper implements storage.CoinStorageHelper.
+type CoinStorageHelper struct {
+	b *storage.BlockStorage
+}
+
+// CurrentBlockIdentifier returns the current head block identifier
+// and is used to comply with the CoinStorageHelper interface.
+func (h *CoinStorageHelper) CurrentBlockIdentifier(
+	ctx context.Context,
+	transaction storage.DatabaseTransaction,
+) (*types.BlockIdentifier, error) {
+	return h.b.GetHeadBlockIdentifierTransactional(ctx, transaction)
+}

indexer/indexer.go

@@ -53,6 +53,9 @@ const (
 	// this is the estimated memory overhead for each
 	// block fetched by the indexer.
 	sizeMultiplier = 15
+
+	// zeroValue is 0 as a string
+	zeroValue = "0"
 )
 
 var (
@@ -74,7 +77,6 @@ type Client interface {
 var _ syncer.Handler = (*Indexer)(nil)
 var _ syncer.Helper = (*Indexer)(nil)
 var _ services.Indexer = (*Indexer)(nil)
-var _ storage.CoinStorageHelper = (*Indexer)(nil)
 
 // Indexer caches blocks and provides balance query functionality.
 type Indexer struct {
@@ -85,11 +87,12 @@ type Indexer struct {
 
 	client Client
 
-	asserter     *asserter.Asserter
-	database     storage.Database
-	blockStorage *storage.BlockStorage
-	coinStorage  *storage.CoinStorage
-	workers      []storage.BlockWorker
+	asserter       *asserter.Asserter
+	database       storage.Database
+	blockStorage   *storage.BlockStorage
+	balanceStorage *storage.BalanceStorage
+	coinStorage    *storage.CoinStorage
+	workers        []storage.BlockWorker
 
 	waiter *waitTable
 }
@@ -197,9 +200,21 @@ func Initialize(
 		asserter:      asserter,
 	}
 
-	coinStorage := storage.NewCoinStorage(localStore, i, asserter)
+	coinStorage := storage.NewCoinStorage(
+		localStore,
+		&CoinStorageHelper{blockStorage},
+		asserter,
+	)
 	i.coinStorage = coinStorage
-	i.workers = []storage.BlockWorker{coinStorage}
+
+	balanceStorage := storage.NewBalanceStorage(localStore)
+	balanceStorage.Initialize(
+		&BalanceStorageHelper{asserter},
+		&BalanceStorageHandler{},
+	)
+	i.balanceStorage = balanceStorage
+
+	i.workers = []storage.BlockWorker{coinStorage, balanceStorage}
 
 	return i, nil
 }
@@ -748,7 +763,11 @@ func (i *Indexer) GetBlockTransaction(
 	blockIdentifier *types.BlockIdentifier,
 	transactionIdentifier *types.TransactionIdentifier,
 ) (*types.Transaction, error) {
-	return i.blockStorage.GetBlockTransaction(ctx, blockIdentifier, transactionIdentifier)
+	return i.blockStorage.GetBlockTransaction(
+		ctx,
+		blockIdentifier,
+		transactionIdentifier,
+	)
 }
 
 // GetCoins returns all unspent coins for a particular *types.AccountIdentifier.
@@ -759,11 +778,42 @@ func (i *Indexer) GetCoins(
 	return i.coinStorage.GetCoins(ctx, accountIdentifier)
 }
 
-// CurrentBlockIdentifier returns the current head block identifier
-// and is used to comply with the CoinStorageHelper interface.
-func (i *Indexer) CurrentBlockIdentifier(
+// GetBalance returns the balance of an account
+// at a particular *types.PartialBlockIdentifier.
+func (i *Indexer) GetBalance(
 	ctx context.Context,
-	transaction storage.DatabaseTransaction,
-) (*types.BlockIdentifier, error) {
-	return i.blockStorage.GetHeadBlockIdentifierTransactional(ctx, transaction)
+	accountIdentifier *types.AccountIdentifier,
+	currency *types.Currency,
+	blockIdentifier *types.PartialBlockIdentifier,
+) (*types.Amount, *types.BlockIdentifier, error) {
+	dbTx := i.database.NewDatabaseTransaction(ctx, false)
+	defer dbTx.Discard(ctx)
+
+	blockResponse, err := i.blockStorage.GetBlockLazyTransactional(
+		ctx,
+		blockIdentifier,
+		dbTx,
+	)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	amount, err := i.balanceStorage.GetBalanceTransactional(
+		ctx,
+		dbTx,
+		accountIdentifier,
+		currency,
+		blockResponse.Block.BlockIdentifier.Index,
+	)
+	if errors.Is(err, storage.ErrAccountMissing) {
+		return &types.Amount{
+			Value:    zeroValue,
+			Currency: currency,
+		}, blockResponse.Block.BlockIdentifier, nil
+	}
+	if err != nil {
+		return nil, nil, err
+	}
+
+	return amount, blockResponse.Block.BlockIdentifier, nil
 }

main.go

@@ -154,7 +154,7 @@ func main() {
 	// requests.
 	asserter, err := asserter.NewServer(
 		bitcoin.OperationTypes,
-		false,
+		services.HistoricalBalanceLookup,
 		[]*types.NetworkIdentifier{cfg.Network},
 		nil,
 	)

mocks/services/indexer.go

@@ -7,10 +7,12 @@
  "http_timeout": 300,
  "max_retries": 5,
  "retry_elapsed_time": 0,
- "max_online_connections": 0,
+ "max_online_connections": 1000,
  "max_sync_concurrency": 0,
  "tip_delay": 1800,
  "log_configuration": false,
+ "compression_disabled": true,
+ "memory_limit_disabled": true,
  "data": {
   "active_reconciliation_concurrency": 0,
   "inactive_reconciliation_concurrency": 0,