Refactor fetching out from repositories to log_fetcher and watcher

pkg/transformers/shared/log_fetcher_test.go

@@ -32,8 +32,12 @@ var _ = Describe("Fetcher", func() {
 			fetcher := shared.NewFetcher(blockChain)
 			header := fakes.FakeHeader
 
-			addresses := []string{"0xfakeAddress", "0xanotherFakeAddress"}
-			topicZeros := [][]common.Hash{{common.BytesToHash([]byte{1, 2, 3, 4, 5})}}
+			addresses := []common.Address{
+				common.HexToAddress("0xfakeAddress"),
+				common.HexToAddress("0xanotherFakeAddress"),
+			}
+
+			topicZeros := []common.Hash{common.BytesToHash([]byte{1, 2, 3, 4, 5})}
 
 			_, err := fetcher.FetchLogs(addresses, topicZeros, header)
 
@@ -45,7 +49,7 @@ var _ = Describe("Fetcher", func() {
 			expectedQuery := ethereum.FilterQuery{
 				BlockHash: &blockHash,
 				Addresses: []common.Address{address1, address2},
-				Topics:    topicZeros,
+				Topics:    [][]common.Hash{topicZeros},
 			}
 			blockChain.AssertGetEthLogsWithCustomQueryCalledWith(expectedQuery)
 		})
@@ -55,7 +59,7 @@ var _ = Describe("Fetcher", func() {
 			blockChain.SetGetEthLogsWithCustomQueryErr(fakes.FakeError)
 			fetcher := shared.NewFetcher(blockChain)
 
-			_, err := fetcher.FetchLogs([]string{}, [][]common.Hash{}, core.Header{})
+			_, err := fetcher.FetchLogs([]common.Address{}, []common.Hash{}, core.Header{})
 
 			Expect(err).To(HaveOccurred())
 			Expect(err).To(MatchError(fakes.FakeError))

pkg/transformers/shared/repository.go

@@ -2,8 +2,11 @@ package shared
 
 import (
 	"database/sql"
+	"database/sql/driver"
+	"fmt"
 	"github.com/vulcanize/vulcanizedb/pkg/core"
 	"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
+	"strings"
 )
 
 func MarkHeaderChecked(headerID int64, db *postgres.DB, checkedHeadersColumn string) error {
@@ -22,7 +25,9 @@ func MarkHeaderCheckedInTransaction(headerID int64, tx *sql.Tx, checkedHeadersCo
 	return err
 }
 
-func MissingHeaders(startingBlockNumber, endingBlockNumber int64, db *postgres.DB, checkedHeadersColumn string) ([]core.Header, error) {
+// Treats a header as missing if it's not in the headers table, or not checked for some log type
+// TODO Revisit definition of "checked header
+func MissingHeaders(startingBlockNumber, endingBlockNumber int64, db *postgres.DB, notCheckedSQL string) ([]core.Header, error) {
 	var result []core.Header
 	var query string
 	var err error
@@ -30,14 +35,14 @@ func MissingHeaders(startingBlockNumber, endingBlockNumber int64, db *postgres.D
 	if endingBlockNumber == -1 {
 		query = `SELECT headers.id, headers.block_number, headers.hash FROM headers
 				LEFT JOIN checked_headers on headers.id = header_id
-				WHERE (header_id ISNULL OR ` + checkedHeadersColumn + ` IS FALSE)
+				WHERE (header_id ISNULL OR ` + notCheckedSQL + `)
 				AND headers.block_number >= $1
 				AND headers.eth_node_fingerprint = $2`
 		err = db.Select(&result, query, startingBlockNumber, db.Node.ID)
 	} else {
 		query = `SELECT headers.id, headers.block_number, headers.hash FROM headers
 				LEFT JOIN checked_headers on headers.id = header_id
-				WHERE (header_id ISNULL OR ` + checkedHeadersColumn + ` IS FALSE)
+				WHERE (header_id ISNULL OR ` + notCheckedSQL + `)
 				AND headers.block_number >= $1
 				AND headers.block_number <= $2
 				AND headers.eth_node_fingerprint = $3`
@@ -46,3 +51,53 @@ func MissingHeaders(startingBlockNumber, endingBlockNumber int64, db *postgres.D
 
 	return result, err
 }
+
+func GetCheckedColumnNames(db *postgres.DB) ([]string, error) {
+	// Query returns `[]driver.Value`, nullable polymorphic interface
+	var queryResult []driver.Value
+	columnNamesQuery :=
+		`SELECT column_name FROM information_schema.columns
+		WHERE table_schema = 'public'
+			AND table_name = 'checked_headers'
+			AND column_name != 'id'
+			AND column_name != 'header_id';`
+
+	err := db.Select(&queryResult, columnNamesQuery)
+	if err != nil {
+		return []string{}, err
+	}
+
+	// Transform column names from `driver.Value` to strings
+	var columnNames []string
+	for _, result := range queryResult {
+		if columnName, ok := result.(string); ok {
+			columnNames = append(columnNames, columnName)
+		} else {
+			return []string{}, fmt.Errorf("incorrect value for checked_headers column name")
+		}
+	}
+
+	return columnNames, nil
+}
+
+// Builds a SQL string that checks if any column value is FALSE, given the column names.
+// Defaults to FALSE when no columns are provided.
+// Ex: ["columnA", "columnB"] => "NOT (columnA AND columnB)"
+//     [] => "FALSE"
+func CreateNotCheckedSQL(boolColumns []string) string {
+	var result strings.Builder
+
+	if len(boolColumns) == 0 {
+		return "FALSE"
+	}
+
+	result.WriteString("NOT (")
+	for _, column := range boolColumns[:len(boolColumns)-1] {
+		result.WriteString(fmt.Sprintf("%v AND ", column))
+	}
+
+	// No trailing "OR" for last column name
+	result.WriteString(fmt.Sprintf("%v)", boolColumns[len(boolColumns)-1]))
+
+	return result.String()
+}

pkg/transformers/shared/repository_utility_test.go

@@ -0,0 +1,90 @@
+// Copyright 2018 Vulcanize
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package shared_test
+
+import (
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+	"github.com/vulcanize/vulcanizedb/pkg/transformers/shared"
+	"github.com/vulcanize/vulcanizedb/test_config"
+)
+
+var _ = Describe("Repository utilities", func() {
+	Describe("GetCheckedColumnNames", func() {
+		It("gets the column names from checked_headers", func() {
+			db := test_config.NewTestDB(test_config.NewTestNode())
+			test_config.CleanTestDB(db)
+			expectedColumnNames := getExpectedColumnNames()
+			actualColumnNames, err := shared.GetCheckedColumnNames(db)
+			Expect(err).NotTo(HaveOccurred())
+			Expect(actualColumnNames).To(Equal(expectedColumnNames))
+		})
+	})
+
+	Describe("CreateNotCheckedSQL", func() {
+		It("generates a correct SQL string for one column", func() {
+			columns := []string{"columnA"}
+			expected := "NOT (columnA)"
+			actual := shared.CreateNotCheckedSQL(columns)
+			Expect(actual).To(Equal(expected))
+		})
+
+		It("generates a correct SQL string for several columns", func() {
+			columns := []string{"columnA", "columnB"}
+			expected := "NOT (columnA AND columnB)"
+			actual := shared.CreateNotCheckedSQL(columns)
+			Expect(actual).To(Equal(expected))
+		})
+
+		It("defaults to FALSE when there are no columns", func() {
+			expected := "FALSE"
+			actual := shared.CreateNotCheckedSQL([]string{})
+			Expect(actual).To(Equal(expected))
+		})
+	})
+})
+
+func getExpectedColumnNames() []string {
+	return []string{
+		"price_feeds_checked",
+		"flip_kick_checked",
+		"frob_checked",
+		"tend_checked",
+		"bite_checked",
+		"dent_checked",
+		"pit_file_debt_ceiling_checked",
+		"pit_file_ilk_checked",
+		"vat_init_checked",
+		"drip_file_ilk_checked",
+		"drip_file_repo_checked",
+		"drip_file_vow_checked",
+		"deal_checked",
+		"drip_drip_checked",
+		"cat_file_chop_lump_checked",
+		"cat_file_flip_checked",
+		"cat_file_pit_vow_checked",
+		"flop_kick_checked",
+		"vat_move_checked",
+		"vat_fold_checked",
+		"vat_heal_checked",
+		"vat_toll_checked",
+		"vat_tune_checked",
+		"vat_grab_checked",
+		"vat_flux_checked",
+		"vat_slip_checked",
+		"vow_flog_checked",
+		"flap_kick_checked",
+	}
+}