feat: optimize code and support running in single process mode (#3142)

* pb * fix: Modifying other fields while setting IsPrivateChat does not take effect * fix: quote message error revoke * refactoring scheduled tasks * refactoring scheduled tasks * refactoring scheduled tasks * refactoring scheduled tasks * refactoring scheduled tasks * refactoring scheduled tasks * upgrading pkg tools * fix * fix * optimize log output * feat: support GetLastMessage * feat: support GetLastMessage * feat: s3 switch * feat: s3 switch * fix: GetUsersOnline * feat: SendBusinessNotification supported configuration parameters * feat: SendBusinessNotification supported configuration parameters * feat: SendBusinessNotification supported configuration parameters * feat: seq conversion failed without exiting * monolithic * fix: DeleteDoc crash * fix: DeleteDoc crash * fix: monolithic * fix: monolithic * fix: fill send time * fix: fill send time * fix: crash caused by withdrawing messages from users who have left the group * fix: mq * fix: mq * fix: user msg timestamp * fix: mq * 1 * 1 * 1 * 1 * 1 * 1 * 1 * seq read config * seq read config * 1 * 1 * fix: the source message of the reference is withdrawn, and the referenced message is deleted * 1 * 1 * 1 * 1 * 1 * 1 * 1 * 1 * 1 * 1 * 1 * 1 * 1 * 1
2026-05-04 09:05:59 +08:00 · 2025-02-14 16:18:27 +08:00
parent 4b3a2b7483
commit f322ddca77
100 changed files with 3044 additions and 1857 deletions
@@ -16,51 +16,35 @@ package msgtransfer

 import (
 	"context"
-	"errors"
 	"fmt"
-	"net"
-	"net/http"
-	"os"
-	"os/signal"
-	"strconv"
-	"syscall"

-	disetcd "github.com/openimsdk/open-im-server/v3/pkg/common/discovery/etcd"
-	"github.com/openimsdk/tools/discovery"
-	"github.com/openimsdk/tools/discovery/etcd"
-	"github.com/openimsdk/tools/utils/jsonutil"
-	"github.com/openimsdk/tools/utils/network"
-
-	"github.com/openimsdk/open-im-server/v3/pkg/common/prommetrics"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache/mcache"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache/redis"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/database/mgo"
-	"github.com/openimsdk/tools/db/mongoutil"
-	"github.com/openimsdk/tools/db/redisutil"
-	"github.com/openimsdk/tools/utils/datautil"
+	"github.com/openimsdk/open-im-server/v3/pkg/dbbuild"
+	"github.com/openimsdk/open-im-server/v3/pkg/mqbuild"
+	"github.com/openimsdk/tools/discovery"
+	"github.com/openimsdk/tools/mq"
 	"github.com/openimsdk/tools/utils/runtimeenv"

 	conf "github.com/openimsdk/open-im-server/v3/pkg/common/config"
-	discRegister "github.com/openimsdk/open-im-server/v3/pkg/common/discovery"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/controller"
-	"github.com/openimsdk/tools/errs"
 	"github.com/openimsdk/tools/log"
-	"github.com/openimsdk/tools/mw"
-	"github.com/openimsdk/tools/system/program"
 	"google.golang.org/grpc"
-	"google.golang.org/grpc/credentials/insecure"
 )

 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
-	ctx            context.Context
-	cancel         context.CancelFunc
-
-	runTimeEnv string
+	historyMongoHandler *OnlineHistoryMongoConsumerHandler
+	ctx                 context.Context
+	//cancel              context.CancelFunc
 }

 type Config struct {
@@ -71,48 +55,59 @@ type Config struct {
 	Share          conf.Share
 	WebhooksConfig conf.Webhooks
 	Discovery      conf.Discovery
+	Index          conf.Index
 }

-func Start(ctx context.Context, index int, config *Config) error {
-	runTimeEnv := runtimeenv.RuntimeEnvironment()
+func Start(ctx context.Context, config *Config, client discovery.Conn, server grpc.ServiceRegistrar) error {
+	builder := mqbuild.NewBuilder(&config.KafkaConfig)

-	log.CInfo(ctx, "MSG-TRANSFER server is initializing", "runTimeEnv", runTimeEnv, "prometheusPorts",
-		config.MsgTransfer.Prometheus.Ports, "index", index)
-
-	mgocli, err := mongoutil.NewMongoDB(ctx, config.MongodbConfig.Build())
+	log.CInfo(ctx, "MSG-TRANSFER server is initializing", "runTimeEnv", runtimeenv.RuntimeEnvironment(), "prometheusPorts",
+		config.MsgTransfer.Prometheus.Ports, "index", config.Index)
+	dbb := dbbuild.NewBuilder(&config.MongodbConfig, &config.RedisConfig)
+	mgocli, err := dbb.Mongo(ctx)
 	if err != nil {
 		return err
 	}
-	rdb, err := redisutil.NewRedisClient(ctx, config.RedisConfig.Build())
+	rdb, err := dbb.Redis(ctx)
 	if err != nil {
 		return err
 	}
-	client, err := discRegister.NewDiscoveryRegister(&config.Discovery, runTimeEnv, nil)
+
+	//if config.Discovery.Enable == conf.ETCD {
+	//	cm := disetcd.NewConfigManager(client.(*etcd.SvcDiscoveryRegistryImpl).GetClient(), []string{
+	//		config.MsgTransfer.GetConfigFileName(),
+	//		config.RedisConfig.GetConfigFileName(),
+	//		config.MongodbConfig.GetConfigFileName(),
+	//		config.KafkaConfig.GetConfigFileName(),
+	//		config.Share.GetConfigFileName(),
+	//		config.WebhooksConfig.GetConfigFileName(),
+	//		config.Discovery.GetConfigFileName(),
+	//		conf.LogConfigFileName,
+	//	})
+	//	cm.Watch(ctx)
+	//}
+	mongoProducer, err := builder.GetTopicProducer(ctx, config.KafkaConfig.ToMongoTopic)
 	if err != nil {
 		return err
 	}
-	client.AddOption(mw.GrpcClient(), grpc.WithTransportCredentials(insecure.NewCredentials()),
-		grpc.WithDefaultServiceConfig(fmt.Sprintf(`{"LoadBalancingPolicy": "%s"}`, "round_robin")))
-
-	if config.Discovery.Enable == conf.ETCD {
-		cm := disetcd.NewConfigManager(client.(*etcd.SvcDiscoveryRegistryImpl).GetClient(), []string{
-			config.MsgTransfer.GetConfigFileName(),
-			config.RedisConfig.GetConfigFileName(),
-			config.MongodbConfig.GetConfigFileName(),
-			config.KafkaConfig.GetConfigFileName(),
-			config.Share.GetConfigFileName(),
-			config.WebhooksConfig.GetConfigFileName(),
-			config.Discovery.GetConfigFileName(),
-			conf.LogConfigFileName,
-		})
-		cm.Watch(ctx)
+	pushProducer, err := builder.GetTopicProducer(ctx, config.KafkaConfig.ToPushTopic)
+	if err != nil {
+		return err
 	}
-
 	msgDocModel, err := mgo.NewMsgMongo(mgocli.GetDB())
 	if err != nil {
 		return err
 	}
-	msgModel := redis.NewMsgCache(rdb, msgDocModel)
+	var msgModel cache.MsgCache
+	if rdb == nil {
+		cm, err := mgo.NewCacheMgo(mgocli.GetDB())
+		if err != nil {
+			return err
+		}
+		msgModel = mcache.NewMsgCache(cm, msgDocModel)
+	} else {
+		msgModel = redis.NewMsgCache(rdb, msgDocModel)
+	}
 	seqConversation, err := mgo.NewSeqConversationMongo(mgocli.GetDB())
 	if err != nil {
 		return err
@@ -123,124 +118,68 @@ func Start(ctx context.Context, index int, config *Config) error {
 		return err
 	}
 	seqUserCache := redis.NewSeqUserCacheRedis(rdb, seqUser)
-	msgTransferDatabase, err := controller.NewMsgTransferDatabase(msgDocModel, msgModel, seqUserCache, seqConversationCache, &config.KafkaConfig)
+	msgTransferDatabase, err := controller.NewMsgTransferDatabase(msgDocModel, msgModel, seqUserCache, seqConversationCache, mongoProducer, pushProducer)
 	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,
-		historyMongoCH: historyMongoCH,
-		runTimeEnv:     runTimeEnv,
+		historyConsumer:      historyConsumer,
+		historyMongoConsumer: historyMongoConsumer,
+		historyHandler:       historyHandler,
+		historyMongoHandler:  historyMongoHandler,
 	}

-	return msgTransfer.Start(index, config, client)
+	return msgTransfer.Start(ctx)
 }

-func (m *MsgTransfer) Start(index int, config *Config, client discovery.SvcDiscoveryRegistry) error {
-	m.ctx, m.cancel = context.WithCancel(context.Background())
-	var (
-		netDone = make(chan struct{}, 1)
-		netErr  error
-	)
+func (m *MsgTransfer) Start(ctx context.Context) error {
+	var cancel context.CancelCauseFunc
+	m.ctx, cancel = context.WithCancelCause(ctx)

-	go m.historyCH.historyConsumerGroup.RegisterHandleAndConsumer(m.ctx, m.historyCH)
-	go m.historyMongoCH.historyConsumerGroup.RegisterHandleAndConsumer(m.ctx, m.historyMongoCH)
-	go m.historyCH.HandleUserHasReadSeqMessages(m.ctx)
-	err := m.historyCH.redisMessageBatches.Start()
+	go func() {
+		for {
+			if err := m.historyConsumer.Subscribe(m.ctx, m.historyHandler.HandlerRedisMessage); err != nil {
+				cancel(fmt.Errorf("history consumer %w", err))
+				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 {
+				cancel(fmt.Errorf("history mongo consumer %w", err))
+				log.ZError(m.ctx, "historyMongoConsumer err", err)
+				return
+			}
+		}
+	}()
+
+	go m.historyHandler.HandleUserHasReadSeqMessages(m.ctx)
+
+	err := m.historyHandler.redisMessageBatches.Start()
 	if err != nil {
 		return err
 	}
-
-	registerIP, err := network.GetRpcRegisterIP("")
-	if err != nil {
-		return err
-	}
-
-	getAutoPort := func() (net.Listener, int, error) {
-		registerAddr := net.JoinHostPort(registerIP, "0")
-		listener, err := net.Listen("tcp", registerAddr)
-		if err != nil {
-			return nil, 0, errs.WrapMsg(err, "listen err", "registerAddr", registerAddr)
-		}
-		_, portStr, _ := net.SplitHostPort(listener.Addr().String())
-		port, _ := strconv.Atoi(portStr)
-		return listener, port, nil
-	}
-
-	if config.MsgTransfer.Prometheus.AutoSetPorts && config.Discovery.Enable != conf.ETCD {
-		return errs.New("only etcd support autoSetPorts", "RegisterName", "api").Wrap()
-	}
-
-	if config.MsgTransfer.Prometheus.Enable {
-		var (
-			listener       net.Listener
-			prometheusPort int
-		)
-
-		if config.MsgTransfer.Prometheus.AutoSetPorts {
-			listener, prometheusPort, err = getAutoPort()
-			if err != nil {
-				return err
-			}
-
-			etcdClient := client.(*etcd.SvcDiscoveryRegistryImpl).GetClient()
-
-			_, err = etcdClient.Put(context.TODO(), prommetrics.BuildDiscoveryKey(prommetrics.MessageTransferKeyName), jsonutil.StructToJsonString(prommetrics.BuildDefaultTarget(registerIP, prometheusPort)))
-			if err != nil {
-				return errs.WrapMsg(err, "etcd put err")
-			}
-		} else {
-			prometheusPort, err = datautil.GetElemByIndex(config.MsgTransfer.Prometheus.Ports, index)
-			if err != nil {
-				return err
-			}
-			listener, err = net.Listen("tcp", fmt.Sprintf(":%d", prometheusPort))
-			if err != nil {
-				return errs.WrapMsg(err, "listen err", "addr", fmt.Sprintf(":%d", prometheusPort))
-			}
-		}
-
-		go func() {
-			defer func() {
-				if r := recover(); r != nil {
-					log.ZPanic(m.ctx, "MsgTransfer Start Panic", errs.ErrPanic(r))
-				}
-			}()
-			if err := prommetrics.TransferInit(listener); err != nil && !errors.Is(err, http.ErrServerClosed) {
-				netErr = errs.WrapMsg(err, "prometheus start error", "prometheusPort", prometheusPort)
-				netDone <- struct{}{}
-			}
-		}()
-	}
-
-	sigs := make(chan os.Signal, 1)
-	signal.Notify(sigs, syscall.SIGTERM)
-	select {
-	case <-sigs:
-		program.SIGTERMExit()
-		// graceful close kafka client.
-		m.cancel()
-		m.historyCH.redisMessageBatches.Close()
-		m.historyCH.Close()
-		m.historyCH.historyConsumerGroup.Close()
-		m.historyMongoCH.historyConsumerGroup.Close()
-		return nil
-	case <-netDone:
-		m.cancel()
-		m.historyCH.redisMessageBatches.Close()
-		m.historyCH.Close()
-		m.historyCH.historyConsumerGroup.Close()
-		m.historyMongoCH.historyConsumerGroup.Close()
-		close(netDone)
-		return netErr
-	}
+	<-m.ctx.Done()
+	return context.Cause(m.ctx)
 }