update direct pg publisher

pkg/snapshot/pg/publisher.go

@@ -17,26 +17,27 @@ package pg
 
 import (
 	"context"
-	"fmt"
 	"math/big"
 	"sync/atomic"
 	"time"
 
-	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/statediff/indexer/models"
-	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/lib/pq"
+
 	"github.com/ipfs/go-cid"
-	blockstore "github.com/ipfs/go-ipfs-blockstore"
-	dshelp "github.com/ipfs/go-ipfs-ds-help"
 	"github.com/multiformats/go-multihash"
 	"github.com/sirupsen/logrus"
 	log "github.com/sirupsen/logrus"
 
-	"github.com/cerc-io/ipld-eth-state-snapshot/pkg/prom"
+	"github.com/ethereum/go-ethereum/common"
-	snapt "github.com/cerc-io/ipld-eth-state-snapshot/pkg/types"
+	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/statediff/indexer/database/sql"
 	"github.com/ethereum/go-ethereum/statediff/indexer/database/sql/postgres"
 	"github.com/ethereum/go-ethereum/statediff/indexer/ipld"
-	"github.com/ethereum/go-ethereum/statediff/indexer/shared"
+	"github.com/ethereum/go-ethereum/statediff/indexer/shared/schema"
+
+	"github.com/cerc-io/ipld-eth-state-snapshot/pkg/prom"
+	snapt "github.com/cerc-io/ipld-eth-state-snapshot/pkg/types"
 )
 
 var _ snapt.Publisher = (*publisher)(nil)
@@ -101,10 +102,14 @@ func (tx pubTx) publishRaw(codec uint64, raw []byte, height *big.Int) (cid, pref
 }
 
 func (tx pubTx) publishIPLD(c cid.Cid, raw []byte, height *big.Int) (string, error) {
-	dbKey := dshelp.MultihashToDsKey(c.Hash())
+	_, err := tx.Exec(schema.TableIPLDBlock.ToInsertStatement(false), height.Uint64(), c.String(), raw)
-	prefixedKey := blockstore.BlockPrefix.String() + dbKey.String()
+	return c.String(), err
-	_, err := tx.Exec(snapt.TableIPLDBlock.ToInsertStatement(), height.Uint64(), prefixedKey, raw)
+}
-	return prefixedKey, err
+
+// PublishIPLD writes an IPLD to the ipld.blocks blockstore
+func (p *publisher) PublishIPLD(c cid.Cid, raw []byte, height *big.Int, snapTx snapt.Tx) (string, error) {
+	tx := snapTx.(pubTx)
+	return tx.publishIPLD(c, raw, height)
 }
 
 // PublishHeader writes the header to the ipfs backing pg datastore and adds secondary indexes in the header_cids table
@@ -130,29 +135,27 @@ func (p *publisher) PublishHeader(header *types.Header) (err error) {
 		return err
 	}
 
-	mhKey := shared.MultihashKeyFromCID(headerNode.Cid())
+	_, err = tx.Exec(schema.TableHeader.ToInsertStatement(false), header.Number.Uint64(), header.Hash().Hex(),
-	_, err = tx.Exec(snapt.TableHeader.ToInsertStatement(), header.Number.Uint64(), header.Hash().Hex(),
+		header.ParentHash.Hex(), headerNode.Cid().String(), "0", pq.StringArray([]string{p.db.NodeID()}), "0",
-		header.ParentHash.Hex(), headerNode.Cid().String(), "0", p.db.NodeID(), "0",
 		header.Root.Hex(), header.TxHash.Hex(), header.ReceiptHash.Hex(), header.UncleHash.Hex(),
-		header.Bloom.Bytes(), header.Time, mhKey, 0, header.Coinbase.String())
+		header.Bloom.Bytes(), header.Time, header.Coinbase.String())
 	return err
 }
 
-// PublishStateNode writes the state node to the ipfs backing datastore and adds secondary indexes in the state_cids table
+// PublishStateLeafNode writes the state leaf node to eth.state_cids
-func (p *publisher) PublishStateNode(node *snapt.Node, headerID string, height *big.Int, snapTx snapt.Tx) error {
+func (p *publisher) PublishStateLeafNode(stateNode *models.StateNodeModel, snapTx snapt.Tx) error {
-	var stateKey string
-	if !snapt.IsNullHash(node.Key) {
-		stateKey = node.Key.Hex()
-	}
-
 	tx := snapTx.(pubTx)
-	stateCIDStr, mhKey, err := tx.publishRaw(ipld.MEthStateTrie, node.Value, height)
+	_, err := tx.Exec(schema.TableStateNode.ToInsertStatement(false),
-	if err != nil {
+		stateNode.BlockNumber,
-		return err
+		stateNode.HeaderID,
-	}
+		stateNode.StateKey,
-
+		stateNode.CID,
-	_, err = tx.Exec(snapt.TableStateNode.ToInsertStatement(),
+		true,
-		height.Uint64(), headerID, stateKey, stateCIDStr, node.Path, node.NodeType, false, mhKey)
+		stateNode.Balance,
+		stateNode.Nonce,
+		stateNode.CodeHash,
+		stateNode.StorageRoot,
+		stateNode.Removed)
 	if err != nil {
 		return err
 	}
@@ -165,21 +168,18 @@ func (p *publisher) PublishStateNode(node *snapt.Node, headerID string, height *
 	return err
 }
 
-// PublishStorageNode writes the storage node to the ipfs backing pg datastore and adds secondary indexes in the storage_cids table
+// PublishStorageLeafNode writes the storage leaf node to eth.storage_cids
-func (p *publisher) PublishStorageNode(node *snapt.Node, headerID string, height *big.Int, statePath []byte, snapTx snapt.Tx) error {
+func (p *publisher) PublishStorageLeafNode(storageNode *models.StorageNodeModel, snapTx snapt.Tx) error {
-	var storageKey string
-	if !snapt.IsNullHash(node.Key) {
-		storageKey = node.Key.Hex()
-	}
-
 	tx := snapTx.(pubTx)
-	storageCIDStr, mhKey, err := tx.publishRaw(ipld.MEthStorageTrie, node.Value, height)
+	_, err := tx.Exec(schema.TableStorageNode.ToInsertStatement(false),
-	if err != nil {
+		storageNode.BlockNumber,
-		return err
+		storageNode.HeaderID,
-	}
+		storageNode.StateKey,
-
+		storageNode.StorageKey,
-	_, err = tx.Exec(snapt.TableStorageNode.ToInsertStatement(),
+		storageNode.CID,
-		height.Uint64(), headerID, statePath, storageKey, storageCIDStr, node.Path, node.NodeType, false, mhKey)
+		true,
+		storageNode.Value,
+		storageNode.Removed)
 	if err != nil {
 		return err
 	}
@@ -194,15 +194,11 @@ func (p *publisher) PublishStorageNode(node *snapt.Node, headerID string, height
 
 // PublishCode writes code to the ipfs backing pg datastore
 func (p *publisher) PublishCode(height *big.Int, codeHash common.Hash, codeBytes []byte, snapTx snapt.Tx) error {
-	// no codec for code, doesn't matter though since blockstore key is multihash-derived
+	c := ipld.Keccak256ToCid(ipld.RawBinary, codeHash.Bytes())
-	mhKey, err := shared.MultihashKeyFromKeccak256(codeHash)
-	if err != nil {
-		return fmt.Errorf("error deriving multihash key from codehash: %v", err)
-	}
 
 	tx := snapTx.(pubTx)
-	if _, err = tx.Exec(snapt.TableIPLDBlock.ToInsertStatement(), height.Uint64(), mhKey, codeBytes); err != nil {
+	if _, err := tx.publishIPLD(c, codeBytes, height); err != nil {
-		return fmt.Errorf("error publishing code IPLD: %v", err)
+		return err
 	}
 
 	// increment code node counter.