chore: adds queue to the worker service pattern

internal/pkg/deploy/cloudformation/stack/transformers.go

@@ -45,6 +45,16 @@ const (
 	capacityProviderFargate     = "FARGATE"
 )
 
+// Time interval options.
+const (
+	retentionMinValueSeconds = 0
+	retentionMaxValueSeconds = 1209600
+	delayMinValueSeconds     = 0
+	delayMaxValueSeconds     = 900
+	timeoutMinValueSeconds   = 0
+	timeoutMaxValueSeconds   = 43200
+)
+
 var (
 	errEphemeralBadSize  = errors.New("ephemeral storage must be between 20 GiB and 200 GiB")
 	errInvalidSpotConfig = errors.New(`"count.spot" and "count.range" cannot be specified together`)
@@ -634,33 +644,125 @@ func convertTopic(t manifest.Topic, accountID, partition, region, app, env, svc
 	}, nil
 }
 
-func convertSubscribe(s *manifest.SubscribeConfig, validTopicARNs []string) (*template.SubscribeOpts, error) {
+func convertSubscribe(s *manifest.SubscribeConfig, validTopicARNs []string, accountID, region, app, env, svc string) (*template.SubscribeOpts, error) {
 	if s == nil || s.Topics == nil {
 		return nil, nil
 	}
 
+	sqsEndpoint, err := endpoints.DefaultResolver().EndpointFor(endpoints.SqsServiceID, region)
+	if err != nil {
+		return nil, err
+	}
+
 	var subscriptions template.SubscribeOpts
 	for _, sb := range *s.Topics {
-		ts, err := convertTopicSubscription(sb, validTopicARNs)
+		ts, err := convertTopicSubscription(sb, validTopicARNs, sqsEndpoint.URL, accountID, app, env, svc)
 		if err != nil {
 			return nil, err
 		}
 
 		subscriptions.Topics = append(subscriptions.Topics, ts)
 	}
+	queue, err := convertQueue(s.Queue, sqsEndpoint.URL, accountID, app, env, svc)
+	if err != nil {
+		return nil, err
+	}
+	subscriptions.Queue = queue
 
 	return &subscriptions, nil
 }
 
-func convertTopicSubscription(t manifest.TopicSubscription, validTopicARNs []string) (*template.TopicSubscription, error) {
-	err := validateTopicSubscription(t, validTopicARNs)
+func convertTopicSubscription(t manifest.TopicSubscription, validTopicARNs []string, url, accountID, app, env, svc string) (*template.TopicSubscription, error) {
+	err := validateTopicSubscription(t, validTopicARNs, app, env)
+	if err != nil {
+		return nil, fmt.Errorf(`invalid topic subscription "%s": %w`, t.Name, err)
+	}
+	queue, err := convertQueue(t.Queue, url, accountID, app, env, svc)
 	if err != nil {
 		return nil, fmt.Errorf(`invalid topic subscription "%s": %w`, t.Name, err)
 	}
 
 	return &template.TopicSubscription{
 		Name:    aws.String(t.Name),
 		Service: aws.String(t.Service),
+		Queue:   queue,
+	}, nil
+}
+
+func convertQueue(q *manifest.SQSQueue, url, accountID, app, env, svc string) (*template.SQSQueue, error) {
+	if q == nil {
+		return nil, nil
+	}
+	retention, err := convertRetention(q.Retention)
+	if err != nil {
+		return nil, fmt.Errorf(" `retention` %w", err)
+	}
+	delay, err := convertDelay(q.Delay)
+	if err != nil {
+		return nil, fmt.Errorf("`delay` %w", err)
+	}
+	timeout, err := convertTimeout(q.Timeout)
+	if err != nil {
+		return nil, fmt.Errorf("`timeout` %w", err)
+	}
+	deadletter, err := convertDeadLetter(q.DeadLetter)
+	if err != nil {
+		return nil, err
+	}
+
+	return &template.SQSQueue{
+		Retention:  retention,
+		Delay:      delay,
+		Timeout:    timeout,
+		DeadLetter: deadletter,
+		FIFO:       convertFIFO(q.FIFO),
+	}, nil
+}
+
+func convertTime(t *time.Duration, floor, ceiling time.Duration) (*int64, error) {
+	if t == nil {
+		return nil, nil
+	}
+
+	if err := validateTime(*t, floor, ceiling); err != nil {
+		return nil, err
+	}
+
+	return aws.Int64(int64(t.Seconds())), nil
+}
+
+func convertRetention(t *time.Duration) (*int64, error) {
+	return convertTime(t, retentionMinValueSeconds*time.Second, retentionMaxValueSeconds*time.Second)
+}
+
+func convertDelay(t *time.Duration) (*int64, error) {
+	return convertTime(t, delayMinValueSeconds*time.Second, delayMaxValueSeconds*time.Second)
+}
+
+func convertTimeout(t *time.Duration) (*int64, error) {
+	return convertTime(t, timeoutMinValueSeconds*time.Second, timeoutMaxValueSeconds*time.Second)
+}
+
+func convertFIFO(f *manifest.FIFOOrBool) *template.FIFOQueue {
+	if f == nil || !aws.BoolValue(f.Enabled) {
+		return nil
+	}
+
+	return &template.FIFOQueue{
+		HighThroughput: aws.BoolValue(f.FIFO.HighThroughput),
+	}
+}
+
+func convertDeadLetter(d *manifest.DeadLetterQueue) (*template.DeadLetterQueue, error) {
+	if d == nil {
+		return nil, nil
+	}
+	if err := validateDeadLetter(d); err != nil {
+		return nil, err
+	}
+
+	return &template.DeadLetterQueue{
+		Tries: d.Tries,
 	}, nil
 }
 

internal/pkg/deploy/cloudformation/stack/transformers_test.go

@@ -547,8 +547,8 @@ func Test_convertAutoscaling(t *testing.T) {
 
 func Test_convertHTTPHealthCheck(t *testing.T) {
 	// These are used by reference to represent the output of the manifest.durationp function.
-	duration15Seconds := time.Duration(15 * time.Second)
-	duration60Seconds := time.Duration(60 * time.Second)
+	duration15Seconds := 15 * time.Second
+	duration60Seconds := 60 * time.Second
 	testCases := map[string]struct {
 		inputPath               *string
 		inputSuccessCodes       *string
@@ -1565,7 +1565,14 @@ func Test_convertPublish(t *testing.T) {
 }
 
 func Test_convertSubscribe(t *testing.T) {
-	validTopics := []string{"arn:aws:us-east-1:123456789012:app-env-svc-name", "arn:aws:us-east-1:123456789012:app-env-svc-name2"}
+	validTopics := []string{"arn:aws:sns:us-west-2:123456789123:app-env-svc-name", "arn:aws:sns:us-west-2:123456789123:app-env-svc-name2"}
+	accountId := "123456789123"
+	region := "us-west-2"
+	app := "app"
+	env := "env"
+	svc := "svc"
+	duration111Seconds := 111 * time.Second
+	duration5Days := 120 * time.Hour
 	testCases := map[string]struct {
 		inSubscribe *manifest.SubscribeConfig
 
@@ -1584,14 +1591,28 @@ func Test_convertSubscribe(t *testing.T) {
 			},
 			wantedError: fmt.Errorf(`invalid topic subscription "": %w`, errMissingPublishTopicField),
 		},
-		"valid publish": {
+		"valid subscribe": {
 			inSubscribe: &manifest.SubscribeConfig{
 				Topics: &[]manifest.TopicSubscription{
 					{
 						Name:    "name",
 						Service: "svc",
 					},
 				},
+				Queue: &manifest.SQSQueue{
+					Retention: &duration111Seconds,
+					Delay:     &duration111Seconds,
+					Timeout:   &duration111Seconds,
+					DeadLetter: &manifest.DeadLetterQueue{
+						Tries: aws.Uint16(35),
+					},
+					FIFO: &manifest.FIFOOrBool{
+						Enabled: aws.Bool(true),
+						FIFO: manifest.FIFOQueue{
+							HighThroughput: aws.Bool(false),
+						},
+					},
+				},
 			},
 			wanted: &template.SubscribeOpts{
 				Topics: []*template.TopicSubscription{
@@ -1600,6 +1621,37 @@ func Test_convertSubscribe(t *testing.T) {
 						Service: aws.String("svc"),
 					},
 				},
+				Queue: &template.SQSQueue{
+					Retention: aws.Int64(111),
+					Delay:     aws.Int64(111),
+					Timeout:   aws.Int64(111),
+					DeadLetter: &template.DeadLetterQueue{
+						Tries: aws.Uint16(35),
+					},
+					FIFO: &template.FIFOQueue{
+						HighThroughput: false,
+					},
+				},
+			},
+		},
+		"valid subscribe with minimal queue": {
+			inSubscribe: &manifest.SubscribeConfig{
+				Topics: &[]manifest.TopicSubscription{
+					{
+						Name:    "name",
+						Service: "svc",
+					},
+				},
+				Queue: &manifest.SQSQueue{},
+			},
+			wanted: &template.SubscribeOpts{
+				Topics: []*template.TopicSubscription{
+					{
+						Name:    aws.String("name"),
+						Service: aws.String("svc"),
+					},
+				},
+				Queue: &template.SQSQueue{},
 			},
 		},
 		"invalid topic name": {
@@ -1635,15 +1687,96 @@ func Test_convertSubscribe(t *testing.T) {
 			},
 			wantedError: fmt.Errorf(`invalid topic subscription "topic1": %w`, errTopicSubscriptionNotAllowed),
 		},
+		"sneaky topic not allowed": {
+			inSubscribe: &manifest.SubscribeConfig{
+				Topics: &[]manifest.TopicSubscription{
+					{
+						Name:    "sneakytopic",
+						Service: "svc-name",
+					},
+				},
+			},
+			wantedError: fmt.Errorf(`invalid topic subscription "sneakytopic": %w`, errTopicSubscriptionNotAllowed),
+		},
+		"subscribe queue delay invalid": {
+			inSubscribe: &manifest.SubscribeConfig{
+				Topics: &[]manifest.TopicSubscription{
+					{
+						Name:    "name",
+						Service: "svc",
+					},
+				},
+				Queue: &manifest.SQSQueue{
+					Delay: &duration5Days,
+				},
+			},
+			wantedError: fmt.Errorf("`delay` must be between 0s and 15m0s"),
+		},
 	}
 	for name, tc := range testCases {
 		t.Run(name, func(t *testing.T) {
-			got, err := convertSubscribe(tc.inSubscribe, validTopics)
+			got, err := convertSubscribe(tc.inSubscribe, validTopics, accountId, region, app, env, svc)
 			if tc.wantedError != nil {
 				require.EqualError(t, err, tc.wantedError.Error())
 			} else {
-				require.Equal(t, got, tc.wanted)
+				require.Equal(t, tc.wanted, got)
 			}
 		})
 	}
 }
+
+func Test_convertFIFO(t *testing.T) {
+	testCases := map[string]struct {
+		inFIFO *manifest.FIFOOrBool
+
+		wanted      *template.FIFOQueue
+		wantedError error
+	}{
+		"empty FIFO": {
+			inFIFO: &manifest.FIFOOrBool{},
+			wanted: nil,
+		},
+		"FIFO with enabled false": {
+			inFIFO: &manifest.FIFOOrBool{
+				Enabled: aws.Bool(false),
+			},
+			wanted: nil,
+		},
+		"FIFO with enabled true and no high throughput": {
+			inFIFO: &manifest.FIFOOrBool{
+				Enabled: aws.Bool(true),
+			},
+			wanted: &template.FIFOQueue{
+				HighThroughput: false,
+			},
+		},
+		"FIFO with enabled true and high throughput false": {
+			inFIFO: &manifest.FIFOOrBool{
+				Enabled: aws.Bool(true),
+				FIFO: manifest.FIFOQueue{
+					HighThroughput: aws.Bool(false),
+				},
+			},
+			wanted: &template.FIFOQueue{
+				HighThroughput: false,
+			},
+		},
+		"FIFO with enabled true and high throughput true": {
+			inFIFO: &manifest.FIFOOrBool{
+				Enabled: aws.Bool(true),
+				FIFO: manifest.FIFOQueue{
+					HighThroughput: aws.Bool(true),
+				},
+			},
+			wanted: &template.FIFOQueue{
+				HighThroughput: true,
+			},
+		},
+	}
+	for name, tc := range testCases {
+		t.Run(name, func(t *testing.T) {
+			got := convertFIFO(tc.inFIFO)
+			require.Equal(t, tc.wanted, got)
+		})
+	}
+}