yumi-golang/internal/service/taskcenter/lifecycle.go

package taskcenter

import (
	"context"
	"encoding/json"
	"errors"
	"log"
	"net/http"
	"strings"
	"time"

	"chatapp3-golang/internal/common"

	rmq "github.com/apache/rocketmq-clients/golang/v5"
	"github.com/apache/rocketmq-clients/golang/v5/credentials"
)

// StartMessageConsumer 启动任务中心事件 MQ 消费链路。
func (s *Service) StartMessageConsumer(ctx context.Context) error {
	if !s.cfg.TaskCenter.MQ.Enabled {
		log.Printf("task center rocketmq consumer disabled")
		return nil
	}
	if strings.TrimSpace(s.cfg.TaskCenter.MQ.Endpoint) == "" ||
		strings.TrimSpace(s.cfg.TaskCenter.MQ.AccessKey) == "" ||
		strings.TrimSpace(s.cfg.TaskCenter.MQ.AccessSecret) == "" ||
		strings.TrimSpace(s.cfg.TaskCenter.MQ.Topic) == "" ||
		strings.TrimSpace(s.cfg.TaskCenter.MQ.ConsumerGroup) == "" {
		log.Printf("task center rocketmq consumer skipped: endpoint, credentials, topic or consumer group missing")
		return nil
	}

	go s.startTaskCenterRocketMQConsumerWithRetry(ctx)
	return nil
}

func (s *Service) startTaskCenterRocketMQConsumerWithRetry(ctx context.Context) {
	retryDelay := 5 * time.Second
	for {
		if ctx.Err() != nil {
			return
		}
		if err := s.startTaskCenterRocketMQConsumer(ctx); err != nil {
			log.Printf("start task center rocketmq consumer failed: %v; retry in %s", err, retryDelay)
			timer := time.NewTimer(retryDelay)
			select {
			case <-ctx.Done():
				timer.Stop()
				return
			case <-timer.C:
			}
			if retryDelay < time.Minute {
				retryDelay *= 2
				if retryDelay > time.Minute {
					retryDelay = time.Minute
				}
			}
			continue
		}
		return
	}
}

func (s *Service) startTaskCenterRocketMQConsumer(ctx context.Context) error {
	filter := rmq.SUB_ALL
	if tag := strings.TrimSpace(s.cfg.TaskCenter.MQ.Tag); tag != "" && tag != "*" {
		filter = rmq.NewFilterExpression(tag)
	}

	consumer, err := rmq.NewPushConsumer(&rmq.Config{
		Endpoint:      s.cfg.TaskCenter.MQ.Endpoint,
		NameSpace:     s.cfg.TaskCenter.MQ.Namespace,
		ConsumerGroup: s.cfg.TaskCenter.MQ.ConsumerGroup,
		Credentials: &credentials.SessionCredentials{
			AccessKey:     s.cfg.TaskCenter.MQ.AccessKey,
			AccessSecret:  s.cfg.TaskCenter.MQ.AccessSecret,
			SecurityToken: s.cfg.TaskCenter.MQ.SecurityToken,
		},
	},
		rmq.WithPushSubscriptionExpressions(map[string]*rmq.FilterExpression{
			s.cfg.TaskCenter.MQ.Topic: filter,
		}),
		rmq.WithPushConsumptionThreadCount(8),
		rmq.WithPushMaxCacheMessageCount(512),
		rmq.WithPushMessageListener(&rmq.FuncMessageListener{
			Consume: func(messageView *rmq.MessageView) rmq.ConsumerResult {
				if messageView == nil {
					return rmq.SUCCESS
				}
				if err := s.processTaskCenterMQMessage(context.Background(), messageView); err != nil {
					log.Printf("task center mq process failed. messageId=%s err=%v", messageView.GetMessageId(), err)
					return rmq.FAILURE
				}
				return rmq.SUCCESS
			},
		}),
	)
	if err != nil {
		return err
	}
	if err := consumer.Start(); err != nil {
		return err
	}
	s.rocketConsumer = consumer
	log.Printf("task center rocketmq consumer started. topic=%s group=%s tag=%s",
		s.cfg.TaskCenter.MQ.Topic,
		s.cfg.TaskCenter.MQ.ConsumerGroup,
		s.cfg.TaskCenter.MQ.Tag,
	)
	go func() {
		<-ctx.Done()
		if s.rocketConsumer != nil {
			_ = s.rocketConsumer.GracefulStop()
		}
	}()
	return nil
}

func (s *Service) processTaskCenterMQMessage(ctx context.Context, messageView *rmq.MessageView) error {
	payload := strings.TrimSpace(string(messageView.GetBody()))
	if payload == "" {
		return nil
	}
	req, err := decodeTaskCenterEventPayload(payload, s.cfg.TaskCenter.MQ.Tag)
	if err != nil {
		var syntaxErr *json.SyntaxError
		var typeErr *json.UnmarshalTypeError
		if errors.As(err, &syntaxErr) || errors.As(err, &typeErr) {
			log.Printf("drop malformed task center message. messageId=%s err=%v", messageView.GetMessageId(), err)
			return nil
		}
		return err
	}
	if strings.TrimSpace(req.EventID) == "" {
		return nil
	}
	if _, err := s.ProcessEvent(ctx, req); err != nil {
		var appErr *common.AppError
		if errors.As(err, &appErr) && appErr.Status >= http.StatusBadRequest && appErr.Status < http.StatusInternalServerError {
			log.Printf("drop invalid task center message. messageId=%s code=%s err=%v", messageView.GetMessageId(), appErr.Code, err)
			return nil
		}
		return err
	}
	return nil
}

func decodeTaskCenterEventPayload(payload string, expectedTag string) (EventRequest, error) {
	var req EventRequest
	if err := json.Unmarshal([]byte(payload), &req); err != nil {
		return req, err
	}
	if strings.TrimSpace(req.EventID) != "" || strings.TrimSpace(req.EventType) != "" || req.UserID.Int64() > 0 {
		return req, nil
	}

	var envelope taskCenterMessageEventEnvelope
	if err := json.Unmarshal([]byte(payload), &envelope); err != nil {
		return req, err
	}
	if shouldSkipTaskCenterEnvelope(envelope.Tag, expectedTag) {
		return EventRequest{}, nil
	}
	bodyPayload, ok, err := envelope.bodyPayload()
	if err != nil || !ok {
		return EventRequest{}, err
	}
	if err := json.Unmarshal([]byte(bodyPayload), &req); err != nil {
		return EventRequest{}, err
	}
	return req, nil
}

type taskCenterMessageEventEnvelope struct {
	Tag  string          `json:"tag"`
	Body json.RawMessage `json:"body"`
}

func (e taskCenterMessageEventEnvelope) bodyPayload() (string, bool, error) {
	raw := strings.TrimSpace(string(e.Body))
	if raw == "" || raw == "null" {
		return "", false, nil
	}
	if strings.HasPrefix(raw, "\"") {
		var body string
		if err := json.Unmarshal(e.Body, &body); err != nil {
			return "", false, err
		}
		body = strings.TrimSpace(body)
		return body, body != "", nil
	}
	return raw, true, nil
}

func shouldSkipTaskCenterEnvelope(actualTag, expectedTag string) bool {
	actualTag = strings.TrimSpace(actualTag)
	expectedTag = strings.TrimSpace(expectedTag)
	if actualTag == "" || expectedTag == "" || expectedTag == "*" {
		return false
	}
	return actualTag != expectedTag
}