channels-1094

README.md

@@ -531,7 +531,7 @@ The `perform` method accepts two arguments, (1) the request client instance (ext
 - `features` - The features available in the request based on the customer's sourceID. Features can only be enabled and/or used by internal Twilio/Segment employees. Features cannot be used for Partner builds.
 - `statsContext` - An object, containing a `statsClient` and `tags`. Stats can only be used by internal Twilio/Segment employees. Stats cannot be used for Partner builds.
 - `logger` - Logger can only be used by internal Twilio/Segment employees. Logger cannot be used for Partner builds.
-- `dataFeedCache` - DataFeedCache can only be used by internal Twilio/Segment employees. DataFeedCache cannot be used for Partner builds.
+- `engageDestinationCache` - EngageDestinationCache can only be used by internal Twilio/Segment employees. EngageDestinationCache should not be used for Partner builds.
 - `transactionContext` - An object, containing transaction variables and a method to update transaction variables which are required for few segment developed actions. Transaction Context cannot be used for Partner builds.
 - `stateContext` - An object, containing context variables and a method to get and set context variables which are required for few segment developed actions. State Context cannot be used for Partner builds.
 

packages/core/src/create-test-integration.ts

@@ -1,7 +1,7 @@
 import { createTestEvent } from './create-test-event'
 import { StateContext, Destination, TransactionContext } from './destination-kit'
 import { mapValues } from './map-values'
-import type { DestinationDefinition, StatsContext, Logger, DataFeedCache, RequestFn } from './destination-kit'
+import type { DestinationDefinition, StatsContext, Logger, EngageDestinationCache, RequestFn } from './destination-kit'
 import type { JSONObject } from './json-object'
 import type { SegmentEvent } from './segment-event'
 import { AuthTokens } from './destination-kit/parse-settings'
@@ -36,15 +36,17 @@ interface InputData<Settings> {
    */
   useDefaultMappings?: boolean
   auth?: AuthTokens
-  /**
-   * The features available in the request based on the customer's sourceID;
-   * `features`, `stats`, `logger`, `dataFeedCache`, and `transactionContext` and `stateContext` are for internal Twilio/Segment use only.
-   */
+  /** Engage internal use only. DO NOT USE. */
   features?: Features
+  /** Engage internal use only. DO NOT USE. */
   statsContext?: StatsContext
+  /** Engage internal use only. DO NOT USE. */
   logger?: Logger
-  dataFeedCache?: DataFeedCache
+  /** Engage internal use only. DO NOT USE. */
+  engageDestinationCache?: EngageDestinationCache
+  /** Engage internal use only. DO NOT USE. */
   transactionContext?: TransactionContext
+  /** Engage internal use only. DO NOT USE. */
   stateContext?: StateContext
 }
 
@@ -77,7 +79,7 @@ class TestDestination<T, AudienceSettings = any> extends Destination<T, Audience
       features,
       statsContext,
       logger,
-      dataFeedCache,
+      engageDestinationCache,
       transactionContext,
       stateContext
     }: InputData<T>
@@ -99,7 +101,7 @@ class TestDestination<T, AudienceSettings = any> extends Destination<T, Audience
       features: features ?? {},
       statsContext: statsContext ?? ({} as StatsContext),
       logger: logger ?? ({ info: noop, error: noop } as Logger),
-      dataFeedCache: dataFeedCache ?? ({} as DataFeedCache),
+      engageDestinationCache: engageDestinationCache ?? ({} as EngageDestinationCache),
       transactionContext: transactionContext ?? ({} as TransactionContext),
       stateContext: stateContext ?? ({} as StateContext)
     })
@@ -121,7 +123,7 @@ class TestDestination<T, AudienceSettings = any> extends Destination<T, Audience
       features,
       statsContext,
       logger,
-      dataFeedCache,
+      engageDestinationCache,
       transactionContext,
       stateContext
     }: Omit<InputData<T>, 'event'> & { events?: SegmentEvent[] }
@@ -147,7 +149,7 @@ class TestDestination<T, AudienceSettings = any> extends Destination<T, Audience
       features: features ?? {},
       statsContext: statsContext ?? ({} as StatsContext),
       logger: logger ?? ({} as Logger),
-      dataFeedCache: dataFeedCache ?? ({} as DataFeedCache),
+      engageDestinationCache: engageDestinationCache ?? ({} as EngageDestinationCache),
       transactionContext: transactionContext ?? ({} as TransactionContext),
       stateContext: stateContext ?? ({} as StateContext)
     })

packages/core/src/destination-kit/action.ts

@@ -12,7 +12,7 @@ import { validateSchema } from '../schema-validation'
 import { AuthTokens } from './parse-settings'
 import { IntegrationError } from '../errors'
 import { removeEmptyValues } from '../remove-empty-values'
-import { Logger, StatsContext, TransactionContext, StateContext, DataFeedCache } from './index'
+import { Logger, StatsContext, TransactionContext, StateContext, EngageDestinationCache } from './index'
 
 type MaybePromise<T> = T | Promise<T>
 type RequestClient = ReturnType<typeof createRequestClient>
@@ -59,11 +59,13 @@ type GenericActionHookBundle = {
   }
 }
 
-// eslint-disable-next-line @typescript-eslint/no-explicit-any
 export interface ActionDefinition<
   Settings,
+  // eslint-disable-next-line @typescript-eslint/no-explicit-any
   Payload = any,
+  // eslint-disable-next-line @typescript-eslint/no-explicit-any
   AudienceSettings = any,
+  // eslint-disable-next-line @typescript-eslint/no-explicit-any
   GeneratedActionHookBundle extends GenericActionHookBundle = any
 > extends BaseActionDefinition {
   /**
@@ -166,7 +168,7 @@ interface ExecuteBundle<T = unknown, Data = unknown, AudienceSettings = any, Act
   features?: Features | undefined
   statsContext?: StatsContext | undefined
   logger?: Logger | undefined
-  dataFeedCache?: DataFeedCache | undefined
+  engageDestinationCache?: EngageDestinationCache | undefined
   transactionContext?: TransactionContext
   stateContext?: StateContext
 }
@@ -274,7 +276,7 @@ export class Action<Settings, Payload extends JSONLikeObject, AudienceSettings =
       features: bundle.features,
       statsContext: bundle.statsContext,
       logger: bundle.logger,
-      dataFeedCache: bundle.dataFeedCache,
+      engageDestinationCache: bundle.engageDestinationCache,
       transactionContext: bundle.transactionContext,
       stateContext: bundle.stateContext,
       audienceSettings: bundle.audienceSettings,
@@ -339,7 +341,7 @@ export class Action<Settings, Payload extends JSONLikeObject, AudienceSettings =
         features: bundle.features,
         statsContext: bundle.statsContext,
         logger: bundle.logger,
-        dataFeedCache: bundle.dataFeedCache,
+        engageDestinationCache: bundle.engageDestinationCache,
         transactionContext: bundle.transactionContext,
         stateContext: bundle.stateContext,
         hookOutputs

packages/core/src/destination-kit/index.ts

@@ -279,7 +279,8 @@ interface EventInput<Settings> {
   readonly features?: Features
   readonly statsContext?: StatsContext
   readonly logger?: Logger
-  readonly dataFeedCache?: DataFeedCache
+  /** Engage internal use only. DO NOT USE. */
+  readonly engageDestinationCache?: EngageDestinationCache
   readonly transactionContext?: TransactionContext
   readonly stateContext?: StateContext
 }
@@ -294,7 +295,8 @@ interface BatchEventInput<Settings> {
   readonly features?: Features
   readonly statsContext?: StatsContext
   readonly logger?: Logger
-  readonly dataFeedCache?: DataFeedCache
+  /** Engage internal use only. DO NOT USE. */
+  readonly engageDestinationCache?: EngageDestinationCache
   readonly transactionContext?: TransactionContext
   readonly stateContext?: StateContext
 }
@@ -310,7 +312,8 @@ interface OnEventOptions {
   features?: Features
   statsContext?: StatsContext
   logger?: Logger
-  readonly dataFeedCache?: DataFeedCache
+  /** Engage internal use only. DO NOT USE. */
+  readonly engageDestinationCache?: EngageDestinationCache
   transactionContext?: TransactionContext
   stateContext?: StateContext
   /** Handler to perform synchronization. If set, the refresh access token method will be synchronized across
@@ -364,11 +367,11 @@ export interface Logger {
   withTags(extraTags: any): void
 }
 
-export interface DataFeedCache {
-  setRequestResponse(requestId: string, response: string, expiryInSeconds: number): Promise<void>
-  getRequestResponse(requestId: string): Promise<string | null>
-  maxResponseSizeBytes: number
-  maxExpirySeconds: number
+export interface EngageDestinationCache {
+  getByKey: (key: string) => Promise<string | null>
+  readonly maxExpirySeconds: number
+  readonly maxResponseSizeBytes: number
+  setByKey: (key: string, value: string, expiryInSeconds?: number) => Promise<void>
 }
 
 export class Destination<Settings = JSONObject, AudienceSettings = JSONObject> {
@@ -401,8 +404,8 @@ export class Destination<Settings = JSONObject, AudienceSettings = JSONObject> {
       this.settingsSchema = fieldsToJsonSchema(this.authentication.fields)
     }
 
-    for (const action of Object.keys(destination.actions)) {
-      this.partnerAction(action, destination.actions[action])
+    for (const [name, action] of Object.entries(destination.actions)) {
+      this.partnerAction(name, action)
     }
   }
 
@@ -547,7 +550,7 @@ export class Destination<Settings = JSONObject, AudienceSettings = JSONObject> {
       features,
       statsContext,
       logger,
-      dataFeedCache,
+      engageDestinationCache,
       transactionContext,
       stateContext
     }: EventInput<Settings>
@@ -571,7 +574,7 @@ export class Destination<Settings = JSONObject, AudienceSettings = JSONObject> {
       features,
       statsContext,
       logger,
-      dataFeedCache,
+      engageDestinationCache,
       transactionContext,
       stateContext
     })
@@ -587,7 +590,7 @@ export class Destination<Settings = JSONObject, AudienceSettings = JSONObject> {
       features,
       statsContext,
       logger,
-      dataFeedCache,
+      engageDestinationCache,
       transactionContext,
       stateContext
     }: BatchEventInput<Settings>
@@ -612,7 +615,7 @@ export class Destination<Settings = JSONObject, AudienceSettings = JSONObject> {
       features,
       statsContext,
       logger,
-      dataFeedCache,
+      engageDestinationCache,
       transactionContext,
       stateContext
     })
@@ -651,7 +654,8 @@ export class Destination<Settings = JSONObject, AudienceSettings = JSONObject> {
       features: options?.features || {},
       statsContext: options?.statsContext || ({} as StatsContext),
       logger: options?.logger,
-      dataFeedCache: options?.dataFeedCache,
+      /** Engage internal use only. DO NOT USE. */
+      engageDestinationCache: options?.engageDestinationCache,
       transactionContext: options?.transactionContext,
       stateContext: options?.stateContext
     }

packages/core/src/destination-kit/types.ts

@@ -1,4 +1,4 @@
-import { StateContext, Logger, StatsContext, TransactionContext, DataFeedCache, ActionHookType } from './index'
+import { StateContext, Logger, StatsContext, TransactionContext, EngageDestinationCache, ActionHookType } from './index'
 import type { RequestOptions } from '../request-client'
 import type { JSONObject } from '../json-object'
 import { AuthTokens } from './parse-settings'
@@ -20,7 +20,9 @@ export interface ExecuteInput<
   Settings,
   Payload,
   AudienceSettings = unknown,
+  /* eslint-disable-next-line @typescript-eslint/no-explicit-any -- Expected any. */
   ActionHookInputs = any,
+  /* eslint-disable-next-line @typescript-eslint/no-explicit-any -- Expected any. */
   ActionHookOutputs = any
 > {
   /** The subscription mapping definition */
@@ -39,15 +41,17 @@ export interface ExecuteInput<
   page?: string
   /** The data needed in OAuth requests */
   readonly auth?: AuthTokens
-  /**
-   * The features available in the request based on the customer's sourceID;
-   * `features`,`stats`, `logger` , `transactionContext` and `stateContext` are for internal Twilio/Segment use only.
-   */
+  /** Engage internal use only. DO NOT USE. */
   readonly features?: Features
+  /** Engage internal use only. DO NOT USE. */
   readonly statsContext?: StatsContext
+  /** Engage internal use only. DO NOT USE. */
   readonly logger?: Logger
-  readonly dataFeedCache?: DataFeedCache
+  /** Engage internal use only. DO NOT USE. */
+  readonly engageDestinationCache?: EngageDestinationCache
+  /** Engage internal use only. DO NOT USE. */
   readonly transactionContext?: TransactionContext
+  /** Engage internal use only. DO NOT USE. */
   readonly stateContext?: StateContext
 }
 

packages/destination-actions/src/destinations/engage/sendgrid/sendEmail/SendEmailPerformer.ts

@@ -361,7 +361,7 @@ export class SendEmailPerformer extends MessageSendPerformer<Settings, Payload>
           this.tags,
           this.settings,
           this.logger.loggerClient,
-          this.dataFeedCache
+          this.engageDestinationCache
         )
         return { name: apiLookup.name, data }
       })

packages/destination-actions/src/destinations/engage/utils/EngageActionPerformer.ts

@@ -19,7 +19,7 @@ import truncate from 'lodash/truncate'
 export abstract class EngageActionPerformer<TSettings = any, TPayload = any, TReturn = any> {
   readonly logger: EngageLogger = new EngageLogger(this)
   readonly statsClient: EngageStats = new EngageStats(this)
-  readonly dataFeedCache = this.executeInput.dataFeedCache
+  readonly engageDestinationCache = this.executeInput.engageDestinationCache
   readonly currentOperation: OperationContext | undefined
 
   readonly payload: TPayload

packages/destination-actions/src/destinations/engage/utils/MessageSendPerformer.ts

@@ -91,9 +91,86 @@ export abstract class MessageSendPerformer<
   TSettings extends MessageSettingsBase,
   TPayload extends MessagePayloadBase
 > extends EngageActionPerformer<TSettings, TPayload> {
+  /**
+   * Cache the response from the server based on messageId and recipientId.
+   * Only messages with valid messageIds, recipientIds, and successful or non
+   * retryable errors are cacheable. All other errors will be ignored.
+   *
+   * @param response The response from the server.
+   * @param messageId The messageId of the message.
+   * @param recipientId The recipientId of the message.
+   * @returns The response from the cache.
+   */
+  @track()
+  async cacheResponse(response: Response, messageId?: string, recipientId?: string) {
+    if (!messageId || !recipientId || !this.engageDestinationCache) {
+      return
+    }
+    // Check if the response is successful.
+    if (response.ok) {
+      const body = await response.text()
+      await this.engageDestinationCache.setByKey(messageId + recipientId, body)
+      // If the response is not retryable, cache the error.
+    } else if (!isRetryableError(response)) {
+      const error = await response.text()
+      await this.engageDestinationCache.setByKey(messageId + recipientId, error)
+    }
+  }
+
+  /**
+   * Read the response from the cache based on messageId and recipientId.
+   *
+   * @param messageId The messageId of the message.
+   * @param recipientId The recipientId of the message.
+   * @returns The response from the cache.
+   */
+  @track()
+  async readCache(messageId?: string, recipientId?: string) {
+    if (!messageId || !recipientId || !this.engageDestinationCache) {
+      return
+    }
+    return this.engageDestinationCache.getByKey(messageId + recipientId)
+  }
+
+  @track()
   async doPerform() {
+    /**
+     * Internal function to process a recepient.
+     *
+     * If a cache exists, it will check to see if the messageId, and recipientId
+     * has already been attempted. The response will be stored if the error
+     * returned from the server is a non retryable error or if the response is
+     * successful.
+     *
+     * @param recepient The recepient to process.
+     * @returns The response from the server.
+     */
+    const processRecipient = async (recepient: ExtId<TPayload>) => {
+      const messageId = (this.executeInput as any)['rawData']?.messageId
+      const recipientId = recepient.id
+      let cachedResponse
+      try {
+        cachedResponse = await this.readCache(messageId, recipientId)
+        if (cachedResponse) {
+          return cachedResponse
+        }
+      } catch (error) {
+        this.logError(`Failed to read cache for messageId: ${messageId}, recipientId: ${recipientId}`, error)
+      }
+
+      const response: Response = await this.sendToRecepient(recepient)
+
+      // This is a non blocking promise.
+      // Let the cacheResponse throw errors if it fails.
+      this.cacheResponse(response, messageId, recipientId).catch((error) => {
+        this.logError(`Failed to cache response for messageId: ${messageId}, recipientId: ${recipientId}`, error)
+      })
+
+      return response
+    }
+
     // sending messages and collecting results, including exceptions
-    const res = this.forAllRecepients((recepient) => this.sendToRecepient(recepient))
+    const res = this.forAllRecepients(processRecipient)
 
     if (this.executeInput.features) {
       this.logInfo('Feature flags:' + JSON.stringify(this.executeInput.features))
@@ -238,6 +315,12 @@ export abstract class MessageSendPerformer<
       : []
   }
 
+  /**
+   * Send message to recipient.
+   *
+   * @param recepient The recipient to send the message to.
+   * @returns The response from the server.
+   */
   abstract sendToRecepient(recepient: ExtId<TPayload>): any
 
   /**