feat: expand IM admin controls
This commit is contained in:
richjohnrambo
@@ -11,6 +11,7 @@ import (
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"hotgo/internal/consts"
|
||||
"hotgo/internal/library/imtenant"
|
||||
"hotgo/internal/model/input/sysin"
|
||||
"hotgo/internal/service"
|
||||
@@ -248,6 +249,213 @@ func (s *sSysImUser) ResetPassword(ctx context.Context, in *sysin.ImUserPassword
|
||||
return err
|
||||
}
|
||||
|
||||
func (s *sSysImUser) RevokeSessions(ctx context.Context, in *sysin.ImUserRevokeSessionsInp) (err error) {
|
||||
if err = in.Filter(ctx); err != nil {
|
||||
return
|
||||
}
|
||||
scope, err := resolveImAdminScope(ctx, in.TenantId)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
userId, err := sysin.ParseOptionalInt64(in.UserId)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
db, _, err := imtenant.DB(ctx, scope.TenantId)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
err = revokeIMUserSessions(ctx, db, scope, userId, "security.sessions_revoked")
|
||||
result := "success"
|
||||
if err != nil {
|
||||
result = "failed"
|
||||
}
|
||||
_ = writeImAdminAudit(ctx, scope.TenantId, "user_sessions_revoke", "user", userId, result, nil)
|
||||
return err
|
||||
}
|
||||
|
||||
func (s *sSysImUser) Status(ctx context.Context, in *sysin.ImUserStatusInp) (err error) {
|
||||
if err = in.Filter(ctx); err != nil {
|
||||
return
|
||||
}
|
||||
if in.State != consts.ImUserStateOk && in.State != consts.ImUserStateSuspended {
|
||||
return gerror.New("账号状态只能设置为正常或已停用")
|
||||
}
|
||||
scope, err := resolveImAdminScope(ctx, in.TenantId)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
userId, err := sysin.ParseOptionalInt64(in.UserId)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
db, _, err := imtenant.DB(ctx, scope.TenantId)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
err = updateIMUserStatus(ctx, db, scope, userId, in.State)
|
||||
result := "success"
|
||||
if err != nil {
|
||||
result = "failed"
|
||||
}
|
||||
_ = writeImAdminAudit(ctx, scope.TenantId, "user_status_update", "user", userId, result, map[string]any{"state": in.State})
|
||||
return err
|
||||
}
|
||||
|
||||
func (s *sSysImUser) Devices(ctx context.Context, in *sysin.ImUserDeviceListInp) (list []*sysin.ImUserDeviceListModel, totalCount int, err error) {
|
||||
if err = in.Filter(ctx); err != nil {
|
||||
return
|
||||
}
|
||||
scope, err := resolveImAdminScope(ctx, in.TenantId)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
userId, err := sysin.ParseOptionalInt64(in.UserId)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
db, tenant, err := imtenant.DB(ctx, scope.TenantId)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
mod := db.Model("im_devices d").Safe().Ctx(ctx).
|
||||
LeftJoin("im_device_push_tokens p", "p.user_id=d.userid AND p.device_id=d.deviceid").
|
||||
Fields(
|
||||
"CAST(d.userid AS CHAR) user_id",
|
||||
"d.deviceid device_id",
|
||||
"COALESCE(d.platform, '') platform",
|
||||
"COALESCE(d.lang, '') lang",
|
||||
"d.lastseen last_seen",
|
||||
"COALESCE(p.provider, '') push_provider",
|
||||
"COALESCE(p.platform, '') push_platform",
|
||||
"COALESCE(p.app_version, '') push_app_version",
|
||||
"(p.id IS NOT NULL) push_token_configured",
|
||||
"p.updated_at push_updated_at",
|
||||
).
|
||||
Where("d.userid", userId).
|
||||
Page(in.Page, in.PerPage).
|
||||
OrderDesc("d.lastseen")
|
||||
|
||||
if err = mod.ScanAndCount(&list, &totalCount, false); err != nil {
|
||||
return nil, 0, gerror.Wrap(err, "获取IM用户设备列表失败，请稍后重试！")
|
||||
}
|
||||
for _, item := range list {
|
||||
item.TenantId = tenant.Id
|
||||
item.TenantName = tenant.Name
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
func updateIMUserStatus(ctx context.Context, db gdb.DB, scope imAdminScope, userId int64, state int) error {
|
||||
securityId, err := imtenant.NextSnowflakeID(ctx)
|
||||
if err != nil {
|
||||
return gerror.Wrap(err, "生成安全记录ID失败")
|
||||
}
|
||||
eventId, err := imtenant.NextSnowflakeID(ctx)
|
||||
if err != nil {
|
||||
return gerror.Wrap(err, "生成安全事件ID失败")
|
||||
}
|
||||
now := time.Now().UTC().Round(time.Millisecond)
|
||||
requestId := scope.RequestId
|
||||
if requestId == "" {
|
||||
requestId = gctx.CtxId(ctx)
|
||||
}
|
||||
if requestId == "" {
|
||||
requestId = guid.S()
|
||||
}
|
||||
eventKeyBytes := sha256.Sum256([]byte(fmt.Sprintf("user-status:%d:%d:%d:%s", scope.TenantId, userId, state, requestId)))
|
||||
eventKey := hex.EncodeToString(eventKeyBytes[:])
|
||||
return db.Transaction(ctx, func(ctx context.Context, tx gdb.TX) error {
|
||||
current, txErr := tx.GetValue("SELECT state FROM im_users WHERE id=? LIMIT 1", userId)
|
||||
if txErr != nil {
|
||||
return txErr
|
||||
}
|
||||
if current.IsNil() {
|
||||
return gerror.New("IM用户不存在")
|
||||
}
|
||||
if current.Int() == consts.ImUserStateDeleted {
|
||||
return gerror.New("已删除账号不能变更状态")
|
||||
}
|
||||
if current.Int() == state {
|
||||
return nil
|
||||
}
|
||||
if _, txErr = tx.Exec("UPDATE im_users SET state=?, updatedat=? WHERE id=?", state, now, userId); txErr != nil {
|
||||
return txErr
|
||||
}
|
||||
if _, txErr = tx.Exec(`INSERT INTO im_user_security
|
||||
(id,user_id,auth_version,created_at,updated_at)
|
||||
VALUES(?,?,?,?,?)
|
||||
ON DUPLICATE KEY UPDATE auth_version=auth_version+1,updated_at=VALUES(updated_at)`,
|
||||
securityId, userId, 2, now, now); txErr != nil {
|
||||
return txErr
|
||||
}
|
||||
version, txErr := tx.GetValue("SELECT auth_version FROM im_user_security WHERE user_id=?", userId)
|
||||
if txErr != nil {
|
||||
return txErr
|
||||
}
|
||||
payload, _ := json.Marshal(map[string]any{
|
||||
"user_id": userId, "operator_id": scope.OperatorId, "request_id": requestId, "state": state, "auth_version": version.Int(),
|
||||
})
|
||||
_, txErr = tx.Exec(`INSERT INTO im_business_events_outbox
|
||||
(id,event_id,target_type,target,actor_user_id,event_type,payload,status,created_at,updated_at)
|
||||
VALUES(?,?,?,?,?,?,CAST(? AS JSON),'pending',?,?)
|
||||
ON DUPLICATE KEY UPDATE event_id=VALUES(event_id)`, eventId, eventKey, "user",
|
||||
fmt.Sprintf("%d", userId), 0, "security.user_status_changed", string(payload), now, now)
|
||||
return txErr
|
||||
})
|
||||
}
|
||||
|
||||
func revokeIMUserSessions(ctx context.Context, db gdb.DB, scope imAdminScope, userId int64, eventType string) error {
|
||||
securityId, err := imtenant.NextSnowflakeID(ctx)
|
||||
if err != nil {
|
||||
return gerror.Wrap(err, "生成安全记录ID失败")
|
||||
}
|
||||
eventId, err := imtenant.NextSnowflakeID(ctx)
|
||||
if err != nil {
|
||||
return gerror.Wrap(err, "生成安全事件ID失败")
|
||||
}
|
||||
now := time.Now().UTC().Round(time.Millisecond)
|
||||
requestId := scope.RequestId
|
||||
if requestId == "" {
|
||||
requestId = gctx.CtxId(ctx)
|
||||
}
|
||||
if requestId == "" {
|
||||
requestId = guid.S()
|
||||
}
|
||||
eventKeyBytes := sha256.Sum256([]byte(fmt.Sprintf("%s:%d:%d:%s", eventType, scope.TenantId, userId, requestId)))
|
||||
eventKey := hex.EncodeToString(eventKeyBytes[:])
|
||||
return db.Transaction(ctx, func(ctx context.Context, tx gdb.TX) error {
|
||||
exists, txErr := tx.GetValue("SELECT EXISTS(SELECT 1 FROM im_users WHERE id=? LIMIT 1)", userId)
|
||||
if txErr != nil {
|
||||
return txErr
|
||||
}
|
||||
if !exists.Bool() {
|
||||
return gerror.New("IM用户不存在")
|
||||
}
|
||||
if _, txErr = tx.Exec(`INSERT INTO im_user_security
|
||||
(id,user_id,auth_version,created_at,updated_at)
|
||||
VALUES(?,?,?,?,?)
|
||||
ON DUPLICATE KEY UPDATE auth_version=auth_version+1,updated_at=VALUES(updated_at)`,
|
||||
securityId, userId, 2, now, now); txErr != nil {
|
||||
return txErr
|
||||
}
|
||||
version, txErr := tx.GetValue("SELECT auth_version FROM im_user_security WHERE user_id=?", userId)
|
||||
if txErr != nil {
|
||||
return txErr
|
||||
}
|
||||
payload, _ := json.Marshal(map[string]any{
|
||||
"user_id": userId, "operator_id": scope.OperatorId, "request_id": requestId, "auth_version": version.Int(),
|
||||
})
|
||||
_, txErr = tx.Exec(`INSERT INTO im_business_events_outbox
|
||||
(id,event_id,target_type,target,actor_user_id,event_type,payload,status,created_at,updated_at)
|
||||
VALUES(?,?,?,?,?,?,CAST(? AS JSON),'pending',?,?)
|
||||
ON DUPLICATE KEY UPDATE event_id=VALUES(event_id)`, eventId, eventKey, "user",
|
||||
fmt.Sprintf("%d", userId), 0, eventType, string(payload), now, now)
|
||||
return txErr
|
||||
})
|
||||
}
|
||||
|
||||
func resetIMUserPassword(ctx context.Context, db gdb.DB, scope imAdminScope, userId int64, newPassword string) error {
|
||||
passwordHash, err := bcrypt.GenerateFromPassword([]byte(newPassword), bcrypt.DefaultCost)
|
||||
if err != nil {
|
||||
|
||||
Reference in New Issue
Block a user