lislgosms/gateway/internal/submitworker/worker_test.go

package submitworker

import (
	"context"
	"testing"
	"time"

	"cmpp-platform/gateway/internal/queue"
	"github.com/alicebob/miniredis/v2"
	"github.com/redis/go-redis/v9"
)

type recordingLimiter struct {
	channelID string
	rate      int
	called    bool
}

func (l *recordingLimiter) Wait(_ context.Context, channelID string, rate int) (time.Duration, error) {
	l.called = true
	l.channelID = channelID
	l.rate = rate
	return 0, nil
}

func TestCommandFromStreamValuesParsesSubmitCommand(t *testing.T) {
	command, err := CommandFromStreamValues(map[string]interface{}{
		"messageType": "SubmitCommand",
		"data": `{
			"schemaVersion": "v1",
			"messageType": "SubmitCommand",
			"traceId": "trace-worker-0001",
			"messageId": "msg-worker-0001",
			"channelId": "channel-1",
			"createdAt": "2026-07-07T10:00:00Z",
			"tenantId": "tenant-1",
			"applicationId": "app-1",
			"submitId": "submit-1",
			"phoneNumber": "13800138000",
			"content": "hello",
			"signature": "测试",
			"templateId": "tpl-1",
			"billingUnits": 1,
			"queuePriority": "normal",
			"route": {
				"channelCode": "CMPP-A",
				"cmppAccountCode": "account-a",
				"priority": 0,
				"rateLimitPerSecond": 100
			},
			"cmpp": {
				"serviceId": "SMS",
				"srcId": "10690000",
				"registeredDelivery": 1,
				"msgFmt": 8
			},
			"upstream": {
				"gatewayHost": "127.0.0.1",
				"gatewayPort": 17890,
				"account": "account-a",
				"passwordCipher": "secret",
				"cmppVersion": "3.0"
			},
			"retry": {
				"attempt": 0,
				"maxAttempts": 1
			}
		}`,
	})
	if err != nil {
		t.Fatalf("parse stream command: %v", err)
	}
	if command.MessageID != "msg-worker-0001" || command.Upstream.Account != "account-a" {
		t.Fatalf("unexpected command: %+v", command)
	}
}

func TestCommandFromStreamValuesRejectsMissingData(t *testing.T) {
	_, err := CommandFromStreamValues(map[string]interface{}{"messageType": "SubmitCommand"})
	if err == nil {
		t.Fatal("expected missing data error")
	}
}

func TestHandleMessageUsesInjectedSubmit(t *testing.T) {
	var got queue.SubmitCommand
	limiter := &recordingLimiter{}
	worker := &Worker{
		Limiter: limiter,
		Submit: func(_ context.Context, command queue.SubmitCommand) (queue.SubmitResult, error) {
			got = command
			return queue.SubmitResult{SubmitStatus: "accepted"}, nil
		},
	}
	if err := worker.handleCommand(context.Background(), queue.SubmitCommand{
		Envelope:      queue.Envelope{MessageID: "msg-worker-0002", ChannelID: "channel-1"},
		SubmitID:      "submit-2",
		PhoneNumber:   "13800138000",
		Content:       "hello",
		Upstream:      queue.UpstreamConfig{GatewayHost: "127.0.0.1", GatewayPort: 17890, Account: "account-a", PasswordCipher: "secret", CMPPVersion: "3.0"},
		CMPP:          queue.CMPP{ServiceID: "SMS", SrcID: "10690000", RegisteredDelivery: 1, MsgFmt: 8},
		Route:         queue.Route{ChannelCode: "CMPP-A", RateLimitPerSecond: 320},
		Retry:         queue.Retry{Attempt: 0, MaxAttempts: 1},
		ApplicationID: "app-1",
		TenantID:      "tenant-1",
	}); err != nil {
		t.Fatalf("handleCommand returned error: %v", err)
	}
	if got.MessageID != "msg-worker-0002" || got.SubmitID != "submit-2" {
		t.Fatalf("unexpected command: %+v", got)
	}
	if !limiter.called || limiter.channelID != "channel-1" || limiter.rate != 320 {
		t.Fatalf("unexpected limiter call: %+v", limiter)
	}
}

func TestMinIdleDefaultsToThirtySeconds(t *testing.T) {
	worker := &Worker{}
	if got := worker.minIdle(); got != defaultMinIdle {
		t.Fatalf("minIdle = %v, want %v", got, defaultMinIdle)
	}
}

func TestProcessMessagesDoesNotLetOneChannelBlockAnother(t *testing.T) {
	mr := miniredis.RunT(t)
	client := redis.NewClient(&redis.Options{Addr: mr.Addr()})
	startedA := make(chan struct{})
	startedB := make(chan struct{})
	releaseA := make(chan struct{})
	worker := &Worker{
		Redis: client,
		Submit: func(_ context.Context, command queue.SubmitCommand) (queue.SubmitResult, error) {
			switch command.ChannelID {
			case "channel-a":
				close(startedA)
				<-releaseA
			case "channel-b":
				close(startedB)
			}
			return queue.SubmitResult{SubmitStatus: "accepted"}, nil
		},
	}
	messages := []redis.XMessage{
		{ID: "1-0", Values: submitCommandValues("message-a", "channel-a")},
		{ID: "2-0", Values: submitCommandValues("message-b", "channel-b")},
	}
	done := make(chan struct{})
	go func() {
		_ = worker.processMessages(context.Background(), messages)
		close(done)
	}()
	select {
	case <-startedA:
	case <-time.After(time.Second):
		t.Fatal("channel-a did not start")
	}
	select {
	case <-startedB:
	case <-time.After(200 * time.Millisecond):
		t.Fatal("channel-b was blocked by channel-a")
	}
	close(releaseA)
	select {
	case <-done:
	case <-time.After(time.Second):
		t.Fatal("message batch did not complete")
	}
}

func TestProcessMessageDeadLettersAfterMaxFailures(t *testing.T) {
	mr := miniredis.RunT(t)
	client := redis.NewClient(&redis.Options{Addr: mr.Addr()})
	reported := []DeadLetterEvent{}
	worker := &Worker{
		Redis:       client,
		Stream:      "gateway.submit.commands",
		Group:       "cmpp-gateway",
		MaxFailures: 2,
		Submit: func(_ context.Context, command queue.SubmitCommand) (queue.SubmitResult, error) {
			return queue.SubmitResult{}, context.DeadlineExceeded
		},
		ReportDeadLetter: func(_ context.Context, event DeadLetterEvent) error {
			reported = append(reported, event)
			return nil
		},
	}
	ctx := context.Background()
	if err := worker.ensureGroup(ctx); err != nil {
		t.Fatalf("ensureGroup: %v", err)
	}
	message := redis.XMessage{
		ID: "1710000000000-0",
		Values: map[string]interface{}{
			"messageType": "SubmitCommand",
			"data": `{
				"schemaVersion": "v1",
				"messageType": "SubmitCommand",
				"traceId": "trace-worker-0003",
				"messageId": "msg-worker-0003",
				"channelId": "channel-1",
				"createdAt": "2026-07-08T10:00:00Z",
				"tenantId": "tenant-1",
				"applicationId": "app-1",
				"submitId": "submit-3",
				"phoneNumber": "13800138000",
				"content": "hello",
				"signature": "测试",
				"templateId": "tpl-1",
				"billingUnits": 1,
				"queuePriority": "normal",
				"route": { "channelCode": "CMPP-A", "cmppAccountCode": "account-a", "priority": 0 },
				"cmpp": { "serviceId": "SMS", "srcId": "10690000", "registeredDelivery": 1, "msgFmt": 8 },
				"upstream": { "gatewayHost": "127.0.0.1", "gatewayPort": 17890, "account": "account-a", "passwordCipher": "secret", "cmppVersion": "3.0" },
				"retry": { "attempt": 0, "maxAttempts": 1 }
			}`,
		},
	}
	if err := client.XAdd(ctx, &redis.XAddArgs{
		Stream: worker.stream(),
		ID:     message.ID,
		Values: message.Values,
	}).Err(); err != nil {
		t.Fatalf("xadd: %v", err)
	}

	if err := worker.processMessage(ctx, message); err == nil {
		t.Fatal("expected first failure")
	}
	if len(reported) != 0 {
		t.Fatalf("unexpected dead letters on first failure: %+v", reported)
	}

	if err := worker.processMessage(ctx, message); err != nil {
		t.Fatalf("second failure should dead-letter and ack, got %v", err)
	}
	if len(reported) != 1 {
		t.Fatalf("dead letters = %d, want 1", len(reported))
	}
	if reported[0].FailureCode != "SUBMIT_PROCESSING_FAILED" || reported[0].Attempts != 2 {
		t.Fatalf("unexpected dead letter: %+v", reported[0])
	}
	if client.HGet(ctx, worker.failureAttemptsKey(), message.ID).Err() != redis.Nil {
		t.Fatalf("failure attempt key was not cleared")
	}
}

func submitCommandValues(messageID string, channelID string) map[string]interface{} {
	return map[string]interface{}{
		"data": `{
			"schemaVersion":"v1","messageType":"SubmitCommand","traceId":"trace-1",
			"messageId":"` + messageID + `","channelId":"` + channelID + `","submitId":"submit-1",
			"tenantId":"tenant-1","applicationId":"app-1","phoneNumber":"13800138000","content":"hello",
			"route":{"channelCode":"CMPP-A","rateLimitPerSecond":100},
			"cmpp":{"serviceId":"SMS","srcId":"10690000","registeredDelivery":1,"msgFmt":8},
			"upstream":{"gatewayHost":"127.0.0.1","gatewayPort":17890,"account":"sp","passwordCipher":"secret","cmppVersion":"3.0"},
			"retry":{"attempt":0,"maxAttempts":1}
		}`,
	}
}