cerc-io/ipld-eth-server
18
0
Fork 1
Code Issues 10 Pull Requests 1 Actions Packages Projects Releases 37 Wiki Activity

Separate repositories (#25)

Browse Source 
* Separate Repository into multiple Repositories

* Use struct scan for transactions

* Use struct scan for blocks

* Remove unused block repo methods

* Update naming

* Rename / Cleanup repository related fields
This commit is contained in:
Matt K 2018-02-02 15:53:16 -06:00 committed by GitHub
parent 3863bcb614
commit aea9c7b5e2
35 changed files with 1440 additions and 1322 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
cmd/sync.go
View File

@ -5,13 +5,14 @@ import (
"time"
"log"
"github.com/spf13/cobra"
"github.com/vulcanize/vulcanizedb/pkg/core"
"github.com/vulcanize/vulcanizedb/pkg/geth"
"github.com/vulcanize/vulcanizedb/pkg/history"
"github.com/vulcanize/vulcanizedb/pkg/repositories"
"github.com/vulcanize/vulcanizedb/pkg/repositories/postgres"
"github.com/vulcanize/vulcanizedb/utils"
"github.com/spf13/cobra"
"log"
)
// syncCmd represents the sync command
@ -48,7 +49,7 @@ func init() {
syncCmd.Flags().IntVarP(&startingBlockNumber, "starting-block-number", "s", 0, "Block number to start syncing from")
}
func backFillAllBlocks(blockchain core.Blockchain, repository repositories.Postgres, missingBlocksPopulated chan int, startingBlockNumber int64) {
func backFillAllBlocks(blockchain core.Blockchain, repository postgres.DB, missingBlocksPopulated chan int, startingBlockNumber int64) {
go func() {
missingBlocksPopulated <- history.PopulateMissingBlocks(blockchain, repository, startingBlockNumber)
}()

17
db/migrations/1517538273_update_gas_limit_gas_used.down.sql Normal file
View File

@ -0,0 +1,17 @@
BEGIN;
ALTER TABLE blocks
ALTER COLUMN block_gaslimit TYPE DOUBLE PRECISION USING block_gaslimit :: DOUBLE PRECISION;
ALTER TABLE blocks
ALTER COLUMN block_gasused TYPE DOUBLE PRECISION USING block_gasused :: DOUBLE PRECISION;
ALTER TABLE blocks
ALTER COLUMN block_time TYPE DOUBLE PRECISION USING block_time :: DOUBLE PRECISION;
ALTER TABLE blocks
ALTER COLUMN block_reward TYPE NUMERIC USING block_time :: NUMERIC;
ALTER TABLE blocks
ALTER COLUMN block_uncles_reward TYPE NUMERIC USING block_time :: NUMERIC;
COMMIT;

17
db/migrations/1517538273_update_gas_limit_gas_used.up.sql Normal file
View File

@ -0,0 +1,17 @@
BEGIN;
ALTER TABLE blocks
ALTER COLUMN block_gaslimit TYPE BIGINT USING block_gaslimit :: BIGINT;
ALTER TABLE blocks
ALTER COLUMN block_gasused TYPE BIGINT USING block_gasused :: BIGINT;
ALTER TABLE blocks
ALTER COLUMN block_time TYPE BIGINT USING block_time :: BIGINT;
ALTER TABLE blocks
ALTER COLUMN block_reward TYPE DOUBLE PRECISION USING block_time :: DOUBLE PRECISION;
ALTER TABLE blocks
ALTER COLUMN block_uncles_reward TYPE DOUBLE PRECISION USING block_time :: DOUBLE PRECISION;
COMMIT;

10
db/schema.sql
View File

@ -69,9 +69,9 @@ CREATE VIEW block_stats AS
CREATE TABLE blocks (
block_number bigint,
block_gaslimit double precision,
block_gasused double precision,
block_time double precision,
block_gaslimit bigint,
block_gasused bigint,
block_time bigint,
id integer NOT NULL,
block_difficulty bigint,
block_hash character varying(66),
@ -83,8 +83,8 @@ CREATE TABLE blocks (
is_final boolean,
block_miner character varying(42),
block_extra_data character varying,
block_reward numeric,
block_uncles_reward numeric
block_reward double precision,
block_uncles_reward double precision
);

10
integration_test/geth_blockchain_test.go
View File

@ -4,12 +4,12 @@ import (
"io/ioutil"
"log"
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
"github.com/vulcanize/vulcanizedb/pkg/config"
"github.com/vulcanize/vulcanizedb/pkg/geth"
"github.com/vulcanize/vulcanizedb/pkg/history"
"github.com/vulcanize/vulcanizedb/pkg/repositories"
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
"github.com/vulcanize/vulcanizedb/pkg/repositories/inmemory"
)
func init() {
@ -19,12 +19,12 @@ func init() {
var _ = Describe("Reading from the Geth blockchain", func() {
var blockchain *geth.Blockchain
var repository *repositories.InMemory
var repository *inmemory.InMemory
BeforeEach(func() {
cfg, _ := config.NewConfig("private")
blockchain = geth.NewBlockchain(cfg.Client.IPCPath)
repository = repositories.NewInMemory()
repository = inmemory.NewInMemory()
})
It("reads two blocks", func(done Done) {

21
pkg/contract_summary/summary_test.go
View File

@ -3,12 +3,13 @@ package contract_summary_test
import (
"math/big"
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
"github.com/vulcanize/vulcanizedb/pkg/contract_summary"
"github.com/vulcanize/vulcanizedb/pkg/core"
"github.com/vulcanize/vulcanizedb/pkg/fakes"
"github.com/vulcanize/vulcanizedb/pkg/repositories"
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
"github.com/vulcanize/vulcanizedb/pkg/repositories/inmemory"
)
func NewCurrentContractSummary(blockchain core.Blockchain, repository repositories.Repository, contractHash string) (contract_summary.ContractSummary, error) {
@ -19,7 +20,7 @@ var _ = Describe("The contract summary", func() {
Context("when the given contract does not exist", func() {
It("returns an error", func() {
repository := repositories.NewInMemory()
repository := inmemory.NewInMemory()
blockchain := fakes.NewBlockchain()
contractSummary, err := NewCurrentContractSummary(blockchain, repository, "0x123")
@ -31,7 +32,7 @@ var _ = Describe("The contract summary", func() {
Context("when the given contract exists", func() {
It("returns the summary", func() {
repository := repositories.NewInMemory()
repository := inmemory.NewInMemory()
contract := core.Contract{Hash: "0x123"}
repository.CreateContract(contract)
blockchain := fakes.NewBlockchain()
@ -43,7 +44,7 @@ var _ = Describe("The contract summary", func() {
})
It("includes the contract hash in the summary", func() {
repository := repositories.NewInMemory()
repository := inmemory.NewInMemory()
contract := core.Contract{Hash: "0x123"}
repository.CreateContract(contract)
blockchain := fakes.NewBlockchain()
@ -54,7 +55,7 @@ var _ = Describe("The contract summary", func() {
})
It("sets the number of transactions", func() {
repository := repositories.NewInMemory()
repository := inmemory.NewInMemory()
contract := core.Contract{Hash: "0x123"}
repository.CreateContract(contract)
block := core.Block{
@ -72,7 +73,7 @@ var _ = Describe("The contract summary", func() {
})
It("sets the last transaction", func() {
repository := repositories.NewInMemory()
repository := inmemory.NewInMemory()
contract := core.Contract{Hash: "0x123"}
repository.CreateContract(contract)
block := core.Block{
@ -90,7 +91,7 @@ var _ = Describe("The contract summary", func() {
})
It("gets contract state attribute for the contract from the blockchain", func() {
repository := repositories.NewInMemory()
repository := inmemory.NewInMemory()
contract := core.Contract{Hash: "0x123"}
repository.CreateContract(contract)
blockchain := fakes.NewBlockchain()
@ -103,7 +104,7 @@ var _ = Describe("The contract summary", func() {
})
It("gets contract state attribute for the contract from the blockchain at specific block height", func() {
repository := repositories.NewInMemory()
repository := inmemory.NewInMemory()
contract := core.Contract{Hash: "0x123"}
repository.CreateContract(contract)
blockchain := fakes.NewBlockchain()
@ -118,7 +119,7 @@ var _ = Describe("The contract summary", func() {
})
It("gets attributes for the contract from the blockchain", func() {
repository := repositories.NewInMemory()
repository := inmemory.NewInMemory()
contract := core.Contract{Hash: "0x123"}
repository.CreateContract(contract)
blockchain := fakes.NewBlockchain()

30
pkg/core/block.go
View File

@ -1,20 +1,20 @@
package core
type Block struct {
Reward float64
Difficulty int64
ExtraData string
GasLimit int64
GasUsed int64
Hash string
IsFinal bool
Miner string
Nonce string
Number int64
ParentHash string
Size int64
Time int64
Reward float64 `db:"block_reward"`
Difficulty int64 `db:"block_difficulty"`
ExtraData string `db:"block_extra_data"`
GasLimit int64 `db:"block_gaslimit"`
GasUsed int64 `db:"block_gasused"`
Hash string `db:"block_hash"`
IsFinal bool `db:"is_final"`
Miner string `db:"block_miner"`
Nonce string `db:"block_nonce"`
Number int64 `db:"block_number"`
ParentHash string `db:"block_parenthash"`
Size int64 `db:"block_size"`
Time int64 `db:"block_time"`
Transactions []Transaction
UncleHash string
UnclesReward float64
UncleHash string `db:"uncle_hash"`
UnclesReward float64 `db:"block_uncles_reward"`
}

16
pkg/core/transaction.go
View File

@ -1,13 +1,13 @@
package core
type Transaction struct {
Hash string
Data string
Nonce uint64
To string
From string
GasLimit int64
GasPrice int64
Hash string `db:"tx_hash"`
Data string `db:"tx_input_data"`
Nonce uint64 `db:"tx_nonce"`
To string `db:"tx_to"`
From string `db:"tx_from"`
GasLimit int64 `db:"tx_gaslimit"`
GasPrice int64 `db:"tx_gasprice"`
Receipt
Value string
Value string `db:"tx_value"`
}

2
pkg/geth/block_to_core_block.go
View File

@ -5,10 +5,10 @@ import (
"log"
"github.com/vulcanize/vulcanizedb/pkg/core"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/common/hexutil"
"github.com/ethereum/go-ethereum/core/types"
"github.com/vulcanize/vulcanizedb/pkg/core"
"golang.org/x/net/context"
)

14
pkg/history/populate_blocks_test.go
View File

@ -1,12 +1,12 @@
package history_test
import (
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
"github.com/vulcanize/vulcanizedb/pkg/core"
"github.com/vulcanize/vulcanizedb/pkg/fakes"
"github.com/vulcanize/vulcanizedb/pkg/history"
"github.com/vulcanize/vulcanizedb/pkg/repositories"
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
"github.com/vulcanize/vulcanizedb/pkg/repositories/inmemory"
)
var _ = Describe("Populating blocks", func() {
@ -17,7 +17,7 @@ var _ = Describe("Populating blocks", func() {
{Number: 2},
}
blockchain := fakes.NewBlockchainWithBlocks(blocks)
repository := repositories.NewInMemory()
repository := inmemory.NewInMemory()
repository.CreateOrUpdateBlock(core.Block{Number: 2})
blocksAdded := history.PopulateMissingBlocks(blockchain, repository, 1)
@ -40,7 +40,7 @@ var _ = Describe("Populating blocks", func() {
{Number: 12},
{Number: 13},
})
repository := repositories.NewInMemory()
repository := inmemory.NewInMemory()
repository.CreateOrUpdateBlock(core.Block{Number: 1})
repository.CreateOrUpdateBlock(core.Block{Number: 2})
repository.CreateOrUpdateBlock(core.Block{Number: 3})
@ -72,7 +72,7 @@ var _ = Describe("Populating blocks", func() {
{Number: 5},
{Number: 6},
})
repository := repositories.NewInMemory()
repository := inmemory.NewInMemory()
repository.CreateOrUpdateBlock(core.Block{Number: 3})
repository.CreateOrUpdateBlock(core.Block{Number: 6})
@ -89,7 +89,7 @@ var _ = Describe("Populating blocks", func() {
{Number: 4},
{Number: 5},
})
repository := repositories.NewInMemory()
repository := inmemory.NewInMemory()
history.RetrieveAndUpdateBlocks(blockchain, repository, history.MakeRange(2, 5))
Expect(repository.BlockCount()).To(Equal(3))

10
pkg/history/validate_blocks_test.go
View File

@ -6,12 +6,12 @@ import (
"io/ioutil"
"log"
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
"github.com/vulcanize/vulcanizedb/pkg/core"
"github.com/vulcanize/vulcanizedb/pkg/fakes"
"github.com/vulcanize/vulcanizedb/pkg/history"
"github.com/vulcanize/vulcanizedb/pkg/repositories"
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
"github.com/vulcanize/vulcanizedb/pkg/repositories/inmemory"
)
func init() {
@ -47,7 +47,7 @@ var _ = Describe("Blocks validator", func() {
{Number: 6},
{Number: 7},
})
repository := repositories.NewInMemory()
repository := inmemory.NewInMemory()
validator := history.NewBlockValidator(blockchain, repository, 2)
window := validator.ValidateBlocks()
@ -62,7 +62,7 @@ var _ = Describe("Blocks validator", func() {
history.ParsedWindowTemplate.Execute(expectedMessage, history.ValidationWindow{5, 7})
blockchain := fakes.NewBlockchainWithBlocks([]core.Block{})
repository := repositories.NewInMemory()
repository := inmemory.NewInMemory()
validator := history.NewBlockValidator(blockchain, repository, 2)
actualMessage := &bytes.Buffer{}
validator.Log(actualMessage, window)

17
pkg/repositories/in_memory_test.go
View File

@ -1,17 +0,0 @@
package repositories_test
import (
"github.com/vulcanize/vulcanizedb/pkg/core"
"github.com/vulcanize/vulcanizedb/pkg/repositories"
"github.com/vulcanize/vulcanizedb/pkg/repositories/testing"
_ "github.com/lib/pq"
. "github.com/onsi/ginkgo"
)
var _ = Describe("In memory repository", func() {
testing.AssertRepositoryBehavior(func(core.Node) repositories.Repository {
return repositories.NewInMemory()
})
})

13
pkg/repositories/in_memory.go → pkg/repositories/inmemory/in_memory.go
View File

@ -1,4 +1,4 @@
package repositories
package inmemory
import (
"fmt"
@ -7,6 +7,11 @@ import (
"github.com/vulcanize/vulcanizedb/pkg/core"
"github.com/vulcanize/vulcanizedb/pkg/filters"
"github.com/vulcanize/vulcanizedb/pkg/repositories"
)
const (
blocksFromHeadBeforeFinal = 20
)
type InMemory struct {
@ -42,7 +47,7 @@ func (repository *InMemory) FindReceipt(txHash string) (core.Receipt, error) {
if receipt, ok := repository.receipts[txHash]; ok {
return receipt, nil
}
return core.Receipt{}, ErrReceiptDoesNotExist(txHash)
return core.Receipt{}, repositories.ErrReceiptDoesNotExist(txHash)
}
func (repository *InMemory) SetBlocksStatus(chainHead int64) {
@ -89,7 +94,7 @@ func (repository *InMemory) ContractExists(contractHash string) bool {
func (repository *InMemory) FindContract(contractHash string) (core.Contract, error) {
contract, ok := repository.contracts[contractHash]
if !ok {
return core.Contract{}, ErrContractDoesNotExist(contractHash)
return core.Contract{}, repositories.ErrContractDoesNotExist(contractHash)
}
for _, block := range repository.blocks {
for _, transaction := range block.Transactions {
@ -129,7 +134,7 @@ func (repository *InMemory) FindBlockByNumber(blockNumber int64) (core.Block, er
if block, ok := repository.blocks[blockNumber]; ok {
return block, nil
}
return core.Block{}, ErrBlockDoesNotExist(blockNumber)
return core.Block{}, repositories.ErrBlockDoesNotExist(blockNumber)
}
func (repository *InMemory) MaxBlockNumber() int64 {

538
pkg/repositories/postgres.go
View File

@ -1,538 +0,0 @@
package repositories
import (
"database/sql"
"context"
"errors"
"fmt"
"github.com/vulcanize/vulcanizedb/pkg/config"
"github.com/vulcanize/vulcanizedb/pkg/core"
"github.com/vulcanize/vulcanizedb/pkg/filters"
"github.com/jmoiron/sqlx"
_ "github.com/lib/pq"
)
type Postgres struct {
Db *sqlx.DB
node core.Node
nodeId int64
}
var (
ErrDBInsertFailed = errors.New("postgres: insert failed")
ErrDBDeleteFailed = errors.New("postgres: delete failed")
ErrDBConnectionFailed = errors.New("postgres: db connection failed")
ErrUnableToSetNode = errors.New("postgres: unable to set node")
)
var ErrReceiptDoesNotExist = func(txHash string) error {
return errors.New(fmt.Sprintf("Receipt for tx: %v does not exist", txHash))
}
var ErrContractDoesNotExist = func(contractHash string) error {
return errors.New(fmt.Sprintf("Contract %v does not exist", contractHash))
}
var ErrBlockDoesNotExist = func(blockNumber int64) error {
return errors.New(fmt.Sprintf("Block number %d does not exist", blockNumber))
}
func NewPostgres(databaseConfig config.Database, node core.Node) (Postgres, error) {
connectString := config.DbConnectionString(databaseConfig)
db, err := sqlx.Connect("postgres", connectString)
if err != nil {
return Postgres{}, ErrDBConnectionFailed
}
pg := Postgres{Db: db, node: node}
err = pg.CreateNode(&node)
if err != nil {
return Postgres{}, ErrUnableToSetNode
}
return pg, nil
}
func (repository Postgres) SetBlocksStatus(chainHead int64) {
cutoff := chainHead - blocksFromHeadBeforeFinal
repository.Db.Exec(`
UPDATE blocks SET is_final = TRUE
WHERE is_final = FALSE AND block_number < $1`,
cutoff)
}
func (repository Postgres) FindLogs(address string, blockNumber int64) []core.Log {
logRows, _ := repository.Db.Query(
`SELECT block_number,
address,
tx_hash,
index,
topic0,
topic1,
topic2,
topic3,
data
FROM logs
WHERE address = $1 AND block_number = $2
ORDER BY block_number DESC`, address, blockNumber)
return repository.loadLogs(logRows)
}
func (repository *Postgres) CreateNode(node *core.Node) error {
var nodeId int64
err := repository.Db.QueryRow(
`INSERT INTO nodes (genesis_block, network_id, node_id, client_name)
VALUES ($1, $2, $3, $4)
ON CONFLICT (genesis_block, network_id, node_id)
DO UPDATE
SET genesis_block = $1,
network_id = $2,
node_id = $3,
client_name = $4
RETURNING id`,
node.GenesisBlock, node.NetworkId, node.Id, node.ClientName).Scan(&nodeId)
if err != nil {
return ErrUnableToSetNode
}
repository.nodeId = nodeId
return nil
}
func (repository Postgres) CreateContract(contract core.Contract) error {
abi := contract.Abi
var abiToInsert *string
if abi != "" {
abiToInsert = &abi
}
_, err := repository.Db.Exec(
`INSERT INTO watched_contracts (contract_hash, contract_abi)
VALUES ($1, $2)
ON CONFLICT (contract_hash)
DO UPDATE
SET contract_hash = $1, contract_abi = $2
`, contract.Hash, abiToInsert)
if err != nil {
return ErrDBInsertFailed
}
return nil
}
func (repository Postgres) ContractExists(contractHash string) bool {
var exists bool
repository.Db.QueryRow(
`SELECT exists(
SELECT 1
FROM watched_contracts
WHERE contract_hash = $1)`, contractHash).Scan(&exists)
return exists
}
func (repository Postgres) FindContract(contractHash string) (core.Contract, error) {
var hash string
var abi string
contract := repository.Db.QueryRow(
`SELECT contract_hash, contract_abi FROM watched_contracts WHERE contract_hash=$1`, contractHash)
err := contract.Scan(&hash, &abi)
if err == sql.ErrNoRows {
return core.Contract{}, ErrContractDoesNotExist(contractHash)
}
savedContract := repository.addTransactions(core.Contract{Hash: hash, Abi: abi})
return savedContract, nil
}
func (repository Postgres) MaxBlockNumber() int64 {
var highestBlockNumber int64
repository.Db.Get(&highestBlockNumber, `SELECT MAX(block_number) FROM blocks`)
return highestBlockNumber
}
func (repository Postgres) MissingBlockNumbers(startingBlockNumber int64, highestBlockNumber int64) []int64 {
numbers := make([]int64, 0)
repository.Db.Select(&numbers,
`SELECT all_block_numbers
FROM (
SELECT generate_series($1::INT, $2::INT) AS all_block_numbers) series
LEFT JOIN blocks
ON block_number = all_block_numbers
WHERE block_number ISNULL`,
startingBlockNumber,
highestBlockNumber)
return numbers
}
func (repository Postgres) FindBlockByNumber(blockNumber int64) (core.Block, error) {
blockRows := repository.Db.QueryRow(
`SELECT id,
block_number,
block_gaslimit,
block_gasused,
block_time,
block_difficulty,
block_hash,
block_nonce,
block_parenthash,
block_size,
uncle_hash,
is_final,
block_miner,
block_extra_data,
block_reward,
block_uncles_reward
FROM blocks
WHERE node_id = $1 AND block_number = $2`, repository.nodeId, blockNumber)
savedBlock, err := repository.loadBlock(blockRows)
if err != nil {
switch err {
case sql.ErrNoRows:
return core.Block{}, ErrBlockDoesNotExist(blockNumber)
default:
return savedBlock, err
}
}
return savedBlock, nil
}
func (repository Postgres) BlockCount() int {
var count int
repository.Db.Get(&count, `SELECT COUNT(*) FROM blocks`)
return count
}
func (repository Postgres) getBlockHash(block core.Block) (string, bool) {
var retrievedBlockHash string
repository.Db.Get(&retrievedBlockHash,
`SELECT block_hash
FROM blocks
WHERE block_number = $1 AND node_id = $2`,
block.Number, repository.nodeId)
return retrievedBlockHash, blockExists(retrievedBlockHash)
}
func blockExists(retrievedBlockHash string) bool {
return retrievedBlockHash != ""
}
func (repository Postgres) CreateOrUpdateBlock(block core.Block) error {
var err error
retrievedBlockHash, ok := repository.getBlockHash(block)
if !ok {
err = repository.insertBlock(block)
return err
}
if ok && retrievedBlockHash != block.Hash {
err = repository.removeBlock(block.Number)
if err != nil {
return err
}
err = repository.insertBlock(block)
return err
}
return nil
}
func (repository Postgres) insertBlock(block core.Block) error {
var blockId int64
tx, _ := repository.Db.BeginTx(context.Background(), nil)
err := tx.QueryRow(
`INSERT INTO blocks
(node_id, block_number, block_gaslimit, block_gasused, block_time, block_difficulty, block_hash, block_nonce, block_parenthash, block_size, uncle_hash, is_final, block_miner, block_extra_data, block_reward, block_uncles_reward)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16)
RETURNING id `,
repository.nodeId, block.Number, block.GasLimit, block.GasUsed, block.Time, block.Difficulty, block.Hash, block.Nonce, block.ParentHash, block.Size, block.UncleHash, block.IsFinal, block.Miner, block.ExtraData, block.Reward, block.UnclesReward).
Scan(&blockId)
if err != nil {
tx.Rollback()
return ErrDBInsertFailed
}
err = repository.createTransactions(tx, blockId, block.Transactions)
if err != nil {
tx.Rollback()
return ErrDBInsertFailed
}
tx.Commit()
return nil
}
func (repository Postgres) removeBlock(blockNumber int64) error {
_, err := repository.Db.Exec(
`DELETE FROM
blocks
WHERE block_number=$1 AND node_id=$2`,
blockNumber, repository.nodeId)
if err != nil {
return ErrDBDeleteFailed
}
return nil
}
func (repository Postgres) FindReceipt(txHash string) (core.Receipt, error) {
row := repository.Db.QueryRow(
`SELECT contract_address,
tx_hash,
cumulative_gas_used,
gas_used,
state_root,
status
FROM receipts
WHERE tx_hash = $1`, txHash)
receipt, err := loadReceipt(row)
if err != nil {
switch err {
case sql.ErrNoRows:
return core.Receipt{}, ErrReceiptDoesNotExist(txHash)
default:
return core.Receipt{}, err
}
}
return receipt, nil
}
func (repository Postgres) createTransactions(tx *sql.Tx, blockId int64, transactions []core.Transaction) error {
for _, transaction := range transactions {
err := repository.createTransaction(tx, blockId, transaction)
if err != nil {
return err
}
}
return nil
}
func (repository Postgres) createTransaction(tx *sql.Tx, blockId int64, transaction core.Transaction) error {
var transactionId int
err := tx.QueryRow(
`INSERT INTO transactions
(block_id, tx_hash, tx_nonce, tx_to, tx_from, tx_gaslimit, tx_gasprice, tx_value, tx_input_data)
VALUES ($1, $2, $3, $4, $5, $6, $7, cast(NULLIF($8, '') AS NUMERIC), $9)
RETURNING id`,
blockId, transaction.Hash, transaction.Nonce, transaction.To, transaction.From, transaction.GasLimit, transaction.GasPrice, transaction.Value, transaction.Data).
Scan(&transactionId)
if err != nil {
return err
}
if hasReceipt(transaction) {
receiptId, err := repository.createReceipt(tx, transactionId, transaction.Receipt)
if err != nil {
return err
}
if hasLogs(transaction) {
err = repository.createLogs(tx, transaction.Receipt.Logs, receiptId)
if err != nil {
return err
}
}
}
return nil
}
func hasLogs(transaction core.Transaction) bool {
return len(transaction.Receipt.Logs) > 0
}
func hasReceipt(transaction core.Transaction) bool {
return transaction.Receipt.TxHash != ""
}
func (repository Postgres) createReceipt(tx *sql.Tx, transactionId int, receipt core.Receipt) (int, error) {
//Not currently persisting log bloom filters
var receiptId int
err := tx.QueryRow(
`INSERT INTO receipts
(contract_address, tx_hash, cumulative_gas_used, gas_used, state_root, status, transaction_id)
VALUES ($1, $2, $3, $4, $5, $6, $7)
RETURNING id`,
receipt.ContractAddress, receipt.TxHash, receipt.CumulativeGasUsed, receipt.GasUsed, receipt.StateRoot, receipt.Status, transactionId).Scan(&receiptId)
if err != nil {
return receiptId, err
}
return receiptId, nil
}
func (repository Postgres) createLogs(tx *sql.Tx, logs []core.Log, receiptId int) error {
for _, tlog := range logs {
_, err := tx.Exec(
`INSERT INTO logs (block_number, address, tx_hash, index, topic0, topic1, topic2, topic3, data, receipt_id)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)
`,
tlog.BlockNumber, tlog.Address, tlog.TxHash, tlog.Index, tlog.Topics[0], tlog.Topics[1], tlog.Topics[2], tlog.Topics[3], tlog.Data, receiptId,
)
if err != nil {
return ErrDBInsertFailed
}
}
return nil
}
func (repository Postgres) CreateLogs(logs []core.Log) error {
tx, _ := repository.Db.BeginTx(context.Background(), nil)
for _, tlog := range logs {
_, err := tx.Exec(
`INSERT INTO logs (block_number, address, tx_hash, index, topic0, topic1, topic2, topic3, data)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
`,
tlog.BlockNumber, tlog.Address, tlog.TxHash, tlog.Index, tlog.Topics[0], tlog.Topics[1], tlog.Topics[2], tlog.Topics[3], tlog.Data,
)
if err != nil {
tx.Rollback()
return ErrDBInsertFailed
}
}
tx.Commit()
return nil
}
func (repository Postgres) AddFilter(query filters.LogFilter) error {
_, err := repository.Db.Exec(
`INSERT INTO log_filters
(name, from_block, to_block, address, topic0, topic1, topic2, topic3)
VALUES ($1, NULLIF($2, -1), NULLIF($3, -1), $4, NULLIF($5, ''), NULLIF($6, ''), NULLIF($7, ''), NULLIF($8, ''))`,
query.Name, query.FromBlock, query.ToBlock, query.Address, query.Topics[0], query.Topics[1], query.Topics[2], query.Topics[3])
if err != nil {
return err
}
return nil
}
func loadReceipt(receiptsRow *sql.Row) (core.Receipt, error) {
var contractAddress string
var txHash string
var cumulativeGasUsed int64
var gasUsed int64
var stateRoot string
var status int
err := receiptsRow.Scan(&contractAddress, &txHash, &cumulativeGasUsed, &gasUsed, &stateRoot, &status)
return core.Receipt{
TxHash: txHash,
ContractAddress: contractAddress,
CumulativeGasUsed: cumulativeGasUsed,
GasUsed: gasUsed,
StateRoot: stateRoot,
Status: status,
}, err
}
func (repository Postgres) loadBlock(blockRows *sql.Row) (core.Block, error) {
var blockId int64
var blockHash string
var blockNonce string
var blockNumber int64
var blockMiner string
var blockExtraData string
var blockParentHash string
var blockSize int64
var blockTime float64
var blockReward float64
var difficulty int64
var gasLimit float64
var gasUsed float64
var uncleHash string
var unclesReward float64
var isFinal bool
err := blockRows.Scan(&blockId, &blockNumber, &gasLimit, &gasUsed, &blockTime, &difficulty, &blockHash, &blockNonce, &blockParentHash, &blockSize, &uncleHash, &isFinal, &blockMiner, &blockExtraData, &blockReward, &unclesReward)
if err != nil {
return core.Block{}, err
}
transactionRows, _ := repository.Db.Query(`
SELECT tx_hash,
tx_nonce,
tx_to,
tx_from,
tx_gaslimit,
tx_gasprice,
tx_value,
tx_input_data
FROM transactions
WHERE block_id = $1
ORDER BY tx_hash`, blockId)
transactions := repository.loadTransactions(transactionRows)
return core.Block{
Reward: blockReward,
Difficulty: difficulty,
ExtraData: blockExtraData,
GasLimit: int64(gasLimit),
GasUsed: int64(gasUsed),
Hash: blockHash,
IsFinal: isFinal,
Miner: blockMiner,
Nonce: blockNonce,
Number: blockNumber,
ParentHash: blockParentHash,
Size: blockSize,
Time: int64(blockTime),
Transactions: transactions,
UncleHash: uncleHash,
UnclesReward: unclesReward,
}, nil
}
func (repository Postgres) loadLogs(logsRows *sql.Rows) []core.Log {
var logs []core.Log
for logsRows.Next() {
var blockNumber int64
var address string
var txHash string
var index int64
var data string
var topics core.Topics
logsRows.Scan(&blockNumber, &address, &txHash, &index, &topics[0], &topics[1], &topics[2], &topics[3], &data)
log := core.Log{
BlockNumber: blockNumber,
TxHash: txHash,
Address: address,
Index: index,
Data: data,
}
for i, topic := range topics {
log.Topics[i] = topic
}
logs = append(logs, log)
}
return logs
}
func (repository Postgres) loadTransactions(transactionRows *sql.Rows) []core.Transaction {
var transactions []core.Transaction
for transactionRows.Next() {
var hash string
var nonce uint64
var to string
var from string
var gasLimit int64
var gasPrice int64
var inputData string
var value string
transactionRows.Scan(&hash, &nonce, &to, &from, &gasLimit, &gasPrice, &value, &inputData)
transaction := core.Transaction{
Hash: hash,
Nonce: nonce,
To: to,
From: from,
GasLimit: gasLimit,
GasPrice: gasPrice,
Value: value,
Data: inputData,
}
transactions = append(transactions, transaction)
}
return transactions
}
func (repository Postgres) addTransactions(contract core.Contract) core.Contract {
transactionRows, _ := repository.Db.Query(`
SELECT tx_hash,
tx_nonce,
tx_to,
tx_from,
tx_gaslimit,
tx_gasprice,
tx_value,
tx_input_data
FROM transactions
WHERE tx_to = $1
ORDER BY block_id DESC`, contract.Hash)
transactions := repository.loadTransactions(transactionRows)
savedContract := core.Contract{Hash: contract.Hash, Transactions: transactions, Abi: contract.Abi}
return savedContract
}

266
pkg/repositories/postgres/block_repository.go Normal file
View File

@ -0,0 +1,266 @@
package postgres
import (
"context"
"database/sql"
"fmt"
"log"
"github.com/jmoiron/sqlx"
"github.com/vulcanize/vulcanizedb/pkg/core"
"github.com/vulcanize/vulcanizedb/pkg/repositories"
)
const (
blocksFromHeadBeforeFinal = 20
)
func (db DB) SetBlocksStatus(chainHead int64) {
cutoff := chainHead - blocksFromHeadBeforeFinal
db.DB.Exec(`
UPDATE blocks SET is_final = TRUE
WHERE is_final = FALSE AND block_number < $1`,
cutoff)
}
func (db DB) CreateOrUpdateBlock(block core.Block) error {
var err error
retrievedBlockHash, ok := db.getBlockHash(block)
if !ok {
err = db.insertBlock(block)
return err
}
if ok && retrievedBlockHash != block.Hash {
err = db.removeBlock(block.Number)
if err != nil {
return err
}
err = db.insertBlock(block)
return err
}
return nil
}
func (db DB) MissingBlockNumbers(startingBlockNumber int64, highestBlockNumber int64) []int64 {
numbers := make([]int64, 0)
db.DB.Select(&numbers,
`SELECT all_block_numbers
FROM (
SELECT generate_series($1::INT, $2::INT) AS all_block_numbers) series
LEFT JOIN blocks
ON block_number = all_block_numbers
WHERE block_number ISNULL`,
startingBlockNumber,
highestBlockNumber)
return numbers
}
func (db DB) FindBlockByNumber(blockNumber int64) (core.Block, error) {
blockRows := db.DB.QueryRowx(
`SELECT id,
block_number,
block_gaslimit,
block_gasused,
block_time,
block_difficulty,
block_hash,
block_nonce,
block_parenthash,
block_size,
uncle_hash,
is_final,
block_miner,
block_extra_data,
block_reward,
block_uncles_reward
FROM blocks
WHERE node_id = $1 AND block_number = $2`, db.nodeId, blockNumber)
savedBlock, err := db.loadBlock(blockRows)
if err != nil {
switch err {
case sql.ErrNoRows:
return core.Block{}, repositories.ErrBlockDoesNotExist(blockNumber)
default:
return savedBlock, err
}
}
return savedBlock, nil
}
func (db DB) insertBlock(block core.Block) error {
var blockId int64
tx, _ := db.DB.BeginTx(context.Background(), nil)
err := tx.QueryRow(
`INSERT INTO blocks
(node_id, block_number, block_gaslimit, block_gasused, block_time, block_difficulty, block_hash, block_nonce, block_parenthash, block_size, uncle_hash, is_final, block_miner, block_extra_data, block_reward, block_uncles_reward)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16)
RETURNING id `,
db.nodeId, block.Number, block.GasLimit, block.GasUsed, block.Time, block.Difficulty, block.Hash, block.Nonce, block.ParentHash, block.Size, block.UncleHash, block.IsFinal, block.Miner, block.ExtraData, block.Reward, block.UnclesReward).
Scan(&blockId)
if err != nil {
tx.Rollback()
return ErrDBInsertFailed
}
err = db.createTransactions(tx, blockId, block.Transactions)
if err != nil {
tx.Rollback()
return ErrDBInsertFailed
}
tx.Commit()
return nil
}
func (db DB) createTransactions(tx *sql.Tx, blockId int64, transactions []core.Transaction) error {
for _, transaction := range transactions {
err := db.createTransaction(tx, blockId, transaction)
if err != nil {
return err
}
}
return nil
}
//Fields like value lose precision if converted to
//int64 so convert to string instead. But nil
//big.Int -> string = "" so convert to "0"
func nullStringToZero(s string) string {
if s == "" {
return "0"
}
return s
}
func (db DB) createTransaction(tx *sql.Tx, blockId int64, transaction core.Transaction) error {
var transactionId int
err := tx.QueryRow(
`INSERT INTO transactions
(block_id, tx_hash, tx_nonce, tx_to, tx_from, tx_gaslimit, tx_gasprice, tx_value, tx_input_data)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8::NUMERIC, $9)
RETURNING id`,
blockId, transaction.Hash, transaction.Nonce, transaction.To, transaction.From, transaction.GasLimit, transaction.GasPrice, nullStringToZero(transaction.Value), transaction.Data).
Scan(&transactionId)
if err != nil {
return err
}
if hasReceipt(transaction) {
receiptId, err := db.createReceipt(tx, transactionId, transaction.Receipt)
if err != nil {
return err
}
if hasLogs(transaction) {
err = db.createLogs(tx, transaction.Receipt.Logs, receiptId)
if err != nil {
return err
}
}
}
return nil
}
func hasLogs(transaction core.Transaction) bool {
return len(transaction.Receipt.Logs) > 0
}
func hasReceipt(transaction core.Transaction) bool {
return transaction.Receipt.TxHash != ""
}
func (db DB) createReceipt(tx *sql.Tx, transactionId int, receipt core.Receipt) (int, error) {
//Not currently persisting log bloom filters
var receiptId int
err := tx.QueryRow(
`INSERT INTO receipts
(contract_address, tx_hash, cumulative_gas_used, gas_used, state_root, status, transaction_id)
VALUES ($1, $2, $3, $4, $5, $6, $7)
RETURNING id`,
receipt.ContractAddress, receipt.TxHash, receipt.CumulativeGasUsed, receipt.GasUsed, receipt.StateRoot, receipt.Status, transactionId).Scan(&receiptId)
if err != nil {
return receiptId, err
}
return receiptId, nil
}
func (db DB) getBlockHash(block core.Block) (string, bool) {
var retrievedBlockHash string
db.DB.Get(&retrievedBlockHash,
`SELECT block_hash
FROM blocks
WHERE block_number = $1 AND node_id = $2`,
block.Number, db.nodeId)
return retrievedBlockHash, blockExists(retrievedBlockHash)
}
func (db DB) createLogs(tx *sql.Tx, logs []core.Log, receiptId int) error {
for _, tlog := range logs {
_, err := tx.Exec(
`INSERT INTO logs (block_number, address, tx_hash, index, topic0, topic1, topic2, topic3, data, receipt_id)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)
`,
tlog.BlockNumber, tlog.Address, tlog.TxHash, tlog.Index, tlog.Topics[0], tlog.Topics[1], tlog.Topics[2], tlog.Topics[3], tlog.Data, receiptId,
)
if err != nil {
return ErrDBInsertFailed
}
}
return nil
}
func blockExists(retrievedBlockHash string) bool {
return retrievedBlockHash != ""
}
func (db DB) removeBlock(blockNumber int64) error {
_, err := db.DB.Exec(
`DELETE FROM
blocks
WHERE block_number=$1 AND node_id=$2`,
blockNumber, db.nodeId)
if err != nil {
return ErrDBDeleteFailed
}
return nil
}
func (db DB) loadBlock(blockRows *sqlx.Row) (core.Block, error) {
type b struct {
ID int
core.Block
}
var block b
err := blockRows.StructScan(&block)
if err != nil {
return core.Block{}, err
}
transactionRows, err := db.DB.Queryx(`
SELECT tx_hash,
tx_nonce,
tx_to,
tx_from,
tx_gaslimit,
tx_gasprice,
tx_value,
tx_input_data
FROM transactions
WHERE block_id = $1
ORDER BY tx_hash`, block.ID)
if err != nil {
return core.Block{}, err
}
block.Transactions = db.loadTransactions(transactionRows)
return block.Block, nil
}
func (db DB) loadTransactions(transactionRows *sqlx.Rows) []core.Transaction {
var transactions []core.Transaction
for transactionRows.Next() {
var transaction core.Transaction
err := transactionRows.StructScan(&transaction)
if err != nil {
fmt.Println(err)
log.Fatal(err)
}
transactions = append(transactions, transaction)
}
return transactions
}

283
pkg/repositories/postgres/block_repository_test.go Normal file
View File

@ -0,0 +1,283 @@
package postgres_test
import (
"math/big"
"strconv"
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
"github.com/vulcanize/vulcanizedb/pkg/core"
"github.com/vulcanize/vulcanizedb/pkg/repositories"
"github.com/vulcanize/vulcanizedb/pkg/repositories/postgres"
)
var _ = Describe("Saving blocks", func() {
var repository repositories.BlockRepository
BeforeEach(func() {
node := core.Node{
GenesisBlock: "GENESIS",
NetworkId: 1,
Id: "b6f90c0fdd8ec9607aed8ee45c69322e47b7063f0bfb7a29c8ecafab24d0a22d24dd2329b5ee6ed4125a03cb14e57fd584e67f9e53e6c631055cbbd82f080845",
ClientName: "Geth/v1.7.2-stable-1db4ecdc/darwin-amd64/go1.9",
}
repository = postgres.BuildRepository(node)
})
It("associates blocks to a node", func() {
block := core.Block{
Number: 123,
}
repository.CreateOrUpdateBlock(block)
nodeTwo := core.Node{
GenesisBlock: "0x456",
NetworkId: 1,
Id: "x123456",
ClientName: "Geth",
}
repositoryTwo := postgres.BuildRepository(nodeTwo)
_, err := repositoryTwo.FindBlockByNumber(123)
Expect(err).To(HaveOccurred())
})
It("saves the attributes of the block", func() {
blockNumber := int64(123)
gasLimit := int64(1000000)
gasUsed := int64(10)
blockHash := "x123"
blockParentHash := "x456"
blockNonce := "0x881db2ca900682e9a9"
miner := "x123"
extraData := "xextraData"
blockTime := int64(1508981640)
uncleHash := "x789"
blockSize := int64(1000)
difficulty := int64(10)
blockReward := float64(5.132)
unclesReward := float64(3.580)
block := core.Block{
Reward: blockReward,
Difficulty: difficulty,
GasLimit: gasLimit,
GasUsed: gasUsed,
Hash: blockHash,
ExtraData: extraData,
Nonce: blockNonce,
Miner: miner,
Number: blockNumber,
ParentHash: blockParentHash,
Size: blockSize,
Time: blockTime,
UncleHash: uncleHash,
UnclesReward: unclesReward,
}
repository.CreateOrUpdateBlock(block)
savedBlock, err := repository.FindBlockByNumber(blockNumber)
Expect(err).NotTo(HaveOccurred())
Expect(savedBlock.Reward).To(Equal(blockReward))
Expect(savedBlock.Difficulty).To(Equal(difficulty))
Expect(savedBlock.GasLimit).To(Equal(gasLimit))
Expect(savedBlock.GasUsed).To(Equal(gasUsed))
Expect(savedBlock.Hash).To(Equal(blockHash))
Expect(savedBlock.Nonce).To(Equal(blockNonce))
Expect(savedBlock.Miner).To(Equal(miner))
Expect(savedBlock.ExtraData).To(Equal(extraData))
Expect(savedBlock.Number).To(Equal(blockNumber))
Expect(savedBlock.ParentHash).To(Equal(blockParentHash))
Expect(savedBlock.Size).To(Equal(blockSize))
Expect(savedBlock.Time).To(Equal(blockTime))
Expect(savedBlock.UncleHash).To(Equal(uncleHash))
Expect(savedBlock.UnclesReward).To(Equal(unclesReward))
})
It("does not find a block when searching for a number that does not exist", func() {
_, err := repository.FindBlockByNumber(111)
Expect(err).To(HaveOccurred())
})
It("saves one transaction associated to the block", func() {
block := core.Block{
Number: 123,
Transactions: []core.Transaction{{}},
}
repository.CreateOrUpdateBlock(block)
savedBlock, _ := repository.FindBlockByNumber(123)
Expect(len(savedBlock.Transactions)).To(Equal(1))
})
It("saves two transactions associated to the block", func() {
block := core.Block{
Number: 123,
Transactions: []core.Transaction{{}, {}},
}
repository.CreateOrUpdateBlock(block)
savedBlock, _ := repository.FindBlockByNumber(123)
Expect(len(savedBlock.Transactions)).To(Equal(2))
})
It(`replaces blocks and transactions associated to the block
when a more new block is in conflict (same block number + nodeid)`, func() {
blockOne := core.Block{
Number: 123,
Hash: "xabc",
Transactions: []core.Transaction{{Hash: "x123"}, {Hash: "x345"}},
}
blockTwo := core.Block{
Number: 123,
Hash: "xdef",
Transactions: []core.Transaction{{Hash: "x678"}, {Hash: "x9ab"}},
}
repository.CreateOrUpdateBlock(blockOne)
repository.CreateOrUpdateBlock(blockTwo)
savedBlock, _ := repository.FindBlockByNumber(123)
Expect(len(savedBlock.Transactions)).To(Equal(2))
Expect(savedBlock.Transactions[0].Hash).To(Equal("x678"))
Expect(savedBlock.Transactions[1].Hash).To(Equal("x9ab"))
})
It(`does not replace blocks when block number is not unique
but block number + node id is`, func() {
blockOne := core.Block{
Number: 123,
Transactions: []core.Transaction{{Hash: "x123"}, {Hash: "x345"}},
}
blockTwo := core.Block{
Number: 123,
Transactions: []core.Transaction{{Hash: "x678"}, {Hash: "x9ab"}},
}
repository.CreateOrUpdateBlock(blockOne)
nodeTwo := core.Node{
GenesisBlock: "0x456",
NetworkId: 1,
}
repositoryTwo := postgres.BuildRepository(nodeTwo)
repository.CreateOrUpdateBlock(blockOne)
repositoryTwo.CreateOrUpdateBlock(blockTwo)
retrievedBlockOne, _ := repository.FindBlockByNumber(123)
retrievedBlockTwo, _ := repositoryTwo.FindBlockByNumber(123)
Expect(retrievedBlockOne.Transactions[0].Hash).To(Equal("x123"))
Expect(retrievedBlockTwo.Transactions[0].Hash).To(Equal("x678"))
})
It("saves the attributes associated to a transaction", func() {
gasLimit := int64(5000)
gasPrice := int64(3)
nonce := uint64(10000)
to := "1234567890"
from := "0987654321"
var value = new(big.Int)
value.SetString("34940183920000000000", 10)
inputData := "0xf7d8c8830000000000000000000000000000000000000000000000000000000000037788000000000000000000000000000000000000000000000000000000000003bd14"
transaction := core.Transaction{
Hash: "x1234",
GasPrice: gasPrice,
GasLimit: gasLimit,
Nonce: nonce,
To: to,
From: from,
Value: value.String(),
Data: inputData,
}
block := core.Block{
Number: 123,
Transactions: []core.Transaction{transaction},
}
repository.CreateOrUpdateBlock(block)
savedBlock, _ := repository.FindBlockByNumber(123)
Expect(len(savedBlock.Transactions)).To(Equal(1))
savedTransaction := savedBlock.Transactions[0]
Expect(savedTransaction.Data).To(Equal(transaction.Data))
Expect(savedTransaction.Hash).To(Equal(transaction.Hash))
Expect(savedTransaction.To).To(Equal(to))
Expect(savedTransaction.From).To(Equal(from))
Expect(savedTransaction.Nonce).To(Equal(nonce))
Expect(savedTransaction.GasLimit).To(Equal(gasLimit))
Expect(savedTransaction.GasPrice).To(Equal(gasPrice))
Expect(savedTransaction.Value).To(Equal(value.String()))
})
Describe("The missing block numbers", func() {
It("is empty the starting block number is the highest known block number", func() {
repository.CreateOrUpdateBlock(core.Block{Number: 1})
Expect(len(repository.MissingBlockNumbers(1, 1))).To(Equal(0))
})
It("is the only missing block number", func() {
repository.CreateOrUpdateBlock(core.Block{Number: 2})
Expect(repository.MissingBlockNumbers(1, 2)).To(Equal([]int64{1}))
})
It("is both missing block numbers", func() {
repository.CreateOrUpdateBlock(core.Block{Number: 3})
Expect(repository.MissingBlockNumbers(1, 3)).To(Equal([]int64{1, 2}))
})
It("goes back to the starting block number", func() {
repository.CreateOrUpdateBlock(core.Block{Number: 6})
Expect(repository.MissingBlockNumbers(4, 6)).To(Equal([]int64{4, 5}))
})
It("only includes missing block numbers", func() {
repository.CreateOrUpdateBlock(core.Block{Number: 4})
repository.CreateOrUpdateBlock(core.Block{Number: 6})
Expect(repository.MissingBlockNumbers(4, 6)).To(Equal([]int64{5}))
})
It("is a list with multiple gaps", func() {
repository.CreateOrUpdateBlock(core.Block{Number: 4})
repository.CreateOrUpdateBlock(core.Block{Number: 5})
repository.CreateOrUpdateBlock(core.Block{Number: 8})
repository.CreateOrUpdateBlock(core.Block{Number: 10})
Expect(repository.MissingBlockNumbers(3, 10)).To(Equal([]int64{3, 6, 7, 9}))
})
It("returns empty array when lower bound exceeds upper bound", func() {
Expect(repository.MissingBlockNumbers(10000, 1)).To(Equal([]int64{}))
})
It("only returns requested range even when other gaps exist", func() {
repository.CreateOrUpdateBlock(core.Block{Number: 3})
repository.CreateOrUpdateBlock(core.Block{Number: 8})
Expect(repository.MissingBlockNumbers(1, 5)).To(Equal([]int64{1, 2, 4, 5}))
})
})
Describe("The block status", func() {
It("sets the status of blocks within n-20 of chain HEAD as final", func() {
blockNumberOfChainHead := 25
for i := 0; i < blockNumberOfChainHead; i++ {
repository.CreateOrUpdateBlock(core.Block{Number: int64(i), Hash: strconv.Itoa(i)})
}
repository.SetBlocksStatus(int64(blockNumberOfChainHead))
blockOne, err := repository.FindBlockByNumber(1)
Expect(err).ToNot(HaveOccurred())
Expect(blockOne.IsFinal).To(Equal(true))
blockTwo, err := repository.FindBlockByNumber(24)
Expect(err).ToNot(HaveOccurred())
Expect(blockTwo.IsFinal).To(BeFalse())
})
})
})

68
pkg/repositories/postgres/contract_repository.go Normal file
View File

@ -0,0 +1,68 @@
package postgres
import (
"database/sql"
"github.com/vulcanize/vulcanizedb/pkg/core"
"github.com/vulcanize/vulcanizedb/pkg/repositories"
)
func (db DB) CreateContract(contract core.Contract) error {
abi := contract.Abi
var abiToInsert *string
if abi != "" {
abiToInsert = &abi
}
_, err := db.DB.Exec(
`INSERT INTO watched_contracts (contract_hash, contract_abi)
VALUES ($1, $2)
ON CONFLICT (contract_hash)
DO UPDATE
SET contract_hash = $1, contract_abi = $2
`, contract.Hash, abiToInsert)
if err != nil {
return ErrDBInsertFailed
}
return nil
}
func (db DB) ContractExists(contractHash string) bool {
var exists bool
db.DB.QueryRow(
`SELECT exists(
SELECT 1
FROM watched_contracts
WHERE contract_hash = $1)`, contractHash).Scan(&exists)
return exists
}
func (db DB) FindContract(contractHash string) (core.Contract, error) {
var hash string
var abi string
contract := db.DB.QueryRow(
`SELECT contract_hash, contract_abi FROM watched_contracts WHERE contract_hash=$1`, contractHash)
err := contract.Scan(&hash, &abi)
if err == sql.ErrNoRows {
return core.Contract{}, repositories.ErrContractDoesNotExist(contractHash)
}
savedContract := db.addTransactions(core.Contract{Hash: hash, Abi: abi})
return savedContract, nil
}
func (db DB) addTransactions(contract core.Contract) core.Contract {
transactionRows, _ := db.DB.Queryx(`
SELECT tx_hash,
tx_nonce,
tx_to,
tx_from,
tx_gaslimit,
tx_gasprice,
tx_value,
tx_input_data
FROM transactions
WHERE tx_to = $1
ORDER BY block_id DESC`, contract.Hash)
transactions := db.loadTransactions(transactionRows)
savedContract := core.Contract{Hash: contract.Hash, Transactions: transactions, Abi: contract.Abi}
return savedContract
}

97
pkg/repositories/postgres/contract_repository_test.go Normal file
View File

@ -0,0 +1,97 @@
package postgres
import (
"sort"
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
"github.com/vulcanize/vulcanizedb/pkg/core"
"github.com/vulcanize/vulcanizedb/pkg/repositories"
)
var _ = Describe("Creating contracts", func() {
var repository repositories.ContractRepository
var node core.Node
BeforeEach(func() {
node = core.Node{
GenesisBlock: "GENESIS",
NetworkId: 1,
Id: "b6f90c0fdd8ec9607aed8ee45c69322e47b7063f0bfb7a29c8ecafab24d0a22d24dd2329b5ee6ed4125a03cb14e57fd584e67f9e53e6c631055cbbd82f080845",
ClientName: "Geth/v1.7.2-stable-1db4ecdc/darwin-amd64/go1.9",
}
repository = BuildRepository(node)
})
It("returns the contract when it exists", func() {
repository.CreateContract(core.Contract{Hash: "x123"})
contract, err := repository.FindContract("x123")
Expect(err).NotTo(HaveOccurred())
Expect(contract.Hash).To(Equal("x123"))
Expect(repository.ContractExists("x123")).To(BeTrue())
Expect(repository.ContractExists("x456")).To(BeFalse())
})
It("returns err if contract does not exist", func() {
_, err := repository.FindContract("x123")
Expect(err).To(HaveOccurred())
})
It("returns empty array when no transactions 'To' a contract", func() {
repository.CreateContract(core.Contract{Hash: "x123"})
contract, err := repository.FindContract("x123")
Expect(err).ToNot(HaveOccurred())
Expect(contract.Transactions).To(BeEmpty())
})
It("returns transactions 'To' a contract", func() {
var blockRepository repositories.BlockRepository
blockRepository = BuildRepository(node)
block := core.Block{
Number: 123,
Transactions: []core.Transaction{
{Hash: "TRANSACTION1", To: "x123", Value: "0"},
{Hash: "TRANSACTION2", To: "x345", Value: "0"},
{Hash: "TRANSACTION3", To: "x123", Value: "0"},
},
}
blockRepository.CreateOrUpdateBlock(block)
repository.CreateContract(core.Contract{Hash: "x123"})
contract, err := repository.FindContract("x123")
Expect(err).ToNot(HaveOccurred())
sort.Slice(contract.Transactions, func(i, j int) bool {
return contract.Transactions[i].Hash < contract.Transactions[j].Hash
})
Expect(contract.Transactions).To(
Equal([]core.Transaction{
{Hash: "TRANSACTION1", To: "x123", Value: "0"},
{Hash: "TRANSACTION3", To: "x123", Value: "0"},
}))
})
It("stores the ABI of the contract", func() {
repository.CreateContract(core.Contract{
Abi: "{\"some\": \"json\"}",
Hash: "x123",
})
contract, err := repository.FindContract("x123")
Expect(err).ToNot(HaveOccurred())
Expect(contract.Abi).To(Equal("{\"some\": \"json\"}"))
})
It("updates the ABI of the contract if hash already present", func() {
repository.CreateContract(core.Contract{
Abi: "{\"some\": \"json\"}",
Hash: "x123",
})
repository.CreateContract(core.Contract{
Abi: "{\"some\": \"different json\"}",
Hash: "x123",
})
contract, err := repository.FindContract("x123")
Expect(err).ToNot(HaveOccurred())
Expect(contract.Abi).To(Equal("{\"some\": \"different json\"}"))
})
})

22
pkg/repositories/postgres/helpers.go Normal file
View File

@ -0,0 +1,22 @@
package postgres
import (
"github.com/vulcanize/vulcanizedb/pkg/config"
"github.com/vulcanize/vulcanizedb/pkg/core"
)
func ClearData(postgres *DB) {
postgres.DB.MustExec("DELETE FROM watched_contracts")
postgres.DB.MustExec("DELETE FROM transactions")
postgres.DB.MustExec("DELETE FROM blocks")
postgres.DB.MustExec("DELETE FROM logs")
postgres.DB.MustExec("DELETE FROM receipts")
postgres.DB.MustExec("DELETE FROM log_filters")
}
func BuildRepository(node core.Node) *DB {
cfg, _ := config.NewConfig("private")
repository, _ := NewDB(cfg.Database, node)
ClearData(repository)
return repository
}

15
pkg/repositories/postgres/log_filter_repository.go Normal file
View File

@ -0,0 +1,15 @@
package postgres
import "github.com/vulcanize/vulcanizedb/pkg/filters"
func (db DB) AddFilter(query filters.LogFilter) error {
_, err := db.DB.Exec(
`INSERT INTO log_filters
(name, from_block, to_block, address, topic0, topic1, topic2, topic3)
VALUES ($1, NULLIF($2, -1), NULLIF($3, -1), $4, NULLIF($5, ''), NULLIF($6, ''), NULLIF($7, ''), NULLIF($8, ''))`,
query.Name, query.FromBlock, query.ToBlock, query.Address, query.Topics[0], query.Topics[1], query.Topics[2], query.Topics[3])
if err != nil {
return err
}
return nil
}

61
pkg/repositories/postgres/log_filter_repository_test.go Normal file
View File

@ -0,0 +1,61 @@
package postgres
import (
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
"github.com/vulcanize/vulcanizedb/pkg/core"
"github.com/vulcanize/vulcanizedb/pkg/filters"
"github.com/vulcanize/vulcanizedb/pkg/repositories"
)
var _ = Describe("Logs Repository", func() {
var repository repositories.FilterRepository
var node core.Node
BeforeEach(func() {
node = core.Node{
GenesisBlock: "GENESIS",
NetworkId: 1,
Id: "b6f90c0fdd8ec9607aed8ee45c69322e47b7063f0bfb7a29c8ecafab24d0a22d24dd2329b5ee6ed4125a03cb14e57fd584e67f9e53e6c631055cbbd82f080845",
ClientName: "Geth/v1.7.2-stable-1db4ecdc/darwin-amd64/go1.9",
}
repository = BuildRepository(node)
})
Describe("LogFilter", func() {
It("inserts filter into watched events", func() {
logFilter := filters.LogFilter{
Name: "TestFilter",
FromBlock: 1,
ToBlock: 2,
Address: "0x8888f1f195afa192cfee860698584c030f4c9db1",
Topics: core.Topics{
"0x000000000000000000000000a94f5374fce5edbc8e2a8697c15331677e6ebf0b",
"",
"0x000000000000000000000000a94f5374fce5edbc8e2a8697c15331677e6ebf0b",
"",
},
}
err := repository.AddFilter(logFilter)
Expect(err).ToNot(HaveOccurred())
})
It("returns error if name is not provided", func() {
logFilter := filters.LogFilter{
FromBlock: 1,
ToBlock: 2,
Address: "0x8888f1f195afa192cfee860698584c030f4c9db1",
Topics: core.Topics{
"0x000000000000000000000000a94f5374fce5edbc8e2a8697c15331677e6ebf0b",
"",
"0x000000000000000000000000a94f5374fce5edbc8e2a8697c15331677e6ebf0b",
"",
},
}
err := repository.AddFilter(logFilter)
Expect(err).To(HaveOccurred())
})
})
})

69
pkg/repositories/postgres/logs.go Normal file
View File

@ -0,0 +1,69 @@
package postgres
import (
"context"
"database/sql"
"github.com/vulcanize/vulcanizedb/pkg/core"
)
func (db DB) CreateLogs(logs []core.Log) error {
tx, _ := db.DB.BeginTx(context.Background(), nil)
for _, tlog := range logs {
_, err := tx.Exec(
`INSERT INTO logs (block_number, address, tx_hash, index, topic0, topic1, topic2, topic3, data)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
`,
tlog.BlockNumber, tlog.Address, tlog.TxHash, tlog.Index, tlog.Topics[0], tlog.Topics[1], tlog.Topics[2], tlog.Topics[3], tlog.Data,
)
if err != nil {
tx.Rollback()
return ErrDBInsertFailed
}
}
tx.Commit()
return nil
}
func (db DB) FindLogs(address string, blockNumber int64) []core.Log {
logRows, _ := db.DB.Query(
`SELECT block_number,
address,
tx_hash,
index,
topic0,
topic1,
topic2,
topic3,
data
FROM logs
WHERE address = $1 AND block_number = $2
ORDER BY block_number DESC`, address, blockNumber)
return db.loadLogs(logRows)
}
func (db DB) loadLogs(logsRows *sql.Rows) []core.Log {
var logs []core.Log
for logsRows.Next() {
var blockNumber int64
var address string
var txHash string
var index int64
var data string
var topics core.Topics
logsRows.Scan(&blockNumber, &address, &txHash, &index, &topics[0], &topics[1], &topics[2], &topics[3], &data)
log := core.Log{
BlockNumber: blockNumber,
TxHash: txHash,
Address: address,
Index: index,
Data: data,
}
for i, topic := range topics {
log.Topics[i] = topic
}
logs = append(logs, log)
}
return logs
}

178
pkg/repositories/postgres/logs_test.go Normal file
View File

@ -0,0 +1,178 @@
package postgres_test
import (
"sort"
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
"github.com/vulcanize/vulcanizedb/pkg/core"
"github.com/vulcanize/vulcanizedb/pkg/repositories"
"github.com/vulcanize/vulcanizedb/pkg/repositories/postgres"
)
var _ = Describe("Logs Repository", func() {
var repository repositories.LogsRepository
var node core.Node
BeforeEach(func() {
node = core.Node{
GenesisBlock: "GENESIS",
NetworkId: 1,
Id: "b6f90c0fdd8ec9607aed8ee45c69322e47b7063f0bfb7a29c8ecafab24d0a22d24dd2329b5ee6ed4125a03cb14e57fd584e67f9e53e6c631055cbbd82f080845",
ClientName: "Geth/v1.7.2-stable-1db4ecdc/darwin-amd64/go1.9",
}
repository = postgres.BuildRepository(node)
})
Describe("Saving logs", func() {
It("returns the log when it exists", func() {
repository.CreateLogs([]core.Log{{
BlockNumber: 1,
Index: 0,
Address: "x123",
TxHash: "x456",
Topics: core.Topics{0: "x777", 1: "x888", 2: "x999"},
Data: "xabc",
}},
)
log := repository.FindLogs("x123", 1)
Expect(log).NotTo(BeNil())
Expect(log[0].BlockNumber).To(Equal(int64(1)))
Expect(log[0].Address).To(Equal("x123"))
Expect(log[0].Index).To(Equal(int64(0)))
Expect(log[0].TxHash).To(Equal("x456"))
Expect(log[0].Topics[0]).To(Equal("x777"))
Expect(log[0].Topics[1]).To(Equal("x888"))
Expect(log[0].Topics[2]).To(Equal("x999"))
Expect(log[0].Data).To(Equal("xabc"))
})
It("returns nil if log does not exist", func() {
log := repository.FindLogs("x123", 1)
Expect(log).To(BeNil())
})
It("filters to the correct block number and address", func() {
repository.CreateLogs([]core.Log{{
BlockNumber: 1,
Index: 0,
Address: "x123",
TxHash: "x456",
Topics: core.Topics{0: "x777", 1: "x888", 2: "x999"},
Data: "xabc",
}},
)
repository.CreateLogs([]core.Log{{
BlockNumber: 1,
Index: 1,
Address: "x123",
TxHash: "x789",
Topics: core.Topics{0: "x111", 1: "x222", 2: "x333"},
Data: "xdef",
}},
)
repository.CreateLogs([]core.Log{{
BlockNumber: 2,
Index: 0,
Address: "x123",
TxHash: "x456",
Topics: core.Topics{0: "x777", 1: "x888", 2: "x999"},
Data: "xabc",
}},
)
log := repository.FindLogs("x123", 1)
type logIndex struct {
blockNumber int64
Index int64
}
var uniqueBlockNumbers []logIndex
for _, log := range log {
uniqueBlockNumbers = append(uniqueBlockNumbers,
logIndex{log.BlockNumber, log.Index})
}
sort.Slice(uniqueBlockNumbers, func(i, j int) bool {
if uniqueBlockNumbers[i].blockNumber < uniqueBlockNumbers[j].blockNumber {
return true
}
if uniqueBlockNumbers[i].blockNumber > uniqueBlockNumbers[j].blockNumber {
return false
}
return uniqueBlockNumbers[i].Index < uniqueBlockNumbers[j].Index
})
Expect(log).NotTo(BeNil())
Expect(len(log)).To(Equal(2))
Expect(uniqueBlockNumbers).To(Equal(
[]logIndex{
{blockNumber: 1, Index: 0},
{blockNumber: 1, Index: 1}},
))
})
It("saves the logs attached to a receipt", func() {
var blockRepository repositories.BlockRepository
blockRepository = postgres.BuildRepository(node)
logs := []core.Log{{
Address: "0x8a4774fe82c63484afef97ca8d89a6ea5e21f973",
BlockNumber: 4745407,
Data: "0x0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000004000000000000000000000000000000000000000000000000000000000000000645a68669900000000000000000000000000000000000000000000003397684ab5869b0000000000000000000000000000000000000000000000000000000000005a36053200000000000000000000000099041f808d598b782d5a3e498681c2452a31da08",
Index: 86,
Topics: core.Topics{
0: "0x5a68669900000000000000000000000000000000000000000000000000000000",
1: "0x000000000000000000000000d0148dad63f73ce6f1b6c607e3413dcf1ff5f030",
2: "0x00000000000000000000000000000000000000000000003397684ab5869b0000",
3: "0x000000000000000000000000000000000000000000000000000000005a360532",
},
TxHash: "0x002c4799161d809b23f67884eb6598c9df5894929fe1a9ead97ca175d360f547",
}, {
Address: "0x99041f808d598b782d5a3e498681c2452a31da08",
BlockNumber: 4745407,
Data: "0x00000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000040000000000000000000000000000000000000000000000000000000000000000418178358",
Index: 87,
Topics: core.Topics{
0: "0x1817835800000000000000000000000000000000000000000000000000000000",
1: "0x0000000000000000000000008a4774fe82c63484afef97ca8d89a6ea5e21f973",
2: "0x0000000000000000000000000000000000000000000000000000000000000000",
3: "0x0000000000000000000000000000000000000000000000000000000000000000",
},
TxHash: "0x002c4799161d809b23f67884eb6598c9df5894929fe1a9ead97ca175d360f547",
}, {
Address: "0x99041f808d598b782d5a3e498681c2452a31da08",
BlockNumber: 4745407,
Data: "0x00000000000000000000000000000000000000000000003338f64c8423af4000",
Index: 88,
Topics: core.Topics{
0: "0x296ba4ca62c6c21c95e828080cb8aec7481b71390585605300a8a76f9e95b527",
},
TxHash: "0x002c4799161d809b23f67884eb6598c9df5894929fe1a9ead97ca175d360f547",
},
}
receipt := core.Receipt{
ContractAddress: "",
CumulativeGasUsed: 7481414,
GasUsed: 60711,
Logs: logs,
Bloom: "0x00000800000000000000001000000000000000400000000080000000000000000000400000010000000000000000000000000000040000000000000002000000000000000000000000000000000000000000000000000000000000000000000000000000020000000000000000000800004000000000000001000000000000000000000000000002000000480000000000000002000000000000000020000000000000000000000000000000000000000080000000000180000c00000000000000002000002000000040000000000000000000000000000010000000000020000000000000000000002000000000000000000000000400800000000000000000",
Status: 1,
TxHash: "0x002c4799161d809b23f67884eb6598c9df5894929fe1a9ead97ca175d360f547",
}
transaction :=
core.Transaction{
Hash: receipt.TxHash,
Receipt: receipt,
}
block := core.Block{Transactions: []core.Transaction{transaction}}
err := blockRepository.CreateOrUpdateBlock(block)
Expect(err).To(Not(HaveOccurred()))
retrievedLogs := repository.FindLogs("0x99041f808d598b782d5a3e498681c2452a31da08", 4745407)
expected := logs[1:]
Expect(retrievedLogs).To(Equal(expected))
})
})
})

27
pkg/repositories/postgres/node_repository.go Normal file
View File

@ -0,0 +1,27 @@
package postgres
import "github.com/vulcanize/vulcanizedb/pkg/core"
type NodeRepository interface {
CreateNode(node *core.Node) error
}
func (db *DB) CreateNode(node *core.Node) error {
var nodeId int64
err := db.DB.QueryRow(
`INSERT INTO nodes (genesis_block, network_id, node_id, client_name)
VALUES ($1, $2, $3, $4)
ON CONFLICT (genesis_block, network_id, node_id)
DO UPDATE
SET genesis_block = $1,
network_id = $2,
node_id = $3,
client_name = $4
RETURNING id`,
node.GenesisBlock, node.NetworkId, node.Id, node.ClientName).Scan(&nodeId)
if err != nil {
return ErrUnableToSetNode
}
db.nodeId = nodeId
return nil
}

37
pkg/repositories/postgres/postgres.go Normal file
View File

@ -0,0 +1,37 @@
package postgres
import (
"errors"
"github.com/jmoiron/sqlx"
_ "github.com/lib/pq"
"github.com/vulcanize/vulcanizedb/pkg/config"
"github.com/vulcanize/vulcanizedb/pkg/core"
)
type DB struct {
*sqlx.DB
node core.Node
nodeId int64
}
var (
ErrDBInsertFailed = errors.New("postgres: insert failed")
ErrDBDeleteFailed = errors.New("postgres: delete failed")
ErrDBConnectionFailed = errors.New("postgres: db connection failed")
ErrUnableToSetNode = errors.New("postgres: unable to set node")
)
func NewDB(databaseConfig config.Database, node core.Node) (*DB, error) {
connectString := config.DbConnectionString(databaseConfig)
db, err := sqlx.Connect("postgres", connectString)
if err != nil {
return &DB{}, ErrDBConnectionFailed
}
pg := DB{DB: db, node: node}
err = pg.CreateNode(&node)
if err != nil {
return &DB{}, ErrUnableToSetNode
}
return &pg, nil
}

10
pkg/core/core_suite_test.go → pkg/repositories/postgres/postgres_suite_test.go
View File

@ -1,13 +1,13 @@
package core_test
package postgres_test
import (
"testing"
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
"testing"
)
func TestVulcanizedb(t *testing.T) {
func TestPostgres(t *testing.T) {
RegisterFailHandler(Fail)
RunSpecs(t, "Vulcanizedb Suite")
RunSpecs(t, "Postgres Suite")
}

37
pkg/repositories/postgres_test.go → pkg/repositories/postgres/postgres_test.go
View File

@ -1,4 +1,4 @@
package repositories_test
package postgres_test
import (
"fmt"
@ -9,14 +9,13 @@ import (
"math/big"
"github.com/vulcanize/vulcanizedb/pkg/config"
"github.com/vulcanize/vulcanizedb/pkg/core"
"github.com/vulcanize/vulcanizedb/pkg/repositories"
"github.com/vulcanize/vulcanizedb/pkg/repositories/testing"
"github.com/jmoiron/sqlx"
_ "github.com/lib/pq"
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
"github.com/vulcanize/vulcanizedb/pkg/config"
"github.com/vulcanize/vulcanizedb/pkg/core"
"github.com/vulcanize/vulcanizedb/pkg/repositories/postgres"
)
func init() {
@ -24,6 +23,7 @@ func init() {
}
var _ = Describe("Postgres repository", func() {
var repository *postgres.DB
It("connects to the database", func() {
cfg, _ := config.NewConfig("private")
@ -33,11 +33,16 @@ var _ = Describe("Postgres repository", func() {
Expect(db).ShouldNot(BeNil())
})
testing.AssertRepositoryBehavior(func(node core.Node) repositories.Repository {
BeforeEach(func() {
node := core.Node{
GenesisBlock: "GENESIS",
NetworkId: 1,
Id: "b6f90c0fdd8ec9607aed8ee45c69322e47b7063f0bfb7a29c8ecafab24d0a22d24dd2329b5ee6ed4125a03cb14e57fd584e67f9e53e6c631055cbbd82f080845",
ClientName: "Geth/v1.7.2-stable-1db4ecdc/darwin-amd64/go1.9",
}
cfg, _ := config.NewConfig("private")
repository, _ := repositories.NewPostgres(cfg.Database, node)
testing.ClearData(repository)
return repository
repository, _ = postgres.NewDB(cfg.Database, node)
postgres.ClearData(repository)
})
It("serializes big.Int to db", func() {
@ -86,7 +91,7 @@ var _ = Describe("Postgres repository", func() {
}
cfg, _ := config.NewConfig("private")
node := core.Node{GenesisBlock: "GENESIS", NetworkId: 1, Id: "x123", ClientName: "geth"}
repository, _ := repositories.NewPostgres(cfg.Database, node)
repository, _ := postgres.NewDB(cfg.Database, node)
err1 := repository.CreateOrUpdateBlock(badBlock)
savedBlock, err2 := repository.FindBlockByNumber(123)
@ -99,16 +104,16 @@ var _ = Describe("Postgres repository", func() {
It("throws error when can't connect to the database", func() {
invalidDatabase := config.Database{}
node := core.Node{GenesisBlock: "GENESIS", NetworkId: 1, Id: "x123", ClientName: "geth"}
_, err := repositories.NewPostgres(invalidDatabase, node)
Expect(err).To(Equal(repositories.ErrDBConnectionFailed))
_, err := postgres.NewDB(invalidDatabase, node)
Expect(err).To(Equal(postgres.ErrDBConnectionFailed))
})
It("throws error when can't create node", func() {
cfg, _ := config.NewConfig("private")
badHash := fmt.Sprintf("x %s", strings.Repeat("1", 100))
node := core.Node{GenesisBlock: badHash, NetworkId: 1, Id: "x123", ClientName: "geth"}
_, err := repositories.NewPostgres(cfg.Database, node)
Expect(err).To(Equal(repositories.ErrUnableToSetNode))
_, err := postgres.NewDB(cfg.Database, node)
Expect(err).To(Equal(postgres.ErrUnableToSetNode))
})
It("does not commit log if log is invalid", func() {
@ -121,7 +126,7 @@ var _ = Describe("Postgres repository", func() {
}
cfg, _ := config.NewConfig("private")
node := core.Node{GenesisBlock: "GENESIS", NetworkId: 1, Id: "x123", ClientName: "geth"}
repository, _ := repositories.NewPostgres(cfg.Database, node)
repository, _ := postgres.NewDB(cfg.Database, node)
err := repository.CreateLogs([]core.Log{badLog})
savedBlock := repository.FindLogs("x123", 1)
@ -140,7 +145,7 @@ var _ = Describe("Postgres repository", func() {
}
cfg, _ := config.NewConfig("private")
node := core.Node{GenesisBlock: "GENESIS", NetworkId: 1, Id: "x123", ClientName: "geth"}
repository, _ := repositories.NewPostgres(cfg.Database, node)
repository, _ := postgres.NewDB(cfg.Database, node)
err1 := repository.CreateOrUpdateBlock(block)
savedBlock, err2 := repository.FindBlockByNumber(123)

49
pkg/repositories/postgres/receipt_repository.go Normal file
View File

@ -0,0 +1,49 @@
package postgres
import (
"database/sql"
"github.com/vulcanize/vulcanizedb/pkg/core"
"github.com/vulcanize/vulcanizedb/pkg/repositories"
)
func (db DB) FindReceipt(txHash string) (core.Receipt, error) {
row := db.DB.QueryRow(
`SELECT contract_address,
tx_hash,
cumulative_gas_used,
gas_used,
state_root,
status
FROM receipts
WHERE tx_hash = $1`, txHash)
receipt, err := loadReceipt(row)
if err != nil {
switch err {
case sql.ErrNoRows:
return core.Receipt{}, repositories.ErrReceiptDoesNotExist(txHash)
default:
return core.Receipt{}, err
}
}
return receipt, nil
}
func loadReceipt(receiptsRow *sql.Row) (core.Receipt, error) {
var contractAddress string
var txHash string
var cumulativeGasUsed int64
var gasUsed int64
var stateRoot string
var status int
err := receiptsRow.Scan(&contractAddress, &txHash, &cumulativeGasUsed, &gasUsed, &stateRoot, &status)
return core.Receipt{
TxHash: txHash,
ContractAddress: contractAddress,
CumulativeGasUsed: cumulativeGasUsed,
GasUsed: gasUsed,
StateRoot: stateRoot,
Status: status,
}, err
}

84
pkg/repositories/postgres/receipts_repository_test.go Normal file
View File

@ -0,0 +1,84 @@
package postgres_test
import (
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
"github.com/vulcanize/vulcanizedb/pkg/core"
"github.com/vulcanize/vulcanizedb/pkg/repositories"
"github.com/vulcanize/vulcanizedb/pkg/repositories/postgres"
)
var _ = Describe("Logs Repository", func() {
var repository repositories.ReceiptRepository
var node core.Node
BeforeEach(func() {
node = core.Node{
GenesisBlock: "GENESIS",
NetworkId: 1,
Id: "b6f90c0fdd8ec9607aed8ee45c69322e47b7063f0bfb7a29c8ecafab24d0a22d24dd2329b5ee6ed4125a03cb14e57fd584e67f9e53e6c631055cbbd82f080845",
ClientName: "Geth/v1.7.2-stable-1db4ecdc/darwin-amd64/go1.9",
}
repository = postgres.BuildRepository(node)
})
Describe("Saving receipts", func() {
It("returns the receipt when it exists", func() {
var blockRepository repositories.BlockRepository
blockRepository = postgres.BuildRepository(node)
expected := core.Receipt{
ContractAddress: "0xde0b295669a9fd93d5f28d9ec85e40f4cb697bae",
CumulativeGasUsed: 7996119,
GasUsed: 21000,
Logs: []core.Log{},
StateRoot: "0x88abf7e73128227370aa7baa3dd4e18d0af70e92ef1f9ef426942fbe2dddb733",
Status: 1,
TxHash: "0xe340558980f89d5f86045ac11e5cc34e4bcec20f9f1e2a427aa39d87114e8223",
}
transaction := core.Transaction{
Hash: expected.TxHash,
Receipt: expected,
}
block := core.Block{Transactions: []core.Transaction{transaction}}
blockRepository.CreateOrUpdateBlock(block)
receipt, err := repository.FindReceipt("0xe340558980f89d5f86045ac11e5cc34e4bcec20f9f1e2a427aa39d87114e8223")
Expect(err).ToNot(HaveOccurred())
//Not currently serializing bloom logs
Expect(receipt.Bloom).To(Equal(core.Receipt{}.Bloom))
Expect(receipt.TxHash).To(Equal(expected.TxHash))
Expect(receipt.CumulativeGasUsed).To(Equal(expected.CumulativeGasUsed))
Expect(receipt.GasUsed).To(Equal(expected.GasUsed))
Expect(receipt.StateRoot).To(Equal(expected.StateRoot))
Expect(receipt.Status).To(Equal(expected.Status))
})
It("returns ErrReceiptDoesNotExist when receipt does not exist", func() {
receipt, err := repository.FindReceipt("DOES NOT EXIST")
Expect(err).To(HaveOccurred())
Expect(receipt).To(BeZero())
})
It("still saves receipts without logs", func() {
var blockRepository repositories.BlockRepository
blockRepository = postgres.BuildRepository(node)
receipt := core.Receipt{
TxHash: "0x002c4799161d809b23f67884eb6598c9df5894929fe1a9ead97ca175d360f547",
}
transaction := core.Transaction{
Hash: receipt.TxHash,
Receipt: receipt,
}
block := core.Block{
Transactions: []core.Transaction{transaction},
}
blockRepository.CreateOrUpdateBlock(block)
_, err := repository.FindReceipt(receipt.TxHash)
Expect(err).To(Not(HaveOccurred()))
})
})
})

6
pkg/repositories/watched_events.go → pkg/repositories/postgres/watched_events.go
View File

@ -1,4 +1,4 @@
package repositories
package postgres
type WatchedEventLog struct {
Name string `json:"name"` // name
@ -17,8 +17,8 @@ type WatchedEventLogs interface {
AllWatchedEventLogs() ([]*WatchedEventLog, error)
}
func (pg *Postgres) AllWatchedEventLogs() ([]*WatchedEventLog, error) {
rows, err := pg.Db.Queryx("SELECT name, block_number, address, tx_hash, index, topic0, topic1, topic2, topic3, data FROM watched_event_logs")
func (db *DB) AllWatchedEventLogs() ([]*WatchedEventLog, error) {
rows, err := db.DB.Queryx(`SELECT name, block_number, address, tx_hash, index, topic0, topic1, topic2, topic3, data FROM watched_event_logs`)
if err != nil {
return nil, err
}

15
pkg/repositories/watched_events_test.go → pkg/repositories/postgres/watched_events_test.go
View File

@ -1,9 +1,6 @@
package repositories_test
package postgres_test
import (
"github.com/vulcanize/vulcanizedb/pkg/repositories"
"github.com/vulcanize/vulcanizedb/pkg/repositories/testing"
"log"
. "github.com/onsi/ginkgo"
@ -11,21 +8,23 @@ import (
"github.com/vulcanize/vulcanizedb/pkg/config"
"github.com/vulcanize/vulcanizedb/pkg/core"
"github.com/vulcanize/vulcanizedb/pkg/filters"
"github.com/vulcanize/vulcanizedb/pkg/repositories/postgres"
)
var _ = Describe("Watched Events Repository", func() {
var repository repositories.Postgres
var repository *postgres.DB
BeforeEach(func() {
cfg, err := config.NewConfig("private")
if err != nil {
log.Fatal(err)
}
repository, err = repositories.NewPostgres(cfg.Database, core.Node{})
repository, err = postgres.NewDB(cfg.Database, core.Node{})
if err != nil {
log.Fatal(err)
}
testing.ClearData(repository)
postgres.ClearData(repository)
})
It("retrieves watched logs that match the event filter", func() {
@ -46,7 +45,7 @@ var _ = Describe("Watched Events Repository", func() {
Data: "",
},
}
expectedWatchedEventLog := []*repositories.WatchedEventLog{
expectedWatchedEventLog := []*postgres.WatchedEventLog{
{
Name: "Filter1",
BlockNumber: 0,

13
pkg/repositories/repositories_suite_test.go
View File

@ -1,13 +0,0 @@
package repositories_test
import (
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
"testing"
)
func TestRepositories(t *testing.T) {
RegisterFailHandler(Fail)
RunSpecs(t, "Repositories Suite")
}

49
pkg/repositories/repository.go
View File

@ -1,26 +1,55 @@
package repositories
import (
"errors"
"fmt"
"github.com/vulcanize/vulcanizedb/pkg/core"
"github.com/vulcanize/vulcanizedb/pkg/filters"
)
const (
blocksFromHeadBeforeFinal = 20
)
type Repository interface {
BlockRepository
ContractRepository
LogsRepository
ReceiptRepository
FilterRepository
}
var ErrBlockDoesNotExist = func(blockNumber int64) error {
return errors.New(fmt.Sprintf("Block number %d does not exist", blockNumber))
}
type BlockRepository interface {
CreateOrUpdateBlock(block core.Block) error
BlockCount() int
FindBlockByNumber(blockNumber int64) (core.Block, error)
MaxBlockNumber() int64
MissingBlockNumbers(startingBlockNumber int64, endingBlockNumber int64) []int64
FindReceipt(txHash string) (core.Receipt, error)
SetBlocksStatus(chainHead int64)
}
var ErrContractDoesNotExist = func(contractHash string) error {
return errors.New(fmt.Sprintf("Contract %v does not exist", contractHash))
}
type ContractRepository interface {
CreateContract(contract core.Contract) error
ContractExists(contractHash string) bool
FindContract(contractHash string) (core.Contract, error)
CreateLogs(log []core.Log) error
FindLogs(address string, blockNumber int64) []core.Log
SetBlocksStatus(chainHead int64)
}
type FilterRepository interface {
AddFilter(filter filters.LogFilter) error
}
type LogsRepository interface {
FindLogs(address string, blockNumber int64) []core.Log
CreateLogs(logs []core.Log) error
}
var ErrReceiptDoesNotExist = func(txHash string) error {
return errors.New(fmt.Sprintf("Receipt for tx: %v does not exist", txHash))
}
type ReceiptRepository interface {
FindReceipt(txHash string) (core.Receipt, error)
}

644
pkg/repositories/testing/helpers.go
View File

@ -1,644 +0,0 @@
package testing
import (
"sort"
"strconv"
"math/big"
"github.com/vulcanize/vulcanizedb/pkg/core"
"github.com/vulcanize/vulcanizedb/pkg/filters"
"github.com/vulcanize/vulcanizedb/pkg/repositories"
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
)
func ClearData(postgres repositories.Postgres) {
postgres.Db.MustExec("DELETE FROM watched_contracts")
postgres.Db.MustExec("DELETE FROM transactions")
postgres.Db.MustExec("DELETE FROM blocks")
postgres.Db.MustExec("DELETE FROM logs")
postgres.Db.MustExec("DELETE FROM receipts")
postgres.Db.MustExec("DELETE FROM log_filters")
}
func AssertRepositoryBehavior(buildRepository func(node core.Node) repositories.Repository) {
var repository repositories.Repository
BeforeEach(func() {
node := core.Node{
GenesisBlock: "GENESIS",
NetworkId: 1,
Id: "b6f90c0fdd8ec9607aed8ee45c69322e47b7063f0bfb7a29c8ecafab24d0a22d24dd2329b5ee6ed4125a03cb14e57fd584e67f9e53e6c631055cbbd82f080845",
ClientName: "Geth/v1.7.2-stable-1db4ecdc/darwin-amd64/go1.9",
}
repository = buildRepository(node)
})
Describe("Saving blocks", func() {
It("starts with no blocks", func() {
count := repository.BlockCount()
Expect(count).Should(Equal(0))
})
It("increments the block count", func() {
block := core.Block{Number: 123}
repository.CreateOrUpdateBlock(block)
Expect(repository.BlockCount()).To(Equal(1))
})
It("associates blocks to a node", func() {
block := core.Block{
Number: 123,
}
repository.CreateOrUpdateBlock(block)
nodeTwo := core.Node{
GenesisBlock: "0x456",
NetworkId: 1,
Id: "x123456",
ClientName: "Geth",
}
repositoryTwo := buildRepository(nodeTwo)
_, err := repositoryTwo.FindBlockByNumber(123)
Expect(err).To(HaveOccurred())
})
It("saves the attributes of the block", func() {
blockNumber := int64(123)
gasLimit := int64(1000000)
gasUsed := int64(10)
blockHash := "x123"
blockParentHash := "x456"
blockNonce := "0x881db2ca900682e9a9"
miner := "x123"
extraData := "xextraData"
blockTime := int64(1508981640)
uncleHash := "x789"
blockSize := int64(1000)
difficulty := int64(10)
blockReward := float64(5.132)
unclesReward := float64(3.580)
block := core.Block{
Reward: blockReward,
Difficulty: difficulty,
GasLimit: gasLimit,
GasUsed: gasUsed,
Hash: blockHash,
ExtraData: extraData,
Nonce: blockNonce,
Miner: miner,
Number: blockNumber,
ParentHash: blockParentHash,
Size: blockSize,
Time: blockTime,
UncleHash: uncleHash,
UnclesReward: unclesReward,
}
repository.CreateOrUpdateBlock(block)
savedBlock, err := repository.FindBlockByNumber(blockNumber)
Expect(err).NotTo(HaveOccurred())
Expect(savedBlock.Reward).To(Equal(blockReward))
Expect(savedBlock.Difficulty).To(Equal(difficulty))
Expect(savedBlock.GasLimit).To(Equal(gasLimit))
Expect(savedBlock.GasUsed).To(Equal(gasUsed))
Expect(savedBlock.Hash).To(Equal(blockHash))
Expect(savedBlock.Nonce).To(Equal(blockNonce))
Expect(savedBlock.Miner).To(Equal(miner))
Expect(savedBlock.ExtraData).To(Equal(extraData))
Expect(savedBlock.Number).To(Equal(blockNumber))
Expect(savedBlock.ParentHash).To(Equal(blockParentHash))
Expect(savedBlock.Size).To(Equal(blockSize))
Expect(savedBlock.Time).To(Equal(blockTime))
Expect(savedBlock.UncleHash).To(Equal(uncleHash))
Expect(savedBlock.UnclesReward).To(Equal(unclesReward))
})
It("does not find a block when searching for a number that does not exist", func() {
_, err := repository.FindBlockByNumber(111)
Expect(err).To(HaveOccurred())
})
It("saves one transaction associated to the block", func() {
block := core.Block{
Number: 123,
Transactions: []core.Transaction{{}},
}
repository.CreateOrUpdateBlock(block)
savedBlock, _ := repository.FindBlockByNumber(123)
Expect(len(savedBlock.Transactions)).To(Equal(1))
})
It("saves two transactions associated to the block", func() {
block := core.Block{
Number: 123,
Transactions: []core.Transaction{{}, {}},
}
repository.CreateOrUpdateBlock(block)
savedBlock, _ := repository.FindBlockByNumber(123)
Expect(len(savedBlock.Transactions)).To(Equal(2))
})
It(`replaces blocks and transactions associated to the block
when a more new block is in conflict (same block number + nodeid)`, func() {
blockOne := core.Block{
Number: 123,
Hash: "xabc",
Transactions: []core.Transaction{{Hash: "x123"}, {Hash: "x345"}},
}
blockTwo := core.Block{
Number: 123,
Hash: "xdef",
Transactions: []core.Transaction{{Hash: "x678"}, {Hash: "x9ab"}},
}
repository.CreateOrUpdateBlock(blockOne)
repository.CreateOrUpdateBlock(blockTwo)
savedBlock, _ := repository.FindBlockByNumber(123)
Expect(len(savedBlock.Transactions)).To(Equal(2))
Expect(savedBlock.Transactions[0].Hash).To(Equal("x678"))
Expect(savedBlock.Transactions[1].Hash).To(Equal("x9ab"))
})
It(`does not replace blocks when block number is not unique
but block number + node id is`, func() {
blockOne := core.Block{
Number: 123,
Transactions: []core.Transaction{{Hash: "x123"}, {Hash: "x345"}},
}
blockTwo := core.Block{
Number: 123,
Transactions: []core.Transaction{{Hash: "x678"}, {Hash: "x9ab"}},
}
repository.CreateOrUpdateBlock(blockOne)
nodeTwo := core.Node{
GenesisBlock: "0x456",
NetworkId: 1,
}
repositoryTwo := buildRepository(nodeTwo)
repository.CreateOrUpdateBlock(blockOne)
repositoryTwo.CreateOrUpdateBlock(blockTwo)
retrievedBlockOne, _ := repository.FindBlockByNumber(123)
retrievedBlockTwo, _ := repositoryTwo.FindBlockByNumber(123)
Expect(retrievedBlockOne.Transactions[0].Hash).To(Equal("x123"))
Expect(retrievedBlockTwo.Transactions[0].Hash).To(Equal("x678"))
})
It("saves the attributes associated to a transaction", func() {
gasLimit := int64(5000)
gasPrice := int64(3)
nonce := uint64(10000)
to := "1234567890"
from := "0987654321"
var value = new(big.Int)
value.SetString("34940183920000000000", 10)
inputData := "0xf7d8c8830000000000000000000000000000000000000000000000000000000000037788000000000000000000000000000000000000000000000000000000000003bd14"
transaction := core.Transaction{
Hash: "x1234",
GasPrice: gasPrice,
GasLimit: gasLimit,
Nonce: nonce,
To: to,
From: from,
Value: value.String(),
Data: inputData,
}
block := core.Block{
Number: 123,
Transactions: []core.Transaction{transaction},
}
repository.CreateOrUpdateBlock(block)
savedBlock, _ := repository.FindBlockByNumber(123)
Expect(len(savedBlock.Transactions)).To(Equal(1))
savedTransaction := savedBlock.Transactions[0]
Expect(savedTransaction.Data).To(Equal(transaction.Data))
Expect(savedTransaction.Hash).To(Equal(transaction.Hash))
Expect(savedTransaction.To).To(Equal(to))
Expect(savedTransaction.From).To(Equal(from))
Expect(savedTransaction.Nonce).To(Equal(nonce))
Expect(savedTransaction.GasLimit).To(Equal(gasLimit))
Expect(savedTransaction.GasPrice).To(Equal(gasPrice))
Expect(savedTransaction.Value).To(Equal(value.String()))
})
})
Describe("The missing block numbers", func() {
It("is empty the starting block number is the highest known block number", func() {
repository.CreateOrUpdateBlock(core.Block{Number: 1})
Expect(len(repository.MissingBlockNumbers(1, 1))).To(Equal(0))
})
It("is the only missing block number", func() {
repository.CreateOrUpdateBlock(core.Block{Number: 2})
Expect(repository.MissingBlockNumbers(1, 2)).To(Equal([]int64{1}))
})
It("is both missing block numbers", func() {
repository.CreateOrUpdateBlock(core.Block{Number: 3})
Expect(repository.MissingBlockNumbers(1, 3)).To(Equal([]int64{1, 2}))
})
It("goes back to the starting block number", func() {
repository.CreateOrUpdateBlock(core.Block{Number: 6})
Expect(repository.MissingBlockNumbers(4, 6)).To(Equal([]int64{4, 5}))
})
It("only includes missing block numbers", func() {
repository.CreateOrUpdateBlock(core.Block{Number: 4})
repository.CreateOrUpdateBlock(core.Block{Number: 6})
Expect(repository.MissingBlockNumbers(4, 6)).To(Equal([]int64{5}))
})
It("is a list with multiple gaps", func() {
repository.CreateOrUpdateBlock(core.Block{Number: 4})
repository.CreateOrUpdateBlock(core.Block{Number: 5})
repository.CreateOrUpdateBlock(core.Block{Number: 8})
repository.CreateOrUpdateBlock(core.Block{Number: 10})
Expect(repository.MissingBlockNumbers(3, 10)).To(Equal([]int64{3, 6, 7, 9}))
})
It("returns empty array when lower bound exceeds upper bound", func() {
Expect(repository.MissingBlockNumbers(10000, 1)).To(Equal([]int64{}))
})
It("only returns requested range even when other gaps exist", func() {
repository.CreateOrUpdateBlock(core.Block{Number: 3})
repository.CreateOrUpdateBlock(core.Block{Number: 8})
Expect(repository.MissingBlockNumbers(1, 5)).To(Equal([]int64{1, 2, 4, 5}))
})
})
Describe("The max block numbers", func() {
It("returns the block number when a single block", func() {
repository.CreateOrUpdateBlock(core.Block{Number: 1})
Expect(repository.MaxBlockNumber()).To(Equal(int64(1)))
})
It("returns highest known block number when multiple blocks", func() {
repository.CreateOrUpdateBlock(core.Block{Number: 1})
repository.CreateOrUpdateBlock(core.Block{Number: 10})
Expect(repository.MaxBlockNumber()).To(Equal(int64(10)))
})
})
Describe("The block status", func() {
It("sets the status of blocks within n-20 of chain HEAD as final", func() {
blockNumberOfChainHead := 25
for i := 0; i < blockNumberOfChainHead; i++ {
repository.CreateOrUpdateBlock(core.Block{Number: int64(i), Hash: strconv.Itoa(i)})
}
repository.SetBlocksStatus(int64(blockNumberOfChainHead))
blockOne, err := repository.FindBlockByNumber(1)
Expect(err).ToNot(HaveOccurred())
Expect(blockOne.IsFinal).To(Equal(true))
blockTwo, err := repository.FindBlockByNumber(24)
Expect(err).ToNot(HaveOccurred())
Expect(blockTwo.IsFinal).To(BeFalse())
})
})
Describe("Creating contracts", func() {
It("returns the contract when it exists", func() {
repository.CreateContract(core.Contract{Hash: "x123"})
contract, err := repository.FindContract("x123")
Expect(err).NotTo(HaveOccurred())
Expect(contract.Hash).To(Equal("x123"))
Expect(repository.ContractExists("x123")).To(BeTrue())
Expect(repository.ContractExists("x456")).To(BeFalse())
})
It("returns err if contract does not exist", func() {
_, err := repository.FindContract("x123")
Expect(err).To(HaveOccurred())
})
It("returns empty array when no transactions 'To' a contract", func() {
repository.CreateContract(core.Contract{Hash: "x123"})
contract, err := repository.FindContract("x123")
Expect(err).ToNot(HaveOccurred())
Expect(contract.Transactions).To(BeEmpty())
})
It("returns transactions 'To' a contract", func() {
block := core.Block{
Number: 123,
Transactions: []core.Transaction{
{Hash: "TRANSACTION1", To: "x123"},
{Hash: "TRANSACTION2", To: "x345"},
{Hash: "TRANSACTION3", To: "x123"},
},
}
repository.CreateOrUpdateBlock(block)
repository.CreateContract(core.Contract{Hash: "x123"})
contract, err := repository.FindContract("x123")
Expect(err).ToNot(HaveOccurred())
sort.Slice(contract.Transactions, func(i, j int) bool {
return contract.Transactions[i].Hash < contract.Transactions[j].Hash
})
Expect(contract.Transactions).To(
Equal([]core.Transaction{
{Hash: "TRANSACTION1", To: "x123"},
{Hash: "TRANSACTION3", To: "x123"},
}))
})
It("stores the ABI of the contract", func() {
repository.CreateContract(core.Contract{
Abi: "{\"some\": \"json\"}",
Hash: "x123",
})
contract, err := repository.FindContract("x123")
Expect(err).ToNot(HaveOccurred())
Expect(contract.Abi).To(Equal("{\"some\": \"json\"}"))
})
It("updates the ABI of the contract if hash already present", func() {
repository.CreateContract(core.Contract{
Abi: "{\"some\": \"json\"}",
Hash: "x123",
})
repository.CreateContract(core.Contract{
Abi: "{\"some\": \"different json\"}",
Hash: "x123",
})
contract, err := repository.FindContract("x123")
Expect(err).ToNot(HaveOccurred())
Expect(contract.Abi).To(Equal("{\"some\": \"different json\"}"))
})
})
Describe("Saving logs", func() {
It("returns the log when it exists", func() {
repository.CreateLogs([]core.Log{{
BlockNumber: 1,
Index: 0,
Address: "x123",
TxHash: "x456",
Topics: core.Topics{0: "x777", 1: "x888", 2: "x999"},
Data: "xabc",
}},
)
log := repository.FindLogs("x123", 1)
Expect(log).NotTo(BeNil())
Expect(log[0].BlockNumber).To(Equal(int64(1)))
Expect(log[0].Address).To(Equal("x123"))
Expect(log[0].Index).To(Equal(int64(0)))
Expect(log[0].TxHash).To(Equal("x456"))
Expect(log[0].Topics[0]).To(Equal("x777"))
Expect(log[0].Topics[1]).To(Equal("x888"))
Expect(log[0].Topics[2]).To(Equal("x999"))
Expect(log[0].Data).To(Equal("xabc"))
})
It("returns nil if log does not exist", func() {
log := repository.FindLogs("x123", 1)
Expect(log).To(BeNil())
})
It("filters to the correct block number and address", func() {
repository.CreateLogs([]core.Log{{
BlockNumber: 1,
Index: 0,
Address: "x123",
TxHash: "x456",
Topics: core.Topics{0: "x777", 1: "x888", 2: "x999"},
Data: "xabc",
}},
)
repository.CreateLogs([]core.Log{{
BlockNumber: 1,
Index: 1,
Address: "x123",
TxHash: "x789",
Topics: core.Topics{0: "x111", 1: "x222", 2: "x333"},
Data: "xdef",
}},
)
repository.CreateLogs([]core.Log{{
BlockNumber: 2,
Index: 0,
Address: "x123",
TxHash: "x456",
Topics: core.Topics{0: "x777", 1: "x888", 2: "x999"},
Data: "xabc",
}},
)
log := repository.FindLogs("x123", 1)
type logIndex struct {
blockNumber int64
Index int64
}
var uniqueBlockNumbers []logIndex
for _, log := range log {
uniqueBlockNumbers = append(uniqueBlockNumbers,
logIndex{log.BlockNumber, log.Index})
}
sort.Slice(uniqueBlockNumbers, func(i, j int) bool {
if uniqueBlockNumbers[i].blockNumber < uniqueBlockNumbers[j].blockNumber {
return true
}
if uniqueBlockNumbers[i].blockNumber > uniqueBlockNumbers[j].blockNumber {
return false
}
return uniqueBlockNumbers[i].Index < uniqueBlockNumbers[j].Index
})
Expect(log).NotTo(BeNil())
Expect(len(log)).To(Equal(2))
Expect(uniqueBlockNumbers).To(Equal(
[]logIndex{
{blockNumber: 1, Index: 0},
{blockNumber: 1, Index: 1}},
))
})
It("saves the logs attached to a receipt", func() {
logs := []core.Log{{
Address: "0x8a4774fe82c63484afef97ca8d89a6ea5e21f973",
BlockNumber: 4745407,
Data: "0x0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000004000000000000000000000000000000000000000000000000000000000000000645a68669900000000000000000000000000000000000000000000003397684ab5869b0000000000000000000000000000000000000000000000000000000000005a36053200000000000000000000000099041f808d598b782d5a3e498681c2452a31da08",
Index: 86,
Topics: core.Topics{
0: "0x5a68669900000000000000000000000000000000000000000000000000000000",
1: "0x000000000000000000000000d0148dad63f73ce6f1b6c607e3413dcf1ff5f030",
2: "0x00000000000000000000000000000000000000000000003397684ab5869b0000",
3: "0x000000000000000000000000000000000000000000000000000000005a360532",
},
TxHash: "0x002c4799161d809b23f67884eb6598c9df5894929fe1a9ead97ca175d360f547",
}, {
Address: "0x99041f808d598b782d5a3e498681c2452a31da08",
BlockNumber: 4745407,
Data: "0x00000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000040000000000000000000000000000000000000000000000000000000000000000418178358",
Index: 87,
Topics: core.Topics{
0: "0x1817835800000000000000000000000000000000000000000000000000000000",
1: "0x0000000000000000000000008a4774fe82c63484afef97ca8d89a6ea5e21f973",
2: "0x0000000000000000000000000000000000000000000000000000000000000000",
3: "0x0000000000000000000000000000000000000000000000000000000000000000",
},
TxHash: "0x002c4799161d809b23f67884eb6598c9df5894929fe1a9ead97ca175d360f547",
}, {
Address: "0x99041f808d598b782d5a3e498681c2452a31da08",
BlockNumber: 4745407,
Data: "0x00000000000000000000000000000000000000000000003338f64c8423af4000",
Index: 88,
Topics: core.Topics{
0: "0x296ba4ca62c6c21c95e828080cb8aec7481b71390585605300a8a76f9e95b527",
},
TxHash: "0x002c4799161d809b23f67884eb6598c9df5894929fe1a9ead97ca175d360f547",
},
}
receipt := core.Receipt{
ContractAddress: "",
CumulativeGasUsed: 7481414,
GasUsed: 60711,
Logs: logs,
Bloom: "0x00000800000000000000001000000000000000400000000080000000000000000000400000010000000000000000000000000000040000000000000002000000000000000000000000000000000000000000000000000000000000000000000000000000020000000000000000000800004000000000000001000000000000000000000000000002000000480000000000000002000000000000000020000000000000000000000000000000000000000080000000000180000c00000000000000002000002000000040000000000000000000000000000010000000000020000000000000000000002000000000000000000000000400800000000000000000",
Status: 1,
TxHash: "0x002c4799161d809b23f67884eb6598c9df5894929fe1a9ead97ca175d360f547",
}
transaction :=
core.Transaction{
Hash: receipt.TxHash,
Receipt: receipt,
}
block := core.Block{Transactions: []core.Transaction{transaction}}
err := repository.CreateOrUpdateBlock(block)
Expect(err).To(Not(HaveOccurred()))
retrievedLogs := repository.FindLogs("0x99041f808d598b782d5a3e498681c2452a31da08", 4745407)
expected := logs[1:]
Expect(retrievedLogs).To(Equal(expected))
})
It("still saves receipts without logs", func() {
receipt := core.Receipt{
TxHash: "0x002c4799161d809b23f67884eb6598c9df5894929fe1a9ead97ca175d360f547",
}
transaction := core.Transaction{
Hash: receipt.TxHash,
Receipt: receipt,
}
block := core.Block{
Transactions: []core.Transaction{transaction},
}
repository.CreateOrUpdateBlock(block)
_, err := repository.FindReceipt(receipt.TxHash)
Expect(err).To(Not(HaveOccurred()))
})
})
Describe("Saving receipts", func() {
It("returns the receipt when it exists", func() {
expected := core.Receipt{
ContractAddress: "0xde0b295669a9fd93d5f28d9ec85e40f4cb697bae",
CumulativeGasUsed: 7996119,
GasUsed: 21000,
Logs: []core.Log{},
StateRoot: "0x88abf7e73128227370aa7baa3dd4e18d0af70e92ef1f9ef426942fbe2dddb733",
Status: 1,
TxHash: "0xe340558980f89d5f86045ac11e5cc34e4bcec20f9f1e2a427aa39d87114e8223",
}
transaction := core.Transaction{
Hash: expected.TxHash,
Receipt: expected,
}
block := core.Block{Transactions: []core.Transaction{transaction}}
repository.CreateOrUpdateBlock(block)
receipt, err := repository.FindReceipt("0xe340558980f89d5f86045ac11e5cc34e4bcec20f9f1e2a427aa39d87114e8223")
Expect(err).ToNot(HaveOccurred())
//Not currently serializing bloom logs
Expect(receipt.Bloom).To(Equal(core.Receipt{}.Bloom))
Expect(receipt.TxHash).To(Equal(expected.TxHash))
Expect(receipt.CumulativeGasUsed).To(Equal(expected.CumulativeGasUsed))
Expect(receipt.GasUsed).To(Equal(expected.GasUsed))
Expect(receipt.StateRoot).To(Equal(expected.StateRoot))
Expect(receipt.Status).To(Equal(expected.Status))
})
It("returns ErrReceiptDoesNotExist when receipt does not exist", func() {
receipt, err := repository.FindReceipt("DOES NOT EXIST")
Expect(err).To(HaveOccurred())
Expect(receipt).To(BeZero())
})
})
Describe("LogFilter", func() {
It("inserts filter into watched events", func() {
logFilter := filters.LogFilter{
Name: "TestFilter",
FromBlock: 1,
ToBlock: 2,
Address: "0x8888f1f195afa192cfee860698584c030f4c9db1",
Topics: core.Topics{
"0x000000000000000000000000a94f5374fce5edbc8e2a8697c15331677e6ebf0b",
"",
"0x000000000000000000000000a94f5374fce5edbc8e2a8697c15331677e6ebf0b",
"",
},
}
err := repository.AddFilter(logFilter)
Expect(err).ToNot(HaveOccurred())
})
It("returns error if name is not provided", func() {
logFilter := filters.LogFilter{
FromBlock: 1,
ToBlock: 2,
Address: "0x8888f1f195afa192cfee860698584c030f4c9db1",
Topics: core.Topics{
"0x000000000000000000000000a94f5374fce5edbc8e2a8697c15331677e6ebf0b",
"",
"0x000000000000000000000000a94f5374fce5edbc8e2a8697c15331677e6ebf0b",
"",
},
}
err := repository.AddFilter(logFilter)
Expect(err).To(HaveOccurred())
})
})
}

8
utils/utils.go
View File

@ -12,7 +12,7 @@ import (
"github.com/vulcanize/vulcanizedb/pkg/config"
"github.com/vulcanize/vulcanizedb/pkg/core"
"github.com/vulcanize/vulcanizedb/pkg/geth"
"github.com/vulcanize/vulcanizedb/pkg/repositories"
"github.com/vulcanize/vulcanizedb/pkg/repositories/postgres"
)
func LoadConfig(environment string) config.Config {
@ -23,12 +23,12 @@ func LoadConfig(environment string) config.Config {
return cfg
}
func LoadPostgres(database config.Database, node core.Node) repositories.Postgres {
repository, err := repositories.NewPostgres(database, node)
func LoadPostgres(database config.Database, node core.Node) postgres.DB {
repository, err := postgres.NewDB(database, node)
if err != nil {
log.Fatalf("Error loading postgres\n%v", err)
}
return repository
return *repository
}
func ReadAbiFile(abiFilepath string) string {