forked from cerc-io/ipld-eth-server
Handle events
- Adds interfaces for developers to build handlers that update data in response to log events - Resolves #29
This commit is contained in:
Matt Krump
@@ -0,0 +1,269 @@
|
||||
package repositories
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
|
||||
"log"
|
||||
|
||||
"github.com/jmoiron/sqlx"
|
||||
"github.com/vulcanize/vulcanizedb/pkg/core"
|
||||
"github.com/vulcanize/vulcanizedb/pkg/datastore"
|
||||
"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
|
||||
)
|
||||
|
||||
const (
|
||||
blocksFromHeadBeforeFinal = 20
|
||||
)
|
||||
|
||||
type BlockRepository struct {
|
||||
*postgres.DB
|
||||
}
|
||||
|
||||
func (blockRepository BlockRepository) SetBlocksStatus(chainHead int64) {
|
||||
cutoff := chainHead - blocksFromHeadBeforeFinal
|
||||
blockRepository.DB.Exec(`
|
||||
UPDATE blocks SET is_final = TRUE
|
||||
WHERE is_final = FALSE AND number < $1`,
|
||||
cutoff)
|
||||
}
|
||||
|
||||
func (blockRepository BlockRepository) CreateOrUpdateBlock(block core.Block) error {
|
||||
var err error
|
||||
retrievedBlockHash, ok := blockRepository.getBlockHash(block)
|
||||
if !ok {
|
||||
err = blockRepository.insertBlock(block)
|
||||
return err
|
||||
}
|
||||
if ok && retrievedBlockHash != block.Hash {
|
||||
err = blockRepository.removeBlock(block.Number)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
err = blockRepository.insertBlock(block)
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (blockRepository BlockRepository) MissingBlockNumbers(startingBlockNumber int64, highestBlockNumber int64) []int64 {
|
||||
numbers := make([]int64, 0)
|
||||
blockRepository.DB.Select(&numbers,
|
||||
`SELECT all_block_numbers
|
||||
FROM (
|
||||
SELECT generate_series($1::INT, $2::INT) AS all_block_numbers) series
|
||||
LEFT JOIN blocks
|
||||
ON number = all_block_numbers
|
||||
WHERE number ISNULL`,
|
||||
startingBlockNumber,
|
||||
highestBlockNumber)
|
||||
return numbers
|
||||
}
|
||||
|
||||
func (blockRepository BlockRepository) GetBlock(blockNumber int64) (core.Block, error) {
|
||||
blockRows := blockRepository.DB.QueryRowx(
|
||||
`SELECT id,
|
||||
number,
|
||||
gaslimit,
|
||||
gasused,
|
||||
time,
|
||||
difficulty,
|
||||
hash,
|
||||
nonce,
|
||||
parenthash,
|
||||
size,
|
||||
uncle_hash,
|
||||
is_final,
|
||||
miner,
|
||||
extra_data,
|
||||
reward,
|
||||
uncles_reward
|
||||
FROM blocks
|
||||
WHERE node_id = $1 AND number = $2`, blockRepository.NodeID, blockNumber)
|
||||
savedBlock, err := blockRepository.loadBlock(blockRows)
|
||||
if err != nil {
|
||||
switch err {
|
||||
case sql.ErrNoRows:
|
||||
return core.Block{}, datastore.ErrBlockDoesNotExist(blockNumber)
|
||||
default:
|
||||
return savedBlock, err
|
||||
}
|
||||
}
|
||||
return savedBlock, nil
|
||||
}
|
||||
|
||||
func (blockRepository BlockRepository) insertBlock(block core.Block) error {
|
||||
var blockId int64
|
||||
tx, _ := blockRepository.DB.BeginTx(context.Background(), nil)
|
||||
err := tx.QueryRow(
|
||||
`INSERT INTO blocks
|
||||
(node_id, number, gaslimit, gasused, time, difficulty, hash, nonce, parenthash, size, uncle_hash, is_final, miner, extra_data, reward, uncles_reward)
|
||||
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16)
|
||||
RETURNING id `,
|
||||
blockRepository.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 postgres.ErrDBInsertFailed
|
||||
}
|
||||
err = blockRepository.createTransactions(tx, blockId, block.Transactions)
|
||||
if err != nil {
|
||||
tx.Rollback()
|
||||
return postgres.ErrDBInsertFailed
|
||||
}
|
||||
tx.Commit()
|
||||
return nil
|
||||
}
|
||||
|
||||
func (blockRepository BlockRepository) createTransactions(tx *sql.Tx, blockId int64, transactions []core.Transaction) error {
|
||||
for _, transaction := range transactions {
|
||||
err := blockRepository.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 (blockRepository BlockRepository) createTransaction(tx *sql.Tx, blockId int64, transaction core.Transaction) error {
|
||||
var transactionId int
|
||||
err := tx.QueryRow(
|
||||
`INSERT INTO transactions
|
||||
(block_id, hash, nonce, tx_to, tx_from, gaslimit, gasprice, value, 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 := blockRepository.createReceipt(tx, transactionId, transaction.Receipt)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if hasLogs(transaction) {
|
||||
err = blockRepository.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 (blockRepository BlockRepository) 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 (blockRepository BlockRepository) getBlockHash(block core.Block) (string, bool) {
|
||||
var retrievedBlockHash string
|
||||
blockRepository.DB.Get(&retrievedBlockHash,
|
||||
`SELECT hash
|
||||
FROM blocks
|
||||
WHERE number = $1 AND node_id = $2`,
|
||||
block.Number, blockRepository.NodeID)
|
||||
return retrievedBlockHash, blockExists(retrievedBlockHash)
|
||||
}
|
||||
|
||||
func (blockRepository BlockRepository) 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 postgres.ErrDBInsertFailed
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func blockExists(retrievedBlockHash string) bool {
|
||||
return retrievedBlockHash != ""
|
||||
}
|
||||
|
||||
func (blockRepository BlockRepository) removeBlock(blockNumber int64) error {
|
||||
_, err := blockRepository.DB.Exec(
|
||||
`DELETE FROM
|
||||
blocks
|
||||
WHERE number=$1 AND node_id=$2`,
|
||||
blockNumber, blockRepository.NodeID)
|
||||
if err != nil {
|
||||
return postgres.ErrDBDeleteFailed
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (blockRepository BlockRepository) 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 := blockRepository.DB.Queryx(`
|
||||
SELECT hash,
|
||||
nonce,
|
||||
tx_to,
|
||||
tx_from,
|
||||
gaslimit,
|
||||
gasprice,
|
||||
value,
|
||||
input_data
|
||||
FROM transactions
|
||||
WHERE block_id = $1
|
||||
ORDER BY hash`, block.ID)
|
||||
if err != nil {
|
||||
return core.Block{}, err
|
||||
}
|
||||
block.Transactions = blockRepository.LoadTransactions(transactionRows)
|
||||
return block.Block, nil
|
||||
}
|
||||
|
||||
func (blockRepository BlockRepository) LoadTransactions(transactionRows *sqlx.Rows) []core.Transaction {
|
||||
var transactions []core.Transaction
|
||||
for transactionRows.Next() {
|
||||
var transaction core.Transaction
|
||||
err := transactionRows.StructScan(&transaction)
|
||||
if err != nil {
|
||||
log.Fatal(err)
|
||||
}
|
||||
transactions = append(transactions, transaction)
|
||||
}
|
||||
return transactions
|
||||
}
|
||||
@@ -0,0 +1,289 @@
|
||||
package repositories_test
|
||||
|
||||
import (
|
||||
"math/big"
|
||||
"strconv"
|
||||
|
||||
. "github.com/onsi/ginkgo"
|
||||
. "github.com/onsi/gomega"
|
||||
"github.com/vulcanize/vulcanizedb/pkg/core"
|
||||
"github.com/vulcanize/vulcanizedb/pkg/datastore"
|
||||
"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
|
||||
"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres/repositories"
|
||||
"github.com/vulcanize/vulcanizedb/test_config"
|
||||
)
|
||||
|
||||
var _ = Describe("Saving blocks", func() {
|
||||
var db *postgres.DB
|
||||
var blockRepository datastore.BlockRepository
|
||||
BeforeEach(func() {
|
||||
node := core.Node{
|
||||
GenesisBlock: "GENESIS",
|
||||
NetworkID: 1,
|
||||
ID: "b6f90c0fdd8ec9607aed8ee45c69322e47b7063f0bfb7a29c8ecafab24d0a22d24dd2329b5ee6ed4125a03cb14e57fd584e67f9e53e6c631055cbbd82f080845",
|
||||
ClientName: "Geth/v1.7.2-stable-1db4ecdc/darwin-amd64/go1.9",
|
||||
}
|
||||
db = test_config.NewTestDB(node)
|
||||
blockRepository = repositories.BlockRepository{DB: db}
|
||||
|
||||
})
|
||||
|
||||
It("associates blocks to a node", func() {
|
||||
block := core.Block{
|
||||
Number: 123,
|
||||
}
|
||||
blockRepository.CreateOrUpdateBlock(block)
|
||||
nodeTwo := core.Node{
|
||||
GenesisBlock: "0x456",
|
||||
NetworkID: 1,
|
||||
ID: "x123456",
|
||||
ClientName: "Geth",
|
||||
}
|
||||
dbTwo := test_config.NewTestDB(nodeTwo)
|
||||
repositoryTwo := repositories.BlockRepository{DB: dbTwo}
|
||||
|
||||
_, err := repositoryTwo.GetBlock(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,
|
||||
}
|
||||
|
||||
blockRepository.CreateOrUpdateBlock(block)
|
||||
|
||||
savedBlock, err := blockRepository.GetBlock(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 := blockRepository.GetBlock(111)
|
||||
|
||||
Expect(err).To(HaveOccurred())
|
||||
})
|
||||
|
||||
It("saves one transaction associated to the block", func() {
|
||||
block := core.Block{
|
||||
Number: 123,
|
||||
Transactions: []core.Transaction{{}},
|
||||
}
|
||||
|
||||
blockRepository.CreateOrUpdateBlock(block)
|
||||
|
||||
savedBlock, _ := blockRepository.GetBlock(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{{}, {}},
|
||||
}
|
||||
|
||||
blockRepository.CreateOrUpdateBlock(block)
|
||||
|
||||
savedBlock, _ := blockRepository.GetBlock(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"}},
|
||||
}
|
||||
|
||||
blockRepository.CreateOrUpdateBlock(blockOne)
|
||||
blockRepository.CreateOrUpdateBlock(blockTwo)
|
||||
|
||||
savedBlock, _ := blockRepository.GetBlock(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"}},
|
||||
}
|
||||
blockRepository.CreateOrUpdateBlock(blockOne)
|
||||
nodeTwo := core.Node{
|
||||
GenesisBlock: "0x456",
|
||||
NetworkID: 1,
|
||||
}
|
||||
dbTwo := test_config.NewTestDB(nodeTwo)
|
||||
repositoryTwo := repositories.BlockRepository{DB: dbTwo}
|
||||
|
||||
blockRepository.CreateOrUpdateBlock(blockOne)
|
||||
repositoryTwo.CreateOrUpdateBlock(blockTwo)
|
||||
retrievedBlockOne, _ := blockRepository.GetBlock(123)
|
||||
retrievedBlockTwo, _ := repositoryTwo.GetBlock(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},
|
||||
}
|
||||
|
||||
blockRepository.CreateOrUpdateBlock(block)
|
||||
|
||||
savedBlock, _ := blockRepository.GetBlock(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() {
|
||||
blockRepository.CreateOrUpdateBlock(core.Block{Number: 1})
|
||||
|
||||
Expect(len(blockRepository.MissingBlockNumbers(1, 1))).To(Equal(0))
|
||||
})
|
||||
|
||||
It("is the only missing block number", func() {
|
||||
blockRepository.CreateOrUpdateBlock(core.Block{Number: 2})
|
||||
|
||||
Expect(blockRepository.MissingBlockNumbers(1, 2)).To(Equal([]int64{1}))
|
||||
})
|
||||
|
||||
It("is both missing block numbers", func() {
|
||||
blockRepository.CreateOrUpdateBlock(core.Block{Number: 3})
|
||||
|
||||
Expect(blockRepository.MissingBlockNumbers(1, 3)).To(Equal([]int64{1, 2}))
|
||||
})
|
||||
|
||||
It("goes back to the starting block number", func() {
|
||||
blockRepository.CreateOrUpdateBlock(core.Block{Number: 6})
|
||||
|
||||
Expect(blockRepository.MissingBlockNumbers(4, 6)).To(Equal([]int64{4, 5}))
|
||||
})
|
||||
|
||||
It("only includes missing block numbers", func() {
|
||||
blockRepository.CreateOrUpdateBlock(core.Block{Number: 4})
|
||||
blockRepository.CreateOrUpdateBlock(core.Block{Number: 6})
|
||||
|
||||
Expect(blockRepository.MissingBlockNumbers(4, 6)).To(Equal([]int64{5}))
|
||||
})
|
||||
|
||||
It("is a list with multiple gaps", func() {
|
||||
blockRepository.CreateOrUpdateBlock(core.Block{Number: 4})
|
||||
blockRepository.CreateOrUpdateBlock(core.Block{Number: 5})
|
||||
blockRepository.CreateOrUpdateBlock(core.Block{Number: 8})
|
||||
blockRepository.CreateOrUpdateBlock(core.Block{Number: 10})
|
||||
|
||||
Expect(blockRepository.MissingBlockNumbers(3, 10)).To(Equal([]int64{3, 6, 7, 9}))
|
||||
})
|
||||
|
||||
It("returns empty array when lower bound exceeds upper bound", func() {
|
||||
Expect(blockRepository.MissingBlockNumbers(10000, 1)).To(Equal([]int64{}))
|
||||
})
|
||||
|
||||
It("only returns requested range even when other gaps exist", func() {
|
||||
blockRepository.CreateOrUpdateBlock(core.Block{Number: 3})
|
||||
blockRepository.CreateOrUpdateBlock(core.Block{Number: 8})
|
||||
|
||||
Expect(blockRepository.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++ {
|
||||
blockRepository.CreateOrUpdateBlock(core.Block{Number: int64(i), Hash: strconv.Itoa(i)})
|
||||
}
|
||||
|
||||
blockRepository.SetBlocksStatus(int64(blockNumberOfChainHead))
|
||||
|
||||
blockOne, err := blockRepository.GetBlock(1)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Expect(blockOne.IsFinal).To(Equal(true))
|
||||
blockTwo, err := blockRepository.GetBlock(24)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Expect(blockTwo.IsFinal).To(BeFalse())
|
||||
})
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,74 @@
|
||||
package repositories
|
||||
|
||||
import (
|
||||
"database/sql"
|
||||
|
||||
"github.com/vulcanize/vulcanizedb/pkg/core"
|
||||
"github.com/vulcanize/vulcanizedb/pkg/datastore"
|
||||
"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
|
||||
)
|
||||
|
||||
type ContractRepository struct {
|
||||
*postgres.DB
|
||||
}
|
||||
|
||||
func (contractRepository ContractRepository) CreateContract(contract core.Contract) error {
|
||||
abi := contract.Abi
|
||||
var abiToInsert *string
|
||||
if abi != "" {
|
||||
abiToInsert = &abi
|
||||
}
|
||||
_, err := contractRepository.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 postgres.ErrDBInsertFailed
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (contractRepository ContractRepository) ContractExists(contractHash string) bool {
|
||||
var exists bool
|
||||
contractRepository.DB.QueryRow(
|
||||
`SELECT exists(
|
||||
SELECT 1
|
||||
FROM watched_contracts
|
||||
WHERE contract_hash = $1)`, contractHash).Scan(&exists)
|
||||
return exists
|
||||
}
|
||||
|
||||
func (contractRepository ContractRepository) GetContract(contractHash string) (core.Contract, error) {
|
||||
var hash string
|
||||
var abi string
|
||||
contract := contractRepository.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{}, datastore.ErrContractDoesNotExist(contractHash)
|
||||
}
|
||||
savedContract := contractRepository.addTransactions(core.Contract{Hash: hash, Abi: abi})
|
||||
return savedContract, nil
|
||||
}
|
||||
|
||||
func (contractRepository ContractRepository) addTransactions(contract core.Contract) core.Contract {
|
||||
transactionRows, _ := contractRepository.DB.Queryx(`
|
||||
SELECT hash,
|
||||
nonce,
|
||||
tx_to,
|
||||
tx_from,
|
||||
gaslimit,
|
||||
gasprice,
|
||||
value,
|
||||
input_data
|
||||
FROM transactions
|
||||
WHERE tx_to = $1
|
||||
ORDER BY block_id DESC`, contract.Hash)
|
||||
blockRepository := &BlockRepository{contractRepository.DB}
|
||||
transactions := blockRepository.LoadTransactions(transactionRows)
|
||||
savedContract := core.Contract{Hash: contract.Hash, Transactions: transactions, Abi: contract.Abi}
|
||||
return savedContract
|
||||
}
|
||||
@@ -0,0 +1,103 @@
|
||||
package repositories_test
|
||||
|
||||
import (
|
||||
"sort"
|
||||
|
||||
. "github.com/onsi/ginkgo"
|
||||
. "github.com/onsi/gomega"
|
||||
"github.com/vulcanize/vulcanizedb/pkg/core"
|
||||
"github.com/vulcanize/vulcanizedb/pkg/datastore"
|
||||
"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
|
||||
"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres/repositories"
|
||||
"github.com/vulcanize/vulcanizedb/test_config"
|
||||
)
|
||||
|
||||
var _ = Describe("Creating contracts", func() {
|
||||
var db *postgres.DB
|
||||
var contractRepository datastore.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",
|
||||
}
|
||||
db = test_config.NewTestDB(node)
|
||||
contractRepository = repositories.ContractRepository{DB: db}
|
||||
})
|
||||
|
||||
It("returns the contract when it exists", func() {
|
||||
contractRepository.CreateContract(core.Contract{Hash: "x123"})
|
||||
|
||||
contract, err := contractRepository.GetContract("x123")
|
||||
Expect(err).NotTo(HaveOccurred())
|
||||
Expect(contract.Hash).To(Equal("x123"))
|
||||
|
||||
Expect(contractRepository.ContractExists("x123")).To(BeTrue())
|
||||
Expect(contractRepository.ContractExists("x456")).To(BeFalse())
|
||||
})
|
||||
|
||||
It("returns err if contract does not exist", func() {
|
||||
_, err := contractRepository.GetContract("x123")
|
||||
Expect(err).To(HaveOccurred())
|
||||
})
|
||||
|
||||
It("returns empty array when no transactions 'To' a contract", func() {
|
||||
contractRepository.CreateContract(core.Contract{Hash: "x123"})
|
||||
contract, err := contractRepository.GetContract("x123")
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Expect(contract.Transactions).To(BeEmpty())
|
||||
})
|
||||
|
||||
It("returns transactions 'To' a contract", func() {
|
||||
var blockRepository datastore.BlockRepository
|
||||
blockRepository = repositories.BlockRepository{DB: db}
|
||||
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)
|
||||
|
||||
contractRepository.CreateContract(core.Contract{Hash: "x123"})
|
||||
contract, err := contractRepository.GetContract("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() {
|
||||
contractRepository.CreateContract(core.Contract{
|
||||
Abi: "{\"some\": \"json\"}",
|
||||
Hash: "x123",
|
||||
})
|
||||
contract, err := contractRepository.GetContract("x123")
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Expect(contract.Abi).To(Equal("{\"some\": \"json\"}"))
|
||||
})
|
||||
|
||||
It("updates the ABI of the contract if hash already present", func() {
|
||||
contractRepository.CreateContract(core.Contract{
|
||||
Abi: "{\"some\": \"json\"}",
|
||||
Hash: "x123",
|
||||
})
|
||||
contractRepository.CreateContract(core.Contract{
|
||||
Abi: "{\"some\": \"different json\"}",
|
||||
Hash: "x123",
|
||||
})
|
||||
contract, err := contractRepository.GetContract("x123")
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Expect(contract.Abi).To(Equal("{\"some\": \"different json\"}"))
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,76 @@
|
||||
package repositories
|
||||
|
||||
import (
|
||||
"database/sql"
|
||||
|
||||
"encoding/json"
|
||||
"errors"
|
||||
|
||||
"github.com/vulcanize/vulcanizedb/pkg/datastore"
|
||||
"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
|
||||
"github.com/vulcanize/vulcanizedb/pkg/filters"
|
||||
)
|
||||
|
||||
type FilterRepository struct {
|
||||
*postgres.DB
|
||||
}
|
||||
|
||||
func (filterRepository FilterRepository) CreateFilter(query filters.LogFilter) error {
|
||||
_, err := filterRepository.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 (filterRepository FilterRepository) GetFilter(name string) (filters.LogFilter, error) {
|
||||
lf := DBLogFilter{}
|
||||
err := filterRepository.DB.Get(&lf,
|
||||
`SELECT
|
||||
id,
|
||||
name,
|
||||
from_block,
|
||||
to_block,
|
||||
address,
|
||||
json_build_array(topic0, topic1, topic2, topic3) AS topics
|
||||
FROM log_filters
|
||||
WHERE name = $1`, name)
|
||||
if err != nil {
|
||||
switch err {
|
||||
case sql.ErrNoRows:
|
||||
return filters.LogFilter{}, datastore.ErrFilterDoesNotExist(name)
|
||||
default:
|
||||
return filters.LogFilter{}, err
|
||||
}
|
||||
}
|
||||
dbLogFilterToCoreLogFilter(lf)
|
||||
return *lf.LogFilter, nil
|
||||
}
|
||||
|
||||
type DBTopics []*string
|
||||
|
||||
func (t *DBTopics) Scan(src interface{}) error {
|
||||
asBytes, ok := src.([]byte)
|
||||
if !ok {
|
||||
return error(errors.New("scan source was not []byte"))
|
||||
}
|
||||
return json.Unmarshal(asBytes, &t)
|
||||
}
|
||||
|
||||
type DBLogFilter struct {
|
||||
ID int
|
||||
*filters.LogFilter
|
||||
Topics DBTopics
|
||||
}
|
||||
|
||||
func dbLogFilterToCoreLogFilter(lf DBLogFilter) {
|
||||
for i, v := range lf.Topics {
|
||||
if v != nil {
|
||||
lf.LogFilter.Topics[i] = *v
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,110 @@
|
||||
package repositories_test
|
||||
|
||||
import (
|
||||
. "github.com/onsi/ginkgo"
|
||||
. "github.com/onsi/gomega"
|
||||
"github.com/vulcanize/vulcanizedb/pkg/core"
|
||||
"github.com/vulcanize/vulcanizedb/pkg/datastore"
|
||||
"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
|
||||
"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres/repositories"
|
||||
"github.com/vulcanize/vulcanizedb/pkg/filters"
|
||||
"github.com/vulcanize/vulcanizedb/test_config"
|
||||
)
|
||||
|
||||
var _ = Describe("Log Filters Repository", func() {
|
||||
var db *postgres.DB
|
||||
var filterRepository datastore.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",
|
||||
}
|
||||
db = test_config.NewTestDB(node)
|
||||
filterRepository = repositories.FilterRepository{DB: db}
|
||||
})
|
||||
|
||||
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 := filterRepository.CreateFilter(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 := filterRepository.CreateFilter(logFilter)
|
||||
Expect(err).To(HaveOccurred())
|
||||
})
|
||||
|
||||
It("gets a log filter", func() {
|
||||
|
||||
expectedLogFilter1 := filters.LogFilter{
|
||||
Name: "TestFilter1",
|
||||
FromBlock: 1,
|
||||
ToBlock: 2,
|
||||
Address: "0x8888f1f195afa192cfee860698584c030f4c9db1",
|
||||
Topics: core.Topics{
|
||||
"0x000000000000000000000000a94f5374fce5edbc8e2a8697c15331677e6ebf0b",
|
||||
"",
|
||||
"0x000000000000000000000000a94f5374fce5edbc8e2a8697c15331677e6ebf0b",
|
||||
"",
|
||||
},
|
||||
}
|
||||
err := filterRepository.CreateFilter(expectedLogFilter1)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
expectedLogFilter2 := filters.LogFilter{
|
||||
Name: "TestFilter2",
|
||||
FromBlock: 10,
|
||||
ToBlock: 20,
|
||||
Address: "0x8888f1f195afa192cfee860698584c030f4c9db1",
|
||||
Topics: core.Topics{
|
||||
"0x000000000000000000000000a94f5374fce5edbc8e2a8697c15331677e6ebf0b",
|
||||
"",
|
||||
"0x000000000000000000000000a94f5374fce5edbc8e2a8697c15331677e6ebf0b",
|
||||
"",
|
||||
},
|
||||
}
|
||||
err = filterRepository.CreateFilter(expectedLogFilter2)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
|
||||
logFilter1, err := filterRepository.GetFilter("TestFilter1")
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Expect(logFilter1).To(Equal(expectedLogFilter1))
|
||||
logFilter2, err := filterRepository.GetFilter("TestFilter2")
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Expect(logFilter2).To(Equal(expectedLogFilter2))
|
||||
})
|
||||
|
||||
It("returns ErrFilterDoesNotExist error when log does not exist", func() {
|
||||
_, err := filterRepository.GetFilter("TestFilter1")
|
||||
Expect(err).To(Equal(datastore.ErrFilterDoesNotExist("TestFilter1")))
|
||||
})
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,74 @@
|
||||
package repositories
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"database/sql"
|
||||
|
||||
"github.com/vulcanize/vulcanizedb/pkg/core"
|
||||
"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
|
||||
)
|
||||
|
||||
type LogRepository struct {
|
||||
*postgres.DB
|
||||
}
|
||||
|
||||
func (logRepository LogRepository) CreateLogs(lgs []core.Log) error {
|
||||
tx, _ := logRepository.DB.BeginTx(context.Background(), nil)
|
||||
for _, tlog := range lgs {
|
||||
_, 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 postgres.ErrDBInsertFailed
|
||||
}
|
||||
}
|
||||
tx.Commit()
|
||||
return nil
|
||||
}
|
||||
|
||||
func (logRepository LogRepository) GetLogs(address string, blockNumber int64) []core.Log {
|
||||
logRows, _ := logRepository.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 logRepository.loadLogs(logRows)
|
||||
}
|
||||
|
||||
func (logRepository LogRepository) loadLogs(logsRows *sql.Rows) []core.Log {
|
||||
var lgs []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)
|
||||
lg := core.Log{
|
||||
BlockNumber: blockNumber,
|
||||
TxHash: txHash,
|
||||
Address: address,
|
||||
Index: index,
|
||||
Data: data,
|
||||
}
|
||||
for i, topic := range topics {
|
||||
lg.Topics[i] = topic
|
||||
}
|
||||
lgs = append(lgs, lg)
|
||||
}
|
||||
return lgs
|
||||
}
|
||||
@@ -0,0 +1,182 @@
|
||||
package repositories_test
|
||||
|
||||
import (
|
||||
"sort"
|
||||
|
||||
. "github.com/onsi/ginkgo"
|
||||
. "github.com/onsi/gomega"
|
||||
"github.com/vulcanize/vulcanizedb/pkg/core"
|
||||
"github.com/vulcanize/vulcanizedb/pkg/datastore"
|
||||
"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
|
||||
"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres/repositories"
|
||||
"github.com/vulcanize/vulcanizedb/test_config"
|
||||
)
|
||||
|
||||
var _ = Describe("Logs Repository", func() {
|
||||
var db *postgres.DB
|
||||
var logsRepository datastore.LogRepository
|
||||
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",
|
||||
}
|
||||
db = test_config.NewTestDB(node)
|
||||
logsRepository = repositories.LogRepository{DB: db}
|
||||
})
|
||||
|
||||
Describe("Saving logs", func() {
|
||||
It("returns the log when it exists", func() {
|
||||
logsRepository.CreateLogs([]core.Log{{
|
||||
BlockNumber: 1,
|
||||
Index: 0,
|
||||
Address: "x123",
|
||||
TxHash: "x456",
|
||||
Topics: core.Topics{0: "x777", 1: "x888", 2: "x999"},
|
||||
Data: "xabc",
|
||||
}},
|
||||
)
|
||||
|
||||
log := logsRepository.GetLogs("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 := logsRepository.GetLogs("x123", 1)
|
||||
Expect(log).To(BeNil())
|
||||
})
|
||||
|
||||
It("filters to the correct block number and address", func() {
|
||||
logsRepository.CreateLogs([]core.Log{{
|
||||
BlockNumber: 1,
|
||||
Index: 0,
|
||||
Address: "x123",
|
||||
TxHash: "x456",
|
||||
Topics: core.Topics{0: "x777", 1: "x888", 2: "x999"},
|
||||
Data: "xabc",
|
||||
}},
|
||||
)
|
||||
logsRepository.CreateLogs([]core.Log{{
|
||||
BlockNumber: 1,
|
||||
Index: 1,
|
||||
Address: "x123",
|
||||
TxHash: "x789",
|
||||
Topics: core.Topics{0: "x111", 1: "x222", 2: "x333"},
|
||||
Data: "xdef",
|
||||
}},
|
||||
)
|
||||
logsRepository.CreateLogs([]core.Log{{
|
||||
BlockNumber: 2,
|
||||
Index: 0,
|
||||
Address: "x123",
|
||||
TxHash: "x456",
|
||||
Topics: core.Topics{0: "x777", 1: "x888", 2: "x999"},
|
||||
Data: "xabc",
|
||||
}},
|
||||
)
|
||||
|
||||
log := logsRepository.GetLogs("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 datastore.BlockRepository
|
||||
blockRepository = repositories.BlockRepository{DB: db}
|
||||
|
||||
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 := logsRepository.GetLogs("0x99041f808d598b782d5a3e498681c2452a31da08", 4745407)
|
||||
|
||||
expected := logs[1:]
|
||||
Expect(retrievedLogs).To(Equal(expected))
|
||||
})
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,54 @@
|
||||
package repositories
|
||||
|
||||
import (
|
||||
"database/sql"
|
||||
|
||||
"github.com/vulcanize/vulcanizedb/pkg/core"
|
||||
"github.com/vulcanize/vulcanizedb/pkg/datastore"
|
||||
"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
|
||||
)
|
||||
|
||||
type ReceiptRepository struct {
|
||||
*postgres.DB
|
||||
}
|
||||
|
||||
func (receiptRepository ReceiptRepository) GetReceipt(txHash string) (core.Receipt, error) {
|
||||
row := receiptRepository.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{}, datastore.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
|
||||
}
|
||||
@@ -0,0 +1,90 @@
|
||||
package repositories_test
|
||||
|
||||
import (
|
||||
. "github.com/onsi/ginkgo"
|
||||
. "github.com/onsi/gomega"
|
||||
"github.com/vulcanize/vulcanizedb/pkg/core"
|
||||
"github.com/vulcanize/vulcanizedb/pkg/datastore"
|
||||
"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
|
||||
"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres/repositories"
|
||||
"github.com/vulcanize/vulcanizedb/test_config"
|
||||
)
|
||||
|
||||
var _ bool = Describe("Logs Repository", func() {
|
||||
var receiptRepository datastore.ReceiptRepository
|
||||
var db *postgres.DB
|
||||
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",
|
||||
}
|
||||
db = test_config.NewTestDB(node)
|
||||
receiptRepository = repositories.ReceiptRepository{DB: db}
|
||||
})
|
||||
|
||||
Describe("Saving receipts", func() {
|
||||
It("returns the receipt when it exists", func() {
|
||||
var blockRepository datastore.BlockRepository
|
||||
db := test_config.NewTestDB(node)
|
||||
blockRepository = repositories.BlockRepository{DB: db}
|
||||
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 := receiptRepository.GetReceipt("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 := receiptRepository.GetReceipt("DOES NOT EXIST")
|
||||
Expect(err).To(HaveOccurred())
|
||||
Expect(receipt).To(BeZero())
|
||||
})
|
||||
|
||||
It("still saves receipts without logs", func() {
|
||||
var blockRepository datastore.BlockRepository
|
||||
db := test_config.NewTestDB(node)
|
||||
blockRepository = repositories.BlockRepository{DB: db}
|
||||
receipt := core.Receipt{
|
||||
TxHash: "0x002c4799161d809b23f67884eb6598c9df5894929fe1a9ead97ca175d360f547",
|
||||
}
|
||||
transaction := core.Transaction{
|
||||
Hash: receipt.TxHash,
|
||||
Receipt: receipt,
|
||||
}
|
||||
|
||||
block := core.Block{
|
||||
Transactions: []core.Transaction{transaction},
|
||||
}
|
||||
blockRepository.CreateOrUpdateBlock(block)
|
||||
|
||||
_, err := receiptRepository.GetReceipt(receipt.TxHash)
|
||||
|
||||
Expect(err).To(Not(HaveOccurred()))
|
||||
})
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,13 @@
|
||||
package repositories_test
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
. "github.com/onsi/ginkgo"
|
||||
. "github.com/onsi/gomega"
|
||||
)
|
||||
|
||||
func TestRepositories(t *testing.T) {
|
||||
RegisterFailHandler(Fail)
|
||||
RunSpecs(t, "Repositories Suite")
|
||||
}
|
||||
@@ -0,0 +1,32 @@
|
||||
package repositories
|
||||
|
||||
import (
|
||||
"github.com/vulcanize/vulcanizedb/pkg/core"
|
||||
"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
|
||||
)
|
||||
|
||||
type WatchedEventRepository struct {
|
||||
*postgres.DB
|
||||
}
|
||||
|
||||
func (watchedEventRepository WatchedEventRepository) GetWatchedEvents(name string) ([]*core.WatchedEvent, error) {
|
||||
rows, err := watchedEventRepository.DB.Queryx(`SELECT id, name, block_number, address, tx_hash, index, topic0, topic1, topic2, topic3, data FROM watched_event_logs where name=$1`, name)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
lgs := make([]*core.WatchedEvent, 0)
|
||||
for rows.Next() {
|
||||
lg := new(core.WatchedEvent)
|
||||
err = rows.StructScan(lg)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
lgs = append(lgs, lg)
|
||||
}
|
||||
if err = rows.Err(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return lgs, nil
|
||||
}
|
||||
@@ -0,0 +1,127 @@
|
||||
package repositories_test
|
||||
|
||||
import (
|
||||
. "github.com/onsi/ginkgo"
|
||||
. "github.com/onsi/gomega"
|
||||
"github.com/vulcanize/vulcanizedb/pkg/core"
|
||||
"github.com/vulcanize/vulcanizedb/pkg/datastore"
|
||||
"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
|
||||
"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres/repositories"
|
||||
"github.com/vulcanize/vulcanizedb/pkg/filters"
|
||||
"github.com/vulcanize/vulcanizedb/test_config"
|
||||
)
|
||||
|
||||
var _ = Describe("Watched Events Repository", func() {
|
||||
var db *postgres.DB
|
||||
var logRepository datastore.LogRepository
|
||||
var filterRepository datastore.FilterRepository
|
||||
var watchedEventRepository datastore.WatchedEventRepository
|
||||
|
||||
BeforeEach(func() {
|
||||
db = test_config.NewTestDB(core.Node{})
|
||||
logRepository = repositories.LogRepository{DB: db}
|
||||
filterRepository = repositories.FilterRepository{DB: db}
|
||||
watchedEventRepository = repositories.WatchedEventRepository{DB: db}
|
||||
})
|
||||
|
||||
It("retrieves watched event logs that match the event filter", func() {
|
||||
filter := filters.LogFilter{
|
||||
Name: "Filter1",
|
||||
FromBlock: 0,
|
||||
ToBlock: 10,
|
||||
Address: "0x123",
|
||||
Topics: core.Topics{0: "event1=10", 2: "event3=hello"},
|
||||
}
|
||||
logs := []core.Log{
|
||||
{
|
||||
BlockNumber: 0,
|
||||
TxHash: "0x1",
|
||||
Address: "0x123",
|
||||
Topics: core.Topics{0: "event1=10", 2: "event3=hello"},
|
||||
Index: 0,
|
||||
Data: "",
|
||||
},
|
||||
}
|
||||
expectedWatchedEventLog := []*core.WatchedEvent{
|
||||
{
|
||||
Name: "Filter1",
|
||||
BlockNumber: 0,
|
||||
TxHash: "0x1",
|
||||
Address: "0x123",
|
||||
Topic0: "event1=10",
|
||||
Topic2: "event3=hello",
|
||||
Index: 0,
|
||||
Data: "",
|
||||
},
|
||||
}
|
||||
err := filterRepository.CreateFilter(filter)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
err = logRepository.CreateLogs(logs)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
matchingLogs, err := watchedEventRepository.GetWatchedEvents("Filter1")
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Expect(len(matchingLogs)).To(Equal(1))
|
||||
Expect(matchingLogs[0].Name).To(Equal(expectedWatchedEventLog[0].Name))
|
||||
Expect(matchingLogs[0].BlockNumber).To(Equal(expectedWatchedEventLog[0].BlockNumber))
|
||||
Expect(matchingLogs[0].TxHash).To(Equal(expectedWatchedEventLog[0].TxHash))
|
||||
Expect(matchingLogs[0].Address).To(Equal(expectedWatchedEventLog[0].Address))
|
||||
Expect(matchingLogs[0].Topic0).To(Equal(expectedWatchedEventLog[0].Topic0))
|
||||
Expect(matchingLogs[0].Topic1).To(Equal(expectedWatchedEventLog[0].Topic1))
|
||||
Expect(matchingLogs[0].Topic2).To(Equal(expectedWatchedEventLog[0].Topic2))
|
||||
Expect(matchingLogs[0].Data).To(Equal(expectedWatchedEventLog[0].Data))
|
||||
|
||||
})
|
||||
|
||||
It("retrieves a watched event log by name", func() {
|
||||
filter := filters.LogFilter{
|
||||
Name: "Filter1",
|
||||
FromBlock: 0,
|
||||
ToBlock: 10,
|
||||
Address: "0x123",
|
||||
Topics: core.Topics{0: "event1=10", 2: "event3=hello"},
|
||||
}
|
||||
logs := []core.Log{
|
||||
{
|
||||
BlockNumber: 0,
|
||||
TxHash: "0x1",
|
||||
Address: "0x123",
|
||||
Topics: core.Topics{0: "event1=10", 2: "event3=hello"},
|
||||
Index: 0,
|
||||
Data: "",
|
||||
},
|
||||
{
|
||||
BlockNumber: 100,
|
||||
TxHash: "",
|
||||
Address: "",
|
||||
Topics: core.Topics{},
|
||||
Index: 0,
|
||||
Data: "",
|
||||
},
|
||||
}
|
||||
expectedWatchedEventLog := []*core.WatchedEvent{{
|
||||
Name: "Filter1",
|
||||
BlockNumber: 0,
|
||||
TxHash: "0x1",
|
||||
Address: "0x123",
|
||||
Topic0: "event1=10",
|
||||
Topic2: "event3=hello",
|
||||
Index: 0,
|
||||
Data: "",
|
||||
}}
|
||||
err := filterRepository.CreateFilter(filter)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
err = logRepository.CreateLogs(logs)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
matchingLogs, err := watchedEventRepository.GetWatchedEvents("Filter1")
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Expect(len(matchingLogs)).To(Equal(1))
|
||||
Expect(matchingLogs[0].Name).To(Equal(expectedWatchedEventLog[0].Name))
|
||||
Expect(matchingLogs[0].BlockNumber).To(Equal(expectedWatchedEventLog[0].BlockNumber))
|
||||
Expect(matchingLogs[0].TxHash).To(Equal(expectedWatchedEventLog[0].TxHash))
|
||||
Expect(matchingLogs[0].Address).To(Equal(expectedWatchedEventLog[0].Address))
|
||||
Expect(matchingLogs[0].Topic0).To(Equal(expectedWatchedEventLog[0].Topic0))
|
||||
Expect(matchingLogs[0].Topic1).To(Equal(expectedWatchedEventLog[0].Topic1))
|
||||
Expect(matchingLogs[0].Topic2).To(Equal(expectedWatchedEventLog[0].Topic2))
|
||||
Expect(matchingLogs[0].Data).To(Equal(expectedWatchedEventLog[0].Data))
|
||||
})
|
||||
})
|
||||
Reference in New Issue
Block a user