golang-im聊天
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 
 

614 lines
16 KiB

  1. package model
  2. import (
  3. "context"
  4. "database/sql"
  5. "fmt"
  6. "gim/internal/logic/domain/message/repo"
  7. "gim/internal/logic/proxy"
  8. "gim/pkg/gerrors"
  9. "gim/pkg/grpclib"
  10. "gim/pkg/logger"
  11. "gim/pkg/pb"
  12. "gim/pkg/rpc"
  13. "gim/pkg/util"
  14. "time"
  15. "go.uber.org/zap"
  16. "google.golang.org/protobuf/proto"
  17. )
  18. const (
  19. UpdateTypeUpdate = 1
  20. UpdateTypeDelete = 2
  21. )
  22. // Group 群组
  23. type Group struct {
  24. Id int64 // 群组id
  25. Name string // 组名
  26. AvatarUrl string // 头像
  27. Introduction string // 群简介
  28. UserNum int32 // 群组人数
  29. IsAllMemberBanned int32 // 是否全员禁言(1:是 2:否)
  30. MasterId int64 // 站长id
  31. Extra string // 附加字段
  32. CreateTime time.Time // 创建时间
  33. UpdateTime time.Time // 更新时间
  34. Members []GroupUser `gorm:"-"` // 群组成员
  35. }
  36. type GroupUser struct {
  37. Id int64 // 自增主键
  38. GroupId int64 // 群组id
  39. UserId int64 // 用户id
  40. MemberType int // 群组类型
  41. Remarks string // 备注
  42. Extra string // 附加属性
  43. Status int // 状态
  44. CreateTime time.Time // 创建时间
  45. UpdateTime time.Time // 更新时间
  46. UpdateType int `gorm:"-"` // 更新类型
  47. }
  48. type GroupUserV2 struct {
  49. Id int64 // 自增主键
  50. GroupId int64 // 群组id
  51. UserId int64 // 用户id
  52. MemberType int // 群组类型
  53. Remarks string // 备注
  54. Extra string // 附加属性
  55. Status sql.NullInt32 // 状态
  56. CreateTime time.Time // 创建时间
  57. UpdateTime time.Time // 更新时间
  58. UpdateType int `gorm:"-"` // 更新类型
  59. }
  60. func (g *Group) ToProto() *pb.Group {
  61. if g == nil {
  62. return nil
  63. }
  64. return &pb.Group{
  65. GroupId: g.Id,
  66. Name: g.Name,
  67. AvatarUrl: g.AvatarUrl,
  68. Introduction: g.Introduction,
  69. UserMum: g.UserNum,
  70. IsAllMemberBanned: g.IsAllMemberBanned,
  71. Extra: g.Extra,
  72. CreateTime: g.CreateTime.Unix(),
  73. UpdateTime: g.UpdateTime.Unix(),
  74. }
  75. }
  76. func CreateGroup(userId, masterId int64, in *pb.CreateGroupReq) *Group {
  77. now := time.Now()
  78. group := &Group{
  79. Name: in.Name,
  80. AvatarUrl: in.AvatarUrl,
  81. Introduction: in.Introduction,
  82. MasterId: masterId,
  83. Extra: in.Extra,
  84. Members: make([]GroupUser, 0, len(in.MemberIds)+1),
  85. IsAllMemberBanned: 2,
  86. CreateTime: now,
  87. UpdateTime: now,
  88. }
  89. // 创建者添加为群主
  90. group.Members = append(group.Members, GroupUser{
  91. GroupId: group.Id,
  92. UserId: userId,
  93. MemberType: int(pb.MemberType_GMT_ADMIN),
  94. CreateTime: now,
  95. UpdateTime: now,
  96. UpdateType: UpdateTypeUpdate,
  97. })
  98. // 其让人添加为成员
  99. for i := range in.MemberIds {
  100. group.Members = append(group.Members, GroupUser{
  101. GroupId: group.Id,
  102. UserId: in.MemberIds[i],
  103. MemberType: int(pb.MemberType_GMT_MEMBER),
  104. CreateTime: now,
  105. UpdateTime: now,
  106. UpdateType: UpdateTypeUpdate,
  107. })
  108. }
  109. return group
  110. }
  111. func (g *Group) Update(ctx context.Context, in *pb.UpdateGroupReq) error {
  112. g.Name = in.Name
  113. g.AvatarUrl = in.AvatarUrl
  114. g.Introduction = in.Introduction
  115. g.Extra = in.Extra
  116. g.UpdateTime = time.Now()
  117. return nil
  118. }
  119. func (g *Group) PushUpdate(ctx context.Context, userId int64, isUpdateIntroduction bool) error {
  120. userResp, err := rpc.GetBusinessIntClient().GetUser(ctx, &pb.GetUserReq{UserId: userId})
  121. if err != nil {
  122. return err
  123. }
  124. err = g.PushMessage(ctx, pb.PushCode_PC_UPDATE_GROUP, &pb.UpdateGroupPush{
  125. OptId: userId,
  126. OptName: userResp.User.Nickname,
  127. OptAvatarUrl: userResp.User.AvatarUrl,
  128. Name: g.Name,
  129. AvatarUrl: g.AvatarUrl,
  130. Introduction: g.Introduction,
  131. IsUpdateIntroduction: isUpdateIntroduction,
  132. Extra: g.Extra,
  133. }, true)
  134. if err != nil {
  135. return err
  136. }
  137. if isUpdateIntroduction {
  138. //_, err = g.SendMessage(ctx,
  139. // &pb.Sender{
  140. // SenderType: pb.SenderType_ST_USER,
  141. // SenderId: 0,
  142. // },
  143. // &pb.SendMessageReq{
  144. // ReceiverType: pb.ReceiverType_RT_GROUP,
  145. // ReceiverId: g.Id,
  146. // ToUserIds: nil,
  147. // MessageType: pb.MessageType_MT_COMMAND,
  148. // MessageContent: commandBuf,
  149. // SendTime: util.UnixMilliTime(time.Now()),
  150. // IsPersist: isPersist,
  151. // })
  152. //if err != nil {
  153. // return err
  154. //}
  155. }
  156. return nil
  157. }
  158. // SendMessage 消息发送至群组
  159. func (g *Group) SendMessage(ctx context.Context, sender *pb.Sender, req *pb.SendMessageReq) (int64, error) {
  160. if sender.SenderType == pb.SenderType_ST_USER && !g.IsMember(sender.SenderId) {
  161. logger.Sugar.Error(ctx, sender.SenderId, req.ReceiverId, "不在群组内")
  162. return 0, gerrors.ErrNotInGroup
  163. }
  164. // 如果发送者是用户,将消息发送给发送者,获取用户seq
  165. var userSeq int64
  166. var err error
  167. if sender.SenderType == pb.SenderType_ST_USER {
  168. userSeq, err = proxy.MessageProxy.SendToUser(ctx, sender, sender.SenderId, req)
  169. if err != nil {
  170. return 0, err
  171. }
  172. }
  173. go func() {
  174. defer util.RecoverPanic()
  175. // 将消息发送给群组用户,使用写扩散
  176. for _, user := range g.Members {
  177. // 前面已经发送过,这里不需要再发送
  178. if sender.SenderType == pb.SenderType_ST_USER && user.UserId == sender.SenderId {
  179. continue
  180. }
  181. _, err := proxy.MessageProxy.SendToUser(grpclib.NewAndCopyRequestId(ctx), sender, user.UserId, req)
  182. if err != nil {
  183. return
  184. }
  185. }
  186. }()
  187. return userSeq, nil
  188. }
  189. // RecallSendMessage 撤回消息发送至群组
  190. func (g *Group) RecallSendMessage(ctx context.Context, sender *pb.Sender, req *pb.RecallMessageReq) (int64, error) {
  191. if sender.SenderType == pb.SenderType_ST_USER && !g.IsMember(sender.SenderId) {
  192. logger.Sugar.Error(ctx, sender.SenderId, req.ReceiverId, "不在群组内")
  193. return 0, gerrors.ErrNotInGroup
  194. }
  195. var err error
  196. //查询到对应seq的消息
  197. msg := &pb.RECALL{}
  198. err = proto.Unmarshal(req.MessageContent, msg)
  199. if err != nil {
  200. return 0, err
  201. }
  202. message, err := repo.MessageRepo.GetMessage(sender.SenderId, msg.RecallSeq)
  203. if err != nil {
  204. return 0, err
  205. }
  206. // 如果发送者是用户,将消息发送给发送者,获取用户seq
  207. var userSeq int64
  208. if sender.SenderType == pb.SenderType_ST_USER {
  209. userSeq, err = proxy.MessageProxy.RecallMessageSendToUser(ctx, sender, sender.SenderId, req, message.SendTime)
  210. if err != nil {
  211. return 0, err
  212. }
  213. }
  214. go func() {
  215. defer util.RecoverPanic()
  216. // 将消息发送给群组用户,使用写扩散
  217. for _, user := range g.Members {
  218. // 前面已经发送过,这里不需要再发送
  219. if sender.SenderType == pb.SenderType_ST_USER && user.UserId == sender.SenderId {
  220. continue
  221. }
  222. msg.RecallSeq, err = repo.MessageRepo.GetMessageSeqForSendTime(user.UserId, message.SendTime)
  223. req.MessageContent, err = proto.Marshal(msg)
  224. if err != nil {
  225. return
  226. }
  227. _, err := proxy.MessageProxy.RecallMessageSendToUser(grpclib.NewAndCopyRequestId(ctx), sender, user.UserId, req, message.SendTime)
  228. if err != nil {
  229. return
  230. }
  231. }
  232. }()
  233. return userSeq, nil
  234. }
  235. // SendRedPackage 发送红包消息发送至群组
  236. func (g *Group) SendRedPackage(ctx context.Context, sender *pb.Sender, req *pb.SendRedPacketReq) (int64, error) {
  237. //TODO::发送红包,暂不过滤群组用户
  238. //if sender.SenderType == pb.SenderType_ST_USER && !g.IsMember(sender.SenderId) {
  239. // logger.Sugar.Error(ctx, sender.SenderId, req.ReceiverId, "不在群组内")
  240. // return 0, gerrors.ErrNotInGroup
  241. //}
  242. // 如果发送者是用户,将消息发送给发送者,获取用户seq
  243. var userSeq int64
  244. var err error
  245. if sender.SenderType == pb.SenderType_ST_USER {
  246. userSeq, err = proxy.MessageProxy.SendRedPackageToUser(ctx, sender, sender.SenderId, req)
  247. if err != nil {
  248. return 0, err
  249. }
  250. }
  251. go func() {
  252. defer util.RecoverPanic()
  253. // 将消息发送给群组用户,使用写扩散
  254. for _, user := range g.Members {
  255. // 前面已经发送过,这里不需要再发送
  256. if sender.SenderType == pb.SenderType_ST_USER && user.UserId == sender.SenderId {
  257. continue
  258. }
  259. _, err := proxy.MessageProxy.SendRedPackageToUser(grpclib.NewAndCopyRequestId(ctx), sender, user.UserId, req)
  260. if err != nil {
  261. return
  262. }
  263. }
  264. }()
  265. return userSeq, nil
  266. }
  267. func (g *Group) IsMember(userId int64) bool {
  268. for i := range g.Members {
  269. if g.Members[i].UserId == userId {
  270. return true
  271. }
  272. }
  273. return false
  274. }
  275. // PushMessage 向群组推送消息
  276. func (g *Group) PushMessage(ctx context.Context, code pb.PushCode, message proto.Message, isPersist bool) error {
  277. logger.Logger.Debug("push_to_group",
  278. zap.Int64("request_id", grpclib.GetCtxRequestId(ctx)),
  279. zap.Int64("group_id", g.Id),
  280. zap.Int32("code", int32(code)),
  281. zap.Any("message", message))
  282. messageBuf, err := proto.Marshal(message)
  283. if err != nil {
  284. return gerrors.WrapError(err)
  285. }
  286. commandBuf, err := proto.Marshal(&pb.Command{Code: int32(code), Data: messageBuf})
  287. if err != nil {
  288. return gerrors.WrapError(err)
  289. }
  290. _, err = g.SendMessage(ctx,
  291. &pb.Sender{
  292. SenderType: pb.SenderType_ST_SYSTEM,
  293. SenderId: 0,
  294. },
  295. &pb.SendMessageReq{
  296. ReceiverType: pb.ReceiverType_RT_GROUP,
  297. ReceiverId: g.Id,
  298. ToUserIds: nil,
  299. MessageType: pb.MessageType_MT_COMMAND,
  300. MessageContent: commandBuf,
  301. SendTime: util.UnixMilliTime(time.Now()),
  302. IsPersist: isPersist,
  303. },
  304. )
  305. if err != nil {
  306. return err
  307. }
  308. return nil
  309. }
  310. // GetMembers 获取群组用户
  311. func (g *Group) GetMembers(ctx context.Context) ([]*pb.GroupMember, error) {
  312. members := g.Members
  313. userIds := make(map[int64]int32, len(members))
  314. for i := range members {
  315. userIds[members[i].UserId] = 0
  316. }
  317. resp, err := rpc.GetBusinessIntClient().GetUsers(ctx, &pb.GetUsersReq{UserIds: userIds})
  318. if err != nil {
  319. return nil, err
  320. }
  321. var infos = make([]*pb.GroupMember, len(members))
  322. for i := range members {
  323. member := pb.GroupMember{
  324. UserId: members[i].UserId,
  325. MemberType: pb.MemberType(members[i].MemberType),
  326. Remarks: members[i].Remarks,
  327. Extra: members[i].Extra,
  328. Status: int32(members[i].Status),
  329. }
  330. user, ok := resp.Users[members[i].UserId]
  331. if ok {
  332. member.Nickname = user.Nickname
  333. member.Sex = user.Sex
  334. member.AvatarUrl = user.AvatarUrl
  335. member.UserExtra = user.Extra
  336. }
  337. infos[i] = &member
  338. }
  339. return infos, nil
  340. }
  341. // AddMembers 给群组添加用户
  342. func (g *Group) AddMembers(ctx context.Context, userIds []int64) ([]int64, []int64, error) {
  343. var existIds []int64
  344. var addedIds []int64
  345. now := time.Now()
  346. for i, userId := range userIds {
  347. if g.IsMember(userId) {
  348. existIds = append(existIds, userIds[i])
  349. continue
  350. }
  351. g.Members = append(g.Members, GroupUser{
  352. GroupId: g.Id,
  353. UserId: userIds[i],
  354. MemberType: int(pb.MemberType_GMT_MEMBER),
  355. CreateTime: now,
  356. UpdateTime: now,
  357. UpdateType: UpdateTypeUpdate,
  358. })
  359. addedIds = append(addedIds, userIds[i])
  360. }
  361. g.UserNum += int32(len(addedIds))
  362. return existIds, addedIds, nil
  363. }
  364. func (g *Group) PushAddMember(ctx context.Context, optUserId int64, addedIds []int64, isFilterGroupLeader bool) error {
  365. var addIdMap = make(map[int64]int32, len(addedIds))
  366. for i := range addedIds {
  367. addIdMap[addedIds[i]] = 0
  368. }
  369. addIdMap[optUserId] = 0
  370. usersResp, err := rpc.GetBusinessIntClient().GetUsers(ctx, &pb.GetUsersReq{UserIds: addIdMap})
  371. if err != nil {
  372. return err
  373. }
  374. var members []*pb.GroupMember
  375. for k, _ := range addIdMap {
  376. if isFilterGroupLeader && k == optUserId {
  377. continue
  378. }
  379. member, ok := usersResp.Users[k]
  380. if !ok {
  381. continue
  382. }
  383. members = append(members, &pb.GroupMember{
  384. UserId: member.UserId,
  385. Nickname: member.Nickname,
  386. Sex: member.Sex,
  387. AvatarUrl: member.AvatarUrl,
  388. UserExtra: member.Extra,
  389. Remarks: "",
  390. Extra: "",
  391. })
  392. }
  393. //
  394. optUser := usersResp.Users[optUserId]
  395. if optUserId == -1 {
  396. optUser.Nickname = "后台操作"
  397. }
  398. err = g.PushMessage(ctx, pb.PushCode_PC_ADD_GROUP_MEMBERS, &pb.AddGroupMembersPush{
  399. OptId: optUser.UserId,
  400. OptName: optUser.Nickname,
  401. Members: members,
  402. }, true)
  403. if err != nil {
  404. logger.Sugar.Error(err)
  405. }
  406. return nil
  407. }
  408. func (g *Group) GetMember(ctx context.Context, userId int64) *GroupUser {
  409. for i := range g.Members {
  410. if g.Members[i].UserId == userId {
  411. return &g.Members[i]
  412. }
  413. }
  414. return nil
  415. }
  416. // UpdateMember 更新群组成员信息
  417. func (g *Group) UpdateMember(ctx context.Context, in *pb.UpdateGroupMemberReq) error {
  418. member := g.GetMember(ctx, in.UserId)
  419. if member == nil {
  420. return nil
  421. }
  422. member.MemberType = int(in.MemberType)
  423. member.Remarks = in.Remarks
  424. member.Extra = in.Extra
  425. member.UpdateTime = time.Now()
  426. member.UpdateType = UpdateTypeUpdate
  427. return nil
  428. }
  429. // DeleteMember 删除用户群组
  430. func (g *Group) DeleteMember(ctx context.Context, userId int64) error {
  431. member := g.GetMember(ctx, userId)
  432. if member == nil {
  433. return nil
  434. }
  435. member.UpdateType = UpdateTypeDelete
  436. return nil
  437. }
  438. func (g *Group) PushUpdateMember(ctx context.Context, optId, userId int64, memberType int32) error {
  439. userResp, err := rpc.GetBusinessIntClient().GetUser(ctx, &pb.GetUserReq{UserId: optId})
  440. if err != nil {
  441. return err
  442. }
  443. if optId == -1 {
  444. userResp.User.Nickname = "后台操作"
  445. }
  446. updateUserResp, err := rpc.GetBusinessIntClient().GetUser(ctx, &pb.GetUserReq{UserId: userId})
  447. if err != nil {
  448. return err
  449. }
  450. err = g.PushMessage(ctx, pb.PushCode_PC_UPDATE_GROUP_MEMBER, &pb.UpdateMemberPush{
  451. OptId: optId,
  452. OptName: userResp.User.Nickname,
  453. UpdateUserId: userId,
  454. UpdateUserName: updateUserResp.User.Nickname,
  455. UpdateUserMemberType: memberType,
  456. }, true)
  457. if err != nil {
  458. return err
  459. }
  460. return nil
  461. }
  462. func (g *Group) PushDeleteMember(ctx context.Context, optId, userId int64) error {
  463. userResp, err := rpc.GetBusinessIntClient().GetUser(ctx, &pb.GetUserReq{UserId: optId})
  464. if err != nil {
  465. return err
  466. }
  467. if optId == -1 {
  468. userResp.User.Nickname = "后台操作"
  469. }
  470. deleteUserResp, err := rpc.GetBusinessIntClient().GetUser(ctx, &pb.GetUserReq{UserId: userId})
  471. if err != nil {
  472. return err
  473. }
  474. err = g.PushMessage(ctx, pb.PushCode_PC_REMOVE_GROUP_MEMBER, &pb.RemoveGroupMemberPush{
  475. OptId: optId,
  476. OptName: userResp.User.Nickname,
  477. DeletedUserId: userId,
  478. DeletedUserName: deleteUserResp.User.Nickname,
  479. }, true)
  480. if err != nil {
  481. return err
  482. }
  483. return nil
  484. }
  485. func (g *Group) PushGroupMemberBanned(ctx context.Context, optId, userId int64, isAllMemberBanned bool) error {
  486. userResp, err := rpc.GetBusinessIntClient().GetUser(ctx, &pb.GetUserReq{UserId: optId})
  487. if err != nil {
  488. return err
  489. }
  490. if optId == -1 {
  491. userResp.User.Nickname = "后台操作"
  492. }
  493. if !isAllMemberBanned && userId > 0 {
  494. bannedUserResp, err := rpc.GetBusinessIntClient().GetUser(ctx, &pb.GetUserReq{UserId: userId})
  495. if err != nil {
  496. return err
  497. }
  498. err = g.PushMessage(ctx, pb.PushCode_PC_BANNED_GROUP_MEMBER, &pb.BannedGroupMemberPush{
  499. OptId: optId,
  500. OptName: userResp.User.Nickname,
  501. BannedUserId: userId,
  502. BannedUserName: bannedUserResp.User.Nickname,
  503. }, true)
  504. if err != nil {
  505. return err
  506. }
  507. return nil
  508. } else {
  509. if userId == 0 {
  510. err = g.PushMessage(ctx, pb.PushCode_PC_BANNED_GROUP_MEMBER, &pb.BannedGroupMemberPush{
  511. OptId: optId,
  512. OptName: userResp.User.Nickname,
  513. BannedUserId: userId,
  514. BannedUserName: fmt.Sprintf("管理员\"%s\"设置群禁言", userResp.User.Nickname),
  515. }, true)
  516. if err != nil {
  517. return err
  518. }
  519. } else {
  520. err = g.PushMessage(ctx, pb.PushCode_PC_BANNED_GROUP_MEMBER, &pb.BannedGroupMemberPush{
  521. OptId: optId,
  522. OptName: userResp.User.Nickname,
  523. BannedUserId: userId,
  524. BannedUserName: fmt.Sprintf("管理员\"%s\"取消群禁言", userResp.User.Nickname),
  525. }, true)
  526. if err != nil {
  527. return err
  528. }
  529. }
  530. return nil
  531. }
  532. }
  533. func (g *Group) PushGroupMemberRemoveBanned(ctx context.Context, optId, userId int64) error {
  534. userResp, err := rpc.GetBusinessIntClient().GetUser(ctx, &pb.GetUserReq{UserId: optId})
  535. if err != nil {
  536. return err
  537. }
  538. if optId == -1 {
  539. userResp.User.Nickname = "后台操作"
  540. }
  541. removeBannedUserResp, err := rpc.GetBusinessIntClient().GetUser(ctx, &pb.GetUserReq{UserId: userId})
  542. if err != nil {
  543. return err
  544. }
  545. err = g.PushMessage(ctx, pb.PushCode_PC_REMOVE_BANNED_GROUP_MEMBER, &pb.RemoveBannedGroupMemberPush{
  546. OptId: optId,
  547. OptName: userResp.User.Nickname,
  548. RemoveBannedUserId: userId,
  549. RemoveBannedUserName: removeBannedUserResp.User.Nickname,
  550. }, true)
  551. if err != nil {
  552. return err
  553. }
  554. return nil
  555. }