feat: add IM governance admin enhancements
This commit is contained in:
@@ -10,6 +10,7 @@ import (
|
||||
"hotgo/internal/service"
|
||||
"hotgo/utility/simple"
|
||||
|
||||
"github.com/gogf/gf/v2/database/gdb"
|
||||
"github.com/gogf/gf/v2/encoding/gjson"
|
||||
"github.com/gogf/gf/v2/errors/gerror"
|
||||
"github.com/gogf/gf/v2/frame/g"
|
||||
@@ -80,6 +81,14 @@ func ensureImTenantActive(ctx context.Context, tenantId int64) (err error) {
|
||||
return nil
|
||||
}
|
||||
|
||||
func resolveImTenantDB(ctx context.Context, selectedTenantId int64) (db gdb.DB, tenant *entity.ImTenant, err error) {
|
||||
scope, err := resolveImAdminScope(ctx, selectedTenantId)
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
return imtenant.DB(ctx, scope.TenantId)
|
||||
}
|
||||
|
||||
func writeImAdminAudit(ctx context.Context, tenantId int64, action string, targetType string, targetId int64, result string, detail g.Map) error {
|
||||
operatorId := contexts.GetUserId(ctx)
|
||||
request := ghttp.RequestFromCtx(ctx)
|
||||
|
||||
@@ -7,12 +7,20 @@ package sys
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"hotgo/internal/consts"
|
||||
"hotgo/internal/library/contexts"
|
||||
"hotgo/internal/library/imtenant"
|
||||
"hotgo/internal/model/input/sysin"
|
||||
"hotgo/internal/service"
|
||||
|
||||
"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{}
|
||||
@@ -31,7 +39,7 @@ func (s *sSysImGroup) List(ctx context.Context, in *sysin.ImGroupListInp) (list
|
||||
return
|
||||
}
|
||||
|
||||
db, tenant, err := imtenant.DB(ctx, in.TenantId)
|
||||
db, tenant, err := resolveImTenantDB(ctx, in.TenantId)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
@@ -114,7 +122,7 @@ func (s *sSysImGroup) View(ctx context.Context, in *sysin.ImGroupViewInp) (res *
|
||||
return
|
||||
}
|
||||
|
||||
db, tenant, err := imtenant.DB(ctx, in.TenantId)
|
||||
db, tenant, err := resolveImTenantDB(ctx, in.TenantId)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
@@ -172,7 +180,7 @@ func (s *sSysImGroup) MemberList(ctx context.Context, in *sysin.ImGroupMemberLis
|
||||
return
|
||||
}
|
||||
|
||||
db, _, err := imtenant.DB(ctx, in.TenantId)
|
||||
db, _, err := resolveImTenantDB(ctx, in.TenantId)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
@@ -284,7 +292,7 @@ func (s *sSysImGroup) Governance(ctx context.Context, in *sysin.ImGroupTopicInp)
|
||||
if err = in.Filter(ctx); err != nil {
|
||||
return
|
||||
}
|
||||
db, tenant, err := imtenant.DB(ctx, in.TenantId)
|
||||
db, tenant, err := resolveImTenantDB(ctx, in.TenantId)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -324,12 +332,276 @@ func (s *sSysImGroup) Governance(ctx context.Context, in *sysin.ImGroupTopicInp)
|
||||
return
|
||||
}
|
||||
|
||||
// GovernanceUpdate 更新IM群治理配置
|
||||
func (s *sSysImGroup) GovernanceUpdate(ctx context.Context, in *sysin.ImGroupGovernanceUpdateInp) (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 {
|
||||
before, txErr := s.groupGovernanceSnapshot(ctx, tx, in.Topic)
|
||||
if txErr != nil {
|
||||
return txErr
|
||||
}
|
||||
if before == nil {
|
||||
return gerror.New("群治理配置不存在")
|
||||
}
|
||||
if before["group_status"].String() == consts.ImGroupStatusDissolved {
|
||||
return gerror.New("已解散群不能修改治理配置")
|
||||
}
|
||||
if _, txErr = tx.Exec(`UPDATE im_group_settings SET
|
||||
join_approval=?,mute_all=?,member_upload_files=?,member_private_chat=?,member_invite=?,
|
||||
max_member_count=?,max_admin_count=?,version=version+1,updated_at=?
|
||||
WHERE topic=?`,
|
||||
in.JoinApproval, in.MuteAll, in.MemberUploadFiles, in.MemberPrivateChat, in.MemberInvite,
|
||||
in.MaxMemberCount, in.MaxAdminCount, now, in.Topic); txErr != nil {
|
||||
return txErr
|
||||
}
|
||||
after, txErr := s.groupGovernanceSnapshot(ctx, tx, in.Topic)
|
||||
if txErr != nil {
|
||||
return txErr
|
||||
}
|
||||
if txErr = s.insertGroupAdminLog(ctx, tx, in.Topic, actor, 0, "governance_update", "group_settings", in.Topic, before.Map(), after.Map(), requestId); txErr != nil {
|
||||
return txErr
|
||||
}
|
||||
return s.insertGroupEvent(ctx, tx, in.Topic, actor, "group.governance_updated", g.Map{
|
||||
"topic": in.Topic, "operator_id": actor, "request_id": requestId,
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
// Dissolve 解散IM群
|
||||
func (s *sSysImGroup) Dissolve(ctx context.Context, in *sysin.ImGroupDissolveInp) (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 {
|
||||
before, txErr := s.groupGovernanceSnapshot(ctx, tx, in.Topic)
|
||||
if txErr != nil {
|
||||
return txErr
|
||||
}
|
||||
if before == nil {
|
||||
return gerror.New("群治理配置不存在")
|
||||
}
|
||||
if before["group_status"].String() == consts.ImGroupStatusDissolved {
|
||||
return gerror.New("群已解散")
|
||||
}
|
||||
result, txErr := tx.Exec(`UPDATE im_group_settings
|
||||
SET group_status=?,dissolved_by=?,dissolved_at=?,version=version+1,updated_at=?
|
||||
WHERE topic=? AND group_status<>?`,
|
||||
consts.ImGroupStatusDissolved, actor, now, now, in.Topic, consts.ImGroupStatusDissolved)
|
||||
if txErr != nil {
|
||||
return txErr
|
||||
}
|
||||
if affected, _ := result.RowsAffected(); affected == 0 {
|
||||
return gerror.New("群已解散")
|
||||
}
|
||||
after, txErr := s.groupGovernanceSnapshot(ctx, tx, in.Topic)
|
||||
if txErr != nil {
|
||||
return txErr
|
||||
}
|
||||
if txErr = s.insertGroupAdminLog(ctx, tx, in.Topic, actor, 0, "group_dissolve", "group", in.Topic, before.Map(), after.Map(), requestId); txErr != nil {
|
||||
return txErr
|
||||
}
|
||||
return s.insertGroupEvent(ctx, tx, in.Topic, actor, "group.dissolved", g.Map{
|
||||
"topic": in.Topic, "operator_id": actor, "request_id": requestId, "reason": in.Reason,
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
// Mute 禁言IM群成员
|
||||
func (s *sSysImGroup) Mute(ctx context.Context, in *sysin.ImGroupMuteInp) (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 txErr := s.ensureActiveGroupMember(ctx, tx, in.Topic, in.UserId); txErr != nil {
|
||||
return txErr
|
||||
}
|
||||
before, txErr := s.groupMuteSnapshot(ctx, tx, in.Topic, in.UserId)
|
||||
if txErr != nil {
|
||||
return txErr
|
||||
}
|
||||
muteId, txErr := imtenant.NextSnowflakeID(ctx)
|
||||
if txErr != nil {
|
||||
return gerror.Wrap(txErr, "生成禁言ID失败")
|
||||
}
|
||||
if _, txErr = tx.Exec(`INSERT INTO im_group_mutes
|
||||
(id,topic,user_id,muted_by,muted_at,expires_at,reason,status,version)
|
||||
VALUES(?,?,?,?,?,?,?, ?,1)
|
||||
ON DUPLICATE KEY UPDATE muted_by=VALUES(muted_by),muted_at=VALUES(muted_at),
|
||||
expires_at=VALUES(expires_at),reason=VALUES(reason),status=VALUES(status),
|
||||
revoked_by=NULL,revoked_at=NULL,version=version+1,updated_at=CURRENT_TIMESTAMP(3)`,
|
||||
muteId, in.Topic, in.UserId, actor, now, in.ExpiresAt, in.Reason, consts.ImGroupMuteStatusActive); txErr != nil {
|
||||
return txErr
|
||||
}
|
||||
after, txErr := s.groupMuteSnapshot(ctx, tx, in.Topic, in.UserId)
|
||||
if txErr != nil {
|
||||
return txErr
|
||||
}
|
||||
if txErr = s.insertGroupAdminLog(ctx, tx, in.Topic, actor, in.UserId, "member_mute", "member", fmt.Sprintf("%d", in.UserId), recordMap(before), recordMap(after), requestId); txErr != nil {
|
||||
return txErr
|
||||
}
|
||||
return s.insertGroupEvent(ctx, tx, in.Topic, actor, "group.mute.updated", g.Map{
|
||||
"topic": in.Topic, "user_id": in.UserId, "muted": true, "expires_at": in.ExpiresAt, "request_id": requestId,
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
// Unmute 解除IM群成员禁言
|
||||
func (s *sSysImGroup) Unmute(ctx context.Context, in *sysin.ImGroupUnmuteInp) (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 {
|
||||
before, txErr := s.groupMuteSnapshot(ctx, tx, in.Topic, in.UserId)
|
||||
if txErr != nil {
|
||||
return txErr
|
||||
}
|
||||
if before == nil || before["status"].String() != consts.ImGroupMuteStatusActive {
|
||||
return gerror.New("成员未处于禁言状态")
|
||||
}
|
||||
result, txErr := tx.Exec(`UPDATE im_group_mutes
|
||||
SET status=?,revoked_by=?,revoked_at=?,version=version+1,updated_at=?
|
||||
WHERE topic=? AND user_id=? AND status=?`,
|
||||
consts.ImGroupMuteStatusRevoked, actor, now, now, in.Topic, in.UserId, consts.ImGroupMuteStatusActive)
|
||||
if txErr != nil {
|
||||
return txErr
|
||||
}
|
||||
if affected, _ := result.RowsAffected(); affected == 0 {
|
||||
return gerror.New("成员未处于禁言状态")
|
||||
}
|
||||
after, txErr := s.groupMuteSnapshot(ctx, tx, in.Topic, in.UserId)
|
||||
if txErr != nil {
|
||||
return txErr
|
||||
}
|
||||
if txErr = s.insertGroupAdminLog(ctx, tx, in.Topic, actor, in.UserId, "member_unmute", "member", fmt.Sprintf("%d", in.UserId), recordMap(before), recordMap(after), requestId); txErr != nil {
|
||||
return txErr
|
||||
}
|
||||
return s.insertGroupEvent(ctx, tx, in.Topic, actor, "group.mute.updated", g.Map{
|
||||
"topic": in.Topic, "user_id": in.UserId, "muted": false, "request_id": requestId,
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
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) 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, topic string, userId int64) error {
|
||||
count, err := tx.Model("im_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 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 := imtenant.DB(ctx, in.TenantId)
|
||||
db, _, err := resolveImTenantDB(ctx, in.TenantId)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
@@ -352,7 +624,7 @@ func (s *sSysImGroup) InvitationList(ctx context.Context, in *sysin.ImGroupInvit
|
||||
if err = in.Filter(ctx); err != nil {
|
||||
return
|
||||
}
|
||||
db, _, err := imtenant.DB(ctx, in.TenantId)
|
||||
db, _, err := resolveImTenantDB(ctx, in.TenantId)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
@@ -378,7 +650,7 @@ func (s *sSysImGroup) AnnouncementList(ctx context.Context, in *sysin.ImGroupAnn
|
||||
if err = in.Filter(ctx); err != nil {
|
||||
return
|
||||
}
|
||||
db, _, err := imtenant.DB(ctx, in.TenantId)
|
||||
db, _, err := resolveImTenantDB(ctx, in.TenantId)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
@@ -398,7 +670,7 @@ func (s *sSysImGroup) MuteList(ctx context.Context, in *sysin.ImGroupMuteListInp
|
||||
if err = in.Filter(ctx); err != nil {
|
||||
return
|
||||
}
|
||||
db, _, err := imtenant.DB(ctx, in.TenantId)
|
||||
db, _, err := resolveImTenantDB(ctx, in.TenantId)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
@@ -421,7 +693,7 @@ func (s *sSysImGroup) AdminLogList(ctx context.Context, in *sysin.ImGroupAdminLo
|
||||
if err = in.Filter(ctx); err != nil {
|
||||
return
|
||||
}
|
||||
db, _, err := imtenant.DB(ctx, in.TenantId)
|
||||
db, _, err := resolveImTenantDB(ctx, in.TenantId)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
@@ -0,0 +1,190 @@
|
||||
// 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"
|
||||
"hotgo/internal/library/imtenant"
|
||||
"hotgo/internal/model/entity"
|
||||
"hotgo/internal/model/input/sysin"
|
||||
"hotgo/internal/service"
|
||||
|
||||
"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/gtime"
|
||||
)
|
||||
|
||||
type sSysImOutbox struct{}
|
||||
|
||||
func NewSysImOutbox() *sSysImOutbox {
|
||||
return &sSysImOutbox{}
|
||||
}
|
||||
|
||||
func init() {
|
||||
service.RegisterSysImOutbox(NewSysImOutbox())
|
||||
}
|
||||
|
||||
func (s *sSysImOutbox) List(ctx context.Context, in *sysin.ImOutboxListInp) (list []*sysin.ImOutboxListModel, totalCount int, err error) {
|
||||
if err = in.Filter(ctx); err != nil {
|
||||
return
|
||||
}
|
||||
db, tenant, err := s.tenantDB(ctx, in.TenantId)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
mod := s.model(ctx, db, in.Source)
|
||||
mod = s.applyWhere(mod, in)
|
||||
mod = mod.Page(in.Page, in.PerPage).OrderDesc("id")
|
||||
if err = mod.ScanAndCount(&list, &totalCount, false); err != nil {
|
||||
err = gerror.Wrap(err, "获取IM Outbox列表失败,请稍后重试!")
|
||||
return
|
||||
}
|
||||
for _, item := range list {
|
||||
item.TenantId = tenant.Id
|
||||
item.TenantName = tenant.Name
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
func (s *sSysImOutbox) View(ctx context.Context, in *sysin.ImOutboxViewInp) (res *sysin.ImOutboxViewModel, err error) {
|
||||
if err = in.Filter(ctx); err != nil {
|
||||
return
|
||||
}
|
||||
db, tenant, err := s.tenantDB(ctx, in.TenantId)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err = s.model(ctx, db, in.Source).Where("id", in.Id).Scan(&res); err != nil {
|
||||
return nil, gerror.Wrap(err, "获取IM Outbox详情失败,请稍后重试!")
|
||||
}
|
||||
if res == nil {
|
||||
return nil, nil
|
||||
}
|
||||
res.TenantId = tenant.Id
|
||||
res.TenantName = tenant.Name
|
||||
return
|
||||
}
|
||||
|
||||
func (s *sSysImOutbox) Retry(ctx context.Context, in *sysin.ImOutboxActionInp) (err error) {
|
||||
if err = in.Filter(ctx); err != nil {
|
||||
return
|
||||
}
|
||||
return s.updateStatus(ctx, in, sysin.ImOutboxStatusPending)
|
||||
}
|
||||
|
||||
func (s *sSysImOutbox) Dead(ctx context.Context, in *sysin.ImOutboxActionInp) (err error) {
|
||||
if err = in.Filter(ctx); err != nil {
|
||||
return
|
||||
}
|
||||
return s.updateStatus(ctx, in, sysin.ImOutboxStatusDead)
|
||||
}
|
||||
|
||||
func (s *sSysImOutbox) tenantDB(ctx context.Context, selectedTenantId int64) (db gdb.DB, tenant *entity.ImTenant, err error) {
|
||||
scope, err := resolveImAdminScope(ctx, selectedTenantId)
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
return imtenant.DB(ctx, scope.TenantId)
|
||||
}
|
||||
|
||||
func (s *sSysImOutbox) model(ctx context.Context, db gdb.DB, source string) *gdb.Model {
|
||||
if source == sysin.ImOutboxSourceGroup {
|
||||
return db.Model("im_group_events_outbox").Safe().Ctx(ctx).Fields(
|
||||
"'group' source",
|
||||
"id",
|
||||
"event_id",
|
||||
"'topic' target_type",
|
||||
"topic target",
|
||||
"actor_user_id",
|
||||
"event_type",
|
||||
"CAST(payload AS CHAR) payload",
|
||||
"status",
|
||||
"attempts",
|
||||
"next_retry_at",
|
||||
"delivered_at",
|
||||
"created_at",
|
||||
"updated_at",
|
||||
)
|
||||
}
|
||||
return db.Model("im_business_events_outbox").Safe().Ctx(ctx).Fields(
|
||||
"'business' source",
|
||||
"id",
|
||||
"event_id",
|
||||
"target_type",
|
||||
"target",
|
||||
"actor_user_id",
|
||||
"event_type",
|
||||
"CAST(payload AS CHAR) payload",
|
||||
"status",
|
||||
"attempts",
|
||||
"next_retry_at",
|
||||
"delivered_at",
|
||||
"created_at",
|
||||
"updated_at",
|
||||
)
|
||||
}
|
||||
|
||||
func (s *sSysImOutbox) applyWhere(mod *gdb.Model, in *sysin.ImOutboxListInp) *gdb.Model {
|
||||
if in.EventId != "" {
|
||||
mod = mod.WhereLike("event_id", "%"+in.EventId+"%")
|
||||
}
|
||||
if in.EventType != "" {
|
||||
mod = mod.Where("event_type", in.EventType)
|
||||
}
|
||||
if in.TargetType != "" && in.Source == sysin.ImOutboxSourceBusiness {
|
||||
mod = mod.Where("target_type", in.TargetType)
|
||||
}
|
||||
if in.Target != "" {
|
||||
if in.Source == sysin.ImOutboxSourceGroup {
|
||||
mod = mod.WhereLike("topic", "%"+in.Target+"%")
|
||||
} else {
|
||||
mod = mod.WhereLike("target", "%"+in.Target+"%")
|
||||
}
|
||||
}
|
||||
if in.Status != "" {
|
||||
mod = mod.Where("status", in.Status)
|
||||
}
|
||||
if len(in.CreatedAt) == 2 {
|
||||
mod = mod.WhereBetween("created_at", in.CreatedAt[0], in.CreatedAt[1])
|
||||
}
|
||||
return mod
|
||||
}
|
||||
|
||||
func (s *sSysImOutbox) updateStatus(ctx context.Context, in *sysin.ImOutboxActionInp, status string) (err error) {
|
||||
db, _, err := s.tenantDB(ctx, in.TenantId)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
table := "im_business_events_outbox"
|
||||
if in.Source == sysin.ImOutboxSourceGroup {
|
||||
table = "im_group_events_outbox"
|
||||
}
|
||||
current, err := db.Model(table).Safe().Ctx(ctx).Fields("status").Where("id", in.Id).Value()
|
||||
if err != nil {
|
||||
return gerror.Wrap(err, "获取IM Outbox事件失败,请稍后重试!")
|
||||
}
|
||||
if current.IsNil() {
|
||||
return gerror.New("IM Outbox事件不存在")
|
||||
}
|
||||
if current.String() == sysin.ImOutboxStatusDelivered {
|
||||
return gerror.New("已投递事件不能重新处置")
|
||||
}
|
||||
|
||||
data := g.Map{
|
||||
"status": status,
|
||||
"updated_at": gtime.Now(),
|
||||
}
|
||||
if status == sysin.ImOutboxStatusPending {
|
||||
data["next_retry_at"] = nil
|
||||
data["delivered_at"] = nil
|
||||
}
|
||||
if _, err = db.Model(table).Safe().Ctx(ctx).Where("id", in.Id).Data(data).Update(); err != nil {
|
||||
return gerror.Wrap(err, "更新IM Outbox事件失败,请稍后重试!")
|
||||
}
|
||||
return
|
||||
}
|
||||
@@ -8,7 +8,6 @@ package sys
|
||||
import (
|
||||
"context"
|
||||
"hotgo/internal/consts"
|
||||
"hotgo/internal/library/imtenant"
|
||||
"hotgo/internal/model/input/sysin"
|
||||
"hotgo/internal/service"
|
||||
|
||||
@@ -33,7 +32,7 @@ func (s *sSysImRealname) List(ctx context.Context, in *sysin.ImRealnameListInp)
|
||||
if err = in.Filter(ctx); err != nil {
|
||||
return
|
||||
}
|
||||
db, tenant, err := imtenant.DB(ctx, in.TenantId)
|
||||
db, tenant, err := resolveImTenantDB(ctx, in.TenantId)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
@@ -91,7 +90,7 @@ func (s *sSysImRealname) View(ctx context.Context, in *sysin.ImRealnameViewInp)
|
||||
if err = in.Filter(ctx); err != nil {
|
||||
return
|
||||
}
|
||||
db, tenant, err := imtenant.DB(ctx, in.TenantId)
|
||||
db, tenant, err := resolveImTenantDB(ctx, in.TenantId)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -129,7 +128,7 @@ func (s *sSysImRealname) Review(ctx context.Context, in *sysin.ImRealnameReviewI
|
||||
if err = in.Filter(ctx); err != nil {
|
||||
return
|
||||
}
|
||||
db, _, err := imtenant.DB(ctx, in.TenantId)
|
||||
db, _, err := resolveImTenantDB(ctx, in.TenantId)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
@@ -7,7 +7,6 @@ package sys
|
||||
|
||||
import (
|
||||
"context"
|
||||
"hotgo/internal/library/imtenant"
|
||||
"hotgo/internal/model/input/sysin"
|
||||
"hotgo/internal/service"
|
||||
|
||||
@@ -32,7 +31,7 @@ func (s *sSysImWorkOrder) List(ctx context.Context, in *sysin.ImWorkOrderListInp
|
||||
if err = in.Filter(ctx); err != nil {
|
||||
return
|
||||
}
|
||||
db, tenant, err := imtenant.DB(ctx, in.TenantId)
|
||||
db, tenant, err := resolveImTenantDB(ctx, in.TenantId)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
@@ -112,7 +111,7 @@ func (s *sSysImWorkOrder) View(ctx context.Context, in *sysin.ImWorkOrderViewInp
|
||||
if err = in.Filter(ctx); err != nil {
|
||||
return
|
||||
}
|
||||
db, tenant, err := imtenant.DB(ctx, in.TenantId)
|
||||
db, tenant, err := resolveImTenantDB(ctx, in.TenantId)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -190,7 +189,7 @@ func (s *sSysImWorkOrder) Status(ctx context.Context, in *sysin.ImWorkOrderStatu
|
||||
if err = in.Filter(ctx); err != nil {
|
||||
return
|
||||
}
|
||||
db, _, err := imtenant.DB(ctx, in.TenantId)
|
||||
db, _, err := resolveImTenantDB(ctx, in.TenantId)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user