feat(pipes-targets): add API destination (#30756)

Add EventBridge API destination as a Pipes target.

CloudFormation groups EventBridge API destination with API Gateway REST API
as [PipeTargetHttpParameters](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-pipes-pipe-pipetargethttpparameters.html#cfn-pipes-pipe-pipetargethttpparameters-pathparametervalues), but I think separating them here similar to [aws-event-targets](https://github.com/aws/aws-cdk/tree/main/packages/aws-cdk-lib/aws-events-targets/lib) 
makes more sense, as API Gateway requires `stage`, `path`, and `method` (see [here](https://github.com/aws/aws-cdk/blob/main/packages/aws-cdk-lib/aws-events-targets/lib/api-gateway.ts#L11-L32)).

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

@@ -32,6 +32,7 @@ The following targets are supported:
 1. `targets.SqsTarget`: [Send event source to a Queue](#amazon-sqs)
 2. `targets.SfnStateMachine`: [Invoke a State Machine from an event source](#aws-step-functions-state-machine)
 3. `targets.LambdaFunction`: [Send event source to a Lambda Function](#aws-lambda-function)
+4. `targets.ApiDestinationTarget`: [Send event source to an EventBridge API Destination](#amazon-eventbridge-api-destination)
 
 ### Amazon SQS
 
@@ -171,3 +172,36 @@ const pipe = new pipes.Pipe(this, 'Pipe', {
     target: pipeTarget
 });
 ```
+
+### Amazon EventBridge API Destination
+
+An EventBridge API destination can be used as a target for a pipe. 
+The API destination will receive the (enriched/filtered) source payload.
+
+```ts
+declare const sourceQueue: sqs.Queue;
+declare const dest: events.ApiDestination;
+
+const apiTarget = new targets.ApiDestinationTarget(dest);
+
+const pipe = new pipes.Pipe(this, 'Pipe', {
+    source: new SqsSource(sourceQueue),
+    target: apiTarget,
+});
+```
+
+The input to the target API destination can be transformed:
+
+```ts
+declare const sourceQueue: sqs.Queue;
+declare const dest: events.ApiDestination;
+
+const apiTarget = new targets.ApiDestinationTarget(dest, {
+  inputTransformation: pipes.InputTransformation.fromObject({ body: "👀" }),
+});
+
+const pipe = new pipes.Pipe(this, 'Pipe', {
+    source: new SqsSource(sourceQueue),
+    target: apiTarget,
+});
+```

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

@@ -0,0 +1,80 @@
+import { IInputTransformation, IPipe, ITarget, TargetConfig } from '@aws-cdk/aws-pipes-alpha';
+import { IApiDestination } from 'aws-cdk-lib/aws-events';
+import { IRole, PolicyStatement } from 'aws-cdk-lib/aws-iam';
+
+/**
+ * EventBridge API destination target properties.
+ */
+export interface ApiDestinationTargetParameters {
+  /**
+   * The input transformation to apply to the message before sending it to the target.
+   *
+   * @see https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-pipes-pipe-pipetargetparameters.html#cfn-pipes-pipe-pipetargetparameters-inputtemplate
+   * @default - none
+   */
+  readonly inputTransformation?: IInputTransformation;
+
+  /**
+   * The headers to send as part of the request invoking the EventBridge API destination.
+   *
+   * The headers are merged with the headers from the API destination.
+   * If there are conflicts, the headers from the API destination take precedence.
+   *
+   * @see https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-pipes-pipe-pipetargethttpparameters.html#cfn-pipes-pipe-pipetargethttpparameters-headerparameters
+   * @default - none
+   */
+  readonly headerParameters?: { [key: string]: string };
+
+  /**
+   * The path parameter values used to populate the EventBridge API destination path wildcards ("*").
+   *
+   * @see https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-pipes-pipe-pipetargethttpparameters.html#cfn-pipes-pipe-pipetargethttpparameters-pathparametervalues
+   * @default - none
+   */
+  readonly pathParameterValues?: string[];
+
+  /**
+   * The query string keys/values that need to be sent as part of request invoking the EventBridge API destination.
+   *
+   * @see https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-pipes-pipe-pipetargethttpparameters.html#cfn-pipes-pipe-pipetargethttpparameters-querystringparameters
+   * @default - none
+   */
+  readonly queryStringParameters?: Record<string, string>;
+}
+
+/**
+ * A EventBridge Pipes target that sends messages to an EventBridge API destination.
+ */
+export class ApiDestinationTarget implements ITarget {
+  private destination: IApiDestination;
+  private apiParameters?: ApiDestinationTargetParameters;
+  public readonly targetArn: string;
+
+  constructor(destination: IApiDestination, parameters?: ApiDestinationTargetParameters) {
+    this.destination = destination;
+    this.apiParameters = parameters;
+    this.targetArn = destination.apiDestinationArn;
+  }
+
+  grantPush(grantee: IRole): void {
+    grantee.addToPrincipalPolicy(new PolicyStatement({
+      resources: [this.destination.apiDestinationArn],
+      actions: ['events:InvokeApiDestination'],
+    }));
+  }
+
+  bind(pipe: IPipe): TargetConfig {
+    if (!this.apiParameters) {
+      return {
+        targetParameters: {},
+      };
+    }
+
+    return {
+      targetParameters: {
+        inputTemplate: this.apiParameters.inputTransformation?.bind(pipe).inputTemplate,
+        httpParameters: this.apiParameters,
+      },
+    };
+  }
+}

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

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

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

@@ -1,10 +1,12 @@
 // 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 sfn from 'aws-cdk-lib/aws-stepfunctions';
 import * as lambda from 'aws-cdk-lib/aws-lambda';
 import { Construct } from 'constructs';
 import * as pipes from '@aws-cdk/aws-pipes-alpha';
+import { SqsSource } from '@aws-cdk/aws-pipes-sources-alpha';
 import * as targets from '@aws-cdk/aws-pipes-targets-alpha';
 
 class SomeSource implements pipes.ISource {

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

@@ -0,0 +1,172 @@
+import { 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 } from './test-classes';
+import { ApiDestinationTarget } from '../lib';
+
+describe('API destination', () => {
+  let app: App;
+  let stack: Stack;
+  let secret: Secret;
+  let connection: events.Connection;
+
+  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',
+    });
+  });
+
+  it('should have only target arn', () => {
+    // ARRANGE
+    const destination = new events.ApiDestination(stack, 'MyApiDestination', {
+      connection,
+      endpoint: 'https://httpbin.org/headers',
+      httpMethod: events.HttpMethod.GET,
+      apiDestinationName: 'MyDestination',
+      rateLimitPerSecond: 1,
+      description: 'Calling example.com with API key x-api-key',
+    });
+    const target = new ApiDestinationTarget(destination);
+
+    new Pipe(stack, 'MyPipe', {
+      source: new TestSource(),
+      target,
+    });
+
+    // ACT
+    const template = Template.fromStack(stack);
+
+    // ASSERT
+    template.hasResourceProperties('AWS::Pipes::Pipe', {
+      Target: {
+        'Fn::GetAtt': [
+          'MyApiDestination07E6A8F9',
+          'Arn',
+        ],
+      },
+      TargetParameters: {},
+    });
+  });
+
+  it('should have target parameters', () => {
+    // ARRANGE
+    const destination = new events.ApiDestination(stack, 'MyApiDestination', {
+      connection,
+      endpoint: 'https://httpbin.org/headers',
+      httpMethod: events.HttpMethod.GET,
+      apiDestinationName: 'MyDestination',
+      rateLimitPerSecond: 1,
+      description: 'Calling example.com with API key x-api-key',
+    });
+    const target = new ApiDestinationTarget(destination, {
+      headerParameters: { headerName: 'headerValue' },
+      pathParameterValues: ['pathValue'],
+      queryStringParameters: { queryName: 'queryValue' },
+    });
+
+    new Pipe(stack, 'MyPipe', {
+      source: new TestSource(),
+      target,
+    });
+
+    // ACT
+    const template = Template.fromStack(stack);
+
+    // ASSERT
+    template.hasResourceProperties('AWS::Pipes::Pipe', {
+      TargetParameters: {
+        HttpParameters: {
+          HeaderParameters: {
+            headerName: 'headerValue',
+          },
+          PathParameterValues: ['pathValue'],
+          QueryStringParameters: {
+            queryName: 'queryValue',
+          },
+        },
+      },
+    });
+  });
+
+  it('should have input transformation', () => {
+    // ARRANGE
+    const destination = new events.ApiDestination(stack, 'MyApiDestination', {
+      connection,
+      endpoint: 'https://httpbin.org/headers',
+      httpMethod: events.HttpMethod.GET,
+      apiDestinationName: 'MyDestination',
+      rateLimitPerSecond: 1,
+      description: 'Calling example.com with API key x-api-key',
+    });
+
+    const inputTransformation = InputTransformation.fromObject({
+      key: 'value',
+    });
+
+    const target = new ApiDestinationTarget(destination, {
+      inputTransformation,
+    });
+
+    new Pipe(stack, 'MyPipe', {
+      source: new TestSource(),
+      target,
+    });
+
+    // ACT
+    const template = Template.fromStack(stack);
+
+    // ASSERT
+    template.hasResourceProperties('AWS::Pipes::Pipe', {
+      TargetParameters: {
+        InputTemplate: '{"key":"value"}',
+      },
+    });
+  });
+
+  it('should grant pipe role push access', () => {
+    // ARRANGE
+    const destination = new events.ApiDestination(stack, 'MyApiDestination', {
+      connection,
+      endpoint: 'https://httpbin.org/headers',
+      httpMethod: events.HttpMethod.GET,
+      apiDestinationName: 'MyDestination',
+      rateLimitPerSecond: 1,
+      description: 'Calling example.com with API key x-api-key',
+    });
+    const target = new ApiDestinationTarget(destination);
+
+    new Pipe(stack, 'MyPipe', {
+      source: new TestSource(),
+      target,
+    });
+
+    // ACT
+    const template = Template.fromStack(stack);
+
+    // ASSERT
+    template.hasResource('AWS::IAM::Policy', {
+      Properties: {
+        Roles: [{
+          Ref: 'MyPipeRoleCBC8E9AB',
+        }],
+        PolicyDocument: {
+          Statement: [{
+            Action: 'events:InvokeApiDestination',
+            Resource: {
+              'Fn::GetAtt': ['MyApiDestination07E6A8F9', 'Arn'],
+            },
+          }],
+        },
+      },
+    });
+  });
+});