diff --git a/packages/core/src/__tests__/batching.test.ts b/packages/core/src/__tests__/batching.test.ts index ccb980f677..5dbdb9a32d 100644 --- a/packages/core/src/__tests__/batching.test.ts +++ b/packages/core/src/__tests__/batching.test.ts @@ -757,6 +757,66 @@ describe('Async Batching', () => { expect(data.audienceMembership).toEqual([true, false]) }) + test('forwards features to audience membership resolution', async () => { + const multiStatusResponse = new MultiStatusResponse() + multiStatusResponse.pushSuccessResponse({ status: 200, body: {}, sent: {} }) + + mockPerformBatch.mockResolvedValue({ + jobId: 'features-job', + status: 200, + multiStatusResponse + } as AsyncBatchResponse) + + // A journey_step event with no membership boolean only resolves to `true` when the + // legacy-journeys flag is passed through to resolveAudienceMembership. + const legacyJourneyEvent = createTestEvent({ + userId: 'user_123', + type: 'track', + event: 'Journey Step Entered', + context: { personas: { computation_class: 'journey_step', computation_key: 'step_1' } }, + properties: {} + }) + + const destination = new Destination(asyncBatchDestination) + await destination.executeAsyncBatch('asyncTestAction', { + events: [legacyJourneyEvent], + mapping: { user_id: { '@path': '$.userId' } }, + settings: {}, + features: { 'actions-legacy-journeys-audience-membership': true } + }) + + expect(mockPerformBatch.mock.calls[0][1].audienceMembership).toEqual([true]) + }) + + test('passes personasContext to performBatch', async () => { + const multiStatusResponse = new MultiStatusResponse() + multiStatusResponse.pushSuccessResponse({ status: 200, body: {}, sent: {} }) + + mockPerformBatch.mockResolvedValue({ + jobId: 'personas-job', + status: 200, + multiStatusResponse + } as AsyncBatchResponse) + + const personas = { computation_class: 'audience', computation_key: 'in_audience' } + const event = createTestEvent({ + userId: 'user_123', + type: 'track', + event: 'Audience Entered', + context: { personas }, + properties: { in_audience: true } + }) + + const destination = new Destination(asyncBatchDestination) + await destination.executeAsyncBatch('asyncTestAction', { + events: [event], + mapping: { user_id: { '@path': '$.userId' } }, + settings: {} + }) + + expect(mockPerformBatch.mock.calls[0][1].personasContext).toEqual(personas) + }) + test('does not throw and returns an empty multi-status response for an empty batch', async () => { const destination = new Destination(asyncBatchDestination) diff --git a/packages/core/src/destination-kit/action.ts b/packages/core/src/destination-kit/action.ts index edd6fb6d8f..3d7f5dce03 100644 --- a/packages/core/src/destination-kit/action.ts +++ b/packages/core/src/destination-kit/action.ts @@ -159,13 +159,14 @@ export type PollResponse = { multiStatusResponse?: MultiStatusResponse } -// eslint-disable-next-line @typescript-eslint/no-explicit-any -export interface ActionDefinition< +/** + * A subset of {@link BaseActionDefinition} that's common to both {@link ActionDefinition} and {@link AsyncActionDefinition}. + */ + +interface CloudActionDefinition< Settings, - // eslint-disable-next-line @typescript-eslint/no-explicit-any - Payload = any, - // eslint-disable-next-line @typescript-eslint/no-explicit-any - AudienceSettings = any, + Payload, + AudienceSettings, // eslint-disable-next-line @typescript-eslint/no-explicit-any GeneratedActionHookInputs = any, // eslint-disable-next-line @typescript-eslint/no-explicit-any @@ -199,12 +200,6 @@ export interface ActionDefinition< : never } - /** The operation to perform when this action is triggered */ - perform: RequestFn - - /** The operation to perform when this action is triggered for a batch of events */ - performBatch?: RequestFn - /** Hooks are triggered at some point in a mappings lifecycle. They may perform a request with the * destination using the provided inputs and return a response. The response may then optionally be stored * in the mapping for later use in the action. @@ -223,7 +218,8 @@ export interface ActionDefinition< syncMode?: SyncModeDefinition } -export interface AsyncActionDefinition< +// eslint-disable-next-line @typescript-eslint/no-explicit-any +export interface ActionDefinition< Settings, // eslint-disable-next-line @typescript-eslint/no-explicit-any Payload = any, @@ -233,58 +229,43 @@ export interface AsyncActionDefinition< GeneratedActionHookInputs = any, // eslint-disable-next-line @typescript-eslint/no-explicit-any GeneratedActionHookOutputs = any -> extends BaseActionDefinition { - /** - * A way to "register" dynamic fields. - * This is likely going to change as we productionalize the data model and definition object - */ - dynamicFields?: { - [K in keyof Payload]?: IsArray extends never - ? Payload[K] extends object | undefined - ? { - [ObjectProperty in keyof NonNullable | '__keys__' | '__values__']?: RequestFn< - Settings, - Payload, - DynamicFieldResponse, - AudienceSettings - > - } - : RequestFn - : IsArray extends object - ? { - [ObjectProperty in keyof NonNullable> | '__keys__' | '__values__']?: RequestFn< - Settings, - Payload, - DynamicFieldResponse, - AudienceSettings - > - } - : never - } +> extends CloudActionDefinition< + Settings, + Payload, + AudienceSettings, + GeneratedActionHookInputs, + GeneratedActionHookOutputs + > { + /** The operation to perform when this action is triggered */ + perform: RequestFn + + /** The operation to perform when this action is triggered for a batch of events */ + performBatch?: RequestFn +} +export interface AsyncActionDefinition< + 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 + GeneratedActionHookInputs = any, + // eslint-disable-next-line @typescript-eslint/no-explicit-any + GeneratedActionHookOutputs = any +> extends CloudActionDefinition< + Settings, + Payload, + AudienceSettings, + GeneratedActionHookInputs, + GeneratedActionHookOutputs + > { /** Async Actions don't support the perform operation, even in case of single payload it should always be handled as a batch */ /** The operation to perform when this action is triggered for a batch of events */ performBatch: RequestFn performPoll: RequestFn - - /** Hooks are triggered at some point in a mappings lifecycle. They may perform a request with the - * destination using the provided inputs and return a response. The response may then optionally be stored - * in the mapping for later use in the action. - */ - hooks?: { - [K in ActionHookType]?: ActionHookDefinition< - Settings, - Payload, - AudienceSettings, - NonNullable, - NonNullable - > - } - - /** The sync mode setting definition. This enables subscription sync mode selection when subscribing to this action. */ - syncMode?: SyncModeDefinition } export const hookTypeStrings = ['onMappingSave', 'retlOnMappingSave'] as const @@ -395,15 +376,26 @@ const isSyncMode = (value: unknown): value is SyncMode => { } /** - * Action is the beginning step for all partner actions. Entrypoints always start with the - * MapAndValidateInput step. + * CloudAction holds the fields and behavior shared by {@link Action} and {@link AsyncAction}: + * schema/hook-schema generation from the definition's fields, dynamic field resolution, hook + * execution, and request client construction. */ -export class Action extends EventEmitter { - readonly definition: ActionDefinition +abstract class CloudAction< + Settings, + Payload extends JSONLikeObject, + AudienceSettings = any, + Definition extends CloudActionDefinition< + Settings, + Payload, + AudienceSettings, + unknown, + unknown + > = CloudActionDefinition +> extends EventEmitter { + readonly definition: Definition readonly destinationName: string readonly schema?: JSONSchema4 readonly hookSchemas?: Record - readonly hasBatchSupport: boolean readonly hasHookSupport: boolean // Payloads may be any type so we use `any` explicitly here. // eslint-disable-next-line @typescript-eslint/no-explicit-any @@ -411,7 +403,7 @@ export class Action, + definition: Definition, // Payloads may be any type so we use `any` explicitly here. // eslint-disable-next-line @typescript-eslint/no-explicit-any extendRequest?: RequestExtension @@ -420,7 +412,6 @@ export class Action): Promise { - // TODO cleanup results... not sure it's even used - const results: Result[] = [] - - // Resolve/transform the mapping with the input data - let payload = transform(bundle.mapping, bundle.data, bundle.statsContext) as Payload - results.push({ output: 'Mappings resolved' }) - - // Remove empty values (`null`, `undefined`, `''`) when not explicitly accepted - payload = removeEmptyValues(payload, this.schema, true) as Payload - - // Validate the resolved payload against the schema - if (this.schema) { - const schemaKey = `${this.destinationName}:${this.definition.title}` - // AJV schema validator removes non mandatory fields post validation - // Refer https://ajv.js.org/guide/modifying-data.html#removing-additional-properties - // https://github.com/segmentio/action-destinations/blob/d245e420e56957e784c29b5c09d80f3e1e64e6c5/packages/core/src/schema-validation.ts#L21 - validateSchema(payload, this.schema, { - schemaKey, - statsContext: bundle.statsContext, - exempt: ['dynamicAuthSettings'] - }) - results.push({ output: 'Payload validated' }) - } - - let hookOutputs = {} - if (this.definition.hooks) { - for (const hookType in this.definition.hooks) { - const hookOutputValues = bundle.mapping?.[hookType] + /* + * Extract the dynamic field context and handler path from a field string. Examples: + * - "structured.first_name" => { dynamicHandlerPath: "structured.first_name" } + * - "unstructuredObject.testProperty" => { dynamicHandlerPath: "unstructuredObject.__values__", dynamicFieldContext: { selectedKey: "testProperty" } } + * - "structuredArray.[0].first_name" => { dynamicHandlerPath: "structuredArray.first_name", dynamicFieldContext: { selectedArrayIndex: 0 } } + */ + private extractFieldContextAndHandler(field: string): { + dynamicHandlerPath: string + dynamicFieldContext?: DynamicFieldContext + } { + const arrayRegex = /(.*)\.\[(\d+)\]\.(.*)/ + const objectRegex = /(.*)\.(.*)/ + let dynamicHandlerPath = field + let dynamicFieldContext: DynamicFieldContext | undefined - if (hookOutputValues) { - hookOutputs = { ...hookOutputs, [hookType]: hookOutputValues } + const match = arrayRegex.exec(field) || objectRegex.exec(field) + if (match) { + const [, parent, indexOrChild, child] = match + if (child) { + // It is an array, so we need to extract the index from parent.[index].child and call parent.child handler + dynamicFieldContext = { selectedArrayIndex: parseInt(indexOrChild, 10) } + dynamicHandlerPath = `${parent}.${child}` + } else { + // It is an object, if there is a dedicated fetcher for child we use it otherwise we use parent.__values__ + const parentFetcher = this.definition.dynamicFields?.[parent] + if (parentFetcher && !(indexOrChild in parentFetcher)) { + dynamicHandlerPath = `${parent}.__values__` + dynamicFieldContext = { selectedKey: indexOrChild } } } } - const syncModeVal = this.definition.syncMode ? bundle.mapping?.['__segment_internal_sync_mode'] : undefined - const syncMode = isSyncMode(syncModeVal) ? syncModeVal : undefined - const matchingKey = bundle.mapping?.['__segment_internal_matching_key'] - const audienceMembership = resolveAudienceMembership(bundle.data, syncMode, bundle.features) + return { dynamicHandlerPath, dynamicFieldContext } + } - // Construct the data bundle to send to an action - const dataBundle = { - rawData: bundle.data, - rawMapping: bundle.mapping, - settings: bundle.settings, - payload, - ...(typeof audienceMembership === 'boolean' ? { audienceMembership } : {}), - auth: bundle.auth, - features: bundle.features, - statsContext: bundle.statsContext, - personasContext: bundle.personasContext, - logger: bundle.logger, - engageDestinationCache: bundle.engageDestinationCache, - transactionContext: bundle.transactionContext, - stateContext: bundle.stateContext, - audienceSettings: bundle.audienceSettings, - hookOutputs, - syncMode, - matchingKey: matchingKey ? String(matchingKey) : undefined, - subscriptionMetadata: bundle.subscriptionMetadata, - signal: bundle?.signal + async executeDynamicField( + field: string, + data: ExecuteDynamicFieldInput, + /** + * The dynamicFn argument is optional since it is only used by dynamic hook input fields. (For now) + */ + dynamicFn?: RequestFn + ): Promise { + if (dynamicFn && typeof dynamicFn === 'function') { + return (await this.performRequest(dynamicFn, { ...data })) as DynamicFieldResponse } - // Construct the request client and perform the action - const output = await this.performRequest(this.definition.perform, dataBundle) - results.push({ data: output as JSONObject, output: 'Action Executed' }) - return results - } + const { dynamicHandlerPath, dynamicFieldContext } = this.extractFieldContextAndHandler(field) - async executeBatch(bundle: ExecuteBundle): Promise { - if (!this.hasBatchSupport) { - throw new IntegrationError('This action does not support batched requests.', 'NotImplemented', 501) + const fn = get>( + this.definition.dynamicFields, + dynamicHandlerPath + ) + + if (typeof fn !== 'function') { + return Promise.resolve({ + choices: [], + nextPage: '', + error: { + message: `No dynamic field named ${field} found.`, + code: '404' + } + }) } - const mapping: JSONObject = bundle.mapping + // fn will always be a dynamic field function, so we can safely cast it to DynamicFieldResponse + return (await this.performRequest(fn, { ...data, dynamicFieldContext })) as DynamicFieldResponse + } - let payloads = transformBatch(mapping, bundle.data, bundle.statsContext) as Payload[] - const batchPayloadLength = payloads.length + async executeHook( + hookType: ActionHookType, + data: ExecuteInput + ): Promise> { + if (!this.hasHookSupport) { + throw new IntegrationError('This action does not support any hooks.', 'NotImplemented', 501) + } + const hookFn = this.definition.hooks?.[hookType]?.performHook - const multiStatusResponse: ResultMultiStatusNode[] = [] - const invalidPayloadIndices = new Set() + if (!hookFn) { + throw new IntegrationError(`Missing implementation for hook: ${hookType}.`, 'NotImplemented', 501) + } - // Validate the resolved payloads against the schema - if (this.schema) { - const schema = this.schema - const validationOptions = { - schemaKey: `${this.destinationName}:${this.definition.title}`, - throwIfInvalid: true, - statsContext: bundle.statsContext, + if (this.hookSchemas?.[hookType]) { + const schema = this.hookSchemas[hookType] + validateSchema(data.hookInputs, schema, { exempt: ['dynamicAuthSettings'] - } + }) + } - // Filter out invalid payloads before sending them to the action - { - const filteredPayload: Payload[] = [] + return (await this.performRequest(hookFn, data)) as ActionHookResponse + } - for (let i = 0; i < payloads.length; i++) { - // Validate payload schema - const payload = removeEmptyValues(payloads[i], schema) as Payload - try { - // AJV schema validator only removes fields that are not defined in the schema (Refer ajv docs) + /** + * Perform a request using the definition's request client + * the given request function + * and given data bundle + */ + protected async performRequest< + T extends Payload | Payload[] | PollPayload, + M extends AudienceMembership | AudienceMembership[] + >( + requestFn: RequestFn, + data: ExecuteInput + ): Promise { + const requestClient = this.createRequestClient(data) + const response = await requestFn(requestClient, data) + return this.parseResponse(response) + } + + // eslint-disable-next-line @typescript-eslint/no-explicit-any + protected createRequestClient(data: ExecuteInput): RequestClient { + // TODO turn `extendRequest` into a beforeRequest hook + const options = this.extendRequest?.(data) ?? {} + return createRequestClient(options, { + afterResponse: [this.afterResponse.bind(this)], + statsContext: data.statsContext, + signal: data?.signal + }) + } + + // Keep track of the request(s) associated with a response + private afterResponse(request: Request, options: NormalizedOptions, response: Response) { + // TODO figure out the types here... + // eslint-disable-next-line @typescript-eslint/no-explicit-any + const modifiedResponse: any = response + modifiedResponse.request = request + modifiedResponse.options = options + + this.emit('response', modifiedResponse) + return modifiedResponse + } + + private parseResponse(response: unknown): unknown { + /** + * Try to use the parsed response `.data` or `.content` string + * @see {@link ../middleware/after-response/prepare-response.ts} + */ + + if (response instanceof Response) { + return (response as ModifiedResponse).data ?? (response as ModifiedResponse).content + } + + // otherwise, we don't really know what this is, so return as-is + return response + } +} + +/** + * Action is the beginning step for all partner actions. Entrypoints always start with the + * MapAndValidateInput step. + */ +export class Action extends CloudAction< + Settings, + Payload, + AudienceSettings, + ActionDefinition +> { + readonly hasBatchSupport: boolean + + constructor( + destinationName: string, + definition: ActionDefinition, + // Payloads may be any type so we use `any` explicitly here. + // eslint-disable-next-line @typescript-eslint/no-explicit-any + extendRequest?: RequestExtension + ) { + super(destinationName, definition, extendRequest) + this.hasBatchSupport = typeof definition.performBatch === 'function' + } + + async execute(bundle: ExecuteBundle): Promise { + // TODO cleanup results... not sure it's even used + const results: Result[] = [] + + // Resolve/transform the mapping with the input data + let payload = transform(bundle.mapping, bundle.data, bundle.statsContext) as Payload + results.push({ output: 'Mappings resolved' }) + + // Remove empty values (`null`, `undefined`, `''`) when not explicitly accepted + payload = removeEmptyValues(payload, this.schema, true) as Payload + + // Validate the resolved payload against the schema + if (this.schema) { + const schemaKey = `${this.destinationName}:${this.definition.title}` + // AJV schema validator removes non mandatory fields post validation + // Refer https://ajv.js.org/guide/modifying-data.html#removing-additional-properties + // https://github.com/segmentio/action-destinations/blob/d245e420e56957e784c29b5c09d80f3e1e64e6c5/packages/core/src/schema-validation.ts#L21 + validateSchema(payload, this.schema, { + schemaKey, + statsContext: bundle.statsContext, + exempt: ['dynamicAuthSettings'] + }) + results.push({ output: 'Payload validated' }) + } + + let hookOutputs = {} + if (this.definition.hooks) { + for (const hookType in this.definition.hooks) { + const hookOutputValues = bundle.mapping?.[hookType] + + if (hookOutputValues) { + hookOutputs = { ...hookOutputs, [hookType]: hookOutputValues } + } + } + } + + const syncModeVal = this.definition.syncMode ? bundle.mapping?.['__segment_internal_sync_mode'] : undefined + const syncMode = isSyncMode(syncModeVal) ? syncModeVal : undefined + const matchingKey = bundle.mapping?.['__segment_internal_matching_key'] + const audienceMembership = resolveAudienceMembership(bundle.data, syncMode, bundle.features) + + // Construct the data bundle to send to an action + const dataBundle = { + rawData: bundle.data, + rawMapping: bundle.mapping, + settings: bundle.settings, + payload, + ...(typeof audienceMembership === 'boolean' ? { audienceMembership } : {}), + auth: bundle.auth, + features: bundle.features, + statsContext: bundle.statsContext, + personasContext: bundle.personasContext, + logger: bundle.logger, + engageDestinationCache: bundle.engageDestinationCache, + transactionContext: bundle.transactionContext, + stateContext: bundle.stateContext, + audienceSettings: bundle.audienceSettings, + hookOutputs, + syncMode, + matchingKey: matchingKey ? String(matchingKey) : undefined, + subscriptionMetadata: bundle.subscriptionMetadata, + signal: bundle?.signal + } + // Construct the request client and perform the action + const output = await this.performRequest(this.definition.perform, dataBundle) + results.push({ data: output as JSONObject, output: 'Action Executed' }) + + return results + } + + async executeBatch(bundle: ExecuteBundle): Promise { + if (!this.hasBatchSupport) { + throw new IntegrationError('This action does not support batched requests.', 'NotImplemented', 501) + } + + const mapping: JSONObject = bundle.mapping + + let payloads = transformBatch(mapping, bundle.data, bundle.statsContext) as Payload[] + const batchPayloadLength = payloads.length + + const multiStatusResponse: ResultMultiStatusNode[] = [] + const invalidPayloadIndices = new Set() + + // Validate the resolved payloads against the schema + if (this.schema) { + const schema = this.schema + const validationOptions = { + schemaKey: `${this.destinationName}:${this.definition.title}`, + throwIfInvalid: true, + statsContext: bundle.statsContext, + exempt: ['dynamicAuthSettings'] + } + + // Filter out invalid payloads before sending them to the action + { + const filteredPayload: Payload[] = [] + + for (let i = 0; i < payloads.length; i++) { + // Validate payload schema + const payload = removeEmptyValues(payloads[i], schema) as Payload + try { + // AJV schema validator only removes fields that are not defined in the schema (Refer ajv docs) // Refer https://ajv.js.org/guide/modifying-data.html#removing-additional-properties // https://github.com/segmentio/action-destinations/blob/d245e420e56957e784c29b5c09d80f3e1e64e6c5/packages/core/src/schema-validation.ts#L21 validateSchema(payload, schema, validationOptions) @@ -740,149 +901,6 @@ export class Action { dynamicHandlerPath: "structured.first_name" } - * - "unstructuredObject.testProperty" => { dynamicHandlerPath: "unstructuredObject.__values__", dynamicFieldContext: { selectedKey: "testProperty" } } - * - "structuredArray.[0].first_name" => { dynamicHandlerPath: "structuredArray.first_name", dynamicFieldContext: { selectedArrayIndex: 0 } } - */ - private extractFieldContextAndHandler(field: string): { - dynamicHandlerPath: string - dynamicFieldContext?: DynamicFieldContext - } { - const arrayRegex = /(.*)\.\[(\d+)\]\.(.*)/ - const objectRegex = /(.*)\.(.*)/ - let dynamicHandlerPath = field - let dynamicFieldContext: DynamicFieldContext | undefined - - const match = arrayRegex.exec(field) || objectRegex.exec(field) - if (match) { - const [, parent, indexOrChild, child] = match - if (child) { - // It is an array, so we need to extract the index from parent.[index].child and call paret.child handler - dynamicFieldContext = { selectedArrayIndex: parseInt(indexOrChild, 10) } - dynamicHandlerPath = `${parent}.${child}` - } else { - // It is an object, if there is a dedicated fetcher for child we use it otherwise we use parent.__values__ - const parentFetcher = this.definition.dynamicFields?.[parent] - if (parentFetcher && !(indexOrChild in parentFetcher)) { - dynamicHandlerPath = `${parent}.__values__` - dynamicFieldContext = { selectedKey: indexOrChild } - } - } - } - - return { dynamicHandlerPath, dynamicFieldContext } - } - - async executeDynamicField( - field: string, - data: ExecuteDynamicFieldInput, - /** - * The dynamicFn argument is optional since it is only used by dynamic hook input fields. (For now) - */ - dynamicFn?: RequestFn - ): Promise { - if (dynamicFn && typeof dynamicFn === 'function') { - return (await this.performRequest(dynamicFn, { ...data })) as DynamicFieldResponse - } - - const { dynamicHandlerPath, dynamicFieldContext } = this.extractFieldContextAndHandler(field) - - const fn = get>( - this.definition.dynamicFields, - dynamicHandlerPath - ) - - if (typeof fn !== 'function') { - return Promise.resolve({ - choices: [], - nextPage: '', - error: { - message: `No dynamic field named ${field} found.`, - code: '404' - } - }) - } - - // fn will always be a dynamic field function, so we can safely cast it to DynamicFieldResponse - return (await this.performRequest(fn, { ...data, dynamicFieldContext })) as DynamicFieldResponse - } - - async executeHook( - hookType: ActionHookType, - data: ExecuteInput - ): Promise> { - if (!this.hasHookSupport) { - throw new IntegrationError('This action does not support any hooks.', 'NotImplemented', 501) - } - const hookFn = this.definition.hooks?.[hookType]?.performHook - - if (!hookFn) { - throw new IntegrationError(`Missing implementation for hook: ${hookType}.`, 'NotImplemented', 501) - } - - if (this.hookSchemas?.[hookType]) { - const schema = this.hookSchemas[hookType] - validateSchema(data.hookInputs, schema, { - exempt: ['dynamicAuthSettings'] - }) - } - - return (await this.performRequest(hookFn, data)) as ActionHookResponse - } - - /** - * Perform a request using the definition's request client - * the given request function - * and given data bundle - */ - private async performRequest( - requestFn: RequestFn, - data: ExecuteInput - ): Promise { - const requestClient = this.createRequestClient(data) - const response = await requestFn(requestClient, data) - return this.parseResponse(response) - } - - // eslint-disable-next-line @typescript-eslint/no-explicit-any - private createRequestClient(data: ExecuteInput): RequestClient { - // TODO turn `extendRequest` into a beforeRequest hook - const options = this.extendRequest?.(data) ?? {} - return createRequestClient(options, { - afterResponse: [this.afterResponse.bind(this)], - statsContext: data.statsContext, - signal: data?.signal - }) - } - - // Keep track of the request(s) associated with a response - private afterResponse(request: Request, options: NormalizedOptions, response: Response) { - // TODO figure out the types here... - // eslint-disable-next-line @typescript-eslint/no-explicit-any - const modifiedResponse: any = response - modifiedResponse.request = request - modifiedResponse.options = options - - this.emit('response', modifiedResponse) - return modifiedResponse - } - - private parseResponse(response: unknown): unknown { - /** - * Try to use the parsed response `.data` or `.content` string - * @see {@link ../middleware/after-response/prepare-response.ts} - */ - - if (response instanceof Response) { - return (response as ModifiedResponse).data ?? (response as ModifiedResponse).content - } - - // otherwise, we don't really know what this is, so return as-is - return response - } - private fillMultiStatusResponse(input: FillMultiStatusResponseInput) { const { multiStatusResponse, batchPayloadLength, status, body, filteredPayloads } = input @@ -932,62 +950,12 @@ export class ActionDestinationErrorResponse { * It does not support single-event perform operations - all events must be processed as batches. * It includes support for polling to check the status of async batch operations. */ -export class AsyncAction extends EventEmitter { - readonly definition: AsyncActionDefinition - readonly destinationName: string - readonly schema?: JSONSchema4 - readonly hookSchemas?: Record - readonly hasHookSupport: boolean - // eslint-disable-next-line @typescript-eslint/no-explicit-any - private extendRequest: RequestExtension | undefined - - constructor( - destinationName: string, - definition: AsyncActionDefinition, - // eslint-disable-next-line @typescript-eslint/no-explicit-any - extendRequest?: RequestExtension - ) { - super() - this.definition = definition - this.destinationName = destinationName - this.extendRequest = extendRequest - this.hasHookSupport = definition.hooks !== undefined - // Generate json schema based on the field definitions - if (Object.keys(definition.fields ?? {}).length) { - this.schema = fieldsToJsonSchema(definition.fields) - } - // Generate a json schema for each defined hook based on the field definitions - if (definition.hooks) { - for (const hookName in definition.hooks) { - const hook = definition.hooks[hookName as ActionHookType] - if (hook?.inputFields) { - if (!this.hookSchemas) { - this.hookSchemas = {} - } - - const castedInputFields: Record = {} - for (const key in hook.inputFields) { - const field = hook.inputFields[key] - - if (field.dynamic) { - castedInputFields[key] = { - ...field, - dynamic: true - } - } else { - castedInputFields[key] = { - ...field, - dynamic: false - } - } - } - - this.hookSchemas[hookName] = fieldsToJsonSchema(castedInputFields) - } - } - } - } - +export class AsyncAction extends CloudAction< + Settings, + Payload, + AudienceSettings, + AsyncActionDefinition +> { async executeBatch(bundle: ExecuteBundle): Promise { const mapping: JSONObject = bundle.mapping @@ -1061,7 +1029,7 @@ export class AsyncAction resolveAudienceMembership(d, syncMode)) + .map((d) => resolveAudienceMembership(d, syncMode, bundle.features)) .filter((_, i) => !invalidPayloadIndices.has(i)) const data = { @@ -1074,6 +1042,7 @@ export class AsyncAction, - dynamicFn?: RequestFn - ): Promise { - if (dynamicFn && typeof dynamicFn === 'function') { - return (await this.performRequest(dynamicFn, { ...data })) as DynamicFieldResponse - } - - const { dynamicHandlerPath, dynamicFieldContext } = this.extractFieldContextAndHandler(field) - - const fn = get>( - this.definition.dynamicFields, - dynamicHandlerPath - ) - - if (typeof fn !== 'function') { - return Promise.resolve({ - choices: [], - nextPage: '', - error: { - message: `No dynamic field named ${field} found.`, - code: '404' - } - }) - } - - return (await this.performRequest(fn, { ...data, dynamicFieldContext })) as DynamicFieldResponse - } - - async executeHook( - hookType: ActionHookType, - data: ExecuteInput - ): Promise> { - if (!this.hasHookSupport) { - throw new IntegrationError('This action does not support any hooks.', 'NotImplemented', 501) - } - const hookFn = this.definition.hooks?.[hookType]?.performHook - - if (!hookFn) { - throw new IntegrationError(`Missing implementation for hook: ${hookType}.`, 'NotImplemented', 501) - } - - if (this.hookSchemas?.[hookType]) { - const schema = this.hookSchemas[hookType] - validateSchema(data.hookInputs, schema, { - exempt: ['dynamicAuthSettings'] - }) - } - - return (await this.performRequest(hookFn, data)) as ActionHookResponse - } - - private async performRequest< - T extends Payload | Payload[] | PollPayload, - M extends AudienceMembership | AudienceMembership[] - >( - requestFn: RequestFn, - data: ExecuteInput - ): Promise { - const requestClient = this.createRequestClient(data) - const response = await requestFn(requestClient, data) - return this.parseResponse(response) - } - - // eslint-disable-next-line @typescript-eslint/no-explicit-any - private createRequestClient(data: ExecuteInput): RequestClient { - const options = this.extendRequest?.(data) ?? {} - return createRequestClient(options, { - afterResponse: [this.afterResponse.bind(this)], - statsContext: data.statsContext, - signal: data?.signal - }) - } - - private afterResponse(request: Request, options: NormalizedOptions, response: Response) { - // eslint-disable-next-line @typescript-eslint/no-explicit-any - const modifiedResponse: any = response - modifiedResponse.request = request - modifiedResponse.options = options - - this.emit('response', modifiedResponse) - return modifiedResponse - } - - private parseResponse(response: unknown): unknown { - if (response instanceof Response) { - return (response as ModifiedResponse).data ?? (response as ModifiedResponse).content - } - - return response - } - private parseBatchError( error: unknown, input: { diff --git a/packages/core/src/destination-kit/index.ts b/packages/core/src/destination-kit/index.ts index a2ca63b73b..f3682f7cb5 100644 --- a/packages/core/src/destination-kit/index.ts +++ b/packages/core/src/destination-kit/index.ts @@ -805,6 +805,8 @@ export class Destination { auth: auth ?? getAuthData(settings as unknown as JSONObject), features, statsContext, + // All events in a batch share the same personas context because batching is keyed on audience/computation. + personasContext: events[0]?.context?.personas as Personas | undefined, logger, engageDestinationCache, transactionContext,