Merged in feature/docinit (pull request #43)

Move queue to config

* movequeue

* passtests

.mockery.yml

@@ -3,7 +3,7 @@ with-expecter: true
 outpkg: "{{.PackageName}}mock"
 dir: "mocks/{{.PackageName}}"
 packages:
-  queryorchestration/internal/server/queue:
+  queryorchestration/internal/serviceconfig/queue:
     interfaces:
       SQSClient:
   queryorchestration/internal/server/runner:

internal/document/init/create.go

@@ -7,7 +7,7 @@ import (
 	"queryorchestration/internal/document"
 	documentclean "queryorchestration/internal/document/clean"
 	"queryorchestration/internal/job"
-	"queryorchestration/internal/server/queue"
+	"queryorchestration/internal/serviceconfig/queue"
 
 	"github.com/aws/aws-sdk-go-v2/service/sqs/types"
 	"github.com/google/uuid"
@@ -79,12 +79,13 @@ func (s *Service) informCreate(ctx context.Context, id uuid.UUID, j *job.Job) er
 		return nil
 	}
 
-	err := queue.Send(ctx, &queue.Config{
-		Client: s.cfg.GetQueueClient(),
-		URL:    s.cfg.GetDocumentCleanURL(),
-	}, documentclean.Create{
-		ID: id,
-	}, map[string]types.MessageAttributeValue{})
+	err := s.cfg.SendToQueue(ctx, &queue.SendParams{
+		QueueURL: s.cfg.GetDocumentCleanURL(),
+		Body: documentclean.Create{
+			ID: id,
+		},
+		Attributes: map[string]types.MessageAttributeValue{},
+	})
 	if err != nil {
 		return err
 	}

internal/objectstorage/get.go

@@ -0,0 +1,8 @@
+package objectstorage
+
+import (
+	"context"
+)
+
+func Get(ctx context.Context) {
+}

internal/server/queue/config.go

@@ -1,6 +0,0 @@
-package queue
-
-type Config struct {
-	URL    string
-	Client SQSClient
-}

internal/server/queue/delete.go

@@ -1,21 +0,0 @@
-package queue
-
-import (
-	"context"
-
-	"github.com/aws/aws-sdk-go-v2/aws"
-	"github.com/aws/aws-sdk-go-v2/service/sqs"
-	"github.com/aws/aws-sdk-go-v2/service/sqs/types"
-)
-
-func Delete(ctx context.Context, config *Config, msg *types.Message) error {
-	_, err := config.Client.DeleteMessage(ctx, &sqs.DeleteMessageInput{
-		QueueUrl:      aws.String(config.URL),
-		ReceiptHandle: msg.ReceiptHandle,
-	})
-	if err != nil {
-		return err
-	}
-
-	return nil
-}

internal/server/queue/receive.go

@@ -1,17 +0,0 @@
-package queue
-
-import (
-	"context"
-
-	"github.com/aws/aws-sdk-go-v2/service/sqs"
-)
-
-func Receive(ctx context.Context, config *Config, attributes []string) (*sqs.ReceiveMessageOutput, error) {
-	return config.Client.ReceiveMessage(ctx, &sqs.ReceiveMessageInput{
-		QueueUrl:              &config.URL,
-		MaxNumberOfMessages:   1,
-		WaitTimeSeconds:       2,
-		VisibilityTimeout:     2,
-		MessageAttributeNames: attributes,
-	})
-}

internal/server/queue/send.go

@@ -1,30 +0,0 @@
-package queue
-
-import (
-	"context"
-	"encoding/json"
-
-	"github.com/aws/aws-sdk-go-v2/aws"
-	"github.com/aws/aws-sdk-go-v2/service/sqs"
-	"github.com/aws/aws-sdk-go-v2/service/sqs/types"
-)
-
-func Send(ctx context.Context, config *Config, body interface{}, attributes map[string]types.MessageAttributeValue) error {
-	jsonBytes, err := json.Marshal(body)
-	if err != nil {
-		return err
-	}
-
-	strBody := string(jsonBytes)
-
-	_, err = config.Client.SendMessage(ctx, &sqs.SendMessageInput{
-		MessageAttributes: attributes,
-		QueueUrl:          aws.String(config.URL),
-		MessageBody:       aws.String(strBody),
-	})
-	if err != nil {
-		return err
-	}
-
-	return nil
-}

internal/server/runner/listener.go

@@ -73,9 +73,7 @@ func New(ctx context.Context, cfg ListenerConfig) (*Server, error) {
 	}
 
 	return &Server{
-		controller: cfg.GetController(),
-		queueURL:   cfg.GetQueueURL(),
-		client:     cfg.GetQueueClient(),
-		cleanup:    cleanup,
+		cfg:     cfg,
+		cleanup: cleanup,
 	}, nil
 }

internal/server/runner/poll.go

@@ -4,14 +4,12 @@ import (
 	"context"
 	"fmt"
 	"log/slog"
-	"queryorchestration/internal/server/queue"
+	"queryorchestration/internal/serviceconfig/queue"
 )
 
 type Server struct {
-	controller Controller
-	queueURL   string
-	client     queue.SQSClient
-	cleanup    func() error
+	cleanup func() error
+	cfg     ListenerConfig
 }
 
 func (c *Server) Listen(ctx context.Context) {
@@ -63,11 +61,9 @@ func (c *Server) Listen(ctx context.Context) {
 //}
 
 func (c *Server) pollMessage(ctx context.Context) error {
-	cfg := &queue.Config{
-		URL:    c.queueURL,
-		Client: c.client,
-	}
-	result, err := queue.Receive(ctx, cfg, []string{})
+	result, err := c.cfg.ReceiveFromQueue(ctx, &queue.ReceiveParams{
+		QueueURL: c.cfg.GetQueueURL(),
+	})
 	if err != nil {
 		return fmt.Errorf("message fetch fail: %v", err)
 	}
@@ -78,12 +74,15 @@ func (c *Server) pollMessage(ctx context.Context) error {
 		// simple case.
 		// if process takes too long to complete then we need to use a thread pool.
 		go func() {
-			err := c.controller.Process(ctx, &message)
+			err := c.cfg.GetController().Process(ctx, &message)
 			if err != nil {
 				slog.Error("message process fail", "err", err)
 			}
 
-			err = queue.Delete(ctx, cfg, &message)
+			err = c.cfg.DeleteFromQueue(ctx, &queue.DeleteParams{
+				QueueURL:      c.cfg.GetQueueURL(),
+				ReceiptHandle: message.ReceiptHandle,
+			})
 			if err != nil {
 				slog.Error("message delete fail", "err", err)
 			}

internal/server/runner/poll_test.go

@@ -2,7 +2,6 @@ package runner
 
 import (
 	"context"
-	"queryorchestration/internal/server/queue"
 	queuemock "queryorchestration/mocks/queue"
 	runnermock "queryorchestration/mocks/runner"
 	"testing"
@@ -17,12 +16,13 @@ import (
 func TestPollMessages(t *testing.T) {
 	ctx := context.Background()
 
-	controller := runnermock.NewMockController(t)
 	mockSQS := queuemock.NewMockSQSClient(t)
-	cfg := &queue.Config{
-		URL:    "/i/am/here",
-		Client: mockSQS,
+	cfg := &BaseConfig{}
+	cfg.QueueClient = mockSQS
+	cfg.ControllerFunc = func() Controller {
+		return runnermock.NewMockController(t)
 	}
+	cfg.QueueURL = "/i/am/here"
 
 	res := sqs.ReceiveMessageOutput{Messages: []types.Message{}}
 
@@ -30,7 +30,7 @@ func TestPollMessages(t *testing.T) {
 		ReceiveMessage(
 			mock.Anything,
 			mock.MatchedBy(func(in *sqs.ReceiveMessageInput) bool {
-				return *in.QueueUrl == cfg.URL
+				return *in.QueueUrl == cfg.QueueURL
 			}),
 			mock.Anything,
 		).
@@ -40,10 +40,8 @@ func TestPollMessages(t *testing.T) {
 	defer cancel()
 
 	scfg := &Server{
-		controller: controller,
-		queueURL:   cfg.URL,
-		client:     cfg.Client,
-		cleanup:    func() error { return nil },
+		cfg:     cfg,
+		cleanup: func() error { return nil },
 	}
 	scfg.Listen(ctx)
 }
@@ -51,26 +49,29 @@ func TestPollMessages(t *testing.T) {
 func TestPollMessage(t *testing.T) {
 	ctx := context.Background()
 
-	controller := runnermock.NewMockController(t)
 	mockSQS := queuemock.NewMockSQSClient(t)
+	cfg := &BaseConfig{}
+	cfg.QueueClient = mockSQS
+	cfg.ControllerFunc = func() Controller {
+		return runnermock.NewMockController(t)
+	}
+	cfg.QueueURL = "/i/am/here"
 
-	cfg := &Server{
-		controller: controller,
-		queueURL:   "/i/am/here",
-		client:     mockSQS,
-		cleanup:    func() error { return nil },
+	ser := &Server{
+		cfg:     cfg,
+		cleanup: func() error { return nil },
 	}
 
 	mockSQS.EXPECT().
 		ReceiveMessage(
 			mock.Anything,
 			mock.MatchedBy(func(in *sqs.ReceiveMessageInput) bool {
-				return *in.QueueUrl == cfg.queueURL
+				return *in.QueueUrl == cfg.QueueURL
 			}),
 			mock.Anything,
 		).
 		Return(&sqs.ReceiveMessageOutput{}, nil)
 
-	err := cfg.pollMessage(ctx)
+	err := ser.pollMessage(ctx)
 	assert.NoError(t, err)
 }

internal/serviceconfig/queue/config.go

@@ -3,7 +3,6 @@ package queue
 import (
 	"context"
 	"fmt"
-	"queryorchestration/internal/server/queue"
 	"queryorchestration/internal/serviceconfig/aws"
 
 	"github.com/aws/aws-sdk-go-v2/config"
@@ -15,17 +14,20 @@ type QueueConfig struct {
 	aws.AWSConfig
 	EnableOtel        bool   `env:"ENABLE_OTEL" envDefault:"false"`
 	AWSEndpointUrlSQS string `env:"AWS_ENDPOINT_URL_SQS"`
-	QueueClient       queue.SQSClient
+	QueueClient       SQSClient
 }
 
 type ConfigProvider interface {
 	aws.ConfigProvider
 	SetQueueClient(context.Context) error
-	GetQueueClient() queue.SQSClient
+	GetQueueClient() SQSClient
 	PingQueueByURL(context.Context, string) error
+	SendToQueue(context.Context, *SendParams) error
+	ReceiveFromQueue(context.Context, *ReceiveParams) (*sqs.ReceiveMessageOutput, error)
+	DeleteFromQueue(context.Context, *DeleteParams) error
 }
 
-func (c *QueueConfig) GetQueueClient() queue.SQSClient {
+func (c *QueueConfig) GetQueueClient() SQSClient {
 	return c.QueueClient
 }
 

internal/serviceconfig/queue/delete.go

@@ -0,0 +1,25 @@
+package queue
+
+import (
+	"context"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/service/sqs"
+)
+
+type DeleteParams struct {
+	QueueURL      string
+	ReceiptHandle *string
+}
+
+func (c *QueueConfig) DeleteFromQueue(ctx context.Context, params *DeleteParams) error {
+	_, err := c.QueueClient.DeleteMessage(ctx, &sqs.DeleteMessageInput{
+		QueueUrl:      aws.String(params.QueueURL),
+		ReceiptHandle: params.ReceiptHandle,
+	})
+	if err != nil {
+		return err
+	}
+
+	return nil
+}

internal/serviceconfig/queue/delete_test.go

@@ -2,12 +2,11 @@ package queue_test
 
 import (
 	"context"
-	"queryorchestration/internal/server/queue"
+	"queryorchestration/internal/serviceconfig/queue"
 	queuemock "queryorchestration/mocks/queue"
 	"testing"
 
 	"github.com/aws/aws-sdk-go-v2/service/sqs"
-	"github.com/aws/aws-sdk-go-v2/service/sqs/types"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
 )
@@ -15,26 +14,24 @@ import (
 func TestDelete(t *testing.T) {
 	ctx := context.Background()
 	mockSQS := queuemock.NewMockSQSClient(t)
-	cfg := &queue.Config{
-		URL:    "/i/am/here",
-		Client: mockSQS,
-	}
+	cfg := queue.QueueConfig{}
+	cfg.QueueClient = mockSQS
 	handle := "iamahandler"
+	params := &queue.DeleteParams{
+		ReceiptHandle: &handle,
+		QueueURL:      "/i/am/here",
+	}
 
 	mockSQS.EXPECT().
 		DeleteMessage(
 			mock.Anything,
 			mock.MatchedBy(func(in *sqs.DeleteMessageInput) bool {
-				return *in.QueueUrl == cfg.URL && *in.ReceiptHandle == handle
+				return *in.QueueUrl == params.QueueURL && *in.ReceiptHandle == handle
 			}),
 			mock.Anything,
 		).
 		Return(&sqs.DeleteMessageOutput{}, nil)
 
-	message := &types.Message{
-		ReceiptHandle: &handle,
-	}
-
-	err := queue.Delete(ctx, cfg, message)
+	err := cfg.DeleteFromQueue(ctx, params)
 	assert.NoError(t, err)
 }

internal/serviceconfig/queue/receive.go

@@ -0,0 +1,22 @@
+package queue
+
+import (
+	"context"
+
+	"github.com/aws/aws-sdk-go-v2/service/sqs"
+)
+
+type ReceiveParams struct {
+	QueueURL   string
+	Attributes []string
+}
+
+func (c *QueueConfig) ReceiveFromQueue(ctx context.Context, params *ReceiveParams) (*sqs.ReceiveMessageOutput, error) {
+	return c.QueueClient.ReceiveMessage(ctx, &sqs.ReceiveMessageInput{
+		QueueUrl:              &params.QueueURL,
+		MaxNumberOfMessages:   1,
+		WaitTimeSeconds:       2,
+		VisibilityTimeout:     2,
+		MessageAttributeNames: params.Attributes,
+	})
+}

internal/serviceconfig/queue/receive_test.go

@@ -2,7 +2,7 @@ package queue_test
 
 import (
 	"context"
-	"queryorchestration/internal/server/queue"
+	"queryorchestration/internal/serviceconfig/queue"
 	queuemock "queryorchestration/mocks/queue"
 	"testing"
 
@@ -15,9 +15,10 @@ import (
 func TestReceive(t *testing.T) {
 	ctx := context.Background()
 	mockSQS := queuemock.NewMockSQSClient(t)
-	cfg := &queue.Config{
-		URL:    "/i/am/here",
-		Client: mockSQS,
+	cfg := queue.QueueConfig{}
+	cfg.QueueClient = mockSQS
+	params := &queue.ReceiveParams{
+		QueueURL: "/i/am/here",
 	}
 
 	res := sqs.ReceiveMessageOutput{Messages: []types.Message{}}
@@ -26,13 +27,13 @@ func TestReceive(t *testing.T) {
 		ReceiveMessage(
 			mock.Anything,
 			mock.MatchedBy(func(in *sqs.ReceiveMessageInput) bool {
-				return *in.QueueUrl == cfg.URL
+				return *in.QueueUrl == params.QueueURL
 			}),
 			mock.Anything,
 		).
 		Return(&res, nil)
 
-	ares, err := queue.Receive(ctx, cfg, []string{})
+	ares, err := cfg.ReceiveFromQueue(ctx, params)
 	assert.NoError(t, err)
 	assert.EqualExportedValues(t, res, *ares)
 }

internal/serviceconfig/queue/send.go

@@ -0,0 +1,36 @@
+package queue
+
+import (
+	"context"
+	"encoding/json"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/service/sqs"
+	"github.com/aws/aws-sdk-go-v2/service/sqs/types"
+)
+
+type SendParams struct {
+	QueueURL   string
+	Body       interface{}
+	Attributes map[string]types.MessageAttributeValue
+}
+
+func (c *QueueConfig) SendToQueue(ctx context.Context, params *SendParams) error {
+	jsonBytes, err := json.Marshal(params.Body)
+	if err != nil {
+		return err
+	}
+
+	strBody := string(jsonBytes)
+
+	_, err = c.QueueClient.SendMessage(ctx, &sqs.SendMessageInput{
+		MessageAttributes: params.Attributes,
+		QueueUrl:          aws.String(params.QueueURL),
+		MessageBody:       aws.String(strBody),
+	})
+	if err != nil {
+		return err
+	}
+
+	return nil
+}

internal/serviceconfig/queue/send_test.go

@@ -2,12 +2,11 @@ package queue_test
 
 import (
 	"context"
-	"queryorchestration/internal/server/queue"
+	"queryorchestration/internal/serviceconfig/queue"
 	queuemock "queryorchestration/mocks/queue"
 	"testing"
 
 	"github.com/aws/aws-sdk-go-v2/service/sqs"
-	"github.com/aws/aws-sdk-go-v2/service/sqs/types"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
 )
@@ -15,24 +14,23 @@ import (
 func TestSend(t *testing.T) {
 	ctx := context.Background()
 	mockSQS := queuemock.NewMockSQSClient(t)
-	cfg := &queue.Config{
-		URL:    "/i/am/here",
-		Client: mockSQS,
+	cfg := queue.QueueConfig{}
+	cfg.QueueClient = mockSQS
+	params := &queue.SendParams{
+		QueueURL: "/i/am/here",
+		Body:     struct{}{},
 	}
 
 	mockSQS.EXPECT().
 		SendMessage(
 			mock.Anything,
 			mock.MatchedBy(func(in *sqs.SendMessageInput) bool {
-				return *in.QueueUrl == cfg.URL && *in.MessageBody == "{}"
+				return *in.QueueUrl == params.QueueURL && *in.MessageBody == "{}"
 			}),
 			mock.Anything,
 		).
 		Return(&sqs.SendMessageOutput{}, nil)
 
-	body := struct{}{}
-	attrs := map[string]types.MessageAttributeValue{}
-
-	err := queue.Send(ctx, cfg, body, attrs)
+	err := cfg.SendToQueue(ctx, params)
 	assert.NoError(t, err)
 }

internal/test/queue.go

@@ -3,8 +3,8 @@ package test
 import (
 	"context"
 	"fmt"
-	"queryorchestration/internal/server/queue"
 	"queryorchestration/internal/serviceconfig"
+	"queryorchestration/internal/serviceconfig/queue"
 	"testing"
 	"time"
 
@@ -131,10 +131,10 @@ func CreateQueue(t *testing.T, ctx context.Context, cfg *serviceconfig.BaseConfi
 	return *queueM.QueueUrl
 }
 
-func AssertMessageWait(t *testing.T, ctx context.Context, cfg *queue.Config, attrs []string) types.Message {
+func AssertMessageWait(t *testing.T, ctx context.Context, cfg serviceconfig.ConfigProvider, params *queue.ReceiveParams) types.Message {
 	time.Sleep(1 * time.Second)
 
-	result, err := queue.Receive(ctx, cfg, attrs)
+	result, err := cfg.ReceiveFromQueue(ctx, params)
 	assert.NoError(t, err)
 	assert.NotNil(t, result.Messages)
 	assert.Len(t, result.Messages, 1)
@@ -143,14 +143,19 @@ func AssertMessageWait(t *testing.T, ctx context.Context, cfg *queue.Config, att
 	return result.Messages[0]
 }
 
-func AssertMessageBodyWait(t *testing.T, ctx context.Context, cfg *queue.Config, body string) {
-	message := AssertMessageWait(t, ctx, cfg, []string{})
+func AssertMessageBodyWait(t *testing.T, ctx context.Context, cfg serviceconfig.ConfigProvider, url string, body string) {
+	message := AssertMessageWait(t, ctx, cfg, &queue.ReceiveParams{
+		QueueURL: url,
+	})
 
 	assert.Equal(t, body, *message.Body)
 }
 
-func AssertMessageAttrWait(t *testing.T, ctx context.Context, cfg *queue.Config, name string, value string) {
-	message := AssertMessageWait(t, ctx, cfg, []string{name})
+func AssertMessageAttrWait(t *testing.T, ctx context.Context, cfg serviceconfig.ConfigProvider, url string, name string, value string) {
+	message := AssertMessageWait(t, ctx, cfg, &queue.ReceiveParams{
+		QueueURL:   url,
+		Attributes: []string{name},
+	})
 
 	assert.NotNil(t, message.MessageAttributes[name])
 	assert.Equal(t, value, *(message.MessageAttributes[name]).StringValue)

internal/test/queue_test.go

@@ -2,7 +2,7 @@ package test
 
 import (
 	"context"
-	"queryorchestration/internal/server/queue"
+	"queryorchestration/internal/serviceconfig/queue"
 	"testing"
 
 	"github.com/aws/aws-sdk-go-v2/aws"
@@ -73,15 +73,15 @@ func TestAssertMessageWait(t *testing.T) {
 	defer cleanup()
 	url := CreateQueue(t, ctx, cfg, "myname")
 
-	qucfg := &queue.Config{
-		URL:    url,
-		Client: cfg.QueueClient,
-	}
-
-	err := queue.Send(ctx, qucfg, "body", map[string]types.MessageAttributeValue{})
+	err := cfg.SendToQueue(ctx, &queue.SendParams{
+		QueueURL: url,
+		Body:     "body",
+	})
 	assert.NoError(t, err)
 
-	msg := AssertMessageWait(t, ctx, qucfg, []string{})
+	msg := AssertMessageWait(t, ctx, cfg, &queue.ReceiveParams{
+		QueueURL: url,
+	})
 	assert.NotNil(t, msg)
 }
 
@@ -99,15 +99,13 @@ func TestAssertMessageBodyWait(t *testing.T) {
 	defer cleanup()
 	url := CreateQueue(t, ctx, cfg, "myname")
 
-	qucfg := &queue.Config{
-		URL:    url,
-		Client: cfg.QueueClient,
-	}
-
-	err := queue.Send(ctx, qucfg, "body", map[string]types.MessageAttributeValue{})
+	err := cfg.SendToQueue(ctx, &queue.SendParams{
+		QueueURL: url,
+		Body:     "body",
+	})
 	assert.NoError(t, err)
 
-	AssertMessageBodyWait(t, ctx, qucfg, "\"body\"")
+	AssertMessageBodyWait(t, ctx, cfg, url, "\"body\"")
 }
 
 func TestAssertMessageAttrWait(t *testing.T) {
@@ -124,21 +122,20 @@ func TestAssertMessageAttrWait(t *testing.T) {
 	defer cleanup()
 	url := CreateQueue(t, ctx, cfg, "myname")
 
-	qucfg := &queue.Config{
-		URL:    url,
-		Client: cfg.QueueClient,
-	}
-
 	name := "name"
 	value := "value"
 
-	err := queue.Send(ctx, qucfg, "body", map[string]types.MessageAttributeValue{
-		name: {
-			DataType:    aws.String("String"),
-			StringValue: aws.String(value),
+	err := cfg.SendToQueue(ctx, &queue.SendParams{
+		QueueURL: url,
+		Body:     "body",
+		Attributes: map[string]types.MessageAttributeValue{
+			name: {
+				DataType:    aws.String("String"),
+				StringValue: aws.String(value),
+			},
 		},
 	})
 	assert.NoError(t, err)
 
-	AssertMessageAttrWait(t, ctx, qucfg, name, value)
+	AssertMessageAttrWait(t, ctx, cfg, url, name, value)
 }

test/docInitRunner/docinitrunner_test.go

@@ -5,12 +5,11 @@ import (
 	"os"
 	"path"
 	documentinit "queryorchestration/internal/document/init"
-	"queryorchestration/internal/server/queue"
+	"queryorchestration/internal/serviceconfig/queue"
 	"queryorchestration/internal/test"
 	queryservice "queryorchestration/pkg/queryService"
 	"testing"
 
-	"github.com/aws/aws-sdk-go-v2/service/sqs/types"
 	"github.com/stretchr/testify/assert"
 )
 
@@ -76,16 +75,13 @@ func TestDocInitRunner(t *testing.T) {
 		Location: "/I/am/here", // TODO
 	}
 
-	qcfg := &queue.Config{
-		URL:    net.Runners[test.DocInitRunner].URI,
-		Client: cfg.QueueClient,
-	}
-
-	err = queue.Send(ctx, qcfg, document, map[string]types.MessageAttributeValue{})
+	err = cfg.SendToQueue(ctx, &queue.SendParams{
+		QueueURL: net.Runners[test.DocInitRunner].URI,
+		Body:     document,
+	})
 	assert.NoError(t, err)
 
-	_ = test.AssertMessageWait(t, ctx, &queue.Config{
-		URL:    doccleanurl,
-		Client: cfg.GetQueueClient(),
-	}, []string{})
+	_ = test.AssertMessageWait(t, ctx, cfg, &queue.ReceiveParams{
+		QueueURL: doccleanurl,
+	})
 }

test/queryRunner/queryrunner_test.go

@@ -3,11 +3,10 @@ package integration_test
 import (
 	"context"
 	"queryorchestration/internal/query"
-	"queryorchestration/internal/server/queue"
+	"queryorchestration/internal/serviceconfig/queue"
 	"queryorchestration/internal/test"
 	"testing"
 
-	"github.com/aws/aws-sdk-go-v2/service/sqs/types"
 	"github.com/google/uuid"
 	"github.com/stretchr/testify/assert"
 )
@@ -29,12 +28,11 @@ func TestQueryRunner(t *testing.T) {
 		CleanVersion: int32(1),
 		TextVersion:  int32(1),
 	}
-	qcfg := &queue.Config{
-		URL:    c.URI,
-		Client: cfg.QueueClient,
-	}
 
-	err := queue.Send(ctx, qcfg, document, map[string]types.MessageAttributeValue{})
+	err := cfg.SendToQueue(ctx, &queue.SendParams{
+		QueueURL: c.URI,
+		Body:     document,
+	})
 	assert.NoError(t, err)
 
 	// TODO - check document output