added cmd fix-any-type

cmd/fireeth/main.go

@@ -99,6 +99,7 @@ func Chain() *firecore.Chain[*pbeth.Block] {
 				parent.AddCommand(compareOneblockRPCCmd)
 				parent.AddCommand(newCompareBlocksRPCCmd(zlog))
 				parent.AddCommand(newFixOrdinalsCmd(zlog))
+				parent.AddCommand(newFixAnyTypeCmd(zlog))
 				parent.AddCommand(newPollRPCBlocksCmd(zlog))
 				parent.AddCommand(newPollerCmd(zlog, tracer))
 				parent.AddCommand(newOptimismPollerCmd(zlog, tracer))

cmd/fireeth/tools_fix_any_type.go

@@ -0,0 +1,110 @@
+package main
+
+import (
+	"fmt"
+	"io"
+	"strings"
+
+	"github.com/spf13/cobra"
+	"github.com/streamingfast/bstream"
+	pbbstream "github.com/streamingfast/bstream/pb/sf/bstream/v1"
+	"github.com/streamingfast/dstore"
+	firecore "github.com/streamingfast/firehose-core"
+	"go.uber.org/zap"
+)
+
+func newFixAnyTypeCmd(logger *zap.Logger) *cobra.Command {
+	return &cobra.Command{
+		Use:   "fix-any-type <src-blocks-store> <dest-blocks-store> <start-block> <stop-block>",
+		Short: "look for blocks with missing type url prefix 'type.googleapis.com' and add it.",
+		Args:  cobra.ExactArgs(4),
+		RunE:  createFixAnyTypeE(logger),
+	}
+}
+
+func createFixAnyTypeE(logger *zap.Logger) firecore.CommandExecutor {
+	return func(cmd *cobra.Command, args []string) error {
+		ctx := cmd.Context()
+
+		srcStore, err := dstore.NewDBinStore(args[0])
+		if err != nil {
+			return fmt.Errorf("unable to create source store: %w", err)
+		}
+
+		destStore, err := dstore.NewDBinStore(args[1])
+		if err != nil {
+			return fmt.Errorf("unable to create destination store: %w", err)
+		}
+
+		start := mustParseUint64(args[2])
+		stop := mustParseUint64(args[3])
+
+		if stop <= start {
+			return fmt.Errorf("stop block must be greater than start block")
+		}
+
+		lastFileProcessed := ""
+		startWalkFrom := fmt.Sprintf("%010d", start-(start%100))
+		err = srcStore.WalkFrom(ctx, "", startWalkFrom, func(filename string) error {
+			logger.Debug("checking merged block file", zap.String("filename", filename))
+
+			startBlock := mustParseUint64(filename)
+
+			if startBlock > stop {
+				logger.Debug("stopping at merged block file above stop block", zap.String("filename", filename), zap.Uint64("stop", stop))
+				return io.EOF
+			}
+
+			if startBlock+100 < start {
+				logger.Debug("skipping merged block file below start block", zap.String("filename", filename))
+				return nil
+			}
+
+			rc, err := srcStore.OpenObject(ctx, filename)
+			if err != nil {
+				return fmt.Errorf("failed to open %s: %w", filename, err)
+			}
+			defer rc.Close()
+
+			br, err := bstream.NewDBinBlockReader(rc)
+			if err != nil {
+				return fmt.Errorf("creating block reader: %w", err)
+			}
+
+			blocks := make([]*pbbstream.Block, 100)
+			i := 0
+			for {
+				block, err := br.Read()
+				if err == io.EOF {
+					break
+				}
+				if !strings.HasPrefix(block.Payload.TypeUrl, "type.googleapis.com/") {
+					block.Payload.TypeUrl = "type.googleapis.com/" + block.Payload.TypeUrl
+				}
+				blocks[i] = block
+				i++
+			}
+			if i != 100 {
+				return fmt.Errorf("expected to have read 100 blocks, we have read %d. Bailing out.", i)
+			}
+			if err := writeMergedBlocks(startBlock, destStore, blocks); err != nil {
+				return fmt.Errorf("writing merged block %d: %w", startBlock, err)
+			}
+
+			lastFileProcessed = filename
+
+			return nil
+		})
+		fmt.Printf("Last file processed: %s.dbin.zst\n", lastFileProcessed)
+
+		if err == io.EOF {
+			return nil
+		}
+
+		if err != nil {
+			return err
+		}
+
+		return nil
+	}
+}