chore: add debugger logs

Author: skwowet

services/apps/organizations_enrichment_worker/src/main.ts

@@ -1,10 +1,13 @@
 import { Config } from '@crowd/archetype-standard'
 import { Options, ServiceWorker } from '@crowd/archetype-worker'
 
-import { scheduleOrganizationsEnrichment } from './schedules'
+// import { scheduleOrganizationsEnrichment } from './schedules'
 
 const config: Config = {
-  envvars: ['CROWD_ORGANIZATION_ENRICHMENT_API_KEY'],
+  envvars: [
+    'CROWD_ORGANIZATION_ENRICHMENT_INTERNAL_API_URL',
+    'CROWD_ORGANIZATION_ENRICHMENT_INTERNAL_API_KEY',
+  ],
   producer: {
     enabled: false,
   },
@@ -30,7 +33,7 @@ export const svc = new ServiceWorker(config, options)
 setImmediate(async () => {
   await svc.init()
 
-  await scheduleOrganizationsEnrichment()
+  // await scheduleOrganizationsEnrichment()
 
   await svc.start()
 })

services/apps/organizations_enrichment_worker/src/types.ts

@@ -44,3 +44,7 @@ export interface IOrganizationEnrichmentDataNormalized {
   attributes?: Record<string, unknown>
   displayName?: string
 }
+
+export interface IGetOrganizationsToEnrichInput {
+  testRun?: boolean
+}

services/apps/organizations_enrichment_worker/src/workflows/enrichOrganization.ts

@@ -36,6 +36,8 @@ export async function enrichOrganization(
   const organization = await findOrganizationById(input.id)
   if (!organization) return
 
+  console.log('Enriching organization:', input.id)
+
   const [cache] = await findOrganizationEnrichmentCache([source], input.id)
 
   // Skip if cache is still fresh
@@ -44,6 +46,8 @@ export async function enrichOrganization(
   // Prepare enrichment input
   const enrichmentInput: IOrganizationEnrichmentSourceInput = await getEnrichmentInput(input)
 
+  console.log('Enrichment input:', enrichmentInput)
+
   // Use LLM to pick the most relevant domain if multiple
   if (enrichmentInput.domains.length > 1) {
     const mostRelevantDomain = await selectMostRelevantDomainWithLLM(
@@ -58,34 +62,41 @@ export async function enrichOrganization(
   // Fetch new enrichment data
   const data = await getEnrichmentData(source, enrichmentInput)
 
+  console.log('Enrichment data:', JSON.stringify(data, null, 2))
+
   // Record enrichment attempt
   await touchOrganizationEnrichmentLastTriedAt(input.id)
 
   let changeInEnrichmentSourceData = false
 
   if (!cache) {
     // First time enriching, create cache entry
+    console.log('Creating cache entry for organization!')
     await createOrganizationEnrichmentCache(source, input.id, data)
     if (data) {
       changeInEnrichmentSourceData = true
     }
   } else if (cache && !data) {
     // No new data, keep the old cache and update the timestamp
+    console.log('No new data, keeping old cache and updating timestamp!')
     await touchOrganizationEnrichmentCacheUpdatedAt(source, input.id)
   } else if (cache && data) {
     // Data changed, update cache
     if (sourceHasDifferentDataComparedToCache(cache, data)) {
+      console.log('Data changed, updating cache!')
       await updateOrganizationEnrichmentCache(source, input.id, data)
       changeInEnrichmentSourceData = true
     } else {
       // Data unchanged, keep the old cache and update the timestamp
+      console.log('Data unchanged, keeping old cache and updating timestamp!')
       await touchOrganizationEnrichmentCacheUpdatedAt(source, input.id)
     }
   }
 
   // Apply enrichment only if there’s new data to apply
   if (changeInEnrichmentSourceData && data) {
     const normalized = await normalizeEnrichmentData(source, data)
+    console.log('Normalized data:', JSON.stringify(normalized, null, 2))
     await applyEnrichmentToOrganization(input.id, normalized)
   }
 }

services/apps/organizations_enrichment_worker/src/workflows/getOrganizationsToEnrich.ts

@@ -9,6 +9,7 @@ import {
 import { IEnrichableOrganization, OrganizationEnrichmentSource } from '@crowd/types'
 
 import * as activities from '../activities'
+import { IGetOrganizationsToEnrichInput } from '../types'
 import { chunkArray } from '../utils/common'
 import { enrichOrganization } from '../workflows'
 
@@ -18,7 +19,9 @@ const { getEnrichableOrganizations, getMaxConcurrentRequests } = proxyActivities
   },
 )
 
-export async function getOrganizationsToEnrich(): Promise<void> {
+export async function getOrganizationsToEnrich(
+  input: IGetOrganizationsToEnrichInput,
+): Promise<void> {
   const QUERY_FOR_ENRICHABLE_ORGANIZATIONS_PER_RUN = 5
   const source = OrganizationEnrichmentSource.INTERNAL_API
 
@@ -59,5 +62,10 @@ export async function getOrganizationsToEnrich(): Promise<void> {
     )
   }
 
-  await continueAsNew<typeof getOrganizationsToEnrich>()
+  if (input.testRun) {
+    console.log('Test run completed - stopping after first batch!')
+    return
+  }
+
+  await continueAsNew<typeof getOrganizationsToEnrich>({ ...input })
 }