feat(pipes-enrichments): support API destination enrichment

packages/@aws-cdk/aws-pipes-enrichments-alpha/README.md

@@ -69,3 +69,22 @@ const pipe = new pipes.Pipe(this, 'Pipe', {
   target: new SomeTarget(targetQueue),
 });
 ```
+
+### API destination
+
+API destination can be used to enrich events of a pipe.
+
+```ts
+declare const sourceQueue: sqs.Queue;
+declare const targetQueue: sqs.Queue;
+
+declare const apiDestination: events.ApiDestination;
+
+const enrichment = new enrichments.ApiDestinationEnrichment(apiDestination);
+
+const pipe = new pipes.Pipe(this, 'Pipe', {
+  source: new SomeSource(sourceQueue),
+  enrichment,
+  target: new SomeTarget(targetQueue),
+});
+```

packages/@aws-cdk/aws-pipes-enrichments-alpha/lib/api-destination.ts

@@ -0,0 +1,86 @@
+import { EnrichmentParametersConfig, IEnrichment, IPipe, InputTransformation } from '@aws-cdk/aws-pipes-alpha';
+import { IApiDestination } from 'aws-cdk-lib/aws-events';
+import { IRole, PolicyStatement } from 'aws-cdk-lib/aws-iam';
+import { CfnPipe } from 'aws-cdk-lib/aws-pipes';
+
+/**
+ * Properties for a ApiDestinationEnrichment
+ */
+export interface ApiDestinationEnrichmentProps {
+  /**
+   * The input transformation for the enrichment
+   * @see https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-pipes-input-transformation.html
+   * @default - None
+   */
+  readonly inputTransformation?: InputTransformation;
+
+  /**
+   * The headers that need to be sent as part of request invoking the EventBridge ApiDestination.
+   *
+   * @default - none
+   */
+  readonly headerParameters?: Record<string, string>;
+
+  /**
+   * The path parameter values used to populate the EventBridge API destination path wildcards ("*").
+   *
+   * @default - none
+   */
+  readonly pathParameterValues?: string[];
+
+  /**
+   * The query string keys/values that need to be sent as part of request invoking the EventBridge API destination.
+   *
+   * @default - none
+   */
+  readonly queryStringParameters?: Record<string, string>;
+}
+
+/**
+ * An API Destination enrichment for a pipe
+ */
+export class ApiDestinationEnrichment implements IEnrichment {
+  public readonly enrichmentArn: string;
+
+  private readonly inputTransformation?: InputTransformation;
+  private readonly headerParameters?: Record<string, string>;
+  private readonly pathParameterValues?: string[];
+  private readonly queryStringParameters?: Record<string, string>;
+
+  constructor(private readonly destination: IApiDestination, props?: ApiDestinationEnrichmentProps) {
+    this.enrichmentArn = destination.apiDestinationArn;
+    this.inputTransformation = props?.inputTransformation;
+    this.headerParameters = props?.headerParameters;
+    this.queryStringParameters = props?.queryStringParameters;
+    this.pathParameterValues = props?.pathParameterValues;
+  }
+
+  bind(pipe: IPipe): EnrichmentParametersConfig {
+
+    const httpParameters: CfnPipe.PipeEnrichmentHttpParametersProperty | undefined =
+      this.headerParameters ??
+        this.pathParameterValues ??
+        this.queryStringParameters
+        ? {
+          headerParameters: this.headerParameters,
+          pathParameterValues: this.pathParameterValues,
+          queryStringParameters: this.queryStringParameters,
+        }
+        : undefined;
+
+    return {
+      enrichmentParameters: {
+        inputTemplate: this.inputTransformation?.bind(pipe).inputTemplate,
+        httpParameters,
+      },
+    };
+  }
+
+  grantInvoke(pipeRole: IRole): void {
+    pipeRole.addToPrincipalPolicy(new PolicyStatement({
+      resources: [this.destination.apiDestinationArn],
+      actions: ['events:InvokeApiDestination'],
+    }));
+  }
+}
+

packages/@aws-cdk/aws-pipes-enrichments-alpha/lib/index.ts

@@ -1,2 +1,3 @@
+export * from './api-destination';
 export * from './lambda';
 export * from './stepfunctions';

packages/@aws-cdk/aws-pipes-enrichments-alpha/rosetta/default.ts-fixture

@@ -1,5 +1,6 @@
 // Fixture with packages imported, but nothing else
 import * as cdk from 'aws-cdk-lib';
+import * as events from 'aws-cdk-lib/aws-events';
 import * as sqs from 'aws-cdk-lib/aws-sqs';
 import * as lambda from 'aws-cdk-lib/aws-lambda';
 import * as stepfunctions from 'aws-cdk-lib/aws-stepfunctions';

packages/@aws-cdk/aws-pipes-enrichments-alpha/test/__snapshots__/api-destination.test.ts.snap

@@ -0,0 +1,54 @@
+// Jest Snapshot v1, https://goo.gl/fbAQLP
+
+exports[`api-destination should grant pipe role invoke access 1`] = `
+{
+  "MyPipeRoleCBC8E9AB": {
+    "Properties": {
+      "AssumeRolePolicyDocument": {
+        "Statement": [
+          {
+            "Action": "sts:AssumeRole",
+            "Effect": "Allow",
+            "Principal": {
+              "Service": "pipes.amazonaws.com",
+            },
+          },
+        ],
+        "Version": "2012-10-17",
+      },
+    },
+    "Type": "AWS::IAM::Role",
+  },
+}
+`;
+
+exports[`api-destination should grant pipe role invoke access 2`] = `
+{
+  "MyPipeRoleDefaultPolicy31387C20": {
+    "Properties": {
+      "PolicyDocument": {
+        "Statement": [
+          {
+            "Action": "events:InvokeApiDestination",
+            "Effect": "Allow",
+            "Resource": {
+              "Fn::GetAtt": [
+                "ApiDestination3AB57A39",
+                "Arn",
+              ],
+            },
+          },
+        ],
+        "Version": "2012-10-17",
+      },
+      "PolicyName": "MyPipeRoleDefaultPolicy31387C20",
+      "Roles": [
+        {
+          "Ref": "MyPipeRoleCBC8E9AB",
+        },
+      ],
+    },
+    "Type": "AWS::IAM::Policy",
+  },
+}
+`;

packages/@aws-cdk/aws-pipes-enrichments-alpha/test/api-destination.test.ts

@@ -0,0 +1,124 @@
+import { DynamicInput, InputTransformation, Pipe } from '@aws-cdk/aws-pipes-alpha';
+import { App, Stack, SecretValue } from 'aws-cdk-lib';
+import { Template } from 'aws-cdk-lib/assertions';
+import * as events from 'aws-cdk-lib/aws-events';
+import { Secret } from 'aws-cdk-lib/aws-secretsmanager';
+import { TestSource, TestTarget } from './test-classes';
+import { ApiDestinationEnrichment } from '../lib';
+
+describe('api-destination', () => {
+  let app: App;
+  let stack: Stack;
+  let secret: Secret;
+  let connection: events.Connection;
+  let apiDestination: events.ApiDestination;
+
+  beforeEach(() => {
+    app = new App();
+    stack = new Stack(app, 'TestStack');
+    secret = new Secret(stack, 'MySecret', {
+      secretStringValue: SecretValue.unsafePlainText('abc123'),
+    });
+    connection = new events.Connection(stack, 'MyConnection', {
+      authorization: events.Authorization.apiKey('x-api-key', secret.secretValue),
+      description: 'Connection with API Key x-api-key',
+      connectionName: 'MyConnection',
+    });
+
+    apiDestination = new events.ApiDestination(stack, 'ApiDestination', {
+      apiDestinationName: 'ApiDestination',
+      connection,
+      description: 'ApiDestination',
+      httpMethod: events.HttpMethod.GET,
+      endpoint: 'someendpoint',
+      rateLimitPerSecond: 60,
+    });
+  });
+
+  it('should have only enrichment arn', () => {
+    // ARRANGE
+    const enrichment = new ApiDestinationEnrichment(apiDestination);
+
+    new Pipe(stack, 'MyPipe', {
+      source: new TestSource(),
+      enrichment,
+      target: new TestTarget(),
+    });
+
+    // ACT
+    const template = Template.fromStack(stack);
+
+    // ASSERT
+    template.hasResourceProperties('AWS::Pipes::Pipe', {
+      Enrichment: {
+        'Fn::GetAtt': [
+          'ApiDestination3AB57A39',
+          'Arn',
+        ],
+      },
+      EnrichmentParameters: {},
+    });
+  });
+
+  it('should have enrichment parameters', () => {
+    // ARRANGE
+    const enrichment = new ApiDestinationEnrichment(apiDestination, {
+      inputTransformation: InputTransformation.fromObject({
+        body: DynamicInput.fromEventPath('$.body'),
+      }),
+      headerParameters: {
+        headerParam: 'headerParam',
+      },
+      pathParameterValues: ['pathParam'],
+      queryStringParameters: {
+        param: 'queryParam',
+      },
+    });
+
+    new Pipe(stack, 'MyPipe', {
+      source: new TestSource(),
+      enrichment,
+      target: new TestTarget(),
+    });
+
+    // ACT
+    const template = Template.fromStack(stack);
+
+    // ASSERT
+    template.hasResourceProperties('AWS::Pipes::Pipe', {
+      EnrichmentParameters: {
+        InputTemplate: '{"body":<$.body>}',
+        HttpParameters: {
+          HeaderParameters: {
+            headerParam: 'headerParam',
+          },
+          PathParameterValues: [
+            'pathParam',
+          ],
+          QueryStringParameters: {
+            param: 'queryParam',
+          },
+        },
+      },
+    });
+  });
+
+  it('should grant pipe role invoke access', () => {
+    // ARRANGE
+    const enrichment = new ApiDestinationEnrichment(apiDestination);
+
+    new Pipe(stack, 'MyPipe', {
+      source: new TestSource(),
+      enrichment,
+      target: new TestTarget(),
+    });
+
+    // ACT
+    const template = Template.fromStack(stack);
+
+    // ASSERT
+    expect(template.findResources('AWS::IAM::Role')).toMatchSnapshot();
+    expect(template.findResources('AWS::IAM::Policy')).toMatchSnapshot();
+  });
+});
+