test(idempotency): add e2e tests for idempotency

packages/commons/tests/utils/e2eUtils.ts

@@ -9,7 +9,7 @@ import {
 } from 'aws-cdk-lib/aws-lambda-nodejs';
 import { Runtime, Tracing } from 'aws-cdk-lib/aws-lambda';
 import { RetentionDays } from 'aws-cdk-lib/aws-logs';
-import { LambdaClient, InvokeCommand } from '@aws-sdk/client-lambda';
+import { InvokeCommand, LambdaClient } from '@aws-sdk/client-lambda';
 import { fromUtf8 } from '@aws-sdk/util-utf8-node';
 
 import { InvocationLogs } from './InvocationLogs';
@@ -38,7 +38,7 @@ export type StackWithLambdaFunctionOptions = {
   timeout?: Duration;
 };
 
-type FunctionPayload = { [key: string]: string | boolean | number };
+type FunctionPayload = { [key: string]: string | boolean | number | Array<Record<string, unknown>> };
 
 export const isValidRuntimeKey = (
   runtime: string
@@ -82,27 +82,26 @@ export const generateUniqueName = (
 
 export const invokeFunction = async (
   functionName: string,
-  times = 1,
+  times: number = 1,
   invocationMode: 'PARALLEL' | 'SEQUENTIAL' = 'PARALLEL',
-  payload: FunctionPayload = {}
+  payload: FunctionPayload = {},
+  includeIndex = true
 ): Promise<InvocationLogs[]> => {
   const invocationLogs: InvocationLogs[] = [];
 
-  const promiseFactory = (index?: number): Promise<void> => {
+  const promiseFactory = (index?: number, includeIndex?: boolean): Promise<void> => {
+
+    // in some cases we need to send a payload without the index, i.e. idempotency tests
+    const payloadToSend = includeIndex ? { invocation: index, ...payload } : { ...payload };
+
     const invokePromise = lambdaClient
       .send(
         new InvokeCommand({
           FunctionName: functionName,
           InvocationType: 'RequestResponse',
           LogType: 'Tail', // Wait until execution completes and return all logs
-          Payload: fromUtf8(
-            JSON.stringify({
-              invocation: index,
-              ...payload,
-            })
-          ),
-        })
-      )
+          Payload: fromUtf8(JSON.stringify(payloadToSend)),
+      }))
       .then((response) => {
         if (response?.LogResult) {
           invocationLogs.push(new InvocationLogs(response?.LogResult));
@@ -117,9 +116,10 @@ export const invokeFunction = async (
   };
 
   const promiseFactories = Array.from({ length: times }, () => promiseFactory);
+
   const invocation =
     invocationMode == 'PARALLEL'
-      ? Promise.all(promiseFactories.map((factory, index) => factory(index)))
+      ? Promise.all(promiseFactories.map((factory, index) => factory(index, includeIndex)))
       : chainPromises(promiseFactories);
   await invocation;
 

packages/idempotency/package.json

@@ -13,10 +13,10 @@
     "commit": "commit",
     "test": "npm run test:unit",
     "test:unit": "jest --group=unit --detectOpenHandles --coverage --verbose",
-    "test:e2e:nodejs14x": "echo \"Not implemented\"",
-    "test:e2e:nodejs16x": "echo \"Not implemented\"",
-    "test:e2e:nodejs18x": "echo \"Not implemented\"",
-    "test:e2e": "echo \"Not implemented\"",
+    "test:e2e:nodejs14x": "RUNTIME=nodejs14x jest --group=e2e",
+    "test:e2e:nodejs16x": "RUNTIME=nodejs16x jest --group=e2e",
+    "test:e2e:nodejs18x": "RUNTIME=nodejs18x jest --group=e2e",
+    "test:e2e": "jest --group=e2e --detectOpenHandles",
     "watch": "jest --watch --group=unit",
     "build": "tsc",
     "lint": "eslint --ext .ts --no-error-on-unmatched-pattern src tests",
@@ -57,6 +57,7 @@
   ],
   "devDependencies": {
     "@types/jmespath": "^0.15.0",
+    "@aws-sdk/client-dynamodb": "^3.231.0",
     "aws-sdk-client-mock": "^2.0.1",
     "aws-sdk-client-mock-jest": "^2.0.1"
   }

packages/idempotency/src/IdempotencyHandler.ts

@@ -1,25 +1,42 @@
-import type { AnyFunctionWithRecord, IdempotencyOptions } from './types';
+import type { AnyFunctionWithRecord, IdempotencyHandlerOptions } from './types';
 import { IdempotencyRecordStatus } from './types';
 import {
   IdempotencyAlreadyInProgressError,
   IdempotencyInconsistentStateError,
   IdempotencyItemAlreadyExistsError,
   IdempotencyPersistenceLayerError,
 } from './Exceptions';
-import { IdempotencyRecord } from './persistence';
+import { BasePersistenceLayer, IdempotencyRecord } from './persistence';
+import { IdempotencyConfig } from './IdempotencyConfig';
 
 export class IdempotencyHandler<U> {
-  public constructor(
-    private functionToMakeIdempotent: AnyFunctionWithRecord<U>,
-    private functionPayloadToBeHashed: Record<string, unknown>,
-    private idempotencyOptions: IdempotencyOptions,
-    private fullFunctionPayload: Record<string, unknown>,
-  ) {
+  private readonly fullFunctionPayload: Record<string, unknown>;
+  private readonly functionPayloadToBeHashed: Record<string, unknown>;
+  private readonly functionToMakeIdempotent: AnyFunctionWithRecord<U>;
+  private readonly idempotencyConfig: IdempotencyConfig;
+  private readonly persistenceStore: BasePersistenceLayer;
+
+  public constructor(options: IdempotencyHandlerOptions<U>) {
+    const {
+      functionToMakeIdempotent,
+      functionPayloadToBeHashed,
+      idempotencyConfig,
+      fullFunctionPayload,
+      persistenceStore
+    } = options;
+    this.functionToMakeIdempotent = functionToMakeIdempotent;
+    this.functionPayloadToBeHashed = functionPayloadToBeHashed;
+    this.idempotencyConfig = idempotencyConfig;
+    this.fullFunctionPayload = fullFunctionPayload;
+
+    this.persistenceStore = persistenceStore;
+
+    this.persistenceStore.configure({
+      config: this.idempotencyConfig
+    });
   }
 
-  public determineResultFromIdempotencyRecord(
-    idempotencyRecord: IdempotencyRecord
-  ): Promise<U> | U {
+  public determineResultFromIdempotencyRecord(idempotencyRecord: IdempotencyRecord): Promise<U> | U {
     if (idempotencyRecord.getStatus() === IdempotencyRecordStatus.EXPIRED) {
       throw new IdempotencyInconsistentStateError(
         'Item has expired during processing and may not longer be valid.'
@@ -40,10 +57,31 @@ export class IdempotencyHandler<U> {
           `There is already an execution in progress with idempotency key: ${idempotencyRecord.idempotencyKey}`
         );
       }
-    } else {
-      // Currently recalling the method as this fulfills FR1. FR3 will address using the previously stored value https://github.com/awslabs/aws-lambda-powertools-typescript/issues/447
-      return this.functionToMakeIdempotent(this.fullFunctionPayload);
     }
+
+    return idempotencyRecord.getResponse() as U;
+  }
+
+  public async getFunctionResult(): Promise<U> {
+    let result: U;
+    try {
+      result = await this.functionToMakeIdempotent(this.fullFunctionPayload);
+
+    } catch (e) {
+      try {
+        await this.persistenceStore.deleteRecord(this.functionPayloadToBeHashed);
+      } catch (e) {
+        throw new IdempotencyPersistenceLayerError('Failed to delete record from idempotency store');
+      }
+      throw e;
+    }
+    try {
+      await this.persistenceStore.saveSuccess(this.functionPayloadToBeHashed, result as Record<string, unknown>);
+    } catch (e) {
+      throw new IdempotencyPersistenceLayerError('Failed to update success record to idempotency store');
+    }
+
+    return result;
   }
 
   /**
@@ -70,13 +108,13 @@ export class IdempotencyHandler<U> {
 
   public async processIdempotency(): Promise<U> {
     try {
-      await this.idempotencyOptions.persistenceStore.saveInProgress(
+      await this.persistenceStore.saveInProgress(
         this.functionPayloadToBeHashed,
       );
     } catch (e) {
       if (e instanceof IdempotencyItemAlreadyExistsError) {
         const idempotencyRecord: IdempotencyRecord =
-          await this.idempotencyOptions.persistenceStore.getRecord(
+          await this.persistenceStore.getRecord(
             this.functionPayloadToBeHashed
           );
 
@@ -86,6 +124,7 @@ export class IdempotencyHandler<U> {
       }
     }
 
-    return this.functionToMakeIdempotent(this.fullFunctionPayload);
+    return this.getFunctionResult();
   }
+
 }

packages/idempotency/src/idempotentDecorator.ts

@@ -1,15 +1,26 @@
-import {
-  GenericTempRecord,
-  IdempotencyOptions,
-} from './types';
+import { GenericTempRecord, IdempotencyFunctionOptions, IdempotencyLambdaHandlerOptions, } from './types';
 import { IdempotencyHandler } from './IdempotencyHandler';
+import { IdempotencyConfig } from './IdempotencyConfig';
 
-const idempotent = function (options: IdempotencyOptions) {
+/**
+ * use this function to narrow the type of options between IdempotencyHandlerOptions and IdempotencyFunctionOptions
+ * @param options
+ */
+const isFunctionOption = (options: IdempotencyLambdaHandlerOptions | IdempotencyFunctionOptions): boolean => (options as IdempotencyFunctionOptions).dataKeywordArgument !== undefined;
+
+const idempotent = function (options: IdempotencyLambdaHandlerOptions | IdempotencyFunctionOptions): (target: unknown, propertyKey: string, descriptor: PropertyDescriptor) => PropertyDescriptor {
   return function (_target: unknown, _propertyKey: string, descriptor: PropertyDescriptor) {
     const childFunction = descriptor.value;
-    // TODO: sort out the type for this
-    descriptor.value = function(record: GenericTempRecord){
-      const idempotencyHandler = new IdempotencyHandler<GenericTempRecord>(childFunction, record[options.dataKeywordArgument], options, record);
+    descriptor.value = function (record: GenericTempRecord) {
+      const functionPayloadtoBeHashed = isFunctionOption(options) ? record[(options as IdempotencyFunctionOptions).dataKeywordArgument] : record;
+      const idempotencyConfig = options.config ? options.config : new IdempotencyConfig({});
+      const idempotencyHandler = new IdempotencyHandler<GenericTempRecord>({
+        functionToMakeIdempotent: childFunction,
+        functionPayloadToBeHashed: functionPayloadtoBeHashed,
+        persistenceStore: options.persistenceStore,
+        idempotencyConfig: idempotencyConfig,
+        fullFunctionPayload: record
+      });
 
       return idempotencyHandler.handle();
     };
@@ -18,4 +29,11 @@ const idempotent = function (options: IdempotencyOptions) {
   };
 };
 
-export { idempotent };
+const idempotentLambdaHandler = function (options: IdempotencyLambdaHandlerOptions): (target: unknown, propertyKey: string, descriptor: PropertyDescriptor) => PropertyDescriptor {
+  return idempotent(options);
+};
+const idempotentFunction = function (options: IdempotencyFunctionOptions): (target: unknown, propertyKey: string, descriptor: PropertyDescriptor) => PropertyDescriptor {
+  return idempotent(options);
+};
+
+export { idempotentLambdaHandler, idempotentFunction };

packages/idempotency/src/makeFunctionIdempotent.ts

@@ -1,17 +1,28 @@
 import type {
-  GenericTempRecord,
-  IdempotencyOptions,
   AnyFunctionWithRecord,
   AnyIdempotentFunction,
+  GenericTempRecord,
+  IdempotencyFunctionOptions,
 } from './types';
 import { IdempotencyHandler } from './IdempotencyHandler';
+import { IdempotencyConfig } from './IdempotencyConfig';
 
 const makeFunctionIdempotent = function <U>(
   fn: AnyFunctionWithRecord<U>,
-  options: IdempotencyOptions
+  options: IdempotencyFunctionOptions,
 ): AnyIdempotentFunction<U> {
   const wrappedFn: AnyIdempotentFunction<U> = function (record: GenericTempRecord): Promise<U> {
-    const idempotencyHandler: IdempotencyHandler<U> = new IdempotencyHandler<U>(fn, record[options.dataKeywordArgument], options, record);
+    if (options.dataKeywordArgument === undefined) {
+      throw new Error(`Missing data keyword argument ${options.dataKeywordArgument}`);
+    }
+    const idempotencyConfig = options.config ? options.config : new IdempotencyConfig({});
+    const idempotencyHandler: IdempotencyHandler<U> = new IdempotencyHandler<U>({
+      functionToMakeIdempotent: fn,
+      functionPayloadToBeHashed: record[options.dataKeywordArgument],
+      idempotencyConfig: idempotencyConfig,
+      persistenceStore: options.persistenceStore,
+      fullFunctionPayload: record
+    });
 
     return idempotencyHandler.handle();
   };

packages/idempotency/src/types/IdempotencyOptions.ts

@@ -1,9 +1,23 @@
 import type { Context } from 'aws-lambda';
 import { BasePersistenceLayer } from '../persistence/BasePersistenceLayer';
+import { AnyFunctionWithRecord } from 'types/AnyFunction';
+import { IdempotencyConfig } from '../IdempotencyConfig';
 
-type IdempotencyOptions = {
+type IdempotencyLambdaHandlerOptions = {
+  persistenceStore: BasePersistenceLayer
+  config?: IdempotencyConfig
+};
+
+type IdempotencyFunctionOptions = IdempotencyLambdaHandlerOptions & {
   dataKeywordArgument: string
+};
+
+type IdempotencyHandlerOptions<U> = {
+  functionToMakeIdempotent: AnyFunctionWithRecord<U>
+  functionPayloadToBeHashed: Record<string, unknown>
   persistenceStore: BasePersistenceLayer
+  idempotencyConfig: IdempotencyConfig
+  fullFunctionPayload: Record<string, unknown>
 };
 
 /**
@@ -45,6 +59,8 @@ type IdempotencyConfigOptions = {
 };
 
 export {
-  IdempotencyOptions,
-  IdempotencyConfigOptions
+  IdempotencyConfigOptions,
+  IdempotencyFunctionOptions,
+  IdempotencyLambdaHandlerOptions,
+  IdempotencyHandlerOptions
 };

packages/idempotency/tests/e2e/constants.ts

@@ -0,0 +1,6 @@
+export const RESOURCE_NAME_PREFIX = 'Idempotency-E2E';
+
+export const ONE_MINUTE = 60 * 1_000;
+export const TEARDOWN_TIMEOUT = 5 * ONE_MINUTE;
+export const SETUP_TIMEOUT = 5 * ONE_MINUTE;
+export const TEST_CASE_TIMEOUT = 5 * ONE_MINUTE;