fm_be/internal/repository/sqs/email_queue_test.go

package sqs

import (
	"context"
	"errors"
	"io"
	"log/slog"
	"testing"
	"time"

	awsv2 "github.com/aws/aws-sdk-go-v2/aws"
	"github.com/aws/aws-sdk-go-v2/service/sqs"
	sqstypes "github.com/aws/aws-sdk-go-v2/service/sqs/types"
	"github.com/aws/smithy-go"

	"wucher/internal/config"
	"wucher/internal/queue"
	"wucher/internal/resilience"
)

type mockAPI struct {
	sendInput             *sqs.SendMessageInput
	receiveInput          *sqs.ReceiveMessageInput
	deleteBatchInput      *sqs.DeleteMessageBatchInput
	changeVisibilityInput *sqs.ChangeMessageVisibilityInput
	getAttributesInput    *sqs.GetQueueAttributesInput
	sendErr               error
	receiveErr            error
	deleteBatchErr        error
	changeVisibilityErr   error
	getAttributesErr      error
	sendCalls             int
	receiveCalls          int
}

func (m *mockAPI) SendMessage(_ context.Context, params *sqs.SendMessageInput, _ ...func(*sqs.Options)) (*sqs.SendMessageOutput, error) {
	m.sendCalls++
	m.sendInput = params
	if m.sendErr != nil {
		return nil, m.sendErr
	}
	return &sqs.SendMessageOutput{}, nil
}

func (m *mockAPI) ReceiveMessage(_ context.Context, params *sqs.ReceiveMessageInput, _ ...func(*sqs.Options)) (*sqs.ReceiveMessageOutput, error) {
	m.receiveCalls++
	m.receiveInput = params
	if m.receiveErr != nil {
		return nil, m.receiveErr
	}
	return &sqs.ReceiveMessageOutput{
		Messages: []sqstypes.Message{{
			MessageId:     awsv2.String("msg-1"),
			ReceiptHandle: awsv2.String("receipt-1"),
			Body:          awsv2.String(`{"version":"v1","message_id":"msg-1","kind":"email.dispatch","occurred_at":"2026-03-19T00:00:00Z","payload":{"to":"user@example.com","subject":"Hello","body":"World"}}`),
			MessageAttributes: map[string]sqstypes.MessageAttributeValue{
				"correlation_id": {DataType: awsv2.String("String"), StringValue: awsv2.String("req-1")},
			},
			Attributes: map[string]string{
				string(sqstypes.MessageSystemAttributeNameApproximateReceiveCount): "3",
				string(sqstypes.MessageSystemAttributeNameSentTimestamp):           "1710806400000",
			},
		}},
	}, nil
}

func (m *mockAPI) DeleteMessageBatch(_ context.Context, params *sqs.DeleteMessageBatchInput, _ ...func(*sqs.Options)) (*sqs.DeleteMessageBatchOutput, error) {
	m.deleteBatchInput = params
	if m.deleteBatchErr != nil {
		return nil, m.deleteBatchErr
	}
	return &sqs.DeleteMessageBatchOutput{}, nil
}

func (m *mockAPI) ChangeMessageVisibility(_ context.Context, params *sqs.ChangeMessageVisibilityInput, _ ...func(*sqs.Options)) (*sqs.ChangeMessageVisibilityOutput, error) {
	m.changeVisibilityInput = params
	if m.changeVisibilityErr != nil {
		return nil, m.changeVisibilityErr
	}
	return &sqs.ChangeMessageVisibilityOutput{}, nil
}

func (m *mockAPI) GetQueueAttributes(_ context.Context, params *sqs.GetQueueAttributesInput, _ ...func(*sqs.Options)) (*sqs.GetQueueAttributesOutput, error) {
	m.getAttributesInput = params
	if m.getAttributesErr != nil {
		return nil, m.getAttributesErr
	}
	return &sqs.GetQueueAttributesOutput{
		Attributes: map[string]string{
			string(sqstypes.QueueAttributeNameApproximateNumberOfMessages):           "5",
			string(sqstypes.QueueAttributeNameApproximateNumberOfMessagesNotVisible): "2",
		},
	}, nil
}

func TestEmailQueuePublishReceiveAckRetryDeadLetter(t *testing.T) {
	api := &mockAPI{}
	cfg := config.SQSConfig{
		Region:            "ap-southeast-1",
		QueueURL:          "https://sqs.ap-southeast-1.amazonaws.com/123/test",
		QueueType:         queue.QueueTypeFIFO,
		MessageGroupID:    "email",
		VisibilityTimeout: 30 * time.Second,
		WaitTimeSeconds:   20,
	}
	q, err := NewEmailQueue(api, cfg)
	if err != nil {
		t.Fatalf("new queue: %v", err)
	}

	err = q.Publish(context.Background(), queue.OutboundMessage{
		ID:              "msg-1",
		Body:            []byte(`{"hello":"world"}`),
		Attributes:      map[string]string{"correlation_id": "req-1"},
		MessageGroupID:  "email",
		DeduplicationID: "dedup-1",
	})
	if err != nil {
		t.Fatalf("publish: %v", err)
	}
	if api.sendInput == nil || awsv2.ToString(api.sendInput.MessageGroupId) != "email" {
		t.Fatalf("expected fifo publish settings")
	}

	deliveries, err := q.Receive(context.Background(), 5)
	if err != nil {
		t.Fatalf("receive: %v", err)
	}
	if len(deliveries) != 1 || deliveries[0].ReceiveCount != 3 {
		t.Fatalf("unexpected deliveries: %+v", deliveries)
	}

	if err := q.Ack(context.Background(), deliveries); err != nil {
		t.Fatalf("ack: %v", err)
	}
	if api.deleteBatchInput == nil || len(api.deleteBatchInput.Entries) != 1 {
		t.Fatalf("expected batch delete")
	}

	if err := q.Retry(context.Background(), deliveries[0], 45*time.Second); err != nil {
		t.Fatalf("retry: %v", err)
	}
	if api.changeVisibilityInput == nil || api.changeVisibilityInput.VisibilityTimeout != 45 {
		t.Fatalf("expected retry visibility timeout 45")
	}

	if err := q.DeadLetter(context.Background(), deliveries[0], "permanent failure", nil); err != nil {
		t.Fatalf("dead-letter: %v", err)
	}
	if got := awsv2.ToString(api.sendInput.MessageBody); got != `{"hello":"world"}` {
		t.Fatalf("expected dead-letter path to leave message managed by SQS redrive, got send body %q", got)
	}

	depth, err := q.ApproximateDepth(context.Background())
	if err != nil {
		t.Fatalf("depth: %v", err)
	}
	if depth != 7 {
		t.Fatalf("expected depth 7, got %d", depth)
	}
}

func TestValidateConfigRejectsInsecureEndpoint(t *testing.T) {
	err := validateConfig(config.SQSConfig{
		Region:   "ap-southeast-1",
		QueueURL: "https://sqs.ap-southeast-1.amazonaws.com/123/test",
		Endpoint: "http://localhost:4566",
	})
	if err == nil {
		t.Fatalf("expected insecure endpoint validation error")
	}
}

func TestResilientAPI_CircuitBreaker(t *testing.T) {
	newExecutor := func() *resilience.Executor {
		return resilience.NewExecutor("sqs", config.CircuitBreakerPolicy{
			Enabled:             true,
			Timeout:             time.Second,
			MinRequests:         2,
			FailureRatio:        0.5,
			ConsecutiveFailures: 2,
		}, slog.New(slog.NewJSONHandler(io.Discard, nil)), resilience.ClassifySQSError)
	}

	t.Run("client faults are excluded", func(t *testing.T) {
		api := &mockAPI{
			sendErr: sqsAPIError{code: "AccessDenied", fault: smithy.FaultClient},
		}
		wrapped := NewResilientAPI(api, newExecutor())

		for range 3 {
			_, err := wrapped.SendMessage(context.Background(), &sqs.SendMessageInput{})
			if err == nil {
				t.Fatalf("expected access denied error")
			}
			var openErr *resilience.OpenError
			if errors.As(err, &openErr) {
				t.Fatalf("expected excluded error to keep breaker closed, got %v", err)
			}
		}
		if api.sendCalls != 3 {
			t.Fatalf("expected all excluded calls to reach API, got %d", api.sendCalls)
		}
	})

	t.Run("server faults open the breaker", func(t *testing.T) {
		api := &mockAPI{
			receiveErr: sqsAPIError{code: "InternalError", fault: smithy.FaultServer},
		}
		wrapped := NewResilientAPI(api, newExecutor())

		for range 2 {
			if _, err := wrapped.ReceiveMessage(context.Background(), &sqs.ReceiveMessageInput{}); err == nil {
				t.Fatalf("expected receive error")
			}
		}
		_, err := wrapped.ReceiveMessage(context.Background(), &sqs.ReceiveMessageInput{})
		var openErr *resilience.OpenError
		if !errors.As(err, &openErr) {
			t.Fatalf("expected circuit open error, got %v", err)
		}
		if api.receiveCalls != 2 {
			t.Fatalf("expected breaker to fail fast after opening, got %d receive calls", api.receiveCalls)
		}
	})
}

type sqsAPIError struct {
	code  string
	fault smithy.ErrorFault
}

func (e sqsAPIError) Error() string                 { return e.code }
func (e sqsAPIError) ErrorCode() string             { return e.code }
func (e sqsAPIError) ErrorMessage() string          { return e.code }
func (e sqsAPIError) ErrorFault() smithy.ErrorFault { return e.fault }