Add fk constraint on logs

db/migrations/1516050071_add_log_fk_constraint.down.sql

@@ -0,0 +1,9 @@
+BEGIN;
+
+ALTER TABLE logs
+  DROP CONSTRAINT receipts_fk;
+
+ALTER TABLE logs
+  DROP COLUMN receipt_id;
+
+COMMIT;

db/migrations/1516050071_add_log_fk_constraint.up.sql

@@ -0,0 +1,12 @@
+BEGIN;
+
+ALTER TABLE logs
+  ADD COLUMN receipt_id INT;
+
+ALTER TABLE logs
+  ADD CONSTRAINT receipts_fk
+FOREIGN KEY (receipt_id)
+REFERENCES receipts (id)
+ON DELETE CASCADE;
+
+COMMIT;

pkg/core/log.go

@@ -5,7 +5,6 @@ type Log struct {
 	TxHash      string
 	Address     string
 	Topics
-	Index   int64
+	Index int64
-	Data    string
+	Data  string
-	Removed bool
 }

pkg/geth/blockchain.go

@@ -5,6 +5,8 @@ import (
 
 	"strings"
 
+	"log"
+
 	"github.com/8thlight/vulcanizedb/pkg/core"
 	"github.com/8thlight/vulcanizedb/pkg/geth/node"
 	"github.com/ethereum/go-ethereum"
@@ -25,7 +27,10 @@ type Blockchain struct {
 
 func NewBlockchain(ipcPath string) *Blockchain {
 	blockchain := Blockchain{}
-	rpcClient, _ := rpc.Dial(ipcPath)
+	rpcClient, err := rpc.Dial(ipcPath)
+	if err != nil {
+		log.Fatal(err)
+	}
 	client := ethclient.NewClient(rpcClient)
 	blockchain.node = node.Info(rpcClient)
 	if infura := isInfuraNode(ipcPath); infura {
@@ -57,7 +62,7 @@ func (blockchain *Blockchain) GetLogs(contract core.Contract, startingBlockNumbe
 	if err != nil {
 		return []core.Log{}, err
 	}
-	logs := LogsToCoreLogs(gethLogs)
+	logs := ToCoreLogs(gethLogs)
 	return logs, nil
 }
 

pkg/geth/log_to_core_log.go

@@ -9,16 +9,16 @@ import (
 	"github.com/ethereum/go-ethereum/core/types"
 )
 
-func LogsToCoreLogs(gethLogs []types.Log) []core.Log {
+func ToCoreLogs(gethLogs []types.Log) []core.Log {
 	var logs []core.Log
 	for _, log := range gethLogs {
-		log := LogToCoreLog(log)
+		log := ToCoreLog(log)
 		logs = append(logs, log)
 	}
 	return logs
 }
 
-func MakeTopics(topics []common.Hash) core.Topics {
+func makeTopics(topics []common.Hash) core.Topics {
 	var hexTopics core.Topics
 	for i, topic := range topics {
 		hexTopics[i] = topic.Hex()
@@ -26,9 +26,9 @@ func MakeTopics(topics []common.Hash) core.Topics {
 	return hexTopics
 }
 
-func LogToCoreLog(gethLog types.Log) core.Log {
+func ToCoreLog(gethLog types.Log) core.Log {
 	topics := gethLog.Topics
-	hexTopics := MakeTopics(topics)
+	hexTopics := makeTopics(topics)
 	return core.Log{
 		Address: strings.ToLower(gethLog.Address.Hex()),
 
@@ -37,6 +37,5 @@ func LogToCoreLog(gethLog types.Log) core.Log {
 		TxHash:      gethLog.TxHash.Hex(),
 		Index:       int64(gethLog.Index),
 		Data:        hexutil.Encode(gethLog.Data),
-		Removed:     gethLog.Removed,
 	}
 }

pkg/geth/log_to_core_log_test.go

@@ -27,7 +27,6 @@ var _ = Describe("Conversion of GethLog to core.Log", func() {
 				common.HexToHash("0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef"),
 				common.HexToHash("0x00000000000000000000000080b2c9d7cbbf30a1b0fc8983c647d754c6525615"),
 			},
-			Removed: true,
 		}
 
 		expected := core.Log{
@@ -37,18 +36,16 @@ var _ = Describe("Conversion of GethLog to core.Log", func() {
 			TxHash:      gethLog.TxHash.Hex(),
 			Index:       2,
 			Topics: core.Topics{
-				0: common.HexToHash("0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef").Hex(),
+				gethLog.Topics[0].Hex(),
-				1: common.HexToHash("0x00000000000000000000000080b2c9d7cbbf30a1b0fc8983c647d754c6525615").Hex(),
+				gethLog.Topics[1].Hex(),
 			},
-			Removed: gethLog.Removed,
 		}
 
-		coreLog := geth.LogToCoreLog(gethLog)
+		coreLog := geth.ToCoreLog(gethLog)
 
 		Expect(coreLog.Address).To(Equal(expected.Address))
 		Expect(coreLog.BlockNumber).To(Equal(expected.BlockNumber))
 		Expect(coreLog.Data).To(Equal(expected.Data))
-		Expect(coreLog.Removed).To(Equal(expected.Removed))
 		Expect(coreLog.Index).To(Equal(expected.Index))
 		Expect(coreLog.Topics[0]).To(Equal(expected.Topics[0]))
 		Expect(coreLog.Topics[1]).To(Equal(expected.Topics[1]))
@@ -84,10 +81,10 @@ var _ = Describe("Conversion of GethLog to core.Log", func() {
 			},
 		}
 
-		expectedOne := geth.LogToCoreLog(gethLogOne)
+		expectedOne := geth.ToCoreLog(gethLogOne)
-		expectedTwo := geth.LogToCoreLog(gethLogTwo)
+		expectedTwo := geth.ToCoreLog(gethLogTwo)
 
-		coreLogs := geth.LogsToCoreLogs([]types.Log{gethLogOne, gethLogTwo})
+		coreLogs := geth.ToCoreLogs([]types.Log{gethLogOne, gethLogTwo})
 
 		Expect(len(coreLogs)).To(Equal(2))
 		Expect(coreLogs[0]).To(Equal(expectedOne))

pkg/geth/receipt_to_core_receipt.go

@@ -49,7 +49,7 @@ func setContractAddress(gethReceipt *types.Receipt) string {
 func dereferenceLogs(gethReceipt *types.Receipt) []core.Log {
 	logs := []core.Log{}
 	for _, log := range gethReceipt.Logs {
-		logs = append(logs, LogToCoreLog(*log))
+		logs = append(logs, ToCoreLog(*log))
 	}
 	return logs
 }

pkg/repositories/postgres.go

@@ -310,19 +310,30 @@ func (repository Postgres) createTransaction(tx *sql.Tx, blockId int64, transact
 	if err != nil {
 		return err
 	}
-	if transaction.Receipt.TxHash != "" {
+	if hasReceipt(transaction) {
-		err = repository.createReceipt(tx, transactionId, transaction.Receipt)
+		receiptId, err := repository.createReceipt(tx, transactionId, transaction.Receipt)
 		if err != nil {
 			return err
 		}
-		if len(transaction.Receipt.Logs) > 0 {
+		if hasLogs(transaction) {
-			err = repository.CreateLogs(transaction.Receipt.Logs)
+			err = repository.createLogs(tx, transaction.Receipt.Logs, receiptId)
+			if err != nil {
+				return err
+			}
 		}
 	}
 	return nil
 }
 
-func (repository Postgres) createReceipt(tx *sql.Tx, transactionId int, receipt core.Receipt) error {
+func hasLogs(transaction core.Transaction) bool {
+	return len(transaction.Receipt.Logs) > 0
+}
+
+func hasReceipt(transaction core.Transaction) bool {
+	return transaction.Receipt.TxHash != ""
+}
+
+func (repository Postgres) createReceipt(tx *sql.Tx, transactionId int, receipt core.Receipt) (int, error) {
 	//Not currently persisting log bloom filters
 	var receiptId int
 	err := tx.QueryRow(
@@ -332,7 +343,33 @@ func (repository Postgres) createReceipt(tx *sql.Tx, transactionId int, receipt
                RETURNING id`,
 		receipt.ContractAddress, receipt.TxHash, receipt.CumulativeGasUsed, receipt.GasUsed, receipt.StateRoot, receipt.Status, transactionId).Scan(&receiptId)
 	if err != nil {
-		return err
+		return receiptId, err
+	}
+	return receiptId, nil
+}
+
+func (repository Postgres) 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)
+                VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
+                ON CONFLICT (index, block_number)
+                  DO UPDATE
+                    SET block_number = $1,
+                   	    address = $2,
+                   	    tx_hash = $3,
+                   	    index = $4,
+                   	    topic0 = $5,
+                   	    topic1 = $6,
+                   	    topic2 = $7,
+                   	    topic3 = $8,
+                   	    data = $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 {
+			return ErrDBInsertFailed
+		}
 	}
 	return nil
 }

pkg/repositories/testing/helpers.go

@@ -399,7 +399,7 @@ func AssertRepositoryBehavior(buildRepository func(node core.Node) repositories.
 				Index:       0,
 				Address:     "x123",
 				TxHash:      "x456",
-				Topics:      [4]string{0: "x777", 1: "x888", 2: "x999"},
+				Topics:      core.Topics{0: "x777", 1: "x888", 2: "x999"},
 				Data:        "xabc",
 			}},
 			)
@@ -428,7 +428,7 @@ func AssertRepositoryBehavior(buildRepository func(node core.Node) repositories.
 				Index:       0,
 				Address:     "x123",
 				TxHash:      "x456",
-				Topics:      [4]string{0: "x777", 1: "x888", 2: "x999"},
+				Topics:      core.Topics{0: "x777", 1: "x888", 2: "x999"},
 				Data:        "xABC",
 			},
 			})
@@ -437,7 +437,7 @@ func AssertRepositoryBehavior(buildRepository func(node core.Node) repositories.
 				Index:       0,
 				Address:     "x123",
 				TxHash:      "x456",
-				Topics:      [4]string{0: "x777", 1: "x888", 2: "x999"},
+				Topics:      core.Topics{0: "x777", 1: "x888", 2: "x999"},
 				Data:        "xXYZ",
 			},
 			})
@@ -452,7 +452,7 @@ func AssertRepositoryBehavior(buildRepository func(node core.Node) repositories.
 				Index:       0,
 				Address:     "x123",
 				TxHash:      "x456",
-				Topics:      [4]string{0: "x777", 1: "x888", 2: "x999"},
+				Topics:      core.Topics{0: "x777", 1: "x888", 2: "x999"},
 				Data:        "xabc",
 			}},
 			)
@@ -461,7 +461,7 @@ func AssertRepositoryBehavior(buildRepository func(node core.Node) repositories.
 				Index:       1,
 				Address:     "x123",
 				TxHash:      "x789",
-				Topics:      [4]string{0: "x111", 1: "x222", 2: "x333"},
+				Topics:      core.Topics{0: "x111", 1: "x222", 2: "x333"},
 				Data:        "xdef",
 			}},
 			)
@@ -470,7 +470,7 @@ func AssertRepositoryBehavior(buildRepository func(node core.Node) repositories.
 				Index:       0,
 				Address:     "x123",
 				TxHash:      "x456",
-				Topics:      [4]string{0: "x777", 1: "x888", 2: "x999"},
+				Topics:      core.Topics{0: "x777", 1: "x888", 2: "x999"},
 				Data:        "xabc",
 			}},
 			)
@@ -561,13 +561,12 @@ func AssertRepositoryBehavior(buildRepository func(node core.Node) repositories.
 			Expect(err).To(Not(HaveOccurred()))
 			retrievedLogs := repository.FindLogs("0x99041f808d598b782d5a3e498681c2452a31da08", 4745407)
 
-			Expect(retrievedLogs).To(Not(BeZero()))
+			expected := logs[1:]
+			Expect(retrievedLogs).To(Equal(expected))
 		})
 
-		It("saves the logs attached to a receipt", func() {
+		It("still saves receipts without logs", func() {
-			logs := make([]core.Log, 0)
 			receipt := core.Receipt{
-				Logs:   logs,
 				TxHash: "0x002c4799161d809b23f67884eb6598c9df5894929fe1a9ead97ca175d360f547",
 			}
 			transaction := core.Transaction{
@@ -580,11 +579,9 @@ func AssertRepositoryBehavior(buildRepository func(node core.Node) repositories.
 			}
 			repository.CreateOrUpdateBlock(block)
 
-			retrievedLogs := repository.FindLogs("0x99041f808d598b782d5a3e498681c2452a31da08", 4745407)
 			_, err := repository.FindReceipt(receipt.TxHash)
 
 			Expect(err).To(Not(HaveOccurred()))
-			Expect(retrievedLogs).To(BeZero())
 		})
 	})