feat: updated pipes implementation

Author: sliedig

Unicorn.Contracts/template.yaml

@@ -15,7 +15,7 @@ Metadata:
         - ES6000 # Rule disabled because SQS DLQs don't need a RedrivePolicy
         - WS2001 # Rule disabled because check does not support !ToJsonString transform
         - ES1001 # Rule disabled because our Lambda functions don't need DestinationConfig.OnFailure
-        - W3002 # Rule disabled as nested templates are being packaged 
+        - W3002 # Rule disabled as nested templates are being packaged
 
 Parameters:
   Stage:
@@ -360,29 +360,37 @@ Resources:
       SourceParameters:
         DynamoDBStreamParameters:
           MaximumRetryAttempts: 3
+          BatchSize: 1
+          StartingPosition: LATEST
           DeadLetterConfig:
             Arn: !GetAtt ContractsTableStreamToEventPipeDLQ.Arn
-          StartingPosition: LATEST
-          OnPartialBatchItemFailure: AUTOMATIC_BISECT
-          BatchSize: 1
         FilterCriteria:
           Filters:
-            - Pattern: !ToJsonString
-                eventName: [INSERT, MODIFY]
-                dynamodb:
-                  NewImage:
-                    ContractStatus:
-                      S: [DRAFT, APPROVED]
-      Target: !GetAtt UnicornContractsEventBus.Arn
+            - Pattern: >-
+                {"eventName":["INSERT","MODIFY"],"dynamodb":{"NewImage":{"ContractStatus":{"S":["DRAFT","APPROVED"]}}}}
+      LogConfiguration:
+        CloudwatchLogsLogDestination:
+          LogGroupArn: !GetAtt ContractsTableStreamToEventPipeLogGroup.Arn
+        Level: ERROR
       TargetParameters:
         EventBridgeEventBusParameters:
-          Source: "{{resolve:ssm:/uni-prop/UnicornContractsNamespace}}"
           DetailType: ContractStatusChanged
-        InputTemplate: !ToJsonString
-          PropertyId: "<$.dynamodb.NewImage.PropertyId.S>"
-          ContractId: "<$.dynamodb.NewImage.ContractId.S>"
-          ContractStatus: "<$.dynamodb.NewImage.ContractStatus.S>"
-          ContractLastModifiedOn: "<$.dynamodb.NewImage.ContractLastModifiedOn.S>"
+          Source: "{{resolve:ssm:/uni-prop/UnicornContractsNamespace}}"
+        InputTemplate: |-
+          {
+            "PropertyId": "<$.dynamodb.Keys.PropertyId.S>",
+            "ContractId": "<$.dynamodb.NewImage.ContractId.S>",
+            "ContractStatus": "<$.dynamodb.NewImage.ContractStatus.S>",
+            "ContractLastModifiedOn": "<$.dynamodb.NewImage.ContractLastModifiedOn.S>"
+          }
+      Target: !GetAtt UnicornContractsEventBus.Arn
+
+  ContractsTableStreamToEventPipeLogGroup:
+    Type: AWS::Logs::LogGroup
+    UpdateReplacePolicy: Delete
+    DeletionPolicy: Delete
+    Properties:
+      RetentionInDays: !FindInMap [LogsRetentionPeriodMap, !Ref Stage, Days]
 
   # IAM Role for Event Bridge Pipe
   ContractsTableStreamToEventPipeRole:
@@ -394,6 +402,9 @@ Resources:
           Effect: Allow
           Principal:
             Service: pipes.amazonaws.com
+          Condition:
+            StringEquals:
+              aws:SourceAccount: !Ref AWS::AccountId
       Policies:
         - PolicyName: ContractsTableStreamToEventPipePolicy
           PolicyDocument:
@@ -408,6 +419,7 @@ Resources:
                   - dynamodb:DescribeStream
                   - dynamodb:GetRecords
                   - dynamodb:GetShardIterator
+                  - dynamodb:ListStreams
                 Resource: !GetAtt ContractsTable.StreamArn
               - Effect: Allow
                 Action: