Update config

Author: jakeloo

cmd/root.go

@@ -125,7 +125,6 @@ func init() {
 	rootCmd.PersistentFlags().Int("storage-staging-postgres-maxConnLifetime", 300, "PostgreSQL max connection lifetime in seconds for staging storage")
 	rootCmd.PersistentFlags().Int("storage-staging-postgres-connectTimeout", 10, "PostgreSQL connection timeout in seconds for staging storage")
 	// Kafka storage flags - only for main storage (where blockchain data is committed)
-	rootCmd.PersistentFlags().Bool("storage-main-kafka-enabled", false, "Enable Kafka storage for main storage")
 	rootCmd.PersistentFlags().String("storage-main-kafka-brokers", "", "Kafka brokers for main storage")
 	rootCmd.PersistentFlags().String("storage-main-kafka-username", "", "Kafka username for main storage")
 	rootCmd.PersistentFlags().String("storage-main-kafka-password", "", "Kafka password for main storage")
@@ -253,8 +252,6 @@ func init() {
 	viper.BindPFlag("storage.staging.postgres.maxIdleConns", rootCmd.PersistentFlags().Lookup("storage-staging-postgres-maxIdleConns"))
 	viper.BindPFlag("storage.staging.postgres.maxConnLifetime", rootCmd.PersistentFlags().Lookup("storage-staging-postgres-maxConnLifetime"))
 	viper.BindPFlag("storage.staging.postgres.connectTimeout", rootCmd.PersistentFlags().Lookup("storage-staging-postgres-connectTimeout"))
-	// Bind Kafka storage flags - only for main storage
-	viper.BindPFlag("storage.main.kafka.enabled", rootCmd.PersistentFlags().Lookup("storage-main-kafka-enabled"))
 	viper.BindPFlag("storage.main.kafka.brokers", rootCmd.PersistentFlags().Lookup("storage-main-kafka-brokers"))
 	viper.BindPFlag("storage.main.kafka.username", rootCmd.PersistentFlags().Lookup("storage-main-kafka-username"))
 	viper.BindPFlag("storage.main.kafka.password", rootCmd.PersistentFlags().Lookup("storage-main-kafka-password"))

configs/config.go

@@ -102,7 +102,6 @@ type PostgresConfig struct {
 }
 
 type KafkaConfig struct {
-	Enabled  bool            `mapstructure:"enabled"`
 	Brokers  string          `mapstructure:"brokers"`
 	Username string          `mapstructure:"username"`
 	Password string          `mapstructure:"password"`

internal/storage/connector.go

@@ -148,12 +148,12 @@ func NewStorageConnector(cfg *config.StorageConfig) (IStorage, error) {
 func NewConnector[T any](cfg *config.StorageConnectionConfig) (T, error) {
 	var conn interface{}
 	var err error
-	if cfg.Postgres != nil {
+	if cfg.Kafka != nil {
+		conn, err = NewKafkaPostgresConnector(cfg.Kafka)
+	} else if cfg.Postgres != nil {
 		conn, err = NewPostgresConnector(cfg.Postgres)
 	} else if cfg.Clickhouse != nil {
 		conn, err = NewClickHouseConnector(cfg.Clickhouse)
-	} else if cfg.Kafka != nil {
-		conn, err = NewKafkaPostgresConnector(cfg.Kafka)
 	} else {
 		return *new(T), fmt.Errorf("no storage driver configured")
 	}

internal/storage/kafka_postgres.go

@@ -56,7 +56,7 @@ func NewKafkaPostgresConnector(cfg *config.KafkaConfig) (*KafkaPostgresConnector
 
 	// Initialize Kafka publisher if enabled
 	var kafkaPublisher *KafkaPublisher
-	if cfg.Enabled && cfg.Brokers != "" {
+	if cfg.Brokers != "" {
 		kafkaPublisher, err = NewKafkaPublisher(cfg)
 		if err != nil {
 			log.Warn().Err(err).Msg("Failed to initialize Kafka publisher, continuing without publishing")