WIP continue on aggregate fetching architecture

libraries/shared/watcher.go

@@ -1,9 +1,10 @@
 package shared
 
 import (
-	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/ethereum/go-ethereum/common"
 	"github.com/vulcanize/vulcanizedb/pkg/core"
 	"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
+	"github.com/vulcanize/vulcanizedb/pkg/transformers"
 	"github.com/vulcanize/vulcanizedb/pkg/transformers/shared"
 )
 
@@ -11,6 +12,35 @@ type Watcher struct {
 	Transformers []shared.Transformer
 	DB           postgres.DB
 	Blockchain   core.BlockChain
+	Fetcher      shared.LogFetcher
+	Chunker      shared.LogChunker
+	Addresses    []common.Address
+	Topics       []common.Hash
+}
+
+func NewWatcher(db postgres.DB, bc core.BlockChain) Watcher {
+	transformerConfigs := transformers.TransformerConfigs()
+	var contractAddresses []common.Address
+	var topic0s []common.Hash
+
+	for _, config := range transformerConfigs {
+		for _, address := range config.ContractAddresses {
+			contractAddresses = append(contractAddresses, common.HexToAddress(address))
+		}
+		topic0s = append(topic0s, common.HexToHash(config.Topic))
+	}
+
+	chunker := shared.NewLogChunker(transformerConfigs)
+	fetcher := shared.NewFetcher(bc)
+
+	return Watcher{
+		DB: db,
+		Blockchain: bc,
+		Fetcher: fetcher,
+		Chunker: chunker,
+		Addresses: contractAddresses,
+		Topics: topic0s,
+	}
 }
 
 func (watcher *Watcher) AddTransformers(us []shared.TransformerInitializer) {
@@ -23,14 +53,24 @@ func (watcher *Watcher) AddTransformers(us []shared.TransformerInitializer) {
 func (watcher *Watcher) Execute() error {
 	// TODO Solve checkedHeadersColumn issue
 	// TODO Handle start and end numbers in transformers?
-	var missingHeaders []core.Header
+	missingHeaders, err := shared.MissingHeaders(0, -1, &watcher.DB, "")
 
-	// TODO Get contract addresses and topic0s
-	var logs []types.Log
+	for _, header := range missingHeaders {
+		// TODO Extend FetchLogs for doing several blocks at a time
+		logs, err := watcher.Fetcher.FetchLogs(watcher.Addresses, watcher.Topics, header)
+		if err != nil {
+			// TODO Handle fetch error in watcher
+			return err
+		}
 
-	var err error
-	for _, transformer := range watcher.Transformers {
-		err = transformer.Execute(logs, missingHeaders)
+		chunkedLogs := watcher.Chunker.ChunkLogs(logs)
+
+		for _, transformer := range watcher.Transformers {
+			// TODO delegate log chunks to respective transformers
+			// Need to get the transformer name... :/
+			logChunk := chunkedLogs["transformerName"]
+			err = transformer.Execute(logChunk, missingHeaders)
+		}
 	}
 	return err
 }