Merge remote-tracking branch 'origin/v2.3.0release' into v2.3.0release

# Conflicts:
#	pkg/common/constant/constant.go

config/config.yaml

@@ -706,6 +706,7 @@ demo:
   superCode: 666666 #超级验证码，建议修改掉，收不到短信验证码时可以用此替代
   #  second
   codeTTL: 60
+  useSuperCode: true
   mail: #仅支持qq邮箱，具体操作参考 https://service.mail.qq.com/cgi-bin/help?subtype=1&id=28&no=1001256 必须修改
     title: "openIM"
     senderMail: "765567899@qq.com"

docker-compose-mongos.yaml

@@ -0,0 +1,139 @@
+version: "3"
+#fixme  Clone openIM Server project before using docker-compose,project address：https://github.com/OpenIMSDK/Open-IM-Server.git
+
+services:
+  mysql:
+    image: mysql:5.7
+    ports:
+      - 13306:3306
+      - 23306:33060
+    container_name: mysql
+    volumes:
+      - ./components/mysql/data:/var/lib/mysql
+      - /etc/localtime:/etc/localtime
+    environment:
+      MYSQL_ROOT_PASSWORD: openIM
+    restart: always
+
+  mongos:
+    image: mongo:latest
+    ports:
+      - 37017:27017
+    container_name: mongos
+    command: mongos --configdb config/103.116.45.174:27019 --bind_ip_all
+    volumes:
+      - ./components/mongos_mongodb/data/db:/data/db
+      - ./components/mongos_mongodb/data/logs:/data/logs
+      - ./components/mongos_mongodb/data/conf:/etc/mongo
+      - ./components/mongos_mongodb/data/configdb:/data/configdb
+    environment:
+      - TZ=Asia/Shanghai
+      # cache
+      - wiredTigerCacheSizeGB=1
+      #    environment:
+      #      - MONGO_INITDB_ROOT_USERNAME=openIM
+      #      - MONGO_INITDB_ROOT_PASSWORD=openIM
+
+
+      #TZ: Asia/Shanghai
+    restart: always
+
+
+  redis:
+    image: redis
+    ports:
+      - 16379:6379
+    container_name: redis
+    volumes:
+      - ./components/redis/data:/data
+      #redis config file
+      - ./components/redis/config/redis.conf:/usr/local/redis/config/redis.conf
+    environment:
+      TZ: Asia/Shanghai
+    restart: always
+    sysctls:
+      net.core.somaxconn: 1024
+    command: redis-server --requirepass openIM --appendonly yes
+
+
+  zookeeper:
+    image: wurstmeister/zookeeper
+    ports:
+      - 2181:2181
+    container_name: zookeeper
+    volumes:
+      - /etc/localtime:/etc/localtime
+    environment:
+      TZ: Asia/Shanghai
+    restart: always
+
+
+  kafka:
+    image: wurstmeister/kafka
+    container_name: kafka
+    restart: always
+    ports:
+      - 9092:9092
+    environment:
+      TZ: Asia/Shanghai
+      KAFKA_BROKER_ID: 0
+      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
+      KAFKA_CREATE_TOPICS: "ws2ms_chat:8:1,ms2ps_chat:8:1,msg_to_mongo:8:1"
+      KAFKA_ADVERTISED_LISTENERS: INSIDE://127.0.0.1:9092,OUTSIDE://103.116.45.174:9092
+      KAFKA_LISTENERS: INSIDE://:9092,OUTSIDE://:9093
+      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: "INSIDE:PLAINTEXT,OUTSIDE:PLAINTEXT"
+      KAFKA_INTER_BROKER_LISTENER_NAME: INSIDE
+    depends_on:
+      - zookeeper
+
+  etcd:
+    image: quay.io/coreos/etcd
+    ports:
+      - 2379:2379
+      - 2380:2380
+    container_name: etcd
+    volumes:
+      - /etc/timezone:/etc/timezone
+      - /etc/localtime:/etc/localtime
+    environment:
+      ETCDCTL_API: 3
+    restart: always
+    command: /usr/local/bin/etcd --name etcd0 --data-dir /etcd-data --listen-client-urls http://0.0.0.0:2379 --advertise-client-urls http://0.0.0.0:2379 --listen-peer-urls http://0.0.0.0:2380 --initial-advertise-peer-urls http://0.0.0.0:2380 --initial-cluster etcd0=http://0.0.0.0:2380 --initial-cluster-token tkn --initial-cluster-state new
+
+  minio:
+    image: minio/minio
+    ports:
+      - 10005:9000
+      - 9090:9090
+    container_name: minio
+    volumes:
+      - /mnt/data:/data
+      - /mnt/config:/root/.minio
+    environment:
+      MINIO_ROOT_USER: user12345
+      MINIO_ROOT_PASSWORD: key12345
+    restart: always
+    command: minio server /data --console-address ':9090'
+
+  open_im_server:
+    image: openim/open_im_server:v2.2.0
+    container_name: open_im_server
+    volumes:
+      - ./logs:/Open-IM-Server/logs
+      - ./config/config.yaml:/Open-IM-Server/config/config.yaml
+      - ./db/sdk:/Open-IM-Server/db/sdk
+      - ./script:/Open-IM-Server/script
+    restart: always
+    depends_on:
+      - kafka
+      - mysql
+      - mongos
+      - redis
+      - etcd
+      - minio
+    network_mode: "host"
+    logging:
+      driver: json-file
+      options:
+        max-size: "1g"
+        max-file: "2"

docker-compose-shard-cluster.yaml

@@ -0,0 +1,113 @@
+version: '3'
+networks:
+  mongo-network:
+    external: false
+services:
+  # 配置服务器configsvr
+  config:
+    image: mongo:latest
+    networks:
+      - mongo-network
+    container_name: config
+    restart: always
+    ports:
+      - 27019:27019
+    command: --configsvr --replSet "config" --bind_ip_all
+    volumes:
+      - ./components/config_mongodb/data/db:/data/db
+      - ./components/config_mongodb/data/logs:/data/logs
+      - ./components/config_mongodb/data/conf:/etc/mongo
+      - ./components/config_mongodb/data/configdb:/data/configdb
+    environment:
+      - TZ=Asia/Shanghai
+      # cache
+      - wiredTigerCacheSizeGB=1
+
+  # shard分片1
+  shard1:
+    image: mongo:latest
+    networks:
+      - mongo-network
+    container_name: shard1
+    restart: always
+    ports:
+      - 27118:27018
+    command: --shardsvr --replSet "shard1" --bind_ip_all
+    volumes:
+      - /data01/mongo/shard1/db:/data/db
+      - /data01/mongo/shard1/configdb:/data/configdb
+    environment:
+      - TZ=Asia/Shanghai
+      # cache
+      - wiredTigerCacheSizeGB=1
+
+  # shard分片2
+  shard2:
+    image: mongo:latest
+    networks:
+      - mongo-network
+    container_name: shard2
+    restart: always
+    ports:
+      - 27218:27018
+    command: --shardsvr --replSet "shard2" --bind_ip_all
+    volumes:
+      - /data02/mongo/shard2/db:/data/db
+      - /data02/mongo/shard2/configdb:/data/configdb
+    environment:
+      - TZ=Asia/Shanghai
+      # cache
+      - wiredTigerCacheSizeGB=1
+  #      - ${PWD}/key.file:/data/mongodb/key.file
+
+  # shard分片3
+  shard3:
+    image: mongo:latest
+    networks:
+      - mongo-network
+    container_name: shard3
+    restart: always
+    ports:
+      - 27318:27018
+    command: --shardsvr --replSet "shard3" --bind_ip_all
+    volumes:
+      - /data03/mongo/shard3/db:/data/db
+      - /data03/mongo/shard3/configdb:/data/configdb
+    environment:
+      - TZ=Asia/Shanghai
+      # cache
+      - wiredTigerCacheSizeGB=1
+  # shard分片4
+  shard4:
+    image: mongo:latest
+    networks:
+      - mongo-network
+    container_name: shard4
+    restart: always
+    ports:
+      - 27418:27018
+    command: --shardsvr --replSet "shard4" --bind_ip_all
+    volumes:
+      - /data04/mongo/shard4/db:/data/db
+      - /data04/mongo/shard4/configdb:/data/configdb
+    environment:
+      - TZ=Asia/Shanghai
+      # cache
+      - wiredTigerCacheSizeGB=1
+  # shard分片5
+  shard5:
+    image: mongo:latest
+    networks:
+      - mongo-network
+    container_name: shard5
+    restart: always
+    ports:
+      - 27518:27018
+    command: --shardsvr --replSet "shard5" --bind_ip_all
+    volumes:
+      - /data05/mongo/shard5/db:/data/db
+      - /data05/mongo/shard5/configdb:/data/configdb
+    environment:
+      - TZ=Asia/Shanghai
+      # cache
+      - wiredTigerCacheSizeGB=1

internal/demo/register/login.go

@@ -15,6 +15,7 @@ import (
 )
 
 type ParamsLogin struct {
+	UserID      string `json:"userID"`
 	Email       string `json:"email"`
 	PhoneNumber string `json:"phoneNumber"`
 	Password    string `json:"password"`
@@ -32,11 +33,13 @@ func Login(c *gin.Context) {
 	var account string
 	if params.Email != "" {
 		account = params.Email
-	} else {
+	} else if params.PhoneNumber != "" {
 		account = params.PhoneNumber
+	} else {
+		account = params.UserID
 	}
 
-	r, err := im_mysql_model.GetRegister(account, params.AreaCode)
+	r, err := im_mysql_model.GetRegister(account, params.AreaCode, params.UserID)
 	if err != nil {
 		log.NewError(params.OperationID, "user have not register", params.Password, account, err.Error())
 		c.JSON(http.StatusOK, gin.H{"errCode": constant.NotRegistered, "errMsg": "Mobile phone number is not registered"})

internal/demo/register/reset_password.go

@@ -34,7 +34,8 @@ func ResetPassword(c *gin.Context) {
 	} else {
 		account = req.PhoneNumber
 	}
-	if req.VerificationCode != config.Config.Demo.SuperCode {
+
+	if (config.Config.Demo.UseSuperCode && req.VerificationCode != config.Config.Demo.SuperCode) || !config.Config.Demo.UseSuperCode {
 		accountKey := req.AreaCode + account + "_" + constant.VerificationCodeForResetSuffix
 		v, err := db.DB.GetAccountCode(accountKey)
 		if err != nil || v != req.VerificationCode {
@@ -43,7 +44,7 @@ func ResetPassword(c *gin.Context) {
 			return
 		}
 	}
-	user, err := im_mysql_model.GetRegister(account, req.AreaCode)
+	user, err := im_mysql_model.GetRegister(account, req.AreaCode, "")
 	if err != nil || user.Account == "" {
 		if err != nil {
 			log.NewError(req.OperationID, utils.GetSelfFuncName(), "get register error", err.Error())

internal/demo/register/send_code.go

@@ -66,7 +66,7 @@ func SendVerificationCode(c *gin.Context) {
 	}
 	switch params.UsedFor {
 	case constant.VerificationCodeForRegister:
-		_, err := im_mysql_model.GetRegister(account, params.AreaCode)
+		_, err := im_mysql_model.GetRegister(account, params.AreaCode, "")
 		if err == nil {
 			log.NewError(params.OperationID, "The phone number has been registered", params)
 			c.JSON(http.StatusOK, gin.H{"errCode": constant.HasRegistered, "errMsg": "The phone number has been registered"})

internal/demo/register/set_password.go

@@ -18,6 +18,7 @@ import (
 )
 
 type ParamsSetPassword struct {
+	UserID           string `json:"userID"`
 	Email            string `json:"email"`
 	Nickname         string `json:"nickname"`
 	PhoneNumber      string `json:"phoneNumber"`
@@ -40,29 +41,37 @@ func SetPassword(c *gin.Context) {
 	var account string
 	if params.Email != "" {
 		account = params.Email
-	} else {
+	} else if params.PhoneNumber != "" {
 		account = params.PhoneNumber
+	} else {
+		account = params.UserID
 	}
 	if params.Nickname == "" {
 		params.Nickname = account
 	}
-	if params.VerificationCode != config.Config.Demo.SuperCode {
-		accountKey := params.AreaCode + account + "_" + constant.VerificationCodeForRegisterSuffix
-		v, err := db.DB.GetAccountCode(accountKey)
-		if err != nil || v != params.VerificationCode {
-			log.NewError(params.OperationID, "password Verification code error", account, params.VerificationCode)
-			data := make(map[string]interface{})
-			data["PhoneNumber"] = account
-			c.JSON(http.StatusOK, gin.H{"errCode": constant.CodeInvalidOrExpired, "errMsg": "Verification code error!", "data": data})
-			return
+	if params.UserID == "" {
+		if (config.Config.Demo.UseSuperCode && params.VerificationCode != config.Config.Demo.SuperCode) || !config.Config.Demo.UseSuperCode {
+			accountKey := params.AreaCode + account + "_" + constant.VerificationCodeForRegisterSuffix
+			v, err := db.DB.GetAccountCode(accountKey)
+			if err != nil || v != params.VerificationCode {
+				log.NewError(params.OperationID, "password Verification code error", account, params.VerificationCode)
+				data := make(map[string]interface{})
+				data["PhoneNumber"] = account
+				c.JSON(http.StatusOK, gin.H{"errCode": constant.CodeInvalidOrExpired, "errMsg": "Verification code error!", "data": data})
+				return
+			}
 		}
 	}
 	//userID := utils.Base64Encode(account)
-
-	userID := utils.Md5(params.OperationID + strconv.FormatInt(time.Now().UnixNano(), 10))
-	bi := big.NewInt(0)
-	bi.SetString(userID[0:8], 16)
-	userID = bi.String()
+	var userID string
+	if params.UserID == "" {
+		userID = utils.Md5(params.OperationID + strconv.FormatInt(time.Now().UnixNano(), 10))
+		bi := big.NewInt(0)
+		bi.SetString(userID[0:8], 16)
+		userID = bi.String()
+	} else {
+		userID = params.UserID
+	}
 
 	url := config.Config.Demo.ImAPIURL + "/auth/user_register"
 	openIMRegisterReq := api.UserRegisterReq{}
@@ -73,6 +82,7 @@ func SetPassword(c *gin.Context) {
 	openIMRegisterReq.Secret = config.Config.Secret
 	openIMRegisterReq.FaceURL = params.FaceURL
 	openIMRegisterResp := api.UserRegisterResp{}
+	log.NewDebug(params.OperationID, utils.GetSelfFuncName(), "register req:", openIMRegisterReq)
 	bMsg, err := http2.Post(url, openIMRegisterReq, 2)
 	if err != nil {
 		log.NewError(params.OperationID, "request openIM register error", account, "err", err.Error())
@@ -97,7 +107,9 @@ func SetPassword(c *gin.Context) {
 	}
 	log.Info(params.OperationID, "end setPassword", account, params.Password)
 	// demo onboarding
-	onboardingProcess(params.OperationID, userID, params.Nickname, params.FaceURL, params.AreaCode+params.PhoneNumber, params.Email)
+	if params.UserID == "" {
+		onboardingProcess(params.OperationID, userID, params.Nickname, params.FaceURL, params.AreaCode+params.PhoneNumber, params.Email)
+	}
 	c.JSON(http.StatusOK, gin.H{"errCode": constant.NoError, "errMsg": "", "data": openIMRegisterResp.UserToken})
 	return
 }

internal/demo/register/verify.go

@@ -40,7 +40,7 @@ func Verify(c *gin.Context) {
 		account = params.AreaCode + params.PhoneNumber
 	}
 
-	if params.VerificationCode == config.Config.Demo.SuperCode {
+	if config.Config.Demo.UseSuperCode && params.VerificationCode == config.Config.Demo.SuperCode {
 		log.NewInfo(params.OperationID, "Super Code Verified successfully", account)
 		data := make(map[string]interface{})
 		data["account"] = account

internal/push/logic/push_to_client.go

@@ -233,30 +233,32 @@ func MsgToSuperGroupUser(pushMsg *pbPush.PushMsgReq) {
 
 			}
 		}
+		if len(onlineFailedUserIDList) > 0 {
+			callbackResp := callbackOfflinePush(pushMsg.OperationID, onlineFailedUserIDList[0], pushMsg.MsgData)
+			log.NewDebug(pushMsg.OperationID, utils.GetSelfFuncName(), "offline callback Resp")
+			if callbackResp.ErrCode != 0 {
+				log.NewError(pushMsg.OperationID, utils.GetSelfFuncName(), "callbackOfflinePush result: ", callbackResp)
+			}
+			if callbackResp.ActionCode != constant.ActionAllow {
+				log.NewDebug(pushMsg.OperationID, utils.GetSelfFuncName(), "offlinePush stop")
+				return
+			}
+			if offlinePusher == nil {
+				return
+			}
+			opts, err := GetOfflinePushOpts(pushMsg)
+			if err != nil {
+				log.NewError(pushMsg.OperationID, utils.GetSelfFuncName(), "GetOfflinePushOpts failed", pushMsg, err.Error())
+			}
+			log.NewInfo(pushMsg.OperationID, utils.GetSelfFuncName(), onlineFailedUserIDList, content, jsonCustomContent, "opts:", opts)
+			pushResult, err := offlinePusher.Push(onlineFailedUserIDList, content, jsonCustomContent, pushMsg.OperationID, opts)
+			if err != nil {
+				log.NewError(pushMsg.OperationID, "offline push error", pushMsg.String(), err.Error())
+			} else {
+				log.NewDebug(pushMsg.OperationID, "offline push return result is ", pushResult, pushMsg.MsgData)
+			}
+		}
 
-		callbackResp := callbackOfflinePush(pushMsg.OperationID, onlineFailedUserIDList[0], pushMsg.MsgData)
-		log.NewDebug(pushMsg.OperationID, utils.GetSelfFuncName(), "offline callback Resp")
-		if callbackResp.ErrCode != 0 {
-			log.NewError(pushMsg.OperationID, utils.GetSelfFuncName(), "callbackOfflinePush result: ", callbackResp)
-		}
-		if callbackResp.ActionCode != constant.ActionAllow {
-			log.NewDebug(pushMsg.OperationID, utils.GetSelfFuncName(), "offlinePush stop")
-			return
-		}
-		if offlinePusher == nil {
-			return
-		}
-		opts, err := GetOfflinePushOpts(pushMsg)
-		if err != nil {
-			log.NewError(pushMsg.OperationID, utils.GetSelfFuncName(), "GetOfflinePushOpts failed", pushMsg, err.Error())
-		}
-		log.NewInfo(pushMsg.OperationID, utils.GetSelfFuncName(), onlineFailedUserIDList, content, jsonCustomContent, "opts:", opts)
-		pushResult, err := offlinePusher.Push(onlineFailedUserIDList, content, jsonCustomContent, pushMsg.OperationID, opts)
-		if err != nil {
-			log.NewError(pushMsg.OperationID, "offline push error", pushMsg.String(), err.Error())
-		} else {
-			log.NewDebug(pushMsg.OperationID, "offline push return result is ", pushResult, pushMsg.MsgData)
-		}
 	}
 }
 

internal/rpc/msg/group_notification.go

@@ -183,7 +183,16 @@ func groupNotification(contentType int32, m proto.Message, sendID, groupID, recv
 	n.SendID = sendID
 	if groupID != "" {
 		n.RecvID = groupID
-		n.SessionType = constant.GroupChatType
+		group, err := imdb.GetGroupInfoByGroupID(groupID)
+		if err != nil {
+			log.NewError(operationID, "GetGroupInfoByGroupID failed ", err.Error(), groupID)
+		}
+		switch group.GroupType {
+		case constant.NormalGroup:
+			n.SessionType = constant.GroupChatType
+		default:
+			n.SessionType = constant.SuperGroupChatType
+		}
 	} else {
 		n.RecvID = recvUserID
 		n.SessionType = constant.SingleChatType

internal/rpc/msg/send_msg.go

@@ -331,8 +331,8 @@ func (rpc *rpcChat) SendMsg(_ context.Context, pb *pbChat.SendMsgReq) (*pbChat.S
 			}
 		}
 		log.Debug(pb.OperationID, "send msg cost time22 ", db.GetCurrentTimestampByMill()-newTime, pb.MsgData.ClientMsgID, "uidList : ", len(addUidList))
-		wg.Add(1)
-		go rpc.sendMsgToGroup(addUidList, *pb, constant.OnlineStatus, &sendTag, &wg)
+		//wg.Add(1)
+		//go rpc.sendMsgToGroup(addUidList, *pb, constant.OnlineStatus, &sendTag, &wg)
 		wg.Wait()
 		newTime = db.GetCurrentTimestampByMill()
 		// callback

pkg/common/config/config.go

@@ -464,9 +464,10 @@ type config struct {
 			VerificationCodeTemplateCode string `yaml:"verificationCodeTemplateCode"`
 			Enable                       bool   `yaml:"enable"`
 		}
-		SuperCode string `yaml:"superCode"`
-		CodeTTL   int    `yaml:"codeTTL"`
-		Mail      struct {
+		SuperCode    string `yaml:"superCode"`
+		CodeTTL      int    `yaml:"codeTTL"`
+		UseSuperCode bool   `yaml:"useSuperCode"`
+		Mail         struct {
 			Title                   string `yaml:"title"`
 			SenderMail              string `yaml:"senderMail"`
 			SenderAuthorizationCode string `yaml:"senderAuthorizationCode"`

pkg/common/constant/constant.go

@@ -164,7 +164,6 @@ const (
 	NormalGroup  = 0
 	SuperGroup   = 1
 	WorkingGroup = 2
-	//	DepartmentGroup = 2
 
 	GroupBaned          = 3
 	GroupBanPrivateChat = 4

pkg/common/db/mysql_model/im_mysql_model/demo_model.go

@@ -5,14 +5,14 @@ import (
 	_ "github.com/jinzhu/gorm"
 )
 
-func GetRegister(account, areaCode string) (*db.Register, error) {
+func GetRegister(account, areaCode, userID string) (*db.Register, error) {
 	dbConn, err := db.DB.MysqlDB.DefaultGormDB()
 	if err != nil {
 		return nil, err
 	}
 	var r db.Register
-	return &r, dbConn.Table("registers").Where("account = ? or account =? and area_code=?",
-		account, account, areaCode).Take(&r).Error
+	return &r, dbConn.Table("registers").Where("user_id = ? and user_id != ? or account = ? or account =? and area_code=?",
+		userID, "", account, account, areaCode).Take(&r).Error
 }
 
 func SetPassword(account, password, ex, userID, areaCode string) error {

pkg/common/db/mysql_model/im_mysql_model/organization_model.go

@@ -302,6 +302,6 @@ func GetRandomDepartmentID() (string, error) {
 		return "", err
 	}
 	department := &db.Department{}
-	err = dbConn.Model(department).Order("RAND()").Where("related_group_id != ? AND department_id != ?", "", "0").First(department).Error
+	err = dbConn.Model(department).Order("RAND()").Where("related_group_id != ? AND department_id != ? AND department_type = ?", "", "0", 1).First(department).Error
 	return department.DepartmentID, err
 }

pkg/common/db/redisModel_test.go

@@ -119,3 +119,9 @@ func Test_GetAccountCode(t *testing.T) {
 	assert.Nil(t, err)
 	fmt.Println(code)
 }
+func Test_GetGroupMemberList(t *testing.T) {
+	groupID := "3791742301"
+	list, err := DB.GetGroupMemberIDListFromCache(groupID)
+	assert.Nil(t, err)
+	fmt.Println(list)
+}