im_server/server/kafka_admin_commands.go

package main

import (
	"context"
	"encoding/base64"
	"encoding/json"
	"errors"
	"strconv"
	"strings"
	"time"

	"github.com/segmentio/kafka-go"
	"github.com/tinode/chat/server/logs"
	"github.com/tinode/chat/server/store"
	"github.com/tinode/chat/server/store/types"
)

const (
	adminCommandGroupProfileUpdate = "group.profile.update"
	adminCommandGroupGovernUpdate  = "group.governance.update"
	adminCommandGroupAdminSet      = "group.admin.set"
	adminCommandGroupAdminRemove   = "group.admin.remove"
	adminCommandGroupMemberMute    = "group.member.mute"
	adminCommandGroupMemberUnmute  = "group.member.unmute"
	adminCommandGroupMemberRemove  = "group.member.remove"
	adminCommandGroupOwnerTransfer = "group.owner.transfer"
	adminCommandUserBan            = "user.ban"
	adminCommandUserUnban          = "user.unban"
	adminCommandUserFreeze         = "user.freeze"
	adminCommandUserUnfreeze       = "user.unfreeze"
	adminCommandUserDeviceKick     = "user.device.kick"
	adminCommandUserLogoutAll      = "user.logout_all"
	adminCommandUserCancel         = "user.account.cancel"
	adminCommandUserDelete         = "user.account.delete"
	adminCommandUserPasswordReset  = "user.password.reset"
	adminCommandMessageHide        = "message.hide"
	adminCommandMessageDelete      = "message.delete"
	adminCommandKafkaDeadReplay    = "kafka.dead_letter.replay"
	adminCommandKafkaDeadIgnore    = "kafka.dead_letter.ignore"
	adminCommandKafkaDeadResolve   = "kafka.dead_letter.resolve"
)

var publishKafkaDeadLetter = publishKafkaEnvelope

type adminCommandPayload struct {
	CommandID        string               `json:"command_id"`
	ActorUserID      string               `json:"actor_user_id"`
	Topic            string               `json:"topic"`
	ConversationType string               `json:"conversation_type"`
	SeqID            int64                `json:"seq_id"`
	UserID           string               `json:"user_id"`
	UserIDs          []string             `json:"user_ids"`
	TargetID         string               `json:"target_user_id"`
	DeviceID         string               `json:"device_id"`
	PasswordHash     string               `json:"password_hash"`
	Version          int                  `json:"version"`
	Reason           string               `json:"reason"`
	DeadID           string               `json:"dead_id"`
	Note             string               `json:"note"`
	Hard             bool                 `json:"hard"`
	ExpiresAt        *time.Time           `json:"expires_at"`
	Profile          groupProfilePatch    `json:"profile"`
	Governance       groupGovernancePatch `json:"governance"`
}

type groupProfilePatch struct {
	Name         *string `json:"name"`
	AvatarFileID *string `json:"avatar_file_id"`
	Description  *string `json:"description"`
}

type groupGovernancePatch struct {
	JoinApproval      *bool `json:"join_approval"`
	MuteAll           *bool `json:"mute_all"`
	MemberUploadFiles *bool `json:"member_upload_files"`
	MemberPrivateChat *bool `json:"member_private_chat"`
	MemberInvite      *bool `json:"member_invite"`
}

type adminCommandResultPayload struct {
	CommandID      string         `json:"command_id"`
	CommandEventID string         `json:"command_event_id"`
	Status         string         `json:"status"`
	Reason         string         `json:"reason,omitempty"`
	Message        string         `json:"message,omitempty"`
	Result         map[string]any `json:"result,omitempty"`
}

func startKafkaAdminCommandConsumer() {
	startKafkaConsumer("admin_command", handleKafkaAdminCommand)
}

func handleKafkaAdminCommand(ctx context.Context, envelope kafkaEnvelope, _ kafka.Message) error {
	if strings.TrimSpace(envelope.EventID) == "" {
		return publishKafkaAdminCommandResult(ctx, envelope, "", "failed", "MALFORMED_EVENT", "missing event_id", nil)
	}
	if strings.TrimSpace(envelope.EventType) == "" {
		return publishKafkaAdminCommandResult(ctx, envelope, "", "failed", "MALFORMED_EVENT", "missing event_type", nil)
	}
	var payload adminCommandPayload
	if len(envelope.Payload) > 0 {
		if err := json.Unmarshal(envelope.Payload, &payload); err != nil {
			return publishKafkaAdminCommandResult(ctx, envelope, "", "failed", "MALFORMED_PAYLOAD", err.Error(), nil)
		}
	}
	if payload.CommandID == "" {
		payload.CommandID = envelope.EventID
	}

	tenantID, tenantErr := kafkaTenantID(envelope.TenantCode)
	if tenantErr != nil {
		return publishKafkaAdminCommandResult(ctx, envelope, payload.CommandID, "failed", kafkaCommandErrorReason(tenantErr), tenantErr.Error(), nil)
	}
	consume, claimed, beginErr := beginKafkaConsumeForTopic(tenantID, envelope, "admin_command")
	if beginErr != nil {
		return beginErr
	}
	if !claimed {
		logs.Info.Printf("kafka admin command skipped: event=%s group=%s status=%s", envelope.EventID, kafkaBus.config.ConsumerGroup, consume.Status)
		return nil
	}

	result, execErr := executeKafkaAdminCommand(tenantID, envelope, payload)
	if execErr != nil {
		reason := kafkaCommandErrorReason(execErr)
		failedRecord, markErr := store.HTTPInfrastructure.MarkKafkaConsumeFailed(tenantID, envelope.EventID, kafkaBus.config.ConsumerGroup, reason, execErr.Error(), time.Now().UTC().Round(time.Millisecond))
		if markErr != nil {
			return markErr
		}
		if shouldDeadLetterKafkaCommand(reason, failedRecord.RetryCount) {
			if deadErr := markKafkaDeadLetter(ctx, tenantID, envelope, failedRecord, reason, execErr.Error()); deadErr != nil {
				return deadErr
			}
			if err := publishKafkaAdminCommandResult(ctx, envelope, payload.CommandID, "dead", reason, execErr.Error(), result); err != nil {
				return err
			}
			logs.Warn.Printf("kafka admin command dead: event=%s type=%s reason=%s err=%v", envelope.EventID, envelope.EventType, reason, execErr)
			return nil
		}
		if kafkaCommandReasonRetryable(reason) {
			sleepKafkaRetryBackoff(failedRecord.RetryCount)
			return execErr
		}
		if err := publishKafkaAdminCommandResult(ctx, envelope, payload.CommandID, "failed", reason, execErr.Error(), result); err != nil {
			return err
		}
		logs.Warn.Printf("kafka admin command failed: event=%s type=%s reason=%s err=%v", envelope.EventID, envelope.EventType, reason, execErr)
		return nil
	}
	if err := store.HTTPInfrastructure.MarkKafkaConsumeSucceeded(tenantID, envelope.EventID, kafkaBus.config.ConsumerGroup, time.Now().UTC().Round(time.Millisecond)); err != nil {
		return err
	}
	if err := publishKafkaAdminCommandResult(ctx, envelope, payload.CommandID, "succeeded", "", "", result); err != nil {
		return err
	}
	if err := enqueueKafkaAuditEvent(ctx, tenantID, "admin", parseKafkaCommandUser(payload.ActorUserID),
		envelope.EventType, "admin_command", payload.CommandID, "succeeded", "", map[string]any{
			"topic":    payload.Topic,
			"biz_id":   envelope.BizID,
			"event_id": envelope.EventID,
			"result":   result,
		}); err != nil {
		logs.Warn.Printf("kafka admin command audit enqueue failed: event=%s err=%v", envelope.EventID, err)
	}
	return nil
}

func executeKafkaAdminCommand(tenantID types.TenantID, envelope kafkaEnvelope, payload adminCommandPayload) (map[string]any, error) {
	eventType := strings.TrimSpace(envelope.EventType)
	if eventType == adminCommandUserPasswordReset {
		return executeKafkaUserPasswordResetCommand(tenantID, payload)
	}
	if eventType == adminCommandMessageHide || eventType == adminCommandMessageDelete {
		return executeKafkaMessageAdminCommand(tenantID, envelope, payload)
	}
	if eventType == adminCommandUserBan || eventType == adminCommandUserUnban ||
		eventType == adminCommandUserFreeze || eventType == adminCommandUserUnfreeze ||
		eventType == adminCommandUserDeviceKick || eventType == adminCommandUserLogoutAll ||
		eventType == adminCommandUserCancel || eventType == adminCommandUserDelete {
		return executeKafkaUserGovernanceCommand(tenantID, envelope, payload, types.ZeroUid)
	}
	actor := parseKafkaCommandUser(payload.ActorUserID)
	if actor.IsZero() {
		return nil, types.ErrMalformed
	}
	switch eventType {
	case adminCommandKafkaDeadReplay:
		dead, err := replayKafkaDeadLetter(context.Background(), tenantID, payload.DeadID, actor, firstNonEmpty(payload.Note, payload.Reason))
		if err != nil {
			return nil, err
		}
		return kafkaDeadLetterResult(dead), nil
	case adminCommandKafkaDeadIgnore:
		if err := resolveKafkaDeadLetter(tenantID, payload.DeadID, types.KafkaDeadStatusIgnored, actor, firstNonEmpty(payload.Note, payload.Reason)); err != nil {
			return nil, err
		}
		return map[string]any{"dead_id": payload.DeadID, "status": types.KafkaDeadStatusIgnored}, nil
	case adminCommandKafkaDeadResolve:
		if err := resolveKafkaDeadLetter(tenantID, payload.DeadID, types.KafkaDeadStatusResolved, actor, firstNonEmpty(payload.Note, payload.Reason)); err != nil {
			return nil, err
		}
		return map[string]any{"dead_id": payload.DeadID, "status": types.KafkaDeadStatusResolved}, nil
	}
	if store.GroupBusiness == nil {
		return nil, errors.New("group business store is not initialized")
	}
	topic := strings.TrimSpace(payload.Topic)
	if topic == "" {
		return nil, types.ErrMalformed
	}
	now := time.Now().UTC().Round(time.Millisecond)

	switch strings.TrimSpace(envelope.EventType) {
	case adminCommandGroupProfileUpdate:
		settings, err := store.GroupBusiness.UpdateSettings(tenantID, topic, types.GroupSettingsPatch{
			Name:         payload.Profile.Name,
			AvatarFileID: payload.Profile.AvatarFileID,
			Description:  payload.Profile.Description,
			Version:      payload.Version,
		})
		if err != nil {
			return nil, err
		}
		dispatchIMGroupEvents(tenantID)
		return map[string]any{"topic": topic, "version": settings.Version, "updated_at": settings.UpdatedAt}, nil

	case adminCommandGroupGovernUpdate:
		settings, err := store.GroupBusiness.UpdateSettings(tenantID, topic, types.GroupSettingsPatch{
			JoinApproval:      payload.Governance.JoinApproval,
			MuteAll:           payload.Governance.MuteAll,
			MemberUploadFiles: payload.Governance.MemberUploadFiles,
			MemberPrivateChat: payload.Governance.MemberPrivateChat,
			MemberInvite:      payload.Governance.MemberInvite,
			Version:           payload.Version,
		})
		if err != nil {
			return nil, err
		}
		dispatchIMGroupEvents(tenantID)
		return map[string]any{"topic": topic, "version": settings.Version, "updated_at": settings.UpdatedAt}, nil

	case adminCommandGroupAdminSet:
		users := payload.UserIDs
		if len(users) == 0 && payload.UserID != "" {
			users = []string{payload.UserID}
		}
		if len(users) == 0 {
			return nil, types.ErrMalformed
		}
		succeeded := make([]map[string]any, 0, len(users))
		for _, raw := range users {
			target := parseKafkaCommandUser(raw)
			if target.IsZero() {
				return map[string]any{"succeeded": succeeded}, types.ErrMalformed
			}
			if err := store.GroupBusiness.UpdateRole(tenantID, topic, actor, target,
				types.GroupMemberRoleAdmin, types.ModeCAdmin|types.ModeCPublic, now); err != nil {
				return map[string]any{"succeeded": succeeded, "failed_user_id": raw}, err
			}
			succeeded = append(succeeded, map[string]any{"user_id": internalAdminUIDRaw(target), "role": types.GroupMemberRoleAdmin})
		}
		dispatchIMGroupEvents(tenantID)
		return map[string]any{"topic": topic, "succeeded": succeeded}, nil

	case adminCommandGroupAdminRemove:
		target := parseKafkaCommandUser(firstNonEmpty(payload.UserID, payload.TargetID))
		if target.IsZero() {
			return nil, types.ErrMalformed
		}
		if err := store.GroupBusiness.UpdateRole(tenantID, topic, actor, target,
			types.GroupMemberRoleMember, types.ModeCPublic, now); err != nil {
			return nil, err
		}
		dispatchIMGroupEvents(tenantID)
		return map[string]any{"topic": topic, "user_id": internalAdminUIDRaw(target), "role": types.GroupMemberRoleMember}, nil

	case adminCommandGroupMemberMute:
		target := parseKafkaCommandUser(firstNonEmpty(payload.UserID, payload.TargetID))
		if target.IsZero() {
			return nil, types.ErrMalformed
		}
		mute, err := store.GroupBusiness.UpsertMute(tenantID, &types.GroupMute{
			Topic:     topic,
			UserID:    target,
			MutedBy:   actor,
			MutedAt:   now,
			ExpiresAt: payload.ExpiresAt,
			Reason:    strings.TrimSpace(payload.Reason),
			Status:    types.GroupMuteStatusActive,
		})
		if err != nil {
			return nil, err
		}
		dispatchIMGroupEvents(tenantID)
		return map[string]any{"topic": topic, "user_id": internalAdminUIDRaw(target), "muted": true, "version": mute.Version}, nil

	case adminCommandGroupMemberUnmute:
		target := parseKafkaCommandUser(firstNonEmpty(payload.UserID, payload.TargetID))
		if target.IsZero() {
			return nil, types.ErrMalformed
		}
		if err := store.GroupBusiness.DeleteMute(tenantID, topic, target, actor, now); err != nil {
			return nil, err
		}
		dispatchIMGroupEvents(tenantID)
		return map[string]any{"topic": topic, "user_id": internalAdminUIDRaw(target), "muted": false}, nil

	case adminCommandGroupMemberRemove:
		target := parseKafkaCommandUser(firstNonEmpty(payload.UserID, payload.TargetID))
		if target.IsZero() {
			return nil, types.ErrMalformed
		}
		if err := store.GroupBusiness.Leave(tenantID, topic, target, now); err != nil {
			return nil, err
		}
		dispatchIMGroupEvents(tenantID)
		return map[string]any{"topic": topic, "user_id": internalAdminUIDRaw(target), "status": "left"}, nil

	case adminCommandGroupOwnerTransfer:
		target := parseKafkaCommandUser(firstNonEmpty(payload.TargetID, payload.UserID))
		if target.IsZero() {
			return nil, types.ErrMalformed
		}
		if err := store.GroupBusiness.TransferOwner(tenantID, topic, actor, target, payload.Version, now); err != nil {
			return nil, err
		}
		dispatchIMGroupEvents(tenantID)
		return map[string]any{"topic": topic, "owner": internalAdminUIDRaw(target), "previous_owner": internalAdminUIDRaw(actor)}, nil
	}
	return nil, errors.New("unsupported admin command: " + envelope.EventType)
}

func executeKafkaMessageAdminCommand(tenantID types.TenantID, envelope kafkaEnvelope, payload adminCommandPayload) (map[string]any, error) {
	conversationType := strings.TrimSpace(payload.ConversationType)
	topic := strings.TrimSpace(payload.Topic)
	if (conversationType != "private" && conversationType != "group") || topic == "" || payload.SeqID <= 0 {
		return nil, types.ErrMalformed
	}
	action := strings.TrimPrefix(strings.TrimSpace(envelope.EventType), "message.")
	actorAdminID := int64(0)
	if envelope.Actor != nil {
		actorAdminID, _ = strconv.ParseInt(strings.TrimSpace(envelope.Actor.ID), 10, 64)
	}
	now := time.Now().UTC().Round(time.Millisecond)
	commandID := firstNonEmpty(payload.CommandID, envelope.EventID)
	actionID := "action_" + envelope.EventID
	if len(actionID) > 96 {
		actionID = actionID[:96]
	}
	requestID := strings.TrimSpace(envelope.TraceID)
	if len(requestID) > 64 {
		requestID = requestID[:64]
	}
	if err := store.SecurityAudit.ApplyMessageAction(tenantID, &types.MessageAdminAction{
		ActionID: actionID, CommandID: commandID, ConversationType: conversationType,
		Topic: topic, SeqID: payload.SeqID, Action: action, Reason: strings.TrimSpace(payload.Reason),
		ActorAdminID: actorAdminID, RequestID: requestID, ProcessedAt: now,
	}); err != nil {
		return nil, err
	}
	if conversationType == "group" {
		if err := store.InvalidateHotMessages(tenantID, topic, []types.Range{{Low: payload.SeqID}}); err != nil {
			logs.Warn.Printf("message admin cache invalidation failed tenant:%d topic:%s seq:%d err:%v", tenantID, topic, payload.SeqID, err)
		}
	}
	return map[string]any{
		"conversation_type": conversationType, "topic": topic, "seq_id": payload.SeqID,
		"action": action, "status": "success", "processed_at": now,
	}, nil
}

func executeKafkaUserGovernanceCommand(tenantID types.TenantID, envelope kafkaEnvelope, payload adminCommandPayload, actor types.Uid) (map[string]any, error) {
	target := parseKafkaCommandUser(firstNonEmpty(payload.TargetID, payload.UserID))
	if target.IsZero() {
		return nil, types.ErrMalformed
	}
	eventType := strings.TrimSpace(envelope.EventType)
	now := time.Now().UTC().Round(time.Millisecond)
	result := map[string]any{
		"user_id": internalAdminUIDRaw(target),
		"action":  eventType,
	}

	switch eventType {
	case adminCommandUserBan, adminCommandUserFreeze:
		if err := store.Users.UpdateState(tenantID, target, types.StateSuspended); err != nil {
			return nil, err
		}
		evictUserSessions(target)
		notifyUserStateChanged(tenantID, target, types.StateSuspended)
		result["state"] = types.StateSuspended.String()
		result["evicted"] = true
		return result, nil
	case adminCommandUserUnban, adminCommandUserUnfreeze:
		if err := store.Users.UpdateState(tenantID, target, types.StateOK); err != nil {
			return nil, err
		}
		notifyUserStateChanged(tenantID, target, types.StateOK)
		result["state"] = types.StateOK.String()
		return result, nil
	case adminCommandUserDeviceKick:
		deviceID := strings.TrimSpace(payload.DeviceID)
		if deviceID == "" {
			return nil, types.ErrMalformed
		}
		evicted := 0
		if globals.sessionStore != nil {
			evicted = globals.sessionStore.EvictUserDevice(target, deviceID)
		}
		if store.Devices != nil {
			if err := store.Devices.Delete(tenantID, target, deviceID); err != nil && !errors.Is(err, types.ErrNotFound) {
				return nil, err
			}
		}
		result["device_id"] = deviceID
		result["evicted_sessions"] = evicted
		return result, nil
	case adminCommandUserLogoutAll:
		evictUserSessions(target)
		result["evicted"] = true
		return result, nil
	case adminCommandUserCancel, adminCommandUserDelete:
		hard := payload.Hard || eventType == adminCommandUserDelete
		if err := deleteUserByAdminCommand(tenantID, target, hard, now); err != nil {
			return nil, err
		}
		result["state"] = types.StateDeleted.String()
		result["hard"] = hard
		return result, nil
	}
	return nil, errors.New("unsupported user governance command: " + eventType)
}

func executeKafkaUserPasswordResetCommand(tenantID types.TenantID, payload adminCommandPayload) (map[string]any, error) {
	target := parseKafkaCommandUser(firstNonEmpty(payload.TargetID, payload.UserID))
	if target.IsZero() || payload.PasswordHash == "" {
		return nil, types.ErrMalformed
	}
	passwordHash, err := base64.StdEncoding.DecodeString(payload.PasswordHash)
	if err != nil || len(passwordHash) == 0 {
		return nil, types.ErrMalformed
	}
	_, authLevel, _, _, err := store.Users.GetAuthRecord(tenantID, target)
	if err != nil {
		return nil, err
	}
	if err = store.Users.UpdateAuthRecord(tenantID, target, authLevel, "", passwordHash, time.Time{}); err != nil {
		return nil, err
	}
	now := time.Now().UTC().Round(time.Millisecond)
	authVersion := 0
	if store.Security != nil {
		version, err := store.Security.BumpAuthVersion(tenantID, target, now)
		if err != nil {
			return nil, err
		}
		authVersion = version
	}
	evictUserSessions(target)
	return map[string]any{
		"user_id":      internalAdminUIDRaw(target),
		"action":       adminCommandUserPasswordReset,
		"auth_version": authVersion,
		"evicted":      true,
	}, nil
}

func evictUserSessions(uid types.Uid) {
	if globals.sessionStore != nil {
		globals.sessionStore.EvictUser(uid, "")
	}
}

func notifyUserStateChanged(tenantID types.TenantID, uid types.Uid, state types.ObjState) {
	usersRemoveUser(tenantID, uid)
	if globals.hub != nil && globals.hub.userStatus != nil {
		globals.hub.userStatus <- &userStatusReq{tenantID: tenantID, forUser: uid, state: state}
	}
}

func deleteUserByAdminCommand(tenantID types.TenantID, uid types.Uid, hard bool, _ time.Time) error {
	evictUserSessions(uid)
	usersRemoveUser(tenantID, uid)
	if globals.hub != nil && globals.hub.unreg != nil {
		done := make(chan bool)
		globals.hub.unreg <- &topicUnreg{tenantID: tenantID, forUser: uid, del: hard, done: done}
		<-done
	}
	return store.Users.Delete(tenantID, uid, hard)
}

func publishKafkaAdminCommandResult(ctx context.Context, command kafkaEnvelope, commandID, status, reason, message string, result map[string]any) error {
	if commandID == "" {
		commandID = command.EventID
	}
	payload := adminCommandResultPayload{
		CommandID:      commandID,
		CommandEventID: command.EventID,
		Status:         status,
		Reason:         reason,
		Message:        message,
		Result:         result,
	}
	encoded, err := json.Marshal(payload)
	if err != nil {
		return err
	}
	eventID := stableIMEventID("admin-command-result", command.EventID, status, reason)
	return publishKafkaEnvelope(ctx, "admin_command_result", commandID, kafkaEnvelope{
		EventID:        eventID,
		TraceID:        command.TraceID,
		TenantCode:     command.TenantCode,
		SourceService:  "im-server",
		TargetService:  command.SourceService,
		EventType:      "admin.command." + status,
		BizType:        command.BizType,
		BizID:          command.BizID,
		IdempotencyKey: command.IdempotencyKey,
		OccurredAt:     time.Now().UTC(),
		ParentEventID:  command.EventID,
		Payload:        encoded,
	})
}

func kafkaTenantID(tenantCode string) (types.TenantID, error) {
	tenant, err := store.Tenants.GetByCode(strings.TrimSpace(tenantCode))
	if err != nil {
		return types.ZeroTenantID, err
	}
	if tenant == nil || tenant.ID.IsZero() {
		return types.ZeroTenantID, types.ErrNotFound
	}
	return tenant.ID, nil
}

func kafkaTenantIDFromEnvelope(envelope kafkaEnvelope) (types.TenantID, error) {
	if !envelope.TenantID.IsZero() {
		return envelope.TenantID, nil
	}
	return kafkaTenantID(envelope.TenantCode)
}

func markKafkaDeadLetter(ctx context.Context, tenantID types.TenantID, envelope kafkaEnvelope, consume *types.KafkaConsumeRecord, reason, message string) error {
	encoded, err := json.Marshal(envelope)
	if err != nil {
		return err
	}
	retryCount := 0
	sourceTopic := kafkaTopic("admin_command")
	if consume != nil {
		retryCount = consume.RetryCount
		if strings.TrimSpace(consume.SourceTopic) != "" {
			sourceTopic = consume.SourceTopic
		}
	}
	deadID := stableIMEventID("dead", envelope.EventID, kafkaBus.config.ConsumerGroup, retryCount)
	dead := &types.KafkaDeadLetter{
		DeadID:        deadID,
		EventID:       envelope.EventID,
		ParentEventID: envelope.ParentEventID,
		SourceTopic:   sourceTopic,
		ConsumerGroup: kafkaBus.config.ConsumerGroup,
		EventType:     envelope.EventType,
		BizType:       envelope.BizType,
		BizID:         envelope.BizID,
		Envelope:      encoded,
		RetryCount:    retryCount,
		LastReason:    reason,
		LastError:     message,
		Status:        types.KafkaDeadStatusDead,
	}
	now := time.Now().UTC().Round(time.Millisecond)
	deadPayload, err := json.Marshal(dead)
	if err != nil {
		return err
	}
	return publishKafkaDeadLetter(ctx, "dead_letter", envelope.EventID, kafkaEnvelope{
		EventID:       deadID,
		TraceID:       envelope.TraceID,
		TenantCode:    envelope.TenantCode,
		TenantID:      tenantID,
		SourceService: "im-server",
		TargetService: "hotgo-admin",
		EventType:     "kafka.dead_letter.created",
		BizType:       envelope.BizType,
		BizID:         envelope.BizID,
		OccurredAt:    now,
		ParentEventID: envelope.EventID,
		Payload:       deadPayload,
	})
}

func shouldDeadLetterKafkaCommand(reason string, retryCount int) bool {
	if !kafkaCommandReasonRetryable(reason) {
		return true
	}
	if kafkaBus == nil {
		return true
	}
	return retryCount >= kafkaBus.config.MaxRetryCount
}

func kafkaCommandReasonRetryable(reason string) bool {
	switch reason {
	case "COMMAND_FAILED":
		return true
	default:
		return false
	}
}

func sleepKafkaRetryBackoff(retryCount int) {
	if kafkaBus == nil || retryCount <= 0 {
		return
	}
	backoffs := kafkaBus.config.RetryBackoffMS
	index := retryCount - 1
	if index >= len(backoffs) {
		index = len(backoffs) - 1
	}
	delay := time.Duration(backoffs[index]) * time.Millisecond
	if delay > 0 {
		time.Sleep(delay)
	}
}

func parseKafkaCommandUser(raw string) types.Uid {
	raw = strings.TrimSpace(raw)
	if raw == "" {
		return types.ZeroUid
	}
	if id, err := strconv.ParseInt(raw, 10, 64); err == nil && id > 0 {
		return store.EncodeUid(id)
	}
	uid := types.ParseUserId(raw)
	if !uid.IsZero() {
		return uid
	}
	return types.ParseUid(raw)
}

func kafkaDeadLetterResult(dead *types.KafkaDeadLetter) map[string]any {
	if dead == nil {
		return nil
	}
	return map[string]any{
		"dead_id":      dead.DeadID,
		"event_id":     dead.EventID,
		"source_topic": dead.SourceTopic,
		"event_type":   dead.EventType,
		"biz_type":     dead.BizType,
		"biz_id":       dead.BizID,
		"status":       dead.Status,
		"replay_count": dead.ReplayCount,
		"updated_at":   dead.UpdatedAt,
	}
}

func kafkaCommandErrorReason(err error) string {
	if err == nil {
		return ""
	}
	switch {
	case errors.Is(err, types.ErrMalformed):
		return "MALFORMED_COMMAND"
	case errors.Is(err, types.ErrPolicy):
		return "POLICY_VIOLATION"
	case errors.Is(err, types.ErrNotFound):
		return "RESOURCE_NOT_FOUND"
	case errors.Is(err, types.ErrPermissionDenied):
		return "FORBIDDEN"
	case errors.Is(err, types.ErrDuplicate):
		return "COMMAND_DUPLICATED"
	case errors.Is(err, types.ErrInvalidResponse):
		return "GROUP_VERSION_CONFLICT"
	default:
		if strings.Contains(err.Error(), "unsupported admin command") {
			return "UNSUPPORTED_COMMAND"
		}
		return "COMMAND_FAILED"
	}
}

func firstNonEmpty(values ...string) string {
	for _, value := range values {
		if strings.TrimSpace(value) != "" {
			return strings.TrimSpace(value)
		}
	}
	return ""
}