Main b 22056 sns sqs deps w endpoint

.envrc

@@ -232,19 +232,22 @@ export TZ="UTC"
 
 # AWS development access
 #
-# To use S3/SES for local builds, you'll need to uncomment the following.
+# To use S3/SES or SNS & SQS for local builds, you'll need to uncomment the following.
 # Do not commit the change:
 #
 #   export STORAGE_BACKEND=s3
 #   export EMAIL_BACKEND=ses
+#   export RECEIVER_BACKEND=sns_sqs
 #
 # Instructions for using S3 storage backend here: https://dp3.atlassian.net/wiki/spaces/MT/pages/1470955567/How+to+test+storing+data+in+S3+locally
 # Instructions for using SES email backend here: https://dp3.atlassian.net/wiki/spaces/MT/pages/1467973894/How+to+test+sending+email+locally
+# Instructions for using SNS&SQS backend here: https://dp3.atlassian.net/wiki/spaces/MT/pages/2793242625/How+to+test+notifications+receiver+locally
 #
 # The default and equivalent to not being set is:
 #
 #   export STORAGE_BACKEND=local
 #   export EMAIL_BACKEND=local
+#   export RECEIVER_BACKEND=local
 #
 # Setting region and profile conditionally while we migrate from com to govcloud.
 if [ "$STORAGE_BACKEND" == "s3" ]; then
@@ -258,6 +261,13 @@ export AWS_S3_KEY_NAMESPACE=$USER
 export AWS_SES_DOMAIN="devlocal.dp3.us"
 export AWS_SES_REGION="us-gov-west-1"
 
+if [ "$RECEIVER_BACKEND" == "sns_sqs" ]; then
+  export SNS_TAGS_UPDATED_TOPIC="app_s3_tag_events"
+  export SNS_REGION="us-gov-west-1"
+# cleanup flag false by default, only used at server startup to wipe receiver artifacts from previous runs
+# export RECEIVER_CLEANUP_ON_START=false
+fi
+
 # To use s3 links aws-bucketname/xx/user/ for local builds,
 # you'll need to add the following to your .envrc.local:
 #
@@ -415,7 +425,7 @@ if [ ! -r .nix-disable  ] && has nix-env; then
 
   # add the NIX_PROFILE bin path so that everything we just installed
   # is available on the path
-  PATH_add ${NIX_PROFILE}/bin
+  PATH_add "${NIX_PROFILE}"/bin
   # Add the node binaries to our path
   PATH_add ./node_modules/.bin
   # nix is immutable, so we need to specify a path for local changes, e.g.
@@ -444,4 +454,4 @@ then
 fi
 
 # Check that all required environment variables are set
-check_required_variables
+check_required_variables

cmd/milmove/serve.go

@@ -478,6 +478,13 @@ func buildRoutingConfig(appCtx appcontext.AppContext, v *viper.Viper, redisPool
 		appCtx.Logger().Fatal("notification sender sending not enabled", zap.Error(err))
 	}
 
+	// Notification Receiver
+	runReceiverCleanup := v.GetBool(cli.ReceiverCleanupOnStartFlag) // Cleanup aws artifacts left over from previous runs
+	notificationReceiver, err := notifications.InitReceiver(v, appCtx.Logger(), runReceiverCleanup)
+	if err != nil {
+		appCtx.Logger().Fatal("notification receiver not enabled", zap.Error(err))
+	}
+
 	routingConfig.BuildRoot = v.GetString(cli.BuildRootFlag)
 	sendProductionInvoice := v.GetBool(cli.GEXSendProdInvoiceFlag)
 
@@ -567,6 +574,7 @@ func buildRoutingConfig(appCtx appcontext.AppContext, v *viper.Viper, redisPool
 		dtodRoutePlanner,
 		fileStorer,
 		notificationSender,
+		notificationReceiver,
 		iwsPersonLookup,
 		sendProductionInvoice,
 		gexSender,

go.mod

@@ -21,6 +21,8 @@ require (
 	github.com/aws/aws-sdk-go-v2/service/rds v1.78.2
 	github.com/aws/aws-sdk-go-v2/service/s3 v1.59.0
 	github.com/aws/aws-sdk-go-v2/service/ses v1.25.3
+	github.com/aws/aws-sdk-go-v2/service/sns v1.31.8
+	github.com/aws/aws-sdk-go-v2/service/sqs v1.34.6
 	github.com/aws/aws-sdk-go-v2/service/ssm v1.52.8
 	github.com/aws/aws-sdk-go-v2/service/sts v1.30.7
 	github.com/aws/smithy-go v1.20.4

go.sum

@@ -82,6 +82,10 @@ github.com/aws/aws-sdk-go-v2/service/s3 v1.59.0 h1:Cso4Ev/XauMVsbwdhYEoxg8rxZWw4
 github.com/aws/aws-sdk-go-v2/service/s3 v1.59.0/go.mod h1:BSPI0EfnYUuNHPS0uqIo5VrRwzie+Fp+YhQOUs16sKI=
 github.com/aws/aws-sdk-go-v2/service/ses v1.25.3 h1:wcfUsE2nqsXhEj68gxr7MnGXNPcBPKx0RW2DzBVgVlM=
 github.com/aws/aws-sdk-go-v2/service/ses v1.25.3/go.mod h1:6Ul/Ir8oOCsI3dFN0prULK9fvpxP+WTYmlHDkFzaAVA=
+github.com/aws/aws-sdk-go-v2/service/sns v1.31.8 h1:vRSk062d1SmaEVbiqFePkvYuhCTnW2JnPkUdt19nqeY=
+github.com/aws/aws-sdk-go-v2/service/sns v1.31.8/go.mod h1:wjhxA9hlVu75dCL/5Wcx8Cwmszvu6t0i8WEDypcB4+s=
+github.com/aws/aws-sdk-go-v2/service/sqs v1.34.6 h1:DbjODDHumQBdJ3T+EO7AXVoFUeUhAsJYOdjStH5Ws4A=
+github.com/aws/aws-sdk-go-v2/service/sqs v1.34.6/go.mod h1:7idt3XszF6sE9WPS1GqZRiDJOxw4oPtlRBXodWnCGjU=
 github.com/aws/aws-sdk-go-v2/service/ssm v1.52.8 h1:7cjN4Wp3U3cud17TsnUxSomTwKzKQGUWdq/N1aWqgMk=
 github.com/aws/aws-sdk-go-v2/service/ssm v1.52.8/go.mod h1:nUSNPaG8mv5rIu7EclHnFqZOjhreEUwRKENtKTtJ9aw=
 github.com/aws/aws-sdk-go-v2/service/sso v1.22.7 h1:pIaGg+08llrP7Q5aiz9ICWbY8cqhTkyy+0SHvfzQpTc=

pkg/cli/receiver.go

@@ -0,0 +1,61 @@
+package cli
+
+import (
+	"fmt"
+
+	"github.com/spf13/pflag"
+	"github.com/spf13/viper"
+)
+
+const (
+	// ReceiverBackendFlag is the Receiver Backend Flag
+	ReceiverBackendFlag string = "receiver-backend"
+	// SNSTagsUpdatedTopicFlag is the SNS Tags Updated Topic Flag
+	SNSTagsUpdatedTopicFlag string = "sns-tags-updated-topic"
+	// SNSRegionFlag is the SNS Region flag
+	SNSRegionFlag string = "sns-region"
+	// SNSAccountId is the application's AWS account id
+	SNSAccountId string = "aws-account-id"
+	// ReceiverCleanupOnStartFlag is the Receiver Cleanup On Start Flag
+	ReceiverCleanupOnStartFlag string = "receiver-cleanup-on-start"
+)
+
+// InitReceiverFlags initializes Storage command line flags
+func InitReceiverFlags(flag *pflag.FlagSet) {
+	flag.String(ReceiverBackendFlag, "local", "Receiver backend to use, either local or sns_sqs.")
+	flag.String(SNSTagsUpdatedTopicFlag, "", "SNS Topic for receiving event messages")
+	flag.String(SNSRegionFlag, "", "Region used for SNS and SQS")
+	flag.String(SNSAccountId, "", "SNS account Id")
+	flag.Bool(ReceiverCleanupOnStartFlag, false, "Receiver will cleanup previous aws artifacts on start.")
+}
+
+// CheckReceiver validates Storage command line flags
+func CheckReceiver(v *viper.Viper) error {
+
+	receiverBackend := v.GetString(ReceiverBackendFlag)
+	if !stringSliceContains([]string{"local", "sns_sqs"}, receiverBackend) {
+		return fmt.Errorf("invalid receiver_backend %s, expecting local or sns_sqs", receiverBackend)
+	}
+
+	receiverCleanupOnStart := v.GetString(ReceiverCleanupOnStartFlag)
+	if !stringSliceContains([]string{"true", "false"}, receiverCleanupOnStart) {
+		return fmt.Errorf("invalid receiver_cleanup_on_start %s, expecting true or false", receiverCleanupOnStart)
+	}
+
+	if receiverBackend == "sns_sqs" {
+		r := v.GetString(SNSRegionFlag)
+		if r == "" {
+			return fmt.Errorf("invalid value for %s: %s", SNSRegionFlag, r)
+		}
+		topic := v.GetString(SNSTagsUpdatedTopicFlag)
+		if topic == "" {
+			return fmt.Errorf("invalid value for %s: %s", SNSTagsUpdatedTopicFlag, topic)
+		}
+		accountId := v.GetString(SNSAccountId)
+		if topic == "" {
+			return fmt.Errorf("invalid value for %s: %s", SNSAccountId, accountId)
+		}
+	}
+
+	return nil
+}

pkg/cli/receiver_test.go

@@ -0,0 +1,6 @@
+package cli
+
+func (suite *cliTestSuite) TestConfigReceiver() {
+	suite.Setup(InitReceiverFlags, []string{})
+	suite.NoError(CheckReceiver(suite.viper))
+}

pkg/gen/internalapi/configure_mymove.go

@@ -4,6 +4,7 @@ package internalapi
 
 import (
 	"crypto/tls"
+	"io"
 	"net/http"
 
 	"github.com/go-openapi/errors"
@@ -60,6 +61,9 @@ func configureAPI(api *internaloperations.MymoveAPI) http.Handler {
 
 	api.BinProducer = runtime.ByteStreamProducer()
 	api.JSONProducer = runtime.JSONProducer()
+	api.TextEventStreamProducer = runtime.ProducerFunc(func(w io.Writer, data interface{}) error {
+		return errors.NotImplemented("textEventStream producer has not yet been implemented")
+	})
 
 	// You may change here the memory limit for this multipart form parser. Below is the default (32 MB).
 	// ppm.CreatePPMUploadMaxParseMemory = 32 << 20
@@ -205,6 +209,11 @@ func configureAPI(api *internaloperations.MymoveAPI) http.Handler {
 			return middleware.NotImplemented("operation transportation_offices.GetTransportationOffices has not yet been implemented")
 		})
 	}
+	if api.UploadsGetUploadStatusHandler == nil {
+		api.UploadsGetUploadStatusHandler = uploads.GetUploadStatusHandlerFunc(func(params uploads.GetUploadStatusParams) middleware.Responder {
+			return middleware.NotImplemented("operation uploads.GetUploadStatus has not yet been implemented")
+		})
+	}
 	if api.EntitlementsIndexEntitlementsHandler == nil {
 		api.EntitlementsIndexEntitlementsHandler = entitlements.IndexEntitlementsHandlerFunc(func(params entitlements.IndexEntitlementsParams) middleware.Responder {
 			return middleware.NotImplemented("operation entitlements.IndexEntitlements has not yet been implemented")