feat: complete cmpp gateway delivery recovery workflows

2026-07-08 16:30:06 +08:00
parent cc628d0214
commit 8144f08652
60 changed files with 8901 additions and 94 deletions
@@ -0,0 +1,431 @@
+package submitworker
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"log"
+	"net/http"
+	"strings"
+	"time"
+
+	"cmpp-platform/gateway/internal/queue"
+	"cmpp-platform/gateway/internal/upstream"
+
+	"github.com/redis/go-redis/v9"
+)
+
+const (
+	defaultStream   = "gateway.submit.commands"
+	defaultGroup    = "cmpp-gateway"
+	defaultConsumer = "gateway-1"
+	defaultMinIdle  = 30 * time.Second
+	defaultMaxFails = 3
+)
+
+type Worker struct {
+	Redis            *redis.Client
+	Upstream         *upstream.Manager
+	Submit           func(context.Context, queue.SubmitCommand) (queue.SubmitResult, error)
+	ReportDeadLetter func(context.Context, DeadLetterEvent) error
+	Stream           string
+	Group            string
+	Consumer         string
+	Block            time.Duration
+	Count            int64
+	MinIdle          time.Duration
+	MaxFailures      int
+	APIBaseURL       string
+	HTTPClient       *http.Client
+	Logger           *log.Logger
+}
+
+type DeadLetterEvent struct {
+	StreamMessageID string                 `json:"streamMessageId"`
+	TraceID         string                 `json:"traceId,omitempty"`
+	MessageID       string                 `json:"messageId,omitempty"`
+	ChannelID       string                 `json:"channelId,omitempty"`
+	TenantID        string                 `json:"tenantId,omitempty"`
+	ApplicationID   string                 `json:"applicationId,omitempty"`
+	SubmitID        string                 `json:"submitId,omitempty"`
+	FailureCode     string                 `json:"failureCode"`
+	FailureMessage  string                 `json:"failureMessage"`
+	Attempts        int                    `json:"attempts"`
+	MaxAttempts     int                    `json:"maxAttempts"`
+	CommandPayload  map[string]interface{} `json:"commandPayload,omitempty"`
+	RawPayload      string                 `json:"rawPayload,omitempty"`
+	DeadLetteredAt  time.Time              `json:"deadLetteredAt"`
+}
+
+func New(redisURL string, manager *upstream.Manager) (*Worker, error) {
+	client, err := redisClient(redisURL)
+	if err != nil {
+		return nil, err
+	}
+	return &Worker{Redis: client, Upstream: manager}, nil
+}
+
+func (w *Worker) Run(ctx context.Context) error {
+	if w.Redis == nil {
+		return fmt.Errorf("redis client is required")
+	}
+	if w.Upstream == nil {
+		return fmt.Errorf("upstream manager is required")
+	}
+	for {
+		if err := w.ensureGroup(ctx); err != nil {
+			if ctx.Err() != nil {
+				return ctx.Err()
+			}
+			w.logf("gateway submit worker ensure group failed: %v", err)
+			sleep(ctx, 3*time.Second)
+			continue
+		}
+		if err := w.recoverPending(ctx); err != nil {
+			if ctx.Err() != nil {
+				return ctx.Err()
+			}
+			w.logf("gateway submit worker pending recovery failed: %v", err)
+			sleep(ctx, time.Second)
+			continue
+		}
+		if err := w.consumeOnce(ctx); err != nil {
+			if ctx.Err() != nil {
+				return ctx.Err()
+			}
+			w.logf("gateway submit worker consume failed: %v", err)
+			sleep(ctx, time.Second)
+		}
+	}
+}
+
+func (w *Worker) ensureGroup(ctx context.Context) error {
+	err := w.Redis.XGroupCreateMkStream(ctx, w.stream(), w.group(), "0").Err()
+	if err == nil || strings.Contains(err.Error(), "BUSYGROUP") {
+		return nil
+	}
+	return err
+}
+
+func (w *Worker) consumeOnce(ctx context.Context) error {
+	streams, err := w.Redis.XReadGroup(ctx, &redis.XReadGroupArgs{
+		Group:    w.group(),
+		Consumer: w.consumer(),
+		Streams:  []string{w.stream(), ">"},
+		Count:    w.count(),
+		Block:    w.block(),
+	}).Result()
+	if errors.Is(err, redis.Nil) {
+		return nil
+	}
+	if err != nil {
+		return err
+	}
+	for _, stream := range streams {
+		if err := w.processMessages(ctx, stream.Messages); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (w *Worker) recoverPending(ctx context.Context) error {
+	start := "0-0"
+	for {
+		messages, next, err := w.Redis.XAutoClaim(ctx, &redis.XAutoClaimArgs{
+			Stream:   w.stream(),
+			Group:    w.group(),
+			Consumer: w.consumer(),
+			MinIdle:  w.minIdle(),
+			Start:    start,
+			Count:    w.count(),
+		}).Result()
+		if errors.Is(err, redis.Nil) {
+			return nil
+		}
+		if err != nil {
+			return err
+		}
+		if len(messages) == 0 {
+			return nil
+		}
+		w.logf("gateway submit worker reclaimed %d pending message(s)", len(messages))
+		if err := w.processMessages(ctx, messages); err != nil {
+			return err
+		}
+		start = next
+		if next == "0-0" {
+			return nil
+		}
+	}
+}
+
+func (w *Worker) processMessages(ctx context.Context, messages []redis.XMessage) error {
+	for _, message := range messages {
+		if err := w.processMessage(ctx, message); err != nil {
+			w.logf("gateway submit worker message %s failed: %v", message.ID, err)
+			continue
+		}
+	}
+	return nil
+}
+
+func (w *Worker) processMessage(ctx context.Context, message redis.XMessage) error {
+	command, err := CommandFromStreamValues(message.Values)
+	if err != nil {
+		return w.deadLetterMalformedMessage(ctx, message, err)
+	}
+	if err := w.handleCommand(ctx, command); err != nil {
+		attempts, attemptsErr := w.incrementFailureAttempt(ctx, message.ID)
+		if attemptsErr != nil {
+			w.logf("gateway submit worker increment failure %s failed: %v", message.ID, attemptsErr)
+		}
+		if attempts >= w.maxFailures() {
+			if reportErr := w.deadLetterCommand(ctx, message, command, attempts, err); reportErr != nil {
+				return reportErr
+			}
+			return w.ackAndClearFailure(ctx, message.ID)
+		}
+		return err
+	}
+	return w.ackAndClearFailure(ctx, message.ID)
+}
+
+func (w *Worker) handleCommand(ctx context.Context, command queue.SubmitCommand) error {
+	submit := w.Submit
+	if submit == nil {
+		if w.Upstream == nil {
+			return fmt.Errorf("upstream manager is required")
+		}
+		submit = w.Upstream.Submit
+	}
+	result, err := submit(ctx, command)
+	if err != nil && (result.SubmitStatus == "" || result.SubmitStatus == "accepted") {
+		return err
+	}
+	return nil
+}
+
+func CommandFromStreamValues(values map[string]interface{}) (queue.SubmitCommand, error) {
+	raw, ok := values["data"]
+	if !ok {
+		return queue.SubmitCommand{}, fmt.Errorf("stream data field is required")
+	}
+	var data string
+	switch value := raw.(type) {
+	case string:
+		data = value
+	case []byte:
+		data = string(value)
+	default:
+		data = fmt.Sprint(value)
+	}
+	var command queue.SubmitCommand
+	if err := json.Unmarshal([]byte(data), &command); err != nil {
+		return queue.SubmitCommand{}, err
+	}
+	if command.MessageType != queue.MessageTypeSubmitCommand {
+		return queue.SubmitCommand{}, fmt.Errorf("unsupported messageType %q", command.MessageType)
+	}
+	return command, nil
+}
+
+func (w *Worker) deadLetterMalformedMessage(ctx context.Context, message redis.XMessage, cause error) error {
+	event := DeadLetterEvent{
+		StreamMessageID: message.ID,
+		FailureCode:     "INVALID_COMMAND_PAYLOAD",
+		FailureMessage:  cause.Error(),
+		Attempts:        1,
+		MaxAttempts:     1,
+		RawPayload:      extractRawPayload(message.Values),
+		DeadLetteredAt:  time.Now().UTC(),
+	}
+	if err := w.reportDeadLetter(ctx, event); err != nil {
+		return err
+	}
+	return w.ackAndClearFailure(ctx, message.ID)
+}
+
+func (w *Worker) deadLetterCommand(ctx context.Context, message redis.XMessage, command queue.SubmitCommand, attempts int, cause error) error {
+	payload, payloadErr := commandPayload(command)
+	if payloadErr != nil {
+		w.logf("gateway submit worker marshal dead-letter command %s failed: %v", message.ID, payloadErr)
+	}
+	return w.reportDeadLetter(ctx, DeadLetterEvent{
+		StreamMessageID: message.ID,
+		TraceID:         command.TraceID,
+		MessageID:       command.MessageID,
+		ChannelID:       command.ChannelID,
+		TenantID:        command.TenantID,
+		ApplicationID:   command.ApplicationID,
+		SubmitID:        command.SubmitID,
+		FailureCode:     "SUBMIT_PROCESSING_FAILED",
+		FailureMessage:  cause.Error(),
+		Attempts:        attempts,
+		MaxAttempts:     w.maxFailures(),
+		CommandPayload:  payload,
+		RawPayload:      extractRawPayload(message.Values),
+		DeadLetteredAt:  time.Now().UTC(),
+	})
+}
+
+func (w *Worker) reportDeadLetter(ctx context.Context, event DeadLetterEvent) error {
+	if w.ReportDeadLetter != nil {
+		return w.ReportDeadLetter(ctx, event)
+	}
+	if w.APIBaseURL == "" {
+		return fmt.Errorf("gateway submit dead-letter reporter is not configured")
+	}
+	client := w.HTTPClient
+	if client == nil {
+		client = &http.Client{Timeout: 10 * time.Second}
+	}
+	body, err := json.Marshal(event)
+	if err != nil {
+		return err
+	}
+	req, err := http.NewRequestWithContext(
+		ctx,
+		http.MethodPost,
+		strings.TrimRight(w.APIBaseURL, "/")+"/gateway/events/dead-letter",
+		bytes.NewReader(body),
+	)
+	if err != nil {
+		return err
+	}
+	req.Header.Set("Content-Type", "application/json")
+	resp, err := client.Do(req)
+	if err != nil {
+		return err
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
+		return fmt.Errorf("dead-letter endpoint returned %d", resp.StatusCode)
+	}
+	return nil
+}
+
+func redisClient(redisURL string) (*redis.Client, error) {
+	if redisURL == "" {
+		redisURL = "redis://127.0.0.1:6379"
+	}
+	options, err := redis.ParseURL(redisURL)
+	if err != nil {
+		return nil, err
+	}
+	return redis.NewClient(options), nil
+}
+
+func (w *Worker) incrementFailureAttempt(ctx context.Context, messageID string) (int, error) {
+	value, err := w.Redis.HIncrBy(ctx, w.failureAttemptsKey(), messageID, 1).Result()
+	return int(value), err
+}
+
+func (w *Worker) ackAndClearFailure(ctx context.Context, messageID string) error {
+	if err := w.Redis.XAck(ctx, w.stream(), w.group(), messageID).Err(); err != nil {
+		return err
+	}
+	if err := w.Redis.HDel(ctx, w.failureAttemptsKey(), messageID).Err(); err != nil {
+		w.logf("gateway submit worker clear failure %s failed: %v", messageID, err)
+	}
+	return nil
+}
+
+func (w *Worker) stream() string {
+	if w.Stream != "" {
+		return w.Stream
+	}
+	return defaultStream
+}
+
+func (w *Worker) group() string {
+	if w.Group != "" {
+		return w.Group
+	}
+	return defaultGroup
+}
+
+func (w *Worker) consumer() string {
+	if w.Consumer != "" {
+		return w.Consumer
+	}
+	return defaultConsumer
+}
+
+func (w *Worker) block() time.Duration {
+	if w.Block > 0 {
+		return w.Block
+	}
+	return 5 * time.Second
+}
+
+func (w *Worker) count() int64 {
+	if w.Count > 0 {
+		return w.Count
+	}
+	return 10
+}
+
+func (w *Worker) minIdle() time.Duration {
+	if w.MinIdle > 0 {
+		return w.MinIdle
+	}
+	return defaultMinIdle
+}
+
+func (w *Worker) maxFailures() int {
+	if w.MaxFailures > 0 {
+		return w.MaxFailures
+	}
+	return defaultMaxFails
+}
+
+func (w *Worker) failureAttemptsKey() string {
+	return w.stream() + ":failure-attempts"
+}
+
+func (w *Worker) logf(format string, args ...interface{}) {
+	if w.Logger != nil {
+		w.Logger.Printf(format, args...)
+		return
+	}
+	log.Printf(format, args...)
+}
+
+func sleep(ctx context.Context, duration time.Duration) {
+	timer := time.NewTimer(duration)
+	defer timer.Stop()
+	select {
+	case <-ctx.Done():
+	case <-timer.C:
+	}
+}
+
+func extractRawPayload(values map[string]interface{}) string {
+	raw, ok := values["data"]
+	if !ok {
+		return ""
+	}
+	switch value := raw.(type) {
+	case string:
+		return value
+	case []byte:
+		return string(value)
+	default:
+		return fmt.Sprint(value)
+	}
+}
+
+func commandPayload(command queue.SubmitCommand) (map[string]interface{}, error) {
+	data, err := json.Marshal(command)
+	if err != nil {
+		return nil, err
+	}
+	var payload map[string]interface{}
+	if err := json.Unmarshal(data, &payload); err != nil {
+		return nil, err
+	}
+	return payload, nil
+}