[DC-789] adds subscription sync modes support in destination-kit

packages/core/src/__tests__/destination-kit.test.ts

@@ -111,6 +111,40 @@ const destinationWithOptions: DestinationDefinition<JSONObject> = {
   }
 }
 
+const destinationWithSyncMode: DestinationDefinition<JSONObject> = {
+  name: 'Actions Google Analytics 4',
+  mode: 'cloud',
+  actions: {
+    customEvent: {
+      title: 'Send a Custom Event',
+      description: 'Send events to a custom event in API',
+      defaultSubscription: 'type = "track"',
+      fields: {},
+      syncMode: {
+        default: 'add',
+        description: 'Select the sync mode for the subscription',
+        label: 'Sync Mode',
+        choices: [
+          {
+            label: 'Insert',
+            value: 'add'
+          },
+          {
+            label: 'Delete',
+            value: 'delete'
+          }
+        ]
+      },
+      perform: (_request, { syncMode }) => {
+        return ['this is a test', syncMode]
+      },
+      performBatch: (_request, { syncMode }) => {
+        return ['this is a test', syncMode]
+      }
+    }
+  }
+}
+
 describe('destination kit', () => {
   describe('event validations', () => {
     test('should return `invalid subscription` when sending an empty subscribe', async () => {
@@ -275,6 +309,55 @@ describe('destination kit', () => {
         { output: 'Action Executed', data: ['this is a test', {}] }
       ])
     })
+
+    test('should inject the syncMode value in the perform handler', async () => {
+      const destinationTest = new Destination(destinationWithSyncMode)
+      const testEvent: SegmentEvent = { type: 'track' }
+      const testSettings = {
+        apiSecret: 'test_key',
+        subscription: {
+          subscribe: 'type = "track"',
+          partnerAction: 'customEvent',
+          mapping: {
+            __segment_internal_sync_mode: 'add'
+          }
+        }
+      }
+
+      const res = await destinationTest.onEvent(testEvent, testSettings)
+
+      expect(res).toEqual([
+        { output: 'Mappings resolved' },
+        {
+          output: 'Action Executed',
+          data: ['this is a test', 'add']
+        }
+      ])
+    })
+
+    test('should inject the syncMode value in the performBatch handler', async () => {
+      const destinationTest = new Destination(destinationWithSyncMode)
+      const testEvent: SegmentEvent = { type: 'track' }
+      const testSettings = {
+        apiSecret: 'test_key',
+        subscription: {
+          subscribe: 'type = "track"',
+          partnerAction: 'customEvent',
+          mapping: {
+            __segment_internal_sync_mode: 'add'
+          }
+        }
+      }
+
+      const res = await destinationTest.onBatch([testEvent], testSettings)
+
+      expect(res).toEqual([
+        {
+          output: 'Action Executed',
+          data: ['this is a test', 'add']
+        }
+      ])
+    })
   })
 
   describe('refresh token', () => {

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

@@ -5,7 +5,16 @@ import { InputData, Features, transform, transformBatch } from '../mapping-kit'
 import { fieldsToJsonSchema } from './fields-to-jsonschema'
 import { Response } from '../fetch'
 import type { ModifiedResponse } from '../types'
-import type { DynamicFieldResponse, InputField, RequestExtension, ExecuteInput, Result } from './types'
+import type {
+  DynamicFieldResponse,
+  InputField,
+  RequestExtension,
+  ExecuteInput,
+  Result,
+  SyncMode,
+  SyncModeDefinition
+} from './types'
+import { syncModeTypes } from './types'
 import { NormalizedOptions } from '../request-client'
 import type { JSONSchema4 } from 'json-schema'
 import { validateSchema } from '../schema-validation'
@@ -93,6 +102,9 @@ export interface ActionDefinition<
       NonNullable<GeneratedActionHookBundle[K]>['inputs']
     >
   }
+
+  /** 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
@@ -171,6 +183,10 @@ interface ExecuteBundle<T = unknown, Data = unknown, AudienceSettings = any, Act
   stateContext?: StateContext
 }
 
+const isSyncMode = (value: unknown): value is SyncMode => {
+  return syncModeTypes.find((validValue) => value === validValue) !== undefined
+}
+
 /**
  * Action is the beginning step for all partner actions. Entrypoints always start with the
  * MapAndValidateInput step.
@@ -246,6 +262,11 @@ export class Action<Settings, Payload extends JSONLikeObject, AudienceSettings =
     // Remove empty values (`null`, `undefined`, `''`) when not explicitly accepted
     payload = removeEmptyValues(payload, this.schema, true) as Payload
 
+    // Remove internal hidden field
+    if (bundle.mapping && '__segment_internal_sync_mode' in bundle.mapping) {
+      delete payload['__segment_internal_sync_mode']
+    }
+
     // Validate the resolved payload against the schema
     if (this.schema) {
       const schemaKey = `${this.destinationName}:${this.definition.title}`
@@ -264,6 +285,8 @@ export class Action<Settings, Payload extends JSONLikeObject, AudienceSettings =
       }
     }
 
+    const syncMode = this.definition.syncMode ? bundle.mapping?.['__segment_internal_sync_mode'] : undefined
+
     // Construct the data bundle to send to an action
     const dataBundle = {
       rawData: bundle.data,
@@ -278,7 +301,8 @@ export class Action<Settings, Payload extends JSONLikeObject, AudienceSettings =
       transactionContext: bundle.transactionContext,
       stateContext: bundle.stateContext,
       audienceSettings: bundle.audienceSettings,
-      hookOutputs
+      hookOutputs,
+      syncMode: isSyncMode(syncMode) ? syncMode : undefined
     }
 
     // Construct the request client and perform the action
@@ -297,6 +321,15 @@ export class Action<Settings, Payload extends JSONLikeObject, AudienceSettings =
 
     let payloads = transformBatch(bundle.mapping, bundle.data) as Payload[]
 
+    // Remove internal hidden field
+    if (bundle.mapping && '__segment_internal_sync_mode' in bundle.mapping) {
+      for (const payload of payloads) {
+        if (payload) {
+          delete payload['__segment_internal_sync_mode']
+        }
+      }
+    }
+
     // Validate the resolved payloads against the schema
     if (this.schema) {
       const schema = this.schema
@@ -329,6 +362,7 @@ export class Action<Settings, Payload extends JSONLikeObject, AudienceSettings =
     }
 
     if (this.definition.performBatch) {
+      const syncMode = this.definition.syncMode ? bundle.mapping?.['__segment_internal_sync_mode'] : undefined
       const data = {
         rawData: bundle.data,
         rawMapping: bundle.mapping,
@@ -342,7 +376,8 @@ export class Action<Settings, Payload extends JSONLikeObject, AudienceSettings =
         dataFeedCache: bundle.dataFeedCache,
         transactionContext: bundle.transactionContext,
         stateContext: bundle.stateContext,
-        hookOutputs
+        hookOutputs,
+        syncMode: isSyncMode(syncMode) ? syncMode : undefined
       }
       const output = await this.performRequest(this.definition.performBatch, data)
       results[0].data = output as JSONObject

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

@@ -37,6 +37,8 @@ export interface ExecuteInput<
   hookOutputs?: Partial<Record<ActionHookType, ActionHookOutputs>>
   /** The page used in dynamic field requests */
   page?: string
+  /** The subscription sync mode */
+  syncMode?: SyncMode
   /** The data needed in OAuth requests */
   readonly auth?: AuthTokens
   /**
@@ -191,18 +193,35 @@ export interface InputField extends InputFieldJSONSchema {
   depends_on?: DependsOnConditions
 }
 
+/** Base interface for conditions  */
+interface BaseCondition {
+  operator: 'is' | 'is_not'
+}
+
 /**
- * A single condition defining whether a field should be shown.
+ * A single condition defining whether a field should be shown based on the value of the field specified by `fieldKey`.
  * fieldKey: The field key in the fields object to look at
  * operator: The operator to use when comparing the field value
  * value: The value we expect that field to have, if undefined, we will match based on whether the field contains a value or not
  */
-export interface Condition {
+export interface FieldCondition extends BaseCondition {
   fieldKey: string
-  operator: 'is' | 'is_not'
+  type?: 'field' // optional for backwards compatibility
   value: Omit<FieldValue, 'Directive'> | Array<Omit<FieldValue, 'Directive'>> | undefined
 }
 
+/**
+ * A single condition defining whether a field should be shown based on the current sync mode.
+ * operator: The operator to use when comparing the sync mode
+ * value: The value to compare against, if undefined, we will match based on whether a sync mode is set or not
+ */
+export interface SyncModeCondition extends BaseCondition {
+  type: 'syncMode'
+  value: SyncMode
+}
+
+export type Condition = FieldCondition | SyncModeCondition
+
 /**
  * If match is not set, it will default to 'all'
  * If match = 'any', then meeting any of the conditions defined will result in the field being shown.
@@ -263,3 +282,26 @@ export type Deletion<Settings, Return = any> = (
   request: RequestClient,
   data: ExecuteInput<Settings, DeletionPayload>
 ) => MaybePromise<Return>
+
+/** The supported sync mode values  */
+export const syncModeTypes = ['add', 'update', 'upsert', 'delete'] as const
+export type SyncMode = typeof syncModeTypes[number]
+
+export interface SyncModeOption {
+  /** The human-readable label for this option */
+  label: string
+  /** The value of this option */
+  value: SyncMode
+}
+
+/** An action sync mode definition */
+export interface SyncModeDefinition {
+  /** The default sync mode that will be selected */
+  default: SyncMode
+  /** The human-readable label for this setting  */
+  label: string
+  /** The human-friendly description of the setting */
+  description: string
+  /** The available sync mode choices */
+  choices: SyncModeOption[]
+}