Merge remote-tracking branch 'origin/main' into list

# Conflicts:
#	go.mod
#	go.sum

config/openim-push.yml

@@ -23,7 +23,9 @@ geTui:
   channelID: ''
   channelName: ''
 fcm:
-  serviceAccount: "x.json"
+  # Prioritize using file paths. If the file path is empty, use URL
+  filePath: ""  # File path is concatenated with the parameters passed in through - c(`mage` default pass in `config/`) and filePath. 
+  authURL: ""  #  Must start with https or http.
 jpns:
   appKey: ''
   masterSecret: ''

go.mod

@@ -176,5 +176,3 @@ require (
 	golang.org/x/crypto v0.21.0 // indirect
 	gopkg.in/ini.v1 v1.67.0 // indirect
 )
-
-//replace github.com/openimsdk/protocol => /Users/chao/Desktop/project/protocol

internal/push/offlinepush/fcm/push.go

@@ -16,8 +16,11 @@ package fcm
 
 import (
 	"context"
+	"fmt"
 	"github.com/openimsdk/open-im-server/v3/internal/push/offlinepush/options"
+	"github.com/openimsdk/tools/utils/httputil"
 	"path/filepath"
+	"strings"
 
 	firebase "firebase.google.com/go"
 	"firebase.google.com/go/messaging"
@@ -40,13 +43,25 @@ type Fcm struct {
 
 // NewClient initializes a new FCM client using the Firebase Admin SDK.
 // It requires the FCM service account credentials file located within the project's configuration directory.
-func NewClient(pushConf *config.Push, cache cache.ThirdCache) (*Fcm, error) {
+func NewClient(pushConf *config.Push, cache cache.ThirdCache, fcmConfigPath string) (*Fcm, error) {
-	projectRoot, err := config.GetProjectRoot()
+	var opt option.ClientOption
-	if err != nil {
+	switch {
-		return nil, err
+	case len(pushConf.FCM.FilePath) != 0:
+		// with file path
+		credentialsFilePath := filepath.Join(fcmConfigPath, pushConf.FCM.FilePath)
+		opt = option.WithCredentialsFile(credentialsFilePath)
+	case len(pushConf.FCM.AuthURL) != 0:
+		// with authentication URL
+		client := httputil.NewHTTPClient(httputil.NewClientConfig())
+		resp, err := client.Get(pushConf.FCM.AuthURL)
+		if err != nil {
+			return nil, err
+		}
+		opt = option.WithCredentialsJSON(resp)
+	default:
+		return nil, errs.New("no FCM config").Wrap()
 	}
-	credentialsFilePath := filepath.Join(projectRoot, "config", pushConf.FCM.ServiceAccount)
+
-	opt := option.WithCredentialsFile(credentialsFilePath)
 	fcmApp, err := firebase.NewApp(context.Background(), nil, opt)
 	if err != nil {
 		return nil, errs.Wrap(err)
@@ -56,7 +71,6 @@ func NewClient(pushConf *config.Push, cache cache.ThirdCache) (*Fcm, error) {
 	if err != nil {
 		return nil, errs.Wrap(err)
 	}
-
 	return &Fcm{fcmMsgCli: fcmMsgClient, cache: cache}, nil
 }
 
@@ -79,6 +93,8 @@ func (f *Fcm) Push(ctx context.Context, userIDs []string, title, content string,
 	notification.Body = content
 	notification.Title = title
 	var messages []*messaging.Message
+	var sendErrBuilder strings.Builder
+	var msgErrBuilder strings.Builder
 	for userID, personTokens := range allTokens {
 		apns := &messaging.APNSConfig{Payload: &messaging.APNSPayload{Aps: &messaging.Aps{Sound: opts.IOSPushSound}}}
 		messageCount := len(messages)
@@ -86,9 +102,21 @@ func (f *Fcm) Push(ctx context.Context, userIDs []string, title, content string,
 			response, err := f.fcmMsgCli.SendAll(ctx, messages)
 			if err != nil {
 				Fail = Fail + messageCount
+				// Record push error
+				sendErrBuilder.WriteString(err.Error())
+				sendErrBuilder.WriteByte('.')
 			} else {
 				Success = Success + response.SuccessCount
 				Fail = Fail + response.FailureCount
+				if response.FailureCount != 0 {
+					// Record message error
+					for i := range response.Responses {
+						if !response.Responses[i].Success {
+							msgErrBuilder.WriteString(response.Responses[i].Error.Error())
+							msgErrBuilder.WriteByte('.')
+						}
+					}
+				}
 			}
 			messages = messages[0:0]
 		}
@@ -134,5 +162,9 @@ func (f *Fcm) Push(ctx context.Context, userIDs []string, title, content string,
 			Fail = Fail + response.FailureCount
 		}
 	}
+	if Fail != 0 {
+		return errs.New(fmt.Sprintf("%d message send failed;send err:%s;message err:%s",
+			Fail, sendErrBuilder.String(), msgErrBuilder.String())).Wrap()
+	}
 	return nil
 }

internal/push/offlinepush/offlinepusher.go

@@ -36,13 +36,13 @@ type OfflinePusher interface {
 	Push(ctx context.Context, userIDs []string, title, content string, opts *options.Opts) error
 }
 
-func NewOfflinePusher(pushConf *config.Push, cache cache.ThirdCache) (OfflinePusher, error) {
+func NewOfflinePusher(pushConf *config.Push, cache cache.ThirdCache, fcmConfigPath string) (OfflinePusher, error) {
 	var offlinePusher OfflinePusher
 	switch pushConf.Enable {
 	case geTUI:
 		offlinePusher = getui.NewClient(pushConf, cache)
 	case firebase:
-		return fcm.NewClient(pushConf, cache)
+		return fcm.NewClient(pushConf, cache, fcmConfigPath)
 	case jPush:
 		offlinePusher = jpush.NewClient(pushConf)
 	default:

internal/push/push.go

@@ -29,6 +29,7 @@ type Config struct {
 	WebhooksConfig     config.Webhooks
 	LocalCacheConfig   config.LocalCache
 	Discovery          config.Discovery
+	FcmConfigPath      string
 }
 
 func (p pushServer) PushMsg(ctx context.Context, req *pbpush.PushMsgReq) (*pbpush.PushMsgResp, error) {
@@ -50,7 +51,7 @@ func Start(ctx context.Context, config *Config, client discovery.SvcDiscoveryReg
 		return err
 	}
 	cacheModel := redis.NewThirdCache(rdb)
-	offlinePusher, err := offlinepush.NewOfflinePusher(&config.RpcConfig, cacheModel)
+	offlinePusher, err := offlinepush.NewOfflinePusher(&config.RpcConfig, cacheModel, config.FcmConfigPath)
 	if err != nil {
 		return err
 	}

internal/rpc/auth/auth.go

@@ -16,6 +16,7 @@ package auth
 
 import (
 	"context"
+
 	"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/db/redisutil"
@@ -32,7 +33,6 @@ import (
 	"github.com/openimsdk/tools/discovery"
 	"github.com/openimsdk/tools/errs"
 	"github.com/openimsdk/tools/log"
-	"github.com/openimsdk/tools/mcontext"
 	"github.com/openimsdk/tools/tokenverify"
 	"google.golang.org/grpc"
 )
@@ -153,21 +153,19 @@ func (s *authServer) ForceLogout(ctx context.Context, req *pbauth.ForceLogoutReq
 	if err := authverify.CheckAdmin(ctx, s.config.Share.IMAdminUserID); err != nil {
 		return nil, err
 	}
-	if err := s.forceKickOff(ctx, req.UserID, req.PlatformID, mcontext.GetOperationID(ctx)); err != nil {
+	if err := s.forceKickOff(ctx, req.UserID, req.PlatformID); err != nil {
 		return nil, err
 	}
 	return &pbauth.ForceLogoutResp{}, nil
 }
 
-func (s *authServer) forceKickOff(ctx context.Context, userID string, platformID int32, operationID string) error {
+func (s *authServer) forceKickOff(ctx context.Context, userID string, platformID int32) error {
 	conns, err := s.RegisterCenter.GetConns(ctx, s.config.Share.RpcRegisterName.MessageGateway)
 	if err != nil {
 		return err
 	}
 	for _, v := range conns {
 		log.ZDebug(ctx, "forceKickOff", "conn", v.Target())
-	}
-	for _, v := range conns {
 		client := msggateway.NewMsgGatewayClient(v)
 		kickReq := &msggateway.KickUserOfflineReq{KickUserIDList: []string{userID}, PlatformID: platformID}
 		_, err := client.KickUserOffline(ctx, kickReq)
@@ -175,8 +173,24 @@ func (s *authServer) forceKickOff(ctx context.Context, userID string, platformID
 			log.ZError(ctx, "forceKickOff", err, "kickReq", kickReq)
 		}
 	}
+
+	m, err := s.authDatabase.GetTokensWithoutError(ctx, userID, int(platformID))
+	if err != nil && err != redis.Nil {
+		return err
+	}
+	for k := range m {
+		m[k] = constant.KickedToken
+		log.ZDebug(ctx, "set token map is ", "token map", m, "userID",
+			userID, "token", k)
+
+		err = s.authDatabase.SetTokenMapByUidPid(ctx, userID, int(platformID), m)
+		if err != nil {
+			return err
+		}
+	}
 	return nil
 }
+
 func (s *authServer) InvalidateToken(ctx context.Context, req *pbauth.InvalidateTokenReq) (*pbauth.InvalidateTokenResp, error) {
 	m, err := s.authDatabase.GetTokensWithoutError(ctx, req.UserID, int(req.PlatformID))
 	if err != nil && err != redis.Nil {

pkg/common/cmd/push.go

@@ -47,6 +47,7 @@ func NewPushRpcCmd() *PushRpcCmd {
 	ret.RootCmd = NewRootCmd(program.GetProcessName(), WithConfigMap(ret.configMap))
 	ret.ctx = context.WithValue(context.Background(), "version", config.Version)
 	ret.Command.RunE = func(cmd *cobra.Command, args []string) error {
+		ret.pushConfig.FcmConfigPath = ret.ConfigPath()
 		return ret.runE()
 	}
 	return ret

pkg/common/cmd/root.go

@@ -31,6 +31,11 @@ type RootCmd struct {
 	prometheusPort int
 	log            config.Log
 	index          int
+	configPath     string
+}
+
+func (r *RootCmd) ConfigPath() string {
+	return r.configPath
 }
 
 func (r *RootCmd) Index() int {
@@ -153,6 +158,7 @@ func (r *RootCmd) getFlag(cmd *cobra.Command) (string, int, error) {
 	if err != nil {
 		return "", 0, errs.Wrap(err)
 	}
+	r.configPath = configDirectory
 	index, err := cmd.Flags().GetInt(FlagTransferIndex)
 	if err != nil {
 		return "", 0, errs.Wrap(err)

pkg/common/config/config.go

@@ -202,7 +202,8 @@ type Push struct {
 		ChannelName  string `mapstructure:"channelName"`
 	} `mapstructure:"geTui"`
 	FCM struct {
-		ServiceAccount string `mapstructure:"serviceAccount"`
+		FilePath string `mapstructure:"filePath"`
+		AuthURL  string `mapstructure:"authURL"`
 	} `mapstructure:"fcm"`
 	JPNS struct {
 		AppKey       string `mapstructure:"appKey"`