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

fix: use unprefixed im kafka tables

Browse Source
This commit is contained in:
richjohnrambo
2026-08-28 14:44:00 +08:00
parent cde4480b51
commit f98f3e146b
2 changed files with 116 additions and 168 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/internal/library/imkafka/command.go
+51 -99
View File
@@ -67,39 +67,20 @@ func PublishAdminCommand(ctx context.Context, in AdminCommandInput) (res *AdminC
db := g.DB() db := g.DB()
err = db.Transaction(ctx, func(ctx context.Context, tx gdb.TX) error { err = db.Transaction(ctx, func(ctx context.Context, tx gdb.TX) error {
data := g.Map{ nowTime := gtime.Now()
"command_id": commandID, if _, err := tx.Exec(`INSERT INTO im_admin_commands
"tenant_id": in.TenantID, (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)
"tenant_code": in.TenantCode, VALUES(?,?,?,?,?,?,?,?,?,CAST(? AS JSON),CAST(? AS JSON),?,?,?)`,
"event_id": eventID, commandID, in.TenantID, in.TenantCode, eventID, in.EventType, in.BizType, in.BizID,
"event_type": in.EventType, in.IdempotencyKey, consts.ImAdminCommandStatusPending, string(payloadBytes), string(envelopeBytes),
"biz_type": in.BizType, in.CreatedBy, nowTime, nowTime); err != nil {
"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 {
return gerror.Wrap(err, "创建IM后台命令失败") return gerror.Wrap(err, "创建IM后台命令失败")
} }
outbox := g.Map{ if _, err := tx.Exec(`INSERT INTO im_admin_outbox
"event_id": eventID, (event_id,command_id,topic,event_type,biz_type,biz_id,key_value,envelope,status,created_at,updated_at)
"command_id": commandID, VALUES(?,?,?,?,?,?,?,CAST(? AS JSON),?,?,?)`,
"topic": consts.QueueImAdminCommandTopic, eventID, commandID, consts.QueueImAdminCommandTopic, in.EventType, in.BizType, in.BizID,
"event_type": in.EventType, in.BizID, string(envelopeBytes), consts.ImAdminOutboxStatusPending, nowTime, nowTime); err != nil {
"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 {
return gerror.Wrap(err, "创建IM后台命令Outbox失败") return gerror.Wrap(err, "创建IM后台命令Outbox失败")
} }
return nil return nil
@@ -126,10 +107,8 @@ func PublishAdminCommand(ctx context.Context, in AdminCommandInput) (res *AdminC
func DispatchOutboxByEvent(ctx context.Context, eventID string) error { func DispatchOutboxByEvent(ctx context.Context, eventID string) error {
var row gdb.Record var row gdb.Record
if err := g.DB().Model(adminOutboxTable).Safe().Ctx(ctx). if err := g.DB().Raw(`SELECT * FROM im_admin_outbox WHERE event_id=? AND status=? LIMIT 1`,
Where("event_id", eventID). eventID, consts.ImAdminOutboxStatusPending).Scan(&row); err != nil {
Where("status", consts.ImAdminOutboxStatusPending).
Scan(&row); err != nil {
return gerror.Wrap(err, "读取IM命令Outbox失败") return gerror.Wrap(err, "读取IM命令Outbox失败")
} }
if row.IsEmpty() { if row.IsEmpty() {
@@ -171,19 +150,12 @@ func PublishConfigEvent(ctx context.Context, in ConfigEventInput) error {
if err != nil { if err != nil {
return gerror.Wrap(err, "序列化IM配置事件Envelope失败") return gerror.Wrap(err, "序列化IM配置事件Envelope失败")
} }
_, err = g.DB().Model(adminOutboxTable).Safe().Ctx(ctx).Data(g.Map{ nowTime := gtime.Now()
"event_id": eventID, _, err = g.DB().Exec(ctx, `INSERT INTO im_admin_outbox
"command_id": "", (event_id,command_id,topic,event_type,biz_type,biz_id,key_value,envelope,status,created_at,updated_at)
"topic": consts.QueueImConfigEventTopic, VALUES(?,?,?,?,?,?,?,CAST(? AS JSON),?,?,?)`,
"event_type": in.EventType, eventID, "", consts.QueueImConfigEventTopic, in.EventType, "config", in.BizID, in.TenantCode,
"biz_type": "config", string(envelopeBytes), consts.ImAdminOutboxStatusPending, nowTime, nowTime)
"biz_id": in.BizID,
"key_value": in.TenantCode,
"envelope": string(envelopeBytes),
"status": consts.ImAdminOutboxStatusPending,
"created_at": gtime.Now(),
"updated_at": gtime.Now(),
}).Insert()
if err != nil { if err != nil {
return gerror.Wrap(err, "创建IM配置事件Outbox失败") return gerror.Wrap(err, "创建IM配置事件Outbox失败")
} }
@@ -230,19 +202,12 @@ func PublishEvent(ctx context.Context, in EventInput) (eventID string, err error
if err != nil { if err != nil {
return "", gerror.Wrap(err, "序列化IM Kafka事件Envelope失败") return "", gerror.Wrap(err, "序列化IM Kafka事件Envelope失败")
} }
_, err = g.DB().Model(adminOutboxTable).Safe().Ctx(ctx).Data(g.Map{ nowTime := gtime.Now()
"event_id": eventID, _, err = g.DB().Exec(ctx, `INSERT INTO im_admin_outbox
"command_id": "", (event_id,command_id,topic,event_type,biz_type,biz_id,key_value,envelope,status,created_at,updated_at)
"topic": in.Topic, VALUES(?,?,?,?,?,?,?,CAST(? AS JSON),?,?,?)`,
"event_type": in.EventType, eventID, "", in.Topic, in.EventType, in.BizType, in.BizID, in.BizID,
"biz_type": in.BizType, string(envelopeBytes), consts.ImAdminOutboxStatusPending, nowTime, nowTime)
"biz_id": in.BizID,
"key_value": in.BizID,
"envelope": string(envelopeBytes),
"status": consts.ImAdminOutboxStatusPending,
"created_at": gtime.Now(),
"updated_at": gtime.Now(),
}).Insert()
if err != nil { if err != nil {
return "", gerror.Wrap(err, "创建IM Kafka事件Outbox失败") return "", gerror.Wrap(err, "创建IM Kafka事件Outbox失败")
} }
@@ -257,12 +222,9 @@ func DispatchPendingOutbox(ctx context.Context, limit int) (sent int, err error)
limit = 100 limit = 100
} }
var rows []gdb.Record var rows []gdb.Record
err = g.DB().Model(adminOutboxTable).Safe().Ctx(ctx). err = g.DB().Raw(`SELECT * FROM im_admin_outbox
WhereIn("status", []string{consts.ImAdminOutboxStatusPending, consts.ImAdminOutboxStatusFailed}). WHERE status IN (?,?) AND (next_retry_at IS NULL OR next_retry_at <= NOW())
Where("next_retry_at IS NULL OR next_retry_at <= NOW()"). ORDER BY id ASC LIMIT ?`, consts.ImAdminOutboxStatusPending, consts.ImAdminOutboxStatusFailed, limit).Scan(&rows)
OrderAsc("id").
Limit(limit).
Scan(&rows)
if err != nil { if err != nil {
return 0, gerror.Wrap(err, "读取待投递IM命令Outbox失败") return 0, gerror.Wrap(err, "读取待投递IM命令Outbox失败")
} }
@@ -294,27 +256,16 @@ func dispatchOutboxRecord(ctx context.Context, row gdb.Record) error {
return err return err
} }
now := gtime.Now() now := gtime.Now()
_, err := g.DB().Model(adminOutboxTable).Safe().Ctx(ctx). _, err := g.DB().Exec(ctx, `UPDATE im_admin_outbox
Where("event_id", eventID). SET status=?,attempts=attempts+1,sent_at=?,last_reason='',last_error='',updated_at=?
Data(g.Map{ WHERE event_id=?`, consts.ImAdminOutboxStatusSent, now, now, eventID)
"status": consts.ImAdminOutboxStatusSent,
"attempts": gdb.Raw("attempts + 1"),
"sent_at": now,
"last_reason": "",
"last_error": "",
"updated_at": now,
}).Update()
if err != nil { if err != nil {
return gerror.Wrap(err, "更新IM命令Outbox投递状态失败") return gerror.Wrap(err, "更新IM命令Outbox投递状态失败")
} }
if commandID != "" { if commandID != "" {
_, err = g.DB().Model(adminCommandTable).Safe().Ctx(ctx). _, err = g.DB().Exec(ctx, `UPDATE im_admin_commands
Where("command_id", commandID). SET status=?,sent_at=?,updated_at=? WHERE command_id=?`,
Data(g.Map{ consts.ImAdminCommandStatusRunning, now, now, commandID)
"status": consts.ImAdminCommandStatusRunning,
"sent_at": now,
"updated_at": now,
}).Update()
if err != nil { if err != nil {
return gerror.Wrap(err, "更新IM后台命令投递状态失败") 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 { func markOutboxFailed(ctx context.Context, eventID string, reason string, message string) error {
now := gtime.Now() now := gtime.Now()
_, err := g.DB().Model(adminOutboxTable).Safe().Ctx(ctx). _, err := g.DB().Exec(ctx, `UPDATE im_admin_outbox
Where("event_id", eventID). SET status=?,attempts=attempts+1,next_retry_at=?,last_reason=?,last_error=?,updated_at=?
Data(g.Map{ WHERE event_id=?`, consts.ImAdminOutboxStatusFailed, now.Add(30*time.Second), reason, message, now, eventID)
"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()
if err != nil { if err != nil {
return gerror.Wrap(err, "更新IM命令Outbox失败状态失败") return gerror.Wrap(err, "更新IM命令Outbox失败状态失败")
} }
@@ -364,13 +308,21 @@ func ApplyCommandResult(ctx context.Context, result CommandResult) error {
if len(result.Result) > 0 { if len(result.Result) > 0 {
data["result"] = string(result.Result) data["result"] = string(result.Result)
} }
mod := g.DB().Model(adminCommandTable).Safe().Ctx(ctx) whereColumn := "event_id"
whereValue := result.CommandEventID
if result.CommandID != "" { if result.CommandID != "" {
mod = mod.Where("command_id", result.CommandID) whereColumn = "command_id"
} else { whereValue = result.CommandID
mod = mod.Where("event_id", result.CommandEventID)
} }
_, 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 { if err != nil {
return gerror.Wrap(err, "更新IM后台命令结果失败") return gerror.Wrap(err, "更新IM后台命令结果失败")
} }
server/internal/logic/sys/im_kafka_command.go
+65 -69
View File
@@ -12,6 +12,7 @@ import (
"hotgo/internal/library/imkafka" "hotgo/internal/library/imkafka"
"hotgo/internal/model/input/sysin" "hotgo/internal/model/input/sysin"
"hotgo/internal/service" "hotgo/internal/service"
"strings"
"github.com/gogf/gf/v2/errors/gerror" "github.com/gogf/gf/v2/errors/gerror"
"github.com/gogf/gf/v2/frame/g" "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 { if err = in.Filter(ctx); err != nil {
return return
} }
mod := g.DB().Model("im_admin_commands").Safe().Ctx(ctx). where, args := buildImKafkaCommandWhere(in)
Fields( if err = g.DB().Raw("SELECT COUNT(*) FROM im_admin_commands "+where, args...).Scan(&totalCount); err != nil {
"id", err = gerror.Wrap(err, "获取IM后台命令总数失败，请稍后重试！")
"command_id", return
"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)
} }
if in.CommandId != "" { page := in.Page
mod = mod.WhereLike("command_id", "%"+in.CommandId+"%") if page <= 0 {
page = 1
} }
if in.EventId != "" { perPage := in.PerPage
mod = mod.WhereLike("event_id", "%"+in.EventId+"%") if perPage <= 0 {
perPage = 20
} }
if in.EventType != "" { offset := (page - 1) * perPage
mod = mod.WhereLike("event_type", "%"+in.EventType+"%") 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
if in.BizType != "" { FROM im_admin_commands ` + where + ` ORDER BY id DESC LIMIT ? OFFSET ?`
mod = mod.Where("biz_type", in.BizType) queryArgs := append(append([]any{}, args...), perPage, offset)
} if err = g.DB().Raw(query, queryArgs...).Scan(&list); err != nil {
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 {
err = gerror.Wrap(err, "获取IM后台命令列表失败，请稍后重试！") err = gerror.Wrap(err, "获取IM后台命令列表失败，请稍后重试！")
} }
return return
@@ -87,40 +61,62 @@ func (s *sSysImKafkaCommand) View(ctx context.Context, in *sysin.ImKafkaCommandV
if err = in.Filter(ctx); err != nil { if err = in.Filter(ctx); err != nil {
return nil, err return nil, err
} }
mod := g.DB().Model("im_admin_commands").Safe().Ctx(ctx). query := `SELECT id,command_id,tenant_id,tenant_code,event_id,event_type,biz_type,biz_id,idempotency_key,
Fields( status,reason,message,CAST(payload AS CHAR) payload,CAST(envelope AS CHAR) envelope,CAST(result AS CHAR) result,
"id", sent_at,completed_at,created_by,created_at,updated_at FROM im_admin_commands WHERE `
"command_id", args := []any{}
"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",
)
if in.Id > 0 { if in.Id > 0 {
mod = mod.Where("id", in.Id) query += "id=?"
args = append(args, in.Id)
} else { } 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 nil, gerror.Wrap(err, "获取IM后台命令详情失败，请稍后重试！")
} }
return 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) { func (s *sSysImKafkaCommand) CacheInvalidate(ctx context.Context, in *sysin.ImKafkaCacheInvalidateInp) (res *sysin.ImKafkaEventModel, err error) {
if err = in.Filter(ctx); err != nil { if err = in.Filter(ctx); err != nil {
return nil, err return nil, err