Repository: michalkvasnicak/aws-lambda-graphql (ts)

Issue:
[DynamoDB] - Filter out expired subscriptions during event processing
Expired records appear in results even if they were marked by DynamoDB to be deleted. When a list of subscribers is returned we need to check the `ttl` field value whether the subscription is already expired. Same applies to connnections.

We'd need to update `ttl` settings to accept `false` value (saying ttl is disabled in case of `false`) so we can disable this check.

Pull request:
feat: filter out ddb expired connections/subscriptions/events
This PR introduces a support for filtering out expired events from DynamoDB and also filters out expired connections and subscriptions.

Closes #91 

Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index a5848bd..9d28960 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -9,6 +9,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### [Unreleased](https://github.com/michalkvasnicak/aws-lambda-graphql/compare/aws-lambda-graphql@1.0.0-alpha.7...HEAD)
 
+#### Added
+
+- Added support for filtering out expired DynamoDB connections, subscriptions and events, see [#92](https://github.com/michalkvasnicak/aws-lambda-graphql/pull/92).
+
 ### [v1.0.0-alpha.8](https://github.com/michalkvasnicak/aws-lambda-graphql/compare/aws-lambda-graphql@1.0.0-alpha.7...aws-lambda-graphql@1.0.0-alpha.8) - 2020-07-06
 
 #### Added
diff --git a/README.md b/README.md
index 50de3e0..a897c74 100644
--- a/README.md
+++ b/README.md
@@ -77,13 +77,22 @@ import {
   DynamoDBSubscriptionManager,
 } from 'aws-lambda-graphql';
 
+/*
+ By default subscriptions and connections use TTL of 2 hours. 
+ This can be changed by `ttl` option in DynamoDBSubscriptionManager and DynamoDBConnectionManager.
+
+ ttl accepts a number in seconds (default is 7200 seconds) or
+ false to turn it off.
+
+ It's your responsibility to set up TTL on your connections and subscriptions tables.
+*/
 const subscriptionManager = new DynamoDBSubscriptionManager();
 const connectionManager = new DynamoDBConnectionManager({
   subscriptionManager,
 });
 ```
 
-**⚠️ in order to clean up stale connections and subscriptions please set up TTL on `ttl` field in Connections, Subscriptions and SubscriptionOperations tables.**
+**⚠️ in order to clean up stale connections and subscriptions please set up TTL on `ttl` field in Connections, Subscriptions and SubscriptionOperations tables. You can turn off the TTL by setting up `ttl` option to `false` in `DynamoDBSubscriptionManager` and `DynamoDBConnectionManager`.**
 
 Redis:
 
@@ -119,6 +128,14 @@ import {
   DynamoDBSubscriptionManager,
 } from 'aws-lambda-graphql';
 
+/*
+ By default event stores uses TTL of 2 hours on every event. 
+ This can be changed by `ttl` option in DynamoDBEventStore.
+ ttl accepts a number in seconds (default is 7200 seconds) or
+ false to turn it off.
+
+ It's your responsibility to set up TTL on your events table.
+*/
 const eventStore = new DynamoDBEventStore();
 const subscriptionManager = new DynamoDBSubscriptionManager();
 const connectionManager = new DynamoDBConnectionManager({
@@ -128,7 +145,7 @@ const connectionManager = new DynamoDBConnectionManager({
 
 That's it for now. Our `eventStore` will use DynamoDB to store messages that we want to broadcast to all subscribed clients.
 
-**⚠️ in order to clean up old events, please set up TTL on `ttl` field in Events store table**
+**⚠️ in order to clean up old events, please set up TTL on `ttl` field in Events store table. This can be turned off by setting up the `ttl` option to `false`.**
 
 #### 1.3 Setting up the GraphQL schema
 
diff --git a/packages/aws-lambda-graphql/src/DynamoDBConnectionManager.ts b/packages/aws-lambda-graphql/src/DynamoDBConnectionManager.ts
index 9c58f90..a12921d 100644
--- a/packages/aws-lambda-graphql/src/DynamoDBConnectionManager.ts
+++ b/packages/aws-lambda-graphql/src/DynamoDBConnectionManager.ts
@@ -9,9 +9,17 @@ import {
   HydrateConnectionOptions,
 } from './types';
 import { computeTTL } from './helpers';
+import { isTTLExpired } from './helpers/isTTLExpired';
 
 const DEFAULT_TTL = 7200;
 
+interface DynamoDBConnection extends IConnection {
+  /**
+   * TTL in UNIX seconds
+   */
+  ttl?: number;
+}
+
 interface DynamoDBConnectionManagerOptions {
   /**
    * Use this to override ApiGatewayManagementApi (for example in usage with serverless-offline)
@@ -32,8 +40,10 @@ interface DynamoDBConnectionManagerOptions {
    * Optional TTL for connections (stored in ttl field) in seconds
    *
    * Default value is 2 hours
+   *
+   * Set to false to turn off TTL
    */
-  ttl?: number;
+  ttl?: number | false;
 }
 
 /**
@@ -50,7 +60,7 @@ export class DynamoDBConnectionManager implements IConnectionManager {
 
   private subscriptions: ISubscriptionManager;
 
-  private ttl: number;
+  private ttl: number | false;
 
   constructor({
     apiGatewayManager,
@@ -68,8 +78,8 @@ export class DynamoDBConnectionManager implements IConnectionManager {
 
   hydrateConnection = async (
     connectionId: string,
-    options: HydrateConnectionOptions,
-  ): Promise<IConnection> => {
+    options?: HydrateConnectionOptions,
+  ): Promise<DynamoDBConnection> => {
     const { retryCount = 0, timeout = 50 } = options || {};
     // if connection is not found, throw so we can terminate connection
     let connection;
@@ -83,16 +93,18 @@ export class DynamoDBConnectionManager implements IConnectionManager {
           },
         })
         .promise();
+
       if (result.Item) {
         // Jump out of loop
-        connection = result.Item as IConnection;
+        connection = result.Item as DynamoDBConnection;
         break;
       }
+
       // wait for another round
       await new Promise(r => setTimeout(r, timeout));
     }
 
-    if (!connection) {
+    if (!connection || isTTLExpired(connection.ttl)) {
       throw new ConnectionNotFoundError(`Connection ${connectionId} not found`);
     }
 
@@ -101,7 +113,7 @@ export class DynamoDBConnectionManager implements IConnectionManager {
 
   setConnectionData = async (
     data: IConnectionData,
-    { id }: IConnection,
+    { id }: DynamoDBConnection,
   ): Promise<void> => {
     await this.db
       .update({
@@ -123,7 +135,7 @@ export class DynamoDBConnectionManager implements IConnectionManager {
   registerConnection = async ({
     connectionId,
     endpoint,
-  }: IConnectEvent): Promise<IConnection> => {
+  }: IConnectEvent): Promise<DynamoDBConnection> => {
     const connection: IConnection = {
       id: connectionId,
       data: { endpoint, context: {}, isInitialized: false },
@@ -136,7 +148,11 @@ export class DynamoDBConnectionManager implements IConnectionManager {
           createdAt: new Date().toString(),
           id: connection.id,
           data: connection.data,
-          ttl: computeTTL(this.ttl),
+          ...(this.ttl === false || this.ttl == null
+            ? {}
+            : {
+                ttl: computeTTL(this.ttl),
+              }),
         },
       })
       .promise();
@@ -145,7 +161,7 @@ export class DynamoDBConnectionManager implements IConnectionManager {
   };
 
   sendToConnection = async (
-    connection: IConnection,
+    connection: DynamoDBConnection,
     payload: string | Buffer,
   ): Promise<void> => {
     try {
@@ -163,7 +179,7 @@ export class DynamoDBConnectionManager implements IConnectionManager {
     }
   };
 
-  unregisterConnection = async ({ id }: IConnection): Promise<void> => {
+  unregisterConnection = async ({ id }: DynamoDBConnection): Promise<void> => {
     await Promise.all([
       this.db
         .delete({
@@ -177,7 +193,7 @@ export class DynamoDBConnectionManager implements IConnectionManager {
     ]);
   };
 
-  closeConnection = async ({ id, data }: IConnection): Promise<void> => {
+  closeConnection = async ({ id, data }: DynamoDBConnection): Promise<void> => {
     await this.createApiGatewayManager(data.endpoint)
       .deleteConnection({ ConnectionId: id })
       .promise();
diff --git a/packages/aws-lambda-graphql/src/DynamoDBEventProcessor.ts b/packages/aws-lambda-graphql/src/DynamoDBEventProcessor.ts
index 81c689f..d670d41 100644
--- a/packages/aws-lambda-graphql/src/DynamoDBEventProcessor.ts
+++ b/packages/aws-lambda-graphql/src/DynamoDBEventProcessor.ts
@@ -3,11 +3,13 @@ import { DynamoDB } from 'aws-sdk';
 import { isAsyncIterable, getAsyncIterator } from 'iterall';
 import { ExecutionResult } from 'graphql';
 import { ArrayPubSub } from './ArrayPubSub';
-import { ISubscriptionEvent, IEventProcessor } from './types';
+import { IEventProcessor } from './types';
 import { formatMessage } from './formatMessage';
 import { execute } from './execute';
 import { SERVER_EVENT_TYPES } from './protocol';
 import { Server } from './Server';
+import { IDynamoDBSubscriptionEvent } from './DynamoDBEventStore';
+import { isTTLExpired } from './helpers/isTTLExpired';
 
 interface DynamoDBEventProcessorOptions {
   onError?: (err: any) => void;
@@ -39,10 +41,15 @@ export class DynamoDBEventProcessor<TServer extends Server = Server>
         }
 
         // now construct event from dynamodb image
-        const event: ISubscriptionEvent = DynamoDB.Converter.unmarshall(
+        const event: IDynamoDBSubscriptionEvent = DynamoDB.Converter.unmarshall(
           record.dynamodb!.NewImage as any,
         ) as any;
 
+        // skip if event is expired
+        if (isTTLExpired(event.ttl)) {
+          continue;
+        }
+
         // iterate over subscribers that listen to this event
         // and for each connection:
         //  - create a schema (so we have subscribers registered in PubSub)
diff --git a/packages/aws-lambda-graphql/src/DynamoDBEventStore.ts b/packages/aws-lambda-graphql/src/DynamoDBEventStore.ts
index ebd328c..121c6bd 100644
--- a/packages/aws-lambda-graphql/src/DynamoDBEventStore.ts
+++ b/packages/aws-lambda-graphql/src/DynamoDBEventStore.ts
@@ -3,6 +3,13 @@ import { ulid } from 'ulid';
 import { IEventStore, ISubscriptionEvent } from './types';
 import { computeTTL } from './helpers';
 
+export interface IDynamoDBSubscriptionEvent extends ISubscriptionEvent {
+  /**
+   * TTL in UNIX seconds
+   */
+  ttl?: number;
+}
+
 const DEFAULT_TTL = 7200;
 
 interface DynamoDBEventStoreOptions {
@@ -18,8 +25,10 @@ interface DynamoDBEventStoreOptions {
    * Optional TTL for events (stored in ttl field) in seconds
    *
    * Default value is 2 hours
+   *
+   * Set to false to turn off TTL
    */
-  ttl?: number;
+  ttl?: number | false;
 }
 
 /**
@@ -34,7 +43,7 @@ export class DynamoDBEventStore implements IEventStore {
 
   private tableName: string;
 
-  private ttl: number;
+  private ttl: number | false;
 
   constructor({
     dynamoDbClient,
@@ -53,7 +62,9 @@ export class DynamoDBEventStore implements IEventStore {
         Item: {
           id: ulid(),
           ...event,
-          ttl: computeTTL(this.ttl),
+          ...(this.ttl === false || this.ttl == null
+            ? {}
+            : { ttl: computeTTL(this.ttl) }),
         },
       })
       .promise();
diff --git a/packages/aws-lambda-graphql/src/DynamoDBSubscriptionManager.ts b/packages/aws-lambda-graphql/src/DynamoDBSubscriptionManager.ts
index cbcf51c..94dd299 100644
--- a/packages/aws-lambda-graphql/src/DynamoDBSubscriptionManager.ts
+++ b/packages/aws-lambda-graphql/src/DynamoDBSubscriptionManager.ts
@@ -20,6 +20,10 @@ interface DynamoDBSubscriber extends ISubscriber {
    * it is in format connectionId:operationId
    */
   subscriptionId: string;
+  /**
+   * TTL in UNIX seconds
+   */
+  ttl?: number;
 }
 
 interface DynamoDBSubscriptionManagerOptions {
@@ -39,8 +43,10 @@ interface DynamoDBSubscriptionManagerOptions {
    * Optional TTL for subscriptions (stored in ttl field) in seconds
    *
    * Default value is 2 hours
+   *
+   * Set to false to turn off TTL
    */
-  ttl?: number;
+  ttl?: number | false;
 }
 
 /**
@@ -64,7 +70,7 @@ export class DynamoDBSubscriptionManager implements ISubscriptionManager {
 
   private db: DynamoDB.DocumentClient;
 
-  private ttl: number;
+  private ttl: number | false;
 
   constructor({
     dynamoDbClient,
@@ -90,14 +96,20 @@ export class DynamoDBSubscriptionManager implements ISubscriptionManager {
           return { value: [], done: true };
         }
 
+        const time = Math.round(Date.now() / 1000);
         const result = await this.db
           .query({
             ExclusiveStartKey,
             TableName: this.subscriptionsTableName,
             Limit: 50,
             KeyConditionExpression: 'event = :event',
+            FilterExpression: '#ttl > :time OR attribute_not_exists(#ttl)',
             ExpressionAttributeValues: {
               ':event': name,
+              ':time': time,
+            },
+            ExpressionAttributeNames: {
+              '#ttl': 'ttl',
             },
           })
           .promise();
@@ -136,7 +148,10 @@ export class DynamoDBSubscriptionManager implements ISubscriptionManager {
     }
     const [name] = names;
 
-    const ttl = computeTTL(this.ttl);
+    const ttlField =
+      this.ttl === false || this.ttl == null
+        ? {}
+        : { ttl: computeTTL(this.ttl) };
 
     await this.db
       .batchWrite({
@@ -150,7 +165,7 @@ export class DynamoDBSubscriptionManager implements ISubscriptionManager {
                   event: name,
                   subscriptionId,
                   operationId: operation.operationId,
-                  ttl,
+                  ...ttlField,
                 } as DynamoDBSubscriber,
               },
             },
@@ -161,7 +176,7 @@ export class DynamoDBSubscriptionManager implements ISubscriptionManager {
                 Item: {
                   subscriptionId,
                   event: name,
-                  ttl,
+                  ...ttlField,
                 },
               },
             },
diff --git a/packages/aws-lambda-graphql/src/helpers/isTTLExpired.ts b/packages/aws-lambda-graphql/src/helpers/isTTLExpired.ts
new file mode 100644
index 0000000..d574861
--- /dev/null
+++ b/packages/aws-lambda-graphql/src/helpers/isTTLExpired.ts
@@ -0,0 +1,7 @@
+export function isTTLExpired(ttl?: null | undefined | false | number): boolean {
+  if (ttl == null || ttl === false) {
+    return false;
+  }
+
+  return ttl * 1000 < Date.now();
+}


Test patch:
diff --git a/packages/aws-lambda-graphql/src/__tests__/DynamoDBConnectionManager.test.ts b/packages/aws-lambda-graphql/src/__tests__/DynamoDBConnectionManager.test.ts
index 7ca1c12..1fe0926 100644
--- a/packages/aws-lambda-graphql/src/__tests__/DynamoDBConnectionManager.test.ts
+++ b/packages/aws-lambda-graphql/src/__tests__/DynamoDBConnectionManager.test.ts
@@ -26,6 +26,7 @@ import {
 } from 'aws-sdk';
 import { DynamoDBConnectionManager } from '../DynamoDBConnectionManager';
 import { ConnectionNotFoundError } from '../errors';
+import { computeTTL } from '../helpers';
 
 const subscriptionManager: any = {
   unsubscribeAllByConnectionId: jest.fn(),
@@ -81,6 +82,34 @@ describe('DynamoDBConnectionManager', () => {
         }),
       );
     });
+
+    it('supports turning off ttl', async () => {
+      const manager = new DynamoDBConnectionManager({
+        subscriptions: subscriptionManager,
+        ttl: false,
+      });
+
+      await expect(
+        manager.registerConnection({ connectionId: 'id', endpoint: '' }),
+      ).resolves.toEqual({
+        id: 'id',
+        data: {
+          endpoint: '',
+          context: {},
+          isInitialized: false,
+        },
+      });
+
+      expect(putMock as jest.Mock).toHaveBeenCalledTimes(1);
+      expect(putMock).not.toHaveBeenCalledWith(
+        expect.objectContaining({
+          Item: {
+            ttl: expect.any(Number),
+          },
+          TableName: 'Connections',
+        }),
+      );
+    });
   });
 
   describe('hydrateConnection', () => {
@@ -98,6 +127,18 @@ describe('DynamoDBConnectionManager', () => {
       expect(getMock as jest.Mock).toHaveBeenCalledTimes(1);
     });
 
+    it('throws ConnectionNotFoundError if connection is expired', async () => {
+      (getPromiseMock as jest.Mock).mockResolvedValueOnce({
+        Item: { ttl: computeTTL(-10) },
+      });
+
+      await expect(manager.hydrateConnection('id')).rejects.toThrowError(
+        ConnectionNotFoundError,
+      );
+
+      expect(getMock as jest.Mock).toHaveBeenCalledTimes(1);
+    });
+
     it('hydrates connection', async () => {
       (getPromiseMock as jest.Mock).mockResolvedValueOnce({
         Item: { id: 'id', data: { endpoint: '' } },
diff --git a/packages/aws-lambda-graphql/src/__tests__/DynamoDBEventProcessor.test.ts b/packages/aws-lambda-graphql/src/__tests__/DynamoDBEventProcessor.test.ts
index e709a6b..2fdaa0a 100644
--- a/packages/aws-lambda-graphql/src/__tests__/DynamoDBEventProcessor.test.ts
+++ b/packages/aws-lambda-graphql/src/__tests__/DynamoDBEventProcessor.test.ts
@@ -9,6 +9,7 @@ import { SERVER_EVENT_TYPES } from '../protocol';
 import { ISubscriber } from '../types';
 import { Server } from '../Server';
 import { PubSub } from '../PubSub';
+import { computeTTL } from '../helpers';
 
 const query = parse(/* GraphQL */ `
   subscription Test($authorId: ID) {
@@ -17,7 +18,7 @@ const query = parse(/* GraphQL */ `
 `);
 
 describe('DynamoDBEventProcessor', () => {
-  it(' payload as JSON', async () => {
+  it('supports payload as JSON', async () => {
     const connectionManager = {
       sendToConnection: jest.fn(),
     };
@@ -143,7 +144,7 @@ describe('DynamoDBEventProcessor', () => {
     );
   });
 
-  it(' payload as object', async () => {
+  it('supports payload as object', async () => {
     const connectionManager = {
       sendToConnection: jest.fn(),
     };
@@ -326,4 +327,141 @@ describe('DynamoDBEventProcessor', () => {
       lambdaContext: expect.any(Object),
     });
   });
+
+  it('skips expired events', async () => {
+    const connectionManager = {
+      sendToConnection: jest.fn(),
+    };
+    const subscriptionManager = {
+      subscribersByEventName: jest.fn(() => ({
+        [$$asyncIterator]: () =>
+          createAsyncIterator([
+            [
+              {
+                connection: { id: '1', data: {} } as any,
+                event: 'test',
+                operationId: '1',
+                operation: { query, variables: { authorId: '1' } },
+              },
+              {
+                connection: { id: '2', data: {} } as any,
+                event: 'test',
+                operationId: '1',
+                operation: { query, variables: { authorId: '2' } },
+              },
+              {
+                connection: { id: '3', data: {} } as any,
+                event: 'test',
+                operationId: '1',
+                operation: { query, variables: { authorId: '2' } },
+              },
+              {
+                connection: { id: '4', data: {} } as any,
+                event: 'test',
+                operationId: '1',
+                operation: { query, variables: { authorId: '1' } },
+              },
+              {
+                connection: {
+                  id: '5',
+                  data: { context: { authorId: '2' } },
+                } as any,
+                event: 'test',
+                operationId: '1',
+                operation: { query, variables: {} },
+              },
+            ] as ISubscriber[],
+          ]),
+      })),
+    };
+
+    const server = new Server({
+      context: {
+        // becuase our test schema relies on this context
+        pubSub: new PubSub({ eventStore: {} as any }),
+      },
+      connectionManager: connectionManager as any,
+      eventProcessor: new DynamoDBEventProcessor(),
+      schema: createSchema(),
+      subscriptionManager: subscriptionManager as any,
+    });
+    const eventProcessor = server.createEventHandler();
+
+    const Records: DynamoDBRecord[] = [
+      {
+        dynamodb: {
+          NewImage: DynamoDB.Converter.marshall({
+            event: 'test',
+            payload: JSON.stringify({ authorId: '1', text: 'test 1' }),
+            ttl: computeTTL(-1),
+          }) as any,
+        },
+        eventName: 'INSERT',
+      },
+      {
+        dynamodb: {
+          NewImage: DynamoDB.Converter.marshall({
+            event: 'test',
+            payload: JSON.stringify({ authorId: '2', text: 'test 2' }),
+            ttl: computeTTL(10),
+          }) as any,
+        },
+        eventName: 'INSERT',
+      },
+      {
+        dynamodb: {
+          NewImage: DynamoDB.Converter.marshall({
+            event: 'test',
+            payload: JSON.stringify({ authorId: '1', text: 'test 3' }),
+          }) as any,
+        },
+        eventName: 'INSERT',
+      },
+    ];
+
+    // now process events
+    await eventProcessor({ Records }, {} as any, {} as any);
+
+    expect(connectionManager.sendToConnection).toHaveBeenCalledTimes(5);
+    expect(connectionManager.sendToConnection).toHaveBeenCalledWith(
+      { id: '2', data: {} },
+      formatMessage({
+        id: '1',
+        payload: { data: { textFeed: 'test 2' } },
+        type: SERVER_EVENT_TYPES.GQL_DATA,
+      }),
+    );
+    expect(connectionManager.sendToConnection).toHaveBeenCalledWith(
+      { id: '3', data: {} },
+      formatMessage({
+        id: '1',
+        payload: { data: { textFeed: 'test 2' } },
+        type: SERVER_EVENT_TYPES.GQL_DATA,
+      }),
+    );
+    expect(connectionManager.sendToConnection).toHaveBeenCalledWith(
+      { id: '5', data: { context: { authorId: '2' } } },
+      formatMessage({
+        id: '1',
+        payload: { data: { textFeed: 'test 2' } },
+        type: SERVER_EVENT_TYPES.GQL_DATA,
+      }),
+    );
+    expect(connectionManager.sendToConnection).toHaveBeenCalledWith(
+      { id: '1', data: {} },
+      formatMessage({
+        id: '1',
+        payload: { data: { textFeed: 'test 3' } },
+        type: SERVER_EVENT_TYPES.GQL_DATA,
+      }),
+    );
+    expect(connectionManager.sendToConnection).toHaveBeenCalledWith(
+      { id: '4', data: {} },
+      formatMessage({
+        id: '1',
+        payload: { data: { textFeed: 'test 3' } },
+        type: SERVER_EVENT_TYPES.GQL_DATA,
+      }),
+    );
+  });
 });
diff --git a/packages/aws-lambda-graphql/src/__tests__/DynamoDBEventStore.test.ts b/packages/aws-lambda-graphql/src/__tests__/DynamoDBEventStore.test.ts
index c84f2ea..8b7f063 100644
--- a/packages/aws-lambda-graphql/src/__tests__/DynamoDBEventStore.test.ts
+++ b/packages/aws-lambda-graphql/src/__tests__/DynamoDBEventStore.test.ts
@@ -35,4 +35,28 @@ describe('DynamoDBEventStore', () => {
       }),
     );
   });
+
+  it('supports turning off the TTL', async () => {
+    const eventStore = new DynamoDBEventStore({
+      ttl: false,
+    });
+
+    await expect(
+      eventStore.publish({
+        event: 'test',
+        payload: {
+          custom: true,
+        },
+      }),
+    ).resolves.toBeUndefined();
+
+    expect(putPromiseMock).toHaveBeenCalledTimes(1);
+    expect(putMock).not.toHaveBeenCalledWith(
+      expect.objectContaining({
+        Item: expect.objectContaining({
+          ttl: expect.any(Number),
+        }),
+      }),
+    );
+  });
 });
diff --git a/packages/aws-lambda-graphql/src/__tests__/DynamoDBSubscriptionManager.test.ts b/packages/aws-lambda-graphql/src/__tests__/DynamoDBSubscriptionManager.test.ts
index 67b4358..d7cf5b4 100644
--- a/packages/aws-lambda-graphql/src/__tests__/DynamoDBSubscriptionManager.test.ts
+++ b/packages/aws-lambda-graphql/src/__tests__/DynamoDBSubscriptionManager.test.ts
@@ -126,6 +126,48 @@ describe('DynamoDBSubscriptionManager', () => {
         }),
       );
     });
+
+    it('supports turning off ttl', async () => {
+      const subscriptionManager = new DynamoDBSubscriptionManager({
+        ttl: false,
+      });
+
+      (batchWritePromiseMock as jest.Mock).mockResolvedValueOnce({});
+
+      await expect(
+        subscriptionManager.subscribe(
+          ['name1'],
+          { id: '1' } as any,
+          { operationId: '1' } as any,
+        ),
+      ).resolves.toBeUndefined();
+
+      expect(batchWritePromiseMock).toHaveBeenCalledTimes(1);
+      expect(batchWriteMock).not.toHaveBeenCalledWith(
+        expect.objectContaining({
+          RequestItems: {
+            SubscriptionOperations: [
+              {
+                PutRequest: {
+                  Item: {
+                    ttl: expect.any(Number),
+                  },
+                },
+              },
+            ],
+            Subscriptions: [
+              {
+                PutRequest: {
+                  Item: {
+                    ttl: expect.any(Number),
+                  },
+                },
+              },
+            ],
+          },
+        }),
+      );
+    });
   });
 
   describe('unsubscribe', () => {