fix: mq

cmd/main.go

@@ -137,11 +137,25 @@ func (x *cmds) parseConf(conf any) error {
 		pkt := getTypePath(field.Type())
 		confData, ok := x.conf[pkt]
 		if !ok {
-			if typeField.Name == "FcmConfigPath" && field.Kind() == reflect.String {
+			switch field.Interface().(type) {
+			case config.Path:
 				field.SetString(x.confPath)
-				continue
+			case config.AllConfig:
+				var allConf config.AllConfig
+				if err := x.parseConf(&allConf); err != nil {
+					return err
+				}
+				field.Set(reflect.ValueOf(allConf))
+			case *config.AllConfig:
+				var allConf config.AllConfig
+				if err := x.parseConf(&allConf); err != nil {
+					return err
+				}
+				field.Set(reflect.ValueOf(&allConf))
+			default:
+				return fmt.Errorf("config field %s %s not found", vof.Type().Name(), typeField.Name)
 			}
-			return fmt.Errorf("config field %s %s not found", vof.Type().Name(), typeField.Name)
+			continue
 		}
 		if confData == nil {
 			continue
@@ -172,15 +186,24 @@ func (x *cmds) run(ctx context.Context) error {
 			return err
 		}
 	}
-	for _, cmd := range x.cmds {
+	if len(x.cmds) == 0 {
-		fmt.Println("start", cmd.Name)
+		return fmt.Errorf("no command to run")
-		if err := cmd.Func(ctx); err != nil {
-			fmt.Println("start failed", cmd.Name, err)
-			return err
-		}
-		fmt.Println("start ok", cmd.Name)
 	}
-	return nil
+	ctx, cancel := context.WithCancelCause(ctx)
+	for i := range x.cmds {
+		cmd := x.cmds[i]
+		go func() {
+			fmt.Println("start", cmd.Name)
+			if err := cmd.Func(ctx); err != nil {
+				fmt.Println("start failed", cmd.Name, err)
+				cancel(err)
+				return
+			}
+			fmt.Println("start end", cmd.Name)
+		}()
+	}
+	<-ctx.Done()
+	return context.Cause(ctx)
 }
 
 type cmdName struct {

internal/api/config_manager.go

@@ -30,16 +30,14 @@ type ConfigManager struct {
 	client        *clientv3.Client
 
 	configPath string
-	runtimeEnv string
 }
 
-func NewConfigManager(IMAdminUserID []string, cfg *config.AllConfig, client *clientv3.Client, configPath string, runtimeEnv string) *ConfigManager {
+func NewConfigManager(IMAdminUserID []string, cfg *config.AllConfig, client *clientv3.Client, configPath string) *ConfigManager {
 	cm := &ConfigManager{
 		imAdminUserID: IMAdminUserID,
 		config:        cfg,
 		client:        client,
 		configPath:    configPath,
-		runtimeEnv:    runtimeEnv,
 	}
 	return cm
 }
@@ -73,7 +71,7 @@ func (cm *ConfigManager) GetConfig(c *gin.Context) {
 func (cm *ConfigManager) GetConfigList(c *gin.Context) {
 	var resp apistruct.GetConfigListResp
 	resp.ConfigNames = cm.config.GetConfigNames()
-	resp.Environment = runtimeenv.PrintRuntimeEnvironment()
+	resp.Environment = runtimeenv.RuntimeEnvironment()
 	resp.Version = version.Version
 
 	apiresp.GinSuccess(c, resp)
@@ -209,13 +207,7 @@ func (cm *ConfigManager) resetConfig(c *gin.Context, checkChange bool, ops ...cl
 
 	changedKeys := make([]string, 0, len(configMap))
 	for k, v := range configMap {
-		err := config.Load(
+		err := config.Load(cm.configPath, k, config.EnvPrefixMap[k], v.new)
-			cm.configPath,
-			k,
-			config.EnvPrefixMap[k],
-			cm.runtimeEnv,
-			v.new,
-		)
 		if err != nil {
 			log.ZError(c, "load config failed", err)
 			continue

internal/api/init.go

@@ -44,10 +44,9 @@ import (
 )
 
 type Config struct {
-	*conf.AllConfig
+	conf.AllConfig
 
-	RuntimeEnv string
+	ConfigPath conf.Path
-	ConfigPath string
 }
 
 func Start(ctx context.Context, index int, config *Config) error {
@@ -56,9 +55,7 @@ func Start(ctx context.Context, index int, config *Config) error {
 		return err
 	}
 
-	config.RuntimeEnv = runtimeenv.PrintRuntimeEnvironment()
+	client, err := kdisc.NewDiscoveryRegister(&config.Discovery, []string{
-
-	client, err := kdisc.NewDiscoveryRegister(&config.Discovery, config.RuntimeEnv, []string{
 		config.Discovery.RpcService.MessageGateway,
 	})
 	if err != nil {
@@ -135,7 +132,7 @@ func Start(ctx context.Context, index int, config *Config) error {
 	address := net.JoinHostPort(network.GetListenIP(config.API.Api.ListenIP), strconv.Itoa(apiPort))
 
 	server := http.Server{Addr: address, Handler: router}
-	log.CInfo(ctx, "API server is initializing", "runtimeEnv", config.RuntimeEnv, "address", address, "apiPort", apiPort, "prometheusPort", prometheusPort)
+	log.CInfo(ctx, "API server is initializing", "runtimeEnv", runtimeenv.RuntimeEnvironment(), "address", address, "apiPort", apiPort, "prometheusPort", prometheusPort)
 	go func() {
 		err = server.ListenAndServe()
 		if err != nil && !errors.Is(err, http.ErrServerClosed) {

internal/api/router.go

@@ -301,7 +301,7 @@ func newGinRouter(ctx context.Context, client discovery.SvcDiscoveryRegistry, cf
 	if cfg.Discovery.Enable == config.ETCD {
 		etcdClient = client.(*etcd.SvcDiscoveryRegistryImpl).GetClient()
 	}
-	cm := NewConfigManager(cfg.Share.IMAdminUserID, cfg.AllConfig, etcdClient, cfg.ConfigPath, cfg.RuntimeEnv)
+	cm := NewConfigManager(cfg.Share.IMAdminUserID, &cfg.AllConfig, etcdClient, string(cfg.ConfigPath))
 	{
 
 		configGroup := r.Group("/config", cm.CheckAdmin)

internal/msggateway/init.go

@@ -33,15 +33,11 @@ type Config struct {
 	RedisConfig    config.Redis
 	WebhooksConfig config.Webhooks
 	Discovery      config.Discovery
-
-	runtimeEnv string
 }
 
 // Start run ws server.
 func Start(ctx context.Context, index int, conf *Config) error {
-	conf.runtimeEnv = runtimeenv.PrintRuntimeEnvironment()
+	log.CInfo(ctx, "MSG-GATEWAY server is initializing", "runtimeEnv", runtimeenv.RuntimeEnvironment(),
-
-	log.CInfo(ctx, "MSG-GATEWAY server is initializing", "runtimeEnv", conf.runtimeEnv,
 		"rpcPorts", conf.MsgGateway.RPC.Ports,
 		"wsPort", conf.MsgGateway.LongConnSvr.Ports, "prometheusPorts", conf.MsgGateway.Prometheus.Ports)
 	wsPort, err := datautil.GetElemByIndex(conf.MsgGateway.LongConnSvr.Ports, index)

internal/msgtransfer/init.go

@@ -29,6 +29,7 @@ import (
 	"github.com/openimsdk/open-im-server/v3/pkg/mqbuild"
 	"github.com/openimsdk/tools/discovery"
 	"github.com/openimsdk/tools/discovery/etcd"
+	"github.com/openimsdk/tools/mq"
 	"github.com/openimsdk/tools/utils/jsonutil"
 	"github.com/openimsdk/tools/utils/network"
 
@@ -52,16 +53,16 @@ import (
 )
 
 type MsgTransfer struct {
+	historyConsumer      mq.Consumer
+	historyMongoConsumer mq.Consumer
 	// This consumer aggregated messages, subscribed to the topic:toRedis,
 	//  the message is stored in redis, Incr Redis, and then the message is sent to toPush topic for push,
 	// and the message is sent to toMongo topic for persistence
-	historyCH *OnlineHistoryRedisConsumerHandler
+	historyHandler *OnlineHistoryRedisConsumerHandler
 	//This consumer handle message to mongo
-	historyMongoCH *OnlineHistoryMongoConsumerHandler
+	historyMongoHandler *OnlineHistoryMongoConsumerHandler
-	ctx            context.Context
+	ctx                 context.Context
-	cancel         context.CancelFunc
+	cancel              context.CancelFunc
-
-	runTimeEnv string
 }
 
 type Config struct {
@@ -75,11 +76,10 @@ type Config struct {
 }
 
 func Start(ctx context.Context, index int, config *Config) error {
-	runTimeEnv := runtimeenv.PrintRuntimeEnvironment()
 
 	builder := mqbuild.NewBuilder(&config.Discovery, &config.KafkaConfig)
 
-	log.CInfo(ctx, "MSG-TRANSFER server is initializing", "runTimeEnv", runTimeEnv, "prometheusPorts",
+	log.CInfo(ctx, "MSG-TRANSFER server is initializing", "runTimeEnv", runtimeenv.RuntimeEnvironment(), "prometheusPorts",
 		config.MsgTransfer.Prometheus.Ports, "index", index)
 
 	mgocli, err := mongoutil.NewMongoDB(ctx, config.MongodbConfig.Build())
@@ -90,7 +90,7 @@ func Start(ctx context.Context, index int, config *Config) error {
 	if err != nil {
 		return err
 	}
-	client, err := discRegister.NewDiscoveryRegister(&config.Discovery, runTimeEnv, nil)
+	client, err := discRegister.NewDiscoveryRegister(&config.Discovery, nil)
 	if err != nil {
 		return err
 	}
@@ -137,19 +137,25 @@ func Start(ctx context.Context, index int, config *Config) error {
 	if err != nil {
 		return err
 	}
-	historyCH, err := NewOnlineHistoryRedisConsumerHandler(ctx, client, config, msgTransferDatabase)
+	historyConsumer, err := builder.GetTopicConsumer(ctx, config.KafkaConfig.ToRedisTopic)
 	if err != nil {
 		return err
 	}
-	historyMongoCH, err := NewOnlineHistoryMongoConsumerHandler(&config.KafkaConfig, msgTransferDatabase)
+	historyMongoConsumer, err := builder.GetTopicConsumer(ctx, config.KafkaConfig.ToMongoTopic)
 	if err != nil {
 		return err
 	}
+	historyHandler, err := NewOnlineHistoryRedisConsumerHandler(ctx, client, config, msgTransferDatabase)
+	if err != nil {
+		return err
+	}
+	historyMongoHandler := NewOnlineHistoryMongoConsumerHandler(msgTransferDatabase)
 
 	msgTransfer := &MsgTransfer{
-		historyCH:      historyCH,
+		historyConsumer:      historyConsumer,
-		historyMongoCH: historyMongoCH,
+		historyMongoConsumer: historyMongoConsumer,
-		runTimeEnv:     runTimeEnv,
+		historyHandler:       historyHandler,
+		historyMongoHandler:  historyMongoHandler,
 	}
 
 	return msgTransfer.Start(index, config, client)
@@ -162,10 +168,30 @@ func (m *MsgTransfer) Start(index int, config *Config, client discovery.SvcDisco
 		netErr  error
 	)
 
-	go m.historyCH.historyConsumerGroup.RegisterHandleAndConsumer(m.ctx, m.historyCH)
+	go func() {
-	go m.historyMongoCH.historyConsumerGroup.RegisterHandleAndConsumer(m.ctx, m.historyMongoCH)
+		for {
-	go m.historyCH.HandleUserHasReadSeqMessages(m.ctx)
+			if err := m.historyConsumer.Subscribe(m.ctx, m.historyHandler.HandlerRedisMessage); err != nil {
-	err := m.historyCH.redisMessageBatches.Start()
+				log.ZError(m.ctx, "historyConsumer err", err)
+				return
+			}
+		}
+	}()
+
+	go func() {
+		fn := func(ctx context.Context, key string, value []byte) error {
+			m.historyMongoHandler.HandleChatWs2Mongo(ctx, key, value)
+			return nil
+		}
+		for {
+			if err := m.historyMongoConsumer.Subscribe(m.ctx, fn); err != nil {
+				log.ZError(m.ctx, "historyMongoConsumer err", err)
+				return
+			}
+		}
+	}()
+
+	go m.historyHandler.HandleUserHasReadSeqMessages(m.ctx)
+	err := m.historyHandler.redisMessageBatches.Start()
 	if err != nil {
 		return err
 	}
@@ -237,18 +263,18 @@ func (m *MsgTransfer) Start(index int, config *Config, client discovery.SvcDisco
 	case <-sigs:
 		program.SIGTERMExit()
 		// graceful close kafka client.
+		_ = m.historyConsumer.Close()
+		_ = m.historyMongoConsumer.Close()
 		m.cancel()
-		m.historyCH.redisMessageBatches.Close()
+		m.historyHandler.redisMessageBatches.Close()
-		m.historyCH.Close()
+		m.historyHandler.Close()
-		m.historyCH.historyConsumerGroup.Close()
-		m.historyMongoCH.historyConsumerGroup.Close()
 		return nil
 	case <-netDone:
+		_ = m.historyConsumer.Close()
+		_ = m.historyMongoConsumer.Close()
 		m.cancel()
-		m.historyCH.redisMessageBatches.Close()
+		m.historyHandler.redisMessageBatches.Close()
-		m.historyCH.Close()
+		m.historyHandler.Close()
-		m.historyCH.historyConsumerGroup.Close()
-		m.historyMongoCH.historyConsumerGroup.Close()
 		close(netDone)
 		return netErr
 	}

internal/msgtransfer/online_history_msg_handler.go

@@ -19,13 +19,12 @@ import (
 	"encoding/json"
 	"errors"
 
-	"github.com/openimsdk/open-im-server/v3/pkg/rpcli"
-	"github.com/openimsdk/tools/discovery"
-	"github.com/openimsdk/tools/mq"
-
 	"sync"
 	"time"
 
+	"github.com/openimsdk/open-im-server/v3/pkg/rpcli"
+	"github.com/openimsdk/tools/discovery"
+
 	"github.com/go-redis/redis"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/prommetrics"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/controller"
@@ -79,7 +78,7 @@ type ConsumerMessage struct {
 	Value []byte
 }
 
-func NewOnlineHistoryRedisConsumerHandler(ctx context.Context, client discovery.SvcDiscoveryRegistry, config *Config, database controller.MsgTransferDatabase, historyConsumer mq.Consumer) (*OnlineHistoryRedisConsumerHandler, error) {
+func NewOnlineHistoryRedisConsumerHandler(ctx context.Context, client discovery.SvcDiscoveryRegistry, config *Config, database controller.MsgTransferDatabase) (*OnlineHistoryRedisConsumerHandler, error) {
 	groupConn, err := client.GetConn(ctx, config.Discovery.RpcService.Group)
 	if err != nil {
 		return nil, err

internal/msgtransfer/online_msg_to_mongo_handler.go

@@ -17,36 +17,24 @@ package msgtransfer
 import (
 	"context"
 
-	"github.com/IBM/sarama"
-	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/prommetrics"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/controller"
 	pbmsg "github.com/openimsdk/protocol/msg"
 	"github.com/openimsdk/tools/log"
-	"github.com/openimsdk/tools/mq/kafka"
 	"google.golang.org/protobuf/proto"
 )
 
 type OnlineHistoryMongoConsumerHandler struct {
-	historyConsumerGroup *kafka.MConsumerGroup
+	msgTransferDatabase controller.MsgTransferDatabase
-	msgTransferDatabase  controller.MsgTransferDatabase
 }
 
-func NewOnlineHistoryMongoConsumerHandler(kafkaConf *config.Kafka, database controller.MsgTransferDatabase) (*OnlineHistoryMongoConsumerHandler, error) {
+func NewOnlineHistoryMongoConsumerHandler(database controller.MsgTransferDatabase) *OnlineHistoryMongoConsumerHandler {
-	historyConsumerGroup, err := kafka.NewMConsumerGroup(kafkaConf.Build(), kafkaConf.ToMongoGroupID, []string{kafkaConf.ToMongoTopic}, true)
+	return &OnlineHistoryMongoConsumerHandler{
-	if err != nil {
+		msgTransferDatabase: database,
-		return nil, err
 	}
-
-	mc := &OnlineHistoryMongoConsumerHandler{
-		historyConsumerGroup: historyConsumerGroup,
-		msgTransferDatabase:  database,
-	}
-	return mc, nil
 }
 
-func (mc *OnlineHistoryMongoConsumerHandler) handleChatWs2Mongo(ctx context.Context, cMsg *sarama.ConsumerMessage, key string, session sarama.ConsumerGroupSession) {
+func (mc *OnlineHistoryMongoConsumerHandler) HandleChatWs2Mongo(ctx context.Context, key string, msg []byte) {
-	msg := cMsg.Value
 	msgFromMQ := pbmsg.MsgDataToMongoByMQ{}
 	err := proto.Unmarshal(msg, &msgFromMQ)
 	if err != nil {
@@ -54,7 +42,7 @@ func (mc *OnlineHistoryMongoConsumerHandler) handleChatWs2Mongo(ctx context.Cont
 		return
 	}
 	if len(msgFromMQ.MsgData) == 0 {
-		log.ZError(ctx, "msgFromMQ.MsgData is empty", nil, "cMsg", cMsg)
+		log.ZError(ctx, "msgFromMQ.MsgData is empty", nil, "key", key, "msg", msg)
 		return
 	}
 	log.ZDebug(ctx, "mongo consumer recv msg", "msgs", msgFromMQ.String())
@@ -78,22 +66,3 @@ func (mc *OnlineHistoryMongoConsumerHandler) handleChatWs2Mongo(ctx context.Cont
 		seqs = append(seqs, msg.Seq)
 	}
 }
-
-func (*OnlineHistoryMongoConsumerHandler) Setup(_ sarama.ConsumerGroupSession) error { return nil }
-
-func (*OnlineHistoryMongoConsumerHandler) Cleanup(_ sarama.ConsumerGroupSession) error { return nil }
-
-func (mc *OnlineHistoryMongoConsumerHandler) ConsumeClaim(sess sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error { // an instance in the consumer group
-	log.ZDebug(context.Background(), "online new session msg come", "highWaterMarkOffset",
-		claim.HighWaterMarkOffset(), "topic", claim.Topic(), "partition", claim.Partition())
-	for msg := range claim.Messages() {
-		ctx := mc.historyConsumerGroup.GetContextFromMsg(msg)
-		if len(msg.Value) != 0 {
-			mc.handleChatWs2Mongo(ctx, msg, string(msg.Key), sess)
-		} else {
-			log.ZError(ctx, "mongo msg get from kafka but is nil", nil, "conversationID", msg.Key)
-		}
-		sess.MarkMessage(msg, "")
-	}
-	return nil
-}

internal/push/onlinepusher.go

@@ -11,6 +11,7 @@ import (
 	"github.com/openimsdk/tools/errs"
 	"github.com/openimsdk/tools/log"
 	"github.com/openimsdk/tools/utils/datautil"
+	"github.com/openimsdk/tools/utils/runtimeenv"
 	"golang.org/x/sync/errgroup"
 	"google.golang.org/grpc"
 
@@ -40,14 +41,14 @@ func (u emptyOnlinePusher) GetOnlinePushFailedUserIDs(ctx context.Context, msg *
 }
 
 func NewOnlinePusher(disCov discovery.Conn, config *Config) OnlinePusher {
-	if config.runTimeEnv == conf.KUBERNETES {
+	if runtimeenv.RuntimeEnvironment() == conf.KUBERNETES {
 		return NewDefaultAllNode(disCov, config)
 	}
 	switch config.Discovery.Enable {
 	case conf.ETCD:
 		return NewDefaultAllNode(disCov, config)
 	default:
-		log.ZError(context.Background(), "NewOnlinePusher is error", errs.Wrap(errors.New("unsupported discovery type")), "type", config.Discovery.Enable)
+		log.ZWarn(context.Background(), "NewOnlinePusher is error", errs.Wrap(errors.New("unsupported discovery type")), "type", config.Discovery.Enable)
 		return nil
 	}
 }

internal/push/push.go

@@ -15,7 +15,6 @@ import (
 	"github.com/openimsdk/tools/discovery"
 	"github.com/openimsdk/tools/log"
 	"github.com/openimsdk/tools/mcontext"
-	"github.com/openimsdk/tools/utils/runtimeenv"
 	"google.golang.org/grpc"
 )
 
@@ -35,9 +34,7 @@ type Config struct {
 	WebhooksConfig     config.Webhooks
 	LocalCacheConfig   config.LocalCache
 	Discovery          config.Discovery
-	FcmConfigPath      string
+	FcmConfigPath      config.Path
-
-	runTimeEnv string
 }
 
 func (p pushServer) DelUserPushToken(ctx context.Context,
@@ -49,14 +46,12 @@ func (p pushServer) DelUserPushToken(ctx context.Context,
 }
 
 func Start(ctx context.Context, config *Config, client discovery.Conn, server grpc.ServiceRegistrar) error {
-	config.runTimeEnv = runtimeenv.PrintRuntimeEnvironment()
-
 	rdb, err := redisutil.NewRedisClient(ctx, config.RedisConfig.Build())
 	if err != nil {
 		return err
 	}
 	cacheModel := redis.NewThirdCache(rdb)
-	offlinePusher, err := offlinepush.NewOfflinePusher(&config.RpcConfig, cacheModel, config.FcmConfigPath)
+	offlinePusher, err := offlinepush.NewOfflinePusher(&config.RpcConfig, cacheModel, string(config.FcmConfigPath))
 	if err != nil {
 		return err
 	}

internal/tools/cron_test.go

@@ -24,7 +24,7 @@ func TestName(t *testing.T) {
 			Address:       []string{"localhost:12379"},
 		},
 	}
-	client, err := kdisc.NewDiscoveryRegister(conf, "source")
+	client, err := kdisc.NewDiscoveryRegister(conf, nil)
 	if err != nil {
 		panic(err)
 	}

pkg/common/config/config.go

@@ -30,6 +30,8 @@ import (
 
 const StructTagName = "yaml"
 
+type Path string
+
 type CacheConfig struct {
 	Topic         string `yaml:"topic"`
 	SlotNum       int    `yaml:"slotNum"`

pkg/common/config/constant.go

@@ -14,12 +14,14 @@
 
 package config
 
+import "github.com/openimsdk/tools/utils/runtimeenv"
+
 const ConfKey = "conf"
 
 const (
 	MountConfigFilePath = "CONFIG_PATH"
 	DeploymentType      = "DEPLOYMENT_TYPE"
-	KUBERNETES          = "kubernetes"
+	KUBERNETES          = runtimeenv.Kubernetes
 	ETCD                = "etcd"
 	Standalone          = "standalone"
 )

pkg/common/config/load_config.go

@@ -7,11 +7,12 @@ import (
 
 	"github.com/mitchellh/mapstructure"
 	"github.com/openimsdk/tools/errs"
+	"github.com/openimsdk/tools/utils/runtimeenv"
 	"github.com/spf13/viper"
 )
 
-func Load(configDirectory string, configFileName string, envPrefix string, runtimeEnv string, config any) error {
+func Load(configDirectory string, configFileName string, envPrefix string, config any) error {
-	if runtimeEnv == KUBERNETES {
+	if runtimeenv.RuntimeEnvironment() == KUBERNETES {
 		mountPath := os.Getenv(MountConfigFilePath)
 		if mountPath == "" {
 			return errs.ErrArgs.WrapMsg(MountConfigFilePath + " env is empty")

pkg/common/discovery/discoveryregister.go

@@ -20,6 +20,7 @@ import (
 	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
 	"github.com/openimsdk/tools/discovery"
 	"github.com/openimsdk/tools/discovery/standalone"
+	"github.com/openimsdk/tools/utils/runtimeenv"
 	"google.golang.org/grpc"
 
 	"github.com/openimsdk/tools/discovery/kubernetes"
@@ -29,8 +30,8 @@ import (
 )
 
 // NewDiscoveryRegister creates a new service discovery and registry client based on the provided environment type.
-func NewDiscoveryRegister(discovery *config.Discovery, runtimeEnv string, watchNames []string) (discovery.SvcDiscoveryRegistry, error) {
+func NewDiscoveryRegister(discovery *config.Discovery, watchNames []string) (discovery.SvcDiscoveryRegistry, error) {
-	if runtimeEnv == config.KUBERNETES {
+	if runtimeenv.RuntimeEnvironment() == config.KUBERNETES {
 		return kubernetes.NewKubernetesConnManager(discovery.Kubernetes.Namespace,
 			grpc.WithDefaultCallOptions(
 				grpc.MaxCallSendMsgSize(1024*1024*20),

pkg/common/startrpc/start.go

@@ -33,8 +33,6 @@ import (
 	"github.com/openimsdk/tools/utils/jsonutil"
 	"google.golang.org/grpc/status"
 
-	"github.com/openimsdk/tools/utils/runtimeenv"
-
 	kdisc "github.com/openimsdk/open-im-server/v3/pkg/common/discovery"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/prommetrics"
 	"github.com/openimsdk/tools/discovery"
@@ -74,8 +72,6 @@ func Start[T any](ctx context.Context, discovery *conf.Discovery, prometheusConf
 		return err
 	}
 
-	runTimeEnv := runtimeenv.PrintRuntimeEnvironment()
-
 	if !autoSetPorts {
 		rpcPort, err := datautil.GetElemByIndex(rpcPorts, index)
 		if err != nil {
@@ -99,7 +95,7 @@ func Start[T any](ctx context.Context, discovery *conf.Discovery, prometheusConf
 	if autoSetPorts && discovery.Enable != conf.ETCD {
 		return errs.New("only etcd support autoSetPorts", "rpcRegisterName", rpcRegisterName).Wrap()
 	}
-	client, err := kdisc.NewDiscoveryRegister(discovery, runTimeEnv, watchServiceNames)
+	client, err := kdisc.NewDiscoveryRegister(discovery, watchServiceNames)
 	if err != nil {
 		return err
 	}

pkg/common/storage/database/black.go

@@ -16,6 +16,7 @@ package database
 
 import (
 	"context"
+
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
 	"github.com/openimsdk/tools/db/pagination"
 )
@@ -29,3 +30,85 @@ type Black interface {
 	FindOwnerBlackInfos(ctx context.Context, ownerUserID string, userIDs []string) (blacks []*model.Black, err error)
 	FindBlackUserIDs(ctx context.Context, ownerUserID string) (blackUserIDs []string, err error)
 }
+
+var (
+	_ Black = (*mgoImpl)(nil)
+	_ Black = (*redisImpl)(nil)
+)
+
+type mgoImpl struct {
+}
+
+func (m *mgoImpl) Create(ctx context.Context, blacks []*model.Black) (err error) {
+	//TODO implement me
+	panic("implement me")
+}
+
+func (m *mgoImpl) Delete(ctx context.Context, blacks []*model.Black) (err error) {
+	//TODO implement me
+	panic("implement me")
+}
+
+func (m *mgoImpl) Find(ctx context.Context, blacks []*model.Black) (blackList []*model.Black, err error) {
+	//TODO implement me
+	panic("implement me")
+}
+
+func (m *mgoImpl) Take(ctx context.Context, ownerUserID, blockUserID string) (black *model.Black, err error) {
+	//TODO implement me
+	panic("implement me")
+}
+
+func (m *mgoImpl) FindOwnerBlacks(ctx context.Context, ownerUserID string, pagination pagination.Pagination) (total int64, blacks []*model.Black, err error) {
+	//TODO implement me
+	panic("implement me")
+}
+
+func (m *mgoImpl) FindOwnerBlackInfos(ctx context.Context, ownerUserID string, userIDs []string) (blacks []*model.Black, err error) {
+	//TODO implement me
+	panic("implement me")
+}
+
+func (m *mgoImpl) FindBlackUserIDs(ctx context.Context, ownerUserID string) (blackUserIDs []string, err error) {
+	//TODO implement me
+	panic("implement me")
+}
+
+type redisImpl struct {
+}
+
+func (r *redisImpl) Create(ctx context.Context, blacks []*model.Black) (err error) {
+
+	//TODO implement me
+	panic("implement me")
+}
+
+func (r *redisImpl) Delete(ctx context.Context, blacks []*model.Black) (err error) {
+	//TODO implement me
+	panic("implement me")
+}
+
+func (r *redisImpl) Find(ctx context.Context, blacks []*model.Black) (blackList []*model.Black, err error) {
+	//TODO implement me
+	panic("implement me")
+}
+
+func (r *redisImpl) Take(ctx context.Context, ownerUserID, blockUserID string) (black *model.Black, err error) {
+	//TODO implement me
+	panic("implement me")
+}
+
+func (r *redisImpl) FindOwnerBlacks(ctx context.Context, ownerUserID string, pagination pagination.Pagination) (total int64, blacks []*model.Black, err error) {
+	//TODO implement me
+	panic("implement me")
+}
+
+func (r *redisImpl) FindOwnerBlackInfos(ctx context.Context, ownerUserID string, userIDs []string) (blacks []*model.Black, err error) {
+	//TODO implement me
+	panic("implement me")
+}
+
+func (r *redisImpl) FindBlackUserIDs(ctx context.Context, ownerUserID string) (blackUserIDs []string, err error) {
+	//TODO implement me
+	panic("implement me")
+}