Merge remote-tracking branch 'origin/staging' into NOBIDS/local-deplo…

…yment-update

.github/workflows/backend-converted-types-check.yml

@@ -0,0 +1,46 @@
+name: Backend-Converted-Types-Check
+on:
+  push:
+    paths:
+      - 'backend/pkg/api/types/**'
+      - 'frontend/types/api/**'
+    branches:
+      - main
+      - staging
+  pull_request:
+    paths:
+      - 'backend/pkg/api/types/**'
+      - 'frontend/types/api/**'
+    branches:
+      - '*'
+
+concurrency:
+  group: ${{ github.workflow }}-${{ github.ref }}
+  cancel-in-progress: true
+
+permissions:
+  contents: read
+  pull-requests: read
+  checks: write
+
+jobs:
+  build:
+    name: converted-types-check
+    runs-on: self-hosted
+    steps:
+      - uses: actions/checkout@v3
+      - uses: actions/setup-go@v4
+        with:
+          go-version-file: 'backend/go.mod'
+          cache-dependency-path: 'backend/go.sum'
+      - name: Check if all backend-types have been converted to frontend-types
+        working-directory: backend
+        run: |
+          currHash=$(find ../frontend/types/api -type f -print0 | sort -z | xargs -0 sha1sum | sha256sum | head -c 64)
+          make frontend-types
+          newHash=$(find ../frontend/types/api -type f -print0 | sort -z | xargs -0 sha1sum | sha256sum | head -c 64)
+          if [ "$currHash" != "$newHash" ]; then
+            echo "frontend-types have changed, please commit the changes"
+            exit 1
+          fi
+

backend/cmd/evm_node_indexer/main.go

@@ -13,6 +13,7 @@ import (
 	"fmt"
 	"io"
 	"net/http"
+	"os"
 	"regexp"
 	"strings"
 	"sync/atomic"
@@ -24,6 +25,7 @@ import (
 	"github.com/gobitfly/beaconchain/pkg/commons/db"
 	"github.com/gobitfly/beaconchain/pkg/commons/hexutil"
 	"github.com/gobitfly/beaconchain/pkg/commons/log"
+	"github.com/gobitfly/beaconchain/pkg/commons/metrics"
 	"github.com/gobitfly/beaconchain/pkg/commons/types"
 	"github.com/gobitfly/beaconchain/pkg/commons/utils"
 	"github.com/gobitfly/beaconchain/pkg/commons/version"
@@ -103,22 +105,32 @@ func init() {
 
 // main
 func Run() {
+	fs := flag.NewFlagSet("fs", flag.ExitOnError)
+
 	// read / set parameter
-	configPath := flag.String("config", "config/default.config.yml", "Path to the config file")
-	startBlockNumber := flag.Int64("start-block-number", -1, "trigger a REEXPORT, only working in combination with end-block-number, defined block is included, will be the first action done and will quite afterwards, ignore every other action")
-	endBlockNumber := flag.Int64("end-block-number", -1, "trigger a REEXPORT, only working in combination with start-block-number, defined block is included, will be the first action done and will quite afterwards, ignore every other action")
-	reorgDepth = flag.Int64("reorg.depth", 32, fmt.Sprintf("lookback to check and handle chain reorgs (MAX %s), you should NEVER reduce this after the first start, otherwise there will be unchecked areas", _formatInt64(MAX_REORG_DEPTH)))
-	concurrency := flag.Int64("concurrency", 8, "maximum threads used (running on maximum whenever possible)")
-	nodeRequestsAtOnce := flag.Int64("node-requests-at-once", 16, fmt.Sprintf("bulk size per node = bt = db request (MAX %s)", _formatInt64(MAX_NODE_REQUESTS_AT_ONCE)))
-	skipHoleCheck := flag.Bool("skip-hole-check", false, "skips the initial check for holes, doesn't go very well with only-hole-check")
-	onlyHoleCheck := flag.Bool("only-hole-check", false, "just check for holes and quit, can be used for a reexport running simulation to a normal setup, just remove entries in postgres and start with this flag, doesn't go very well with skip-hole-check")
-	noNewBlocks := flag.Bool("ignore-new-blocks", false, "there are no new blocks, at all")
-	noNewBlocksThresholdSeconds := flag.Int("fatal-if-no-new-block-for-x-seconds", 600, "will fatal if there is no new block for x seconds (MIN 30), will start throwing errors at 2/3 of the time, will start throwing warnings at 1/3 of the time, doesn't go very well with ignore-new-blocks")
-	discordWebhookBlockThreshold := flag.Int64("discord-block-threshold", 100000, "every x blocks an update is send to Discord")
-	discordWebhookReportUrl := flag.String("discord-url", "", "report progress to discord url")
-	discordWebhookUser := flag.String("discord-user", "", "report progress to discord user")
-	discordWebhookAddTextFatal := flag.String("discord-fatal-text", "", "this text will be added to the discord message in the case of an fatal")
-	flag.Parse()
+	configPath := fs.String("config", "config/default.config.yml", "Path to the config file")
+	versionFlag := fs.Bool("version", false, "print version and exit")
+	startBlockNumber := fs.Int64("start-block-number", -1, "trigger a REEXPORT, only working in combination with end-block-number, defined block is included, will be the first action done and will quite afterwards, ignore every other action")
+	endBlockNumber := fs.Int64("end-block-number", -1, "trigger a REEXPORT, only working in combination with start-block-number, defined block is included, will be the first action done and will quite afterwards, ignore every other action")
+	reorgDepth = fs.Int64("reorg.depth", 32, fmt.Sprintf("lookback to check and handle chain reorgs (MAX %s), you should NEVER reduce this after the first start, otherwise there will be unchecked areas", _formatInt64(MAX_REORG_DEPTH)))
+	concurrency := fs.Int64("concurrency", 8, "maximum threads used (running on maximum whenever possible)")
+	nodeRequestsAtOnce := fs.Int64("node-requests-at-once", 16, fmt.Sprintf("bulk size per node = bt = db request (MAX %s)", _formatInt64(MAX_NODE_REQUESTS_AT_ONCE)))
+	skipHoleCheck := fs.Bool("skip-hole-check", false, "skips the initial check for holes, doesn't go very well with only-hole-check")
+	onlyHoleCheck := fs.Bool("only-hole-check", false, "just check for holes and quit, can be used for a reexport running simulation to a normal setup, just remove entries in postgres and start with this flag, doesn't go very well with skip-hole-check")
+	noNewBlocks := fs.Bool("ignore-new-blocks", false, "there are no new blocks, at all")
+	noNewBlocksThresholdSeconds := fs.Int("fatal-if-no-new-block-for-x-seconds", 600, "will fatal if there is no new block for x seconds (MIN 30), will start throwing errors at 2/3 of the time, will start throwing warnings at 1/3 of the time, doesn't go very well with ignore-new-blocks")
+	discordWebhookBlockThreshold := fs.Int64("discord-block-threshold", 100000, "every x blocks an update is send to Discord")
+	discordWebhookReportUrl := fs.String("discord-url", "", "report progress to discord url")
+	discordWebhookUser := fs.String("discord-user", "", "report progress to discord user")
+	discordWebhookAddTextFatal := fs.String("discord-fatal-text", "", "this text will be added to the discord message in the case of an fatal")
+	err := fs.Parse(os.Args[2:])
+	if err != nil {
+		log.Fatal(err, "error parsing flags", 0)
+	}
+	if *versionFlag {
+		log.Info(version.Version)
+		return
+	}
 
 	// tell the user about all parameter
 	{
@@ -161,6 +173,15 @@ func Run() {
 		} else {
 			eth1RpcEndpoint = utils.Config.Eth1GethEndpoint
 		}
+
+		if utils.Config.Metrics.Enabled {
+			go func() {
+				log.Infof("serving metrics on %v", utils.Config.Metrics.Address)
+				if err := metrics.Serve(utils.Config.Metrics.Address, utils.Config.Metrics.Pprof, utils.Config.Metrics.PprofExtra); err != nil {
+					log.Fatal(err, "error serving metrics", 0)
+				}
+			}()
+		}
 	}
 
 	// check parameters

backend/cmd/exporter/main.go

@@ -193,7 +193,19 @@ func Run() {
 		go services.StartHistoricPriceService()
 	}
 
-	go modules.StartAll(context)
+	usedModules := []modules.ModuleInterface{}
+
+	if cfg.JustV2 {
+		usedModules = append(usedModules, modules.NewDashboardDataModule(context))
+	} else {
+		usedModules = append(usedModules,
+			modules.NewSlotExporter(context),
+			modules.NewExecutionDepositsExporter(context),
+			modules.NewExecutionPayloadsExporter(context),
+		)
+	}
+
+	go modules.StartAll(context, usedModules, cfg.JustV2)
 
 	// Keep the program alive until Ctrl+C is pressed
 	utils.WaitForCtrlC()

backend/cmd/misc/main.go

@@ -564,6 +564,40 @@ func collectNotifications(startEpoch uint64) error {
 	if len(notifications[0]) > 0 {
 		spew.Dump(notifications[0])
 	}
+
+	emails, err := notification.RenderEmailsForUserEvents(0, notifications)
+	if err != nil {
+		return err
+	}
+
+	for _, email := range emails {
+		// if email.Address == "" {
+		log.Infof("to: %v", email.Address)
+		log.Infof("subject: %v", email.Subject)
+		log.Infof("body: %v", email.Email.Body)
+		log.Info("-----")
+		// }
+	}
+
+	// pushMessages, err := notification.RenderPushMessagesForUserEvents(0, notifications)
+	// if err != nil {
+	// 	return err
+	// }
+
+	// for _, pushMessage := range pushMessages {
+	// 	message := pushMessage.Messages[0]
+	// 	log.Infof("title: %v body: %v", message.Notification.Title, message.Notification.Body)
+
+	// 	if message.Token == "" {
+	// 		log.Info("sending test message")
+
+	// 		err = notification.SendPushBatch(pushMessage.UserId, []*messaging.Message{message}, false)
+	// 		if err != nil {
+	// 			log.Error(err, "error sending firebase batch job", 0)
+	// 		}
+	// 	}
+	// }
+
 	return nil
 }
 
@@ -606,7 +640,7 @@ func collectUserDbNotifications(startEpoch uint64) error {
 		if message.Token == "" {
 			log.Info("sending test message")
 
-			err = notification.SendPushBatch(pushMessage.UserId, []*messaging.Message{message}, false)
+			err = notification.SendPushBatch(pushMessage.UserId, []*messaging.Message{message}, true)
 			if err != nil {
 				log.Error(err, "error sending firebase batch job", 0)
 			}

backend/cmd/notification_collector/main.go

@@ -150,6 +150,31 @@ func Run() {
 		}, "pgx", "postgres")
 	}()
 
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		// clickhouse
+		db.ClickHouseWriter, db.ClickHouseReader = db.MustInitDB(&types.DatabaseConfig{
+			Username:     cfg.ClickHouse.WriterDatabase.Username,
+			Password:     cfg.ClickHouse.WriterDatabase.Password,
+			Name:         cfg.ClickHouse.WriterDatabase.Name,
+			Host:         cfg.ClickHouse.WriterDatabase.Host,
+			Port:         cfg.ClickHouse.WriterDatabase.Port,
+			MaxOpenConns: cfg.ClickHouse.WriterDatabase.MaxOpenConns,
+			SSL:          true,
+			MaxIdleConns: cfg.ClickHouse.WriterDatabase.MaxIdleConns,
+		}, &types.DatabaseConfig{
+			Username:     cfg.ClickHouse.ReaderDatabase.Username,
+			Password:     cfg.ClickHouse.ReaderDatabase.Password,
+			Name:         cfg.ClickHouse.ReaderDatabase.Name,
+			Host:         cfg.ClickHouse.ReaderDatabase.Host,
+			Port:         cfg.ClickHouse.ReaderDatabase.Port,
+			MaxOpenConns: cfg.ClickHouse.ReaderDatabase.MaxOpenConns,
+			SSL:          true,
+			MaxIdleConns: cfg.ClickHouse.ReaderDatabase.MaxIdleConns,
+		}, "clickhouse", "clickhouse")
+	}()
+
 	wg.Add(1)
 	go func() {
 		defer wg.Done()
@@ -184,6 +209,8 @@ func Run() {
 	defer db.FrontendWriterDB.Close()
 	defer db.AlloyReader.Close()
 	defer db.AlloyWriter.Close()
+	defer db.ClickHouseReader.Close()
+	defer db.ClickHouseWriter.Close()
 	defer db.BigtableClient.Close()
 
 	log.Infof("database connection established")

backend/pkg/api/auth.go

@@ -13,8 +13,10 @@ import (
 	"github.com/gorilla/csrf"
 )
 
+var day time.Duration = time.Hour * 24
+var sessionDuration time.Duration = day * 365
+
 func newSessionManager(cfg *types.Config) *scs.SessionManager {
-	// TODO: replace redis with user db down the line (or replace sessions with oauth2)
 	pool := &redis.Pool{
 		MaxIdle: 10,
 		Dial: func() (redis.Conn, error) {
@@ -23,7 +25,7 @@ func newSessionManager(cfg *types.Config) *scs.SessionManager {
 	}
 
 	scs := scs.New()
-	scs.Lifetime = time.Hour * 24 * 7
+	scs.Lifetime = sessionDuration
 	scs.Cookie.Name = "session_id"
 	scs.Cookie.HttpOnly = true
 	scs.Cookie.Persist = true
@@ -42,6 +44,19 @@ func newSessionManager(cfg *types.Config) *scs.SessionManager {
 	return scs
 }
 
+// returns a middleware that extends the session expiration if the session is older than 1 day
+func getSlidingSessionExpirationMiddleware(scs *scs.SessionManager) func(http.Handler) http.Handler {
+	return func(next http.Handler) http.Handler {
+		return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			deadline := scs.Deadline(r.Context()) // unauthenticated requests have deadline set to now+sessionDuration
+			if time.Until(deadline) < sessionDuration-day {
+				scs.SetDeadline(r.Context(), time.Now().Add(sessionDuration).UTC()) // setting to utc because library also does that internally
+			}
+			next.ServeHTTP(w, r)
+		})
+	}
+}
+
 // returns goriila/csrf middleware with the given config settings
 func getCsrfProtectionMiddleware(cfg *types.Config) func(http.Handler) http.Handler {
 	csrfBytes, err := hex.DecodeString(cfg.Frontend.CsrfAuthKey)

backend/pkg/api/data_access/archiver.go

@@ -27,7 +27,7 @@ func (d *DataAccessService) GetValidatorDashboardsCountInfo(ctx context.Context)
 	}
 
 	var dbReturn []DashboardInfo
-	err := d.readerDb.Select(&dbReturn, `
+	err := d.readerDb.SelectContext(ctx, &dbReturn, `
 		WITH dashboards_groups AS
 			(SELECT
 				dashboard_id,

backend/pkg/api/data_access/block.go

@@ -74,71 +74,71 @@ func (d *DataAccessService) GetBlockBlobs(ctx context.Context, chainId, block ui
 }
 
 func (d *DataAccessService) GetSlot(ctx context.Context, chainId, slot uint64) (*t.BlockSummary, error) {
-	block, err := d.GetBlockHeightAt(slot)
+	block, err := d.GetBlockHeightAt(ctx, slot)
 	if err != nil {
 		return nil, err
 	}
 	return d.GetBlock(ctx, chainId, block)
 }
 
 func (d *DataAccessService) GetSlotOverview(ctx context.Context, chainId, slot uint64) (*t.BlockOverview, error) {
-	block, err := d.GetBlockHeightAt(slot)
+	block, err := d.GetBlockHeightAt(ctx, slot)
 	if err != nil {
 		return nil, err
 	}
 	return d.GetBlockOverview(ctx, chainId, block)
 }
 
 func (d *DataAccessService) GetSlotTransactions(ctx context.Context, chainId, slot uint64) ([]t.BlockTransactionTableRow, error) {
-	block, err := d.GetBlockHeightAt(slot)
+	block, err := d.GetBlockHeightAt(ctx, slot)
 	if err != nil {
 		return nil, err
 	}
 	return d.GetBlockTransactions(ctx, chainId, block)
 }
 
 func (d *DataAccessService) GetSlotVotes(ctx context.Context, chainId, slot uint64) ([]t.BlockVoteTableRow, error) {
-	block, err := d.GetBlockHeightAt(slot)
+	block, err := d.GetBlockHeightAt(ctx, slot)
 	if err != nil {
 		return nil, err
 	}
 	return d.GetBlockVotes(ctx, chainId, block)
 }
 
 func (d *DataAccessService) GetSlotAttestations(ctx context.Context, chainId, slot uint64) ([]t.BlockAttestationTableRow, error) {
-	block, err := d.GetBlockHeightAt(slot)
+	block, err := d.GetBlockHeightAt(ctx, slot)
 	if err != nil {
 		return nil, err
 	}
 	return d.GetBlockAttestations(ctx, chainId, block)
 }
 
 func (d *DataAccessService) GetSlotWithdrawals(ctx context.Context, chainId, slot uint64) ([]t.BlockWithdrawalTableRow, error) {
-	block, err := d.GetBlockHeightAt(slot)
+	block, err := d.GetBlockHeightAt(ctx, slot)
 	if err != nil {
 		return nil, err
 	}
 	return d.GetBlockWithdrawals(ctx, chainId, block)
 }
 
 func (d *DataAccessService) GetSlotBlsChanges(ctx context.Context, chainId, slot uint64) ([]t.BlockBlsChangeTableRow, error) {
-	block, err := d.GetBlockHeightAt(slot)
+	block, err := d.GetBlockHeightAt(ctx, slot)
 	if err != nil {
 		return nil, err
 	}
 	return d.GetBlockBlsChanges(ctx, chainId, block)
 }
 
 func (d *DataAccessService) GetSlotVoluntaryExits(ctx context.Context, chainId, slot uint64) ([]t.BlockVoluntaryExitTableRow, error) {
-	block, err := d.GetBlockHeightAt(slot)
+	block, err := d.GetBlockHeightAt(ctx, slot)
 	if err != nil {
 		return nil, err
 	}
 	return d.GetBlockVoluntaryExits(ctx, chainId, block)
 }
 
 func (d *DataAccessService) GetSlotBlobs(ctx context.Context, chainId, slot uint64) ([]t.BlockBlobTableRow, error) {
-	block, err := d.GetBlockHeightAt(slot)
+	block, err := d.GetBlockHeightAt(ctx, slot)
 	if err != nil {
 		return nil, err
 	}

backend/pkg/api/data_access/data_access.go

@@ -35,16 +35,15 @@ type DataAccessor interface {
 
 	Close()
 
-	GetLatestFinalizedEpoch() (uint64, error)
-	GetLatestSlot() (uint64, error)
-	GetLatestBlock() (uint64, error)
-	GetBlockHeightAt(slot uint64) (uint64, error)
-	GetLatestExchangeRates() ([]t.EthConversionRate, error)
+	GetLatestFinalizedEpoch(ctx context.Context) (uint64, error)
+	GetLatestSlot(ctx context.Context) (uint64, error)
+	GetLatestBlock(ctx context.Context) (uint64, error)
+	GetLatestExchangeRates(ctx context.Context) ([]t.EthConversionRate, error)
 
 	GetProductSummary(ctx context.Context) (*t.ProductSummary, error)
 	GetFreeTierPerks(ctx context.Context) (*t.PremiumPerks, error)
 
-	GetValidatorsFromSlices(indices []uint64, publicKeys []string) ([]t.VDBValidator, error)
+	GetValidatorsFromSlices(ctx context.Context, indices []uint64, publicKeys []string) ([]t.VDBValidator, error)
 }
 
 type DataAccessService struct {