Enrichment monitoring fixes

backend/src/database/migrations/R__memberEnrichmentMaterializedViews.sql

@@ -314,7 +314,8 @@ from unique_members),
             select distinct mi."memberId"
             from "memberIdentities" mi
             left join "membersGlobalActivityCount" on "membersGlobalActivityCount"."memberId" = mi."memberId"
-            where mi.verified 
+            where "membersGlobalActivityCount".total_count > 1000 
+              and mi.verified 
               and mi.platform = 'linkedin'
             group by mi."memberId"),
         unique_members as (
@@ -359,7 +360,8 @@ from unique_members),
             select distinct mi."memberId"
             from "memberIdentities" mi
             left join "membersGlobalActivityCount" on "membersGlobalActivityCount"."memberId" = mi."memberId"
-            where mi.verified
+            where "membersGlobalActivityCount".total_count > 1000 
+            and mi.verified
             and mi.platform = 'linkedin'
             group by mi."memberId"),
       unique_members as (

services/apps/premium/members_enrichment_worker/src/activities.ts

@@ -2,6 +2,8 @@ import {
   findMemberEnrichmentCache,
   findMemberIdentityWithTheMostActivityInPlatform,
   getEnrichmentData,
+  getEnrichmentInput,
+  getObsoleteSourcesOfMember,
   insertMemberEnrichmentCache,
   isCacheObsolete,
   isEnrichableBySource,
@@ -10,7 +12,7 @@ import {
   touchMemberEnrichmentCacheUpdatedAt,
   updateMemberEnrichmentCache,
 } from './activities/enrichment'
-import { getEnrichableMembers } from './activities/getMembers'
+import { getEnrichableMembers, getMaxConcurrentRequests } from './activities/getMembers'
 import { refreshToken } from './activities/lf-auth0/authenticateLFAuth0'
 import {
   getIdentitiesExistInOtherMembers,
@@ -54,4 +56,7 @@ export {
   isEnrichableBySource,
   findMemberIdentityWithTheMostActivityInPlatform,
   refreshMemberEnrichmentMaterializedView,
+  getEnrichmentInput,
+  getMaxConcurrentRequests,
+  getObsoleteSourcesOfMember,
 }

services/apps/premium/members_enrichment_worker/src/activities/enrichment.ts

@@ -9,9 +9,12 @@ import {
 import { refreshMaterializedView } from '@crowd/data-access-layer/src/utils'
 import { RedisCache } from '@crowd/redis'
 import {
+  IEnrichableMember,
   IEnrichableMemberIdentityActivityAggregate,
   IMemberEnrichmentCache,
   MemberEnrichmentSource,
+  MemberIdentityType,
+  PlatformType,
 } from '@crowd/types'
 
 import { EnrichmentSourceServiceFactory } from '../factory'
@@ -41,6 +44,51 @@ export async function getEnrichmentData(
   return null
 }
 
+export async function getEnrichmentInput(
+  input: IEnrichableMember,
+): Promise<IEnrichmentSourceInput> {
+  const enrichmentInput: IEnrichmentSourceInput = {
+    memberId: input.id,
+    email: input.identities.find((i) => i.verified && i.type === MemberIdentityType.EMAIL),
+    linkedin: input.identities.find(
+      (i) =>
+        i.verified &&
+        i.platform === PlatformType.LINKEDIN &&
+        i.type === MemberIdentityType.USERNAME,
+    ),
+    displayName: input.displayName || undefined,
+    website: input.website || undefined,
+    location: input.location || undefined,
+    activityCount: input.activityCount || 0,
+  }
+
+  // there can be multiple verified identities in github, we select the one with the most activities
+  const verifiedGithubIdentities = input.identities.filter(
+    (i) =>
+      i.verified && i.platform === PlatformType.GITHUB && i.type === MemberIdentityType.USERNAME,
+  )
+
+  if (verifiedGithubIdentities.length > 1) {
+    const ghIdentityWithTheMostActivities = await findMemberIdentityWithTheMostActivityInPlatform(
+      input.id,
+      PlatformType.GITHUB,
+    )
+    if (ghIdentityWithTheMostActivities) {
+      enrichmentInput.github = input.identities.find(
+        (i) =>
+          i.verified &&
+          i.platform === PlatformType.GITHUB &&
+          i.type === MemberIdentityType.USERNAME &&
+          i.value === ghIdentityWithTheMostActivities.username,
+      )
+    }
+  } else {
+    enrichmentInput.github = verifiedGithubIdentities?.[0] || undefined
+  }
+
+  return enrichmentInput
+}
+
 export async function normalizeEnrichmentData(
   source: MemberEnrichmentSource,
   data: IMemberEnrichmentData,
@@ -53,6 +101,13 @@ export async function isCacheObsolete(
   source: MemberEnrichmentSource,
   cache: IMemberEnrichmentCache<IMemberEnrichmentData>,
 ): Promise<boolean> {
+  return isCacheObsoleteSync(source, cache)
+}
+
+export function isCacheObsoleteSync(
+  source: MemberEnrichmentSource,
+  cache: IMemberEnrichmentCache<IMemberEnrichmentData>,
+): boolean {
   const service = EnrichmentSourceServiceFactory.getEnrichmentSourceService(source, svc.log)
   return (
     !cache ||
@@ -104,8 +159,13 @@ export async function findMemberEnrichmentCache(
 
 export async function findMemberEnrichmentCacheForAllSources(
   memberId: string,
+  returnRowsWithoutData = false,
 ): Promise<IMemberEnrichmentCache<IMemberEnrichmentData>[]> {
-  return findMemberEnrichmentCacheForAllSourcesDb(svc.postgres.reader.connection(), memberId)
+  return findMemberEnrichmentCacheForAllSourcesDb(
+    svc.postgres.reader.connection(),
+    memberId,
+    returnRowsWithoutData,
+  )
 }
 
 export async function insertMemberEnrichmentCache(
@@ -138,6 +198,20 @@ export async function findMemberIdentityWithTheMostActivityInPlatform(
   return findMemberIdentityWithTheMostActivityInPlatformQuestDb(svc.questdbSQL, memberId, platform)
 }
 
+export async function getObsoleteSourcesOfMember(
+  memberId: string,
+  possibleSources: MemberEnrichmentSource[],
+): Promise<MemberEnrichmentSource[]> {
+  const caches = await findMemberEnrichmentCacheForAllSources(memberId, true)
+  const obsoleteSources = possibleSources.filter((source) =>
+    isCacheObsoleteSync(
+      source,
+      caches.find((c) => c.source === source),
+    ),
+  )
+  return obsoleteSources
+}
+
 export async function refreshMemberEnrichmentMaterializedView(mvName: string): Promise<void> {
   await refreshMaterializedView(svc.postgres.writer.connection(), mvName)
 }

services/apps/premium/members_enrichment_worker/src/activities/getMembers.ts

@@ -7,11 +7,11 @@ import {
 
 import { EnrichmentSourceServiceFactory } from '../factory'
 import { svc } from '../main'
+import { IEnrichmentService } from '../types'
 
 export async function getEnrichableMembers(
   limit: number,
   sources: MemberEnrichmentSource[],
-  afterCursor: { activityCount: number; memberId: string } | null,
 ): Promise<IEnrichableMember[]> {
   let rows: IEnrichableMember[] = []
   const sourceInputs: IMemberEnrichmentSourceQueryInput[] = sources.map((s) => {
@@ -23,7 +23,34 @@ export async function getEnrichableMembers(
     }
   })
   const db = svc.postgres.reader
-  rows = await fetchMembersForEnrichment(db, limit, sourceInputs, afterCursor)
+  rows = await fetchMembersForEnrichment(db, limit, sourceInputs)
 
   return rows
 }
+
+// Get the most strict parallelism among existing and enrichable sources
+// We only check sources that has activity count cutoff in current range
+export async function getMaxConcurrentRequests(
+  members: IEnrichableMember[],
+  possibleSources: MemberEnrichmentSource[],
+  concurrencyLimit: number,
+): Promise<number> {
+  const serviceMap: Partial<Record<MemberEnrichmentSource, IEnrichmentService>> = {}
+  const currentProcessingActivityCount = members[0].activityCount
+
+  let maxConcurrentRequestsInAllSources = concurrencyLimit
+
+  for (const source of possibleSources) {
+    serviceMap[source] = EnrichmentSourceServiceFactory.getEnrichmentSourceService(source, svc.log)
+    const activityCountCutoff = serviceMap[source].enrichMembersWithActivityMoreThan
+    if (!activityCountCutoff || activityCountCutoff <= currentProcessingActivityCount) {
+      maxConcurrentRequestsInAllSources = Math.min(
+        maxConcurrentRequestsInAllSources,
+        serviceMap[source].maxConcurrentRequests,
+      )
+    }
+  }
+  svc.log.info('Setting max concurrent requests', { maxConcurrentRequestsInAllSources })
+
+  return maxConcurrentRequestsInAllSources
+}

services/apps/premium/members_enrichment_worker/src/schedules/getMembersToEnrich.ts

@@ -29,11 +29,7 @@ export const scheduleMembersEnrichment = async () => {
           backoffCoefficient: 2,
           maximumAttempts: 3,
         },
-        args: [
-          {
-            afterCursor: null,
-          },
-        ],
+        args: [],
       },
     })
   } catch (err) {

services/apps/premium/members_enrichment_worker/src/sources/clearbit/service.ts

@@ -31,6 +31,8 @@ export default class EnrichmentServiceClearbit extends LoggerBase implements IEn
 
   public enrichableBySql = `"membersGlobalActivityCount".total_count > ${this.enrichMembersWithActivityMoreThan} AND mi.type = 'email' and mi.verified`
 
+  public maxConcurrentRequests = 15
+
   // bust cache after 120 days
   public cacheObsoleteAfterSeconds = 60 * 60 * 24 * 120
 
@@ -57,7 +59,11 @@ export default class EnrichmentServiceClearbit extends LoggerBase implements IEn
   }
 
   async isEnrichableBySource(input: IEnrichmentSourceInput): Promise<boolean> {
-    return !!input.email?.value && input.email?.verified
+    return (
+      input.activityCount > this.enrichMembersWithActivityMoreThan &&
+      !!input.email?.value &&
+      input.email?.verified
+    )
   }
 
   async getData(input: IEnrichmentSourceInput): Promise<IMemberEnrichmentDataClearbit | null> {

services/apps/premium/members_enrichment_worker/src/sources/crustdata/service.ts

@@ -47,6 +47,8 @@ export default class EnrichmentServiceCrustdata extends LoggerBase implements IE
 
   public cacheObsoleteAfterSeconds = 60 * 60 * 24 * 90
 
+  public maxConcurrentRequests = 5
+
   public attributeSettings: IMemberEnrichmentAttributeSettings = {
     [MemberAttributeName.AVATAR_URL]: {
       fields: ['profile_picture_url'],

services/apps/premium/members_enrichment_worker/src/sources/progai-linkedin-scraper/service.ts

@@ -37,6 +37,8 @@ export default class EnrichmentServiceProgAILinkedinScraper
 
   public cacheObsoleteAfterSeconds = 60 * 60 * 24 * 90
 
+  public maxConcurrentRequests = 1000
+
   constructor(public readonly log: Logger) {
     super(log)
   }

services/apps/premium/members_enrichment_worker/src/sources/progai/service.ts

@@ -33,11 +33,13 @@ export default class EnrichmentServiceProgAI extends LoggerBase implements IEnri
   public source: MemberEnrichmentSource = MemberEnrichmentSource.PROGAI
   public platform = `enrichment-${this.source}`
 
-  enrichableBySql = `mi.verified and ((mi.type = 'username' AND mi.platform = 'github') OR (mi.type = 'email'))`
+  public enrichableBySql = `mi.verified and ((mi.type = 'username' AND mi.platform = 'github') OR (mi.type = 'email'))`
 
   // bust cache after 90 days
   public cacheObsoleteAfterSeconds = 60 * 60 * 24 * 90
 
+  public maxConcurrentRequests = 1000
+
   public attributeSettings: IMemberEnrichmentAttributeSettings = {
     [MemberAttributeName.AVATAR_URL]: {
       fields: ['profile_pic_url'],

services/apps/premium/members_enrichment_worker/src/sources/serp/service.ts

@@ -32,6 +32,8 @@ export default class EnrichmentServiceSerpApi extends LoggerBase implements IEnr
   // bust cache after 120 days
   public cacheObsoleteAfterSeconds = 60 * 60 * 24 * 120
 
+  public maxConcurrentRequests = 300
+
   constructor(public readonly log: Logger) {
     super(log)
   }

services/apps/premium/members_enrichment_worker/src/types.ts

@@ -41,6 +41,9 @@ export interface IEnrichmentService {
   // cache rows with older updatedAt than this will be considered obsolete and will be re-enriched
   cacheObsoleteAfterSeconds: number
 
+  // max concurrent requests that can be made to the source
+  maxConcurrentRequests: number
+
   // can the source enrich using this input
   isEnrichableBySource(input: IEnrichmentSourceInput): Promise<boolean>
 
@@ -54,6 +57,9 @@ export interface IEnrichmentService {
   // activity count is available in "membersGlobalActivityCount" alias, "membersGlobalActivityCount".total_count field
   enrichableBySql: string
 
+  // only enrich members with activity more than this number
+  enrichMembersWithActivityMoreThan?: number
+
   // should either return the data or null if it's a miss
   getData(input: IEnrichmentSourceInput): Promise<IMemberEnrichmentData | null>
   normalize(
@@ -88,10 +94,6 @@ export interface IMemberEnrichmentLinkedinScraperMetadata {
   isFromVerifiedSource: boolean
 }
 
-export interface IGetMembersForEnrichmentArgs {
-  afterCursor: { activityCount: number; memberId: string } | null
-}
-
 export interface IMemberEnrichmentSocialData {
   platform: PlatformType
   handle: string

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

@@ -1,11 +1,6 @@
 import { proxyActivities } from '@temporalio/workflow'
 
-import {
-  IEnrichableMember,
-  MemberEnrichmentSource,
-  MemberIdentityType,
-  PlatformType,
-} from '@crowd/types'
+import { IEnrichableMember, MemberEnrichmentSource } from '@crowd/types'
 
 import * as activities from '../activities'
 import { IEnrichmentSourceInput } from '../types'
@@ -19,7 +14,7 @@ const {
   updateMemberEnrichmentCache,
   isCacheObsolete,
   normalizeEnrichmentData,
-  findMemberIdentityWithTheMostActivityInPlatform,
+  getEnrichmentInput,
 } = proxyActivities<typeof activities>({
   startToCloseTimeout: '1 minute',
   retry: {
@@ -42,44 +37,7 @@ export async function enrichMember(
 
     // cache is obsolete when it's not found or cache.updatedAt is older than cacheObsoleteAfterSeconds
     if (await isCacheObsolete(source, cache)) {
-      const enrichmentInput: IEnrichmentSourceInput = {
-        memberId: input.id,
-        email: input.identities.find((i) => i.verified && i.type === MemberIdentityType.EMAIL),
-        linkedin: input.identities.find(
-          (i) =>
-            i.verified &&
-            i.platform === PlatformType.LINKEDIN &&
-            i.type === MemberIdentityType.USERNAME,
-        ),
-        displayName: input.displayName || undefined,
-        website: input.website || undefined,
-        location: input.location || undefined,
-        activityCount: input.activityCount || 0,
-      }
-
-      // there can be multiple verified identities in github, we select the one with the most activities
-      const verifiedGithubIdentities = input.identities.filter(
-        (i) =>
-          i.verified &&
-          i.platform === PlatformType.GITHUB &&
-          i.type === MemberIdentityType.USERNAME,
-      )
-
-      if (verifiedGithubIdentities.length > 1) {
-        const ghIdentityWithTheMostActivities =
-          await findMemberIdentityWithTheMostActivityInPlatform(input.id, PlatformType.GITHUB)
-        if (ghIdentityWithTheMostActivities) {
-          enrichmentInput.github = input.identities.find(
-            (i) =>
-              i.verified &&
-              i.platform === PlatformType.GITHUB &&
-              i.type === MemberIdentityType.USERNAME &&
-              i.value === ghIdentityWithTheMostActivities.username,
-          )
-        }
-      } else {
-        enrichmentInput.github = verifiedGithubIdentities?.[0] || undefined
-      }
+      const enrichmentInput: IEnrichmentSourceInput = await getEnrichmentInput(input)
 
       const data = await getEnrichmentData(source, enrichmentInput)