330 lines
11 KiB
Go
330 lines
11 KiB
Go
package imkafka
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"hotgo/internal/consts"
|
|
"hotgo/internal/library/imtenant"
|
|
"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"
|
|
)
|
|
|
|
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 := adminDB()
|
|
err = db.Transaction(ctx, func(ctx context.Context, tx gdb.TX) error {
|
|
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后台命令失败")
|
|
}
|
|
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
|
|
})
|
|
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 {
|
|
row, err := adminDB().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() {
|
|
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失败")
|
|
}
|
|
nowTime := gtime.Now()
|
|
_, err = adminDB().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失败")
|
|
}
|
|
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失败")
|
|
}
|
|
nowTime := gtime.Now()
|
|
_, err = adminDB().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失败")
|
|
}
|
|
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
|
|
}
|
|
rows, err := adminDB().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)
|
|
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 := adminDB().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 = adminDB().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后台命令投递状态失败")
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func markOutboxFailed(ctx context.Context, eventID string, reason string, message string) error {
|
|
now := gtime.Now()
|
|
_, err := adminDB().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失败状态失败")
|
|
}
|
|
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)
|
|
}
|
|
whereColumn := "event_id"
|
|
whereValue := result.CommandEventID
|
|
if result.CommandID != "" {
|
|
whereColumn = "command_id"
|
|
whereValue = result.CommandID
|
|
}
|
|
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 := adminDB().Exec(ctx, sql, args...)
|
|
if err != nil {
|
|
return gerror.Wrap(err, "更新IM后台命令结果失败")
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func adminDB() gdb.DB {
|
|
return imtenant.RegistryDB()
|
|
}
|