refactor code and bump chunk size

Author: skwowet

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

@@ -1,22 +1,18 @@
 import { continueAsNew, proxyActivities } from '@temporalio/workflow'
 
-import * as commonActivities from '../activities/common'
-import * as activities from '../activities/fix-deleted-member-org-affilations'
-import * as syncActivities from '../activities/sync/member'
+import * as activities from '../activities'
 import { IFixDeletedMemberOrgAffilationsArgs } from '../types'
 
-const activity = proxyActivities<typeof activities>({
+const {
+  getProcessedMemberOrgAffiliations,
+  calculateMemberAffiliations,
+  syncMembersBatch,
+  queueOrgForAggComputation,
+  deleteProcessedMemberOrgAffiliations,
+} = proxyActivities<typeof activities>({
   startToCloseTimeout: '45 minutes',
 })
 
-const syncActivity = proxyActivities<typeof syncActivities>({
-  startToCloseTimeout: '30 minutes',
-})
-
-const commonActivity = proxyActivities<typeof commonActivities>({
-  startToCloseTimeout: '30 minutes',
-})
-
 export async function fixDeletedMemberOrgAffilations(
   args: IFixDeletedMemberOrgAffilationsArgs,
 ): Promise<void> {
@@ -27,29 +23,29 @@ export async function fixDeletedMemberOrgAffilations(
   )
 
   // Find affected memberId and orgId
-  const affectedMembers = await activity.getProcessedMemberOrgAffiliations(BATCH_SIZE)
+  const affectedMembers = await getProcessedMemberOrgAffiliations(BATCH_SIZE)
 
   if (affectedMembers.length === 0) {
     console.log('No processed member org affiliations found!')
     return
   }
 
-  const CHUNK_SIZE = 10
+  const CHUNK_SIZE = 25
   for (let i = 0; i < affectedMembers.length; i += CHUNK_SIZE) {
     const chunk = affectedMembers.slice(i, i + CHUNK_SIZE)
     await Promise.all(
       chunk.map(async ({ memberId, organizationId }) => {
         // Calculate affiliation
-        await activity.calculateMemberAffiliations(memberId)
+        await calculateMemberAffiliations(memberId)
 
         // Sync member
-        await syncActivity.syncMembersBatch([memberId], true)
+        await syncMembersBatch([memberId], true)
 
         // Add orgId to redisCache
         // It will be picked up by the spawnOrganizationAggregatesComputation workflow
-        await commonActivity.queueOrganizationForAggComputation(organizationId)
+        await queueOrgForAggComputation(organizationId)
 
-        await activity.deleteProcessedMemberOrgAffiliations(memberId, organizationId)
+        await deleteProcessedMemberOrgAffiliations(memberId, organizationId)
       }),
     )
   }