refactor: move deposits-aggregation from exporter to statistics and run only once per day

Author: guybrush

backend/cmd/statistics/main.go

@@ -1,6 +1,7 @@
 package statistics
 
 import (
+	"database/sql"
 	"errors"
 	"flag"
 	"fmt"
@@ -33,6 +34,7 @@ type options struct {
 	statisticsValidatorToggle bool
 	statisticsChartToggle     bool
 	statisticsGraffitiToggle  bool
+	statisticsDepositsToggle  bool
 	resetStatus               bool
 }
 
@@ -46,6 +48,7 @@ func Run() {
 	fs.BoolVar(&opt.statisticsValidatorToggle, "validators.enabled", false, "Toggle exporting validator statistics")
 	fs.BoolVar(&opt.statisticsChartToggle, "charts.enabled", false, "Toggle exporting chart series")
 	fs.BoolVar(&opt.statisticsGraffitiToggle, "graffiti.enabled", false, "Toggle exporting graffiti statistics")
+	fs.BoolVar(&opt.statisticsDepositsToggle, "deposits.enabled", false, "Toggle aggregating deposits")
 	fs.BoolVar(&opt.resetStatus, "validators.reset", false, "Export stats independent if they have already been exported previously")
 
 	versionFlag := fs.Bool("version", false, "Show version and exit")
@@ -349,6 +352,12 @@ func statisticsLoop(client rpc.Client) {
 			}
 		}
 
+		if opt.statisticsDepositsToggle {
+			err := aggregateDeposits()
+			log.Error(err, fmt.Errorf("error aggregating deposits"), 0)
+			loopError = err
+		}
+
 		if loopError == nil {
 			services.ReportStatus("statistics", "Running", nil)
 		} else {
@@ -365,3 +374,47 @@ func clearStatsStatusTable(day uint64) {
 		log.Fatal(err, "error resetting status for day", 0, log.Fields{"day": day})
 	}
 }
+
+func aggregateDeposits() error {
+	start := time.Now()
+	defer func() {
+		metrics.TaskDuration.WithLabelValues("statistics_aggregate_eth1_deposits").Observe(time.Since(start).Seconds())
+	}()
+	_, err := db.WriterDb.Exec(`
+		INSERT INTO eth1_deposits_aggregated (from_address, amount, validcount, invalidcount, slashedcount, totalcount, activecount, pendingcount, voluntary_exit_count)
+		SELECT
+			eth1.from_address,
+			SUM(eth1.amount) as amount,
+			SUM(eth1.validcount) AS validcount,
+			SUM(eth1.invalidcount) AS invalidcount,
+			COUNT(CASE WHEN v.status = 'slashed' THEN 1 END) AS slashedcount,
+			COUNT(v.pubkey) AS totalcount,
+			COUNT(CASE WHEN v.status = 'active_online' OR v.status = 'active_offline' THEN 1 END) as activecount,
+			COUNT(CASE WHEN v.status = 'deposited' THEN 1 END) AS pendingcount,
+			COUNT(CASE WHEN v.status = 'exited' THEN 1 END) AS voluntary_exit_count
+		FROM (
+			SELECT
+				from_address,
+				publickey,
+				SUM(amount) AS amount,
+				COUNT(CASE WHEN valid_signature = 't' THEN 1 END) AS validcount,
+				COUNT(CASE WHEN valid_signature = 'f' THEN 1 END) AS invalidcount
+			FROM eth1_deposits
+			GROUP BY from_address, publickey
+		) eth1
+		LEFT JOIN (SELECT pubkey, status FROM validators) v ON v.pubkey = eth1.publickey
+		GROUP BY eth1.from_address
+		ON CONFLICT (from_address) DO UPDATE SET
+			amount               = excluded.amount,
+			validcount           = excluded.validcount,
+			invalidcount         = excluded.invalidcount,
+			slashedcount         = excluded.slashedcount,
+			totalcount           = excluded.totalcount,
+			activecount          = excluded.activecount,
+			pendingcount         = excluded.pendingcount,
+			voluntary_exit_count = excluded.voluntary_exit_count`)
+	if err != nil && err != sql.ErrNoRows {
+		return nil
+	}
+	return err
+}