fix: Adding in dynamodb stream configuration (#174)

kcwinner · Mar 28, 2021 · 08594aa · 08594aa
1 parent 6b53b69
commit 08594aa
Show file tree

Hide file tree

Showing 4 changed files with 208 additions and 15 deletions.
diff --git a/API.md b/API.md
@@ -19,6 +19,7 @@ Name|Description
 [CdkTransformerTable](#cdk-appsync-transformer-cdktransformertable)|*No description*
 [CdkTransformerTableKey](#cdk-appsync-transformer-cdktransformertablekey)|*No description*
 [CdkTransformerTableTtl](#cdk-appsync-transformer-cdktransformertablettl)|*No description*
+[DynamoDBStreamProps](#cdk-appsync-transformer-dynamodbstreamprops)|*No description*
 [SchemaTransformerOutputs](#cdk-appsync-transformer-schematransformeroutputs)|*No description*
 
 
@@ -45,6 +46,7 @@ new AppSyncTransformer(scope: Construct, id: string, props: AppSyncTransformerPr
   * **schemaPath** (<code>string</code>)  Relative path where schema.graphql exists. 
   * **apiName** (<code>string</code>)  String value representing the api name. __*Default*__: `${id}-api`
   * **authorizationConfig** (<code>[AuthorizationConfig](#aws-cdk-aws-appsync-authorizationconfig)</code>)  Optional. __*Default*__: API_KEY authorization config
+  * **dynamoDbStreamConfig** (<code>Map<string, [StreamViewType](#aws-cdk-aws-dynamodb-streamviewtype)></code>)  A map of @model type names to stream view type e.g { Blog: StreamViewType.NEW_IMAGE }. __*Optional*__
   * **enableDynamoPointInTimeRecovery** (<code>boolean</code>)  Whether to enable dynamo Point In Time Recovery. __*Default*__: false
   * **fieldLogLevel** (<code>[FieldLogLevel](#aws-cdk-aws-appsync-fieldloglevel)</code>)  Optional. __*Default*__: FieldLogLevel.NONE
   * **postCdkTransformers** (<code>Array<any></code>)  Optional. __*Default*__: undefined
@@ -65,11 +67,27 @@ Name | Type | Description
 **nestedAppsyncStack**🔹 | <code>[NestedStack](#aws-cdk-core-nestedstack)</code> | The NestedStack that contains the AppSync resources.
 **outputs**🔹 | <code>[SchemaTransformerOutputs](#cdk-appsync-transformer-schematransformeroutputs)</code> | The outputs from the SchemaTransformer.
 **resolvers**🔹 | <code>any</code> | The AppSync resolvers from the transformer minus any function resolvers.
+**tableMap**🔹 | <code>Map<string, [Table](#aws-cdk-aws-dynamodb-table)></code> | Map of cdk table keys to L2 Table e.g. { 'TaskTable': Table }.
 **tableNameMap**🔹 | <code>Map<string, any></code> | Map of cdk table tokens to table names.
 
 ### Methods
 
 
+#### addDynamoDBStream(props)🔹 <a id="cdk-appsync-transformer-appsynctransformer-adddynamodbstream"></a>
+
+Adds a stream to the dynamodb table associated with the type.
+
+```ts
+addDynamoDBStream(props: DynamoDBStreamProps): string
+```
+
+* **props** (<code>[DynamoDBStreamProps](#cdk-appsync-transformer-dynamodbstreamprops)</code>)  *No description*
+  * **modelTypeName** (<code>string</code>)  The @model type name from the graph schema e.g. Blog. 
+  * **streamViewType** (<code>[StreamViewType](#aws-cdk-aws-dynamodb-streamviewtype)</code>)  *No description* 
+
+__Returns__:
+* <code>string</code>
+
 #### addLambdaDataSourceAndResolvers(functionName, id, lambdaFunction, options?)🔹 <a id="cdk-appsync-transformer-appsynctransformer-addlambdadatasourceandresolvers"></a>
 
 Adds the function as a lambdaDataSource to the AppSync api Adds all of the functions resolvers to the AppSync api.
@@ -102,6 +120,7 @@ Name | Type | Description
 **schemaPath**🔹 | <code>string</code> | Relative path where schema.graphql exists.
 **apiName**?🔹 | <code>string</code> | String value representing the api name.<br/>__*Default*__: `${id}-api`
 **authorizationConfig**?🔹 | <code>[AuthorizationConfig](#aws-cdk-aws-appsync-authorizationconfig)</code> | Optional.<br/>__*Default*__: API_KEY authorization config
+**dynamoDbStreamConfig**?🔹 | <code>Map<string, [StreamViewType](#aws-cdk-aws-dynamodb-streamviewtype)></code> | A map of @model type names to stream view type e.g { Blog: StreamViewType.NEW_IMAGE }.<br/>__*Optional*__
 **enableDynamoPointInTimeRecovery**?🔹 | <code>boolean</code> | Whether to enable dynamo Point In Time Recovery.<br/>__*Default*__: false
 **fieldLogLevel**?🔹 | <code>[FieldLogLevel](#aws-cdk-aws-appsync-fieldloglevel)</code> | Optional.<br/>__*Default*__: FieldLogLevel.NONE
 **postCdkTransformers**?🔹 | <code>Array<any></code> | Optional.<br/>__*Default*__: undefined
@@ -221,6 +240,20 @@ Name | Type | Description
 
 
 
+## struct DynamoDBStreamProps 🔹 <a id="cdk-appsync-transformer-dynamodbstreamprops"></a>
+
+
+
+
+
+
+Name | Type | Description 
+-----|------|-------------
+**modelTypeName**🔹 | <code>string</code> | The @model type name from the graph schema e.g. Blog.
+**streamViewType**🔹 | <code>[StreamViewType](#aws-cdk-aws-dynamodb-streamviewtype)</code> | <span></span>
+
+
+
 ## struct SchemaTransformerOutputs 🔹 <a id="cdk-appsync-transformer-schematransformeroutputs"></a>
 
 

diff --git a/README.md b/README.md
@@ -232,6 +232,38 @@ Often you will need to access your table names in a lambda function or elsewhere
 }
 ```
 
+### Table Map
+
+You may need to access your dynamo table L2 constructs. These can be accessed via `appSyncTransformer.tableMap`.
+
+### DynamoDB Streams
+
+There are two ways to enable DynamoDB streams for a table. The first version is probably most preferred. You pass in the `@model` type name and the StreamViewType as properties when creating the AppSyncTransformer. This will also allow you to access the `tableStreamArn` property of the L2 table construct from the `tableMap`.
+
+```ts
+const appSyncTransformer = new AppSyncTransformer(stack, 'test-transformer', {
+  schemaPath: testSchemaPath,
+  dynamoDbStreamConfig: {
+    Order: StreamViewType.NEW_IMAGE,
+    Blog: StreamViewType.NEW_AND_OLD_IMAGES
+  }
+});
+
+const orderTable = appSyncTransformer.tableMap.OrderTable;
+// Do something with the table stream arn - orderTable.tableStreamArn
+```
+
+A convenience method is also available. It returns the stream arn because the L2 Table construct does not seem to update with the value since we are updating the underlying CfnTable. Normally a Table construct must pass in the stream specification as a prop
+
+```ts
+const streamArn = appSyncTransformer.addDynamoDBStream({
+  modelTypeName: 'Order',
+  streamViewType: StreamViewType.NEW_IMAGE,
+});
+
+// Do something with the streamArn
+```
+
 ### DataStore Support
 
 1. Pass `syncEnabled: true` to the `AppSyncTransformerProps`

diff --git a/src/appsync-transformer.ts b/src/appsync-transformer.ts
@@ -12,10 +12,13 @@ import {
 } from '@aws-cdk/aws-appsync';
 
 import {
+  CfnTable,
   Table,
   AttributeType,
   ProjectionType,
   BillingMode,
+  StreamViewType,
+  TableProps,
 } from '@aws-cdk/aws-dynamodb';
 import { Effect, PolicyStatement } from '@aws-cdk/aws-iam';
 import { IFunction } from '@aws-cdk/aws-lambda';
@@ -76,6 +79,12 @@ export interface AppSyncTransformerProps {
    */
   readonly xrayEnabled?: boolean;
 
+  /**
+   * A map of @model type names to stream view type
+   * e.g { Blog: StreamViewType.NEW_IMAGE }
+   */
+  readonly dynamoDbStreamConfig?: { [name: string]: StreamViewType };
+
   /**
    * Optional. Additonal custom transformers to run prior to the CDK resource generations.
    * Particularly useful for custom directives.
@@ -85,7 +94,6 @@ export interface AppSyncTransformerProps {
 
   readonly preCdkTransformers?: any[];
 
-
   /**
    * Optional. Additonal custom transformers to run after the CDK resource generations.
    * Mostly useful for deep level customization of the generated CDK CloudFormation resources.
@@ -125,6 +133,12 @@ export class AppSyncTransformer extends Construct {
    */
   public readonly tableNameMap: { [name: string]: any };
 
+  /**
+   * Map of cdk table keys to L2 Table
+   * e.g. { 'TaskTable': Table }
+   */
+  public readonly tableMap: { [name: string]: Table };
+
   /**
    * The outputs from the SchemaTransformer
    */
@@ -147,13 +161,16 @@ export class AppSyncTransformer extends Construct {
     [name: string]: CdkTransformerHttpResolver[];
   };
 
+  private props: AppSyncTransformerProps
   private isSyncEnabled: boolean;
   private syncTable: Table | undefined;
   private pointInTimeRecovery: boolean;
 
   constructor(scope: Construct, id: string, props: AppSyncTransformerProps) {
     super(scope, id);
 
+    this.props = props;
+    this.tableMap = {};
     this.isSyncEnabled = props.syncEnabled ? props.syncEnabled : false;
     this.pointInTimeRecovery = props.enableDynamoPointInTimeRecovery ?? false;
 
@@ -316,6 +333,8 @@ export class AppSyncTransformer extends Construct {
 
     Object.keys(tableData).forEach((tableKey: any) => {
       const table = this.createTable(tableData[tableKey]);
+      this.tableMap[tableKey] = table;
+
       const dataSource = this.appsyncAPI.addDynamoDbDataSource(tableKey, table);
 
       // https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-appsync-datasource-deltasyncconfig.html
@@ -394,26 +413,25 @@ export class AppSyncTransformer extends Construct {
   }
 
   private createTable(tableData: CdkTransformerTable) {
-    let tableProps: any = {
+    // I do not want to force people to pass `TypeTable` - this way they are only passing the @model Type name
+    const modelTypeName = tableData.tableName.replace('Table', '');
+    const streamSpecification = this.props.dynamoDbStreamConfig && this.props.dynamoDbStreamConfig[modelTypeName];
+    const tableProps: TableProps = {
       billingMode: BillingMode.PAY_PER_REQUEST,
       partitionKey: {
         name: tableData.partitionKey.name,
         type: this.convertAttributeType(tableData.partitionKey.type),
       },
       pointInTimeRecovery: this.pointInTimeRecovery,
+      sortKey: tableData.sortKey && tableData.sortKey.name
+        ? {
+          name: tableData.sortKey.name,
+          type: this.convertAttributeType(tableData.sortKey.type),
+        } : undefined,
+      timeToLiveAttribute: tableData?.ttl?.enabled ? tableData.ttl.attributeName : undefined,
+      stream: streamSpecification,
     };
 
-    if (tableData.sortKey && tableData.sortKey.name) {
-      tableProps.sortKey = {
-        name: tableData.sortKey.name,
-        type: this.convertAttributeType(tableData.sortKey.type),
-      };
-    }
-
-    if (tableData.ttl && tableData.ttl.enabled) {
-      tableProps.timeToLiveAttribute = tableData.ttl.attributeName;
-    }
-
     const table = new Table(
       this.nestedAppsyncStack,
       tableData.tableName,
@@ -557,4 +575,31 @@ export class AppSyncTransformer extends Construct {
 
     return functionDataSource;
   }
+
+  /**
+   * Adds a stream to the dynamodb table associated with the type
+   * @param props
+   * @returns string - the stream arn token
+   */
+  public addDynamoDBStream(props: DynamoDBStreamProps): string {
+    const tableName = `${props.modelTypeName}Table`;
+    const table = this.tableMap[tableName];
+    if (!table) throw new Error(`Table with name '${tableName}' not found.`);
+
+    const cfnTable = table.node.defaultChild as CfnTable;
+    cfnTable.streamSpecification = {
+      streamViewType: props.streamViewType,
+    };
+
+    return cfnTable.attrStreamArn;
+  }
+}
+
+export interface DynamoDBStreamProps {
+  /**
+   * The @model type name from the graph schema
+   * e.g. Blog
+   */
+  readonly modelTypeName: string;
+  readonly streamViewType: StreamViewType;
 }
diff --git a/test/index.test.ts b/test/index.test.ts
@@ -2,7 +2,7 @@ import '@aws-cdk/assert/jest';
 import * as path from 'path';
 import { AuthorizationType, AuthorizationConfig, UserPoolDefaultAction } from '@aws-cdk/aws-appsync';
 import { UserPool, UserPoolClient } from '@aws-cdk/aws-cognito';
-import { CfnTable } from '@aws-cdk/aws-dynamodb';
+import { StreamViewType } from '@aws-cdk/aws-dynamodb';
 import { Runtime, Code, Function } from '@aws-cdk/aws-lambda';
 import { App, Stack } from '@aws-cdk/core';
 
@@ -129,7 +129,7 @@ test('Model Tables Created and PITR', () => {
   }
 
   // Make sure ttl is on Order table
-  const orderTable = appSyncTransformer.nestedAppsyncStack.node.findChild('OrderTable') as CfnTable;
+  const orderTable = appSyncTransformer.nestedAppsyncStack.node.findChild('OrderTable');
   expect(orderTable).toBeTruthy();
 
   expect(appSyncTransformer.nestedAppsyncStack).toHaveResource('AWS::DynamoDB::Table', {
@@ -352,4 +352,87 @@ test('Invalid Transformer', () => {
   };
 
   expect(willThrow).toThrow();
+});
+
+test('DynamoDB Stream Config Property', () => {
+  const mockApp = new App();
+  const stack = new Stack(mockApp, 'user-pool-auth-stack');
+
+  const userPool = new UserPool(stack, 'test-userpool');
+  const userPoolClient = new UserPoolClient(stack, 'test-userpool-client', {
+    userPool: userPool,
+  });
+
+  const appSyncTransformer = new AppSyncTransformer(stack, 'test-transformer', {
+    schemaPath: testSchemaPath,
+    apiName: 'user-pool-auth-api',
+    authorizationConfig: {
+      defaultAuthorization: {
+        authorizationType: AuthorizationType.USER_POOL,
+        userPoolConfig: {
+          userPool: userPool,
+          appIdClientRegex: userPoolClient.userPoolClientId,
+          defaultAction: UserPoolDefaultAction.ALLOW,
+        },
+      },
+    },
+    dynamoDbStreamConfig: {
+      Order: StreamViewType.NEW_IMAGE,
+    },
+  });
+
+  const tableData = appSyncTransformer.outputs.cdkTables;
+  if (!tableData) throw new Error('Expected table data');
+
+  // Make sure order table exists
+  const orderTable = appSyncTransformer.tableMap.OrderTable;
+  expect(orderTable.tableStreamArn).toBeTruthy();
+
+  expect(appSyncTransformer.nestedAppsyncStack).toHaveResource('AWS::DynamoDB::Table', {
+    StreamSpecification: {
+      StreamViewType: StreamViewType.NEW_IMAGE,
+    },
+  });
+});
+
+test('DynamoDB Stream Enabled Convenience Method', () => {
+  const mockApp = new App();
+  const stack = new Stack(mockApp, 'user-pool-auth-stack');
+
+  const userPool = new UserPool(stack, 'test-userpool');
+  const userPoolClient = new UserPoolClient(stack, 'test-userpool-client', {
+    userPool: userPool,
+  });
+
+  const appSyncTransformer = new AppSyncTransformer(stack, 'test-transformer', {
+    schemaPath: testSchemaPath,
+    apiName: 'user-pool-auth-api',
+    authorizationConfig: {
+      defaultAuthorization: {
+        authorizationType: AuthorizationType.USER_POOL,
+        userPoolConfig: {
+          userPool: userPool,
+          appIdClientRegex: userPoolClient.userPoolClientId,
+          defaultAction: UserPoolDefaultAction.ALLOW,
+        },
+      },
+    },
+  });
+
+  const tableData = appSyncTransformer.outputs.cdkTables;
+  if (!tableData) throw new Error('Expected table data');
+
+  const streamArn = appSyncTransformer.addDynamoDBStream({
+    modelTypeName: 'Order',
+    streamViewType: StreamViewType.NEW_IMAGE,
+  });
+
+  expect(streamArn).toBeTruthy();
+
+  expect(appSyncTransformer.nestedAppsyncStack).toHaveResource('AWS::DynamoDB::Table', {
+    StreamSpecification: {
+      StreamViewType: StreamViewType.NEW_IMAGE,
+    },
+  });
+
 });