Add event filtering patterns to stream event type

docs/providers/aws/events/streams.md

@@ -329,3 +329,22 @@ functions:
           arn: arn:aws:dynamodb:region:XXXXXX:table/foo/stream/1970-01-01T00:00:00.000
           tumblingWindowInSeconds: 30
 ```
+
+## Setting FilterCriteria
+
+This configuration allows customers to filter event before lambda invocation. It accepts up to 5 filter criterion. If one event matches at least 1 pattern, lambda will process it.
+
+For more information and examples, read the [AWS event filtering documentation](https://docs.aws.amazon.com/lambda/latest/dg/invocation-eventfiltering.html)
+
+Note: Serverless only sets this property if you explicitly add it to the stream configuration (see example below). The following exemple will only process inserted item in the DynamoDB table (not the removed and modified items).
+
+```yml
+functions:
+  handleInsertedDynamoDBItem:
+    handler: handler.preprocess
+    events:
+      - stream:
+          arn: arn:aws:dynamodb:region:XXXXXX:table/foo/stream/1970-01-01T00:00:00.000
+          patterns:
+            - eventName: [INSERT]
+```

docs/providers/aws/guide/serverless.yml.md

@@ -458,6 +458,8 @@ functions:
           startingPosition: LATEST
           enabled: true
           functionResponseType: ReportBatchItemFailures
+          patterns:
+            - partitionKey: [ 1 ]
       - msk:
           arn: arn:aws:kafka:us-east-1:111111111111:cluster/ClusterName/a1a1a1a1a1a1a1a1a # ARN of MSK Cluster
           topic: kafkaTopic # name of Kafka topic to consume from

lib/plugins/aws/package/compile/events/stream.js

@@ -69,6 +69,12 @@ class AwsCompileStreamEvents {
               required: ['onFailure'],
             },
             tumblingWindowInSeconds: { type: 'integer', minimum: 0, maximum: 900 },
+            patterns: {
+              type: 'array',
+              minItems: 1,
+              maxItems: 10,
+              items: { type: 'object' },
+            },
           },
           additionalProperties: false,
           anyOf: [
@@ -273,6 +279,14 @@ class AwsCompileStreamEvents {
               };
             }
 
+            if (event.stream.patterns) {
+              streamResource.Properties.FilterCriteria = {
+                Filters: event.stream.patterns.map((pattern) => ({
+                  Pattern: JSON.stringify(pattern),
+                })),
+              };
+            }
+
             const newStreamObject = {
               [streamLogicalId]: streamResource,
             };

test/unit/lib/plugins/aws/package/compile/events/stream.test.js

@@ -1724,4 +1724,43 @@ describe('AwsCompileStreamEvents #2', () => {
       );
     });
   });
+  describe('with patterns', () => {
+    let eventSourceMappingResource;
+
+    before(async () => {
+      const { awsNaming, cfTemplate } = await runServerless({
+        fixture: 'function',
+        configExt: {
+          functions: {
+            basic: {
+              events: [
+                {
+                  stream: {
+                    arn: 'arn:aws:dynamodb:region:account:table/foo/stream/1',
+                    patterns: [{ eventName: ['INSERT'] }, { eventName: ['MODIFY'] }],
+                  },
+                },
+              ],
+            },
+          },
+        },
+        command: 'package',
+      });
+      const streamLogicalId = awsNaming.getStreamLogicalId('basic', 'dynamodb', 'foo');
+      eventSourceMappingResource = cfTemplate.Resources[streamLogicalId];
+    });
+
+    it('should wrap patterns within FilterCriteria property', () => {
+      expect(eventSourceMappingResource.Properties.FilterCriteria).to.deep.equal({
+        Filters: [
+          {
+            Pattern: JSON.stringify({ eventName: ['INSERT'] }),
+          },
+          {
+            Pattern: JSON.stringify({ eventName: ['MODIFY'] }),
+          },
+        ],
+      });
+    });
+  });
 });