work on mocks and unit tests

pkg/ipfs/converter.go

@@ -18,7 +18,6 @@ package ipfs
 
 import (
 	"context"
-	"math/big"
 
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/core/types"
@@ -28,36 +27,25 @@ import (
 	"github.com/vulcanize/vulcanizedb/pkg/core"
 )
 
-// Converter interface is used to convert a geth statediff.Payload to our IPLDPayload type
-type Converter interface {
+// PayloadConverter interface is used to convert a geth statediff.Payload to our IPLDPayload type
+type PayloadConverter interface {
 	Convert(payload statediff.Payload) (*IPLDPayload, error)
 }
 
-// PayloadConverter is the underlying struct for the Converter interface
-type PayloadConverter struct {
+// Converter is the underlying struct for the PayloadConverter interface
+type Converter struct {
 	client core.EthClient
 }
 
-// IPLDPayload is a custom type which packages ETH data for the IPFS publisher
-type IPLDPayload struct {
-	HeaderRLP    []byte
-	BlockNumber  *big.Int
-	BlockHash    common.Hash
-	BlockBody    *types.Body
-	Receipts     types.Receipts
-	StateLeafs   map[common.Hash][]byte
-	StorageLeafs map[common.Hash]map[common.Hash][]byte
-}
-
-// NewPayloadConverter creates a pointer to a new PayloadConverter which satisfies the Converter interface
-func NewPayloadConverter(client core.EthClient) *PayloadConverter {
-	return &PayloadConverter{
+// NewPayloadConverter creates a pointer to a new Converter which satisfies the PayloadConverter interface
+func NewPayloadConverter(client core.EthClient) *Converter {
+	return &Converter{
 		client: client,
 	}
 }
 
 // Convert method is used to convert a geth statediff.Payload to a IPLDPayload
-func (pc *PayloadConverter) Convert(payload statediff.Payload) (*IPLDPayload, error) {
+func (pc *Converter) Convert(payload statediff.Payload) (*IPLDPayload, error) {
 	// Unpack block rlp to access fields
 	block := new(types.Block)
 	err := rlp.DecodeBytes(payload.BlockRlp, block)
@@ -66,21 +54,49 @@ func (pc *PayloadConverter) Convert(payload statediff.Payload) (*IPLDPayload, er
 	if err != nil {
 		return nil, err
 	}
+	trxLen := len(block.Transactions())
 	convertedPayload := &IPLDPayload{
-		BlockHash:    block.Hash(),
-		BlockNumber:  block.Number(),
-		HeaderRLP:    headerRlp,
-		BlockBody:    block.Body(),
-		Receipts:     make(types.Receipts, 0),
-		StateLeafs:   make(map[common.Hash][]byte),
-		StorageLeafs: make(map[common.Hash]map[common.Hash][]byte),
+		BlockHash:       block.Hash(),
+		BlockNumber:     block.Number(),
+		HeaderRLP:       headerRlp,
+		BlockBody:       block.Body(),
+		TrxMetaData:     make([]*TrxMetaData, 0, trxLen),
+		Receipts:        make(types.Receipts, 0, trxLen),
+		ReceiptMetaData: make([]*ReceiptMetaData, 0, trxLen),
+		StateLeafs:      make(map[common.Hash][]byte),
+		StorageLeafs:    make(map[common.Hash]map[common.Hash][]byte),
 	}
-	for _, trx := range block.Transactions() {
+	for gethTransactionIndex, trx := range block.Transactions() {
+		// Extract to and from data from the the transactions for indexing
+		from, err := pc.client.TransactionSender(context.Background(), trx, block.Hash(), uint(gethTransactionIndex))
+		if err != nil {
+			return nil, err
+		}
+		txMeta := &TrxMetaData{
+			To:   trx.To().Hex(),
+			From: from.Hex(),
+		}
+		// txMeta will have same index as its corresponding trx in the convertedPayload.BlockBody
+		convertedPayload.TrxMetaData = append(convertedPayload.TrxMetaData, txMeta)
+
+		// Retrieve receipt for this transaction
 		gethReceipt, err := pc.client.TransactionReceipt(context.Background(), trx.Hash())
 		if err != nil {
 			return nil, err
 		}
+		// Extract topic0 data from the receipt's logs for indexing
+		rctMeta := &ReceiptMetaData{
+			Topic0s: make([]string, 0, len(gethReceipt.Logs)),
+		}
+		for _, log := range gethReceipt.Logs {
+			if len(log.Topics[0]) < 1 {
+				continue
+			}
+			rctMeta.Topic0s = append(rctMeta.Topic0s, log.Topics[0].Hex())
+		}
+		// receipt and rctMeta will have same indexes
 		convertedPayload.Receipts = append(convertedPayload.Receipts, gethReceipt)
+		convertedPayload.ReceiptMetaData = append(convertedPayload.ReceiptMetaData, rctMeta)
 	}
 
 	// Unpack state diff rlp to access fields

pkg/ipfs/converter_test.go

@@ -15,3 +15,28 @@
 // along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
 package ipfs_test
+
+import (
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+
+	"github.com/vulcanize/vulcanizedb/pkg/ipfs/test_helpers"
+	"github.com/vulcanize/vulcanizedb/pkg/ipfs/test_helpers/mocks"
+)
+
+var _ = Describe("Converter", func() {
+	Describe("Convert", func() {
+		It("Converts StatediffPayloads into IPLDPayloads", func() {
+			mockConverter := mocks.PayloadConverter{}
+			mockConverter.ReturnIPLDPayload = &test_helpers.MockIPLDPayload
+			ipldPayload, err := mockConverter.Convert(test_helpers.MockStatediffPayload)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(ipldPayload).To(Equal(&test_helpers.MockIPLDPayload))
+			Expect(mockConverter.PassedStatediffPayload).To(Equal(test_helpers.MockStatediffPayload))
+		})
+
+		It("Fails if", func() {
+
+		})
+	})
+})

pkg/ipfs/ipfs_suite_test.go

@@ -15,3 +15,21 @@
 // along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
 package ipfs_test
+
+import (
+	"github.com/sirupsen/logrus"
+	"io/ioutil"
+	"testing"
+
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+)
+
+func TestIPFS(t *testing.T) {
+	RegisterFailHandler(Fail)
+	RunSpecs(t, "IPFS Suite Test")
+}
+
+var _ = BeforeSuite(func() {
+	logrus.SetOutput(ioutil.Discard)
+})

pkg/ipfs/processor.go

@@ -28,31 +28,36 @@ import (
 
 const payloadChanBufferSize = 800 // 1/10th max eth sub buffer size
 
-// Indexer is an interface for streaming, converting to IPLDs, publishing, and indexing all Ethereum data
+// SyncAndPublish is an interface for streaming, converting to IPLDs, publishing, and indexing all Ethereum data
 // This is the top-level interface used by the syncAndPublish command
-type Indexer interface {
-	Index(wg *sync.WaitGroup) error
+type SyncAndPublish interface {
+	Process(wg *sync.WaitGroup) error
 }
 
-// ipfsIndexer is the underlying struct for the Indexer interface
-// we are not exporting this to enforce proper initialization through the NewIPFSIndexer function
-type ipfsIndexer struct {
-	Streamer    Streamer
-	Converter   Converter
-	Publisher   Publisher
-	Repository  Repository
+// Processor is the underlying struct for the SyncAndPublish interface
+type Processor struct {
+	// Interface for streaming statediff payloads over a geth rpc subscription
+	Streamer StateDiffStreamer
+	// Interface for converting statediff payloads into ETH-IPLD object payloads
+	Converter PayloadConverter
+	// Interface for publishing the ETH-IPLD payloads to IPFS
+	Publisher IPLDPublisher
+	// Interface for indexing the CIDs of the published ETH-IPLDs in Postgres
+	Repository CIDRepository
+	// Chan the processor uses to subscribe to state diff payloads from the Streamer
 	PayloadChan chan statediff.Payload
-	QuitChan    chan bool
+	// Chan used to shut down the Processor
+	QuitChan chan bool
 }
 
-// NewIPFSIndexer creates a new Indexer interface using an underlying ipfsIndexer struct
-func NewIPFSIndexer(ipfsPath string, db *postgres.DB, ethClient core.EthClient, rpcClient core.RpcClient, qc chan bool) (Indexer, error) {
+// NewIPFSProcessor creates a new Processor interface using an underlying Processor struct
+func NewIPFSProcessor(ipfsPath string, db *postgres.DB, ethClient core.EthClient, rpcClient core.RpcClient, qc chan bool) (SyncAndPublish, error) {
 	publisher, err := NewIPLDPublisher(ipfsPath)
 	if err != nil {
 		return nil, err
 	}
-	return &ipfsIndexer{
-		Streamer:    NewStateDiffSyncer(rpcClient),
+	return &Processor{
+		Streamer:    NewStateDiffStreamer(rpcClient),
 		Repository:  NewCIDRepository(db),
 		Converter:   NewPayloadConverter(ethClient),
 		Publisher:   publisher,
@@ -61,15 +66,14 @@ func NewIPFSIndexer(ipfsPath string, db *postgres.DB, ethClient core.EthClient,
 	}, nil
 }
 
-// Index is the main processing loop
-func (i *ipfsIndexer) Index(wg *sync.WaitGroup) error {
+// Process is the main processing loop
+func (i *Processor) Process(wg *sync.WaitGroup) error {
 	sub, err := i.Streamer.Stream(i.PayloadChan)
 	if err != nil {
 		return err
 	}
 	go func() {
 		wg.Add(1)
-		defer wg.Done()
 		for {
 			select {
 			case payload := <-i.PayloadChan:
@@ -80,10 +84,12 @@ func (i *ipfsIndexer) Index(wg *sync.WaitGroup) error {
 				ipldPayload, err := i.Converter.Convert(payload)
 				if err != nil {
 					log.Error(err)
+					continue
 				}
 				cidPayload, err := i.Publisher.Publish(ipldPayload)
 				if err != nil {
 					log.Error(err)
+					continue
 				}
 				err = i.Repository.Index(cidPayload)
 				if err != nil {
@@ -92,7 +98,8 @@ func (i *ipfsIndexer) Index(wg *sync.WaitGroup) error {
 			case err = <-sub.Err():
 				log.Error(err)
 			case <-i.QuitChan:
-				log.Info("quiting IPFSIndexer")
+				log.Info("quiting IPFSProcessor")
+				wg.Done()
 				return
 			}
 		}

pkg/ipfs/processor_test.go

@@ -0,0 +1,69 @@
+// 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 ipfs_test
+
+import (
+	"sync"
+
+	"github.com/ethereum/go-ethereum/rpc"
+	"github.com/ethereum/go-ethereum/statediff"
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+
+	"github.com/vulcanize/vulcanizedb/pkg/ipfs"
+	"github.com/vulcanize/vulcanizedb/pkg/ipfs/test_helpers"
+	"github.com/vulcanize/vulcanizedb/pkg/ipfs/test_helpers/mocks"
+)
+
+var _ = Describe("Processor", func() {
+
+	Describe("Process", func() {
+		It("Streams StatediffPayloads, converts them to IPLDPayloads, publishes IPLDPayloads, and indexes CIDPayloads", func() {
+			wg := new(sync.WaitGroup)
+			payloadChan := make(chan statediff.Payload, 800)
+			processor := ipfs.Processor{
+				Repository: &mocks.CIDRepository{
+					ReturnErr: nil,
+				},
+				Publisher: &mocks.IPLDPublisher{
+					ReturnCIDPayload: &test_helpers.MockCIDPayload,
+					ReturnErr:        nil,
+				},
+				Streamer: &mocks.StateDiffStreamer{
+					ReturnSub: &rpc.ClientSubscription{},
+					StreamPayloads: []statediff.Payload{
+						test_helpers.MockStatediffPayload,
+					},
+					ReturnErr: nil,
+					WaitGroup: wg,
+				},
+				Converter: &mocks.PayloadConverter{
+					ReturnIPLDPayload: &test_helpers.MockIPLDPayload,
+					ReturnErr:         nil,
+				},
+				PayloadChan: payloadChan,
+			}
+			err := processor.Process(wg)
+			Expect(err).ToNot(HaveOccurred())
+			wg.Wait()
+		})
+
+		It("Fails if", func() {
+
+		})
+	})
+})

pkg/ipfs/publisher.go

@@ -21,6 +21,7 @@ import (
 
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/core/types"
+	rlp2 "github.com/ethereum/go-ethereum/rlp"
 
 	"github.com/vulcanize/eth-block-extractor/pkg/ipfs"
 	"github.com/vulcanize/eth-block-extractor/pkg/ipfs/eth_block_header"
@@ -31,13 +32,13 @@ import (
 	"github.com/vulcanize/eth-block-extractor/pkg/wrappers/rlp"
 )
 
-// Publisher is the interface for publishing an IPLD payload
-type Publisher interface {
+// IPLDPublisher is the interface for publishing an IPLD payload
+type IPLDPublisher interface {
 	Publish(payload *IPLDPayload) (*CIDPayload, error)
 }
 
-// IPLDPublisher is the underlying struct for the Publisher interface
-type IPLDPublisher struct {
+// Publisher is the underlying struct for the IPLDPublisher interface
+type Publisher struct {
 	Node              *ipfs.IPFS
 	HeaderPutter      *eth_block_header.BlockHeaderDagPutter
 	TransactionPutter *eth_block_transactions.BlockTransactionsDagPutter
@@ -46,27 +47,15 @@ type IPLDPublisher struct {
 	StoragePutter     *eth_storage_trie.StorageTrieDagPutter
 }
 
-// CID payload is a struct to hold all the CIDs and their meta data
-type CIDPayload struct {
-	BlockNumber     string
-	BlockHash       string
-	HeaderCID       string
-	TransactionCIDs map[common.Hash]string
-	ReceiptCIDs     map[common.Hash]string
-	StateLeafCIDs   map[common.Hash]string
-	StorageLeafCIDs map[common.Hash]map[common.Hash]string
-}
-
-// NewIPLDPublisher creates a pointer to a new IPLDPublisher which satisfies the Publisher interface
-func NewIPLDPublisher(ipfsPath string) (*IPLDPublisher, error) {
+// NewIPLDPublisher creates a pointer to a new Publisher which satisfies the IPLDPublisher interface
+func NewIPLDPublisher(ipfsPath string) (*Publisher, error) {
 	node, err := ipfs.InitIPFSNode(ipfsPath)
 	if err != nil {
 		return nil, err
 	}
-	decoder := rlp.RlpDecoder{}
-	return &IPLDPublisher{
+	return &Publisher{
 		Node:              node,
-		HeaderPutter:      eth_block_header.NewBlockHeaderDagPutter(node, decoder),
+		HeaderPutter:      eth_block_header.NewBlockHeaderDagPutter(node, rlp.RlpDecoder{}),
 		TransactionPutter: eth_block_transactions.NewBlockTransactionsDagPutter(node),
 		ReceiptPutter:     eth_block_receipts.NewEthBlockReceiptDagPutter(node),
 		StatePutter:       eth_state_trie.NewStateTrieDagPutter(node),
@@ -75,21 +64,35 @@ func NewIPLDPublisher(ipfsPath string) (*IPLDPublisher, error) {
 }
 
 // Publish publishes an IPLDPayload to IPFS and returns the corresponding CIDPayload
-func (pub *IPLDPublisher) Publish(payload *IPLDPayload) (*CIDPayload, error) {
+func (pub *Publisher) Publish(payload *IPLDPayload) (*CIDPayload, error) {
 	// Process and publish headers
 	headerCid, err := pub.publishHeaders(payload.HeaderRLP)
 	if err != nil {
 		return nil, err
 	}
 
+	// Process and publish uncles
+	uncleCids := make(map[common.Hash]string)
+	for _, uncle := range payload.BlockBody.Uncles {
+		uncleRlp, err := rlp2.EncodeToBytes(uncle)
+		if err != nil {
+			return nil, err
+		}
+		cid, err := pub.publishHeaders(uncleRlp)
+		if err != nil {
+			return nil, err
+		}
+		uncleCids[uncle.Hash()] = cid
+	}
+
 	// Process and publish transactions
-	transactionCids, err := pub.publishTransactions(payload.BlockBody)
+	transactionCids, err := pub.publishTransactions(payload.BlockBody, payload.TrxMetaData)
 	if err != nil {
 		return nil, err
 	}
 
 	// Process and publish receipts
-	receiptsCids, err := pub.publishReceipts(payload.Receipts)
+	receiptsCids, err := pub.publishReceipts(payload.Receipts, payload.ReceiptMetaData)
 	if err != nil {
 		return nil, err
 	}
@@ -111,6 +114,7 @@ func (pub *IPLDPublisher) Publish(payload *IPLDPayload) (*CIDPayload, error) {
 		BlockHash:       payload.BlockHash.Hex(),
 		BlockNumber:     payload.BlockNumber.String(),
 		HeaderCID:       headerCid,
+		UncleCIDS:       uncleCids,
 		TransactionCIDs: transactionCids,
 		ReceiptCIDs:     receiptsCids,
 		StateLeafCIDs:   stateLeafCids,
@@ -118,7 +122,7 @@ func (pub *IPLDPublisher) Publish(payload *IPLDPayload) (*CIDPayload, error) {
 	}, nil
 }
 
-func (pub *IPLDPublisher) publishHeaders(headerRLP []byte) (string, error) {
+func (pub *Publisher) publishHeaders(headerRLP []byte) (string, error) {
 	headerCids, err := pub.HeaderPutter.DagPut(headerRLP)
 	if err != nil {
 		return "", err
@@ -129,7 +133,7 @@ func (pub *IPLDPublisher) publishHeaders(headerRLP []byte) (string, error) {
 	return headerCids[0], nil
 }
 
-func (pub *IPLDPublisher) publishTransactions(blockBody *types.Body) (map[common.Hash]string, error) {
+func (pub *Publisher) publishTransactions(blockBody *types.Body, trxMeta []*TrxMetaData) (map[common.Hash]*TrxMetaData, error) {
 	transactionCids, err := pub.TransactionPutter.DagPut(blockBody)
 	if err != nil {
 		return nil, err
@@ -137,14 +141,15 @@ func (pub *IPLDPublisher) publishTransactions(blockBody *types.Body) (map[common
 	if len(transactionCids) != len(blockBody.Transactions) {
 		return nil, errors.New("expected one CID for each transaction")
 	}
-	mappedTrxCids := make(map[common.Hash]string, len(transactionCids))
+	mappedTrxCids := make(map[common.Hash]*TrxMetaData, len(transactionCids))
 	for i, trx := range blockBody.Transactions {
-		mappedTrxCids[trx.Hash()] = transactionCids[i]
+		mappedTrxCids[trx.Hash()] = trxMeta[i]
+		mappedTrxCids[trx.Hash()].CID = transactionCids[i]
 	}
 	return mappedTrxCids, nil
 }
 
-func (pub *IPLDPublisher) publishReceipts(receipts types.Receipts) (map[common.Hash]string, error) {
+func (pub *Publisher) publishReceipts(receipts types.Receipts, receiptMeta []*ReceiptMetaData) (map[common.Hash]*ReceiptMetaData, error) {
 	receiptsCids, err := pub.ReceiptPutter.DagPut(receipts)
 	if err != nil {
 		return nil, err
@@ -152,14 +157,15 @@ func (pub *IPLDPublisher) publishReceipts(receipts types.Receipts) (map[common.H
 	if len(receiptsCids) != len(receipts) {
 		return nil, errors.New("expected one CID for each receipt")
 	}
-	mappedRctCids := make(map[common.Hash]string, len(receiptsCids))
+	mappedRctCids := make(map[common.Hash]*ReceiptMetaData, len(receiptsCids))
 	for i, rct := range receipts {
-		mappedRctCids[rct.TxHash] = receiptsCids[i]
+		mappedRctCids[rct.TxHash] = receiptMeta[i]
+		mappedRctCids[rct.TxHash].CID = receiptsCids[i]
 	}
 	return mappedRctCids, nil
 }
 
-func (pub *IPLDPublisher) publishStateLeafs(stateLeafs map[common.Hash][]byte) (map[common.Hash]string, error) {
+func (pub *Publisher) publishStateLeafs(stateLeafs map[common.Hash][]byte) (map[common.Hash]string, error) {
 	stateLeafCids := make(map[common.Hash]string)
 	for addr, leaf := range stateLeafs {
 		stateLeafCid, err := pub.StatePutter.DagPut(leaf)
@@ -174,7 +180,7 @@ func (pub *IPLDPublisher) publishStateLeafs(stateLeafs map[common.Hash][]byte) (
 	return stateLeafCids, nil
 }
 
-func (pub *IPLDPublisher) publishStorageLeafs(storageLeafs map[common.Hash]map[common.Hash][]byte) (map[common.Hash]map[common.Hash]string, error) {
+func (pub *Publisher) publishStorageLeafs(storageLeafs map[common.Hash]map[common.Hash][]byte) (map[common.Hash]map[common.Hash]string, error) {
 	storageLeafCids := make(map[common.Hash]map[common.Hash]string)
 	for addr, storageTrie := range storageLeafs {
 		storageLeafCids[addr] = make(map[common.Hash]string)

pkg/ipfs/publisher_test.go

@@ -15,3 +15,28 @@
 // along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
 package ipfs_test
+
+import (
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+
+	"github.com/vulcanize/vulcanizedb/pkg/ipfs/test_helpers"
+	"github.com/vulcanize/vulcanizedb/pkg/ipfs/test_helpers/mocks"
+)
+
+var _ = Describe("Publisher", func() {
+	Describe("Publish", func() {
+		It("Publishes IPLDPayload to IPFS", func() {
+			mockPublisher := mocks.IPLDPublisher{}
+			mockPublisher.ReturnCIDPayload = &test_helpers.MockCIDPayload
+			cidPayload, err := mockPublisher.Publish(&test_helpers.MockIPLDPayload)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(cidPayload).To(Equal(&test_helpers.MockCIDPayload))
+			Expect(mockPublisher.PassedIPLDPayload).To(Equal(&test_helpers.MockIPLDPayload))
+		})
+
+		It("Fails if", func() {
+
+		})
+	})
+})

pkg/ipfs/repository.go

@@ -18,29 +18,30 @@ package ipfs
 
 import (
 	"github.com/jmoiron/sqlx"
+	"github.com/lib/pq"
 
 	"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
 )
 
-// Repository is an interface for indexing CIDPayloads
-type Repository interface {
+// CIDRepository is an interface for indexing CIDPayloads
+type CIDRepository interface {
 	Index(cidPayload *CIDPayload) error
 }
 
-// CIDRepository is the underlying struct for the Repository interface
-type CIDRepository struct {
+// Repository is the underlying struct for the CIDRepository interface
+type Repository struct {
 	db *postgres.DB
 }
 
-// NewCIDRepository creates a new pointer to a CIDRepository which satisfies the Repository interface
-func NewCIDRepository(db *postgres.DB) *CIDRepository {
-	return &CIDRepository{
+// NewCIDRepository creates a new pointer to a Repository which satisfies the CIDRepository interface
+func NewCIDRepository(db *postgres.DB) *Repository {
+	return &Repository{
 		db: db,
 	}
 }
 
-// IndexCIDs indexes a cidPayload in Postgres
-func (repo *CIDRepository) Index(cidPayload *CIDPayload) error {
+// Index indexes a cidPayload in Postgres
+func (repo *Repository) Index(cidPayload *CIDPayload) error {
 	tx, _ := repo.db.Beginx()
 	headerID, err := repo.indexHeaderCID(tx, cidPayload.HeaderCID, cidPayload.BlockNumber, cidPayload.BlockHash)
 	if err != nil {
@@ -60,7 +61,7 @@ func (repo *CIDRepository) Index(cidPayload *CIDPayload) error {
 	return tx.Commit()
 }
 
-func (repo *CIDRepository) indexHeaderCID(tx *sqlx.Tx, cid, blockNumber, hash string) (int64, error) {
+func (repo *Repository) indexHeaderCID(tx *sqlx.Tx, cid, blockNumber, hash string) (int64, error) {
 	var headerID int64
 	err := tx.QueryRowx(`INSERT INTO public.header_cids (block_number, block_hash, cid) VALUES ($1, $2, $3)
 								ON CONFLICT DO UPDATE SET cid = $3
@@ -69,19 +70,19 @@ func (repo *CIDRepository) indexHeaderCID(tx *sqlx.Tx, cid, blockNumber, hash st
 	return headerID, err
 }
 
-func (repo *CIDRepository) indexTransactionAndReceiptCIDs(tx *sqlx.Tx, payload *CIDPayload, headerID int64) error {
-	for hash, trxCid := range payload.TransactionCIDs {
+func (repo *Repository) indexTransactionAndReceiptCIDs(tx *sqlx.Tx, payload *CIDPayload, headerID int64) error {
+	for hash, trxCidMeta := range payload.TransactionCIDs {
 		var txID int64
-		err := tx.QueryRowx(`INSERT INTO public.transaction_cids (header_id, tx_hash, cid) VALUES ($1, $2, $3) 
-										ON CONFLICT DO UPDATE SET cid = $3
-										RETURNING id`,
-			headerID, hash.Hex(), trxCid).Scan(&txID)
+		err := tx.QueryRowx(`INSERT INTO public.transaction_cids (header_id, tx_hash, cid, dst, src) VALUES ($1, $2, $3, $4, $5) 
+									ON CONFLICT DO UPDATE SET (cid, dst, src) = ($3, $4, $5)
+									RETURNING id`,
+			headerID, hash.Hex(), trxCidMeta.CID, trxCidMeta.To, trxCidMeta.From).Scan(&txID)
 		if err != nil {
 			return err
 		}
-		receiptCid, ok := payload.ReceiptCIDs[hash]
+		receiptCidMeta, ok := payload.ReceiptCIDs[hash]
 		if ok {
-			err = repo.indexReceiptCID(tx, receiptCid, txID)
+			err = repo.indexReceiptCID(tx, receiptCidMeta, txID)
 			if err != nil {
 				return err
 			}
@@ -90,18 +91,18 @@ func (repo *CIDRepository) indexTransactionAndReceiptCIDs(tx *sqlx.Tx, payload *
 	return nil
 }
 
-func (repo *CIDRepository) indexReceiptCID(tx *sqlx.Tx, cid string, txId int64) error {
-	_, err := tx.Exec(`INSERT INTO public.receipt_cids (tx_id, cid) VALUES ($1, $2) 
-										ON CONFLICT DO UPDATE SET cid = $2`, txId, cid)
+func (repo *Repository) indexReceiptCID(tx *sqlx.Tx, cidMeta *ReceiptMetaData, txID int64) error {
+	_, err := tx.Exec(`INSERT INTO public.receipt_cids (tx_id, cid, topic0s) VALUES ($1, $2, $3) 
+							  ON CONFLICT DO UPDATE SET (cid, topic0s) = ($2, $3)`, txID, cidMeta.CID, pq.Array(cidMeta.Topic0s))
 	return err
 }
 
-func (repo *CIDRepository) indexStateAndStorageCIDs(tx *sqlx.Tx, payload *CIDPayload, headerID int64) error {
+func (repo *Repository) indexStateAndStorageCIDs(tx *sqlx.Tx, payload *CIDPayload, headerID int64) error {
 	for accountKey, stateCID := range payload.StateLeafCIDs {
 		var stateID int64
-		err := tx.QueryRowx(`INSERT INTO public.state_cids (header_id, account_key, cid) VALUES ($1, $2, $3) 
-										ON CONFLICT DO UPDATE SET cid = $3
-										RETURNING id`,
+		err := tx.QueryRowx(`INSERT INTO public.state_cids (header_id, account_key, cid) VALUES ($1, $2, $3)
+									ON CONFLICT DO UPDATE SET cid = $3
+									RETURNING id`,
 			headerID, accountKey.Hex(), stateCID).Scan(&stateID)
 		if err != nil {
 			return err
@@ -116,8 +117,8 @@ func (repo *CIDRepository) indexStateAndStorageCIDs(tx *sqlx.Tx, payload *CIDPay
 	return nil
 }
 
-func (repo *CIDRepository) indexStorageCID(tx *sqlx.Tx, key, cid string, stateId int64) error {
+func (repo *Repository) indexStorageCID(tx *sqlx.Tx, key, cid string, stateID int64) error {
 	_, err := repo.db.Exec(`INSERT INTO public.storage_cids (state_id, storage_key, cid) VALUES ($1, $2, $3) 
-										ON CONFLICT DO UPDATE SET cid = $3`, stateId, key, cid)
+								   ON CONFLICT DO UPDATE SET cid = $3`, stateID, key, cid)
 	return err
 }

pkg/ipfs/repository_test.go

@@ -15,3 +15,26 @@
 // along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
 package ipfs_test
+
+import (
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+
+	"github.com/vulcanize/vulcanizedb/pkg/ipfs/test_helpers"
+	"github.com/vulcanize/vulcanizedb/pkg/ipfs/test_helpers/mocks"
+)
+
+var _ = Describe("Repository", func() {
+	Describe("Index", func() {
+		It("Indexes CIDs against their metadata", func() {
+			mockRepo := mocks.CIDRepository{}
+			err := mockRepo.Index(&test_helpers.MockCIDPayload)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(mockRepo.PassedCIDPayload).To(Equal(&test_helpers.MockCIDPayload))
+		})
+
+		It("Fails if", func() {
+
+		})
+	})
+})

pkg/ipfs/streamer.go

@@ -23,25 +23,25 @@ import (
 	"github.com/vulcanize/vulcanizedb/pkg/core"
 )
 
-// Streamer is the interface for streaming a statediff subscription
-type Streamer interface {
+// StateDiffStreamer is the interface for streaming a statediff subscription
+type StateDiffStreamer interface {
 	Stream(payloadChan chan statediff.Payload) (*rpc.ClientSubscription, error)
 }
 
-// StateDiffStreamer is the underlying struct for the Streamer interface
-type StateDiffStreamer struct {
+// Streamer is the underlying struct for the StateDiffStreamer interface
+type Streamer struct {
 	Client      core.RpcClient
 	PayloadChan chan statediff.Payload
 }
 
-// NewStateDiffStreamer creates a pointer to a new StateDiffStreamer which satisfies the Streamer interface
-func NewStateDiffSyncer(client core.RpcClient) *StateDiffStreamer {
-	return &StateDiffStreamer{
+// NewStateDiffStreamer creates a pointer to a new Streamer which satisfies the StateDiffStreamer interface
+func NewStateDiffStreamer(client core.RpcClient) *Streamer {
+	return &Streamer{
 		Client: client,
 	}
 }
 
 // Stream is the main loop for subscribing to data from the Geth state diff process
-func (i *StateDiffStreamer) Stream(payloadChan chan statediff.Payload) (*rpc.ClientSubscription, error) {
-	return i.Client.Subscribe("statediff", i.PayloadChan)
+func (sds *Streamer) Stream(payloadChan chan statediff.Payload) (*rpc.ClientSubscription, error) {
+	return sds.Client.Subscribe("statediff", sds.PayloadChan)
 }

pkg/ipfs/streamer_test.go

@@ -15,3 +15,41 @@
 // along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
 package ipfs_test
+
+import (
+	"sync"
+
+	"github.com/ethereum/go-ethereum/rpc"
+	"github.com/ethereum/go-ethereum/statediff"
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+
+	"github.com/vulcanize/vulcanizedb/pkg/ipfs/test_helpers"
+	"github.com/vulcanize/vulcanizedb/pkg/ipfs/test_helpers/mocks"
+)
+
+var _ = Describe("Streamer", func() {
+	Describe("Stream", func() {
+		It("Streams StatediffPayloads from a Geth RPC subscription", func() {
+			wg := new(sync.WaitGroup)
+			mockStreamer := mocks.StateDiffStreamer{}
+			mockStreamer.ReturnSub = &rpc.ClientSubscription{}
+			mockStreamer.WaitGroup = wg
+			mockStreamer.StreamPayloads = []statediff.Payload{
+				test_helpers.MockStatediffPayload,
+			}
+			payloadChan := make(chan statediff.Payload, 1)
+			sub, err := mockStreamer.Stream(payloadChan)
+			wg.Wait()
+			Expect(err).ToNot(HaveOccurred())
+			Expect(sub).To(Equal(&rpc.ClientSubscription{}))
+			Expect(mockStreamer.PassedPayloadChan).To(Equal(payloadChan))
+			streamedPayload := <-payloadChan
+			Expect(streamedPayload).To(Equal(test_helpers.MockStatediffPayload))
+		})
+
+		It("Fails if", func() {
+
+		})
+	})
+})

pkg/ipfs/test_helpers/fakes.go

@@ -1,17 +0,0 @@
-// 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 test_helpers

pkg/ipfs/test_helpers/mocks/converter.go

@@ -15,3 +15,22 @@
 // along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
 package mocks
+
+import (
+	"github.com/ethereum/go-ethereum/statediff"
+
+	"github.com/vulcanize/vulcanizedb/pkg/ipfs"
+)
+
+// PayloadConverter is the underlying struct for the Converter interface
+type PayloadConverter struct {
+	PassedStatediffPayload statediff.Payload
+	ReturnIPLDPayload      *ipfs.IPLDPayload
+	ReturnErr              error
+}
+
+// Convert method is used to convert a geth statediff.Payload to a IPLDPayload
+func (pc *PayloadConverter) Convert(payload statediff.Payload) (*ipfs.IPLDPayload, error) {
+	pc.PassedStatediffPayload = payload
+	return pc.ReturnIPLDPayload, pc.ReturnErr
+}

pkg/ipfs/test_helpers/mocks/publisher.go

@@ -15,3 +15,20 @@
 // along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
 package mocks
+
+import (
+	"github.com/vulcanize/vulcanizedb/pkg/ipfs"
+)
+
+// IPLDPublisher is the underlying struct for the Publisher interface
+type IPLDPublisher struct {
+	PassedIPLDPayload *ipfs.IPLDPayload
+	ReturnCIDPayload  *ipfs.CIDPayload
+	ReturnErr         error
+}
+
+// Publish publishes an IPLDPayload to IPFS and returns the corresponding CIDPayload
+func (pub *IPLDPublisher) Publish(payload *ipfs.IPLDPayload) (*ipfs.CIDPayload, error) {
+	pub.PassedIPLDPayload = payload
+	return pub.ReturnCIDPayload, pub.ReturnErr
+}

pkg/ipfs/test_helpers/mocks/repository.go

@@ -15,3 +15,17 @@
 // along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
 package mocks
+
+import "github.com/vulcanize/vulcanizedb/pkg/ipfs"
+
+// CIDRepository is the underlying struct for the Repository interface
+type CIDRepository struct {
+	PassedCIDPayload *ipfs.CIDPayload
+	ReturnErr        error
+}
+
+// Index indexes a cidPayload in Postgres
+func (repo *CIDRepository) Index(cidPayload *ipfs.CIDPayload) error {
+	repo.PassedCIDPayload = cidPayload
+	return repo.ReturnErr
+}

pkg/ipfs/test_helpers/mocks/streamer.go

@@ -15,3 +15,34 @@
 // along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
 package mocks
+
+import (
+	"sync"
+
+	"github.com/ethereum/go-ethereum/rpc"
+	"github.com/ethereum/go-ethereum/statediff"
+)
+
+// StateDiffStreamer is the underlying struct for the Streamer interface
+type StateDiffStreamer struct {
+	PassedPayloadChan chan statediff.Payload
+	ReturnSub         *rpc.ClientSubscription
+	ReturnErr         error
+	StreamPayloads    []statediff.Payload
+	WaitGroup         *sync.WaitGroup
+}
+
+// Stream is the main loop for subscribing to data from the Geth state diff process
+func (sds *StateDiffStreamer) Stream(payloadChan chan statediff.Payload) (*rpc.ClientSubscription, error) {
+	sds.PassedPayloadChan = payloadChan
+
+	go func() {
+		sds.WaitGroup.Add(1)
+		for _, payload := range sds.StreamPayloads {
+			sds.PassedPayloadChan <- payload
+		}
+		sds.WaitGroup.Done()
+	}()
+
+	return sds.ReturnSub, sds.ReturnErr
+}

pkg/ipfs/test_helpers/test_data.go

@@ -0,0 +1,179 @@
+// 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 test_helpers
+
+import (
+	"errors"
+	"math/big"
+	"math/rand"
+
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/core/state"
+	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/ethereum/go-ethereum/crypto"
+	"github.com/ethereum/go-ethereum/rlp"
+	"github.com/ethereum/go-ethereum/statediff"
+
+	"github.com/vulcanize/vulcanizedb/pkg/ipfs"
+)
+
+// AddressToLeafKey hashes an returns an address
+func AddressToLeafKey(address common.Address) common.Hash {
+	return common.BytesToHash(crypto.Keccak256(address[:]))
+}
+
+// Test variables
+var (
+	BlockNumber     = rand.Int63()
+	BlockHash       = "0xfa40fbe2d98d98b3363a778d52f2bcd29d6790b9b3f3cab2b167fd12d3550f73"
+	CodeHash        = common.Hex2Bytes("0xc5d2460186f7233c927e7db2dcc703c0e500b653ca82273b7bfad8045d85a470")
+	NewNonceValue   = rand.Uint64()
+	NewBalanceValue = rand.Int63()
+	ContractRoot    = common.HexToHash("0x56e81f171bcc55a6ff8345e692c0f86e5b48e01b996cadc001622fb5e363b421")
+	StoragePath     = common.HexToHash("0xc5d2460186f7233c927e7db2dcc703c0e500b653ca82273b7bfad8045d85a470").Bytes()
+	StorageKey      = common.HexToHash("0000000000000000000000000000000000000000000000000000000000000001").Bytes()
+	StorageValue    = common.Hex2Bytes("0x03")
+	storage         = []statediff.StorageDiff{{
+		Key:   StorageKey,
+		Value: StorageValue,
+		Path:  StoragePath,
+		Proof: [][]byte{},
+	}}
+	emptyStorage           = make([]statediff.StorageDiff, 0)
+	address                = common.HexToAddress("0xaE9BEa628c4Ce503DcFD7E305CaB4e29E7476592")
+	ContractLeafKey        = AddressToLeafKey(address)
+	anotherAddress         = common.HexToAddress("0xaE9BEa628c4Ce503DcFD7E305CaB4e29E7476593")
+	AnotherContractLeafKey = AddressToLeafKey(anotherAddress)
+	testAccount            = state.Account{
+		Nonce:    NewNonceValue,
+		Balance:  big.NewInt(NewBalanceValue),
+		Root:     ContractRoot,
+		CodeHash: CodeHash,
+	}
+	valueBytes, _       = rlp.EncodeToBytes(testAccount)
+	CreatedAccountDiffs = statediff.AccountDiffsMap{
+		ContractLeafKey: {
+			Key:     ContractLeafKey.Bytes(),
+			Value:   valueBytes,
+			Storage: storage,
+		},
+		AnotherContractLeafKey: {
+			Key:     AnotherContractLeafKey.Bytes(),
+			Value:   valueBytes,
+			Storage: emptyStorage,
+		},
+	}
+
+	UpdatedAccountDiffs = statediff.AccountDiffsMap{ContractLeafKey: {
+		Key:     ContractLeafKey.Bytes(),
+		Value:   valueBytes,
+		Storage: storage,
+	}}
+
+	DeletedAccountDiffs = statediff.AccountDiffsMap{ContractLeafKey: {
+		Key:     ContractLeafKey.Bytes(),
+		Value:   valueBytes,
+		Storage: storage,
+	}}
+
+	MockStateDiff = statediff.StateDiff{
+		BlockNumber:     BlockNumber,
+		BlockHash:       common.HexToHash(BlockHash),
+		CreatedAccounts: CreatedAccountDiffs,
+		DeletedAccounts: DeletedAccountDiffs,
+		UpdatedAccounts: UpdatedAccountDiffs,
+	}
+	MockStateDiffRlp, _ = rlp.EncodeToBytes(MockStateDiff)
+
+	mockTransaction1 = types.NewTransaction(0, common.HexToAddress("0x0"), big.NewInt(1000), 50, big.NewInt(100), nil)
+	mockTransaction2 = types.NewTransaction(1, common.HexToAddress("0x1"), big.NewInt(2000), 100, big.NewInt(200), nil)
+	MockTransactions = types.Transactions{mockTransaction1, mockTransaction2}
+
+	mockReceipt1 = types.NewReceipt(common.HexToHash("0x0").Bytes(), false, 50)
+	mockReceipt2 = types.NewReceipt(common.HexToHash("0x1").Bytes(), false, 100)
+	MockReceipts = types.Receipts{mockReceipt1, mockReceipt2}
+
+	MockHeader = types.Header{
+		Time:        0,
+		Number:      big.NewInt(1),
+		Root:        common.HexToHash("0x0"),
+		TxHash:      common.HexToHash("0x0"),
+		ReceiptHash: common.HexToHash("0x0"),
+	}
+	MockBlock       = types.NewBlock(&MockHeader, MockTransactions, nil, MockReceipts)
+	MockBlockRlp, _ = rlp.EncodeToBytes(MockBlock)
+
+	MockStatediffPayload = statediff.Payload{
+		BlockRlp:     MockBlockRlp,
+		StateDiffRlp: MockStateDiffRlp,
+		Err:          nil,
+	}
+
+	EmptyStatediffPayload = statediff.Payload{
+		BlockRlp:     []byte{},
+		StateDiffRlp: []byte{},
+		Err:          nil,
+	}
+
+	ErrStatediffPayload = statediff.Payload{
+		BlockRlp:     []byte{},
+		StateDiffRlp: []byte{},
+		Err:          errors.New("mock error"),
+	}
+
+	MockIPLDPayload = ipfs.IPLDPayload{}
+
+	MockCIDPayload = ipfs.CIDPayload{
+		BlockNumber: "1",
+		BlockHash:   "0x0",
+		HeaderCID:   "mockHeaderCID",
+		TransactionCIDs: map[common.Hash]*ipfs.TrxMetaData{
+			common.HexToHash("0x0"): {
+				CID:  "mockTrxCID1",
+				To:   "mockTo1",
+				From: "mockFrom1",
+			},
+			common.HexToHash("0x1"): {
+				CID:  "mockTrxCID2",
+				To:   "mockTo2",
+				From: "mockFrom2",
+			},
+		},
+		ReceiptCIDs: map[common.Hash]*ipfs.ReceiptMetaData{
+			common.HexToHash("0x0"): {
+				CID:     "mockReceiptCID1",
+				Topic0s: []string{"mockTopic1"},
+			},
+			common.HexToHash("0x1"): {
+				CID:     "mockReceiptCID2",
+				Topic0s: []string{"mockTopic1", "mockTopic2"},
+			},
+		},
+		StateLeafCIDs: map[common.Hash]string{
+			common.HexToHash("0x0"): "mockStateCID1",
+			common.HexToHash("0x1"): "mockStateCID2",
+		},
+		StorageLeafCIDs: map[common.Hash]map[common.Hash]string{
+			common.HexToHash("0x0"): {
+				common.HexToHash("0x0"): "mockStorageCID1",
+			},
+			common.HexToHash("0x1"): {
+				common.HexToHash("0x1"): "mockStorageCID2",
+			},
+		},
+	}
+)

pkg/ipfs/types.go

@@ -0,0 +1,61 @@
+// 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 ipfs
+
+import (
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/core/types"
+	"math/big"
+)
+
+// IPLDPayload is a custom type which packages ETH data for the IPFS publisher
+type IPLDPayload struct {
+	HeaderRLP       []byte
+	BlockNumber     *big.Int
+	BlockHash       common.Hash
+	BlockBody       *types.Body
+	TrxMetaData     []*TrxMetaData
+	Receipts        types.Receipts
+	ReceiptMetaData []*ReceiptMetaData
+	StateLeafs      map[common.Hash][]byte
+	StorageLeafs    map[common.Hash]map[common.Hash][]byte
+}
+
+// CIDPayload is a struct to hold all the CIDs and their meta data
+type CIDPayload struct {
+	BlockNumber     string
+	BlockHash       string
+	HeaderCID       string
+	UncleCIDS       map[common.Hash]string
+	TransactionCIDs map[common.Hash]*TrxMetaData
+	ReceiptCIDs     map[common.Hash]*ReceiptMetaData
+	StateLeafCIDs   map[common.Hash]string
+	StorageLeafCIDs map[common.Hash]map[common.Hash]string
+}
+
+// ReceiptMetaData wraps some additional data around our receipt CIDs for indexing
+type ReceiptMetaData struct {
+	CID     string
+	Topic0s []string
+}
+
+// TrxMetaData wraps some additional data around our transaction CID for indexing
+type TrxMetaData struct {
+	CID  string
+	To   string
+	From string
+}