feat: add redis support in shopify pixel for id stitching

src/v0/sources/shopify/util.js

@@ -272,6 +272,7 @@ module.exports = {
   createPropertiesForEcomEvent,
   extractEmailFromPayload,
   getAnonymousIdAndSessionId,
+  getCartToken,
   checkAndUpdateCartItems,
   getHashLineItems,
   getDataFromRedis,

src/v1/sources/shopify/pixelEventsMappings/campaignObjectMappings.json

@@ -0,0 +1,18 @@
+[
+  {
+    "sourceKeys": "utm_campaign",
+    "destKeys": "name"
+  },
+  {
+    "sourceKeys": "utm_medium",
+    "destKeys": "medium"
+  },
+  {
+    "sourceKeys": "utm_term",
+    "destKeys": "term"
+  },
+  {
+    "sourceKeys": "utm_content",
+    "destKeys": "content"
+  }
+]

src/v1/sources/shopify/transform.js

@@ -4,10 +4,14 @@
 const {
   process: processPixelWebhookEvents,
 } = require('./webhookTransformations/serverSideTransform');
+const { processIdentifierEvent, isIdentifierEvent } = require('./utils');
 
 const process = async (inputEvent) => {
   const { event } = inputEvent;
   const { query_parameters } = event;
+  if (isIdentifierEvent(event)) {
+    return processIdentifierEvent(event);
+  }
   // check identify the event is from the web pixel based on the pixelEventLabel property.
   const { pixelEventLabel: pixelClientEventLabel } = event;
   if (pixelClientEventLabel) {

src/v1/sources/shopify/utils.js

@@ -0,0 +1,22 @@
+const { RedisDB } = require('../../../util/redis/redisConnector');
+
+const NO_OPERATION_SUCCESS = {
+  outputToSource: {
+    body: Buffer.from('OK').toString('base64'),
+    contentType: 'text/plain',
+  },
+  statusCode: 200,
+};
+
+const isIdentifierEvent = (payload) => ['rudderIdentifier'].includes(payload?.event);
+
+const processIdentifierEvent = async (event) => {
+  const { cartToken, anonymousId } = event;
+  await RedisDB.setVal(`${cartToken}`, ['anonymousId', anonymousId]);
+  return NO_OPERATION_SUCCESS;
+};
+
+module.exports = {
+  processIdentifierEvent,
+  isIdentifierEvent,
+};

src/v1/sources/shopify/utils.test.js

@@ -0,0 +1,45 @@
+const { isIdentifierEvent, processIdentifierEvent } = require('./utils');
+const { RedisDB } = require('../../../util/redis/redisConnector');
+
+describe('Identifier Utils Tests', () => {
+  describe('test isIdentifierEvent', () => {
+    it('should return true if the event is rudderIdentifier', () => {
+      const event = { event: 'rudderIdentifier' };
+      expect(isIdentifierEvent(event)).toBe(true);
+    });
+
+    it('should return false if the event is not rudderIdentifier', () => {
+      const event = { event: 'checkout started' };
+      expect(isIdentifierEvent(event)).toBe(false);
+    });
+  });
+
+  describe('test processIdentifierEvent', () => {
+    beforeEach(() => {
+      jest.clearAllMocks();
+    });
+
+    it('should set the anonymousId in redis and return NO_OPERATION_SUCCESS', async () => {
+      const setValSpy = jest.spyOn(RedisDB, 'setVal').mockResolvedValue('OK');
+      const event = { cartToken: 'cartTokenTest1', anonymousId: 'anonymousIdTest1' };
+
+      const response = await processIdentifierEvent(event);
+
+      expect(setValSpy).toHaveBeenCalledWith('cartTokenTest1', ['anonymousId', 'anonymousIdTest1']);
+      expect(response).toEqual({
+        outputToSource: {
+          body: Buffer.from('OK').toString('base64'),
+          contentType: 'text/plain',
+        },
+        statusCode: 200,
+      });
+    });
+
+    it('should handle redis errors', async () => {
+      jest.spyOn(RedisDB, 'setVal').mockRejectedValue(new Error('Redis connection failed'));
+      const event = { cartToken: 'cartTokenTest1', anonymousId: 'anonymousIdTest1' };
+
+      await expect(processIdentifierEvent(event)).rejects.toThrow('Redis connection failed');
+    });
+  });
+});

src/v1/sources/shopify/webhookTransformations/serverSideTransform.js

@@ -1,9 +1,8 @@
-/* eslint-disable @typescript-eslint/naming-convention */
 const lodash = require('lodash');
 const get = require('get-value');
 const stats = require('../../../../util/stats');
-const { getShopifyTopic, extractEmailFromPayload } = require('../../../../v0/sources/shopify/util');
-const { removeUndefinedAndNullValues, isDefinedAndNotNull } = require('../../../../v0/util');
+const { getShopifyTopic } = require('../../../../v0/sources/shopify/util');
+const { removeUndefinedAndNullValues } = require('../../../../v0/util');
 const Message = require('../../../../v0/sources/message');
 const { EventType } = require('../../../../constants');
 const {
@@ -19,7 +18,8 @@ const { RUDDER_ECOM_MAP } = require('../config');
 const {
   createPropertiesForEcomEventFromWebhook,
   getProductsFromLineItems,
-  getAnonymousIdFromAttributes,
+  handleAnonymousId,
+  handleCommonProperties,
 } = require('./serverSideUtlis');
 
 const NO_OPERATION_SUCCESS = {
@@ -65,9 +65,6 @@ const ecomPayloadBuilder = (event, shopifyTopic) => {
   if (event.billing_address) {
     message.setProperty('traits.billingAddress', event.billing_address);
   }
-  if (!message.userId && event.user_id) {
-    message.setProperty('userId', event.user_id);
-  }
   return message;
 };
 
@@ -110,38 +107,12 @@ const processEvent = async (inputEvent, metricMetadata) => {
       message = trackPayloadBuilder(event, shopifyTopic);
       break;
   }
-
-  if (message.userId) {
-    message.userId = String(message.userId);
-  }
-  if (!get(message, 'traits.email')) {
-    const email = extractEmailFromPayload(event);
-    if (email) {
-      message.setProperty('traits.email', email);
-    }
-  }
   // attach anonymousId if the event is track event using note_attributes
   if (message.type !== EventType.IDENTIFY) {
-    const anonymousId = getAnonymousIdFromAttributes(event);
-    if (isDefinedAndNotNull(anonymousId)) {
-      message.setProperty('anonymousId', anonymousId);
-    }
-  }
-  message.setProperty(`integrations.${INTEGERATION}`, true);
-  message.setProperty('context.library', {
-    eventOrigin: 'server',
-    name: 'RudderStack Shopify Cloud',
-    version: '2.0.0',
-  });
-  message.setProperty('context.topic', shopifyTopic);
-  // attaching cart, checkout and order tokens in context object
-  message.setProperty(`context.cart_token`, event.cart_token);
-  message.setProperty(`context.checkout_token`, event.checkout_token);
-  // raw shopify payload passed inside context object under shopifyDetails
-  message.setProperty('context.shopifyDetails', event);
-  if (shopifyTopic === 'orders_updated') {
-    message.setProperty(`context.order_token`, event.token);
+    handleAnonymousId(message, event);
   }
+  // attach userId, email and other contextual properties
+  message = handleCommonProperties(message, event, shopifyTopic);
   message = removeUndefinedAndNullValues(message);
   return message;
 };

src/v1/sources/shopify/webhookTransformations/serverSideUtils.test.js

@@ -1,15 +1,13 @@
+const { processEvent } = require('./serverSideTransform');
 const {
   getProductsFromLineItems,
   createPropertiesForEcomEventFromWebhook,
   getAnonymousIdFromAttributes,
+  getCartToken,
 } = require('./serverSideUtlis');
+const { RedisDB } = require('../../../../util/redis/redisConnector');
 
-const { constructPayload } = require('../../../../v0/util');
-
-const {
-  lineItemsMappingJSON,
-  productMappingJSON,
-} = require('../../../../v0/sources/shopify/config');
+const { lineItemsMappingJSON } = require('../../../../v0/sources/shopify/config');
 const Message = require('../../../../v0/sources/message');
 jest.mock('../../../../v0/sources/message');
 
@@ -63,7 +61,6 @@ describe('serverSideUtils.js', () => {
     });
 
     it('should return array of products', () => {
-      const mapping = {};
       const result = getProductsFromLineItems(LINEITEMS, lineItemsMappingJSON);
       expect(result).toEqual([
         { brand: 'Hydrogen Vendor', price: '600.00', product_id: 7234590408818, quantity: 1 },
@@ -115,16 +112,62 @@ describe('serverSideUtils.js', () => {
     // Handles empty note_attributes array gracefully
     it('should return null when note_attributes is an empty array', async () => {
       const event = { note_attributes: [] };
-      const result = await getAnonymousIdFromAttributes(event);
+      const result = getAnonymousIdFromAttributes(event);
+      expect(result).toBeNull();
+    });
+
+    it('should return null when note_attributes is not present', async () => {
+      const event = {};
+      const result = getAnonymousIdFromAttributes(event);
       expect(result).toBeNull();
     });
 
     it('get anonymousId from noteAttributes', async () => {
       const event = {
         note_attributes: [{ name: 'rudderAnonymousId', value: '123456' }],
       };
-      const result = await getAnonymousIdFromAttributes(event);
+      const result = getAnonymousIdFromAttributes(event);
       expect(result).toEqual('123456');
     });
   });
+
+  describe('getCartToken', () => {
+    it('should return null if cart_token is not present', () => {
+      const event = {};
+      const result = getCartToken(event);
+      expect(result).toBeNull();
+    });
+
+    it('should return cart_token if it is present', () => {
+      const event = { cart_token: 'cartTokenTest1' };
+      const result = getCartToken(event);
+      expect(result).toEqual('cartTokenTest1');
+    });
+  });
+});
+
+describe('Redis cart token tests', () => {
+  it('should get anonymousId property from redis', async () => {
+    const getValSpy = jest
+      .spyOn(RedisDB, 'getVal')
+      .mockResolvedValue({ anonymousId: 'anonymousIdTest1' });
+    const event = {
+      cart_token: `cartTokenTest1`,
+      id: 5778367414385,
+      line_items: [
+        {
+          id: 14234727743601,
+        },
+      ],
+      query_parameters: {
+        topic: ['orders_updated'],
+        version: ['pixel'],
+        writeKey: ['dummy-write-key'],
+      },
+    };
+    const message = await processEvent(event);
+    expect(getValSpy).toHaveBeenCalledTimes(1);
+    expect(getValSpy).toHaveBeenCalledWith('cartTokenTest1');
+    expect(message.setProperty).toHaveBeenCalledWith('anonymousId', 'anonymousIdTest1');
+  });
 });

src/v1/sources/shopify/webhookTransformations/serverSideUtlis.js

@@ -1,6 +1,10 @@
+/* eslint-disable no-param-reassign */
+const get = require('get-value');
 const { isDefinedAndNotNull } = require('@rudderstack/integrations-lib');
+const { extractEmailFromPayload } = require('../../../../v0/sources/shopify/util');
 const { constructPayload } = require('../../../../v0/util');
-const { lineItemsMappingJSON, productMappingJSON } = require('../config');
+const { INTEGERATION, lineItemsMappingJSON, productMappingJSON } = require('../config');
+const { RedisDB } = require('../../../../util/redis/redisConnector');
 
 /**
  * Returns an array of products from the lineItems array received from the webhook event
@@ -54,8 +58,73 @@ const getAnonymousIdFromAttributes = (event) => {
   return rudderAnonymousIdObject ? rudderAnonymousIdObject.value : null;
 };
 
+/**
+ * Returns the cart_token from the event message
+ * @param {Object} event
+ * @returns {String} cart_token
+ */
+const getCartToken = (event) => event?.cart_token || null;
+
+/**
+ * Handles the anonymousId assignment for the message, based on the event attributes and redis data
+ * @param {Object} message
+ * @param {Object} event
+ */
+const handleAnonymousId = async (message, event) => {
+  const anonymousId = getAnonymousIdFromAttributes(event);
+  if (isDefinedAndNotNull(anonymousId)) {
+    message.setProperty('anonymousId', anonymousId);
+  } else {
+    // if anonymousId is not present in note_attributes or note_attributes is not present, query redis for anonymousId
+    const cartToken = getCartToken(event);
+    if (cartToken) {
+      const redisData = await RedisDB.getVal(cartToken);
+      if (redisData?.anonymousId) {
+        message.setProperty('anonymousId', redisData.anonymousId);
+      }
+    }
+  }
+};
+
+/**
+  Handles userId, email and contextual properties enrichment for the message payload
+ * @param {Object} message
+ * @param {Object} event
+ * @param {String} shopifyTopic
+*/
+const handleCommonProperties = (message, event, shopifyTopic) => {
+  if (message.userId) {
+    message.userId = String(message.userId);
+  }
+  if (!get(message, 'traits.email')) {
+    const email = extractEmailFromPayload(event);
+    if (email) {
+      message.setProperty('traits.email', email);
+    }
+  }
+  message.setProperty(`integrations.${INTEGERATION}`, true);
+  message.setProperty('context.library', {
+    eventOrigin: 'server',
+    name: 'RudderStack Shopify Cloud',
+    version: '2.0.0',
+  });
+  message.setProperty('context.topic', shopifyTopic);
+  // attaching cart, checkout and order tokens in context object
+  message.setProperty(`context.cart_token`, event.cart_token);
+  message.setProperty(`context.checkout_token`, event.checkout_token);
+  // raw shopify payload passed inside context object under shopifyDetails
+  message.setProperty('context.shopifyDetails', event);
+  if (shopifyTopic === 'orders_updated') {
+    message.setProperty(`context.order_token`, event.token);
+  }
+  return message;
+};
+
 module.exports = {
   createPropertiesForEcomEventFromWebhook,
   getProductsFromLineItems,
   getAnonymousIdFromAttributes,
+  getCartToken,
+  handleAnonymousId,
+  handleCommonProperties,
 };