lislgosms/gateway/internal/submitworker/worker_test.go

package submitworker

import (
	"context"
	"testing"

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

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
	worker := &Worker{
		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"},
		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"},
		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)
	}
}

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

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")
	}
}