lislgosms/gateway/internal/resultoutbox/outbox_test.go

package resultoutbox

import (
	"context"
	"encoding/json"
	"net/http"
	"net/http/httptest"
	"sync/atomic"
	"testing"
	"time"

	"cmpp-platform/gateway/internal/queue"

	"github.com/alicebob/miniredis/v2"
	"github.com/redis/go-redis/v9"
)

func TestPublishSubmitSegmentIsIdempotent(t *testing.T) {
	mr := miniredis.RunT(t)
	client := redis.NewClient(&redis.Options{Addr: mr.Addr()})
	outbox := New(client)
	command := testCommand()
	segment := queue.SubmitSegmentResult{SegmentTotal: 1, SegmentIndex: 1, SequenceID: 7, GatewayMessageID: "88", SubmitStatus: "accepted"}

	if err := outbox.PublishSubmitSegment(context.Background(), command, segment); err != nil {
		t.Fatalf("first publish: %v", err)
	}
	if err := outbox.PublishSubmitSegment(context.Background(), command, segment); err != nil {
		t.Fatalf("duplicate publish: %v", err)
	}
	if got := client.XLen(context.Background(), outbox.StreamName()).Val(); got != 1 {
		t.Fatalf("stream length = %d, want 1", got)
	}
}

func TestBatchCallbackSendsMultipleEventsInOneRequest(t *testing.T) {
	mr := miniredis.RunT(t)
	client := redis.NewClient(&redis.Options{Addr: mr.Addr()})
	var requests atomic.Int32
	var eventCount atomic.Int32
	server := httptest.NewServer(http.HandlerFunc(func(response http.ResponseWriter, request *http.Request) {
		requests.Add(1)
		var batch callbackBatch
		if err := json.NewDecoder(request.Body).Decode(&batch); err != nil {
			t.Errorf("decode batch: %v", err)
			return
		}
		eventCount.Store(int32(len(batch.Events)))
		result := callbackBatchResponse{BatchID: batch.BatchID}
		for _, event := range batch.Events {
			result.Results = append(result.Results, callbackEventResult{EventID: event.EventID, Accepted: true})
		}
		_ = json.NewEncoder(response).Encode(result)
	}))
	defer server.Close()
	outbox := New(client)
	outbox.APIBaseURL = server.URL
	outbox.BatchEnabled = true
	outbox.BatchSize = 50
	outbox.BatchWait = 10 * time.Millisecond
	outbox.GatewayInstanceID = "gateway-test"
	command := testCommand()
	for index := 1; index <= 2; index++ {
		if err := outbox.PublishSubmitSegment(context.Background(), command, queue.SubmitSegmentResult{SegmentTotal: 2, SegmentIndex: index, SequenceID: uint32(index), GatewayMessageID: "88", SubmitStatus: "accepted"}); err != nil {
			t.Fatal(err)
		}
	}
	ctx, cancel := context.WithCancel(context.Background())
	done := make(chan error, 1)
	go func() { done <- outbox.Run(ctx) }()
	deadline := time.Now().Add(3 * time.Second)
	for client.XLen(context.Background(), outbox.StreamName()).Val() != 0 {
		if time.Now().After(deadline) {
			t.Fatal("batch did not drain")
		}
		time.Sleep(10 * time.Millisecond)
	}
	cancel()
	<-done
	if requests.Load() != 1 || eventCount.Load() != 2 {
		t.Fatalf("requests/events=%d/%d, want 1/2", requests.Load(), eventCount.Load())
	}
}

func TestBatchCallbackReplaysWholeRequestAfterHTTPFailure(t *testing.T) {
	mr := miniredis.RunT(t)
	client := redis.NewClient(&redis.Options{Addr: mr.Addr()})
	var calls atomic.Int32
	server := httptest.NewServer(http.HandlerFunc(func(response http.ResponseWriter, request *http.Request) {
		var batch callbackBatch
		_ = json.NewDecoder(request.Body).Decode(&batch)
		if calls.Add(1) == 1 {
			http.Error(response, "busy", http.StatusServiceUnavailable)
			return
		}
		result := callbackBatchResponse{BatchID: batch.BatchID}
		for _, event := range batch.Events {
			result.Results = append(result.Results, callbackEventResult{EventID: event.EventID, Accepted: true})
		}
		_ = json.NewEncoder(response).Encode(result)
	}))
	defer server.Close()
	outbox := New(client)
	outbox.APIBaseURL = server.URL
	outbox.BatchEnabled = true
	outbox.BatchWait = 5 * time.Millisecond
	outbox.MinIdle = 5 * time.Millisecond
	outbox.GatewayInstanceID = "g"
	if err := outbox.PublishSubmitSegment(context.Background(), testCommand(), queue.SubmitSegmentResult{SegmentTotal: 1, SegmentIndex: 1, SequenceID: 1, GatewayMessageID: "1", SubmitStatus: "accepted"}); err != nil {
		t.Fatal(err)
	}
	ctx, cancel := context.WithCancel(context.Background())
	done := make(chan error, 1)
	go func() { done <- outbox.Run(ctx) }()
	deadline := time.Now().Add(3 * time.Second)
	for client.XLen(context.Background(), outbox.StreamName()).Val() != 0 {
		if time.Now().After(deadline) {
			t.Fatal("replayed batch did not drain")
		}
		time.Sleep(10 * time.Millisecond)
	}
	cancel()
	<-done
	if calls.Load() < 2 {
		t.Fatalf("calls=%d want replay", calls.Load())
	}
}

func TestPublishAggregateAndCommandAckAreAtomicAndIdempotent(t *testing.T) {
	mr := miniredis.RunT(t)
	client := redis.NewClient(&redis.Options{Addr: mr.Addr()})
	ctx := context.Background()
	commandStream := "gateway.submit.commands"
	commandGroup := "cmpp-gateway"
	if err := client.XGroupCreateMkStream(ctx, commandStream, commandGroup, "0").Err(); err != nil {
		t.Fatalf("create command group: %v", err)
	}
	commandID, err := client.XAdd(ctx, &redis.XAddArgs{Stream: commandStream, Values: map[string]interface{}{"data": "command"}}).Result()
	if err != nil {
		t.Fatalf("add command: %v", err)
	}
	if _, err := client.XReadGroup(ctx, &redis.XReadGroupArgs{Group: commandGroup, Consumer: "gateway-1", Streams: []string{commandStream, ">"}, Count: 1}).Result(); err != nil {
		t.Fatalf("claim command: %v", err)
	}
	outbox := New(client)
	command := testCommand()
	result := queue.SubmitResult{Envelope: command.Envelope, SubmitID: command.SubmitID, GatewayMessageID: "99", SubmitStatus: "accepted"}

	for attempt := 0; attempt < 2; attempt++ {
		if err := outbox.PublishSubmitResultAndAck(ctx, commandStream, commandGroup, commandID, command, result); err != nil {
			t.Fatalf("publish attempt %d: %v", attempt+1, err)
		}
	}
	pending, err := client.XPending(ctx, commandStream, commandGroup).Result()
	if err != nil {
		t.Fatalf("command pending: %v", err)
	}
	if pending.Count != 0 {
		t.Fatalf("command pending = %d, want 0", pending.Count)
	}
	if got := client.XLen(ctx, outbox.StreamName()).Val(); got != 1 {
		t.Fatalf("result stream length = %d, want 1", got)
	}
}

func TestCallbackWorkerRetriesAndOnlyDeletesAfterSuccess(t *testing.T) {
	mr := miniredis.RunT(t)
	client := redis.NewClient(&redis.Options{Addr: mr.Addr()})
	var calls atomic.Int32
	server := httptest.NewServer(http.HandlerFunc(func(response http.ResponseWriter, request *http.Request) {
		if request.Header.Get("X-CMPP-Result-Event-ID") == "" {
			t.Error("missing result event id header")
		}
		if calls.Add(1) == 1 {
			http.Error(response, "temporary failure", http.StatusServiceUnavailable)
			return
		}
		response.WriteHeader(http.StatusCreated)
	}))
	defer server.Close()

	outbox := New(client)
	outbox.APIBaseURL = server.URL
	outbox.MinIdle = 10 * time.Millisecond
	outbox.Concurrency = 1
	command := testCommand()
	if err := outbox.PublishSubmitSegment(context.Background(), command, queue.SubmitSegmentResult{
		SegmentTotal: 1, SegmentIndex: 1, SequenceID: 7, GatewayMessageID: "88", SubmitStatus: "accepted",
	}); err != nil {
		t.Fatalf("publish segment: %v", err)
	}

	ctx, cancel := context.WithCancel(context.Background())
	done := make(chan error, 1)
	go func() { done <- outbox.Run(ctx) }()
	deadline := time.Now().Add(5 * time.Second)
	for calls.Load() < 2 || client.XLen(context.Background(), outbox.StreamName()).Val() != 0 {
		if time.Now().After(deadline) {
			t.Fatalf("calls=%d streamLength=%d", calls.Load(), client.XLen(context.Background(), outbox.StreamName()).Val())
		}
		time.Sleep(10 * time.Millisecond)
	}
	cancel()
	select {
	case <-done:
	case <-time.After(3 * time.Second):
		t.Fatal("worker did not stop")
	}
}

func testCommand() queue.SubmitCommand {
	return queue.SubmitCommand{
		Envelope: queue.Envelope{
			SchemaVersion: queue.SchemaVersion,
			MessageType:   queue.MessageTypeSubmitCommand,
			TraceID:       "trace-1",
			MessageID:     "message-1",
			ChannelID:     "channel-1",
			CreatedAt:     time.Now().UTC(),
		},
		SubmitID: "submit-1",
	}
}