From f98f3e146b48cfb2f45ed0a4cc26ef5e0670f772 Mon Sep 17 00:00:00 2001 From: richjohnrambo Date: Fri, 28 Aug 2026 14:44:00 +0800 Subject: [PATCH] fix: use unprefixed im kafka tables --- server/internal/library/imkafka/command.go | 150 ++++++------------ server/internal/logic/sys/im_kafka_command.go | 134 ++++++++-------- 2 files changed, 116 insertions(+), 168 deletions(-) diff --git a/server/internal/library/imkafka/command.go b/server/internal/library/imkafka/command.go index fc38efd..4859204 100644 --- a/server/internal/library/imkafka/command.go +++ b/server/internal/library/imkafka/command.go @@ -67,39 +67,20 @@ func PublishAdminCommand(ctx context.Context, in AdminCommandInput) (res *AdminC db := g.DB() err = db.Transaction(ctx, func(ctx context.Context, tx gdb.TX) error { - data := g.Map{ - "command_id": commandID, - "tenant_id": in.TenantID, - "tenant_code": in.TenantCode, - "event_id": eventID, - "event_type": in.EventType, - "biz_type": in.BizType, - "biz_id": in.BizID, - "idempotency_key": in.IdempotencyKey, - "status": consts.ImAdminCommandStatusPending, - "payload": string(payloadBytes), - "envelope": string(envelopeBytes), - "created_by": in.CreatedBy, - "created_at": gtime.Now(), - "updated_at": gtime.Now(), - } - if _, err := tx.Model(adminCommandTable).Safe().Ctx(ctx).Data(data).Insert(); err != nil { + nowTime := gtime.Now() + if _, err := tx.Exec(`INSERT INTO im_admin_commands + (command_id,tenant_id,tenant_code,event_id,event_type,biz_type,biz_id,idempotency_key,status,payload,envelope,created_by,created_at,updated_at) + VALUES(?,?,?,?,?,?,?,?,?,CAST(? AS JSON),CAST(? AS JSON),?,?,?)`, + commandID, in.TenantID, in.TenantCode, eventID, in.EventType, in.BizType, in.BizID, + in.IdempotencyKey, consts.ImAdminCommandStatusPending, string(payloadBytes), string(envelopeBytes), + in.CreatedBy, nowTime, nowTime); err != nil { return gerror.Wrap(err, "创建IM后台命令失败") } - outbox := g.Map{ - "event_id": eventID, - "command_id": commandID, - "topic": consts.QueueImAdminCommandTopic, - "event_type": in.EventType, - "biz_type": in.BizType, - "biz_id": in.BizID, - "key_value": in.BizID, - "envelope": string(envelopeBytes), - "status": consts.ImAdminOutboxStatusPending, - "created_at": gtime.Now(), - "updated_at": gtime.Now(), - } - if _, err := tx.Model(adminOutboxTable).Safe().Ctx(ctx).Data(outbox).Insert(); err != nil { + if _, err := tx.Exec(`INSERT INTO im_admin_outbox + (event_id,command_id,topic,event_type,biz_type,biz_id,key_value,envelope,status,created_at,updated_at) + VALUES(?,?,?,?,?,?,?,CAST(? AS JSON),?,?,?)`, + eventID, commandID, consts.QueueImAdminCommandTopic, in.EventType, in.BizType, in.BizID, + in.BizID, string(envelopeBytes), consts.ImAdminOutboxStatusPending, nowTime, nowTime); err != nil { return gerror.Wrap(err, "创建IM后台命令Outbox失败") } return nil @@ -126,10 +107,8 @@ 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().Model(adminOutboxTable).Safe().Ctx(ctx). - Where("event_id", eventID). - Where("status", consts.ImAdminOutboxStatusPending). - Scan(&row); err != nil { + if err := g.DB().Raw(`SELECT * FROM im_admin_outbox WHERE event_id=? AND status=? LIMIT 1`, + eventID, consts.ImAdminOutboxStatusPending).Scan(&row); err != nil { return gerror.Wrap(err, "读取IM命令Outbox失败") } if row.IsEmpty() { @@ -171,19 +150,12 @@ func PublishConfigEvent(ctx context.Context, in ConfigEventInput) error { if err != nil { return gerror.Wrap(err, "序列化IM配置事件Envelope失败") } - _, err = g.DB().Model(adminOutboxTable).Safe().Ctx(ctx).Data(g.Map{ - "event_id": eventID, - "command_id": "", - "topic": consts.QueueImConfigEventTopic, - "event_type": in.EventType, - "biz_type": "config", - "biz_id": in.BizID, - "key_value": in.TenantCode, - "envelope": string(envelopeBytes), - "status": consts.ImAdminOutboxStatusPending, - "created_at": gtime.Now(), - "updated_at": gtime.Now(), - }).Insert() + nowTime := gtime.Now() + _, err = g.DB().Exec(ctx, `INSERT INTO im_admin_outbox + (event_id,command_id,topic,event_type,biz_type,biz_id,key_value,envelope,status,created_at,updated_at) + VALUES(?,?,?,?,?,?,?,CAST(? AS JSON),?,?,?)`, + eventID, "", consts.QueueImConfigEventTopic, in.EventType, "config", in.BizID, in.TenantCode, + string(envelopeBytes), consts.ImAdminOutboxStatusPending, nowTime, nowTime) if err != nil { return gerror.Wrap(err, "创建IM配置事件Outbox失败") } @@ -230,19 +202,12 @@ func PublishEvent(ctx context.Context, in EventInput) (eventID string, err error if err != nil { return "", gerror.Wrap(err, "序列化IM Kafka事件Envelope失败") } - _, err = g.DB().Model(adminOutboxTable).Safe().Ctx(ctx).Data(g.Map{ - "event_id": eventID, - "command_id": "", - "topic": in.Topic, - "event_type": in.EventType, - "biz_type": in.BizType, - "biz_id": in.BizID, - "key_value": in.BizID, - "envelope": string(envelopeBytes), - "status": consts.ImAdminOutboxStatusPending, - "created_at": gtime.Now(), - "updated_at": gtime.Now(), - }).Insert() + nowTime := gtime.Now() + _, err = g.DB().Exec(ctx, `INSERT INTO im_admin_outbox + (event_id,command_id,topic,event_type,biz_type,biz_id,key_value,envelope,status,created_at,updated_at) + VALUES(?,?,?,?,?,?,?,CAST(? AS JSON),?,?,?)`, + eventID, "", in.Topic, in.EventType, in.BizType, in.BizID, in.BizID, + string(envelopeBytes), consts.ImAdminOutboxStatusPending, nowTime, nowTime) if err != nil { return "", gerror.Wrap(err, "创建IM Kafka事件Outbox失败") } @@ -257,12 +222,9 @@ func DispatchPendingOutbox(ctx context.Context, limit int) (sent int, err error) limit = 100 } var rows []gdb.Record - err = g.DB().Model(adminOutboxTable).Safe().Ctx(ctx). - WhereIn("status", []string{consts.ImAdminOutboxStatusPending, consts.ImAdminOutboxStatusFailed}). - Where("next_retry_at IS NULL OR next_retry_at <= NOW()"). - OrderAsc("id"). - Limit(limit). - Scan(&rows) + err = g.DB().Raw(`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) if err != nil { return 0, gerror.Wrap(err, "读取待投递IM命令Outbox失败") } @@ -294,27 +256,16 @@ func dispatchOutboxRecord(ctx context.Context, row gdb.Record) error { return err } now := gtime.Now() - _, err := g.DB().Model(adminOutboxTable).Safe().Ctx(ctx). - Where("event_id", eventID). - Data(g.Map{ - "status": consts.ImAdminOutboxStatusSent, - "attempts": gdb.Raw("attempts + 1"), - "sent_at": now, - "last_reason": "", - "last_error": "", - "updated_at": now, - }).Update() + _, err := g.DB().Exec(ctx, `UPDATE im_admin_outbox + SET status=?,attempts=attempts+1,sent_at=?,last_reason='',last_error='',updated_at=? + WHERE event_id=?`, consts.ImAdminOutboxStatusSent, now, now, eventID) if err != nil { return gerror.Wrap(err, "更新IM命令Outbox投递状态失败") } if commandID != "" { - _, err = g.DB().Model(adminCommandTable).Safe().Ctx(ctx). - Where("command_id", commandID). - Data(g.Map{ - "status": consts.ImAdminCommandStatusRunning, - "sent_at": now, - "updated_at": now, - }).Update() + _, err = g.DB().Exec(ctx, `UPDATE im_admin_commands + SET status=?,sent_at=?,updated_at=? WHERE command_id=?`, + consts.ImAdminCommandStatusRunning, now, now, commandID) if err != nil { return gerror.Wrap(err, "更新IM后台命令投递状态失败") } @@ -324,16 +275,9 @@ func dispatchOutboxRecord(ctx context.Context, row gdb.Record) error { func markOutboxFailed(ctx context.Context, eventID string, reason string, message string) error { now := gtime.Now() - _, err := g.DB().Model(adminOutboxTable).Safe().Ctx(ctx). - Where("event_id", eventID). - Data(g.Map{ - "status": consts.ImAdminOutboxStatusFailed, - "attempts": gdb.Raw("attempts + 1"), - "next_retry_at": now.Add(30 * time.Second), - "last_reason": reason, - "last_error": message, - "updated_at": now, - }).Update() + _, err := g.DB().Exec(ctx, `UPDATE im_admin_outbox + SET status=?,attempts=attempts+1,next_retry_at=?,last_reason=?,last_error=?,updated_at=? + WHERE event_id=?`, consts.ImAdminOutboxStatusFailed, now.Add(30*time.Second), reason, message, now, eventID) if err != nil { return gerror.Wrap(err, "更新IM命令Outbox失败状态失败") } @@ -364,13 +308,21 @@ func ApplyCommandResult(ctx context.Context, result CommandResult) error { if len(result.Result) > 0 { data["result"] = string(result.Result) } - mod := g.DB().Model(adminCommandTable).Safe().Ctx(ctx) + whereColumn := "event_id" + whereValue := result.CommandEventID if result.CommandID != "" { - mod = mod.Where("command_id", result.CommandID) - } else { - mod = mod.Where("event_id", result.CommandEventID) + whereColumn = "command_id" + whereValue = result.CommandID } - _, err := mod.Data(data).Update() + sql := `UPDATE im_admin_commands SET status=?,reason=?,message=?,completed_at=?,updated_at=?` + args := []any{data["status"], data["reason"], data["message"], data["completed_at"], data["updated_at"]} + if resultValue, ok := data["result"]; ok { + sql += `,result=CAST(? AS JSON)` + args = append(args, resultValue) + } + sql += ` WHERE ` + whereColumn + `=?` + args = append(args, whereValue) + _, err := g.DB().Exec(ctx, sql, args...) if err != nil { return gerror.Wrap(err, "更新IM后台命令结果失败") } diff --git a/server/internal/logic/sys/im_kafka_command.go b/server/internal/logic/sys/im_kafka_command.go index 620eeea..cba6426 100644 --- a/server/internal/logic/sys/im_kafka_command.go +++ b/server/internal/logic/sys/im_kafka_command.go @@ -12,6 +12,7 @@ import ( "hotgo/internal/library/imkafka" "hotgo/internal/model/input/sysin" "hotgo/internal/service" + "strings" "github.com/gogf/gf/v2/errors/gerror" "github.com/gogf/gf/v2/frame/g" @@ -32,52 +33,25 @@ func (s *sSysImKafkaCommand) List(ctx context.Context, in *sysin.ImKafkaCommandL if err = in.Filter(ctx); err != nil { return } - mod := g.DB().Model("im_admin_commands").Safe().Ctx(ctx). - Fields( - "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", - ) - if in.TenantId > 0 { - mod = mod.Where("tenant_id", in.TenantId) + where, args := buildImKafkaCommandWhere(in) + if err = g.DB().Raw("SELECT COUNT(*) FROM im_admin_commands "+where, args...).Scan(&totalCount); err != nil { + err = gerror.Wrap(err, "获取IM后台命令总数失败，请稍后重试！") + return } - if in.CommandId != "" { - mod = mod.WhereLike("command_id", "%"+in.CommandId+"%") + page := in.Page + if page <= 0 { + page = 1 } - if in.EventId != "" { - mod = mod.WhereLike("event_id", "%"+in.EventId+"%") + perPage := in.PerPage + if perPage <= 0 { + perPage = 20 } - if in.EventType != "" { - mod = mod.WhereLike("event_type", "%"+in.EventType+"%") - } - if in.BizType != "" { - mod = mod.Where("biz_type", in.BizType) - } - if in.BizId != "" { - mod = mod.WhereLike("biz_id", "%"+in.BizId+"%") - } - 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]) - } - mod = mod.Page(in.Page, in.PerPage).OrderDesc("id") - if err = mod.ScanAndCount(&list, &totalCount, false); err != nil { + 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) + if err = g.DB().Raw(query, queryArgs...).Scan(&list); err != nil { err = gerror.Wrap(err, "获取IM后台命令列表失败，请稍后重试！") } return @@ -87,40 +61,62 @@ func (s *sSysImKafkaCommand) View(ctx context.Context, in *sysin.ImKafkaCommandV if err = in.Filter(ctx); err != nil { return nil, err } - mod := g.DB().Model("im_admin_commands").Safe().Ctx(ctx). - Fields( - "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", - ) + 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 { - mod = mod.Where("id", in.Id) + query += "id=?" + args = append(args, in.Id) } else { - mod = mod.Where("command_id", in.CommandId) + query += "command_id=?" + args = append(args, in.CommandId) } - if err = mod.Scan(&res); err != nil { + query += " LIMIT 1" + if err = g.DB().Raw(query, args...).Scan(&res); err != nil { return nil, gerror.Wrap(err, "获取IM后台命令详情失败，请稍后重试！") } return } +func buildImKafkaCommandWhere(in *sysin.ImKafkaCommandListInp) (string, []any) { + parts := []string{"1=1"} + args := []any{} + if in.TenantId > 0 { + parts = append(parts, "tenant_id=?") + args = append(args, in.TenantId) + } + 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