Merge branch 'main' into crowd-linux

CrowdDotDev · Oct 16, 2023 · 638eb74 · 638eb74
2 parents 387c569 + fa96db5
commit 638eb74
Show file tree

Hide file tree

Showing 5 changed files with 31 additions and 16 deletions.
diff --git a/services/apps/data_sink_worker/src/bin/process-results.ts b/services/apps/data_sink_worker/src/bin/process-results.ts
@@ -1,11 +1,12 @@
-import { DB_CONFIG, REDIS_CONFIG, SQS_CONFIG } from '../conf'
+import { DB_CONFIG, REDIS_CONFIG, SQS_CONFIG, SENTIMENT_CONFIG } from '../conf'
 import DataSinkRepository from '../repo/dataSink.repo'
 import DataSinkService from '../service/dataSink.service'
 import { DbStore, getDbConnection } from '@crowd/database'
 import { getServiceTracer } from '@crowd/tracing'
 import { getServiceLogger } from '@crowd/logging'
 import { getRedisClient } from '@crowd/redis'
 import { NodejsWorkerEmitter, SearchSyncWorkerEmitter, getSqsClient } from '@crowd/sqs'
+import { initializeSentimentAnalysis } from '@crowd/sentiment'
 
 const tracer = getServiceTracer()
 const log = getServiceLogger()
@@ -23,8 +24,12 @@ setImmediate(async () => {
   const sqsClient = getSqsClient(SQS_CONFIG())
   const redisClient = await getRedisClient(REDIS_CONFIG())
 
+  initializeSentimentAnalysis(SENTIMENT_CONFIG())
+
   const nodejsWorkerEmitter = new NodejsWorkerEmitter(sqsClient, tracer, log)
+  await nodejsWorkerEmitter.init()
   const searchSyncWorkerEmitter = new SearchSyncWorkerEmitter(sqsClient, tracer, log)
+  await searchSyncWorkerEmitter.init()
 
   const dbConnection = await getDbConnection(DB_CONFIG())
   const store = new DbStore(log, dbConnection)

diff --git a/services/apps/data_sink_worker/src/repo/member.repo.ts b/services/apps/data_sink_worker/src/repo/member.repo.ts
@@ -152,26 +152,30 @@ export default class MemberRepository extends RepositoryBase<MemberRepository> {
       },
     })
 
+    const updatedAt = new Date()
+
     const prepared = RepositoryBase.prepare(
       {
         ...data,
         ...(data?.weakIdentities &&
           data?.weakIdentities?.length > 0 && {
             weakIdentities: JSON.stringify(data.weakIdentities),
           }),
-        updatedAt: new Date(),
+        updatedAt,
       },
       dynamicColumnSet,
     )
     const query = this.dbInstance.helpers.update(prepared, dynamicColumnSet)
 
-    const condition = this.format('where id = $(id) and "tenantId" = $(tenantId)', {
-      id,
-      tenantId,
-    })
-    const result = await this.db().result(`${query} ${condition}`)
-
-    this.checkUpdateRowCount(result.rowCount, 1)
+    const condition = this.format(
+      'where id = $(id) and "tenantId" = $(tenantId) and "updatedAt" < $(updatedAt)',
+      {
+        id,
+        tenantId,
+        updatedAt,
+      },
+    )
+    await this.db().result(`${query} ${condition}`)
   }
 
   public async getIdentities(memberId: string, tenantId: string): Promise<IMemberIdentity[]> {

diff --git a/services/apps/data_sink_worker/src/repo/organization.repo.ts b/services/apps/data_sink_worker/src/repo/organization.repo.ts
@@ -458,24 +458,26 @@ export class OrganizationRepository extends RepositoryBase<OrganizationRepositor
         table: 'organizations',
       },
     })
+    const updatedAt = new Date()
     const prepared = RepositoryBase.prepare(
       {
         ...data,
         ...(data?.weakIdentities &&
           data?.weakIdentities?.length > 0 && {
             weakIdentities: JSON.stringify(data.weakIdentities),
           }),
-        updatedAt: new Date(),
+        updatedAt,
       },
       dynamicColumnSet,
     )
 
     const query = this.dbInstance.helpers.update(prepared, dynamicColumnSet)
-    const condition = this.format('where id = $(id)', { id })
-
-    const result = await this.db().result(`${query} ${condition}`)
+    const condition = this.format('where id = $(id) and "updatedAt" < $(updatedAt)', {
+      id,
+      updatedAt,
+    })
 
-    this.checkUpdateRowCount(result.rowCount, 1)
+    await this.db().result(`${query} ${condition}`)
   }
 
   public async addIdentity(

diff --git a/services/libs/integrations/src/integrations/reddit/api/getMoreComments.ts b/services/libs/integrations/src/integrations/reddit/api/getMoreComments.ts
@@ -37,7 +37,7 @@ async function getMoreComments(
       url: `http://oauth.reddit.com/api/morechildren?api_type=json`,
       params: {
         depth: 99,
-        link_id: input.postId,
+        link_id: `t3_${input.postId}`,
         children: input.children,
       },
       headers: {

diff --git a/services/libs/integrations/src/integrations/reddit/processStream.ts b/services/libs/integrations/src/integrations/reddit/processStream.ts
@@ -156,7 +156,11 @@ const processMoreCommentStream: ProcessStreamHandler = async (ctx) => {
     ctx,
   )
 
-  const comments = response.json.data.things
+  const comments = response?.json?.data?.things
+
+  if (!comments) {
+    return
+  }
 
   if (comments.length === 0) {
     return