From bfe8bcd9ad2e455466999daa9f39329d4b28eeab Mon Sep 17 00:00:00 2001
From: withchao <993506633@qq.com>
Date: Fri, 3 Jul 2026 17:16:50 +0800
Subject: [PATCH] refactor: replace Kafka topic configuration with mqbuild
 constants for improved readability

---
 cmd/main.go                  | 1 +
 internal/msgtransfer/init.go | 8 ++++----
 internal/push/push.go        | 6 +++---
 internal/rpc/msg/server.go   | 2 +-
 pkg/common/config/mq.go      | 2 +-
 5 files changed, 10 insertions(+), 9 deletions(-)

diff --git a/cmd/main.go b/cmd/main.go
index e5eefece5..7b0a7a42f 100644
--- a/cmd/main.go
+++ b/cmd/main.go
@@ -498,6 +498,7 @@ func startRedisServerRegister(ctx context.Context, cfg *serverConfig, client dis
 			log.ZWarn(ctx, "gateway unregister failed", err, "address", selfAddr)
 		}
 	}()
+	register()
 	for {
 		select {
 		case <-timer.C:
diff --git a/internal/msgtransfer/init.go b/internal/msgtransfer/init.go
index bb392ff97..2b483fcd7 100644
--- a/internal/msgtransfer/init.go
+++ b/internal/msgtransfer/init.go
@@ -73,11 +73,11 @@ func Start(ctx context.Context, config *Config, client discovery.SvcDiscoveryReg
 	if err != nil {
 		return err
 	}
-	mongoProducer, err := builder.GetTopicProducer(ctx, config.KafkaConfig.ToMongoTopic)
+	mongoProducer, err := builder.GetTopicProducer(ctx, mqbuild.TopicToMongo)
 	if err != nil {
 		return err
 	}
-	pushProducer, err := builder.GetTopicProducer(ctx, config.KafkaConfig.ToPushTopic)
+	pushProducer, err := builder.GetTopicProducer(ctx, mqbuild.TopicToPush)
 	if err != nil {
 		return err
 	}
@@ -100,11 +100,11 @@ func Start(ctx context.Context, config *Config, client discovery.SvcDiscoveryReg
 	if err != nil {
 		return err
 	}
-	historyConsumer, err := builder.GetTopicConsumer(ctx, config.KafkaConfig.ToRedisTopic)
+	historyConsumer, err := builder.GetTopicConsumer(ctx, mqbuild.TopicToRedis)
 	if err != nil {
 		return err
 	}
-	historyMongoConsumer, err := builder.GetTopicConsumer(ctx, config.KafkaConfig.ToMongoTopic)
+	historyMongoConsumer, err := builder.GetTopicConsumer(ctx, mqbuild.TopicToMongo)
 	if err != nil {
 		return err
 	}
diff --git a/internal/push/push.go b/internal/push/push.go
index 78e6a107f..d68708422 100644
--- a/internal/push/push.go
+++ b/internal/push/push.go
@@ -65,17 +65,17 @@ func Start(ctx context.Context, config *Config, client discovery.SvcDiscoveryReg
 	if err != nil {
 		return err
 	}
-	offlinePushProducer, err := builder.GetTopicProducer(ctx, config.KafkaConfig.ToOfflinePushTopic)
+	offlinePushProducer, err := builder.GetTopicProducer(ctx, mqbuild.TopicToOfflinePush)
 	if err != nil {
 		return err
 	}
 	database := controller.NewPushDatabase(cacheModel, offlinePushProducer)
 
-	pushConsumer, err := builder.GetTopicConsumer(ctx, config.KafkaConfig.ToPushTopic)
+	pushConsumer, err := builder.GetTopicConsumer(ctx, mqbuild.TopicToPush)
 	if err != nil {
 		return err
 	}
-	offlinePushConsumer, err := builder.GetTopicConsumer(ctx, config.KafkaConfig.ToOfflinePushTopic)
+	offlinePushConsumer, err := builder.GetTopicConsumer(ctx, mqbuild.TopicToOfflinePush)
 	if err != nil {
 		return err
 	}
diff --git a/internal/rpc/msg/server.go b/internal/rpc/msg/server.go
index 1d88ad933..e7c589b8c 100644
--- a/internal/rpc/msg/server.go
+++ b/internal/rpc/msg/server.go
@@ -92,7 +92,7 @@ func Start(ctx context.Context, config *Config, client discovery.SvcDiscoveryReg
 	if err != nil {
 		return err
 	}
-	redisProducer, err := builder.GetTopicProducer(ctx, config.KafkaConfig.ToRedisTopic)
+	redisProducer, err := builder.GetTopicProducer(ctx, mqbuild.TopicToRedis)
 	if err != nil {
 		return err
 	}
diff --git a/pkg/common/config/mq.go b/pkg/common/config/mq.go
index 2cba9dd38..76fdb26e6 100644
--- a/pkg/common/config/mq.go
+++ b/pkg/common/config/mq.go
@@ -14,7 +14,7 @@ const (
 
 func NormalizeQueueEngine(engine string) string {
 	switch strings.ToLower(strings.TrimSpace(engine)) {
-	case "kafka":
+	case "", "kafka":
 		return QueueEngineKafka
 	case "redis":
 		return QueueEngineRedis