feat(fb_custom_audience): handle invalid fields and events (#5023)

## Summary
- Validate email fields using `validator.isEmail`; emit
`fb_custom_audience_invalid_email` stats metric for invalid emails
instead of silently passing them through
- Validate country codes must be exactly 2 characters; emit
`fb_custom_audience_invalid_country_code` stats metric for invalid codes
- Throw `InstrumentationError` when all user properties for a record are
null/invalid, replacing the previous silent stats counter
(`fb_custom_audience_event_having_all_null_field_values_for_a_user`)
- Pass `workspaceId` and `destinationId` through the call chain to
enable proper metric attribution

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Author: shekhar-rudder

src/v0/destinations/fb_custom_audience/config.ts

@@ -126,6 +126,17 @@ function getMaxPayloadSize(workspaceId: string): number {
   return DEFAULT_MAX_PAYLOAD_SIZE;
 }
 
+/**
+ * Whether to reject invalid field values (e.g., malformed emails, invalid country codes)
+ * by replacing them with empty strings. When disabled, invalid values are passed through as-is.
+ *
+ * Controlled via env var: FB_CUSTOM_AUDIENCE_REJECT_INVALID_FIELDS=true
+ * Default: false
+ */
+function isRejectInvalidFieldsEnabled(): boolean {
+  return process.env.FB_CUSTOM_AUDIENCE_REJECT_INVALID_FIELDS === 'true';
+}
+
 export {
   DESTINATION,
   ENDPOINT_PATH,
@@ -137,4 +148,5 @@ export {
   typeFields,
   subTypeFields,
   getMaxPayloadSize,
+  isRejectInvalidFieldsEnabled,
 };

src/v0/destinations/fb_custom_audience/recordTransform.ts

@@ -17,7 +17,6 @@ import type {
   FbRecordEvent,
 } from './types';
 import { schemaFields, MAX_USER_COUNT } from './config';
-import stats from '../../../util/stats';
 import {
   getDestinationExternalIDInfoForRetl,
   checkSubsetOfArray,
@@ -51,6 +50,8 @@ const processRecord = (
   userSchema: string[],
   isHashRequired: boolean,
   disableFormat: boolean | undefined,
+  workspaceId: string,
+  destinationId: string,
 ): { dataElement: unknown[]; metadata: Metadata } => {
   const fields = record.message.fields!;
   let dataElement: unknown[] = [];
@@ -61,7 +62,12 @@ const processRecord = (
     let updatedProperty: unknown = userProperty;
 
     if (isHashRequired && !disableFormat) {
-      updatedProperty = ensureApplicableFormat(eachProperty, userProperty);
+      updatedProperty = ensureApplicableFormat(
+        eachProperty,
+        userProperty,
+        workspaceId,
+        destinationId,
+      );
     }
 
     dataElement = getUpdatedDataElement(
@@ -79,10 +85,9 @@ const processRecord = (
   });
 
   if (nullUserData) {
-    stats.increment('fb_custom_audience_event_having_all_null_field_values_for_a_user', {
-      destinationId: record.destination.ID,
-      nullFields: userSchema,
-    });
+    throw new InstrumentationError(
+      `All user properties [${userSchema.join(', ')}] are invalid or null. At least one valid field is required.`,
+    );
   }
 
   return { dataElement, metadata: record.metadata };
@@ -115,6 +120,8 @@ const processRecordEventArray = async (
         userSchema,
         isHashRequired,
         disableFormat,
+        input.metadata.workspaceId,
+        destination.ID,
       );
       metadata.push(recordMetadata);
       return dataElement;

src/v0/destinations/fb_custom_audience/util.ts

@@ -2,6 +2,7 @@ import lodash from 'lodash';
 import sha256 from 'sha256';
 import crypto from 'crypto';
 import jsonSize from 'json-size';
+import validator from 'validator';
 import {
   InstrumentationError,
   ConfigurationError,
@@ -15,7 +16,13 @@ import type {
   FbRecordMessage,
   WrappedResponse,
 } from './types';
-import { typeFields, subTypeFields, getEndPoint, DESTINATION } from './config';
+import {
+  typeFields,
+  subTypeFields,
+  getEndPoint,
+  isRejectInvalidFieldsEnabled,
+  DESTINATION,
+} from './config';
 import {
   defaultRequestConfig,
   defaultPostRequestConfig,
@@ -77,20 +84,37 @@ const getSchemaForEventMappedToDest = (message: FbRecordMessage): string[] => {
   return userSchema;
 };
 
-// function responsible to ensure the user inputs are passed according to the allowed format
-const ensureApplicableFormat = (userProperty: string, userInformation: unknown): unknown => {
+/**
+ * Ensures user inputs are in the format required by Facebook Custom Audiences.
+ * Returns empty string for invalid field values.
+ */
+const ensureApplicableFormat = (
+  userProperty: string,
+  userInformation: unknown,
+  workspaceId: string,
+  destinationId: string,
+): unknown => {
   let updatedProperty: unknown;
   let userInformationTrimmed: string;
   if (isDefinedAndNotNull(userInformation)) {
     const stringifiedUserInformation = convertToString(userInformation);
     switch (userProperty) {
-      case 'EMAIL':
-        updatedProperty = stringifiedUserInformation.trim().toLowerCase();
+      case 'EMAIL': {
+        const emailValue = stringifiedUserInformation.trim().toLowerCase();
+        if (validator.isEmail(emailValue)) {
+          updatedProperty = emailValue;
+        } else {
+          stats.increment('fb_custom_audience_invalid_email', { workspaceId, destinationId });
+          updatedProperty = isRejectInvalidFieldsEnabled() ? '' : emailValue;
+        }
         break;
-      case 'PHONE':
+      }
+      case 'PHONE': {
         // remove all non-numerical characters, then remove all leading zeros
         updatedProperty = stringifiedUserInformation.replace(/\D/g, '').replace(/^0+/g, '');
+        // Note: libphonenumber-js is not used here as it requires a country code to validate, which may not always be present.
         break;
+      }
       case 'GEN':
         updatedProperty =
           stringifiedUserInformation.toLowerCase() === 'f' ||
@@ -124,9 +148,19 @@ const ensureApplicableFormat = (userProperty: string, userInformation: unknown):
       case 'MADID':
         updatedProperty = stringifiedUserInformation.toLowerCase();
         break;
-      case 'COUNTRY':
-        updatedProperty = stringifiedUserInformation.toLowerCase();
+      case 'COUNTRY': {
+        const countryCode = stringifiedUserInformation.toLowerCase();
+        if (countryCode.length === 2) {
+          updatedProperty = countryCode;
+        } else {
+          stats.increment('fb_custom_audience_invalid_country_code', {
+            workspaceId,
+            destinationId,
+          });
+          updatedProperty = isRejectInvalidFieldsEnabled() ? '' : countryCode;
+        }
         break;
+      }
       case 'ZIP':
         userInformationTrimmed = stringifiedUserInformation.replace(/\s/g, '');
         updatedProperty = userInformationTrimmed.toLowerCase();
@@ -224,7 +258,12 @@ const prepareDataField = (
       let updatedProperty: unknown = userProperty;
 
       if (isHashRequired && !disableFormat) {
-        updatedProperty = ensureApplicableFormat(eachProperty, userProperty);
+        updatedProperty = ensureApplicableFormat(
+          eachProperty,
+          userProperty,
+          workspaceId,
+          destinationId,
+        );
       }
 
       dataElement = getUpdatedDataElement(
@@ -243,10 +282,9 @@ const prepareDataField = (
     });
 
     if (nullUserData) {
-      stats.increment('fb_custom_audience_event_having_all_null_field_values_for_a_user', {
-        destinationId,
-        nullFields: userSchema,
-      });
+      throw new InstrumentationError(
+        `All user properties [${userSchema.join(', ')}] are invalid or null. At least one valid field is required.`,
+      );
     }
 
     data.push(dataElement);

test/integrations/destinations/fb_custom_audience/processor/data.ts

@@ -2029,47 +2029,17 @@ export const data = [
         status: 200,
         body: [
           {
-            output: {
-              version: '1',
-              type: 'REST',
-              method: 'POST',
-              endpoint: 'https://graph.facebook.com/v23.0/aud1/users',
-              endpointPath: 'users',
-              headers: {},
-              userId: '',
-              params: {
-                access_token: 'ABC',
-                payload: {
-                  is_raw: true,
-                  data_source: {
-                    type: 'UNKNOWN',
-                    sub_type: 'ANYTHING',
-                  },
-                  schema: [
-                    'EMAIL',
-                    'DOBM',
-                    'DOBD',
-                    'DOBY',
-                    'PHONE',
-                    'GEN',
-                    'FI',
-                    'MADID',
-                    'ZIP',
-                    'ST',
-                    'COUNTRY',
-                  ],
-                  data: [['', '', '', '', '', '', '', '', '', '', '']],
-                },
-              },
-              body: {
-                JSON: {},
-                JSON_ARRAY: {},
-                XML: {},
-                FORM: {},
-              },
-              files: {},
+            error:
+              'All user properties [EMAIL, DOBM, DOBD, DOBY, PHONE, GEN, FI, MADID, ZIP, ST, COUNTRY] are invalid or null. At least one valid field is required.',
+            statTags: {
+              destType: 'FB_CUSTOM_AUDIENCE',
+              errorCategory: 'dataValidation',
+              errorType: 'instrumentation',
+              feature: 'processor',
+              implementation: 'native',
+              module: 'destination',
             },
-            statusCode: 200,
+            statusCode: 400,
           },
         ],
       },

test/integrations/destinations/fb_custom_audience/router/data.ts

@@ -884,44 +884,9 @@ export const data = [
         body: {
           output: [
             {
-              batchedRequest: [
-                {
-                  version: '1',
-                  type: 'REST',
-                  method: 'POST',
-                  endpoint: 'https://graph.facebook.com/v23.0/23848494844100489/users',
-                  endpointPath: 'users',
-                  headers: {},
-                  params: {
-                    access_token: 'ABC',
-                    payload: {
-                      schema: ['EMAIL', 'FI'],
-                      data: [
-                        [
-                          'b100c2ec0718fe6b4805b623aeec6710719d042ceea55f5c8135b010ec1c7b36',
-                          '1e14a2f476f7611a8b22bc85d14237fdc88aac828737e739416c32c5bce3bd16',
-                        ],
-                        [
-                          'b100c2ec0718fe6b4805b623aeec6710719d042ceea55f5c8135b010ec1c7b36',
-                          '1e14a2f476f7611a8b22bc85d14237fdc88aac828737e739416c32c5bce3bd16',
-                        ],
-                        [
-                          'b100c2ec0718fe6b4805b623aeec6710719d042ceea55f5c8135b010ec1c7b36',
-                          '1e14a2f476f7611a8b22bc85d14237fdc88aac828737e739416c32c5bce3bd16',
-                        ],
-                        ['', ''],
-                      ],
-                    },
-                  },
-                  body: {
-                    JSON: {},
-                    JSON_ARRAY: {},
-                    XML: {},
-                    FORM: {},
-                  },
-                  files: {},
-                },
-              ],
+              batched: false,
+              error:
+                'All user properties [EMAIL, FI] are invalid or null. At least one valid field is required.',
               metadata: [
                 {
                   attemptNum: 1,
@@ -972,31 +937,16 @@ export const data = [
                   workspaceId: 'default-workspaceId',
                 },
               ],
-              batched: true,
-              statusCode: 200,
-              destination: {
-                Config: {
-                  accessToken: 'ABC',
-                  disableFormat: false,
-                  isHashRequired: true,
-                  isRaw: false,
-                  skipVerify: false,
-                  subType: 'NA',
-                  type: 'NA',
-                },
-                Name: 'FB_CUSTOM_AUDIENCE',
-                Enabled: true,
-                WorkspaceID: '1TSN08muJTZwH8iCDmnnRt1pmLd',
-                DestinationDefinition: {
-                  Config: {},
-                  DisplayName: 'FB_CUSTOM_AUDIENCE',
-                  ID: '1aIXqM806xAVm92nx07YwKbRrO9',
-                  Name: 'FB_CUSTOM_AUDIENCE',
-                },
-                ID: '1mMy5cqbtfuaKZv1IhVQKnBdVwe',
-                Transformations: [],
-                IsConnectionEnabled: true,
-                IsProcessorEnabled: true,
+              statusCode: 400,
+              statTags: {
+                errorCategory: 'dataValidation',
+                errorType: 'instrumentation',
+                destType: 'FB_CUSTOM_AUDIENCE',
+                module: 'destination',
+                implementation: 'native',
+                feature: 'router',
+                destinationId: 'default-destinationId',
+                workspaceId: 'default-workspaceId',
               },
             },
           ],