Merge branch 'main' of https://github.com/CrowdDotDev/crowd.dev into CM-2082

Author: Sameh16

services/apps/cache_worker/src/activities.ts

@@ -1,7 +1,7 @@
 import {
   checkOrganizationExists,
   dropOrgIdFromRedis,
-  getOrgIdsFromRedis,
+  getOrganizationIdsFromRedis,
   syncOrganization,
 } from './activities/computeAggs/organization'
 import {
@@ -44,7 +44,7 @@ export {
   getActivePlatforms,
   findNewActivityPlatforms,
   updateMemberMergeSuggestionsLastGeneratedAt,
-  getOrgIdsFromRedis,
+  getOrganizationIdsFromRedis,
   dropOrgIdFromRedis,
   checkOrganizationExists,
   syncOrganization,

services/apps/cache_worker/src/activities/computeAggs/organization.ts

@@ -5,18 +5,22 @@ import { OrganizationSyncService } from '@crowd/opensearch'
 import { svc } from '../../main'
 
 interface IScanResult {
-  cursor: string
   organizationIds: string[]
+  totalSize: number
 }
 
-export async function getOrgIdsFromRedis(cursor = '0', count = 100): Promise<IScanResult> {
+export async function getOrganizationIdsFromRedis(batchSize = 100): Promise<IScanResult> {
   try {
-    const result = await svc.redis.sScan('organizationIdsForAggComputation', Number(cursor), {
-      COUNT: count,
-    })
-    return { organizationIds: result.members, cursor: result.cursor.toString() }
+    const totalSize = await svc.redis.sCard('organizationIdsForAggComputation')
+    const organizationIds = await svc.redis.sendCommand<string[]>([
+      'SRANDMEMBER',
+      'organizationIdsForAggComputation',
+      batchSize.toString(),
+    ])
+
+    return { organizationIds, totalSize }
   } catch (e) {
-    this.log.error(e, 'Failed to get organization IDs from Redis!')
+    svc.log.error(e, 'Failed to get organization IDs from Redis!')
     throw e
   }
 }
@@ -33,7 +37,7 @@ export async function checkOrganizationExists(orgId: string): Promise<boolean> {
     const results = await repo.checkOrganizationsExists([orgId])
     exists = results.length > 0
   } catch (e) {
-    this.log.error(e, 'Failed to check if organization exists!')
+    svc.log.error(e, 'Failed to check if organization exists!')
   }
 
   return exists

services/apps/cache_worker/src/schedules/computeOrgAggsDaily.ts

@@ -1,12 +1,12 @@
 import { ScheduleAlreadyRunning, ScheduleOverlapPolicy } from '@temporalio/client'
 
 import { svc } from '../main'
-import { dailyGetAndComputeOrgAggs } from '../workflows'
+import { spawnOrganizationAggregatesComputation } from '../workflows'
 
 export const scheduleComputeOrgAggsDaily = async () => {
   try {
     await svc.temporal.schedule.create({
-      scheduleId: 'compute-org-aggs-daily',
+      scheduleId: 'computeOrgAggsDaily',
       spec: {
         cronExpressions: ['0 8 * * *'],
       },
@@ -16,9 +16,8 @@ export const scheduleComputeOrgAggsDaily = async () => {
       },
       action: {
         type: 'startWorkflow',
-        workflowType: dailyGetAndComputeOrgAggs,
+        workflowType: spawnOrganizationAggregatesComputation,
         taskQueue: 'cache',
-        workflowExecutionTimeout: '5 minutes',
       },
     })
   } catch (err) {

services/apps/cache_worker/src/workflows.ts

@@ -1,11 +1,11 @@
-import { computeOrgAggsAndUpdate } from './workflows/compute-orgs-agg/computeOrgAggsAndUpdate'
-import { dailyGetAndComputeOrgAggs } from './workflows/compute-orgs-agg/getAndComputeOrgAggs'
+import { processOrganizationAggregates } from './workflows/processOrganizationAggregates'
 import { refreshDashboardCache } from './workflows/refreshDashboardCache'
 import { spawnDashboardCacheRefresh } from './workflows/spawnDashboardCacheRefresh'
+import { spawnOrganizationAggregatesComputation } from './workflows/spawnOrganizationAggregatesComputation'
 
 export {
   spawnDashboardCacheRefresh,
   refreshDashboardCache,
-  dailyGetAndComputeOrgAggs,
-  computeOrgAggsAndUpdate,
+  spawnOrganizationAggregatesComputation,
+  processOrganizationAggregates,
 }

services/apps/cache_worker/src/workflows/compute-orgs-agg/getAndComputeOrgAggs.ts

@@ -1,11 +1,11 @@
 import { proxyActivities } from '@temporalio/workflow'
 
-import * as activities from '../../activities/computeAggs/organization'
-import { IProcessComputeOrgAggs } from '../../types'
+import * as activities from '../activities/computeAggs/organization'
+import { IProcessComputeOrgAggs } from '../types'
 
-const activity = proxyActivities<typeof activities>({ startToCloseTimeout: '10 minutes' })
+const activity = proxyActivities<typeof activities>({ startToCloseTimeout: '30 minutes' })
 
-export async function computeOrgAggsAndUpdate(args: IProcessComputeOrgAggs): Promise<void> {
+export async function processOrganizationAggregates(args: IProcessComputeOrgAggs): Promise<void> {
   const orgId = args.organizationId
 
   const orgExists = await activity.checkOrganizationExists(orgId)

services/apps/cache_worker/src/workflows/compute-orgs-agg/computeOrgAggsAndUpdate.ts services/apps/cache_worker/src/workflows/processOrganizationAggregates.ts

@@ -0,0 +1,54 @@
+import {
+  ChildWorkflowCancellationType,
+  ParentClosePolicy,
+  continueAsNew,
+  executeChild,
+  proxyActivities,
+  workflowInfo,
+} from '@temporalio/workflow'
+
+import * as activities from '../activities/computeAggs/organization'
+
+import { processOrganizationAggregates } from './processOrganizationAggregates'
+
+const activity = proxyActivities<typeof activities>({ startToCloseTimeout: '15 minutes' })
+
+/*
+spawnOrganizationAggregatesComputation is a Temporal workflow that:
+  - [Activity]: Get organization IDs from Redis.
+  - [Child Workflow]: Re-compute and update aggregates for each organization 
+    in batches of 100. Child workflows run independently and won't be 
+    cancelled if the parent workflow stops.
+*/
+export async function spawnOrganizationAggregatesComputation(): Promise<void> {
+  const { organizationIds, totalSize } = await activity.getOrganizationIdsFromRedis()
+
+  if (!totalSize) {
+    console.log('No organizations found - finishing workflow!')
+    return
+  }
+
+  console.log(`Found ${totalSize} organizations for aggs computation!`)
+
+  const info = workflowInfo()
+
+  await Promise.all(
+    organizationIds.map((organizationId) => {
+      return executeChild(processOrganizationAggregates, {
+        workflowId: `${info.workflowId}/${organizationId}`,
+        cancellationType: ChildWorkflowCancellationType.ABANDON,
+        parentClosePolicy: ParentClosePolicy.PARENT_CLOSE_POLICY_ABANDON,
+        retry: {
+          backoffCoefficient: 2,
+          initialInterval: 2 * 1000,
+          maximumInterval: 30 * 1000,
+          maximumAttempts: 3,
+        },
+        args: [{ organizationId }],
+      })
+    }),
+  )
+
+  // Continue with the next batch
+  await continueAsNew<typeof spawnOrganizationAggregatesComputation>()
+}

services/apps/cache_worker/src/workflows/spawnOrganizationAggregatesComputation.ts

@@ -1,4 +1,6 @@
+import { partition } from '@crowd/common'
 import {
+  IActivityRelationsCreateData,
   createOrUpdateRelations,
   getActivityRelationsSortedByTimestamp,
 } from '@crowd/data-access-layer'
@@ -49,23 +51,30 @@ export async function getActivitiesToCopy(latestSyncedActivityTimestamp: string,
 }
 
 export async function createRelations(activitiesRedisKey): Promise<void> {
-  const activities = await getActivitiyDataFromRedis(activitiesRedisKey)
-  const promises = activities.map(async (activity) =>
-    createOrUpdateRelations(pgpQx(svc.postgres.writer.connection()), {
-      activityId: activity.id,
-      memberId: activity.memberId,
-      platform: activity.platform,
-      segmentId: activity.segmentId,
-      username: activity.username,
-      conversationId: activity.conversationId,
-      objectMemberId: activity.objectMemberId,
-      objectMemberUsername: activity.objectMemberUsername,
-      organizationId: activity.organizationId,
-      parentId: activity.parentId,
-    }),
-  )
+  const activities: IActivityRelationsCreateData[] =
+    await getActivitiyDataFromRedis(activitiesRedisKey)
+
+  const chunkSize = 1000
+  const activityChunks = partition(activities, chunkSize)
 
-  await Promise.all(promises)
+  for (const chunk of activityChunks) {
+    const promises = chunk.map((activity) =>
+      createOrUpdateRelations(pgpQx(svc.postgres.writer.connection()), {
+        activityId: activity.id,
+        memberId: activity.memberId,
+        platform: activity.platform,
+        segmentId: activity.segmentId,
+        username: activity.username,
+        conversationId: activity.conversationId,
+        objectMemberId: activity.objectMemberId,
+        objectMemberUsername: activity.objectMemberUsername,
+        organizationId: activity.organizationId,
+        parentId: activity.parentId,
+      }),
+    )
+
+    await Promise.all(promises)
+  }
 }
 
 export async function saveActivityDataToRedis(key: string, activities): Promise<void> {

services/apps/script_executor_worker/src/activities/populate-activity-relations/index.ts

@@ -22,20 +22,34 @@ export async function populateActivityRelations(
       args.latestSyncedActivityTimestamp || (await activity.getLatestSyncedActivityTimestamp())
   }
 
-  const { activitiesLength, activitiesRedisKey, lastTimestamp } =
-    await activity.getActivitiesToCopy(
-      latestSyncedActivityTimestamp ?? undefined,
-      BATCH_SIZE_PER_RUN,
-    )
+  let { activitiesLength, activitiesRedisKey, lastTimestamp } = await activity.getActivitiesToCopy(
+    latestSyncedActivityTimestamp ?? undefined,
+    BATCH_SIZE_PER_RUN,
+  )
 
   if (activitiesLength === 0) {
     return
   }
 
-  if (activitiesLength < BATCH_SIZE_PER_RUN) {
-    if (lastTimestamp === args.latestSyncedActivityTimestamp) {
+  if (lastTimestamp === args.latestSyncedActivityTimestamp) {
+    if (activitiesLength < BATCH_SIZE_PER_RUN) {
       return
     }
+
+    // All activities returned have the same timestamp, we need a bigger batch size
+    // then the default one to pass this point
+    let batchSizeMultiplier = 2
+
+    while (lastTimestamp === args.latestSyncedActivityTimestamp) {
+      const result = await activity.getActivitiesToCopy(
+        lastTimestamp,
+        BATCH_SIZE_PER_RUN * batchSizeMultiplier,
+      )
+      activitiesLength = result.activitiesLength
+      activitiesRedisKey = result.activitiesRedisKey
+      lastTimestamp = result.lastTimestamp
+      batchSizeMultiplier += 1
+    }
   }
 
   await activity.createRelations(activitiesRedisKey)

services/apps/script_executor_worker/src/workflows/populateActivityRelations.ts

@@ -1741,11 +1741,26 @@ export async function moveActivityRelationsToAnotherOrganization(
   } while (rowsUpdated === batchSize)
 }
 
+export interface IActivityRelationsCreateData {
+  id: string
+  memberId: string
+  timestamp: string
+  createdAt: string
+  objectMemberId?: string
+  organizationId?: string
+  conversationId?: string
+  parentId?: string
+  segmentId: string
+  platform: string
+  username: string
+  objectMemberUsername?: string
+}
+
 export async function getActivityRelationsSortedByTimestamp(
   qdbConn: DbConnOrTx,
   cursorActivityTimestamp?: string,
   limit = 100,
-) {
+): Promise<IActivityRelationsCreateData[]> {
   let cursorQuery = ''
 
   if (cursorActivityTimestamp) {