From 7ace0d29c4508d66418a733457b38005e8e18858 Mon Sep 17 00:00:00 2001 From: anilb Date: Fri, 8 Nov 2024 22:45:39 +0100 Subject: [PATCH 01/16] fix crustdata monitoring query --- .../migrations/R__memberEnrichmentMaterializedViews.sql | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/backend/src/database/migrations/R__memberEnrichmentMaterializedViews.sql b/backend/src/database/migrations/R__memberEnrichmentMaterializedViews.sql index d9588ea398..68d35392b1 100644 --- a/backend/src/database/migrations/R__memberEnrichmentMaterializedViews.sql +++ b/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 ( From 08677f081eb5dbff0a1184c0c3da4fb3afaa3bc3 Mon Sep 17 00:00:00 2001 From: anilb Date: Sat, 9 Nov 2024 19:04:47 +0100 Subject: [PATCH 02/16] setting max concurrent requests dynamically by using enrichable sources --- .../src/activities.ts | 5 +- .../src/activities/enrichment.ts | 48 +++++++++++++++++++ .../src/activities/getMembers.ts | 36 ++++++++++++++ .../src/sources/clearbit/service.ts | 8 +++- .../src/sources/crustdata/service.ts | 2 + .../progai-linkedin-scraper/service.ts | 2 + .../src/sources/progai/service.ts | 4 +- .../src/sources/serp/service.ts | 2 + .../members_enrichment_worker/src/types.ts | 3 ++ .../src/workflows/enrichMember.ts | 48 ++----------------- .../src/workflows/getMembersToEnrich.ts | 7 +-- 11 files changed, 114 insertions(+), 51 deletions(-) diff --git a/services/apps/premium/members_enrichment_worker/src/activities.ts b/services/apps/premium/members_enrichment_worker/src/activities.ts index addb2befb5..6550e8a127 100644 --- a/services/apps/premium/members_enrichment_worker/src/activities.ts +++ b/services/apps/premium/members_enrichment_worker/src/activities.ts @@ -2,6 +2,7 @@ import { findMemberEnrichmentCache, findMemberIdentityWithTheMostActivityInPlatform, getEnrichmentData, + getEnrichmentInput, insertMemberEnrichmentCache, isCacheObsolete, isEnrichableBySource, @@ -10,7 +11,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 +55,6 @@ export { isEnrichableBySource, findMemberIdentityWithTheMostActivityInPlatform, refreshMemberEnrichmentMaterializedView, + getEnrichmentInput, + getMaxConcurrentRequests, } diff --git a/services/apps/premium/members_enrichment_worker/src/activities/enrichment.ts b/services/apps/premium/members_enrichment_worker/src/activities/enrichment.ts index d19031f52b..99efa7da49 100644 --- a/services/apps/premium/members_enrichment_worker/src/activities/enrichment.ts +++ b/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 { + 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, diff --git a/services/apps/premium/members_enrichment_worker/src/activities/getMembers.ts b/services/apps/premium/members_enrichment_worker/src/activities/getMembers.ts index 09698a6512..5def7a0c97 100644 --- a/services/apps/premium/members_enrichment_worker/src/activities/getMembers.ts +++ b/services/apps/premium/members_enrichment_worker/src/activities/getMembers.ts @@ -7,6 +7,9 @@ import { import { EnrichmentSourceServiceFactory } from '../factory' import { svc } from '../main' +import { IEnrichmentService } from '../types' + +import { getEnrichmentInput } from './enrichment' export async function getEnrichableMembers( limit: number, @@ -27,3 +30,36 @@ export async function getEnrichableMembers( return rows } + +export async function getMaxConcurrentRequests( + members: IEnrichableMember[], + possibleSources: MemberEnrichmentSource[], +): Promise { + const serviceMap: Partial> = {} + const set = new Set() + + for (const source of possibleSources) { + serviceMap[source] = EnrichmentSourceServiceFactory.getEnrichmentSourceService(source, svc.log) + } + for (const member of members) { + const enrichmentInput = await getEnrichmentInput(member) + Object.keys(serviceMap).forEach(async (source) => { + if (await serviceMap[source].isEnrichableBySource(enrichmentInput)) { + set.add(source as MemberEnrichmentSource) + } + }) + } + + let smallestMaxConcurrentRequests = Infinity + + Array.from(set).forEach(async (source) => { + smallestMaxConcurrentRequests = Math.min( + smallestMaxConcurrentRequests, + serviceMap[source].maxConcurrentRequests, + ) + }) + + svc.log.info('Setting max concurrent requests', { smallestMaxConcurrentRequests }) + + return smallestMaxConcurrentRequests +} diff --git a/services/apps/premium/members_enrichment_worker/src/sources/clearbit/service.ts b/services/apps/premium/members_enrichment_worker/src/sources/clearbit/service.ts index 7cfbb25df6..b505cfe52f 100644 --- a/services/apps/premium/members_enrichment_worker/src/sources/clearbit/service.ts +++ b/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 { - return !!input.email?.value && input.email?.verified + return ( + input.activityCount > this.enrichMembersWithActivityMoreThan && + !!input.email?.value && + input.email?.verified + ) } async getData(input: IEnrichmentSourceInput): Promise { diff --git a/services/apps/premium/members_enrichment_worker/src/sources/crustdata/service.ts b/services/apps/premium/members_enrichment_worker/src/sources/crustdata/service.ts index 80aa3abebb..ba65dddbff 100644 --- a/services/apps/premium/members_enrichment_worker/src/sources/crustdata/service.ts +++ b/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'], diff --git a/services/apps/premium/members_enrichment_worker/src/sources/progai-linkedin-scraper/service.ts b/services/apps/premium/members_enrichment_worker/src/sources/progai-linkedin-scraper/service.ts index b0b75f9733..0e683ac2b3 100644 --- a/services/apps/premium/members_enrichment_worker/src/sources/progai-linkedin-scraper/service.ts +++ b/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) } diff --git a/services/apps/premium/members_enrichment_worker/src/sources/progai/service.ts b/services/apps/premium/members_enrichment_worker/src/sources/progai/service.ts index 5b6dc736f4..4d5bf149a2 100644 --- a/services/apps/premium/members_enrichment_worker/src/sources/progai/service.ts +++ b/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'], diff --git a/services/apps/premium/members_enrichment_worker/src/sources/serp/service.ts b/services/apps/premium/members_enrichment_worker/src/sources/serp/service.ts index 7559480dae..6a93a08042 100644 --- a/services/apps/premium/members_enrichment_worker/src/sources/serp/service.ts +++ b/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) } diff --git a/services/apps/premium/members_enrichment_worker/src/types.ts b/services/apps/premium/members_enrichment_worker/src/types.ts index e51bba2547..738d11c3cf 100644 --- a/services/apps/premium/members_enrichment_worker/src/types.ts +++ b/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 diff --git a/services/apps/premium/members_enrichment_worker/src/workflows/enrichMember.ts b/services/apps/premium/members_enrichment_worker/src/workflows/enrichMember.ts index 90b24dfcf1..c1ca4fc3ea 100644 --- a/services/apps/premium/members_enrichment_worker/src/workflows/enrichMember.ts +++ b/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({ 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) diff --git a/services/apps/premium/members_enrichment_worker/src/workflows/getMembersToEnrich.ts b/services/apps/premium/members_enrichment_worker/src/workflows/getMembersToEnrich.ts index b5255357d6..5d02a12cf9 100644 --- a/services/apps/premium/members_enrichment_worker/src/workflows/getMembersToEnrich.ts +++ b/services/apps/premium/members_enrichment_worker/src/workflows/getMembersToEnrich.ts @@ -14,13 +14,12 @@ import { chunkArray } from '../utils/common' import { enrichMember } from './enrichMember' -const { getEnrichableMembers } = proxyActivities({ +const { getEnrichableMembers, getMaxConcurrentRequests } = proxyActivities({ startToCloseTimeout: '2 minutes', }) export async function getMembersToEnrich(args: IGetMembersForEnrichmentArgs): Promise { const QUERY_FOR_ENRICHABLE_MEMBERS_PER_RUN = 1000 - const PARALLEL_ENRICHMENT_WORKFLOWS = 5 const afterCursor = args?.afterCursor || null const sources = [ MemberEnrichmentSource.PROGAI, @@ -40,7 +39,9 @@ export async function getMembersToEnrich(args: IGetMembersForEnrichmentArgs): Pr return } - const chunks = chunkArray(members, PARALLEL_ENRICHMENT_WORKFLOWS) + const parallelEnrichmentWorkflows = await getMaxConcurrentRequests(members, sources) + + const chunks = chunkArray(members, parallelEnrichmentWorkflows) for (const chunk of chunks) { await Promise.all( From 2ffa4a4f6fbc99d5daf8f7d62abaa894acd6ab4a Mon Sep 17 00:00:00 2001 From: anilb Date: Sat, 9 Nov 2024 19:06:13 +0100 Subject: [PATCH 03/16] variable name update --- .../members_enrichment_worker/src/activities/getMembers.ts | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/services/apps/premium/members_enrichment_worker/src/activities/getMembers.ts b/services/apps/premium/members_enrichment_worker/src/activities/getMembers.ts index 5def7a0c97..f609ea2b92 100644 --- a/services/apps/premium/members_enrichment_worker/src/activities/getMembers.ts +++ b/services/apps/premium/members_enrichment_worker/src/activities/getMembers.ts @@ -36,7 +36,7 @@ export async function getMaxConcurrentRequests( possibleSources: MemberEnrichmentSource[], ): Promise { const serviceMap: Partial> = {} - const set = new Set() + const distinctEnrichableSources = new Set() for (const source of possibleSources) { serviceMap[source] = EnrichmentSourceServiceFactory.getEnrichmentSourceService(source, svc.log) @@ -45,14 +45,14 @@ export async function getMaxConcurrentRequests( const enrichmentInput = await getEnrichmentInput(member) Object.keys(serviceMap).forEach(async (source) => { if (await serviceMap[source].isEnrichableBySource(enrichmentInput)) { - set.add(source as MemberEnrichmentSource) + distinctEnrichableSources.add(source as MemberEnrichmentSource) } }) } let smallestMaxConcurrentRequests = Infinity - Array.from(set).forEach(async (source) => { + Array.from(distinctEnrichableSources).forEach(async (source) => { smallestMaxConcurrentRequests = Math.min( smallestMaxConcurrentRequests, serviceMap[source].maxConcurrentRequests, From f1e15327ef6b6e1fe14edeeabb575ebe7895cf6e Mon Sep 17 00:00:00 2001 From: anilb Date: Sun, 10 Nov 2024 12:29:41 +0100 Subject: [PATCH 04/16] improved getMaxConcurrentRequests --- .../src/activities.ts | 2 ++ .../src/activities/enrichment.ts | 21 ++++++++++++++++++- .../src/activities/getMembers.ts | 8 +++++-- .../members_enrichment_worker/index.ts | 4 +++- 4 files changed, 31 insertions(+), 4 deletions(-) diff --git a/services/apps/premium/members_enrichment_worker/src/activities.ts b/services/apps/premium/members_enrichment_worker/src/activities.ts index 6550e8a127..0805b9a6a7 100644 --- a/services/apps/premium/members_enrichment_worker/src/activities.ts +++ b/services/apps/premium/members_enrichment_worker/src/activities.ts @@ -3,6 +3,7 @@ import { findMemberIdentityWithTheMostActivityInPlatform, getEnrichmentData, getEnrichmentInput, + getObsoleteSourcesOfMember, insertMemberEnrichmentCache, isCacheObsolete, isEnrichableBySource, @@ -57,4 +58,5 @@ export { refreshMemberEnrichmentMaterializedView, getEnrichmentInput, getMaxConcurrentRequests, + getObsoleteSourcesOfMember, } diff --git a/services/apps/premium/members_enrichment_worker/src/activities/enrichment.ts b/services/apps/premium/members_enrichment_worker/src/activities/enrichment.ts index 99efa7da49..80edafd6cb 100644 --- a/services/apps/premium/members_enrichment_worker/src/activities/enrichment.ts +++ b/services/apps/premium/members_enrichment_worker/src/activities/enrichment.ts @@ -152,8 +152,13 @@ export async function findMemberEnrichmentCache( export async function findMemberEnrichmentCacheForAllSources( memberId: string, + returnRowsWithoutData = false, ): Promise[]> { - return findMemberEnrichmentCacheForAllSourcesDb(svc.postgres.reader.connection(), memberId) + return findMemberEnrichmentCacheForAllSourcesDb( + svc.postgres.reader.connection(), + memberId, + returnRowsWithoutData, + ) } export async function insertMemberEnrichmentCache( @@ -186,6 +191,20 @@ export async function findMemberIdentityWithTheMostActivityInPlatform( return findMemberIdentityWithTheMostActivityInPlatformQuestDb(svc.questdbSQL, memberId, platform) } +export async function getObsoleteSourcesOfMember( + memberId: string, + possibleSources: MemberEnrichmentSource[], +): Promise { + const caches = await findMemberEnrichmentCacheForAllSources(memberId, true) + const obsoleteSources = possibleSources.filter((source) => + isCacheObsolete( + source, + caches.find((s) => s.source === source), + ), + ) + return obsoleteSources +} + export async function refreshMemberEnrichmentMaterializedView(mvName: string): Promise { await refreshMaterializedView(svc.postgres.writer.connection(), mvName) } diff --git a/services/apps/premium/members_enrichment_worker/src/activities/getMembers.ts b/services/apps/premium/members_enrichment_worker/src/activities/getMembers.ts index f609ea2b92..98d11b4c2c 100644 --- a/services/apps/premium/members_enrichment_worker/src/activities/getMembers.ts +++ b/services/apps/premium/members_enrichment_worker/src/activities/getMembers.ts @@ -9,7 +9,7 @@ import { EnrichmentSourceServiceFactory } from '../factory' import { svc } from '../main' import { IEnrichmentService } from '../types' -import { getEnrichmentInput } from './enrichment' +import { getEnrichmentInput, getObsoleteSourcesOfMember } from './enrichment' export async function getEnrichableMembers( limit: number, @@ -44,7 +44,11 @@ export async function getMaxConcurrentRequests( for (const member of members) { const enrichmentInput = await getEnrichmentInput(member) Object.keys(serviceMap).forEach(async (source) => { - if (await serviceMap[source].isEnrichableBySource(enrichmentInput)) { + const obsoleteSources = await getObsoleteSourcesOfMember(member.id, possibleSources) + if ( + (await serviceMap[source].isEnrichableBySource(enrichmentInput)) && + (obsoleteSources as string[]).includes(source) + ) { distinctEnrichableSources.add(source as MemberEnrichmentSource) } }) diff --git a/services/libs/data-access-layer/src/old/apps/premium/members_enrichment_worker/index.ts b/services/libs/data-access-layer/src/old/apps/premium/members_enrichment_worker/index.ts index 46d719e3f0..25edabab02 100644 --- a/services/libs/data-access-layer/src/old/apps/premium/members_enrichment_worker/index.ts +++ b/services/libs/data-access-layer/src/old/apps/premium/members_enrichment_worker/index.ts @@ -523,13 +523,15 @@ export async function findMemberEnrichmentCacheDb( export async function findMemberEnrichmentCacheForAllSourcesDb( tx: DbConnOrTx, memberId: string, + returnRowsWithoutData = false, ): Promise[]> { + const dataFilter = returnRowsWithoutData ? '' : 'and data is not null' const result = await tx.manyOrNone( ` select * from "memberEnrichmentCache" where - "memberId" = $(memberId) and data is not null; + "memberId" = $(memberId) ${dataFilter}; `, { memberId }, ) From 59595855227714ff86e08dfbbf9e4415b9d29eb5 Mon Sep 17 00:00:00 2001 From: anilb Date: Sun, 10 Nov 2024 12:57:32 +0100 Subject: [PATCH 05/16] add some logging --- .../members_enrichment_worker/src/activities/getMembers.ts | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/services/apps/premium/members_enrichment_worker/src/activities/getMembers.ts b/services/apps/premium/members_enrichment_worker/src/activities/getMembers.ts index 98d11b4c2c..3d463f1a62 100644 --- a/services/apps/premium/members_enrichment_worker/src/activities/getMembers.ts +++ b/services/apps/premium/members_enrichment_worker/src/activities/getMembers.ts @@ -43,8 +43,9 @@ export async function getMaxConcurrentRequests( } for (const member of members) { const enrichmentInput = await getEnrichmentInput(member) + const obsoleteSources = await getObsoleteSourcesOfMember(member.id, possibleSources) + Object.keys(serviceMap).forEach(async (source) => { - const obsoleteSources = await getObsoleteSourcesOfMember(member.id, possibleSources) if ( (await serviceMap[source].isEnrichableBySource(enrichmentInput)) && (obsoleteSources as string[]).includes(source) @@ -56,6 +57,8 @@ export async function getMaxConcurrentRequests( let smallestMaxConcurrentRequests = Infinity + svc.log.info('Distinct enrichable sources', { distinctEnrichableSources }) + Array.from(distinctEnrichableSources).forEach(async (source) => { smallestMaxConcurrentRequests = Math.min( smallestMaxConcurrentRequests, From 03d8a43b76c15e31e5a23b6e404724431690b39c Mon Sep 17 00:00:00 2001 From: anilb Date: Sun, 10 Nov 2024 13:10:35 +0100 Subject: [PATCH 06/16] testing --- .../src/sources/clearbit/service.ts | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/services/apps/premium/members_enrichment_worker/src/sources/clearbit/service.ts b/services/apps/premium/members_enrichment_worker/src/sources/clearbit/service.ts index b505cfe52f..0fc3ef138c 100644 --- a/services/apps/premium/members_enrichment_worker/src/sources/clearbit/service.ts +++ b/services/apps/premium/members_enrichment_worker/src/sources/clearbit/service.ts @@ -61,8 +61,9 @@ export default class EnrichmentServiceClearbit extends LoggerBase implements IEn async isEnrichableBySource(input: IEnrichmentSourceInput): Promise { return ( input.activityCount > this.enrichMembersWithActivityMoreThan && - !!input.email?.value && - input.email?.verified + input.email && + input.email.value && + input.email.verified ) } From 594574ca76af77cfd708c8e1b490f05114433e04 Mon Sep 17 00:00:00 2001 From: anilb Date: Sun, 10 Nov 2024 13:25:28 +0100 Subject: [PATCH 07/16] comments, tweaks --- .../members_enrichment_worker/src/activities/getMembers.ts | 5 +++-- .../src/sources/clearbit/service.ts | 5 ++--- 2 files changed, 5 insertions(+), 5 deletions(-) diff --git a/services/apps/premium/members_enrichment_worker/src/activities/getMembers.ts b/services/apps/premium/members_enrichment_worker/src/activities/getMembers.ts index 3d463f1a62..492f217fd6 100644 --- a/services/apps/premium/members_enrichment_worker/src/activities/getMembers.ts +++ b/services/apps/premium/members_enrichment_worker/src/activities/getMembers.ts @@ -31,6 +31,9 @@ export async function getEnrichableMembers( return rows } +// Get the most strict parallelism among existing and enrichable sources +// If current members are only enrichable by one source, we will use the maxConcurrentRequests of that source +// If current members are enrichable by multiple sources, we will use the min(maxConcurrentRequests) among sources export async function getMaxConcurrentRequests( members: IEnrichableMember[], possibleSources: MemberEnrichmentSource[], @@ -57,8 +60,6 @@ export async function getMaxConcurrentRequests( let smallestMaxConcurrentRequests = Infinity - svc.log.info('Distinct enrichable sources', { distinctEnrichableSources }) - Array.from(distinctEnrichableSources).forEach(async (source) => { smallestMaxConcurrentRequests = Math.min( smallestMaxConcurrentRequests, diff --git a/services/apps/premium/members_enrichment_worker/src/sources/clearbit/service.ts b/services/apps/premium/members_enrichment_worker/src/sources/clearbit/service.ts index 0fc3ef138c..b505cfe52f 100644 --- a/services/apps/premium/members_enrichment_worker/src/sources/clearbit/service.ts +++ b/services/apps/premium/members_enrichment_worker/src/sources/clearbit/service.ts @@ -61,9 +61,8 @@ export default class EnrichmentServiceClearbit extends LoggerBase implements IEn async isEnrichableBySource(input: IEnrichmentSourceInput): Promise { return ( input.activityCount > this.enrichMembersWithActivityMoreThan && - input.email && - input.email.value && - input.email.verified + !!input.email?.value && + input.email?.verified ) } From 9f08e8372600780f4418a1438fb4d3cde9f8938c Mon Sep 17 00:00:00 2001 From: anilb Date: Sun, 10 Nov 2024 13:28:23 +0100 Subject: [PATCH 08/16] improve comments --- .../members_enrichment_worker/src/activities/getMembers.ts | 1 - 1 file changed, 1 deletion(-) diff --git a/services/apps/premium/members_enrichment_worker/src/activities/getMembers.ts b/services/apps/premium/members_enrichment_worker/src/activities/getMembers.ts index 492f217fd6..f2b597e804 100644 --- a/services/apps/premium/members_enrichment_worker/src/activities/getMembers.ts +++ b/services/apps/premium/members_enrichment_worker/src/activities/getMembers.ts @@ -32,7 +32,6 @@ export async function getEnrichableMembers( } // Get the most strict parallelism among existing and enrichable sources -// If current members are only enrichable by one source, we will use the maxConcurrentRequests of that source // If current members are enrichable by multiple sources, we will use the min(maxConcurrentRequests) among sources export async function getMaxConcurrentRequests( members: IEnrichableMember[], From e3a4fb3941cf42d1de3ed6283bfe40a265b8793b Mon Sep 17 00:00:00 2001 From: anilb Date: Sun, 10 Nov 2024 14:14:07 +0100 Subject: [PATCH 09/16] fix getObsoleteSourcesOfMember --- .../src/activities/enrichment.ts | 16 ++++++++++------ 1 file changed, 10 insertions(+), 6 deletions(-) diff --git a/services/apps/premium/members_enrichment_worker/src/activities/enrichment.ts b/services/apps/premium/members_enrichment_worker/src/activities/enrichment.ts index 80edafd6cb..84a7a8e8b2 100644 --- a/services/apps/premium/members_enrichment_worker/src/activities/enrichment.ts +++ b/services/apps/premium/members_enrichment_worker/src/activities/enrichment.ts @@ -196,13 +196,17 @@ export async function getObsoleteSourcesOfMember( possibleSources: MemberEnrichmentSource[], ): Promise { const caches = await findMemberEnrichmentCacheForAllSources(memberId, true) - const obsoleteSources = possibleSources.filter((source) => - isCacheObsolete( - source, - caches.find((s) => s.source === source), - ), + + const obsoleteSourcesPromises = possibleSources.map(async (source) => { + const cache = caches.find((s) => s.source === source) + return (await isCacheObsolete(source, cache)) ? source : null + }) + + const obsoleteSources = (await Promise.all(obsoleteSourcesPromises)).filter( + (source) => source !== null, ) - return obsoleteSources + + return obsoleteSources as MemberEnrichmentSource[] } export async function refreshMemberEnrichmentMaterializedView(mvName: string): Promise { From 3cb5ef3626c4624e9ef6b6355366671c1c78be26 Mon Sep 17 00:00:00 2001 From: anilb Date: Sun, 10 Nov 2024 14:28:24 +0100 Subject: [PATCH 10/16] fix iisCacheObsoleteSync --- .../src/activities/enrichment.ts | 24 +++++++------------ .../src/utils/common.ts | 21 +++++++++++++++- 2 files changed, 29 insertions(+), 16 deletions(-) diff --git a/services/apps/premium/members_enrichment_worker/src/activities/enrichment.ts b/services/apps/premium/members_enrichment_worker/src/activities/enrichment.ts index 84a7a8e8b2..ae77e84c46 100644 --- a/services/apps/premium/members_enrichment_worker/src/activities/enrichment.ts +++ b/services/apps/premium/members_enrichment_worker/src/activities/enrichment.ts @@ -24,6 +24,7 @@ import { IMemberEnrichmentData, IMemberEnrichmentDataNormalized, } from '../types' +import { isCacheObsolete as isCacheObsoleteSync } from '../utils/common' export async function isEnrichableBySource( source: MemberEnrichmentSource, @@ -101,11 +102,7 @@ export async function isCacheObsolete( source: MemberEnrichmentSource, cache: IMemberEnrichmentCache, ): Promise { - const service = EnrichmentSourceServiceFactory.getEnrichmentSourceService(source, svc.log) - return ( - !cache || - Date.now() - new Date(cache.updatedAt).getTime() > 1000 * service.cacheObsoleteAfterSeconds - ) + return isCacheObsoleteSync(source, cache, svc.log) } export async function setHasRemainingCredits( @@ -196,17 +193,14 @@ export async function getObsoleteSourcesOfMember( possibleSources: MemberEnrichmentSource[], ): Promise { const caches = await findMemberEnrichmentCacheForAllSources(memberId, true) - - const obsoleteSourcesPromises = possibleSources.map(async (source) => { - const cache = caches.find((s) => s.source === source) - return (await isCacheObsolete(source, cache)) ? source : null - }) - - const obsoleteSources = (await Promise.all(obsoleteSourcesPromises)).filter( - (source) => source !== null, + const obsoleteSources = possibleSources.filter((source) => + isCacheObsoleteSync( + source, + caches.find((s) => s.source === source), + svc.log, + ), ) - - return obsoleteSources as MemberEnrichmentSource[] + return obsoleteSources } export async function refreshMemberEnrichmentMaterializedView(mvName: string): Promise { diff --git a/services/apps/premium/members_enrichment_worker/src/utils/common.ts b/services/apps/premium/members_enrichment_worker/src/utils/common.ts index 0a62747909..949222b869 100644 --- a/services/apps/premium/members_enrichment_worker/src/utils/common.ts +++ b/services/apps/premium/members_enrichment_worker/src/utils/common.ts @@ -1,7 +1,14 @@ import lodash from 'lodash' -import { IMemberEnrichmentCache, MemberIdentityType, PlatformType } from '@crowd/types' +import { Logger } from '@crowd/logging' +import { + IMemberEnrichmentCache, + MemberEnrichmentSource, + MemberIdentityType, + PlatformType, +} from '@crowd/types' +import { EnrichmentSourceServiceFactory } from '../factory' import { IMemberEnrichmentAttributeSettings, IMemberEnrichmentData, @@ -108,3 +115,15 @@ export function chunkArray(array: T[], chunkSize: number): T[][] { } return chunks } + +export function isCacheObsolete( + source: MemberEnrichmentSource, + cache: IMemberEnrichmentCache, + logger: Logger, +): boolean { + const service = EnrichmentSourceServiceFactory.getEnrichmentSourceService(source, logger) + return ( + !cache || + Date.now() - new Date(cache.updatedAt).getTime() > 1000 * service.cacheObsoleteAfterSeconds + ) +} From 1980bd9fc81963f36a0020e8e07affa3285f8d61 Mon Sep 17 00:00:00 2001 From: anilb Date: Sun, 10 Nov 2024 14:36:50 +0100 Subject: [PATCH 11/16] fix import issues --- .../src/activities/enrichment.ts | 15 ++++++++++++--- .../members_enrichment_worker/src/utils/common.ts | 14 +------------- 2 files changed, 13 insertions(+), 16 deletions(-) diff --git a/services/apps/premium/members_enrichment_worker/src/activities/enrichment.ts b/services/apps/premium/members_enrichment_worker/src/activities/enrichment.ts index ae77e84c46..d0f82f106e 100644 --- a/services/apps/premium/members_enrichment_worker/src/activities/enrichment.ts +++ b/services/apps/premium/members_enrichment_worker/src/activities/enrichment.ts @@ -24,7 +24,6 @@ import { IMemberEnrichmentData, IMemberEnrichmentDataNormalized, } from '../types' -import { isCacheObsolete as isCacheObsoleteSync } from '../utils/common' export async function isEnrichableBySource( source: MemberEnrichmentSource, @@ -102,7 +101,18 @@ export async function isCacheObsolete( source: MemberEnrichmentSource, cache: IMemberEnrichmentCache, ): Promise { - return isCacheObsoleteSync(source, cache, svc.log) + return isCacheObsoleteSync(source, cache) +} + +export function isCacheObsoleteSync( + source: MemberEnrichmentSource, + cache: IMemberEnrichmentCache, +): boolean { + const service = EnrichmentSourceServiceFactory.getEnrichmentSourceService(source, svc.log) + return ( + !cache || + Date.now() - new Date(cache.updatedAt).getTime() > 1000 * service.cacheObsoleteAfterSeconds + ) } export async function setHasRemainingCredits( @@ -197,7 +207,6 @@ export async function getObsoleteSourcesOfMember( isCacheObsoleteSync( source, caches.find((s) => s.source === source), - svc.log, ), ) return obsoleteSources diff --git a/services/apps/premium/members_enrichment_worker/src/utils/common.ts b/services/apps/premium/members_enrichment_worker/src/utils/common.ts index 949222b869..96f1149bc0 100644 --- a/services/apps/premium/members_enrichment_worker/src/utils/common.ts +++ b/services/apps/premium/members_enrichment_worker/src/utils/common.ts @@ -114,16 +114,4 @@ export function chunkArray(array: T[], chunkSize: number): T[][] { chunks.push(array.slice(i, i + chunkSize)) } return chunks -} - -export function isCacheObsolete( - source: MemberEnrichmentSource, - cache: IMemberEnrichmentCache, - logger: Logger, -): boolean { - const service = EnrichmentSourceServiceFactory.getEnrichmentSourceService(source, logger) - return ( - !cache || - Date.now() - new Date(cache.updatedAt).getTime() > 1000 * service.cacheObsoleteAfterSeconds - ) -} +} \ No newline at end of file From 3788cc0ec0c0888bca2b2e32ff0befd684bb3c63 Mon Sep 17 00:00:00 2001 From: anilb Date: Sun, 10 Nov 2024 14:58:33 +0100 Subject: [PATCH 12/16] force redeploy --- .../members_enrichment_worker/src/activities/enrichment.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/services/apps/premium/members_enrichment_worker/src/activities/enrichment.ts b/services/apps/premium/members_enrichment_worker/src/activities/enrichment.ts index d0f82f106e..fc59fead34 100644 --- a/services/apps/premium/members_enrichment_worker/src/activities/enrichment.ts +++ b/services/apps/premium/members_enrichment_worker/src/activities/enrichment.ts @@ -206,7 +206,7 @@ export async function getObsoleteSourcesOfMember( const obsoleteSources = possibleSources.filter((source) => isCacheObsoleteSync( source, - caches.find((s) => s.source === source), + caches.find((c) => c.source === source), ), ) return obsoleteSources From 877368bde7dff9851078b5223f8e60770b856762 Mon Sep 17 00:00:00 2001 From: anilb Date: Sun, 10 Nov 2024 15:07:38 +0100 Subject: [PATCH 13/16] fix --- .../src/activities/getMembers.ts | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/services/apps/premium/members_enrichment_worker/src/activities/getMembers.ts b/services/apps/premium/members_enrichment_worker/src/activities/getMembers.ts index f2b597e804..e97654ec6d 100644 --- a/services/apps/premium/members_enrichment_worker/src/activities/getMembers.ts +++ b/services/apps/premium/members_enrichment_worker/src/activities/getMembers.ts @@ -39,9 +39,16 @@ export async function getMaxConcurrentRequests( ): Promise { const serviceMap: Partial> = {} const distinctEnrichableSources = new Set() + let maxConcurrentRequestsInAllSources = 0 for (const source of possibleSources) { serviceMap[source] = EnrichmentSourceServiceFactory.getEnrichmentSourceService(source, svc.log) + if (serviceMap[source].maxConcurrentRequests > maxConcurrentRequestsInAllSources) { + maxConcurrentRequestsInAllSources = Math.max( + maxConcurrentRequestsInAllSources, + serviceMap[source].maxConcurrentRequests, + ) + } } for (const member of members) { const enrichmentInput = await getEnrichmentInput(member) @@ -57,7 +64,7 @@ export async function getMaxConcurrentRequests( }) } - let smallestMaxConcurrentRequests = Infinity + let smallestMaxConcurrentRequests = maxConcurrentRequestsInAllSources Array.from(distinctEnrichableSources).forEach(async (source) => { smallestMaxConcurrentRequests = Math.min( From 50aedb7ddc6609f0c9e10d301188f4877f03b753 Mon Sep 17 00:00:00 2001 From: anilb Date: Mon, 11 Nov 2024 00:06:27 +0100 Subject: [PATCH 14/16] improve and simplify pagination --- .../src/activities/getMembers.ts | 3 +-- .../src/schedules/getMembersToEnrich.ts | 6 +---- .../members_enrichment_worker/src/types.ts | 4 ---- .../src/workflows/getMembersToEnrich.ts | 17 +++---------- .../members_enrichment_worker/index.ts | 24 ++++++++++--------- 5 files changed, 18 insertions(+), 36 deletions(-) diff --git a/services/apps/premium/members_enrichment_worker/src/activities/getMembers.ts b/services/apps/premium/members_enrichment_worker/src/activities/getMembers.ts index e97654ec6d..fdf03713b9 100644 --- a/services/apps/premium/members_enrichment_worker/src/activities/getMembers.ts +++ b/services/apps/premium/members_enrichment_worker/src/activities/getMembers.ts @@ -14,7 +14,6 @@ import { getEnrichmentInput, getObsoleteSourcesOfMember } from './enrichment' export async function getEnrichableMembers( limit: number, sources: MemberEnrichmentSource[], - afterCursor: { activityCount: number; memberId: string } | null, ): Promise { let rows: IEnrichableMember[] = [] const sourceInputs: IMemberEnrichmentSourceQueryInput[] = sources.map((s) => { @@ -26,7 +25,7 @@ export async function getEnrichableMembers( } }) const db = svc.postgres.reader - rows = await fetchMembersForEnrichment(db, limit, sourceInputs, afterCursor) + rows = await fetchMembersForEnrichment(db, limit, sourceInputs) return rows } diff --git a/services/apps/premium/members_enrichment_worker/src/schedules/getMembersToEnrich.ts b/services/apps/premium/members_enrichment_worker/src/schedules/getMembersToEnrich.ts index 948fc69400..e8aa7629a3 100644 --- a/services/apps/premium/members_enrichment_worker/src/schedules/getMembersToEnrich.ts +++ b/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) { diff --git a/services/apps/premium/members_enrichment_worker/src/types.ts b/services/apps/premium/members_enrichment_worker/src/types.ts index 738d11c3cf..1003739a65 100644 --- a/services/apps/premium/members_enrichment_worker/src/types.ts +++ b/services/apps/premium/members_enrichment_worker/src/types.ts @@ -91,10 +91,6 @@ export interface IMemberEnrichmentLinkedinScraperMetadata { isFromVerifiedSource: boolean } -export interface IGetMembersForEnrichmentArgs { - afterCursor: { activityCount: number; memberId: string } | null -} - export interface IMemberEnrichmentSocialData { platform: PlatformType handle: string diff --git a/services/apps/premium/members_enrichment_worker/src/workflows/getMembersToEnrich.ts b/services/apps/premium/members_enrichment_worker/src/workflows/getMembersToEnrich.ts index 5d02a12cf9..2c3515934d 100644 --- a/services/apps/premium/members_enrichment_worker/src/workflows/getMembersToEnrich.ts +++ b/services/apps/premium/members_enrichment_worker/src/workflows/getMembersToEnrich.ts @@ -9,7 +9,6 @@ import { import { IEnrichableMember, MemberEnrichmentSource } from '@crowd/types' import * as activities from '../activities/getMembers' -import { IGetMembersForEnrichmentArgs } from '../types' import { chunkArray } from '../utils/common' import { enrichMember } from './enrichMember' @@ -18,9 +17,8 @@ const { getEnrichableMembers, getMaxConcurrentRequests } = proxyActivities { +export async function getMembersToEnrich(): Promise { const QUERY_FOR_ENRICHABLE_MEMBERS_PER_RUN = 1000 - const afterCursor = args?.afterCursor || null const sources = [ MemberEnrichmentSource.PROGAI, MemberEnrichmentSource.CLEARBIT, @@ -29,11 +27,7 @@ export async function getMembersToEnrich(args: IGetMembersForEnrichmentArgs): Pr MemberEnrichmentSource.CRUSTDATA, ] - const members = await getEnrichableMembers( - QUERY_FOR_ENRICHABLE_MEMBERS_PER_RUN, - sources, - afterCursor, - ) + const members = await getEnrichableMembers(QUERY_FOR_ENRICHABLE_MEMBERS_PER_RUN, sources) if (members.length === 0) { return @@ -66,10 +60,5 @@ export async function getMembersToEnrich(args: IGetMembersForEnrichmentArgs): Pr ) } - await continueAsNew({ - afterCursor: { - memberId: chunks[chunks.length - 1][chunks[chunks.length - 1].length - 1].id, - activityCount: chunks[chunks.length - 1][chunks[chunks.length - 1].length - 1].activityCount, - }, - }) + await continueAsNew() } diff --git a/services/libs/data-access-layer/src/old/apps/premium/members_enrichment_worker/index.ts b/services/libs/data-access-layer/src/old/apps/premium/members_enrichment_worker/index.ts index 25edabab02..63d939c6cf 100644 --- a/services/libs/data-access-layer/src/old/apps/premium/members_enrichment_worker/index.ts +++ b/services/libs/data-access-layer/src/old/apps/premium/members_enrichment_worker/index.ts @@ -12,21 +12,24 @@ import { OrganizationSource, } from '@crowd/types' +/** + * Gets enrichable members using the provided sources + * If a member is enrichable in one source, and not enrichable in another, the member will be returned + * Members with at least one missing or old source cache rows will be returned + * The reason we're not checking enrichable members and cache age in the same subquery is because of linkedin scraper sources. + * These sources also use data from other sources and it's costly to check cache data jsons. + * This check is instead done in the application layer. + */ export async function fetchMembersForEnrichment( db: DbStore, limit: number, sourceInputs: IMemberEnrichmentSourceQueryInput[], - afterCursor: { activityCount: number; memberId: string } | null, ): Promise { - const cursorFilter = afterCursor - ? `AND ((coalesce("membersGlobalActivityCount".total_count, 0) < $2) OR (coalesce("membersGlobalActivityCount".total_count, 0) = $2 AND members.id < $3))` - : '' - - const sourceInnerQueryItems = [] + const cacheAgeInnerQueryItems = [] const enrichableBySqlConditions = [] sourceInputs.forEach((input) => { - sourceInnerQueryItems.push( + cacheAgeInnerQueryItems.push( ` ( NOT EXISTS ( SELECT 1 FROM "memberEnrichmentCache" mec @@ -70,13 +73,12 @@ export async function fetchMembersForEnrichment( ${enrichableBySqlJoined} AND tenants."deletedAt" IS NULL AND members."deletedAt" IS NULL - AND (${sourceInnerQueryItems.join(' OR ')}) - ${cursorFilter} + AND (${cacheAgeInnerQueryItems.join(' OR ')}) GROUP BY members.id - ORDER BY "activityCount" DESC, members.id DESC + ORDER BY "activityCount" DESC LIMIT $1; `, - afterCursor ? [limit, afterCursor.activityCount, afterCursor.memberId] : [limit], + [limit], ) } From 761611b11916b8f00b00f5cd7aeaed28536ca183 Mon Sep 17 00:00:00 2001 From: anilb Date: Mon, 11 Nov 2024 01:21:05 +0100 Subject: [PATCH 15/16] better concurrency handling --- .../src/activities/getMembers.ts | 40 +++++-------------- .../members_enrichment_worker/src/types.ts | 3 ++ .../src/workflows/getMembersToEnrich.ts | 6 ++- 3 files changed, 18 insertions(+), 31 deletions(-) diff --git a/services/apps/premium/members_enrichment_worker/src/activities/getMembers.ts b/services/apps/premium/members_enrichment_worker/src/activities/getMembers.ts index fdf03713b9..0f45bcbd42 100644 --- a/services/apps/premium/members_enrichment_worker/src/activities/getMembers.ts +++ b/services/apps/premium/members_enrichment_worker/src/activities/getMembers.ts @@ -31,48 +31,28 @@ export async function getEnrichableMembers( } // Get the most strict parallelism among existing and enrichable sources -// If current members are enrichable by multiple sources, we will use the min(maxConcurrentRequests) among sources +// We only check sources that has activity count cutoff in current range export async function getMaxConcurrentRequests( members: IEnrichableMember[], possibleSources: MemberEnrichmentSource[], + concurrencyLimit: number, ): Promise { const serviceMap: Partial> = {} - const distinctEnrichableSources = new Set() - let maxConcurrentRequestsInAllSources = 0 + const currentProcessingActivityCount = members[0].activityCount + + let maxConcurrentRequestsInAllSources = concurrencyLimit for (const source of possibleSources) { serviceMap[source] = EnrichmentSourceServiceFactory.getEnrichmentSourceService(source, svc.log) - if (serviceMap[source].maxConcurrentRequests > maxConcurrentRequestsInAllSources) { - maxConcurrentRequestsInAllSources = Math.max( + const activityCountCutoff = serviceMap[source].enrichMembersWithActivityMoreThan + if (!activityCountCutoff || activityCountCutoff <= currentProcessingActivityCount) { + maxConcurrentRequestsInAllSources = Math.min( maxConcurrentRequestsInAllSources, serviceMap[source].maxConcurrentRequests, ) } } - for (const member of members) { - const enrichmentInput = await getEnrichmentInput(member) - const obsoleteSources = await getObsoleteSourcesOfMember(member.id, possibleSources) - - Object.keys(serviceMap).forEach(async (source) => { - if ( - (await serviceMap[source].isEnrichableBySource(enrichmentInput)) && - (obsoleteSources as string[]).includes(source) - ) { - distinctEnrichableSources.add(source as MemberEnrichmentSource) - } - }) - } - - let smallestMaxConcurrentRequests = maxConcurrentRequestsInAllSources - - Array.from(distinctEnrichableSources).forEach(async (source) => { - smallestMaxConcurrentRequests = Math.min( - smallestMaxConcurrentRequests, - serviceMap[source].maxConcurrentRequests, - ) - }) - - svc.log.info('Setting max concurrent requests', { smallestMaxConcurrentRequests }) + svc.log.info('Setting max concurrent requests', { maxConcurrentRequestsInAllSources }) - return smallestMaxConcurrentRequests + return maxConcurrentRequestsInAllSources } diff --git a/services/apps/premium/members_enrichment_worker/src/types.ts b/services/apps/premium/members_enrichment_worker/src/types.ts index 1003739a65..22d896c34c 100644 --- a/services/apps/premium/members_enrichment_worker/src/types.ts +++ b/services/apps/premium/members_enrichment_worker/src/types.ts @@ -57,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 normalize( diff --git a/services/apps/premium/members_enrichment_worker/src/workflows/getMembersToEnrich.ts b/services/apps/premium/members_enrichment_worker/src/workflows/getMembersToEnrich.ts index 2c3515934d..8e4482a250 100644 --- a/services/apps/premium/members_enrichment_worker/src/workflows/getMembersToEnrich.ts +++ b/services/apps/premium/members_enrichment_worker/src/workflows/getMembersToEnrich.ts @@ -33,7 +33,11 @@ export async function getMembersToEnrich(): Promise { return } - const parallelEnrichmentWorkflows = await getMaxConcurrentRequests(members, sources) + const parallelEnrichmentWorkflows = await getMaxConcurrentRequests( + members, + sources, + QUERY_FOR_ENRICHABLE_MEMBERS_PER_RUN, + ) const chunks = chunkArray(members, parallelEnrichmentWorkflows) From d893c8f4baa46e5fe28100782fc358eb3d407560 Mon Sep 17 00:00:00 2001 From: anilb Date: Mon, 11 Nov 2024 14:54:45 +0100 Subject: [PATCH 16/16] linting --- .../src/activities/getMembers.ts | 2 -- .../members_enrichment_worker/src/utils/common.ts | 11 ++--------- 2 files changed, 2 insertions(+), 11 deletions(-) diff --git a/services/apps/premium/members_enrichment_worker/src/activities/getMembers.ts b/services/apps/premium/members_enrichment_worker/src/activities/getMembers.ts index 0f45bcbd42..ebabb35501 100644 --- a/services/apps/premium/members_enrichment_worker/src/activities/getMembers.ts +++ b/services/apps/premium/members_enrichment_worker/src/activities/getMembers.ts @@ -9,8 +9,6 @@ import { EnrichmentSourceServiceFactory } from '../factory' import { svc } from '../main' import { IEnrichmentService } from '../types' -import { getEnrichmentInput, getObsoleteSourcesOfMember } from './enrichment' - export async function getEnrichableMembers( limit: number, sources: MemberEnrichmentSource[], diff --git a/services/apps/premium/members_enrichment_worker/src/utils/common.ts b/services/apps/premium/members_enrichment_worker/src/utils/common.ts index 96f1149bc0..0a62747909 100644 --- a/services/apps/premium/members_enrichment_worker/src/utils/common.ts +++ b/services/apps/premium/members_enrichment_worker/src/utils/common.ts @@ -1,14 +1,7 @@ import lodash from 'lodash' -import { Logger } from '@crowd/logging' -import { - IMemberEnrichmentCache, - MemberEnrichmentSource, - MemberIdentityType, - PlatformType, -} from '@crowd/types' +import { IMemberEnrichmentCache, MemberIdentityType, PlatformType } from '@crowd/types' -import { EnrichmentSourceServiceFactory } from '../factory' import { IMemberEnrichmentAttributeSettings, IMemberEnrichmentData, @@ -114,4 +107,4 @@ export function chunkArray(array: T[], chunkSize: number): T[][] { chunks.push(array.slice(i, i + chunkSize)) } return chunks -} \ No newline at end of file +}