feat(appsync): add support for data source integrations (#34248)

### Issue # (if applicable)

Closes #34264

### Reason for this change

This change is to support the release of data source integrations for AWS AppSync Events. The implementation is almost an exact replica of the existing data sources for AppSync GraphQL APIs.

### Description of changes

- New `data-source-common.ts` file which contains the data source implementation. This new file was needed given the previous `data-source.ts` implementation only works for the `IGraphqlApi` Interface. The common implementation makes it to the new generic `IApi` type is used.
- Changes to `eventapi.ts` to support adding data sources to the API.
- Changes to `channel-namespace.ts` to support adding event handler configuration with the data sources configured for the API.
- New unit tests to validate functionality for data sources.
- New integration tests to validate end-to-end functionality across all data source types and the different configurations for the Lambda data source type including direct Lambda invoke and sync/async invoke type.

### Describe any new or updated permissions being added

There are no new permissions being added. The data source implementation was a copy/paste from the GraphQL implementation and the grant methods used in those have not changed in this implementation.

### Description of how you validated changes

- New unit tests to validate functionality for data sources.
- New integration tests to validate end-to-end functionality across all data source types and the different configurations for the Lambda data source type including direct Lambda invoke and sync/async invoke type.
 - `integ.appsync-eventapi-dynamodb.ts`
 - `integ.appsync-eventapi-eventbridge.ts`
 - `integ.appsync-eventapi-http.ts`
 - `integ.appsync-eventapi-lambda-direct-async.ts`
 - `integ.appsync-eventapi-lambda-direct.ts`
 - `integ.appsync-eventapi-lambda.ts`
 - `integ.appsync-eventapi-opensearch.ts`
 - `integ.appsync-eventapi-rds.ts`

### Checklist
- [X] My code adheres to the [CONTRIBUTING GUIDE](https://github.com/aws/aws-cdk/blob/main/CONTRIBUTING.md) and [DESIGN GUIDELINES](https://github.com/aws/aws-cdk/blob/main/docs/DESIGN_GUIDELINES.md)

----

*By submitting this pull request, I confirm that my contribution is made under the terms of the Apache-2.0 license*

Author: kwwendt

packages/@aws-cdk-testing/framework-integ/test/aws-appsync/test/integ-assets/eventapi-grant-assertion/index.js

@@ -225,10 +225,12 @@ async function getPublishAuthHeader(authMode, event={}, authToken='') {
  * @param {string} authMode the authorization mode for the request
  * @param {string} authToken the token used for Lambda auth mode
  * @param {boolean} triggerPub whether to also publish in the method
+ * @param {array} eventPayload the payload to publish
  * @returns {Object}
  */
-async function subscribe(channel, authMode, authToken, triggerPub=false) {
+async function subscribe(channel, authMode, authToken, triggerPub=false, eventPayload=[]) {
   const response = {};
+  const pubMsg = [];
   const authHeader = await getPublishAuthHeader(authMode, {}, authToken);
   const auth = getAuthProtocolForIAM(authHeader);
   const socket = await new Promise((resolve, reject) => {
@@ -255,7 +257,7 @@ async function subscribe(channel, authMode, authToken, triggerPub=false) {
       } else if (payload.type === 'data') {
         console.log('Data received');
         response.pubStatusCode = 200;
-        response.pubMsg = JSON.parse(payload.event).message;
+        pubMsg.push(JSON.parse(payload.event));
       } else if (payload.type === 'subscribe_error') {
         console.log(payload);
         if (payload.errors.some((error) => error.errorType === 'UnauthorizedException')) {
@@ -276,7 +278,7 @@ async function subscribe(channel, authMode, authToken, triggerPub=false) {
     socket.onerror = (event) => console.log(event)
   });
 
-  const subChannel = `/${channel}/*`;
+  const subChannel = `${channel}/*`;
   socket.send(JSON.stringify({
     type: 'subscribe',
     id: crypto.randomUUID(),
@@ -285,35 +287,39 @@ async function subscribe(channel, authMode, authToken, triggerPub=false) {
   }));
 
   if (triggerPub) {
-    await sleep(1000);
-    await publish(channel, authMode, authToken);
+    await sleep(2000);
+    console.log("Hello in here for publishing")
+    await publish(channel, eventPayload, authMode, authToken);
   }
   await sleep(3000);
+
+  if (pubMsg.length > 0)
+    response.pubMsg = pubMsg;
+
   return response;
 }
 
 /**
  * Publishes to a channel and returns the response
  *
  * @param {string} channel the channel to publish to
+ * @param {array} eventPayload the payload to publish
  * @param {string} authMode the auth mode to use for publishing
  * @param {string} authToken the auth token to use for Lambda auth mode
  * @returns {Object}
  */
-async function publish(channel, authMode, authToken) {
+async function publish(channel, eventPayload, authMode, authToken) {
   const event = {
-    'channel': `/${channel}/test`,
-    'events': [
-      JSON.stringify({message:'Hello World!'})
-    ]
-  }
-  
+    'channel': `${channel}/test`,
+    'events': eventPayload.map((_payload) => JSON.stringify(_payload))
+  };
+
   const response = await fetch(`${httpUrl}`, {
     method: 'POST',
     headers: await getPublishAuthHeader(authMode, event, authToken),
     body: JSON.stringify(event)
   });
-  
+
   if (!response.ok) {
     return {
       statusCode: response.status,
@@ -323,7 +329,7 @@ async function publish(channel, authMode, authToken) {
   const output = await response.json();
   return {
     statusCode: 200,
-    msg: output.successful.length == 1 ? 'publish_success' : 'publish_fail',
+    msg: output.successful.length == eventPayload.length ? 'publish_success' : 'publish_fail',
   }
 }
 
@@ -337,6 +343,7 @@ exports.handler = async function(event) {
   const channel = event.channel;
   const authMode = event.authMode;
   const authToken = event.authToken ?? '';
+  const eventPayload = event.eventPayload ?? [{message:'Hello World!'}];
   const isCustomEndpoint = event.customEndpoint ?? false;
 
   // If custom endpoint, wait for 60 seconds for DNS to propagate
@@ -350,13 +357,13 @@ exports.handler = async function(event) {
 
   let res;
   if (pubSubAction === 'publish') {
-    res = await publish(channel, authMode, authToken);
+    res = await publish(channel, eventPayload, authMode, authToken);
     console.log(res);
   } else if (pubSubAction === 'subscribe') {
     res = await subscribe(channel, authMode, authToken, false);
     console.log(res);
   } else if (pubSubAction === 'pubSub') {
-    res = await subscribe(channel, authMode, authToken, true);
+    res = await subscribe(channel, authMode, authToken, true, eventPayload);
     console.log(res);
   }
 

packages/@aws-cdk-testing/framework-integ/test/aws-appsync/test/integ-assets/eventapi-handlers/ddb.js

@@ -0,0 +1,35 @@
+import { util } from '@aws-appsync/utils';
+
+const TABLE = 'event-messages';
+
+export const onPublish = {
+  request: (ctx) => {
+    return {
+      operation: 'BatchPutItem',
+      tables: {
+        [TABLE]: ctx.events.map((event) => util.dynamodb.toMapValues({
+            ...event.payload,
+            ddb: true,
+            eventId: event.id
+        })),
+      },
+    };
+  },
+  response: (ctx) => {
+    const { error, result } = ctx;
+    if (error) {
+        return util.appendError(error.message, error.type, result);
+    }
+    return ctx.result.data[TABLE].map((item) => ({
+        id: item.eventId,
+        payload: {
+            ...item
+        }
+    }));
+  },
+}
+
+export function onSubscribe(ctx) {
+  // Reject a subscription attempt
+  // util.unauthorized();
+}

packages/@aws-cdk-testing/framework-integ/test/aws-appsync/test/integ-assets/eventapi-handlers/eb.js

@@ -0,0 +1,38 @@
+import { util } from '@aws-appsync/utils';
+
+export const onPublish = {
+  request: (ctx) => {
+    const events = ctx.events.map((event) => ({
+      source: 'appsync.eventapi',
+      detail: {
+        id: event.id,
+        ...event.payload,
+      },
+      detailType: 'AppSyncEvent',
+    }));
+  
+    return {
+      operation: "PutEvents",
+      events: events,
+    };
+  },
+  response: (ctx) => {
+    const { error, result } = ctx;
+    if (error) {
+      return util.appendError(error.message, error.type, result);
+    }
+
+    return ctx.events.map((event, index) => ({
+      id: event.id,
+      payload: {
+        ...event.payload,
+        ...ctx.result.Entries[index],
+      }
+    }));
+  },
+}
+
+export function onSubscribe(ctx) {
+  // Reject a subscription attempt
+  // util.unauthorized();
+}

packages/@aws-cdk-testing/framework-integ/test/aws-appsync/test/integ-assets/eventapi-handlers/http.js

@@ -0,0 +1,35 @@
+import { util } from '@aws-appsync/utils';
+
+export const onPublish = {
+  request: (ctx) => {
+    return {
+      "version": "2018-05-29",
+      "method": "GET",
+      "params": {
+        "headers": {
+          "Content-Type": "application/json"
+        }
+      },
+      "resourcePath": `/prod/random`
+    };
+  },
+  response: (ctx) => {
+    const { error, result } = ctx;
+    if (error) {
+      return util.appendError(error.message, error.type, result);
+    }
+    const randomValue = result.body;
+    return ctx.events.map((event) => ({
+      id: event.id,
+      payload: {
+        ...event.payload,
+        random: randomValue
+      }
+    }));
+  },
+}
+
+export function onSubscribe(ctx) {
+  // Reject a subscription attempt
+  // util.unauthorized();
+}

packages/@aws-cdk-testing/framework-integ/test/aws-appsync/test/integ-assets/eventapi-handlers/lambda.js

@@ -0,0 +1,25 @@
+import { util } from '@aws-appsync/utils';
+
+export const onPublish = {
+  request: (ctx) => {
+    return {
+      operation: 'Invoke',
+      payload: {
+        events: ctx.events
+      },
+    };
+  },
+  response: (ctx) => {
+    const { error, result } = ctx;
+    if (error) {
+      return util.appendError(error.message, error.type, result);
+    }
+
+    return result;
+  },
+}
+
+export function onSubscribe(ctx) {
+  // Reject a subscription attempt
+  // util.unauthorized();
+}

packages/@aws-cdk-testing/framework-integ/test/aws-appsync/test/integ-assets/eventapi-handlers/oss.js

@@ -0,0 +1,45 @@
+import { util } from '@aws-appsync/utils';
+
+const INDEX_NAME = 'movies';
+
+export const onPublish = {
+  request: (ctx) => {
+    const docs = ctx.events.map(event => ({
+      _index: INDEX_NAME,
+      _id: event.payload.id
+    }));
+
+    return {
+      operation: "GET",
+      path: "/_mget",
+      params: {
+        body: { docs }
+      }
+    };
+  },
+  response: (ctx) => {
+    const { error, result } = ctx;
+
+    if (error) {
+      return util.appendError(error.message, error.type, result);
+    }
+
+    return result.docs.map((item, idx) => {
+      if (item.found) {
+        return {
+          id: ctx.events[idx].id,
+          payload: {
+            ...item._source
+          }
+        }
+      } else {
+        return ctx.events[idx]
+      }
+    });
+  },
+}
+
+export function onSubscribe(ctx) {
+  // Reject a subscription attempt
+  // util.unauthorized();
+}

packages/@aws-cdk-testing/framework-integ/test/aws-appsync/test/integ-assets/eventapi-handlers/rds.js

@@ -0,0 +1,36 @@
+import { toJsonObject, createPgStatement as pg } from '@aws-appsync/utils/rds';
+import { util } from '@aws-appsync/utils';
+
+const TABLE_NAME = 'events';
+
+export const onPublish = {
+  request: (ctx) => {
+    const values = ctx.events.map((event) => {
+      return `('${event.id}', ${Object.values(event.payload).map(val => typeof val === 'string' ? `'${val}'` : val).join(',')}, 'rds')`
+    }).join(', ');
+
+    const statement = `INSERT INTO ${TABLE_NAME} (event_id, message, ds_type) VALUES ${values} RETURNING *`;
+    return pg(statement);
+  },
+  response: (ctx) => {
+    const { error, result } = ctx;
+    console.log(ctx);
+    if (error) {
+      return util.appendError(error.message, error.type, result);
+    }
+
+    const parsedRes = toJsonObject(result)[0];
+    return parsedRes.map((res) => ({
+      id: res.event_id,
+      payload: {
+        message: res.message,
+        ds_type: res.ds_type
+      }
+    }));
+  },
+}
+
+export function onSubscribe(ctx) {
+  // Reject a subscription attempt
+  // util.unauthorized();
+}