golang-im聊天
選択できるのは25トピックまでです。 トピックは、先頭が英数字で、英数字とダッシュ('-')を使用した35文字以内のものにしてください。

group.go 16 KiB

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