STRATCONN 2552 - LiveRamp Audiences Destination (segmentio#1251)

* Initial scaffolding for LiveRamp

* temporarily remove warnings from scaffold

* STRATCONN-2553 Add mapping and destination settings for LiveRamp

* STRATCONN-2556 Add support for an S3 upload module

* improve formatting

* add support for arrays

* fix bugs

* add testcases

* fix documentation

* last fixes

* last fixes

* update lock file

* add missing description

* refactor S3 into its own action

* Revert "refactor S3 into its own action"

This reverts commit 508566e.

* fix enquoting to be global

package.json

@@ -33,6 +33,7 @@
   },
   "devDependencies": {
     "@peculiar/webcrypto": "^1.2.3",
+    "@types/aws4": "^1.11.2",
     "@types/chance": "^1.1.3",
     "@types/cors": "^2.8.12",
     "@types/express": "^4.17.16",
@@ -107,6 +108,7 @@
     }
   },
   "dependencies": {
+    "aws4": "^1.12.0",
     "chance": "^1.1.8"
   }
 }

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

@@ -98,6 +98,7 @@ register('643fdf094cfdbcf1bcccbc42', './usermaven')
 register('6440068936c4fb9f699b0645', './the-trade-desk-crm')
 register('6447ca8bfaa773a2ba0777a0', './tiktok-offline-conversions')
 register('645babd9362d97b777391325', './iterable')
+register('644ad6c6c4a87a3290450602', './liveramp-audiences')
 
 function register(id: MetadataId, destinationPath: string) {
   // eslint-disable-next-line @typescript-eslint/no-var-requires

packages/destination-actions/src/destinations/liveramp-audiences/audienceEntered/__tests__/__snapshots__/snapshot.test.ts.snap

@@ -0,0 +1,47 @@
+// Jest Snapshot v1, https://goo.gl/fbAQLP
+
+exports[`Testing snapshot for LiverampAudiences's audienceEntered destination action: all fields 1`] = `
+"audience_key,testType
+\\"PywYAY\\",\\"PywYAY\\""
+`;
+
+exports[`Testing snapshot for LiverampAudiences's audienceEntered destination action: enquotated indentifier data 1`] = `
+Array [
+  "\\"LCD TV,50\\"\\"\\"",
+  "\\"\\"\\"early-bird\\"\\" special\\"",
+  "\\"5'8\\"\\"\\"",
+]
+`;
+
+exports[`Testing snapshot for LiverampAudiences's audienceEntered destination action: required fields 1`] = `
+"audience_key,testType
+\\"PywYAY\\",\\"PywYAY\\""
+`;
+
+exports[`Testing snapshot for LiverampAudiences's audienceEntered destination action: required fields 2`] = `
+Headers {
+  Symbol(map): Object {
+    "authorization": Array [
+      "AWS4-HMAC-SHA256 Credential=PywYAY/19700101/PywYAY/s3/aws4_request, SignedHeaders=content-length;content-type;host;x-amz-content-sha256;x-amz-date, Signature=3273c678e9edaf86b444eeadfd6021f8814141ee94adcb7409c7931c1f286dff",
+    ],
+    "content-length": Array [
+      "39",
+    ],
+    "content-type": Array [
+      "application/x-www-form-urlencoded; charset=utf-8",
+    ],
+    "host": Array [
+      "PywYAY.s3.amazonaws.com",
+    ],
+    "user-agent": Array [
+      "Segment (Actions)",
+    ],
+    "x-amz-content-sha256": Array [
+      "c52b8202716894946461f92ea7d3ae902483d4c0339c06e616618d84bce9d642",
+    ],
+    "x-amz-date": Array [
+      "19700101T000012Z",
+    ],
+  },
+}
+`;

packages/destination-actions/src/destinations/liveramp-audiences/audienceEntered/__tests__/snapshot.test.ts

@@ -0,0 +1,91 @@
+import { createTestEvent, createTestIntegration } from '@segment/actions-core'
+import { generateTestData } from '../../../../lib/test-data'
+import destination from '../../index'
+import nock from 'nock'
+import { enquoteIdentifier } from '../operations'
+
+const testDestination = createTestIntegration(destination)
+const actionSlug = 'audienceEntered'
+const destinationSlug = 'LiverampAudiences'
+const seedName = `${destinationSlug}#${actionSlug}`
+
+describe(`Testing snapshot for ${destinationSlug}'s ${actionSlug} destination action:`, () => {
+  beforeAll(() => {
+    const mockDate = new Date(12345)
+    jest.spyOn(global, 'Date').mockImplementation(() => mockDate as unknown as string)
+  })
+
+  it('required fields', async () => {
+    const action = destination.actions[actionSlug]
+    const [eventData, settingsData] = generateTestData(seedName, destination, action, true)
+    eventData.delimiter = ','
+    settingsData.upload_mode = 'S3'
+    nock(/.*/).persist().get(/.*/).reply(200)
+    nock(/.*/).persist().post(/.*/).reply(200)
+    nock(/.*/).persist().put(/.*/).reply(200)
+
+    const event = createTestEvent({
+      properties: eventData
+    })
+
+    const responses = await testDestination.testAction(actionSlug, {
+      event: event,
+      mapping: event.properties,
+      settings: settingsData,
+      auth: undefined
+    })
+
+    const request = responses[0].request
+    const rawBody = await request.text()
+
+    try {
+      const json = JSON.parse(rawBody)
+      expect(json).toMatchSnapshot()
+      return
+    } catch (err) {
+      expect(rawBody).toMatchSnapshot()
+    }
+
+    expect(request.headers).toMatchSnapshot()
+  })
+
+  it('all fields', async () => {
+    const action = destination.actions[actionSlug]
+    const [eventData, settingsData] = generateTestData(seedName, destination, action, false)
+    eventData.delimiter = ','
+    settingsData.upload_mode = 'S3'
+
+    nock(/.*/).persist().get(/.*/).reply(200)
+    nock(/.*/).persist().post(/.*/).reply(200)
+    nock(/.*/).persist().put(/.*/).reply(200)
+
+    const event = createTestEvent({
+      properties: eventData
+    })
+
+    const responses = await testDestination.testAction(actionSlug, {
+      event: event,
+      mapping: event.properties,
+      settings: settingsData,
+      auth: undefined
+    })
+
+    const request = responses[0].request
+    const rawBody = await request.text()
+
+    try {
+      const json = JSON.parse(rawBody)
+      expect(json).toMatchSnapshot()
+      return
+    } catch (err) {
+      expect(rawBody).toMatchSnapshot()
+    }
+  })
+
+  it('enquotated indentifier data', async () => {
+    const identifiers = [`LCD TV,50"`, `"early-bird" special`, `5'8"`]
+    const enquotedIdentifiers = identifiers.map(enquoteIdentifier)
+
+    expect(enquotedIdentifiers).toMatchSnapshot()
+  })
+})

packages/destination-actions/src/destinations/liveramp-audiences/audienceEntered/index.ts

@@ -0,0 +1,56 @@
+import { ActionDefinition } from '@segment/actions-core'
+import type { Settings } from '../generated-types'
+import type { Payload } from './generated-types'
+import { processData } from './operations'
+
+const action: ActionDefinition<Settings, Payload> = {
+  title: 'Audience Entered',
+  description: 'Uploads audience membership data to a file for LiveRamp ingestion.',
+  defaultSubscription: 'event = "Audience Entered"',
+  fields: {
+    audience_key: {
+      label: 'Audience Key',
+      description: 'Identifies the user within the entered audience.',
+      type: 'string',
+      required: true,
+      default: { '@path': '$.userId' }
+    },
+    identifier_data: {
+      label: 'Identifier Data',
+      description: `Additional data pertaining to the user.`,
+      type: 'object',
+      required: false,
+      defaultObjectUI: 'keyvalue:only',
+      default: { '@path': '$.context.traits' }
+    },
+    delimiter: {
+      label: 'Delimeter',
+      description: `Character used to separate tokens in the resulting file.`,
+      type: 'string',
+      required: true,
+      default: ','
+    },
+    audience_name: {
+      label: 'Audience name',
+      description: `Name of the audience the user has entered.`,
+      type: 'string',
+      required: true,
+      default: { '@path': '$.properties.audience_key' }
+    },
+    received_at: {
+      label: 'Received At',
+      description: `Datetime at which the event was received. Used to disambiguate the resulting file.`,
+      type: 'datetime',
+      required: true,
+      default: { '@path': '$.receivedAt' }
+    }
+  },
+  perform: async (request, { settings, payload }) => {
+    return processData(request, settings, [payload])
+  },
+  performBatch: (request, { settings, payload }) => {
+    return processData(request, settings, payload)
+  }
+}
+
+export default action

packages/destination-actions/src/destinations/liveramp-audiences/audienceEntered/operations.ts

@@ -0,0 +1,59 @@
+import { PayloadValidationError, RequestClient } from '@segment/actions-core'
+import type { Settings } from '../generated-types'
+import type { Payload } from './generated-types'
+import { uploadS3, validateS3 } from '../s3'
+
+async function processData(request: RequestClient, settings: Settings, payloads: Payload[]) {
+  // STRATCONN-2554: Add support for SFTP
+  if (settings.upload_mode == 'S3') {
+    validateS3(settings)
+  } else {
+    throw new PayloadValidationError(`Unrecognized upload mode: ${settings.upload_mode}`)
+  }
+  // STRATCONN-2584: error if less than 25 elements in payload
+
+  // Prepare header row. Expected format:
+  // liveramp_audience_key[1],identifier_data[0..n]
+  const rows = []
+  const headers = ['audience_key']
+  if (payloads[0].identifier_data) {
+    for (const identifier of Object.getOwnPropertyNames(payloads[0].identifier_data)) {
+      headers.push(identifier)
+    }
+  }
+  rows.push(headers.join(payloads[0].delimiter))
+
+  // Prepare data rows
+  for (const payload of payloads) {
+    const row = []
+    row.push(payload.audience_key)
+    if (payload.identifier_data) {
+      for (const identifier of Object.getOwnPropertyNames(payload.identifier_data)) {
+        row.push(payload.identifier_data[identifier] as string)
+      }
+    }
+    rows.push(row.map(enquoteIdentifier).join(payload.delimiter))
+  }
+
+  // STRATCONN-2584: verify multiple emails are handled
+  const filename = `${payloads[0].audience_name}_PII_${payloads[0].received_at}.csv`
+  const fileContent = rows.join('\n')
+
+  if (settings.upload_mode == 'S3') {
+    return await uploadS3(settings, filename, fileContent, request)
+  }
+}
+
+/*
+  To avoid collision with delimeters, we should surround identifiers with quotation marks.
+  https://docs.liveramp.com/connect/en/formatting-file-data.html#idm45998667347936
+
+  Examples:
+  LCD TV -> "LCD TV"
+  LCD TV,50" -> "LCD TV,50"""
+*/
+function enquoteIdentifier(identifier: string) {
+  return `"${identifier.replace(/"/g, '""')}"`
+}
+
+export { processData, enquoteIdentifier }