// 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/consts" "hotgo/internal/library/contexts" "hotgo/internal/library/imkafka" "hotgo/internal/library/imtenant" "hotgo/internal/model/input/sysin" "hotgo/internal/service" "strings" "github.com/gogf/gf/v2/errors/gerror" "github.com/gogf/gf/v2/util/gconv" ) type sSysImKafkaCommand struct{} func NewSysImKafkaCommand() *sSysImKafkaCommand { return &sSysImKafkaCommand{} } func init() { service.RegisterSysImKafkaCommand(NewSysImKafkaCommand()) } func (s *sSysImKafkaCommand) List(ctx context.Context, in *sysin.ImKafkaCommandListInp) (list []*sysin.ImKafkaCommandListModel, totalCount int, err error) { if err = in.Filter(ctx); err != nil { return } // 强制租户范围:商户绑定账号忽略前端所选租户,平台管理员使用所选租户 scope, err := resolveImAdminScope(ctx, in.TenantId) if err != nil { return } where, args := buildImKafkaCommandWhere(in, scope.TenantId) db := imtenant.RegistryDB() totalValue, err := db.GetValue(ctx, "SELECT COUNT(*) FROM im_admin_commands "+where, args...) if err != nil { err = gerror.Wrap(err, "获取IM后台命令总数失败,请稍后重试!") return } totalCount = totalValue.Int() page := in.Page if page <= 0 { page = 1 } perPage := in.PerPage if perPage <= 0 { perPage = 20 } offset := (page - 1) * perPage query := `SELECT id,command_id,tenant_id,tenant_code,event_id,event_type,biz_type,biz_id,idempotency_key, status,reason,message,sent_at,completed_at,created_by,created_at,updated_at FROM im_admin_commands ` + where + ` ORDER BY id DESC LIMIT ? OFFSET ?` queryArgs := append(append([]any{}, args...), perPage, offset) rows, err := db.GetAll(ctx, query, queryArgs...) if err != nil { err = gerror.Wrap(err, "获取IM后台命令列表失败,请稍后重试!") return } if err = rows.Structs(&list); err != nil { err = gerror.Wrap(err, "解析IM后台命令列表失败,请稍后重试!") } return } func (s *sSysImKafkaCommand) View(ctx context.Context, in *sysin.ImKafkaCommandViewInp) (res *sysin.ImKafkaCommandViewModel, err error) { if err = in.Filter(ctx); err != nil { return nil, err } query := `SELECT id,command_id,tenant_id,tenant_code,event_id,event_type,biz_type,biz_id,idempotency_key, status,reason,message,CAST(payload AS CHAR) payload,CAST(envelope AS CHAR) envelope,CAST(result AS CHAR) result, sent_at,completed_at,created_by,created_at,updated_at FROM im_admin_commands WHERE ` args := []any{} if in.Id > 0 { query += "id=?" args = append(args, in.Id) } else { query += "command_id=?" args = append(args, in.CommandId) } query += " LIMIT 1" row, err := imtenant.RegistryDB().GetOne(ctx, query, args...) if err != nil { return nil, gerror.Wrap(err, "获取IM后台命令详情失败,请稍后重试!") } if row.IsEmpty() { return nil, nil } if err = row.Struct(&res); err != nil { return nil, gerror.Wrap(err, "解析IM后台命令详情失败,请稍后重试!") } // 后置校验记录归属租户是否在操作者允许范围内 if err = ensureImRecordTenantAllowed(ctx, res.TenantId); err != nil { return nil, err } return } func buildImKafkaCommandWhere(in *sysin.ImKafkaCommandListInp, scopeTenantId int64) (string, []any) { parts := []string{"tenant_id=?"} args := []any{scopeTenantId} if in.CommandId != "" { parts = append(parts, "command_id LIKE ?") args = append(args, "%"+in.CommandId+"%") } if in.EventId != "" { parts = append(parts, "event_id LIKE ?") args = append(args, "%"+in.EventId+"%") } if in.EventType != "" { parts = append(parts, "event_type LIKE ?") args = append(args, "%"+in.EventType+"%") } if in.BizType != "" { parts = append(parts, "biz_type=?") args = append(args, in.BizType) } if in.BizId != "" { parts = append(parts, "biz_id LIKE ?") args = append(args, "%"+in.BizId+"%") } if in.Status != "" { parts = append(parts, "status=?") args = append(args, in.Status) } if len(in.CreatedAt) == 2 { parts = append(parts, "created_at BETWEEN ? AND ?") args = append(args, in.CreatedAt[0], in.CreatedAt[1]) } return "WHERE " + strings.Join(parts, " AND "), args } func (s *sSysImKafkaCommand) CacheInvalidate(ctx context.Context, in *sysin.ImKafkaCacheInvalidateInp) (res *sysin.ImKafkaEventModel, err error) { if err = in.Filter(ctx); err != nil { return nil, err } // 校验目标租户是否在操作者允许范围内 tenant, err := lookupImTenantByCode(ctx, in.TenantCode) if err != nil { return nil, err } if err = ensureImRecordTenantAllowed(ctx, tenant.Id); err != nil { return nil, err } user := contexts.GetUser(ctx) actor := &imkafka.Actor{ Type: "admin", ID: gconv.String(contexts.GetUserId(ctx)), } if user != nil { actor.DisplayName = user.RealName } payload := map[string]any{ "tenant_code": in.TenantCode, "namespace": in.Namespace, "keys": in.Keys, "reason": in.Reason, } eventID, err := imkafka.PublishEvent(ctx, imkafka.EventInput{ Topic: consts.QueueImCacheInvalidateTopic, TenantCode: in.TenantCode, EventType: "cache.invalidate", BizType: "cache", BizID: in.Namespace, Actor: actor, Payload: payload, CreatedBy: contexts.GetUserId(ctx), }) if err != nil { return nil, err } return &sysin.ImKafkaEventModel{EventId: eventID}, nil } func (s *sSysImKafkaCommand) WorkbenchEvent(ctx context.Context, in *sysin.ImKafkaWorkbenchEventInp) (res *sysin.ImKafkaEventModel, err error) { if err = in.Filter(ctx); err != nil { return nil, err } // 校验目标租户是否在操作者允许范围内 tenant, err := lookupImTenantByCode(ctx, in.TenantCode) if err != nil { return nil, err } if err = ensureImRecordTenantAllowed(ctx, tenant.Id); err != nil { return nil, err } user := contexts.GetUser(ctx) actor := &imkafka.Actor{ Type: "admin", ID: gconv.String(contexts.GetUserId(ctx)), } if user != nil { actor.DisplayName = user.RealName } payload := in.Payload payload["tenant_code"] = in.TenantCode payload["source"] = "admin" eventID, err := imkafka.PublishEvent(ctx, imkafka.EventInput{ Topic: consts.QueueImWorkbenchEventTopic, TenantCode: in.TenantCode, EventType: in.EventType, BizType: "workbench", BizID: in.BizId, Actor: actor, Payload: payload, CreatedBy: contexts.GetUserId(ctx), }) if err != nil { return nil, err } return &sysin.ImKafkaEventModel{EventId: eventID}, nil }