// Package sys // @Link https://github.com/bufanyun/hotgo // @Copyright Copyright (c) 2026 HotGo // @Author Ms <133814250@qq.com> // @License https://github.com/bufanyun/hotgo/blob/master/LICENSE package sys import ( "context" "encoding/json" "fmt" "hotgo/internal/consts" "hotgo/internal/library/contexts" "hotgo/internal/library/imkafka" "hotgo/internal/library/imtenant" "hotgo/internal/model/input/sysin" "hotgo/internal/service" "strconv" "github.com/gogf/gf/v2/database/gdb" "github.com/gogf/gf/v2/errors/gerror" "github.com/gogf/gf/v2/frame/g" "github.com/gogf/gf/v2/os/gctx" "github.com/gogf/gf/v2/os/gtime" "github.com/gogf/gf/v2/util/guid" ) type sSysImGroup struct{} const ( imGroupModeMember = "JRWPS" imGroupModeAdmin = "JRWPASO" imGroupModeOwner = "JRWPASDO" ) func NewSysImGroup() *sSysImGroup { return &sSysImGroup{} } func init() { service.RegisterSysImGroup(NewSysImGroup()) } // List 获取IM群组列表 func (s *sSysImGroup) List(ctx context.Context, in *sysin.ImGroupListInp) (list []*sysin.ImGroupListModel, totalCount int, err error) { if err = in.Filter(ctx); err != nil { return } db, tenant, err := resolveImTenantDB(ctx, in.TenantId) if err != nil { return } mod := s.groupModel(ctx, db, tenant.Id) mod = s.applyGroupWhere(mod, in) mod = mod.Page(in.Page, in.PerPage).OrderDesc("t.touchedat").OrderDesc("t.id") if err = mod.ScanAndCount(&list, &totalCount, false); err != nil { err = gerror.Wrap(err, "获取IM群组列表失败,请稍后重试!") return } for _, item := range list { item.TenantId = tenant.Id item.TenantName = tenant.Name } return } func (s *sSysImGroup) groupModel(ctx context.Context, db gdb.DB, tenantId int64) *gdb.Model { return db.Model(tenantTablePrefix(tenantId)+"topics t").Safe().Ctx(ctx). LeftJoin("im_group_settings gs", "gs.tenant_id = t.tenant_id AND gs.topic = t.name"). LeftJoin("im_users owner", "owner.tenant_id = t.tenant_id AND owner.id = COALESCE(gs.owner_user_id, t.owner)"). LeftJoin("im_user_profiles owner_profile", "owner_profile.tenant_id = t.tenant_id AND owner_profile.user_id = COALESCE(gs.owner_user_id, t.owner)"). Fields( "t.name topic", "t.state", "COALESCE(gs.owner_user_id, t.owner) owner_id", "COALESCE(NULLIF(owner_profile.nickname, ''), owner.username, '') owner_nickname", "COALESCE(owner.username, '') owner_username", "t.seqid seq_id", "t.delid del_id", "t.subcnt sub_cnt", "t.touchedat touched_at", "t.createdat created_at", "t.updatedat updated_at", "t.public public_json", "t.trusted trusted_json", "t.tags tags_json", "t.aux aux_json", "COALESCE(NULLIF(gs.name, ''), NULLIF(JSON_UNQUOTE(JSON_EXTRACT(t.public, '$.fn')), ''), NULLIF(JSON_UNQUOTE(JSON_EXTRACT(t.public, '$.name')), ''), t.name) group_name", "COALESCE(gs.avatar_file_id, '') avatar_file_id", "COALESCE(gs.description, '') description", "COALESCE(gs.group_status, '') group_status", "COALESCE(gs.join_approval, 0) join_approval", "COALESCE(gs.mute_all, 0) mute_all", "COALESCE(gs.member_upload_files, 0) member_upload_files", "COALESCE(gs.member_private_chat, 0) member_private_chat", "COALESCE(gs.member_invite, 0) member_invite", "COALESCE(gs.max_member_count, 0) max_member_count", "COALESCE(gs.max_admin_count, 0) max_admin_count", "COALESCE(gs.member_count, 0) business_member_count", "COALESCE(gs.version, 0) settings_version", "gs.updated_at settings_updated_at", ). Where("t.tenant_id", tenantId). WhereLike("t.name", "grp%") } func (s *sSysImGroup) applyGroupWhere(mod *gdb.Model, in *sysin.ImGroupListInp) *gdb.Model { if in.Topic != "" { mod = mod.WhereLike("t.name", "%"+in.Topic+"%") } if in.OwnerId > 0 { mod = mod.Where("COALESCE(gs.owner_user_id, t.owner)", in.OwnerId) } if in.State != nil { mod = mod.Where("t.state", *in.State) } if in.GroupStatus != "" { mod = mod.Where("gs.group_status", in.GroupStatus) } if len(in.CreatedAt) == 2 { mod = mod.WhereBetween("t.createdat", in.CreatedAt[0], in.CreatedAt[1]) } if len(in.TouchedAt) == 2 { mod = mod.WhereBetween("t.touchedat", in.TouchedAt[0], in.TouchedAt[1]) } return mod } // View 获取IM群组指定信息 func (s *sSysImGroup) View(ctx context.Context, in *sysin.ImGroupViewInp) (res *sysin.ImGroupViewModel, err error) { if err = in.Filter(ctx); err != nil { return } db, tenant, err := resolveImTenantDB(ctx, in.TenantId) if err != nil { return } err = db.Model(tenantTablePrefix(tenant.Id)+"topics t").Safe().Ctx(ctx). LeftJoin("im_group_settings gs", "gs.tenant_id = t.tenant_id AND gs.topic = t.name"). LeftJoin("im_users owner", "owner.tenant_id = t.tenant_id AND owner.id = COALESCE(gs.owner_user_id, t.owner)"). LeftJoin("im_user_profiles owner_profile", "owner_profile.tenant_id = t.tenant_id AND owner_profile.user_id = COALESCE(gs.owner_user_id, t.owner)"). Fields( "t.name topic", "t.state", "COALESCE(gs.owner_user_id, t.owner) owner_id", "COALESCE(NULLIF(owner_profile.nickname, ''), owner.username, '') owner_nickname", "COALESCE(owner.username, '') owner_username", "t.seqid seq_id", "t.delid del_id", "t.subcnt sub_cnt", "t.touchedat touched_at", "t.createdat created_at", "t.updatedat updated_at", "t.public public_json", "t.trusted trusted_json", "t.tags tags_json", "t.aux aux_json", "t.access access_json", "t.usebt use_bt", "COALESCE(NULLIF(gs.name, ''), NULLIF(JSON_UNQUOTE(JSON_EXTRACT(t.public, '$.fn')), ''), NULLIF(JSON_UNQUOTE(JSON_EXTRACT(t.public, '$.name')), ''), t.name) group_name", "COALESCE(gs.avatar_file_id, '') avatar_file_id", "COALESCE(gs.description, '') description", "COALESCE(gs.group_status, '') group_status", "COALESCE(gs.join_approval, 0) join_approval", "COALESCE(gs.mute_all, 0) mute_all", "COALESCE(gs.member_upload_files, 0) member_upload_files", "COALESCE(gs.member_private_chat, 0) member_private_chat", "COALESCE(gs.member_invite, 0) member_invite", "COALESCE(gs.max_member_count, 0) max_member_count", "COALESCE(gs.max_admin_count, 0) max_admin_count", "COALESCE(gs.member_count, 0) business_member_count", "COALESCE(gs.version, 0) settings_version", "gs.updated_at settings_updated_at", ). Where("t.tenant_id", tenant.Id). Where("t.name", in.Topic). WhereLike("t.name", "grp%"). Scan(&res) if err != nil { return nil, gerror.Wrap(err, "获取IM群组信息失败,请稍后重试!") } if res == nil { return nil, nil } res.TenantId = tenant.Id res.TenantName = tenant.Name return } // MemberList 获取IM群成员列表 func (s *sSysImGroup) MemberList(ctx context.Context, in *sysin.ImGroupMemberListInp) (list []*sysin.ImGroupMemberListModel, totalCount int, err error) { if err = in.Filter(ctx); err != nil { return } db, tenant, err := resolveImTenantDB(ctx, in.TenantId) if err != nil { return } if s.hasGroupMemberTruth(ctx, db, tenant.Id, in.Topic) { return s.memberListFromBusiness(ctx, db, tenant.Id, in) } return s.memberListFromProjection(ctx, db, tenant.Id, in) } func (s *sSysImGroup) hasGroupMemberTruth(ctx context.Context, db gdb.DB, tenantId int64, topic string) bool { count, err := db.Model(tenantTablePrefix(tenantId)+"group_members").Safe().Ctx(ctx).Where("tenant_id", tenantId).Where("topic", topic).Count() return err == nil && count > 0 } func (s *sSysImGroup) memberListFromBusiness(ctx context.Context, db gdb.DB, tenantId int64, in *sysin.ImGroupMemberListInp) (list []*sysin.ImGroupMemberListModel, totalCount int, err error) { mod := db.Model(tenantTablePrefix(tenantId)+"group_members gm").Safe().Ctx(ctx). LeftJoin("im_users u", "u.tenant_id = gm.tenant_id AND u.id = gm.user_id"). LeftJoin("im_user_profiles up", "up.tenant_id = gm.tenant_id AND up.user_id = gm.user_id"). LeftJoin(tenantTablePrefix(tenantId)+"subscriptions s", "s.tenant_id = gm.tenant_id AND s.topic = gm.topic AND s.userid = gm.user_id"). Fields( "gm.user_id", "up.nickname", "u.state", "s.modewant mode_want", "s.modegiven mode_given", "s.recvseqid recv_seq_id", "s.readseqid read_seq_id", "s.delid del_id", "s.deletedat deleted_at", "s.createdat created_at", "s.updatedat updated_at", "s.private private_json", "gm.role", "gm.member_status", "gm.nickname group_nickname", "gm.pinned", "gm.mute_notifications", "gm.join_source", "gm.joined_by", "gm.joined_at", "gm.left_at", "gm.removed_by", "gm.version", ). Where("gm.tenant_id", tenantId). Where("gm.topic", in.Topic) if in.UserId > 0 { mod = mod.Where("gm.user_id", in.UserId) } if in.Role != "" { mod = mod.Where("gm.role", in.Role) } if in.MemberStatus != "" { mod = mod.Where("gm.member_status", in.MemberStatus) } mod = mod.Page(in.Page, in.PerPage).OrderAsc("gm.member_status").OrderAsc("gm.role").OrderAsc("gm.id") if err = mod.ScanAndCount(&list, &totalCount, false); err != nil { err = gerror.Wrap(err, "获取IM群成员列表失败,请稍后重试!") return } return } func (s *sSysImGroup) memberListFromProjection(ctx context.Context, db gdb.DB, tenantId int64, in *sysin.ImGroupMemberListInp) (list []*sysin.ImGroupMemberListModel, totalCount int, err error) { mod := db.Model(tenantTablePrefix(tenantId)+"subscriptions s").Safe().Ctx(ctx). LeftJoin("im_users u", "u.tenant_id = s.tenant_id AND u.id = s.userid"). LeftJoin("im_user_profiles up", "up.tenant_id = s.tenant_id AND up.user_id = s.userid"). Fields( "s.userid user_id", "up.nickname", "u.state", "s.modewant mode_want", "s.modegiven mode_given", "s.recvseqid recv_seq_id", "s.readseqid read_seq_id", "s.delid del_id", "s.deletedat deleted_at", "s.createdat created_at", "s.updatedat updated_at", "s.private private_json", "'member' role", "IF(s.deletedat IS NULL, 'active', 'left') member_status", "'' group_nickname", "0 pinned", "0 mute_notifications", "'projection' join_source", "0 joined_by", "s.createdat joined_at", "s.deletedat left_at", "0 removed_by", "0 version", ). Where("s.tenant_id", tenantId). Where("s.topic", in.Topic) if in.UserId > 0 { mod = mod.Where("s.userid", in.UserId) } mod = mod.Page(in.Page, in.PerPage).OrderAsc("s.deletedat").OrderAsc("s.userid") if err = mod.ScanAndCount(&list, &totalCount, false); err != nil { err = gerror.Wrap(err, "获取IM群成员列表失败,请稍后重试!") return } return } // Governance 获取IM群治理配置 func (s *sSysImGroup) Governance(ctx context.Context, in *sysin.ImGroupTopicInp) (res *sysin.ImGroupGovernanceModel, err error) { if err = in.Filter(ctx); err != nil { return } db, tenant, err := resolveImTenantDB(ctx, in.TenantId) if err != nil { return nil, err } err = db.Model("im_group_settings").Safe().Ctx(ctx). Fields( "topic", "owner_user_id", "name", "avatar_file_id", "description", "group_status", "join_approval", "mute_all", "member_upload_files", "member_private_chat", "member_invite", "max_member_count", "max_admin_count", "member_count", "version", "created_by", "dissolved_by", "dissolved_at", "created_at", "updated_at", ). Where("tenant_id", tenant.Id). Where("topic", in.Topic). Scan(&res) if err != nil { return nil, gerror.Wrap(err, "获取IM群治理配置失败,请稍后重试!") } if res == nil { return nil, nil } res.TenantId = tenant.Id res.TenantName = tenant.Name return } // GovernanceUpdate 更新IM群治理配置 func (s *sSysImGroup) GovernanceUpdate(ctx context.Context, in *sysin.ImGroupGovernanceUpdateInp) (res *sysin.ImGroupCommandModel, err error) { if err = in.Filter(ctx); err != nil { return nil, err } payload := map[string]any{ "topic": in.Topic, "actor_user_id": strconv.FormatInt(contexts.GetUserId(ctx), 10), "governance": map[string]any{ "join_approval": in.JoinApproval == 1, "mute_all": in.MuteAll == 1, "member_upload_files": in.MemberUploadFiles == 1, "member_private_chat": in.MemberPrivateChat == 1, "member_invite": in.MemberInvite == 1, "max_member_count": in.MaxMemberCount, "max_admin_count": in.MaxAdminCount, }, } return s.publishGroupCommand(ctx, in.TenantId, in.Topic, consts.ImAdminCommandGroupGovernUpdate, payload, 0) } func (s *sSysImGroup) ProfileUpdate(ctx context.Context, in *sysin.ImGroupProfileUpdateInp) (res *sysin.ImGroupCommandModel, err error) { if err = in.Filter(ctx); err != nil { return nil, err } payload := map[string]any{ "topic": in.Topic, "actor_user_id": strconv.FormatInt(contexts.GetUserId(ctx), 10), "profile": map[string]any{ "name": in.Name, "avatar_file_id": in.AvatarFileId, "description": in.Description, }, } return s.publishGroupCommand(ctx, in.TenantId, in.Topic, consts.ImAdminCommandGroupProfileUpdate, payload, in.Version) } // Dissolve 解散IM群 func (s *sSysImGroup) Dissolve(ctx context.Context, in *sysin.ImGroupDissolveInp) (res *sysin.ImGroupCommandModel, err error) { if err = in.Filter(ctx); err != nil { return nil, err } payload := map[string]any{ "topic": in.Topic, "actor_user_id": strconv.FormatInt(contexts.GetUserId(ctx), 10), "reason": in.Reason, } return s.publishGroupCommand(ctx, in.TenantId, in.Topic, consts.ImAdminCommandGroupDissolve, payload, 0) } // Mute 禁言IM群成员 func (s *sSysImGroup) Mute(ctx context.Context, in *sysin.ImGroupMuteInp) (res *sysin.ImGroupCommandModel, err error) { return s.MemberMute(ctx, in) } func (s *sSysImGroup) MemberMute(ctx context.Context, in *sysin.ImGroupMuteInp) (res *sysin.ImGroupCommandModel, err error) { if err = in.Filter(ctx); err != nil { return nil, err } payload := map[string]any{ "topic": in.Topic, "actor_user_id": strconv.FormatInt(contexts.GetUserId(ctx), 10), "user_id": strconv.FormatInt(in.UserId, 10), "reason": in.Reason, } if in.ExpiresAt != nil { payload["expires_at"] = in.ExpiresAt.Time.Format("2006-01-02T15:04:05.000Z07:00") } return s.publishGroupCommand(ctx, in.TenantId, in.Topic, consts.ImAdminCommandGroupMemberMute, payload, 0) } // Unmute 解除IM群成员禁言 func (s *sSysImGroup) Unmute(ctx context.Context, in *sysin.ImGroupUnmuteInp) (res *sysin.ImGroupCommandModel, err error) { return s.MemberUnmute(ctx, in) } func (s *sSysImGroup) MemberUnmute(ctx context.Context, in *sysin.ImGroupUnmuteInp) (res *sysin.ImGroupCommandModel, err error) { if err = in.Filter(ctx); err != nil { return nil, err } payload := map[string]any{ "topic": in.Topic, "actor_user_id": strconv.FormatInt(contexts.GetUserId(ctx), 10), "user_id": strconv.FormatInt(in.UserId, 10), } return s.publishGroupCommand(ctx, in.TenantId, in.Topic, consts.ImAdminCommandGroupMemberUnmute, payload, 0) } func (s *sSysImGroup) MemberRemove(ctx context.Context, in *sysin.ImGroupMemberRemoveInp) (res *sysin.ImGroupCommandModel, err error) { if err = in.Filter(ctx); err != nil { return nil, err } payload := map[string]any{ "topic": in.Topic, "actor_user_id": strconv.FormatInt(contexts.GetUserId(ctx), 10), "user_id": strconv.FormatInt(in.UserId, 10), "reason": in.Reason, } return s.publishGroupCommand(ctx, in.TenantId, in.Topic, consts.ImAdminCommandGroupMemberRemove, payload, 0) } // JoinRequestApprove 通过IM入群申请 func (s *sSysImGroup) JoinRequestApprove(ctx context.Context, in *sysin.ImGroupJoinRequestResolveInp) (err error) { if err = in.Filter(ctx); err != nil { return } return s.resolveJoinRequest(ctx, in, consts.ImGroupRequestStatusAccepted) } // JoinRequestReject 拒绝IM入群申请 func (s *sSysImGroup) JoinRequestReject(ctx context.Context, in *sysin.ImGroupJoinRequestResolveInp) (err error) { if err = in.Filter(ctx); err != nil { return } return s.resolveJoinRequest(ctx, in, consts.ImGroupRequestStatusRejected) } func (s *sSysImGroup) resolveJoinRequest(ctx context.Context, in *sysin.ImGroupJoinRequestResolveInp, status string) (err error) { db, tenant, err := resolveImTenantDB(ctx, in.TenantId) if err != nil { return err } actor := contexts.GetUserId(ctx) requestId := gctx.CtxId(ctx) if requestId == "" { requestId = guid.S() } now := gtime.Now() return db.Transaction(ctx, func(ctx context.Context, tx gdb.TX) error { if err := s.ensureActiveGroup(ctx, tx, in.Topic); err != nil { return err } before, txErr := s.groupJoinRequestSnapshot(ctx, tx, in.Topic, in.RequestId) if txErr != nil { return txErr } if before == nil { return gerror.New("入群申请不存在") } if before["status"].String() != consts.ImGroupRequestStatusPending { return gerror.New("入群申请已处理") } applicantUserId := before["applicant_user_id"].Int64() if status == consts.ImGroupRequestStatusAccepted { if txErr = s.upsertGroupSubscription(ctx, tx, tenant.Id, in.Topic, applicantUserId, imGroupModeMember, now); txErr != nil { return txErr } if txErr = s.upsertGroupMember(ctx, tx, tenant.Id, in.Topic, applicantUserId, consts.ImGroupRoleMember, "approval", actor, now); txErr != nil { return txErr } } result, txErr := tx.Exec(`UPDATE im_group_join_requests SET status=?,reject_reason=?,handled_by=?,handled_at=?,updated_at=?,version=version+1 WHERE topic=? AND request_id=? AND status=?`, status, in.RejectReason, actor, now, now, in.Topic, in.RequestId, consts.ImGroupRequestStatusPending) if txErr != nil { return txErr } if affected, _ := result.RowsAffected(); affected == 0 { return gerror.New("入群申请已处理") } if txErr = s.recountGroupMembers(ctx, tx, tenant.Id, in.Topic); txErr != nil { return txErr } after, txErr := s.groupJoinRequestSnapshot(ctx, tx, in.Topic, in.RequestId) if txErr != nil { return txErr } action := "join_request_reject" if status == consts.ImGroupRequestStatusAccepted { action = "join_request_accept" } if txErr = s.insertGroupAdminLog(ctx, tx, in.Topic, actor, applicantUserId, action, "join_request", in.RequestId, recordMap(before), recordMap(after), requestId); txErr != nil { return txErr } return s.insertGroupEvent(ctx, tx, in.Topic, actor, "group.join_request."+status, g.Map{ "topic": in.Topic, "request_id": in.RequestId, "applicant_user_id": applicantUserId, "status": status, "request_id_admin": requestId, }) }) } // RoleUpdate 更新IM群成员角色 func (s *sSysImGroup) RoleUpdate(ctx context.Context, in *sysin.ImGroupRoleUpdateInp) (res *sysin.ImGroupCommandModel, err error) { if err = in.Filter(ctx); err != nil { return nil, err } switch in.Role { case consts.ImGroupRoleAdmin: return s.AdminSet(ctx, in) case consts.ImGroupRoleMember: return s.AdminRemove(ctx, in) default: return nil, gerror.New("只能设置管理员或普通成员") } } func (s *sSysImGroup) AdminSet(ctx context.Context, in *sysin.ImGroupRoleUpdateInp) (res *sysin.ImGroupCommandModel, err error) { if err = in.Filter(ctx); err != nil { return nil, err } payload := map[string]any{ "topic": in.Topic, "actor_user_id": strconv.FormatInt(contexts.GetUserId(ctx), 10), "user_ids": []string{strconv.FormatInt(in.UserId, 10)}, } return s.publishGroupCommand(ctx, in.TenantId, in.Topic, consts.ImAdminCommandGroupAdminSet, payload, 0) } func (s *sSysImGroup) AdminRemove(ctx context.Context, in *sysin.ImGroupRoleUpdateInp) (res *sysin.ImGroupCommandModel, err error) { if err = in.Filter(ctx); err != nil { return nil, err } payload := map[string]any{ "topic": in.Topic, "actor_user_id": strconv.FormatInt(contexts.GetUserId(ctx), 10), "user_id": strconv.FormatInt(in.UserId, 10), } return s.publishGroupCommand(ctx, in.TenantId, in.Topic, consts.ImAdminCommandGroupAdminRemove, payload, 0) } // OwnerTransfer 转让IM群主 func (s *sSysImGroup) OwnerTransfer(ctx context.Context, in *sysin.ImGroupOwnerTransferInp) (res *sysin.ImGroupCommandModel, err error) { if err = in.Filter(ctx); err != nil { return nil, err } payload := map[string]any{ "topic": in.Topic, "actor_user_id": strconv.FormatInt(contexts.GetUserId(ctx), 10), "target_user_id": strconv.FormatInt(in.UserId, 10), } return s.publishGroupCommand(ctx, in.TenantId, in.Topic, consts.ImAdminCommandGroupOwnerTransfer, payload, 0) } func (s *sSysImGroup) publishGroupCommand(ctx context.Context, tenantId int64, topic string, eventType string, payload map[string]any, version int64) (*sysin.ImGroupCommandModel, error) { _, tenant, err := resolveImTenantDB(ctx, tenantId) if err != nil { return nil, err } user := contexts.GetUser(ctx) actorName := "" if user != nil { actorName = user.RealName } command, err := imkafka.PublishAdminCommand(ctx, imkafka.AdminCommandInput{ TenantID: tenant.Id, TenantCode: tenant.Code, EventType: eventType, BizType: "group", BizID: topic, Version: version, Actor: &imkafka.Actor{ Type: "admin", ID: strconv.FormatInt(contexts.GetUserId(ctx), 10), DisplayName: actorName, }, Payload: payload, CreatedBy: contexts.GetUserId(ctx), }) if err != nil { return nil, err } return &sysin.ImGroupCommandModel{ CommandID: command.CommandID, EventID: command.EventID, Status: command.Status, TenantID: command.TenantID, TenantCode: command.TenantCode, EventType: command.EventType, IdempotencyKey: command.IdempotencyKey, }, nil } func (s *sSysImGroup) groupGovernanceSnapshot(ctx context.Context, tx gdb.TX, topic string) (res gdb.Record, err error) { return tx.Model("im_group_settings").Safe().Ctx(ctx). Fields( "topic", "owner_user_id", "name", "group_status", "join_approval", "mute_all", "member_upload_files", "member_private_chat", "member_invite", "max_member_count", "max_admin_count", "member_count", "version", "created_by", "dissolved_by", "dissolved_at", "created_at", "updated_at", ). Where("topic", topic). One() } func (s *sSysImGroup) groupJoinRequestSnapshot(ctx context.Context, tx gdb.TX, topic string, requestId string) (res gdb.Record, err error) { return tx.Model("im_group_join_requests").Safe().Ctx(ctx). Where("topic", topic). Where("request_id", requestId). One() } func (s *sSysImGroup) groupMemberSnapshot(ctx context.Context, tx gdb.TX, tenantId int64, topic string, userId int64) (res gdb.Record, err error) { return tx.Model(tenantTablePrefix(tenantId)+"group_members").Safe().Ctx(ctx). Where("topic", topic). Where("user_id", userId). One() } func (s *sSysImGroup) groupFileSnapshot(ctx context.Context, tx gdb.TX, topic string, groupFileId string) (res gdb.Record, err error) { return tx.Model("im_group_files").Safe().Ctx(ctx). Where("topic", topic). Where("group_file_id", groupFileId). One() } func (s *sSysImGroup) ensureActiveGroup(ctx context.Context, tx gdb.TX, topic string) error { status, err := tx.Model("im_group_settings").Safe().Ctx(ctx).Where("topic", topic).Value("group_status") if err != nil { return err } if status.IsNil() { return gerror.New("群治理配置不存在") } if status.String() == consts.ImGroupStatusDissolved { return gerror.New("已解散群不能执行该操作") } return nil } func (s *sSysImGroup) groupMuteSnapshot(ctx context.Context, tx gdb.TX, topic string, userId int64) (res gdb.Record, err error) { return tx.Model("im_group_mutes").Safe().Ctx(ctx). Fields( "topic", "user_id", "muted_by", "muted_at", "expires_at", "reason", "status", "revoked_by", "revoked_at", "version", "created_at", "updated_at", ). Where("topic", topic). Where("user_id", userId). One() } func (s *sSysImGroup) ensureActiveGroupMember(ctx context.Context, tx gdb.TX, tenantId int64, topic string, userId int64) error { count, err := tx.Model(tenantTablePrefix(tenantId)+"group_members").Safe().Ctx(ctx). Where("topic", topic). Where("user_id", userId). Where("member_status", consts.ImGroupMemberStatusActive). Count() if err != nil { return err } if count == 0 { return gerror.New("成员不在群内或不是正常状态") } return nil } func (s *sSysImGroup) ensureAdminLimit(ctx context.Context, tx gdb.TX, tenantId int64, topic string, userId int64) error { currentRole, err := tx.Model(tenantTablePrefix(tenantId)+"group_members").Safe().Ctx(ctx). Where("topic", topic). Where("user_id", userId). Where("member_status", consts.ImGroupMemberStatusActive). Value("role") if err != nil { return err } if currentRole.String() == consts.ImGroupRoleAdmin { return nil } maxAdminCount, err := tx.Model("im_group_settings").Safe().Ctx(ctx).Where("topic", topic).Value("max_admin_count") if err != nil { return err } if maxAdminCount.Int() <= 0 { return nil } adminCount, err := tx.Model(tenantTablePrefix(tenantId)+"group_members").Safe().Ctx(ctx). Where("topic", topic). Where("role", consts.ImGroupRoleAdmin). Where("member_status", consts.ImGroupMemberStatusActive). Count() if err != nil { return err } if adminCount >= maxAdminCount.Int() { return gerror.New("群管理员数量已达上限") } return nil } func (s *sSysImGroup) upsertGroupSubscription(ctx context.Context, tx gdb.TX, tenantId int64, topic string, userId int64, modeGiven string, now *gtime.Time) error { table := tenantTablePrefix(tenantId) + "subscriptions" _, err := tx.Exec(fmt.Sprintf(`INSERT INTO %s(tenant_id,createdat,updatedat,deletedat,userid,topic,modewant,modegiven,private) VALUES(?,?,?,NULL,?,?,?,?,CAST('{}' AS JSON)) ON DUPLICATE KEY UPDATE createdat=VALUES(createdat),updatedat=VALUES(updatedat),deletedat=NULL, modewant=VALUES(modewant),modegiven=VALUES(modegiven),delid=0,recvseqid=0,readseqid=0`, table), tenantId, now, now, userId, topic, imGroupModeMember, modeGiven) return err } func (s *sSysImGroup) upsertGroupMember(ctx context.Context, tx gdb.TX, tenantId int64, topic string, userId int64, role string, joinSource string, joinedBy int64, now *gtime.Time) error { table := tenantTablePrefix(tenantId) + "group_members" _, err := tx.Exec(fmt.Sprintf(`INSERT INTO %s(tenant_id,topic,user_id,role,member_status,join_source,joined_by,joined_at,version,created_at,updated_at) VALUES(?,?,?,?,?,?,?,?,?,?,?) ON DUPLICATE KEY UPDATE role=VALUES(role),member_status=VALUES(member_status),join_source=VALUES(join_source), joined_by=VALUES(joined_by),joined_at=VALUES(joined_at),left_at=NULL,removed_by=NULL,version=version+1,updated_at=VALUES(updated_at)`, table), tenantId, topic, userId, role, consts.ImGroupMemberStatusActive, joinSource, joinedBy, now, 1, now, now) return err } func (s *sSysImGroup) recountGroupMembers(ctx context.Context, tx gdb.TX, tenantId int64, topic string) error { membersTable := tenantTablePrefix(tenantId) + "group_members" subscriptionsTable := tenantTablePrefix(tenantId) + "subscriptions" topicsTable := tenantTablePrefix(tenantId) + "topics" if _, err := tx.Exec(fmt.Sprintf(`UPDATE im_group_settings SET member_count=(SELECT COUNT(*) FROM %s WHERE tenant_id=? AND topic=? AND member_status=?),updated_at=CURRENT_TIMESTAMP(3) WHERE tenant_id=? AND topic=?`, membersTable), tenantId, topic, consts.ImGroupMemberStatusActive, tenantId, topic); err != nil { return err } _, err := tx.Exec(fmt.Sprintf(`UPDATE %s SET subcnt=(SELECT COUNT(*) FROM %s WHERE tenant_id=? AND topic=? AND deletedat IS NULL),updatedat=CURRENT_TIMESTAMP(3) WHERE tenant_id=? AND name=?`, topicsTable, subscriptionsTable), tenantId, topic, tenantId, topic) return err } func recordMap(record gdb.Record) any { if record == nil { return nil } return record.Map() } func (s *sSysImGroup) insertGroupAdminLog(ctx context.Context, tx gdb.TX, topic string, operatorId int64, targetUserId int64, action string, objectType string, objectId string, beforeData any, afterData any, requestId string) error { id, err := imtenant.NextSnowflakeID(ctx) if err != nil { return gerror.Wrap(err, "生成群管理日志ID失败") } logId := fmt.Sprintf("gal_%d", id) beforeJson, _ := json.Marshal(beforeData) afterJson, _ := json.Marshal(afterData) _, err = tx.Exec(`INSERT INTO im_group_admin_logs (log_id,topic,operator_id,target_user_id,action,object_type,object_id,before_data,after_data,request_id,idempotency_key) VALUES(?,?,?,?,?,?,?,CAST(? AS JSON),CAST(? AS JSON),?,?)`, logId, topic, operatorId, targetUserId, action, objectType, objectId, string(beforeJson), string(afterJson), requestId, requestId) return err } func (s *sSysImGroup) insertGroupEvent(ctx context.Context, tx gdb.TX, topic string, actorId int64, eventType string, payload any) error { id, err := imtenant.NextSnowflakeID(ctx) if err != nil { return gerror.Wrap(err, "生成群事件ID失败") } payloadJson, _ := json.Marshal(payload) _, err = tx.Exec(`INSERT INTO im_group_events_outbox (event_id,topic,actor_user_id,event_type,payload,status) VALUES(?,?,?,?,CAST(? AS JSON),'pending')`, fmt.Sprintf("gev_%d", id), topic, actorId, eventType, string(payloadJson)) return err } // JoinRequestList 获取IM入群申请列表 func (s *sSysImGroup) JoinRequestList(ctx context.Context, in *sysin.ImGroupJoinRequestListInp) (list []*sysin.ImGroupJoinRequestListModel, totalCount int, err error) { if err = in.Filter(ctx); err != nil { return } db, _, err := resolveImTenantDB(ctx, in.TenantId) if err != nil { return } mod := db.Model("im_group_join_requests").Safe().Ctx(ctx).Where("topic", in.Topic) if in.Status != "" { mod = mod.Where("status", in.Status) } if in.ApplicantUserId > 0 { mod = mod.Where("applicant_user_id", in.ApplicantUserId) } mod = mod.Page(in.Page, in.PerPage).OrderDesc("id") if err = mod.ScanAndCount(&list, &totalCount, false); err != nil { err = gerror.Wrap(err, "获取IM入群申请列表失败,请稍后重试!") } return } // FileList 获取IM群文件列表 func (s *sSysImGroup) FileList(ctx context.Context, in *sysin.ImGroupFileListInp) (list []*sysin.ImGroupFileListModel, totalCount int, err error) { if err = in.Filter(ctx); err != nil { return } db, _, err := resolveImTenantDB(ctx, in.TenantId) if err != nil { return } mod := db.Model("im_group_files").Safe().Ctx(ctx).Where("topic", in.Topic) if in.Status != "" { mod = mod.Where("status", in.Status) } if in.FileType != "" { mod = mod.Where("file_type", in.FileType) } if in.FileName != "" { mod = mod.WhereLike("file_name", "%"+in.FileName+"%") } if in.UploaderId > 0 { mod = mod.Where("uploader_id", in.UploaderId) } mod = mod.Page(in.Page, in.PerPage).OrderDesc("id") if err = mod.ScanAndCount(&list, &totalCount, false); err != nil { err = gerror.Wrap(err, "获取IM群文件列表失败,请稍后重试!") } return } // FileDelete 删除IM群文件 func (s *sSysImGroup) FileDelete(ctx context.Context, in *sysin.ImGroupFileDeleteInp) (err error) { if err = in.Filter(ctx); err != nil { return } db, _, err := resolveImTenantDB(ctx, in.TenantId) if err != nil { return err } actor := contexts.GetUserId(ctx) requestId := gctx.CtxId(ctx) if requestId == "" { requestId = guid.S() } now := gtime.Now() return db.Transaction(ctx, func(ctx context.Context, tx gdb.TX) error { if err := s.ensureActiveGroup(ctx, tx, in.Topic); err != nil { return err } before, txErr := s.groupFileSnapshot(ctx, tx, in.Topic, in.GroupFileId) if txErr != nil { return txErr } if before == nil { return gerror.New("群文件不存在") } if before["status"].String() == consts.ImGroupFileStatusDeleted { return gerror.New("群文件已删除") } result, txErr := tx.Exec(`UPDATE im_group_files SET status=?,deleted_by=?,deleted_at=?,updated_at=? WHERE topic=? AND group_file_id=? AND status<>?`, consts.ImGroupFileStatusDeleted, actor, now, now, in.Topic, in.GroupFileId, consts.ImGroupFileStatusDeleted) if txErr != nil { return txErr } if affected, _ := result.RowsAffected(); affected == 0 { return gerror.New("群文件已删除") } after, txErr := s.groupFileSnapshot(ctx, tx, in.Topic, in.GroupFileId) if txErr != nil { return txErr } if txErr = s.insertGroupAdminLog(ctx, tx, in.Topic, actor, before["uploader_id"].Int64(), "group_file_delete", "group_file", in.GroupFileId, recordMap(before), recordMap(after), requestId); txErr != nil { return txErr } return s.insertGroupEvent(ctx, tx, in.Topic, actor, "group.file.deleted", g.Map{ "topic": in.Topic, "group_file_id": in.GroupFileId, "file_id": before["file_id"].String(), "request_id": requestId, }) }) } // InvitationList 获取IM群邀请列表 func (s *sSysImGroup) InvitationList(ctx context.Context, in *sysin.ImGroupInvitationListInp) (list []*sysin.ImGroupInvitationListModel, totalCount int, err error) { if err = in.Filter(ctx); err != nil { return } db, _, err := resolveImTenantDB(ctx, in.TenantId) if err != nil { return } mod := db.Model("im_group_invitations").Safe().Ctx(ctx).Where("topic", in.Topic) if in.Status != "" { mod = mod.Where("status", in.Status) } if in.TargetUserId > 0 { mod = mod.Where("target_user_id", in.TargetUserId) } if in.InviterUserId > 0 { mod = mod.Where("inviter_user_id", in.InviterUserId) } mod = mod.Page(in.Page, in.PerPage).OrderDesc("id") if err = mod.ScanAndCount(&list, &totalCount, false); err != nil { err = gerror.Wrap(err, "获取IM群邀请列表失败,请稍后重试!") } return } // AnnouncementList 获取IM群公告列表 func (s *sSysImGroup) AnnouncementList(ctx context.Context, in *sysin.ImGroupAnnouncementListInp) (list []*sysin.ImGroupAnnouncementListModel, totalCount int, err error) { if err = in.Filter(ctx); err != nil { return } db, _, err := resolveImTenantDB(ctx, in.TenantId) if err != nil { return } mod := db.Model("im_group_announcements").Safe().Ctx(ctx).Where("topic", in.Topic) if in.Status != "" { mod = mod.Where("status", in.Status) } mod = mod.Page(in.Page, in.PerPage).OrderDesc("published_at").OrderDesc("id") if err = mod.ScanAndCount(&list, &totalCount, false); err != nil { err = gerror.Wrap(err, "获取IM群公告列表失败,请稍后重试!") } return } // MuteList 获取IM群禁言列表 func (s *sSysImGroup) MuteList(ctx context.Context, in *sysin.ImGroupMuteListInp) (list []*sysin.ImGroupMuteListModel, totalCount int, err error) { if err = in.Filter(ctx); err != nil { return } db, _, err := resolveImTenantDB(ctx, in.TenantId) if err != nil { return } mod := db.Model("im_group_mutes").Safe().Ctx(ctx).Where("topic", in.Topic) if in.Status != "" { mod = mod.Where("status", in.Status) } if in.UserId > 0 { mod = mod.Where("user_id", in.UserId) } mod = mod.Page(in.Page, in.PerPage).OrderDesc("muted_at").OrderDesc("user_id") if err = mod.ScanAndCount(&list, &totalCount, false); err != nil { err = gerror.Wrap(err, "获取IM群禁言列表失败,请稍后重试!") } return } // AdminLogList 获取IM群管理日志列表 func (s *sSysImGroup) AdminLogList(ctx context.Context, in *sysin.ImGroupAdminLogListInp) (list []*sysin.ImGroupAdminLogListModel, totalCount int, err error) { if err = in.Filter(ctx); err != nil { return } db, _, err := resolveImTenantDB(ctx, in.TenantId) if err != nil { return } mod := db.Model("im_group_admin_logs").Safe().Ctx(ctx). Fields( "id", "log_id", "topic", "operator_id", "target_user_id", "action", "object_type", "object_id", "CAST(before_data AS CHAR) before_data", "CAST(after_data AS CHAR) after_data", "request_id", "idempotency_key", "created_at", ). Where("topic", in.Topic) if in.Action != "" { mod = mod.WhereLike("action", "%"+in.Action+"%") } if in.OperatorId > 0 { mod = mod.Where("operator_id", in.OperatorId) } mod = mod.Page(in.Page, in.PerPage).OrderDesc("id") if err = mod.ScanAndCount(&list, &totalCount, false); err != nil { err = gerror.Wrap(err, "获取IM群管理日志列表失败,请稍后重试!") } return }