feat: Persist prepared data from grants.gov to DynamoDB (#46)

* feat: Persist prepared data from grants.gov to DynamoDB

* init commit v2

* updated lambda code

* Fix missing permissions boundary in localstack environments

* basic functionality appears to work now

* clean up + trying to make tests work

* troubleshooting unit tests

* cleaning up lambda and tests

* removing unused vars

* adding updated go dep files

* fixing tf fmt issues

* hardcoding region for dynamo tests

* oops -- no need to use dynamodb.Client here when iface will do

* readding SendMetric, removing unused function

* fixing go mod/sum

* removing old unused env var

* addressing various review items

* addressing more review items

* tf lint

* putting ddb contributor insights behind a bool

* oops

* removing unused iam permissions + moving bucket notificiation resource

* updating the comments

* Add newline to end of file

* Build(deps): Bump gopkg.in/DataDog/dd-trace-go.v1 from 1.50.0 to 1.50.1 (#74)

Bumps gopkg.in/DataDog/dd-trace-go.v1 from 1.50.0 to 1.50.1.

---
updated-dependencies:
- dependency-name: gopkg.in/DataDog/dd-trace-go.v1
  dependency-type: direct:production
  update-type: version-update:semver-patch
...

Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>

* removing tf lock file

* fixes based on review feedback

* Bump go.mod versions

* Ignore .DS_Store files

---------

Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: TylerHendrickson <hendrickson.tsh@gmail.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>

.gitignore

@@ -5,6 +5,9 @@
 *.so
 *.dylib
 
+# MacOS
+.DS_Store
+
 # IDE settings
 .idea/
 .vscode/

Taskfile.yml

@@ -110,6 +110,7 @@ tasks:
       - build-DownloadGrantsGovDB
       - build-SplitGrantsGovXMLDB
       - build-EnqueueFFISDownload
+      - build-PersistGrantsGovXMLDB
       - build-DownloadFFISSpreadsheet
 
   build-DownloadGrantsGovDB:
@@ -133,6 +134,13 @@ tasks:
         vars:
           LAMBDA_CMD: EnqueueFFISDownload
 
+  build-PersistGrantsGovXMLDB:
+    desc: Compiles PersistGrantsGovXMLDB
+    cmds:
+      - task: build-lambda
+        vars:
+          LAMBDA_CMD: PersistGrantsGovXMLDB
+
   build-DownloadFFISSpreadsheet:
     desc: Compiles DownloadFFISSpreadsheet
     cmds:

cmd/PersistGrantsGovXMLDB/dynamodb.go

@@ -0,0 +1,55 @@
+package main
+
+import (
+	"context"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/feature/dynamodb/attributevalue"
+	"github.com/aws/aws-sdk-go-v2/feature/dynamodb/expression"
+	"github.com/aws/aws-sdk-go-v2/service/dynamodb"
+	"github.com/aws/aws-sdk-go-v2/service/dynamodb/types"
+)
+
+type DynamoDBUpdateItemAPI interface {
+	UpdateItem(ctx context.Context, params *dynamodb.UpdateItemInput, optFns ...func(*dynamodb.Options)) (*dynamodb.UpdateItemOutput, error)
+}
+
+func UpdateDynamoDBItem(ctx context.Context, c DynamoDBUpdateItemAPI, table string, opp opportunity) error {
+	key, err := buildKey(opp)
+	if err != nil {
+		return err
+	}
+	expr, err := buildUpdateExpression(opp)
+	if err != nil {
+		return err
+	}
+	_, err = c.UpdateItem(ctx, &dynamodb.UpdateItemInput{
+		TableName:                 aws.String(table),
+		Key:                       key,
+		ExpressionAttributeNames:  expr.Names(),
+		ExpressionAttributeValues: expr.Values(),
+		UpdateExpression:          expr.Update(),
+		ReturnValues:              types.ReturnValueUpdatedNew,
+	})
+	return err
+}
+
+func buildKey(o opportunity) (map[string]types.AttributeValue, error) {
+	oid, err := attributevalue.Marshal(o.OpportunityID)
+
+	return map[string]types.AttributeValue{"grant_id": oid}, err
+}
+
+func buildUpdateExpression(o opportunity) (expression.Expression, error) {
+	oppAttr, err := attributevalue.MarshalMap(o)
+	if err != nil {
+		return expression.Expression{}, err
+	}
+
+	update := expression.UpdateBuilder{}
+	for k, v := range oppAttr {
+		update = update.Set(expression.Name(k), expression.Value(v))
+	}
+
+	return expression.NewBuilder().WithUpdate(update).Build()
+}

cmd/PersistGrantsGovXMLDB/dynamodb_test.go

@@ -0,0 +1,76 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/service/dynamodb"
+	"github.com/aws/aws-sdk-go-v2/service/dynamodb/types"
+	"github.com/stretchr/testify/assert"
+	grantsgov "github.com/usdigitalresponse/grants-ingest/pkg/grantsSchemas/grants.gov"
+)
+
+type mockUpdateItemAPI func(ctx context.Context, params *dynamodb.UpdateItemInput, optFns ...func(*dynamodb.Options)) (*dynamodb.UpdateItemOutput, error)
+
+func (m mockUpdateItemAPI) UpdateItem(ctx context.Context, params *dynamodb.UpdateItemInput, optFns ...func(*dynamodb.Options)) (*dynamodb.UpdateItemOutput, error) {
+	return m(ctx, params, optFns...)
+}
+
+type mockDynamoDBUpdateItemAPI struct {
+	mockUpdateItemAPI
+}
+
+func TestUploadDynamoDBItem(t *testing.T) {
+	now := time.Now()
+	testTableName := "test-table"
+	testHashKey := map[string]types.AttributeValue{}
+	testHashKey["grant_id"] = &types.AttributeValueMemberS{Value: "123456"}
+	testError := fmt.Errorf("oh no this is an error")
+	testOpportunity := opportunity{
+		OpportunityID:   "123456",
+		LastUpdatedDate: grantsgov.MMDDYYYYType(now.Format(grantsgov.TimeLayoutMMDDYYYYType)),
+	}
+
+	for _, tt := range []struct {
+		name   string
+		client func(t *testing.T) DynamoDBUpdateItemAPI
+		expErr error
+	}{
+		{
+			"UpdateItem successful",
+			func(t *testing.T) DynamoDBUpdateItemAPI {
+				return mockUpdateItemAPI(func(ctx context.Context, params *dynamodb.UpdateItemInput, optFns ...func(*dynamodb.Options)) (*dynamodb.UpdateItemOutput, error) {
+					t.Helper()
+					assert.Equal(t, aws.String(testTableName), params.TableName)
+					assert.Equal(t, testHashKey, params.Key)
+					return &dynamodb.UpdateItemOutput{}, nil
+				})
+			},
+			nil,
+		},
+		{
+			"UpdateItem returns error",
+			func(t *testing.T) DynamoDBUpdateItemAPI {
+				return mockUpdateItemAPI(func(ctx context.Context, params *dynamodb.UpdateItemInput, optFns ...func(*dynamodb.Options)) (*dynamodb.UpdateItemOutput, error) {
+					t.Helper()
+					assert.Equal(t, aws.String(testTableName), params.TableName)
+					assert.Equal(t, testHashKey, params.Key)
+					return &dynamodb.UpdateItemOutput{}, testError
+				})
+			},
+			testError,
+		},
+	} {
+		t.Run(tt.name, func(t *testing.T) {
+			err := UpdateDynamoDBItem(context.TODO(), tt.client(t), testTableName, testOpportunity)
+			if tt.expErr != nil {
+				assert.EqualError(t, err, tt.expErr.Error())
+			} else {
+				assert.NoError(t, err)
+			}
+		})
+	}
+}

cmd/PersistGrantsGovXMLDB/handler.go

@@ -0,0 +1,98 @@
+package main
+
+import (
+	"context"
+	"encoding/xml"
+	"io"
+
+	"github.com/aws/aws-lambda-go/events"
+	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/service/s3"
+	"github.com/hashicorp/go-multierror"
+	"github.com/usdigitalresponse/grants-ingest/internal/log"
+	grantsgov "github.com/usdigitalresponse/grants-ingest/pkg/grantsSchemas/grants.gov"
+	"gopkg.in/DataDog/dd-trace-go.v1/ddtrace/tracer"
+)
+
+const (
+	MB                         = int64(1024 * 1024)
+	GRANT_OPPORTUNITY_XML_NAME = "OpportunitySynopsisDetail_1_0"
+)
+
+type opportunity grantsgov.OpportunitySynopsisDetail_1_0
+
+// handleS3Event handles events representing S3 bucket notifications of type "ObjectCreated:*"
+// for XML DB extracts saved from Grants.gov and split into separate files via the SplitGrantsGovXMLDB Lambda.
+// The XML data from the source S3 object provided represents an individual grant opportunity.
+// Returns an error that represents any and all errors accumulated during the invocation,
+// either while handling a source object or while processing its contents; an error may indicate
+// a partial or complete invocation failure.
+// Returns nil when all grant opportunities are successfully processed from all source records,
+// indicating complete success.
+func handleS3EventWithConfig(s3svc *s3.Client, dynamodbsvc DynamoDBUpdateItemAPI, ctx context.Context, s3Event events.S3Event) error {
+	wg := multierror.Group{}
+	for _, record := range s3Event.Records {
+		func(record events.S3EventRecord) {
+			wg.Go(func() (err error) {
+				span, ctx := tracer.StartSpanFromContext(ctx, "handle.record")
+				defer span.Finish(tracer.WithError(err))
+				defer func() {
+					if err != nil {
+						sendMetric("opportunity.failed", 1)
+					}
+				}()
+
+				sourceBucket := record.S3.Bucket.Name
+				sourceKey := record.S3.Object.Key
+				logger := log.With(logger, "event_name", record.EventName,
+					"source_bucket", sourceBucket, "source_object_key", sourceKey)
+
+				resp, err := s3svc.GetObject(ctx, &s3.GetObjectInput{
+					Bucket: aws.String(sourceBucket),
+					Key:    aws.String(sourceKey),
+				})
+				if err != nil {
+					log.Error(logger, "Error getting source S3 object", err)
+					return err
+				}
+
+				data, err := io.ReadAll(resp.Body)
+				if err != nil {
+					log.Error(logger, "Error reading source opportunity from S3", err)
+					return err
+				}
+
+				var opp opportunity
+				if err := xml.Unmarshal(data, &opp); err != nil {
+					log.Error(logger, "Error parsing opportunity from XML", err)
+					return err
+				}
+				return processOpportunity(ctx, dynamodbsvc, opp)
+			})
+		}(record)
+	}
+
+	errs := wg.Wait()
+	if err := errs.ErrorOrNil(); err != nil {
+		log.Warn(logger, "Failures occurred during invocation; check logs for details",
+			"count_errors", errs.Len(),
+			"count_s3_events", len(s3Event.Records))
+		return err
+	}
+	return nil
+}
+
+// processOpportunity takes a single opportunity and uploads an XML representation of the
+// opportunity to its configured DynamoDB table.
+func processOpportunity(ctx context.Context, svc DynamoDBUpdateItemAPI, opp opportunity) error {
+	logger := log.With(logger,
+		"opportunity_id", opp.OpportunityID, "opportunity_number", opp.OpportunityNumber)
+
+	if err := UpdateDynamoDBItem(ctx, svc, env.DestinationTable, opp); err != nil {
+		return log.Errorf(logger, "Error uploading prepared grant opportunity to DynamoDB", err)
+	}
+
+	log.Info(logger, "Successfully uploaded opportunity")
+	sendMetric("opportunity.saved", 1)
+	return nil
+}