add baseline rewards to event (#5)

* add baseline rewards to event

* pass flag to process on different chains

* add abis

* store chain and only most recently processed block

* updated postgres

* pass chains in settings, update config to store all chain-related info

* move event definition

* WIP process chains simultaneously, write events to same topic

* all for single shutdown command across all channels

* indicate chain in logs

* add chain id to blocks table

* use error groups

* sql fix

* fix migration to drop unneeded rows

* set chain id to polygon

* types

* One chain per process. Extreme Helm configuration

* Fix some Helm references

---------

Co-authored-by: Dylan Moreland <[email protected]>

Author: Allyson-English

abi/baseline_rewards.json

@@ -0,0 +1,9 @@
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: {{ include "contract-event-processor.fullname" . }}-abi-config-map
+  namespace: {{ .Release.Namespace }}
+  labels:
+    {{- include "contract-event-processor.labels" . | nindent 4 }}
+data:
+{{ tpl (.Files.Glob "abi/*").AsConfig $ | indent 2 }}

abi/tokenMainnet.json

@@ -0,0 +1,9 @@
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: {{ include "contract-event-processor.fullname" . }}-config-config-map
+  namespace: {{ .Release.Namespace }}
+  labels:
+{{ include "contract-event-processor.fullname" . | indent 4 }}
+data:
+  config.yaml: {{ .Values.config | toYaml | quote }}

abi/tokenMumbai.json

@@ -70,6 +70,19 @@ spec:
               port: mon-http
           resources:
             {{- toYaml .Values.resources | nindent 12 }}
+          volumeMounts:
+            - name: abi
+              mountPath: /abi
+            - name: config
+              mountPath: /config.yaml
+              subPath: config.yaml
+      volumes:
+        - name: abi
+          configMap:
+            name: {{ include "contract-event-processor.fullname" . }}-abi-config-map
+        - name: config
+          configMap:
+            name: {{ include "contract-event-processor.fullname" . }}-config-config-map
       {{- with .Values.nodeSelector }}
       nodeSelector:
         {{- toYaml . | nindent 8 }}

abi/tokenPolygon.json

@@ -17,6 +17,9 @@ spec:
   - remoteRef:
       key: {{ .Release.Namespace }}/contract_event/db/pass
     secretKey: DB_PASSWORD
+  - remoteRef:
+      key: {{ .Release.Namespace }}/contract_event/rpc_url/{{ .Values.config.chain }}
+    secretKey: BLOCKCHAIN_RPC_URL
   secretStoreRef:
     kind: ClusterSecretStore
     name: aws-secretsmanager-secret-store

charts/contract-event-processor/abi/Ethereum_Dimo.json

@@ -9,6 +9,13 @@ env:
   ENVIRONMENT: prod
   KAFKA_BROKERS: kafka-prod-dimo-kafka-kafka-brokers:9092
   BLOCK_CONFIRMATIONS: 5
+config:
+  chain: polygon
+  contracts:
+    - address: '0xe261d618a959afffd53168cd07d12e37b26761db'
+      abi: Polygon_Dimo.json
+    - address: '0xba5738a18d83d41847dffbdc6101d37c69c9b0cf'
+      abi: VehicleId.json
 ingress:
   enabled: false
 resources:

charts/contract-event-processor/abi/Mumbai_OmidV2.json

@@ -23,6 +23,13 @@ securityContext:
 cronJob:
   enabled: false
   schedule: 0 0 * * 0
+config:
+  chain: mumbai
+  contracts:
+    - address: '0x80ee7ec4493a1d7975ab900f94db25ba7c688201'
+      abi: Mumbai_OmidV2.json
+    - address: '0x90c4d6113ec88dd4bdf12f26db2b3998fd13a144'
+      abi: VehicleId.json
 env:
   ENVIRONMENT: dev
   MONITORING_PORT: 8888

charts/contract-event-processor/abi/Polygon_Dimo.json

@@ -1,14 +1,17 @@
 package main
 
 import (
-	"fmt"
+	"context"
+	"flag"
 	"log"
 	"math/big"
 	"os"
-	"strconv"
+	"os/signal"
+	"syscall"
 
 	"github.com/DIMO-Network/contract-event-processor/internal/config"
 	"github.com/DIMO-Network/contract-event-processor/internal/services"
+	"golang.org/x/sync/errgroup"
 
 	"github.com/DIMO-Network/shared"
 	"github.com/Shopify/sarama"
@@ -19,13 +22,18 @@ import (
 )
 
 func main() {
-	logger := zerolog.New(os.Stdout).With().Timestamp().Str("app", "event-stream-processor").Logger()
+	logger := zerolog.New(os.Stdout).With().Timestamp().Str("app", "contract-stream-processor").Logger()
 	settings, err := shared.LoadConfig[config.Settings]("settings.yaml")
 	if err != nil {
-		logger.Fatal().Err(err)
+		logger.Fatal().Err(err).Msg("couldn't load settings")
 	}
 
-	var blockNum *big.Int
+	ctx, cancel := context.WithCancel(context.Background())
+	var group errgroup.Group
+
+	limit := flag.Int("limit", -1, "limit number of block iterations during development")
+	flag.Parse()
+
 	if len(os.Args) > 1 {
 		switch subCommand := os.Args[1]; subCommand {
 		case "migrate":
@@ -38,19 +46,10 @@ func main() {
 			}
 			migrateDatabase(logger, &settings, command)
 			return
-		case "override":
-			if len(os.Args) > 2 {
-				n, err := strconv.Atoi(os.Args[2])
-				if err != nil {
-					logger.Fatal().Err(err)
-				}
-				blockNum = big.NewInt(int64(n))
-
-			}
 		}
 	}
 
-	monApp := serveMonitoring(settings.MonitoringPort, &logger)
+	group.Go(func() error { return serveMonitoring(ctx, settings.MonitoringPort, &logger) })
 
 	kafkaClient, err := services.StartKafkaStream(settings)
 	if err != nil {
@@ -63,19 +62,29 @@ func main() {
 		log.Fatal(err)
 	}
 
-	listener, err := services.NewBlockListener(settings, logger, producer)
+	listener, err := services.NewBlockListener(settings, logger, producer, "config.yaml")
 	if err != nil {
-		log.Fatal(err)
+		logger.Fatal().Err(err).Msg("Failed creating block listener.")
 	}
 
-	listener.CompileRegistryMap(fmt.Sprintf("config-%s.yaml", settings.Environment))
-	listener.ChainIndexer(blockNum)
+	listener.Limit = *limit
+	if listener.Limit > 0 {
+		listener.DevTest = true
+	}
+
+	chainChan := make(chan *big.Int)
+	group.Go(func() error { return listener.PollNewBlocks(ctx, listener.StartBlock, chainChan) })
+	group.Go(func() error { return listener.ProcessBlocks(ctx, chainChan) })
+
+	sc := make(chan os.Signal, 1)
+	signal.Notify(sc, os.Interrupt, syscall.SIGTERM)
+	<-sc
+	cancel()
+	_ = group.Wait()
 
-	// TODO(elffjs): Log this.
-	_ = monApp.Shutdown()
 }
 
-func serveMonitoring(port string, logger *zerolog.Logger) *fiber.App {
+func serveMonitoring(ctx context.Context, port string, logger *zerolog.Logger) error {
 	monApp := fiber.New(fiber.Config{DisableStartupMessage: true})
 
 	// Health check.
@@ -89,6 +98,7 @@ func serveMonitoring(port string, logger *zerolog.Logger) *fiber.App {
 	}()
 
 	logger.Info().Str("port", port).Msg("Started monitoring web server.")
+	<-ctx.Done()
 
-	return monApp
+	return monApp.Shutdown()
 }

charts/contract-event-processor/abi/VehicleId.json

@@ -1,3 +1,35 @@
 contracts:
   - address: 0x90C4D6113Ec88dd4BDf12f26DB2b3998fd13A144
+    chain: polygon
     abi: abi/VehicleId.json
+  - address: 0xE261D618a959aFfFd53168Cd07D12E37B26761db
+    chain: polygon
+    abi: abi/tokenPolygon.json
+  - address: 0x5fab9761d60419c9eeebe3915a8fa1ed7e8d2e1b
+    chain: ethereum
+    abi: abi/tokenMainnet.json
+  - address: 0x80Ee7ec4493A1d7975ab900F94dB25ba7C688201
+    chain: mumbai
+    abi: abi/tokenMumbai.json
+
+chains:
+  - chain: polygon
+    rpc_url: wss://polygon-mainnet.g.alchemy.com/v2/
+    startBlock:
+    contracts:
+      - address: 0x90C4D6113Ec88dd4BDf12f26DB2b3998fd13A144
+        abi: abi/VehicleId.json
+      - address: 0xE261D618a959aFfFd53168Cd07D12E37B26761db
+        abi: abi/tokenPolygon.json
+  - chain: ethereum
+    rpc_url: wss://eth-mainnet.ws.alchemyapi.io/v2/
+    startBlock:
+    contracts:
+      - address: 0x5fab9761d60419c9eeebe3915a8fa1ed7e8d2e1b
+        abi: abi/tokenMainnet.json
+  - chain: mumbai
+    rpc_url: wss://polygon-mumbai.g.alchemy.com/v2/
+    startBlock:
+    contracts:
+      - address: 0x80Ee7ec4493A1d7975ab900F94dB25ba7C688201
+        abi: abi/tokenMumbai.json

charts/contract-event-processor/templates/abi-configmap.yaml

@@ -1,3 +1,12 @@
 contracts:
-  - address: 0xbA5738a18d83D41847dfFbDC6101d37C69c9B0cF
+  - address: 0x90C4D6113Ec88dd4BDf12f26DB2b3998fd13A144
     abi: abi/VehicleId.json
+  - address: 0xE261D618a959aFfFd53168Cd07D12E37B26761db
+    chain: polygon
+    abi: abi/tokenPolygon.json
+  - address: 0x5fab9761d60419c9eeebe3915a8fa1ed7e8d2e1b
+    chain: ethereum
+    abi: abi/tokenMainnet.json
+  - address: 0x80Ee7ec4493A1d7975ab900F94dB25ba7C688201
+    chain: mumbai
+    abi: abi/tokenMumbai.json

charts/contract-event-processor/templates/config-configmap.yaml

@@ -13,6 +13,7 @@ require (
 	github.com/volatiletech/null/v8 v8.1.2
 	github.com/volatiletech/sqlboiler/v4 v4.14.0
 	github.com/volatiletech/strmangle v0.0.4
+	golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4
 	gopkg.in/yaml.v3 v3.0.1
 )
 

charts/contract-event-processor/templates/deployment.yaml

@@ -3,14 +3,18 @@ package config
 import "github.com/DIMO-Network/shared/db"
 
 type Settings struct {
-	Environment        string `yaml:"ENVIRONMENT"`
-	EthereumRPCURL     string `yaml:"ETHEREUM_RPC_URL"`
-	BlockConfirmations int    `yaml:"BLOCK_CONFIRMATIONS"`
+	Environment string `yaml:"ENVIRONMENT"`
+
+	BlockConfirmations int64  `yaml:"BLOCK_CONFIRMATIONS"`
 	ContractEventTopic string `yaml:"CONTRACT_EVENT_TOPIC"`
 
 	KafkaBrokers string `yaml:"KAFKA_BROKERS"`
 
 	DB db.Settings `yaml:"DB"`
 
 	MonitoringPort string `yaml:"MONITORING_PORT"`
+
+	BlockchainRPCURL string `yaml:"BLOCKCHAIN_RPC_URL"`
+
+	APIKey string `yaml:"ALCHEMY_API_KEY"`
 }