Enhancement/organizations in opensearch c 1978

backend/src/database/migrations/U1691485202__organization-sync-timestamp.sql

@@ -0,0 +1,2 @@
+alter table organizations
+    drop column "searchSyncedAt";

backend/src/database/migrations/V1691485202__organization-sync-timestamp.sql

@@ -0,0 +1,7 @@
+-- Add the column with NOT NULL and a default value
+alter table organizations
+    add column "searchSyncedAt" timestamptz not null default now();
+
+-- Then, make the column nullable
+alter table organizations
+    alter column "searchSyncedAt" drop not null;

backend/src/database/repositories/organizationRepository.ts

@@ -1,5 +1,7 @@
 import lodash from 'lodash'
-import { SyncStatus } from '@crowd/types'
+import { FieldTranslatorFactory, OpensearchQueryParser } from '@crowd/opensearch'
+import { PageData } from '@crowd/common'
+import { OpenSearchIndex, SyncStatus } from '@crowd/types'
 import Sequelize, { QueryTypes } from 'sequelize'
 import SequelizeRepository from './sequelizeRepository'
 import AuditLogRepository from './auditLogRepository'
@@ -9,6 +11,8 @@ import { IRepositoryOptions } from './IRepositoryOptions'
 import QueryParser from './filters/queryParser'
 import { QueryOutput } from './filters/queryTypes'
 import OrganizationSyncRemoteRepository from './organizationSyncRemoteRepository'
+import isFeatureEnabled from '@/feature-flags/isFeatureEnabled'
+import { FeatureFlag } from '@/types/common'
 
 const { Op } = Sequelize
 
@@ -460,6 +464,9 @@ class OrganizationRepository {
       }
     }
 
+    // compatibility issue
+    delete result.searchSyncedAt
+
     return result
   }
 
@@ -569,6 +576,87 @@ class OrganizationRepository {
     })
   }
 
+  static async findAndCountAllOpensearch(
+    {
+      filter = {} as any,
+      limit = 20,
+      offset = 0,
+      orderBy = 'joinedAt_DESC',
+      countOnly = false,
+      segments = [] as string[],
+      customSortFunction = undefined,
+    },
+    options: IRepositoryOptions,
+  ): Promise<PageData<any>> {
+    const tenant = SequelizeRepository.getCurrentTenant(options)
+
+    const segmentsEnabled = await isFeatureEnabled(FeatureFlag.SEGMENTS, options)
+
+    const segment = segments[0]
+
+    const translator = FieldTranslatorFactory.getTranslator(OpenSearchIndex.ORGANIZATIONS)
+
+    const parsed = OpensearchQueryParser.parse(
+      { filter, limit, offset, orderBy },
+      OpenSearchIndex.ORGANIZATIONS,
+      translator,
+    )
+
+    // add tenant filter to parsed query
+    parsed.query.bool.must.push({
+      term: {
+        uuid_tenantId: tenant.id,
+      },
+    })
+
+    if (segmentsEnabled) {
+      // add segment filter
+      parsed.query.bool.must.push({
+        term: {
+          uuid_segmentId: segment,
+        },
+      })
+    }
+
+    // exclude empty filters if any
+    parsed.query.bool.must = parsed.query.bool.must.filter((obj) => {
+      // Check if the object has a non-empty 'term' property
+      if (obj.term) {
+        return Object.keys(obj.term).length !== 0
+      }
+      return true
+    })
+
+    if (customSortFunction) {
+      parsed.sort = customSortFunction
+    }
+
+    const countResponse = await options.opensearch.count({
+      index: OpenSearchIndex.ORGANIZATIONS,
+      body: { query: parsed.query },
+    })
+
+    if (countOnly) {
+      return {
+        rows: [],
+        count: countResponse.body.count,
+        limit,
+        offset,
+      }
+    }
+
+    const response = await options.opensearch.search({
+      index: OpenSearchIndex.ORGANIZATIONS,
+      body: parsed,
+    })
+
+    const translatedRows = response.body.hits.hits.map((o) =>
+      translator.translateObjectToCrowd(o._source),
+    )
+
+    return { rows: translatedRows, count: countResponse.body.count, limit, offset }
+  }
+
   static async findAndCountAll(
     {
       filter = {} as any,

backend/src/services/organizationService.ts

@@ -9,6 +9,7 @@ import Plans from '../security/plans'
 import telemetryTrack from '../segment/telemetryTrack'
 import { IServiceOptions } from './IServiceOptions'
 import { enrichOrganization } from './helpers/enrichment'
+import { getSearchSyncWorkerEmitter } from '@/serverless/utils/serviceSQS'
 
 export default class OrganizationService extends LoggerBase {
   options: IServiceOptions
@@ -120,6 +121,9 @@ export default class OrganizationService extends LoggerBase {
 
       await SequelizeRepository.commitTransaction(transaction)
 
+      const searchSyncEmitter = await getSearchSyncWorkerEmitter()
+      await searchSyncEmitter.triggerOrganizationSync(this.options.currentTenant.id, record.id)
+
       return record
     } catch (error) {
       await SequelizeRepository.rollbackTransaction(transaction)
@@ -151,6 +155,9 @@ export default class OrganizationService extends LoggerBase {
         await SequelizeRepository.commitTransaction(transaction)
       }
 
+      const searchSyncEmitter = await getSearchSyncWorkerEmitter()
+      await searchSyncEmitter.triggerOrganizationSync(this.options.currentTenant.id, record.id)
+
       return record
     } catch (error) {
       await SequelizeRepository.rollbackTransaction(transaction)
@@ -177,6 +184,12 @@ export default class OrganizationService extends LoggerBase {
       }
 
       await SequelizeRepository.commitTransaction(transaction)
+
+      const searchSyncEmitter = await getSearchSyncWorkerEmitter()
+
+      for (const id of ids) {
+        await searchSyncEmitter.triggerRemoveOrganization(this.options.currentTenant.id, id)
+      }
     } catch (error) {
       await SequelizeRepository.rollbackTransaction(transaction)
       throw error
@@ -204,8 +217,8 @@ export default class OrganizationService extends LoggerBase {
     const orderBy = data.orderBy
     const limit = data.limit
     const offset = data.offset
-    return OrganizationRepository.findAndCountAll(
-      { advancedFilter, orderBy, limit, offset },
+    return OrganizationRepository.findAndCountAllOpensearch(
+      { filter: advancedFilter, orderBy, limit, offset, segments: data.segments },
       this.options,
     )
   }
@@ -224,6 +237,11 @@ export default class OrganizationService extends LoggerBase {
       )
 
       await SequelizeRepository.commitTransaction(transaction)
+
+      const searchSyncEmitter = await getSearchSyncWorkerEmitter()
+      for (const id of ids) {
+        await searchSyncEmitter.triggerRemoveOrganization(this.options.currentTenant.id, id)
+      }
     } catch (error) {
       await SequelizeRepository.rollbackTransaction(transaction)
       throw error

frontend/src/modules/organization/config/filters/main.ts

@@ -34,7 +34,7 @@ export const organizationSearchFilter: SearchFilterConfig = {
     return [
       {
         or: [
-          { name: { textContains: value } },
+          { displayName: { textContains: value } },
         ],
       },
     ];

services/apps/data_sink_worker/src/service/member.service.ts

@@ -93,6 +93,9 @@ export default class MemberService extends LoggerBase {
 
       if (organizationIds.length > 0) {
         await this.nodejsWorkerEmitter.enrichMemberOrganizations(tenantId, id, organizationIds)
+        for (const orgId of organizationIds) {
+          await this.searchSyncWorkerEmitter.triggerOrganizationSync(tenantId, orgId)
+        }
       }
 
       return id
@@ -112,7 +115,7 @@ export default class MemberService extends LoggerBase {
     fireSync = true,
   ): Promise<void> {
     try {
-      const updated = await this.store.transactionally(async (txStore) => {
+      const { updated, organizationIds } = await this.store.transactionally(async (txStore) => {
         const txRepo = new MemberRepository(txStore, this.log)
         const txMemberAttributeService = new MemberAttributeService(txStore, this.log)
         const dbIdentities = await txRepo.getIdentities(id, tenantId)
@@ -176,12 +179,19 @@ export default class MemberService extends LoggerBase {
           }
         }
 
-        return updated
+        return { updated, organizationIds }
       })
 
       if (updated && fireSync) {
         await this.searchSyncWorkerEmitter.triggerMemberSync(tenantId, id)
       }
+
+      if (organizationIds.length > 0) {
+        await this.nodejsWorkerEmitter.enrichMemberOrganizations(tenantId, id, organizationIds)
+        for (const orgId of organizationIds) {
+          await this.searchSyncWorkerEmitter.triggerOrganizationSync(tenantId, orgId)
+        }
+      }
     } catch (err) {
       this.log.error(err, { memberId: id }, 'Error while updating a member!')
       throw err

services/apps/search_sync_worker/package.json

@@ -17,14 +17,19 @@
     "script:update-mappings": "SERVICE=script TS_NODE_TRANSPILE_ONLY=true node -r tsconfig-paths/register -r ts-node/register src/bin/update-mappings.ts",
     "script:sync-all-members": "SERVICE=script TS_NODE_TRANSPILE_ONLY=true node -r tsconfig-paths/register -r ts-node/register src/bin/sync-all-members.ts",
     "script:sync-all-activities": "SERVICE=script TS_NODE_TRANSPILE_ONLY=true node -r tsconfig-paths/register -r ts-node/register src/bin/sync-all-activities.ts",
+    "script:sync-all-organizations": "SERVICE=script TS_NODE_TRANSPILE_ONLY=true node -r tsconfig-paths/register -r ts-node/register src/bin/sync-all-organizations.ts",
     "script:sync-member": "SERVICE=script TS_NODE_TRANSPILE_ONLY=true node -r tsconfig-paths/register -r ts-node/register src/bin/sync-member.ts",
     "script:sync-activity": "SERVICE=script TS_NODE_TRANSPILE_ONLY=true node -r tsconfig-paths/register -r ts-node/register src/bin/sync-activity.ts",
+    "script:sync-organization": "SERVICE=script TS_NODE_TRANSPILE_ONLY=true node -r tsconfig-paths/register -r ts-node/register src/bin/sync-organization.ts",
     "script:sync-tenant-members": "SERVICE=script TS_NODE_TRANSPILE_ONLY=true node -r tsconfig-paths/register -r ts-node/register src/bin/sync-tenant-members.ts",
     "script:sync-tenant-activities": "SERVICE=script TS_NODE_TRANSPILE_ONLY=true node -r tsconfig-paths/register -r ts-node/register src/bin/sync-tenant-activities.ts",
+    "script:sync-tenant-organizations": "SERVICE=script TS_NODE_TRANSPILE_ONLY=true node -r tsconfig-paths/register -r ts-node/register src/bin/sync-tenant-organizations.ts",
     "script:cleanup-tenant-members": "SERVICE=script TS_NODE_TRANSPILE_ONLY=true node -r tsconfig-paths/register -r ts-node/register src/bin/cleanup-tenant-members.ts",
     "script:cleanup-tenant-activities": "SERVICE=script TS_NODE_TRANSPILE_ONLY=true node -r tsconfig-paths/register -r ts-node/register src/bin/cleanup-tenant-activities.ts",
+    "script:cleanup-tenant-organizations": "SERVICE=script TS_NODE_TRANSPILE_ONLY=true node -r tsconfig-paths/register -r ts-node/register src/bin/cleanup-tenant-organizations.ts",
     "script:cleanup-all-members": "SERVICE=script TS_NODE_TRANSPILE_ONLY=true node -r tsconfig-paths/register -r ts-node/register src/bin/cleanup-all-members.ts",
     "script:cleanup-all-activities": "SERVICE=script TS_NODE_TRANSPILE_ONLY=true node -r tsconfig-paths/register -r ts-node/register src/bin/cleanup-all-activities.ts",
+    "script:cleanup-all-organizations": "SERVICE=script TS_NODE_TRANSPILE_ONLY=true node -r tsconfig-paths/register -r ts-node/register src/bin/cleanup-all-activities.ts",
     "script:delete-index": "SERVICE=script TS_NODE_TRANSPILE_ONLY=true node -r tsconfig-paths/register -r ts-node/register src/bin/delete-index.ts"
   },
   "dependencies": {

services/apps/search_sync_worker/src/bin/cleanup-all-organizations.ts

@@ -0,0 +1,36 @@
+import { DB_CONFIG } from '@/conf'
+import { InitService } from '@/service/init.service'
+import { OpenSearchService } from '@/service/opensearch.service'
+import { OrganizationSyncService } from '@/service/organization.sync.service'
+import { DbStore, getDbConnection } from '@crowd/database'
+import { getServiceLogger } from '@crowd/logging'
+
+const log = getServiceLogger()
+
+setImmediate(async () => {
+  const openSearchService = new OpenSearchService(log)
+  await openSearchService.initialize()
+
+  const dbConnection = getDbConnection(DB_CONFIG())
+  const store = new DbStore(log, dbConnection)
+
+  const service = new OrganizationSyncService(store, openSearchService, log)
+
+  const pageSize = 100
+  let results = await service.getAllIndexedTenantIds(pageSize)
+
+  while (results.data.length > 0) {
+    for (const id of results.data) {
+      if (id !== InitService.FAKE_TENANT_ID) {
+        await service.cleanupOrganizationIndex(id)
+      }
+    }
+    if (results.afterKey) {
+      results = await service.getAllIndexedTenantIds(pageSize, results.afterKey)
+    } else {
+      results = { data: [] }
+    }
+  }
+
+  process.exit(0)
+})

services/apps/search_sync_worker/src/bin/cleanup-tenant-organizations.ts

@@ -0,0 +1,23 @@
+import { SQS_CONFIG } from '@/conf'
+import { getServiceLogger } from '@crowd/logging'
+import { SearchSyncWorkerEmitter, getSqsClient } from '@crowd/sqs'
+
+const log = getServiceLogger()
+
+const processArguments = process.argv.slice(2)
+
+if (processArguments.length !== 1) {
+  log.error('Expected 1 argument: tenantId')
+  process.exit(1)
+}
+
+const tenantId = processArguments[0]
+
+setImmediate(async () => {
+  const sqsClient = getSqsClient(SQS_CONFIG())
+  const emitter = new SearchSyncWorkerEmitter(sqsClient, log)
+  await emitter.init()
+
+  await emitter.triggerOrganizationCleanup(tenantId)
+  process.exit(0)
+})

services/apps/search_sync_worker/src/bin/sync-all-organizations.ts

@@ -0,0 +1,27 @@
+import { DB_CONFIG } from '@/conf'
+import { OpenSearchService } from '@/service/opensearch.service'
+import { DbStore, getDbConnection } from '@crowd/database'
+import { getServiceLogger } from '@crowd/logging'
+import { OrganizationRepository } from '@/repo/organization.repo'
+import { OrganizationSyncService } from '@/service/organization.sync.service'
+
+const log = getServiceLogger()
+
+setImmediate(async () => {
+  const openSearchService = new OpenSearchService(log)
+  await openSearchService.initialize()
+
+  const dbConnection = getDbConnection(DB_CONFIG())
+  const store = new DbStore(log, dbConnection)
+
+  const repo = new OrganizationRepository(store, log)
+
+  const tenantIds = await repo.getTenantIds()
+
+  const service = new OrganizationSyncService(store, openSearchService, log)
+
+  for (const tenantId of tenantIds) {
+    await service.syncTenantOrganizations(tenantId, 500)
+  }
+  process.exit(0)
+})

services/apps/search_sync_worker/src/bin/sync-tenant-organizations.ts

@@ -0,0 +1,23 @@
+import { SQS_CONFIG } from '@/conf'
+import { getServiceLogger } from '@crowd/logging'
+import { SearchSyncWorkerEmitter, getSqsClient } from '@crowd/sqs'
+
+const log = getServiceLogger()
+
+const processArguments = process.argv.slice(2)
+
+if (processArguments.length !== 1) {
+  log.error('Expected 1 argument: tenantId')
+  process.exit(1)
+}
+
+const tenantId = processArguments[0]
+
+setImmediate(async () => {
+  const sqsClient = getSqsClient(SQS_CONFIG())
+  const emitter = new SearchSyncWorkerEmitter(sqsClient, log)
+  await emitter.init()
+
+  await emitter.triggerTenantOrganizationSync(tenantId)
+  process.exit(0)
+})