refat: coinbase - created TransactionDetails table to prevent storing empty data; adjusted queries to use TransactionDetails; changed and improved the coinbase scripts

Author: mschead

backfill/Dockerfile.create-coinbase

@@ -0,0 +1,11 @@
+FROM golang:1.23.7 AS builder
+WORKDIR /app
+COPY . .
+RUN go mod download
+RUN CGO_ENABLED=0 GOOS=linux GOARCH=arm64 go build -o create-coinbase ./create-coinbase/create-coinbase.go
+
+FROM scratch
+WORKDIR /app
+COPY ./global-bundle.pem ./global-bundle.pem
+COPY --from=builder /app/create-coinbase .
+CMD ["./create-coinbase"]

backfill/Dockerfile.migrate-coinbase

@@ -0,0 +1,11 @@
+FROM golang:1.23.7 AS builder
+WORKDIR /app
+COPY . .
+RUN go mod download
+RUN CGO_ENABLED=0 GOOS=linux GOARCH=arm64 go build -o migrate-coinbase ./migrate-coinbase/migrate-coinbase.go
+
+FROM scratch
+WORKDIR /app
+COPY ./global-bundle.pem ./global-bundle.pem
+COPY --from=builder /app/migrate-coinbase .
+CMD ["./migrate-coinbase"]

backfill/create-coinbase/create-coinbase.go

@@ -0,0 +1,179 @@
+package main
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"go-backfill/config"
+	"go-backfill/fetch"
+	"go-backfill/process"
+	"go-backfill/repository"
+	"log"
+	"strconv"
+	"time"
+
+	"github.com/jackc/pgx/v5"
+)
+
+const (
+	coinbaseBatchSize = 1000
+)
+
+type CoinbaseData struct {
+	ID           int64  `json:"id"`
+	Coinbase     string `json:"coinbase"`
+	ChainId      int    `json:"chainId"`
+	CreationTime string `json:"creationTime"`
+}
+
+func createBatchCoinbase(conn *pgx.Conn, lastId int64, network string) (bool, int64, error) {
+	startTime := time.Now()
+
+	// Start transaction for writes
+	tx, err := conn.Begin(context.Background())
+	if err != nil {
+		return false, lastId, fmt.Errorf("failed to begin transaction: %v", err)
+	}
+	defer tx.Rollback(context.Background())
+
+	// Fetch blocks with coinbase data using cursor pagination
+	query := `
+		SELECT id, coinbase, "chainId", "creationTime"
+		FROM "Blocks"
+		WHERE id > $1
+		ORDER BY id ASC
+		LIMIT $2
+	`
+
+	rows, err := conn.Query(context.Background(), query, lastId, coinbaseBatchSize)
+	if err != nil {
+		return false, lastId, fmt.Errorf("failed to execute query: %v", err)
+	}
+	defer rows.Close()
+
+	var blocks []CoinbaseData
+	for rows.Next() {
+		var block CoinbaseData
+		if err := rows.Scan(&block.ID, &block.Coinbase, &block.ChainId, &block.CreationTime); err != nil {
+			return false, lastId, fmt.Errorf("failed to scan row: %v", err)
+		}
+		blocks = append(blocks, block)
+	}
+
+	if len(blocks) == 0 {
+		return false, lastId, nil
+	}
+
+	// Process each block's coinbase transaction
+	var transactions []repository.TransactionAttributes
+	var transactionIds []int64
+	for _, block := range blocks {
+		creationTime, err := strconv.ParseInt(block.CreationTime, 10, 64)
+		if err != nil {
+			return false, lastId, fmt.Errorf("failed to parse creation time for block %d: %v", block.ID, err)
+		}
+		tx, err := process.ProcessCoinbaseTransaction(block.Coinbase, block.ID, creationTime, int64(block.ChainId))
+		if err != nil {
+			return false, lastId, fmt.Errorf("failed to process coinbase for block %d: %v", block.ID, err)
+		}
+		transactions = append(transactions, tx)
+	}
+
+	// Save transactions to database
+	if len(transactions) > 0 {
+		ids, err := repository.SaveTransactions(tx, transactions, repository.TransactionAttributes{})
+		if err != nil {
+			return false, lastId, fmt.Errorf("failed to save transactions: %v", err)
+		}
+		transactionIds = ids
+
+		// Process and save events and transfers for each coinbase transaction
+		for i, block := range blocks {
+			transactionId := transactionIds[i]
+
+			// Create a ProcessedPayload structure for the coinbase events
+			processedPayload := fetch.ProcessedPayload{
+				Header: fetch.Header{
+					ChainId: block.ChainId,
+				},
+				Coinbase: []byte(block.Coinbase),
+			}
+
+			// Prepare and save events
+			events, err := process.PrepareEvents(network, processedPayload, []int64{transactionId})
+			if err != nil {
+				return false, lastId, fmt.Errorf("failed to prepare events for block %d: %v", block.ID, err)
+			}
+
+			if err := repository.SaveEventsToDatabase(events, tx); err != nil {
+				return false, lastId, fmt.Errorf("failed to save events for block %d: %v", block.ID, err)
+			}
+
+			// Prepare and save transfers
+			transfers, err := process.PrepareTransfers(network, processedPayload, []int64{transactionId})
+			if err != nil {
+				return false, lastId, fmt.Errorf("failed to prepare transfers for block %d: %v", block.ID, err)
+			}
+
+			if err := repository.SaveTransfersToDatabase(transfers, tx); err != nil {
+				return false, lastId, fmt.Errorf("failed to save transfers for block %d: %v", block.ID, err)
+			}
+		}
+	}
+
+	if err := tx.Commit(context.Background()); err != nil {
+		return false, lastId, fmt.Errorf("failed to commit transaction: %v", err)
+	}
+
+	elapsed := time.Since(startTime)
+	log.Printf("Processed %d coinbase transactions, their events, and transfers. Batch time: %.2fs", len(transactions), elapsed.Seconds())
+
+	// Return the last processed ID as the new cursor
+	lastProcessedId := blocks[len(blocks)-1].ID
+	return len(blocks) == coinbaseBatchSize, lastProcessedId, nil
+}
+
+func main() {
+	envFile := flag.String("env", ".env", "Path to the .env file")
+	flag.Parse()
+
+	config.InitEnv(*envFile)
+	env := config.GetConfig()
+
+	// Database connection
+	connStr := fmt.Sprintf("host=%s port=%s user=%s password=%s dbname=%s sslmode=disable",
+		env.DbHost, env.DbPort, env.DbUser, env.DbPassword, env.DbName)
+
+	conn, err := pgx.Connect(context.Background(), connStr)
+	if err != nil {
+		log.Fatalf("Failed to connect to database: %v", err)
+	}
+	defer conn.Close(context.Background())
+
+	log.Println("Connected to database")
+
+	lastId := int64(0)
+	hasMore := true
+	totalBlocks := int64(104813544)
+	processedBlocks := int64(0)
+
+	for hasMore {
+		var err error
+		hasMore, lastId, err = createBatchCoinbase(conn, lastId, env.Network)
+		if err != nil {
+			log.Fatalf("Error during batch processing: %v", err)
+		}
+		processedBlocks += coinbaseBatchSize
+		progress := float64(processedBlocks) / float64(totalBlocks) * 100
+
+		if hasMore {
+			log.Printf("Progress: %.2f%% (%d/%d blocks processed)", progress, processedBlocks, totalBlocks)
+		} else {
+			log.Printf("Progress: 100.00%%")
+		}
+
+		time.Sleep(100 * time.Millisecond)
+	}
+
+	log.Println("Coinbase creation completed successfully")
+}

backfill/fetch/process_payloads.go

@@ -17,8 +17,8 @@ type Event struct {
 }
 
 type Module struct {
-	Namespace string `json:"namespace"`
-	Name      string `json:"name"`
+	Namespace *string `json:"namespace"`
+	Name      string  `json:"name"`
 }
 
 type Result struct {

backfill/middle-backfill/middle-backfill.go

@@ -37,7 +37,6 @@ func main() {
 		ChainIdStr := strconv.FormatInt(maxHeight.ChainID, 10)
 		Height := cuts.Hashes[ChainIdStr].Height
 		Hash := cuts.Hashes[ChainIdStr].Hash
-		// fmt.Println("Hash: ", Hash, " ChainId: ", ChainIdStr, " Height: ", Height, " MaxHeight: ", maxHeight.MaxHeight)
 		process.StartBackfill(Height, Hash, int(maxHeight.ChainID), int(maxHeight.MaxHeight+1), pool)
 	}
 }

backfill/migrate-coinbase/migrate-coinbase.go

@@ -0,0 +1,124 @@
+package main
+
+import (
+	"database/sql"
+	"flag"
+	"fmt"
+	"go-backfill/config"
+	"log"
+	"time"
+
+	_ "github.com/lib/pq" // PostgreSQL driver
+)
+
+const (
+	batchSize = 1000
+	stopID    = 173921979
+)
+
+func migrateBatchTransactions(db *sql.DB, lastID int) (int, bool, error) {
+	startTime := time.Now()
+
+	tx, err := db.Begin()
+	if err != nil {
+		return 0, false, fmt.Errorf("failed to begin transaction: %v", err)
+	}
+	defer tx.Rollback()
+
+	// Insert batch into TransactionDetails
+	insertQuery := `
+		INSERT INTO "TransactionDetails" (
+			"transactionId", code, continuation, data, gas, gaslimit, gasprice,
+			nonce, pactid, proof, rollback, sigs, step, ttl, "createdAt", "updatedAt"
+		)
+		SELECT 
+			id, code, continuation, data, gas, gaslimit, gasprice,
+			nonce, pactid, proof, rollback, sigs, step, ttl, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP
+		FROM "Transactions"
+		WHERE id > $1
+		ORDER BY id ASC
+		LIMIT $2
+		RETURNING "transactionId";
+	`
+
+	rows, err := tx.Query(insertQuery, lastID, batchSize)
+	if err != nil {
+		return 0, false, fmt.Errorf("failed to execute query: %v", err)
+	}
+	defer rows.Close()
+
+	var newLastID int
+	var insertedCount int
+
+	// Scan through all rows and keep the last ID
+	for rows.Next() {
+		if err := rows.Scan(&newLastID); err != nil {
+			return 0, false, fmt.Errorf("failed to scan row: %v", err)
+		}
+		insertedCount++
+	}
+
+	if err := tx.Commit(); err != nil {
+		return 0, false, fmt.Errorf("failed to commit transaction: %v", err)
+	}
+
+	elapsed := time.Since(startTime)
+	log.Printf("Inserted %d non-coinbase transactions. Last ID: %d. Batch time: %.2fs",
+		insertedCount, newLastID, elapsed.Seconds())
+
+	hasNext := insertedCount == batchSize
+
+	return newLastID, hasNext, nil
+}
+
+func migrateTransactionDbInformation(envFile string) error {
+	config.InitEnv(envFile)
+	env := config.GetConfig()
+
+	connStr := fmt.Sprintf("host=%s port=%s user=%s password=%s dbname=%s sslmode=disable",
+		env.DbHost, env.DbPort, env.DbUser, env.DbPassword, env.DbName)
+
+	db, err := sql.Open("postgres", connStr)
+	if err != nil {
+		return fmt.Errorf("failed to connect to database: %v", err)
+	}
+	defer db.Close()
+
+	if err := db.Ping(); err != nil {
+		return fmt.Errorf("failed to ping database: %v", err)
+	}
+
+	log.Println("Connected to database")
+
+	lastIdInserted := 0
+
+	hasNext := true
+	for hasNext {
+		newLastId, newHasNext, err := migrateBatchTransactions(db, lastIdInserted)
+		if err != nil {
+			return fmt.Errorf("error during batch insert: %v", err)
+		}
+
+		lastIdInserted = newLastId
+		hasNext = newHasNext
+		time.Sleep(100 * time.Millisecond)
+		if hasNext {
+			log.Printf("Progress: %.2f%%", float64(lastIdInserted)/float64(stopID)*100)
+		} else {
+			log.Printf("Progress: 100.00%%")
+		}
+	}
+
+	log.Println("Migration completed successfully")
+	return nil
+}
+
+func main() {
+	envFile := flag.String("env", ".env", "Path to the .env file")
+	flag.Parse()
+
+	// Run the migration
+	if err := migrateTransactionDbInformation(*envFile); err != nil {
+		log.Fatalf("Migration failed: %v", err)
+	}
+}