new kafka updates

Author: nischitpra

cmd/migrate_valid.go

@@ -104,7 +104,7 @@ type Migrator struct {
 	targetConn         *storage.ClickHouseConnector
 	migrationBatchSize int
 	rpcBatchSize       int
-	newkafka           *newkafka.Publisher
+	newkafka           *newkafka.NewKafka
 	psql               *storage.PostgresConnector
 }
 

cmd/root.go

@@ -149,6 +149,9 @@ func init() {
 	rootCmd.PersistentFlags().String("publisher-events-topicName", "", "Kafka topic name for events")
 	rootCmd.PersistentFlags().String("publisher-events-addressFilter", "", "Filter events by address")
 	rootCmd.PersistentFlags().String("publisher-events-topic0Filter", "", "Filter events by topic0")
+	rootCmd.PersistentFlags().String("newKafka-brokers", "", "New Kafka brokers (comma-separated)")
+	rootCmd.PersistentFlags().String("newKafka-username", "", "New Kafka username")
+	rootCmd.PersistentFlags().String("newKafka-password", "", "New Kafka password")
 	rootCmd.PersistentFlags().Int("workMode-checkIntervalMinutes", 10, "How often to check work mode in minutes")
 	rootCmd.PersistentFlags().Int64("workMode-liveModeThreshold", 500, "How many blocks the indexer can be behind before switching to live mode")
 	rootCmd.PersistentFlags().String("validation-mode", "strict", "Validation mode. Strict will validate logsBloom and transactionsRoot. Minimal will validate transaction count and logs existence.")
@@ -265,6 +268,9 @@ func init() {
 	viper.BindPFlag("publisher.events.topicName", rootCmd.PersistentFlags().Lookup("publisher-events-topicName"))
 	viper.BindPFlag("publisher.events.addressFilter", rootCmd.PersistentFlags().Lookup("publisher-events-addressFilter"))
 	viper.BindPFlag("publisher.events.topic0Filter", rootCmd.PersistentFlags().Lookup("publisher-events-topic0Filter"))
+	viper.BindPFlag("newKafka.brokers", rootCmd.PersistentFlags().Lookup("newKafka-brokers"))
+	viper.BindPFlag("newKafka.username", rootCmd.PersistentFlags().Lookup("newKafka-username"))
+	viper.BindPFlag("newKafka.password", rootCmd.PersistentFlags().Lookup("newKafka-password"))
 	viper.BindPFlag("workMode.checkIntervalMinutes", rootCmd.PersistentFlags().Lookup("workMode-checkIntervalMinutes"))
 	viper.BindPFlag("workMode.liveModeThreshold", rootCmd.PersistentFlags().Lookup("workMode-liveModeThreshold"))
 	viper.BindPFlag("validation.mode", rootCmd.PersistentFlags().Lookup("validation-mode"))

configs/config.go

@@ -182,6 +182,12 @@ type PublisherConfig struct {
 	Events       EventPublisherConfig       `mapstructure:"events"`
 }
 
+type NewKafkaConfig struct {
+	Brokers  string `mapstructure:"brokers"`
+	Username string `mapstructure:"username"`
+	Password string `mapstructure:"password"`
+}
+
 type WorkModeConfig struct {
 	CheckIntervalMinutes int   `mapstructure:"checkIntervalMinutes"`
 	LiveModeThreshold    int64 `mapstructure:"liveModeThreshold"`
@@ -201,6 +207,7 @@ type Config struct {
 	Storage          StorageConfig          `mapstructure:"storage"`
 	API              APIConfig              `mapstructure:"api"`
 	Publisher        PublisherConfig        `mapstructure:"publisher"`
+	NewKafka         NewKafkaConfig         `mapstructure:"newKafka"`
 	WorkMode         WorkModeConfig         `mapstructure:"workMode"`
 	Validation       ValidationConfig       `mapstructure:"validation"`
 }

internal/publisher/newkafka/publisherNewKafka.go

@@ -18,13 +18,13 @@ import (
 	"github.com/twmb/franz-go/pkg/sasl/plain"
 )
 
-type Publisher struct {
+type NewKafka struct {
 	client *kgo.Client
 	mu     sync.RWMutex
 }
 
 var (
-	instance *Publisher
+	instance *NewKafka
 	once     sync.Once
 )
 
@@ -34,31 +34,26 @@ type PublishableMessage[T common.BlockModel | common.TransactionModel | common.L
 }
 
 // GetInstance returns the singleton Publisher instance
-func GetInstance() *Publisher {
+func GetInstance() *NewKafka {
 	once.Do(func() {
-		instance = &Publisher{}
+		instance = &NewKafka{}
 		if err := instance.initialize(); err != nil {
 			log.Error().Err(err).Msg("Failed to initialize publisher")
 		}
 	})
 	return instance
 }
 
-func (p *Publisher) initialize() error {
-	if !config.Cfg.Publisher.Enabled {
-		log.Debug().Msg("Publisher is disabled, skipping initialization")
-		return nil
-	}
-
+func (p *NewKafka) initialize() error {
 	p.mu.Lock()
 	defer p.mu.Unlock()
 
-	if config.Cfg.Publisher.Brokers == "" {
+	if config.Cfg.NewKafka.Brokers == "" {
 		log.Info().Msg("No Kafka brokers configured, skipping publisher initialization")
 		return nil
 	}
 
-	brokers := strings.Split(config.Cfg.Publisher.Brokers, ",")
+	brokers := strings.Split(config.Cfg.NewKafka.Brokers, ",")
 	opts := []kgo.Opt{
 		kgo.SeedBrokers(brokers...),
 		kgo.AllowAutoTopicCreation(),
@@ -71,10 +66,10 @@ func (p *Publisher) initialize() error {
 		kgo.DialTimeout(10 * time.Second),
 	}
 
-	if config.Cfg.Publisher.Username != "" && config.Cfg.Publisher.Password != "" {
+	if config.Cfg.NewKafka.Username != "" && config.Cfg.NewKafka.Password != "" {
 		opts = append(opts, kgo.SASL(plain.Auth{
-			User: config.Cfg.Publisher.Username,
-			Pass: config.Cfg.Publisher.Password,
+			User: config.Cfg.NewKafka.Username,
+			Pass: config.Cfg.NewKafka.Password,
 		}.AsMechanism()))
 		tlsDialer := &tls.Dialer{NetDialer: &net.Dialer{Timeout: 10 * time.Second}}
 		opts = append(opts, kgo.Dialer(tlsDialer.DialContext))
@@ -96,22 +91,11 @@ func (p *Publisher) initialize() error {
 	return nil
 }
 
-func (p *Publisher) PublishBlockData(blockData []common.BlockData) error {
+func (p *NewKafka) PublishBlockData(blockData []common.BlockData) error {
 	return p.publishBlockData(blockData, false)
 }
 
-func (p *Publisher) PublishReorg(oldData []common.BlockData, newData []common.BlockData) error {
-	if err := p.publishBlockData(oldData, true); err != nil {
-		return fmt.Errorf("failed to publish old block data: %v", err)
-	}
-
-	if err := p.publishBlockData(newData, false); err != nil {
-		return fmt.Errorf("failed to publish new block data: %v", err)
-	}
-	return nil
-}
-
-func (p *Publisher) Close() error {
+func (p *NewKafka) Close() error {
 	p.mu.Lock()
 	defer p.mu.Unlock()
 
@@ -122,40 +106,7 @@ func (p *Publisher) Close() error {
 	return nil
 }
 
-func (p *Publisher) publishMessages(ctx context.Context, messages []*kgo.Record) error {
-	if len(messages) == 0 {
-		return nil
-	}
-
-	if !config.Cfg.Publisher.Enabled {
-		log.Debug().Msg("Publisher is disabled, skipping publish")
-		return nil
-	}
-
-	p.mu.RLock()
-	defer p.mu.RUnlock()
-
-	if p.client == nil {
-		return nil // Skip if no client configured
-	}
-
-	var wg sync.WaitGroup
-	wg.Add(len(messages))
-	// Publish to all configured producers
-	for _, msg := range messages {
-		p.client.Produce(ctx, msg, func(_ *kgo.Record, err error) {
-			defer wg.Done()
-			if err != nil {
-				log.Error().Err(err).Msg("Failed to publish message to Kafka")
-			}
-		})
-	}
-	wg.Wait()
-
-	return nil
-}
-
-func (p *Publisher) publishBlockData(blockData []common.BlockData, isReorg bool) error {
+func (p *NewKafka) publishBlockData(blockData []common.BlockData, isReorg bool) error {
 	if p.client == nil || len(blockData) == 0 {
 		return nil
 	}
@@ -176,7 +127,7 @@ func (p *Publisher) publishBlockData(blockData []common.BlockData, isReorg bool)
 			return fmt.Errorf("failed to marshal block data: %v", err)
 		}
 		blockdataMessages[i] = &kgo.Record{
-			Topic: p.getTopicName("block_data"),
+			Topic: "block_data",
 			Key:   []byte(fmt.Sprintf("block-%s-%s-%s", status, data.Block.ChainId.String(), data.Block.Hash)),
 			Value: msgJson,
 		}
@@ -198,143 +149,30 @@ func (p *Publisher) publishBlockData(blockData []common.BlockData, isReorg bool)
 	return nil
 }
 
-func (p *Publisher) createBlockMessage(block common.Block, status string) (*kgo.Record, error) {
-	msg := PublishableMessage[common.BlockModel]{
-		Data:   block.Serialize(),
-		Status: status,
-	}
-	msgJson, err := json.Marshal(msg)
-	if err != nil {
-		return nil, fmt.Errorf("failed to marshal block data: %v", err)
-	}
-	return &kgo.Record{
-		Topic: p.getTopicName("blocks"),
-		Key:   []byte(fmt.Sprintf("block-%s-%s-%s", status, block.ChainId.String(), block.Hash)),
-		Value: msgJson,
-	}, nil
-}
-
-func (p *Publisher) createTransactionMessage(tx common.Transaction, status string) (*kgo.Record, error) {
-	msg := PublishableMessage[common.TransactionModel]{
-		Data:   tx.Serialize(),
-		Status: status,
-	}
-	msgJson, err := json.Marshal(msg)
-	if err != nil {
-		return nil, fmt.Errorf("failed to marshal transaction data: %v", err)
-	}
-	return &kgo.Record{
-		Topic: p.getTopicName("transactions"),
-		Key:   []byte(fmt.Sprintf("transaction-%s-%s-%s", status, tx.ChainId.String(), tx.Hash)),
-		Value: msgJson,
-	}, nil
-}
-
-func (p *Publisher) createTraceMessage(trace common.Trace, status string) (*kgo.Record, error) {
-	msg := PublishableMessage[common.TraceModel]{
-		Data:   trace.Serialize(),
-		Status: status,
-	}
-	msgJson, err := json.Marshal(msg)
-	if err != nil {
-		return nil, fmt.Errorf("failed to marshal trace data: %v", err)
-	}
-	traceAddressStr := make([]string, len(trace.TraceAddress))
-	for i, addr := range trace.TraceAddress {
-		traceAddressStr[i] = fmt.Sprint(addr)
-	}
-	return &kgo.Record{
-		Topic: p.getTopicName("traces"),
-		Key:   []byte(fmt.Sprintf("trace-%s-%s-%s-%v", status, trace.ChainID.String(), trace.TransactionHash, strings.Join(traceAddressStr, ","))),
-		Value: msgJson,
-	}, nil
-}
-
-func (p *Publisher) createEventMessage(event common.Log, status string) (*kgo.Record, error) {
-	msg := PublishableMessage[common.LogModel]{
-		Data:   event.Serialize(),
-		Status: status,
-	}
-	msgJson, err := json.Marshal(msg)
-	if err != nil {
-		return nil, fmt.Errorf("failed to marshal event data: %v", err)
-	}
-	return &kgo.Record{
-		Topic: p.getTopicName("events"),
-		Key:   []byte(fmt.Sprintf("event-%s-%s-%s-%d", status, event.ChainId.String(), event.TransactionHash, event.LogIndex)),
-		Value: msgJson,
-	}, nil
-}
-
-func (p *Publisher) shouldPublishEvent(event common.Log) bool {
-	if len(config.Cfg.Publisher.Events.AddressFilter) > 0 {
-		for _, addr := range config.Cfg.Publisher.Events.AddressFilter {
-			if addr == event.Address {
-				return true
-			}
-		}
-		return false
+func (p *NewKafka) publishMessages(ctx context.Context, messages []*kgo.Record) error {
+	if len(messages) == 0 {
+		return nil
 	}
 
-	if len(config.Cfg.Publisher.Events.Topic0Filter) > 0 {
-		for _, topic0 := range config.Cfg.Publisher.Events.Topic0Filter {
-			if topic0 == event.Topic0 {
-				return true
-			}
-		}
-		return false
-	}
-	return true
-}
+	p.mu.RLock()
+	defer p.mu.RUnlock()
 
-func (p *Publisher) shouldPublishTransaction(tx common.Transaction) bool {
-	if len(config.Cfg.Publisher.Transactions.ToFilter) > 0 {
-		for _, addr := range config.Cfg.Publisher.Transactions.ToFilter {
-			if addr == tx.ToAddress {
-				return true
-			}
-		}
-		return false
+	if p.client == nil {
+		return nil // Skip if no client configured
 	}
 
-	if len(config.Cfg.Publisher.Transactions.FromFilter) > 0 {
-		for _, addr := range config.Cfg.Publisher.Transactions.FromFilter {
-			if addr == tx.FromAddress {
-				return true
+	var wg sync.WaitGroup
+	wg.Add(len(messages))
+	// Publish to all configured producers
+	for _, msg := range messages {
+		p.client.Produce(ctx, msg, func(_ *kgo.Record, err error) {
+			defer wg.Done()
+			if err != nil {
+				log.Error().Err(err).Msg("Failed to publish message to Kafka")
 			}
-		}
-		return false
+		})
 	}
-	return true
-}
+	wg.Wait()
 
-func (p *Publisher) getTopicName(entity string) string {
-	chainIdSuffix := ""
-	if config.Cfg.RPC.ChainID != "" {
-		chainIdSuffix = fmt.Sprintf(".%s", config.Cfg.RPC.ChainID)
-	}
-	switch entity {
-	case "blocks":
-		if config.Cfg.Publisher.Blocks.TopicName != "" {
-			return config.Cfg.Publisher.Blocks.TopicName
-		}
-		return fmt.Sprintf("insight.blocks%s", chainIdSuffix)
-	case "transactions":
-		if config.Cfg.Publisher.Transactions.TopicName != "" {
-			return config.Cfg.Publisher.Transactions.TopicName
-		}
-		return fmt.Sprintf("insight.transactions%s", chainIdSuffix)
-	case "traces":
-		if config.Cfg.Publisher.Traces.TopicName != "" {
-			return config.Cfg.Publisher.Traces.TopicName
-		}
-		return fmt.Sprintf("insight.traces%s", chainIdSuffix)
-	case "events":
-		if config.Cfg.Publisher.Events.TopicName != "" {
-			return config.Cfg.Publisher.Events.TopicName
-		}
-		return fmt.Sprintf("insight.events%s", chainIdSuffix)
-	default:
-		return ""
-	}
+	return nil
 }