[event-watcher] Create service boilerplate

Makefile

@@ -1,31 +1,21 @@
-SHELL := /bin/bash
-
-
-## help: print this help message
-.PHONY: help
-help:
- @echo 'Usage:'
- @sed -n 's/^##//p' ${MAKEFILE_LIST} | column -t -s ':' | sed -e 's/^/ /'
-
 build:
  make -C analytics/ build
  make -C api/ build
+ make -C core-contract-watcher/ build
+ make -C contract-watcher/ build
  make -C fly/ build
- make -C spy/ build
  make -C parser/ build
+ make -C spy/ build
  make -C tx-tracker/ build
- make -C contract-watcher/ build
 
-doc:
- swag init -pd
-
 test:
  cd analytics && go test -v -cover ./...
  cd api && go test -v -cover ./...
+ cd core-contract-watcher && go test -v -cover ./...
+ cd contract-watcher && go test -v -cover ./...
  cd fly && go test -v -cover ./...
- cd spy && go test -v -cover ./...
  cd parser && go test -v -cover ./...
+ cd spy && go test -v -cover ./...
  cd tx-tracker && go test -v -cover ./...
- cd contract-watcher && go test -v -cover ./...
 
-.PHONY: build doc test
+.PHONY: build test

common/health/http.go

@@ -0,0 +1,119 @@
+package health
+
+import (
+ "fmt"
+
+ "github.com/gofiber/fiber/v2"
+ "github.com/gofiber/fiber/v2/middleware/pprof"
+ "go.uber.org/zap"
+)
+
+type Server struct {
+ app *fiber.App
+ port string
+ logger *zap.Logger
+}
+
+func NewServer(logger *zap.Logger, port string, pprofEnabled bool, checks ...Check) *Server {
+
+ app := fiber.New(fiber.Config{DisableStartupMessage: true})
+
+ // config use of middlware.
+ if pprofEnabled {
+ app.Use(pprof.New())
+ }
+
+ ctrl := newController(checks, logger)
+ api := app.Group("/api")
+ api.Get("/health", ctrl.healthCheck)
+ api.Get("/ready", ctrl.readinessCheck)
+
+ return &Server{
+ app: app,
+ port: port,
+ logger: logger,
+ }
+}
+
+// Start initiates the serving of HTTP requests.
+func (s *Server) Start() {
+
+ addr := ":" + s.port
+ s.logger.Info("Monitoring server starting", zap.String("bindAddress", addr))
+
+ go func() {
+ err := s.app.Listen(addr)
+ if err != nil {
+ s.logger.Error("Failed to start monitoring server", zap.Error(err), zap.String("bindAddress", addr))
+ }
+ }()
+}
+
+// Stop gracefully shuts down the server.
+//
+// Blocks until all active connections are closed.
+func (s *Server) Stop() {
+ _ = s.app.Shutdown()
+}
+
+type controller struct {
+ checks []Check
+ logger *zap.Logger
+}
+
+// newController creates a Controller instance.
+func newController(checks []Check, logger *zap.Logger) *controller {
+ return &controller{checks: checks, logger: logger}
+}
+
+// healthCheck is the HTTP handler for the route `GET /health`.
+func (c *controller) healthCheck(ctx *fiber.Ctx) error {
+
+ response := ctx.JSON(struct {
+ Status string `json:"status"`
+ }{
+ Status: "OK",
+ })
+
+ return response
+}
+
+// readinessCheck is the HTTP handler for the route `GET /ready`.
+func (c *controller) readinessCheck(ctx *fiber.Ctx) error {
+
+ requestCtx := ctx.Context()
+ requestID := fmt.Sprintf("%v", requestCtx.Value("requestid"))
+
+ // For every callback, check whether it is passing
+ for _, check := range c.checks {
+ if err := check(requestCtx); err != nil {
+
+ c.logger.Error(
+ "Readiness check failed",
+ zap.Error(err),
+ zap.String("requestID", requestID),
+ )
+
+ // Return error information to the caller
+ response := ctx.
+ Status(fiber.StatusInternalServerError).
+ JSON(struct {
+ Ready string `json:"ready"`
+ Error string `json:"error"`
+ }{
+ Ready: "NO",
+ Error: err.Error(),
+ })
+ return response
+ }
+ }
+
+ // All checks passed
+ response := ctx.Status(fiber.StatusOK).
+ JSON(struct {
+ Ready string `json:"ready"`
+ }{
+ Ready: "OK",
+ })
+ return response
+}

common/mongohelpers/helpers.go

@@ -0,0 +1,73 @@
+package mongohelpers
+
+import (
+ "context"
+ "fmt"
+ "time"
+
+ "go.mongodb.org/mongo-driver/mongo"
+ "go.mongodb.org/mongo-driver/mongo/options"
+ "go.mongodb.org/mongo-driver/mongo/readpref"
+)
+
+const (
+ connectTimeout = 10 * time.Second
+ disconnectTimeout = 10 * time.Second
+)
+
+// DB is a POD struct that represents a handle to a MongoDB database.
+type DB struct {
+ Client *mongo.Client
+ Database *mongo.Database
+}
+
+// Connect to a MongoDB database.
+//
+// Returns a struct that represents a handle to the database.
+//
+// Most of the time, you probably want to defer a call to `DB.Disconnect()`
+// after calling this function.
+func Connect(ctx context.Context, uri, databaseName string) (*DB, error) {
+
+ // Create a timed sub-context for the connection attempt
+ subContext, cancelFunc := context.WithTimeout(ctx, connectTimeout)
+ defer cancelFunc()
+
+ // Connect to MongoDB
+ client, err := mongo.Connect(subContext, options.Client().ApplyURI(uri))
+ if err != nil {
+ return nil, fmt.Errorf("failed to connect to MongoDB: %w", err)
+ }
+
+ // Ping the database to make sure we're actually connected
+ //
+ // This can detect a misconfuiguration error when a service is being initialized,
+ // rather than waiting for the first query to fail in the service's processing loop.
+ err = client.Ping(subContext, readpref.Primary())
+ if err != nil {
+ return nil, fmt.Errorf("failed to ping MongoDB database: %w", err)
+ }
+
+ // Populate the result struct and return
+ db := &DB{
+ Client: client,
+ Database: client.Database(databaseName),
+ }
+ return db, nil
+}
+
+// Disconnect from a MongoDB database.
+func (db *DB) Disconnect(ctx context.Context) error {
+
+ // Create a timed sub-context for the disconnection attempt
+ subContext, cancelFunc := context.WithTimeout(ctx, connectTimeout)
+ defer cancelFunc()
+
+ // Attempt to disconnect
+ err := db.Client.Disconnect(subContext)
+ if err != nil {
+ return fmt.Errorf("failed to disconnect from MongoDB: %w", err)
+ }
+
+ return nil
+}

common/settings/structs.go

@@ -0,0 +1,45 @@
+package settings
+
+import (
+ "fmt"
+
+ "github.com/joho/godotenv"
+ "github.com/kelseyhightower/envconfig"
+)
+
+// MongoDB contains configuration settings for a MongoDB database.
+type MongoDB struct {
+ MongodbURI string `split_words:"true" required:"true"`
+ MongodbDatabase string `split_words:"true" required:"true"`
+}
+
+// Logger contains configuration settings for a logger.
+type Logger struct {
+ LogLevel string `split_words:"true" default:"INFO"`
+}
+
+// Monitoring contains configuration settings for the monitoring endpoints.
+type Monitoring struct {
+ // MonitoringPort defines the TCP port for the monitoring endpoints.
+ MonitoringPort string `split_words:"true" default:"8000"`
+ PprofEnabled bool `split_words:"true" default:"false"`
+}
+
+// LoadFromEnv loads the configuration settings from environment variables.
+//
+// If there is a .env file in the current directory, it will be used to
+// populate the environment variables.
+func LoadFromEnv[T any]() (*T, error) {
+
+ // Load .env file (if it exists)
+ _ = godotenv.Load()
+
+ // Load environment variables into a struct
+ var settings T
+ err := envconfig.Process("", &settings)
+ if err != nil {
+ return nil, fmt.Errorf("failed to read config from environment: %w", err)
+ }
+
+ return &settings, nil
+}

core-contract-watcher/.gitignore

@@ -0,0 +1,2 @@
+bin/service
+.env

core-contract-watcher/Makefile

@@ -0,0 +1,9 @@
+
+build:
+ CGO_ENABLED=0 GOOS=linux go build -o "./bin/service" cmd/service/main.go
+
+test:
+ go test -v -cover ./...
+
+
+.PHONY: build test

core-contract-watcher/cmd/service/main.go

@@ -0,0 +1,66 @@
+package main
+
+import (
+ "context"
+ "log"
+ "os"
+ "os/signal"
+ "syscall"
+
+ "github.com/wormhole-foundation/wormhole-explorer/common/health"
+ "github.com/wormhole-foundation/wormhole-explorer/common/logger"
+ "github.com/wormhole-foundation/wormhole-explorer/common/mongohelpers"
+ "github.com/wormhole-foundation/wormhole-explorer/common/settings"
+ "github.com/wormhole-foundation/wormhole-explorer/core-contract-watcher/config"
+ "go.uber.org/zap"
+)
+
+func main() {
+
+ // Load config
+ cfg, err := settings.LoadFromEnv[config.ServiceSettings]()
+ if err != nil {
+ log.Fatal("Error loading config: ", err)
+ }
+
+ // Build rootLogger
+ rootLogger := logger.New("wormhole-explorer-core-contract-watcher", logger.WithLevel(cfg.LogLevel))
+
+ // Create top-level context
+ rootCtx, rootCtxCancel := context.WithCancel(context.Background())
+
+ // Connect to MongoDB
+ rootLogger.Info("connecting to MongoDB...")
+ db, err := mongohelpers.Connect(rootCtx, cfg.MongodbURI, cfg.MongodbDatabase)
+ if err != nil {
+ rootLogger.Fatal("Error connecting to MongoDB", zap.Error(err))
+ }
+
+ // Start serving the monitoring endpoints.
+ plugins := []health.Check{health.Mongo(db.Database)}
+ server := health.NewServer(
+ rootLogger,
+ cfg.MonitoringPort,
+ cfg.PprofEnabled,
+ plugins...,
+ )
+ server.Start()
+
+ // Block until we get a termination signal or the context is cancelled
+ rootLogger.Info("waiting for termination signal or context cancellation...")
+ sigterm := make(chan os.Signal, 1)
+ signal.Notify(sigterm, syscall.SIGINT, syscall.SIGTERM)
+ select {
+ case <-rootCtx.Done():
+ rootLogger.Warn("terminating (root context cancelled)")
+ case signal := <-sigterm:
+ rootLogger.Info("terminating (signal received)", zap.String("signal", signal.String()))
+ }
+
+ // Shut down gracefully
+ rootLogger.Info("disconnecting from MongoDB...")
+ db.Disconnect(rootCtx)
+ rootLogger.Info("cancelling root context...")
+ rootCtxCancel()
+ rootLogger.Info("terminated")
+}

core-contract-watcher/config/structs.go

@@ -0,0 +1,12 @@
+package config
+
+import (
+ "github.com/wormhole-foundation/wormhole-explorer/common/settings"
+)
+
+// ServiceSettings models the configuration settings for the core-contract-watcher service.
+type ServiceSettings struct {
+ settings.Logger
+ settings.MongoDB
+ settings.Monitoring
+}

core-contract-watcher/go.mod

@@ -0,0 +1,3 @@
+module github.com/wormhole-foundation/wormhole-explorer/core-contract-watcher
+
+go 1.19

go.work

@@ -5,6 +5,7 @@ use (
  ./api
  ./common
  ./contract-watcher
+ ./core-contract-watcher
  ./fly
  ./parser
  ./pipeline