sync to opensearch

services/apps/premium/organizations_enrichment_worker/package.json

@@ -22,6 +22,7 @@
     "@crowd/redis": "file:../../../libs/redis",
     "@crowd/feature-flags": "file:../../../libs/feature-flags",
     "@crowd/common_services": "file:../../../libs/common_services",
+    "@crowd/opensearch": "file:../../../libs/opensearch",
     "@crowd/types": "file:../../../libs/types",
     "@temporalio/client": "~1.8.6",
     "@temporalio/workflow": "~1.8.6",

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

@@ -5,6 +5,7 @@ import {
   tryEnrichOrganization,
   getApplicableTenants,
   hasTenantOrganizationEnrichmentEnabled,
+  syncToOpensearch,
 } from './activities/organizationEnrichment'
 
 export {
@@ -14,4 +15,5 @@ export {
   getTenantOrganizationsForEnrichment,
   tryEnrichOrganization,
   hasTenantOrganizationEnrichmentEnabled,
+  syncToOpensearch,
 }

services/apps/premium/organizations_enrichment_worker/src/activities/organizationEnrichment.ts

@@ -18,9 +18,34 @@ import {
 } from '../repos/organization.repo'
 import { IPremiumTenantInfo, TenantRepository } from '../repos/tenant.repo'
 import { isFeatureEnabled } from '@crowd/feature-flags'
+import { OrganizationSyncService } from '@crowd/opensearch'
 
 /* eslint-disable @typescript-eslint/no-explicit-any */
 
+const syncOrganizations = new OrganizationSyncService(
+  svc.postgres.writer,
+  svc.opensearch,
+  svc.log,
+  {
+    edition: process.env['CROWD_EDITION'],
+  },
+)
+
+export async function syncToOpensearch(organizationId: string): Promise<void> {
+  const log = getChildLogger(syncToOpensearch.name, svc.log, {
+    organizationId,
+  })
+
+  try {
+    await syncOrganizations.syncOrganizations([organizationId])
+  } catch (err) {
+    log.error(err, 'Error while syncing organization to OpenSearch!')
+    throw new Error(err)
+  }
+
+  return null
+}
+
 export async function getApplicableTenants(
   page: number,
   lastId?: string,

services/apps/premium/organizations_enrichment_worker/src/main.ts

@@ -20,7 +20,7 @@ const options: Options = {
     enabled: true,
   },
   opensearch: {
-    enabled: false,
+    enabled: true,
   },
 }
 

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

@@ -2,7 +2,9 @@ import { proxyActivities } from '@temporalio/workflow'
 import * as activities from '../activities/organizationEnrichment'
 import { TenantPlans } from '@crowd/types'
 
-const { tryEnrichOrganization, incrementTenantCredits } = proxyActivities<typeof activities>({
+const { tryEnrichOrganization, incrementTenantCredits, syncToOpensearch } = proxyActivities<
+  typeof activities
+>({
   startToCloseTimeout: '75 seconds',
 })
 
@@ -16,6 +18,7 @@ export async function enrichOrganization(input: IEnrichOrganizationInput): Promi
   const wasEnriched = await tryEnrichOrganization(input.tenantId, input.organizationId)
 
   if (wasEnriched) {
+    await syncToOpensearch(input.organizationId)
     await incrementTenantCredits(input.tenantId, input.plan)
   }
 }