// Package sys // @Link https://github.com/bufanyun/hotgo // @Copyright Copyright (c) 2026 HotGo // @Author Ms <133814250@qq.com> // @License https://github.com/bufanyun/hotgo/blob/master/LICENSE package sys import ( "context" "encoding/json" "fmt" "hotgo/internal/consts" "hotgo/internal/dao" "hotgo/internal/library/contexts" "hotgo/internal/library/imadmin" "hotgo/internal/library/imkafka" "hotgo/internal/library/imtenant" "hotgo/internal/model/entity" "hotgo/internal/model/input/sysin" "hotgo/internal/service" "net/url" "strconv" "github.com/gogf/gf/v2/errors/gerror" ) type sSysImDeadLetter struct{} func NewSysImDeadLetter() *sSysImDeadLetter { return &sSysImDeadLetter{} } func init() { service.RegisterSysImDeadLetter(NewSysImDeadLetter()) } // List 获取IM Kafka死信列表 func (s *sSysImDeadLetter) List(ctx context.Context, in *sysin.ImDeadLetterListInp) (res *sysin.ImDeadLetterPageModel, err error) { if err = in.Filter(ctx); err != nil { return nil, err } tenant, err := s.getTenant(ctx, in.TenantId) if err != nil { return nil, err } values := url.Values{} values.Set("limit", strconv.Itoa(in.Limit)) if in.Status != "" { values.Set("status", in.Status) } if in.BizType != "" { values.Set("biz_type", in.BizType) } if in.BizId != "" { values.Set("biz_id", in.BizId) } if in.EventType != "" { values.Set("event_type", in.EventType) } if in.Cursor != "" { values.Set("cursor", in.Cursor) } var raw sysin.ImDeadLetterPageRaw path := fmt.Sprintf("/internal/admin-query/tenants/%s/kafka/dead-letters?%s", url.PathEscape(tenant.Code), values.Encode()) if err = imadmin.DefaultClient(ctx).InternalGet(ctx, path, &raw); err != nil { return nil, err } res = &sysin.ImDeadLetterPageModel{ List: make([]*sysin.ImDeadLetterListModel, 0, len(raw.Items)), NextCursor: raw.NextCursor, HasMore: raw.HasMore, } for _, item := range raw.Items { res.List = append(res.List, convertDeadLetterModel(item)) } return res, nil } // View 获取IM Kafka死信详情 func (s *sSysImDeadLetter) View(ctx context.Context, in *sysin.ImDeadLetterViewInp) (res *sysin.ImDeadLetterViewModel, err error) { if err = in.Filter(ctx); err != nil { return nil, err } tenant, err := s.getTenant(ctx, in.TenantId) if err != nil { return nil, err } var raw sysin.ImDeadLetterViewRaw path := fmt.Sprintf("/internal/admin-query/tenants/%s/kafka/dead-letters/%s", url.PathEscape(tenant.Code), url.PathEscape(in.DeadId)) if err = imadmin.DefaultClient(ctx).InternalGet(ctx, path, &raw); err != nil { return nil, err } res = &sysin.ImDeadLetterViewModel{ ImDeadLetterListModel: *convertDeadLetterModel(&raw.ImDeadLetterListModelRaw), EnvelopeRaw: raw.EnvelopeRaw, } if raw.Envelope != nil { b, _ := json.MarshalIndent(raw.Envelope, "", " ") res.Envelope = string(b) } return res, nil } // Replay 重放IM Kafka死信 func (s *sSysImDeadLetter) Replay(ctx context.Context, in *sysin.ImDeadLetterActionInp) (res *sysin.ImGroupCommandModel, err error) { return s.publishActionCommand(ctx, in, consts.ImAdminCommandKafkaDeadReplay) } // Ignore 忽略IM Kafka死信 func (s *sSysImDeadLetter) Ignore(ctx context.Context, in *sysin.ImDeadLetterActionInp) (res *sysin.ImGroupCommandModel, err error) { return s.publishActionCommand(ctx, in, consts.ImAdminCommandKafkaDeadIgnore) } // Resolve 标记IM Kafka死信已处理 func (s *sSysImDeadLetter) Resolve(ctx context.Context, in *sysin.ImDeadLetterActionInp) (res *sysin.ImGroupCommandModel, err error) { return s.publishActionCommand(ctx, in, consts.ImAdminCommandKafkaDeadResolve) } func (s *sSysImDeadLetter) publishActionCommand(ctx context.Context, in *sysin.ImDeadLetterActionInp, eventType string) (*sysin.ImGroupCommandModel, error) { if err := in.Filter(ctx); err != nil { return nil, err } tenant, err := s.getTenant(ctx, in.TenantId) if err != nil { return nil, err } user := contexts.GetUser(ctx) actorName := "" if user != nil { actorName = user.RealName } command, err := imkafka.PublishAdminCommand(ctx, imkafka.AdminCommandInput{ TenantID: tenant.Id, TenantCode: tenant.Code, EventType: eventType, BizType: "kafka_dead_letter", BizID: in.DeadId, Actor: &imkafka.Actor{ Type: "admin", ID: strconv.FormatInt(contexts.GetUserId(ctx), 10), DisplayName: actorName, }, Payload: map[string]any{ "command_id": "", "actor_user_id": strconv.FormatInt(in.ActorUserId, 10), "dead_id": in.DeadId, "note": in.Note, }, CreatedBy: contexts.GetUserId(ctx), }) if err != nil { return nil, err } return &sysin.ImGroupCommandModel{ CommandID: command.CommandID, EventID: command.EventID, Status: command.Status, TenantID: command.TenantID, TenantCode: command.TenantCode, EventType: command.EventType, IdempotencyKey: command.IdempotencyKey, }, nil } func (s *sSysImDeadLetter) getTenant(ctx context.Context, tenantId int64) (tenant *entity.ImTenant, err error) { if tenantId <= 0 { return nil, gerror.New("请选择IM租户") } if err = imtenant.RegistryDB().Model(dao.ImTenant.Table()).Safe().Ctx(ctx).WherePri(tenantId).Scan(&tenant); err != nil { return nil, gerror.Wrap(err, "获取IM租户失败") } if tenant == nil || tenant.Id <= 0 { return nil, gerror.New("IM租户不存在") } if tenant.State != consts.ImTenantStateActive { return nil, gerror.New("IM租户未启用") } return tenant, nil } func convertDeadLetterModel(in *sysin.ImDeadLetterListModelRaw) *sysin.ImDeadLetterListModel { if in == nil { return &sysin.ImDeadLetterListModel{} } return &sysin.ImDeadLetterListModel{ Id: in.Id, DeadId: in.DeadId, EventId: in.EventId, ParentEventId: in.ParentEventId, SourceTopic: in.SourceTopic, ConsumerGroup: in.ConsumerGroup, EventType: in.EventType, BizType: in.BizType, BizId: in.BizId, RetryCount: in.RetryCount, LastReason: in.LastReason, LastError: in.LastError, Status: in.Status, ReplayCount: in.ReplayCount, CreatedAt: in.CreatedAt, UpdatedAt: in.UpdatedAt, } }