feat: add worker services to svc init

internal/pkg/cli/flag.go

@@ -52,6 +52,9 @@ const (
 	deleteSecretFlag      = "delete-secret"
 	svcPortFlag           = "port"
 
+	noSubscriptionFlag  = "no-subscribe"
+	subscribeTopicsFlag = "subscribe-topics"
+
 	storageTypeFlag              = "storage-type"
 	storagePartitionKeyFlag      = "partition-key"
 	storageSortKeyFlag           = "sort-key"
@@ -218,6 +221,10 @@ Defaults to all logs. Only one of end-time / follow may be used.`
 	deleteSecretFlagDescription      = "Deletes AWS Secrets Manager secret associated with a pipeline source repository."
 	svcPortFlagDescription           = "The port on which your service listens."
 
+	noSubscriptionFlagDescription  = "Optional. Turn off selection for adding subscriptions for worker services."
+	subscribeTopicsFlagDescription = `Optional. SNS Topics to subscribe to from other services in your application.
+Must be of format '<svcName>:<topicName>'`
+
 	storageFlagDescription             = "Name of the storage resource to create."
 	storageWorkloadFlagDescription     = "Name of the service or job to associate with storage."
 	storagePartitionKeyFlagDescription = `Partition key for the DDB table.

internal/pkg/cli/init.go

@@ -7,6 +7,7 @@ package cli
 import (
 	"fmt"
 
+	"github.com/aws/copilot-cli/internal/pkg/deploy"
 	"github.com/aws/copilot-cli/internal/pkg/docker/dockerengine"
 
 	"github.com/aws/aws-sdk-go/aws"
@@ -103,6 +104,11 @@ func newInitOpts(vars initVars) (*initOpts, error) {
 	}
 	prompt := prompt.New()
 	sel := selector.NewWorkspaceSelect(prompt, ssm, ws)
+	deployStore, err := deploy.NewStore(ssm)
+	if err != nil {
+		return nil, err
+	}
+	snsSel := selector.NewDeploySelect(prompt, ssm, deployStore)
 	spin := termprogress.NewSpinner(log.DiagnosticWriter)
 	id := identity.New(defaultSess)
 	deployer := cloudformation.New(defaultSess)
@@ -240,6 +246,7 @@ func newInitOpts(vars initVars) (*initOpts, error) {
 					fs:           fs,
 					init:         wlInitializer,
 					sel:          sel,
+					topicSel:     snsSel,
 					prompt:       prompt,
 					dockerEngine: dockerengine.New(cmd),
 				}

internal/pkg/cli/init_test.go

@@ -121,6 +121,10 @@ func TestInitOpts_Run(t *testing.T) {
 						Value: manifest.BackendServiceType,
 						Hint:  "ECS on Fargate",
 					},
+					{
+						Value: manifest.WorkerServiceType,
+						Hint:  "Events to SQS to ECS on Fargate",
+					},
 					{
 						Value: manifest.ScheduledJobType,
 						Hint:  "Scheduled event to State Machine to Fargate",

internal/pkg/cli/interfaces.go

@@ -506,6 +506,10 @@ type dockerfileSelector interface {
 	Dockerfile(selPrompt, notFoundPrompt, selHelp, notFoundHelp string, pv prompt.ValidatorFunc) (string, error)
 }
 
+type topicSelector interface {
+	Topics(prompt, help, app string) ([]deploy.Topic, error)
+}
+
 type ec2Selector interface {
 	VPC(prompt, help string) (string, error)
 	PublicSubnets(prompt, help, vpcID string) ([]string, error)

internal/pkg/cli/svc_init.go

@@ -12,6 +12,7 @@ import (
 
 	"github.com/aws/copilot-cli/internal/pkg/aws/sessions"
 	"github.com/aws/copilot-cli/internal/pkg/config"
+	"github.com/aws/copilot-cli/internal/pkg/deploy"
 	"github.com/aws/copilot-cli/internal/pkg/deploy/cloudformation"
 	"github.com/aws/copilot-cli/internal/pkg/exec"
 	"github.com/aws/copilot-cli/internal/pkg/initialize"
@@ -61,12 +62,17 @@ Deployed resources (such as your ECR repository, logs) will contain this %[1]s's
 	svcInitSvcPortPrompt     = "Which %s do you want customer traffic sent to?"
 	svcInitSvcPortHelpPrompt = `The port will be used by the load balancer to route incoming traffic to this service.
 You should set this to the port which your Dockerfile uses to communicate with the internet.`
+
+	svcInitPublisherPrompt     = "Which publishers do you want to subscribe to?"
+	svcInitPublisherHelpPrompt = `A publisher is an existing SNS Topic to which a service publishes messages. 
+These messages can be consumed by the Worker Service.`
 )
 
 var serviceTypeHints = map[string]string{
 	manifest.RequestDrivenWebServiceType: "App Runner",
 	manifest.LoadBalancedWebServiceType:  "Internet to ECS on Fargate",
 	manifest.BackendServiceType:          "ECS on Fargate",
+	manifest.WorkerServiceType:           "Events to SQS to ECS on Fargate",
 }
 
 type initWkldVars struct {
@@ -75,6 +81,8 @@ type initWkldVars struct {
 	name           string
 	dockerfilePath string
 	image          string
+	subscriptions  []string
+	noSubscribe    bool
 }
 
 type initSvcVars struct {
@@ -92,10 +100,12 @@ type initSvcOpts struct {
 	prompt       prompter
 	dockerEngine dockerEngine
 	sel          dockerfileSelector
+	topicSel     topicSelector
 
 	// Outputs stored on successful actions.
 	manifestPath string
 	platform     *string
+	topics       []manifest.TopicSubscription
 
 	// Cache variables
 	df dockerfileParser
@@ -122,6 +132,11 @@ func newInitSvcOpts(vars initSvcVars) (*initSvcOpts, error) {
 	}
 	prompter := prompt.New()
 	sel := selector.NewWorkspaceSelect(prompter, store, ws)
+	deployStore, err := deploy.NewStore(store)
+	if err != nil {
+		return nil, err
+	}
+	snsSel := selector.NewDeploySelect(prompter, store, deployStore)
 
 	initSvc := &initialize.WorkloadInitializer{
 		Store:    store,
@@ -137,6 +152,7 @@ func newInitSvcOpts(vars initSvcVars) (*initSvcOpts, error) {
 		init:         initSvc,
 		prompt:       prompter,
 		sel:          sel,
+		topicSel:     snsSel,
 		dockerEngine: dockerengine.New(exec.NewCmd()),
 	}
 	opts.dockerfile = func(path string) dockerfileParser {
@@ -182,6 +198,9 @@ func (o *initSvcOpts) Validate() error {
 			return err
 		}
 	}
+	if err := validateSubscribe(o.noSubscribe, o.subscriptions); err != nil {
+		return err
+	}
 	return nil
 }
 
@@ -208,6 +227,10 @@ func (o *initSvcOpts) Ask() error {
 		return err
 	}
 
+	if err := o.askSvcPublishers(); err != nil {
+		return err
+	}
+
 	return nil
 }
 
@@ -243,6 +266,7 @@ func (o *initSvcOpts) Execute() error {
 			DockerfilePath: o.dockerfilePath,
 			Image:          o.image,
 			Platform:       o.platform,
+			Topics:         o.topics,
 		},
 		Port:        o.port,
 		HealthCheck: hc,
@@ -273,6 +297,7 @@ func (o *initSvcOpts) askSvcType() error {
 		manifest.RequestDrivenWebServiceType,
 		manifest.LoadBalancedWebServiceType,
 		manifest.BackendServiceType,
+		manifest.WorkerServiceType,
 	)
 	msg := fmt.Sprintf(fmtSvcInitSvcTypePrompt, color.Emphasize("service type"))
 
@@ -393,8 +418,8 @@ func (o *initSvcOpts) askSvcPort() (err error) {
 			defaultPort = strconv.Itoa(int(ports[0]))
 		}
 	}
-	// Skip asking if it is a backend service.
-	if o.wkldType == manifest.BackendServiceType {
+	// Skip asking if it is a backend or worker service.
+	if o.wkldType == manifest.BackendServiceType || o.wkldType == manifest.WorkerServiceType {
 		return nil
 	}
 
@@ -419,6 +444,58 @@ func (o *initSvcOpts) askSvcPort() (err error) {
 	return nil
 }
 
+func (o *initSvcOpts) askSvcPublishers() (err error) {
+	if o.wkldType != manifest.WorkerServiceType {
+		return nil
+	}
+	// publishers already specified by flags
+	if len(o.subscriptions) > 0 {
+		var topicSubscriptions []manifest.TopicSubscription
+		for _, sub := range o.subscriptions {
+			sub, err := parseSerializedSubscription(sub)
+			if err != nil {
+				return err
+			}
+			topicSubscriptions = append(topicSubscriptions, sub)
+		}
+		o.topics = topicSubscriptions
+		return nil
+	}
+
+	// if --no-subscriptions flag specified, no need to ask for publishers
+	if o.noSubscribe {
+		return nil
+	}
+
+	topics, err := o.topicSel.Topics(svcInitPublisherPrompt, svcInitPublisherHelpPrompt, o.appName)
+	if err != nil {
+		return fmt.Errorf("select publisher: %w", err)
+	}
+
+	subscriptions := make([]manifest.TopicSubscription, 0, len(topics))
+	for _, t := range topics {
+		subscriptions = append(subscriptions, manifest.TopicSubscription{
+			Name:    t.Name(),
+			Service: t.Workload(),
+		})
+	}
+	o.topics = subscriptions
+
+	return nil
+}
+
+// parseSerializedSubscription parses the service and topic name out of keys specified in the form "service:topicName"
+func parseSerializedSubscription(input string) (manifest.TopicSubscription, error) {
+	attrs := regexpMatchSubscription.FindStringSubmatch(input)
+	if len(attrs) == 0 {
+		return manifest.TopicSubscription{}, fmt.Errorf("parse subscription from key: %s", input)
+	}
+	return manifest.TopicSubscription{
+		Name:    attrs[2],
+		Service: attrs[1],
+	}, nil
+}
+
 func parseHealthCheck(df dockerfileParser) (*manifest.ContainerHealthCheck, error) {
 	hc, err := df.GetHealthCheck()
 	if err != nil {
@@ -489,5 +566,8 @@ This command is also run as part of "copilot init".`,
 	cmd.Flags().StringVarP(&vars.dockerfilePath, dockerFileFlag, dockerFileFlagShort, "", dockerFileFlagDescription)
 	cmd.Flags().StringVarP(&vars.image, imageFlag, imageFlagShort, "", imageFlagDescription)
 	cmd.Flags().Uint16Var(&vars.port, svcPortFlag, 0, svcPortFlagDescription)
+	cmd.Flags().StringArrayVar(&vars.subscriptions, subscribeTopicsFlag, []string{}, subscribeTopicsFlagDescription)
+	cmd.Flags().BoolVar(&vars.noSubscribe, noSubscriptionFlag, false, noSubscriptionFlagDescription)
+
 	return cmd
 }