diff --git a/server/internal/library/imkafka/command.go b/server/internal/library/imkafka/command.go index 4859204..2cfcf85 100644 --- a/server/internal/library/imkafka/command.go +++ b/server/internal/library/imkafka/command.go @@ -16,11 +16,6 @@ import ( "github.com/gogf/gf/v2/util/guid" ) -const ( - adminCommandTable = "im_admin_commands" - adminOutboxTable = "im_admin_outbox" -) - func PublishAdminCommand(ctx context.Context, in AdminCommandInput) (res *AdminCommandModel, err error) { if in.TenantID <= 0 || in.TenantCode == "" { return nil, gerror.New("IM租户信息不完整") @@ -106,9 +101,9 @@ func PublishAdminCommand(ctx context.Context, in AdminCommandInput) (res *AdminC } func DispatchOutboxByEvent(ctx context.Context, eventID string) error { - var row gdb.Record - if err := g.DB().Raw(`SELECT * FROM im_admin_outbox WHERE event_id=? AND status=? LIMIT 1`, - eventID, consts.ImAdminOutboxStatusPending).Scan(&row); err != nil { + row, err := g.DB().GetOne(ctx, `SELECT * FROM im_admin_outbox WHERE event_id=? AND status=? LIMIT 1`, + eventID, consts.ImAdminOutboxStatusPending) + if err != nil { return gerror.Wrap(err, "读取IM命令Outbox失败") } if row.IsEmpty() { @@ -221,10 +216,9 @@ func DispatchPendingOutbox(ctx context.Context, limit int) (sent int, err error) if limit <= 0 { limit = 100 } - var rows []gdb.Record - err = g.DB().Raw(`SELECT * FROM im_admin_outbox + rows, err := g.DB().GetAll(ctx, `SELECT * FROM im_admin_outbox WHERE status IN (?,?) AND (next_retry_at IS NULL OR next_retry_at <= NOW()) - ORDER BY id ASC LIMIT ?`, consts.ImAdminOutboxStatusPending, consts.ImAdminOutboxStatusFailed, limit).Scan(&rows) + ORDER BY id ASC LIMIT ?`, consts.ImAdminOutboxStatusPending, consts.ImAdminOutboxStatusFailed, limit) if err != nil { return 0, gerror.Wrap(err, "读取待投递IM命令Outbox失败") } diff --git a/server/internal/logic/sys/im_kafka_command.go b/server/internal/logic/sys/im_kafka_command.go index cba6426..85c3495 100644 --- a/server/internal/logic/sys/im_kafka_command.go +++ b/server/internal/logic/sys/im_kafka_command.go @@ -34,10 +34,12 @@ func (s *sSysImKafkaCommand) List(ctx context.Context, in *sysin.ImKafkaCommandL return } where, args := buildImKafkaCommandWhere(in) - if err = g.DB().Raw("SELECT COUNT(*) FROM im_admin_commands "+where, args...).Scan(&totalCount); err != nil { + totalValue, err := g.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 @@ -51,8 +53,13 @@ func (s *sSysImKafkaCommand) List(ctx context.Context, in *sysin.ImKafkaCommandL 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) - if err = g.DB().Raw(query, queryArgs...).Scan(&list); err != nil { + rows, err := g.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 } @@ -73,9 +80,16 @@ func (s *sSysImKafkaCommand) View(ctx context.Context, in *sysin.ImKafkaCommandV args = append(args, in.CommandId) } query += " LIMIT 1" - if err = g.DB().Raw(query, args...).Scan(&res); err != nil { + row, err := g.DB().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后台命令详情失败,请稍后重试!") + } return }