Feature/c 867 organization enrichment

backend/.env.dist.local

@@ -139,6 +139,9 @@ CROWD_COHERE_API_KEY=
 CROWD_ENRICHMENT_URL=
 CROWD_ENRICHMENT_API_KEY=
 
+# PDL Organization Enrichment settings
+CROWD_ORGANIZATION_ENRICHMENT_API_KEY=
+
 # EagleEye settings
 CROWD_EAGLE_EYE_URL=
 CROWD_EAGLE_EYE_API_KEY=

backend/config/custom-environment-variables.json

@@ -145,6 +145,9 @@
     "url": "CROWD_ENRICHMENT_URL",
     "apiKey": "CROWD_ENRICHMENT_API_KEY"
   },
+  "organizationEnrichment": {
+    "apiKey": "CROWD_ORGANIZATION_ENRICHMENT_API_KEY"
+  },
   "eagleEye": {
     "url": "CROWD_EAGLE_EYE_URL",
     "apiKey": "CROWD_EAGLE_EYE_API_KEY"

backend/config/default.json

@@ -39,6 +39,7 @@
   "github": {},
   "stackexchange": {},
   "enrichment": {},
+  "organizationEnrichment": {},
   "eagleEye": {},
   "unleash": {
     "db": {}

backend/package.json

@@ -101,6 +101,7 @@
     "passport-google-oauth": "2.0.0",
     "passport-google-oauth20": "^2.0.0",
     "passport-slack": "0.0.7",
+    "peopledatalabs": "^5.0.3",
     "pg": "^8.7.3",
     "pm2": "^5.2.0",
     "redis": "^4.5.0",

backend/src/bin/jobs/index.ts

@@ -11,6 +11,7 @@ import mergeSuggestions from './mergeSuggestions'
 import refreshSampleData from './refreshSampleData'
 import cleanUp from './cleanUp'
 import checkStuckIntegrationRuns from './checkStuckIntegrationRuns'
+import enrichOrganizations from './organizationEnricher'
 
 const jobs: CrowdJob[] = [
   weeklyAnalyticsEmailsCoordinator,
@@ -25,6 +26,7 @@ const jobs: CrowdJob[] = [
   refreshSampleData,
   cleanUp,
   checkStuckIntegrationRuns,
+  enrichOrganizations,
 ]
 
 export default jobs

backend/src/bin/jobs/organizationEnricher.ts

@@ -0,0 +1,32 @@
+import cronGenerator from 'cron-time-generator'
+import SequelizeRepository from '../../database/repositories/sequelizeRepository'
+import { CrowdJob } from '../../types/jobTypes'
+import { sendNodeWorkerMessage } from '../../serverless/utils/nodeWorkerSQS'
+import { NodeWorkerMessageBase } from '../../types/mq/nodeWorkerMessageBase'
+import { NodeWorkerMessageType } from '../../serverless/types/workerTypes'
+import TenantRepository from '../../database/repositories/tenantRepository'
+import { getServiceLogger } from '../../utils/logging'
+
+const job: CrowdJob = {
+  name: 'organization enricher',
+  cronTime: cronGenerator.everyDay(),
+  onTrigger: sendWorkerMessage,
+}
+
+async function sendWorkerMessage() {
+  const options = await SequelizeRepository.getDefaultIRepositoryOptions()
+  const log = getServiceLogger()
+  const tenants = await TenantRepository.getPayingTenantIds(options)
+  log.info(tenants)
+  for (const { id } of tenants) {
+    const payload = {
+      type: NodeWorkerMessageType.NODE_MICROSERVICE,
+      service: 'enrich-organizations',
+      tenantId: id,
+    } as NodeWorkerMessageBase
+    log.info({ payload }, 'enricher worker payload')
+    await sendNodeWorkerMessage(id, payload)
+  }
+}
+
+export default job

backend/src/config/configTypes.ts

@@ -186,6 +186,10 @@ export interface EnrichmentConfiguration {
   apiKey: string
 }
 
+export interface OrganizationEnrichmentConfiguration {
+  apiKey: string
+}
+
 export interface EagleEyeConfiguration {
   url: string
   apiKey: string

backend/src/config/index.ts

@@ -30,6 +30,7 @@ import {
   SampleDataConfiguration,
   IntegrationProcessingConfiguration,
   SlackNotifierConfiguration,
+  OrganizationEnrichmentConfiguration,
 } from './configTypes'
 
 // TODO-kube
@@ -228,6 +229,12 @@ export const ENRICHMENT_CONFIG: EnrichmentConfiguration = KUBE_MODE
       apiKey: process.env.ENRICHMENT_SECRET_KEY,
     }
 
+export const ORGANIZATION_ENRICHMENT_CONFIG: OrganizationEnrichmentConfiguration = KUBE_MODE
+  ? config.get<OrganizationEnrichmentConfiguration>('organizationEnrichment')
+  : {
+      apiKey: process.env.ORGANIZATION_ENRICHMENT_SECRET_KEY,
+    }
+
 export const EAGLE_EYE_CONFIG: EagleEyeConfiguration = KUBE_MODE
   ? config.get<EagleEyeConfiguration>('eagleEye')
   : {

backend/src/database/initializers/sample-data.json

@@ -882,7 +882,7 @@
         "country": {
           "enrichment": "USA"
         },
-        "yearsofExperience": {
+        "yearsOfExperience": {
           "enrichment": 25
         },
         "education": {
@@ -1042,7 +1042,7 @@
         "country": {
           "enrichment": "USA"
         },
-        "yearsofExperience": {
+        "yearsOfExperience": {
           "enrichment": 5
         },
         "education": {
@@ -1187,7 +1187,7 @@
         "country": {
           "enrichment": "USA"
         },
-        "yearsofExperience": {
+        "yearsOfExperience": {
           "enrichment": 10
         },
         "education": {
@@ -1333,7 +1333,7 @@
         "country": {
           "enrichment": "USA"
         },
-        "yearsofExperience": {
+        "yearsOfExperience": {
           "enrichment": 20
         },
         "education": {

backend/src/database/migrations/U1682865647__enrichOrganizationCacheWithMissingFields.sql

@@ -0,0 +1,13 @@
+ALTER TABLE public."organizationCaches" DROP COLUMN "lastEnrichedAt";
+ALTER TABLE public."organizationCaches" DROP COLUMN "employeeCountByCountry";
+ALTER TABLE public."organizationCaches" DROP COLUMN "type";
+ALTER TABLE public."organizationCaches" DROP COLUMN "geoLocation";
+ALTER TABLE public."organizationCaches" DROP COLUMN "size";
+ALTER TABLE public."organizationCaches" DROP COLUMN "ticker";
+ALTER TABLE public."organizationCaches" DROP COLUMN "headline";
+ALTER TABLE public."organizationCaches" DROP COLUMN "profiles";
+ALTER TABLE public."organizationCaches" DROP COLUMN "naics";
+ALTER TABLE public."organizationCaches" DROP COLUMN "address";
+ALTER TABLE public."organizationCaches" DROP COLUMN "industry";
+ALTER TABLE public."organizationCaches" DROP COLUMN "founded";
+ALTER TABLE public."organizationCaches" DROP COLUMN "location";

backend/src/database/migrations/U1682865663__enrichOrganizationWithMissingFields.sql

@@ -0,0 +1,13 @@
+ALTER TABLE public."organizations" DROP COLUMN "lastEnrichedAt";
+ALTER TABLE public."organizations" DROP COLUMN "employeeCountByCountry";
+ALTER TABLE public."organizations" DROP COLUMN "type";
+ALTER TABLE public."organizations" DROP COLUMN "geoLocation";
+ALTER TABLE public."organizations" DROP COLUMN "size";
+ALTER TABLE public."organizations" DROP COLUMN "ticker";
+ALTER TABLE public."organizations" DROP COLUMN "headline";
+ALTER TABLE public."organizations" DROP COLUMN "profiles";
+ALTER TABLE public."organizations" DROP COLUMN "naics";
+ALTER TABLE public."organizations" DROP COLUMN "address";
+ALTER TABLE public."organizations" DROP COLUMN "industry";
+ALTER TABLE public."organizations" DROP COLUMN "founded";
+ALTER TABLE public."organizations" DROP COLUMN "location";

backend/src/database/migrations/V1682865647__enrichOrganizationCacheWithMissingFields.sql

@@ -0,0 +1,12 @@
+ALTER TABLE public."organizationCaches" ADD COLUMN "lastEnrichedAt" TIMESTAMP WITH TIME ZONE NULL;
+ALTER TABLE public."organizationCaches" ADD COLUMN "employeeCountByCountry" JSONB NULL;
+ALTER TABLE public."organizationCaches" ADD COLUMN "type" TEXT NULL;
+ALTER TABLE public."organizationCaches" ADD COLUMN "geoLocation" TEXT NULL;
+ALTER TABLE public."organizationCaches" ADD COLUMN "size" TEXT NULL;
+ALTER TABLE public."organizationCaches" ADD COLUMN "ticker" TEXT NULL;
+ALTER TABLE public."organizationCaches" ADD COLUMN "headline" TEXT NULL;
+ALTER TABLE public."organizationCaches" ADD COLUMN "profiles" TEXT[] NULL;
+ALTER TABLE public."organizationCaches" ADD COLUMN "naics" JSONB[] NULL;
+ALTER TABLE public."organizationCaches" ADD COLUMN "address" JSONB NULL;
+ALTER TABLE public."organizationCaches" ADD COLUMN "industry" TEXT NULL;
+ALTER TABLE public."organizationCaches" ADD COLUMN "founded" INTEGER NULL;

backend/src/database/migrations/V1682865663__enrichOrganizationWithMissingFields.sql

@@ -0,0 +1,12 @@
+ALTER TABLE public."organizations" ADD COLUMN "lastEnrichedAt" TIMESTAMP WITH TIME ZONE NULL;
+ALTER TABLE public."organizations" ADD COLUMN "employeeCountByCountry" JSONB NULL;
+ALTER TABLE public."organizations" ADD COLUMN "type" TEXT NULL;
+ALTER TABLE public."organizations" ADD COLUMN "geoLocation" TEXT NULL;
+ALTER TABLE public."organizations" ADD COLUMN "size" TEXT NULL;
+ALTER TABLE public."organizations" ADD COLUMN "ticker" TEXT NULL;
+ALTER TABLE public."organizations" ADD COLUMN "headline" TEXT NULL;
+ALTER TABLE public."organizations" ADD COLUMN "profiles" TEXT[] NULL;
+ALTER TABLE public."organizations" ADD COLUMN "naics" JSONB[] NULL;
+ALTER TABLE public."organizations" ADD COLUMN "address" JSONB NULL;
+ALTER TABLE public."organizations" ADD COLUMN "industry" TEXT NULL;
+ALTER TABLE public."organizations" ADD COLUMN "founded" INTEGER NULL;