feat(exporter): add legacy transformers to new cl event indexer

Author: invis-bitfly

backend/pkg/commons/db/db.go

@@ -2593,115 +2593,3 @@ func HasEventsForEpoch(epoch uint64) (bool, error) {
 
 	return count > 0, nil
 }
-
-func TransformSwitchToCompoundingRequests(firstSlot, lastSlot uint64, tx *sqlx.Tx) (int64, error) {
-	res, err := tx.Exec(`
-	INSERT INTO blocks_switch_to_compounding_requests (block_slot, block_root, request_index, address, validator_index)
-		SELECT
-			cle.slot AS slot,
-			cle.block_root AS block_root,
-			cle.event_index AS request_index,
-			decode((cle.data->>'address'), 'base64') AS address,
-			vali.validatorindex AS validator_index
-		FROM consensus_layer_events cle
-		JOIN validators AS vali
-			ON vali.pubkey = decode(cle.data ->> 'pubkey', 'base64')
-		WHERE event_name = 'SwitchToCompoundingEvent' AND slot >= $1 AND slot <= $2 AND version = $3 ON CONFLICT DO NOTHING;
-	`, firstSlot, lastSlot, types.ConsensusLayerEventVersion)
-	if err != nil {
-		return 0, fmt.Errorf("error transforming consolidation requests: %w", err)
-	}
-
-	consolidationRequestsProcessed, err := res.RowsAffected()
-	if err != nil {
-		return 0, fmt.Errorf("error getting the amount of processed consolidation requests: %w", err)
-	}
-
-	return consolidationRequestsProcessed, nil
-}
-
-func TransformConsolidationRequests(firstSlot, lastSlot uint64, tx *sqlx.Tx) (int64, error) {
-	res, err := tx.Exec(`
-	INSERT INTO blocks_consolidation_requests (block_slot, block_root, request_index, source_index, target_index, amount_consolidated)
-		SELECT
-			cle.slot					AS block_slot,
-			cle.block_root				AS block_root,
-			cle.event_index				AS request_index,
-			src.validatorindex			AS source_index,
-			tgt.validatorindex			AS target_index,
-			(data->>'amount')::bigint 	AS amount_consolidated
-		FROM consensus_layer_events cle
-		JOIN validators AS src
-			ON src.pubkey = decode(cle.data ->> 'source_pubkey', 'base64')
-		JOIN validators AS tgt
-			ON tgt.pubkey = decode(cle.data ->> 'target_pubkey', 'base64')
-		WHERE
-			event_name = 'ConsolidationProcessedEvent' AND
-			slot >= $1 AND slot <= $2 AND
-			version = $3
-		ON CONFLICT DO NOTHING;
-	`, firstSlot, lastSlot, types.ConsensusLayerEventVersion)
-	if err != nil {
-		return 0, fmt.Errorf("error transforming consolidation requests: %w", err)
-	}
-
-	consolidationRequestsProcessed, err := res.RowsAffected()
-	if err != nil {
-		return 0, fmt.Errorf("error getting the amount of processed consolidation requests: %w", err)
-	}
-
-	return consolidationRequestsProcessed, nil
-}
-
-func TransformDepositRequests(firstSlot, lastSlot uint64, tx *sqlx.Tx) (int64, error) {
-	res, err := tx.Exec(`
-	INSERT INTO blocks_deposit_requests (block_slot, block_root, request_index, pubkey, withdrawal_credentials, amount, signature)
-		SELECT
-				slot AS block_slot,
-				block_root AS block_root,
-				event_index AS request_index,
-				decode((data->>'pubkey'), 'base64') AS pubkey,
-				decode((data->>'withdrawal_credentials'), 'base64')::bytea AS withdrawal_credentials,
-				(data->>'amount')::bigint AS amount,
-				decode((data->>'signature'), 'base64')::bytea AS signature
-		FROM consensus_layer_events WHERE event_name = 'DepositProcessedEvent' AND slot >= $1 AND slot <= $2 AND version = $3 ON CONFLICT DO NOTHING;
-`, firstSlot, lastSlot, types.ConsensusLayerEventVersion)
-	if err != nil {
-		return 0, fmt.Errorf("error transforming deposit requests: %w", err)
-	}
-
-	depositRequestsProcessed, err := res.RowsAffected()
-	if err != nil {
-		return 0, fmt.Errorf("error getting the amount of processed deposit requests: %w", err)
-	}
-
-	return depositRequestsProcessed, nil
-}
-
-func TransformRemovedExcessBalanceEvents(firstSlot, lastSlot uint64, tx *sqlx.Tx) (int64, error) {
-	// we offset by -20000 to avoid conflicts with normal withdrawals in the blocks
-	res, err := tx.Exec(`
-	INSERT INTO blocks_withdrawals (block_slot, block_root, withdrawalindex, validatorindex, address, amount)
-		SELECT
-			cle.slot 						AS block_slot,
-			cle.block_root 					AS block_root,
-			-20000 + cle.event_index 		AS withdrawalindex,
-			(vali.validatorindex)::int 		AS validatorindex,
-			''::bytea 						as address,
-			(cle.data->>'amount')::bigint 	AS amount
-		FROM consensus_layer_events cle
-		JOIN validators AS vali
-			ON vali.pubkey = decode(cle.data ->> 'pubkey', 'base64')
-		WHERE event_name = 'RemovedExcessBalance' AND slot >= $1 AND slot <= $2 AND version = $3 ON CONFLICT DO NOTHING;
-`, firstSlot, lastSlot, types.ConsensusLayerEventVersion)
-	if err != nil {
-		return 0, fmt.Errorf("error transforming excess balance requests: %w", err)
-	}
-
-	excessBalanceRequestsProcessed, err := res.RowsAffected()
-	if err != nil {
-		return 0, fmt.Errorf("error getting the amount of processed excess balance requests: %w", err)
-	}
-
-	return excessBalanceRequestsProcessed, nil
-}

backend/pkg/exporter/modules/cl_transformers/17_legacy_switch_to_compounding.go

@@ -0,0 +1,92 @@
+package cl_transformers
+
+import (
+	"fmt"
+
+	"github.com/doug-martin/goqu/v9"
+	"github.com/gobitfly/beaconchain/pkg/commons/types"
+	"github.com/jmoiron/sqlx"
+	"github.com/pkg/errors"
+
+	"encoding/json"
+)
+
+type LegacySwitchToCompoundingEventTransformer struct{}
+
+func (d *LegacySwitchToCompoundingEventTransformer) Transform(tx *sqlx.Tx, events []types.ConsensusLayerEvent) (int, error) {
+	eventName := types.SwitchToCompoundingEventName
+	// filter events for the relevant type
+	events, err := FilterEventsByTypeAndSort(events, eventName)
+	if err != nil {
+		return 0, errors.Wrap(err, "error filtering events")
+	}
+	if len(events) == 0 {
+		return 0, nil
+	}
+	// convert raw data from events into actual structs
+	data := make([]types.SwitchToCompoundingEvent, len(events))
+	for i, event := range events {
+		// unmarshal from data field to DepositQueuedEvent using json
+		err = json.Unmarshal(event.RawData, &data[i])
+		if err != nil {
+			return 0, errors.Wrap(err, "error unmarshalling event data")
+		}
+	}
+
+	filters := make([]types.ConsensusLayerEventFilter, 0)
+	for _, event := range events {
+		filters = append(filters, types.ConsensusLayerEventFilter{
+			Slot:      uint64(event.Slot),
+			BlockRoot: event.BlockRoot,
+		})
+	}
+
+	var orConditions []goqu.Expression
+	for _, filter := range filters {
+		orConditions = append(orConditions, goqu.Ex{
+			"slot":       filter.Slot,
+			"block_root": filter.BlockRoot,
+		})
+	}
+
+	ds := goqu.Dialect("postgres").Insert("blocks_switch_to_compounding_requests").
+		Cols(
+			"block_slot",
+			"block_root",
+			"request_index",
+			"address",
+			"validator_index").
+		FromQuery(
+			goqu.From("consensus_layer_events").As("cle").
+				Select(
+					"cle.slot",
+					"cle.block_root",
+					"cle.event_index",
+					goqu.L("decode((cle.data->>'address'), 'base64') AS address"),
+					goqu.L("vali.validatorindex AS validator_index"),
+				).
+				Join(goqu.T("validators"), goqu.On(goqu.Ex{
+					"vali.pubkey": goqu.L(`decode(cle.data ->> 'pubkey', 'base64')`),
+				})).As("vali").
+				Where(goqu.And(
+					goqu.Ex{"event_name": eventName},
+					goqu.Or(orConditions...),
+					goqu.Ex{"version": types.ConsensusLayerEventVersion},
+				)),
+		)
+	query, args, err := ds.Prepared(true).ToSQL()
+	if err != nil {
+		return 0, fmt.Errorf("error preparing query: %w", err)
+	}
+
+	res, err := tx.Exec(query, args...)
+	if err != nil {
+		return 0, fmt.Errorf("error executing query: %w", err)
+	}
+	c, err := res.RowsAffected()
+	if err != nil {
+		return 0, fmt.Errorf("error getting rows affected: %w", err)
+	}
+
+	return int(c), nil
+}

backend/pkg/exporter/modules/cl_transformers/18_legacy_consolidation_processed copy.go

@@ -0,0 +1,95 @@
+package cl_transformers
+
+import (
+	"fmt"
+
+	"github.com/doug-martin/goqu/v9"
+	"github.com/gobitfly/beaconchain/pkg/commons/types"
+	"github.com/jmoiron/sqlx"
+	"github.com/pkg/errors"
+
+	"encoding/json"
+)
+
+type LegacyRemovedExcessBalanceEventTransformer struct{}
+
+func (d *LegacyRemovedExcessBalanceEventTransformer) Transform(tx *sqlx.Tx, events []types.ConsensusLayerEvent) (int, error) {
+	eventName := types.RemovedExcessBalanceEventName
+	// filter events for the relevant type
+	events, err := FilterEventsByTypeAndSort(events, eventName)
+	if err != nil {
+		return 0, errors.Wrap(err, "error filtering events")
+	}
+	if len(events) == 0 {
+		return 0, nil
+	}
+	// convert raw data from events into actual structs
+	data := make([]types.RemovedExcessBalance, len(events))
+	for i, event := range events {
+		// unmarshal from data field to DepositQueuedEvent using json
+		err = json.Unmarshal(event.RawData, &data[i])
+		if err != nil {
+			return 0, errors.Wrap(err, "error unmarshalling event data")
+		}
+	}
+
+	filters := make([]types.ConsensusLayerEventFilter, 0)
+	for _, event := range events {
+		filters = append(filters, types.ConsensusLayerEventFilter{
+			Slot:      uint64(event.Slot),
+			BlockRoot: event.BlockRoot,
+		})
+	}
+
+	var orConditions []goqu.Expression
+	for _, filter := range filters {
+		orConditions = append(orConditions, goqu.Ex{
+			"slot":       filter.Slot,
+			"block_root": filter.BlockRoot,
+		})
+	}
+
+	ds := goqu.Dialect("postgres").Insert("blocks_withdrawals").
+		Cols(
+			"block_slot",
+			"block_root",
+			"withdrawalindex",
+			"validatorindex",
+			"address",
+			"amount").
+		FromQuery(
+			goqu.From("consensus_layer_events").As("cle").
+				Select(
+					"cle.slot AS block_slot",
+					"cle.block_root",
+					goqu.L("-20000 + cle.event_index AS withdrawalindex"),
+					goqu.L("vali.validatorindex::int AS validatorindex"),
+					goqu.L("''::bytea AS address"),
+					goqu.L("(cle.data->>'amount')::bigint AS amount"),
+				).
+				Join(goqu.T("validators").As("vali"), goqu.On(goqu.Ex{
+					"vali.pubkey": goqu.L(`decode(cle.data ->> 'pubkey', 'base64')`),
+				})).
+				Where(goqu.And(
+					goqu.Ex{"event_name": eventName},
+					goqu.Or(orConditions...),
+					goqu.Ex{"version": types.ConsensusLayerEventVersion},
+				)),
+		).OnConflict(goqu.DoNothing())
+
+	query, args, err := ds.Prepared(true).ToSQL()
+	if err != nil {
+		return 0, fmt.Errorf("error preparing query: %w", err)
+	}
+
+	res, err := tx.Exec(query, args...)
+	if err != nil {
+		return 0, fmt.Errorf("error executing query: %w", err)
+	}
+	c, err := res.RowsAffected()
+	if err != nil {
+		return 0, fmt.Errorf("error getting rows affected: %w", err)
+	}
+
+	return int(c), nil
+}

backend/pkg/exporter/modules/cl_transformers/19_legacy_deposits_processed.go

@@ -0,0 +1,93 @@
+package cl_transformers
+
+import (
+	"fmt"
+
+	"github.com/doug-martin/goqu/v9"
+	"github.com/gobitfly/beaconchain/pkg/commons/types"
+	"github.com/jmoiron/sqlx"
+	"github.com/pkg/errors"
+
+	"encoding/json"
+)
+
+type LegacyDepositProcessedEventTransformer struct{}
+
+func (d *LegacyDepositProcessedEventTransformer) Transform(tx *sqlx.Tx, events []types.ConsensusLayerEvent) (int, error) {
+	eventName := types.DepositProcessedEventName
+	// filter events for the relevant type
+	events, err := FilterEventsByTypeAndSort(events, eventName)
+	if err != nil {
+		return 0, errors.Wrap(err, "error filtering events")
+	}
+	if len(events) == 0 {
+		return 0, nil
+	}
+	// convert raw data from events into actual structs
+	data := make([]types.DepositProcessedEvent, len(events))
+	for i, event := range events {
+		// unmarshal from data field to DepositQueuedEvent using json
+		err = json.Unmarshal(event.RawData, &data[i])
+		if err != nil {
+			return 0, errors.Wrap(err, "error unmarshalling event data")
+		}
+	}
+
+	filters := make([]types.ConsensusLayerEventFilter, 0)
+	for _, event := range events {
+		filters = append(filters, types.ConsensusLayerEventFilter{
+			Slot:      uint64(event.Slot),
+			BlockRoot: event.BlockRoot,
+		})
+	}
+
+	var orConditions []goqu.Expression
+	for _, filter := range filters {
+		orConditions = append(orConditions, goqu.Ex{
+			"slot":       filter.Slot,
+			"block_root": filter.BlockRoot,
+		})
+	}
+
+	ds := goqu.Dialect("postgres").Insert("blocks_deposit_requests").
+		Cols(
+			"block_slot",
+			"block_root",
+			"request_index",
+			"pubkey",
+			"withdrawal_credentials",
+			"amount",
+			"signature").
+		FromQuery(
+			goqu.From("consensus_layer_events").As("cle").
+				Select(
+					"cle.slot",
+					"cle.block_root",
+					"cle.event_index",
+					goqu.L("decode((cle.data->>'pubkey'), 'base64') AS pubkey"),
+					goqu.L("decode((cle.data->>'withdrawal_credentials'), 'base64')::bytea AS withdrawal_credentials"),
+					goqu.L("(cle.data->>'amount')::bigint AS amount"),
+					goqu.L("decode((cle.data->>'signature'), 'base64')::bytea AS signature"),
+				).
+				Where(goqu.And(
+					goqu.Ex{"event_name": eventName},
+					goqu.Or(orConditions...),
+					goqu.Ex{"version": types.ConsensusLayerEventVersion},
+				)),
+		).OnConflict(goqu.DoNothing())
+	query, args, err := ds.Prepared(true).ToSQL()
+	if err != nil {
+		return 0, fmt.Errorf("error preparing query: %w", err)
+	}
+
+	res, err := tx.Exec(query, args...)
+	if err != nil {
+		return 0, fmt.Errorf("error executing query: %w", err)
+	}
+	c, err := res.RowsAffected()
+	if err != nil {
+		return 0, fmt.Errorf("error getting rows affected: %w", err)
+	}
+
+	return int(c), nil
+}