feat: implement offline push using kafka

config/kafka.yml

@@ -14,12 +14,16 @@ toRedisTopic: toRedis
 toMongoTopic: toMongo
 # Kafka topic for push notifications
 toPushTopic: toPush
+# Kafka topic for offline push notifications
+toOfflinePushTopic: toOfflinePush
 # Consumer group ID for Redis topic
 toRedisGroupID: redis
 # Consumer group ID for MongoDB topic
 toMongoGroupID: mongo
 # Consumer group ID for push notifications topic
 toPushGroupID: push
+# Consumer group ID for offline push notifications topic
+toOfflinePushGroupID: offlinePush
 # TLS (Transport Layer Security) configuration
 tls:
   # Enable or disable TLS

go.mod

@@ -12,7 +12,7 @@ require (
 	github.com/gorilla/websocket v1.5.1
 	github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0
 	github.com/mitchellh/mapstructure v1.5.0
-	github.com/openimsdk/protocol v0.0.72-alpha.13
+	github.com/openimsdk/protocol v0.0.72-alpha.17
 	github.com/openimsdk/tools v0.0.50-alpha.11
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/prometheus/client_golang v1.18.0

go.sum

@@ -319,8 +319,8 @@ github.com/onsi/gomega v1.25.0 h1:Vw7br2PCDYijJHSfBOWhov+8cAnUf8MfMaIOV323l6Y=
 github.com/onsi/gomega v1.25.0/go.mod h1:r+zV744Re+DiYCIPRlYOTxn0YkOLcAnW8k1xXdMPGhM=
 github.com/openimsdk/gomake v0.0.14-alpha.5 h1:VY9c5x515lTfmdhhPjMvR3BBRrRquAUCFsz7t7vbv7Y=
 github.com/openimsdk/gomake v0.0.14-alpha.5/go.mod h1:PndCozNc2IsQIciyn9mvEblYWZwJmAI+06z94EY+csI=
-github.com/openimsdk/protocol v0.0.72-alpha.13 h1:ILpvuxWGrVJMVCPRodOQcrSMFKUBzLahBPb8GkITWSc=
-github.com/openimsdk/protocol v0.0.72-alpha.13/go.mod h1:OZQA9FR55lseYoN2Ql1XAHYKHJGu7OMNkUbuekrKCM8=
+github.com/openimsdk/protocol v0.0.72-alpha.17 h1:kB7eyjJHdkc8lpSlLIHskHzbodxkIG4eaK908iQLVdI=
+github.com/openimsdk/protocol v0.0.72-alpha.17/go.mod h1:OZQA9FR55lseYoN2Ql1XAHYKHJGu7OMNkUbuekrKCM8=
 github.com/openimsdk/tools v0.0.50-alpha.11 h1:ClhkRjUVJWbmOiQ14G6do/ES1a6ZueDITv40Apwq/Tc=
 github.com/openimsdk/tools v0.0.50-alpha.11/go.mod h1:h1cYmfyaVtgFbKmb1Cfsl8XwUOMTt8ubVUQrdGtsUh4=
 github.com/pelletier/go-toml/v2 v2.1.0 h1:FnwAJ4oYMvbT/34k9zzHuZNrhlz48GB3/s6at6/MHO4=

internal/msgtransfer/init.go

@@ -111,6 +111,7 @@ func Start(ctx context.Context, index int, config *Config) error {
 	if err != nil {
 		return err
 	}
+
 	msgTransfer := &MsgTransfer{
 		historyCH:      historyCH,
 		historyMongoCH: historyMongoCH,

internal/push/offlinepush/getui/push.go

@@ -18,11 +18,12 @@ import (
 	"context"
 	"crypto/sha256"
 	"encoding/hex"
-	"github.com/openimsdk/open-im-server/v3/internal/push/offlinepush/options"
 	"strconv"
 	"sync"
 	"time"
 
+	"github.com/openimsdk/open-im-server/v3/internal/push/offlinepush/options"
+
 	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache"
 	"github.com/openimsdk/tools/errs"
@@ -91,6 +92,16 @@ func (g *Client) Push(ctx context.Context, userIDs []string, title, content stri
 			for i, v := range s.GetSplitResult() {
 				go func(index int, userIDs []string) {
 					defer wg.Done()
+					for i := 0; i < len(userIDs); i += maxNum {
+						end := i + maxNum
+						if end > len(userIDs) {
+							end = len(userIDs)
+						}
+						if err = g.batchPush(ctx, token, userIDs[i:end], pushReq); err != nil {
+							log.ZError(ctx, "batchPush failed", err, "index", index, "token", token, "req", pushReq)
+						}
+
+					}
 					if err = g.batchPush(ctx, token, userIDs, pushReq); err != nil {
 						log.ZError(ctx, "batchPush failed", err, "index", index, "token", token, "req", pushReq)
 					}

internal/push/offlinepush_handler.go

@@ -0,0 +1,115 @@
+package push
+
+import (
+	"context"
+
+	"github.com/IBM/sarama"
+	"github.com/openimsdk/open-im-server/v3/internal/push/offlinepush"
+	"github.com/openimsdk/open-im-server/v3/internal/push/offlinepush/options"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/prommetrics"
+	"github.com/openimsdk/protocol/constant"
+	pbpush "github.com/openimsdk/protocol/push"
+	"github.com/openimsdk/protocol/sdkws"
+	"github.com/openimsdk/tools/log"
+	"github.com/openimsdk/tools/mq/kafka"
+	"github.com/openimsdk/tools/utils/jsonutil"
+	"google.golang.org/protobuf/proto"
+)
+
+type OfflinePushConsumerHandler struct {
+	OfflinePushConsumerGroup *kafka.MConsumerGroup
+	offlinePusher            offlinepush.OfflinePusher
+}
+
+func NewOfflinePushConsumerHandler(config *Config) (*OfflinePushConsumerHandler, error) {
+	var offlinePushConsumerHandler OfflinePushConsumerHandler
+	var err error
+	offlinePushConsumerHandler.OfflinePushConsumerGroup, err = kafka.NewMConsumerGroup(config.KafkaConfig.Build(), config.KafkaConfig.ToOfflineGroupID,
+		[]string{config.KafkaConfig.ToOfflinePushTopic}, true)
+	if err != nil {
+		return nil, err
+	}
+	return &offlinePushConsumerHandler, nil
+}
+
+func (*OfflinePushConsumerHandler) Setup(sarama.ConsumerGroupSession) error   { return nil }
+func (*OfflinePushConsumerHandler) Cleanup(sarama.ConsumerGroupSession) error { return nil }
+func (o *OfflinePushConsumerHandler) ConsumeClaim(sess sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
+	for msg := range claim.Messages() {
+		ctx := o.OfflinePushConsumerGroup.GetContextFromMsg(msg)
+		o.handleMsg2OfflinePush(ctx, msg.Value)
+		sess.MarkMessage(msg, "")
+	}
+	return nil
+}
+
+func (o *OfflinePushConsumerHandler) handleMsg2OfflinePush(ctx context.Context, msg []byte) {
+	offlinePushMsg := pbpush.PushMsgReq{}
+	if err := proto.Unmarshal(msg, &offlinePushMsg); err != nil {
+		log.ZError(ctx, "offline push Unmarshal msg err", err, "msg", string(msg))
+		return
+	}
+	err := o.offlinePushMsg(ctx, offlinePushMsg.MsgData, offlinePushMsg.UserIDs)
+	if err != nil {
+		log.ZWarn(ctx, "offline push failed", err, "msg", offlinePushMsg.String())
+	}
+
+}
+
+func (c *OfflinePushConsumerHandler) getOfflinePushInfos(msg *sdkws.MsgData) (title, content string, opts *options.Opts, err error) {
+	type AtTextElem struct {
+		Text       string   `json:"text,omitempty"`
+		AtUserList []string `json:"atUserList,omitempty"`
+		IsAtSelf   bool     `json:"isAtSelf"`
+	}
+
+	opts = &options.Opts{Signal: &options.Signal{}}
+	if msg.OfflinePushInfo != nil {
+		opts.IOSBadgeCount = msg.OfflinePushInfo.IOSBadgeCount
+		opts.IOSPushSound = msg.OfflinePushInfo.IOSPushSound
+		opts.Ex = msg.OfflinePushInfo.Ex
+	}
+
+	if msg.OfflinePushInfo != nil {
+		title = msg.OfflinePushInfo.Title
+		content = msg.OfflinePushInfo.Desc
+	}
+	if title == "" {
+		switch msg.ContentType {
+		case constant.Text:
+			fallthrough
+		case constant.Picture:
+			fallthrough
+		case constant.Voice:
+			fallthrough
+		case constant.Video:
+			fallthrough
+		case constant.File:
+			title = constant.ContentType2PushContent[int64(msg.ContentType)]
+		case constant.AtText:
+			ac := AtTextElem{}
+			_ = jsonutil.JsonStringToStruct(string(msg.Content), &ac)
+		case constant.SignalingNotification:
+			title = constant.ContentType2PushContent[constant.SignalMsg]
+		default:
+			title = constant.ContentType2PushContent[constant.Common]
+		}
+	}
+	if content == "" {
+		content = title
+	}
+	return
+}
+
+func (c *OfflinePushConsumerHandler) offlinePushMsg(ctx context.Context, msg *sdkws.MsgData, offlinePushUserIDs []string) error {
+	title, content, opts, err := c.getOfflinePushInfos(msg)
+	if err != nil {
+		return err
+	}
+	err = c.offlinePusher.Push(ctx, offlinePushUserIDs, title, content, opts)
+	if err != nil {
+		prommetrics.MsgOfflinePushFailedCounter.Inc()
+		return err
+	}
+	return nil
+}

internal/push/push.go

@@ -2,6 +2,7 @@ package push
 
 import (
 	"context"
+
 	"github.com/openimsdk/open-im-server/v3/internal/push/offlinepush"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache/redis"
@@ -17,12 +18,12 @@ type pushServer struct {
 	disCov        discovery.SvcDiscoveryRegistry
 	offlinePusher offlinepush.OfflinePusher
 	pushCh        *ConsumerHandler
+	offlinePushCh *OfflinePushConsumerHandler
 }
 
 type Config struct {
 	RpcConfig          config.Push
 	RedisConfig        config.Redis
-	MongodbConfig      config.Mongo
 	KafkaConfig        config.Kafka
 	NotificationConfig config.Notification
 	Share              config.Share
@@ -55,18 +56,30 @@ func Start(ctx context.Context, config *Config, client discovery.SvcDiscoveryReg
 	if err != nil {
 		return err
 	}
-	database := controller.NewPushDatabase(cacheModel)
 
-	consumer, err := NewConsumerHandler(config, offlinePusher, rdb, client)
+	database := controller.NewPushDatabase(cacheModel, &config.KafkaConfig)
+
+	consumer, err := NewConsumerHandler(config, database, offlinePusher, rdb, client)
+	if err != nil {
+		return err
+	}
+
+	offlinePushConsumer, err := NewOfflinePushConsumerHandler(config)
 	if err != nil {
 		return err
 	}
+
 	pbpush.RegisterPushMsgServiceServer(server, &pushServer{
 		database:      database,
 		disCov:        client,
 		offlinePusher: offlinePusher,
 		pushCh:        consumer,
+		offlinePushCh: offlinePushConsumer,
 	})
+
 	go consumer.pushConsumerGroup.RegisterHandleAndConsumer(ctx, consumer)
+
+	go offlinePushConsumer.OfflinePushConsumerGroup.RegisterHandleAndConsumer(ctx, offlinePushConsumer)
+
 	return nil
 }