junhong_cmp_fiber/internal/infrastructure/wecom/approval_recovery_task.go

package wecom

import (
	"context"
	"strings"
	"time"

	"github.com/bytedance/sonic"
	"github.com/hibiken/asynq"

	"github.com/break/junhong_cmp_fiber/pkg/constants"
	"github.com/break/junhong_cmp_fiber/pkg/errors"
	"github.com/break/junhong_cmp_fiber/pkg/queue"
)

// ApprovalRecoveryTaskHandler 周期恢复结果未知审批并轮询未终态详情。
type ApprovalRecoveryTaskHandler struct {
	contexts *ApprovalContextRepository
	infos    *ApprovalInfoClient
	queue    *queue.Client
	now      func() time.Time
}

// NewApprovalRecoveryTaskHandler 创建企业微信审批主动恢复任务 Handler。
func NewApprovalRecoveryTaskHandler(contexts *ApprovalContextRepository, infos *ApprovalInfoClient, queueClient *queue.Client) *ApprovalRecoveryTaskHandler {
	return &ApprovalRecoveryTaskHandler{contexts: contexts, infos: infos, queue: queueClient, now: time.Now}
}

// Handle 支持全局扫描任务与按实例人工任务，二者共享同一 unknown 确认算法。
func (h *ApprovalRecoveryTaskHandler) Handle(ctx context.Context, task *asynq.Task) error {
	if h == nil || h.contexts == nil || h.infos == nil || h.queue == nil {
		return errors.New(errors.CodeServiceUnavailable, "企业微信审批主动恢复任务未配置")
	}
	var payload struct {
		InstanceID uint `json:"instance_id"`
	}
	if task != nil && len(task.Payload()) > 0 {
		_ = sonic.Unmarshal(task.Payload(), &payload)
	}
	now := h.now().UTC()
	if payload.InstanceID > 0 {
		return h.recoverUnknownInstance(ctx, payload.InstanceID, now)
	}
	if err := h.contexts.PromoteStaleSendingToUnknown(ctx, now.Add(-constants.WeComApprovalSendingLease)); err != nil {
		return err
	}
	if err := h.enqueuePendingSync(ctx, now); err != nil {
		return err
	}
	return h.recoverUnknown(ctx, now)
}

func (h *ApprovalRecoveryTaskHandler) enqueuePendingSync(ctx context.Context, now time.Time) error {
	records, err := h.contexts.ListPendingSync(ctx, now.Add(-constants.WeComApprovalPollingInterval), constants.WeComApprovalRecoveryBatchSize)
	if err != nil {
		return err
	}
	for _, record := range records {
		if err := h.enqueueDetailSync(ctx, record.ApplicationID, record.SPNo); err != nil {
			return err
		}
	}
	return nil
}

func (h *ApprovalRecoveryTaskHandler) recoverUnknownInstance(ctx context.Context, instanceID uint, now time.Time) error {
	record, err := h.contexts.GetRecoveryRecord(ctx, instanceID)
	if err != nil || record == nil {
		return err
	}
	claimed, err := h.contexts.ClaimUnknownRecovery(ctx, instanceID, now.Add(-constants.WeComApprovalPollingInterval))
	if err != nil || !claimed {
		return err
	}
	return h.recoverUnknownRecord(ctx, *record, now)
}

func (h *ApprovalRecoveryTaskHandler) recoverUnknownRecord(ctx context.Context, record ApprovalRecoveryRecord, now time.Time) error {
	spNo, submissionIntegrationID, err := h.contexts.FindSuccessfulSubmissionSPNo(ctx, record.InstanceID)
	if err != nil {
		return err
	}
	integrationIDs := []string{submissionIntegrationID}
	if spNo == "" {
		var ids []string
		spNo, ids, err = h.findUniqueSPNo(ctx, record, now)
		if err != nil {
			return err
		}
		integrationIDs = append(integrationIDs, ids...)
	}
	if spNo == "" {
		return nil
	}
	recovered, err := h.contexts.RecoverSubmitted(ctx, record.InstanceID, spNo, integrationIDs, constants.AuditActorScheduledJob, constants.ApprovalAuditActorRecoveryJob, constants.AuditSourceScheduler)
	if err != nil || !recovered {
		return err
	}
	return h.enqueueDetailSync(ctx, record.ApplicationID, spNo)
}

func (h *ApprovalRecoveryTaskHandler) recoverUnknown(ctx context.Context, now time.Time) error {
	records, err := h.contexts.ListUnknownRecovery(ctx, now.Add(-constants.WeComApprovalPollingInterval), constants.WeComApprovalUnknownRecoveryBatchSize)
	if err != nil {
		return err
	}
	for _, record := range records {
		claimed, err := h.contexts.ClaimUnknownRecovery(ctx, record.InstanceID, now.Add(-constants.WeComApprovalPollingInterval))
		if err != nil {
			return err
		}
		if !claimed {
			continue
		}
		if err := h.recoverUnknownRecord(ctx, record, now); err != nil {
			return err
		}
	}
	return nil
}

// findUniqueSPNo 使用提交时间附近的固定窄窗口分页查询，并只接受唯一未关联候选。
func (h *ApprovalRecoveryTaskHandler) findUniqueSPNo(ctx context.Context, record ApprovalRecoveryRecord, now time.Time) (string, []string, error) {
	startTime := record.SubmissionAttemptedAt.Add(-constants.WeComApprovalRecoveryWindow)
	endTime := record.SubmissionAttemptedAt.Add(constants.WeComApprovalRecoveryWindow)
	if endTime.After(now) {
		endTime = now
	}
	if !startTime.Before(endTime) {
		return "", nil, nil
	}
	cursor := ""
	seenCursors := make(map[string]struct{})
	seenCandidates := make(map[string]struct{})
	unbound := make([]string, 0, 2)
	integrationIDs := make([]string, 0, 2)
	for {
		page, err := h.infos.List(ctx, ApprovalInfoQuery{
			ApplicationID: record.ApplicationID, StartTime: startTime, EndTime: endTime,
			TemplateID: record.TemplateID, CreatorUserID: record.CreatorUserID,
			Cursor: cursor, Size: constants.WeComApprovalInfoMaxPageSize,
		})
		if err != nil {
			return "", nil, err
		}
		integrationIDs = append(integrationIDs, page.IntegrationID)
		existing, err := h.contexts.ExistingSPNos(ctx, record.ApplicationID, page.SPNos)
		if err != nil {
			return "", nil, err
		}
		for _, candidate := range page.SPNos {
			if _, seen := seenCandidates[candidate]; seen {
				continue
			}
			seenCandidates[candidate] = struct{}{}
			if _, exists := existing[candidate]; !exists {
				unbound = append(unbound, candidate)
				if len(unbound) > 1 {
					return "", integrationIDs, nil
				}
			}
		}
		next := strings.TrimSpace(page.NextCursor)
		if next == "" {
			break
		}
		if _, exists := seenCursors[next]; exists {
			return "", nil, errors.New(errors.CodeServiceUnavailable, "企业微信批量审批单号分页游标重复")
		}
		seenCursors[next] = struct{}{}
		cursor = next
	}
	if len(unbound) != 1 {
		return "", integrationIDs, nil
	}
	return unbound[0], integrationIDs, nil
}

func (h *ApprovalRecoveryTaskHandler) enqueueDetailSync(ctx context.Context, applicationID uint, spNo string) error {
	return h.queue.EnqueueTask(ctx, constants.TaskTypeWeComApprovalSync, ApprovalDetailSyncTask{
		ApplicationID: applicationID, SPNo: spNo, Source: constants.ApprovalSyncSourcePolling,
	})
}