im_manager/server/internal/logic/sys/im_dead_letter.go

// 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,
	}
}