Merge branch 'openimsdk:main' into main

pull/1651/head
chao 2 years ago committed by GitHub
commit 5fd85aedb0
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23

@ -312,6 +312,14 @@ callback:
enable: false
timeout: 5
failedContinue: true
beforeUpdateUserInfoEx:
enable: false
timeout: 5
failedContinue: true
afterUpdateUserInfoEx:
enable: false
timeout: 5
failedContinue: true
afterSendSingleMsg:
enable: false
timeout: 5

@ -4,7 +4,7 @@ go 1.19
require (
firebase.google.com/go v3.13.0+incompatible
github.com/OpenIMSDK/protocol v0.0.39
github.com/OpenIMSDK/protocol v0.0.40
github.com/OpenIMSDK/tools v0.0.21
github.com/bwmarrin/snowflake v0.3.0 // indirect
github.com/dtm-labs/rockscache v0.1.1
@ -156,3 +156,5 @@ require (
golang.org/x/crypto v0.14.0 // indirect
gopkg.in/ini.v1 v1.67.0 // indirect
)
replace github.com/OpenIMSDK/protocol v0.0.40 => github.com/luhaoling/protocol v0.0.0-20231227040641-2f934a0d64a3

@ -18,8 +18,8 @@ firebase.google.com/go v3.13.0+incompatible/go.mod h1:xlah6XbEyW6tbfSklcfe5FHJIw
github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
github.com/IBM/sarama v1.41.3 h1:MWBEJ12vHC8coMjdEXFq/6ftO6DUZnQlFYcxtOJFa7c=
github.com/IBM/sarama v1.41.3/go.mod h1:Xxho9HkHd4K/MDUo/T/sOqwtX/17D33++E9Wib6hUdQ=
github.com/OpenIMSDK/protocol v0.0.39 h1:DfvFcNGBcfj2vtT7W3uw4U/ipnI7NecTzQdlSYGuQz8=
github.com/OpenIMSDK/protocol v0.0.39/go.mod h1:F25dFrwrIx3lkNoiuf6FkCfxuwf8L4Z8UIsdTHP/r0Y=
github.com/OpenIMSDK/protocol v0.0.40 h1:1/Oij6RSAaePCPrWGwp9Cyz976/8Uxr94hM5M5FXzlg=
github.com/OpenIMSDK/protocol v0.0.40/go.mod h1:F25dFrwrIx3lkNoiuf6FkCfxuwf8L4Z8UIsdTHP/r0Y=
github.com/OpenIMSDK/tools v0.0.21 h1:iTapc2mIEVH/xl5Nd6jfwPub11Pgp44tVcE1rjB3a48=
github.com/OpenIMSDK/tools v0.0.21/go.mod h1:eg+q4A34Qmu73xkY0mt37FHGMCMfC6CtmOnm0kFEGFI=
github.com/QcloudApi/qcloud_sign_golang v0.0.0-20141224014652-e4130a326409/go.mod h1:1pk82RBxDY/JZnPQrtqHlUFfCctgdorsd9M06fMynOM=
@ -227,6 +227,8 @@ github.com/lestrrat-go/strftime v1.0.6 h1:CFGsDEt1pOpFNU+TJB0nhz9jl+K0hZSLE205Ah
github.com/lestrrat-go/strftime v1.0.6/go.mod h1:f7jQKgV5nnJpYgdEasS+/y7EsTb8ykN2z68n3TtcTaw=
github.com/lithammer/shortuuid v3.0.0+incompatible h1:NcD0xWW/MZYXEHa6ITy6kaXN5nwm/V115vj2YXfhS0w=
github.com/lithammer/shortuuid v3.0.0+incompatible/go.mod h1:FR74pbAuElzOUuenUHTK2Tciko1/vKuIKS9dSkDrA4w=
github.com/luhaoling/protocol v0.0.0-20231227040641-2f934a0d64a3 h1:HZz2U/M3T4x9SqPxWdrD9MZy7jxx7nS+nx/aRN9m3RQ=
github.com/luhaoling/protocol v0.0.0-20231227040641-2f934a0d64a3/go.mod h1:F25dFrwrIx3lkNoiuf6FkCfxuwf8L4Z8UIsdTHP/r0Y=
github.com/mattn/go-isatty v0.0.19 h1:JITubQf0MOLdlGRuRq+jtsDlekdYPia9ZFsB8h/APPA=
github.com/mattn/go-isatty v0.0.19/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y=
github.com/matttproud/golang_protobuf_extensions v1.0.4 h1:mmDVorXM7PCGKw94cs5zkfA9PSy5pEvNWRP0ET0TIVo=

@ -92,3 +92,6 @@ func (o *FriendApi) GetFriendIDs(c *gin.Context) {
func (o *FriendApi) GetSpecifiedFriendsInfo(c *gin.Context) {
a2r.Call(friend.FriendClient.GetSpecifiedFriendsInfo, o.Client, c)
}
func (o *FriendApi) UpdateFriends(c *gin.Context) {
a2r.Call(friend.FriendClient.UpdateFriends, o.Client, c)
}

@ -172,6 +172,7 @@ func (m *MessageApi) getSendMsgReq(c *gin.Context, req apistruct.SendMsg) (sendM
if err = m.userRpcClient.GetNotificationByID(c, req.SendID); err != nil {
return nil, err
}
default:
return nil, errs.ErrArgs.WithDetail("not support err contentType")
}
@ -185,38 +186,63 @@ func (m *MessageApi) getSendMsgReq(c *gin.Context, req apistruct.SendMsg) (sendM
return m.newUserSendMsgReq(c, &req), nil
}
// SendMessage handles the sending of a message. It's an HTTP handler function to be used with Gin framework.
func (m *MessageApi) SendMessage(c *gin.Context) {
// Initialize a request struct for sending a message.
req := apistruct.SendMsgReq{}
// Bind the JSON request body to the request struct.
if err := c.BindJSON(&req); err != nil {
// Respond with an error if request body binding fails.
apiresp.GinError(c, errs.ErrArgs.WithDetail(err.Error()).Wrap())
return
}
// Check if the user has the app manager role.
if !authverify.IsAppManagerUid(c) {
// Respond with a permission error if the user is not an app manager.
apiresp.GinError(c, errs.ErrNoPermission.Wrap("only app manager can send message"))
return
}
// Prepare the message request with additional required data.
sendMsgReq, err := m.getSendMsgReq(c, req.SendMsg)
if err != nil {
// Log and respond with an error if preparation fails.
log.ZError(c, "decodeData failed", err)
apiresp.GinError(c, err)
return
}
// Set the receiver ID in the message data.
sendMsgReq.MsgData.RecvID = req.RecvID
// Declare a variable to store the message sending status.
var status int
// Attempt to send the message using the client.
respPb, err := m.Client.SendMsg(c, sendMsgReq)
if err != nil {
// Set the status to failed and respond with an error if sending fails.
status = constant.MsgSendFailed
log.ZError(c, "send message err", err)
apiresp.GinError(c, err)
return
}
// Set the status to successful if the message is sent.
status = constant.MsgSendSuccessed
// Attempt to update the message sending status in the system.
_, err = m.Client.SetSendMsgStatus(c, &msg.SetSendMsgStatusReq{
Status: int32(status),
})
if err != nil {
// Log the error if updating the status fails.
log.ZError(c, "SetSendMsgStatus failed", err)
}
// Respond with a success message and the response payload.
apiresp.GinSuccess(c, respPb)
}

@ -67,6 +67,7 @@ func NewGinRouter(discov discoveryregistry.SvcDiscoveryRegistry, rdb redis.Unive
{
userRouterGroup.POST("/user_register", u.UserRegister)
userRouterGroup.POST("/update_user_info", ParseToken, u.UpdateUserInfo)
userRouterGroup.POST("/update_user_info_ex", ParseToken, u.UpdateUserInfoEx)
userRouterGroup.POST("/set_global_msg_recv_opt", ParseToken, u.SetGlobalRecvMessageOpt)
userRouterGroup.POST("/get_users_info", ParseToken, u.GetUsersPublicInfo)
userRouterGroup.POST("/get_all_users_uid", ParseToken, u.GetAllUsersID)
@ -107,7 +108,7 @@ func NewGinRouter(discov discoveryregistry.SvcDiscoveryRegistry, rdb redis.Unive
friendRouterGroup.POST("/is_friend", f.IsFriend)
friendRouterGroup.POST("/get_friend_id", f.GetFriendIDs)
friendRouterGroup.POST("/get_specified_friends_info", f.GetSpecifiedFriendsInfo)
//friendRouterGroup.POST("/set_pin_friend", f.SetPinFriends)
friendRouterGroup.POST("/update_friends", f.UpdateFriends)
}
g := NewGroupApi(*groupRpc)
groupRouterGroup := r.Group("/group", ParseToken)

@ -41,7 +41,9 @@ func (u *UserApi) UserRegister(c *gin.Context) {
func (u *UserApi) UpdateUserInfo(c *gin.Context) {
a2r.Call(user.UserClient.UpdateUserInfo, u.Client, c)
}
func (u *UserApi) UpdateUserInfoEx(c *gin.Context) {
a2r.Call(user.UserClient.UpdateUserInfoEx, u.Client, c)
}
func (u *UserApi) SetGlobalRecvMessageOpt(c *gin.Context) {
a2r.Call(user.UserClient.SetGlobalRecvMessageOpt, u.Client, c)
}

@ -87,6 +87,7 @@ func newClient(ctx *UserConnContext, conn LongConn, isCompress bool) *Client {
}
}
// ResetClient updates the client's state with new connection and context information.
func (c *Client) ResetClient(
ctx *UserConnContext,
conn LongConn,
@ -108,11 +109,13 @@ func (c *Client) ResetClient(
c.token = token
}
// pingHandler handles ping messages and sends pong responses.
func (c *Client) pingHandler(_ string) error {
_ = c.conn.SetReadDeadline(pongWait)
return c.writePongMsg()
}
// readMessage continuously reads messages from the connection.
func (c *Client) readMessage() {
defer func() {
if r := recover(); r != nil {
@ -164,6 +167,7 @@ func (c *Client) readMessage() {
}
}
// handleMessage processes a single message received by the client.
func (c *Client) handleMessage(message []byte) error {
if c.IsCompress {
var err error

@ -53,10 +53,6 @@ type friendServer struct {
RegisterCenter registry.SvcDiscoveryRegistry
}
func (s *friendServer) UpdateFriends(ctx context.Context, req *pbfriend.UpdateFriendsReq) (*pbfriend.UpdateFriendsResp, error) {
return nil, errs.ErrInternalServer.Wrap("not implemented")
}
func Start(client registry.SvcDiscoveryRegistry, server *grpc.Server) error {
// Initialize MongoDB
mongo, err := unrelation.NewMongo()
@ -440,7 +436,7 @@ func (s *friendServer) GetSpecifiedFriendsInfo(ctx context.Context, req *pbfrien
}
return resp, nil
}
func (s *friendServer) PinFriends(
func (s *friendServer) UpdateFriends(
ctx context.Context,
req *pbfriend.UpdateFriendsReq,
) (*pbfriend.UpdateFriendsResp, error) {
@ -450,25 +446,35 @@ func (s *friendServer) PinFriends(
if utils.Duplicate(req.FriendUserIDs) {
return nil, errs.ErrArgs.Wrap("friendIDList repeated")
}
var isPinned bool
if req.IsPinned != nil {
isPinned = req.IsPinned.Value
} else {
return nil, errs.ErrArgs.Wrap("isPinned is nil")
}
//check whther in friend list
_, err := s.friendDatabase.FindFriendsWithError(ctx, req.OwnerUserID, req.FriendUserIDs)
if err != nil {
return nil, err
}
//set friendslist friend pin status to isPinned
for _, friendID := range req.FriendUserIDs {
if err := s.friendDatabase.UpdateFriendPinStatus(ctx, req.OwnerUserID, friendID, isPinned); err != nil {
return nil, err
if req.IsPinned != nil {
if err = s.friendDatabase.UpdateFriendPinStatus(ctx, req.OwnerUserID, friendID, req.IsPinned.Value); err != nil {
return nil, err
}
}
if req.Remark != nil {
if err = s.friendDatabase.UpdateFriendRemark(ctx, req.OwnerUserID, friendID, req.Remark.Value); err != nil {
return nil, err
}
}
if req.Ex != nil {
if err = s.friendDatabase.UpdateFriendEx(ctx, req.OwnerUserID, friendID, req.Ex.Value); err != nil {
return nil, err
}
}
}
resp := &pbfriend.UpdateFriendsResp{}
err = s.notificationSender.FriendsInfoUpdateNotification(ctx, req.OwnerUserID, req.FriendUserIDs)
if err != nil {
return nil, errs.Wrap(err, "FriendsInfoUpdateNotification Error")
}
return resp, nil
}

@ -100,7 +100,7 @@ func callbackAfterSendSingleMsg(ctx context.Context, msg *pbchat.SendMsgReq) err
}
func callbackBeforeSendGroupMsg(ctx context.Context, msg *pbchat.SendMsgReq) error {
if !config.Config.Callback.CallbackBeforeSendSingleMsg.Enable {
if !config.Config.Callback.CallbackBeforeSendGroupMsg.Enable {
return nil
}
req := &cbapi.CallbackBeforeSendGroupMsgReq{

@ -44,7 +44,6 @@ func CallbackBeforeUpdateUserInfo(ctx context.Context, req *pbuser.UpdateUserInf
utils.NotNilReplace(&req.UserInfo.Nickname, resp.Nickname)
return nil
}
func CallbackAfterUpdateUserInfo(ctx context.Context, req *pbuser.UpdateUserInfoReq) error {
if !config.Config.Callback.CallbackAfterUpdateUserInfo.Enable {
return nil
@ -61,6 +60,41 @@ func CallbackAfterUpdateUserInfo(ctx context.Context, req *pbuser.UpdateUserInfo
}
return nil
}
func CallbackBeforeUpdateUserInfoEx(ctx context.Context, req *pbuser.UpdateUserInfoExReq) error {
if !config.Config.Callback.CallbackBeforeUpdateUserInfoEx.Enable {
return nil
}
cbReq := &cbapi.CallbackBeforeUpdateUserInfoExReq{
CallbackCommand: cbapi.CallbackBeforeUpdateUserInfoExCommand,
UserID: req.UserInfo.UserID,
FaceURL: &req.UserInfo.FaceURL,
Nickname: &req.UserInfo.Nickname,
}
resp := &cbapi.CallbackBeforeUpdateUserInfoExResp{}
if err := http.CallBackPostReturn(ctx, config.Config.Callback.CallbackUrl, cbReq, resp, config.Config.Callback.CallbackBeforeUpdateUserInfoEx); err != nil {
return err
}
utils.NotNilReplace(&req.UserInfo.FaceURL, resp.FaceURL)
utils.NotNilReplace(req.UserInfo.Ex, resp.Ex)
utils.NotNilReplace(&req.UserInfo.Nickname, resp.Nickname)
return nil
}
func CallbackAfterUpdateUserInfoEx(ctx context.Context, req *pbuser.UpdateUserInfoExReq) error {
if !config.Config.Callback.CallbackAfterUpdateUserInfoEx.Enable {
return nil
}
cbReq := &cbapi.CallbackAfterUpdateUserInfoExReq{
CallbackCommand: cbapi.CallbackAfterUpdateUserInfoExCommand,
UserID: req.UserInfo.UserID,
FaceURL: req.UserInfo.FaceURL,
Nickname: req.UserInfo.Nickname,
}
resp := &cbapi.CallbackAfterUpdateUserInfoExResp{}
if err := http.CallBackPostReturn(ctx, config.Config.Callback.CallbackUrl, cbReq, resp, config.Config.Callback.CallbackBeforeUpdateUserInfoEx); err != nil {
return err
}
return nil
}
func CallbackBeforeUserRegister(ctx context.Context, req *pbuser.UserRegisterReq) error {
if !config.Config.Callback.CallbackBeforeUserRegister.Enable {

@ -17,6 +17,7 @@ package user
import (
"context"
"errors"
"github.com/openimsdk/open-im-server/v3/pkg/common/db/table/relation"
"math/rand"
"strings"
"time"
@ -57,10 +58,6 @@ type userServer struct {
RegisterCenter registry.SvcDiscoveryRegistry
}
func (s *userServer) UpdateUserInfoEx(ctx context.Context, req *pbuser.UpdateUserInfoExReq) (*pbuser.UpdateUserInfoExResp, error) {
return nil, errs.ErrInternalServer.Wrap("not implemented")
}
func Start(client registry.SvcDiscoveryRegistry, server *grpc.Server) error {
rdb, err := cache.NewRedis()
if err != nil {
@ -152,7 +149,41 @@ func (s *userServer) UpdateUserInfo(ctx context.Context, req *pbuser.UpdateUserI
}
return resp, nil
}
func (s *userServer) UpdateUserInfoEx(ctx context.Context, req *pbuser.UpdateUserInfoExReq) (resp *pbuser.UpdateUserInfoExResp, err error) {
resp = &pbuser.UpdateUserInfoExResp{}
err = authverify.CheckAccessV3(ctx, req.UserInfo.UserID)
if err != nil {
return nil, err
}
if err = CallbackBeforeUpdateUserInfoEx(ctx, req); err != nil {
return nil, err
}
data := convert.UserPb2DBMapEx(req.UserInfo)
if err = s.UpdateByMap(ctx, req.UserInfo.UserID, data); err != nil {
return nil, err
}
_ = s.friendNotificationSender.UserInfoUpdatedNotification(ctx, req.UserInfo.UserID)
friends, err := s.friendRpcClient.GetFriendIDs(ctx, req.UserInfo.UserID)
if err != nil {
return nil, err
}
if req.UserInfo.Nickname != "" || req.UserInfo.FaceURL != "" {
if err := s.groupRpcClient.NotificationUserInfoUpdate(ctx, req.UserInfo.UserID); err != nil {
log.ZError(ctx, "NotificationUserInfoUpdate", err)
}
}
for _, friendID := range friends {
s.friendNotificationSender.FriendInfoUpdatedNotification(ctx, req.UserInfo.UserID, friendID)
}
if err := CallbackAfterUpdateUserInfoEx(ctx, req); err != nil {
return nil, err
}
if err := s.groupRpcClient.NotificationUserInfoUpdate(ctx, req.UserInfo.UserID); err != nil {
log.ZError(ctx, "NotificationUserInfoUpdate", err, "userID", req.UserInfo.UserID)
}
return resp, nil
}
func (s *userServer) SetGlobalRecvMessageOpt(ctx context.Context, req *pbuser.SetGlobalRecvMessageOptReq) (resp *pbuser.SetGlobalRecvMessageOptResp, err error) {
resp = &pbuser.SetGlobalRecvMessageOptResp{}
if _, err := s.FindWithError(ctx, []string{req.UserID}); err != nil {
@ -466,26 +497,31 @@ func (s *userServer) SearchNotificationAccount(ctx context.Context, req *pbuser.
return nil, err
}
if req.NickName != "" {
users, err := s.UserDatabase.FindByNickname(ctx, req.NickName)
if err != nil {
return nil, err
}
resp := s.userModelToResp(users)
return resp, nil
}
if req.UserID != "" {
users, err := s.UserDatabase.Find(ctx, []string{req.UserID})
if err != nil {
return nil, err
}
resp := s.userModelToResp(users)
return resp, nil
}
_, users, err := s.UserDatabase.Page(ctx, req.Pagination)
if err != nil {
return nil, err
}
var total int64
accounts := make([]*pbuser.NotificationAccountInfo, 0, len(users))
for _, v := range users {
if v.AppMangerLevel != constant.AppNotificationAdmin {
continue
}
temp := &pbuser.NotificationAccountInfo{
UserID: v.UserID,
FaceURL: v.FaceURL,
NickName: v.Nickname,
}
accounts = append(accounts, temp)
total += 1
}
return &pbuser.SearchNotificationAccountResp{Total: total, NotificationAccounts: accounts}, nil
resp := s.userModelToResp(users)
return resp, nil
}
func (s *userServer) GetNotificationAccount(ctx context.Context, req *pbuser.GetNotificationAccountReq) (*pbuser.GetNotificationAccountResp, error) {
@ -517,3 +553,20 @@ func (s *userServer) genUserID() string {
}
return string(data)
}
func (s *userServer) userModelToResp(users []*relation.UserModel) *pbuser.SearchNotificationAccountResp {
accounts := make([]*pbuser.NotificationAccountInfo, 0)
var total int64
for _, v := range users {
if v.AppMangerLevel == constant.AppNotificationAdmin || v.AppMangerLevel == constant.AppAdmin {
temp := &pbuser.NotificationAccountInfo{
UserID: v.UserID,
FaceURL: v.FaceURL,
NickName: v.Nickname,
}
accounts = append(accounts, temp)
total += 1
}
}
return &pbuser.SearchNotificationAccountResp{Total: total, NotificationAccounts: accounts}
}

@ -37,6 +37,8 @@ const (
CallbackGroupMsgReadCommand = "callbackGroupMsgReadCommand"
CallbackMsgModifyCommand = "callbackMsgModifyCommand"
CallbackAfterUpdateUserInfoCommand = "callbackAfterUpdateUserInfoCommand"
CallbackAfterUpdateUserInfoExCommand = "callbackAfterUpdateUserInfoExCommand"
CallbackBeforeUpdateUserInfoExCommand = "callbackBeforeUpdateUserInfoExCommand"
CallbackBeforeUserRegisterCommand = "callbackBeforeUserRegisterCommand"
CallbackAfterUserRegisterCommand = "callbackAfterUserRegisterCommand"
CallbackTransferGroupOwnerAfter = "callbackTransferGroupOwnerAfter"

@ -14,7 +14,10 @@
package callbackstruct
import "github.com/OpenIMSDK/protocol/sdkws"
import (
"github.com/OpenIMSDK/protocol/sdkws"
"github.com/OpenIMSDK/protocol/wrapperspb"
)
type CallbackBeforeUpdateUserInfoReq struct {
CallbackCommand `json:"callbackCommand"`
@ -41,6 +44,31 @@ type CallbackAfterUpdateUserInfoResp struct {
CommonCallbackResp
}
type CallbackBeforeUpdateUserInfoExReq struct {
CallbackCommand `json:"callbackCommand"`
UserID string `json:"userID"`
Nickname *string `json:"nickName"`
FaceURL *string `json:"faceURL"`
Ex *wrapperspb.StringValue `json:"ex"`
}
type CallbackBeforeUpdateUserInfoExResp struct {
CommonCallbackResp
Nickname *string `json:"nickName"`
FaceURL *string `json:"faceURL"`
Ex *wrapperspb.StringValue `json:"ex"`
}
type CallbackAfterUpdateUserInfoExReq struct {
CallbackCommand `json:"callbackCommand"`
UserID string `json:"userID"`
Nickname string `json:"nickName"`
FaceURL string `json:"faceURL"`
Ex *wrapperspb.StringValue `json:"ex"`
}
type CallbackAfterUpdateUserInfoExResp struct {
CommonCallbackResp
}
type CallbackBeforeUserRegisterReq struct {
CallbackCommand `json:"callbackCommand"`
Secret string `json:"secret"`

@ -282,6 +282,8 @@ type configStruct struct {
CallbackBeforeSetFriendRemark CallBackConfig `yaml:"callbackBeforeSetFriendRemark"`
CallbackAfterSetFriendRemark CallBackConfig `yaml:"callbackAfterSetFriendRemark"`
CallbackBeforeUpdateUserInfo CallBackConfig `yaml:"beforeUpdateUserInfo"`
CallbackBeforeUpdateUserInfoEx CallBackConfig `yaml:"beforeUpdateUserInfoEx"`
CallbackAfterUpdateUserInfoEx CallBackConfig `yaml:"afterUpdateUserInfoEx"`
CallbackBeforeUserRegister CallBackConfig `yaml:"beforeUserRegister"`
CallbackAfterUpdateUserInfo CallBackConfig `yaml:"updateUserInfo"`
CallbackAfterUserRegister CallBackConfig `yaml:"afterUserRegister"`

@ -64,7 +64,7 @@ func UserPb2DBMap(user *sdkws.UserInfo) map[string]any {
"global_recv_msg_opt": user.GlobalRecvMsgOpt,
}
for key, value := range fields {
if v, ok := value.(string); ok {
if v, ok := value.(string); ok && v != "" {
val[key] = v
} else if v, ok := value.(int32); ok && v != 0 {
val[key] = v
@ -72,3 +72,22 @@ func UserPb2DBMap(user *sdkws.UserInfo) map[string]any {
}
return val
}
func UserPb2DBMapEx(user *sdkws.UserInfoWithEx) map[string]any {
if user == nil {
return nil
}
val := make(map[string]any)
// Map fields from UserInfoWithEx to val
val["nickname"] = user.Nickname
val["face_url"] = user.FaceURL
if user.Ex != nil {
val["ex"] = user.Ex.Value
}
if user.GlobalRecvMsgOpt != 0 {
val["global_recv_msg_opt"] = user.GlobalRecvMsgOpt
}
return val
}

@ -87,7 +87,15 @@ func NewRedis() (redis.UniversalClient, error) {
// overrideConfigFromEnv overrides configuration fields with environment variables if present.
func overrideConfigFromEnv() {
if envAddr := os.Getenv("REDIS_ADDRESS"); envAddr != "" {
config.Config.Redis.Address = strings.Split(envAddr, ",") // Assuming addresses are comma-separated
if envPort := os.Getenv("REDIS_PORT"); envPort != "" {
addresses := strings.Split(envAddr, ",")
for i, addr := range addresses {
addresses[i] = addr + ":" + envPort
}
config.Config.Redis.Address = addresses
} else {
config.Config.Redis.Address = strings.Split(envAddr, ",")
}
}
if envUser := os.Getenv("REDIS_USERNAME"); envUser != "" {
config.Config.Redis.Username = envUser

@ -32,33 +32,57 @@ import (
)
type FriendDatabase interface {
// 检查user2是否在user1的好友列表中(inUser1Friends==true) 检查user1是否在user2的好友列表中(inUser2Friends==true)
// CheckIn checks if user2 is in user1's friend list (inUser1Friends==true) and if user1 is in user2's friend list (inUser2Friends==true)
CheckIn(ctx context.Context, user1, user2 string) (inUser1Friends bool, inUser2Friends bool, err error)
// 增加或者更新好友申请
// AddFriendRequest adds or updates a friend request
AddFriendRequest(ctx context.Context, fromUserID, toUserID string, reqMsg string, ex string) (err error)
// 先判断是否在好友表,如果在则不插入
// BecomeFriends first checks if the users are already in the friends table; if not, it inserts them as friends
BecomeFriends(ctx context.Context, ownerUserID string, friendUserIDs []string, addSource int32) (err error)
// 拒绝好友申请
// RefuseFriendRequest refuses a friend request
RefuseFriendRequest(ctx context.Context, friendRequest *relation.FriendRequestModel) (err error)
// 同意好友申请
// AgreeFriendRequest accepts a friend request
AgreeFriendRequest(ctx context.Context, friendRequest *relation.FriendRequestModel) (err error)
// 删除好友
// Delete removes a friend or friends from the owner's friend list
Delete(ctx context.Context, ownerUserID string, friendUserIDs []string) (err error)
// 更新好友备注
// UpdateRemark updates the remark for a friend
UpdateRemark(ctx context.Context, ownerUserID, friendUserID, remark string) (err error)
// 获取ownerUserID的好友列表
// PageOwnerFriends retrieves the friend list of ownerUserID with pagination
PageOwnerFriends(ctx context.Context, ownerUserID string, pagination pagination.Pagination) (total int64, friends []*relation.FriendModel, err error)
// friendUserID在哪些人的好友列表中
// PageInWhoseFriends finds the users who have friendUserID in their friend list with pagination
PageInWhoseFriends(ctx context.Context, friendUserID string, pagination pagination.Pagination) (total int64, friends []*relation.FriendModel, err error)
// 获取我发出去的好友申请
// PageFriendRequestFromMe retrieves the friend requests sent by the user with pagination
PageFriendRequestFromMe(ctx context.Context, userID string, pagination pagination.Pagination) (total int64, friends []*relation.FriendRequestModel, err error)
// 获取我收到的的好友申请
// PageFriendRequestToMe retrieves the friend requests received by the user with pagination
PageFriendRequestToMe(ctx context.Context, userID string, pagination pagination.Pagination) (total int64, friends []*relation.FriendRequestModel, err error)
// 获取某人指定好友的信息
// FindFriendsWithError fetches specified friends of a user and returns an error if any do not exist
FindFriendsWithError(ctx context.Context, ownerUserID string, friendUserIDs []string) (friends []*relation.FriendModel, err error)
// FindFriendUserIDs retrieves the friend IDs of a user
FindFriendUserIDs(ctx context.Context, ownerUserID string) (friendUserIDs []string, err error)
// FindBothFriendRequests finds friend requests sent and received
FindBothFriendRequests(ctx context.Context, fromUserID, toUserID string) (friends []*relation.FriendRequestModel, err error)
// UpdateFriendPinStatus updates the pinned status of a friend
UpdateFriendPinStatus(ctx context.Context, ownerUserID string, friendUserID string, isPinned bool) (err error)
// UpdateFriendRemark updates the remark for a friend
UpdateFriendRemark(ctx context.Context, ownerUserID string, friendUserID string, remark string) (err error)
// UpdateFriendEx updates the 'ex' field for a friend
UpdateFriendEx(ctx context.Context, ownerUserID string, friendUserID string, ex string) (err error)
}
type friendDatabase struct {
@ -305,3 +329,15 @@ func (f *friendDatabase) UpdateFriendPinStatus(ctx context.Context, ownerUserID
}
return f.cache.DelFriend(ownerUserID, friendUserID).ExecDel(ctx)
}
func (f *friendDatabase) UpdateFriendRemark(ctx context.Context, ownerUserID string, friendUserID string, remark string) (err error) {
if err := f.friend.UpdateFriendRemark(ctx, ownerUserID, friendUserID, remark); err != nil {
return err
}
return f.cache.DelFriend(ownerUserID, friendUserID).ExecDel(ctx)
}
func (f *friendDatabase) UpdateFriendEx(ctx context.Context, ownerUserID string, friendUserID string, ex string) (err error) {
if err := f.friend.UpdateFriendEx(ctx, ownerUserID, friendUserID, ex); err != nil {
return err
}
return f.cache.DelFriend(ownerUserID, friendUserID).ExecDel(ctx)
}

@ -38,6 +38,8 @@ type UserDatabase interface {
FindWithError(ctx context.Context, userIDs []string) (users []*relation.UserModel, err error)
// Find Get the information of the specified user If the userID is not found, no error will be returned
Find(ctx context.Context, userIDs []string) (users []*relation.UserModel, err error)
// Find userInfo By Nickname
FindByNickname(ctx context.Context, nickname string) (users []*relation.UserModel, err error)
// Create Insert multiple external guarantees that the userID is not repeated and does not exist in the db
Create(ctx context.Context, users []*relation.UserModel) (err error)
// Update update (non-zero value) external guarantee userID exists
@ -133,6 +135,11 @@ func (u *userDatabase) Find(ctx context.Context, userIDs []string) (users []*rel
return u.cache.GetUsersInfo(ctx, userIDs)
}
// Find userInfo By Nickname
func (u *userDatabase) FindByNickname(ctx context.Context, nickname string) (users []*relation.UserModel, err error) {
return u.userDB.TakeByNickname(ctx, nickname)
}
// Create Insert multiple external guarantees that the userID is not repeated and does not exist in the db.
func (u *userDatabase) Create(ctx context.Context, users []*relation.UserModel) (err error) {
return u.tx.Transaction(ctx, func(ctx context.Context) error {

@ -160,3 +160,33 @@ func (f *FriendMgo) UpdatePinStatus(ctx context.Context, ownerUserID string, fri
return nil
}
func (f *FriendMgo) UpdateFriendRemark(ctx context.Context, ownerUserID string, friendUserID string, remark string) (err error) {
filter := bson.M{"owner_user_id": ownerUserID, "friend_user_id": friendUserID}
// Create an update operation to set the "is_pinned" field to isPinned for all documents.
update := bson.M{"$set": bson.M{"remark": remark}}
// Perform the update operation for all documents in the collection.
_, err = f.coll.UpdateMany(ctx, filter, update)
if err != nil {
return errs.Wrap(err, "update remark error")
}
return nil
}
func (f *FriendMgo) UpdateFriendEx(ctx context.Context, ownerUserID string, friendUserID string, ex string) (err error) {
filter := bson.M{"owner_user_id": ownerUserID, "friend_user_id": friendUserID}
// Create an update operation to set the "is_pinned" field to isPinned for all documents.
update := bson.M{"$set": bson.M{"ex": ex}}
// Perform the update operation for all documents in the collection.
_, err = f.coll.UpdateMany(ctx, filter, update)
if err != nil {
return errs.Wrap(err, "update ex error")
}
return nil
}

@ -65,6 +65,10 @@ func (u *UserMgo) Take(ctx context.Context, userID string) (user *relation.UserM
return mgoutil.FindOne[*relation.UserModel](ctx, u.coll, bson.M{"user_id": userID})
}
func (u *UserMgo) TakeByNickname(ctx context.Context, nickname string) (user []*relation.UserModel, err error) {
return mgoutil.Find[*relation.UserModel](ctx, u.coll, bson.M{"nickname": nickname})
}
func (u *UserMgo) Page(ctx context.Context, pagination pagination.Pagination) (count int64, users []*relation.UserModel, err error) {
return mgoutil.FindPage[*relation.UserModel](ctx, u.coll, bson.M{}, pagination)
}

@ -59,4 +59,8 @@ type FriendModelInterface interface {
FindFriendUserIDs(ctx context.Context, ownerUserID string) (friendUserIDs []string, err error)
// UpdatePinStatus update friend's pin status
UpdatePinStatus(ctx context.Context, ownerUserID string, friendUserID string, isPinned bool) (err error)
// UpdateFriendRemark update friend's remark
UpdateFriendRemark(ctx context.Context, ownerUserID string, friendUserID string, remark string) (err error)
// UpdateFriendEx update friend's ex
UpdateFriendEx(ctx context.Context, ownerUserID string, friendUserID string, ex string) (err error)
}

@ -53,6 +53,7 @@ type UserModelInterface interface {
UpdateByMap(ctx context.Context, userID string, args map[string]any) (err error)
Find(ctx context.Context, userIDs []string) (users []*UserModel, err error)
Take(ctx context.Context, userID string) (user *UserModel, err error)
TakeByNickname(ctx context.Context, nickname string) (user []*UserModel, err error)
Page(ctx context.Context, pagination pagination.Pagination) (count int64, users []*UserModel, err error)
Exist(ctx context.Context, userID string) (exist bool, err error)
GetAllUserID(ctx context.Context, pagination pagination.Pagination) (count int64, userIDs []string, err error)

@ -74,6 +74,10 @@ func buildMongoURI() string {
return uri
}
if config.Config.Mongo.Uri != "" {
return config.Config.Mongo.Uri
}
username := os.Getenv("MONGO_USERNAME")
password := os.Getenv("MONGO_PASSWORD")
address := os.Getenv("MONGO_ADDRESS")

@ -24,7 +24,8 @@ import (
func setupTestEnvironment() {
os.Setenv("ZOOKEEPER_SCHEMA", "openim")
os.Setenv("ZOOKEEPER_ADDRESS", "172.28.0.1:12181")
os.Setenv("ZOOKEEPER_ADDRESS", "172.28.0.1")
os.Setenv("ZOOKEEPER_PORT", "12181")
os.Setenv("ZOOKEEPER_USERNAME", "")
os.Setenv("ZOOKEEPER_PASSWORD", "")
}

@ -52,10 +52,18 @@ func getEnv(key, fallback string) string {
return fallback
}
// getZkAddrFromEnv returns the value of an environment variable if it exists, otherwise it returns the fallback value.
// getZkAddrFromEnv returns the Zookeeper addresses combined from the ZOOKEEPER_ADDRESS and ZOOKEEPER_PORT environment variables.
// If the environment variables are not set, it returns the fallback value.
func getZkAddrFromEnv(fallback []string) []string {
if value, exists := os.LookupEnv("ZOOKEEPER_ADDRESS"); exists {
return strings.Split(value, ",")
address, addrExists := os.LookupEnv("ZOOKEEPER_ADDRESS")
port, portExists := os.LookupEnv("ZOOKEEPER_PORT")
if addrExists && portExists {
addresses := strings.Split(address, ",")
for i, addr := range addresses {
addresses[i] = addr + ":" + port
}
return addresses
}
return fallback
}

@ -31,15 +31,14 @@ import (
"github.com/openimsdk/open-im-server/v3/pkg/common/config"
)
const (
maxRetry = 10 // Maximum number of retries for producer creation
)
const maxRetry = 10 // number of retries
var errEmptyMsg = errors.New("binary msg is empty")
var errEmptyMsg = errors.New("kafka binary msg is empty")
// Producer represents a Kafka producer.
type Producer struct {
topic string
addr []string
topic string
config *sarama.Config
producer sarama.SyncProducer
}
@ -68,7 +67,7 @@ func NewKafkaProducer(addr []string, topic string) *Producer {
// Get Kafka configuration from environment variables or fallback to config file
kafkaUsername := getEnvOrConfig("KAFKA_USERNAME", config.Config.Kafka.Username)
kafkaPassword := getEnvOrConfig("KAFKA_PASSWORD", config.Config.Kafka.Password)
kafkaAddr := getEnvOrConfig("KAFKA_ADDRESS", addr[0]) // Assuming addr[0] contains address from config
kafkaAddr := getKafkaAddrFromEnv(addr) // Updated to use the new function
// Configure SASL authentication if credentials are provided
if kafkaUsername != "" && kafkaPassword != "" {
@ -78,7 +77,7 @@ func NewKafkaProducer(addr []string, topic string) *Producer {
}
// Set the Kafka address
p.addr = []string{kafkaAddr}
p.addr = kafkaAddr
// Set up TLS configuration (if required)
SetupTLSConfig(p.config)

@ -15,7 +15,9 @@
package kafka
import (
"fmt"
"os"
"strings"
"github.com/IBM/sarama"
@ -44,3 +46,20 @@ func getEnvOrConfig(envName string, configValue string) string {
}
return configValue
}
// getKafkaAddrFromEnv returns the Kafka addresses combined from the KAFKA_ADDRESS and KAFKA_PORT environment variables.
// If the environment variables are not set, it returns the fallback value.
func getKafkaAddrFromEnv(fallback []string) []string {
envAddr := os.Getenv("KAFKA_ADDRESS")
envPort := os.Getenv("KAFKA_PORT")
if envAddr != "" && envPort != "" {
addresses := strings.Split(envAddr, ",")
for i, addr := range addresses {
addresses[i] = fmt.Sprintf("%s:%s", addr, envPort)
}
return addresses
}
return fallback
}

@ -30,10 +30,9 @@ func NewOptions(opts ...OptionsOpt) Options {
options[constant.IsOfflinePush] = false
options[constant.IsUnreadCount] = false
options[constant.IsConversationUpdate] = false
options[constant.IsSenderSync] = false
options[constant.IsSenderSync] = true
options[constant.IsNotPrivate] = false
options[constant.IsSenderConversationUpdate] = false
options[constant.IsSenderNotificationPush] = false
options[constant.IsReactionFromCache] = false
for _, opt := range opts {
opt(options)
@ -114,12 +113,6 @@ func WithSenderConversationUpdate() OptionsOpt {
}
}
func WithSenderNotificationPush() OptionsOpt {
return func(options Options) {
options[constant.IsSenderNotificationPush] = true
}
}
func WithReactionFromCache() OptionsOpt {
return func(options Options) {
options[constant.IsReactionFromCache] = true
@ -174,10 +167,6 @@ func (o Options) IsSenderConversationUpdate() bool {
return o.Is(constant.IsSenderConversationUpdate)
}
func (o Options) IsSenderNotificationPush() bool {
return o.Is(constant.IsSenderNotificationPush)
}
func (o Options) IsReactionFromCache() bool {
return o.Is(constant.IsReactionFromCache)
}

@ -68,6 +68,7 @@ func newContentTypeConf() map[int32]config.NotificationConf {
constant.BlackAddedNotification: config.Config.Notification.BlackAdded,
constant.BlackDeletedNotification: config.Config.Notification.BlackDeleted,
constant.FriendInfoUpdatedNotification: config.Config.Notification.FriendInfoUpdated,
constant.FriendsInfoUpdateNotification: config.Config.Notification.FriendInfoUpdated, //use the same FriendInfoUpdated
// conversation
constant.ConversationChangeNotification: config.Config.Notification.ConversationChanged,
constant.ConversationUnreadNotification: config.Config.Notification.ConversationChanged,
@ -115,6 +116,7 @@ func newSessionTypeConf() map[int32]int32 {
constant.BlackAddedNotification: constant.SingleChatType,
constant.BlackDeletedNotification: constant.SingleChatType,
constant.FriendInfoUpdatedNotification: constant.SingleChatType,
constant.FriendsInfoUpdateNotification: constant.SingleChatType,
// conversation
constant.ConversationChangeNotification: constant.SingleChatType,
constant.ConversationUnreadNotification: constant.SingleChatType,
@ -256,6 +258,7 @@ func (s *NotificationSender) NotificationWithSesstionType(ctx context.Context, s
optionsConfig.ReliabilityLevel = constant.UnreliableNotification
}
options := config.GetOptionsByNotification(optionsConfig)
s.SetOptionsByContentType(ctx, options, contentType)
msg.Options = options
offlineInfo.Title = title
offlineInfo.Desc = desc
@ -274,3 +277,11 @@ func (s *NotificationSender) NotificationWithSesstionType(ctx context.Context, s
func (s *NotificationSender) Notification(ctx context.Context, sendID, recvID string, contentType int32, m proto.Message, opts ...NotificationOptions) error {
return s.NotificationWithSesstionType(ctx, sendID, recvID, contentType, s.sessionTypeConf[contentType], m, opts...)
}
func (s *NotificationSender) SetOptionsByContentType(_ context.Context, options map[string]bool, contentType int32) {
switch contentType {
case constant.UserStatusChangeNotification:
options[constant.IsSenderSync] = false
default:
}
}

@ -196,7 +196,12 @@ func (f *FriendNotificationSender) FriendRemarkSetNotification(ctx context.Conte
tips.FromToUserID.ToUserID = toUserID
return f.Notification(ctx, fromUserID, toUserID, constant.FriendRemarkSetNotification, &tips)
}
func (f *FriendNotificationSender) FriendsInfoUpdateNotification(ctx context.Context, toUserID string, friendIDs []string) error {
tips := sdkws.FriendsInfoUpdateTips{}
tips.FromToUserID.ToUserID = toUserID
tips.FriendIDs = friendIDs
return f.Notification(ctx, toUserID, toUserID, constant.FriendsInfoUpdateNotification, &tips)
}
func (f *FriendNotificationSender) BlackAddedNotification(ctx context.Context, req *pbfriend.AddBlackReq) error {
tips := sdkws.BlackAddedTips{FromToUserID: &sdkws.FromToUserID{}}
tips.FromToUserID.FromUserID = req.OwnerUserID

@ -16,6 +16,7 @@ package main
import (
"context"
"errors"
"flag"
"fmt"
"net"
@ -285,10 +286,23 @@ func checkZookeeper() (string, error) {
// Connect to Zookeeper
str := "the addr is:" + address
c, _, err := zk.Connect(zookeeperAddresses, time.Second) // Adjust the timeout as necessary
c, eventChan, err := zk.Connect(zookeeperAddresses, time.Second) // Adjust the timeout as necessary
if err != nil {
return "", errs.Wrap(errStr(err, str))
}
timeout := time.After(5 * time.Second)
for {
select {
case event := <-eventChan:
if event.State == zk.StateConnected {
fmt.Println("Connected to Zookeeper")
goto Connected
}
case <-timeout:
return "", errs.Wrap(errors.New("timeout waiting for Zookeeper connection"), "Zookeeper Addr: "+strings.Join(config.Config.Zookeeper.ZkAddr, " "))
}
}
Connected:
defer c.Close()
// Set authentication if username and password are provided
@ -298,12 +312,6 @@ func checkZookeeper() (string, error) {
}
}
// Check if Zookeeper is reachable
_, _, err = c.Get("/")
if err != nil {
return "", errs.Wrap(err, str)
}
return str, nil
}

@ -21,22 +21,10 @@ import (
"time"
"github.com/redis/go-redis/v9"
"github.com/stretchr/testify/assert"
"github.com/openimsdk/open-im-server/v3/pkg/common/config"
)
func TestCheckMysql(t *testing.T) {
err := mockInitCfg()
assert.NoError(t, err, "Initialization should not produce errors")
err = checkMysql()
if err != nil {
// You might expect an error if MySQL isn't running locally with the mock credentials.
t.Logf("Expected error due to mock configuration: %v", err)
}
}
// Mock for initCfg for testing purpose
func mockInitCfg() error {
config.Config.Mysql.Username = "root"

Loading…
Cancel
Save