379 lines
11 KiB
Go
379 lines
11 KiB
Go
package imkafka
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"hotgo/internal/consts"
|
|
"hotgo/internal/library/queue"
|
|
"time"
|
|
|
|
"github.com/gogf/gf/v2/database/gdb"
|
|
"github.com/gogf/gf/v2/errors/gerror"
|
|
"github.com/gogf/gf/v2/frame/g"
|
|
"github.com/gogf/gf/v2/os/gctx"
|
|
"github.com/gogf/gf/v2/os/gtime"
|
|
"github.com/gogf/gf/v2/util/gconv"
|
|
"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租户信息不完整")
|
|
}
|
|
if in.EventType == "" {
|
|
return nil, gerror.New("IM命令类型不能为空")
|
|
}
|
|
if in.IdempotencyKey == "" {
|
|
in.IdempotencyKey = "cmd_" + guid.S()
|
|
}
|
|
commandID := "cmd_" + guid.S()
|
|
eventID := "evt_" + guid.S()
|
|
now := time.Now().UTC().Round(time.Millisecond)
|
|
|
|
payload := in.Payload
|
|
if payload == nil {
|
|
payload = map[string]any{}
|
|
}
|
|
payload["command_id"] = commandID
|
|
|
|
payloadBytes, err := json.Marshal(payload)
|
|
if err != nil {
|
|
return nil, gerror.Wrap(err, "序列化IM命令Payload失败")
|
|
}
|
|
envelope := Envelope{
|
|
EventID: eventID,
|
|
TraceID: gctx.CtxId(ctx),
|
|
TenantCode: in.TenantCode,
|
|
SourceService: consts.ImKafkaSourceService,
|
|
TargetService: consts.ImKafkaTargetService,
|
|
EventType: in.EventType,
|
|
BizType: in.BizType,
|
|
BizID: in.BizID,
|
|
IdempotencyKey: in.IdempotencyKey,
|
|
Version: in.Version,
|
|
OccurredAt: now,
|
|
Actor: in.Actor,
|
|
Payload: payloadBytes,
|
|
}
|
|
envelopeBytes, err := json.Marshal(envelope)
|
|
if err != nil {
|
|
return nil, gerror.Wrap(err, "序列化IM命令Envelope失败")
|
|
}
|
|
|
|
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 {
|
|
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 {
|
|
return gerror.Wrap(err, "创建IM后台命令Outbox失败")
|
|
}
|
|
return nil
|
|
})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
res = &AdminCommandModel{
|
|
CommandID: commandID,
|
|
EventID: eventID,
|
|
Status: consts.ImAdminCommandStatusPending,
|
|
TenantID: in.TenantID,
|
|
TenantCode: in.TenantCode,
|
|
EventType: in.EventType,
|
|
IdempotencyKey: in.IdempotencyKey,
|
|
}
|
|
if err = DispatchOutboxByEvent(ctx, eventID); err != nil {
|
|
return res, err
|
|
}
|
|
res.Status = consts.ImAdminCommandStatusRunning
|
|
return res, nil
|
|
}
|
|
|
|
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 {
|
|
return gerror.Wrap(err, "读取IM命令Outbox失败")
|
|
}
|
|
if row.IsEmpty() {
|
|
return nil
|
|
}
|
|
return dispatchOutboxRecord(ctx, row)
|
|
}
|
|
|
|
func PublishConfigEvent(ctx context.Context, in ConfigEventInput) error {
|
|
if in.EventType == "" {
|
|
return gerror.New("IM配置事件类型不能为空")
|
|
}
|
|
eventID := "evt_" + guid.S()
|
|
now := time.Now().UTC().Round(time.Millisecond)
|
|
payload := map[string]any{
|
|
"tenant_code": in.TenantCode,
|
|
"tenant_id": in.TenantID,
|
|
"group": in.Group,
|
|
"keys": in.Keys,
|
|
}
|
|
payloadBytes, err := json.Marshal(payload)
|
|
if err != nil {
|
|
return gerror.Wrap(err, "序列化IM配置事件Payload失败")
|
|
}
|
|
envelope := Envelope{
|
|
EventID: eventID,
|
|
TraceID: gctx.CtxId(ctx),
|
|
TenantCode: in.TenantCode,
|
|
SourceService: consts.ImKafkaSourceService,
|
|
TargetService: consts.ImKafkaTargetService,
|
|
EventType: in.EventType,
|
|
BizType: "config",
|
|
BizID: in.BizID,
|
|
OccurredAt: now,
|
|
Actor: in.Actor,
|
|
Payload: payloadBytes,
|
|
}
|
|
envelopeBytes, err := json.Marshal(envelope)
|
|
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()
|
|
if err != nil {
|
|
return gerror.Wrap(err, "创建IM配置事件Outbox失败")
|
|
}
|
|
return DispatchOutboxByEvent(ctx, eventID)
|
|
}
|
|
|
|
func PublishEvent(ctx context.Context, in EventInput) (eventID string, err error) {
|
|
if in.Topic == "" {
|
|
return "", gerror.New("IM Kafka Topic不能为空")
|
|
}
|
|
if in.EventType == "" {
|
|
return "", gerror.New("IM Kafka事件类型不能为空")
|
|
}
|
|
eventID = "evt_" + guid.S()
|
|
now := time.Now().UTC().Round(time.Millisecond)
|
|
payload := in.Payload
|
|
if payload == nil {
|
|
payload = map[string]any{}
|
|
}
|
|
payloadBytes, err := json.Marshal(payload)
|
|
if err != nil {
|
|
return "", gerror.Wrap(err, "序列化IM Kafka事件Payload失败")
|
|
}
|
|
targetService := in.TargetService
|
|
if targetService == "" {
|
|
targetService = consts.ImKafkaTargetService
|
|
}
|
|
envelope := Envelope{
|
|
EventID: eventID,
|
|
TraceID: gctx.CtxId(ctx),
|
|
TenantCode: in.TenantCode,
|
|
SourceService: consts.ImKafkaSourceService,
|
|
TargetService: targetService,
|
|
EventType: in.EventType,
|
|
BizType: in.BizType,
|
|
BizID: in.BizID,
|
|
IdempotencyKey: in.IdempotencyKey,
|
|
Version: in.Version,
|
|
OccurredAt: now,
|
|
Actor: in.Actor,
|
|
Payload: payloadBytes,
|
|
}
|
|
envelopeBytes, err := json.Marshal(envelope)
|
|
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()
|
|
if err != nil {
|
|
return "", gerror.Wrap(err, "创建IM Kafka事件Outbox失败")
|
|
}
|
|
if err = DispatchOutboxByEvent(ctx, eventID); err != nil {
|
|
return eventID, err
|
|
}
|
|
return eventID, nil
|
|
}
|
|
|
|
func DispatchPendingOutbox(ctx context.Context, limit int) (sent int, err error) {
|
|
if limit <= 0 {
|
|
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)
|
|
if err != nil {
|
|
return 0, gerror.Wrap(err, "读取待投递IM命令Outbox失败")
|
|
}
|
|
for _, row := range rows {
|
|
if err = dispatchOutboxRecord(ctx, row); err != nil {
|
|
g.Log().Warningf(ctx, "IM命令Outbox投递失败 event:%s err:%+v", row["event_id"].String(), err)
|
|
continue
|
|
}
|
|
sent++
|
|
}
|
|
return sent, nil
|
|
}
|
|
|
|
func dispatchOutboxRecord(ctx context.Context, row gdb.Record) error {
|
|
eventID := row["event_id"].String()
|
|
commandID := row["command_id"].String()
|
|
topic := row["topic"].String()
|
|
body := row["envelope"].String()
|
|
if topic == "" || body == "" {
|
|
return markOutboxFailed(ctx, eventID, "MALFORMED_OUTBOX", "topic or envelope is empty")
|
|
}
|
|
if g.Cfg().MustGet(ctx, "queue.driver").String() != "kafka" {
|
|
err := gerror.New("IM Kafka未启用，请将queue.driver配置为kafka")
|
|
_ = markOutboxFailed(ctx, eventID, "KAFKA_DISABLED", err.Error())
|
|
return err
|
|
}
|
|
if err := queue.Push(topic, body); err != nil {
|
|
_ = markOutboxFailed(ctx, eventID, "KAFKA_SEND_FAILED", err.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()
|
|
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()
|
|
if err != nil {
|
|
return gerror.Wrap(err, "更新IM后台命令投递状态失败")
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
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()
|
|
if err != nil {
|
|
return gerror.Wrap(err, "更新IM命令Outbox失败状态失败")
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func ApplyCommandResult(ctx context.Context, result CommandResult) error {
|
|
commandID := result.CommandID
|
|
if commandID == "" {
|
|
commandID = result.CommandEventID
|
|
}
|
|
if commandID == "" && result.CommandEventID == "" {
|
|
return gerror.New("IM命令结果缺少command_id")
|
|
}
|
|
status := gconv.String(result.Status)
|
|
switch status {
|
|
case consts.ImAdminCommandStatusSucceeded, consts.ImAdminCommandStatusFailed, consts.ImAdminCommandStatusDead:
|
|
default:
|
|
status = consts.ImAdminCommandStatusFailed
|
|
}
|
|
data := g.Map{
|
|
"status": status,
|
|
"reason": result.Reason,
|
|
"message": result.Message,
|
|
"completed_at": gtime.Now(),
|
|
"updated_at": gtime.Now(),
|
|
}
|
|
if len(result.Result) > 0 {
|
|
data["result"] = string(result.Result)
|
|
}
|
|
mod := g.DB().Model(adminCommandTable).Safe().Ctx(ctx)
|
|
if result.CommandID != "" {
|
|
mod = mod.Where("command_id", result.CommandID)
|
|
} else {
|
|
mod = mod.Where("event_id", result.CommandEventID)
|
|
}
|
|
_, err := mod.Data(data).Update()
|
|
if err != nil {
|
|
return gerror.Wrap(err, "更新IM后台命令结果失败")
|
|
}
|
|
return nil
|
|
}
|