pro/im_manager
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: query im kafka tables with gf raw helpers

Browse Source
This commit is contained in:
richjohnrambo
2026-08-28 14:56:02 +08:00
parent f98f3e146b
commit 6b5ef0a911
2 changed files with 22 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/internal/library/imkafka/command.go
+5 -11
View File
@@ -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失败")
}
server/internal/logic/sys/im_kafka_command.go
+17 -3
View File
@@ -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
}