* update transformer to able to recheck headers (#4)

* update transformer to able to recheck headers
* put cap on rechecking header
* integration test for recheck headers
* use enum for recheck headers param; make recheck cap configurable
* update integration tests with new test config
* update omni pkg with new recheck header column type
* update migration with new migration tool and final tweaks needed to accommodate changes in omni pkg

pkg/transformers/shared/constants/checked_headers.go

@@ -1,5 +1,13 @@
 package constants
 
+type TransformerExecution bool
+
+const (
+	HeaderRecheck    TransformerExecution = true
+	HeaderMissing    TransformerExecution = false
+	RecheckHeaderCap                      = "4"
+)
+
 var (
 	BiteChecked               = "bite_checked"
 	CatFileChopLumpChecked    = "cat_file_chop_lump_checked"

pkg/transformers/shared/repository.go

@@ -12,17 +12,17 @@ import (
 
 func MarkHeaderChecked(headerID int64, db *postgres.DB, checkedHeadersColumn string) error {
 	_, err := db.Exec(`INSERT INTO public.checked_headers (header_id, `+checkedHeadersColumn+`)
-		VALUES ($1, $2) 
+		VALUES ($1, $2)
 		ON CONFLICT (header_id) DO
-			UPDATE SET `+checkedHeadersColumn+` = $2`, headerID, true)
+			UPDATE SET `+checkedHeadersColumn+` = checked_headers.`+checkedHeadersColumn+` + 1`, headerID, 1)
 	return err
 }
 
 func MarkHeaderCheckedInTransaction(headerID int64, tx *sql.Tx, checkedHeadersColumn string) error {
 	_, err := tx.Exec(`INSERT INTO public.checked_headers (header_id, `+checkedHeadersColumn+`)
-		VALUES ($1, $2) 
+		VALUES ($1, $2)
 		ON CONFLICT (header_id) DO
-			UPDATE SET `+checkedHeadersColumn+` = $2`, headerID, true)
+			UPDATE SET `+checkedHeadersColumn+` = checked_headers.`+checkedHeadersColumn+` + 1`, headerID, 1)
 	return err
 }
 
@@ -52,6 +52,31 @@ func MissingHeaders(startingBlockNumber, endingBlockNumber int64, db *postgres.D
 	return result, err
 }
 
+func RecheckHeaders(startingBlockNumber, endingBlockNumber int64, db *postgres.DB, checkedHeadersColumn string) ([]core.Header, error) {
+	var result []core.Header
+	var query string
+	var err error
+
+	if endingBlockNumber == -1 {
+		query = `SELECT headers.id, headers.block_number, headers.hash FROM headers
+				LEFT JOIN checked_headers on headers.id = header_id
+				WHERE ` + checkedHeadersColumn + ` between 1 and ` + constants.RecheckHeaderCap + `
+				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 ` + checkedHeadersColumn + ` between 1 and ` + constants.RecheckHeaderCap + `
+				AND headers.block_number >= $1
+				AND headers.block_number <= $2
+				AND headers.eth_node_fingerprint = $3`
+		err = db.Select(&result, query, startingBlockNumber, endingBlockNumber, db.Node.ID)
+	}
+
+	return result, err
+}
+
 func GetCheckedColumnNames(db *postgres.DB) ([]string, error) {
 	// Query returns `[]driver.Value`, nullable polymorphic interface
 	var queryResult []driver.Value
@@ -79,11 +104,12 @@ func GetCheckedColumnNames(db *postgres.DB) ([]string, error) {
 	return columnNames, nil
 }
 
-// Builds a SQL string that checks if any column value is FALSE, given the column names.
+// Builds a SQL string that checks if any column value is 0, given the column names.
 // Defaults to FALSE when no columns are provided.
-// Ex: ["columnA", "columnB"] => "NOT (columnA AND columnB)"
+// Ex: ["columnA", "columnB"] => "NOT (columnA!=0 AND columnB!=0)"
 //     [] => "FALSE"
-func CreateNotCheckedSQL(boolColumns []string) string {
+func CreateNotCheckedSQL(boolColumns []string, recheckHeaders constants.TransformerExecution) string {
+
 	var result bytes.Buffer
 
 	if len(boolColumns) == 0 {
@@ -91,13 +117,24 @@ func CreateNotCheckedSQL(boolColumns []string) string {
 	}
 
 	result.WriteString("NOT (")
+
 	// Loop excluding last column name
 	for _, column := range boolColumns[:len(boolColumns)-1] {
-		result.WriteString(fmt.Sprintf("%v AND ", column))
+
+		if recheckHeaders {
+			result.WriteString(fmt.Sprintf("%v>=%s AND ", column, constants.RecheckHeaderCap))
+		} else {
+			result.WriteString(fmt.Sprintf("%v!=0 AND ", column))
+		}
 	}
 
 	// No trailing "OR" for the last column name
-	result.WriteString(fmt.Sprintf("%v)", boolColumns[len(boolColumns)-1]))
+	if recheckHeaders {
+		result.WriteString(fmt.Sprintf("%v>=%s)", boolColumns[len(boolColumns)-1], constants.RecheckHeaderCap))
+	} else {
+		result.WriteString(fmt.Sprintf("%v!=0)", boolColumns[len(boolColumns)-1]))
+
+	}
 
 	return result.String()
 }

pkg/transformers/shared/repository_utility_test.go

@@ -17,6 +17,7 @@
 package shared_test
 
 import (
+	"fmt"
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
 	"github.com/vulcanize/vulcanizedb/pkg/core"
@@ -25,6 +26,7 @@ import (
 	"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres/repositories"
 	"github.com/vulcanize/vulcanizedb/pkg/fakes"
 	"github.com/vulcanize/vulcanizedb/pkg/transformers/shared"
+	"github.com/vulcanize/vulcanizedb/pkg/transformers/shared/constants"
 	"github.com/vulcanize/vulcanizedb/test_config"
 	"math/rand"
 )
@@ -50,7 +52,7 @@ var _ = Describe("Repository utilities", func() {
 
 			columnNames, err := shared.GetCheckedColumnNames(db)
 			Expect(err).NotTo(HaveOccurred())
-			notCheckedSQL = shared.CreateNotCheckedSQL(columnNames)
+			notCheckedSQL = shared.CreateNotCheckedSQL(columnNames, constants.HeaderMissing)
 
 			startingBlockNumber = rand.Int63()
 			eventSpecificBlockNumber = startingBlockNumber + 1
@@ -118,21 +120,28 @@ var _ = Describe("Repository utilities", func() {
 	Describe("CreateNotCheckedSQL", func() {
 		It("generates a correct SQL string for one column", func() {
 			columns := []string{"columnA"}
-			expected := "NOT (columnA)"
-			actual := shared.CreateNotCheckedSQL(columns)
+			expected := "NOT (columnA!=0)"
+			actual := shared.CreateNotCheckedSQL(columns, constants.HeaderMissing)
 			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)
+			expected := "NOT (columnA!=0 AND columnB!=0)"
+			actual := shared.CreateNotCheckedSQL(columns, constants.HeaderMissing)
 			Expect(actual).To(Equal(expected))
 		})
 
 		It("defaults to FALSE when there are no columns", func() {
 			expected := "FALSE"
-			actual := shared.CreateNotCheckedSQL([]string{})
+			actual := shared.CreateNotCheckedSQL([]string{}, constants.HeaderMissing)
+			Expect(actual).To(Equal(expected))
+		})
+
+		It("generates a correct SQL string for rechecking headers", func() {
+			columns := []string{"columnA", "columnB"}
+			expected := fmt.Sprintf("NOT (columnA>=%s AND columnB>=%s)", constants.RecheckHeaderCap, constants.RecheckHeaderCap)
+			actual := shared.CreateNotCheckedSQL(columns, constants.HeaderRecheck)
 			Expect(actual).To(Equal(expected))
 		})
 	})

pkg/transformers/shared/transformer.go

@@ -22,10 +22,11 @@ import (
 	"github.com/vulcanize/vulcanizedb/pkg/core"
 
 	"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
+	"github.com/vulcanize/vulcanizedb/pkg/transformers/shared/constants"
 )
 
 type Transformer interface {
-	Execute(logs []types.Log, header core.Header) error
+	Execute(logs []types.Log, header core.Header, recheckHeaders constants.TransformerExecution) error
 	GetConfig() TransformerConfig
 }