[Hubspot] Add de-duplication logic in Hubspot

packages/destination-actions/src/destinations/hubspot/upsertObject/__tests__/validate.test.ts

@@ -1,4 +1,4 @@
-import { validate } from '../functions/validation-functions'
+import { mergeAndDeduplicateById, validate } from '../functions/validation-functions'
 import { Payload } from '../generated-types'
 
 const payload: Payload = {
@@ -93,3 +93,111 @@ describe('Hubspot.upsertObject', () => {
     expect(validatedPayload).toEqual(expectedValidatedPayload)
   })
 })
+
+describe('mergeAndDeduplicateById', () => {
+  const basePayload = (overrides: Partial<Payload> = {}) => ({
+    object_details: {
+      id_field_name: 'email',
+      id_field_value: 'user@example.com',
+      object_type: 'contact',
+      ...((overrides.object_details as object) || {})
+    },
+    association_sync_mode: 'upsert',
+    enable_batching: true,
+    batch_size: 100,
+    properties: { foo: 'bar' },
+    sensitive_properties: { secret: '123' },
+    timestamp: '2024-01-01T00:00:00.000Z',
+    ...overrides
+  })
+
+  it('returns empty array for empty input', () => {
+    expect(mergeAndDeduplicateById([])).toEqual([])
+  })
+
+  it('deduplicates payloads by id_field_value', () => {
+    const payloads: Payload[] = [
+      basePayload({ properties: { a: 1 }, timestamp: '2024-01-01T00:00:00.000Z' }),
+      basePayload({ properties: { b: 2 }, timestamp: '2024-01-02T00:00:00.000Z' })
+    ]
+    const result = mergeAndDeduplicateById(payloads)
+    expect(result).toHaveLength(1)
+    expect(result[0].properties).toEqual({ a: 1, b: 2 })
+    expect(result[0]).not.toHaveProperty('timestamp')
+  })
+
+  it('merges properties preferring latest timestamp', () => {
+    const payloads: Payload[] = [
+      basePayload({ properties: { foo: 'old', bar: 'keep' }, timestamp: '2024-01-01T00:00:00.000Z' }),
+      basePayload({ properties: { foo: 'new' }, timestamp: '2024-01-03T00:00:00.000Z' })
+    ]
+    const result = mergeAndDeduplicateById(payloads)
+    expect(result[0].properties).toEqual({ foo: 'new', bar: 'keep' })
+  })
+
+  it('merges sensitive_properties preferring latest timestamp', () => {
+    const payloads: Payload[] = [
+      basePayload({ sensitive_properties: { secret: 'old', keep: 'yes' }, timestamp: '2024-01-01T00:00:00.000Z' }),
+      basePayload({ sensitive_properties: { secret: 'new' }, timestamp: '2024-01-03T00:00:00.000Z' })
+    ]
+    const result = mergeAndDeduplicateById(payloads)
+    expect(result[0].sensitive_properties).toEqual({ secret: 'new', keep: 'yes' })
+  })
+
+  it('merges associations', () => {
+    const assoc1 = {
+      id_field_name: 'email',
+      id_field_value: 'user@example.com',
+      object_type: 'contact' as string,
+      association_label: 'primary'
+    }
+    const assoc2 = {
+      id_field_name: 'email',
+      id_field_value: 'user@example.com',
+      object_type: 'contact' as string,
+      association_label: 'secondary'
+    }
+    const payloads: Payload[] = [
+      basePayload({ associations: [assoc1], timestamp: '2024-01-01T00:00:00.000Z' }),
+      basePayload({ associations: [assoc2], timestamp: '2024-01-02T00:00:00.000Z' })
+    ]
+    const result = mergeAndDeduplicateById(payloads)
+    expect(result[0].associations).toHaveLength(2)
+    expect(result[0].associations).toEqual(expect.arrayContaining([assoc1, assoc2]))
+  })
+
+  it('handles multiple ids and returns merged payloads for each', () => {
+    const payloads: Payload[] = [
+      basePayload({
+        object_details: { id_field_name: 'email', id_field_value: 'a@example.com', object_type: 'contact' },
+        properties: { foo: 1 }
+      }),
+      basePayload({
+        object_details: { id_field_name: 'email', id_field_value: 'b@example.com', object_type: 'contact' },
+        properties: { bar: 2 }
+      })
+    ]
+    const result = mergeAndDeduplicateById(payloads)
+    expect(result).toHaveLength(2)
+    expect(result.map((r) => r.object_details.id_field_value).sort()).toEqual(['a@example.com', 'b@example.com'])
+  })
+
+  it('skips payloads without id_field_value', () => {
+    const payloads: Payload[] = [
+      basePayload({ object_details: { id_field_name: 'email', id_field_value: '', object_type: 'contact' } }),
+      basePayload({ object_details: { id_field_name: 'email', id_field_value: null as any, object_type: 'contact' } }),
+      basePayload({
+        object_details: { id_field_name: 'email', id_field_value: 'valid@example.com', object_type: 'contact' }
+      })
+    ]
+    const result = mergeAndDeduplicateById(payloads)
+    expect(result).toHaveLength(1)
+    expect(result[0].object_details.id_field_value).toBe('valid@example.com')
+  })
+
+  it('removes timestamp from final output', () => {
+    const payloads: Payload[] = [basePayload({ timestamp: '2024-01-01T00:00:00.000Z' })]
+    const result = mergeAndDeduplicateById(payloads)
+    expect(result[0]).not.toHaveProperty('timestamp')
+  })
+})

packages/destination-actions/src/destinations/hubspot/upsertObject/common-fields.ts

@@ -137,5 +137,14 @@ export const commonFields: Record<string, InputField> = {
     required: true,
     unsafe_hidden: true,
     default: MAX_HUBSPOT_BATCH_SIZE
+  },
+  timestamp: {
+    label: 'Timestamp',
+    description:
+      'The time the event occurred. This will be used to de-duplicate the events before sending them to hubspot.',
+    type: 'string',
+    required: false,
+    default: { '@path': '$.timestamp' },
+    disabledInputMethods: ['literal', 'variable', 'function', 'freeform', 'enrichment']
   }
 }

packages/destination-actions/src/destinations/hubspot/upsertObject/constants.ts

@@ -18,3 +18,5 @@ export const SUPPORTED_HUBSPOT_OBJECT_TYPES = [
 ]
 
 export const MAX_HUBSPOT_BATCH_SIZE = 100
+
+export const HUBSPOT_DEDUPLICATION_FLAGON = 'hubspot-object-deduplication'

packages/destination-actions/src/destinations/hubspot/upsertObject/functions/validation-functions.ts

@@ -1,5 +1,6 @@
 import { PayloadValidationError } from '@segment/actions-core'
 import { Payload } from '../generated-types'
+import { Association } from '../types'
 
 export function validate(payloads: Payload[]): Payload[] {
   const length = payloads.length
@@ -85,3 +86,114 @@ function cleanProp(str: string): string {
   }
   return str
 }
+
+/**
+ * Merges an array of payloads by their unique `id_field_value`, deduplicating entries and merging their properties.
+ *
+ * For each unique ID:
+ * - Properties and sensitive properties are merged, preferring values from the payload with the latest timestamp.
+ * - Associations are merged.
+ * - The resulting payload for each ID contains merged properties, sensitive properties, and associations.
+ * - The `timestamp` field is used only for determining recency and is removed from the final output.
+ *
+ * @param payloads - An array of payloads to merge and deduplicate.
+ * @returns An array of merged and deduplicated payloads, with timestamps removed.
+ */
+export function mergeAndDeduplicateById(payloads: Payload[]): Payload[] {
+  const mergedMap = new Map<string, Payload>()
+
+  for (const incoming of payloads) {
+    const id = incoming.object_details?.id_field_value
+    if (!id) continue
+
+    const incomingTimestamp = incoming.timestamp ? new Date(incoming.timestamp).getTime() : 0
+
+    const existing = mergedMap.get(id)
+    if (!existing) {
+      mergedMap.set(id, { ...incoming })
+      continue
+    }
+
+    const existingTimestamp = existing.timestamp ? new Date(existing.timestamp).getTime() : 0
+
+    // Merge properties
+    const mergedProps: Record<string, unknown> = { ...existing.properties }
+    for (const [key, value] of Object.entries(incoming.properties || {})) {
+      if (!(key in mergedProps) || incomingTimestamp >= existingTimestamp) {
+        mergedProps[key] = value
+      }
+    }
+
+    // Merge sensitive_properties
+    const mergedSensProps: Record<string, unknown> = { ...existing.sensitive_properties }
+    for (const [key, value] of Object.entries(incoming.sensitive_properties || {})) {
+      if (!(key in mergedSensProps) || incomingTimestamp >= existingTimestamp) {
+        mergedSensProps[key] = value
+      }
+    }
+
+    // Merge associations
+    const existingAssociations = existing.associations || []
+    const incomingAssociations = incoming.associations || []
+
+    const associationKey = (assoc: Association) =>
+      `${assoc.object_type}|${assoc.association_label}|${assoc.id_field_name}|${assoc.id_field_value}`
+
+    const existingAssocMap = new Map(existingAssociations.map((a) => [associationKey(a), a]))
+
+    for (const assoc of incomingAssociations) {
+      const key = associationKey(assoc)
+      if (!existingAssocMap.has(key)) {
+        existingAssocMap.set(key, assoc)
+      }
+    }
+
+    const mergedAssociations = Array.from(existingAssocMap.values())
+
+    // Save merged record with the latest timestamp
+    mergedMap.set(id, {
+      ...existing,
+      properties: mergedProps,
+      sensitive_properties: mergedSensProps,
+      associations: mergedAssociations,
+      timestamp: incomingTimestamp >= existingTimestamp ? incoming.timestamp : existing.timestamp
+    })
+  }
+
+  // Final output with timestamp removed
+  return Array.from(mergedMap.values()).map(({ timestamp, ...rest }) => rest)
+}
+
+/**
+ * Ensures that each payload in the provided array has a valid timestamp.
+ * If a payload's timestamp is invalid, it is replaced with the provided fallback timestamp.
+ *
+ * @param payloads - An array of payload objects, each potentially containing a `timestamp` property.
+ * @param fallbackTimestamp - The timestamp string to use when a payload's timestamp is invalid.
+ * @returns A new array of payloads with valid timestamps.
+ */
+export function ensureValidTimestamps(payloads: Payload[], fallbackTimestamp?: string): Payload[] {
+  const safeFallback = isValidTimestamp(fallbackTimestamp) ? fallbackTimestamp : new Date().toISOString()
+
+  return payloads.map((p) => ({
+    ...p,
+    timestamp: isValidTimestamp(p.timestamp) ? p.timestamp : safeFallback
+  }))
+}
+
+/**
+ * Checks if the provided value is a valid timestamp.
+ *
+ * Accepts strings, numbers, or Date objects and verifies if they can be converted
+ * to a valid date. Returns `true` if the value represents a valid timestamp,
+ * otherwise returns `false`.
+ *
+ * @param ts - The value to validate as a timestamp.
+ * @returns `true` if `ts` is a valid timestamp, otherwise `false`.
+ */
+function isValidTimestamp(ts: unknown): ts is string | number | Date {
+  if (typeof ts === 'string' || typeof ts === 'number' || ts instanceof Date) {
+    return !isNaN(new Date(ts).getTime())
+  }
+  return false
+}

packages/destination-actions/src/destinations/hubspot/upsertObject/index.ts

@@ -1,13 +1,13 @@
-import { ActionDefinition, RequestClient, IntegrationError, StatsContext } from '@segment/actions-core'
+import { ActionDefinition, RequestClient, IntegrationError, StatsContext, Features } from '@segment/actions-core'
 import type { Settings } from '../generated-types'
 import type { Payload } from './generated-types'
 import { commonFields } from './common-fields'
 import { SubscriptionMetadata } from '@segment/actions-core/destination-kit'
 import { Client } from './client'
-import { AssociationSyncMode, SyncMode, SchemaMatch } from './types'
+import { AssociationSyncMode, SyncMode, SchemaMatch, RequestData } from './types'
 import { dynamicFields } from './functions/dynamic-field-functions'
 import { getSchemaFromCache, saveSchemaToCache } from './functions/cache-functions'
-import { validate } from './functions/validation-functions'
+import { ensureValidTimestamps, mergeAndDeduplicateById, validate } from './functions/validation-functions'
 import { objectSchema, compareSchemas } from './functions/schema-functions'
 import { sendFromRecords } from './functions/hubspot-record-functions'
 import {
@@ -16,6 +16,7 @@ import {
   sendAssociations
 } from './functions/hubspot-association-functions'
 import { getSchemaFromHubspot, createProperties } from './functions/hubspot-properties-functions'
+import { HUBSPOT_DEDUPLICATION_FLAGON } from './constants'
 
 const action: ActionDefinition<Settings, Payload> = {
   title: 'Custom Object V2',
@@ -39,9 +40,11 @@ const action: ActionDefinition<Settings, Payload> = {
     statsContext?.tags?.push('action:custom_object')
     return await send(request, [payload], syncMode as SyncMode, subscriptionMetadata, statsContext)
   },
-  performBatch: async (request, { payload, syncMode, subscriptionMetadata, statsContext }) => {
+  performBatch: async (request, data) => {
+    const requestData = data as RequestData<Settings, Payload[]>
+    const { payload, syncMode, subscriptionMetadata, statsContext, features, rawData } = requestData
     statsContext?.tags?.push('action:custom_object_batch')
-    return await send(request, payload, syncMode as SyncMode, subscriptionMetadata, statsContext)
+    return await send(request, payload, syncMode, subscriptionMetadata, statsContext, features, rawData?.timestamp)
   }
 }
 
@@ -50,8 +53,15 @@ const send = async (
   payloads: Payload[],
   syncMode: SyncMode,
   subscriptionMetadata?: SubscriptionMetadata,
-  statsContext?: StatsContext
+  statsContext?: StatsContext,
+  features?: Features,
+  fallbackTimestamp?: string
 ) => {
+  if (features && features[HUBSPOT_DEDUPLICATION_FLAGON] && (syncMode === 'upsert' || syncMode === 'update')) {
+    payloads = ensureValidTimestamps(payloads, fallbackTimestamp)
+    payloads = mergeAndDeduplicateById(payloads)
+  }
+
   const {
     object_details: { object_type: objectType, property_group: propertyGroup },
     association_sync_mode: assocationSyncMode

packages/destination-actions/src/destinations/hubspot/upsertObject/types.ts

@@ -1,4 +1,6 @@
+import { Features, StatsContext } from '@segment/actions-core/*'
 import { Payload } from './generated-types'
+import { SubscriptionMetadata } from '@segment/actions-core/destination-kit'
 
 export const SyncMode = {
   Upsert: 'upsert',
@@ -208,3 +210,23 @@ export interface CreatePropsReqItem {
   fieldType: string
   options?: Array<{ label: string; value: string; hidden: boolean; description: string; displayOrder: number }>
 }
+
+export interface RequestData<Settings, Payload> {
+  rawData: {
+    timestamp: string
+  }
+  payload: Payload
+  syncMode: SyncMode
+  subscriptionMetadata: SubscriptionMetadata
+  statsContext?: StatsContext
+  features?: Features
+  settings: Settings
+}
+
+export interface Association {
+  object_type?: string
+  association_label?: string
+  id_field_name?: string
+  id_field_value?: string
+  from_record_id?: string
+}