feat: enrich with geo location data during destination transformation

src/helpers/geoLocation.ts

@@ -0,0 +1,54 @@
+import cloneDeep from 'lodash/cloneDeep';
+import isEmpty from 'lodash/isEmpty';
+import set from 'set-value';
+import { FixMe } from '../util/types';
+import { getFirstMatchingKeyAndValue } from '../v0/util';
+import * as GenericFieldMappingJson from '../v0/util/data/GenericFieldMapping.json';
+
+export default class GeoLocationHelper {
+  public static getAddressKeyAndValue(message: Record<string, FixMe>): {
+    key: string;
+    value: Record<string, FixMe>;
+  } {
+    const { value, key: foundKey } = getFirstMatchingKeyAndValue(
+      message,
+      GenericFieldMappingJson.address,
+    );
+    const { key: traitsKey } = getFirstMatchingKeyAndValue(message, GenericFieldMappingJson.traits);
+    const addressKey =
+      foundKey || (traitsKey ? `${traitsKey}.address` : GenericFieldMappingJson.address[0]);
+    return { key: addressKey, value };
+  }
+
+  public static enrichMessageAddressWithGeoData(
+    message: Record<string, FixMe>,
+  ): Record<string, FixMe> {
+    let msg = message;
+    if (isEmpty(msg?.context?.geo)) {
+      // geo-location data was not sent
+      return {};
+    }
+    const { value: address, key: addressKey } = GeoLocationHelper.getAddressKeyAndValue(msg);
+    const addressFieldToGeoFieldMap = {
+      city: 'city',
+      country: 'country',
+      postalCode: 'postal',
+      state: 'region',
+    };
+
+    const mappedAddress = Object.entries(addressFieldToGeoFieldMap).reduce(
+      (agg, [addressFieldKey, geoFieldKey]) => {
+        if (!address?.[addressFieldKey] && msg?.context?.geo?.[geoFieldKey]) {
+          return { [addressFieldKey]: msg.context.geo[geoFieldKey], ...agg };
+        }
+        return agg;
+      },
+      {},
+    );
+    if (!isEmpty(address) || !isEmpty(mappedAddress)) {
+      msg = cloneDeep(message || {});
+      set(msg, addressKey, { ...address, ...mappedAddress });
+    }
+    return msg;
+  }
+}

src/middlewares/enricher.ts

@@ -0,0 +1,54 @@
+import { Context, Next } from 'koa';
+import {
+  ProcessorTransformationRequest,
+  RouterTransformationRequest,
+  RouterTransformationRequestData,
+} from '../types';
+import GeoLocationHelper from '../helpers/geoLocation';
+
+export type DTRequest = RouterTransformationRequest | ProcessorTransformationRequest[];
+
+export default class GeoEnricher {
+  private static enrichWithGeoInfo(
+    data: RouterTransformationRequestData[],
+  ): RouterTransformationRequestData[] {
+    return data.map((inputEvent) => {
+      const geoEnrichedMessage = GeoLocationHelper.enrichMessageAddressWithGeoData(
+        inputEvent.message,
+      );
+      return {
+        ...inputEvent,
+        message: {
+          ...inputEvent.message,
+          ...geoEnrichedMessage,
+        },
+      };
+    });
+  }
+
+  public static async enrich(ctx: Context, next: Next) {
+    const transformationRequest = ctx.request.body;
+    let transformationReq: DTRequest;
+    let reqBody: unknown;
+    const isRouterTransform = Array.isArray(
+      (transformationRequest as RouterTransformationRequest)?.input,
+    );
+    if (isRouterTransform) {
+      // Router or batch transformation request
+      transformationReq = transformationRequest as RouterTransformationRequest;
+      const enrichedEvents: RouterTransformationRequestData[] = GeoEnricher.enrichWithGeoInfo(
+        transformationReq.input,
+      );
+      reqBody = {
+        input: enrichedEvents,
+        destType: transformationReq.destType,
+      };
+    } else {
+      // Processor transformation
+      transformationReq = transformationRequest as ProcessorTransformationRequest[];
+      reqBody = GeoEnricher.enrichWithGeoInfo(transformationReq);
+    }
+    ctx.request.body = reqBody;
+    await next();
+  }
+}

src/routes/destination.ts

@@ -3,6 +3,7 @@ import DestinationController from '../controllers/destination';
 import RegulationController from '../controllers/regulation';
 import FeatureFlagController from '../middlewares/featureFlag';
 import RouteActivationController from '../middlewares/routeActivation';
+import GeoEnricher from '../middlewares/enricher';
 
 const router = new Router();
 
@@ -11,20 +12,23 @@ router.post(
   RouteActivationController.isDestinationRouteActive,
   RouteActivationController.destinationProcFilter,
   FeatureFlagController.handle,
+  GeoEnricher.enrich,
   DestinationController.destinationTransformAtProcessor,
 );
 router.post(
   '/routerTransform',
   RouteActivationController.isDestinationRouteActive,
   RouteActivationController.destinationRtFilter,
   FeatureFlagController.handle,
+  GeoEnricher.enrich,
   DestinationController.destinationTransformAtRouter,
 );
 router.post(
   '/batch',
   RouteActivationController.isDestinationRouteActive,
   RouteActivationController.destinationBatchFilter,
   FeatureFlagController.handle,
+  GeoEnricher.enrich,
   DestinationController.batchProcess,
 );
 

src/types/index.ts

@@ -84,21 +84,44 @@ type UserTransformationLibrary = {
   VersionID: string;
 };
 
-type ProcessorTransformationRequest = {
-  request?: object;
-  message: object;
-  metadata: Metadata;
-  destination: Destination;
-  libraries: UserTransformationLibrary[];
-};
+type RudderTimestamp = Date | string;
+type RudderMessageType =
+  | 'identify'
+  | 'track'
+  | 'page'
+  | 'screen'
+  | 'group'
+  | 'alias'
+  | 'audiencelist';
+
+type RudderMessageSchema = Partial<{
+  userId: string;
+  anonymousId: string;
+  type: RudderMessageType;
+  channel: string;
+  context: object;
+  originalTimestamp: RudderTimestamp;
+  sentAt: RudderTimestamp;
+  timestamp: RudderTimestamp;
+  event: string;
+  integrations: object;
+  messageId: string;
+  properties: object;
+  traits: object;
+  [key: string]: FixMe;
+}>;
 
 type RouterTransformationRequestData = {
   request?: object;
-  message: object;
+  message: RudderMessageSchema;
   metadata: Metadata;
   destination: Destination;
 };
 
+type ProcessorTransformationRequest = {
+  libraries: UserTransformationLibrary[];
+} & RouterTransformationRequestData;
+
 type RouterTransformationRequest = {
   input: RouterTransformationRequestData[];
   destType: string;
@@ -275,4 +298,5 @@ export {
   ComparatorInput,
   SourceInput,
   Source,
+  RudderMessageSchema,
 };

src/v0/destinations/hs/util.js

@@ -607,7 +607,8 @@ const splitEventsForCreateUpdate = async (inputs, destination) => {
     const { destinationExternalId } = getDestinationExternalIDInfoForRetl(message, DESTINATION);
 
     const filteredInfo = updateHubspotIds.filter(
-      (update) => update.property.toString().toLowerCase() === destinationExternalId.toString().toLowerCase(),
+      (update) =>
+        update.property.toString().toLowerCase() === destinationExternalId.toString().toLowerCase(),
     );
 
     if (filteredInfo.length > 0) {

src/v0/sources/shopify/util.js

@@ -32,7 +32,10 @@ const getDataFromRedis = async (key, metricMetadata) => {
       ...metricMetadata,
     });
     const redisData = await RedisDB.getVal(key);
-    if (redisData === null || (typeof redisData === "object" && Object.keys(redisData).length === 0)) {
+    if (
+      redisData === null ||
+      (typeof redisData === 'object' && Object.keys(redisData).length === 0)
+    ) {
       stats.increment('shopify_redis_no_val', {
         ...metricMetadata,
       });

src/v0/util/index.js

@@ -659,6 +659,33 @@ const getValueFromMessage = (message, sourceKeys) => {
   return null;
 };
 
+const getFirstMatchingKeyAndValue = (message, sourceKeys) => {
+  let srcKeys = sourceKeys;
+  if (
+    typeof message !== 'object' ||
+    (!Array.isArray(sourceKeys) && typeof sourceKeys !== 'string')
+  ) {
+    // sourceKeys => boolean, number, object, function etc,.
+    // message => anything other than object,.
+    return { key: '' };
+  }
+  if (typeof sourceKeys === 'string') {
+    srcKeys = [sourceKeys];
+  }
+  // got the possible sourceKeys
+  // eslint-disable-next-line no-restricted-syntax
+  for (const sourceKey of srcKeys) {
+    const val = get(message, sourceKey);
+    if (!lodash.isNil(val)) {
+      // return only if the value is valid.
+      // else look for next possible source in precedence
+      return { value: val, key: sourceKey };
+    }
+  }
+
+  return { value: null, key: '' };
+};
+
 // get a field value from message.
 // if sourceFromGenericMap is true get its value from GenericFieldMapping.json and use it as sourceKey
 // else use sourceKey from `data/message.json` for actual field precedence
@@ -2192,4 +2219,5 @@ module.exports = {
   isValidInteger,
   isNewStatusCodesAccepted,
   IsGzipSupported,
+  getFirstMatchingKeyAndValue,
 };

test/apitests/data_scenarios/source/v1/pipedream.json

@@ -46,4 +46,4 @@
       }
     }
   ]
-}
+}