feat: add support for identity stitching for shopify pixel flow (#3818)

src/util/prometheus.js

@@ -439,6 +439,24 @@ class Prometheus {
         type: 'counter',
         labelNames: ['writeKey', 'source', 'shopifyTopic'],
       },
+      {
+        name: 'shopify_pixel_cart_token_not_found',
+        help: 'shopify_pixel_cart_token_not_found',
+        type: 'counter',
+        labelNames: ['event', 'writeKey'],
+      },
+      {
+        name: 'shopify_pixel_cart_token_set',
+        help: 'shopify_pixel_cart_token_set',
+        type: 'counter',
+        labelNames: ['event', 'writeKey'],
+      },
+      {
+        name: 'shopify_pixel_cart_token_redis_error',
+        help: 'shopify_pixel_cart_token_redis_error',
+        type: 'counter',
+        labelNames: ['event', 'writeKey'],
+      },
       {
         name: 'outgoing_request_count',
         help: 'Outgoing HTTP requests count',

src/v0/sources/shopify/util.js

@@ -124,11 +124,11 @@ const extractEmailFromPayload = (event) => {
 };
 
 const getCartToken = (message) => {
-  const { event, properties } = message;
+  const { event, properties, context } = message;
   if (event === SHOPIFY_TRACK_MAP.carts_update) {
     return properties?.id || properties?.token;
   }
-  return properties?.cart_token || null;
+  return properties?.cart_token || context?.cart_token || null;
 };
 
 /**

src/v1/sources/shopify/config.js

@@ -1,6 +1,8 @@
 const path = require('path');
 const fs = require('fs');
 
+const commonCartTokenLocation = 'context.document.location.pathname';
+
 const PIXEL_EVENT_TOPICS = {
   CART_VIEWED: 'cart_viewed',
   PRODUCT_ADDED_TO_CART: 'product_added_to_cart',
@@ -61,6 +63,16 @@ const checkoutStartedCompletedEventMappingJSON = JSON.parse(
   ),
 );
 
+const pixelEventToCartTokenLocationMapping = {
+  cart_viewed: 'properties.cart_id',
+  checkout_address_info_submitted: commonCartTokenLocation,
+  checkout_contact_info_submitted: commonCartTokenLocation,
+  checkout_shipping_info_submitted: commonCartTokenLocation,
+  payment_info_submitted: commonCartTokenLocation,
+  checkout_started: commonCartTokenLocation,
+  checkout_completed: commonCartTokenLocation,
+};
+
 const INTEGERATION = 'SHOPIFY';
 
 module.exports = {
@@ -73,4 +85,5 @@ module.exports = {
   productViewedEventMappingJSON,
   productToCartEventMappingJSON,
   checkoutStartedCompletedEventMappingJSON,
+  pixelEventToCartTokenLocationMapping,
 };

src/v1/sources/shopify/pixelTransform.js

@@ -1,6 +1,11 @@
+/* eslint-disable no-param-reassign */
+// eslint-disable-next-line @typescript-eslint/naming-convention
+const _ = require('lodash');
+const { isDefinedNotNullNotEmpty } = require('@rudderstack/integrations-lib');
 const stats = require('../../../util/stats');
 const logger = require('../../../logger');
 const { removeUndefinedAndNullValues } = require('../../../v0/util');
+const { RedisDB } = require('../../../util/redis/redisConnector');
 const {
   pageViewedEventBuilder,
   cartViewedEventBuilder,
@@ -11,7 +16,11 @@ const {
   checkoutStepEventBuilder,
   searchEventBuilder,
 } = require('./pixelUtils');
-const { INTEGERATION, PIXEL_EVENT_TOPICS } = require('./config');
+const {
+  INTEGERATION,
+  PIXEL_EVENT_TOPICS,
+  pixelEventToCartTokenLocationMapping,
+} = require('./config');
 
 const NO_OPERATION_SUCCESS = {
   outputToSource: {
@@ -21,6 +30,59 @@ const NO_OPERATION_SUCCESS = {
   statusCode: 200,
 };
 
+/**
+ * Parses and extracts cart token value from the input event
+ * @param {Object} inputEvent
+ * @returns {String} cartToken
+ */
+function extractCartToken(inputEvent) {
+  const cartTokenLocation = pixelEventToCartTokenLocationMapping[inputEvent.name];
+  if (!cartTokenLocation) {
+    stats.increment('shopify_pixel_cart_token_not_found', {
+      event: inputEvent.name,
+      writeKey: inputEvent.query_parameters.writeKey,
+    });
+    return undefined;
+  }
+  // the unparsedCartToken is a string like '/checkout/cn/1234'
+  const unparsedCartToken = _.get(inputEvent, cartTokenLocation);
+  if (typeof unparsedCartToken !== 'string') {
+    logger.error(`Cart token is not a string`);
+    stats.increment('shopify_pixel_cart_token_not_found', {
+      event: inputEvent.name,
+      writeKey: inputEvent.query_parameters.writeKey,
+    });
+    return undefined;
+  }
+  const cartTokenParts = unparsedCartToken.split('/');
+  const cartToken = cartTokenParts[3];
+  return cartToken;
+}
+
+/**
+ * Handles storing cart token and anonymousId (clientId) in Redis
+ * @param {Object} inputEvent
+ * @param {String} clientId
+ */
+const handleCartTokenRedisOperations = async (inputEvent, clientId) => {
+  const cartToken = extractCartToken(inputEvent);
+  try {
+    if (isDefinedNotNullNotEmpty(clientId) && isDefinedNotNullNotEmpty(cartToken)) {
+      await RedisDB.setVal(cartToken, ['anonymousId', clientId]);
+      stats.increment('shopify_pixel_cart_token_set', {
+        event: inputEvent.name,
+        writeKey: inputEvent.query_parameters.writeKey,
+      });
+    }
+  } catch (error) {
+    logger.error(`Error handling Redis operations for event: ${inputEvent.name}`, error);
+    stats.increment('shopify_pixel_cart_token_redis_error', {
+      event: inputEvent.name,
+      writeKey: inputEvent.query_parameters.writeKey,
+    });
+  }
+};
+
 function processPixelEvent(inputEvent) {
   // eslint-disable-next-line @typescript-eslint/naming-convention
   const { name, query_parameters, clientId, data, id } = inputEvent;
@@ -37,6 +99,7 @@ function processPixelEvent(inputEvent) {
       message = pageViewedEventBuilder(inputEvent);
       break;
     case PIXEL_EVENT_TOPICS.CART_VIEWED:
+      handleCartTokenRedisOperations(inputEvent, clientId);
       message = cartViewedEventBuilder(inputEvent);
       break;
     case PIXEL_EVENT_TOPICS.COLLECTION_VIEWED:
@@ -52,13 +115,15 @@ function processPixelEvent(inputEvent) {
     case PIXEL_EVENT_TOPICS.CHECKOUT_STARTED:
     case PIXEL_EVENT_TOPICS.CHECKOUT_COMPLETED:
       if (customer.id) message.userId = customer.id || '';
+      handleCartTokenRedisOperations(inputEvent, clientId);
       message = checkoutEventBuilder(inputEvent);
       break;
     case PIXEL_EVENT_TOPICS.CHECKOUT_ADDRESS_INFO_SUBMITTED:
     case PIXEL_EVENT_TOPICS.CHECKOUT_CONTACT_INFO_SUBMITTED:
     case PIXEL_EVENT_TOPICS.CHECKOUT_SHIPPING_INFO_SUBMITTED:
     case PIXEL_EVENT_TOPICS.PAYMENT_INFO_SUBMITTED:
       if (customer.id) message.userId = customer.id || '';
+      handleCartTokenRedisOperations(inputEvent, clientId);
       message = checkoutStepEventBuilder(inputEvent);
       break;
     case PIXEL_EVENT_TOPICS.SEARCH_SUBMITTED:
@@ -94,4 +159,6 @@ const processEventFromPixel = async (event) => {
 
 module.exports = {
   processEventFromPixel,
+  handleCartTokenRedisOperations,
+  extractCartToken,
 };

src/v1/sources/shopify/pixelTransform.redisCartToken.test.js

@@ -0,0 +1,116 @@
+const { extractCartToken, handleCartTokenRedisOperations } = require('./pixelTransform');
+const { RedisDB } = require('../../../util/redis/redisConnector');
+const stats = require('../../../util/stats');
+const logger = require('../../../logger');
+const { pixelEventToCartTokenLocationMapping } = require('./config');
+
+jest.mock('../../../util/redis/redisConnector', () => ({
+  RedisDB: {
+    setVal: jest.fn(),
+  },
+}));
+
+jest.mock('../../../util/stats', () => ({
+  increment: jest.fn(),
+}));
+
+jest.mock('../../../logger', () => ({
+  info: jest.fn(),
+  error: jest.fn(),
+}));
+
+jest.mock('./config', () => ({
+  pixelEventToCartTokenLocationMapping: { cart_viewed: 'properties.cart_id' },
+}));
+
+describe('extractCartToken', () => {
+  it('should return undefined if cart token location is not found', () => {
+    const inputEvent = { name: 'unknownEvent', query_parameters: { writeKey: 'testWriteKey' } };
+
+    const result = extractCartToken(inputEvent);
+
+    expect(result).toBeUndefined();
+    expect(stats.increment).toHaveBeenCalledWith('shopify_pixel_cart_token_not_found', {
+      event: 'unknownEvent',
+      writeKey: 'testWriteKey',
+    });
+  });
+
+  it('should return undefined if cart token is not a string', () => {
+    const inputEvent = {
+      name: 'cart_viewed',
+      properties: { cart_id: 12345 },
+      query_parameters: { writeKey: 'testWriteKey' },
+    };
+
+    const result = extractCartToken(inputEvent);
+
+    expect(result).toBeUndefined();
+    expect(logger.error).toHaveBeenCalledWith('Cart token is not a string');
+    expect(stats.increment).toHaveBeenCalledWith('shopify_pixel_cart_token_not_found', {
+      event: 'cart_viewed',
+      writeKey: 'testWriteKey',
+    });
+  });
+
+  it('should return the cart token if it is a valid string', () => {
+    const inputEvent = {
+      name: 'cart_viewed',
+      properties: { cart_id: '/checkout/cn/1234' },
+      query_parameters: { writeKey: 'testWriteKey' },
+    };
+
+    const result = extractCartToken(inputEvent);
+
+    expect(result).toBe('1234');
+  });
+});
+
+describe('handleCartTokenRedisOperations', () => {
+  beforeEach(() => {
+    jest.clearAllMocks();
+  });
+
+  it('should handle undefined or null cart token gracefully', async () => {
+    const inputEvent = {
+      name: 'unknownEvent',
+      query_parameters: {
+        writeKey: 'testWriteKey',
+      },
+    };
+    const clientId = 'testClientId';
+
+    await handleCartTokenRedisOperations(inputEvent, clientId);
+
+    expect(stats.increment).toHaveBeenCalledWith('shopify_pixel_cart_token_not_found', {
+      event: 'unknownEvent',
+      writeKey: 'testWriteKey',
+    });
+  });
+
+  it('should log error and increment stats when exception occurs', async () => {
+    const inputEvent = {
+      name: 'cart_viewed',
+      properties: {
+        cart_id: '/checkout/cn/1234',
+      },
+      query_parameters: {
+        writeKey: 'testWriteKey',
+      },
+    };
+    const clientId = 'testClientId';
+    const error = new Error('Redis error');
+    RedisDB.setVal.mockRejectedValue(error);
+
+    await handleCartTokenRedisOperations(inputEvent, clientId);
+
+    expect(logger.error).toHaveBeenCalledWith(
+      'Error handling Redis operations for event: cart_viewed',
+      error,
+    );
+    expect(stats.increment).toHaveBeenCalledWith('shopify_pixel_cart_token_redis_error', {
+      event: 'cart_viewed',
+      writeKey: 'testWriteKey',
+    });
+  });
+});