Querying members with opensearch

backend/config/custom-environment-variables.json

@@ -51,7 +51,8 @@
     "nodejsWorkerPassword": "CROWD_DB_NODEJS_WORKER_PASSWORD",
     "jobGeneratorUsername": "CROWD_DB_JOB_GENERATOR_USERNAME",
     "jobGeneratorPassword": "CROWD_DB_JOB_GENERATOR_PASSWORD",
-    "database": "CROWD_DB_DATABASE"
+    "database": "CROWD_DB_DATABASE",
+    "logging": "CROWD_DB_LOGGING"
   },
   "cubejs": {
     "url": "CROWD_CUBEJS_URL",
@@ -168,5 +169,11 @@
       "password": "CROWD_UNLEASH_DB_PASSWORD",
       "database": "CROWD_UNLEASH_DB_DATABASE"
     }
+  },
+  "opensearch": {
+    "node": "CROWD_OPENSEARCH_NODE",
+    "region": "CROWD_OPENSEARCH_AWS_REGION",
+    "accessKeyId": "CROWD_OPENSEARCH_AWS_ACCESS_KEY_ID",
+    "secretAccessKey": "CROWD_OPENSEARCH_AWS_SECRET_ACCESS_KEY"
   }
 }

backend/package.json

@@ -46,18 +46,20 @@
     "@aws-sdk/s3-request-presigner": "^3.229.0",
     "@aws-sdk/url-parser": "^3.226.0",
     "@aws-sdk/util-format-url": "^3.226.0",
+    "@crowd/alerting": "file:../services/libs/alerting",
     "@crowd/common": "file:../services/libs/common",
+    "@crowd/integrations": "file:../services/libs/integrations",
     "@crowd/logging": "file:../services/libs/logging",
     "@crowd/redis": "file:../services/libs/redis",
-    "@crowd/integrations": "file:../services/libs/integrations",
-    "@crowd/types": "file:../services/libs/types",
     "@crowd/sqs": "file:../services/libs/sqs",
-    "@crowd/alerting": "file:../services/libs/alerting",
+    "@crowd/opensearch": "file:../services/libs/opensearch",
+    "@crowd/types": "file:../services/libs/types",
     "@cubejs-client/core": "^0.30.4",
     "@google-cloud/storage": "5.3.0",
     "@octokit/auth-app": "^3.6.1",
     "@octokit/graphql": "^4.8.0",
     "@octokit/request": "^5.6.3",
+    "@opensearch-project/opensearch": "^1.2.0",
     "@pm2/io": "^5.0.0",
     "@sendgrid/eventwebhook": "^7.7.0",
     "@sendgrid/mail": "7.2.6",

backend/src/api/index.ts

@@ -8,7 +8,8 @@ import { Unleash } from 'unleash-client'
 import { getRedisClient, getRedisPubSubPair, RedisPubSubReceiver } from '@crowd/redis'
 import { getServiceLogger } from '@crowd/logging'
 import { ApiWebsocketMessage } from '@crowd/types'
-import { API_CONFIG, REDIS_CONFIG, UNLEASH_CONFIG } from '../conf'
+import { getOpensearchClient } from '@crowd/opensearch'
+import { API_CONFIG, REDIS_CONFIG, UNLEASH_CONFIG, OPENSEARCH_CONFIG } from '../conf'
 import { authMiddleware } from '../middlewares/authMiddleware'
 import { tenantMiddleware } from '../middlewares/tenantMiddleware'
 import { databaseMiddleware } from '../middlewares/databaseMiddleware'
@@ -22,6 +23,7 @@ import { passportStrategyMiddleware } from '../middlewares/passportStrategyMiddl
 import { redisMiddleware } from '../middlewares/redisMiddleware'
 import WebSockets from './websockets'
 import { Edition } from '../types/common'
+import { opensearchMiddleware } from '../middlewares/opensearchMiddleware'
 
 const serviceLogger = getServiceLogger()
 
@@ -32,6 +34,8 @@ const server = http.createServer(app)
 setImmediate(async () => {
   const redis = await getRedisClient(REDIS_CONFIG, true)
 
+  const opensearch = getOpensearchClient(OPENSEARCH_CONFIG)
+
   const redisPubSubPair = await getRedisPubSubPair(REDIS_CONFIG)
   const userNamespace = await WebSockets.initialize(server)
 
@@ -77,6 +81,9 @@ setImmediate(async () => {
   // Bind redis to request
   app.use(redisMiddleware(redis))
 
+  // bind opensearch
+  app.use(opensearchMiddleware(opensearch))
+
   // Bind unleash to request
   if (UNLEASH_CONFIG.url && API_CONFIG.edition === Edition.CROWD_HOSTED) {
     const unleash = new Unleash({

backend/src/conf/configTypes.ts

@@ -215,3 +215,10 @@ export interface SampleDataConfiguration {
 export interface IntegrationProcessingConfiguration {
   maxRetries: number
 }
+
+export interface IOpenSearchConfig {
+  node: string
+  region?: string
+  accessKeyId?: string
+  secretAccessKey?: string
+}

backend/src/conf/index.ts

@@ -30,6 +30,7 @@ import {
   IntegrationProcessingConfiguration,
   SlackNotifierConfiguration,
   OrganizationEnrichmentConfiguration,
+  IOpenSearchConfig,
 } from './configTypes'
 
 // TODO-kube
@@ -107,6 +108,8 @@ export const EAGLE_EYE_CONFIG: EagleEyeConfiguration = config.get<EagleEyeConfig
 
 export const UNLEASH_CONFIG: UnleashConfiguration = config.get<UnleashConfiguration>('unleash')
 
+export const OPENSEARCH_CONFIG: IOpenSearchConfig = config.get<IOpenSearchConfig>('opensearch')
+
 export const STACKEXCHANGE_CONFIG: StackExchangeConfiguration =
   config.get<StackExchangeConfiguration>('stackexchange') ?? {
     key: process.env.STACKEXCHANGE_KEY,

backend/src/database/repositories/IRepositoryOptions.ts

@@ -10,4 +10,5 @@ export interface IRepositoryOptions {
   database: any
   transaction?: any
   bypassPermissionValidation?: any
+  opensearch?: any
 }

backend/src/database/repositories/memberRepository.ts

@@ -5,7 +5,10 @@ import {
   ALL_PLATFORM_TYPES,
   MemberAttributeType,
   PlatformType,
+  OpenSearchIndex,
 } from '@crowd/types'
+
+import { FieldTranslatorFactory, OpensearchQueryParser } from '@crowd/opensearch'
 import { KUBE_MODE, SERVICE } from '../../conf'
 import { ServiceType } from '../../conf/configTypes'
 import Error404 from '../../errors/Error404'
@@ -1297,6 +1300,72 @@ where m."deletedAt" is null
     }
   }
 
+  static async findAndCountAllOpensearch(
+    {
+      filter = {} as any,
+      limit = 20,
+      offset = 0,
+      orderBy = 'joinedAt_DESC',
+      countOnly = false,
+      attributesSettings = [] as AttributeData[],
+    },
+    options: IRepositoryOptions,
+  ): Promise<PageData<any>> {
+    const tenant = SequelizeRepository.getCurrentTenant(options)
+
+    const translator = FieldTranslatorFactory.getTranslator(
+      OpenSearchIndex.MEMBERS,
+      attributesSettings,
+      [
+        'default',
+        'custom',
+        'enrichment',
+        ...(await TenantRepository.getAvailablePlatforms(options.currentTenant.id, options)).map(
+          (p) => p.platform,
+        ),
+      ],
+    )
+
+    const parsed = OpensearchQueryParser.parse(
+      { filter, limit, offset, orderBy },
+      OpenSearchIndex.MEMBERS,
+      translator,
+    )
+
+    // add tenant filter to parsed query
+    parsed.query.bool.must.push({
+      term: {
+        uuid_tenantId: tenant.id,
+      },
+    })
+
+    const countResponse = await options.opensearch.count({
+      index: OpenSearchIndex.MEMBERS,
+      body: { query: parsed.query },
+    })
+
+    if (countOnly) {
+      return {
+        rows: [],
+        count: countResponse.body.count,
+        limit,
+        offset,
+      }
+    }
+
+    const response = await options.opensearch.search({
+      index: OpenSearchIndex.MEMBERS,
+      body: parsed,
+    })
+
+    // const translated = response.body.hits.hits[0]._source
+    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,