feat: apply select enrichment attributes from a single source (CM-1355)

services/apps/members_enrichment_worker/src/types.ts

@@ -117,6 +117,7 @@ export type IMemberEnrichmentAttributeSettings = {
 
 export interface IProcessMemberSourcesArgs {
   memberId: string
+  activityCount: number
   sources: MemberEnrichmentSource[]
 }
 

services/apps/members_enrichment_worker/src/utils/config.ts

@@ -1,2 +1,12 @@
+import { MemberAttributeName } from '@crowd/types'
+
 export const ALSO_USE_EMAIL_IDENTITIES_FOR_ENRICHMENT = false
 export const ENRICH_EMAIL_IDENTITIES = false
+
+/**
+ * Attributes safe to write from a single enrichment source.
+ */
+export const SINGLE_SOURCE_ENRICHMENT_ATTRIBUTES: MemberAttributeName[] = [
+  MemberAttributeName.LOCATION,
+  MemberAttributeName.COUNTRY,
+]

services/apps/members_enrichment_worker/src/workflows/enrichMember.ts

@@ -86,8 +86,7 @@ export async function enrichMember(
 
   const changeInEnrichmentSourceData = sourceResults.some(Boolean)
 
-  if (changeInEnrichmentSourceData && input.activityCount > 100) {
-    // Member enrichment data has been updated, use squasher again!
+  if (changeInEnrichmentSourceData) {
     await executeChild(processMemberSources, {
       workflowId: 'member-enrichment/' + input.id + '/processMemberSources',
       cancellationType: ChildWorkflowCancellationType.WAIT_CANCELLATION_COMPLETED,
@@ -102,6 +101,7 @@ export async function enrichMember(
       args: [
         {
           memberId: input.id,
+          activityCount: input.activityCount ?? 0,
           sources,
         },
       ],

services/apps/members_enrichment_worker/src/workflows/processMemberSources.ts

@@ -4,6 +4,7 @@ import { MemberEnrichmentSource, PlatformType } from '@crowd/types'
 
 import * as activities from '../activities'
 import { IMemberEnrichmentDataNormalized, IProcessMemberSourcesArgs } from '../types'
+import { SINGLE_SOURCE_ENRICHMENT_ATTRIBUTES } from '../utils/config'
 
 const {
   findMemberEnrichmentCache,
@@ -35,9 +36,25 @@ const { squashMultipleValueAttributesWithLLM, squashWorkExperiencesWithLLM } = p
   },
 })
 
+function getEnrichmentAttributeValue(
+  attributes: IMemberEnrichmentDataNormalized['attributes'],
+  attributeName: string,
+) {
+  const values = attributes?.[attributeName]
+  if (!values) {
+    return undefined
+  }
+
+  const enrichmentKey = Object.keys(values).find((key) => key.startsWith('enrichment-'))
+  return enrichmentKey ? values[enrichmentKey] : undefined
+}
+
 export async function processMemberSources(args: IProcessMemberSourcesArgs): Promise<boolean> {
   // without contributions since they take a lot of space
-  const toBeSquashed = {}
+  const toBeSquashed: Record<
+    string,
+    IMemberEnrichmentDataNormalized | IMemberEnrichmentDataNormalized[]
+  > = {}
 
   let hasContributions = false
 
@@ -46,10 +63,9 @@ export async function processMemberSources(args: IProcessMemberSourcesArgs): Pro
   for (const source of args.sources) {
     const cache = caches.find((c) => c.source === source)
     if (cache && cache.data) {
-      const normalized = (await normalizeEnrichmentData(
-        source,
-        cache.data,
-      )) as IMemberEnrichmentDataNormalized
+      const normalized = (await normalizeEnrichmentData(source, cache.data)) as
+        | IMemberEnrichmentDataNormalized
+        | IMemberEnrichmentDataNormalized[]
 
       if (Array.isArray(normalized)) {
         for (const n of normalized) {
@@ -64,43 +80,65 @@ export async function processMemberSources(args: IProcessMemberSourcesArgs): Pro
             delete n.reach
           }
         }
-      }
-
-      if (normalized.contributions) {
-        delete normalized.contributions
-      }
+      } else {
+        if (normalized.contributions) {
+          delete normalized.contributions
+        }
 
-      if (normalized.reach) {
-        delete normalized.reach
+        if (normalized.reach) {
+          delete normalized.reach
+        }
       }
 
       toBeSquashed[source] = normalized
     }
   }
 
-  if (Object.keys(toBeSquashed).length > 1) {
-    const existingMemberData = await fetchMemberDataForLLMSquashing(args.memberId)
-
-    let progaiLinkedinScraperProfileSelected: IMemberEnrichmentDataNormalized = null
-    let crustDataProfileSelected: IMemberEnrichmentDataNormalized = null
+  let existingMemberData = null
+
+  const arraySources = Object.keys(toBeSquashed).filter((source) =>
+    Array.isArray(toBeSquashed[source]),
+  )
+
+  // Only resolve arrays when an apply path can run afterward.
+  if (
+    arraySources.length > 0 &&
+    (Object.keys(toBeSquashed).length === 1 || args.activityCount > 100)
+  ) {
+    existingMemberData = await fetchMemberDataForLLMSquashing(args.memberId)
+
+    const orderedArraySources = [
+      ...arraySources.filter((source) => source === MemberEnrichmentSource.CRUSTDATA),
+      ...arraySources.filter((source) => source === MemberEnrichmentSource.PROGAI_LINKEDIN_SCRAPER),
+      ...arraySources.filter(
+        (source) =>
+          source !== MemberEnrichmentSource.CRUSTDATA &&
+          source !== MemberEnrichmentSource.PROGAI_LINKEDIN_SCRAPER,
+      ),
+    ]
+
+    for (const source of orderedArraySources) {
+      const normalized = toBeSquashed[source]
+      if (!Array.isArray(normalized)) {
+        continue
+      }
 
-    if (toBeSquashed[MemberEnrichmentSource.CRUSTDATA]) {
       const categorizationResult = await findWhichLinkedinProfileToUseAmongScraperResult(
         args.memberId,
         existingMemberData,
-        toBeSquashed[MemberEnrichmentSource.CRUSTDATA],
+        normalized,
       )
 
-      crustDataProfileSelected = categorizationResult.selected
-
-      if (crustDataProfileSelected) {
-        toBeSquashed[MemberEnrichmentSource.CRUSTDATA] = crustDataProfileSelected
+      if (categorizationResult.selected) {
+        toBeSquashed[source] = categorizationResult.selected
+      } else {
+        delete toBeSquashed[source]
       }
 
       // check if there are any discarded profiles
       if (categorizationResult.discarded.length > 0) {
         for (const discardedProfile of categorizationResult.discarded) {
-          const discardedLinkedinIdentity = discardedProfile.identities.find(
+          const discardedLinkedinIdentity = discardedProfile.identities?.find(
             (i) => i.platform === PlatformType.LINKEDIN,
           )
 
@@ -110,63 +148,80 @@ export async function processMemberSources(args: IProcessMemberSourcesArgs): Pro
           }
 
           // remove the root source where the discarded linkedin profile is coming from
-          for (const source of Object.keys(toBeSquashed)) {
+          for (const otherSource of Object.keys(toBeSquashed)) {
+            const profile = toBeSquashed[otherSource]
+            if (Array.isArray(profile)) {
+              continue
+            }
+
             if (
-              (toBeSquashed[source].identities || []).some(
+              (profile.identities || []).some(
                 (i) =>
                   i.value.trim().toLowerCase() ===
                     discardedLinkedinIdentity.value.trim().toLowerCase() &&
                   i.platform === PlatformType.LINKEDIN,
               )
             ) {
-              delete toBeSquashed[source]
+              delete toBeSquashed[otherSource]
             }
           }
         }
       }
     }
+  }
 
-    if (toBeSquashed[MemberEnrichmentSource.PROGAI_LINKEDIN_SCRAPER]) {
-      const categorizationResult = await findWhichLinkedinProfileToUseAmongScraperResult(
-        args.memberId,
-        existingMemberData,
-        toBeSquashed[MemberEnrichmentSource.PROGAI_LINKEDIN_SCRAPER],
-      )
-
-      progaiLinkedinScraperProfileSelected = categorizationResult.selected
+  const sourceKeys = Object.keys(toBeSquashed)
 
-      if (progaiLinkedinScraperProfileSelected) {
-        toBeSquashed[MemberEnrichmentSource.PROGAI_LINKEDIN_SCRAPER] =
-          progaiLinkedinScraperProfileSelected
-      }
+  // A single source isn't reliable enough for a full profile update, but some attributes
+  // are still useful for aggregate consumers, where coverage matters more than certainty.
+  if (sourceKeys.length === 1) {
+    const source = sourceKeys[0]
+    const normalized = toBeSquashed[source] as IMemberEnrichmentDataNormalized
 
-      if (categorizationResult.discarded.length > 0) {
-        for (const discardedProfile of categorizationResult.discarded) {
-          const discardedLinkedinIdentity = discardedProfile.identities.find(
-            (i) => i.platform === PlatformType.LINKEDIN,
-          )
+    if (normalized.attributes) {
+      const attributes = {}
 
-          // Skip if no LinkedIn identity found
-          if (!discardedLinkedinIdentity) {
-            continue
+      for (const attributeName of SINGLE_SOURCE_ENRICHMENT_ATTRIBUTES) {
+        const value = getEnrichmentAttributeValue(normalized.attributes, attributeName)
+        if (value) {
+          attributes[attributeName] = {
+            enrichment: await cleanAttributeValue(value),
           }
+        }
+      }
 
-          // remove the root source where the discarded linkedin profile is coming from
-          for (const source of Object.keys(toBeSquashed)) {
-            if (
-              (toBeSquashed[source].identities || []).some(
-                (i) =>
-                  i.value.trim().toLowerCase() ===
-                    discardedLinkedinIdentity.value.trim().toLowerCase() &&
-                  i.platform === PlatformType.LINKEDIN,
-              )
-            ) {
-              delete toBeSquashed[source]
-            }
-          }
+      if (Object.keys(attributes).length > 0) {
+        if (!existingMemberData) {
+          existingMemberData = await fetchMemberDataForLLMSquashing(args.memberId)
         }
+
+        return updateMemberUsingSquashedPayload(
+          args.memberId,
+          existingMemberData,
+          {
+            identities: [],
+            attributes,
+            memberOrganizations: [],
+            reach: {},
+          },
+          false,
+          false,
+        )
       }
     }
+  }
+
+  if (sourceKeys.length > 1 && args.activityCount > 100) {
+    if (!existingMemberData) {
+      existingMemberData = await fetchMemberDataForLLMSquashing(args.memberId)
+    }
+
+    const crustDataProfileSelected = toBeSquashed[
+      MemberEnrichmentSource.CRUSTDATA
+    ] as IMemberEnrichmentDataNormalized
+    const progaiLinkedinScraperProfileSelected = toBeSquashed[
+      MemberEnrichmentSource.PROGAI_LINKEDIN_SCRAPER
+    ] as IMemberEnrichmentDataNormalized
 
     // start squashing the data
     const squashedPayload: IMemberEnrichmentDataNormalized = {
@@ -178,8 +233,9 @@ export async function processMemberSources(args: IProcessMemberSourcesArgs): Pro
 
     // 1) squash identities
     for (const source of Object.keys(toBeSquashed)) {
-      if (toBeSquashed[source].identities) {
-        for (const identity of toBeSquashed[source].identities) {
+      const profile = toBeSquashed[source] as IMemberEnrichmentDataNormalized
+      if (profile.identities) {
+        for (const identity of profile.identities) {
           const sameIdentity = (i: { platform: string; type: string; value: string }) =>
             i.platform === identity.platform &&
             i.type === identity.type &&
@@ -201,23 +257,21 @@ export async function processMemberSources(args: IProcessMemberSourcesArgs): Pro
 
     // 2) squash attributes
     for (const source of Object.keys(toBeSquashed)) {
-      if (toBeSquashed[source].attributes) {
-        for (const attribute of Object.keys(toBeSquashed[source].attributes)) {
-          if (toBeSquashed[source].attributes[attribute][`enrichment-${source}`]) {
+      const profile = toBeSquashed[source] as IMemberEnrichmentDataNormalized
+      if (profile.attributes) {
+        for (const attribute of Object.keys(profile.attributes)) {
+          const value = getEnrichmentAttributeValue(profile.attributes, attribute)
+          if (value) {
             if (attributeCountMap[attribute]) {
               attributeCountMap[attribute] = attributeCountMap[attribute] + 1
               delete attributesSquashed[attribute]
-              attributeValues[attribute].push(
-                toBeSquashed[source].attributes[attribute][`enrichment-${source}`],
-              )
+              attributeValues[attribute].push(value)
             } else {
               attributeCountMap[attribute] = 1
               attributesSquashed[attribute] = {
-                enrichment: toBeSquashed[source].attributes[attribute][`enrichment-${source}`],
+                enrichment: value,
               }
-              attributeValues[attribute] = [
-                toBeSquashed[source].attributes[attribute][`enrichment-${source}`],
-              ]
+              attributeValues[attribute] = [value]
             }
           }
         }
@@ -263,11 +317,9 @@ export async function processMemberSources(args: IProcessMemberSourcesArgs): Pro
       // check if there are multiple work experiences from different sources
       const workExperienceDataInDifferentSources = []
       for (const source of Object.keys(toBeSquashed)) {
-        if (
-          toBeSquashed[source].memberOrganizations &&
-          toBeSquashed[source].memberOrganizations.length > 0
-        ) {
-          workExperienceDataInDifferentSources.push(toBeSquashed[source].memberOrganizations)
+        const profile = toBeSquashed[source] as IMemberEnrichmentDataNormalized
+        if (profile.memberOrganizations && profile.memberOrganizations.length > 0) {
+          workExperienceDataInDifferentSources.push(profile.memberOrganizations)
         }
       }
 
@@ -308,7 +360,7 @@ export async function processMemberSources(args: IProcessMemberSourcesArgs): Pro
       args.memberId,
       existingMemberData,
       squashedPayload,
-      progaiLinkedinScraperProfileSelected && hasContributions,
+      !!progaiLinkedinScraperProfileSelected && hasContributions,
       !!crustDataProfileSelected,
     )