(VDB-560) Rename lightSync to headerSync

pkg/contract_watcher/header/repository/header_repository.go

@@ -0,0 +1,293 @@
+// 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 repository
+
+import (
+	"fmt"
+
+	"github.com/hashicorp/golang-lru"
+	"github.com/jmoiron/sqlx"
+
+	"github.com/vulcanize/vulcanizedb/pkg/core"
+	"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
+)
+
+const columnCacheSize = 1000
+
+type HeaderRepository interface {
+	AddCheckColumn(id string) error
+	AddCheckColumns(ids []string) error
+	MarkHeaderChecked(headerID int64, eventID string) error
+	MarkHeaderCheckedForAll(headerID int64, ids []string) error
+	MarkHeadersCheckedForAll(headers []core.Header, ids []string) error
+	MissingHeaders(startingBlockNumber int64, endingBlockNumber int64, eventID string) ([]core.Header, error)
+	MissingMethodsCheckedEventsIntersection(startingBlockNumber, endingBlockNumber int64, methodIds, eventIds []string) ([]core.Header, error)
+	MissingHeadersForAll(startingBlockNumber, endingBlockNumber int64, ids []string) ([]core.Header, error)
+	CheckCache(key string) (interface{}, bool)
+}
+
+type headerRepository struct {
+	db      *postgres.DB
+	columns *lru.Cache // Cache created columns to minimize db connections
+}
+
+func NewHeaderRepository(db *postgres.DB) *headerRepository {
+	ccs, _ := lru.New(columnCacheSize)
+	return &headerRepository{
+		db:      db,
+		columns: ccs,
+	}
+}
+
+// Adds a checked_header column for the provided column id
+func (r *headerRepository) AddCheckColumn(id string) error {
+	// Check cache to see if column already exists before querying pg
+	_, ok := r.columns.Get(id)
+	if ok {
+		return nil
+	}
+
+	pgStr := "ALTER TABLE public.checked_headers ADD COLUMN IF NOT EXISTS "
+	pgStr = pgStr + id + " INTEGER NOT NULL DEFAULT 0"
+	_, err := r.db.Exec(pgStr)
+	if err != nil {
+		return err
+	}
+
+	// Add column name to cache
+	r.columns.Add(id, true)
+
+	return nil
+}
+
+// Adds a checked_header column for all of the provided column ids
+func (r *headerRepository) AddCheckColumns(ids []string) error {
+	var err error
+	baseQuery := "ALTER TABLE public.checked_headers"
+	input := make([]string, 0, len(ids))
+	for _, id := range ids {
+		_, ok := r.columns.Get(id)
+		if !ok {
+			baseQuery += " ADD COLUMN IF NOT EXISTS " + id + " INTEGER NOT NULL DEFAULT 0,"
+			input = append(input, id)
+		}
+	}
+	if len(input) > 0 {
+		_, err = r.db.Exec(baseQuery[:len(baseQuery)-1])
+		if err == nil {
+			for _, id := range input {
+				r.columns.Add(id, true)
+			}
+		}
+	}
+
+	return err
+}
+
+// Marks the header checked for the provided column id
+func (r *headerRepository) MarkHeaderChecked(headerID int64, id string) error {
+	_, err := r.db.Exec(`INSERT INTO public.checked_headers (header_id, `+id+`)
+		VALUES ($1, $2) 
+		ON CONFLICT (header_id) DO
+			UPDATE SET `+id+` = checked_headers.`+id+` + 1`, headerID, 1)
+	return err
+}
+
+// Marks the header checked for all of the provided column ids
+func (r *headerRepository) MarkHeaderCheckedForAll(headerID int64, ids []string) error {
+	pgStr := "INSERT INTO public.checked_headers (header_id, "
+	for _, id := range ids {
+		pgStr += id + ", "
+	}
+	pgStr = pgStr[:len(pgStr)-2] + ") VALUES ($1, "
+	for i := 0; i < len(ids); i++ {
+		pgStr += "1, "
+	}
+	pgStr = pgStr[:len(pgStr)-2] + ") ON CONFLICT (header_id) DO UPDATE SET "
+	for _, id := range ids {
+		pgStr += id + `= checked_headers.` + id + ` + 1, `
+	}
+	pgStr = pgStr[:len(pgStr)-2]
+	_, err := r.db.Exec(pgStr, headerID)
+	return err
+}
+
+// Marks all of the provided headers checked for each of the provided column ids
+func (r *headerRepository) MarkHeadersCheckedForAll(headers []core.Header, ids []string) error {
+	tx, err := r.db.Beginx()
+	if err != nil {
+		return err
+	}
+	for _, header := range headers {
+		pgStr := "INSERT INTO public.checked_headers (header_id, "
+		for _, id := range ids {
+			pgStr += id + ", "
+		}
+		pgStr = pgStr[:len(pgStr)-2] + ") VALUES ($1, "
+		for i := 0; i < len(ids); i++ {
+			pgStr += "1, "
+		}
+		pgStr = pgStr[:len(pgStr)-2] + ") ON CONFLICT (header_id) DO UPDATE SET "
+		for _, id := range ids {
+			pgStr += fmt.Sprintf("%s = checked_headers.%s + 1, ", id, id)
+		}
+		pgStr = pgStr[:len(pgStr)-2]
+		_, err = tx.Exec(pgStr, header.Id)
+		if err != nil {
+			tx.Rollback()
+			return err
+		}
+	}
+	err = tx.Commit()
+	return err
+}
+
+// Returns missing headers for the provided checked_headers column id
+func (r *headerRepository) MissingHeaders(startingBlockNumber, endingBlockNumber int64, id 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 (header_id ISNULL OR checked_headers.` + id + `=0)
+				AND headers.block_number >= $1
+				AND headers.eth_node_fingerprint = $2
+				ORDER BY headers.block_number`
+		err = r.db.Select(&result, query, startingBlockNumber, r.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 checked_headers.` + id + `=0)
+				AND headers.block_number >= $1
+				AND headers.block_number <= $2
+				AND headers.eth_node_fingerprint = $3
+				ORDER BY headers.block_number`
+		err = r.db.Select(&result, query, startingBlockNumber, endingBlockNumber, r.db.Node.ID)
+	}
+	return contiguousHeaders(result, startingBlockNumber), err
+}
+
+// Returns missing headers for all of the provided checked_headers column ids
+func (r *headerRepository) MissingHeadersForAll(startingBlockNumber, endingBlockNumber int64, ids []string) ([]core.Header, error) {
+	var result []core.Header
+	var query string
+	var err error
+	baseQuery := `SELECT headers.id, headers.block_number, headers.hash FROM headers
+				  LEFT JOIN checked_headers on headers.id = header_id
+				  WHERE (header_id ISNULL`
+	for _, id := range ids {
+		baseQuery += ` OR checked_headers.` + id + `= 0`
+	}
+	if endingBlockNumber == -1 {
+		endStr := `) AND headers.block_number >= $1
+				  AND headers.eth_node_fingerprint = $2
+				  ORDER BY headers.block_number`
+		query = baseQuery + endStr
+		err = r.db.Select(&result, query, startingBlockNumber, r.db.Node.ID)
+	} else {
+		endStr := `) AND headers.block_number >= $1
+				  AND headers.block_number <= $2
+				  AND headers.eth_node_fingerprint = $3
+				  ORDER BY headers.block_number`
+		query = baseQuery + endStr
+		err = r.db.Select(&result, query, startingBlockNumber, endingBlockNumber, r.db.Node.ID)
+	}
+	return contiguousHeaders(result, startingBlockNumber), err
+}
+
+// Returns a continuous set of headers that is contiguous with the provided startingBlockNumber
+func contiguousHeaders(headers []core.Header, startingBlockNumber int64) []core.Header {
+	if len(headers) < 1 {
+		return headers
+	}
+	previousHeader := headers[0].BlockNumber
+	if previousHeader != startingBlockNumber {
+		return []core.Header{}
+	}
+	for i := 1; i < len(headers); i++ {
+		previousHeader++
+		if headers[i].BlockNumber != previousHeader {
+			return headers[:i]
+		}
+	}
+
+	return headers
+}
+
+// Returns headers that have been checked for all of the provided event ids but not for the provided method ids
+func (r *headerRepository) MissingMethodsCheckedEventsIntersection(startingBlockNumber, endingBlockNumber int64, methodIds, eventIds []string) ([]core.Header, error) {
+	var result []core.Header
+	var query string
+	var err error
+	baseQuery := `SELECT headers.id, headers.block_number, headers.hash FROM headers
+				  LEFT JOIN checked_headers on headers.id = header_id
+				  WHERE (header_id IS NOT NULL`
+	for _, id := range eventIds {
+		baseQuery += ` AND ` + id + `!=0`
+	}
+	baseQuery += `) AND (`
+	for _, id := range methodIds {
+		baseQuery += id + ` =0 AND `
+	}
+	baseQuery = baseQuery[:len(baseQuery)-5] + `) `
+	if endingBlockNumber == -1 {
+		endStr := `AND headers.block_number >= $1
+				  AND headers.eth_node_fingerprint = $2
+				  ORDER BY headers.block_number`
+		query = baseQuery + endStr
+		err = r.db.Select(&result, query, startingBlockNumber, r.db.Node.ID)
+	} else {
+		endStr := `AND headers.block_number >= $1
+				  AND headers.block_number <= $2
+				  AND headers.eth_node_fingerprint = $3
+				  ORDER BY headers.block_number`
+		query = baseQuery + endStr
+		err = r.db.Select(&result, query, startingBlockNumber, endingBlockNumber, r.db.Node.ID)
+	}
+	return continuousHeaders(result), err
+}
+
+// Returns a continuous set of headers
+func continuousHeaders(headers []core.Header) []core.Header {
+	if len(headers) < 1 {
+		return headers
+	}
+	previousHeader := headers[0].BlockNumber
+	for i := 1; i < len(headers); i++ {
+		previousHeader++
+		if headers[i].BlockNumber != previousHeader {
+			return headers[:i]
+		}
+	}
+
+	return headers
+}
+
+// Check the repositories column id cache for a value
+func (r *headerRepository) CheckCache(key string) (interface{}, bool) {
+	return r.columns.Get(key)
+}
+
+// Used to mark a header checked as part of some external transaction so as to group into one commit
+func (r *headerRepository) MarkHeaderCheckedInTransaction(headerID int64, tx *sqlx.Tx, eventID string) error {
+	_, err := tx.Exec(`INSERT INTO public.checked_headers (header_id, `+eventID+`)
+		VALUES ($1, $2) 
+		ON CONFLICT (header_id) DO
+			UPDATE SET `+eventID+` = checked_headers.`+eventID+` + 1`, headerID, 1)
+	return err
+}

pkg/contract_watcher/header/repository/header_repository_test.go

@@ -0,0 +1,347 @@
+// 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 repository_test
+
+import (
+	"fmt"
+	"github.com/vulcanize/vulcanizedb/pkg/core"
+
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+
+	"github.com/vulcanize/vulcanizedb/pkg/contract_watcher/header/repository"
+	"github.com/vulcanize/vulcanizedb/pkg/contract_watcher/shared/helpers/test_helpers"
+	"github.com/vulcanize/vulcanizedb/pkg/contract_watcher/shared/helpers/test_helpers/mocks"
+	"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
+	"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres/repositories"
+)
+
+var _ = Describe("Repository", func() {
+	var db *postgres.DB
+	var contractHeaderRepo repository.HeaderRepository // contract_watcher headerSync header repository
+	var coreHeaderRepo repositories.HeaderRepository   // pkg/datastore header repository
+	var eventIDs = []string{
+		"eventName_contractAddr",
+		"eventName_contractAddr2",
+		"eventName_contractAddr3",
+	}
+	var methodIDs = []string{
+		"methodName_contractAddr",
+		"methodName_contractAddr2",
+		"methodName_contractAddr3",
+	}
+
+	BeforeEach(func() {
+		db, _ = test_helpers.SetupDBandBC()
+		contractHeaderRepo = repository.NewHeaderRepository(db)
+		coreHeaderRepo = repositories.NewHeaderRepository(db)
+	})
+
+	AfterEach(func() {
+		test_helpers.TearDown(db)
+	})
+
+	Describe("AddCheckColumn", func() {
+		It("Creates a column for the given eventID to mark if the header has been checked for that event", func() {
+			query := fmt.Sprintf("SELECT %s FROM checked_headers", eventIDs[0])
+			_, err := db.Exec(query)
+			Expect(err).To(HaveOccurred())
+
+			err = contractHeaderRepo.AddCheckColumn(eventIDs[0])
+			Expect(err).ToNot(HaveOccurred())
+
+			_, err = db.Exec(query)
+			Expect(err).ToNot(HaveOccurred())
+		})
+
+		It("Caches column it creates so that it does not need to repeatedly query the database to check for it's existence", func() {
+			_, ok := contractHeaderRepo.CheckCache(eventIDs[0])
+			Expect(ok).To(Equal(false))
+
+			err := contractHeaderRepo.AddCheckColumn(eventIDs[0])
+			Expect(err).ToNot(HaveOccurred())
+
+			v, ok := contractHeaderRepo.CheckCache(eventIDs[0])
+			Expect(ok).To(Equal(true))
+			Expect(v).To(Equal(true))
+		})
+	})
+
+	Describe("AddCheckColumns", func() {
+		It("Creates a column for the given eventIDs to mark if the header has been checked for those events", func() {
+			for _, id := range eventIDs {
+				_, err := db.Exec(fmt.Sprintf("SELECT %s FROM checked_headers", id))
+				Expect(err).To(HaveOccurred())
+			}
+
+			err := contractHeaderRepo.AddCheckColumns(eventIDs)
+			Expect(err).ToNot(HaveOccurred())
+
+			for _, id := range eventIDs {
+				_, err := db.Exec(fmt.Sprintf("SELECT %s FROM checked_headers", id))
+				Expect(err).ToNot(HaveOccurred())
+			}
+		})
+
+		It("Caches columns it creates so that it does not need to repeatedly query the database to check for it's existence", func() {
+			for _, id := range eventIDs {
+				_, ok := contractHeaderRepo.CheckCache(id)
+				Expect(ok).To(Equal(false))
+			}
+
+			err := contractHeaderRepo.AddCheckColumns(eventIDs)
+			Expect(err).ToNot(HaveOccurred())
+
+			for _, id := range eventIDs {
+				v, ok := contractHeaderRepo.CheckCache(id)
+				Expect(ok).To(Equal(true))
+				Expect(v).To(Equal(true))
+			}
+		})
+	})
+
+	Describe("MissingHeaders", func() {
+		It("Returns all unchecked headers for the given eventID", func() {
+			addHeaders(coreHeaderRepo)
+			err := contractHeaderRepo.AddCheckColumn(eventIDs[0])
+			Expect(err).ToNot(HaveOccurred())
+
+			missingHeaders, err := contractHeaderRepo.MissingHeaders(6194632, 6194635, eventIDs[0])
+			Expect(err).ToNot(HaveOccurred())
+			Expect(len(missingHeaders)).To(Equal(3))
+		})
+
+		It("Returns unchecked headers in ascending order", func() {
+			addHeaders(coreHeaderRepo)
+			err := contractHeaderRepo.AddCheckColumn(eventIDs[0])
+			Expect(err).ToNot(HaveOccurred())
+
+			missingHeaders, err := contractHeaderRepo.MissingHeaders(6194632, 6194635, eventIDs[0])
+			Expect(err).ToNot(HaveOccurred())
+			Expect(len(missingHeaders)).To(Equal(3))
+
+			h1 := missingHeaders[0]
+			h2 := missingHeaders[1]
+			h3 := missingHeaders[2]
+			Expect(h1.BlockNumber).To(Equal(int64(6194632)))
+			Expect(h2.BlockNumber).To(Equal(int64(6194633)))
+			Expect(h3.BlockNumber).To(Equal(int64(6194634)))
+		})
+
+		It("Returns only contiguous chunks of headers", func() {
+			addDiscontinuousHeaders(coreHeaderRepo)
+			err := contractHeaderRepo.AddCheckColumns(eventIDs)
+			Expect(err).ToNot(HaveOccurred())
+
+			missingHeaders, err := contractHeaderRepo.MissingHeaders(6194632, 6194635, eventIDs[0])
+			Expect(err).ToNot(HaveOccurred())
+			Expect(len(missingHeaders)).To(Equal(2))
+			Expect(missingHeaders[0].BlockNumber).To(Equal(int64(6194632)))
+			Expect(missingHeaders[1].BlockNumber).To(Equal(int64(6194633)))
+		})
+
+		It("Fails if eventID does not yet exist in check_headers table", func() {
+			addHeaders(coreHeaderRepo)
+			err := contractHeaderRepo.AddCheckColumn(eventIDs[0])
+			Expect(err).ToNot(HaveOccurred())
+
+			_, err = contractHeaderRepo.MissingHeaders(6194632, 6194635, "notEventId")
+			Expect(err).To(HaveOccurred())
+		})
+	})
+
+	Describe("MissingHeadersForAll", func() { // HERE
+		It("Returns all headers that have not been checked for all of the ids provided", func() {
+			addHeaders(coreHeaderRepo)
+			err := contractHeaderRepo.AddCheckColumns(eventIDs)
+			Expect(err).ToNot(HaveOccurred())
+
+			missingHeaders, err := contractHeaderRepo.MissingHeadersForAll(6194632, 6194635, eventIDs)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(len(missingHeaders)).To(Equal(3))
+
+			err = contractHeaderRepo.MarkHeaderChecked(missingHeaders[0].Id, eventIDs[0])
+			Expect(err).ToNot(HaveOccurred())
+
+			missingHeaders, err = contractHeaderRepo.MissingHeadersForAll(6194632, 6194635, eventIDs)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(len(missingHeaders)).To(Equal(3))
+
+			err = contractHeaderRepo.MarkHeaderChecked(missingHeaders[0].Id, eventIDs[1])
+			Expect(err).ToNot(HaveOccurred())
+			err = contractHeaderRepo.MarkHeaderChecked(missingHeaders[0].Id, eventIDs[2])
+			Expect(err).ToNot(HaveOccurred())
+
+			missingHeaders, err = contractHeaderRepo.MissingHeadersForAll(6194633, 6194635, eventIDs)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(len(missingHeaders)).To(Equal(2))
+		})
+
+		It("Returns only contiguous chunks of headers", func() {
+			addDiscontinuousHeaders(coreHeaderRepo)
+			err := contractHeaderRepo.AddCheckColumns(eventIDs)
+			Expect(err).ToNot(HaveOccurred())
+
+			missingHeaders, err := contractHeaderRepo.MissingHeadersForAll(6194632, 6194635, eventIDs)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(len(missingHeaders)).To(Equal(2))
+			Expect(missingHeaders[0].BlockNumber).To(Equal(int64(6194632)))
+			Expect(missingHeaders[1].BlockNumber).To(Equal(int64(6194633)))
+		})
+
+		It("Fails if one of the eventIDs does not yet exist in check_headers table", func() {
+			addHeaders(coreHeaderRepo)
+			err := contractHeaderRepo.AddCheckColumns(eventIDs)
+			Expect(err).ToNot(HaveOccurred())
+			badEventIDs := append(eventIDs, "notEventId")
+
+			_, err = contractHeaderRepo.MissingHeadersForAll(6194632, 6194635, badEventIDs)
+			Expect(err).To(HaveOccurred())
+		})
+	})
+
+	Describe("MarkHeaderChecked", func() {
+		It("Marks the header checked for the given eventID", func() {
+			addHeaders(coreHeaderRepo)
+			err := contractHeaderRepo.AddCheckColumn(eventIDs[0])
+			Expect(err).ToNot(HaveOccurred())
+
+			missingHeaders, err := contractHeaderRepo.MissingHeaders(6194632, 6194635, eventIDs[0])
+			Expect(err).ToNot(HaveOccurred())
+			Expect(len(missingHeaders)).To(Equal(3))
+
+			headerID := missingHeaders[0].Id
+			err = contractHeaderRepo.MarkHeaderChecked(headerID, eventIDs[0])
+			Expect(err).ToNot(HaveOccurred())
+
+			missingHeaders, err = contractHeaderRepo.MissingHeaders(6194633, 6194635, eventIDs[0])
+			Expect(err).ToNot(HaveOccurred())
+			Expect(len(missingHeaders)).To(Equal(2))
+		})
+
+		It("Fails if eventID does not yet exist in check_headers table", func() {
+			addHeaders(coreHeaderRepo)
+			err := contractHeaderRepo.AddCheckColumn(eventIDs[0])
+			Expect(err).ToNot(HaveOccurred())
+
+			missingHeaders, err := contractHeaderRepo.MissingHeaders(6194632, 6194635, eventIDs[0])
+			Expect(err).ToNot(HaveOccurred())
+			Expect(len(missingHeaders)).To(Equal(3))
+
+			headerID := missingHeaders[0].Id
+			err = contractHeaderRepo.MarkHeaderChecked(headerID, "notEventId")
+			Expect(err).To(HaveOccurred())
+
+			missingHeaders, err = contractHeaderRepo.MissingHeaders(6194632, 6194635, eventIDs[0])
+			Expect(err).ToNot(HaveOccurred())
+			Expect(len(missingHeaders)).To(Equal(3))
+		})
+	})
+
+	Describe("MarkHeaderCheckedForAll", func() {
+		It("Marks the header checked for all provided column ids", func() {
+			addHeaders(coreHeaderRepo)
+			err := contractHeaderRepo.AddCheckColumns(eventIDs)
+			Expect(err).ToNot(HaveOccurred())
+
+			missingHeaders, err := contractHeaderRepo.MissingHeadersForAll(6194632, 6194635, eventIDs)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(len(missingHeaders)).To(Equal(3))
+
+			headerID := missingHeaders[0].Id
+			err = contractHeaderRepo.MarkHeaderCheckedForAll(headerID, eventIDs)
+			Expect(err).ToNot(HaveOccurred())
+
+			missingHeaders, err = contractHeaderRepo.MissingHeaders(6194633, 6194635, eventIDs[0])
+			Expect(err).ToNot(HaveOccurred())
+			Expect(len(missingHeaders)).To(Equal(2))
+		})
+	})
+
+	Describe("MarkHeadersCheckedForAll", func() {
+		It("Marks the headers checked for all provided column ids", func() {
+			addHeaders(coreHeaderRepo)
+			methodIDs := []string{
+				"methodName_contractAddr",
+				"methodName_contractAddr2",
+				"methodName_contractAddr3",
+			}
+
+			var missingHeaders []core.Header
+			for _, id := range methodIDs {
+				err := contractHeaderRepo.AddCheckColumn(id)
+				Expect(err).ToNot(HaveOccurred())
+				missingHeaders, err = contractHeaderRepo.MissingHeaders(6194632, 6194635, id)
+				Expect(err).ToNot(HaveOccurred())
+				Expect(len(missingHeaders)).To(Equal(3))
+			}
+
+			err := contractHeaderRepo.MarkHeadersCheckedForAll(missingHeaders, methodIDs)
+			Expect(err).ToNot(HaveOccurred())
+			for _, id := range methodIDs {
+				missingHeaders, err = contractHeaderRepo.MissingHeaders(6194632, 6194635, id)
+				Expect(err).ToNot(HaveOccurred())
+				Expect(len(missingHeaders)).To(Equal(0))
+			}
+		})
+	})
+
+	Describe("MissingMethodsCheckedEventsIntersection", func() {
+		It("Returns headers that have been checked for all the provided events but have not been checked for all the provided methods", func() {
+			addHeaders(coreHeaderRepo)
+			for i, id := range eventIDs {
+				err := contractHeaderRepo.AddCheckColumn(id)
+				Expect(err).ToNot(HaveOccurred())
+				err = contractHeaderRepo.AddCheckColumn(methodIDs[i])
+				Expect(err).ToNot(HaveOccurred())
+			}
+
+			missingHeaders, err := contractHeaderRepo.MissingHeaders(6194632, 6194635, eventIDs[0])
+			Expect(err).ToNot(HaveOccurred())
+			Expect(len(missingHeaders)).To(Equal(3))
+
+			headerID := missingHeaders[0].Id
+			headerID2 := missingHeaders[1].Id
+			for i, id := range eventIDs {
+				err = contractHeaderRepo.MarkHeaderChecked(headerID, id)
+				Expect(err).ToNot(HaveOccurred())
+				err = contractHeaderRepo.MarkHeaderChecked(headerID2, id)
+				Expect(err).ToNot(HaveOccurred())
+				err = contractHeaderRepo.MarkHeaderChecked(headerID, methodIDs[i])
+				Expect(err).ToNot(HaveOccurred())
+			}
+
+			intersectionHeaders, err := contractHeaderRepo.MissingMethodsCheckedEventsIntersection(6194632, 6194635, methodIDs, eventIDs)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(len(intersectionHeaders)).To(Equal(1))
+			Expect(intersectionHeaders[0].Id).To(Equal(headerID2))
+
+		})
+	})
+})
+
+func addHeaders(coreHeaderRepo repositories.HeaderRepository) {
+	coreHeaderRepo.CreateOrUpdateHeader(mocks.MockHeader1)
+	coreHeaderRepo.CreateOrUpdateHeader(mocks.MockHeader2)
+	coreHeaderRepo.CreateOrUpdateHeader(mocks.MockHeader3)
+}
+
+func addDiscontinuousHeaders(coreHeaderRepo repositories.HeaderRepository) {
+	coreHeaderRepo.CreateOrUpdateHeader(mocks.MockHeader1)
+	coreHeaderRepo.CreateOrUpdateHeader(mocks.MockHeader2)
+	coreHeaderRepo.CreateOrUpdateHeader(mocks.MockHeader4)
+}

pkg/contract_watcher/header/repository/repository_suite_test.go

@@ -0,0 +1,35 @@
+// 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 repository_test
+
+import (
+	"github.com/sirupsen/logrus"
+	"io/ioutil"
+	"testing"
+
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+)
+
+func TestRepository(t *testing.T) {
+	RegisterFailHandler(Fail)
+	RunSpecs(t, "Header Sync Repository Suite Test")
+}
+
+var _ = BeforeSuite(func() {
+	logrus.SetOutput(ioutil.Discard)
+})