Tidy up packages, settings, database connection

Author: elffjs

cmd/contract-event-processor/main.go

@@ -6,7 +6,8 @@ import (
 	"os"
 	"strconv"
 
-	services "event-stream/internal"
+	"github.com/DIMO-Network/contract-event-processor/internal/config"
+	"github.com/DIMO-Network/contract-event-processor/internal/services"
 
 	"github.com/DIMO-Network/shared"
 	"github.com/Shopify/sarama"
@@ -18,7 +19,7 @@ import (
 
 func main() {
 	logger := zerolog.New(os.Stdout).With().Timestamp().Str("app", "event-stream-processor").Logger()
-	settings, err := shared.LoadConfig[services.Settings]("settings.yaml")
+	settings, err := shared.LoadConfig[config.Settings]("settings.yaml")
 	if err != nil {
 		logger.Fatal().Err(err)
 	}

go.mod

@@ -1,13 +1,16 @@
-module event-stream
+module github.com/DIMO-Network/contract-event-processor
 
 go 1.19
 
 require (
 	github.com/Shopify/sarama v1.33.0
 	github.com/ethereum/go-ethereum v1.10.26
 	github.com/friendsofgo/errors v0.9.2
+	github.com/gofiber/adaptor/v2 v2.1.31
+	github.com/gofiber/fiber/v2 v2.41.0
 	github.com/lib/pq v1.10.7
 	github.com/pressly/goose/v3 v3.7.0
+	github.com/prometheus/client_golang v1.12.1
 	github.com/volatiletech/null/v8 v8.1.2
 	github.com/volatiletech/sqlboiler/v4 v4.14.0
 	github.com/volatiletech/strmangle v0.0.4
@@ -16,17 +19,21 @@ require (
 
 require (
 	github.com/Shopify/toxiproxy/v2 v2.4.0 // indirect
+	github.com/andybalholm/brotli v1.0.4 // indirect
 	github.com/avast/retry-go/v4 v4.1.0 // indirect
 	github.com/aws/aws-sdk-go-v2 v1.16.5 // indirect
 	github.com/aws/aws-sdk-go-v2/internal/configsources v1.1.12 // indirect
 	github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.4.6 // indirect
 	github.com/aws/aws-sdk-go-v2/service/kms v1.17.3 // indirect
 	github.com/aws/smithy-go v1.11.3 // indirect
+	github.com/beorn7/perks v1.0.1 // indirect
+	github.com/cespare/xxhash/v2 v2.1.2 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/eapache/go-resiliency v1.2.0 // indirect
 	github.com/eapache/go-xerial-snappy v0.0.0-20180814174437-776d5712da21 // indirect
 	github.com/eapache/queue v1.1.0 // indirect
 	github.com/gofrs/uuid v4.2.0+incompatible // indirect
+	github.com/golang/protobuf v1.5.2 // indirect
 	github.com/golang/snappy v0.0.4 // indirect
 	github.com/hashicorp/errwrap v1.0.0 // indirect
 	github.com/hashicorp/go-multierror v1.1.1 // indirect
@@ -36,18 +43,28 @@ require (
 	github.com/jcmturner/gofork v1.0.0 // indirect
 	github.com/jcmturner/gokrb5/v8 v8.4.2 // indirect
 	github.com/jcmturner/rpc/v2 v2.0.3 // indirect
-	github.com/klauspost/compress v1.15.6 // indirect
-	github.com/mattn/go-colorable v0.1.12 // indirect
-	github.com/mattn/go-isatty v0.0.16 // indirect
+	github.com/klauspost/compress v1.15.9 // indirect
+	github.com/mattn/go-colorable v0.1.13 // indirect
+	github.com/mattn/go-isatty v0.0.17 // indirect
+	github.com/mattn/go-runewidth v0.0.14 // indirect
+	github.com/matttproud/golang_protobuf_extensions v1.0.1 // indirect
 	github.com/pierrec/lz4 v2.6.1+incompatible // indirect
 	github.com/pkg/errors v0.9.1 // indirect
+	github.com/prometheus/client_model v0.2.0 // indirect
+	github.com/prometheus/common v0.32.1 // indirect
+	github.com/prometheus/procfs v0.7.3 // indirect
 	github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475 // indirect
+	github.com/rivo/uniseg v0.2.0 // indirect
 	github.com/spf13/cast v1.5.0 // indirect
+	github.com/valyala/bytebufferpool v1.0.0 // indirect
+	github.com/valyala/fasthttp v1.44.0 // indirect
+	github.com/valyala/tcplisten v1.0.0 // indirect
 	github.com/volatiletech/inflect v0.0.1 // indirect
 	github.com/volatiletech/randomize v0.0.1 // indirect
 	golang.org/x/net v0.1.0 // indirect
 	golang.org/x/tools v0.2.0 // indirect
 	golang.org/x/xerrors v0.0.0-20220609144429-65e65417b02f // indirect
+	google.golang.org/protobuf v1.28.0 // indirect
 )
 
 require (

go.sum

@@ -0,0 +1,18 @@
+package config
+
+import "github.com/DIMO-Network/shared/db"
+
+type Settings struct {
+	EthereumRPCURL string `yaml:"ETHEREUM_RPC_URL"`
+
+	BlockConfirmations int `yaml:"BLOCK_CONFIRMATIONS"`
+
+	EventStreamTopic string `yaml:"EVENT_STREAM_TOPIC"`
+
+	KafkaBroker string `yaml:"KAFKA_BROKER"`
+	Partitions  int    `yaml:"PARTITIONS"`
+
+	DB db.Settings `yaml:"DB"`
+
+	MonitoringPort string `yaml:"MONITORING_PORT"`
+}

internal/config/settings.go

@@ -14,7 +14,10 @@ import (
 	"syscall"
 	"time"
 
+	"github.com/DIMO-Network/contract-event-processor/internal/config"
+	"github.com/DIMO-Network/contract-event-processor/models"
 	"github.com/DIMO-Network/shared"
+	"github.com/DIMO-Network/shared/db"
 	"github.com/Shopify/sarama"
 	ethereum "github.com/ethereum/go-ethereum"
 	"github.com/ethereum/go-ethereum/accounts/abi"
@@ -28,30 +31,8 @@ import (
 	"gopkg.in/yaml.v3"
 
 	_ "github.com/lib/pq"
-
-	"event-stream/models"
 )
 
-type Settings struct {
-	WebSocketAddress string `yaml:"WEB_SOCKET_ADDRESS"`
-	AlchemyAPIKey    string `yaml:"API_KEY"`
-
-	BlockConfirmations int `yaml:"BLOCK_CONFIRMATIONS"`
-
-	EventStreamTopic string `yaml:"EVENT_STREAM_TOPIC"`
-
-	KafkaBroker string `yaml:"KAFKA_BROKER"`
-	Partitions  int    `yaml:"PARTITIONS"`
-
-	PostgresUser     string `yaml:"POSTGRES_USER"`
-	PostgresPassword string `yaml:"POSTGRES_PASSWORD"`
-	PostgresDB       string `yaml:"POSTGRES_DB"`
-	PostgresHOST     string `yaml:"POSTGRES_HOST"`
-	PostgresPort     int    `yaml:"POSTGRES_PORT"`
-
-	MonitoringPort string `yaml:"MONITORING_PORT"`
-}
-
 type BlockListener struct {
 	Client           *ethclient.Client
 	Contracts        []common.Address
@@ -60,7 +41,7 @@ type BlockListener struct {
 	EventStreamTopic string
 	Registry         map[common.Address]map[common.Hash]abi.Event
 	Confirmations    *big.Int
-	DB               *sql.DB
+	DB               db.Store
 	ABIs             map[common.Address]abi.ABI
 }
 
@@ -78,32 +59,21 @@ type Block struct {
 	Number *big.Int
 }
 
-func NewBlockListener(s Settings, logger zerolog.Logger, producer sarama.SyncProducer) (BlockListener, error) {
-	c, err := ethclient.Dial(s.WebSocketAddress + s.AlchemyAPIKey)
+func NewBlockListener(s config.Settings, logger zerolog.Logger, producer sarama.SyncProducer) (BlockListener, error) {
+	c, err := ethclient.Dial(s.EthereumRPCURL)
 	if err != nil {
 		return BlockListener{}, err
 	}
 
-	psqlInfo := fmt.Sprintf("host=%s port=%d user=%s "+
-		"password=%s dbname=%s sslmode=disable",
-		s.PostgresHOST, s.PostgresPort, s.PostgresUser, s.PostgresPassword, s.PostgresDB)
-
-	pg, err := sql.Open("postgres", psqlInfo)
-	if err != nil {
-		return BlockListener{}, err
-	}
-	err = pg.Ping()
-	if err != nil {
-		return BlockListener{}, err
-	}
+	pdb := db.NewDbConnectionFromSettings(context.TODO(), &s.DB, true)
 
 	return BlockListener{
 		Client:           c,
 		EventStreamTopic: s.EventStreamTopic,
 		Logger:           logger,
 		Producer:         producer,
 		Confirmations:    big.NewInt(int64(s.BlockConfirmations)),
-		DB:               pg,
+		DB:               pdb,
 	}, nil
 }
 
@@ -149,7 +119,7 @@ func (bl *BlockListener) GetBlockHead(blockNum *big.Int) (*types.Header, error)
 		return bl.Client.HeaderByNumber(context.Background(), blockNum)
 	}
 
-	resp, err := models.Blocks(qm.OrderBy(models.BlockColumns.Number+" DESC")).One(context.Background(), bl.DB)
+	resp, err := models.Blocks(qm.OrderBy(models.BlockColumns.Number+" DESC")).One(context.Background(), bl.DB.DBS().Reader)
 	if err != nil {
 		if errors.Is(err, sql.ErrNoRows) {
 			bl.Logger.Info().Msg("no value passed or found in db; setting block head to current head minus five")
@@ -178,7 +148,7 @@ func (bl *BlockListener) RecordBlock(block *types.Header) error {
 		Hash:   block.Hash().Bytes(),
 	}
 
-	return processedBlock.Insert(context.Background(), bl.DB, boil.Infer())
+	return processedBlock.Insert(context.Background(), bl.DB.DBS().Writer, boil.Infer())
 }
 
 func (bl *BlockListener) ChainIndexer(blockNum *big.Int) {

internal/block_listener.go renamed to internal/services/block_listener.go

@@ -1,6 +1,7 @@
 package services
 
 import (
+	"github.com/DIMO-Network/contract-event-processor/internal/config"
 	"github.com/Shopify/sarama"
 )
 
@@ -13,7 +14,7 @@ type Event struct {
 	EventName       string         `json:"eventName,omitempty"`
 }
 
-func StartKafkaStream(s Settings) (sarama.Client, error) {
+func StartKafkaStream(s config.Settings) (sarama.Client, error) {
 
 	config := sarama.NewConfig()
 	config.Producer.Return.Successes = true

internal/kafka_service.go renamed to internal/services/kafka_service.go

@@ -4,24 +4,27 @@ import (
 	"database/sql"
 	"fmt"
 
+	"github.com/DIMO-Network/contract-event-processor/internal/config"
 	"github.com/pressly/goose/v3"
 	"github.com/rs/zerolog"
 
 	_ "github.com/lib/pq"
 )
 
-func MigrateDatabase(logger zerolog.Logger, s *Settings, command, schemaName string) {
+func MigrateDatabase(logger zerolog.Logger, s *config.Settings, command, schemaName string) {
+	var db *sql.DB
 	// setup database
-	psqlInfo := fmt.Sprintf("host=%s port=%d user=%s "+
-		"password=%s dbname=%s sslmode=disable",
-		s.PostgresHOST, s.PostgresPort, s.PostgresUser, s.PostgresPassword, s.PostgresDB)
-
-	pg, err := sql.Open("postgres", psqlInfo)
+	db, err := sql.Open("postgres", s.DB.BuildConnectionString(true))
+	defer func() {
+		if err := db.Close(); err != nil {
+			logger.Fatal().Msgf("goose: failed to close DB: %v\n", err)
+		}
+	}()
 	if err != nil {
-		logger.Fatal().Msgf("failed to establish db connection: %v\n", err)
+		logger.Fatal().Msgf("failed to open db connection: %v\n", err)
 	}
 
-	if err = pg.Ping(); err != nil {
+	if err = db.Ping(); err != nil {
 		logger.Fatal().Msgf("failed to ping db: %v\n", err)
 	}
 
@@ -30,12 +33,12 @@ func MigrateDatabase(logger zerolog.Logger, s *Settings, command, schemaName str
 		command = "up"
 	}
 	// must create schema so that can set migrations table to that schema
-	_, err = pg.Exec(fmt.Sprintf("CREATE SCHEMA IF NOT EXISTS %s;", schemaName))
+	_, err = db.Exec(fmt.Sprintf("CREATE SCHEMA IF NOT EXISTS %s;", schemaName))
 	if err != nil {
 		logger.Fatal().Err(err).Msgf("could not create schema, %s", schemaName)
 	}
 	goose.SetTableName(fmt.Sprintf("%s.migrations", schemaName))
-	if err := goose.Run(command, pg, "migrations"); err != nil {
+	if err := goose.Run(command, db, "migrations"); err != nil {
 		logger.Fatal().Msgf("failed to apply go code migrations: %v\n", err)
 	}
 	// if we add any code migrations import _ "github.com/DIMO-Network/users-api/migrations" // migrations won't work without this

internal/migrate.go renamed to internal/services/migrate.go

@@ -1,5 +1,4 @@
-WEB_SOCKET_ADDRESS: wss://polygon-mainnet.g.alchemy.com/v2/
-API_KEY:
+ETHEREUM_RPC_URL: wss://polygon-mainnet.g.alchemy.com/v2/
 EVENT_STREAM_TOPIC: topic.contract.event
 PARTITIONS: 1
 KAFKA_BROKER: localhost:9092