From 10a577c3107ac19fd9e9aaed8c82887ba65e4bdb Mon Sep 17 00:00:00 2001 From: chao <48119764+withchao@users.noreply.github.com> Date: Fri, 3 Jul 2026 18:28:06 +0800 Subject: [PATCH] feat: Supports launching multiple monolithic instances, load-balanced by nginx, requiring only MongoDB and Redis at a minimum (#3759) * fix: performance issues with Kafka caused by encapsulating the MQ interface * fix: admin token in standalone mode * fix: full id version * fix: resolve deadlock in cache eviction and improve GetBatch implementation * refactor: replace LongConn with ClientConn interface and simplify message handling * refactor: replace LongConn with ClientConn interface and simplify message handling * fix: seq use $setOnInsert for min_seq in conversation update * feat: add error code for handled friend requests and improve error handling in friend operations * refactor(msg): update regex pattern for conversationID to include a trailing colon * refactor: streamline cache initialization and enhance queue engine configuration * refactor: streamline cache initialization and enhance queue engine configuration * feat: add RegisterIP to API config and implement Redis server registration * feat(redis): implement standalone gateway registration with Redis * chore: update openimsdk/tools dependency to v0.0.50-alpha.121 * fix: change timer to ticker for improved periodic execution * refactor: replace Kafka topic configuration with mqbuild constants for improved readability * feat(redis): implement Redis-based locking mechanism for cron tasks --- cmd/main.go | 154 +++++++++++++--- config/openim-api.yml | 2 + config/share.yml | 8 + go.mod | 2 +- go.sum | 4 +- internal/api/init.go | 15 +- internal/api/router.go | 54 +++++- internal/msggateway/init.go | 42 +---- internal/msgtransfer/init.go | 44 ++--- internal/push/push.go | 36 ++-- internal/rpc/auth/auth.go | 24 +-- internal/rpc/msg/server.go | 29 ++- internal/rpc/third/third.go | 27 +-- internal/tools/cron/cron_task.go | 34 ++-- .../cron/{dist_look.go => etcd_locker.go} | 7 +- internal/tools/cron/locker.go | 14 ++ internal/tools/cron/redis_locker.go | 68 +++++++ pkg/common/cmd/cron_task.go | 4 +- pkg/common/config/config.go | 11 +- pkg/common/config/load_config.go | 27 ++- pkg/common/config/mq.go | 41 +++++ pkg/common/config/selector.go | 43 +++++ pkg/common/discovery/discoveryregister.go | 7 +- pkg/common/storage/cache/mcache/minio.go | 50 ------ pkg/common/storage/cache/mcache/msg_cache.go | 132 -------------- pkg/common/storage/cache/mcache/online.go | 82 --------- .../storage/cache/mcache/seq_conversation.go | 79 --------- pkg/common/storage/cache/mcache/third.go | 98 ----------- pkg/common/storage/cache/mcache/token.go | 166 ------------------ pkg/common/storage/cache/mcache/tools.go | 63 ------- pkg/common/storage/cache/redis/online.go | 8 +- .../storage/cache/redis/seq_conversation.go | 7 +- .../storage/cache/redis/standalone_gateway.go | 54 ++++++ .../cache/redis/standalone_gateway_test.go | 52 ++++++ pkg/mqbuild/builder.go | 143 +++++++++++++-- 35 files changed, 703 insertions(+), 928 deletions(-) rename internal/tools/cron/{dist_look.go => etcd_locker.go} (97%) create mode 100644 internal/tools/cron/locker.go create mode 100644 internal/tools/cron/redis_locker.go create mode 100644 pkg/common/config/mq.go create mode 100644 pkg/common/config/selector.go delete mode 100644 pkg/common/storage/cache/mcache/minio.go delete mode 100644 pkg/common/storage/cache/mcache/msg_cache.go delete mode 100644 pkg/common/storage/cache/mcache/online.go delete mode 100644 pkg/common/storage/cache/mcache/seq_conversation.go delete mode 100644 pkg/common/storage/cache/mcache/third.go delete mode 100644 pkg/common/storage/cache/mcache/token.go delete mode 100644 pkg/common/storage/cache/mcache/tools.go create mode 100644 pkg/common/storage/cache/redis/standalone_gateway.go create mode 100644 pkg/common/storage/cache/redis/standalone_gateway_test.go diff --git a/cmd/main.go b/cmd/main.go index 7e19f1c98..7b0a7a42f 100644 --- a/cmd/main.go +++ b/cmd/main.go @@ -3,6 +3,7 @@ package main import ( "bytes" "context" + "errors" "flag" "fmt" "net" @@ -12,12 +13,15 @@ import ( "path/filepath" "reflect" "runtime" + "strconv" "strings" "sync" "syscall" "time" - "github.com/mitchellh/mapstructure" + "github.com/spf13/viper" + "google.golang.org/grpc" + "github.com/openimsdk/open-im-server/v3/internal/api" "github.com/openimsdk/open-im-server/v3/internal/msggateway" "github.com/openimsdk/open-im-server/v3/internal/msgtransfer" @@ -30,33 +34,40 @@ import ( "github.com/openimsdk/open-im-server/v3/internal/rpc/third" "github.com/openimsdk/open-im-server/v3/internal/rpc/user" "github.com/openimsdk/open-im-server/v3/internal/tools/cron" + "github.com/openimsdk/open-im-server/v3/pkg/authverify" "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/cache/redis" + "github.com/openimsdk/open-im-server/v3/pkg/dbbuild" "github.com/openimsdk/open-im-server/v3/version" "github.com/openimsdk/tools/discovery" - "github.com/openimsdk/tools/discovery/standalone" + "github.com/openimsdk/tools/discovery/inprocess" "github.com/openimsdk/tools/log" "github.com/openimsdk/tools/system/program" "github.com/openimsdk/tools/utils/datautil" - "github.com/spf13/viper" - "google.golang.org/grpc" + "github.com/openimsdk/tools/utils/network" ) func init() { config.SetStandalone() prommetrics.RegistryAll() + inprocess.SetContextAdminFunc(authverify.IsAdmin) } func main() { - var configPath string + var ( + configPath string + index int + ) flag.StringVar(&configPath, "c", "", "config path") + flag.IntVar(&index, "i", 0, "start index") flag.Parse() if configPath == "" { _, _ = fmt.Fprintln(os.Stderr, "config path is empty") os.Exit(1) return } - cmd := newCmds(configPath) + cmd := newCmds(configPath, index) putCmd(cmd, false, auth.Start) putCmd(cmd, false, conversation.Start) putCmd(cmd, false, relation.Start) @@ -69,6 +80,7 @@ func main() { putCmd(cmd, true, msgtransfer.Start) putCmd(cmd, true, api.Start) putCmd(cmd, true, cron.Start) + putCmd(cmd, true, startRedisServerRegister) ctx := context.Background() if err := cmd.run(ctx); err != nil { _, _ = fmt.Fprintf(os.Stderr, "server exit %s", err) @@ -77,8 +89,8 @@ func main() { } } -func newCmds(confPath string) *cmds { - return &cmds{confPath: confPath} +func newCmds(confPath string, index int) *cmds { + return &cmds{confPath: confPath, index: index} } type cmdName struct { @@ -88,6 +100,7 @@ type cmdName struct { } type cmds struct { confPath string + index int cmds []cmdName config config.AllConfig conf map[string]reflect.Value @@ -116,6 +129,9 @@ func (x *cmds) initDiscovery() { func (x *cmds) initAllConfig() error { x.conf = make(map[string]reflect.Value) + if err := x.loadShareConfig(); err != nil { + return err + } vof := reflect.ValueOf(&x.config).Elem() num := vof.NumField() for i := 0; i < num; i++ { @@ -129,23 +145,17 @@ func (x *cmds) initAllConfig() error { } x.conf[x.getTypePath(field.Type())] = field val := field.Addr().Interface() - name := val.(interface{ GetConfigFileName() string }).GetConfigFileName() - confData, err := os.ReadFile(filepath.Join(x.confPath, name)) - if err != nil { - if os.IsNotExist(err) { - continue - } - return err + // Fields without a config file (e.g. Timer) are still registered above + // for parseConf distribution, but there is nothing to load for them. + fc, ok := val.(config.FileConfig) + if !ok { + continue } - v := viper.New() - v.SetConfigType("yaml") - if err := v.ReadConfig(bytes.NewReader(confData)); err != nil { - return err + name := fc.GetConfigFileName() + if name == config.ShareFileName || x.skipKafkaConfig(name) { + continue } - opt := func(conf *mapstructure.DecoderConfig) { - conf.TagName = config.StructTagName - } - if err := v.Unmarshal(val, opt); err != nil { + if err := x.loadFileConfig(name, val); err != nil { return err } } @@ -156,6 +166,31 @@ func (x *cmds) initAllConfig() error { return nil } +func (x *cmds) loadShareConfig() error { + return x.loadFileConfig(config.ShareFileName, &x.config.Share) +} + +func (x *cmds) skipKafkaConfig(name string) bool { + return name == config.KafkaConfigFileName && + config.NormalizeQueueEngine(x.config.Share.Queue.Engine) != config.QueueEngineKafka +} + +func (x *cmds) loadFileConfig(name string, val any) error { + confData, err := os.ReadFile(filepath.Join(x.confPath, name)) + if err != nil { + if os.IsNotExist(err) { + return nil + } + return err + } + v := viper.New() + v.SetConfigType(config.StructTagName) + if err := v.ReadConfig(bytes.NewReader(confData)); err != nil { + return err + } + return v.Unmarshal(val, config.ApplyDecoderConfig) +} + func (x *cmds) parseConf(conf any) error { vof := reflect.ValueOf(conf) for { @@ -178,6 +213,7 @@ func (x *cmds) parseConf(conf any) error { if !ok { switch field.Interface().(type) { case config.Index: + field.Set(reflect.ValueOf(config.Index(x.index))) case config.Path: field.SetString(x.confPath) case config.AllConfig: @@ -201,7 +237,7 @@ func (x *cmds) add(name string, block bool, fn func(ctx context.Context) error) func (x *cmds) initLog() error { conf := x.config.Log if err := log.InitLoggerFromConfig( - "openim-server", + "openim-service-log", program.GetProcessName(), "", "", conf.RemainLogLevel, @@ -339,7 +375,7 @@ func putCmd[C any](cmd *cmds, block bool, fn func(ctx context.Context, config *C if err := cmd.parseConf(&conf); err != nil { return err } - return fn(ctx, &conf, standalone.GetSvcDiscoveryRegistry(), standalone.GetServiceRegistrar()) + return fn(ctx, &conf, inprocess.GetSvcDiscoveryRegistry(), inprocess.GetServiceRegistrar()) }) } @@ -404,3 +440,71 @@ func (x *cmdManger) Running() []string { } return names } + +type serverConfig struct { + API config.API + Share config.Share + RedisConfig config.Redis + Index config.Index +} + +func startRedisServerRegister(ctx context.Context, cfg *serverConfig, client discovery.SvcDiscoveryRegistry, service grpc.ServiceRegistrar) error { + apiPort, err := datautil.GetElemByIndex(cfg.API.Api.Ports, int(cfg.Index)) + if err != nil { + return err + } + if apiPort <= 0 { + return errors.New("standalone api port is 0") + } + registerIP, err := network.GetRpcRegisterIP(cfg.API.Api.RegisterIP) + if err != nil { + return err + } + const validTime = time.Second * 10 + dbb := dbbuild.NewBuilder(nil, &cfg.RedisConfig) + rdb, err := dbb.Redis(ctx) + if err != nil { + return err + } + gateway := redis.NewStandaloneGatewayRedis(rdb, validTime) + selfAddr := net.JoinHostPort(registerIP, strconv.Itoa(apiPort)) + inprocess.SetLocalTarget(selfAddr) + inprocess.SetBroadcastAddress(cfg.Share.Secret, func(ctx context.Context) ([]string, error) { + address, err := gateway.GetGatewayAddrs(ctx) + if err != nil { + return nil, err + } + notSelf := make([]string, 0, len(address)) + for _, addr := range address { + if addr != selfAddr { + notSelf = append(notSelf, addr) + } + } + return notSelf, nil + }) + register := func() { + ctx, cancel := context.WithTimeout(ctx, validTime/2) + defer cancel() + if err := gateway.RegisterGateway(ctx, selfAddr); err != nil { + log.ZWarn(ctx, "gateway register failed", err, "address", selfAddr) + } + } + timer := time.NewTicker(validTime / 2) + defer func() { + timer.Stop() + ctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), time.Second) + defer cancel() + if err := gateway.UnregisterGateway(ctx, selfAddr); err != nil { + log.ZWarn(ctx, "gateway unregister failed", err, "address", selfAddr) + } + }() + register() + for { + select { + case <-timer.C: + register() + case <-ctx.Done(): + return context.Cause(ctx) + } + } +} diff --git a/config/openim-api.yml b/config/openim-api.yml index 89be50123..a879fcb1e 100644 --- a/config/openim-api.yml +++ b/config/openim-api.yml @@ -3,6 +3,8 @@ api: listenIP: 0.0.0.0 # Listening ports; if multiple are configured, multiple instances will be launched, must be consistent with the number of prometheus.ports ports: [ 10002 ] + # Fallback for IP resolution issues in multi-instance standalone mode. + # registerIP: # API compression level; 0: default compression, 1: best compression, 2: best speed, -1: no compression compressionLevel: 0 diff --git a/config/share.yml b/config/share.yml index a42bdcdd7..df0b55be2 100644 --- a/config/share.yml +++ b/config/share.yml @@ -9,6 +9,14 @@ imAdminUser: # Each entry here corresponds by index to the matching entry in the userIDs list above. nicknames: [superAdmin] +# queue: choose message queue engine +# Supported values: +# - kafka (default) +# - redis +# - memory (standalone only; microservices cannot use memory) +queue: kafka + + # 1: For Android, iOS, Windows, Mac, and web platforms, only one instance can be online at a time multiLogin: policy: 1 diff --git a/go.mod b/go.mod index f00a6ee40..8a8dae98f 100644 --- a/go.mod +++ b/go.mod @@ -13,7 +13,7 @@ require ( github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0 github.com/mitchellh/mapstructure v1.5.0 github.com/openimsdk/protocol v0.0.73-alpha.19 - github.com/openimsdk/tools v0.0.50-alpha.117 + github.com/openimsdk/tools v0.0.50-alpha.121 github.com/pkg/errors v0.9.1 // indirect github.com/prometheus/client_golang v1.18.0 github.com/stretchr/testify v1.11.1 diff --git a/go.sum b/go.sum index 0da371b9f..5d0c9e012 100644 --- a/go.sum +++ b/go.sum @@ -363,8 +363,8 @@ github.com/openimsdk/gomake v0.0.17 h1:q8haP48VOH45WhJRiLj1YSBJyUFJqD8CTedH65i1Y github.com/openimsdk/gomake v0.0.17/go.mod h1:nnjS8yCtrPJAt1knMbyPiUwCH2gpyBzj/EZAONfUOXg= github.com/openimsdk/protocol v0.0.73-alpha.19 h1:CvXoDF2U73UcMhLnrtMFks2Aw+bXiDgH8AITEt783/s= github.com/openimsdk/protocol v0.0.73-alpha.19/go.mod h1:WF7EuE55vQvpyUAzDXcqg+B+446xQyEba0X35lTINmw= -github.com/openimsdk/tools v0.0.50-alpha.117 h1:ACfijEVCeBcttT7OOkNGOOOvq14pJtb9szNIMHLm6Vc= -github.com/openimsdk/tools v0.0.50-alpha.117/go.mod h1:I0WESSa7ghPIo9BL+ETlH/qEIbO6+KZioM1jwNuDwz0= +github.com/openimsdk/tools v0.0.50-alpha.121 h1:TXKKgtkeMeqIs0vpolbW8rIEngE9xlESq+0NV+FoLH0= +github.com/openimsdk/tools v0.0.50-alpha.121/go.mod h1:I0WESSa7ghPIo9BL+ETlH/qEIbO6+KZioM1jwNuDwz0= github.com/pelletier/go-toml/v2 v2.2.2 h1:aYUidT7k73Pcl9nb2gScu7NSrKCSHIDE89b3+6Wq+LM= github.com/pelletier/go-toml/v2 v2.2.2/go.mod h1:1t835xjRzz80PqgE6HHgN2JOsmgYu/h4qDAS4n929Rs= github.com/pierrec/lz4/v4 v4.1.21 h1:yOVMLb6qSIDP67pl/5F7RepeKYu/VmTyEXvuMI5d9mQ= diff --git a/internal/api/init.go b/internal/api/init.go index f3548e29a..77011d0ad 100644 --- a/internal/api/init.go +++ b/internal/api/init.go @@ -23,13 +23,14 @@ import ( "strconv" "time" + "google.golang.org/grpc" + conf "github.com/openimsdk/open-im-server/v3/pkg/common/config" "github.com/openimsdk/tools/discovery" "github.com/openimsdk/tools/log" "github.com/openimsdk/tools/utils/datautil" "github.com/openimsdk/tools/utils/network" "github.com/openimsdk/tools/utils/runtimeenv" - "google.golang.org/grpc" ) type Config struct { @@ -77,18 +78,6 @@ func Start(ctx context.Context, config *Config, client discovery.SvcDiscoveryReg apiCancel(err) }() - //if config.Discovery.Enable == conf.ETCD { - // cm := disetcd.NewConfigManager(client.(*etcd.SvcDiscoveryRegistryImpl).GetClient(), config.GetConfigNames()) - // cm.Watch(ctx) - //} - //sigs := make(chan os.Signal, 1) - //signal.Notify(sigs, syscall.SIGTERM) - //select { - //case val := <-sigs: - // log.ZDebug(ctx, "recv exit", "signal", val.String()) - // cancel(fmt.Errorf("signal %s", val.String())) - //case <-ctx.Done(): - //} <-apiCtx.Done() exitCause := context.Cause(apiCtx) log.ZWarn(ctx, "api server exit", exitCause) diff --git a/internal/api/router.go b/internal/api/router.go index 81787f9a9..c5f93c6c0 100644 --- a/internal/api/router.go +++ b/internal/api/router.go @@ -9,6 +9,8 @@ import ( "github.com/gin-gonic/gin" "github.com/gin-gonic/gin/binding" "github.com/go-playground/validator/v10" + clientv3 "go.etcd.io/etcd/client/v3" + "github.com/openimsdk/open-im-server/v3/internal/api/jssdk" "github.com/openimsdk/open-im-server/v3/pkg/authverify" "github.com/openimsdk/open-im-server/v3/pkg/common/config" @@ -23,13 +25,15 @@ import ( "github.com/openimsdk/protocol/relation" "github.com/openimsdk/protocol/third" "github.com/openimsdk/protocol/user" + "github.com/openimsdk/tools/a2r" "github.com/openimsdk/tools/apiresp" "github.com/openimsdk/tools/discovery" "github.com/openimsdk/tools/discovery/etcd" + "github.com/openimsdk/tools/discovery/inprocess" "github.com/openimsdk/tools/log" + "github.com/openimsdk/tools/mcontext" "github.com/openimsdk/tools/mw" "github.com/openimsdk/tools/mw/api" - clientv3 "go.etcd.io/etcd/client/v3" ) const ( @@ -341,9 +345,55 @@ func newGinRouter(ctx context.Context, client discovery.SvcDiscoveryRegistry, cf { r.POST("/restart", cm.CheckAdmin, cm.Restart) } + + if config.Standalone() { + a := internalApi{secret: cfg.Share.Secret, client: client} + r.POST(inprocess.BroadcastPath, a.RpcInvoke) + } return r, nil } +type internalApi struct { + secret string + client discovery.SvcDiscoveryRegistry +} + +func (x *internalApi) RpcInvoke(c *gin.Context) { + req, err := a2r.ParseRequestNotCheck[inprocess.InvokeRequest](c) + if err != nil { + apiresp.GinError(c, err) + return + } + // Request length is deliberately not validated: a proto request whose fields + // are all default values marshals to zero bytes. + if req.Service == "" || req.Method == "" || req.Secret == "" { + apiresp.GinError(c, servererrs.ErrArgs) + return + } + if req.Secret != x.secret { + apiresp.GinError(c, servererrs.ErrNoPermission.WrapMsg("secret not match")) + return + } + ctx := context.Context(c) + if req.OpUserID != "" { + ctx = mcontext.SetOpUserID(ctx, req.OpUserID) + } + if req.Admin { + ctx = authverify.WithTempAdmin(ctx) + } + cc, err := x.client.GetConn(ctx, req.Service) + if err != nil { + apiresp.GinError(c, err) + return + } + var resp []byte + if err := cc.Invoke(ctx, req.Method, req.Request, &resp); err != nil { + apiresp.GinError(c, err) + return + } + apiresp.GinSuccess(c, resp) +} + func GinParseToken(authClient *rpcli.AuthClient) gin.HandlerFunc { return func(c *gin.Context) { switch c.Request.Method { @@ -385,4 +435,6 @@ func setGinIsAdmin(imAdminUserID []string) gin.HandlerFunc { var Whitelist = []string{ "/auth/get_admin_token", "/auth/parse_token", + // cross-instance rpc invoke authenticates with share.secret instead of token + inprocess.BroadcastPath, } diff --git a/internal/msggateway/init.go b/internal/msggateway/init.go index 40a57b1da..59b33ad85 100644 --- a/internal/msggateway/init.go +++ b/internal/msggateway/init.go @@ -18,13 +18,14 @@ import ( "context" "time" + "google.golang.org/grpc" + "github.com/openimsdk/open-im-server/v3/pkg/common/config" "github.com/openimsdk/open-im-server/v3/pkg/dbbuild" "github.com/openimsdk/open-im-server/v3/pkg/rpccache" "github.com/openimsdk/tools/discovery" "github.com/openimsdk/tools/utils/datautil" "github.com/openimsdk/tools/utils/runtimeenv" - "google.golang.org/grpc" "github.com/openimsdk/tools/log" ) @@ -76,42 +77,3 @@ func Start(ctx context.Context, conf *Config, client discovery.SvcDiscoveryRegis return hubServer.LongConnServer.Run(ctx) } - -// -//// Start run ws server. -//func Start(ctx context.Context, index int, conf *Config) error { -// log.CInfo(ctx, "MSG-GATEWAY server is initializing", "runtimeEnv", runtimeenv.RuntimeEnvironment(), -// "rpcPorts", conf.MsgGateway.RPC.Ports, -// "wsPort", conf.MsgGateway.LongConnSvr.Ports, "prometheusPorts", conf.MsgGateway.Prometheus.Ports) -// wsPort, err := datautil.GetElemByIndex(conf.MsgGateway.LongConnSvr.Ports, index) -// if err != nil { -// return err -// } -// -// rdb, err := redisutil.NewRedisClient(ctx, conf.RedisConfig.Build()) -// if err != nil { -// return err -// } -// longServer := NewWsServer( -// conf, -// WithPort(wsPort), -// WithMaxConnNum(int64(conf.MsgGateway.LongConnSvr.WebsocketMaxConnNum)), -// WithHandshakeTimeout(time.Duration(conf.MsgGateway.LongConnSvr.WebsocketTimeout)*time.Second), -// WithMessageMaxMsgLength(conf.MsgGateway.LongConnSvr.WebsocketMaxMsgLen), -// ) -// -// hubServer := NewServer(longServer, conf, func(srv *Server) error { -// var err error -// longServer.online, err = rpccache.NewOnlineCache(srv.userClient, nil, rdb, false, longServer.subscriberUserOnlineStatusChanges) -// return err -// }) -// -// go longServer.ChangeOnlineStatus(4) -// -// netDone := make(chan error) -// go func() { -// err = hubServer.Start(ctx, index, conf) -// netDone <- err -// }() -// return hubServer.LongConnServer.Run(netDone) -//} diff --git a/internal/msgtransfer/init.go b/internal/msgtransfer/init.go index 35026c79a..2b483fcd7 100644 --- a/internal/msgtransfer/init.go +++ b/internal/msgtransfer/init.go @@ -18,8 +18,6 @@ import ( "context" "fmt" - "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/open-im-server/v3/pkg/dbbuild" @@ -28,10 +26,11 @@ import ( "github.com/openimsdk/tools/mq" "github.com/openimsdk/tools/utils/runtimeenv" + "google.golang.org/grpc" + conf "github.com/openimsdk/open-im-server/v3/pkg/common/config" "github.com/openimsdk/open-im-server/v3/pkg/common/storage/controller" "github.com/openimsdk/tools/log" - "google.golang.org/grpc" ) type MsgTransfer struct { @@ -59,8 +58,6 @@ type Config struct { } func Start(ctx context.Context, config *Config, client discovery.SvcDiscoveryRegistry, server grpc.ServiceRegistrar) error { - builder := mqbuild.NewBuilder(&config.KafkaConfig) - 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) @@ -72,25 +69,15 @@ func Start(ctx context.Context, config *Config, client discovery.SvcDiscoveryReg if err != nil { return err } - - //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) + builder, err := mqbuild.NewBuilder(config.Share.Queue, &config.KafkaConfig, rdb) if err != nil { return err } - pushProducer, err := builder.GetTopicProducer(ctx, config.KafkaConfig.ToPushTopic) + mongoProducer, err := builder.GetTopicProducer(ctx, mqbuild.TopicToMongo) + if err != nil { + return err + } + pushProducer, err := builder.GetTopicProducer(ctx, mqbuild.TopicToPush) if err != nil { return err } @@ -98,16 +85,7 @@ func Start(ctx context.Context, config *Config, client discovery.SvcDiscoveryReg if err != nil { return err } - 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) - } + msgModel := redis.NewMsgCache(rdb, msgDocModel) seqConversation, err := mgo.NewSeqConversationMongo(mgocli.GetDB()) if err != nil { return err @@ -122,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 bf95b6acc..d68708422 100644 --- a/internal/push/push.go +++ b/internal/push/push.go @@ -2,25 +2,24 @@ package push import ( "context" - "github.com/openimsdk/tools/mq" "math/rand" "strconv" + "github.com/openimsdk/tools/mq" + + "google.golang.org/grpc" + "github.com/openimsdk/open-im-server/v3/internal/push/offlinepush" "github.com/openimsdk/open-im-server/v3/pkg/authverify" "github.com/openimsdk/open-im-server/v3/pkg/common/config" - "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/controller" - "github.com/openimsdk/open-im-server/v3/pkg/common/storage/database/mgo" "github.com/openimsdk/open-im-server/v3/pkg/dbbuild" "github.com/openimsdk/open-im-server/v3/pkg/mqbuild" pbpush "github.com/openimsdk/protocol/push" "github.com/openimsdk/tools/discovery" "github.com/openimsdk/tools/log" "github.com/openimsdk/tools/mcontext" - "google.golang.org/grpc" ) type pushServer struct { @@ -57,37 +56,26 @@ func Start(ctx context.Context, config *Config, client discovery.SvcDiscoveryReg if err != nil { return err } - var cacheModel cache.ThirdCache - if rdb == nil { - mdb, err := dbb.Mongo(ctx) - if err != nil { - return err - } - mc, err := mgo.NewCacheMgo(mdb.GetDB()) - if err != nil { - return err - } - cacheModel = mcache.NewThirdCache(mc) - } else { - cacheModel = redis.NewThirdCache(rdb) - } + cacheModel := redis.NewThirdCache(rdb) offlinePusher, err := offlinepush.NewOfflinePusher(&config.RpcConfig, cacheModel, string(config.FcmConfigPath)) if err != nil { return err } - builder := mqbuild.NewBuilder(&config.KafkaConfig) - - offlinePushProducer, err := builder.GetTopicProducer(ctx, config.KafkaConfig.ToOfflinePushTopic) + builder, err := mqbuild.NewBuilder(config.Share.Queue, &config.KafkaConfig, rdb) + if err != nil { + return err + } + 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/auth/auth.go b/internal/rpc/auth/auth.go index a78a714ca..c6f64deff 100644 --- a/internal/rpc/auth/auth.go +++ b/internal/rpc/auth/auth.go @@ -19,18 +19,18 @@ import ( "errors" "github.com/openimsdk/open-im-server/v3/pkg/common/convert" - "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/database/mgo" "github.com/openimsdk/open-im-server/v3/pkg/dbbuild" "github.com/openimsdk/open-im-server/v3/pkg/localcache" "github.com/openimsdk/open-im-server/v3/pkg/rpccache" "github.com/openimsdk/open-im-server/v3/pkg/rpcli" + "github.com/redis/go-redis/v9" + "github.com/openimsdk/open-im-server/v3/pkg/common/config" redis2 "github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache/redis" "github.com/openimsdk/tools/utils/datautil" - "github.com/redis/go-redis/v9" + + "google.golang.org/grpc" "github.com/openimsdk/open-im-server/v3/pkg/authverify" "github.com/openimsdk/open-im-server/v3/pkg/common/prommetrics" @@ -43,7 +43,6 @@ import ( "github.com/openimsdk/tools/errs" "github.com/openimsdk/tools/log" "github.com/openimsdk/tools/tokenverify" - "google.golang.org/grpc" ) type authServer struct { @@ -71,20 +70,7 @@ func Start(ctx context.Context, config *Config, client discovery.SvcDiscoveryReg if err != nil { return err } - var token cache.TokenModel - if rdb == nil { - mdb, err := dbb.Mongo(ctx) - if err != nil { - return err - } - mc, err := mgo.NewCacheMgo(mdb.GetDB()) - if err != nil { - return err - } - token = mcache.NewTokenCacheModel(mc, config.RpcConfig.TokenPolicy.Expire) - } else { - token = redis2.NewTokenCacheModel(rdb, &config.LocalCacheConfig, config.RpcConfig.TokenPolicy.Expire) - } + token := redis2.NewTokenCacheModel(rdb, &config.LocalCacheConfig, config.RpcConfig.TokenPolicy.Expire) userConn, err := client.GetConn(ctx, config.Discovery.RpcService.User) if err != nil { return err diff --git a/internal/rpc/msg/server.go b/internal/rpc/msg/server.go index 48101cdd7..e7c589b8c 100644 --- a/internal/rpc/msg/server.go +++ b/internal/rpc/msg/server.go @@ -17,12 +17,11 @@ package msg import ( "context" - "github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache" - "github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache/mcache" + "google.golang.org/grpc" + "github.com/openimsdk/open-im-server/v3/pkg/dbbuild" "github.com/openimsdk/open-im-server/v3/pkg/mqbuild" "github.com/openimsdk/open-im-server/v3/pkg/rpcli" - "google.golang.org/grpc" "github.com/openimsdk/open-im-server/v3/pkg/common/config" "github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache/redis" @@ -80,11 +79,6 @@ func (m *msgServer) addInterceptorHandler(interceptorFunc ...MessageInterceptorF } func Start(ctx context.Context, config *Config, client discovery.SvcDiscoveryRegistry, server grpc.ServiceRegistrar) error { - builder := mqbuild.NewBuilder(&config.KafkaConfig) - redisProducer, err := builder.GetTopicProducer(ctx, config.KafkaConfig.ToRedisTopic) - if err != nil { - return err - } dbb := dbbuild.NewBuilder(&config.MongodbConfig, &config.RedisConfig) mgocli, err := dbb.Mongo(ctx) if err != nil { @@ -94,20 +88,19 @@ func Start(ctx context.Context, config *Config, client discovery.SvcDiscoveryReg if err != nil { return err } + builder, err := mqbuild.NewBuilder(config.Share.Queue, &config.KafkaConfig, rdb) + if err != nil { + return err + } + redisProducer, err := builder.GetTopicProducer(ctx, mqbuild.TopicToRedis) + if err != nil { + return err + } msgDocModel, err := mgo.NewMsgMongo(mgocli.GetDB()) if err != nil { return err } - 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) - } + msgModel := redis.NewMsgCache(rdb, msgDocModel) seqConversation, err := mgo.NewSeqConversationMongo(mgocli.GetDB()) if err != nil { return err diff --git a/internal/rpc/third/third.go b/internal/rpc/third/third.go index cea6a8522..90a16012d 100644 --- a/internal/rpc/third/third.go +++ b/internal/rpc/third/third.go @@ -20,8 +20,6 @@ import ( "time" "github.com/openimsdk/open-im-server/v3/pkg/authverify" - "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/dbbuild" "github.com/openimsdk/open-im-server/v3/pkg/rpcli" "github.com/openimsdk/tools/s3/disable" @@ -33,6 +31,8 @@ import ( "github.com/openimsdk/tools/s3/aws" "github.com/openimsdk/tools/s3/kodo" + "google.golang.org/grpc" + "github.com/openimsdk/open-im-server/v3/pkg/common/storage/controller" "github.com/openimsdk/protocol/third" "github.com/openimsdk/tools/discovery" @@ -40,7 +40,6 @@ import ( "github.com/openimsdk/tools/s3/cos" "github.com/openimsdk/tools/s3/minio" "github.com/openimsdk/tools/s3/oss" - "google.golang.org/grpc" ) type thirdServer struct { @@ -83,30 +82,12 @@ func Start(ctx context.Context, config *Config, client discovery.SvcDiscoveryReg if err != nil { return err } - var thirdCache cache.ThirdCache - if rdb == nil { - tc, err := mgo.NewCacheMgo(mgocli.GetDB()) - if err != nil { - return err - } - thirdCache = mcache.NewThirdCache(tc) - } else { - thirdCache = redis.NewThirdCache(rdb) - } + thirdCache := redis.NewThirdCache(rdb) // Select the oss method according to the profile policy var o s3.Interface switch enable := config.RpcConfig.Object.Enable; enable { case "minio": - var minioCache minio.Cache - if rdb == nil { - mc, err := mgo.NewCacheMgo(mgocli.GetDB()) - if err != nil { - return err - } - minioCache = mcache.NewMinioCache(mc) - } else { - minioCache = redis.NewMinioCache(rdb) - } + minioCache := redis.NewMinioCache(rdb) o, err = minio.NewMinio(ctx, minioCache, *config.MinioConfig.Build()) case "cos": o, err = cos.NewCos(*config.RpcConfig.Object.Cos.Build()) diff --git a/internal/tools/cron/cron_task.go b/internal/tools/cron/cron_task.go index 2c8655d4a..e4644435b 100644 --- a/internal/tools/cron/cron_task.go +++ b/internal/tools/cron/cron_task.go @@ -3,8 +3,12 @@ package cron import ( "context" + "github.com/robfig/cron/v3" + "google.golang.org/grpc" + "github.com/openimsdk/open-im-server/v3/pkg/common/config" disetcd "github.com/openimsdk/open-im-server/v3/pkg/common/discovery/etcd" + "github.com/openimsdk/open-im-server/v3/pkg/dbbuild" pbconversation "github.com/openimsdk/protocol/conversation" "github.com/openimsdk/protocol/msg" "github.com/openimsdk/protocol/third" @@ -14,14 +18,13 @@ import ( "github.com/openimsdk/tools/log" "github.com/openimsdk/tools/mcontext" "github.com/openimsdk/tools/utils/runtimeenv" - "github.com/robfig/cron/v3" - "google.golang.org/grpc" ) type Config struct { - CronTask config.CronTask - Share config.Share - Discovery config.Discovery + CronTask config.CronTask + Share config.Share + Discovery config.Discovery + RedisConfig config.Redis } func Start(ctx context.Context, conf *Config, client discovery.SvcDiscoveryRegistry, service grpc.ServiceRegistrar) error { @@ -60,10 +63,13 @@ func Start(ctx context.Context, conf *Config, client discovery.SvcDiscoveryRegis if err != nil { return err } - } - - if locker == nil { - locker = emptyLocker{} + } else { + builder := dbbuild.NewBuilder(nil, &conf.RedisConfig) + rdb, err := builder.Redis(ctx) + if err != nil { + return err + } + locker = NewRedisLocker(rdb) } srv := &cronServer{ @@ -95,16 +101,6 @@ func Start(ctx context.Context, conf *Config, client discovery.SvcDiscoveryRegis return nil } -type Locker interface { - ExecuteWithLock(ctx context.Context, taskName string, task func()) -} - -type emptyLocker struct{} - -func (emptyLocker) ExecuteWithLock(ctx context.Context, taskName string, task func()) { - task() -} - type cronServer struct { ctx context.Context config *Config diff --git a/internal/tools/cron/dist_look.go b/internal/tools/cron/etcd_locker.go similarity index 97% rename from internal/tools/cron/dist_look.go rename to internal/tools/cron/etcd_locker.go index e46b9206c..959110fea 100644 --- a/internal/tools/cron/dist_look.go +++ b/internal/tools/cron/etcd_locker.go @@ -6,13 +6,10 @@ import ( "os" "time" - "github.com/openimsdk/tools/log" clientv3 "go.etcd.io/etcd/client/v3" "go.etcd.io/etcd/client/v3/concurrency" -) -const ( - lockLeaseTTL = 300 + "github.com/openimsdk/tools/log" ) type EtcdLocker struct { @@ -35,7 +32,7 @@ func NewEtcdLocker(client *clientv3.Client) (*EtcdLocker, error) { } func (e *EtcdLocker) ExecuteWithLock(ctx context.Context, taskName string, task func()) { - session, err := concurrency.NewSession(e.client, concurrency.WithTTL(lockLeaseTTL)) + session, err := concurrency.NewSession(e.client, concurrency.WithTTL(int(lockLeaseTTL/time.Second))) if err != nil { log.ZWarn(ctx, "Failed to create etcd session", err, "taskName", taskName, diff --git a/internal/tools/cron/locker.go b/internal/tools/cron/locker.go new file mode 100644 index 000000000..4a1115486 --- /dev/null +++ b/internal/tools/cron/locker.go @@ -0,0 +1,14 @@ +package cron + +import ( + "context" + "time" +) + +const ( + lockLeaseTTL = time.Second * 300 +) + +type Locker interface { + ExecuteWithLock(ctx context.Context, taskName string, task func()) +} diff --git a/internal/tools/cron/redis_locker.go b/internal/tools/cron/redis_locker.go new file mode 100644 index 000000000..329e9cb77 --- /dev/null +++ b/internal/tools/cron/redis_locker.go @@ -0,0 +1,68 @@ +package cron + +import ( + "context" + "strings" + "time" + + "github.com/google/uuid" + "github.com/redis/go-redis/v9" + + "github.com/openimsdk/tools/log" +) + +func NewRedisLocker(client redis.UniversalClient) *RedisLocker { + return &RedisLocker{ + client: client, + script: redis.NewScript(strings.TrimSpace(` +if redis.call("get", KEYS[1]) == ARGV[1] then + return redis.call("del", KEYS[1]) +else + return 0 +end +`)), + } +} + +type RedisLocker struct { + client redis.UniversalClient + script *redis.Script +} + +func (e *RedisLocker) getKey(name string) string { + return "CRON_LOCKED:" + name +} + +func (e *RedisLocker) lock(ctx context.Context, name string, owner string) (bool, error) { + ctx, cancel := context.WithTimeout(ctx, time.Second) + defer cancel() + return e.client.SetNX(ctx, e.getKey(name), owner, lockLeaseTTL).Result() +} + +func (e *RedisLocker) unlock(ctx context.Context, name string, owner string) error { + ctx, cancel := context.WithTimeout(ctx, time.Second) + defer cancel() + return e.script.Run(ctx, e.client, []string{e.getKey(name)}, owner).Err() +} + +func (e *RedisLocker) ExecuteWithLock(ctx context.Context, taskName string, task func()) { + owner := uuid.New().String() + ok, err := e.lock(ctx, taskName, owner) + if err != nil { + log.ZWarn(ctx, "cron lock get lock", err, "taskName", taskName) + return + } + log.ZDebug(ctx, "cron lock get lock", "taskName", taskName, "ok", ok, "owner", owner) + if !ok { + return + } + defer func() { + err := e.unlock(ctx, taskName, owner) + if err == nil { + log.ZDebug(ctx, "cron lock unlock", "taskName", taskName, "owner", owner) + } else { + log.ZWarn(ctx, "cron lock unlock", err, "taskName", taskName, "owner", owner) + } + }() + task() +} diff --git a/pkg/common/cmd/cron_task.go b/pkg/common/cmd/cron_task.go index c666bd021..041deea67 100644 --- a/pkg/common/cmd/cron_task.go +++ b/pkg/common/cmd/cron_task.go @@ -17,12 +17,13 @@ package cmd import ( "context" + "github.com/spf13/cobra" + "github.com/openimsdk/open-im-server/v3/internal/tools/cron" "github.com/openimsdk/open-im-server/v3/pkg/common/config" "github.com/openimsdk/open-im-server/v3/pkg/common/startrpc" "github.com/openimsdk/open-im-server/v3/version" "github.com/openimsdk/tools/system/program" - "github.com/spf13/cobra" ) type CronTaskCmd struct { @@ -39,6 +40,7 @@ func NewCronTaskCmd() *CronTaskCmd { config.OpenIMCronTaskCfgFileName: &cronTaskConfig.CronTask, config.ShareFileName: &cronTaskConfig.Share, config.DiscoveryConfigFilename: &cronTaskConfig.Discovery, + config.RedisConfigFileName: &cronTaskConfig.RedisConfig, } ret.RootCmd = NewRootCmd(program.GetProcessName(), WithConfigMap(ret.configMap)) ret.ctx = context.WithValue(context.Background(), "version", version.Version) diff --git a/pkg/common/config/config.go b/pkg/common/config/config.go index 695684d15..8ec648359 100644 --- a/pkg/common/config/config.go +++ b/pkg/common/config/config.go @@ -28,6 +28,13 @@ import ( "github.com/openimsdk/tools/s3/oss" ) +// FileConfig is implemented by every config struct that is backed by a config +// file; GetConfigFileName reports that file's name (e.g. "redis.yml"). It is the +// single marker used to decide whether a struct field is loaded from a file. +type FileConfig interface { + GetConfigFileName() string +} + const StructTagName = "yaml" type Path string @@ -135,6 +142,7 @@ type API struct { Api struct { ListenIP string `yaml:"listenIP"` Ports []int `yaml:"ports"` + RegisterIP string `yaml:"registerIP"` CompressionLevel int `yaml:"compressionLevel"` } `yaml:"api"` Prometheus struct { @@ -418,7 +426,8 @@ type AfterConfig struct { } type Share struct { - Secret string `yaml:"secret"` + Secret string `yaml:"secret"` + Queue EngineSelector `yaml:"queue"` IMAdminUser struct { UserIDs []string `yaml:"userIDs"` Nicknames []string `yaml:"nicknames"` diff --git a/pkg/common/config/load_config.go b/pkg/common/config/load_config.go index 142b704e1..de493ceeb 100644 --- a/pkg/common/config/load_config.go +++ b/pkg/common/config/load_config.go @@ -3,12 +3,14 @@ package config import ( "os" "path/filepath" + "reflect" "strings" "github.com/mitchellh/mapstructure" + "github.com/spf13/viper" + "github.com/openimsdk/tools/errs" "github.com/openimsdk/tools/utils/runtimeenv" - "github.com/spf13/viper" ) func Load(configDirectory string, configFileName string, envPrefix string, config any) error { @@ -35,10 +37,27 @@ func loadConfig(path string, envPrefix string, config any) error { return errs.WrapMsg(err, "failed to read config file", "path", path, "envPrefix", envPrefix) } - if err := v.Unmarshal(config, func(config *mapstructure.DecoderConfig) { - config.TagName = StructTagName - }); err != nil { + if err := v.Unmarshal(config, ApplyDecoderConfig); err != nil { return errs.WrapMsg(err, "failed to unmarshal config", "path", path, "envPrefix", envPrefix) } return nil } + +func ApplyDecoderConfig(config *mapstructure.DecoderConfig) { + config.TagName = StructTagName + config.DecodeHook = mapstructure.ComposeDecodeHookFunc( + mapstructure.StringToTimeDurationHookFunc(), + mapstructure.StringToSliceHookFunc(","), + stringToEngineSelectorHookFunc(), + ) +} + +func stringToEngineSelectorHookFunc() mapstructure.DecodeHookFuncType { + engineSelectorType := reflect.TypeOf(EngineSelector{}) + return func(from reflect.Type, to reflect.Type, data any) (any, error) { + if to != engineSelectorType || from.Kind() != reflect.String { + return data, nil + } + return EngineSelector{Engine: data.(string)}, nil + } +} diff --git a/pkg/common/config/mq.go b/pkg/common/config/mq.go new file mode 100644 index 000000000..76fdb26e6 --- /dev/null +++ b/pkg/common/config/mq.go @@ -0,0 +1,41 @@ +package config + +import ( + "strings" + + "github.com/openimsdk/tools/errs" +) + +const ( + QueueEngineKafka = "kafka" + QueueEngineRedis = "redis" + QueueEngineMemory = "memory" +) + +func NormalizeQueueEngine(engine string) string { + switch strings.ToLower(strings.TrimSpace(engine)) { + case "", "kafka": + return QueueEngineKafka + case "redis": + return QueueEngineRedis + case "memory": + return QueueEngineMemory + default: + return strings.ToLower(strings.TrimSpace(engine)) + } +} + +func ValidateQueueEngine(engine string, standalone bool) (string, error) { + normalized := NormalizeQueueEngine(engine) + switch normalized { + case QueueEngineKafka, QueueEngineRedis: + return normalized, nil + case QueueEngineMemory: + if standalone { + return normalized, nil + } + return "", errs.ErrArgs.WrapMsg("unsupported queue engine for microservice deployment", "engine", engine) + default: + return "", errs.ErrArgs.WrapMsg("unsupported queue engine", "engine", engine) + } +} diff --git a/pkg/common/config/selector.go b/pkg/common/config/selector.go new file mode 100644 index 000000000..8aeeeaaa3 --- /dev/null +++ b/pkg/common/config/selector.go @@ -0,0 +1,43 @@ +package config + +import "encoding/json" + +type EngineSelector struct { + Engine string `yaml:"engine" mapstructure:"engine" json:"engine"` +} + +func (e EngineSelector) String() string { + return e.Engine +} + +func (e *EngineSelector) UnmarshalYAML(unmarshal func(any) error) error { + var engine string + if err := unmarshal(&engine); err == nil { + e.Engine = engine + return nil + } + var cfg struct { + Engine string `yaml:"engine"` + } + if err := unmarshal(&cfg); err != nil { + return err + } + e.Engine = cfg.Engine + return nil +} + +func (e *EngineSelector) UnmarshalJSON(data []byte) error { + var engine string + if err := json.Unmarshal(data, &engine); err == nil { + e.Engine = engine + return nil + } + var cfg struct { + Engine string `json:"engine"` + } + if err := json.Unmarshal(data, &cfg); err != nil { + return err + } + e.Engine = cfg.Engine + return nil +} diff --git a/pkg/common/discovery/discoveryregister.go b/pkg/common/discovery/discoveryregister.go index 87333fcac..29ed5929f 100644 --- a/pkg/common/discovery/discoveryregister.go +++ b/pkg/common/discovery/discoveryregister.go @@ -17,11 +17,12 @@ package discovery import ( "time" + "google.golang.org/grpc" + "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/discovery/inprocess" "github.com/openimsdk/tools/utils/runtimeenv" - "google.golang.org/grpc" "github.com/openimsdk/tools/discovery/kubernetes" @@ -32,7 +33,7 @@ import ( // NewDiscoveryRegister creates a new service discovery and registry client based on the provided environment type. func NewDiscoveryRegister(discovery *config.Discovery, watchNames []string) (discovery.SvcDiscoveryRegistry, error) { if config.Standalone() { - return standalone.GetSvcDiscoveryRegistry(), nil + return inprocess.GetSvcDiscoveryRegistry(), nil } if runtimeenv.RuntimeEnvironment() == config.KUBERNETES { return kubernetes.NewConnManager(discovery.Kubernetes.Namespace, nil, diff --git a/pkg/common/storage/cache/mcache/minio.go b/pkg/common/storage/cache/mcache/minio.go deleted file mode 100644 index f07203cc2..000000000 --- a/pkg/common/storage/cache/mcache/minio.go +++ /dev/null @@ -1,50 +0,0 @@ -package mcache - -import ( - "context" - "time" - - "github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache/cachekey" - "github.com/openimsdk/open-im-server/v3/pkg/common/storage/database" - "github.com/openimsdk/tools/s3/minio" -) - -func NewMinioCache(cache database.Cache) minio.Cache { - return &minioCache{ - cache: cache, - expireTime: time.Hour * 24 * 7, - } -} - -type minioCache struct { - cache database.Cache - expireTime time.Duration -} - -func (g *minioCache) getObjectImageInfoKey(key string) string { - return cachekey.GetObjectImageInfoKey(key) -} - -func (g *minioCache) getMinioImageThumbnailKey(key string, format string, width int, height int) string { - return cachekey.GetMinioImageThumbnailKey(key, format, width, height) -} - -func (g *minioCache) DelObjectImageInfoKey(ctx context.Context, keys ...string) error { - ks := make([]string, 0, len(keys)) - for _, key := range keys { - ks = append(ks, g.getObjectImageInfoKey(key)) - } - return g.cache.Del(ctx, ks) -} - -func (g *minioCache) DelImageThumbnailKey(ctx context.Context, key string, format string, width int, height int) error { - return g.cache.Del(ctx, []string{g.getMinioImageThumbnailKey(key, format, width, height)}) -} - -func (g *minioCache) GetImageObjectKeyInfo(ctx context.Context, key string, fn func(ctx context.Context) (*minio.ImageInfo, error)) (*minio.ImageInfo, error) { - return getCache[*minio.ImageInfo](ctx, g.cache, g.getObjectImageInfoKey(key), g.expireTime, fn) -} - -func (g *minioCache) GetThumbnailKey(ctx context.Context, key string, format string, width int, height int, minioCache func(ctx context.Context) (string, error)) (string, error) { - return getCache[string](ctx, g.cache, g.getMinioImageThumbnailKey(key, format, width, height), g.expireTime, minioCache) -} diff --git a/pkg/common/storage/cache/mcache/msg_cache.go b/pkg/common/storage/cache/mcache/msg_cache.go deleted file mode 100644 index 6fd5f80a1..000000000 --- a/pkg/common/storage/cache/mcache/msg_cache.go +++ /dev/null @@ -1,132 +0,0 @@ -package mcache - -import ( - "context" - "strconv" - "sync" - "time" - - "github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache" - "github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache/cachekey" - "github.com/openimsdk/open-im-server/v3/pkg/common/storage/database" - "github.com/openimsdk/open-im-server/v3/pkg/common/storage/model" - "github.com/openimsdk/open-im-server/v3/pkg/localcache" - "github.com/openimsdk/open-im-server/v3/pkg/localcache/lru" - "github.com/openimsdk/tools/errs" - "github.com/openimsdk/tools/utils/datautil" - "github.com/redis/go-redis/v9" -) - -var ( - memMsgCache lru.LRU[string, *model.MsgInfoModel] - initMemMsgCache sync.Once -) - -func NewMsgCache(cache database.Cache, msgDocDatabase database.Msg) cache.MsgCache { - initMemMsgCache.Do(func() { - memMsgCache = lru.NewLazyLRU[string, *model.MsgInfoModel](1024*8, time.Hour, time.Second*10, localcache.EmptyTarget{}, nil) - }) - return &msgCache{ - cache: cache, - msgDocDatabase: msgDocDatabase, - memMsgCache: memMsgCache, - } -} - -type msgCache struct { - cache database.Cache - msgDocDatabase database.Msg - memMsgCache lru.LRU[string, *model.MsgInfoModel] -} - -func (x *msgCache) getSendMsgKey(id string) string { - return cachekey.GetSendMsgKey(id) -} - -func (x *msgCache) SetSendMsgStatus(ctx context.Context, id string, status int32) error { - return x.cache.Set(ctx, x.getSendMsgKey(id), strconv.Itoa(int(status)), time.Hour*24) -} - -func (x *msgCache) GetSendMsgStatus(ctx context.Context, id string) (int32, error) { - key := x.getSendMsgKey(id) - res, err := x.cache.Get(ctx, []string{key}) - if err != nil { - return 0, err - } - val, ok := res[key] - if !ok { - return 0, errs.Wrap(redis.Nil) - } - status, err := strconv.Atoi(val) - if err != nil { - return 0, errs.WrapMsg(err, "GetSendMsgStatus strconv.Atoi error", "val", val) - } - return int32(status), nil -} - -func (x *msgCache) getMsgCacheKey(conversationID string, seq int64) string { - return cachekey.GetMsgCacheKey(conversationID, seq) - -} - -func (x *msgCache) GetMessageBySeqs(ctx context.Context, conversationID string, seqs []int64) ([]*model.MsgInfoModel, error) { - if len(seqs) == 0 { - return nil, nil - } - keys := make([]string, 0, len(seqs)) - keySeq := make(map[string]int64, len(seqs)) - for _, seq := range seqs { - key := x.getMsgCacheKey(conversationID, seq) - keys = append(keys, key) - keySeq[key] = seq - } - res, err := x.memMsgCache.GetBatch(keys, func(keys []string) (map[string]*model.MsgInfoModel, error) { - findSeqs := make([]int64, 0, len(keys)) - for _, key := range keys { - seq, ok := keySeq[key] - if !ok { - continue - } - findSeqs = append(findSeqs, seq) - } - res, err := x.msgDocDatabase.FindSeqs(ctx, conversationID, seqs) - if err != nil { - return nil, err - } - kv := make(map[string]*model.MsgInfoModel) - for i := range res { - msg := res[i] - if msg == nil || msg.Msg == nil || msg.Msg.Seq <= 0 { - continue - } - key := x.getMsgCacheKey(conversationID, msg.Msg.Seq) - kv[key] = msg - } - return kv, nil - }) - if err != nil { - return nil, err - } - return datautil.Values(res), nil -} - -func (x msgCache) DelMessageBySeqs(ctx context.Context, conversationID string, seqs []int64) error { - if len(seqs) == 0 { - return nil - } - for _, seq := range seqs { - x.memMsgCache.Del(x.getMsgCacheKey(conversationID, seq)) - } - return nil -} - -func (x *msgCache) SetMessageBySeqs(ctx context.Context, conversationID string, msgs []*model.MsgInfoModel) error { - for i := range msgs { - msg := msgs[i] - if msg == nil || msg.Msg == nil || msg.Msg.Seq <= 0 { - continue - } - x.memMsgCache.Set(x.getMsgCacheKey(conversationID, msg.Msg.Seq), msg) - } - return nil -} diff --git a/pkg/common/storage/cache/mcache/online.go b/pkg/common/storage/cache/mcache/online.go deleted file mode 100644 index f018da03e..000000000 --- a/pkg/common/storage/cache/mcache/online.go +++ /dev/null @@ -1,82 +0,0 @@ -package mcache - -import ( - "context" - "sync" - - "github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache" -) - -var ( - globalOnlineCache cache.OnlineCache - globalOnlineOnce sync.Once -) - -func NewOnlineCache() cache.OnlineCache { - globalOnlineOnce.Do(func() { - globalOnlineCache = &onlineCache{ - user: make(map[string]map[int32]struct{}), - } - }) - return globalOnlineCache -} - -type onlineCache struct { - lock sync.RWMutex - user map[string]map[int32]struct{} -} - -func (x *onlineCache) GetOnline(ctx context.Context, userID string) ([]int32, error) { - x.lock.RLock() - defer x.lock.RUnlock() - pSet, ok := x.user[userID] - if !ok { - return nil, nil - } - res := make([]int32, 0, len(pSet)) - for k := range pSet { - res = append(res, k) - } - return res, nil -} - -func (x *onlineCache) SetUserOnline(ctx context.Context, userID string, online, offline []int32) error { - x.lock.Lock() - defer x.lock.Unlock() - pSet, ok := x.user[userID] - if ok { - for _, p := range offline { - delete(pSet, p) - } - } - if len(online) > 0 { - if !ok { - pSet = make(map[int32]struct{}) - x.user[userID] = pSet - } - for _, p := range online { - pSet[p] = struct{}{} - } - } - if len(pSet) == 0 { - delete(x.user, userID) - } - return nil -} - -func (x *onlineCache) GetAllOnlineUsers(ctx context.Context, cursor uint64) (map[string][]int32, uint64, error) { - if cursor != 0 { - return nil, 0, nil - } - x.lock.RLock() - defer x.lock.RUnlock() - res := make(map[string][]int32) - for k, v := range x.user { - pSet := make([]int32, 0, len(v)) - for p := range v { - pSet = append(pSet, p) - } - res[k] = pSet - } - return res, 0, nil -} diff --git a/pkg/common/storage/cache/mcache/seq_conversation.go b/pkg/common/storage/cache/mcache/seq_conversation.go deleted file mode 100644 index 879b03535..000000000 --- a/pkg/common/storage/cache/mcache/seq_conversation.go +++ /dev/null @@ -1,79 +0,0 @@ -package mcache - -import ( - "context" - - "github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache" - "github.com/openimsdk/open-im-server/v3/pkg/common/storage/database" -) - -func NewSeqConversationCache(sc database.SeqConversation) cache.SeqConversationCache { - return &seqConversationCache{ - sc: sc, - } -} - -type seqConversationCache struct { - sc database.SeqConversation -} - -func (x *seqConversationCache) Malloc(ctx context.Context, conversationID string, size int64) (int64, error) { - return x.sc.Malloc(ctx, conversationID, size) -} - -func (x *seqConversationCache) SetMinSeq(ctx context.Context, conversationID string, seq int64) error { - return x.sc.SetMinSeq(ctx, conversationID, seq) -} - -func (x *seqConversationCache) GetMinSeq(ctx context.Context, conversationID string) (int64, error) { - return x.sc.GetMinSeq(ctx, conversationID) -} - -func (x *seqConversationCache) GetMaxSeqs(ctx context.Context, conversationIDs []string) (map[string]int64, error) { - res := make(map[string]int64) - for _, conversationID := range conversationIDs { - seq, err := x.GetMaxSeq(ctx, conversationID) - if err != nil { - return nil, err - } - res[conversationID] = seq - } - return res, nil -} - -func (x *seqConversationCache) GetMaxSeqsWithTime(ctx context.Context, conversationIDs []string) (map[string]database.SeqTime, error) { - res := make(map[string]database.SeqTime) - for _, conversationID := range conversationIDs { - seq, err := x.GetMaxSeq(ctx, conversationID) - if err != nil { - return nil, err - } - res[conversationID] = database.SeqTime{Seq: seq} - } - return res, nil -} - -func (x *seqConversationCache) GetMaxSeq(ctx context.Context, conversationID string) (int64, error) { - return x.sc.GetMaxSeq(ctx, conversationID) -} - -func (x *seqConversationCache) GetMaxSeqWithTime(ctx context.Context, conversationID string) (database.SeqTime, error) { - seq, err := x.GetMinSeq(ctx, conversationID) - if err != nil { - return database.SeqTime{}, err - } - return database.SeqTime{Seq: seq}, nil -} - -func (x *seqConversationCache) SetMinSeqs(ctx context.Context, seqs map[string]int64) error { - for conversationID, seq := range seqs { - if err := x.sc.SetMinSeq(ctx, conversationID, seq); err != nil { - return err - } - } - return nil -} - -func (x *seqConversationCache) GetCacheMaxSeqWithTime(ctx context.Context, conversationIDs []string) (map[string]database.SeqTime, error) { - return x.GetMaxSeqsWithTime(ctx, conversationIDs) -} diff --git a/pkg/common/storage/cache/mcache/third.go b/pkg/common/storage/cache/mcache/third.go deleted file mode 100644 index 6918ae784..000000000 --- a/pkg/common/storage/cache/mcache/third.go +++ /dev/null @@ -1,98 +0,0 @@ -package mcache - -import ( - "context" - "strconv" - "time" - - "github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache" - "github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache/cachekey" - "github.com/openimsdk/open-im-server/v3/pkg/common/storage/database" - "github.com/openimsdk/tools/errs" - "github.com/redis/go-redis/v9" -) - -func NewThirdCache(cache database.Cache) cache.ThirdCache { - return &thirdCache{ - cache: cache, - } -} - -type thirdCache struct { - cache database.Cache -} - -func (c *thirdCache) getGetuiTokenKey() string { - return cachekey.GetGetuiTokenKey() -} - -func (c *thirdCache) getGetuiTaskIDKey() string { - return cachekey.GetGetuiTaskIDKey() -} - -func (c *thirdCache) getUserBadgeUnreadCountSumKey(userID string) string { - return cachekey.GetUserBadgeUnreadCountSumKey(userID) -} - -func (c *thirdCache) getFcmAccountTokenKey(account string, platformID int) string { - return cachekey.GetFcmAccountTokenKey(account, platformID) -} - -func (c *thirdCache) get(ctx context.Context, key string) (string, error) { - res, err := c.cache.Get(ctx, []string{key}) - if err != nil { - return "", err - } - if val, ok := res[key]; ok { - return val, nil - } - return "", errs.Wrap(redis.Nil) -} - -func (c *thirdCache) SetFcmToken(ctx context.Context, account string, platformID int, fcmToken string, expireTime int64) (err error) { - return errs.Wrap(c.cache.Set(ctx, c.getFcmAccountTokenKey(account, platformID), fcmToken, time.Duration(expireTime)*time.Second)) -} - -func (c *thirdCache) GetFcmToken(ctx context.Context, account string, platformID int) (string, error) { - return c.get(ctx, c.getFcmAccountTokenKey(account, platformID)) -} - -func (c *thirdCache) DelFcmToken(ctx context.Context, account string, platformID int) error { - return c.cache.Del(ctx, []string{c.getFcmAccountTokenKey(account, platformID)}) -} - -func (c *thirdCache) IncrUserBadgeUnreadCountSum(ctx context.Context, userID string) (int, error) { - return c.cache.Incr(ctx, c.getUserBadgeUnreadCountSumKey(userID), 1) -} - -func (c *thirdCache) SetUserBadgeUnreadCountSum(ctx context.Context, userID string, value int) error { - return c.cache.Set(ctx, c.getUserBadgeUnreadCountSumKey(userID), strconv.Itoa(value), 0) -} - -func (c *thirdCache) GetUserBadgeUnreadCountSum(ctx context.Context, userID string) (int, error) { - str, err := c.get(ctx, c.getUserBadgeUnreadCountSumKey(userID)) - if err != nil { - return 0, err - } - val, err := strconv.Atoi(str) - if err != nil { - return 0, errs.WrapMsg(err, "strconv.Atoi", "str", str) - } - return val, nil -} - -func (c *thirdCache) SetGetuiToken(ctx context.Context, token string, expireTime int64) error { - return c.cache.Set(ctx, c.getGetuiTokenKey(), token, time.Duration(expireTime)*time.Second) -} - -func (c *thirdCache) GetGetuiToken(ctx context.Context) (string, error) { - return c.get(ctx, c.getGetuiTokenKey()) -} - -func (c *thirdCache) SetGetuiTaskID(ctx context.Context, taskID string, expireTime int64) error { - return c.cache.Set(ctx, c.getGetuiTaskIDKey(), taskID, time.Duration(expireTime)*time.Second) -} - -func (c *thirdCache) GetGetuiTaskID(ctx context.Context) (string, error) { - return c.get(ctx, c.getGetuiTaskIDKey()) -} diff --git a/pkg/common/storage/cache/mcache/token.go b/pkg/common/storage/cache/mcache/token.go deleted file mode 100644 index 98b9cc066..000000000 --- a/pkg/common/storage/cache/mcache/token.go +++ /dev/null @@ -1,166 +0,0 @@ -package mcache - -import ( - "context" - "fmt" - "strconv" - "strings" - "time" - - "github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache" - "github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache/cachekey" - "github.com/openimsdk/open-im-server/v3/pkg/common/storage/database" - "github.com/openimsdk/tools/errs" - "github.com/openimsdk/tools/log" -) - -func NewTokenCacheModel(cache database.Cache, accessExpire int64) cache.TokenModel { - c := &tokenCache{cache: cache} - c.accessExpire = c.getExpireTime(accessExpire) - return c -} - -type tokenCache struct { - cache database.Cache - accessExpire time.Duration -} - -func (x *tokenCache) getTokenKey(userID string, platformID int, token string) string { - return cachekey.GetTokenKey(userID, platformID) + ":" + token -} - -func (x *tokenCache) SetTokenFlag(ctx context.Context, userID string, platformID int, token string, flag int) error { - return x.cache.Set(ctx, x.getTokenKey(userID, platformID, token), strconv.Itoa(flag), x.accessExpire) -} - -// SetTokenFlagEx set token and flag with expire time -func (x *tokenCache) SetTokenFlagEx(ctx context.Context, userID string, platformID int, token string, flag int) error { - return x.SetTokenFlag(ctx, userID, platformID, token, flag) -} - -func (x *tokenCache) GetTokensWithoutError(ctx context.Context, userID string, platformID int) (map[string]int, error) { - prefix := x.getTokenKey(userID, platformID, "") - m, err := x.cache.Prefix(ctx, prefix) - if err != nil { - return nil, errs.Wrap(err) - } - mm := make(map[string]int) - for k, v := range m { - state, err := strconv.Atoi(v) - if err != nil { - log.ZError(ctx, "token value is not int", err, "value", v, "userID", userID, "platformID", platformID) - continue - } - mm[strings.TrimPrefix(k, prefix)] = state - } - return mm, nil -} - -func (x *tokenCache) HasTemporaryToken(ctx context.Context, userID string, platformID int, token string) error { - key := cachekey.GetTemporaryTokenKey(userID, platformID, token) - if _, err := x.cache.Get(ctx, []string{key}); err != nil { - return err - } - return nil -} - -func (x *tokenCache) GetAllTokensWithoutError(ctx context.Context, userID string) (map[int]map[string]int, error) { - prefix := cachekey.UidPidToken + userID + ":" - tokens, err := x.cache.Prefix(ctx, prefix) - if err != nil { - return nil, err - } - res := make(map[int]map[string]int) - for key, flagStr := range tokens { - flag, err := strconv.Atoi(flagStr) - if err != nil { - log.ZError(ctx, "token value is not int", err, "key", key, "value", flagStr, "userID", userID) - continue - } - arr := strings.SplitN(strings.TrimPrefix(key, prefix), ":", 2) - if len(arr) != 2 { - log.ZError(ctx, "token value is not int", err, "key", key, "value", flagStr, "userID", userID) - continue - } - platformID, err := strconv.Atoi(arr[0]) - if err != nil { - log.ZError(ctx, "token value is not int", err, "key", key, "value", flagStr, "userID", userID) - continue - } - token := arr[1] - if token == "" { - log.ZError(ctx, "token value is not int", err, "key", key, "value", flagStr, "userID", userID) - continue - } - tk, ok := res[platformID] - if !ok { - tk = make(map[string]int) - res[platformID] = tk - } - tk[token] = flag - } - return res, nil -} - -func (x *tokenCache) SetTokenMapByUidPid(ctx context.Context, userID string, platformID int, m map[string]int) error { - for token, flag := range m { - err := x.SetTokenFlag(ctx, userID, platformID, token, flag) - if err != nil { - return err - } - } - return nil -} - -func (x *tokenCache) BatchSetTokenMapByUidPid(ctx context.Context, tokens map[string]map[string]any) error { - for prefix, tokenFlag := range tokens { - for token, flag := range tokenFlag { - flagStr := fmt.Sprintf("%v", flag) - if err := x.cache.Set(ctx, prefix+":"+token, flagStr, x.accessExpire); err != nil { - return err - } - } - } - return nil -} - -func (x *tokenCache) DeleteTokenByUidPid(ctx context.Context, userID string, platformID int, fields []string) error { - keys := make([]string, 0, len(fields)) - for _, token := range fields { - keys = append(keys, x.getTokenKey(userID, platformID, token)) - } - return x.cache.Del(ctx, keys) -} - -func (x *tokenCache) getExpireTime(t int64) time.Duration { - return time.Hour * 24 * time.Duration(t) -} - -func (x *tokenCache) DeleteTokenByTokenMap(ctx context.Context, userID string, tokens map[int][]string) error { - keys := make([]string, 0, len(tokens)) - for platformID, ts := range tokens { - for _, t := range ts { - keys = append(keys, x.getTokenKey(userID, platformID, t)) - } - } - return x.cache.Del(ctx, keys) -} - -func (x *tokenCache) DeleteAndSetTemporary(ctx context.Context, userID string, platformID int, fields []string) error { - keys := make([]string, 0, len(fields)) - for _, f := range fields { - keys = append(keys, x.getTokenKey(userID, platformID, f)) - } - if err := x.cache.Del(ctx, keys); err != nil { - return err - } - - for _, f := range fields { - k := cachekey.GetTemporaryTokenKey(userID, platformID, f) - if err := x.cache.Set(ctx, k, "", time.Minute*5); err != nil { - return errs.Wrap(err) - } - } - - return nil -} diff --git a/pkg/common/storage/cache/mcache/tools.go b/pkg/common/storage/cache/mcache/tools.go deleted file mode 100644 index f3c4265cd..000000000 --- a/pkg/common/storage/cache/mcache/tools.go +++ /dev/null @@ -1,63 +0,0 @@ -package mcache - -import ( - "context" - "encoding/json" - "time" - - "github.com/openimsdk/open-im-server/v3/pkg/common/storage/database" - "github.com/openimsdk/tools/log" -) - -func getCache[V any](ctx context.Context, cache database.Cache, key string, expireTime time.Duration, fn func(ctx context.Context) (V, error)) (V, error) { - getDB := func() (V, bool, error) { - res, err := cache.Get(ctx, []string{key}) - if err != nil { - var val V - return val, false, err - } - var val V - if str, ok := res[key]; ok { - if json.Unmarshal([]byte(str), &val) != nil { - return val, false, err - } - return val, true, nil - } - return val, false, nil - } - dbVal, ok, err := getDB() - if err != nil { - return dbVal, err - } - if ok { - return dbVal, nil - } - lockValue, err := cache.Lock(ctx, key, time.Minute) - if err != nil { - return dbVal, err - } - defer func() { - if err := cache.Unlock(ctx, key, lockValue); err != nil { - log.ZError(ctx, "unlock cache key", err, "key", key, "value", lockValue) - } - }() - dbVal, ok, err = getDB() - if err != nil { - return dbVal, err - } - if ok { - return dbVal, nil - } - val, err := fn(ctx) - if err != nil { - return val, err - } - data, err := json.Marshal(val) - if err != nil { - return val, err - } - if err := cache.Set(ctx, key, string(data), expireTime); err != nil { - return val, err - } - return val, nil -} diff --git a/pkg/common/storage/cache/redis/online.go b/pkg/common/storage/cache/redis/online.go index d09c44e9a..fe5d7ecb9 100644 --- a/pkg/common/storage/cache/redis/online.go +++ b/pkg/common/storage/cache/redis/online.go @@ -7,20 +7,16 @@ import ( "strings" "time" - "github.com/openimsdk/open-im-server/v3/pkg/common/config" + "github.com/redis/go-redis/v9" + "github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache" "github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache/cachekey" - "github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache/mcache" "github.com/openimsdk/protocol/constant" "github.com/openimsdk/tools/errs" "github.com/openimsdk/tools/log" - "github.com/redis/go-redis/v9" ) func NewUserOnline(rdb redis.UniversalClient) cache.OnlineCache { - if rdb == nil || config.Standalone() { - return mcache.NewOnlineCache() - } return &userOnline{ rdb: rdb, expire: cachekey.OnlineExpire, diff --git a/pkg/common/storage/cache/redis/seq_conversation.go b/pkg/common/storage/cache/redis/seq_conversation.go index 604826598..524f07372 100644 --- a/pkg/common/storage/cache/redis/seq_conversation.go +++ b/pkg/common/storage/cache/redis/seq_conversation.go @@ -7,20 +7,17 @@ import ( "strconv" "time" + "github.com/redis/go-redis/v9" + "github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache" "github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache/cachekey" - "github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache/mcache" "github.com/openimsdk/open-im-server/v3/pkg/common/storage/database" "github.com/openimsdk/open-im-server/v3/pkg/msgprocessor" "github.com/openimsdk/tools/errs" "github.com/openimsdk/tools/log" - "github.com/redis/go-redis/v9" ) func NewSeqConversationCacheRedis(rdb redis.UniversalClient, mgo database.SeqConversation) cache.SeqConversationCache { - if rdb == nil { - return mcache.NewSeqConversationCache(mgo) - } return &seqConversationCacheRedis{ mgo: mgo, lockTime: time.Second * 3, diff --git a/pkg/common/storage/cache/redis/standalone_gateway.go b/pkg/common/storage/cache/redis/standalone_gateway.go new file mode 100644 index 000000000..bb46fb476 --- /dev/null +++ b/pkg/common/storage/cache/redis/standalone_gateway.go @@ -0,0 +1,54 @@ +package redis + +import ( + "context" + "strconv" + "time" + + "github.com/redis/go-redis/v9" + + "github.com/openimsdk/tools/errs" +) + +const standaloneGatewayHashKey = "STANDALONE_GATEWAY_REGISTRY" + +type StandaloneGatewayRedis struct { + rdb redis.UniversalClient + validTime time.Duration +} + +func NewStandaloneGatewayRedis(rdb redis.UniversalClient, validTime time.Duration) *StandaloneGatewayRedis { + return &StandaloneGatewayRedis{rdb: rdb, validTime: validTime} +} + +func (s *StandaloneGatewayRedis) RegisterGateway(ctx context.Context, addr string) error { + pipe := s.rdb.Pipeline() + pipe.HSet(ctx, standaloneGatewayHashKey, addr, strconv.FormatInt(time.Now().UnixMilli(), 10)) + pipe.Expire(ctx, standaloneGatewayHashKey, s.validTime*2) + _, err := pipe.Exec(ctx) + return errs.Wrap(err) +} + +func (s *StandaloneGatewayRedis) UnregisterGateway(ctx context.Context, addr string) error { + return errs.Wrap(s.rdb.HDel(ctx, standaloneGatewayHashKey, addr).Err()) +} + +func (s *StandaloneGatewayRedis) GetGatewayAddrs(ctx context.Context) ([]string, error) { + gateways, err := s.rdb.HGetAll(ctx, standaloneGatewayHashKey).Result() + if err != nil { + return nil, errs.Wrap(err) + } + + now := time.Now() + addrs := make([]string, 0, len(gateways)) + for addr, registeredAt := range gateways { + registeredAtMs, err := strconv.ParseInt(registeredAt, 10, 64) + if err != nil { + return nil, errs.WrapMsg(err, "redis gateway register time is not int64", "addr", addr, "value", registeredAt) + } + if now.Sub(time.UnixMilli(registeredAtMs)) <= s.validTime { + addrs = append(addrs, addr) + } + } + return addrs, nil +} diff --git a/pkg/common/storage/cache/redis/standalone_gateway_test.go b/pkg/common/storage/cache/redis/standalone_gateway_test.go new file mode 100644 index 000000000..3b88df104 --- /dev/null +++ b/pkg/common/storage/cache/redis/standalone_gateway_test.go @@ -0,0 +1,52 @@ +package redis + +import ( + "context" + "strconv" + "testing" + "time" + + "github.com/go-redis/redismock/v9" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestStandaloneGatewayRedisRegisterGateway(t *testing.T) { + rdb, mock := redismock.NewClientMock() + cache := NewStandaloneGatewayRedis(rdb, time.Second*10) + + mock.Regexp().ExpectHSet(standaloneGatewayHashKey, "127.0.0.1:10001", `^[0-9]+$`).SetVal(1) + mock.ExpectExpire(standaloneGatewayHashKey, time.Second*20).SetVal(true) + + err := cache.RegisterGateway(context.Background(), "127.0.0.1:10001") + require.NoError(t, err) + assert.NoError(t, mock.ExpectationsWereMet()) +} + +func TestStandaloneGatewayRedisUnregisterGateway(t *testing.T) { + rdb, mock := redismock.NewClientMock() + cache := NewStandaloneGatewayRedis(rdb, time.Second) + + mock.ExpectHDel(standaloneGatewayHashKey, "127.0.0.1:10001").SetVal(1) + + err := cache.UnregisterGateway(context.Background(), "127.0.0.1:10001") + require.NoError(t, err) + assert.NoError(t, mock.ExpectationsWereMet()) +} + +func TestStandaloneGatewayRedisGetGatewayAddrs(t *testing.T) { + rdb, mock := redismock.NewClientMock() + cache := NewStandaloneGatewayRedis(rdb, time.Second*10) + + now := time.Now() + mock.ExpectHGetAll(standaloneGatewayHashKey).SetVal(map[string]string{ + "127.0.0.1:10001": strconv.FormatInt(now.Add(-time.Second).UnixMilli(), 10), + "127.0.0.1:10002": strconv.FormatInt(now.Add(-time.Second*20).UnixMilli(), 10), + "127.0.0.1:10003": strconv.FormatInt(now.Add(time.Second).UnixMilli(), 10), + }) + + addrs, err := cache.GetGatewayAddrs(context.Background()) + require.NoError(t, err) + assert.ElementsMatch(t, []string{"127.0.0.1:10001", "127.0.0.1:10003"}, addrs) + assert.NoError(t, mock.ExpectationsWereMet()) +} diff --git a/pkg/mqbuild/builder.go b/pkg/mqbuild/builder.go index 938159372..c2f2d3b28 100644 --- a/pkg/mqbuild/builder.go +++ b/pkg/mqbuild/builder.go @@ -4,9 +4,12 @@ import ( "context" "fmt" + "github.com/redis/go-redis/v9" + "github.com/openimsdk/open-im-server/v3/pkg/common/config" "github.com/openimsdk/tools/mq" "github.com/openimsdk/tools/mq/kafka" + "github.com/openimsdk/tools/mq/redismq" "github.com/openimsdk/tools/mq/simmq" ) @@ -15,19 +18,120 @@ type Builder interface { GetTopicConsumer(ctx context.Context, topic string) (mq.Consumer, error) } -func NewBuilder(kafka *config.Kafka) Builder { - if config.Standalone() { - return standaloneBuilder{} +const ( + TopicToRedis = "toRedis" + TopicToMongo = "toMongo" + TopicToPush = "toPush" + TopicToOfflinePush = "toOfflinePush" +) + +const ( + GroupRedis = "redis" + GroupMongo = "mongo" + GroupPush = "push" + GroupOfflinePush = "offlinePush" +) + +func NewBuilder(queue config.EngineSelector, kafka *config.Kafka, redis redis.UniversalClient) (Builder, error) { + engine, err := config.ValidateQueueEngine(queue.Engine, config.Standalone()) + if err != nil { + return nil, err } + switch engine { + case config.QueueEngineKafka: + if kafka == nil { + return nil, fmt.Errorf("nil kafka config") + } + return newKafkaBuilder(kafka), nil + case config.QueueEngineRedis: + return newRedisBuilder(redis), nil + case config.QueueEngineMemory: + return standaloneBuilder{}, nil + default: + return nil, fmt.Errorf("unsupported queue engine %s", queue.Engine) + } +} + +func newKafkaBuilder(kafka *config.Kafka) Builder { + topics := MergeTopics(KafkaTopics(kafka)) return &kafkaBuilder{ - addr: kafka.Address, - config: kafka.Build(), - topicGroupID: map[string]string{ - kafka.ToRedisTopic: kafka.ToRedisGroupID, - kafka.ToMongoTopic: kafka.ToMongoGroupID, - kafka.ToPushTopic: kafka.ToPushGroupID, - kafka.ToOfflinePushTopic: kafka.ToOfflineGroupID, - }, + addr: kafka.Address, + config: kafka.Build(), + logicalTopic: LogicalTopicNames(topics), + topicGroupID: TopicGroupID(topics), + } +} + +func newRedisBuilder(redis redis.UniversalClient) Builder { + return redismq.NewBuilder(redis, TopicGroupID(DefaultTopics()), redismq.Config{StreamPrefix: "mq:"}) +} + +type QueueTopicsConfig struct { + ToRedis QueueTopicConfig + ToMongo QueueTopicConfig + ToPush QueueTopicConfig + ToOfflinePush QueueTopicConfig +} + +type QueueTopicConfig struct { + Topic string + GroupID string +} + +func DefaultTopics() QueueTopicsConfig { + return QueueTopicsConfig{ + ToRedis: QueueTopicConfig{Topic: TopicToRedis, GroupID: GroupRedis}, + ToMongo: QueueTopicConfig{Topic: TopicToMongo, GroupID: GroupMongo}, + ToPush: QueueTopicConfig{Topic: TopicToPush, GroupID: GroupPush}, + ToOfflinePush: QueueTopicConfig{Topic: TopicToOfflinePush, GroupID: GroupOfflinePush}, + } +} + +func KafkaTopics(kafka *config.Kafka) QueueTopicsConfig { + if kafka == nil { + return QueueTopicsConfig{} + } + return QueueTopicsConfig{ + ToRedis: QueueTopicConfig{Topic: kafka.ToRedisTopic, GroupID: kafka.ToRedisGroupID}, + ToMongo: QueueTopicConfig{Topic: kafka.ToMongoTopic, GroupID: kafka.ToMongoGroupID}, + ToPush: QueueTopicConfig{Topic: kafka.ToPushTopic, GroupID: kafka.ToPushGroupID}, + ToOfflinePush: QueueTopicConfig{Topic: kafka.ToOfflinePushTopic, GroupID: kafka.ToOfflineGroupID}, + } +} + +func MergeTopics(topics QueueTopicsConfig) QueueTopicsConfig { + defaults := DefaultTopics() + fillTopic(&topics.ToRedis, defaults.ToRedis) + fillTopic(&topics.ToMongo, defaults.ToMongo) + fillTopic(&topics.ToPush, defaults.ToPush) + fillTopic(&topics.ToOfflinePush, defaults.ToOfflinePush) + return topics +} + +func fillTopic(topic *QueueTopicConfig, defaultTopic QueueTopicConfig) { + if topic.Topic == "" { + topic.Topic = defaultTopic.Topic + } + if topic.GroupID == "" { + topic.GroupID = defaultTopic.GroupID + } +} + +func LogicalTopicNames(topics QueueTopicsConfig) map[string]string { + return map[string]string{ + TopicToRedis: topics.ToRedis.Topic, + TopicToMongo: topics.ToMongo.Topic, + TopicToPush: topics.ToPush.Topic, + TopicToOfflinePush: topics.ToOfflinePush.Topic, + } +} + +func TopicGroupID(topics QueueTopicsConfig) map[string]string { + return map[string]string{ + topics.ToRedis.Topic: topics.ToRedis.GroupID, + topics.ToMongo.Topic: topics.ToMongo.GroupID, + topics.ToPush.Topic: topics.ToPush.GroupID, + topics.ToOfflinePush.Topic: topics.ToOfflinePush.GroupID, } } @@ -44,17 +148,26 @@ func (standaloneBuilder) GetTopicConsumer(ctx context.Context, topic string) (mq type kafkaBuilder struct { addr []string config *kafka.Config + logicalTopic map[string]string topicGroupID map[string]string } func (x *kafkaBuilder) GetTopicProducer(ctx context.Context, topic string) (mq.Producer, error) { - return kafka.NewKafkaProducerV2(x.config, x.addr, topic) + realTopic, ok := x.logicalTopic[topic] + if !ok { + return nil, fmt.Errorf("topic %s not found", topic) + } + return kafka.NewKafkaProducerV2(x.config, x.addr, realTopic) } func (x *kafkaBuilder) GetTopicConsumer(ctx context.Context, topic string) (mq.Consumer, error) { - groupID, ok := x.topicGroupID[topic] + realTopic, ok := x.logicalTopic[topic] if !ok { - return nil, fmt.Errorf("topic %s groupID not found", topic) + return nil, fmt.Errorf("topic %s not found", topic) } - return kafka.NewMConsumerGroupV2(ctx, x.config, groupID, []string{topic}, true) + groupID, ok := x.topicGroupID[realTopic] + if !ok { + return nil, fmt.Errorf("topic %s groupID not found", realTopic) + } + return kafka.NewMConsumerGroupV2(ctx, x.config, groupID, []string{realTopic}, true) }