combined/directp publish and indexing for btc; tests

2020-04-30 17:22:22 -05:00
parent 65da8cafe7
commit 059664ffc9
11 changed files with 306 additions and 133 deletions
@@ -47,7 +47,7 @@ func (in *CIDIndexer) Index(cids shared.CIDsForIndexing) error {
 	if err != nil {
 		return err
 	}
-	headerID, err := in.indexHeaderCID(tx, cidWrapper.HeaderCID, in.db.NodeID)
+	headerID, err := in.indexHeaderCID(tx, cidWrapper.HeaderCID)
 	if err != nil {
 		logrus.Error("btc indexer error when indexing header")
 		return err
@@ -59,13 +59,13 @@ func (in *CIDIndexer) Index(cids shared.CIDsForIndexing) error {
 	return tx.Commit()
 }

-func (in *CIDIndexer) indexHeaderCID(tx *sqlx.Tx, header HeaderModel, nodeID int64) (int64, error) {
+func (in *CIDIndexer) indexHeaderCID(tx *sqlx.Tx, header HeaderModel) (int64, error) {
 	var headerID int64
 	err := tx.QueryRowx(`INSERT INTO btc.header_cids (block_number, block_hash, parent_hash, cid, timestamp, bits, node_id, times_validated)
 							VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
 							ON CONFLICT (block_number, block_hash) DO UPDATE SET (parent_hash, cid, timestamp, bits, node_id, times_validated) = ($3, $4, $5, $6, $7, btc.header_cids.times_validated + 1)
 							RETURNING id`,
-		header.BlockNumber, header.BlockHash, header.ParentHash, header.CID, header.Timestamp, header.Bits, nodeID, 1).Scan(&headerID)
+		header.BlockNumber, header.BlockHash, header.ParentHash, header.CID, header.Timestamp, header.Bits, in.db.NodeID, 1).Scan(&headerID)
 	return headerID, err
 }

@@ -0,0 +1,121 @@
+// VulcanizeDB
+// Copyright © 2019 Vulcanize
+
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+
+// You should have received a copy of the GNU Affero General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+package btc
+
+import (
+	"fmt"
+	"strconv"
+
+	"github.com/vulcanize/vulcanizedb/pkg/ipfs/ipld"
+	"github.com/vulcanize/vulcanizedb/pkg/postgres"
+	"github.com/vulcanize/vulcanizedb/pkg/super_node/shared"
+)
+
+// IPLDPublisherAndIndexer satisfies the IPLDPublisher interface for bitcoin
+// It interfaces directly with the public.blocks table of PG-IPFS rather than going through an ipfs intermediary
+// It publishes and indexes IPLDs together in a single sqlx.Tx
+type IPLDPublisherAndIndexer struct {
+	indexer *CIDIndexer
+}
+
+// NewIPLDPublisherAndIndexer creates a pointer to a new IPLDPublisherAndIndexer which satisfies the IPLDPublisher interface
+func NewIPLDPublisherAndIndexer(db *postgres.DB) *IPLDPublisherAndIndexer {
+	return &IPLDPublisherAndIndexer{
+		indexer: NewCIDIndexer(db),
+	}
+}
+
+// Publish publishes an IPLDPayload to IPFS and returns the corresponding CIDPayload
+func (pub *IPLDPublisherAndIndexer) Publish(payload shared.ConvertedData) (shared.CIDsForIndexing, error) {
+	ipldPayload, ok := payload.(ConvertedPayload)
+	if !ok {
+		return nil, fmt.Errorf("btc publisher expected payload type %T got %T", ConvertedPayload{}, payload)
+	}
+	// Generate the iplds
+	headerNode, txNodes, txTrieNodes, err := ipld.FromHeaderAndTxs(ipldPayload.Header, ipldPayload.Txs)
+	if err != nil {
+		return nil, err
+	}
+
+	// Begin new db tx
+	tx, err := pub.indexer.db.Beginx()
+	if err != nil {
+		return nil, err
+	}
+
+	// Publish trie nodes
+	for _, node := range txTrieNodes {
+		if err := shared.PublishIPLD(tx, node); err != nil {
+			shared.Rollback(tx)
+			return nil, err
+		}
+	}
+
+	// Publish and index header
+	if err := shared.PublishIPLD(tx, headerNode); err != nil {
+		shared.Rollback(tx)
+		return nil, err
+	}
+	header := HeaderModel{
+		CID:         headerNode.Cid().String(),
+		ParentHash:  ipldPayload.Header.PrevBlock.String(),
+		BlockNumber: strconv.Itoa(int(ipldPayload.BlockPayload.BlockHeight)),
+		BlockHash:   ipldPayload.Header.BlockHash().String(),
+		Timestamp:   ipldPayload.Header.Timestamp.UnixNano(),
+		Bits:        ipldPayload.Header.Bits,
+	}
+	headerID, err := pub.indexer.indexHeaderCID(tx, header)
+	if err != nil {
+		shared.Rollback(tx)
+		return nil, err
+	}
+
+	// Publish and index txs
+	for i, txNode := range txNodes {
+		if err := shared.PublishIPLD(tx, txNode); err != nil {
+			shared.Rollback(tx)
+			return nil, err
+		}
+		txModel := ipldPayload.TxMetaData[i]
+		txModel.CID = txNode.Cid().String()
+		txID, err := pub.indexer.indexTransactionCID(tx, txModel, headerID)
+		if err != nil {
+			shared.Rollback(tx)
+			return nil, err
+		}
+		for _, input := range txModel.TxInputs {
+			if err := pub.indexer.indexTxInput(tx, input, txID); err != nil {
+				shared.Rollback(tx)
+				return nil, err
+			}
+		}
+		for _, output := range txModel.TxOutputs {
+			if err := pub.indexer.indexTxOutput(tx, output, txID); err != nil {
+				shared.Rollback(tx)
+				return nil, err
+			}
+		}
+	}
+
+	// This IPLDPublisher does both publishing and indexing, we do not need to pass anything forward to the indexer
+	return nil, tx.Commit()
+}
+
+// Index satisfies the shared.CIDIndexer interface
+func (pub *IPLDPublisherAndIndexer) Index(cids shared.CIDsForIndexing) error {
+	return nil
+}
@@ -0,0 +1,121 @@
+// VulcanizeDB
+// Copyright © 2019 Vulcanize
+
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+
+// You should have received a copy of the GNU Affero General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+package btc_test
+
+import (
+	"bytes"
+
+	"github.com/ipfs/go-cid"
+	"github.com/ipfs/go-ipfs-blockstore"
+	"github.com/ipfs/go-ipfs-ds-help"
+	"github.com/multiformats/go-multihash"
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+	"github.com/vulcanize/vulcanizedb/pkg/ipfs/ipld"
+
+	"github.com/vulcanize/vulcanizedb/pkg/postgres"
+	"github.com/vulcanize/vulcanizedb/pkg/super_node/btc"
+	"github.com/vulcanize/vulcanizedb/pkg/super_node/btc/mocks"
+	"github.com/vulcanize/vulcanizedb/pkg/super_node/shared"
+)
+
+var _ = Describe("PublishAndIndexer", func() {
+	var (
+		db        *postgres.DB
+		err       error
+		repo      *btc.IPLDPublisherAndIndexer
+		ipfsPgGet = `SELECT data FROM public.blocks
+					WHERE key = $1`
+	)
+	BeforeEach(func() {
+		db, err = shared.SetupDB()
+		Expect(err).ToNot(HaveOccurred())
+		repo = btc.NewIPLDPublisherAndIndexer(db)
+	})
+	AfterEach(func() {
+		btc.TearDownDB(db)
+	})
+
+	Describe("Publish", func() {
+		It("Published and indexes header and transaction IPLDs in a single tx", func() {
+			emptyReturn, err := repo.Publish(mocks.MockConvertedPayload)
+			Expect(emptyReturn).To(BeNil())
+			Expect(err).ToNot(HaveOccurred())
+			pgStr := `SELECT * FROM btc.header_cids
+				WHERE block_number = $1`
+			// check header was properly indexed
+			buf := bytes.NewBuffer(make([]byte, 0, 80))
+			err = mocks.MockBlock.Header.Serialize(buf)
+			Expect(err).ToNot(HaveOccurred())
+			headerBytes := buf.Bytes()
+			c, _ := ipld.RawdataToCid(ipld.MBitcoinHeader, headerBytes, multihash.DBL_SHA2_256)
+			header := new(btc.HeaderModel)
+			err = db.Get(header, pgStr, mocks.MockHeaderMetaData.BlockNumber)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(header.CID).To(Equal(c.String()))
+			Expect(header.BlockNumber).To(Equal(mocks.MockHeaderMetaData.BlockNumber))
+			Expect(header.Bits).To(Equal(mocks.MockHeaderMetaData.Bits))
+			Expect(header.Timestamp).To(Equal(mocks.MockHeaderMetaData.Timestamp))
+			Expect(header.BlockHash).To(Equal(mocks.MockHeaderMetaData.BlockHash))
+			Expect(header.ParentHash).To(Equal(mocks.MockHeaderMetaData.ParentHash))
+			dc, err := cid.Decode(header.CID)
+			Expect(err).ToNot(HaveOccurred())
+			mhKey := dshelp.CidToDsKey(dc)
+			prefixedKey := blockstore.BlockPrefix.String() + mhKey.String()
+			var data []byte
+			err = db.Get(&data, ipfsPgGet, prefixedKey)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(data).To(Equal(headerBytes))
+
+			// check that txs were properly indexed
+			trxs := make([]btc.TxModel, 0)
+			pgStr = `SELECT transaction_cids.id, transaction_cids.header_id, transaction_cids.index,
+				transaction_cids.tx_hash, transaction_cids.cid, transaction_cids.segwit, transaction_cids.witness_hash
+				FROM btc.transaction_cids INNER JOIN btc.header_cids ON (transaction_cids.header_id = header_cids.id)
+				WHERE header_cids.block_number = $1`
+			err = db.Select(&trxs, pgStr, mocks.MockHeaderMetaData.BlockNumber)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(len(trxs)).To(Equal(3))
+			txData := make([][]byte, len(mocks.MockTransactions))
+			txCIDs := make([]string, len(mocks.MockTransactions))
+			for i, m := range mocks.MockTransactions {
+				buf := bytes.NewBuffer(make([]byte, 0))
+				err = m.MsgTx().Serialize(buf)
+				Expect(err).ToNot(HaveOccurred())
+				tx := buf.Bytes()
+				txData[i] = tx
+				c, _ := ipld.RawdataToCid(ipld.MBitcoinTx, tx, multihash.DBL_SHA2_256)
+				txCIDs[i] = c.String()
+			}
+			for _, tx := range trxs {
+				Expect(tx.SegWit).To(Equal(false))
+				Expect(tx.HeaderID).To(Equal(header.ID))
+				Expect(tx.WitnessHash).To(Equal(""))
+				Expect(tx.CID).To(Equal(txCIDs[tx.Index]))
+				Expect(tx.TxHash).To(Equal(mocks.MockBlock.Transactions[tx.Index].TxHash().String()))
+				dc, err := cid.Decode(tx.CID)
+				Expect(err).ToNot(HaveOccurred())
+				mhKey := dshelp.CidToDsKey(dc)
+				prefixedKey := blockstore.BlockPrefix.String() + mhKey.String()
+				var data []byte
+				err = db.Get(&data, ipfsPgGet, prefixedKey)
+				Expect(err).ToNot(HaveOccurred())
+				Expect(data).To(Equal(txData[tx.Index]))
+			}
+		})
+	})
+})