reorganizing omni directory and beginning light watcher work

pkg/core/blockchain.go

@@ -16,13 +16,19 @@
 
 package core
 
-import "math/big"
+import (
+	"math/big"
+
+	"github.com/ethereum/go-ethereum"
+	"github.com/ethereum/go-ethereum/core/types"
+)
 
 type BlockChain interface {
 	ContractDataFetcher
 	GetBlockByNumber(blockNumber int64) (Block, error)
 	GetHeaderByNumber(blockNumber int64) (Header, error)
 	GetLogs(contract Contract, startingBlockNumber *big.Int, endingBlockNumber *big.Int) ([]Log, error)
+	GetEthLogsWithCustomQuery(query ethereum.FilterQuery) ([]types.Log, error)
 	LastBlock() *big.Int
 	Node() Node
 }

pkg/crypto/parser.go

@@ -18,7 +18,7 @@ package crypto
 
 import (
 	"github.com/ethereum/go-ethereum/crypto"
-	"github.com/ethereum/go-ethereum/p2p/discover"
+	"github.com/ethereum/go-ethereum/p2p/discv5"
 )
 
 type PublicKeyParser interface {
@@ -32,6 +32,6 @@ func (EthPublicKeyParser) ParsePublicKey(privateKey string) (string, error) {
 	if err != nil {
 		return "", err
 	}
-	pubKey := discover.PubkeyID(&np.PublicKey)
+	pubKey := discv5.PubkeyID(&np.PublicKey)
 	return pubKey.String(), nil
 }

pkg/datastore/ethereum/level/database.go

@@ -52,5 +52,5 @@ func (l LevelDatabase) GetBlockReceipts(blockHash []byte, blockNumber int64) typ
 func (l LevelDatabase) GetHeadBlockNumber() int64 {
 	h := l.reader.GetHeadBlockHash()
 	n := l.reader.GetBlockNumber(h)
-	return int64(n)
+	return int64(*n)
 }

pkg/datastore/ethereum/level/database_reader.go

@@ -18,42 +18,42 @@ package level
 
 import (
 	"github.com/ethereum/go-ethereum/common"
-	"github.com/ethereum/go-ethereum/core"
+	"github.com/ethereum/go-ethereum/core/rawdb"
 	"github.com/ethereum/go-ethereum/core/types"
 )
 
 type Reader interface {
 	GetBlock(hash common.Hash, number uint64) *types.Block
-	GetBlockNumber(hash common.Hash) uint64
+	GetBlockNumber(hash common.Hash) *uint64
 	GetBlockReceipts(hash common.Hash, number uint64) types.Receipts
 	GetCanonicalHash(number uint64) common.Hash
 	GetHeadBlockHash() common.Hash
 }
 
 type LevelDatabaseReader struct {
-	reader core.DatabaseReader
+	reader rawdb.DatabaseReader
 }
 
-func NewLevelDatabaseReader(reader core.DatabaseReader) *LevelDatabaseReader {
+func NewLevelDatabaseReader(reader rawdb.DatabaseReader) *LevelDatabaseReader {
 	return &LevelDatabaseReader{reader: reader}
 }
 
 func (ldbr *LevelDatabaseReader) GetBlock(hash common.Hash, number uint64) *types.Block {
-	return core.GetBlock(ldbr.reader, hash, number)
+	return rawdb.ReadBlock(ldbr.reader, hash, number)
 }
 
-func (ldbr *LevelDatabaseReader) GetBlockNumber(hash common.Hash) uint64 {
-	return core.GetBlockNumber(ldbr.reader, hash)
+func (ldbr *LevelDatabaseReader) GetBlockNumber(hash common.Hash) *uint64 {
+	return rawdb.ReadHeaderNumber(ldbr.reader, hash)
 }
 
 func (ldbr *LevelDatabaseReader) GetBlockReceipts(hash common.Hash, number uint64) types.Receipts {
-	return core.GetBlockReceipts(ldbr.reader, hash, number)
+	return rawdb.ReadReceipts(ldbr.reader, hash, number)
 }
 
 func (ldbr *LevelDatabaseReader) GetCanonicalHash(number uint64) common.Hash {
-	return core.GetCanonicalHash(ldbr.reader, number)
+	return rawdb.ReadCanonicalHash(ldbr.reader, number)
 }
 
 func (ldbr *LevelDatabaseReader) GetHeadBlockHash() common.Hash {
-	return core.GetHeadBlockHash(ldbr.reader)
+	return rawdb.ReadHeadBlockHash(ldbr.reader)
 }

pkg/fakes/mock_blockchain.go

@@ -19,6 +19,8 @@ package fakes
 import (
 	"math/big"
 
+	"github.com/ethereum/go-ethereum"
+	"github.com/ethereum/go-ethereum/core/types"
 	. "github.com/onsi/gomega"
 
 	"github.com/vulcanize/vulcanizedb/pkg/core"
@@ -33,6 +35,9 @@ type MockBlockChain struct {
 	fetchContractDataPassedResult      interface{}
 	fetchContractDataPassedBlockNumber int64
 	getBlockByNumberErr                error
+	logQuery                           ethereum.FilterQuery
+	logQueryErr                        error
+	logQueryReturnLogs                 []types.Log
 	lastBlock                          *big.Int
 	node                               core.Node
 }
@@ -55,6 +60,14 @@ func (blockChain *MockBlockChain) SetGetBlockByNumberErr(err error) {
 	blockChain.getBlockByNumberErr = err
 }
 
+func (chain *MockBlockChain) SetGetEthLogsWithCustomQueryErr(err error) {
+	chain.logQueryErr = err
+}
+
+func (chain *MockBlockChain) SetGetEthLogsWithCustomQueryReturnLogs(logs []types.Log) {
+	chain.logQueryReturnLogs = logs
+}
+
 func (blockChain *MockBlockChain) GetHeaderByNumber(blockNumber int64) (core.Header, error) {
 	return core.Header{BlockNumber: blockNumber}, nil
 }
@@ -69,31 +82,42 @@ func (blockChain *MockBlockChain) FetchContractData(abiJSON string, address stri
 	return blockChain.fetchContractDataErr
 }
 
-func (blockChain *MockBlockChain) CallContract(contractHash string, input []byte, blockNumber *big.Int) ([]byte, error) {
-	return []byte{}, nil
+func (chain *MockBlockChain) GetBlockByNumber(blockNumber int64) (core.Block, error) {
+	return core.Block{Number: blockNumber}, chain.getBlockByNumberErr
 }
 
-func (blockChain *MockBlockChain) LastBlock() *big.Int {
-	return blockChain.lastBlock
+func (blockChain *MockBlockChain) GetEthLogsWithCustomQuery(query ethereum.FilterQuery) ([]types.Log, error) {
+	blockChain.logQuery = query
+	return blockChain.logQueryReturnLogs, blockChain.logQueryErr
 }
 
-func (blockChain *MockBlockChain) GetLogs(contract core.Contract, startingBlock *big.Int, endingBlock *big.Int) ([]core.Log, error) {
+func (chain *MockBlockChain) GetLogs(contract core.Contract, startingBlockNumber, endingBlockNumber *big.Int) ([]core.Log, error) {
 	return []core.Log{}, nil
 }
 
-func (blockChain *MockBlockChain) Node() core.Node {
-	return blockChain.node
+func (chain *MockBlockChain) CallContract(contractHash string, input []byte, blockNumber *big.Int) ([]byte, error) {
+	return []byte{}, nil
 }
 
-func (blockChain *MockBlockChain) GetBlockByNumber(blockNumber int64) (core.Block, error) {
-	return core.Block{Number: blockNumber}, blockChain.getBlockByNumberErr
+func (chain *MockBlockChain) LastBlock() *big.Int {
+	return chain.lastBlock
 }
 
-// TODO: handle methodArg being nil (can't match nil to nil in Gomega)
-func (blockChain *MockBlockChain) AssertFetchContractDataCalledWith(abiJSON string, address string, method string, methodArgs []interface{}, result interface{}, blockNumber int64) {
-	Expect(blockChain.fetchContractDataPassedAbi).To(Equal(abiJSON))
-	Expect(blockChain.fetchContractDataPassedAddress).To(Equal(address))
-	Expect(blockChain.fetchContractDataPassedMethod).To(Equal(method))
-	Expect(blockChain.fetchContractDataPassedResult).To(Equal(result))
-	Expect(blockChain.fetchContractDataPassedBlockNumber).To(Equal(blockNumber))
+func (chain *MockBlockChain) Node() core.Node {
+	return chain.node
+}
+
+func (chain *MockBlockChain) AssertFetchContractDataCalledWith(abiJSON string, address string, method string, methodArgs []interface{}, result interface{}, blockNumber int64) {
+	Expect(chain.fetchContractDataPassedAbi).To(Equal(abiJSON))
+	Expect(chain.fetchContractDataPassedAddress).To(Equal(address))
+	Expect(chain.fetchContractDataPassedMethod).To(Equal(method))
+	if methodArgs != nil {
+		Expect(chain.fetchContractDataPassedMethodArgs).To(Equal(methodArgs))
+	}
+	Expect(chain.fetchContractDataPassedResult).To(BeAssignableToTypeOf(result))
+	Expect(chain.fetchContractDataPassedBlockNumber).To(Equal(blockNumber))
+}
+
+func (blockChain *MockBlockChain) AssertGetEthLogsWithCustomQueryCalledWith(query ethereum.FilterQuery) {
+	Expect(blockChain.logQuery).To(Equal(query))
 }

pkg/fakes/mock_level_database_reader.go

@@ -98,10 +98,10 @@ func (mldr *MockLevelDatabaseReader) GetBlockReceipts(hash common.Hash, number u
 	return mldr.returnReceipts
 }
 
-func (mldr *MockLevelDatabaseReader) GetBlockNumber(hash common.Hash) uint64 {
+func (mldr *MockLevelDatabaseReader) GetBlockNumber(hash common.Hash) *uint64 {
 	mldr.getBlockNumberCalled = true
 	mldr.getBlockNumberPassedHash = hash
-	return mldr.returnBlockNumber
+	return &mldr.returnBlockNumber
 }
 
 func (mldr *MockLevelDatabaseReader) GetCanonicalHash(number uint64) common.Hash {

pkg/geth/converters/common/block_converter_test.go

@@ -234,7 +234,7 @@ var _ = Describe("Conversion of GethBlock to core.Block", func() {
 				CumulativeGasUsed: uint64(7996119),
 				GasUsed:           uint64(21000),
 				Logs:              []*types.Log{},
-				Status:            uint(1),
+				Status:            uint64(1),
 				TxHash:            gethTransaction.Hash(),
 			}
 

pkg/geth/converters/common/receipt_converter_test.go

@@ -70,7 +70,7 @@ var _ = Describe("Conversion of GethReceipt to core.Receipt", func() {
 			CumulativeGasUsed: uint64(7996119),
 			GasUsed:           uint64(21000),
 			Logs:              []*types.Log{},
-			Status:            uint(1),
+			Status:            uint64(1),
 			TxHash:            common.HexToHash("0xe340558980f89d5f86045ac11e5cc34e4bcec20f9f1e2a427aa39d87114e8223"),
 		}
 

pkg/omni/full/converter/converter.go

@@ -18,6 +18,7 @@ package converter
 
 import (
 	"errors"
+	"fmt"
 	"math/big"
 	"strconv"
 
@@ -25,9 +26,9 @@ import (
 	"github.com/ethereum/go-ethereum/common"
 
 	"github.com/vulcanize/vulcanizedb/pkg/core"
-	"github.com/vulcanize/vulcanizedb/pkg/omni/contract"
-	"github.com/vulcanize/vulcanizedb/pkg/omni/helpers"
-	"github.com/vulcanize/vulcanizedb/pkg/omni/types"
+	"github.com/vulcanize/vulcanizedb/pkg/omni/shared/contract"
+	"github.com/vulcanize/vulcanizedb/pkg/omni/shared/helpers"
+	"github.com/vulcanize/vulcanizedb/pkg/omni/shared/types"
 )
 
 // Converter is used to convert watched event logs to
@@ -38,27 +39,22 @@ type Converter interface {
 }
 
 type converter struct {
-	contractInfo *contract.Contract
+	ContractInfo *contract.Contract
 }
 
 func NewConverter(info *contract.Contract) *converter {
-
 	return &converter{
-		contractInfo: info,
+		ContractInfo: info,
 	}
 }
 
 func (c *converter) Update(info *contract.Contract) {
-	c.contractInfo = info
-}
-
-func (c *converter) CheckInfo() *contract.Contract {
-	return c.contractInfo
+	c.ContractInfo = info
 }
 
 // Convert the given watched event log into a types.Log for the given event
 func (c *converter) Convert(watchedEvent core.WatchedEvent, event types.Event) (*types.Log, error) {
-	contract := bind.NewBoundContract(common.HexToAddress(c.contractInfo.Address), c.contractInfo.ParsedAbi, nil, nil, nil)
+	contract := bind.NewBoundContract(common.HexToAddress(c.ContractInfo.Address), c.ContractInfo.ParsedAbi, nil, nil, nil)
 	values := make(map[string]interface{})
 
 	for _, field := range event.Fields {
@@ -85,7 +81,7 @@ func (c *converter) Convert(watchedEvent core.WatchedEvent, event types.Event) (
 			var a common.Address
 			a = input.(common.Address)
 			strValues[fieldName] = a.String()
-			c.contractInfo.AddTokenHolderAddress(a.String()) // cache address in a list of contract's token holder addresses
+			c.ContractInfo.AddTokenHolderAddress(a.String()) // cache address in a list of contract's token holder addresses
 		case common.Hash:
 			var h common.Hash
 			h = input.(common.Hash)
@@ -95,12 +91,12 @@ func (c *converter) Convert(watchedEvent core.WatchedEvent, event types.Event) (
 		case bool:
 			strValues[fieldName] = strconv.FormatBool(input.(bool))
 		default:
-			return nil, errors.New("error: unhandled abi type")
+			return nil, errors.New(fmt.Sprintf("error: unhandled abi type %T", input))
 		}
 	}
 
 	// Only hold onto logs that pass our address filter, if any
-	if c.contractInfo.PassesEventFilter(strValues) {
+	if c.ContractInfo.PassesEventFilter(strValues) {
 		eventLog := &types.Log{
 			Event:  event,
 			Id:     watchedEvent.LogID,
@@ -109,7 +105,7 @@ func (c *converter) Convert(watchedEvent core.WatchedEvent, event types.Event) (
 			Tx:     watchedEvent.TxHash,
 		}
 
-		return eventLog, err
+		return eventLog, nil
 	}
 
 	return nil, nil

pkg/omni/full/converter/converter_suite_test.go

@@ -27,7 +27,7 @@ import (
 
 func TestConverter(t *testing.T) {
 	RegisterFailHandler(Fail)
-	RunSpecs(t, "Converter Suite Test")
+	RunSpecs(t, "Full Converter Suite Test")
 }
 
 var _ = BeforeSuite(func() {

pkg/omni/full/converter/converter_test.go

@@ -21,28 +21,12 @@ import (
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
 
-	"github.com/vulcanize/vulcanizedb/pkg/core"
-	"github.com/vulcanize/vulcanizedb/pkg/omni/constants"
-	"github.com/vulcanize/vulcanizedb/pkg/omni/contract"
-	"github.com/vulcanize/vulcanizedb/pkg/omni/converter"
-	"github.com/vulcanize/vulcanizedb/pkg/omni/helpers"
-	"github.com/vulcanize/vulcanizedb/pkg/omni/helpers/test_helpers"
+	"github.com/vulcanize/vulcanizedb/pkg/omni/full/converter"
+	"github.com/vulcanize/vulcanizedb/pkg/omni/shared/contract"
+	"github.com/vulcanize/vulcanizedb/pkg/omni/shared/helpers"
+	"github.com/vulcanize/vulcanizedb/pkg/omni/shared/helpers/test_helpers"
 )
 
-var mockEvent = core.WatchedEvent{
-	LogID:       1,
-	Name:        constants.TransferEvent.String(),
-	BlockNumber: 5488076,
-	Address:     constants.TusdContractAddress,
-	TxHash:      "0x135391a0962a63944e5908e6fedfff90fb4be3e3290a21017861099bad6546ae",
-	Index:       110,
-	Topic0:      constants.TransferEvent.Signature(),
-	Topic1:      "0x000000000000000000000000000000000000000000000000000000000000af21",
-	Topic2:      "0x9dd48110dcc444fdc242510c09bbbbe21a5975cac061d82f7b843bce061ba391",
-	Topic3:      "",
-	Data:        "0x000000000000000000000000c02aaa39b223fe8d0a0e5c4f27ead9083c756cc200000000000000000000000089d24a6b4ccb1b6faa2625fe562bdd9a23260359000000000000000000000000000000000000000000000000392d2e2bda9c00000000000000000000000000000000000000000000000000927f41fa0a4a418000000000000000000000000000000000000000000000000000000000005adcfebe",
-}
-
 var _ = Describe("Converter", func() {
 	var info *contract.Contract
 	var wantedEvents = []string{"Transfer"}
@@ -55,13 +39,11 @@ var _ = Describe("Converter", func() {
 	Describe("Update", func() {
 		It("Updates contract info held by the converter", func() {
 			c := converter.NewConverter(info)
-			i := c.CheckInfo()
-			Expect(i).To(Equal(info))
+			Expect(c.ContractInfo).To(Equal(info))
 
 			info := test_helpers.SetupTusdContract([]string{}, []string{})
 			c.Update(info)
-			i = c.CheckInfo()
-			Expect(i).To(Equal(info))
+			Expect(c.ContractInfo).To(Equal(info))
 		})
 	})
 
@@ -76,7 +58,7 @@ var _ = Describe("Converter", func() {
 			Expect(err).ToNot(HaveOccurred())
 
 			c := converter.NewConverter(info)
-			log, err := c.Convert(mockEvent, event)
+			log, err := c.Convert(test_helpers.MockTranferEvent, event)
 			Expect(err).ToNot(HaveOccurred())
 
 			from := common.HexToAddress("0x000000000000000000000000000000000000000000000000000000000000af21")
@@ -93,7 +75,7 @@ var _ = Describe("Converter", func() {
 		It("Fails with an empty contract", func() {
 			event := info.Events["Transfer"]
 			c := converter.NewConverter(&contract.Contract{})
-			_, err = c.Convert(mockEvent, event)
+			_, err = c.Convert(test_helpers.MockTranferEvent, event)
 			Expect(err).To(HaveOccurred())
 		})
 	})

pkg/omni/full/repository/event_repository.go

@@ -22,31 +22,31 @@ import (
 	"strings"
 
 	"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
-	"github.com/vulcanize/vulcanizedb/pkg/omni/types"
+	"github.com/vulcanize/vulcanizedb/pkg/omni/shared/types"
 )
 
-// Event datastore is used to persist event data into custom tables
-type EventDatastore interface {
+// Event repository is used to persist event data into custom tables
+type EventRepository interface {
 	PersistLog(event types.Log, contractAddr, contractName string) error
 	CreateEventTable(contractName string, event types.Log) (bool, error)
 	CreateContractSchema(contractName string) (bool, error)
 }
 
-type eventDatastore struct {
-	*postgres.DB
+type eventRepository struct {
+	db *postgres.DB
 }
 
-func NewEventDataStore(db *postgres.DB) *eventDatastore {
+func NewEventRepository(db *postgres.DB) *eventRepository {
 
-	return &eventDatastore{
-		DB: db,
+	return &eventRepository{
+		db: db,
 	}
 }
 
 // Creates a schema for the contract if needed
 // Creates table for the watched contract event if needed
 // Persists converted event log data into this custom table
-func (d *eventDatastore) PersistLog(event types.Log, contractAddr, contractName string) error {
+func (d *eventRepository) PersistLog(event types.Log, contractAddr, contractName string) error {
 	_, err := d.CreateContractSchema(contractAddr)
 	if err != nil {
 		return err
@@ -61,7 +61,7 @@ func (d *eventDatastore) PersistLog(event types.Log, contractAddr, contractName
 }
 
 // Creates a custom postgres command to persist logs for the given event
-func (d *eventDatastore) persistLog(event types.Log, contractAddr, contractName string) error {
+func (d *eventRepository) persistLog(event types.Log, contractAddr, contractName string) error {
 	// Begin postgres string
 	pgStr := fmt.Sprintf("INSERT INTO c%s.%s_event ", strings.ToLower(contractAddr), strings.ToLower(event.Name))
 	pgStr = pgStr + "(vulcanize_log_id, token_name, block, tx"
@@ -91,7 +91,7 @@ func (d *eventDatastore) persistLog(event types.Log, contractAddr, contractName
 	}
 	pgStr = pgStr + ") ON CONFLICT (vulcanize_log_id) DO NOTHING"
 
-	_, err := d.DB.Exec(pgStr, data...)
+	_, err := d.db.Exec(pgStr, data...)
 	if err != nil {
 		return err
 	}
@@ -100,7 +100,7 @@ func (d *eventDatastore) persistLog(event types.Log, contractAddr, contractName
 }
 
 // Checks for event table and creates it if it does not already exist
-func (d *eventDatastore) CreateEventTable(contractAddr string, event types.Log) (bool, error) {
+func (d *eventRepository) CreateEventTable(contractAddr string, event types.Log) (bool, error) {
 	tableExists, err := d.checkForTable(contractAddr, event.Name)
 	if err != nil {
 		return false, err
@@ -117,7 +117,7 @@ func (d *eventDatastore) CreateEventTable(contractAddr string, event types.Log)
 }
 
 // Creates a table for the given contract and event
-func (d *eventDatastore) newEventTable(contractAddr string, event types.Log) error {
+func (d *eventRepository) newEventTable(contractAddr string, event types.Log) error {
 	// Begin pg string
 	pgStr := fmt.Sprintf("CREATE TABLE IF NOT EXISTS c%s.%s_event ", strings.ToLower(contractAddr), strings.ToLower(event.Name))
 	pgStr = pgStr + "(id SERIAL, vulcanize_log_id INTEGER NOT NULL UNIQUE, token_name CHARACTER VARYING(66) NOT NULL, block INTEGER NOT NULL, tx CHARACTER VARYING(66) NOT NULL,"
@@ -128,23 +128,23 @@ func (d *eventDatastore) newEventTable(contractAddr string, event types.Log) err
 	}
 
 	pgStr = pgStr + " CONSTRAINT log_index_fk FOREIGN KEY (vulcanize_log_id) REFERENCES logs (id) ON DELETE CASCADE)"
-	_, err := d.DB.Exec(pgStr)
+	_, err := d.db.Exec(pgStr)
 
 	return err
 }
 
 // Checks if a table already exists for the given contract and event
-func (d *eventDatastore) checkForTable(contractAddr string, eventName string) (bool, error) {
+func (d *eventRepository) checkForTable(contractAddr string, eventName string) (bool, error) {
 	pgStr := fmt.Sprintf("SELECT EXISTS (SELECT 1 FROM information_schema.tables WHERE table_schema = 'c%s' AND table_name = '%s_event')", strings.ToLower(contractAddr), strings.ToLower(eventName))
 
 	var exists bool
-	err := d.DB.Get(&exists, pgStr)
+	err := d.db.Get(&exists, pgStr)
 
 	return exists, err
 }
 
 // Checks for contract schema and creates it if it does not already exist
-func (d *eventDatastore) CreateContractSchema(contractAddr string) (bool, error) {
+func (d *eventRepository) CreateContractSchema(contractAddr string) (bool, error) {
 	if contractAddr == "" {
 		return false, errors.New("error: no contract address specified")
 	}
@@ -165,18 +165,18 @@ func (d *eventDatastore) CreateContractSchema(contractAddr string) (bool, error)
 }
 
 // Creates a schema for the given contract
-func (d *eventDatastore) newContractSchema(contractAddr string) error {
-	_, err := d.DB.Exec("CREATE SCHEMA IF NOT EXISTS c" + strings.ToLower(contractAddr))
+func (d *eventRepository) newContractSchema(contractAddr string) error {
+	_, err := d.db.Exec("CREATE SCHEMA IF NOT EXISTS c" + strings.ToLower(contractAddr))
 
 	return err
 }
 
 // Checks if a schema already exists for the given contract
-func (d *eventDatastore) checkForSchema(contractAddr string) (bool, error) {
+func (d *eventRepository) checkForSchema(contractAddr string) (bool, error) {
 	pgStr := fmt.Sprintf("SELECT EXISTS (SELECT schema_name FROM information_schema.schemata WHERE schema_name = 'c%s')", strings.ToLower(contractAddr))
 
 	var exists bool
-	err := d.DB.QueryRow(pgStr).Scan(&exists)
+	err := d.db.QueryRow(pgStr).Scan(&exists)
 
 	return exists, err
 }

pkg/omni/full/repository/event_repository_test.go

@@ -24,12 +24,12 @@ import (
 
 	"github.com/vulcanize/vulcanizedb/pkg/core"
 	"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
-	"github.com/vulcanize/vulcanizedb/pkg/omni/constants"
-	"github.com/vulcanize/vulcanizedb/pkg/omni/contract"
-	"github.com/vulcanize/vulcanizedb/pkg/omni/converter"
-	"github.com/vulcanize/vulcanizedb/pkg/omni/helpers/test_helpers"
-	"github.com/vulcanize/vulcanizedb/pkg/omni/repository"
-	"github.com/vulcanize/vulcanizedb/pkg/omni/types"
+	"github.com/vulcanize/vulcanizedb/pkg/omni/full/converter"
+	"github.com/vulcanize/vulcanizedb/pkg/omni/full/repository"
+	"github.com/vulcanize/vulcanizedb/pkg/omni/shared/constants"
+	"github.com/vulcanize/vulcanizedb/pkg/omni/shared/contract"
+	"github.com/vulcanize/vulcanizedb/pkg/omni/shared/helpers/test_helpers"
+	"github.com/vulcanize/vulcanizedb/pkg/omni/shared/types"
 )
 
 var mockEvent = core.WatchedEvent{
@@ -47,7 +47,7 @@ var mockEvent = core.WatchedEvent{
 
 var _ = Describe("Repository", func() {
 	var db *postgres.DB
-	var dataStore repository.EventDatastore
+	var dataStore repository.EventRepository
 	var err error
 	var log *types.Log
 	var con *contract.Contract
@@ -67,7 +67,7 @@ var _ = Describe("Repository", func() {
 		log, err = c.Convert(mockEvent, event)
 		Expect(err).ToNot(HaveOccurred())
 
-		dataStore = repository.NewEventDataStore(db)
+		dataStore = repository.NewEventRepository(db)
 	})
 
 	AfterEach(func() {

pkg/omni/full/repository/repository_suite_test.go

@@ -27,7 +27,7 @@ import (
 
 func TestRepository(t *testing.T) {
 	RegisterFailHandler(Fail)
-	RunSpecs(t, "Repository Suite Test")
+	RunSpecs(t, "Full Repository Suite Test")
 }
 
 var _ = BeforeSuite(func() {

pkg/omni/full/retriever/address_retriever.go

@@ -24,7 +24,7 @@ import (
 	"github.com/ethereum/go-ethereum/common"
 
 	"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
-	"github.com/vulcanize/vulcanizedb/pkg/omni/contract"
+	"github.com/vulcanize/vulcanizedb/pkg/omni/shared/contract"
 )
 
 // Address retriever is used to retrieve the addresses associated with a contract
@@ -35,13 +35,12 @@ type AddressRetriever interface {
 }
 
 type addressRetriever struct {
-	*postgres.DB
+	db *postgres.DB
 }
 
 func NewAddressRetriever(db *postgres.DB) (r *addressRetriever) {
-
 	return &addressRetriever{
-		DB: db,
+		db: db,
 	}
 }
 
@@ -86,7 +85,7 @@ func (r *addressRetriever) retrieveTransferAddresses(con contract.Contract) ([]s
 		if field.Type.T == abi.AddressTy { // If they have address type, retrieve those addresses
 			addrs := make([]string, 0)
 			pgStr := fmt.Sprintf("SELECT %s_ FROM c%s.%s_event", strings.ToLower(field.Name), strings.ToLower(con.Address), strings.ToLower(event.Name))
-			err := r.DB.Select(&addrs, pgStr)
+			err := r.db.Select(&addrs, pgStr)
 			if err != nil {
 				return []string{}, err
 			}
@@ -107,7 +106,7 @@ func (r *addressRetriever) retrieveTokenMintees(con contract.Contract) ([]string
 		if field.Type.T == abi.AddressTy { // If they have address type, retrieve those addresses
 			addrs := make([]string, 0)
 			pgStr := fmt.Sprintf("SELECT %s_ FROM c%s.%s_event", strings.ToLower(field.Name), strings.ToLower(con.Address), strings.ToLower(event.Name))
-			err := r.DB.Select(&addrs, pgStr)
+			err := r.db.Select(&addrs, pgStr)
 			if err != nil {
 				return []string{}, err
 			}

pkg/omni/full/retriever/address_retriever_test.go

@@ -20,16 +20,16 @@ import (
 	"github.com/ethereum/go-ethereum/common"
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
-	"github.com/vulcanize/vulcanizedb/pkg/omni/types"
+	"github.com/vulcanize/vulcanizedb/pkg/omni/shared/types"
 
 	"github.com/vulcanize/vulcanizedb/pkg/core"
 	"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
-	"github.com/vulcanize/vulcanizedb/pkg/omni/constants"
-	"github.com/vulcanize/vulcanizedb/pkg/omni/contract"
-	"github.com/vulcanize/vulcanizedb/pkg/omni/converter"
-	"github.com/vulcanize/vulcanizedb/pkg/omni/helpers/test_helpers"
-	"github.com/vulcanize/vulcanizedb/pkg/omni/repository"
-	"github.com/vulcanize/vulcanizedb/pkg/omni/retriever"
+	"github.com/vulcanize/vulcanizedb/pkg/omni/full/converter"
+	"github.com/vulcanize/vulcanizedb/pkg/omni/full/repository"
+	"github.com/vulcanize/vulcanizedb/pkg/omni/full/retriever"
+	"github.com/vulcanize/vulcanizedb/pkg/omni/shared/constants"
+	"github.com/vulcanize/vulcanizedb/pkg/omni/shared/contract"
+	"github.com/vulcanize/vulcanizedb/pkg/omni/shared/helpers/test_helpers"
 )
 
 var mockEvent = core.WatchedEvent{
@@ -47,7 +47,7 @@ var mockEvent = core.WatchedEvent{
 
 var _ = Describe("Address Retriever Test", func() {
 	var db *postgres.DB
-	var dataStore repository.EventDatastore
+	var dataStore repository.EventRepository
 	var err error
 	var info *contract.Contract
 	var vulcanizeLogId int64
@@ -68,7 +68,7 @@ var _ = Describe("Address Retriever Test", func() {
 		log, err = c.Convert(mockEvent, event)
 		Expect(err).ToNot(HaveOccurred())
 
-		dataStore = repository.NewEventDataStore(db)
+		dataStore = repository.NewEventRepository(db)
 		dataStore.PersistLog(*log, info.Address, info.Name)
 		Expect(err).ToNot(HaveOccurred())
 

pkg/omni/full/retriever/block_retriever.go

@@ -28,13 +28,12 @@ type BlockRetriever interface {
 }
 
 type blockRetriever struct {
-	*postgres.DB
+	db *postgres.DB
 }
 
 func NewBlockRetriever(db *postgres.DB) (r *blockRetriever) {
-
 	return &blockRetriever{
-		DB: db,
+		db: db,
 	}
 }
 
@@ -51,7 +50,7 @@ func (r *blockRetriever) RetrieveFirstBlock(contractAddr string) (int64, error)
 // For some contracts the contract creation transaction receipt doesn't have the contract address so this doesn't work (e.g. Sai)
 func (r *blockRetriever) retrieveFirstBlockFromReceipts(contractAddr string) (int64, error) {
 	var firstBlock int
-	err := r.DB.Get(
+	err := r.db.Get(
 		&firstBlock,
 		`SELECT number FROM blocks
 		       WHERE id = (SELECT block_id FROM receipts
@@ -67,7 +66,7 @@ func (r *blockRetriever) retrieveFirstBlockFromReceipts(contractAddr string) (in
 // In which case this servers as a heuristic to find the first block by finding the first contract event log
 func (r *blockRetriever) retrieveFirstBlockFromLogs(contractAddr string) (int64, error) {
 	var firstBlock int
-	err := r.DB.Get(
+	err := r.db.Get(
 		&firstBlock,
 		"SELECT block_number FROM logs WHERE address = $1 ORDER BY block_number ASC LIMIT 1",
 		contractAddr,
@@ -79,7 +78,7 @@ func (r *blockRetriever) retrieveFirstBlockFromLogs(contractAddr string) (int64,
 // Method to retrieve the most recent block in vDB
 func (r *blockRetriever) RetrieveMostRecentBlock() (int64, error) {
 	var lastBlock int64
-	err := r.DB.Get(
+	err := r.db.Get(
 		&lastBlock,
 		"SELECT number FROM blocks ORDER BY number DESC LIMIT 1",
 	)

pkg/omni/full/retriever/block_retriever_test.go

@@ -23,9 +23,9 @@ import (
 	"github.com/vulcanize/vulcanizedb/pkg/core"
 	"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
 	"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres/repositories"
-	"github.com/vulcanize/vulcanizedb/pkg/omni/constants"
-	"github.com/vulcanize/vulcanizedb/pkg/omni/helpers/test_helpers"
-	"github.com/vulcanize/vulcanizedb/pkg/omni/retriever"
+	"github.com/vulcanize/vulcanizedb/pkg/omni/full/retriever"
+	"github.com/vulcanize/vulcanizedb/pkg/omni/shared/constants"
+	"github.com/vulcanize/vulcanizedb/pkg/omni/shared/helpers/test_helpers"
 )
 
 var _ = Describe("Block Retriever", func() {

pkg/omni/full/retriever/retriever_suite_test.go

@@ -27,7 +27,7 @@ import (
 
 func TestRetriever(t *testing.T) {
 	RegisterFailHandler(Fail)
-	RunSpecs(t, "Retriever Suite Test")
+	RunSpecs(t, "Full Retriever Suite Test")
 }
 
 var _ = BeforeSuite(func() {

pkg/omni/full/transformer/transformer.go

@@ -25,34 +25,20 @@ import (
 	"github.com/vulcanize/vulcanizedb/pkg/datastore"
 	"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
 	"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres/repositories"
-	"github.com/vulcanize/vulcanizedb/pkg/omni/contract"
-	"github.com/vulcanize/vulcanizedb/pkg/omni/converter"
-	"github.com/vulcanize/vulcanizedb/pkg/omni/parser"
-	"github.com/vulcanize/vulcanizedb/pkg/omni/poller"
-	"github.com/vulcanize/vulcanizedb/pkg/omni/repository"
-	"github.com/vulcanize/vulcanizedb/pkg/omni/retriever"
+	"github.com/vulcanize/vulcanizedb/pkg/omni/full/converter"
+	"github.com/vulcanize/vulcanizedb/pkg/omni/full/repository"
+	"github.com/vulcanize/vulcanizedb/pkg/omni/full/retriever"
+	"github.com/vulcanize/vulcanizedb/pkg/omni/shared/contract"
+	"github.com/vulcanize/vulcanizedb/pkg/omni/shared/parser"
+	"github.com/vulcanize/vulcanizedb/pkg/omni/shared/poller"
 )
 
-// Omni transformer
-// Used to extract all or a subset of event and method data
-// for any contract and persist it to postgres in a manner
-// that requires no prior knowledge of the contract
-// other than its address and which network it is on
-type Transformer interface {
-	SetEvents(contractAddr string, filterSet []string)
-	SetEventAddrs(contractAddr string, filterSet []string)
-	SetMethods(contractAddr string, filterSet []string)
-	SetMethodAddrs(contractAddr string, filterSet []string)
-	SetRange(contractAddr string, rng [2]int64)
-	Init() error
-	Execute() error
-}
-
+// Requires a fully synced vDB and a running eth node (or infura)
 type transformer struct {
 	// Database interfaces
 	datastore.FilterRepository       // Log filters repo; accepts filters generated by Contract.GenerateFilters()
 	datastore.WatchedEventRepository // Watched event log views, created by the log filters
-	repository.EventDatastore        // Holds transformed watched event log data
+	repository.EventRepository       // Holds transformed watched event log data
 
 	// Pre-processing interfaces
 	parser.Parser            // Parses events and methods out of contract abi fetched using contract address
@@ -84,7 +70,6 @@ type transformer struct {
 
 // Transformer takes in config for blockchain, database, and network id
 func NewTransformer(network string, BC core.BlockChain, DB *postgres.DB) *transformer {
-
 	return &transformer{
 		Poller:                 poller.NewPoller(BC, DB),
 		Parser:                 parser.NewParser(network),
@@ -93,7 +78,7 @@ func NewTransformer(network string, BC core.BlockChain, DB *postgres.DB) *transf
 		Contracts:              map[string]*contract.Contract{},
 		WatchedEventRepository: repositories.WatchedEventRepository{DB: DB},
 		FilterRepository:       repositories.FilterRepository{DB: DB},
-		EventDatastore:         repository.NewEventDataStore(DB),
+		EventRepository:        repository.NewEventRepository(DB),
 		WatchedEvents:          map[string][]string{},
 		WantedMethods:          map[string][]string{},
 		ContractRanges:         map[string][2]int64{},
@@ -107,7 +92,6 @@ func NewTransformer(network string, BC core.BlockChain, DB *postgres.DB) *transf
 // Uses parser to pull event info from abi
 // Use this info to generate event filters
 func (t *transformer) Init() error {
-
 	for contractAddr, subset := range t.WatchedEvents {
 		// Get Abi
 		err := t.Parser.Parse(contractAddr)
@@ -153,6 +137,7 @@ func (t *transformer) Init() error {
 		// Aggregate info into contract object
 		info := &contract.Contract{
 			Name:           *name,
+			Network:        t.Network,
 			Address:        contractAddr,
 			Abi:            t.Abi(),
 			ParsedAbi:      t.ParsedAbi(),
@@ -196,7 +181,7 @@ func (tr transformer) Execute() error {
 	for _, con := range tr.Contracts {
 
 		// Update converter with current contract
-		tr.Converter.Update(con)
+		tr.Update(con)
 
 		// Iterate through contract filters and get watched event logs
 		for eventName, filter := range con.Filters {
@@ -213,8 +198,11 @@ func (tr transformer) Execute() error {
 				if err != nil {
 					return err
 				}
+				if log == nil {
+					break
+				}
 
-				// And immediately persist converted logs in repo
+				// If log is not empty, immediately persist in repo
 				// Run this in seperate goroutine?
 				err = tr.PersistLog(*log, con.Address, con.Name)
 				if err != nil {

pkg/omni/full/transformer/transformer_suite_test.go

@@ -27,7 +27,7 @@ import (
 
 func TestTransformer(t *testing.T) {
 	RegisterFailHandler(Fail)
-	RunSpecs(t, "Transformer Suite Test")
+	RunSpecs(t, "Full Transformer Suite Test")
 }
 
 var _ = BeforeSuite(func() {

pkg/omni/full/transformer/transformer_test.go

@@ -27,61 +27,11 @@ import (
 	"github.com/vulcanize/vulcanizedb/pkg/core"
 	"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
 	"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres/repositories"
-	"github.com/vulcanize/vulcanizedb/pkg/omni/constants"
-	"github.com/vulcanize/vulcanizedb/pkg/omni/helpers/test_helpers"
-	"github.com/vulcanize/vulcanizedb/pkg/omni/transformer"
+	"github.com/vulcanize/vulcanizedb/pkg/omni/full/transformer"
+	"github.com/vulcanize/vulcanizedb/pkg/omni/shared/constants"
+	"github.com/vulcanize/vulcanizedb/pkg/omni/shared/helpers/test_helpers"
 )
 
-var block1 = core.Block{
-	Hash:   "0x135391a0962a63944e5908e6fedfff90fb4be3e3290a21017861099bad123ert",
-	Number: 6194633,
-	Transactions: []core.Transaction{{
-		Hash: "0x135391a0962a63944e5908e6fedfff90fb4be3e3290a21017861099bad654aaa",
-		Receipt: core.Receipt{
-			TxHash:          "0x135391a0962a63944e5908e6fedfff90fb4be3e3290a21017861099bad654aaa",
-			ContractAddress: "",
-			Logs: []core.Log{{
-				BlockNumber: 6194633,
-				TxHash:      "0x135391a0962a63944e5908e6fedfff90fb4be3e3290a21017861099bad654aaa",
-				Address:     constants.TusdContractAddress,
-				Topics: core.Topics{
-					constants.TransferEvent.Signature(),
-					"0x000000000000000000000000000000000000000000000000000000000000af21",
-					"0x9dd48110dcc444fdc242510c09bbbbe21a5975cac061d82f7b843bce061ba391",
-					"",
-				},
-				Index: 1,
-				Data:  "0x000000000000000000000000c02aaa39b223fe8d0a0e5c4f27ead9083c756cc200000000000000000000000089d24a6b4ccb1b6faa2625fe562bdd9a23260359000000000000000000000000000000000000000000000000392d2e2bda9c00000000000000000000000000000000000000000000000000927f41fa0a4a418000000000000000000000000000000000000000000000000000000000005adcfebe",
-			}},
-		},
-	}},
-}
-
-var block2 = core.Block{
-	Hash:   "0x135391a0962a63944e5908e6fedfff90fb4be3e3290a21017861099bad123ooo",
-	Number: 6194634,
-	Transactions: []core.Transaction{{
-		Hash: "0x135391a0962a63944e5908e6fedfff90fb4be3e3290a21017861099bad654eee",
-		Receipt: core.Receipt{
-			TxHash:          "0x135391a0962a63944e5908e6fedfff90fb4be3e3290a21017861099bad654eee",
-			ContractAddress: "",
-			Logs: []core.Log{{
-				BlockNumber: 6194634,
-				TxHash:      "0x135391a0962a63944e5908e6fedfff90fb4be3e3290a21017861099bad654eee",
-				Address:     constants.TusdContractAddress,
-				Topics: core.Topics{
-					constants.TransferEvent.Signature(),
-					"0x000000000000000000000000000000000000000000000000000000000000af21",
-					"0x9dd48110dcc444fdc242510c09bbbbe21a5975cac061d82f7b843bce061ba391",
-					"",
-				},
-				Index: 1,
-				Data:  "0x000000000000000000000000c02aaa39b223fe8d0a0e5c4f27ead9083c756cc200000000000000000000000089d24a6b4ccb1b6faa2625fe562bdd9a23260359000000000000000000000000000000000000000000000000392d2e2bda9c00000000000000000000000000000000000000000000000000927f41fa0a4a418000000000000000000000000000000000000000000000000000000000005adcfebe",
-			}},
-		},
-	}},
-}
-
 var _ = Describe("Transformer", func() {
 	var db *postgres.DB
 	var err error
@@ -145,8 +95,8 @@ var _ = Describe("Transformer", func() {
 
 	Describe("Init", func() {
 		It("Initializes transformer's contract objects", func() {
-			blockRepository.CreateOrUpdateBlock(block1)
-			blockRepository.CreateOrUpdateBlock(block2)
+			blockRepository.CreateOrUpdateBlock(test_helpers.TransferBlock1)
+			blockRepository.CreateOrUpdateBlock(test_helpers.TransferBlock2)
 			t := transformer.NewTransformer("", blockChain, db)
 			t.SetEvents(constants.TusdContractAddress, []string{"Transfer"})
 			err = t.Init()
@@ -169,9 +119,9 @@ var _ = Describe("Transformer", func() {
 			Expect(err).To(HaveOccurred())
 		})
 
-		It("Does nothing if watched events are nil", func() {
-			blockRepository.CreateOrUpdateBlock(block1)
-			blockRepository.CreateOrUpdateBlock(block2)
+		It("Does nothing if watched events are unset", func() {
+			blockRepository.CreateOrUpdateBlock(test_helpers.TransferBlock1)
+			blockRepository.CreateOrUpdateBlock(test_helpers.TransferBlock2)
 			t := transformer.NewTransformer("", blockChain, db)
 			err = t.Init()
 			Expect(err).ToNot(HaveOccurred())
@@ -183,8 +133,8 @@ var _ = Describe("Transformer", func() {
 
 	Describe("Execute", func() {
 		BeforeEach(func() {
-			blockRepository.CreateOrUpdateBlock(block1)
-			blockRepository.CreateOrUpdateBlock(block2)
+			blockRepository.CreateOrUpdateBlock(test_helpers.TransferBlock1)
+			blockRepository.CreateOrUpdateBlock(test_helpers.TransferBlock2)
 		})
 
 		It("Transforms watched contract data into custom repositories", func() {

pkg/omni/light/converter/converter.go

@@ -0,0 +1,50 @@
+// VulcanizeDB
+// Copyright © 2018 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 converter
+
+import (
+	"errors"
+
+	geth "github.com/ethereum/go-ethereum/core/types"
+
+	"github.com/vulcanize/vulcanizedb/pkg/omni/shared/contract"
+	"github.com/vulcanize/vulcanizedb/pkg/omni/shared/types"
+)
+
+type Converter interface {
+	Convert(log geth.Log, event types.Event) (*types.Log, error)
+	Update(info *contract.Contract)
+}
+
+type converter struct {
+	ContractInfo *contract.Contract
+}
+
+func NewConverter(info *contract.Contract) *converter {
+	return &converter{
+		ContractInfo: info,
+	}
+}
+
+func (c *converter) Update(info *contract.Contract) {
+	c.ContractInfo = info
+}
+
+// Convert the given watched event log into a types.Log for the given event
+func (c *converter) Convert(log geth.Log, event types.Event) (*types.Log, error) {
+	return nil, errors.New("implement me")
+}

pkg/omni/light/converter/converter_suite_test.go

@@ -0,0 +1,35 @@
+// VulcanizeDB
+// Copyright © 2018 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 converter_test
+
+import (
+	"io/ioutil"
+	"log"
+	"testing"
+
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+)
+
+func TestConverter(t *testing.T) {
+	RegisterFailHandler(Fail)
+	RunSpecs(t, "Light Converter Suite Test")
+}
+
+var _ = BeforeSuite(func() {
+	log.SetOutput(ioutil.Discard)
+})

pkg/omni/light/converter/converter_test.go

@@ -0,0 +1,17 @@
+// VulcanizeDB
+// Copyright © 2018 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 converter_test

pkg/omni/light/fetcher/fetcher.go

@@ -0,0 +1,70 @@
+// VulcanizeDB
+// Copyright © 2018 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 fetcher
+
+import (
+	"github.com/ethereum/go-ethereum"
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/core/types"
+
+	"github.com/vulcanize/vulcanizedb/pkg/core"
+)
+
+type LogFetcher interface {
+	FetchLogs(contractAddresses []string, topics [][]common.Hash, header core.Header) ([]types.Log, error)
+}
+
+type SettableLogFetcher interface {
+	LogFetcher
+	SetBC(bc core.BlockChain)
+}
+
+type Fetcher struct {
+	blockChain core.BlockChain
+}
+
+func (fetcher *Fetcher) SetBC(bc core.BlockChain) {
+	fetcher.blockChain = bc
+}
+
+func NewFetcher(blockchain core.BlockChain) Fetcher {
+	return Fetcher{
+		blockChain: blockchain,
+	}
+}
+
+func (fetcher Fetcher) FetchLogs(contractAddresses []string, topics [][]common.Hash, header core.Header) ([]types.Log, error) {
+	addresses := hexStringsToAddresses(contractAddresses)
+	blockHash := common.HexToHash(header.Hash)
+	query := ethereum.FilterQuery{
+		BlockHash: &blockHash,
+		Addresses: addresses,
+		Topics:    topics,
+	}
+
+	return fetcher.blockChain.GetEthLogsWithCustomQuery(query)
+}
+
+func hexStringsToAddresses(hexStrings []string) []common.Address {
+	var addresses []common.Address
+	for _, hexString := range hexStrings {
+		address := common.HexToAddress(hexString)
+		addresses = append(addresses, address)
+	}
+
+	return addresses
+}

pkg/omni/light/fetcher/fetcher_suite_test.go

@@ -0,0 +1,35 @@
+// VulcanizeDB
+// Copyright © 2018 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 fetcher_test
+
+import (
+	"io/ioutil"
+	"log"
+	"testing"
+
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+)
+
+func TestFetcher(t *testing.T) {
+	RegisterFailHandler(Fail)
+	RunSpecs(t, "Fetcher Suite Test")
+}
+
+var _ = BeforeSuite(func() {
+	log.SetOutput(ioutil.Discard)
+})

pkg/omni/light/fetcher/fetcher_test.go

@@ -0,0 +1,66 @@
+// VulcanizeDB
+// Copyright © 2018 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 fetcher_test
+
+import (
+	"github.com/ethereum/go-ethereum"
+	"github.com/ethereum/go-ethereum/common"
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+
+	"github.com/vulcanize/vulcanizedb/pkg/core"
+	"github.com/vulcanize/vulcanizedb/pkg/fakes"
+	"github.com/vulcanize/vulcanizedb/pkg/omni/light/fetcher"
+)
+
+var _ = Describe("Fetcher", func() {
+	Describe("FetchLogs", func() {
+		It("fetches logs based on the given query", func() {
+			blockChain := fakes.NewMockBlockChain()
+			fetcher := fetcher.NewFetcher(blockChain)
+			header := fakes.FakeHeader
+
+			addresses := []string{"0xfakeAddress", "0xanotherFakeAddress"}
+			topicZeros := [][]common.Hash{{common.BytesToHash([]byte{1, 2, 3, 4, 5})}}
+
+			_, err := fetcher.FetchLogs(addresses, topicZeros, header)
+
+			address1 := common.HexToAddress("0xfakeAddress")
+			address2 := common.HexToAddress("0xanotherFakeAddress")
+			Expect(err).NotTo(HaveOccurred())
+
+			blockHash := common.HexToHash(header.Hash)
+			expectedQuery := ethereum.FilterQuery{
+				BlockHash: &blockHash,
+				Addresses: []common.Address{address1, address2},
+				Topics:    topicZeros,
+			}
+			blockChain.AssertGetEthLogsWithCustomQueryCalledWith(expectedQuery)
+		})
+
+		It("returns an error if fetching the logs fails", func() {
+			blockChain := fakes.NewMockBlockChain()
+			blockChain.SetGetEthLogsWithCustomQueryErr(fakes.FakeError)
+			fetcher := fetcher.NewFetcher(blockChain)
+
+			_, err := fetcher.FetchLogs([]string{}, [][]common.Hash{}, core.Header{})
+
+			Expect(err).To(HaveOccurred())
+			Expect(err).To(MatchError(fakes.FakeError))
+		})
+	})
+})

pkg/omni/light/repository/event_repository.go

@@ -0,0 +1,181 @@
+// VulcanizeDB
+// Copyright © 2018 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 (
+	"errors"
+	"fmt"
+	"strings"
+
+	"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
+	"github.com/vulcanize/vulcanizedb/pkg/omni/shared/types"
+)
+
+type EventRepository interface {
+	PersistLog(event types.Log, contractAddr, contractName string) error
+	CreateEventTable(contractName string, event types.Log) (bool, error)
+	CreateContractSchema(contractName string) (bool, error)
+}
+
+type eventRepository struct {
+	db *postgres.DB
+}
+
+func NewEventRepository(db *postgres.DB) *eventRepository {
+	return &eventRepository{
+		db: db,
+	}
+}
+
+// Creates a schema for the contract if needed
+// Creates table for the watched contract event if needed
+// Persists converted event log data into this custom table
+func (r *eventRepository) PersistLog(event types.Log, contractAddr, contractName string) error {
+	_, err := r.CreateContractSchema(contractAddr)
+	if err != nil {
+		return err
+	}
+
+	_, err = r.CreateEventTable(contractAddr, event)
+	if err != nil {
+		return err
+	}
+
+	return r.persistLog(event, contractAddr, contractName)
+}
+
+// Creates a custom postgres command to persist logs for the given event
+func (r *eventRepository) persistLog(event types.Log, contractAddr, contractName string) error {
+	// Begin postgres string
+	pgStr := fmt.Sprintf("INSERT INTO l%s.%s_event ", strings.ToLower(contractAddr), strings.ToLower(event.Name))
+	pgStr = pgStr + "(header_id, token_name, raw_log, log_idx, tx_idx"
+
+	// Pack the corresponding variables in a slice
+	var data []interface{}
+	data = append(data,
+		event.Id,
+		contractName,
+		event.Raw,
+		event.LogIndex,
+		event.TransactionIndex)
+
+	// Iterate over name-value pairs in the log adding
+	// names to the string and pushing values to the slice
+	counter := 0 // Keep track of number of inputs
+	for inputName, input := range event.Values {
+		counter += 1
+		pgStr = pgStr + fmt.Sprintf(", %s_", strings.ToLower(inputName)) // Add underscore after to avoid any collisions with reserved pg words
+		data = append(data, input)
+	}
+
+	// Finish off the string and execute the command using the packed data
+	// For each input entry we created we add its postgres command variable to the string
+	pgStr = pgStr + ") VALUES ($1, $2, $3, $4, $5"
+	for i := 0; i < counter; i++ {
+		pgStr = pgStr + fmt.Sprintf(", $%d", i+6)
+	}
+	pgStr = pgStr + ")"
+
+	_, err := r.db.Exec(pgStr, data...)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
+// Checks for event table and creates it if it does not already exist
+func (r *eventRepository) CreateEventTable(contractAddr string, event types.Log) (bool, error) {
+	tableExists, err := r.checkForTable(contractAddr, event.Name)
+	if err != nil {
+		return false, err
+	}
+
+	if !tableExists {
+		err = r.newEventTable(contractAddr, event)
+		if err != nil {
+			return false, err
+		}
+	}
+
+	return !tableExists, nil
+}
+
+// Creates a table for the given contract and event
+func (r *eventRepository) newEventTable(contractAddr string, event types.Log) error {
+	// Begin pg string
+	pgStr := fmt.Sprintf("CREATE TABLE IF NOT EXISTS l%s.%s_event ", strings.ToLower(contractAddr), strings.ToLower(event.Name))
+	pgStr = pgStr + "(id SERIAL, header_id INTEGER NOT NULL REFERENCES headers (id) ON DELETE CASCADE, token_name CHARACTER VARYING(66) NOT NULL, raw_log JSONB, log_idx INTEGER NOT NULL, tx_idx INTEGER NOT NULL,"
+
+	// Iterate over event fields, using their name and pgType to grow the string
+	for _, field := range event.Fields {
+		pgStr = pgStr + fmt.Sprintf(" %s_ %s NOT NULL,", strings.ToLower(field.Name), field.PgType)
+	}
+
+	pgStr = pgStr + " UNIQUE (header_id, tx_idx, log_idx))"
+	_, err := r.db.Exec(pgStr)
+
+	return err
+}
+
+// Checks if a table already exists for the given contract and event
+func (r *eventRepository) checkForTable(contractAddr string, eventName string) (bool, error) {
+	pgStr := fmt.Sprintf("SELECT EXISTS (SELECT 1 FROM information_schema.tables WHERE table_schema = 'l%s' AND table_name = '%s_event')", strings.ToLower(contractAddr), strings.ToLower(eventName))
+
+	var exists bool
+	err := r.db.Get(&exists, pgStr)
+
+	return exists, err
+}
+
+// Checks for contract schema and creates it if it does not already exist
+func (r *eventRepository) CreateContractSchema(contractAddr string) (bool, error) {
+	if contractAddr == "" {
+		return false, errors.New("error: no contract address specified")
+	}
+
+	schemaExists, err := r.checkForSchema(contractAddr)
+	if err != nil {
+		return false, err
+	}
+
+	if !schemaExists {
+		err = r.newContractSchema(contractAddr)
+		if err != nil {
+			return false, err
+		}
+	}
+
+	return !schemaExists, nil
+}
+
+// Creates a schema for the given contract
+func (r *eventRepository) newContractSchema(contractAddr string) error {
+	_, err := r.db.Exec("CREATE SCHEMA IF NOT EXISTS l" + strings.ToLower(contractAddr))
+
+	return err
+}
+
+// Checks if a schema already exists for the given contract
+func (r *eventRepository) checkForSchema(contractAddr string) (bool, error) {
+	pgStr := fmt.Sprintf("SELECT EXISTS (SELECT schema_name FROM information_schema.schemata WHERE schema_name = 'l%s')", strings.ToLower(contractAddr))
+
+	var exists bool
+	err := r.db.QueryRow(pgStr).Scan(&exists)
+
+	return exists, err
+}

pkg/omni/light/repository/event_repository_test.go

@@ -0,0 +1,17 @@
+// VulcanizeDB
+// Copyright © 2018 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

pkg/omni/light/repository/header_repository.go

@@ -0,0 +1,70 @@
+// VulcanizeDB
+// Copyright © 2018 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 (
+	"github.com/vulcanize/vulcanizedb/pkg/core"
+	"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
+)
+
+type HeaderRepository interface {
+	MarkHeaderChecked(headerID int64, eventID string) error
+	MissingHeaders(startingBlockNumber int64, endingBlockNumber int64, eventID string) ([]core.Header, error)
+}
+
+type headerRepository struct {
+	db *postgres.DB
+}
+
+func NewHeaderRepository(db *postgres.DB) *headerRepository {
+	return &headerRepository{
+		db: db,
+	}
+}
+
+func (r *headerRepository) MarkHeaderChecked(headerID int64, eventID string) error {
+	_, err := r.db.Exec(`INSERT INTO public.checked_headers (header_id, `+eventID+`)
+		VALUES ($1, $2) 
+		ON CONFLICT (header_id) DO
+			UPDATE SET `+eventID+` = $2`, headerID, true)
+	return err
+}
+
+func (r *headerRepository) MissingHeaders(startingBlockNumber int64, endingBlockNumber int64, eventID 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 ` + eventID + ` IS FALSE)
+				AND headers.block_number >= $1
+				AND headers.eth_node_fingerprint = $2`
+		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 ` + eventID + ` IS FALSE)
+				AND headers.block_number >= $1
+				AND headers.block_number <= $2
+				AND headers.eth_node_fingerprint = $3`
+		err = r.db.Select(&result, query, startingBlockNumber, endingBlockNumber, r.db.Node.ID)
+	}
+
+	return result, err
+}

pkg/omni/light/repository/header_repository_test.go

@@ -0,0 +1,17 @@
+// VulcanizeDB
+// Copyright © 2018 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

pkg/omni/light/repository/repository_suite_test.go

@@ -0,0 +1,35 @@
+// VulcanizeDB
+// Copyright © 2018 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 (
+	"io/ioutil"
+	"log"
+	"testing"
+
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+)
+
+func TestRepository(t *testing.T) {
+	RegisterFailHandler(Fail)
+	RunSpecs(t, "Light Repository Suite Test")
+}
+
+var _ = BeforeSuite(func() {
+	log.SetOutput(ioutil.Discard)
+})

pkg/omni/light/retriever/block_retriever.go

@@ -0,0 +1,60 @@
+// VulcanizeDB
+// Copyright © 2018 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 retriever
+
+import (
+	"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
+)
+
+// Block retriever is used to retrieve the first block for a given contract and the most recent block
+// It requires a vDB synced database with blocks, transactions, receipts, and logs
+type BlockRetriever interface {
+	RetrieveFirstBlock() (int64, error)
+	RetrieveMostRecentBlock() (int64, error)
+}
+
+type blockRetriever struct {
+	db *postgres.DB
+}
+
+func NewBlockRetriever(db *postgres.DB) (r *blockRetriever) {
+	return &blockRetriever{
+		db: db,
+	}
+}
+
+// Retrieve block number of earliest header in repo
+func (r *blockRetriever) RetrieveFirstBlock() (int64, error) {
+	var firstBlock int
+	err := r.db.Get(
+		&firstBlock,
+		"SELECT block_number FROM headers ORDER BY block_number ASC LIMIT 1",
+	)
+
+	return int64(firstBlock), err
+}
+
+// Retrieve block number of latest header in repo
+func (r *blockRetriever) RetrieveMostRecentBlock() (int64, error) {
+	var lastBlock int
+	err := r.db.Get(
+		&lastBlock,
+		"SELECT block_number FROM headers ORDER BY block_number DESC LIMIT 1",
+	)
+
+	return int64(lastBlock), err
+}

pkg/omni/light/retriever/block_retriever_test.go

@@ -0,0 +1,78 @@
+// VulcanizeDB
+// Copyright © 2018 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 retriever_test
+
+import (
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+
+	"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
+	"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres/repositories"
+	"github.com/vulcanize/vulcanizedb/pkg/omni/light/retriever"
+	"github.com/vulcanize/vulcanizedb/pkg/omni/shared/helpers/test_helpers"
+)
+
+var _ = Describe("Block Retriever", func() {
+	var db *postgres.DB
+	var r retriever.BlockRetriever
+	var headerRepository repositories.HeaderRepository
+
+	BeforeEach(func() {
+		db, _ = test_helpers.SetupDBandBC()
+		headerRepository = repositories.NewHeaderRepository(db)
+		r = retriever.NewBlockRetriever(db)
+	})
+
+	AfterEach(func() {
+		test_helpers.TearDown(db)
+	})
+
+	Describe("RetrieveFirstBlock", func() {
+		It("Retrieves block number of earliest header in the database", func() {
+			headerRepository.CreateOrUpdateHeader(test_helpers.MockHeader1)
+			headerRepository.CreateOrUpdateHeader(test_helpers.MockHeader2)
+			headerRepository.CreateOrUpdateHeader(test_helpers.MockHeader3)
+
+			i, err := r.RetrieveFirstBlock()
+			Expect(err).NotTo(HaveOccurred())
+			Expect(i).To(Equal(int64(6194632)))
+		})
+
+		It("Fails if no headers can be found in the database", func() {
+			_, err := r.RetrieveFirstBlock()
+			Expect(err).To(HaveOccurred())
+		})
+	})
+
+	Describe("RetrieveMostRecentBlock", func() {
+		It("Retrieves the latest header's block number", func() {
+			headerRepository.CreateOrUpdateHeader(test_helpers.MockHeader1)
+			headerRepository.CreateOrUpdateHeader(test_helpers.MockHeader2)
+			headerRepository.CreateOrUpdateHeader(test_helpers.MockHeader3)
+
+			i, err := r.RetrieveMostRecentBlock()
+			Expect(err).ToNot(HaveOccurred())
+			Expect(i).To(Equal(int64(6194634)))
+		})
+
+		It("Fails if no headers can be found in the database", func() {
+			i, err := r.RetrieveMostRecentBlock()
+			Expect(err).To(HaveOccurred())
+			Expect(i).To(Equal(int64(0)))
+		})
+	})
+})

pkg/omni/light/retriever/retriever_suite_test.go

@@ -0,0 +1,35 @@
+// VulcanizeDB
+// Copyright © 2018 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 retriever_test
+
+import (
+	"io/ioutil"
+	"log"
+	"testing"
+
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+)
+
+func TestRetriever(t *testing.T) {
+	RegisterFailHandler(Fail)
+	RunSpecs(t, "Light Retriever Suite Test")
+}
+
+var _ = BeforeSuite(func() {
+	log.SetOutput(ioutil.Discard)
+})

pkg/omni/light/transformer/transformer.go

@@ -0,0 +1,238 @@
+// VulcanizeDB
+// Copyright © 2018 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 transformer
+
+import (
+	"errors"
+	"fmt"
+
+	"github.com/ethereum/go-ethereum/common"
+
+	"github.com/vulcanize/vulcanizedb/pkg/core"
+	"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
+	"github.com/vulcanize/vulcanizedb/pkg/omni/light/converter"
+	"github.com/vulcanize/vulcanizedb/pkg/omni/light/fetcher"
+	"github.com/vulcanize/vulcanizedb/pkg/omni/light/repository"
+	"github.com/vulcanize/vulcanizedb/pkg/omni/light/retriever"
+	"github.com/vulcanize/vulcanizedb/pkg/omni/shared/contract"
+	"github.com/vulcanize/vulcanizedb/pkg/omni/shared/parser"
+	"github.com/vulcanize/vulcanizedb/pkg/omni/shared/poller"
+)
+
+// Requires a light synced vDB (headers) and a running eth node (or infura)
+type transformer struct {
+	// Database interfaces
+	repository.EventRepository  // Holds transformed watched event log data
+	repository.HeaderRepository // Interface for interaction with header repositories
+
+	// Pre-processing interfaces
+	parser.Parser            // Parses events and methods out of contract abi fetched using contract address
+	retriever.BlockRetriever // Retrieves first block for contract and current block height
+
+	// Processing interfaces
+	fetcher.Fetcher     // Fetches event logs, using header hashes
+	converter.Converter // Converts watched event logs into custom log
+	poller.Poller       // Polls methods using contract's token holder addresses and persists them using method datastore
+
+	// Ethereum network name; default "" is mainnet
+	Network string
+
+	// Store contract info as mapping to contract address
+	Contracts map[string]*contract.Contract
+
+	// Targeted subset of events/methods
+	// Stored as map sof contract address to events/method names of interest
+	WatchedEvents map[string][]string // Default/empty event list means all are watched
+	WantedMethods map[string][]string // Default/empty method list means none are polled
+
+	// Block ranges to watch contracts
+	ContractRanges map[string][2]int64
+
+	// Lists of addresses to filter event or method data
+	// before persisting; if empty no filter is applied
+	EventAddrs  map[string][]string
+	MethodAddrs map[string][]string
+}
+
+// Transformer takes in config for blockchain, database, and network id
+func NewTransformer(network string, bc core.BlockChain, db *postgres.DB) *transformer {
+
+	return &transformer{
+		Poller:           poller.NewPoller(bc, db),
+		Fetcher:          fetcher.NewFetcher(bc),
+		Parser:           parser.NewParser(network),
+		HeaderRepository: repository.NewHeaderRepository(db),
+		BlockRetriever:   retriever.NewBlockRetriever(db),
+		Converter:        converter.NewConverter(&contract.Contract{}),
+		Contracts:        map[string]*contract.Contract{},
+		EventRepository:  repository.NewEventRepository(db),
+		WatchedEvents:    map[string][]string{},
+		WantedMethods:    map[string][]string{},
+		ContractRanges:   map[string][2]int64{},
+		EventAddrs:       map[string][]string{},
+		MethodAddrs:      map[string][]string{},
+	}
+}
+
+// Use after creating and setting transformer
+// Loops over all of the addr => filter sets
+// Uses parser to pull event info from abi
+// Use this info to generate event filters
+func (tr *transformer) Init() error {
+
+	for contractAddr, subset := range tr.WatchedEvents {
+		// Get Abi
+		err := tr.Parser.Parse(contractAddr)
+		if err != nil {
+			return err
+		}
+
+		// Get first block for contract and most recent block for the chain
+		firstBlock, err := tr.BlockRetriever.RetrieveFirstBlock()
+		if err != nil {
+			return err
+		}
+		lastBlock, err := tr.BlockRetriever.RetrieveMostRecentBlock()
+		if err != nil {
+			return err
+		}
+
+		// Set to specified range if it falls within the contract's bounds
+		if firstBlock < tr.ContractRanges[contractAddr][0] {
+			firstBlock = tr.ContractRanges[contractAddr][0]
+		}
+		if lastBlock > tr.ContractRanges[contractAddr][1] && tr.ContractRanges[contractAddr][1] > firstBlock {
+			lastBlock = tr.ContractRanges[contractAddr][1]
+		}
+
+		// Get contract name
+		var name = new(string)
+		err = tr.FetchContractData(tr.Abi(), contractAddr, "name", nil, &name, lastBlock)
+		if err != nil {
+			return errors.New(fmt.Sprintf("unable to fetch contract name: %v\r\n", err))
+		}
+
+		// Remove any accidental duplicate inputs in filter addresses
+		EventAddrs := map[string]bool{}
+		for _, addr := range tr.EventAddrs[contractAddr] {
+			EventAddrs[addr] = true
+		}
+		MethodAddrs := map[string]bool{}
+		for _, addr := range tr.MethodAddrs[contractAddr] {
+			MethodAddrs[addr] = true
+		}
+
+		// Aggregate info into contract object
+		info := &contract.Contract{
+			Name:           *name,
+			Network:        tr.Network,
+			Address:        contractAddr,
+			Abi:            tr.Abi(),
+			StartingBlock:  firstBlock,
+			LastBlock:      lastBlock,
+			Events:         tr.GetEvents(subset),
+			Methods:        tr.GetAddrMethods(tr.WantedMethods[contractAddr]),
+			EventAddrs:     EventAddrs,
+			MethodAddrs:    MethodAddrs,
+			TknHolderAddrs: map[string]bool{},
+		}
+
+		// Store contract info for further processing
+		tr.Contracts[contractAddr] = info
+	}
+
+	return nil
+}
+
+func (tr *transformer) Execute() error {
+	if len(tr.Contracts) == 0 {
+		return errors.New("error: transformer has no initialized contracts to work with")
+	}
+	// Iterate through all internal contracts
+	for _, con := range tr.Contracts {
+
+		// Update converter with current contract
+		tr.Update(con)
+
+		for _, event := range con.Events {
+			topics := [][]common.Hash{{common.HexToHash(event.Sig())}}
+			eventId := event.Name + "_" + con.Address
+			missingHeaders, err := tr.MissingHeaders(con.StartingBlock, con.LastBlock, eventId)
+			if err != nil {
+				return err
+			}
+
+			for _, header := range missingHeaders {
+				logs, err := tr.FetchLogs([]string{con.Address}, topics, header)
+				if err != nil {
+					return err
+				}
+
+				if len(logs) < 1 {
+					err = tr.MarkHeaderChecked(header.Id, eventId)
+					if err != nil {
+						return err
+					}
+
+					continue
+				}
+
+				for _, l := range logs {
+					mapping, err := tr.Convert(l, event)
+					if err != nil {
+						return err
+					}
+					if mapping == nil {
+						break
+					}
+
+					err = tr.PersistLog(*mapping, con.Address, con.Name)
+					if err != nil {
+						return err
+					}
+				}
+			}
+		}
+	}
+
+	return nil
+}
+
+// Used to set which contract addresses and which of their events to watch
+func (tr *transformer) SetEvents(contractAddr string, filterSet []string) {
+	tr.WatchedEvents[contractAddr] = filterSet
+}
+
+// Used to set subset of account addresses to watch events for
+func (tr *transformer) SetEventAddrs(contractAddr string, filterSet []string) {
+	tr.EventAddrs[contractAddr] = filterSet
+}
+
+// Used to set which contract addresses and which of their methods to call
+func (tr *transformer) SetMethods(contractAddr string, filterSet []string) {
+	tr.WantedMethods[contractAddr] = filterSet
+}
+
+// Used to set subset of account addresses to poll methods on
+func (tr *transformer) SetMethodAddrs(contractAddr string, filterSet []string) {
+	tr.MethodAddrs[contractAddr] = filterSet
+}
+
+// Used to set the block range to watch for a given address
+func (tr *transformer) SetRange(contractAddr string, rng [2]int64) {
+	tr.ContractRanges[contractAddr] = rng
+}

pkg/omni/light/transformer/transformer_suite_test.go

@@ -0,0 +1,35 @@
+// VulcanizeDB
+// Copyright © 2018 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 transformer_test
+
+import (
+	"io/ioutil"
+	"log"
+	"testing"
+
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+)
+
+func TestTransformer(t *testing.T) {
+	RegisterFailHandler(Fail)
+	RunSpecs(t, "Light Transformer Suite Test")
+}
+
+var _ = BeforeSuite(func() {
+	log.SetOutput(ioutil.Discard)
+})

pkg/omni/light/transformer/transformer_test.go

@@ -0,0 +1,136 @@
+// VulcanizeDB
+// Copyright © 2018 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 transformer_test
+
+import (
+	"math/rand"
+	"time"
+
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+
+	"github.com/vulcanize/vulcanizedb/pkg/core"
+	"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
+	"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres/repositories"
+	"github.com/vulcanize/vulcanizedb/pkg/omni/light/transformer"
+	"github.com/vulcanize/vulcanizedb/pkg/omni/shared/constants"
+	"github.com/vulcanize/vulcanizedb/pkg/omni/shared/helpers/test_helpers"
+)
+
+var _ = Describe("Transformer", func() {
+	var db *postgres.DB
+	var err error
+	var blockChain core.BlockChain
+	var headerRepository repositories.HeaderRepository
+	rand.Seed(time.Now().UnixNano())
+
+	BeforeEach(func() {
+		db, blockChain = test_helpers.SetupDBandBC()
+		headerRepository = repositories.NewHeaderRepository(db)
+	})
+
+	AfterEach(func() {
+		test_helpers.TearDown(db)
+	})
+
+	Describe("SetEvents", func() {
+		It("Sets which events to watch from the given contract address", func() {
+			watchedEvents := []string{"Transfer", "Mint"}
+			t := transformer.NewTransformer("", blockChain, db)
+			t.SetEvents(constants.TusdContractAddress, watchedEvents)
+			Expect(t.WatchedEvents[constants.TusdContractAddress]).To(Equal(watchedEvents))
+		})
+	})
+
+	Describe("SetEventAddrs", func() {
+		It("Sets which account addresses to watch events for", func() {
+			eventAddrs := []string{"test1", "test2"}
+			t := transformer.NewTransformer("", blockChain, db)
+			t.SetEventAddrs(constants.TusdContractAddress, eventAddrs)
+			Expect(t.EventAddrs[constants.TusdContractAddress]).To(Equal(eventAddrs))
+		})
+	})
+
+	Describe("SetMethods", func() {
+		It("Sets which methods to poll at the given contract address", func() {
+			watchedMethods := []string{"balanceOf", "totalSupply"}
+			t := transformer.NewTransformer("", blockChain, db)
+			t.SetMethods(constants.TusdContractAddress, watchedMethods)
+			Expect(t.WantedMethods[constants.TusdContractAddress]).To(Equal(watchedMethods))
+		})
+	})
+
+	Describe("SetMethodAddrs", func() {
+		It("Sets which account addresses to poll methods against", func() {
+			methodAddrs := []string{"test1", "test2"}
+			t := transformer.NewTransformer("", blockChain, db)
+			t.SetMethodAddrs(constants.TusdContractAddress, methodAddrs)
+			Expect(t.MethodAddrs[constants.TusdContractAddress]).To(Equal(methodAddrs))
+		})
+	})
+
+	Describe("SetRange", func() {
+		It("Sets the block range that the contract should be watched within", func() {
+			rng := [2]int64{1, 100000}
+			t := transformer.NewTransformer("", blockChain, db)
+			t.SetRange(constants.TusdContractAddress, rng)
+			Expect(t.ContractRanges[constants.TusdContractAddress]).To(Equal(rng))
+		})
+	})
+
+	Describe("Init", func() {
+		It("Initializes transformer's contract objects", func() {
+			headerRepository.CreateOrUpdateHeader(test_helpers.MockHeader1)
+			headerRepository.CreateOrUpdateHeader(test_helpers.MockHeader3)
+			t := transformer.NewTransformer("", blockChain, db)
+			t.SetEvents(constants.TusdContractAddress, []string{"Transfer"})
+			err = t.Init()
+			Expect(err).ToNot(HaveOccurred())
+
+			c, ok := t.Contracts[constants.TusdContractAddress]
+			Expect(ok).To(Equal(true))
+
+			Expect(c.StartingBlock).To(Equal(int64(6194632)))
+			Expect(c.LastBlock).To(Equal(int64(6194634)))
+			Expect(c.Abi).To(Equal(constants.TusdAbiString))
+			Expect(c.Name).To(Equal("TrueUSD"))
+			Expect(c.Address).To(Equal(constants.TusdContractAddress))
+		})
+
+		It("Fails to initialize if first and most recent blocks cannot be fetched from vDB", func() {
+			t := transformer.NewTransformer("", blockChain, db)
+			t.SetEvents(constants.TusdContractAddress, []string{"Transfer"})
+			err = t.Init()
+			Expect(err).To(HaveOccurred())
+		})
+
+		It("Does nothing if watched events are unset", func() {
+			headerRepository.CreateOrUpdateHeader(test_helpers.MockHeader1)
+			headerRepository.CreateOrUpdateHeader(test_helpers.MockHeader3)
+			t := transformer.NewTransformer("", blockChain, db)
+			err = t.Init()
+			Expect(err).ToNot(HaveOccurred())
+
+			_, ok := t.Contracts[constants.TusdContractAddress]
+			Expect(ok).To(Equal(false))
+		})
+	})
+
+	Describe("Execute", func() {
+
+	})
+})

pkg/omni/shared/constants/constants.go

@@ -19,7 +19,7 @@ package constants
 import (
 	"github.com/vulcanize/vulcanizedb/pkg/core"
 	"github.com/vulcanize/vulcanizedb/pkg/filters"
-	"github.com/vulcanize/vulcanizedb/pkg/omni/helpers"
+	"github.com/vulcanize/vulcanizedb/pkg/omni/shared/helpers"
 )
 
 // Event enums

pkg/omni/shared/contract/contract.go

@@ -23,17 +23,19 @@ import (
 
 	"github.com/vulcanize/vulcanizedb/pkg/core"
 	"github.com/vulcanize/vulcanizedb/pkg/filters"
-	"github.com/vulcanize/vulcanizedb/pkg/omni/helpers"
-	"github.com/vulcanize/vulcanizedb/pkg/omni/types"
+	"github.com/vulcanize/vulcanizedb/pkg/omni/shared/helpers"
+	"github.com/vulcanize/vulcanizedb/pkg/omni/shared/types"
 )
 
+// Contract object to hold our contract data
 type Contract struct {
-	Name           string
-	Address        string
-	StartingBlock  int64
-	LastBlock      int64
-	Abi            string
-	ParsedAbi      abi.ABI
+	Name           string                       // Name of the contract
+	Address        string                       // Address of the contract
+	Network        string                       // Network on which the contract is deployed; default empty "" is Ethereum mainnet
+	StartingBlock  int64                        // Starting block of the contract
+	LastBlock      int64                        // Most recent block on the network
+	Abi            string                       // Abi string
+	ParsedAbi      abi.ABI                      // Parsed abi
 	Events         map[string]types.Event       // Map of events to their names
 	Methods        map[string]types.Method      // Map of methods to their names
 	Filters        map[string]filters.LogFilter // Map of event filters to their names
@@ -55,7 +57,7 @@ func (c *Contract) GenerateFilters() error {
 			Topics:    core.Topics{helpers.GenerateSignature(event.Sig())}, // move generate signatrue to pkg
 		}
 	}
-	// If no filters we generated, throw an error (no point in continuing)
+	// If no filters were generated, throw an error (no point in continuing with this contract)
 	if len(c.Filters) == 0 {
 		return errors.New("error: no filters created")
 	}

pkg/omni/shared/contract/contract_test.go

@@ -20,8 +20,8 @@ import (
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
 
-	"github.com/vulcanize/vulcanizedb/pkg/omni/contract"
-	"github.com/vulcanize/vulcanizedb/pkg/omni/helpers/test_helpers"
+	"github.com/vulcanize/vulcanizedb/pkg/omni/shared/contract"
+	"github.com/vulcanize/vulcanizedb/pkg/omni/shared/helpers/test_helpers"
 )
 
 var _ = Describe("Contract", func() {

pkg/omni/shared/helpers/mocks/parser.go

@@ -20,7 +20,7 @@ import (
 	"github.com/ethereum/go-ethereum/accounts/abi"
 
 	"github.com/vulcanize/vulcanizedb/pkg/geth"
-	"github.com/vulcanize/vulcanizedb/pkg/omni/types"
+	"github.com/vulcanize/vulcanizedb/pkg/omni/shared/types"
 )
 
 // Mock parser

pkg/omni/shared/helpers/test_helpers/database.go

@@ -27,32 +27,15 @@ import (
 	"github.com/vulcanize/vulcanizedb/pkg/core"
 	"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
 	"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres/repositories"
-	"github.com/vulcanize/vulcanizedb/pkg/filters"
 	"github.com/vulcanize/vulcanizedb/pkg/geth"
 	"github.com/vulcanize/vulcanizedb/pkg/geth/client"
 	rpc2 "github.com/vulcanize/vulcanizedb/pkg/geth/converters/rpc"
 	"github.com/vulcanize/vulcanizedb/pkg/geth/node"
-	"github.com/vulcanize/vulcanizedb/pkg/omni/constants"
-	"github.com/vulcanize/vulcanizedb/pkg/omni/contract"
-	"github.com/vulcanize/vulcanizedb/pkg/omni/parser"
+	"github.com/vulcanize/vulcanizedb/pkg/omni/shared/constants"
+	"github.com/vulcanize/vulcanizedb/pkg/omni/shared/contract"
+	"github.com/vulcanize/vulcanizedb/pkg/omni/shared/parser"
 )
 
-var ExpectedTransferFilter = filters.LogFilter{
-	Name:      "Transfer",
-	Address:   constants.TusdContractAddress,
-	ToBlock:   -1,
-	FromBlock: 6194634,
-	Topics:    core.Topics{constants.TransferEvent.Signature()},
-}
-
-var ExpectedApprovalFilter = filters.LogFilter{
-	Name:      "Approval",
-	Address:   constants.TusdContractAddress,
-	ToBlock:   -1,
-	FromBlock: 6194634,
-	Topics:    core.Topics{constants.ApprovalEvent.Signature()},
-}
-
 type TransferLog struct {
 	Id             int64  `db:"id"`
 	VulvanizeLogId int64  `db:"vulcanize_log_id"`
@@ -159,6 +142,12 @@ func TearDown(db *postgres.DB) {
 	_, err := db.Query(`DELETE FROM blocks`)
 	Expect(err).NotTo(HaveOccurred())
 
+	_, err = db.Query(`DELETE FROM headers`)
+	Expect(err).NotTo(HaveOccurred())
+
+	_, err = db.Query(`DELETE FROM checked_headers`)
+	Expect(err).NotTo(HaveOccurred())
+
 	_, err = db.Query(`DELETE FROM logs`)
 	Expect(err).NotTo(HaveOccurred())
 

pkg/omni/shared/helpers/test_helpers/entities.go

@@ -0,0 +1,128 @@
+// VulcanizeDB
+// Copyright © 2018 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 (
+	"encoding/json"
+
+	"github.com/vulcanize/vulcanizedb/pkg/core"
+	"github.com/vulcanize/vulcanizedb/pkg/filters"
+	"github.com/vulcanize/vulcanizedb/pkg/omni/shared/constants"
+)
+
+var TransferBlock1 = core.Block{
+	Hash:   "0x135391a0962a63944e5908e6fedfff90fb4be3e3290a21017861099bad123ert",
+	Number: 6194633,
+	Transactions: []core.Transaction{{
+		Hash: "0x135391a0962a63944e5908e6fedfff90fb4be3e3290a21017861099bad654aaa",
+		Receipt: core.Receipt{
+			TxHash:          "0x135391a0962a63944e5908e6fedfff90fb4be3e3290a21017861099bad654aaa",
+			ContractAddress: "",
+			Logs: []core.Log{{
+				BlockNumber: 6194633,
+				TxHash:      "0x135391a0962a63944e5908e6fedfff90fb4be3e3290a21017861099bad654aaa",
+				Address:     constants.TusdContractAddress,
+				Topics: core.Topics{
+					constants.TransferEvent.Signature(),
+					"0x000000000000000000000000000000000000000000000000000000000000af21",
+					"0x9dd48110dcc444fdc242510c09bbbbe21a5975cac061d82f7b843bce061ba391",
+					"",
+				},
+				Index: 1,
+				Data:  "0x000000000000000000000000c02aaa39b223fe8d0a0e5c4f27ead9083c756cc200000000000000000000000089d24a6b4ccb1b6faa2625fe562bdd9a23260359000000000000000000000000000000000000000000000000392d2e2bda9c00000000000000000000000000000000000000000000000000927f41fa0a4a418000000000000000000000000000000000000000000000000000000000005adcfebe",
+			}},
+		},
+	}},
+}
+
+var TransferBlock2 = core.Block{
+	Hash:   "0x135391a0962a63944e5908e6fedfff90fb4be3e3290a21017861099bad123ooo",
+	Number: 6194634,
+	Transactions: []core.Transaction{{
+		Hash: "0x135391a0962a63944e5908e6fedfff90fb4be3e3290a21017861099bad654eee",
+		Receipt: core.Receipt{
+			TxHash:          "0x135391a0962a63944e5908e6fedfff90fb4be3e3290a21017861099bad654eee",
+			ContractAddress: "",
+			Logs: []core.Log{{
+				BlockNumber: 6194634,
+				TxHash:      "0x135391a0962a63944e5908e6fedfff90fb4be3e3290a21017861099bad654eee",
+				Address:     constants.TusdContractAddress,
+				Topics: core.Topics{
+					constants.TransferEvent.Signature(),
+					"0x000000000000000000000000000000000000000000000000000000000000af21",
+					"0x9dd48110dcc444fdc242510c09bbbbe21a5975cac061d82f7b843bce061ba391",
+					"",
+				},
+				Index: 1,
+				Data:  "0x000000000000000000000000c02aaa39b223fe8d0a0e5c4f27ead9083c756cc200000000000000000000000089d24a6b4ccb1b6faa2625fe562bdd9a23260359000000000000000000000000000000000000000000000000392d2e2bda9c00000000000000000000000000000000000000000000000000927f41fa0a4a418000000000000000000000000000000000000000000000000000000000005adcfebe",
+			}},
+		},
+	}},
+}
+
+var ExpectedTransferFilter = filters.LogFilter{
+	Name:      "Transfer",
+	Address:   constants.TusdContractAddress,
+	ToBlock:   -1,
+	FromBlock: 6194634,
+	Topics:    core.Topics{constants.TransferEvent.Signature()},
+}
+
+var ExpectedApprovalFilter = filters.LogFilter{
+	Name:      "Approval",
+	Address:   constants.TusdContractAddress,
+	ToBlock:   -1,
+	FromBlock: 6194634,
+	Topics:    core.Topics{constants.ApprovalEvent.Signature()},
+}
+
+var MockTranferEvent = core.WatchedEvent{
+	LogID:       1,
+	Name:        constants.TransferEvent.String(),
+	BlockNumber: 5488076,
+	Address:     constants.TusdContractAddress,
+	TxHash:      "0x135391a0962a63944e5908e6fedfff90fb4be3e3290a21017861099bad6546ae",
+	Index:       110,
+	Topic0:      constants.TransferEvent.Signature(),
+	Topic1:      "0x000000000000000000000000000000000000000000000000000000000000af21",
+	Topic2:      "0x9dd48110dcc444fdc242510c09bbbbe21a5975cac061d82f7b843bce061ba391",
+	Topic3:      "",
+	Data:        "0x000000000000000000000000c02aaa39b223fe8d0a0e5c4f27ead9083c756cc200000000000000000000000089d24a6b4ccb1b6faa2625fe562bdd9a23260359000000000000000000000000000000000000000000000000392d2e2bda9c00000000000000000000000000000000000000000000000000927f41fa0a4a418000000000000000000000000000000000000000000000000000000000005adcfebe",
+}
+
+var rawFakeHeader, _ = json.Marshal(core.Header{})
+
+var MockHeader1 = core.Header{
+	Hash:        "0x135391a0962a63944e5908e6fedfff90fb4be3e3290a21017861099bad123ert",
+	BlockNumber: 6194632,
+	Raw:         rawFakeHeader,
+	Timestamp:   "50000000",
+}
+
+var MockHeader2 = core.Header{
+	Hash:        "0x135391a0962a63944e5908e6fedfff90fb4be3e3290a21017861099bad456yui",
+	BlockNumber: 6194633,
+	Raw:         rawFakeHeader,
+	Timestamp:   "50000015",
+}
+
+var MockHeader3 = core.Header{
+	Hash:        "0x135391a0962a63944e5908e6fedfff90fb4be3e3290a21017861099bad234hfs",
+	BlockNumber: 6194634,
+	Raw:         rawFakeHeader,
+	Timestamp:   "50000030",
+}

pkg/omni/shared/parser/parser.go

@@ -20,7 +20,7 @@ import (
 	"github.com/ethereum/go-ethereum/accounts/abi"
 
 	"github.com/vulcanize/vulcanizedb/pkg/geth"
-	"github.com/vulcanize/vulcanizedb/pkg/omni/types"
+	"github.com/vulcanize/vulcanizedb/pkg/omni/shared/types"
 )
 
 // Parser is used to fetch and parse contract ABIs

pkg/omni/shared/parser/parser_test.go

@@ -22,9 +22,9 @@ import (
 	. "github.com/onsi/gomega"
 
 	"github.com/vulcanize/vulcanizedb/pkg/geth"
-	"github.com/vulcanize/vulcanizedb/pkg/omni/constants"
-	"github.com/vulcanize/vulcanizedb/pkg/omni/helpers/mocks"
-	"github.com/vulcanize/vulcanizedb/pkg/omni/parser"
+	"github.com/vulcanize/vulcanizedb/pkg/omni/shared/constants"
+	"github.com/vulcanize/vulcanizedb/pkg/omni/shared/helpers/mocks"
+	"github.com/vulcanize/vulcanizedb/pkg/omni/shared/parser"
 )
 
 var _ = Describe("Parser", func() {

pkg/omni/shared/poller/poller.go

@@ -26,9 +26,9 @@ import (
 
 	"github.com/vulcanize/vulcanizedb/pkg/core"
 	"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
-	"github.com/vulcanize/vulcanizedb/pkg/omni/contract"
-	"github.com/vulcanize/vulcanizedb/pkg/omni/repository"
-	"github.com/vulcanize/vulcanizedb/pkg/omni/types"
+	"github.com/vulcanize/vulcanizedb/pkg/omni/shared/contract"
+	"github.com/vulcanize/vulcanizedb/pkg/omni/shared/repository"
+	"github.com/vulcanize/vulcanizedb/pkg/omni/shared/types"
 )
 
 type Poller interface {
@@ -37,7 +37,7 @@ type Poller interface {
 }
 
 type poller struct {
-	repository.MethodDatastore
+	repository.MethodRepository
 	bc       core.BlockChain
 	contract contract.Contract
 }
@@ -45,8 +45,8 @@ type poller struct {
 func NewPoller(blockChain core.BlockChain, db *postgres.DB) *poller {
 
 	return &poller{
-		MethodDatastore: repository.NewMethodDatastore(db),
-		bc:              blockChain,
+		MethodRepository: repository.NewMethodRepository(db),
+		bc:               blockChain,
 	}
 }
 

pkg/omni/shared/poller/poller_test.go

@@ -24,10 +24,10 @@ import (
 
 	"github.com/vulcanize/vulcanizedb/pkg/core"
 	"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
-	"github.com/vulcanize/vulcanizedb/pkg/omni/constants"
-	"github.com/vulcanize/vulcanizedb/pkg/omni/contract"
-	"github.com/vulcanize/vulcanizedb/pkg/omni/helpers/test_helpers"
-	"github.com/vulcanize/vulcanizedb/pkg/omni/poller"
+	"github.com/vulcanize/vulcanizedb/pkg/omni/shared/constants"
+	"github.com/vulcanize/vulcanizedb/pkg/omni/shared/contract"
+	"github.com/vulcanize/vulcanizedb/pkg/omni/shared/helpers/test_helpers"
+	"github.com/vulcanize/vulcanizedb/pkg/omni/shared/poller"
 )
 
 var _ = Describe("Poller", func() {

pkg/omni/shared/repository/method_repository.go

@@ -22,27 +22,27 @@ import (
 	"strings"
 
 	"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
-	"github.com/vulcanize/vulcanizedb/pkg/omni/types"
+	"github.com/vulcanize/vulcanizedb/pkg/omni/shared/types"
 )
 
-type MethodDatastore interface {
+type MethodRepository interface {
 	PersistResult(method types.Result, contractAddr, contractName string) error
 	CreateMethodTable(contractAddr string, method types.Result) (bool, error)
 	CreateContractSchema(contractAddr string) (bool, error)
 }
 
-type methodDatastore struct {
+type methodRepository struct {
 	*postgres.DB
 }
 
-func NewMethodDatastore(db *postgres.DB) *methodDatastore {
+func NewMethodRepository(db *postgres.DB) *methodRepository {
 
-	return &methodDatastore{
+	return &methodRepository{
 		DB: db,
 	}
 }
 
-func (d *methodDatastore) PersistResult(method types.Result, contractAddr, contractName string) error {
+func (d *methodRepository) PersistResult(method types.Result, contractAddr, contractName string) error {
 	if len(method.Args) != len(method.Inputs) {
 		return errors.New("error: given number of inputs does not match number of method arguments")
 	}
@@ -64,7 +64,7 @@ func (d *methodDatastore) PersistResult(method types.Result, contractAddr, contr
 }
 
 // Creates a custom postgres command to persist logs for the given event
-func (d *methodDatastore) persistResult(method types.Result, contractAddr, contractName string) error {
+func (d *methodRepository) persistResult(method types.Result, contractAddr, contractName string) error {
 	// Begin postgres string
 	pgStr := fmt.Sprintf("INSERT INTO c%s.%s_method ", strings.ToLower(contractAddr), strings.ToLower(method.Name))
 	pgStr = pgStr + "(token_name, block"
@@ -103,7 +103,7 @@ func (d *methodDatastore) persistResult(method types.Result, contractAddr, contr
 }
 
 // Checks for event table and creates it if it does not already exist
-func (d *methodDatastore) CreateMethodTable(contractAddr string, method types.Result) (bool, error) {
+func (d *methodRepository) CreateMethodTable(contractAddr string, method types.Result) (bool, error) {
 	tableExists, err := d.checkForTable(contractAddr, method.Name)
 	if err != nil {
 		return false, err
@@ -120,7 +120,7 @@ func (d *methodDatastore) CreateMethodTable(contractAddr string, method types.Re
 }
 
 // Creates a table for the given contract and event
-func (d *methodDatastore) newMethodTable(contractAddr string, method types.Result) error {
+func (d *methodRepository) newMethodTable(contractAddr string, method types.Result) error {
 	// Begin pg string
 	pgStr := fmt.Sprintf("CREATE TABLE IF NOT EXISTS c%s.%s_method ", strings.ToLower(contractAddr), strings.ToLower(method.Name))
 	pgStr = pgStr + "(id SERIAL, token_name CHARACTER VARYING(66) NOT NULL, block INTEGER NOT NULL,"
@@ -138,7 +138,7 @@ func (d *methodDatastore) newMethodTable(contractAddr string, method types.Resul
 }
 
 // Checks if a table already exists for the given contract and event
-func (d *methodDatastore) checkForTable(contractAddr string, methodName string) (bool, error) {
+func (d *methodRepository) checkForTable(contractAddr string, methodName string) (bool, error) {
 	pgStr := fmt.Sprintf("SELECT EXISTS (SELECT 1 FROM information_schema.tables WHERE table_schema = 'c%s' AND table_name = '%s_method')", strings.ToLower(contractAddr), strings.ToLower(methodName))
 	var exists bool
 	err := d.DB.Get(&exists, pgStr)
@@ -147,7 +147,7 @@ func (d *methodDatastore) checkForTable(contractAddr string, methodName string)
 }
 
 // Checks for contract schema and creates it if it does not already exist
-func (d *methodDatastore) CreateContractSchema(contractAddr string) (bool, error) {
+func (d *methodRepository) CreateContractSchema(contractAddr string) (bool, error) {
 	if contractAddr == "" {
 		return false, errors.New("error: no contract address specified")
 	}
@@ -168,14 +168,14 @@ func (d *methodDatastore) CreateContractSchema(contractAddr string) (bool, error
 }
 
 // Creates a schema for the given contract
-func (d *methodDatastore) newContractSchema(contractAddr string) error {
+func (d *methodRepository) newContractSchema(contractAddr string) error {
 	_, err := d.DB.Exec("CREATE SCHEMA IF NOT EXISTS c" + strings.ToLower(contractAddr))
 
 	return err
 }
 
 // Checks if a schema already exists for the given contract
-func (d *methodDatastore) checkForSchema(contractAddr string) (bool, error) {
+func (d *methodRepository) checkForSchema(contractAddr string) (bool, error) {
 	pgStr := fmt.Sprintf("SELECT EXISTS (SELECT schema_name FROM information_schema.schemata WHERE schema_name = 'c%s')", strings.ToLower(contractAddr))
 
 	var exists bool

pkg/omni/shared/repository/method_repository_test.go

@@ -23,16 +23,16 @@ import (
 	. "github.com/onsi/gomega"
 
 	"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
-	"github.com/vulcanize/vulcanizedb/pkg/omni/constants"
-	"github.com/vulcanize/vulcanizedb/pkg/omni/contract"
-	"github.com/vulcanize/vulcanizedb/pkg/omni/helpers/test_helpers"
-	"github.com/vulcanize/vulcanizedb/pkg/omni/repository"
-	"github.com/vulcanize/vulcanizedb/pkg/omni/types"
+	"github.com/vulcanize/vulcanizedb/pkg/omni/shared/constants"
+	"github.com/vulcanize/vulcanizedb/pkg/omni/shared/contract"
+	"github.com/vulcanize/vulcanizedb/pkg/omni/shared/helpers/test_helpers"
+	"github.com/vulcanize/vulcanizedb/pkg/omni/shared/repository"
+	"github.com/vulcanize/vulcanizedb/pkg/omni/shared/types"
 )
 
 var _ = Describe("Repository", func() {
 	var db *postgres.DB
-	var dataStore repository.MethodDatastore
+	var dataStore repository.MethodRepository
 	var con *contract.Contract
 	var err error
 	var mockResult types.Result
@@ -50,7 +50,7 @@ var _ = Describe("Repository", func() {
 		mockResult.Inputs[0] = "0xfE9e8709d3215310075d67E3ed32A380CCf451C8"
 		mockResult.Output = "66386309548896882859581786"
 		db, _ = test_helpers.SetupDBandBC()
-		dataStore = repository.NewMethodDatastore(db)
+		dataStore = repository.NewMethodRepository(db)
 	})
 
 	AfterEach(func() {

pkg/omni/shared/repository/repository_suite_test.go

@@ -0,0 +1,35 @@
+// VulcanizeDB
+// Copyright © 2018 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 (
+	"io/ioutil"
+	"log"
+	"testing"
+
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+)
+
+func TestRepository(t *testing.T) {
+	RegisterFailHandler(Fail)
+	RunSpecs(t, "Shared Repository Suite Test")
+}
+
+var _ = BeforeSuite(func() {
+	log.SetOutput(ioutil.Discard)
+})

pkg/omni/shared/transformer/interface.go

@@ -0,0 +1,29 @@
+// VulcanizeDB
+// Copyright © 2018 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 transformer
+
+// Used to extract any/all events and a subset of method (state variable)
+// data for any contract and persists it to custom postgres tables in vDB
+type Transformer interface {
+	SetEvents(contractAddr string, filterSet []string)
+	SetEventAddrs(contractAddr string, filterSet []string)
+	SetMethods(contractAddr string, filterSet []string)
+	SetMethodAddrs(contractAddr string, filterSet []string)
+	SetRange(contractAddr string, rng []int64)
+	Init() error
+	Execute() error
+}

pkg/omni/shared/types/event.go

@@ -0,0 +1,98 @@
+// VulcanizeDB
+// Copyright © 2018 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 types
+
+import (
+	"fmt"
+	"strings"
+
+	"github.com/ethereum/go-ethereum/accounts/abi"
+	"github.com/i-norden/go-ethereum/core/types"
+)
+
+type Event struct {
+	Name      string
+	Anonymous bool
+	Fields    []Field
+}
+
+type Field struct {
+	abi.Argument        // Name, Type, Indexed
+	PgType       string // Holds type used when committing data held in this field to postgres
+}
+
+// Struct to hold instance of an event log data
+type Log struct {
+	Event
+	Id     int64             // VulcanizeIdLog for full sync and header ID for light sync omni watcher
+	Values map[string]string // Map of event input names to their values
+
+	// Used for full sync only
+	Block int64
+	Tx    string
+
+	// Used for lightSync only
+	LogIndex         uint
+	TransactionIndex uint
+	Raw              types.Log
+}
+
+// Unpack abi.Event into our custom Event struct
+func NewEvent(e abi.Event) Event {
+	fields := make([]Field, len(e.Inputs))
+	for i, input := range e.Inputs {
+		fields[i] = Field{}
+		fields[i].Name = input.Name
+		fields[i].Type = input.Type
+		fields[i].Indexed = input.Indexed
+		// Fill in pg type based on abi type
+		switch fields[i].Type.T {
+		case abi.StringTy, abi.HashTy, abi.AddressTy:
+			fields[i].PgType = "CHARACTER VARYING(66)"
+		case abi.IntTy, abi.UintTy:
+			fields[i].PgType = "DECIMAL"
+		case abi.BoolTy:
+			fields[i].PgType = "BOOLEAN"
+		case abi.BytesTy, abi.FixedBytesTy:
+			fields[i].PgType = "BYTEA"
+		case abi.ArrayTy:
+			fields[i].PgType = "TEXT[]"
+		case abi.FixedPointTy:
+			fields[i].PgType = "MONEY" // use shopspring/decimal for fixed point numbers in go and money type in postgres?
+		case abi.FunctionTy:
+			fields[i].PgType = "TEXT"
+		default:
+			fields[i].PgType = "TEXT"
+		}
+	}
+
+	return Event{
+		Name:      e.Name,
+		Anonymous: e.Anonymous,
+		Fields:    fields,
+	}
+}
+
+func (e Event) Sig() string {
+	types := make([]string, len(e.Fields))
+
+	for i, input := range e.Fields {
+		types[i] = input.Type.String()
+	}
+
+	return fmt.Sprintf("%v(%v)", e.Name, strings.Join(types, ","))
+}

pkg/omni/shared/types/method.go

@@ -23,12 +23,6 @@ import (
 	"github.com/ethereum/go-ethereum/accounts/abi"
 )
 
-type Event struct {
-	Name      string
-	Anonymous bool
-	Fields    []Field
-}
-
 type Method struct {
 	Name   string
 	Const  bool
@@ -36,11 +30,6 @@ type Method struct {
 	Return []Field
 }
 
-type Field struct {
-	abi.Argument        // Name, Type, Indexed
-	PgType       string // Holds type used when committing data held in this field to postgres
-}
-
 // Struct to hold instance of result from method call with given inputs and block
 type Result struct {
 	Method
@@ -50,51 +39,6 @@ type Result struct {
 	Block  int64
 }
 
-// Struct to hold instance of an event log data
-type Log struct {
-	Event
-	Id     int64             // VulcanizeIdLog
-	Values map[string]string // Map of event input names to their values
-	Block  int64
-	Tx     string
-}
-
-// Unpack abi.Event into our custom Event struct
-func NewEvent(e abi.Event) Event {
-	fields := make([]Field, len(e.Inputs))
-	for i, input := range e.Inputs {
-		fields[i] = Field{}
-		fields[i].Name = input.Name
-		fields[i].Type = input.Type
-		fields[i].Indexed = input.Indexed
-		// Fill in pg type based on abi type
-		switch fields[i].Type.T {
-		case abi.StringTy, abi.HashTy, abi.AddressTy:
-			fields[i].PgType = "CHARACTER VARYING(66)"
-		case abi.IntTy, abi.UintTy:
-			fields[i].PgType = "DECIMAL"
-		case abi.BoolTy:
-			fields[i].PgType = "BOOLEAN"
-		case abi.BytesTy, abi.FixedBytesTy:
-			fields[i].PgType = "BYTEA"
-		case abi.ArrayTy:
-			fields[i].PgType = "TEXT[]"
-		case abi.FixedPointTy:
-			fields[i].PgType = "MONEY" // use shopspring/decimal for fixed point numbers in go and money type in postgres?
-		case abi.FunctionTy:
-			fields[i].PgType = "TEXT"
-		default:
-			fields[i].PgType = "TEXT"
-		}
-	}
-
-	return Event{
-		Name:      e.Name,
-		Anonymous: e.Anonymous,
-		Fields:    fields,
-	}
-}
-
 // Unpack abi.Method into our custom Method struct
 func NewMethod(m abi.Method) Method {
 	inputs := make([]Field, len(m.Inputs))
@@ -157,16 +101,6 @@ func NewMethod(m abi.Method) Method {
 	}
 }
 
-func (e Event) Sig() string {
-	types := make([]string, len(e.Fields))
-
-	for i, input := range e.Fields {
-		types[i] = input.Type.String()
-	}
-
-	return fmt.Sprintf("%v(%v)", e.Name, strings.Join(types, ","))
-}
-
 func (m Method) Sig() string {
 	types := make([]string, len(m.Args))
 	i := 0