package submitworker

import (
	"context"
	"sync/atomic"
	"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
}

type acknowledgingResultOutbox struct {
	client *redis.Client
}

func (o acknowledgingResultOutbox) PublishSubmitResultAndAck(
	ctx context.Context,
	commandStream string,
	commandGroup string,
	commandMessageID string,
	_ queue.SubmitCommand,
	_ queue.SubmitResult,
) error {
	if o.client == nil {
		return nil
	}
	return o.client.XAck(ctx, commandStream, commandGroup, commandMessageID).Err()
}

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 TestConcurrencyUsesBoundedDefaultAndMaximum(t *testing.T) {
	if got := (&Worker{}).ConfiguredConcurrency(); got != defaultConcurrency {
		t.Fatalf("default concurrency = %d, want %d", got, defaultConcurrency)
	}
	if got := (&Worker{Concurrency: 2048}).ConfiguredConcurrency(); got != 1024 {
		t.Fatalf("capped concurrency = %d, want 1024", got)
	}
}

func TestMessageWorkPoolContinuouslyRefillsWithoutWaitingForSlowSibling(t *testing.T) {
	mr := miniredis.RunT(t)
	client := redis.NewClient(&redis.Options{Addr: mr.Addr()})
	startedA := make(chan struct{})
	startedB := make(chan struct{})
	startedC := make(chan struct{})
	releaseA := make(chan struct{})
	worker := &Worker{
		Redis:        client,
		ResultOutbox: acknowledgingResultOutbox{client: 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)
			case "channel-c":
				close(startedC)
			}
			return queue.SubmitResult{SubmitStatus: "accepted"}, nil
		},
	}
	pool := newMessageWorkPool(context.Background(), worker, 2)
	if !pool.dispatch(redis.XMessage{ID: "1-0", Values: submitCommandValues("message-a", "channel-a")}) ||
		!pool.dispatch(redis.XMessage{ID: "2-0", Values: submitCommandValues("message-b", "channel-b")}) {
		t.Fatal("initial messages were not dispatched")
	}
	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")
	}
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	if _, err := pool.waitForCapacity(ctx); err != nil {
		t.Fatalf("wait for refill capacity: %v", err)
	}
	if !pool.dispatch(redis.XMessage{ID: "3-0", Values: submitCommandValues("message-c", "channel-c")}) {
		t.Fatal("refill message was not dispatched")
	}
	select {
	case <-startedC:
	case <-time.After(200 * time.Millisecond):
		t.Fatal("pool waited for the slow sibling instead of refilling its free slot")
	}
	close(releaseA)
	pool.wait()
}

func TestMessageWorkPoolAcknowledgesFastMessageBeforeSlowSiblingCompletes(t *testing.T) {
	mr := miniredis.RunT(t)
	client := redis.NewClient(&redis.Options{Addr: mr.Addr()})
	worker := &Worker{Redis: client, Stream: "gateway.submit.commands", Group: "cmpp-gateway", ResultOutbox: acknowledgingResultOutbox{client: client}}
	ctx := context.Background()
	if err := worker.ensureGroup(ctx); err != nil {
		t.Fatalf("ensureGroup: %v", err)
	}
	for _, entry := range []struct{ id, messageID, channelID string }{
		{"1-0", "message-slow", "channel-slow"},
		{"2-0", "message-fast", "channel-fast"},
	} {
		if err := client.XAdd(ctx, &redis.XAddArgs{Stream: worker.stream(), ID: entry.id, Values: submitCommandValues(entry.messageID, entry.channelID)}).Err(); err != nil {
			t.Fatalf("xadd %s: %v", entry.id, err)
		}
	}
	streams, err := client.XReadGroup(ctx, &redis.XReadGroupArgs{Group: worker.group(), Consumer: worker.consumer(), Streams: []string{worker.stream(), ">"}, Count: 2}).Result()
	if err != nil || len(streams) != 1 || len(streams[0].Messages) != 2 {
		t.Fatalf("xreadgroup: streams=%+v err=%v", streams, err)
	}
	slowStarted := make(chan struct{})
	fastReturned := make(chan struct{})
	releaseSlow := make(chan struct{})
	worker.Submit = func(_ context.Context, command queue.SubmitCommand) (queue.SubmitResult, error) {
		if command.ChannelID == "channel-slow" {
			close(slowStarted)
			<-releaseSlow
		} else {
			close(fastReturned)
		}
		return queue.SubmitResult{SubmitStatus: "accepted"}, nil
	}
	pool := newMessageWorkPool(ctx, worker, 2)
	for _, message := range streams[0].Messages {
		if !pool.dispatch(message) {
			t.Fatalf("message %s was not dispatched", message.ID)
		}
	}
	<-slowStarted
	<-fastReturned
	deadline := time.Now().Add(time.Second)
	for {
		pending, pendingErr := client.XPending(ctx, worker.stream(), worker.group()).Result()
		if pendingErr != nil {
			t.Fatalf("xpending: %v", pendingErr)
		}
		if pending.Count == 1 {
			break
		}
		if time.Now().After(deadline) {
			t.Fatalf("pending count = %d, want 1 while slow sibling is still running", pending.Count)
		}
		time.Sleep(10 * time.Millisecond)
	}
	close(releaseSlow)
	pool.wait()
}

func TestPendingRecoveryDoesNotDuplicateAnActiveOrAlreadyAcknowledgedMessage(t *testing.T) {
	mr := miniredis.RunT(t)
	client := redis.NewClient(&redis.Options{Addr: mr.Addr()})
	worker := &Worker{Redis: client, Stream: "gateway.submit.commands", Group: "cmpp-gateway", MinIdle: time.Millisecond, ResultOutbox: acknowledgingResultOutbox{client: client}}
	ctx := context.Background()
	if err := worker.ensureGroup(ctx); err != nil {
		t.Fatalf("ensureGroup: %v", err)
	}
	if err := client.XAdd(ctx, &redis.XAddArgs{Stream: worker.stream(), ID: "3-0", Values: submitCommandValues("message-active", "channel-active")}).Err(); err != nil {
		t.Fatalf("xadd: %v", err)
	}
	streams, err := client.XReadGroup(ctx, &redis.XReadGroupArgs{Group: worker.group(), Consumer: worker.consumer(), Streams: []string{worker.stream(), ">"}, Count: 1}).Result()
	if err != nil || len(streams) != 1 || len(streams[0].Messages) != 1 {
		t.Fatalf("xreadgroup: streams=%+v err=%v", streams, err)
	}
	message := streams[0].Messages[0]
	started := make(chan struct{})
	release := make(chan struct{})
	var submits atomic.Int32
	worker.Submit = func(_ context.Context, _ queue.SubmitCommand) (queue.SubmitResult, error) {
		submits.Add(1)
		close(started)
		<-release
		return queue.SubmitResult{SubmitStatus: "accepted"}, nil
	}
	pool := newMessageWorkPool(ctx, worker, 2)
	if !pool.dispatch(message) {
		t.Fatal("active message was not dispatched")
	}
	<-started
	time.Sleep(5 * time.Millisecond)
	if err := worker.recoverPending(ctx, pool); err != nil {
		t.Fatalf("recoverPending: %v", err)
	}
	if got := submits.Load(); got != 1 {
		t.Fatalf("active message submit count = %d, want 1", got)
	}
	close(release)
	pool.wait()
	if err := pool.dispatchRecovered(ctx, message); err != nil {
		t.Fatalf("dispatch acknowledged recovery: %v", err)
	}
	if got := submits.Load(); got != 1 {
		t.Fatalf("acknowledged message submit count = %d, want 1", got)
	}
}

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}
		}`,
	}
}