feat: add lfx-internal-api source to organization enrichment (CM-686)

backend/src/database/migrations/R__organizationEnrichmentMaterializedViews.sql

@@ -0,0 +1,22 @@
+-- drop old materialized view if exists
+drop materialized view if exists "organizationsGlobalActivityCount" cascade;
+
+-- create the materialized view
+create materialized view "organizationsGlobalActivityCount" as
+select 
+    osa."organizationId",
+    sum(osa."activityCount") as total_count
+from "organizationSegmentsAgg" osa
+where osa."segmentId" in (
+    select id
+    from segments
+    where "grandparentId" is not null
+      and "parentId" is not null
+)
+group by osa."organizationId"
+order by sum(osa."activityCount") desc;
+
+create unique index ix_organization_global_activity_count_organization_id
+    on "organizationsGlobalActivityCount" ("organizationId");
+
+create index ix_organization_global_activity_count on "organizationsGlobalActivityCount" (total_count);

backend/src/database/migrations/V1760602430__alter_organizationEnrichmentCache_data_nullable.sql

@@ -0,0 +1,2 @@
+alter table "organizationEnrichmentCache"
+    alter column "data" drop not null;

backend/src/database/repositories/member/memberAffiliationsRepository.ts

@@ -1,9 +1,9 @@
+import { OrganizationField, queryOrgs } from '@crowd/data-access-layer'
 import {
   deleteMemberAffiliations,
   fetchMemberAffiliations,
   insertMemberAffiliations,
 } from '@crowd/data-access-layer/src/member_segment_affiliations'
-import { OrganizationField, queryOrgs } from '@crowd/data-access-layer/src/orgs'
 import { fetchManySegments } from '@crowd/data-access-layer/src/segments'
 import { IMemberAffiliation, IOrganization, SegmentData } from '@crowd/types'
 

backend/src/database/repositories/memberRepository.ts

@@ -12,9 +12,11 @@ import {
 } from '@crowd/common'
 import { CommonMemberService } from '@crowd/common_services'
 import {
+  OrganizationField,
   getActiveMembers,
   getLastActivitiesForMembers,
   queryActivityRelations,
+  queryOrgs,
 } from '@crowd/data-access-layer'
 import { findManyLfxMemberships } from '@crowd/data-access-layer/src/lfx_memberships'
 import { findMaintainerRoles } from '@crowd/data-access-layer/src/maintainers'
@@ -41,7 +43,6 @@ import {
   includeMemberToSegments,
 } from '@crowd/data-access-layer/src/members/segments'
 import { IDbMemberData } from '@crowd/data-access-layer/src/members/types'
-import { OrganizationField, queryOrgs } from '@crowd/data-access-layer/src/orgs'
 import { optionsQx } from '@crowd/data-access-layer/src/queryExecutor'
 import {
   fetchManySegments,

backend/src/database/repositories/organizationRepository.ts

@@ -19,6 +19,7 @@ import {
   IDbOrgAttribute,
   IDbOrganization,
   OrgIdentityField,
+  OrganizationField,
   addOrgIdentity,
   addOrgsToSegments,
   cleanUpOrgIdentities,
@@ -1131,7 +1132,24 @@ class OrganizationRepository {
       (a, b) => b.identities.length - a.identities.length,
     )[0].organizationId
 
-    const result = await findOrgById(qx, orgIdWithMostIdentities)
+    const result = await findOrgById(qx, orgIdWithMostIdentities, [
+      OrganizationField.ID,
+      OrganizationField.DISPLAY_NAME,
+      OrganizationField.DESCRIPTION,
+      OrganizationField.LOGO,
+      OrganizationField.TAGS,
+      OrganizationField.EMPLOYEES,
+      OrganizationField.REVENUE_RANGE,
+      OrganizationField.IMPORT_HASH,
+      OrganizationField.LOCATION,
+      OrganizationField.TYPE,
+      OrganizationField.SIZE,
+      OrganizationField.HEADLINE,
+      OrganizationField.INDUSTRY,
+      OrganizationField.FOUNDED,
+      OrganizationField.IS_TEAM_ORGANIZATION,
+      OrganizationField.MANUALLY_CREATED,
+    ])
 
     return result
   }

backend/src/services/collectionService.ts

@@ -1,7 +1,7 @@
 import { uniq } from 'lodash'
 
 import { getCleanString } from '@crowd/common'
-import { QueryExecutor } from '@crowd/data-access-layer'
+import { OrganizationField, QueryExecutor, findOrgById, queryOrgs } from '@crowd/data-access-layer'
 import { listCategoriesByIds } from '@crowd/data-access-layer/src/categories'
 import {
   CollectionField,
@@ -29,7 +29,6 @@ import {
   fetchIntegrationsForSegment,
   removePlainGitHubRepoMapping,
 } from '@crowd/data-access-layer/src/integrations'
-import { OrganizationField, findOrgById, queryOrgs } from '@crowd/data-access-layer/src/orgs'
 import { QueryFilter } from '@crowd/data-access-layer/src/query'
 import {
   ICreateRepositoryGroup,

backend/src/services/dataIssueService.ts

@@ -1,9 +1,9 @@
 import axios from 'axios'
 
 import { createHeading, createParagraph } from '@crowd/common'
+import { OrganizationField, findOrgById } from '@crowd/data-access-layer'
 import { createDataIssue } from '@crowd/data-access-layer/src/data_issues'
 import { MemberField, findMemberById } from '@crowd/data-access-layer/src/members'
-import { OrganizationField, findOrgById } from '@crowd/data-access-layer/src/orgs'
 import { PgPromiseQueryExecutor } from '@crowd/data-access-layer/src/queryExecutor'
 import { LoggerBase } from '@crowd/logging'
 import { DataIssueEntity } from '@crowd/types'

services/apps/members_enrichment_worker/src/activities.ts

@@ -13,10 +13,10 @@ import {
   isCacheObsolete,
   normalizeEnrichmentData,
   refreshMemberEnrichmentMaterializedView,
-  setMemberEnrichmentTryDate,
   squashMultipleValueAttributesWithLLM,
   squashWorkExperiencesWithLLM,
   touchMemberEnrichmentCacheUpdatedAt,
+  touchMemberEnrichmentLastTriedAt,
   updateMemberEnrichmentCache,
   updateMemberUsingSquashedPayload,
 } from './activities/enrichment'
@@ -45,7 +45,7 @@ import {
 } from './activities/syncEnrichedData'
 
 export {
-  setMemberEnrichmentTryDate,
+  touchMemberEnrichmentLastTriedAt,
   getMemberById,
   getEnrichableMembers,
   getEnrichmentData,

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

@@ -22,8 +22,8 @@ import {
   findMemberEnrichmentCacheForAllSourcesDb,
   insertMemberEnrichmentCacheDb,
   insertWorkExperience,
-  setMemberEnrichmentTryDate as setMemberEnrichmentTryDateDb,
-  setMemberEnrichmentUpdateDate as setMemberEnrichmentUpdateDateDb,
+  setMemberEnrichmentLastTriedAt,
+  setMemberEnrichmentUpdatedAt,
   touchMemberEnrichmentCacheUpdatedAtDb,
   updateMemberEnrichmentCacheDb,
   updateMemberOrg,
@@ -63,7 +63,7 @@ async function setRateLimitBackoff(
   source: MemberEnrichmentSource,
   backoffSeconds: number,
 ): Promise<void> {
-  const redisCache = new RedisCache(`enrichment-${source}`, svc.redis, svc.log)
+  const redisCache = new RedisCache(`member-enrichment-${source}`, svc.redis, svc.log)
   const backoff = new RateLimitBackoff(redisCache, 'rate-limit-backoff')
   await backoff.set(backoffSeconds)
 }
@@ -147,7 +147,7 @@ export async function setHasRemainingCredits(
   source: MemberEnrichmentSource,
   hasCredits: boolean,
 ): Promise<void> {
-  const redisCache = new RedisCache(`enrichment-${source}`, svc.redis, svc.log)
+  const redisCache = new RedisCache(`member-enrichment-${source}`, svc.redis, svc.log)
   if (hasCredits) {
     await redisCache.set('hasRemainingCredits', 'true', 60)
   } else {
@@ -156,12 +156,12 @@ export async function setHasRemainingCredits(
 }
 
 export async function getHasRemainingCredits(source: MemberEnrichmentSource): Promise<boolean> {
-  const redisCache = new RedisCache(`enrichment-${source}`, svc.redis, svc.log)
+  const redisCache = new RedisCache(`member-enrichment-${source}`, svc.redis, svc.log)
   return (await redisCache.get('hasRemainingCredits')) === 'true'
 }
 
 export async function hasRemainingCreditsExists(source: MemberEnrichmentSource): Promise<boolean> {
-  const redisCache = new RedisCache(`enrichment-${source}`, svc.redis, svc.log)
+  const redisCache = new RedisCache(`member-enrichment-${source}`, svc.redis, svc.log)
   return await redisCache.exists('hasRemainingCredits')
 }
 
@@ -221,6 +221,10 @@ export async function updateMemberEnrichmentCache(
   await updateMemberEnrichmentCacheDb(svc.postgres.writer.connection(), data, memberId, source)
 }
 
+export async function touchMemberEnrichmentLastTriedAt(memberId: string): Promise<void> {
+  await setMemberEnrichmentLastTriedAt(svc.postgres.writer.connection(), memberId)
+}
+
 export async function touchMemberEnrichmentCacheUpdatedAt(
   source: MemberEnrichmentSource,
   memberId: string,
@@ -453,10 +457,8 @@ export async function updateMemberUsingSquashedPayload(
     await Promise.all(promises)
 
     if (updated) {
-      await setMemberEnrichmentUpdateDateDb(tx.transaction(), memberId)
+      await setMemberEnrichmentUpdatedAt(tx.transaction(), memberId)
       await syncMember(memberId)
-    } else {
-      await setMemberEnrichmentTryDateDb(tx.transaction(), memberId)
     }
 
     svc.log.debug({ memberId }, 'Member sources processed successfully!')
@@ -508,10 +510,6 @@ export function doesIncomingOrgExistInExistingOrgs(
   )
 }
 
-export async function setMemberEnrichmentTryDate(memberId: string): Promise<void> {
-  await setMemberEnrichmentTryDateDb(svc.postgres.writer.connection(), memberId)
-}
-
 export async function getObsoleteSourcesOfMember(
   memberId: string,
   possibleSources: MemberEnrichmentSource[],

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

@@ -3,7 +3,7 @@ import { fetchMembersForEnrichment } from '@crowd/data-access-layer/src/old/apps
 import { RateLimitBackoff, RedisCache } from '@crowd/redis'
 import {
   IEnrichableMember,
-  IMemberEnrichmentSourceQueryInput,
+  IEnrichmentSourceQueryInput,
   MemberEnrichmentSource,
 } from '@crowd/types'
 
@@ -30,14 +30,16 @@ export async function getEnrichableMembers(
   ).filter((s): s is MemberEnrichmentSource => s !== null)
 
   let rows: IEnrichableMember[] = []
-  const sourceInputs: IMemberEnrichmentSourceQueryInput[] = availableSources.map((s) => {
-    const srv = EnrichmentSourceServiceFactory.getEnrichmentSourceService(s, svc.log)
-    return {
-      source: s,
-      cacheObsoleteAfterSeconds: srv.cacheObsoleteAfterSeconds,
-      enrichableBySql: srv.enrichableBySql,
-    }
-  })
+  const sourceInputs: IEnrichmentSourceQueryInput<MemberEnrichmentSource>[] = availableSources.map(
+    (s) => {
+      const srv = EnrichmentSourceServiceFactory.getEnrichmentSourceService(s, svc.log)
+      return {
+        source: s,
+        cacheObsoleteAfterSeconds: srv.cacheObsoleteAfterSeconds,
+        enrichableBySql: srv.enrichableBySql,
+      }
+    },
+  )
 
   const db = svc.postgres.reader
   rows = await fetchMembersForEnrichment(db, limit, sourceInputs)

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

@@ -18,6 +18,7 @@ const {
   findMemberEnrichmentCache,
   insertMemberEnrichmentCache,
   touchMemberEnrichmentCacheUpdatedAt,
+  touchMemberEnrichmentLastTriedAt,
   updateMemberEnrichmentCache,
   isCacheObsolete,
   getEnrichmentInput,
@@ -61,6 +62,9 @@ export async function enrichMember(
 
       const data = await getEnrichmentData(source, enrichmentInput)
 
+      // Record enrichment attempt
+      await touchMemberEnrichmentLastTriedAt(input.id)
+
       if (!cache) {
         await insertMemberEnrichmentCache(source, input.id, data)
         if (data) {

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

@@ -13,7 +13,6 @@ const {
   squashMultipleValueAttributesWithLLM,
   squashWorkExperiencesWithLLM,
   updateMemberUsingSquashedPayload,
-  setMemberEnrichmentTryDate,
   cleanAttributeValue,
 } = proxyActivities<typeof activities>({
   startToCloseTimeout: '2 minutes',
@@ -310,7 +309,5 @@ export async function processMemberSources(args: IProcessMemberSourcesArgs): Pro
     return memberUpdated
   }
 
-  await setMemberEnrichmentTryDate(args.memberId)
-
   return false
 }

services/apps/organizations_enrichment_worker/package.json

@@ -24,7 +24,10 @@
     "@temporalio/workflow": "~1.11.8",
     "peopledatalabs": "~6.1.5",
     "tsx": "^4.7.1",
-    "typescript": "^5.6.3"
+    "typescript": "^5.6.3",
+    "lodash.isequal": "^4.5.0",
+    "lodash.uniqby": "^4.7.0",
+    "axios": "~1.6.2"
   },
   "devDependencies": {
     "@types/node": "^20.8.2",

services/apps/organizations_enrichment_worker/src/activities.ts

@@ -1,13 +1,34 @@
 import {
-  getMaxEnrichedOrganizationsPerExecution,
-  getOrganizationsToEnrich,
-  syncToOpensearch,
-  tryEnrichOrganization,
+  applyEnrichmentToOrganization,
+  createOrganizationEnrichmentCache,
+  findOrganizationEnrichmentCache,
+  getEnrichmentData,
+  getEnrichmentInput,
+  getMaxConcurrentRequests,
+  isCacheObsolete,
+  normalizeEnrichmentData,
+  refreshOrganizationEnrichmentMaterializedView,
+  touchOrganizationEnrichmentCacheUpdatedAt,
+  touchOrganizationEnrichmentLastTriedAt,
+  updateOrganizationEnrichmentCache,
 } from './activities/enrichment'
+import { selectMostRelevantDomainWithLLM } from './activities/llm'
+import { findOrganizationById, getEnrichableOrganizations } from './activities/organization'
 
 export {
-  syncToOpensearch,
-  getOrganizationsToEnrich,
-  tryEnrichOrganization,
-  getMaxEnrichedOrganizationsPerExecution,
+  getMaxConcurrentRequests,
+  findOrganizationById,
+  getEnrichableOrganizations,
+  findOrganizationEnrichmentCache,
+  isCacheObsolete,
+  refreshOrganizationEnrichmentMaterializedView,
+  getEnrichmentInput,
+  getEnrichmentData,
+  createOrganizationEnrichmentCache,
+  updateOrganizationEnrichmentCache,
+  touchOrganizationEnrichmentCacheUpdatedAt,
+  touchOrganizationEnrichmentLastTriedAt,
+  normalizeEnrichmentData,
+  applyEnrichmentToOrganization,
+  selectMostRelevantDomainWithLLM,
 }