feat: updated Pipes implementation

Author: sliedig

unicorn_contracts/template.yaml

@@ -359,29 +359,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:
-                    contract_status:
-                      S: [DRAFT, APPROVED]
-      Target: !GetAtt UnicornContractsEventBus.Arn
+            - Pattern: >-
+                {"eventName":["INSERT","MODIFY"],"dynamodb":{"NewImage":{"contract_status":{"S":["DRAFT","APPROVED"]}}}}
+      LogConfiguration:
+        CloudwatchLogsLogDestination:
+          LogGroupArn: !GetAtt ContractsTableStreamToEventPipeLogGroup.Arn
+        Level: ERROR
       TargetParameters:
         EventBridgeEventBusParameters:
-          Source: "{{resolve:ssm:/uni-prop/UnicornContractsNamespace}}"
           DetailType: ContractStatusChanged
-        InputTemplate: !ToJsonString
-          property_id: "<$.dynamodb.NewImage.property_id.S>"
-          contract_id: "<$.dynamodb.NewImage.contract_id.S>"
-          contract_status: "<$.dynamodb.NewImage.contract_status.S>"
-          contract_last_modified_on: "<$.dynamodb.NewImage.contract_last_modified_on.S>"
+          Source: "{{resolve:ssm:/uni-prop/UnicornContractsNamespace}}"
+        InputTemplate: |-
+          {
+            "property_id": "<$.dynamodb.Keys.property_id.S>",
+            "contract_id": "<$.dynamodb.NewImage.contract_id.S>",
+            "contract_status": "<$.dynamodb.NewImage.contract_status.S>",
+            "contract_last_modified_on": "<$.dynamodb.NewImage.contract_last_modified_on.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:
@@ -393,6 +401,9 @@ Resources:
           Effect: Allow
           Principal:
             Service: pipes.amazonaws.com
+          Condition:
+            StringEquals:
+              aws:SourceAccount: !Ref AWS::AccountId
       Policies:
         - PolicyName: ContractsTableStreamToEventPipePolicy
           PolicyDocument:
@@ -407,6 +418,7 @@ Resources:
                   - dynamodb:DescribeStream
                   - dynamodb:GetRecords
                   - dynamodb:GetShardIterator
+                  - dynamodb:ListStreams
                 Resource: !GetAtt ContractsTable.StreamArn
               - Effect: Allow
                 Action: