junhong_cmp_fiber/internal/infrastructure/wecom/callback_service.go

package wecom

import (
	"context"
	"encoding/xml"
	"strconv"
	"strings"

	"github.com/break/junhong_cmp_fiber/internal/infrastructure/integrationlog"
	"github.com/break/junhong_cmp_fiber/internal/model"
	"github.com/break/junhong_cmp_fiber/pkg/constants"
	"github.com/break/junhong_cmp_fiber/pkg/errors"
	"github.com/break/junhong_cmp_fiber/pkg/middleware"
	"github.com/break/junhong_cmp_fiber/pkg/queue"
	"go.uber.org/zap"
)

type callbackIntegrationLog interface {
	RecordInbound(ctx context.Context, input integrationlog.InboundAttempt) (*model.IntegrationLog, bool, error)
}

type encryptedCallbackEnvelope struct {
	Encrypt string `xml:"Encrypt"`
}

type approvalChangeEvent struct {
	Event            string `xml:"Event"`
	SPNo             string `xml:"ApprovalInfo>SpNoStr"`
	SPStatus         int    `xml:"ApprovalInfo>SpStatus"`
	TemplateID       string `xml:"ApprovalInfo>TemplateId"`
	StatusChangeType int    `xml:"ApprovalInfo>StatuChangeEvent"`
}

// ApprovalDetailSyncTask 是回调快速响应后提交给 Worker 的结构化任务载荷。
type ApprovalDetailSyncTask struct {
	ApplicationID uint   `json:"application_id"`
	SPNo          string `json:"sp_no"`
	Source        string `json:"source"`
	IntegrationID string `json:"integration_id"`
}

// CallbackService 校验解密企微回调、记录入站幂等事实并异步拉取详情。
type CallbackService struct {
	applications *ApplicationRepository
	integration  callbackIntegrationLog
	queue        *queue.Client
	logger       *zap.Logger
}

// NewCallbackService 创建企业微信审批回调服务。
func NewCallbackService(applications *ApplicationRepository, integration callbackIntegrationLog, queueClient *queue.Client, logger *zap.Logger) *CallbackService {
	return &CallbackService{applications: applications, integration: integration, queue: queueClient, logger: logger}
}

// VerifyURL 校验企微回调 URL 并返回 echostr 明文。
func (s *CallbackService) VerifyURL(ctx context.Context, applicationID uint, signature, timestamp, nonce, echo string) ([]byte, error) {
	crypto, err := s.callbackCrypto(ctx, applicationID)
	if err != nil {
		return nil, err
	}
	return crypto.VerifyAndDecrypt(signature, timestamp, nonce, echo)
}

// Receive 校验并解密审批事件，持久化入站事实后用 struct 载荷提交详情同步任务。
func (s *CallbackService) Receive(ctx context.Context, applicationID uint, signature, timestamp, nonce string, body []byte) error {
	if s == nil || s.integration == nil || s.queue == nil {
		return errors.New(errors.CodeServiceUnavailable, "企业微信审批回调服务未配置")
	}
	var envelope encryptedCallbackEnvelope
	if err := xml.Unmarshal(body, &envelope); err != nil || strings.TrimSpace(envelope.Encrypt) == "" {
		return errors.New(errors.CodeInvalidParam, "企业微信回调 XML 无效")
	}
	crypto, err := s.callbackCrypto(ctx, applicationID)
	if err != nil {
		return err
	}
	plaintext, err := crypto.VerifyAndDecrypt(signature, timestamp, nonce, envelope.Encrypt)
	if err != nil {
		return err
	}
	if s.logger != nil {
		s.logger.Info("企业微信审批回调解密成功", zap.Uint("application_id", applicationID), zap.ByteString("content", plaintext))
	}
	var event approvalChangeEvent
	if err := xml.Unmarshal(plaintext, &event); err != nil || event.Event != "sys_approval_change" || strings.TrimSpace(event.SPNo) == "" {
		return errors.New(errors.CodeInvalidParam, "企业微信审批回调事件无效")
	}
	resourceKey := strings.TrimSpace(event.SPNo)
	requestID := middleware.GetRequestIDFromContext(ctx)
	log, created, err := s.integration.RecordInbound(ctx, integrationlog.InboundAttempt{
		IdempotencyKey: applicationCallbackIdempotencyKey(applicationID, signature),
		Provider:       constants.IntegrationProviderWeCom, Operation: constants.IntegrationOperationWeComApprovalCallback,
		ExternalID: event.SPNo, ResourceType: constants.WeComApprovalInstanceResourceType, ResourceKey: &resourceKey,
		RawPayload: body, ContentType: "application/xml", RequestID: requestID, CorrelationID: &resourceKey,
	})
	if err != nil {
		return err
	}
	if !created && log.Result != constants.IntegrationResultPending {
		return nil
	}
	return s.queue.EnqueueTask(ctx, constants.TaskTypeWeComApprovalSync, ApprovalDetailSyncTask{
		ApplicationID: applicationID, SPNo: event.SPNo, Source: constants.ApprovalSyncSourceCallback,
		IntegrationID: log.IntegrationID,
	})
}

// callbackCrypto 每次读取数据库当前回调凭据，使凭据更新无需重启进程。
func (s *CallbackService) callbackCrypto(ctx context.Context, applicationID uint) (*CallbackCrypto, error) {
	if s == nil || s.applications == nil || applicationID == 0 {
		return nil, errors.New(errors.CodeServiceUnavailable, "企业微信审批回调服务未配置")
	}
	application, err := s.applications.GetEnabled(ctx, applicationID)
	if err != nil {
		return nil, err
	}
	return NewCallbackCrypto(application.CallbackToken, application.EncodingAESKey, application.CorpID)
}

func applicationCallbackIdempotencyKey(applicationID uint, signature string) string {
	return "wecom-approval:" + strconv.FormatUint(uint64(applicationID), 10) + ":" + strings.ToLower(strings.TrimSpace(signature))
}