Merge pull request #357 from hack-a-chain-software/balances-backfill

refactor: improved balances backfill; normalized the gas price format in the database to a decimal to ensure consistency

Author: 0xneves

backfill/process/process_balances.go

@@ -0,0 +1,104 @@
+package process
+
+import (
+	"encoding/json"
+	"fmt"
+	"go-backfill/repository"
+)
+
+// PrepareFungibleBalancesData processes regular token transfers to prepare fungible balance data
+func PrepareFungibleBalancesData(transfers []repository.TransferAttributes) []repository.BalanceAttributes {
+	// Pre-allocate with estimated size (2 balances per transfer)
+	balances := make([]repository.BalanceAttributes, 0, len(transfers)*2)
+
+	// Process transfers
+	for _, transfer := range transfers {
+
+		// Add from account if not empty
+		if transfer.FromAcct != "" {
+			balances = append(balances, repository.BalanceAttributes{
+				ChainId:    transfer.ChainId,
+				Account:    transfer.FromAcct,
+				Module:     transfer.ModuleName,
+				HasTokenId: transfer.HasTokenId,
+				TokenId:    stringValueOrEmpty(transfer.TokenId),
+			})
+		}
+
+		// Add to account if not empty
+		if transfer.ToAcct != "" {
+			balances = append(balances, repository.BalanceAttributes{
+				ChainId:    transfer.ChainId,
+				Account:    transfer.ToAcct,
+				Module:     transfer.ModuleName,
+				HasTokenId: transfer.HasTokenId,
+				TokenId:    stringValueOrEmpty(transfer.TokenId),
+			})
+		}
+	}
+
+	return balances
+}
+
+// PrepareNonFungibleBalancesData processes Marmalade events to prepare NFT balance data
+func PrepareNonFungibleBalancesData(events []repository.EventAttributes) ([]repository.BalanceAttributes, error) {
+	// Pre-allocate with estimated size (1 balance per mint/transfer event)
+	balances := make([]repository.BalanceAttributes, 0, len(events))
+
+	// Process mint and transfer events for Marmalade NFTs
+	for _, event := range events {
+		var isMarmaladeEvent = event.Module == "marmalade-v2.ledger" || event.Module == "marmalade.ledger"
+		var isMarmaladeTransferEvent = event.Name == "TRANSFER"
+		var isMarmaladeMintEvent = event.Name == "MINT"
+		if isMarmaladeEvent && (isMarmaladeTransferEvent || isMarmaladeMintEvent) {
+			var params []interface{}
+			if err := json.Unmarshal(event.Params, &params); err != nil {
+				return []repository.BalanceAttributes{}, fmt.Errorf("unmarshaling params for event %s: %w", event.RequestKey, err)
+			}
+
+			switch event.Name {
+			case "MINT":
+				if len(params) < 2 {
+					continue // Skip if we don't have enough params
+				}
+				tokenId, ok1 := params[0].(string)
+				account, ok2 := params[1].(string)
+				if !ok1 || !ok2 {
+					continue
+				}
+				balances = append(balances, repository.BalanceAttributes{
+					ChainId:    event.ChainId,
+					TokenId:    tokenId,
+					Account:    account,
+					Module:     event.Module,
+					HasTokenId: true,
+				})
+			case "TRANSFER":
+				if len(params) < 3 {
+					continue // Skip if we don't have enough params
+				}
+				tokenId, ok1 := params[0].(string)
+				recipient, ok2 := params[2].(string)
+				if !ok1 || !ok2 {
+					continue
+				}
+				balances = append(balances, repository.BalanceAttributes{
+					ChainId:    event.ChainId,
+					TokenId:    tokenId,
+					Account:    recipient, // params[2] is the recipient account
+					Module:     event.Module,
+					HasTokenId: true,
+				})
+			}
+		}
+	}
+
+	return balances, nil
+}
+
+func stringValueOrEmpty(s *string) string {
+	if s == nil {
+		return ""
+	}
+	return *s
+}

backfill/process/process_guards.go

@@ -16,13 +16,15 @@ type Counters struct {
 	Events       int
 	Transfers    int
 	Signers      int
+	Balances     int
 }
 
 type DataSizeTracker struct {
 	TransactionsKB int
 	EventsKB       int
 	TransfersKB    int
 	SignersKB      int
+	BalancesKB     int
 }
 
 func savePayloads(network string, chainId int, processedPayloads []fetch.ProcessedPayload, pool *pgxpool.Pool) (Counters, DataSizeTracker, error) {
@@ -33,13 +35,15 @@ func savePayloads(network string, chainId int, processedPayloads []fetch.Process
 		Events:       0,
 		Transfers:    0,
 		Signers:      0,
+		Balances:     0,
 	}
 
 	dataSizeTracker := DataSizeTracker{
 		TransactionsKB: 0,
 		EventsKB:       0,
 		TransfersKB:    0,
 		SignersKB:      0,
+		BalancesKB:     0,
 	}
 
 	conn, err := pool.Acquire(context.Background())
@@ -100,6 +104,18 @@ func savePayloads(network string, chainId int, processedPayloads []fetch.Process
 			return Counters{}, DataSizeTracker{}, fmt.Errorf("saving events -> %w", err)
 		}
 
+		balances, err := PrepareNonFungibleBalancesData(events)
+		if err != nil {
+			return Counters{}, DataSizeTracker{}, fmt.Errorf("preparing non fungible balances -> %w", err)
+		}
+		if err := repository.SaveBalancesToDatabase(balances, tx); err != nil {
+			return Counters{}, DataSizeTracker{}, fmt.Errorf("saving non fungible balances: %w", err)
+		}
+
+		balancesSize := approximateSize(balances)
+		dataSizeTracker.BalancesKB += balancesSize
+		counters.Balances += len(balances)
+
 		size := approximateSize(events)
 		dataSizeTracker.EventsKB += size
 		counters.Events += len(events)
@@ -114,6 +130,15 @@ func savePayloads(network string, chainId int, processedPayloads []fetch.Process
 			return Counters{}, DataSizeTracker{}, fmt.Errorf("saving transfers: %w", err)
 		}
 
+		balances := PrepareFungibleBalancesData(transfers)
+		if err := repository.SaveBalancesToDatabase(balances, tx); err != nil {
+			return Counters{}, DataSizeTracker{}, fmt.Errorf("saving fungible balances: %w", err)
+		}
+
+		balancesSize := approximateSize(balances)
+		dataSizeTracker.BalancesKB += balancesSize
+		counters.Balances += len(balances)
+
 		size := approximateSize(transfers)
 		dataSizeTracker.TransfersKB += size
 		counters.Transfers += len(transfers)
@@ -130,20 +155,6 @@ func savePayloads(network string, chainId int, processedPayloads []fetch.Process
 		counters.Signers += len(signers)
 	}
 
-	// for _, processedPayload := range processedPayloads {
-	// 	guards, err := PrepareGuards(processedPayload)
-	// 	if err != nil {
-	// 		return Counters{}, DataSizeTracker{}, fmt.Errorf("preparing guards -> %w", err)
-	// 	}
-	// 	if err := repository.SaveGuardsToDatabase(guards, tx); err != nil {
-	// 		return Counters{}, DataSizeTracker{}, fmt.Errorf("saving guards -> %w", err)
-	// 	}
-
-	// 	size := approximateSize(guards)
-	// 	dataSizeTracker.GuardsKB += size
-	// 	counters.Guards += len(guards)
-	// }
-
 	log.Printf("Saved payloads in %fs\n", time.Since(startTime).Seconds())
 
 	commitStartTime := time.Now()
@@ -156,6 +167,7 @@ func savePayloads(network string, chainId int, processedPayloads []fetch.Process
 	dataSizeTracker.EventsKB /= 1024
 	dataSizeTracker.TransfersKB /= 1024
 	dataSizeTracker.SignersKB /= 1024
+	dataSizeTracker.BalancesKB /= 1024
 
 	return counters, dataSizeTracker, nil
 }