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

Use GORM to retrieve data from database

Browse Source
This commit is contained in:
nabarun
2022-06-03 15:08:45 +05:30
parent b3d9e01d67
commit 6bd563e3d5
7 changed files with 138 additions and 302 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/eth/cid_retriever.go
+84 -133
View File
@@ -26,7 +26,8 @@ import (
"github.com/jmoiron/sqlx"
"github.com/lib/pq"
log "github.com/sirupsen/logrus"
"github.com/thoas/go-funk"
"gorm.io/driver/postgres"
"gorm.io/gorm"
"github.com/vulcanize/ipld-eth-server/v3/pkg/shared"
)
@@ -40,13 +41,70 @@ type Retriever interface {
// CIDRetriever satisfies the CIDRetriever interface for ethereum
type CIDRetriever struct {
db *sqlx.DB
db *sqlx.DB
gormDB *gorm.DB
}
type IPLD struct {
models.IPLDModel
}
// TableName overrides the table name used by IPLD
func (IPLD) TableName() string {
return "public.blocks"
}
type HeaderCid struct {
CID string `gorm:"column:cid"`
BlockHash string `gorm:"primaryKey"`
BlockNumber string
ParentHash string
Timestamp uint64
StateRoot string
TotalDifficulty string `gorm:"column:td"`
TxRoot string
RctRoot string `gorm:"column:receipt_root"`
UncleRoot string
Bloom []byte
MhKey string
TransactionCids []TransactionCid `gorm:"foreignKey:HeaderID;references:BlockHash"`
IPLD IPLD `gorm:"foreignKey:MhKey;references:Key"`
}
// TableName overrides the table name used by HeaderCid
func (HeaderCid) TableName() string {
return "eth.header_cids"
}
type TransactionCid struct {
CID string `gorm:"column:cid"`
TxHash string `gorm:"primaryKey"`
HeaderID string `gorm:"column:header_id"`
Index int64
Src string
Dst string
MhKey string
IPLD IPLD `gorm:"foreignKey:MhKey;references:Key"`
}
// TableName overrides the table name used by TransactionCid
func (TransactionCid) TableName() string {
return "eth.transaction_cids"
}
// NewCIDRetriever returns a pointer to a new CIDRetriever which supports the CIDRetriever interface
func NewCIDRetriever(db *sqlx.DB) *CIDRetriever {
gormDB, err := gorm.Open(postgres.New(postgres.Config{
Conn: db,
}), &gorm.Config{})
if err != nil {
log.Error(err)
return nil
}
return &CIDRetriever{
db: db,
db: db,
gormDB: gormDB,
}
}
@@ -594,29 +652,6 @@ func (ecr *CIDRetriever) RetrieveTxCIDsByHeaderID(tx *sqlx.Tx, headerID string)
return txCIDs, tx.Select(&txCIDs, pgStr, headerID)
}
// RetrieveTxCIDsByHeaderIDs retrieves all tx CIDs for the given headerIDs
func (ecr *CIDRetriever) RetrieveTxCIDsByHeaderIDs(tx *sqlx.Tx, headerIDs []string) ([]models.TxModel, error) {
log.Debug("retrieving tx cids for headerIDs ", headerIDs)
pgStr := `SELECT header_id, index, tx_hash, cid, mh_key,
dst, src, tx_data, tx_type, value
FROM eth.transaction_cids
WHERE header_id in (?)
ORDER BY index`
var txCIDs []models.TxModel
if len(headerIDs) < 1 {
return txCIDs, nil
}
query, args, err := sqlx.In(pgStr, headerIDs)
if err != nil {
return txCIDs, err
}
query = tx.Rebind(query)
return txCIDs, tx.Select(&txCIDs, query, args...)
}
// RetrieveReceiptCIDsByTxIDs retrieves receipt CIDs by their associated tx IDs
func (ecr *CIDRetriever) RetrieveReceiptCIDsByTxIDs(tx *sqlx.Tx, txHashes []string) ([]models.ReceiptModel, error) {
log.Debugf("retrieving receipt cids for tx hashes %v", txHashes)
@@ -631,134 +666,50 @@ func (ecr *CIDRetriever) RetrieveReceiptCIDsByTxIDs(tx *sqlx.Tx, txHashes []stri
}
// RetrieveHeaderAndTxCIDsByBlockNumber retrieves header CIDs and their associated tx CIDs by block number
func (ecr *CIDRetriever) RetrieveHeaderAndTxCIDsByBlockNumber(blockNumber int64) ([]models.HeaderModel, [][]models.TxModel, error) {
func (ecr *CIDRetriever) RetrieveHeaderAndTxCIDsByBlockNumber(blockNumber int64) ([]HeaderCid, error) {
log.Debug("retrieving header cids and tx cids for block number ", blockNumber)
// Begin new db tx
tx, err := ecr.db.Beginx()
if err != nil {
return nil, nil, err
}
defer func() {
if p := recover(); p != nil {
shared.Rollback(tx)
panic(p)
} else if err != nil {
shared.Rollback(tx)
} else {
err = tx.Commit()
}
}()
var headerCIDs []HeaderCid
var headerCIDs []models.HeaderModel
headerCIDs, err = ecr.RetrieveHeaderCIDs(tx, blockNumber)
// https://github.com/go-gorm/gorm/issues/4083#issuecomment-778883283
// Will use join for TransactionCids once preload for 1:N is supported.
err := ecr.gormDB.Preload("TransactionCids").Joins("IPLD").Find(&headerCIDs, "block_number = ?", blockNumber).Error
if err != nil {
log.Error("header cid retrieval error")
return nil, nil, err
return nil, err
}
var allTxCIDs [][]models.TxModel
txHeaderIDs := funk.Map(
headerCIDs,
func(headerCID models.HeaderModel) string {
return headerCID.BlockHash
},
)
txCIDs, err := ecr.RetrieveTxCIDsByHeaderIDs(tx, txHeaderIDs.([]string))
if err != nil {
log.Error("tx cid retrieval error")
return nil, nil, err
}
txCIDsByHeaderID := funk.Reduce(
txCIDs,
func(acc map[string][]models.TxModel, txCID models.TxModel) map[string][]models.TxModel {
if _, ok := acc[txCID.HeaderID]; !ok {
acc[txCID.HeaderID] = []models.TxModel{}
}
txCIDs = append(acc[txCID.HeaderID], txCID)
acc[txCID.HeaderID] = txCIDs
return acc
},
make(map[string][]models.TxModel),
)
txCIDsByHeaderIDMap := txCIDsByHeaderID.(map[string][]models.TxModel)
for _, headerCID := range headerCIDs {
txCIDs := txCIDsByHeaderIDMap[headerCID.BlockHash]
allTxCIDs = append(allTxCIDs, txCIDs)
}
return headerCIDs, allTxCIDs, nil
return headerCIDs, nil
}
// RetrieveHeaderAndTxCIDsByBlockHash retrieves header CID and their associated tx CIDs by block hash
func (ecr *CIDRetriever) RetrieveHeaderAndTxCIDsByBlockHash(blockHash common.Hash) (models.HeaderModel, []models.TxModel, error) {
func (ecr *CIDRetriever) RetrieveHeaderAndTxCIDsByBlockHash(blockHash common.Hash) (HeaderCid, error) {
log.Debug("retrieving header cid and tx cids for block hash ", blockHash.String())
// Begin new db tx
tx, err := ecr.db.Beginx()
if err != nil {
return models.HeaderModel{}, nil, err
}
defer func() {
if p := recover(); p != nil {
shared.Rollback(tx)
panic(p)
} else if err != nil {
shared.Rollback(tx)
} else {
err = tx.Commit()
}
}()
var headerCID HeaderCid
var headerCID models.HeaderModel
headerCID, err = ecr.RetrieveHeaderCIDByHash(tx, blockHash)
// https://github.com/go-gorm/gorm/issues/4083#issuecomment-778883283
// Will use join for TransactionCids once preload for 1:N is supported.
err := ecr.gormDB.Preload("TransactionCids").Joins("IPLD").First(&headerCID, "block_hash = ?", blockHash.String()).Error
if err != nil {
log.Error("header cid retrieval error")
return models.HeaderModel{}, nil, err
}
if err != nil {
return models.HeaderModel{}, nil, err
return headerCID, err
}
var txCIDs []models.TxModel
txCIDs, err = ecr.RetrieveTxCIDsByHeaderID(tx, headerCID.BlockHash)
if err != nil {
log.Error("tx cid retrieval error")
return models.HeaderModel{}, nil, err
}
return headerCID, txCIDs, nil
return headerCID, nil
}
// RetrieveTxCIDByHash returns the tx for the given tx hash
func (ecr *CIDRetriever) RetrieveTxCIDByHash(txHash string) (models.TxModel, error) {
func (ecr *CIDRetriever) RetrieveTxCIDByHash(txHash string) (TransactionCid, error) {
log.Debug("retrieving tx cid for tx hash ", txHash)
// Begin new db tx
tx, err := ecr.db.Beginx()
if err != nil {
return models.TxModel{}, err
}
defer func() {
if p := recover(); p != nil {
shared.Rollback(tx)
panic(p)
} else if err != nil {
shared.Rollback(tx)
} else {
err = tx.Commit()
}
}()
var txCID TransactionCid
pgStr := `SELECT header_id, index, tx_hash, cid, mh_key,
dst, src, tx_data, tx_type, value
FROM eth.transaction_cids
WHERE tx_hash = $1
ORDER BY index`
var txCID models.TxModel
return txCID, tx.Get(&txCID, pgStr, txHash)
err := ecr.gormDB.Joins("IPLD").First(&txCID, "tx_hash = ?", txHash).Error
if err != nil {
log.Error("header cid retrieval error")
return txCID, err
}
return txCID, nil
}
pkg/eth/ipld_fetcher.go
-35
View File
@@ -25,7 +25,6 @@ import (
"github.com/ethereum/go-ethereum/statediff/indexer/models"
"github.com/jmoiron/sqlx"
log "github.com/sirupsen/logrus"
"github.com/thoas/go-funk"
"github.com/vulcanize/ipld-eth-server/v3/pkg/shared"
)
@@ -113,40 +112,6 @@ func (f *IPLDFetcher) FetchHeader(tx *sqlx.Tx, c models.HeaderModel) (models.IPL
}, nil
}
// FetchHeaders fetches headers
func (f *IPLDFetcher) FetchHeaders(tx *sqlx.Tx, cids []models.HeaderModel) ([]models.IPLDModel, error) {
log.Debug("fetching header iplds")
headerIPLDs := make([]models.IPLDModel, len(cids))
if len(cids) < 1 {
return headerIPLDs, nil
}
mhKeys := make([]string, len(cids))
for i, c := range cids {
var err error
mhKeys[i] = c.MhKey
if err != nil {
return nil, err
}
}
fetchedIPLDs, err := shared.FetchIPLDsByMhKeys(tx, mhKeys)
if err != nil {
return nil, err
}
for i, c := range cids {
headerIPLD := funk.Find(fetchedIPLDs, func(ipld models.IPLDModel) bool {
return ipld.Key == c.MhKey
}).(models.IPLDModel)
headerIPLDs[i] = headerIPLD
}
return headerIPLDs, nil
}
// FetchUncles fetches uncles
func (f *IPLDFetcher) FetchUncles(tx *sqlx.Tx, cids []models.UncleModel) ([]models.IPLDModel, error) {
log.Debug("fetching uncle iplds")
pkg/graphql/graphql.go
+11 -42
View File
@@ -33,7 +33,6 @@ import (
"github.com/ethereum/go-ethereum/eth/filters"
"github.com/ethereum/go-ethereum/rlp"
"github.com/ethereum/go-ethereum/rpc"
"github.com/ethereum/go-ethereum/statediff/indexer/models"
"github.com/vulcanize/ipld-eth-server/v3/pkg/eth"
"github.com/vulcanize/ipld-eth-server/v3/pkg/shared"
@@ -1264,19 +1263,17 @@ type EthHeaderCidCondition struct {
func (r *Resolver) AllEthHeaderCids(ctx context.Context, args struct {
Condition *EthHeaderCidCondition
}) (*EthHeaderCidsConnection, error) {
var headerCIDs []models.HeaderModel
var allTxCIDs [][]models.TxModel
var headerCIDs []eth.HeaderCid
var err error
if args.Condition.BlockHash != nil {
headerCID, txCIDs, err := r.backend.Retriever.RetrieveHeaderAndTxCIDsByBlockHash(common.HexToHash(*args.Condition.BlockHash))
headerCID, err := r.backend.Retriever.RetrieveHeaderAndTxCIDsByBlockHash(common.HexToHash(*args.Condition.BlockHash))
if err != nil {
return nil, err
}
headerCIDs = append(headerCIDs, headerCID)
allTxCIDs = append(allTxCIDs, txCIDs)
} else if args.Condition.BlockNumber != nil {
headerCIDs, allTxCIDs, err = r.backend.Retriever.RetrieveHeaderAndTxCIDsByBlockNumber(args.Condition.BlockNumber.ToInt().Int64())
headerCIDs, err = r.backend.Retriever.RetrieveHeaderAndTxCIDsByBlockNumber(args.Condition.BlockNumber.ToInt().Int64())
if err != nil {
return nil, err
}
@@ -1300,13 +1297,8 @@ func (r *Resolver) AllEthHeaderCids(ctx context.Context, args struct {
}
}()
headerIPLDs, err := r.backend.Fetcher.FetchHeaders(tx, headerCIDs)
if err != nil {
return nil, err
}
var resultNodes []*EthHeaderCid
for idx, headerCID := range headerCIDs {
for _, headerCID := range headerCIDs {
var blockNumber BigInt
blockNumber.UnmarshalText([]byte(headerCID.BlockNumber))
@@ -1328,10 +1320,13 @@ func (r *Resolver) AllEthHeaderCids(ctx context.Context, args struct {
receiptRoot: headerCID.RctRoot,
uncleRoot: headerCID.UncleRoot,
bloom: Bytes(headerCID.Bloom).String(),
ipfsBlock: IPFSBlock{
key: headerCID.IPLD.Key,
data: Bytes(headerCID.IPLD.Data).String(),
},
}
txCIDs := allTxCIDs[idx]
for _, txCID := range txCIDs {
for _, txCID := range headerCID.TransactionCids {
ethHeaderCidNode.transactions = append(ethHeaderCidNode.transactions, &EthTransactionCid{
cid: txCID.CID,
txHash: txCID.TxHash,
@@ -1341,11 +1336,6 @@ func (r *Resolver) AllEthHeaderCids(ctx context.Context, args struct {
})
}
ethHeaderCidNode.ipfsBlock = IPFSBlock{
key: headerIPLDs[idx].Key,
data: Bytes(headerIPLDs[idx].Data).String(),
}
resultNodes = append(resultNodes, &ethHeaderCidNode)
}
@@ -1363,27 +1353,6 @@ func (r *Resolver) EthTransactionCidByTxHash(ctx context.Context, args struct {
return nil, err
}
// Begin tx
tx, err := r.backend.DB.Beginx()
if err != nil {
return nil, err
}
defer func() {
if p := recover(); p != nil {
shared.Rollback(tx)
panic(p)
} else if err != nil {
shared.Rollback(tx)
} else {
err = tx.Commit()
}
}()
txIPLDs, err := r.backend.Fetcher.FetchTrxs(tx, []models.TxModel{txCID})
if err != nil {
return nil, err
}
return &EthTransactionCid{
cid: txCID.CID,
txHash: txCID.TxHash,
@@ -1391,8 +1360,8 @@ func (r *Resolver) EthTransactionCidByTxHash(ctx context.Context, args struct {
src: txCID.Src,
dst: txCID.Dst,
ipfsBlock: IPFSBlock{
key: txIPLDs[0].Key,
data: Bytes(txIPLDs[0].Data).String(),
key: txCID.IPLD.Key,
data: Bytes(txCID.IPLD.Data).String(),
},
}, nil
}
pkg/graphql/graphql_test.go
+10 -63
View File
@@ -31,7 +31,6 @@ import (
"github.com/ethereum/go-ethereum/params"
"github.com/ethereum/go-ethereum/rpc"
"github.com/ethereum/go-ethereum/statediff"
"github.com/ethereum/go-ethereum/statediff/indexer/models"
sdtypes "github.com/ethereum/go-ethereum/statediff/types"
"github.com/jmoiron/sqlx"
. "github.com/onsi/ginkgo"
@@ -258,30 +257,13 @@ var _ = Describe("GraphQL", func() {
allEthHeaderCidsResp, err := client.AllEthHeaderCids(ctx, graphql.EthHeaderCidCondition{BlockNumber: new(graphql.BigInt).SetUint64(2)})
Expect(err).ToNot(HaveOccurred())
headerCIDs, txCIDs, err := backend.Retriever.RetrieveHeaderAndTxCIDsByBlockNumber(2)
Expect(err).ToNot(HaveOccurred())
// Begin tx
tx, err := backend.DB.Beginx()
Expect(err).ToNot(HaveOccurred())
defer func() {
if p := recover(); p != nil {
shared.Rollback(tx)
panic(p)
} else if err != nil {
shared.Rollback(tx)
} else {
err = tx.Commit()
}
}()
headerIPLDs, err := backend.Fetcher.FetchHeaders(tx, headerCIDs)
headerCIDs, err := backend.Retriever.RetrieveHeaderAndTxCIDsByBlockNumber(2)
Expect(err).ToNot(HaveOccurred())
for idx, headerCID := range headerCIDs {
ethHeaderCid := allEthHeaderCidsResp.Nodes[idx]
compareEthHeaderCid(ethHeaderCid, headerCID, txCIDs[idx], headerIPLDs[idx])
compareEthHeaderCid(ethHeaderCid, headerCID)
}
})
@@ -290,30 +272,12 @@ var _ = Describe("GraphQL", func() {
allEthHeaderCidsResp, err := client.AllEthHeaderCids(ctx, graphql.EthHeaderCidCondition{BlockHash: &blockHash})
Expect(err).ToNot(HaveOccurred())
headerCID, txCIDs, err := backend.Retriever.RetrieveHeaderAndTxCIDsByBlockHash(blocks[1].Hash())
headerCID, err := backend.Retriever.RetrieveHeaderAndTxCIDsByBlockHash(blocks[1].Hash())
Expect(err).ToNot(HaveOccurred())
// Begin tx
tx, err := backend.DB.Beginx()
Expect(err).ToNot(HaveOccurred())
defer func() {
if p := recover(); p != nil {
shared.Rollback(tx)
panic(p)
} else if err != nil {
shared.Rollback(tx)
} else {
err = tx.Commit()
}
}()
headerIPLDs, err := backend.Fetcher.FetchHeaders(tx, []models.HeaderModel{headerCID})
Expect(err).ToNot(HaveOccurred())
Expect(len(allEthHeaderCidsResp.Nodes)).To(Equal(1))
Expect(len(allEthHeaderCidsResp.Nodes)).To(Equal(1))
ethHeaderCid := allEthHeaderCidsResp.Nodes[0]
compareEthHeaderCid(ethHeaderCid, headerCID, txCIDs, headerIPLDs[0])
compareEthHeaderCid(ethHeaderCid, headerCID)
})
})
@@ -328,29 +292,12 @@ var _ = Describe("GraphQL", func() {
compareEthTxCid(*ethTransactionCidResp, txCID)
// Begin tx
tx, err := backend.DB.Beginx()
Expect(err).ToNot(HaveOccurred())
defer func() {
if p := recover(); p != nil {
shared.Rollback(tx)
panic(p)
} else if err != nil {
shared.Rollback(tx)
} else {
err = tx.Commit()
}
}()
txIPLDs, err := backend.Fetcher.FetchTrxs(tx, []models.TxModel{txCID})
Expect(err).ToNot(HaveOccurred())
Expect(len(txIPLDs)).To(Equal(1))
Expect(ethTransactionCidResp.BlockByMhKey.Data).To(Equal(graphql.Bytes(txIPLDs[0].Data).String()))
Expect(ethTransactionCidResp.BlockByMhKey.Data).To(Equal(graphql.Bytes(txCID.IPLD.Data).String()))
})
})
})
func compareEthHeaderCid(ethHeaderCid graphql.EthHeaderCidResp, headerCID models.HeaderModel, txCIDs []models.TxModel, headerIPLD models.IPLDModel) {
func compareEthHeaderCid(ethHeaderCid graphql.EthHeaderCidResp, headerCID eth.HeaderCid) {
blockNumber, err := strconv.ParseInt(headerCID.BlockNumber, 10, 64)
Expect(err).ToNot(HaveOccurred())
@@ -369,16 +316,16 @@ func compareEthHeaderCid(ethHeaderCid graphql.EthHeaderCidResp, headerCID models
Expect(ethHeaderCid.UncleRoot).To(Equal(headerCID.UncleRoot))
Expect(ethHeaderCid.Bloom).To(Equal(graphql.Bytes(headerCID.Bloom).String()))
for tIdx, txCID := range txCIDs {
for tIdx, txCID := range headerCID.TransactionCids {
ethTxCid := ethHeaderCid.EthTransactionCidsByHeaderId.Nodes[tIdx]
compareEthTxCid(ethTxCid, txCID)
}
Expect(ethHeaderCid.BlockByMhKey.Data).To(Equal(graphql.Bytes(headerIPLD.Data).String()))
Expect(ethHeaderCid.BlockByMhKey.Key).To(Equal(headerIPLD.Key))
Expect(ethHeaderCid.BlockByMhKey.Data).To(Equal(graphql.Bytes(headerCID.IPLD.Data).String()))
Expect(ethHeaderCid.BlockByMhKey.Key).To(Equal(headerCID.IPLD.Key))
}
func compareEthTxCid(ethTxCid graphql.EthTransactionCidResp, txCID models.TxModel) {
func compareEthTxCid(ethTxCid graphql.EthTransactionCidResp, txCID eth.TransactionCid) {
Expect(ethTxCid.Cid).To(Equal(txCID.CID))
Expect(ethTxCid.TxHash).To(Equal(txCID.TxHash))
Expect(ethTxCid.Index).To(Equal(int32(txCID.Index)))
pkg/shared/functions.go
-14
View File
@@ -19,7 +19,6 @@ package shared
import (
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/statediff/indexer/ipld"
"github.com/ethereum/go-ethereum/statediff/indexer/models"
"github.com/ipfs/go-cid"
blockstore "github.com/ipfs/go-ipfs-blockstore"
dshelp "github.com/ipfs/go-ipfs-ds-help"
@@ -78,19 +77,6 @@ func FetchIPLDByMhKey(tx *sqlx.Tx, mhKey string) ([]byte, error) {
return block, tx.Get(&block, pgStr, mhKey)
}
// FetchIPLDByMhKeysAndBlockNumbers is used to retrieve iplds from Postgres blockstore with the provided tx, mhkey strings and blockNumbers
func FetchIPLDsByMhKeys(tx *sqlx.Tx, mhKeys []string) ([]models.IPLDModel, error) {
var blocks []models.IPLDModel
pgStr := `SELECT key, data FROM public.blocks WHERE key IN (?)`
query, args, err := sqlx.In(pgStr, mhKeys)
if err != nil {
return blocks, err
}
query = tx.Rebind(query)
return blocks, tx.Select(&blocks, query, args...)
}
// MultihashKeyFromCID converts a cid into a blockstore-prefixed multihash db key string
func MultihashKeyFromCID(c cid.Cid) string {
dbKey := dshelp.MultihashToDsKey(c.Hash())