migrate with destination storage

jakeloo · jakeloo · commit debc231eb7f1 · 2025-08-26T08:05:30.000Z
diff --git a/cmd/migrate_valid.go b/cmd/migrate_valid.go
diff --git a/cmd/root.go b/cmd/root.go
diff --git a/configs/config.go b/configs/config.go
@@ -52,13 +52,6 @@ type StorageConfig struct {
 	Main         StorageConnectionConfig `mapstructure:"main"`
 	Orchestrator StorageConnectionConfig `mapstructure:"orchestrator"`
 }
-type StorageType string
-
-const (
-	StorageTypeMain         StorageType = "main"
-	StorageTypeStaging      StorageType = "staging"
-	StorageTypeOrchestrator StorageType = "orchestrator"
-)
 
 type StorageConnectionConfig struct {
 	Type       string            `mapstructure:"type"` // "auto", "clickhouse", "postgres", "kafka", "badger", "s3"
@@ -116,6 +109,7 @@ type ClickhouseConfig struct {
 	EnableParallelViewProcessing bool                           `mapstructure:"enableParallelViewProcessing"`
 	MaxQueryTime                 int                            `mapstructure:"maxQueryTime"`
 	MaxMemoryUsage               int                            `mapstructure:"maxMemoryUsage"`
+	EnableCompression            bool                           `mapstructure:"enableCompression"`
 }
 
 type PostgresConfig struct {
@@ -238,6 +232,14 @@ type ValidationConfig struct {
 	Mode string `mapstructure:"mode"` // "disabled", "minimal", "strict"
 }
 
+type MigratorConfig struct {
+	Destination      StorageConnectionConfig `mapstructure:"destination"`
+	StartBlock       uint                    `mapstructure:"startBlock"`
+	EndBlock         uint                    `mapstructure:"endBlock"`
+	StorageBatchSize uint                    `mapstructure:"storageBatchSize"`
+	RpcBatchSize     uint                    `mapstructure:"rpcBatchSize"`
+}
+
 type Config struct {
 	RPC              RPCConfig              `mapstructure:"rpc"`
 	Log              LogConfig              `mapstructure:"log"`
@@ -250,6 +252,7 @@ type Config struct {
 	Publisher        PublisherConfig        `mapstructure:"publisher"`
 	WorkMode         WorkModeConfig         `mapstructure:"workMode"`
 	Validation       ValidationConfig       `mapstructure:"validation"`
+	Migrator         MigratorConfig         `mapstructure:"migrator"`
 }
 
 var Cfg Config
diff --git a/internal/common/block.go b/internal/common/block.go
@@ -59,7 +59,6 @@ type BlockModel struct {
 }
 
 type BlockData struct {
-	ChainId      uint64        `json:"chain_id"`
 	Block        Block         `json:"block"`
 	Transactions []Transaction `json:"transactions"`
 	Logs         []Log         `json:"logs"`
@@ -103,7 +102,6 @@ func (b *Block) Serialize() BlockModel {
 
 func (b *BlockData) Serialize() BlockData {
 	data := BlockData{
-		ChainId:      b.ChainId,
 		Block:        b.Block,
 		Transactions: b.Transactions,
 		Logs:         b.Logs,
diff --git a/internal/orchestrator/failure_recoverer.go b/internal/orchestrator/failure_recoverer.go
@@ -110,7 +110,6 @@ func (fr *FailureRecoverer) handleWorkerResults(blockFailures []common.BlockFail
 			})
 		} else {
 			successfulResults = append(successfulResults, common.BlockData{
-				ChainId:      fr.rpc.GetChainID().Uint64(),
 				Block:        result.Data.Block,
 				Logs:         result.Data.Logs,
 				Transactions: result.Data.Transactions,
diff --git a/internal/orchestrator/poller.go b/internal/orchestrator/poller.go
@@ -261,7 +261,6 @@ func (p *Poller) convertPollResultsToBlockData(results []rpc.GetFullBlockResult)
 	blockData := make([]common.BlockData, 0, len(successfulResults))
 	for _, result := range successfulResults {
 		blockData = append(blockData, common.BlockData{
-			ChainId:      p.rpc.GetChainID().Uint64(),
 			Block:        result.Data.Block,
 			Logs:         result.Data.Logs,
 			Transactions: result.Data.Transactions,
diff --git a/internal/orchestrator/reorg_handler.go b/internal/orchestrator/reorg_handler.go
@@ -274,7 +274,6 @@ func (rh *ReorgHandler) handleReorg(ctx context.Context, reorgedBlockNumbers []*
 			return fmt.Errorf("cannot fix reorg: failed block %s: %w", result.BlockNumber.String(), result.Error)
 		}
 		data = append(data, common.BlockData{
-			ChainId:      rh.rpc.GetChainID().Uint64(),
 			Block:        result.Data.Block,
 			Logs:         result.Data.Logs,
 			Transactions: result.Data.Transactions,
diff --git a/internal/storage/badger.go b/internal/storage/badger.go
@@ -237,7 +237,7 @@ func (bc *BadgerConnector) InsertStagingData(data []common.BlockData) error {
 
 	return bc.db.Update(func(txn *badger.Txn) error {
 		for _, blockData := range data {
-			key := blockKey(big.NewInt(int64(blockData.ChainId)), blockData.Block.Number)
+			key := blockKey(blockData.Block.ChainId, blockData.Block.Number)
 
 			var buf bytes.Buffer
 			if err := gob.NewEncoder(&buf).Encode(blockData); err != nil {
@@ -348,7 +348,7 @@ func (bc *BadgerConnector) DeleteStagingData(data []common.BlockData) error {
 
 	return bc.db.Update(func(txn *badger.Txn) error {
 		for _, blockData := range data {
-			key := blockKey(big.NewInt(int64(blockData.ChainId)), blockData.Block.Number)
+			key := blockKey(blockData.Block.ChainId, blockData.Block.Number)
 			if err := txn.Delete(key); err != nil && err != badger.ErrKeyNotFound {
 				return err
 			}
diff --git a/internal/storage/clickhouse.go b/internal/storage/clickhouse.go
@@ -107,6 +107,14 @@ func connectDB(cfg *config.ClickhouseConfig) (clickhouse.Conn, error) {
 		},
 		MaxOpenConns: cfg.MaxOpenConns,
 		MaxIdleConns: cfg.MaxIdleConns,
+		Compression: func() *clickhouse.Compression {
+			c := &clickhouse.Compression{}
+			if cfg.EnableCompression {
+				zLog.Debug().Msg("ClickHouse LZ4 compression is enabled")
+				c.Method = clickhouse.CompressionLZ4
+			}
+			return c
+		}(),
 		Settings: func() clickhouse.Settings {
 			settings := clickhouse.Settings{
 				"do_not_merge_across_partitions_select_final": "1",
@@ -901,6 +909,19 @@ func (c *ClickHouseConnector) GetMaxBlockNumberInRange(chainId *big.Int, startBl
 	return maxBlockNumber, nil
 }
 
+func (c *ClickHouseConnector) GetBlockCount(chainId *big.Int, startBlock *big.Int, endBlock *big.Int) (blockCount *big.Int, err error) {
+	tableName := c.getTableName(chainId, "blocks")
+	query := fmt.Sprintf("SELECT COUNT(DISTINCT block_number) FROM %s.%s WHERE chain_id = ? AND block_number >= ? AND block_number <= ?", c.cfg.Database, tableName)
+	err = c.conn.QueryRow(context.Background(), query, chainId, startBlock, endBlock).Scan(&blockCount)
+	if err != nil {
+		if err == sql.ErrNoRows {
+			return big.NewInt(0), nil
+		}
+		return nil, err
+	}
+	return blockCount, nil
+}
+
 func (c *ClickHouseConnector) getMaxBlockNumberConsistent(chainId *big.Int) (maxBlockNumber *big.Int, err error) {
 	tableName := c.getTableName(chainId, "blocks")
 	query := fmt.Sprintf("SELECT block_number FROM %s.%s WHERE chain_id = ? ORDER BY block_number DESC LIMIT 1 SETTINGS select_sequential_consistency = 1", c.cfg.Database, tableName)
@@ -1976,7 +1997,6 @@ func (c *ClickHouseConnector) GetValidationBlockData(chainId *big.Int, startBloc
 	for i, block := range blocksResult.blocks {
 		blockNum := block.Number.String()
 		blockData[i] = common.BlockData{
-			ChainId:      chainId.Uint64(),
 			Block:        block,
 			Logs:         logsResult.logMap[blockNum],
 			Transactions: txsResult.txMap[blockNum],
@@ -2156,7 +2176,6 @@ func (c *ClickHouseConnector) GetFullBlockData(chainId *big.Int, blockNumbers []
 	for i, block := range blocksResult.blocks {
 		blockNum := block.Number.String()
 		blockData[i] = common.BlockData{
-			ChainId:      chainId.Uint64(),
 			Block:        block,
 			Logs:         logsResult.logMap[blockNum],
 			Transactions: txsResult.txMap[blockNum],
diff --git a/internal/storage/connector.go b/internal/storage/connector.go
@@ -130,6 +130,8 @@ type IMainStorage interface {
 
 	GetMaxBlockNumber(chainId *big.Int) (maxBlockNumber *big.Int, err error)
 	GetMaxBlockNumberInRange(chainId *big.Int, startBlock *big.Int, endBlock *big.Int) (maxBlockNumber *big.Int, err error)
+	GetBlockCount(chainId *big.Int, startBlock *big.Int, endBlock *big.Int) (blockCount *big.Int, err error)
+
 	/**
 	 * Get block headers ordered from latest to oldest.
 	 */
@@ -175,13 +177,13 @@ func NewStorageConnector(cfg *config.StorageConfig) (IStorage, error) {
 func NewConnector[T any](cfg *config.StorageConnectionConfig) (T, error) {
 	var conn interface{}
 	var err error
-	
+
 	// Default to "auto" if Type is not specified
 	storageType := cfg.Type
 	if storageType == "" {
 		storageType = "auto"
 	}
-	
+
 	// Handle explicit type selection
 	if storageType != "auto" {
 		switch storageType {
diff --git a/internal/storage/kafka_publisher.go b/internal/storage/kafka_publisher.go
@@ -18,9 +18,8 @@ import (
 )
 
 type KafkaPublisher struct {
-	client  *kgo.Client
-	mu      sync.RWMutex
-	chainID string
+	client *kgo.Client
+	mu     sync.RWMutex
 }
 
 type MessageType string
@@ -37,6 +36,7 @@ type PublishableMessagePayload struct {
 
 type PublishableMessageBlockData struct {
 	common.BlockData
+	ChainId         uint64    `json:"chain_id"`
 	IsDeleted       int8      `json:"is_deleted"`
 	InsertTimestamp time.Time `json:"insert_timestamp"`
 }
@@ -104,8 +104,7 @@ func NewKafkaPublisher(cfg *config.KafkaConfig) (*KafkaPublisher, error) {
 	}
 
 	publisher := &KafkaPublisher{
-		client:  client,
-		chainID: chainID,
+		client: client,
 	}
 
 	return publisher, nil
@@ -116,9 +115,10 @@ func (p *KafkaPublisher) PublishBlockData(blockData []common.BlockData) error {
 }
 
 func (p *KafkaPublisher) PublishReorg(oldData []common.BlockData, newData []common.BlockData) error {
+	chainId := newData[0].Block.ChainId.Uint64()
 	newHead := uint64(newData[0].Block.Number.Uint64())
 	// Publish revert the revert to the new head - 1, so that the new updated block data can be re-processed
-	if err := p.publishBlockRevert(newData[0].ChainId, newHead-1); err != nil {
+	if err := p.publishBlockRevert(chainId, newHead-1); err != nil {
 		return fmt.Errorf("failed to revert: %v", err)
 	}
 
@@ -233,6 +233,7 @@ func (p *KafkaPublisher) createBlockDataMessage(block common.BlockData, isDelete
 
 	data := PublishableMessageBlockData{
 		BlockData:       block,
+		ChainId:         block.Block.ChainId.Uint64(),
 		IsDeleted:       0,
 		InsertTimestamp: timestamp,
 	}
@@ -251,7 +252,7 @@ func (p *KafkaPublisher) createBlockDataMessage(block common.BlockData, isDelete
 		return nil, fmt.Errorf("failed to marshal block data: %v", err)
 	}
 
-	return p.createRecord(data.GetType(), block.ChainId, block.Block.Number.Uint64(), timestamp, msgJson)
+	return p.createRecord(data.GetType(), data.ChainId, block.Block.Number.Uint64(), timestamp, msgJson)
 }
 
 func (p *KafkaPublisher) createBlockRevertMessage(chainId uint64, blockNumber uint64) (*kgo.Record, error) {
diff --git a/internal/storage/kafka_redis.go b/internal/storage/kafka_redis.go
@@ -212,26 +212,11 @@ func (kr *KafkaRedisConnector) GetMaxBlockNumber(chainId *big.Int) (*big.Int, er
 }
 
 func (kr *KafkaRedisConnector) GetMaxBlockNumberInRange(chainId *big.Int, startBlock *big.Int, endBlock *big.Int) (*big.Int, error) {
-	// Get the last published block number
-	lastPublished, err := kr.GetLastPublishedBlockNumber(chainId)
-	if err != nil {
-		return nil, err
-	}
-
-	// Check if it's within the range
-	if lastPublished.Cmp(startBlock) >= 0 && lastPublished.Cmp(endBlock) <= 0 {
-		return lastPublished, nil
-	}
-
-	// If outside range, return appropriate boundary
-	if lastPublished.Cmp(endBlock) > 0 {
-		return endBlock, nil
-	}
-	if lastPublished.Cmp(startBlock) < 0 {
-		return big.NewInt(0), nil
-	}
+	return nil, fmt.Errorf("query operations are not supported with KafkaRedis connector - this is a write-only connector for streaming")
+}
 
-	return lastPublished, nil
+func (kr *KafkaRedisConnector) GetBlockCount(chainId *big.Int, startBlock *big.Int, endBlock *big.Int) (*big.Int, error) {
+	return nil, fmt.Errorf("query operations are not supported with KafkaRedis connector - this is a write-only connector for streaming")
 }
 
 func (kr *KafkaRedisConnector) GetBlockHeadersDescending(chainId *big.Int, from *big.Int, to *big.Int) ([]common.BlockHeader, error) {
diff --git a/internal/storage/s3.go b/internal/storage/s3.go
diff --git a/test/mocks/MockIMainStorage.go b/test/mocks/MockIMainStorage.go

Original file line number	Diff line number	Diff line change
`@@ -274,7 +274,6 @@ func (rh ReorgHandler) handleReorg(ctx context.Context, reorgedBlockNumbers []`
`274`	`274`	`return fmt.Errorf("cannot fix reorg: failed block %s: %w", result.BlockNumber.String(), result.Error)`
`275`	`275`	`}`
`276`	`276`	`data = append(data, common.BlockData{`
`277`		`- ChainId: rh.rpc.GetChainID().Uint64(),`
`278`	`277`	`Block: result.Data.Block,`
`279`	`278`	`Logs: result.Data.Logs,`
`280`	`279`	`Transactions: result.Data.Transactions,`